From 6dce6b928fd33eecc6d7b8103d6c454469cf4016 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 10 Oct 2026 20:46:40 +0000 Subject: [PATCH 1/3] Keep condition replay drift recoverable across worker tasks --- docs/sdk-reference.md | 9 +- src/durable_workflow/worker.py | 44 ++++-- src/durable_workflow/workflow.py | 27 ++-- ...dition-fingerprint-drift-task-failure.json | 13 ++ .../condition-key-drift-task-failure.json | 13 ++ tests/integration/test_condition_recovery.py | 131 ++++++++++++++++++ tests/test_condition_replay_drift.py | 104 ++++++++++++++ tests/test_replay_regression_corpus.py | 2 + tests/test_wait_condition.py | 25 ++-- 9 files changed, 327 insertions(+), 41 deletions(-) create mode 100644 tests/fixtures/replay_regressions/condition-fingerprint-drift-task-failure.json create mode 100644 tests/fixtures/replay_regressions/condition-key-drift-task-failure.json create mode 100644 tests/integration/test_condition_recovery.py create mode 100644 tests/test_condition_replay_drift.py diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index 94e2198..5c3de44 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -488,8 +488,13 @@ Use `yield ctx.wait_condition(lambda: self.approved, key="approved", timeout=30)` to wait for signal- or update-mutated workflow state without polling timers by hand. The SDK sends a stable predicate fingerprint with the durable wait command and rejects replay if history records a different wait -key or predicate fingerprint, so condition changes fail visibly instead of -silently resolving a different wait. +key or predicate fingerprint. A mismatch raises `NonDeterministicReplayError` +with the original wait identity, workflow sequence and both definitions. The +worker fails the current task without recording a terminal workflow failure. +Deploy compatible code to resume the original run. Accepted updates also remain +available for retry when their state replay encounters this error. Query replay +reports a query failure with the mismatch as its cause. Exceptions raised by +application predicates remain ordinary workflow failures. Workers fingerprint registered workflow class definitions and advertise those fingerprints during registration. Re-registering the same `worker_id` with a diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index 291bf56..2d032e4 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -69,6 +69,7 @@ DurableWorkflowError, ExternalPayloadError, InvalidArgument, + NonDeterministicReplayError, NonRetryableError, QueryFailed, ServerError, @@ -2502,20 +2503,35 @@ async def _run_workflow_task_core(self, task: dict[str, Any]) -> list[dict[str, update_id = task.get("workflow_update_id") if isinstance(update_id, str) and update_id: - update_command = apply_update( - cls, - history, - start_input, - update_id, - workflow_id=task.get("workflow_id"), - run_id=run_id, - payload_codec=codec, - external_storage=self.external_storage, - external_storage_cache=self.external_storage_cache, - prepare_local_activities=self._prepared_local_activities_supported, - prepare_local_activity_groups=self._prepared_local_activity_groups_supported, - local_activity_cancellation_policies=self._local_activity_cancellation_policies, - ) + try: + update_command = apply_update( + cls, + history, + start_input, + update_id, + workflow_id=task.get("workflow_id"), + run_id=run_id, + payload_codec=codec, + external_storage=self.external_storage, + external_storage_cache=self.external_storage_cache, + prepare_local_activities=self._prepared_local_activities_supported, + prepare_local_activity_groups=self._prepared_local_activity_groups_supported, + local_activity_cancellation_policies=self._local_activity_cancellation_policies, + ) + except NonDeterministicReplayError as exc: + log.exception("workflow update task %s replay failed for run %s", task_id, run_id) + try: + await self.client.fail_workflow_task( + task_id=task_id, + lease_owner=self.worker_id, + workflow_task_attempt=attempt, + message=f"workflow replay failed before update {update_id!r}: {exc}", + failure_type=type(exc).__name__, + stack_trace=traceback.format_exc(), + ) + except Exception as failure_error: + log.warning("failed to report update replay failure: %s", failure_error) + return None command = update_command.to_server_command( self.task_queue, payload_codec=command_codec, diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 1234f10..72d0ff8 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -3067,6 +3067,8 @@ def apply_update( prepare_local_activity_groups=prepare_local_activity_groups, local_activity_cancellation_policies=local_activity_cancellation_policies, ) + except NonDeterministicReplayError: + raise except Exception as exc: return _fail_update_from_exception( update_id, @@ -5144,16 +5146,20 @@ def _apply_condition_wait_receivers(condition_wait_id: str | None) -> None: for _ in range(boundary): _apply_receiver(pending_receivers.pop(0)) - def _condition_wait_mismatch(opened: Mapping[str, Any], cmd: WaitCondition) -> FailWorkflow | None: + def _condition_wait_mismatch( + opened: Mapping[str, Any], cmd: WaitCondition, + ) -> NonDeterministicReplayError | None: + step = recorded_wait_steps[wait_yield_count] + identity = f"condition wait {opened.get('condition_wait_id')!r}" opened_key = opened.get("condition_key") if isinstance(opened_key, str) and opened_key != (cmd.condition_key or ""): - return FailWorkflow( - message=( - "wait_condition key changed during replay: " + return NonDeterministicReplayError( + step.workflow_sequence, "condition wait", step.event_types, + detail=( + f"condition_key_changed: {identity} wait_condition key changed during replay: " f"history has {opened_key!r}, workflow yielded " f"{cmd.condition_key!r}" ), - exception_type="NonDeterministicWaitCondition", ) opened_fingerprint = opened.get("condition_definition_fingerprint") @@ -5161,13 +5167,14 @@ def _condition_wait_mismatch(opened: Mapping[str, Any], cmd: WaitCondition) -> F isinstance(opened_fingerprint, str) and cmd.condition_definition_fingerprint != opened_fingerprint ): - return FailWorkflow( - message=( + return NonDeterministicReplayError( + step.workflow_sequence, "condition wait", step.event_types, + detail=( + f"condition_fingerprint_changed: {identity} " "wait_condition predicate fingerprint changed during replay: " f"history has {opened_fingerprint!r}, workflow yielded " f"{cmd.condition_definition_fingerprint!r}" ), - exception_type="NonDeterministicWaitCondition", ) return None @@ -6655,7 +6662,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None next_value = True continue if next_opened_mismatch is not None: - return _state([next_opened_mismatch]) + raise next_opened_mismatch step = _next_unconsumed_recorded_step() if step is not None: _assert_step_matches(cmd, step) @@ -6707,7 +6714,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None opened = wait_opened[wait_yield_count] mismatch = _condition_wait_mismatch(opened, cmd) if mismatch is not None: - return _state([mismatch]) + raise mismatch opened_id = opened.get("condition_wait_id") if isinstance(opened_id, str): resolution = wait_resolutions.get(opened_id) diff --git a/tests/fixtures/replay_regressions/condition-fingerprint-drift-task-failure.json b/tests/fixtures/replay_regressions/condition-fingerprint-drift-task-failure.json new file mode 100644 index 0000000..dce922b --- /dev/null +++ b/tests/fixtures/replay_regressions/condition-fingerprint-drift-task-failure.json @@ -0,0 +1,13 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "python-condition-fingerprint-drift-task-failure", + "protocol_version": "1.19", + "bindings": ["python"], + "workflow": {"type": "tests.replay.condition-identity", "input": [], "payload_codec": "avro"}, + "history": [ + {"event_type": "ConditionWaitOpened", "payload": {"sequence": 7, "condition_wait_id": "wait-original", "condition_key": "approved", "condition_definition_fingerprint": "sha256:different"}} + ], + "expected": {"command_sequence": []}, + "expected_replay_error": {"type": "NonDeterministicReplayError", "message_contains": "condition_fingerprint_changed", "workflow_sequence": 7} +} diff --git a/tests/fixtures/replay_regressions/condition-key-drift-task-failure.json b/tests/fixtures/replay_regressions/condition-key-drift-task-failure.json new file mode 100644 index 0000000..af9a14f --- /dev/null +++ b/tests/fixtures/replay_regressions/condition-key-drift-task-failure.json @@ -0,0 +1,13 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "python-condition-key-drift-task-failure", + "protocol_version": "1.19", + "bindings": ["python"], + "workflow": {"type": "tests.replay.condition-identity", "input": [], "payload_codec": "avro"}, + "history": [ + {"event_type": "ConditionWaitOpened", "payload": {"sequence": 7, "condition_wait_id": "wait-original", "condition_key": "old-key"}} + ], + "expected": {"command_sequence": []}, + "expected_replay_error": {"type": "NonDeterministicReplayError", "message_contains": "condition_key_changed", "workflow_sequence": 7} +} diff --git a/tests/integration/test_condition_recovery.py b/tests/integration/test_condition_recovery.py new file mode 100644 index 0000000..aa7a5c0 --- /dev/null +++ b/tests/integration/test_condition_recovery.py @@ -0,0 +1,131 @@ +"""A rejected condition deployment must leave the durable run recoverable.""" + +from __future__ import annotations + +import asyncio +import json +import os +import sys +import uuid +from typing import Any + +import pytest + +from durable_workflow import Client, Worker, workflow + + +@workflow.defn(name="tests.condition-deployment-recovery") +class ConditionRecoveryWorkflow: + def __init__(self) -> None: + self.approved = False + + @workflow.signal("wake") + def wake(self) -> None: + pass + + @workflow.signal("approve") + def approve(self) -> None: + self.approved = True + + def run(self, ctx: Any) -> Any: + # Deliberately unsafe configuration drift keeps the class fingerprint + # stable, exercising the recorded condition's own identity protection. + mode = os.environ.get("CONDITION_RECOVERY_MODE", "correct") + key = "changed" if mode == "key" else "approved" + predicate = (lambda: bool(self.approved)) if mode == "fingerprint" else (lambda: self.approved) + yield ctx.wait_condition(predicate, key=key) + return "approved" + + +async def execute_one() -> None: + async with Client( + os.environ["DURABLE_WORKFLOW_SERVER_URL"], + token=os.environ["DURABLE_WORKFLOW_AUTH_TOKEN"], namespace="default", + ) as client: + worker = Worker( + client, task_queue=os.environ["CONDITION_RECOVERY_QUEUE"], + workflows=[ConditionRecoveryWorkflow], + worker_id=os.environ["CONDITION_RECOVERY_WORKER"], + max_concurrent_workflow_tasks=1, + ) + await worker._register() + task = await client.poll_workflow_task( + worker_id=worker.worker_id, task_queue=worker.task_queue, timeout=20, + ) + assert task is not None + commands = await worker._run_workflow_task(task) + print(json.dumps({ + "task_id": task["task_id"], "run_id": task["run_id"], + "attempt": task["workflow_task_attempt"], "commands": commands, + "fingerprint": worker.workflow_definition_fingerprints, + }), flush=True) + + +async def cold_worker(server_url: str, token: str, queue: str, mode: str) -> dict[str, Any]: + environment = { + **os.environ, "DURABLE_WORKFLOW_SERVER_URL": server_url, + "DURABLE_WORKFLOW_AUTH_TOKEN": token, "CONDITION_RECOVERY_QUEUE": queue, + "CONDITION_RECOVERY_MODE": mode, + "CONDITION_RECOVERY_WORKER": f"condition-worker-{uuid.uuid4().hex}", + } + process = await asyncio.create_subprocess_exec( + sys.executable, "-c", + "import asyncio; from tests.integration.test_condition_recovery import execute_one; asyncio.run(execute_one())", + env=environment, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + try: + stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=45) + finally: + if process.returncode is None: + process.kill() + await process.wait() + assert process.returncode == 0, stderr.decode() + return json.loads(stdout.decode()) + + +def history_events(document: dict[str, Any]) -> list[dict[str, Any]]: + events = document.get("events", document.get("history_events")) + assert isinstance(events, list) + return events + + +@pytest.mark.asyncio +@pytest.mark.parametrize("drift", ["key", "fingerprint"]) +async def test_bad_deployment_then_cold_replacement_resumes_original_run( + server_url: str, server_token: str, drift: str, +) -> None: + suffix = uuid.uuid4().hex + queue = f"condition-recovery-{suffix}" + async with Client(server_url, token=server_token, namespace="default") as client: + handle = await client.start_workflow( + workflow_type="tests.condition-deployment-recovery", task_queue=queue, + workflow_id=f"condition-recovery-{suffix}", input=[], + ) + initial = await cold_worker(server_url, server_token, queue, "correct") + assert [command["type"] for command in initial["commands"]] == ["open_condition_wait"] + original_history = history_events(await handle.get_history()) + await handle.signal("wake") + + rejected = await cold_worker(server_url, server_token, queue, drift) + assert rejected["commands"] is None + assert rejected["run_id"] == initial["run_id"] == handle.run_id + assert rejected["fingerprint"] == initial["fingerprint"] + assert (await handle.describe()).status not in {"failed", "completed", "terminated"} + after_rejection = history_events(await handle.get_history()) + assert after_rejection[:len(original_history)] == original_history + assert all(event["event_type"] != "WorkflowFailed" for event in after_rejection) + + resumed = await cold_worker(server_url, server_token, queue, "correct") + assert resumed["run_id"] == handle.run_id + assert resumed["fingerprint"] == initial["fingerprint"] + assert [command["type"] for command in resumed["commands"]] == ["open_condition_wait"] + assert resumed["attempt"] > rejected["attempt"] + await handle.signal("approve") + finished = await cold_worker(server_url, server_token, queue, "correct") + assert [command["type"] for command in finished["commands"]] == ["complete_workflow"] + assert await handle.result(timeout=15) == "approved" + final_history = history_events(await handle.get_history()) + assert final_history[:len(original_history)] == original_history + kinds = [event["event_type"] for event in final_history] + assert "WorkflowFailed" not in kinds + assert kinds.count("WorkflowCompleted") == 1 diff --git a/tests/test_condition_replay_drift.py b/tests/test_condition_replay_drift.py new file mode 100644 index 0000000..7e720c9 --- /dev/null +++ b/tests/test_condition_replay_drift.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +from unittest.mock import AsyncMock + +import pytest + +from durable_workflow import Worker, serializer, workflow +from durable_workflow.client import Client +from durable_workflow.errors import NonDeterministicReplayError, QueryFailed +from durable_workflow.workflow import WorkflowContext, apply_update, query_state, replay + + +@workflow.defn(name="tests.replay.condition-identity") +class ConditionIdentityWorkflow: + @workflow.query("state") + def state(self) -> str: + return "waiting" + + @workflow.update("approve") + def approve(self) -> str: + return "approved" + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(lambda: False, key="approved") + return "done" + + +def condition_history(field: str) -> list[dict]: + return [{ + "event_type": "ConditionWaitOpened", + "payload": { + "sequence": 7, + "condition_wait_id": "wait-original", + "condition_key": "approved", + field: "old-key" if field == "condition_key" else "sha256:different", + }, + }] + + +@pytest.mark.parametrize("field", ["condition_key", "condition_definition_fingerprint"]) +@pytest.mark.parametrize("consumer", ["replay", "query", "update"]) +def test_condition_drift_is_not_a_business_failure(field: str, consumer: str) -> None: + history = condition_history(field) + error = QueryFailed if consumer == "query" else NonDeterministicReplayError + with pytest.raises(error) as captured: + if consumer == "query": + query_state(ConditionIdentityWorkflow, history, [], "state") + elif consumer == "update": + apply_update(ConditionIdentityWorkflow, history, [], "update-original") + else: + replay(ConditionIdentityWorkflow, history, []) + + mismatch = captured.value.__cause__ if consumer == "query" else captured.value + assert isinstance(mismatch, NonDeterministicReplayError) + assert mismatch.workflow_sequence == 7 + assert "wait-original" in str(mismatch) + assert "changed" in str(mismatch) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("field", ["condition_key", "condition_definition_fingerprint"]) +@pytest.mark.parametrize("consumer", ["workflow", "update"]) +async def test_worker_reports_drift_once_without_a_completion(field: str, consumer: str) -> None: + client = AsyncMock(spec=Client) + worker = Worker(client, task_queue="condition-drift", workflows=[ConditionIdentityWorkflow]) + task = { + "task_id": "task-original", + "workflow_id": "workflow-original", + "run_id": "run-original", + "workflow_type": "tests.replay.condition-identity", + "workflow_task_attempt": 2, + "history_events": condition_history(field), + "arguments": serializer.envelope([], codec="avro"), + "payload_codec": "avro", + } + if consumer == "update": + task["workflow_update_id"] = "update-original" + + assert await worker._run_workflow_task_core(task) is None + + client.complete_workflow_task.assert_not_awaited() + client.fail_workflow_task.assert_awaited_once() + report = client.fail_workflow_task.await_args.kwargs + assert report["task_id"] == "task-original" + assert report["workflow_task_attempt"] == 2 + assert report["lease_owner"] == worker.worker_id + assert report["failure_type"] == "NonDeterministicReplayError" + assert "wait-original" in report["message"] + + +def test_application_predicate_failure_remains_terminal() -> None: + @workflow.defn(name="tests.condition-predicate-failure") + class PredicateFailure: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + def predicate() -> bool: + raise ValueError("application failed") + + yield ctx.wait_condition(predicate, key="approved") + + outcome = replay(PredicateFailure, [], []) + assert len(outcome.commands) == 1 + command = outcome.commands[0].to_server_command("condition-drift") + assert command["type"] == "fail_workflow" + assert command["exception_type"] == "ValueError" diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 2f1713f..0e61313 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -17,6 +17,7 @@ commands_to_server_commands, query_state, ) +from tests.test_condition_replay_drift import ConditionIdentityWorkflow from tests.test_golden_history_replay import ( GoldenSagaCompensationWorkflow, GoldenSignalWaitWorkflow, @@ -291,6 +292,7 @@ async def run(self, ctx: WorkflowContext) -> str: YieldedContinueAfterMetadataWorkflow, ] WORKFLOW_TYPES = {str(getattr(workflow, "__workflow_name__", workflow.__name__)): workflow for workflow in WORKFLOWS} +WORKFLOW_TYPES["tests.replay.condition-identity"] = ConditionIdentityWorkflow def _fixture_paths() -> list[Path]: diff --git a/tests/test_wait_condition.py b/tests/test_wait_condition.py index 9a22295..9795320 100644 --- a/tests/test_wait_condition.py +++ b/tests/test_wait_condition.py @@ -6,7 +6,6 @@ from durable_workflow.errors import NonDeterministicReplayError from durable_workflow.workflow import ( CompleteWorkflow, - FailWorkflow, WaitCondition, WorkflowContext, query_state, @@ -1052,14 +1051,12 @@ def test_replay_rejects_changed_condition_key(self) -> None: }, ] - outcome = replay(WaitUntilApproved, history, []) + with pytest.raises(NonDeterministicReplayError, match="key changed") as captured: + replay(WaitUntilApproved, history, []) - assert len(outcome.commands) == 1 - assert not isinstance(outcome.commands[0], CompleteWorkflow) - cmd = outcome.commands[0] - assert isinstance(cmd, FailWorkflow) - assert cmd.exception_type == "NonDeterministicWaitCondition" - assert "key changed" in cmd.message + assert captured.value.workflow_sequence == 1 + assert "old-key" in str(captured.value) + assert "approved" in str(captured.value) def test_replay_rejects_changed_condition_fingerprint(self) -> None: initial = replay(WaitUntilApproved, [], []) @@ -1077,14 +1074,12 @@ def test_replay_rejects_changed_condition_fingerprint(self) -> None: }, ] - outcome = replay(WaitUntilApproved, history, []) + with pytest.raises(NonDeterministicReplayError, match="predicate fingerprint changed") as captured: + replay(WaitUntilApproved, history, []) - assert len(outcome.commands) == 1 - cmd = outcome.commands[0] - assert isinstance(cmd, FailWorkflow) - assert cmd.exception_type == "NonDeterministicWaitCondition" - assert "predicate fingerprint changed" in cmd.message - assert fingerprint in cmd.message + assert captured.value.workflow_sequence == 1 + assert "sha256:different" in str(captured.value) + assert fingerprint in str(captured.value) def test_replay_accepts_matching_condition_fingerprint(self) -> None: initial = replay(WaitUntilApproved, [], []) From f72a51ae37e9a488f3a93223b001cbc8ed7b84a5 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 10 Oct 2026 20:49:21 +0000 Subject: [PATCH 2/3] Qualify explicit repair with cold workers and prepare Python 2.5.3 --- docs/sdk-reference.md | 6 ++-- pyproject.toml | 6 ++-- tests/integration/test_condition_recovery.py | 37 +++++++++++++++++--- tests/test_condition_replay_drift.py | 28 +++++++++++++++ 4 files changed, 68 insertions(+), 9 deletions(-) diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index 5c3de44..5bee7d5 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -491,8 +491,10 @@ durable wait command and rejects replay if history records a different wait key or predicate fingerprint. A mismatch raises `NonDeterministicReplayError` with the original wait identity, workflow sequence and both definitions. The worker fails the current task without recording a terminal workflow failure. -Deploy compatible code to resume the original run. Accepted updates also remain -available for retry when their state replay encounters this error. Query replay +Deploy compatible code, then use `await handle.repair()` to release the blocked +task for the original run. The Server holds replay failures instead of +continuously retrying incompatible code. Accepted updates also remain available +for repair when their state replay encounters this error. Query replay reports a query failure with the mismatch as its cause. Exceptions raised by application predicates remain ordinary workflow failures. diff --git a/pyproject.toml b/pyproject.toml index 17b84af..610c1d3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "durable-workflow" -version = "2.5.2" +version = "2.5.3" description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server" readme = "README.md" requires-python = ">=3.10" @@ -71,8 +71,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main" durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main" [tool.durable-workflow] -product-train = "2.5.2" -registry-version = "2.5.2" +product-train = "2.5.3" +registry-version = "2.5.3" supported-server-versions = "2.5.16" worker-protocol-version = "1.19" control-plane-version = "2" diff --git a/tests/integration/test_condition_recovery.py b/tests/integration/test_condition_recovery.py index aa7a5c0..66675bf 100644 --- a/tests/integration/test_condition_recovery.py +++ b/tests/integration/test_condition_recovery.py @@ -3,10 +3,14 @@ from __future__ import annotations import asyncio +import hashlib +import importlib.metadata +import inspect import json import os import sys import uuid +from pathlib import Path from typing import Any import pytest @@ -49,27 +53,40 @@ async def execute_one() -> None: max_concurrent_workflow_tasks=1, ) await worker._register() + expect_empty = os.environ.get("CONDITION_RECOVERY_EXPECT_EMPTY") == "1" task = await client.poll_workflow_task( - worker_id=worker.worker_id, task_queue=worker.task_queue, timeout=20, + worker_id=worker.worker_id, task_queue=worker.task_queue, timeout=1 if expect_empty else 20, ) + if expect_empty: + assert task is None + print(json.dumps({"empty": True}), flush=True) + return assert task is not None commands = await worker._run_workflow_task(task) print(json.dumps({ "task_id": task["task_id"], "run_id": task["run_id"], "attempt": task["workflow_task_attempt"], "commands": commands, "fingerprint": worker.workflow_definition_fingerprints, + "sdk_version": importlib.metadata.version("durable-workflow"), + "replay_source_sha256": hashlib.sha256( + Path(inspect.getfile(workflow)).read_bytes(), + ).hexdigest(), }), flush=True) -async def cold_worker(server_url: str, token: str, queue: str, mode: str) -> dict[str, Any]: +async def cold_worker( + server_url: str, token: str, queue: str, mode: str, *, expect_empty: bool = False, + python_executable: str | None = None, +) -> dict[str, Any]: environment = { **os.environ, "DURABLE_WORKFLOW_SERVER_URL": server_url, "DURABLE_WORKFLOW_AUTH_TOKEN": token, "CONDITION_RECOVERY_QUEUE": queue, "CONDITION_RECOVERY_MODE": mode, + "CONDITION_RECOVERY_EXPECT_EMPTY": "1" if expect_empty else "0", "CONDITION_RECOVERY_WORKER": f"condition-worker-{uuid.uuid4().hex}", } process = await asyncio.create_subprocess_exec( - sys.executable, "-c", + python_executable or sys.executable, "-c", "import asyncio; from tests.integration.test_condition_recovery import execute_one; asyncio.run(execute_one())", env=environment, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) @@ -101,7 +118,10 @@ async def test_bad_deployment_then_cold_replacement_resumes_original_run( workflow_type="tests.condition-deployment-recovery", task_queue=queue, workflow_id=f"condition-recovery-{suffix}", input=[], ) - initial = await cold_worker(server_url, server_token, queue, "correct") + initial = await cold_worker( + server_url, server_token, queue, "correct", + python_executable=os.environ.get("CONDITION_RECOVERY_INITIAL_PYTHON"), + ) assert [command["type"] for command in initial["commands"]] == ["open_condition_wait"] original_history = history_events(await handle.get_history()) await handle.signal("wake") @@ -115,6 +135,8 @@ async def test_bad_deployment_then_cold_replacement_resumes_original_run( assert after_rejection[:len(original_history)] == original_history assert all(event["event_type"] != "WorkflowFailed" for event in after_rejection) + assert await cold_worker(server_url, server_token, queue, "correct", expect_empty=True) == {"empty": True} + await handle.repair() resumed = await cold_worker(server_url, server_token, queue, "correct") assert resumed["run_id"] == handle.run_id assert resumed["fingerprint"] == initial["fingerprint"] @@ -129,3 +151,10 @@ async def test_bad_deployment_then_cold_replacement_resumes_original_run( kinds = [event["event_type"] for event in final_history] assert "WorkflowFailed" not in kinds assert kinds.count("WorkflowCompleted") == 1 + print(json.dumps({ + "drift": drift, "initial": initial, "rejected": rejected, + "resumed": resumed, "finished": finished, + "preserved_prefix_events": len(original_history), + "workflow_failed_events": 0, "workflow_completed_events": 1, + "explicit_repair": True, "same_run": True, + }), flush=True) diff --git a/tests/test_condition_replay_drift.py b/tests/test_condition_replay_drift.py index 7e720c9..efd262f 100644 --- a/tests/test_condition_replay_drift.py +++ b/tests/test_condition_replay_drift.py @@ -102,3 +102,31 @@ def predicate() -> bool: command = outcome.commands[0].to_server_command("condition-drift") assert command["type"] == "fail_workflow" assert command["exception_type"] == "ValueError" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("field", ["condition_key", "condition_definition_fingerprint"]) +async def test_query_worker_reports_unavailable_state_without_changing_the_run(field: str) -> None: + client = AsyncMock(spec=Client) + worker = Worker(client, task_queue="condition-drift", workflows=[ConditionIdentityWorkflow]) + outcome = await worker._run_query_task({ + "query_task_id": "query-original", + "query_task_attempt": 2, + "workflow_id": "workflow-original", + "run_id": "run-original", + "workflow_type": "tests.replay.condition-identity", + "query_name": "state", + "history_events": condition_history(field), + "workflow_arguments": serializer.envelope([], codec="avro"), + "query_arguments": serializer.envelope([], codec="avro"), + "payload_codec": "avro", + }) + + assert outcome == "failed" + client.complete_query_task.assert_not_awaited() + client.complete_workflow_task.assert_not_awaited() + client.fail_workflow_task.assert_not_awaited() + client.fail_query_task.assert_awaited_once() + report = client.fail_query_task.await_args.kwargs + assert report["reason"] == "query_workflow_state_unavailable" + assert "wait-original" in report["message"] From 0effc8fa89978a141740834d378d6be16bebbac3 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 10 Oct 2026 20:55:23 +0000 Subject: [PATCH 3/3] Verify accepted updates remain recoverable through published HTTP --- tests/integration/test_condition_recovery.py | 35 +++++++++++++++++--- 1 file changed, 30 insertions(+), 5 deletions(-) diff --git a/tests/integration/test_condition_recovery.py b/tests/integration/test_condition_recovery.py index 66675bf..58d51d5 100644 --- a/tests/integration/test_condition_recovery.py +++ b/tests/integration/test_condition_recovery.py @@ -31,6 +31,10 @@ def wake(self) -> None: def approve(self) -> None: self.approved = True + @workflow.update("acknowledge") + def acknowledge(self) -> str: + return "acknowledged" + def run(self, ctx: Any) -> Any: # Deliberately unsafe configuration drift keeps the class fingerprint # stable, exercising the recorded condition's own identity protection. @@ -53,6 +57,9 @@ async def execute_one() -> None: max_concurrent_workflow_tasks=1, ) await worker._register() + if os.environ.get("CONDITION_RECOVERY_REGISTER_ONLY") == "1": + print(json.dumps({"registered": True}), flush=True) + return expect_empty = os.environ.get("CONDITION_RECOVERY_EXPECT_EMPTY") == "1" task = await client.poll_workflow_task( worker_id=worker.worker_id, task_queue=worker.task_queue, timeout=1 if expect_empty else 20, @@ -66,6 +73,7 @@ async def execute_one() -> None: print(json.dumps({ "task_id": task["task_id"], "run_id": task["run_id"], "attempt": task["workflow_task_attempt"], "commands": commands, + "update_id": task.get("workflow_update_id"), "fingerprint": worker.workflow_definition_fingerprints, "sdk_version": importlib.metadata.version("durable-workflow"), "replay_source_sha256": hashlib.sha256( @@ -76,13 +84,14 @@ async def execute_one() -> None: async def cold_worker( server_url: str, token: str, queue: str, mode: str, *, expect_empty: bool = False, - python_executable: str | None = None, + python_executable: str | None = None, register_only: bool = False, ) -> dict[str, Any]: environment = { **os.environ, "DURABLE_WORKFLOW_SERVER_URL": server_url, "DURABLE_WORKFLOW_AUTH_TOKEN": token, "CONDITION_RECOVERY_QUEUE": queue, "CONDITION_RECOVERY_MODE": mode, "CONDITION_RECOVERY_EXPECT_EMPTY": "1" if expect_empty else "0", + "CONDITION_RECOVERY_REGISTER_ONLY": "1" if register_only else "0", "CONDITION_RECOVERY_WORKER": f"condition-worker-{uuid.uuid4().hex}", } process = await asyncio.create_subprocess_exec( @@ -108,12 +117,17 @@ def history_events(document: dict[str, Any]) -> list[dict[str, Any]]: @pytest.mark.asyncio @pytest.mark.parametrize("drift", ["key", "fingerprint"]) +@pytest.mark.parametrize("consumer", ["workflow", "update"]) async def test_bad_deployment_then_cold_replacement_resumes_original_run( - server_url: str, server_token: str, drift: str, + server_url: str, server_token: str, drift: str, consumer: str, ) -> None: suffix = uuid.uuid4().hex queue = f"condition-recovery-{suffix}" async with Client(server_url, token=server_token, namespace="default") as client: + assert await cold_worker( + server_url, server_token, queue, "correct", register_only=True, + python_executable=os.environ.get("CONDITION_RECOVERY_INITIAL_PYTHON"), + ) == {"registered": True} handle = await client.start_workflow( workflow_type="tests.condition-deployment-recovery", task_queue=queue, workflow_id=f"condition-recovery-{suffix}", input=[], @@ -124,7 +138,10 @@ async def test_bad_deployment_then_cold_replacement_resumes_original_run( ) assert [command["type"] for command in initial["commands"]] == ["open_condition_wait"] original_history = history_events(await handle.get_history()) - await handle.signal("wake") + if consumer == "update": + await handle.update("acknowledge", wait_for="accepted", request_id=f"ack-{suffix}") + else: + await handle.signal("wake") rejected = await cold_worker(server_url, server_token, queue, drift) assert rejected["commands"] is None @@ -134,13 +151,18 @@ async def test_bad_deployment_then_cold_replacement_resumes_original_run( after_rejection = history_events(await handle.get_history()) assert after_rejection[:len(original_history)] == original_history assert all(event["event_type"] != "WorkflowFailed" for event in after_rejection) + assert all(event["event_type"] != "UpdateFailed" for event in after_rejection) assert await cold_worker(server_url, server_token, queue, "correct", expect_empty=True) == {"empty": True} await handle.repair() resumed = await cold_worker(server_url, server_token, queue, "correct") assert resumed["run_id"] == handle.run_id assert resumed["fingerprint"] == initial["fingerprint"] - assert [command["type"] for command in resumed["commands"]] == ["open_condition_wait"] + expected_command = "complete_update" if consumer == "update" else "open_condition_wait" + assert [command["type"] for command in resumed["commands"]] == [expected_command] + if consumer == "update": + assert resumed["update_id"] == rejected["update_id"] + assert resumed["update_id"] is not None assert resumed["attempt"] > rejected["attempt"] await handle.signal("approve") finished = await cold_worker(server_url, server_token, queue, "correct") @@ -151,8 +173,11 @@ async def test_bad_deployment_then_cold_replacement_resumes_original_run( kinds = [event["event_type"] for event in final_history] assert "WorkflowFailed" not in kinds assert kinds.count("WorkflowCompleted") == 1 + if consumer == "update": + assert kinds.count("UpdateCompleted") == 1 + assert "UpdateFailed" not in kinds print(json.dumps({ - "drift": drift, "initial": initial, "rejected": rejected, + "drift": drift, "consumer": consumer, "initial": initial, "rejected": rejected, "resumed": resumed, "finished": finished, "preserved_prefix_events": len(original_history), "workflow_failed_events": 0, "workflow_completed_events": 1,