From 1f729a632ecb88154cd1465178fa98e8dfe0ebb0 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 10:12:42 +0000 Subject: [PATCH 1/3] Pin legacy parallel failure suspension before introducing policy metadata --- ...gacy-parallel-failure-remains-pending.json | 99 +++++++++++++++++++ tests/test_parallel_group_policy.py | 45 +++++++++ tests/test_replay_regression_corpus.py | 2 + 3 files changed, 146 insertions(+) create mode 100644 tests/fixtures/replay_regressions/legacy-parallel-failure-remains-pending.json create mode 100644 tests/test_parallel_group_policy.py diff --git a/tests/fixtures/replay_regressions/legacy-parallel-failure-remains-pending.json b/tests/fixtures/replay_regressions/legacy-parallel-failure-remains-pending.json new file mode 100644 index 0000000..980e702 --- /dev/null +++ b/tests/fixtures/replay_regressions/legacy-parallel-failure-remains-pending.json @@ -0,0 +1,99 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "legacy-parallel-failure-remains-pending", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.parallel-failure-policy", + "input": [], + "payload_codec": "avro", + "query": { + "name": "caught", + "arguments": [] + } + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-01-01T00:00:00Z", + "payload": {} + }, + { + "event_type": "ActivityScheduled", + "recorded_at": "2026-01-01T00:00:00Z", + "payload": { + "sequence": 1, + "activity_execution_id": "execution-0", + "activity_type": "slow-first", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ] + } + }, + { + "event_type": "ActivityScheduled", + "recorded_at": "2026-01-01T00:00:00Z", + "payload": { + "sequence": 2, + "activity_execution_id": "execution-1", + "activity_type": "fast-second", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ] + } + }, + { + "event_type": "ActivityFailed", + "recorded_at": "2026-01-01T00:00:01Z", + "payload": { + "sequence": 2, + "activity_execution_id": "execution-1", + "activity_type": "fast-second", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ], + "message": "higher position failed first" + } + } + ], + "expected": { + "command_sequence": [], + "query_result": null + } +} diff --git a/tests/test_parallel_group_policy.py b/tests/test_parallel_group_policy.py new file mode 100644 index 0000000..846ab48 --- /dev/null +++ b/tests/test_parallel_group_policy.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from durable_workflow import workflow +from durable_workflow.errors import ActivityFailed +from durable_workflow.workflow import CompleteWorkflow, WorkflowContext, query_state, replay + +FIXTURE = Path(__file__).parent / "fixtures/replay_regressions/legacy-parallel-failure-remains-pending.json" + + +@workflow.defn(name="tests.replay.parallel-failure-policy") +class ParallelFailurePolicyWorkflow: + def __init__(self) -> None: + self.caught: str | None = None + + @workflow.query("caught") + def caught_failure(self) -> str | None: + return self.caught + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + try: + yield [ctx.schedule_activity("slow-first", []), ctx.schedule_activity("fast-second", [])] + except ActivityFailed as error: + self.caught = error.activity_type + return self.caught + + +def test_absent_policy_keeps_the_failure_handler_suspended_until_all_members_settle() -> None: + history = json.loads(FIXTURE.read_text())["history"] + outcome = replay(ParallelFailurePolicyWorkflow, history, []) + assert not any(isinstance(command, CompleteWorkflow) for command in outcome.commands) + assert query_state(ParallelFailurePolicyWorkflow, history, [], "caught") is None + + +def test_absent_policy_preserves_positional_failure_after_a_later_sibling_settles() -> None: + history = json.loads(FIXTURE.read_text())["history"] + first = dict(history[1]["payload"]) + first["message"] = "lower position failed later" + history.append({"event_type": "ActivityFailed", "recorded_at": "2026-01-01T00:00:03Z", "payload": first}) + outcome = replay(ParallelFailurePolicyWorkflow, history, []) + assert len(outcome.commands) == 1 + assert isinstance(outcome.commands[0], CompleteWorkflow) + assert outcome.commands[0].result == "slow-first" diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 6b11249..86e0c4a 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -47,6 +47,7 @@ GoldenTimeoutWaitWorkflow, GoldenVersionMarkerWorkflow, ) +from tests.test_parallel_group_policy import ParallelFailurePolicyWorkflow from tests.test_patch_history import ( MixedPatchHelpersWorkflow, PatchCancellationWorkflow, @@ -470,6 +471,7 @@ def run(self, ctx: WorkflowContext, message: str) -> None: WORKFLOW_TYPES = {str(getattr(workflow, "__workflow_name__", workflow.__name__)): workflow for workflow in WORKFLOWS} WORKFLOW_TYPES["tests.replay.condition-identity"] = ConditionIdentityWorkflow WORKFLOW_TYPES["tests.replay.entry-arguments-required"] = RequiredEntryArgumentWorkflow +WORKFLOW_TYPES["tests.replay.parallel-failure-policy"] = ParallelFailurePolicyWorkflow def _fixture_paths() -> list[Path]: From 461e0ec52b7312373e797a86c77f17c06e82a366 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 10:37:50 +0000 Subject: [PATCH 2/3] Prepare disabled explicit parallel failure propagation --- CHANGELOG.md | 4 + src/durable_workflow/replay_clock.py | 13 + src/durable_workflow/workflow.py | 182 ++++++++++- ...cit-parallel-policy-requires-consumer.json | 108 +++++++ ...gacy-parallel-failure-remains-pending.json | 24 +- tests/test_parallel_group_policy.py | 296 +++++++++++++++++- 6 files changed, 609 insertions(+), 18 deletions(-) create mode 100644 tests/fixtures/replay_regressions/explicit-parallel-policy-requires-consumer.json diff --git a/CHANGELOG.md b/CHANGELOG.md index d7ed657..5fa16b3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed + +- Replay refuses explicit `parallel_group_policy` histories before constructing application state until a qualified policy consumer is enabled. Absent-policy histories retain their wait, failure-selection and clock behavior. Explicit propagation is being prepared behind disabled worker admission. (#307) + ### Added - Timeout cleanup interruption (workflow#863, workflow#1129). `CancellationContext.from_dict()` reads the context's optional top-level `cause`: absent or `"cancellation"` is an ordinary cancellation, `"timeout"` marks a timeout-cleanup request (the run passed its work deadline and is now in cleanup), and any other value raises `ValueError`. The context carries no other timeout fields: its existing `cleanup_deadline_at` is the cleanup deadline, and the timeout kind and work deadline stay on the `WorkflowTimeoutCleanupStarted` event. `cause` and `is_timeout` are readable on the context, and `to_dict()` writes `cause` back exactly as received, including an explicit `"cancellation"`, leaving ordinary contexts byte-identical. A timeout request is raised at the authored call as the new experimental `WorkflowTimeoutCleanupRequested`, a subclass of `WorkflowCancelled`, so existing `except WorkflowCancelled:` handlers, such as saga compensation, run on timeout too. It stays `BaseException`-rooted and is not an `asyncio.CancelledError`. Test `cause`, the cross-SDK discriminator, to tell the two apart. `cause` is never inherited: a child cancelled under a timeout request, scoped or not, receives an ordinary cancellation without `cause`. diff --git a/src/durable_workflow/replay_clock.py b/src/durable_workflow/replay_clock.py index 171a709..bd78da0 100644 --- a/src/durable_workflow/replay_clock.py +++ b/src/durable_workflow/replay_clock.py @@ -343,6 +343,19 @@ def observe_activity_group( events.append(event) self._pending = (None, _activity_group_time(events, failed_member)) + def observe_propagated_activity_failure(self, member: _GroupMember, *, reconstruction: bool) -> None: + """Use only the qualified first-recorded failure of an explicit group.""" + self._pending = (None, None) + if reconstruction: + return + event = self._activity_group_boundary(member) + if event is None or _kind(event) == "ActivityCompleted": + return + time = _group_event_time(event) + milliseconds = canonical_milliseconds(time) if time is not None else None + if milliseconds is not None: + self._pending = (member[0], from_canonical_milliseconds(milliseconds)) + def _activity_group_boundary(self, member: _GroupMember) -> Mapping[str, Any] | None: index, sequence, activity_type, path = member indices = self._by_sequence.get(sequence, []) diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 0948e20..76c4a1a 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -4026,6 +4026,72 @@ def _parallel_group_entry(payload: Mapping[str, Any]) -> tuple[str, str, int, in return group_id, kind, base_sequence, size, index +def _read_parallel_group_policies( + events: list[dict[str, Any]], supported: tuple[str, ...], +) -> dict[tuple[str, str, int, int], str | None]: + """Validate recorded policy before application code can observe the history. + + This draft consumer has no worker negotiation or advertisement. The private + replay option enables source tests only until producer admission is qualified. + """ + if any(policy != "propagate" for policy in supported): + raise ValueError("parallel_group_policy consumer is not implemented") + policies: dict[tuple[str, str, int, int], str | None] = {} + for event in events: + payload = event.get("payload") + if not isinstance(payload, Mapping): + continue + path = payload.get("parallel_group_path") + frames = path if isinstance(path, list) and path else [payload] + has_explicit = "parallel_group_policy" in payload or any( + isinstance(frame, Mapping) and "parallel_group_policy" in frame for frame in frames + ) + if has_explicit and path is not None and not isinstance(path, list): + raise ValueError("parallel_group_policy requires a canonical group path") + if isinstance(path, list) and path and isinstance(path[-1], Mapping) and ( + ("parallel_group_policy" in payload) != ("parallel_group_policy" in path[-1]) + or payload.get("parallel_group_policy") != path[-1].get("parallel_group_policy") + ): + raise ValueError("parallel_group_policy disagrees with its canonical path") + if isinstance(path, list) and path and any( + isinstance(frame, Mapping) and "parallel_group_policy" in frame for frame in path + ) and (not isinstance(path[-1], Mapping) + or _parallel_group_entry(payload) != _parallel_group_entry(path[-1])): + raise ValueError("parallel_group_policy requires matching flattened group identity") + ordinary_policies: list[str | None] = [] + for frame in frames: + if not isinstance(frame, Mapping): + if has_explicit: + raise ValueError("parallel_group_policy requires canonical group metadata") + continue + present = "parallel_group_policy" in frame + policy = frame.get("parallel_group_policy") + if present and (not isinstance(policy, str) or policy not in {"propagate", "cancel_remaining", "settle"}): + raise ValueError("parallel_group_policy is unknown or malformed") + if frame.get("parallel_group_mode") == "select": + if present: + raise ValueError("parallel_group_policy does not apply to select frames") + continue + entry = _parallel_group_entry(frame) + if entry is None: + if has_explicit: + raise ValueError("parallel_group_policy requires canonical group metadata") + continue + key = entry[:4] + recorded = cast(str | None, policy) if present else None + ordinary_policies.append(recorded) + if key in policies and policies[key] != recorded: + raise ValueError("parallel_group_policy is inconsistent across members or history") + policies[key] = recorded + if has_explicit and None in ordinary_policies: + raise LocalActivityExecutionAborted("parallel_group_policy mixed nested composition is not supported") + if any(policy is not None and policy not in supported for policy in policies.values()): + raise LocalActivityExecutionAborted( + "parallel_group_policy_not_supported: explicit history requires a qualified policy consumer", + ) + return policies + + def _selection_group_metadata(payload: Mapping[str, Any]) -> Mapping[str, Any] | None: """Return the outer durable-selection metadata entry, if present.""" raw_path = payload.get("parallel_group_path") @@ -4864,6 +4930,7 @@ def _replay_state( reconstruction_only: bool = False, allow_cancellation_scope_authoring: bool = False, allow_cancellation_scope_delivery: bool = False, + parallel_group_policies: tuple[str, ...] = (), ) -> _ReplayState: _validate_workflow_entrypoint(workflow_cls) _workflow_handler_registries(workflow_cls) @@ -4887,6 +4954,7 @@ def _replay_state( # an all() group and never supplies replay time: the workflow had closed # past it. Every history reader below sees the activity as still open. events = [event for event in history_events if not _settled_at_close(event)] + recorded_group_policies = _read_parallel_group_policies(events, parallel_group_policies) # A malformed delivery span refuses replay before any history parser can # read it more permissively. _assert_delivery_spans(events) @@ -5312,7 +5380,11 @@ def _append_resolved_result(value: Any, shape: str, event: Mapping[str, Any]) -> _append_selection_step(shape, event, fallback_sequence=workflow_sequence) return resolved_results.append(value) - resolved_parallel_members.append(_parallel_group_member(payload)) + resolved_parallel_members.append( + None if any(frame.get("parallel_group_policy") is not None + for frame in (payload.get("parallel_group_path") or []) if isinstance(frame, Mapping)) + or payload.get("parallel_group_policy") is not None else _parallel_group_member(payload) + ) recorded_steps.append(step) def _recorded_step( @@ -5456,7 +5528,8 @@ def _unconsumed_recorded_steps() -> list[_RecordedStep]: candidates.extend(recorded_wait_steps[wait_yield_count:]) if pending_step_cursor < len(recorded_pending_steps): candidates.extend(recorded_pending_steps[pending_step_cursor:]) - return sorted(candidates, key=lambda step: step.workflow_sequence) + return sorted((step for step in candidates if step.workflow_sequence not in consumed_policy_sequences), + key=lambda step: step.workflow_sequence) def _next_unconsumed_recorded_step() -> _RecordedStep | None: candidates = _unconsumed_recorded_steps() @@ -6179,6 +6252,7 @@ def _same_logical_condition_wait( ) result_cursor = 0 + consumed_policy_sequences: set[int] = set() pending_step_cursor = 0 wait_yield_count = 0 has_terminal_history = any(_history_event_type(event) in { @@ -6381,13 +6455,25 @@ def _parallel_group_entry(base_sequence: int, size: int, index: int, kind: str) "condition": "parallel-conditions", "mixed": "parallel-calls", }[kind] - return { + entry: dict[str, Any] = { "parallel_group_id": f"{prefix}:{base_sequence}:{size}", "parallel_group_kind": kind, "parallel_group_base_sequence": base_sequence, "parallel_group_size": size, "parallel_group_index": index, } + key = (entry["parallel_group_id"], kind, base_sequence, size) + if key in recorded_group_policies: + policy = recorded_group_policies[key] + elif any(base_sequence <= step.workflow_sequence < base_sequence + size + for step in recorded_steps + recorded_wait_steps + recorded_pending_steps): + # Histories predating group metadata also keep their recorded semantics. + policy = None + else: + policy = "propagate" if "propagate" in parallel_group_policies else None + if policy is not None: + entry["parallel_group_policy"] = policy + return entry def _annotate_parallel_commands( commands: list[Any], @@ -6407,7 +6493,7 @@ def _annotate_parallel_commands( recorded_sequences = [ step.workflow_sequence for step in recorded_steps + recorded_wait_steps + recorded_pending_steps ] - base_sequence = max(recorded_sequences, default=0) + len(pending) + 1 + base_sequence = max([*recorded_sequences, *consumed_policy_sequences], default=0) + len(pending) + 1 def _build(group: list[Any], group_base: int) -> tuple[list[Any], list[Any]]: size = len(_parallel_leaves(group, nested=group is not commands)) @@ -7288,6 +7374,43 @@ def _consume_scope_delivery(command: Any, delivery: ScopeBoundary) -> _ReplaySta except StopIteration as stop: return _terminal_state(stop.value, include_pending=True) + def _consume_propagating_group( + commands: list[Any], + ) -> tuple[list[Any], list[_RecordedStep], ActivityFailed | ChildWorkflowFailed | None] | None: + base = commands[0]._parallel_group_path[0]["parallel_group_base_sequence"] + slots = [index for index in range(result_cursor, len(recorded_steps)) + if base <= recorded_steps[index].workflow_sequence < base + len(commands)] + if len({recorded_steps[index].workflow_sequence for index in slots}) != len(slots): + raise ValueError("parallel_group_policy history has duplicate member outcomes") + for index in slots: + _assert_step_matches(commands[recorded_steps[index].workflow_sequence - base], recorded_steps[index]) + failures = [index for index in slots + if isinstance(resolved_results[index], ActivityFailed | ChildWorkflowFailed)] + selected = min(failures, key=lambda index: cast(int, recorded_steps[index].history_index)) if failures else None + if selected is None and len(slots) != len(commands): + return None + ordered = sorted(slots, key=lambda index: recorded_steps[index].workflow_sequence) + values = [resolved_results[index] for index in ordered] + steps = [recorded_steps[index] for index in ordered] + failure = cast(ActivityFailed | ChildWorkflowFailed | None, + resolved_results[selected] if selected is not None else None) + boundary = (recorded_steps[selected].history_index if selected is not None + else max(cast(int, step.history_index) for step in steps)) + while pending_receivers and pending_receivers[0].history_index < cast(int, boundary): + if pending_receivers[0].condition_wait_id is not None: + break + _apply_receiver(pending_receivers.pop(0)) + # A late loser can settle on either side of successor work. Remove its + # outcome by durable sequence so it never occupies the successor cursor. + # Keep receiver positions relative to the remaining ordinary outcomes. + for receiver in pending_receivers: + receiver.result_index -= sum(index < receiver.result_index for index in slots) + for index in sorted(slots, reverse=True): + del resolved_results[index] + del recorded_steps[index] + consumed_policy_sequences.update(range(base, base + len(commands))) + return values, steps, failure + def _advance_selection_clock(base: int, size: int, failure: BaseException | None) -> None: indexes = [ selection_resolution_indexes[sequence] for sequence in range(base, base + size) @@ -7576,6 +7699,57 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None prepared_group = prepare_local_activities and _contains_local_activity(cmd) cmd, result_shape = _annotate_parallel_commands(cmd, current_call_sequence if prepared_group else None) needed = len(cmd) + if cmd and cmd[0]._parallel_group_path[0].get("parallel_group_policy") == "propagate": + if prepared_group or any(isinstance(leaf, RecordLocalActivity) for leaf in cmd): + raise LocalActivityExecutionAborted("parallel_group_policy requires qualified local admission") + consumed = _consume_propagating_group(cmd) + if consumed is None: + _set_replaying(False) + for offset, leaf in enumerate(cmd): + _assert_pending_step_matches(leaf, offset) + return _state([*pending, *cmd]) + vals, consumed_steps, failed = consumed + if failed is not None: + failed_step = next(step for step, value in zip(consumed_steps, vals, strict=True) + if value is failed) + _advance_cancellation_clock([failed_step.history_index]) + leaf = cmd[failed_step.workflow_sequence - cmd[0]._parallel_group_path[0][ + "parallel_group_base_sequence"]] + if isinstance(leaf, ScheduleActivity): + ctx._business_clock.observe_propagated_activity_failure( + (failed_step.history_index, failed_step.workflow_sequence, + leaf.activity_type, leaf._parallel_group_path), reconstruction=reconstruction_only, + ) + else: + ctx._business_clock.unavailable() + try: + _refresh_replay_logging() + advanced_cmd = _resume_business(gen.throw, failed) + continue + except StopIteration as stop: + return _terminal_state(stop.value, include_pending=False) + _advance_cancellation_clock(step.history_index for step in consumed_steps) + try: + typed_values = [_typed_activity_result(leaf, value) + for leaf, value in zip(cmd, vals, strict=True)] + except TypedValueError as error: + ctx._business_clock.unavailable() + try: + _refresh_replay_logging() + advanced_cmd = _resume_business(gen.throw, error) + continue + except StopIteration as stop: + return _terminal_state(stop.value, include_pending=False) + if all(isinstance(leaf, ScheduleActivity) for leaf in cmd): + ctx._business_clock.observe_activity_group( + [(step.history_index, step.workflow_sequence, leaf.activity_type, leaf._parallel_group_path) + for step, leaf in zip(consumed_steps, cmd, strict=True)], + None, reconstruction=reconstruction_only, + ) + else: + ctx._business_clock.unavailable() + next_value = _parallel_result_shape(result_shape, typed_values) + continue if prepared_group: group = _prepared_group(cmd) if group is not None: diff --git a/tests/fixtures/replay_regressions/explicit-parallel-policy-requires-consumer.json b/tests/fixtures/replay_regressions/explicit-parallel-policy-requires-consumer.json new file mode 100644 index 0000000..f632c04 --- /dev/null +++ b/tests/fixtures/replay_regressions/explicit-parallel-policy-requires-consumer.json @@ -0,0 +1,108 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "explicit-parallel-policy-requires-consumer", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.parallel-failure-policy", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-01-01T00:00:00Z", + "payload": {}, + "sequence": 1 + }, + { + "event_type": "ActivityScheduled", + "recorded_at": "2026-01-01T00:00:00Z", + "payload": { + "sequence": 1, + "activity_execution_id": "execution-0", + "activity_type": "slow-first", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_policy": "propagate" + } + ], + "parallel_group_policy": "propagate" + }, + "sequence": 2 + }, + { + "event_type": "ActivityScheduled", + "recorded_at": "2026-01-01T00:00:00Z", + "payload": { + "sequence": 2, + "activity_execution_id": "execution-1", + "activity_type": "fast-second", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_policy": "propagate" + } + ], + "parallel_group_policy": "propagate" + }, + "sequence": 3 + }, + { + "event_type": "ActivityFailed", + "recorded_at": "2026-01-01T00:00:01Z", + "payload": { + "sequence": 2, + "activity_execution_id": "execution-1", + "activity_type": "fast-second", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_policy": "propagate" + } + ], + "message": "higher position failed first", + "parallel_group_policy": "propagate" + }, + "sequence": 4 + } + ], + "expected": { + "command_sequence": [] + }, + "expected_replay_error": { + "type": "LocalActivityExecutionAborted", + "message_contains": "parallel_group_policy_not_supported" + } +} diff --git a/tests/fixtures/replay_regressions/legacy-parallel-failure-remains-pending.json b/tests/fixtures/replay_regressions/legacy-parallel-failure-remains-pending.json index 980e702..42c86d5 100644 --- a/tests/fixtures/replay_regressions/legacy-parallel-failure-remains-pending.json +++ b/tests/fixtures/replay_regressions/legacy-parallel-failure-remains-pending.json @@ -9,17 +9,14 @@ "workflow": { "type": "tests.replay.parallel-failure-policy", "input": [], - "payload_codec": "avro", - "query": { - "name": "caught", - "arguments": [] - } + "payload_codec": "avro" }, "history": [ { "event_type": "WorkflowStarted", "recorded_at": "2026-01-01T00:00:00Z", - "payload": {} + "payload": {}, + "sequence": 1 }, { "event_type": "ActivityScheduled", @@ -42,7 +39,8 @@ "parallel_group_index": 0 } ] - } + }, + "sequence": 2 }, { "event_type": "ActivityScheduled", @@ -65,7 +63,8 @@ "parallel_group_index": 1 } ] - } + }, + "sequence": 3 }, { "event_type": "ActivityFailed", @@ -89,11 +88,14 @@ } ], "message": "higher position failed first" - } + }, + "sequence": 4 } ], "expected": { - "command_sequence": [], - "query_result": null + "command_sequence": [ + {"type": "schedule_activity", "activity_type": "slow-first"}, + {"type": "schedule_activity", "activity_type": "fast-second"} + ] } } diff --git a/tests/test_parallel_group_policy.py b/tests/test_parallel_group_policy.py index 846ab48..b396b8f 100644 --- a/tests/test_parallel_group_policy.py +++ b/tests/test_parallel_group_policy.py @@ -1,11 +1,25 @@ from __future__ import annotations import json +from copy import deepcopy from pathlib import Path -from durable_workflow import workflow -from durable_workflow.errors import ActivityFailed -from durable_workflow.workflow import CompleteWorkflow, WorkflowContext, query_state, replay +import pytest + +from durable_workflow import serializer, workflow +from durable_workflow.errors import ActivityFailed, ChildWorkflowFailed, TypedValueError +from durable_workflow.workflow import ( + CompleteWorkflow, + LocalActivityExecutionAborted, + WorkflowContext, + _read_parallel_group_policies, + _replay_state, + commands_to_server_commands, + query_state, + replay, +) +from tests.business_clock_workflows import ParallelFailureBusinessClockWorkflow, clock_text +from tests.test_parallel_completion_order_replay import ParallelCompletionOrderMixed, _terminal_event FIXTURE = Path(__file__).parent / "fixtures/replay_regressions/legacy-parallel-failure-remains-pending.json" @@ -43,3 +57,279 @@ def test_absent_policy_preserves_positional_failure_after_a_later_sibling_settle assert len(outcome.commands) == 1 assert isinstance(outcome.commands[0], CompleteWorkflow) assert outcome.commands[0].result == "slow-first" + + +def explicit_history(policy: str = "propagate") -> list[dict]: + history = json.loads(FIXTURE.read_text())["history"] + for event in history: + payload = event["payload"] + if "parallel_group_id" in payload: + payload["parallel_group_policy"] = policy + for frame in payload["parallel_group_path"]: + frame["parallel_group_policy"] = policy + return history + + +def draft_replay(history: list[dict], definition: type = ParallelFailurePolicyWorkflow): + return _replay_state( + definition, history, [], payload_codec="avro", parallel_group_policies=("propagate",), + ).outcome + + +@pytest.mark.parametrize("policy", ["propagate", "cancel_remaining", "settle"]) +def test_public_replay_does_not_silently_interpret_an_unsupported_recorded_policy(policy: str) -> None: + with pytest.raises(LocalActivityExecutionAborted, match="parallel_group_policy_not_supported"): + replay(ParallelFailurePolicyWorkflow, explicit_history(policy), []) + + +def test_explicit_propagate_raises_while_the_other_member_is_open() -> None: + outcome = draft_replay(explicit_history()) + assert len(outcome.commands) == 1 + assert isinstance(outcome.commands[0], CompleteWorkflow) + assert outcome.commands[0].result == "fast-second" + + +def test_enabling_draft_support_does_not_change_absent_policy_history() -> None: + history = json.loads(FIXTURE.read_text())["history"] + assert not any(isinstance(command, CompleteWorkflow) for command in draft_replay(history).commands) + + +def test_enabling_draft_support_keeps_the_legacy_failure_clock_gap() -> None: + fixture = json.loads(FIXTURE.with_name("business-clock-parallel-failure-selection-gap.json").read_text()) + outcome = draft_replay(fixture["history"], ParallelFailureBusinessClockWorkflow) + assert outcome.commands[0].result == fixture["expected"]["command_sequence"][0]["result"] + + +def test_new_draft_groups_record_the_policy_but_public_authoring_stays_legacy() -> None: + ordinary = commands_to_server_commands(replay(ParallelFailurePolicyWorkflow, [], []).commands, "queue") + explicit = commands_to_server_commands(draft_replay([]).commands, "queue") + for encoded in ordinary: + assert "parallel_group_policy" not in encoded + for encoded in explicit: + assert encoded["parallel_group_policy"] == "propagate" + assert all(frame["parallel_group_policy"] == "propagate" for frame in encoded["parallel_group_path"]) + + +def test_recorded_event_order_selects_the_failure_even_when_timestamps_tie() -> None: + history = explicit_history() + payload = deepcopy(history[1]["payload"]) + payload["message"] = "lower position failed second" + history.append({"event_type": "ActivityFailed", "recorded_at": history[-1]["recorded_at"], "payload": payload}) + assert draft_replay(history).commands[0].result == "fast-second" + + +@pytest.mark.parametrize("value", [None, "", "unknown", 1, True, ["propagate"]]) +def test_malformed_policy_fails_closed(value: object) -> None: + history = explicit_history() + history[1]["payload"]["parallel_group_policy"] = value + history[1]["payload"]["parallel_group_path"][0]["parallel_group_policy"] = value + with pytest.raises(ValueError, match="parallel_group_policy"): + draft_replay(history) + + +def test_inconsistent_members_or_flattened_policy_fail_closed() -> None: + history = explicit_history() + del history[1]["payload"]["parallel_group_policy"] + with pytest.raises(ValueError, match="canonical path"): + draft_replay(history) + del history[1]["payload"]["parallel_group_path"][0]["parallel_group_policy"] + with pytest.raises(ValueError, match="inconsistent"): + draft_replay(history) + + +def test_policy_cannot_change_flattened_group_identity() -> None: + history = explicit_history() + history[1]["payload"]["parallel_group_id"] = "another-group" + with pytest.raises(ValueError, match="flattened group identity"): + draft_replay(history) + + +def test_mixed_legacy_and_explicit_nested_policies_are_not_interpreted() -> None: + history = explicit_history() + for event in history[1:]: + payload = event["payload"] + outer = {**payload["parallel_group_path"][0], "parallel_group_id": "outer"} + del outer["parallel_group_policy"] + payload["parallel_group_path"].insert(0, outer) + with pytest.raises(LocalActivityExecutionAborted, match="mixed nested composition"): + draft_replay(history) + + +def test_select_frame_refuses_policy_but_an_enclosing_all_frame_can_have_it() -> None: + root = deepcopy(explicit_history()[1]["payload"]["parallel_group_path"][0]) + selection = {**root, "parallel_group_id": "selection-group", "parallel_group_size": 1, + "parallel_group_mode": "select", "selection_member_key": "first"} + del selection["parallel_group_policy"] + payload = {**selection, "sequence": 1, "parallel_group_path": [root, selection]} + history = [{"event_type": "ActivityScheduled", "payload": payload}] + policies = _read_parallel_group_policies(history, ("propagate",)) + assert list(policies.values()) == ["propagate"] + payload["parallel_group_policy"] = selection["parallel_group_policy"] = "propagate" + with pytest.raises(ValueError, match="select frames"): + _read_parallel_group_policies(history, ("propagate",)) + + +@workflow.defn(name="tests.replay.parallel-policy-tail") +class ParallelPolicyTailWorkflow: + def __init__(self) -> None: + self.signals: list[str] = [] + + @workflow.signal("mark") + def mark(self, value: str) -> None: + self.signals.append(value) + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + try: + yield [ctx.schedule_activity("slow-first", []), ctx.schedule_activity("fast-second", [])] + except ActivityFailed: + at_failure = {"clock": clock_text(ctx), "signals": list(self.signals)} + tail = yield ctx.schedule_activity("after-group", []) + return {"at_failure": at_failure, "tail": tail, "clock": clock_text(ctx), "signals": self.signals} + + +def tail_history() -> list[dict]: + history = explicit_history() + history.extend([ + {"event_type": "ActivityScheduled", "recorded_at": "2026-01-01T00:00:01Z", "payload": { + "sequence": 3, "activity_type": "after-group", "activity_execution_id": "tail", + }}, + {"event_type": "ActivityCompleted", "recorded_at": "2026-01-01T00:00:02Z", "payload": { + "sequence": 3, "activity_type": "after-group", "activity_execution_id": "tail", + "result": serializer.encode("tail-result", codec="avro"), + }}, + {"event_type": "ActivityCompleted", "recorded_at": "2026-01-01T00:00:09Z", "payload": { + **deepcopy(history[1]["payload"]), "result": serializer.encode("late-loser", codec="avro"), + }}, + ]) + return history + + +def test_late_sibling_cannot_take_a_successor_result_or_advance_the_failure_clock() -> None: + outcome = draft_replay(tail_history(), ParallelPolicyTailWorkflow) + assert outcome.commands[0].result == { + "at_failure": {"clock": "2026-01-01T00:00:01.000000+00:00", "signals": []}, + "tail": "tail-result", "clock": "2026-01-01T00:00:02.000000+00:00", "signals": [], + } + + +def test_signals_stay_on_their_side_of_the_first_failure_boundary() -> None: + history = tail_history() + def signal(value: str) -> dict: + return {"event_type": "SignalReceived", "payload": { + "signal_name": "mark", "arguments": serializer.encode([value], codec="avro"), "payload_codec": "avro", + }} + history.insert(3, signal("before")) + history.insert(5, signal("after")) + result = draft_replay(history, ParallelPolicyTailWorkflow).commands[0].result + assert result["at_failure"]["signals"] == ["before"] + assert result["signals"] == ["before", "after"] + assert result["tail"] == "tail-result" + + +def test_successful_explicit_groups_keep_member_order() -> None: + @workflow.defn(name="tests.parallel-policy-success") + class Success: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + return (yield [ctx.schedule_activity("slow-first", []), ctx.schedule_activity("fast-second", [])]) + history = explicit_history()[:3] + for index in (1, 0): + payload = deepcopy(history[index + 1]["payload"]) + payload["result"] = serializer.encode(f"position-{index}", codec="avro") + history.append({"event_type": "ActivityCompleted", "recorded_at": "2026-01-01T00:00:02Z", "payload": payload}) + assert draft_replay(history, Success).commands[0].result == ["position-0", "position-1"] + + +def test_invalid_policy_is_refused_before_constructing_application_state() -> None: + @workflow.defn(name="tests.parallel-policy-no-application-side-effect") + class NoApplication(ParallelFailurePolicyWorkflow): + calls = 0 + + def __init__(self) -> None: + type(self).calls += 1 + super().__init__() + with pytest.raises(ValueError, match="parallel_group_policy"): + draft_replay(explicit_history("unknown"), NoApplication) + assert NoApplication.calls == 0 + + +def test_typed_success_decoding_errors_are_raised_at_the_group_yield() -> None: + @workflow.defn(name="tests.parallel-policy-typed-failure") + class Typed: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + try: + yield [ctx.schedule_activity("slow-first", [], result_type=int), + ctx.schedule_activity("fast-second", [], result_type=int)] + except TypedValueError: + return "handled" + history = explicit_history()[:3] + for index in (1, 0): + payload = deepcopy(history[index + 1]["payload"]) + payload["result"] = serializer.encode("not-an-int", codec="avro") + history.append({"event_type": "ActivityCompleted", "payload": payload}) + assert draft_replay(history, Typed).commands[0].result == "handled" + + +def test_nested_groups_propagate_without_waiting_for_open_members() -> None: + @workflow.defn(name="tests.parallel-policy-nested") + class Nested: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + try: + yield [ctx.schedule_activity("slow-first", []), + [ctx.schedule_activity("middle", []), ctx.schedule_activity("fast-third", [])]] + except ActivityFailed as error: + return error.activity_type + history = [{"event_type": "WorkflowStarted", "payload": {}}] + for index, name in enumerate(("slow-first", "middle", "fast-third")): + root = {"parallel_group_id": "parallel-activities:1:3", "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, "parallel_group_size": 3, "parallel_group_index": index, + "parallel_group_policy": "propagate"} + path = [root] + if index: + path.append({**root, "parallel_group_id": "parallel-activities:2:2", "parallel_group_base_sequence": 2, + "parallel_group_size": 2, "parallel_group_index": index - 1}) + history.append({"event_type": "ActivityScheduled", "payload": { + "sequence": index + 1, "activity_type": name, **path[-1], "parallel_group_path": path, + }}) + history.append({"event_type": "ActivityFailed", "payload": {**deepcopy(history[-1]["payload"]), + "message": "failed first"}}) + assert draft_replay(history, Nested).commands[0].result == "fast-third" + + +def with_explicit_policy(history: list[dict]) -> list[dict]: + for event in history: + payload = event["payload"] + payload["parallel_group_policy"] = "propagate" + for frame in payload["parallel_group_path"]: + frame["parallel_group_policy"] = "propagate" + return history + + +def test_explicit_mixed_success_keeps_activity_child_and_timer_positions() -> None: + history = with_explicit_policy([ + _terminal_event("TimerFired", "mixed", 3, 2), + _terminal_event("ChildRunCompleted", "mixed", 3, 1, value="child-result", + child_workflow_type="mixed-child"), + _terminal_event("ActivityCompleted", "mixed", 3, 0, value="activity-result", + activity_type="mixed-activity"), + ]) + assert draft_replay(history, ParallelCompletionOrderMixed).commands[0].result == [ + "activity-result", "child-result", None, + ] + + +def test_explicit_child_failure_does_not_wait_for_a_timer_or_activity() -> None: + @workflow.defn(name="tests.parallel-policy-mixed-child-failure") + class Mixed: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + try: + yield [ctx.schedule_activity("open", []), ctx.start_child_workflow("failed-child", []), + ctx.start_timer(30)] + except ChildWorkflowFailed as error: + return error.child_workflow_type + history = with_explicit_policy([ + _terminal_event("ActivityScheduled", "mixed", 3, 0, activity_type="open"), + _terminal_event("ChildWorkflowScheduled", "mixed", 3, 1, workflow_type="failed-child"), + _terminal_event("TimerScheduled", "mixed", 3, 2), + _terminal_event("ChildRunFailed", "mixed", 3, 1, child_workflow_type="failed-child", message="failed"), + ]) + assert draft_replay(history, Mixed).commands[0].result == "failed-child" From 7fe3d7eb57170ca585ecdad7ae7e5a41b517514f Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 10:42:01 +0000 Subject: [PATCH 3/3] Separate parallel policy validation boundaries --- src/durable_workflow/workflow.py | 117 +++++++++++++++++----------- tests/test_parallel_group_policy.py | 6 +- 2 files changed, 77 insertions(+), 46 deletions(-) diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 76c4a1a..6ce6349 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -4026,6 +4026,78 @@ def _parallel_group_entry(payload: Mapping[str, Any]) -> tuple[str, str, int, in return group_id, kind, base_sequence, size, index +def _parallel_policy_frames(payload: Mapping[str, Any]) -> tuple[list[Any], bool]: + path = payload.get("parallel_group_path") + frames = path if isinstance(path, list) and path else [payload] + has_explicit = "parallel_group_policy" in payload or any( + isinstance(frame, Mapping) and "parallel_group_policy" in frame for frame in frames + ) + if has_explicit and path is not None and not isinstance(path, list): + raise ValueError("parallel_group_policy requires a canonical group path") + if isinstance(path, list) and path: + _assert_parallel_policy_alias(payload, path[-1], has_explicit) + return frames, has_explicit + + +def _assert_parallel_policy_alias(payload: Mapping[str, Any], inner: Any, has_explicit: bool) -> None: + if not isinstance(inner, Mapping): + if has_explicit: + raise ValueError("parallel_group_policy requires matching flattened group identity") + return + if (("parallel_group_policy" in payload) != ("parallel_group_policy" in inner) + or payload.get("parallel_group_policy") != inner.get("parallel_group_policy")): + raise ValueError("parallel_group_policy disagrees with its canonical path") + if has_explicit and _parallel_group_entry(payload) != _parallel_group_entry(inner): + raise ValueError("parallel_group_policy requires matching flattened group identity") + + +def _parallel_policy_value(frame: Mapping[str, Any]) -> str | None: + if "parallel_group_policy" not in frame: + return None + policy = frame["parallel_group_policy"] + if not isinstance(policy, str) or policy not in {"propagate", "cancel_remaining", "settle"}: + raise ValueError("parallel_group_policy is unknown or malformed") + if frame.get("parallel_group_mode") == "select": + raise ValueError("parallel_group_policy does not apply to select frames") + return policy + + +def _parallel_policy_entry( + frame: Any, *, strict: bool, +) -> tuple[tuple[str, str, int, int], str | None] | None: + if not isinstance(frame, Mapping): + if strict: + raise ValueError("parallel_group_policy requires canonical group metadata") + return None + policy = _parallel_policy_value(frame) + if frame.get("parallel_group_mode") == "select": + return None + entry = _parallel_group_entry(frame) + if entry is None: + if strict: + raise ValueError("parallel_group_policy requires canonical group metadata") + return None + return entry[:4], policy + + +def _merge_parallel_group_policies( + payload: Mapping[str, Any], policies: dict[tuple[str, str, int, int], str | None], +) -> None: + frames, has_explicit = _parallel_policy_frames(payload) + ordinary_policies: list[str | None] = [] + for frame in frames: + parsed = _parallel_policy_entry(frame, strict=has_explicit) + if parsed is None: + continue + key, recorded = parsed + ordinary_policies.append(recorded) + if key in policies and policies[key] != recorded: + raise ValueError("parallel_group_policy is inconsistent across members or history") + policies[key] = recorded + if has_explicit and None in ordinary_policies: + raise LocalActivityExecutionAborted("parallel_group_policy mixed nested composition is not supported") + + def _read_parallel_group_policies( events: list[dict[str, Any]], supported: tuple[str, ...], ) -> dict[tuple[str, str, int, int], str | None]: @@ -4041,50 +4113,7 @@ def _read_parallel_group_policies( payload = event.get("payload") if not isinstance(payload, Mapping): continue - path = payload.get("parallel_group_path") - frames = path if isinstance(path, list) and path else [payload] - has_explicit = "parallel_group_policy" in payload or any( - isinstance(frame, Mapping) and "parallel_group_policy" in frame for frame in frames - ) - if has_explicit and path is not None and not isinstance(path, list): - raise ValueError("parallel_group_policy requires a canonical group path") - if isinstance(path, list) and path and isinstance(path[-1], Mapping) and ( - ("parallel_group_policy" in payload) != ("parallel_group_policy" in path[-1]) - or payload.get("parallel_group_policy") != path[-1].get("parallel_group_policy") - ): - raise ValueError("parallel_group_policy disagrees with its canonical path") - if isinstance(path, list) and path and any( - isinstance(frame, Mapping) and "parallel_group_policy" in frame for frame in path - ) and (not isinstance(path[-1], Mapping) - or _parallel_group_entry(payload) != _parallel_group_entry(path[-1])): - raise ValueError("parallel_group_policy requires matching flattened group identity") - ordinary_policies: list[str | None] = [] - for frame in frames: - if not isinstance(frame, Mapping): - if has_explicit: - raise ValueError("parallel_group_policy requires canonical group metadata") - continue - present = "parallel_group_policy" in frame - policy = frame.get("parallel_group_policy") - if present and (not isinstance(policy, str) or policy not in {"propagate", "cancel_remaining", "settle"}): - raise ValueError("parallel_group_policy is unknown or malformed") - if frame.get("parallel_group_mode") == "select": - if present: - raise ValueError("parallel_group_policy does not apply to select frames") - continue - entry = _parallel_group_entry(frame) - if entry is None: - if has_explicit: - raise ValueError("parallel_group_policy requires canonical group metadata") - continue - key = entry[:4] - recorded = cast(str | None, policy) if present else None - ordinary_policies.append(recorded) - if key in policies and policies[key] != recorded: - raise ValueError("parallel_group_policy is inconsistent across members or history") - policies[key] = recorded - if has_explicit and None in ordinary_policies: - raise LocalActivityExecutionAborted("parallel_group_policy mixed nested composition is not supported") + _merge_parallel_group_policies(payload, policies) if any(policy is not None and policy not in supported for policy in policies.values()): raise LocalActivityExecutionAborted( "parallel_group_policy_not_supported: explicit history requires a qualified policy consumer", diff --git a/tests/test_parallel_group_policy.py b/tests/test_parallel_group_policy.py index b396b8f..a2ad6b0 100644 --- a/tests/test_parallel_group_policy.py +++ b/tests/test_parallel_group_policy.py @@ -78,8 +78,9 @@ def draft_replay(history: list[dict], definition: type = ParallelFailurePolicyWo @pytest.mark.parametrize("policy", ["propagate", "cancel_remaining", "settle"]) def test_public_replay_does_not_silently_interpret_an_unsupported_recorded_policy(policy: str) -> None: + history = explicit_history(policy) with pytest.raises(LocalActivityExecutionAborted, match="parallel_group_policy_not_supported"): - replay(ParallelFailurePolicyWorkflow, explicit_history(policy), []) + replay(ParallelFailurePolicyWorkflow, history, []) def test_explicit_propagate_raises_while_the_other_member_is_open() -> None: @@ -247,8 +248,9 @@ class NoApplication(ParallelFailurePolicyWorkflow): def __init__(self) -> None: type(self).calls += 1 super().__init__() + history = explicit_history("unknown") with pytest.raises(ValueError, match="parallel_group_policy"): - draft_replay(explicit_history("unknown"), NoApplication) + draft_replay(history, NoApplication) assert NoApplication.calls == 0