From cc654327c0c0873df660ded8b5626a3cef2d9f50 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 7 Oct 2026 07:33:02 +0000 Subject: [PATCH] Acknowledge pending replay through waiting-for-history protocol --- CHANGELOG.md | 9 ++++++ docs/sdk-reference.md | 7 +++++ pyproject.toml | 6 ++-- src/durable_workflow/client.py | 14 ++++++++- tests/test_client.py | 19 ++++++++++++ tests/test_pending_operation_replay.py | 41 +++++++++++++++++++++++++- 6 files changed, 91 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f11db86..be80423 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [2.4.2] - 2026-10-07 + +### Fixed +- Acknowledge replay waiting on an already pending timer, activity or child + through the existing waiting-for-history protocol. Empty command lists no + longer produce rejected workflow task completions. +- Reject message-stream completion metadata without commands before sending a + waiting acknowledgement, preventing stream progress from being discarded. + ## [2.4.1] - 2026-10-07 ### Fixed diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index c2cafae..c7110a8 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -12,6 +12,13 @@ in-process `WorkflowEnvironment` test harness, and invocable activity carriers. Python workers execute server-routed query tasks after the Server advertises the query-tasks capability through cluster discovery. +When replay is waiting on an already scheduled operation, the worker acknowledges +waiting for history with its original lease and attempt. It does not schedule the +operation again or fail the workflow. Low-level adapters calling +`complete_workflow_task(commands=[])` use the same acknowledgement. Message-stream +completion metadata requires nonempty commands so this path cannot discard stream +progress. + ## Install ```bash diff --git a/pyproject.toml b/pyproject.toml index 0347da4..579c708 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "durable-workflow" -version = "2.4.1" +version = "2.4.2" description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server" readme = "README.md" requires-python = ">=3.10" @@ -71,8 +71,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main" durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main" [tool.durable-workflow] -product-train = "2.4.1" -registry-version = "2.4.1" +product-train = "2.4.2" +registry-version = "2.4.2" supported-server-versions = "2.5.0" worker-protocol-version = "1.19" control-plane-version = "2" diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index 2eb1a35..f044831 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -5115,11 +5115,23 @@ async def complete_workflow_task( Worker-plane endpoint, called by :class:`~durable_workflow.Worker`. ``commands`` is the list of serialized commands the workflow yielded - for this task. + for this task. An empty list acknowledges waiting for already scheduled + history without reporting a workflow failure or scheduling another operation. """ if (message_stream_cursors or message_stream_waits) and not _worker_protocol_supports_message_streams(): raise ValueError("message stream completion metadata requires worker protocol 1.15 or newer") + if not commands: + if message_stream_cursors or message_stream_waits: + raise ValueError("Message stream completion metadata requires nonempty commands.") + return await self.fail_workflow_task( + task_id=task_id, + lease_owner=lease_owner, + workflow_task_attempt=workflow_task_attempt, + message="Workflow task waiting for scheduled history.", + failure_type="WorkflowTaskWaitingForHistory", + ) + body: dict[str, Any] = { "lease_owner": lease_owner, "workflow_task_attempt": workflow_task_attempt, diff --git a/tests/test_client.py b/tests/test_client.py index 12c13e6..384ef54 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -2903,6 +2903,25 @@ async def test_complete_workflow_task_matches_polyglot_fixture(self, client: Cli assert mock.call_args.args[:2] == (fixture["request"]["method"], f"/api{fixture['request']['path']}") assert mock.call_args.kwargs["json"] == fixture["request"]["body"] + @pytest.mark.asyncio + @pytest.mark.parametrize("metadata", [ + {"message_stream_cursors": [{"stream_name": "input", "sequence": 7}]}, + {"message_stream_waits": [{"stream_name": "input"}]}, + ]) + async def test_empty_completion_does_not_discard_message_stream_metadata( + self, client: Client, metadata: dict + ) -> None: + with ( + patch.object(client._http, "request", new_callable=AsyncMock, + return_value=_mock_response(200, {"completed": True})) as mock, + pytest.raises(ValueError, match="nonempty commands"), + ): + await client.complete_workflow_task( + task_id="task", lease_owner="owner", workflow_task_attempt=3, + commands=[], **metadata, + ) + mock.assert_not_awaited() + @pytest.mark.asyncio async def test_fail_workflow_task_matches_polyglot_fixture(self, client: Client) -> None: fixture_path = Path(__file__).parent / "fixtures" / "control-plane" / "workflow-task-fail-parity.json" diff --git a/tests/test_pending_operation_replay.py b/tests/test_pending_operation_replay.py index ba9353e..d621e50 100644 --- a/tests/test_pending_operation_replay.py +++ b/tests/test_pending_operation_replay.py @@ -2,10 +2,12 @@ from __future__ import annotations from typing import Any +from unittest.mock import AsyncMock, patch +import httpx import pytest -from durable_workflow import serializer, workflow +from durable_workflow import Client, Worker, serializer, workflow from durable_workflow.errors import NonDeterministicReplayError from durable_workflow.workflow import CompleteWorkflow, WorkflowContext, query_state, replay @@ -78,6 +80,43 @@ def test_signal_replay_does_not_reschedule_pending_operation(kind: str) -> None: ) +@pytest.mark.asyncio +@pytest.mark.parametrize("kind", OPENINGS) +async def test_worker_acknowledges_pending_signal_replay_with_original_lease(kind: str) -> None: + history = initial_history() + [OPENINGS[kind], { + "event_type": "SignalReceived", "payload": { + "signal_name": "increment", "arguments": serializer.envelope([7]), + }, + }] + requests = [] + + async def server(method, path, **kwargs): # type: ignore[no-untyped-def] + body = kwargs["json"] + requests.append((method, path, body, kwargs["headers"]["Authorization"])) + if path.endswith("/complete") and not body["commands"]: + return httpx.Response(422, json={"message": "The commands field is required."}, + request=httpx.Request(method, "http://test" + path)) + assert path.endswith("/fail") + return httpx.Response(200, json={"outcome": "waiting_for_history", "recorded": True}, + request=httpx.Request(method, "http://test" + path)) + + async with Client("http://test", worker_token="worker-token") as client: + worker = Worker(client, task_queue="queue", worker_id="original-owner", workflows=[PendingOperation]) + with patch.object(client._http, "request", new=AsyncMock(side_effect=server)): + commands = await worker._run_workflow_task({ + "task_id": "original-task", "workflow_task_attempt": 4, "run_id": "original-run", + "workflow_type": "tests.replay.pending-operation", "history_events": history, + "payload_codec": "avro", "arguments": serializer.envelope([kind]), + }) + + assert commands == [] + assert requests == [("POST", "/api/worker/workflow-tasks/original-task/fail", { + "lease_owner": "original-owner", "workflow_task_attempt": 4, + "failure": {"message": "Workflow task waiting for scheduled history.", + "type": "WorkflowTaskWaitingForHistory"}, + }, "Bearer worker-token")] + + @pytest.mark.parametrize("kind", OPENINGS) def test_new_operation_is_still_scheduled(kind: str) -> None: outcome = replay(PendingOperation, initial_history(), [kind])