Replace df.break() JSON sentinel with typed NodeError (#148)

[crprashant]

Summary

Refactors df.break() propagation from an in-band JSON sentinel string
({"__break__": true, "value": ...}) to a typed NodeError::Break enum, so Rust's
? operator auto-propagates break through every compound node (THEN / IF / JOIN / RACE
and the sub-orchestration boundary). The loop node is now the sole break catcher,
which turns "forgetting to propagate a break" from a recurring silent-bug class into a
compile error.

Fixes #148.

Changes

• Typed control flow: new NodeError { Break, Failure } + NodeResult;
  From<String> / From<&str> map plain errors to Failure so activity/helper calls
  keep their ? ergonomics.
• Subtree boundary: break now crosses the sub-orchestration boundary via a typed
  control: Option<SubtreeControl> field on SubtreeEnvelope instead of a sentinel
  smuggled in result. The Option makes an absent field (an envelope written by a
  <= 0.2.2 binary) distinguishable from a new-binary Some(Normal), so the legacy
  sentinel fallback runs only on genuinely old envelopes — a fresh JOIN/RACE branch
  result can never impersonate a break.
• In-flight-upgrade safety: when control is absent (None), parse_subtree_envelope
  re-raises a legacy {"__break__"} sentinel (written by <= 0.2.2 binaries) as
  NodeError::Break, so a break persisted by an older binary is not silently swallowed
  across a binary upgrade.
• Docs: documented the Break-is-control-flow contract on the execute() entry point
  and in docs/ARCHITECTURE.md.
• Tests: 11 new unit tests for envelope parsing (legacy + typed, incl. a guard that a
  new-binary Normal result shaped like the legacy sentinel is not re-raised as a break);
  tightened the break-in-join-race E2E to assert the exact break value.

Validation

• cargo fmt -p pg_durable -- --check — clean
• cargo build --features pg17 — clean
• cargo clippy --features pg17 — clean (only the pre-existing extract_host warning)
• ./scripts/test-unit.sh — 156 passed, 0 failed
• ./scripts/test-e2e-local.sh — 32 passed, 0 failed

Upgrade notes

The only behavior change for already-running instances affects a misuse case: an
orchestration started under <= 0.2.2 that hits a top-level df.break() (one used
outside any df.loop()) previously completed carrying the sentinel value; under the new
binary that uncaught break surfaces as a clear failure ("df.break() was called outside of a
loop"). JOIN/RACE-in-loop breaks persisted by an old binary still unwind correctly via the
legacy-sentinel fallback. No schema migration is required.

Review

Reviewed via a multi-specialist pass (correctness, edge-cases, architecture,
maintainability, testing) plus synthesis. All three must-fix findings are addressed in
this PR; remaining should-fix / consider items were deferred per the synthesis
dispositions.

Notes

A pre-existing, unrelated bug exists where a JOIN/RACE inside a loop running

= 2 iterations can stall due to continue_as_new sub-orchestration id collisions. It is
independent of this change and is tracked separately in #230.

[pinodeca]

Reviewed with Opus 4.8

One real finding

The legacy fallback re-introduces the exact sentinel-collision the refactor claims to eliminate, for the JOIN/RACE subtree path.

In parse_subtree_envelope:

SubtreeControl::Normal => match parse_legacy_break_sentinel(&envelope.result) {
    Some(value) => Err(NodeError::Break(value)),
    None => Ok(envelope.result),
},

Because control uses #[serde(default)], a new-binary normal branch (which writes control = Normal) is indistinguishable at parse time from an old-binary envelope (no field). So parse_legacy_break_sentinel runs on every normal subtree result, including fresh ones. A JOIN/RACE branch whose legitimate SQL result is shaped like {"__break__": true, "value": ...} would be falsely re-raised as a Break and exit the loop.

Issue #148 explicitly lists "user payloads cannot impersonate control flow" as benefit #2 — this is honored for the top-level/THEN/IF paths but silently reintroduced for subtree branch results. Low probability, but it's the same bug class, and it's a data-driven control-flow hijack rather than a crash.

Concrete fix — make field presence distinguishable instead of defaulting:

struct SubtreeEnvelope {
    #[serde(default)]
    control: Option<SubtreeControl>,  // None = pre-#148 binary
    result: String,
    results: HashMap<String, String>,
}

match envelope.control {
    Some(SubtreeControl::Break) => Err(NodeError::Break(envelope.result)),
    Some(SubtreeControl::Normal) => Ok(envelope.result), // new binary: NO legacy check
    None => match parse_legacy_break_sentinel(&envelope.result) {  // only old envelopes
        Some(value) => Err(NodeError::Break(value)),
        None => Ok(envelope.result),
    },
}

This keeps in-flight-upgrade safety (old envelopes still decode) while closing the collision for all envelopes written by the new binary. The existing tests would need their control arg adjusted, but the test helper already supports omitting the field for the legacy case.

Minor notes

• Behavior change on resume: an orchestration started under ≤0.2.2 that later hits a top-level break will now fail instead of complete if it resumes under the new binary. This only affects the misuse case the PR intends to fix, but it's worth a one-line mention in the PR's upgrade notes.
• The two status-recording arms (Ok and Err(Break)) in execute_function_node_with_vars build a near-identical completed update_node_status payload. These could be collapsed by extracting the value first, then having a single failed vs completed branch. ~12 duplicated lines.

Verdict

Solid, well-motivated refactor that achieves its primary goal with good attention to replay determinism and test coverage. I'd treat the SubtreeControl collision via #[serde(default)] as a should-fix before merge (simple Option change), since leaving it half-undercuts the stated security benefit. Everything else is polish.

[crprashant]

@pinodeca thanks for the careful review — all three points are addressed in 5f2e9a6.

Main finding (legacy fallback ran on every normal result). You're right: with #[serde(default)] defaulting to Normal, a new-binary normal envelope was indistinguishable from a pre-#148 one, so parse_legacy_break_sentinel ran on every normal subtree result and re-opened the exact payload-impersonates-break collision this PR set out to close. Fixed by making the field control: Option<SubtreeControl>:

• None → envelope written by a <= 0.2.2 binary (no control field) → run the legacy sentinel fallback.
• Some(SubtreeControl::Normal) → new binary → pass through untouched, no legacy check.
• Some(SubtreeControl::Break) → typed break.

I also dropped Default/#[default] from SubtreeControl since the default is no longer used. The wire format is unchanged — Some(Normal) still serializes to "control":"Normal" and an absent field still deserializes (now to None) — so replay across this change is byte-identical; the only behavior change is closing the collision. Added a regression test, envelope_new_format_normal_with_sentinel_shaped_result_is_not_reraised, which asserts a Some(Normal) envelope whose genuine result is shaped like {"__break__": true, ...} passes through as Ok rather than being re-raised.

Minor — duplicated status-recording arms. Collapsed the three near-identical arms in execute_function_node_with_vars into a single (status, result) match feeding one update_node_status schedule. Kept it to exactly one scheduled activity in every case so the recorded orchestration history stays identical for replay.

Minor — resume behavior change. Added an ## Upgrade notes section to the PR description: an orchestration started under <= 0.2.2 that hits a top-level break now surfaces a clear failure instead of completing with the sentinel value (JOIN/RACE-in-loop breaks from old binaries still unwind via the legacy fallback).

Validation after the change: cargo fmt --check clean, clippy clean (only the pre-existing extract_host warning), unit 156/0, e2e 32/0.

[pinodeca]

Thanks for addressing the issues raised - and for the fast turnaround!

LGTM - let's merge it!