From f993a39a05b6505be72ef23097c01508b5bdd6e7 Mon Sep 17 00:00:00 2001 From: Fantix King Date: Wed, 23 Sep 2026 20:56:51 -0400 Subject: [PATCH] workflow: check remaining history before exit Catch replay mismatches before a run ends, preparing for hook lifecycle writes in #410. --- .../20260928-terminal-replay-check.bugfix.md | 1 + .../tests/unit/test_workflow_determinism.py | 257 +++++++++++++++++- .../vercel/workflow/_internal/runtime.py | 63 +++-- 3 files changed, 304 insertions(+), 17 deletions(-) create mode 100644 changes/vercel-workflow/20260928-terminal-replay-check.bugfix.md diff --git a/changes/vercel-workflow/20260928-terminal-replay-check.bugfix.md b/changes/vercel-workflow/20260928-terminal-replay-check.bugfix.md new file mode 100644 index 00000000..8f54c28f --- /dev/null +++ b/changes/vercel-workflow/20260928-terminal-replay-check.bugfix.md @@ -0,0 +1 @@ +Detect replay divergence in remaining recorded history when a workflow returns or raises. diff --git a/src/vercel-workflow/tests/unit/test_workflow_determinism.py b/src/vercel-workflow/tests/unit/test_workflow_determinism.py index 116b732b..3da5bf97 100644 --- a/src/vercel-workflow/tests/unit/test_workflow_determinism.py +++ b/src/vercel-workflow/tests/unit/test_workflow_determinism.py @@ -29,7 +29,7 @@ async def _greet(*, name: str) -> str: @dataclasses.dataclass -class _HookPayload: +class _HookPayload(core.BaseHook): value: str @@ -400,6 +400,62 @@ async def _reraising() -> str: raise runtime.NondeterminismError("surfaced by the body") from None +@_run_registry.workflow +async def _unawaited_record(fail: bool = False) -> str: + ctx = runtime.WorkflowOrchestratorContext.current() + pending = ctx.run_step(_record, name="a") + # Completing historical operations after the body finishes must not start + # user continuations. The isolated loop has already shut down by then. + pending.add_done_callback(lambda _: ctx.run_step(_record, name="unexpected continuation")) + if fail: + raise ValueError("body failed") + return "done" + + +@_run_registry.workflow +async def _unawaited_wait_and_attributes() -> str: + ctx = runtime.WorkflowOrchestratorContext.current() + pending = [ + ctx.run_wait(datetime(2026, 1, 1, tzinfo=timezone.utc)), + ctx.set_attributes([w.AttributeChange(key="phase", value="done")], allow_reserved=False), + ] + for future in pending: + future.add_done_callback(lambda _: ctx.run_step(_record, name="unexpected continuation")) + return "done" + + +@_run_registry.workflow +async def _operation_from_task_cleanup() -> str: + async def background() -> None: + try: + await asyncio.Future() + finally: + runtime.WorkflowOrchestratorContext.current().run_step(_record, name="a") + + asyncio.create_task(background()) + await asyncio.sleep(0) + return "done" + + +@_run_registry.workflow +async def _no_operations() -> str: + return "done" + + +@_run_registry.workflow +async def _unawaited_hook(token: str) -> str: + hook = _HookPayload.wait(token=token) + + async def consume() -> None: + await hook + await _record(name="unexpected continuation") + + asyncio.create_task(consume()) + asyncio.create_task(hook.get_conflict()) + await asyncio.sleep(0) + return "done" + + def _running_run(workflow_id: str) -> w.WorkflowRun: now = datetime(2026, 1, 1, tzinfo=timezone.utc) return w.NonFinalWorkflowRun( @@ -447,6 +503,205 @@ async def test_nondeterminism_cannot_be_suppressed_by_the_body() -> None: ctx.run_workflow(_running_run(_suppressing.workflow_id)) +@pytest.mark.parametrize("fail", [False, True]) +@pytest.mark.parametrize("outcome", ["pending", "completed", "failed"]) +async def test_terminal_replay_applies_unawaited_operation_history(fail, outcome) -> None: + cid = f"step_{_context([]).generate_ulid()}" + events = [_created(_record, cid)] + if outcome == "completed": + events.append(_completed(cid, "recorded result")) + elif outcome == "failed": + events.append( + w.StepFailedEventData( + error=PLAIN_ENCODER.encode_error(ValueError("step failed")) + ).into_event(cid) + ) + ctx = runtime.WorkflowOrchestratorContext( + events, run_id="wrun_test", seed="wrun_test", started_at=0, registry=_run_registry + ) + run = _running_run(_unawaited_record.workflow_id).model_copy( + update={"input": PLAIN_ENCODER.encode(ser.argument_array((fail,), {}))} + ) + + if fail: + with pytest.raises(ValueError, match="body failed"): + ctx.run_workflow(run) + else: + assert ctx.run_workflow(run) == PLAIN_ENCODER.encode("done") + + assert ctx.replay_index == len(events) + assert not ctx.suspended + assert set(ctx.suspensions) == ({cid} if outcome == "pending" else set()) + if outcome == "pending": + assert ctx.suspensions[cid].has_created_event + + +@pytest.mark.parametrize("diverged", [False, True]) +async def test_terminal_replay_applies_waits_and_validates_attributes(diverged) -> None: + probe = _context([]) + wait_id = f"wait_{probe.generate_ulid()}" + attr_id = f"attr_{probe.generate_ulid()}" + events: list[w.Event] = [ + w.WaitCreatedEventData(resume_at=datetime(2026, 1, 1, tzinfo=timezone.utc)).into_event( + wait_id + ), + w.WaitCompletedEvent(correlation_id=wait_id), + w.AttrSetEventData( + changes=[w.AttributeChange(key="phase", value="other" if diverged else "done")], + writer=w.WorkflowAttributeWriter(), + ).into_event(attr_id), + ] + ctx = runtime.WorkflowOrchestratorContext( + events, run_id="wrun_test", seed="wrun_test", started_at=0, registry=_run_registry + ) + run = _running_run(_unawaited_wait_and_attributes.workflow_id) + + if diverged: + with pytest.raises(runtime.NondeterminismError, match="recorded attributes"): + ctx.run_workflow(run) + else: + assert ctx.run_workflow(run) == PLAIN_ENCODER.encode("done") + + assert ctx.replay_index == len(events) + assert not ctx.suspensions + await asyncio.sleep(0) + + +async def test_terminal_replay_runs_after_task_cleanup() -> None: + cid = f"step_{_context([]).generate_ulid()}" + events = [_created(_record, cid), _completed(cid, "recorded result")] + ctx = runtime.WorkflowOrchestratorContext( + events, run_id="wrun_test", seed="wrun_test", started_at=0, registry=_run_registry + ) + + assert ctx.run_workflow(_running_run(_operation_from_task_cleanup.workflow_id)) == ( + PLAIN_ENCODER.encode("done") + ) + assert ctx.replay_index == len(events) + assert not ctx.suspended + assert not ctx.suspensions + + +@pytest.mark.parametrize("fail", [False, True]) +async def test_terminal_replay_validates_unawaited_operation_input(fail) -> None: + cid = f"step_{_context([]).generate_ulid()}" + event = w.StepCreatedEventData(step_name=_record.name, input=_args(name="other")).into_event( + cid + ) + ctx = runtime.WorkflowOrchestratorContext( + [event], run_id="wrun_test", seed="wrun_test", started_at=0, registry=_run_registry + ) + run = _running_run(_unawaited_record.workflow_id).model_copy( + update={"input": PLAIN_ENCODER.encode(ser.argument_array((fail,), {}))} + ) + + with pytest.raises(runtime.NondeterminismError, match="different arguments"): + ctx.run_workflow(run) + + # Terminal validation runs outside the isolated loop. It must not cancel + # tasks in the caller's event loop when it finds a divergence. + await asyncio.sleep(0) + + +async def test_terminal_replay_rejects_an_operation_the_body_no_longer_declares() -> None: + cid = f"step_{_context([]).generate_ulid()}" + ctx = runtime.WorkflowOrchestratorContext( + [_created(_record, cid)], + run_id="wrun_test", + seed="wrun_test", + started_at=0, + registry=_run_registry, + ) + + with pytest.raises(runtime.NondeterminismError, match="has not registered"): + ctx.run_workflow(_running_run(_no_operations.workflow_id)) + + +@pytest.mark.parametrize("outcome", ["created", "conflict", "received", "disposed"]) +async def test_terminal_replay_applies_hook_history_after_task_cleanup(outcome: str) -> None: + cid = f"hook_{_context([]).generate_ulid()}" + events = [_hook_registration_event(cid, "same-token", conflict=outcome == "conflict")] + if outcome in {"received", "disposed"}: + events.append( + w.HookReceivedEventData( + token="same-token", payload=PLAIN_ENCODER.encode({"value": "recorded payload"}) + ).into_event(cid) + ) + if outcome == "disposed": + events.append(w.HookDisposedEvent(correlation_id=cid)) + ctx = runtime.WorkflowOrchestratorContext( + events, run_id="wrun_test", seed="wrun_test", started_at=0, registry=_run_registry + ) + run = _running_run(_unawaited_hook.workflow_id).model_copy( + update={"input": PLAIN_ENCODER.encode(ser.argument_array(("same-token",), {}))} + ) + + assert ctx.run_workflow(run) == PLAIN_ENCODER.encode("done") + + hook = ctx.hooks[cid] + assert ctx.replay_index == len(events) + assert not ctx.suspended + assert hook.has_created_event == (outcome != "conflict") + assert (hook.conflict_error is not None) == (outcome == "conflict") + assert hook.has_dispose_event == (outcome == "disposed") + assert not hook.buffered_results + assert set(ctx.suspensions) == ({cid} if outcome in {"created", "received"} else set()) + await asyncio.sleep(0) + + +async def test_terminal_hook_disposal_does_not_resolve_a_waiter_on_a_closed_loop() -> None: + ctx = _context([]) + hook_event = ctx.create_hook("same-token", _HookPayload) + cid = hook_event._correlation_id + hook = ctx.hooks[cid] + ctx.events.extend( + [ + _hook_registration_event(cid, "same-token", conflict=False), + w.HookDisposedEvent(correlation_id=cid), + ] + ) + + # A task spawned during shutdown can leave a waiter on the closed loop. + # Keep a callback attached so resolving it would try to schedule user code. + isolated_loop = asyncio.new_event_loop() + try: + future = isolated_loop.create_future() + future.add_done_callback(lambda _: None) + hook.futures.append(future) + finally: + isolated_loop.close() + + ctx._finish_replay() + + assert ctx.replay_index == len(ctx.events) + assert hook.has_created_event + assert hook.has_dispose_event + assert hook.disposed + assert not ctx.suspensions + assert not future.done() + + +@pytest.mark.parametrize("conflict", [False, True], ids=["created", "conflict"]) +async def test_terminal_replay_validates_hook_tokens(conflict: bool) -> None: + cid = f"hook_{_context([]).generate_ulid()}" + ctx = runtime.WorkflowOrchestratorContext( + [_hook_registration_event(cid, "old-token", conflict=conflict)], + run_id="wrun_test", + seed="wrun_test", + started_at=0, + registry=_run_registry, + ) + run = _running_run(_unawaited_hook.workflow_id).model_copy( + update={"input": PLAIN_ENCODER.encode(ser.argument_array(("new-token",), {}))} + ) + + with pytest.raises(runtime.NondeterminismError, match="recorded hook token 'old-token'"): + ctx.run_workflow(run) + + assert not ctx.suspended + await asyncio.sleep(0) + + # --- now(): deterministic clock anchored to replay progress, not list tail ------ diff --git a/src/vercel-workflow/vercel/workflow/_internal/runtime.py b/src/vercel-workflow/vercel/workflow/_internal/runtime.py index 87b9c2c8..59678b70 100644 --- a/src/vercel-workflow/vercel/workflow/_internal/runtime.py +++ b/src/vercel-workflow/vercel/workflow/_internal/runtime.py @@ -922,14 +922,18 @@ def run_workflow(self: Self, workflow_run: w.WorkflowRun) -> bytes | None: token = self._ctx.set(self) try: - result = self.payload_encoder.encode( - obj.codec.dump_return( - _run_isolated( - obj.func(*args, **kwargs), - loop_factory=lambda: loop.WorkflowLoop(workflow=self), + try: + result = self.payload_encoder.encode( + obj.codec.dump_return( + _run_isolated( + obj.func(*args, **kwargs), + loop_factory=lambda: loop.WorkflowLoop(workflow=self), + ) ) ) - ) + finally: + if not self.suspended: + self._finish_replay() except BaseException as ex: if self.resume_exception is not None: # Since resume_exception actually got raised on a @@ -1098,25 +1102,30 @@ async def run_hook_conflict(self, *, correlation_id: str) -> Run[Any] | None: hook.conflict_futures.append(future) return await future - def dispose_hook(self, *, correlation_id: str) -> None: + def dispose_hook(self, *, correlation_id: str, deliver: bool = True) -> None: hook = self.hooks[correlation_id] hook.disposed = True - while hook.futures: - fut = hook.futures.popleft() - if not fut.done(): - fut.set_exception(StopAsyncIteration) + if deliver: + while hook.futures: + fut = hook.futures.popleft() + if not fut.done(): + fut.set_exception(StopAsyncIteration) self.suspensions.pop(correlation_id, None) - def _fail_nondeterminism(self, sus: BaseSuspension | None, exc: Exception) -> None: + def _fail_nondeterminism( + self, sus: BaseSuspension | None, exc: Exception, *, deliver: bool + ) -> None: """Fail the run with a replay-divergence error the body cannot suppress. The diverged suspension may be absent or not be what the body is blocked on, so failing its future alone might never surface anywhere -- and a body that is awaiting it could catch the error. So the exception is also stashed for ``run_workflow`` to raise, and the run is suspended - so nothing else executes. + so nothing else executes. During finalization, raise directly instead. """ self.resume_exception = exc + if not deliver: + raise exc if sus is not None: sus.fail(exc) self.suspend() @@ -1152,6 +1161,13 @@ def resume(self) -> None: # multiple deliveries bunched up before a resume(), which could # lead to mismatches between a recording trace and a replaying # one. + self._replay_next_event(deliver=True) + + def _finish_replay(self) -> None: + while self.replay_index < len(self.events): + self._replay_next_event(deliver=False) + + def _replay_next_event(self, *, deliver: bool) -> None: event = self.events[self.replay_index] self.replay_index += 1 if event.correlation_id in self.hooks: @@ -1169,6 +1185,7 @@ def resume(self) -> None: f"recorded hook token {event_token!r}, but the body now uses " f"{registered_hook.token!r}" ), + deliver=deliver, ) return elif event.correlation_id not in self.suspensions: @@ -1205,6 +1222,7 @@ def resume(self) -> None: f"issues a {_correlation_kind(sus.correlation_id)!r} call. " "The workflow body is non-deterministic." ), + deliver=deliver, ) return self._fail_nondeterminism( @@ -1213,6 +1231,7 @@ def resume(self) -> None: f"workflow replay cannot deliver {slot_id!r}: " "the workflow body has not registered its suspension" ), + deliver=deliver, ) return @@ -1236,6 +1255,7 @@ def resume(self) -> None: f"step {name!r}, but the body now calls {sus.step.name!r} with " "different arguments. The workflow body is non-deterministic." ), + deliver=deliver, ) return sus.has_created_event = True @@ -1247,6 +1267,8 @@ def resume(self) -> None: hook = self.suspensions[event.correlation_id] hook.has_created_event = True if isinstance(hook, Hook): + if not deliver: + return while hook.conflict_futures: future = hook.conflict_futures.popleft() if not future.cancelled(): @@ -1271,14 +1293,17 @@ def resume(self) -> None: f"{recorded_changes!r}, but the body now sets " f"{attr_sus.changes!r}. The workflow body is non-deterministic." ), + deliver=deliver, ) return - if not attr_sus.future.cancelled(): + if deliver and not attr_sus.future.cancelled(): attr_sus.future.set_result(None) case w.StepCompletedEvent(event_data=w.StepCompletedEventData(result=data)): sus = self.suspensions.pop(event.correlation_id) assert isinstance(sus, Suspension) + if not deliver: + return result = ser.hydrate( data, what=f"the result of step {event.correlation_id}", @@ -1295,12 +1320,14 @@ def resume(self) -> None: case w.WaitCompletedEvent(): wait = self.suspensions.pop(event.correlation_id) assert isinstance(wait, Wait) - if not wait.future.cancelled(): + if deliver and not wait.future.cancelled(): wait.future.set_result(None) case w.StepFailedEvent(event_data=w.StepFailedEventData(error=data)): sus = self.suspensions.pop(event.correlation_id) assert isinstance(sus, Suspension) + if not deliver: + return what = f"the error of step {event.correlation_id}" try: failure = ser.hydrate_error(data, what=what, key=self.run_key) @@ -1330,6 +1357,8 @@ def resume(self) -> None: conflicting_hook.conflicting_run = ( Run(conflicting_run_id) if conflicting_run_id else None ) + if not deliver: + return while conflicting_hook.futures: future = conflicting_hook.futures.popleft() if not future.cancelled(): @@ -1357,6 +1386,8 @@ def resume(self) -> None: return assert isinstance(hook, Hook) + if not deliver: + return try: result = ser.hydrate( data, @@ -1370,7 +1401,7 @@ def resume(self) -> None: case w.HookDisposedEvent(): self.hooks[event.correlation_id].has_dispose_event = True - self.dispose_hook(correlation_id=event.correlation_id) + self.dispose_hook(correlation_id=event.correlation_id, deliver=deliver) # ── lazy hook resume ───────────────────────────────────────────────────────