From bb6e7f2f45114b4d3f06a07409d6784d6c358dfa Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 02:52:14 +0000 Subject: [PATCH 1/3] Replay histories whose run close settled open activities Workflow 2.28.0 (workflow#1117, workflow#1121) and Server (server#1193) record ActivityCancelled with cancellation_reason run_closed for each activity a run never awaited when its own code completes, fails or continues as new, just before the terminal event. Abandon-policy activities are listed in the terminal event's abandoned_activities. Replay now drops those run-close cancellations before any history reader sees them, so the activity stays open as the workflow saw it: the cancellation never resolves a select() member, never supplies replay time and never becomes a list group's first positional failure. Before, a close's cancellation of a lower-index member could replace the failure the workflow handled. Activity workers treat a run_closed refusal as final: a completion accepted with recorded false counts as completed (the abandon-policy case), a refused failure or cancellation report is not repeated and logs at info, and a claim refused with run_closed is an answered poll without failure backoff. Heartbeat refusals keep raising ActivityLeaseAuthorityLost without publishing an outcome. Adds replay tests and two corpus fixtures shaped like workflow#1121's events, and worker tests for each refusal. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 5 + src/durable_workflow/worker.py | 29 + src/durable_workflow/workflow.py | 22 +- ...osed-cancellation-not-a-group-outcome.json | 343 +++++++++ .../selection-loser-settled-at-close.json | 390 +++++++++++ tests/test_activity_run_closed_refusal.py | 186 +++++ tests/test_replay_regression_corpus.py | 3 + tests/test_run_closed_settlement_replay.py | 658 ++++++++++++++++++ 8 files changed, 1635 insertions(+), 1 deletion(-) create mode 100644 tests/fixtures/replay_regressions/run-closed-cancellation-not-a-group-outcome.json create mode 100644 tests/fixtures/replay_regressions/selection-loser-settled-at-close.json create mode 100644 tests/test_activity_run_closed_refusal.py create mode 100644 tests/test_run_closed_settlement_replay.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 8d3f6ae..e42df0b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- Replay supports histories in which a run's own close settled the work it never awaited (server#1193, workflow#1117/#1121, Workflow 2.28.0). When a run completes, fails or continues as new, each still-open activity, such as a `select()` loser, gets an `ActivityCancelled` with `cancellation_reason: run_closed` just before the terminal event. Replay, query and update replay now skip that cancellation and leave the activity open, as the workflow saw it. It never resolves a selection member, never supplies replay time, and is never taken as a list group's first positional failure. Before this fix, a close's cancellation of a lower-index member could replace the failure the workflow handled. Replay also accepts the new `abandoned_activities` key on `WorkflowCompleted`, `WorkflowFailed` and `WorkflowContinuedAsNew`, which lists abandon-policy activities left running at close. +- Activity workers treat a `run_closed` refusal as final. A completion refused with `recorded: false, reason: run_closed` counts as completed. This is how Server accepts the outcome of an abandon-policy activity that kept running after its run closed. A failure or cancellation report refused with `run_closed` is not reported again and logs at info level. A claim refused with `run_closed` counts as an answered poll, with no failure backoff. A heartbeat refused with `run_closed` raises `ActivityLeaseAuthorityLost` and publishes no outcome, as before. + ## [2.16.0] - 2026-11-12 ### Added diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index 797a441..08af72b 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -165,6 +165,19 @@ _MULTIPLEXED_WORKFLOW_POLL_ENDPOINT = "/worker/workflow-tasks/poll" _TERMINAL_WORKFLOW_STATUSES = {"completed", "failed", "terminated", "canceled", "cancelled"} + +# The typed refusal for late activity work on a run whose own code closed it +# (server#1193, workflow#1117). The run settled the activity at close, or left +# it running under the abandon policy, and nothing is ever appended after the +# terminal event, so the refusal is final: never retried or reported again. +_RUN_CLOSED_REFUSAL = "run_closed" + + +def _run_closed_refusal(error: BaseException) -> bool: + """Whether a claim or outcome request was refused because its run closed.""" + if isinstance(error, ActivityLeaseAuthorityLost): + return error.reason == _RUN_CLOSED_REFUSAL + return isinstance(error, ServerError) and error.reason() == _RUN_CLOSED_REFUSAL _QUERY_TASK_FINAL_REJECTION_REASONS = { "lease_expired", "query_task_not_found", @@ -3380,6 +3393,11 @@ async def _run_activity_task_body(self, task: dict[str, Any]) -> str: **self._external_storage_completion_kwargs(), ) except Exception as e: + if _run_closed_refusal(e): + # Final: an activity the run left running under the abandon + # policy ends with this outcome, any other was settled at close. + log.info("activity task %s outcome not recorded: its workflow run closed", task_id) + return "completed" log.warning("failed to complete activity task %s: %s", task_id, e) return "complete_error" return "completed" @@ -3435,6 +3453,9 @@ async def _report_activity_task_failure(self, task_id: str, attempt_id: str, task_id=task_id, activity_attempt_id=attempt_id, lease_owner=self.worker_id, **failure, ) except Exception as error: + if _run_closed_refusal(error): + log.info("%s for activity task %s not recorded: its workflow run closed", context, task_id) + return log.warning("failed to report %s: %s", context, error) async def _invoke_remote_activity(self, task: dict[str, Any], activity_type: str, @@ -4045,6 +4066,14 @@ async def _poll_activity_tasks(self) -> None: if self._stop.is_set(): self._status.request_abandoned("activity_poll") return + if _run_closed_refusal(e): + # The claim's run closed before it started: the Server + # settled that task, so this answered poll is no failure. + log.info("activity claim refused because its workflow run closed: %s", e) + failures.reset() + self._status.request_succeeded("activity_poll") + self._record_poll_metrics("activity", "empty", time.perf_counter() - poll_start) + continue delay = _poll_capacity_delay(e, "activity_task", self.task_queue) self._record_poll_status("activity_poll", e, answered=delay is not None) if _is_storage_admission_error(e): diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 8999efe..373fbbc 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -3955,6 +3955,23 @@ def _history_event_type(event: Mapping[str, Any]) -> str | None: return value if isinstance(value, str) and value else None +# ``cancellation_reason`` of an ActivityCancelled that the run's own close +# recorded (Workflow 2.28.0, workflow#1117 and server#1193). Completing, +# failing or continuing as new cancels each activity the workflow never +# awaited, such as a select() loser, in the closing transaction just before the +# terminal event. The workflow had already moved past those operations, so the +# cancellation is never an outcome replay can observe: the activity stays open. +RUN_CLOSED_CANCELLATION_REASON = "run_closed" + + +def _settled_at_close(event: Any) -> bool: + """Whether this is an activity cancellation recorded by its run's close.""" + if not isinstance(event, Mapping) or _history_event_type(event) != "ActivityCancelled": + return False + payload = event.get("payload") + return isinstance(payload, Mapping) and payload.get("cancellation_reason") == RUN_CLOSED_CANCELLATION_REASON + + def _workflow_sequence(payload: Mapping[str, Any]) -> int | None: value = payload.get("sequence") or payload.get("workflow_sequence") if isinstance(value, int): @@ -4865,7 +4882,10 @@ def _replay_state( exception_type=type(exception).__name__, ) from exception - events = list(history_events) + # A run-close cancellation never resolves a select() member, never fails + # 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)] # A malformed delivery span refuses replay before any history parser can # read it more permissively. _assert_delivery_spans(events) diff --git a/tests/fixtures/replay_regressions/run-closed-cancellation-not-a-group-outcome.json b/tests/fixtures/replay_regressions/run-closed-cancellation-not-a-group-outcome.json new file mode 100644 index 0000000..4a4c533 --- /dev/null +++ b/tests/fixtures/replay_regressions/run-closed-cancellation-not-a-group-outcome.json @@ -0,0 +1,343 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "run-closed-cancellation-not-a-group-outcome", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "run_closed_cancellation_not_a_group_outcome": { + "issues": [ + "durable-workflow/server#1193", + "durable-workflow/workflow#1117", + "durable-workflow/workflow#1121" + ], + "note": "Core's V2OpenOperationsAtCloseTest all() vector with frozen time. The run's close cancels the still-open lower-index member in the same instant as the other member's failure. That cancellation is not an outcome the workflow observed: the list group stays as the workflow saw it, so the handler state never holds the close's cancellation (before the fix it held 'activity failed')." + }, + "workflow": { + "type": "run-closed-all-failure", + "input": [], + "payload_codec": "avro", + "query": { + "name": "caught", + "arguments": [] + } + }, + "history": [ + { + "id": "01m795h48s87krk3abrcesqf39", + "sequence": 1, + "event_type": "StartAccepted", + "payload": { + "workflow_command_id": "01m795h48pxb9ks6zxaxxzcxkg", + "workflow_instance_id": "gen-1117", + "workflow_run_id": "01m795h48k4a44yjrzbhx5hvaa", + "workflow_class": "run-closed-all-failure", + "workflow_type": "run-closed-all-failure", + "business_key": null, + "visibility_labels": null, + "memo": null, + "search_attributes": null, + "outcome": "started_new" + }, + "recorded_at": "2026-11-13T00:39:42.103872Z" + }, + { + "id": "01m795h48vrbq18nwm3g08va0x", + "sequence": 2, + "event_type": "WorkflowStarted", + "payload": { + "workflow_class": "run-closed-all-failure", + "workflow_type": "run-closed-all-failure", + "workflow_instance_id": "gen-1117", + "workflow_run_id": "01m795h48k4a44yjrzbhx5hvaa", + "workflow_command_id": "01m795h48pxb9ks6zxaxxzcxkg", + "business_key": null, + "visibility_labels": null, + "memo": null, + "search_attributes": null, + "execution_timeout_seconds": null, + "run_timeout_seconds": null, + "execution_deadline_at": null, + "run_deadline_at": null, + "declared_queries": [], + "declared_query_contracts": [], + "declared_signals": [], + "declared_signal_contracts": [], + "declared_updates": [], + "declared_update_contracts": [], + "declared_entry_method": "handle", + "declared_entry_mode": "canonical" + }, + "recorded_at": "2026-11-13T00:39:42.107521Z" + }, + { + "id": "scheduled-1", + "event_type": "ActivityScheduled", + "payload": { + "activity_execution_id": "activity-1", + "activity_class": "still-running", + "activity_type": "still-running", + "sequence": 1, + "activity": { + "id": "activity-1", + "idempotency_key": "activity-1", + "sequence": 1, + "type": "still-running", + "class": "still-running", + "local_activity": false, + "status": "pending", + "payload_codec": "avro", + "attempt_count": 0, + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 1, + "backoff_seconds": [ + 1, + 2, + 5, + 10, + 15, + 30, + 60, + 120 + ], + "start_to_close_timeout": null, + "schedule_to_start_timeout": null, + "schedule_to_close_timeout": null, + "heartbeat_timeout": null, + "non_retryable_error_types": [] + }, + "connection": "redis", + "queue": "default", + "created_at": "2026-11-13T00:39:42.143497Z", + "arguments": "wwHioz3/VYAiNwwA", + "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_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 + } + ] + }, + "recorded_at": "2026-11-13T00:39:42.150118Z" + }, + { + "id": "scheduled-2", + "event_type": "ActivityScheduled", + "payload": { + "activity_execution_id": "activity-2", + "activity_class": "failing", + "activity_type": "failing", + "sequence": 2, + "activity": { + "id": "activity-2", + "idempotency_key": "activity-2", + "sequence": 2, + "type": "failing", + "class": "failing", + "local_activity": false, + "status": "pending", + "payload_codec": "avro", + "attempt_count": 0, + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 1, + "backoff_seconds": [ + 1, + 2, + 5, + 10, + 15, + 30, + 60, + 120 + ], + "start_to_close_timeout": null, + "schedule_to_start_timeout": null, + "schedule_to_close_timeout": null, + "heartbeat_timeout": null, + "non_retryable_error_types": [] + }, + "connection": "redis", + "queue": "default", + "created_at": "2026-11-13T00:39:42.143497Z", + "arguments": "wwHioz3/VYAiNwwA", + "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_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 + } + ] + }, + "recorded_at": "2026-11-13T00:39:42.150118Z" + }, + { + "id": "failed-2", + "event_type": "ActivityFailed", + "payload": { + "activity_execution_id": "activity-2", + "activity_class": "failing", + "activity_type": "failing", + "sequence": 2, + "attempt_number": 1, + "failure_category": "application", + "non_retryable": false, + "exception_type": "RuntimeError", + "exception_class": "RuntimeError", + "message": "boom", + "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 + } + ] + }, + "recorded_at": "2026-11-13T00:39:42.150118Z" + }, + { + "id": "cancelled-1", + "event_type": "ActivityCancelled", + "payload": { + "workflow_command_id": null, + "activity_execution_id": "activity-1", + "activity_attempt_id": null, + "activity_class": "still-running", + "activity_type": "still-running", + "sequence": 1, + "attempt_number": 0, + "cancelled_at": "2026-11-13T00:39:42.150118Z", + "activity": { + "id": "activity-1", + "idempotency_key": "activity-1", + "sequence": 1, + "type": "still-running", + "class": "still-running", + "local_activity": false, + "status": "cancelled", + "payload_codec": "avro", + "attempt_count": 0, + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 1, + "backoff_seconds": [ + 1, + 2, + 5, + 10, + 15, + 30, + 60, + 120 + ], + "start_to_close_timeout": null, + "schedule_to_start_timeout": null, + "schedule_to_close_timeout": null, + "heartbeat_timeout": null, + "non_retryable_error_types": [] + }, + "connection": "redis", + "queue": "default", + "created_at": "2026-11-13T00:39:42.143497Z", + "arguments": "wwHioz3/VYAiNwwA", + "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 + } + ] + }, + "activity_attempt": null, + "cancellation_reason": "run_closed", + "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 + } + ] + }, + "recorded_at": "2026-11-13T00:39:42.150118Z" + }, + { + "id": "completed", + "event_type": "WorkflowCompleted", + "payload": { + "output": "wwHioz3/VYAiNwoYY2F1Z2h0OiBib29t", + "payload_codec": "avro" + }, + "recorded_at": "2026-11-13T00:39:42.150118Z" + } + ], + "expected": { + "query_result": null + } +} diff --git a/tests/fixtures/replay_regressions/selection-loser-settled-at-close.json b/tests/fixtures/replay_regressions/selection-loser-settled-at-close.json new file mode 100644 index 0000000..6f4e648 --- /dev/null +++ b/tests/fixtures/replay_regressions/selection-loser-settled-at-close.json @@ -0,0 +1,390 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "selection-loser-settled-at-close", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "selection_loser_settled_at_close": { + "issues": [ + "durable-workflow/server#1193", + "durable-workflow/workflow#1117", + "durable-workflow/workflow#1121" + ], + "note": "Python port of workflow#1121's selection-loser-settled-at-close corpus fixture, with the close recorded. The zero timer wins the select(); completing the run records ActivityCancelled with cancellation_reason run_closed for the never-awaited activity loser just before WorkflowCompleted. Replay never treats that cancellation as an outcome and reaches the same close." + }, + "workflow": { + "type": "test-selection-loser-at-close-workflow", + "input": [ + "complete" + ], + "payload_codec": "avro", + "run_id": "01m795h48k4a44yjrzbhx5hvaa" + }, + "history": [ + { + "id": "01m795h48s87krk3abrcesqf39", + "sequence": 1, + "event_type": "StartAccepted", + "payload": { + "workflow_command_id": "01m795h48pxb9ks6zxaxxzcxkg", + "workflow_instance_id": "gen-1117", + "workflow_run_id": "01m795h48k4a44yjrzbhx5hvaa", + "workflow_class": "test-selection-loser-at-close-workflow", + "workflow_type": "test-selection-loser-at-close-workflow", + "business_key": null, + "visibility_labels": null, + "memo": null, + "search_attributes": null, + "outcome": "started_new" + }, + "recorded_at": "2026-11-13T00:39:42.103872Z" + }, + { + "id": "01m795h48vrbq18nwm3g08va0x", + "sequence": 2, + "event_type": "WorkflowStarted", + "payload": { + "workflow_class": "test-selection-loser-at-close-workflow", + "workflow_type": "test-selection-loser-at-close-workflow", + "workflow_instance_id": "gen-1117", + "workflow_run_id": "01m795h48k4a44yjrzbhx5hvaa", + "workflow_command_id": "01m795h48pxb9ks6zxaxxzcxkg", + "business_key": null, + "visibility_labels": null, + "memo": null, + "search_attributes": null, + "execution_timeout_seconds": null, + "run_timeout_seconds": null, + "execution_deadline_at": null, + "run_deadline_at": null, + "declared_queries": [], + "declared_query_contracts": [], + "declared_signals": [], + "declared_signal_contracts": [], + "declared_updates": [], + "declared_update_contracts": [], + "declared_entry_method": "handle", + "declared_entry_mode": "canonical" + }, + "recorded_at": "2026-11-13T00:39:42.107521Z" + }, + { + "id": "01m795h4a14c2beezexqk6zwqd", + "event_type": "ActivityScheduled", + "payload": { + "activity_execution_id": "01m795h49zm0xftwdqam8nqj81", + "activity_class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "sequence": 1, + "activity": { + "id": "01m795h49zm0xftwdqam8nqj81", + "idempotency_key": "01m795h49zm0xftwdqam8nqj81", + "sequence": 1, + "type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "local_activity": false, + "status": "pending", + "payload_codec": "avro", + "attempt_count": 0, + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 1, + "backoff_seconds": [ + 1, + 2, + 5, + 10, + 15, + 30, + 60, + 120 + ], + "start_to_close_timeout": null, + "schedule_to_start_timeout": null, + "schedule_to_close_timeout": null, + "heartbeat_timeout": null, + "non_retryable_error_types": [] + }, + "connection": "redis", + "queue": "default", + "created_at": "2026-11-13T00:39:42.143497Z", + "arguments": "wwHioz3/VYAiNwwA", + "parallel_group_id": "select-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_mode": "select", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "selection_member_key": "work", + "selection_member_index": 0, + "selection_member_base_sequence": 1, + "selection_member_size": 1, + "selection_member_kind": "activity", + "parallel_group_path": [ + { + "parallel_group_id": "select-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_mode": "select", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "selection_member_key": "work", + "selection_member_index": 0, + "selection_member_base_sequence": 1, + "selection_member_size": 1, + "selection_member_kind": "activity" + } + ] + }, + "parallel_group_id": "select-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_mode": "select", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "selection_member_key": "work", + "selection_member_index": 0, + "selection_member_base_sequence": 1, + "selection_member_size": 1, + "selection_member_kind": "activity", + "parallel_group_path": [ + { + "parallel_group_id": "select-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_mode": "select", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "selection_member_key": "work", + "selection_member_index": 0, + "selection_member_base_sequence": 1, + "selection_member_size": 1, + "selection_member_kind": "activity" + } + ] + }, + "recorded_at": "2026-11-13T00:39:42.144915Z", + "sequence": 3 + }, + { + "id": "01m795h4a344awn80w7wnvztxr", + "sequence": 4, + "event_type": "TimerScheduled", + "payload": { + "timer_id": "01m795h4a2vty5vv2b0j103gpv", + "sequence": 2, + "delay_seconds": 0, + "fire_at": "2026-11-13T00:39:42.146411Z", + "parallel_group_id": "select-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_mode": "select", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "selection_member_key": "deadline", + "selection_member_index": 1, + "selection_member_base_sequence": 2, + "selection_member_size": 1, + "selection_member_kind": "timer", + "parallel_group_path": [ + { + "parallel_group_id": "select-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_mode": "select", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "selection_member_key": "deadline", + "selection_member_index": 1, + "selection_member_base_sequence": 2, + "selection_member_size": 1, + "selection_member_kind": "timer" + } + ] + }, + "recorded_at": "2026-11-13T00:39:42.146930Z" + }, + { + "id": "01m795h4a466hy5ckc65z8bgwa", + "sequence": 5, + "event_type": "TimerFired", + "payload": { + "timer_id": "01m795h4a2vty5vv2b0j103gpv", + "sequence": 2, + "delay_seconds": 0, + "fired_at": "2026-11-13T00:39:42.146411Z", + "parallel_group_id": "select-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_mode": "select", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "selection_member_key": "deadline", + "selection_member_index": 1, + "selection_member_base_sequence": 2, + "selection_member_size": 1, + "selection_member_kind": "timer", + "parallel_group_path": [ + { + "parallel_group_id": "select-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_mode": "select", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "selection_member_key": "deadline", + "selection_member_index": 1, + "selection_member_base_sequence": 2, + "selection_member_size": 1, + "selection_member_kind": "timer" + } + ] + }, + "recorded_at": "2026-11-13T00:39:42.147973Z" + }, + { + "id": "01m795h4a5xa436wzevyv0zqq2", + "sequence": 6, + "event_type": "SelectionResolved", + "payload": { + "selection_group_id": "select-calls:1:2", + "selection_group_base_sequence": 1, + "selection_group_size": 2, + "member_key": "deadline", + "member_index": 1, + "member_base_sequence": 2, + "member_size": 1, + "operation_kind": "timer", + "operation_identity": "01m795h4a2vty5vv2b0j103gpv", + "outcome": "completed", + "resolution_event_id": "01m795h4a466hy5ckc65z8bgwa", + "resolution_event_type": "TimerFired" + }, + "recorded_at": "2026-11-13T00:39:42.149382Z" + }, + { + "id": "01m795h4a6cancelledatclose0", + "event_type": "ActivityCancelled", + "payload": { + "workflow_command_id": null, + "activity_execution_id": "01m795h49zm0xftwdqam8nqj81", + "activity_attempt_id": null, + "activity_class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "activity_type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "sequence": 1, + "attempt_number": 0, + "cancelled_at": "2026-11-13T00:39:42.150118Z", + "activity": { + "id": "01m795h49zm0xftwdqam8nqj81", + "idempotency_key": "01m795h49zm0xftwdqam8nqj81", + "sequence": 1, + "type": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "class": "Tests\\Fixtures\\V2\\TestGreetingActivity", + "local_activity": false, + "status": "cancelled", + "payload_codec": "avro", + "attempt_count": 0, + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 1, + "backoff_seconds": [ + 1, + 2, + 5, + 10, + 15, + 30, + 60, + 120 + ], + "start_to_close_timeout": null, + "schedule_to_start_timeout": null, + "schedule_to_close_timeout": null, + "heartbeat_timeout": null, + "non_retryable_error_types": [] + }, + "connection": "redis", + "queue": "default", + "created_at": "2026-11-13T00:39:42.143497Z", + "arguments": "wwHioz3/VYAiNwwA", + "parallel_group_id": "select-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_mode": "select", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "selection_member_key": "work", + "selection_member_index": 0, + "selection_member_base_sequence": 1, + "selection_member_size": 1, + "selection_member_kind": "activity", + "parallel_group_path": [ + { + "parallel_group_id": "select-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_mode": "select", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "selection_member_key": "work", + "selection_member_index": 0, + "selection_member_base_sequence": 1, + "selection_member_size": 1, + "selection_member_kind": "activity" + } + ] + }, + "activity_attempt": null, + "cancellation_reason": "run_closed", + "parallel_group_id": "select-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_mode": "select", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "selection_member_key": "work", + "selection_member_index": 0, + "selection_member_base_sequence": 1, + "selection_member_size": 1, + "selection_member_kind": "activity", + "parallel_group_path": [ + { + "parallel_group_id": "select-calls:1:2", + "parallel_group_kind": "mixed", + "parallel_group_mode": "select", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "selection_member_key": "work", + "selection_member_index": 0, + "selection_member_base_sequence": 1, + "selection_member_size": 1, + "selection_member_kind": "activity" + } + ] + }, + "recorded_at": "2026-11-13T00:39:42.150118Z", + "sequence": 7 + }, + { + "id": "01m795h4a7terminal00000000", + "sequence": 8, + "event_type": "WorkflowCompleted", + "payload": { + "output": "wwHioz3/VYAiNwooY2xvc2VkLXdpdGgtZGVhZGxpbmU=", + "payload_codec": "avro" + }, + "recorded_at": "2026-11-13T00:39:42.150118Z" + } + ], + "expected": { + "command_sequence": [ + { + "type": "complete_workflow", + "result": "closed-with-deadline" + } + ] + } +} diff --git a/tests/test_activity_run_closed_refusal.py b/tests/test_activity_run_closed_refusal.py new file mode 100644 index 0000000..1fe56de --- /dev/null +++ b/tests/test_activity_run_closed_refusal.py @@ -0,0 +1,186 @@ +"""Activity workers treat a ``run_closed`` refusal as final (server#1193). + +When a run's own code closes it, Workflow 2.28.0 (workflow#1117/#1121) and +Server cancel each activity it never awaited in the closing transaction, or +leave an abandon-policy activity running, and append nothing after the +terminal event. A late claim, heartbeat or outcome for such an activity is +answered with the typed ``run_closed`` refusal. The worker must stop there: +no retry, no second report, and no poll failure backoff. +""" + +from __future__ import annotations + +import asyncio +import logging +from typing import Any +from unittest.mock import AsyncMock + +import pytest + +from durable_workflow import activity, serializer +from durable_workflow.activity import ActivityContext, ActivityInfo +from durable_workflow.client import Client +from durable_workflow.errors import ActivityCancelled, ActivityLeaseAuthorityLost, ServerError +from durable_workflow.worker import Worker + +RUN_CLOSED_OUTCOME = { + "task_id": "late-task", + "activity_attempt_id": "late-attempt", + "outcome": "completed", + "recorded": False, + "reason": "run_closed", + "next_task_id": None, +} + + +def _task(activity_type: str) -> dict[str, Any]: + return { + "task_id": "late-task", "activity_attempt_id": "late-attempt", + "activity_type": activity_type, "attempt_number": 1, + "arguments": serializer.encode([], codec="avro"), "payload_codec": "avro", + } + + +def _worker(client: AsyncMock, handler: Any) -> Worker: + return Worker(client, task_queue="late-queue", workflows=[], activities=[handler]) + + +@pytest.mark.asyncio +async def test_completion_refused_because_the_run_closed_is_final(caplog: pytest.LogCaptureFixture) -> None: + """An abandon-policy activity kept running after close reports its outcome; + the Server accepts it with ``recorded: false, reason: run_closed``.""" + client = AsyncMock(spec=Client) + client.complete_activity_task.side_effect = ServerError(409, {**RUN_CLOSED_OUTCOME}) + + @activity.defn(name="abandoned-at-close") + async def handler() -> str: + return "independent result" + + with caplog.at_level(logging.INFO, logger="durable_workflow.worker"): + assert await _worker(client, handler)._run_activity_task(_task("abandoned-at-close")) == "completed" + + client.complete_activity_task.assert_awaited_once() + client.fail_activity_task.assert_not_called() + assert not [record for record in caplog.records if record.levelno >= logging.WARNING] + + +@pytest.mark.asyncio +async def test_cancellation_at_close_is_reported_once_and_its_refusal_is_final( + caplog: pytest.LogCaptureFixture, +) -> None: + """A running select() loser cancelled at close: its heartbeat reports the + cancellation, and the late report is refused with run_closed.""" + client = AsyncMock(spec=Client) + client.heartbeat_activity_task.return_value = { + "can_continue": False, "cancel_requested": True, "reason": "activity_cancelled", + } + client.fail_activity_task.side_effect = ServerError( + 409, {**RUN_CLOSED_OUTCOME, "outcome": "failed"}, + ) + cleaned_up: list[str] = [] + + @activity.defn(name="loser-cancelled-at-close") + async def handler() -> str: + try: + await activity.context().heartbeat() + except ActivityCancelled: + cleaned_up.append("cleanup") + raise + return "late result" + + with caplog.at_level(logging.INFO, logger="durable_workflow.worker"): + assert await _worker(client, handler)._run_activity_task(_task("loser-cancelled-at-close")) == "cancelled" + + assert cleaned_up == ["cleanup"] + client.complete_activity_task.assert_not_called() + client.fail_activity_task.assert_awaited_once() + assert client.fail_activity_task.await_args.kwargs["non_retryable"] is True + assert client.fail_activity_task.await_args.kwargs["failure_type"] == "ActivityCancelled" + assert not [record for record in caplog.records if record.levelno >= logging.WARNING] + + +@pytest.mark.asyncio +async def test_failure_refused_because_the_run_closed_is_final() -> None: + client = AsyncMock(spec=Client) + client.fail_activity_task.side_effect = ServerError(409, {**RUN_CLOSED_OUTCOME, "outcome": "failed"}) + + @activity.defn(name="fails-after-close") + async def handler() -> str: + raise RuntimeError("late failure") + + assert await _worker(client, handler)._run_activity_task(_task("fails-after-close")) == "failed" + + client.fail_activity_task.assert_awaited_once() + client.complete_activity_task.assert_not_called() + + +@pytest.mark.asyncio +async def test_heartbeat_refused_because_the_run_closed_stops_without_any_outcome() -> None: + client = AsyncMock(spec=Client) + client.heartbeat_activity_task.return_value = { + "can_continue": False, "cancel_requested": False, "reason": "run_closed", + } + continued: list[str] = [] + + @activity.defn(name="heartbeat-after-close") + async def handler() -> str: + await activity.context().heartbeat() + continued.append("after refusal") + return "late result" + + assert await _worker(client, handler)._run_activity_task(_task("heartbeat-after-close")) == "claim_aborted" + + assert continued == [] + client.heartbeat_activity_task.assert_awaited_once() + client.complete_activity_task.assert_not_called() + client.fail_activity_task.assert_not_called() + + +@pytest.mark.asyncio +async def test_heartbeat_of_an_abandoned_activity_keeps_running() -> None: + client = AsyncMock(spec=Client) + client.heartbeat_activity_task.return_value = {"can_continue": True, "cancel_requested": False} + ctx = ActivityContext(info=ActivityInfo( + task_id="late-task", activity_type="abandoned-at-close", activity_attempt_id="late-attempt", + attempt_number=1, task_queue="late-queue", worker_id="late-worker", + ), client=client) + + await ctx.heartbeat({"message": "still going"}) + + assert not ctx.is_cancelled + + +@pytest.mark.asyncio +async def test_claim_refused_because_the_run_closed_is_an_answered_poll( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, +) -> None: + client = AsyncMock(spec=Client) + + @activity.defn(name="claimed-after-close") + async def handler() -> str: + return "never runs" + + worker = Worker(client, task_queue="late-queue", workflows=[], activities=[handler], poll_timeout=0.01) + polls: list[str] = [] + + async def poll(**_: object) -> None: + polls.append("poll") + if len(polls) == 1: + raise ActivityLeaseAuthorityLost("late-task", "run_closed") + await worker.stop() + return None + + async def no_failure_backoff(*_: object) -> None: + raise AssertionError("a run_closed claim refusal is not a poll failure") + + client.poll_activity_task.side_effect = poll + monkeypatch.setattr(asyncio, "sleep", no_failure_backoff) + dispatched: list[dict[str, Any]] = [] + monkeypatch.setattr(worker, "_dispatch_activity_task", lambda task: dispatched.append(task)) + + with caplog.at_level(logging.INFO, logger="durable_workflow.worker"): + await worker._poll_activity_tasks() + + assert polls == ["poll", "poll"] + assert dispatched == [] + assert not [record for record in caplog.records if record.levelno >= logging.WARNING] diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index e16c59d..96bb41d 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -65,6 +65,7 @@ RedriveVisibilityCleanupWorkflow, RedriveVisibilityWorkflow, ) +from tests.test_run_closed_settlement_replay import AllFailureClosedWorkflow, SelectionLoserAtCloseWorkflow from tests.test_typed_values import TypedActivityResultsWorkflow from tests.test_typed_workflow_values import TypedOnboarding, TypedTerminalResult from tests.test_update_signal_condition_replay import ( @@ -431,6 +432,8 @@ def run(self, ctx: WorkflowContext, message: str) -> None: SelectionAwaitMarkerWorkflow, SelectionCancellationQueryWorkflow, SelectionMissingNestedOpeningWorkflow, + SelectionLoserAtCloseWorkflow, + AllFailureClosedWorkflow, UpdateSignalConditionTimerWorkflow, WorkflowStreamAuthorWorkflow, WorkflowMemoAuthorWorkflow, diff --git a/tests/test_run_closed_settlement_replay.py b/tests/test_run_closed_settlement_replay.py new file mode 100644 index 0000000..4bf036b --- /dev/null +++ b/tests/test_run_closed_settlement_replay.py @@ -0,0 +1,658 @@ +"""Replay of histories whose run close settled still-open work (server#1193). + +Workflow 2.28.0 (workflow#1117, workflow#1121) and Server (server#1193) record +``ActivityCancelled`` with ``cancellation_reason: run_closed`` for each +activity a run never awaited when its own code completes, fails or continues +as new, in the closing transaction just before the terminal event. Pending +timers get ``TimerCancelled``. An activity scheduled with the abandon policy is +listed in the terminal event's ``abandoned_activities`` instead. + +The workflow had already moved past those operations, so replay must never +treat such a cancellation as an outcome: it must not resolve a ``select()`` +member and must not replace the failure an ``all()`` group raised. + +The event shapes below are copied from workflow#1121: the corpus fixture +``selection-loser-settled-at-close.json`` and ``ActivityCancellation::record``. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from durable_workflow import serializer, workflow +from durable_workflow.errors import ActivityFailed +from durable_workflow.workflow import ( + CompleteWorkflow, + ContinueAsNew, + FailWorkflow, + SelectionResult, + WorkflowContext, + query_state, + replay, +) + +RUN_ID = "01m795h48k4a44yjrzbhx5hvaa" +INSTANCE_ID = "gen-1117" +CLOSED_AT = "2026-11-13T00:39:42.150118Z" + + +def _started(workflow_type: str) -> list[dict[str, Any]]: + return [ + { + "id": "01m795h48s87krk3abrcesqf39", + "sequence": 1, + "event_type": "StartAccepted", + "payload": { + "workflow_command_id": "01m795h48pxb9ks6zxaxxzcxkg", + "workflow_instance_id": INSTANCE_ID, + "workflow_run_id": RUN_ID, + "workflow_class": workflow_type, + "workflow_type": workflow_type, + "business_key": None, + "visibility_labels": None, + "memo": None, + "search_attributes": None, + "outcome": "started_new", + }, + "recorded_at": "2026-11-13T00:39:42.103872Z", + }, + { + "id": "01m795h48vrbq18nwm3g08va0x", + "sequence": 2, + "event_type": "WorkflowStarted", + "payload": { + "workflow_class": workflow_type, + "workflow_type": workflow_type, + "workflow_instance_id": INSTANCE_ID, + "workflow_run_id": RUN_ID, + "workflow_command_id": "01m795h48pxb9ks6zxaxxzcxkg", + "business_key": None, + "visibility_labels": None, + "memo": None, + "search_attributes": None, + "execution_timeout_seconds": None, + "run_timeout_seconds": None, + "execution_deadline_at": None, + "run_deadline_at": None, + "declared_queries": [], + "declared_query_contracts": [], + "declared_signals": [], + "declared_signal_contracts": [], + "declared_updates": [], + "declared_update_contracts": [], + "declared_entry_method": "handle", + "declared_entry_mode": "canonical", + }, + "recorded_at": "2026-11-13T00:39:42.107521Z", + }, + ] + + +def _member( + *, + group_id: str, + kind: str, + size: int, + index: int, + key: str | None = None, + member_index: int | None = None, + member_base: int | None = None, + member_size: int = 1, + member_kind: str | None = None, +) -> dict[str, Any]: + entry: dict[str, Any] = { + "parallel_group_id": group_id, + "parallel_group_kind": kind, + } + if key is not None: + entry["parallel_group_mode"] = "select" + entry.update( + { + "parallel_group_base_sequence": 1, + "parallel_group_size": size, + "parallel_group_index": index, + } + ) + if key is not None: + entry.update( + { + "selection_member_key": key, + "selection_member_index": member_index, + "selection_member_base_sequence": member_base, + "selection_member_size": member_size, + "selection_member_kind": member_kind, + } + ) + return entry + + +def _with_path(path: list[dict[str, Any]]) -> dict[str, Any]: + return {**path[-1], "parallel_group_path": path} + + +def _activity_snapshot( + execution_id: str, activity_type: str, sequence: int, status: str, path: list[dict[str, Any]], +) -> dict[str, Any]: + return { + "id": execution_id, + "idempotency_key": execution_id, + "sequence": sequence, + "type": activity_type, + "class": activity_type, + "local_activity": False, + "status": status, + "payload_codec": "avro", + "attempt_count": 0, + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 1, + "backoff_seconds": [1, 2, 5, 10, 15, 30, 60, 120], + "start_to_close_timeout": None, + "schedule_to_start_timeout": None, + "schedule_to_close_timeout": None, + "heartbeat_timeout": None, + "non_retryable_error_types": [], + }, + "connection": "redis", + "queue": "default", + "created_at": "2026-11-13T00:39:42.143497Z", + "arguments": serializer.envelope([])["blob"], + **_with_path(path), + } + + +def _activity_scheduled( + event_id: str, sequence: int, execution_id: str, activity_type: str, path: list[dict[str, Any]], + *, recorded_at: str = "2026-11-13T00:39:42.144915Z", +) -> dict[str, Any]: + return { + "id": event_id, + "event_type": "ActivityScheduled", + "payload": { + "activity_execution_id": execution_id, + "activity_class": activity_type, + "activity_type": activity_type, + "sequence": sequence, + "activity": _activity_snapshot(execution_id, activity_type, sequence, "pending", path), + **_with_path(path), + }, + "recorded_at": recorded_at, + } + + +def _activity_cancelled_at_close( + event_id: str, sequence: int, execution_id: str, activity_type: str, path: list[dict[str, Any]], + *, recorded_at: str = CLOSED_AT, +) -> dict[str, Any]: + """``ActivityCancellation::record`` with ``cancellationReason: run_closed``.""" + return { + "id": event_id, + "event_type": "ActivityCancelled", + "payload": { + "workflow_command_id": None, + "activity_execution_id": execution_id, + "activity_attempt_id": None, + "activity_class": activity_type, + "activity_type": activity_type, + "sequence": sequence, + "attempt_number": 0, + "cancelled_at": recorded_at, + "activity": _activity_snapshot(execution_id, activity_type, sequence, "cancelled", path), + "activity_attempt": None, + "cancellation_reason": "run_closed", + **_with_path(path), + }, + "recorded_at": recorded_at, + } + + +# --- select(): the loser is cancelled when the run closes ------------------- + +SELECT_WORK = [ + _member( + group_id="select-calls:1:2", kind="mixed", size=2, index=0, + key="work", member_index=0, member_base=1, member_kind="activity", + ) +] +SELECT_DEADLINE = [ + _member( + group_id="select-calls:1:2", kind="mixed", size=2, index=1, + key="deadline", member_index=1, member_base=2, member_kind="timer", + ) +] + + +@workflow.defn(name="test-selection-loser-at-close-workflow") +class SelectionLoserAtCloseWorkflow: + """Python port of workflow#1121's ``TestSelectionLoserAtCloseWorkflow``.""" + + def run(self, ctx: WorkflowContext, mode: str, round: int = 0): # type: ignore[no-untyped-def] + if round > 0: + return "continued" + selected: SelectionResult = yield ctx.select( + { + "work": ctx.schedule_activity("Tests\\Fixtures\\V2\\TestGreetingActivity", [mode]), + "deadline": ctx.start_timer(0), + } + ) + if mode == "fail": + raise RuntimeError("Closing with an open select() loser.") + if mode == "continue": + return (yield ctx.continue_as_new(mode, 1)) + return "closed-with-" + selected.key + + +def _selection_history_before_close() -> list[dict[str, Any]]: + """The ``history`` of ``selection-loser-settled-at-close.json``.""" + timer_fields = {"timer_id": "01m795h4a2vty5vv2b0j103gpv", "sequence": 2, "delay_seconds": 0} + return [ + *_started("test-selection-loser-at-close-workflow"), + { + **_activity_scheduled( + "01m795h4a14c2beezexqk6zwqd", 1, "01m795h49zm0xftwdqam8nqj81", + "Tests\\Fixtures\\V2\\TestGreetingActivity", SELECT_WORK, + ), + "sequence": 3, + }, + { + "id": "01m795h4a344awn80w7wnvztxr", + "sequence": 4, + "event_type": "TimerScheduled", + "payload": { + **timer_fields, + "fire_at": "2026-11-13T00:39:42.146411Z", + **_with_path(SELECT_DEADLINE), + }, + "recorded_at": "2026-11-13T00:39:42.146930Z", + }, + { + "id": "01m795h4a466hy5ckc65z8bgwa", + "sequence": 5, + "event_type": "TimerFired", + "payload": { + **timer_fields, + "fired_at": "2026-11-13T00:39:42.146411Z", + **_with_path(SELECT_DEADLINE), + }, + "recorded_at": "2026-11-13T00:39:42.147973Z", + }, + { + "id": "01m795h4a5xa436wzevyv0zqq2", + "sequence": 6, + "event_type": "SelectionResolved", + "payload": { + "selection_group_id": "select-calls:1:2", + "selection_group_base_sequence": 1, + "selection_group_size": 2, + "member_key": "deadline", + "member_index": 1, + "member_base_sequence": 2, + "member_size": 1, + "operation_kind": "timer", + "operation_identity": "01m795h4a2vty5vv2b0j103gpv", + "outcome": "completed", + "resolution_event_id": "01m795h4a466hy5ckc65z8bgwa", + "resolution_event_type": "TimerFired", + }, + "recorded_at": "2026-11-13T00:39:42.149382Z", + }, + ] + + +def _loser_cancelled_at_close() -> dict[str, Any]: + return { + **_activity_cancelled_at_close( + "01m795h4a6cancelledatclose0", 1, "01m795h49zm0xftwdqam8nqj81", + "Tests\\Fixtures\\V2\\TestGreetingActivity", SELECT_WORK, + ), + "sequence": 7, + } + + +def _terminal(mode: str) -> dict[str, Any]: + if mode == "complete": + return { + "id": "01m795h4a7terminal00000000", + "sequence": 8, + "event_type": "WorkflowCompleted", + "payload": { + "output": serializer.envelope("closed-with-deadline")["blob"], + "payload_codec": "avro", + }, + "recorded_at": CLOSED_AT, + } + if mode == "fail": + return { + "id": "01m795h4a7terminal00000000", + "sequence": 8, + "event_type": "WorkflowFailed", + "payload": { + "failure_id": "01m795h4a7failure000000000", + "source_kind": "workflow_run", + "source_id": RUN_ID, + "failure_category": "application", + "non_retryable": False, + "exception_type": "RuntimeError", + "exception_class": "RuntimeError", + "message": "Closing with an open select() loser.", + "failed_step_kind": "workflow_code", + }, + "recorded_at": CLOSED_AT, + } + return { + "id": "01m795h4a7terminal00000000", + "sequence": 8, + "event_type": "WorkflowContinuedAsNew", + "payload": { + "sequence": 3, + "continued_to_run_id": "01m795h4a8continuedrun0000", + "continued_to_run_number": 2, + "workflow_link_id": "01m795h4a8link000000000000", + "closed_reason": "continued", + }, + "recorded_at": CLOSED_AT, + } + + +def _assert_close(outcome: Any, mode: str) -> None: + assert len(outcome.commands) == 1 + command = outcome.commands[0] + if mode == "complete": + assert isinstance(command, CompleteWorkflow) + assert command.result == "closed-with-deadline" + elif mode == "fail": + assert isinstance(command, FailWorkflow) + assert command.message == "Closing with an open select() loser." + else: + assert isinstance(command, ContinueAsNew) + assert command.arguments == ["continue", 1] + + +@pytest.mark.parametrize("mode", ["complete", "fail", "continue"]) +def test_select_loser_cancelled_at_close_replays_to_the_same_close(mode: str) -> None: + open_history = _selection_history_before_close() + closed_history = [*open_history, _loser_cancelled_at_close(), _terminal(mode)] + + live = replay(SelectionLoserAtCloseWorkflow, open_history, [mode], run_id=RUN_ID, payload_codec="avro") + cold = replay(SelectionLoserAtCloseWorkflow, closed_history, [mode], run_id=RUN_ID, payload_codec="avro") + + _assert_close(live, mode) + _assert_close(cold, mode) + + +@pytest.mark.parametrize("mode", ["complete", "fail", "continue"]) +def test_abandoned_activities_and_unknown_keys_on_the_terminal_event_are_tolerated(mode: str) -> None: + terminal = _terminal(mode) + terminal["payload"]["abandoned_activities"] = [ + {"activity_execution_id": "01m795h49zm0xftwdqam8nqj81", "sequence": 1}, + ] + terminal["payload"]["a_key_from_a_later_server"] = {"nested": True} + cancelled = _loser_cancelled_at_close() + cancelled["payload"]["a_key_from_a_later_server"] = "ignored" + + outcome = replay( + SelectionLoserAtCloseWorkflow, + [*_selection_history_before_close(), cancelled, terminal], + [mode], + run_id=RUN_ID, + payload_codec="avro", + ) + + _assert_close(outcome, mode) + + +def test_abandoned_loser_has_no_cancellation_and_replays_to_the_same_close() -> None: + terminal = _terminal("complete") + terminal["payload"]["abandoned_activities"] = [ + {"activity_execution_id": "01m795h49zm0xftwdqam8nqj81", "sequence": 1}, + ] + + outcome = replay( + SelectionLoserAtCloseWorkflow, + [*_selection_history_before_close(), terminal], + ["complete"], + run_id=RUN_ID, + payload_codec="avro", + ) + + _assert_close(outcome, "complete") + + +# --- select() whose winner is a failed nested group ------------------------ + + +def _nested_outer(flat_index: int) -> dict[str, Any]: + return _member( + group_id="select-calls:1:3", kind="mixed", size=3, index=flat_index, + key="nested", member_index=0, member_base=1, member_size=2, member_kind="group", + ) + + +def _nested_inner(index: int) -> dict[str, Any]: + return _member(group_id="parallel-activities:1:2", kind="activity", size=2, index=index) + + +NESTED_DEADLINE = _member( + group_id="select-calls:1:3", kind="mixed", size=3, index=2, + key="deadline", member_index=1, member_base=3, member_kind="timer", +) + + +@workflow.defn(name="run-closed-nested-selection") +class NestedSelectionClosedWorkflow: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + selected: SelectionResult = yield ctx.select( + { + "nested": [ + ctx.schedule_activity("nested-first", []), + ctx.schedule_activity("nested-second", []), + ], + "deadline": ctx.start_timer(3600), + } + ) + try: + selected.result() + except ActivityFailed as failure: + return {"winner": selected.key, "failure": str(failure)} + return {"winner": selected.key, "failure": None} + + +def test_cancellation_at_close_never_becomes_the_failed_nested_winner() -> None: + """A run-close cancellation of the winning group's lower-index member in + the same millisecond as its failure must not replace the failure.""" + failed_at = CLOSED_AT + history: list[dict[str, Any]] = [] + for index, activity_type in enumerate(["nested-first", "nested-second"]): + history.append( + _activity_scheduled( + f"scheduled-{index + 1}", index + 1, f"activity-{index + 1}", activity_type, + [_nested_outer(index), _nested_inner(index)], + ) + ) + history.extend( + [ + { + "id": "timer-open", + "event_type": "TimerScheduled", + "payload": {"sequence": 3, "timer_id": "timer-3", "delay_seconds": 3600, + **_with_path([NESTED_DEADLINE])}, + "recorded_at": "2026-11-13T00:39:42.146930Z", + }, + { + "id": "nested-failure", + "event_type": "ActivityFailed", + "payload": { + "activity_execution_id": "activity-2", + "activity_type": "nested-second", + "sequence": 2, + "message": "boom", + "exception_type": "NestedFailure", + **_with_path([_nested_outer(1), _nested_inner(1)]), + }, + "recorded_at": failed_at, + }, + { + "event_type": "SelectionResolved", + "payload": { + "selection_group_id": "select-calls:1:3", + "selection_group_base_sequence": 1, + "selection_group_size": 3, + "member_key": "nested", + "member_index": 0, + "member_base_sequence": 1, + "member_size": 2, + "operation_kind": "group", + "operation_identity": "group:1:2", + "outcome": "failed", + "resolution_event_id": "nested-failure", + "resolution_event_type": "ActivityFailed", + }, + "recorded_at": failed_at, + }, + _activity_cancelled_at_close( + "first-cancelled-at-close", 1, "activity-1", "nested-first", + [_nested_outer(0), _nested_inner(0)], recorded_at=failed_at, + ), + { + "id": "timer-cancelled-at-close", + "event_type": "TimerCancelled", + "payload": {"sequence": 3, "timer_id": "timer-3", "delay_seconds": 3600, + **_with_path([NESTED_DEADLINE])}, + "recorded_at": failed_at, + }, + ] + ) + + outcome = replay(NestedSelectionClosedWorkflow, history, []) + + assert len(outcome.commands) == 1 + assert isinstance(outcome.commands[0], CompleteWorkflow) + assert outcome.commands[0].result == {"winner": "nested", "failure": "boom"} + + +# --- all(): the failure the workflow handled stays the failure -------------- + + +@workflow.defn(name="run-closed-all-failure") +class AllFailureClosedWorkflow: + def __init__(self) -> None: + self.caught: str | None = None + + @workflow.query("caught") + def caught_message(self) -> str | None: + return self.caught + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + try: + yield [ + ctx.schedule_activity("still-running", []), + ctx.schedule_activity("failing", []), + ] + except ActivityFailed as failure: + self.caught = str(failure) + return "caught: " + str(self.caught) + + +def _all_failure_history(*, cancelled_at_close: bool) -> list[dict[str, Any]]: + """``testACancellationRecordedAtCloseNeverReplacesTheFailureTheWorkflowHandled`` + with frozen time: the failure and the close's cancellation share a timestamp.""" + frozen = CLOSED_AT + + def path(index: int) -> list[dict[str, Any]]: + return [_member(group_id="parallel-activities:1:2", kind="activity", size=2, index=index)] + + history = [ + *_started("run-closed-all-failure"), + _activity_scheduled("scheduled-1", 1, "activity-1", "still-running", path(0), recorded_at=frozen), + _activity_scheduled("scheduled-2", 2, "activity-2", "failing", path(1), recorded_at=frozen), + { + "id": "failed-2", + "event_type": "ActivityFailed", + "payload": { + "activity_execution_id": "activity-2", + "activity_class": "failing", + "activity_type": "failing", + "sequence": 2, + "attempt_number": 1, + "failure_category": "application", + "non_retryable": False, + "exception_type": "RuntimeError", + "exception_class": "RuntimeError", + "message": "boom", + **_with_path(path(1)), + }, + "recorded_at": frozen, + }, + ] + if cancelled_at_close: + history.append( + _activity_cancelled_at_close( + "cancelled-1", 1, "activity-1", "still-running", path(0), recorded_at=frozen, + ) + ) + history.append( + { + "id": "completed", + "event_type": "WorkflowCompleted", + "payload": {"output": serializer.envelope("caught: boom")["blob"], "payload_codec": "avro"}, + "recorded_at": frozen, + } + ) + return history + + +def test_cancellation_at_close_never_replaces_the_failure_the_workflow_handled() -> None: + """Core's all() vector. A Python list group resumes once every member is + terminal, so the close's cancellation of the still-open lower-index member + must leave the group exactly as the workflow saw it, never complete it with + that cancellation as the group's first positional failure.""" + live = replay(AllFailureClosedWorkflow, _all_failure_history(cancelled_at_close=False), []) + cold = replay(AllFailureClosedWorkflow, _all_failure_history(cancelled_at_close=True), []) + + assert cold.commands == live.commands + assert not any(isinstance(command, CompleteWorkflow) for command in cold.commands) + assert query_state( + AllFailureClosedWorkflow, _all_failure_history(cancelled_at_close=True), [], "caught", + ) is None + + +def test_ordinary_cancellation_is_still_an_activity_outcome() -> None: + """Only ``cancellation_reason: run_closed`` is skipped.""" + history = _all_failure_history(cancelled_at_close=True)[:-1] + del history[-1]["payload"]["cancellation_reason"] + + outcome = replay(AllFailureClosedWorkflow, history, []) + + assert len(outcome.commands) == 1 + assert isinstance(outcome.commands[0], CompleteWorkflow) + assert outcome.commands[0].result == "caught: activity failed" + + +@workflow.defn(name="test-selection-loser-at-close-query-workflow") +class SelectionLoserAtCloseQueryWorkflow(SelectionLoserAtCloseWorkflow): + def __init__(self) -> None: + self.winner: str | None = None + + @workflow.query("winner") + def winner_key(self) -> str | None: + return self.winner + + def run(self, ctx: WorkflowContext, mode: str, round: int = 0): # type: ignore[no-untyped-def] + selected: SelectionResult = yield ctx.select( + { + "work": ctx.schedule_activity("Tests\\Fixtures\\V2\\TestGreetingActivity", [mode]), + "deadline": ctx.start_timer(0), + } + ) + self.winner = selected.key + return "closed-with-" + selected.key + + +def test_query_replay_of_a_closed_run_keeps_the_selection_winner() -> None: + history = [*_selection_history_before_close(), _loser_cancelled_at_close(), _terminal("complete")] + + assert query_state( + SelectionLoserAtCloseQueryWorkflow, history, ["complete"], "winner", run_id=RUN_ID, payload_codec="avro", + ) == "deadline" From e26e7c52bdd9c10856a55e2c49c9c28fdb594982 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 03:22:01 +0000 Subject: [PATCH 2/3] Split run_closed completion refusals on activity_status A completion refused with recorded:false, reason run_closed now follows the PHP worker. An abandon-policy activity whose own outcome closed it (activity_status completed or failed) is logged at info and counts as completed. An activity the close cancelled is logged as a warning and reported as aborted, never completed. Neither is retried. Refs durable-workflow/server#1193, durable-workflow/workflow#1117. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 2 +- src/durable_workflow/worker.py | 37 ++++++++++++++++++++--- tests/test_activity_run_closed_refusal.py | 37 ++++++++++++++++++++++- 3 files changed, 70 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e42df0b..18f39e3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - Replay supports histories in which a run's own close settled the work it never awaited (server#1193, workflow#1117/#1121, Workflow 2.28.0). When a run completes, fails or continues as new, each still-open activity, such as a `select()` loser, gets an `ActivityCancelled` with `cancellation_reason: run_closed` just before the terminal event. Replay, query and update replay now skip that cancellation and leave the activity open, as the workflow saw it. It never resolves a selection member, never supplies replay time, and is never taken as a list group's first positional failure. Before this fix, a close's cancellation of a lower-index member could replace the failure the workflow handled. Replay also accepts the new `abandoned_activities` key on `WorkflowCompleted`, `WorkflowFailed` and `WorkflowContinuedAsNew`, which lists abandon-policy activities left running at close. -- Activity workers treat a `run_closed` refusal as final. A completion refused with `recorded: false, reason: run_closed` counts as completed. This is how Server accepts the outcome of an abandon-policy activity that kept running after its run closed. A failure or cancellation report refused with `run_closed` is not reported again and logs at info level. A claim refused with `run_closed` counts as an answered poll, with no failure backoff. A heartbeat refused with `run_closed` raises `ActivityLeaseAuthorityLost` and publishes no outcome, as before. +- Activity workers treat a `run_closed` refusal as final. A completion refused with `recorded: false, reason: run_closed` is split on its `activity_status`, as in the PHP worker. When it is `completed` or `failed`, the activity was an abandon-policy activity that kept running after its run closed and its own outcome closed it: the task counts as completed and logs at info level. Otherwise the close cancelled the activity: its outcome is dropped, the task is reported as aborted (never completed), and a warning is logged. A failure or cancellation report refused with `run_closed` is not reported again and logs at info level. A claim refused with `run_closed` counts as an answered poll, with no failure backoff. A heartbeat refused with `run_closed` raises `ActivityLeaseAuthorityLost` and publishes no outcome, as before. ## [2.16.0] - 2026-11-12 diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index 08af72b..eaea0dc 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -178,6 +178,23 @@ def _run_closed_refusal(error: BaseException) -> bool: if isinstance(error, ActivityLeaseAuthorityLost): return error.reason == _RUN_CLOSED_REFUSAL return isinstance(error, ServerError) and error.reason() == _RUN_CLOSED_REFUSAL + + +def _run_closed_completion_kept(error: BaseException) -> bool | None: + """Classify a completion refused with ``run_closed``, as the PHP worker does. + + ``True``: an abandon-policy activity its run left running at close, whose + own outcome closed it (``activity_status`` completed or failed); nothing + is recorded on the closed run. ``False``: the close cancelled the activity + and its outcome was dropped. ``None``: not a ``run_closed`` refusal with + ``recorded: false``. + """ + if not isinstance(error, ServerError) or error.status != 409 or error.reason() != _RUN_CLOSED_REFUSAL: + return None + body = error.body + if not isinstance(body, dict) or body.get("recorded") is not False: + return None + return body.get("activity_status") in ("completed", "failed") _QUERY_TASK_FINAL_REJECTION_REASONS = { "lease_expired", "query_task_not_found", @@ -3393,11 +3410,23 @@ async def _run_activity_task_body(self, task: dict[str, Any]) -> str: **self._external_storage_completion_kwargs(), ) except Exception as e: - if _run_closed_refusal(e): - # Final: an activity the run left running under the abandon - # policy ends with this outcome, any other was settled at close. - log.info("activity task %s outcome not recorded: its workflow run closed", task_id) + kept = _run_closed_completion_kept(e) + if kept is True: + # The run left this activity running under the abandon policy: + # its outcome closed it, and nothing is appended to the run. + log.info( + "activity task %s completed; its workflow run closed, so the outcome is not recorded on the run", + task_id, + ) return "completed" + if kept is False: + # The run's close cancelled this activity: the outcome is + # dropped, and the refusal is final, never retried. + log.warning( + "activity task %s outcome not recorded: its workflow run closed and cancelled the activity", + task_id, + ) + return "claim_aborted" log.warning("failed to complete activity task %s: %s", task_id, e) return "complete_error" return "completed" diff --git a/tests/test_activity_run_closed_refusal.py b/tests/test_activity_run_closed_refusal.py index 1fe56de..e4f611f 100644 --- a/tests/test_activity_run_closed_refusal.py +++ b/tests/test_activity_run_closed_refusal.py @@ -50,7 +50,9 @@ async def test_completion_refused_because_the_run_closed_is_final(caplog: pytest """An abandon-policy activity kept running after close reports its outcome; the Server accepts it with ``recorded: false, reason: run_closed``.""" client = AsyncMock(spec=Client) - client.complete_activity_task.side_effect = ServerError(409, {**RUN_CLOSED_OUTCOME}) + client.complete_activity_task.side_effect = ServerError( + 409, {**RUN_CLOSED_OUTCOME, "activity_status": "completed"}, + ) @activity.defn(name="abandoned-at-close") async def handler() -> str: @@ -62,6 +64,39 @@ async def handler() -> str: client.complete_activity_task.assert_awaited_once() client.fail_activity_task.assert_not_called() assert not [record for record in caplog.records if record.levelno >= logging.WARNING] + assert any( + record.levelno == logging.INFO and "late-task completed" in record.getMessage() + for record in caplog.records + ) + + +@pytest.mark.asyncio +async def test_completion_refused_because_the_close_cancelled_the_activity_is_aborted( + caplog: pytest.LogCaptureFixture, +) -> None: + """A select() loser the close cancelled finishes anyway and reports its + result; the Server drops it with ``recorded: false, reason: run_closed`` + and ``activity_status: cancelled``. The worker reports an abort, never a + completion, and never retries or reports again.""" + client = AsyncMock(spec=Client) + client.complete_activity_task.side_effect = ServerError( + 409, {**RUN_CLOSED_OUTCOME, "activity_status": "cancelled"}, + ) + + @activity.defn(name="loser-finished-after-close") + async def handler() -> str: + return "late result" + + with caplog.at_level(logging.INFO, logger="durable_workflow.worker"): + outcome = await _worker(client, handler)._run_activity_task(_task("loser-finished-after-close")) + + assert outcome == "claim_aborted" + client.complete_activity_task.assert_awaited_once() + client.fail_activity_task.assert_not_called() + warnings = [record for record in caplog.records if record.levelno == logging.WARNING] + assert len(warnings) == 1 + assert "closed and cancelled the activity" in warnings[0].getMessage() + assert not [record for record in caplog.records if "late-task completed" in record.getMessage()] @pytest.mark.asyncio From 84d91fdb8e95fde9d185261581a9c23e56fdf75e Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 09:04:56 +0000 Subject: [PATCH 3/3] Keep run-close fixes in the unreleased changelog --- CHANGELOG.md | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9383618..42b1875 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- Replay supports histories in which a run's own close settled the work it never awaited (server#1193, workflow#1117/#1121, Workflow 2.28.0). When a run completes, fails or continues as new, each still-open activity, such as a `select()` loser, gets an `ActivityCancelled` with `cancellation_reason: run_closed` just before the terminal event. Replay, query and update replay now skip that cancellation and leave the activity open, as the workflow saw it. It never resolves a selection member, never supplies replay time, and is never taken as a list group's first positional failure. Before this fix, a close's cancellation of a lower-index member could replace the failure the workflow handled. Replay also accepts the new `abandoned_activities` key on `WorkflowCompleted`, `WorkflowFailed` and `WorkflowContinuedAsNew`, which lists abandon-policy activities left running at close. +- Activity workers treat a `run_closed` refusal as final. A completion refused with `recorded: false, reason: run_closed` is split on its `activity_status`, as in the PHP worker. When it is `completed` or `failed`, the activity was an abandon-policy activity that kept running after its run closed and its own outcome closed it: the task counts as completed and logs at info level. Otherwise the close cancelled the activity: its outcome is dropped, the task is reported as aborted (never completed), and a warning is logged. A failure or cancellation report refused with `run_closed` is not reported again and logs at info level. A claim refused with `run_closed` counts as an answered poll, with no failure backoff. A heartbeat refused with `run_closed` raises `ActivityLeaseAuthorityLost` and publishes no outcome, as before. + ## [2.17.0] - 2026-11-13 ### Added @@ -21,8 +26,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - A workflow-task completion the Server refuses as too large (Server 3.6.0's typed `413 payload_too_large` with `limit_kind: workflow_task_completion_bytes`, over 16 payload limits, 32 MiB by default) now raises `WorkflowTaskCompletionTooLarge` with the Server's `limit` and the encoded request `size` (durable-workflow/.github#228). The worker no longer reports this as a workflow-task failure, which made the Server retry the same oversized completion or block replay. It never resends the refused completion: it completes the same task once with a non-retryable `fail_workflow` command whose message starts with `Structural limit exceeded: workflow_task_completion_bytes:`, so the run fails with the `structural_limit` failure category and guidance to split the fan-out, use child workflows or continue-as-new. If that follow-up is rejected too, the worker logs it and never calls `/fail`. Other 413 responses and 5xx retries are unchanged. -- Replay supports histories in which a run's own close settled the work it never awaited (server#1193, workflow#1117/#1121, Workflow 2.28.0). When a run completes, fails or continues as new, each still-open activity, such as a `select()` loser, gets an `ActivityCancelled` with `cancellation_reason: run_closed` just before the terminal event. Replay, query and update replay now skip that cancellation and leave the activity open, as the workflow saw it. It never resolves a selection member, never supplies replay time, and is never taken as a list group's first positional failure. Before this fix, a close's cancellation of a lower-index member could replace the failure the workflow handled. Replay also accepts the new `abandoned_activities` key on `WorkflowCompleted`, `WorkflowFailed` and `WorkflowContinuedAsNew`, which lists abandon-policy activities left running at close. -- Activity workers treat a `run_closed` refusal as final. A completion refused with `recorded: false, reason: run_closed` is split on its `activity_status`, as in the PHP worker. When it is `completed` or `failed`, the activity was an abandon-policy activity that kept running after its run closed and its own outcome closed it: the task counts as completed and logs at info level. Otherwise the close cancelled the activity: its outcome is dropped, the task is reported as aborted (never completed), and a warning is logged. A failure or cancellation report refused with `run_closed` is not reported again and logs at info level. A claim refused with `run_closed` counts as an answered poll, with no failure backoff. A heartbeat refused with `run_closed` raises `ActivityLeaseAuthorityLost` and publishes no outcome, as before. ## [2.16.0] - 2026-11-12