diff --git a/CHANGELOG.md b/CHANGELOG.md index 52a6f2a..4bea38e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Condition wait opens carry a deterministic `python:condition-wait:{ordinal}` occurrence ID for each authored call, including selection members. False-wake reopens preserve the first open's ID and remaining timeout; legacy chains without an ID keep that absence even when later physical opens have an ID. Separate calls and grace waits receive new ordinals. These identities enable unresolved-open deduplication on supporting runtimes. (#259) + - Opt-in typed workflow values (#184). `@workflow.defn(..., typed=True)` decodes the arguments of `run` (after `ctx`) and of every signal, query, update and update-validator handler to their annotations. `ctx.start_child_workflow(..., result_type=T)` decodes a child's result, and `WorkflowHandle.result(result_type=T)` / `Client.get_result(..., result_type=T)` decode a workflow's result on the client. Dataclass and `Enum` values in client workflow, signal, update, query and schedule arguments, child and continue-as-new arguments, and workflow and update results are sent as their neutral values. A typed `TypedValueError` names the path but never the value. Untyped workflows receive plain values as before, so in-flight histories replay unchanged. ### Fixed diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 772a2a2..aba9ac8 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -1645,6 +1645,7 @@ class WaitCondition: condition_key: str | None = None condition_definition_fingerprint: str | None = None timeout_seconds: int | None = None + condition_wait_occurrence_id: str | None = None _cancellation_scope_id: str = field( default_factory=_authored_cancellation_scope, init=False, repr=False, compare=False, ) @@ -1670,6 +1671,8 @@ def to_server_command( cmd["condition_definition_fingerprint"] = self.condition_definition_fingerprint if self.timeout_seconds is not None: cmd["timeout_seconds"] = self.timeout_seconds + if self.condition_wait_occurrence_id is not None: + cmd["condition_wait_occurrence_id"] = self.condition_wait_occurrence_id return cmd @@ -2213,6 +2216,7 @@ def __init__( self._rng = random.Random(seed) self._uuid7_counter = 0 self._nexus_call_counter = 0 + self._condition_wait_counter = 0 self._message_stream_messages: dict[str, list[MessageStreamMessage]] = {} self._message_stream_cursors: dict[str, int] = {} self._message_stream_waits: dict[str, int] = {} @@ -2513,15 +2517,21 @@ def wait_condition( The deadline is fixed when the wait first opens. A signal that leaves the predicate false re-drives the workflow, and the wait reopens with only the time left before that deadline, so signals never extend it. + + Each authored call gets a deterministic occurrence ID. Reopens keep + the first recorded ID, including its absence in legacy histories. """ timeout_seconds: int | None = None if timeout is not None: timeout_seconds = max(0, math.ceil(timeout)) + occurrence_id = f"python:condition-wait:{self._condition_wait_counter}" + self._condition_wait_counter += 1 return WaitCondition( predicate=predicate, condition_key=key, condition_definition_fingerprint=_condition_predicate_fingerprint(predicate), timeout_seconds=timeout_seconds, + condition_wait_occurrence_id=occurrence_id, ) def select( @@ -7143,6 +7153,10 @@ def _reopened_with_remaining_timeout(command: WaitCondition, first_index: int) - if first_index >= len(wait_opened): return command reopened = copy(command) + # An upgraded worker must not add an ID to a legacy chain or inherit + # a later physical open's ID when duplicate chains merge (#259). + occurrence = wait_opened[first_index].get("condition_wait_occurrence_id") + reopened.condition_wait_occurrence_id = occurrence if isinstance(occurrence, str) and occurrence else None # A reopen sends the fingerprint its first open recorded, so a wait # first opened under v1 (2.14.0 and earlier) keeps v1 across the # upgrade and links v1 to v1, never v1 to v2 (#258). diff --git a/tests/fixtures/replay_regressions/condition-occurrence-first-open-preserved.json b/tests/fixtures/replay_regressions/condition-occurrence-first-open-preserved.json new file mode 100644 index 0000000..a6e084a --- /dev/null +++ b/tests/fixtures/replay_regressions/condition-occurrence-first-open-preserved.json @@ -0,0 +1,91 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "id": "condition-occurrence-first-open-preserved", + "workflow": { + "type": "tests.condition-occurrence.approval", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-04T00:00:00+00:00", + "payload": { + "workflow_type": "tests.condition-occurrence.approval" + }, + "sequence": 1 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:00+00:00", + "payload": { + "condition_wait_id": "w1", + "sequence": 1, + "condition_key": "approval", + "timeout_seconds": 60, + "condition_wait_occurrence_id": "original-occurrence" + }, + "sequence": 2 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:10+00:00", + "payload": { + "signal_name": "note", + "workflow_sequence": 1, + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwA" + }, + "payload_codec": "avro" + }, + "sequence": 3 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:10+00:00", + "payload": { + "condition_wait_id": "w1", + "sequence": 1 + }, + "sequence": 4 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:11+00:00", + "payload": { + "condition_wait_id": "w2", + "sequence": 2, + "condition_key": "approval", + "timeout_seconds": 50, + "condition_wait_occurrence_id": "later-occurrence" + }, + "sequence": 5 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:20+00:00", + "payload": { + "signal_name": "note", + "workflow_sequence": 2, + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwA" + }, + "payload_codec": "avro" + }, + "sequence": 6 + } + ], + "expected": { + "type": "open_condition_wait", + "condition_key": "approval", + "condition_wait_occurrence_id": "original-occurrence", + "timeout_seconds": 40 + } +} diff --git a/tests/fixtures/replay_regressions/condition-occurrence-selection-authoring-order.json b/tests/fixtures/replay_regressions/condition-occurrence-selection-authoring-order.json new file mode 100644 index 0000000..d51e45a --- /dev/null +++ b/tests/fixtures/replay_regressions/condition-occurrence-selection-authoring-order.json @@ -0,0 +1,67 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "id": "condition-occurrence-selection-authoring-order", + "workflow": { + "type": "tests.condition-occurrence.selection", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "payload": { + "workflow_type": "tests.condition-occurrence.selection" + }, + "sequence": 1 + } + ], + "command_sequence": [ + { + "type": "open_condition_wait", + "condition_key": "first", + "condition_wait_occurrence_id": "python:condition-wait:0" + }, + { + "type": "open_condition_wait", + "condition_key": "nested", + "condition_wait_occurrence_id": "python:condition-wait:1" + }, + { + "type": "start_timer", + "delay_seconds": 5 + }, + { + "type": "open_condition_wait", + "condition_key": "last", + "condition_wait_occurrence_id": "python:condition-wait:2" + } + ], + "expected": { + "command_sequence": [ + { + "type": "open_condition_wait", + "condition_key": "first", + "condition_wait_occurrence_id": "python:condition-wait:0" + }, + { + "type": "open_condition_wait", + "condition_key": "nested", + "condition_wait_occurrence_id": "python:condition-wait:1" + }, + { + "type": "start_timer", + "delay_seconds": 5 + }, + { + "type": "open_condition_wait", + "condition_key": "last", + "condition_wait_occurrence_id": "python:condition-wait:2" + } + ] + } +} diff --git a/tests/test_condition_occurrence_id.py b/tests/test_condition_occurrence_id.py new file mode 100644 index 0000000..97d7fd3 --- /dev/null +++ b/tests/test_condition_occurrence_id.py @@ -0,0 +1,245 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +import pytest + +from durable_workflow import serializer, workflow +from durable_workflow.workflow import ( + CompleteWorkflow, + WaitCondition, + WorkflowContext, + commands_to_server_commands, + replay, +) +from tests.test_durable_selection import ( + DurableSelectionConditionTerminalWorkflow, + _condition_terminal_selection_history, +) +from tests.test_wait_condition import OrphanedDuplicateDecision, _orphaned_duplicate_history + +ROOT = Path(__file__).resolve().parent.parent + + +def _event(kind: str, seconds: int, **payload: Any) -> dict[str, Any]: + at = datetime(2026, 11, 4, tzinfo=timezone.utc) + timedelta(seconds=seconds) + return {"event_type": kind, "recorded_at": at.isoformat(), "payload": payload} + + +START = _event("WorkflowStarted", 0) + + +def _opened(wait_id: str, sequence: int, seconds: int, occurrence: str | None, timeout: int = 60) -> dict[str, Any]: + event = _event("ConditionWaitOpened", seconds, condition_wait_id=wait_id, sequence=sequence, + condition_key="approval", timeout_seconds=timeout) + if occurrence is not None: + event["payload"]["condition_wait_occurrence_id"] = occurrence + return event + + +def _signal(name: str, sequence: int, seconds: int) -> dict[str, Any]: + return _event("SignalReceived", seconds, signal_name=name, workflow_sequence=sequence, + value=serializer.envelope([]), payload_codec="avro") + + +def _satisfied(wait_id: str, sequence: int, seconds: int) -> dict[str, Any]: + return _event("ConditionWaitSatisfied", seconds, condition_wait_id=wait_id, sequence=sequence) + + +@workflow.defn(name="tests.condition-occurrence.approval") +class OccurrenceApproval: + def __init__(self) -> None: + self.ready = False + + @workflow.signal("note") + def note(self) -> None: + return None + + @workflow.signal("approve") + def approve(self) -> None: + self.ready = True + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(lambda: self.ready, key="approval", timeout=60) + return "approved" + + +@workflow.defn(name="tests.condition-occurrence.two-calls") +class TwoOccurrenceCalls(OccurrenceApproval): + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(lambda: self.ready, key="approval", timeout=60) + self.ready = False + yield ctx.wait_condition(lambda: self.ready, key="second", timeout=30) + return "approved" + + +@workflow.defn(name="tests.condition-occurrence.no-timeout") +class NoTimeoutOccurrence(OccurrenceApproval): + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(lambda: self.ready, key="approval") + return "approved" + + +@workflow.defn(name="tests.condition-occurrence.grace") +class OccurrenceGrace(OccurrenceApproval): + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + ready = yield ctx.wait_condition(lambda: self.ready, key="approval", timeout=10) + if not ready: + yield ctx.wait_condition(lambda: self.ready, key="approval", timeout=30) + return "approved" + + +@workflow.defn(name="tests.condition-occurrence.transient") +class TransientThenOccurrence: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(lambda: True) + yield ctx.wait_condition(lambda: False) + + +@workflow.defn(name="tests.condition-occurrence.selection") +class SelectionOccurrences: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.select({ + "first": ctx.wait_condition(lambda: False, key="first"), + "nested": [ctx.wait_condition(lambda: False, key="nested"), ctx.start_timer(5)], + "last": ctx.wait_condition(lambda: False, key="last"), + }) + + +def test_occurrences_count_authored_calls_and_reset_for_each_run() -> None: + first = WorkflowContext(run_id="one") + second = WorkflowContext(run_id="two") + commands = [first.wait_condition(lambda: False), first.wait_condition(lambda: False), + second.wait_condition(lambda: False)] + assert [command.to_server_command("q")["condition_wait_occurrence_id"] for command in commands] == [ + "python:condition-wait:0", "python:condition-wait:1", "python:condition-wait:0", + ] + + +def test_a_transient_satisfied_call_still_owns_its_ordinal() -> None: + (command,) = replay(TransientThenOccurrence, [], []).commands + assert isinstance(command, WaitCondition) + assert command.to_server_command("q")["condition_wait_occurrence_id"] == "python:condition-wait:1" + + +def test_explicit_occurrence_is_serialized() -> None: + command = WaitCondition(lambda: False, condition_wait_occurrence_id="recorded-occurrence") + assert command.to_server_command("q") == { + "type": "open_condition_wait", "condition_wait_occurrence_id": "recorded-occurrence", + } + + +def test_a_false_wake_reuses_the_original_occurrence_and_deadline() -> None: + history = [START, _opened("w1", 1, 0, "python:condition-wait:0"), _signal("note", 1, 10)] + (command,) = replay(OccurrenceApproval, history, []).commands + assert isinstance(command, WaitCondition) + wire = command.to_server_command("q") + assert wire["condition_wait_occurrence_id"] == "python:condition-wait:0" + assert wire["timeout_seconds"] == 50 + + +@pytest.mark.parametrize("first", [None, "unbounded-first-id"]) +def test_a_no_timeout_reopen_without_history_clocks_keeps_its_first_id(first: str | None) -> None: + opened = _opened("w1", 1, 0, first) + opened["payload"].pop("timeout_seconds") + history = [dict(START), opened, _signal("note", 1, 10)] + for event in history: + event.pop("recorded_at", None) + (command,) = replay(NoTimeoutOccurrence, history, []).commands + assert isinstance(command, WaitCondition) + wire = command.to_server_command("q") + assert wire.get("condition_wait_occurrence_id") == first + assert ("condition_wait_occurrence_id" in wire) == bool(first) + assert "timeout_seconds" not in wire + + +@pytest.mark.parametrize(("first", "later"), [ + (None, None), (None, "new-worker-id"), ("", "new-worker-id"), + ("first-open-id", None), ("first-open-id", "different-later-id"), +]) +def test_upgrade_crossing_keeps_the_first_open_id_or_its_absence(first: str | None, later: str | None) -> None: + history = [START, _opened("w1", 1, 0, first), _signal("note", 1, 10), _satisfied("w1", 1, 10), + _opened("w2", 2, 11, later, timeout=50), _signal("note", 2, 20)] + (command,) = replay(OccurrenceApproval, history, []).commands + assert isinstance(command, WaitCondition) + wire = command.to_server_command("q") + assert wire.get("condition_wait_occurrence_id") == (first or None) + assert ("condition_wait_occurrence_id" in wire) == bool(first) + assert wire["timeout_seconds"] == 40 + + +@pytest.mark.parametrize("first", [None, "original-orphan-id"]) +def test_an_orphaned_duplicate_chain_keeps_its_first_open_id(first: str | None) -> None: + history = _orphaned_duplicate_history(180, True) + opens = [event for event in history if event["event_type"] == "ConditionWaitOpened"] + if first is not None: + opens[0]["payload"]["condition_wait_occurrence_id"] = first + for event in opens[1:]: + event["payload"]["condition_wait_occurrence_id"] = "later-duplicate-id" + (command,) = replay(OrphanedDuplicateDecision, history, []).commands + assert isinstance(command, WaitCondition) + wire = command.to_server_command("q") + assert wire.get("condition_wait_occurrence_id") == first + assert ("condition_wait_occurrence_id" in wire) == bool(first) + assert wire["timeout_seconds"] == 0 + + +def test_the_next_authored_call_after_a_false_wake_gets_a_new_ordinal() -> None: + history = [START, _opened("w1", 1, 0, "python:condition-wait:0"), _signal("note", 1, 10), + _satisfied("w1", 1, 10), _opened("w2", 2, 11, "python:condition-wait:0", timeout=50), + _signal("approve", 2, 20), _satisfied("w2", 2, 20)] + (command,) = replay(TwoOccurrenceCalls, history, []).commands + assert isinstance(command, WaitCondition) + wire = command.to_server_command("q") + assert wire["condition_wait_occurrence_id"] == "python:condition-wait:1" + assert wire["condition_key"] == "second" + assert wire["timeout_seconds"] == 30 + + +def test_a_grace_wait_after_timeout_gets_a_new_ordinal() -> None: + history = [START, _opened("w1", 1, 0, "python:condition-wait:0", timeout=10), + _event("TimerFired", 10, timer_kind="condition_timeout", condition_wait_id="w1", sequence=1)] + (command,) = replay(OccurrenceGrace, history, []).commands + assert isinstance(command, WaitCondition) + wire = command.to_server_command("q") + assert wire["condition_wait_occurrence_id"] == "python:condition-wait:1" + assert wire["timeout_seconds"] == 30 + + +def test_selection_members_use_authoring_order_including_nested_members() -> None: + wire = commands_to_server_commands(replay(SelectionOccurrences, [], []).commands, "q") + assert [command["condition_wait_occurrence_id"] for command in wire + if command["type"] == "open_condition_wait"] == [ + "python:condition-wait:0", "python:condition-wait:1", "python:condition-wait:2", + ] + + +@pytest.mark.parametrize("recorded", [None, "foreign-selection-id"]) +def test_a_recorded_selection_does_not_refuse_an_old_or_foreign_occurrence(recorded: str | None) -> None: + history = _condition_terminal_selection_history("ConditionWaitSatisfied", "valid") + if recorded is not None: + history[0]["payload"]["condition_wait_occurrence_id"] = recorded + (command,) = replay(DurableSelectionConditionTerminalWorkflow, history, []).commands + assert command == CompleteWorkflow({"winner": "condition", "winner_value": True}) + + +def test_occurrences_are_deterministic_in_fresh_replay_processes() -> None: + script = ( + "import json; from durable_workflow.workflow import replay, commands_to_server_commands;" + "from tests.test_condition_occurrence_id import SelectionOccurrences;" + "print(json.dumps([c['condition_wait_occurrence_id'] for c in " + "commands_to_server_commands(replay(SelectionOccurrences, [], []).commands, 'q') " + "if c['type'] == 'open_condition_wait']))" + ) + for _ in range(2): + result = subprocess.run([sys.executable, "-c", script], check=True, capture_output=True, text=True, + cwd=ROOT, env={**os.environ, "PYTHONPATH": str(ROOT / "src")}) + assert json.loads(result.stdout) == [ + "python:condition-wait:0", "python:condition-wait:1", "python:condition-wait:2", + ] diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 0cbcb86..143ad7b 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -27,6 +27,7 @@ ) from tests.condition_fingerprint_workflows import FingerprintApproval, FingerprintGenexpr, FingerprintTerminalDuplicate from tests.deregistration_recovery import RecoveredWorkflow as FencedRecoveryWorkflow +from tests.test_condition_occurrence_id import OccurrenceApproval, SelectionOccurrences from tests.test_condition_replay_drift import ConditionIdentityWorkflow from tests.test_generator_lifecycle import CleanupSideEffect, PendingFinally from tests.test_golden_history_replay import ( @@ -428,6 +429,8 @@ def run(self, ctx: WorkflowContext, message: str) -> None: FingerprintApproval, FingerprintGenexpr, FingerprintTerminalDuplicate, + OccurrenceApproval, + SelectionOccurrences, ] WORKFLOW_TYPES = {str(getattr(workflow, "__workflow_name__", workflow.__name__)): workflow for workflow in WORKFLOWS} WORKFLOW_TYPES["tests.replay.condition-identity"] = ConditionIdentityWorkflow