diff --git a/CHANGELOG.md b/CHANGELOG.md index 0cddff1..ad196de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] -## [2.11.0] - Unreleased +## [2.11.1] - 2026-10-24 + +### Fixed + +- Keep the bounded `Worker.run_until` helper polling after a matching typed + backend-lock refusal. Backoff stays inside the original caller deadline, + releases reserved capacity and stops promptly on cancellation. Malformed, + foreign and fatal refusals retain their existing error behavior. + +## [2.11.0] - 2026-10-23 ### Fixed @@ -15,7 +24,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 markers, and respects lower configured or discovered limits. Preserve failure type, retry policy and claim identity; refuse an identity that cannot fit. -## [2.10.1] - Unreleased +## [2.10.1] - 2026-10-23 ### Fixed diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index bef59c4..6e6564c 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -84,6 +84,12 @@ own `/api` routes. For example, use `http://server:8080`, not `http://server:8080/api`. Managed-runtime paths such as `https://cloud.example/api/runtime/v1/namespaces/acme` are valid as written. +`Worker.run_until` is a bounded helper for examples and single-workflow scripts. +Its timeout includes registration and waits through matching typed backend-lock +poll refusals without restarting the budget. Persistent pressure still times +out; malformed, foreign and authorization errors still raise. Worker shutdown +uses its separate drain timeout. Use `Worker.run` for long-running workers. + ### Duplicate workflow IDs A workflow ID is the durable identity of an instance. The PHP Server start diff --git a/pyproject.toml b/pyproject.toml index a00fdb5..0b9b00a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "durable-workflow" -version = "2.11.0" +version = "2.11.1" description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server" readme = "README.md" requires-python = ">=3.10" @@ -76,8 +76,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.11.0" -registry-version = "2.11.0" +product-train = "2.11.1" +registry-version = "2.11.1" supported-server-versions = "2.8.0" worker-protocol-version = "1.20" control-plane-version = "2" diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index 6e5a837..28ee9a4 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -259,8 +259,8 @@ def reset(self) -> None: self._last_summary: str | None = None self._last_warning_at = 0.0 - def retry_delay(self, error: Exception, task_kind: str) -> float: - delay = random.uniform(self._ceiling / 2, self._ceiling) + def retry_delay(self, error: Exception, task_kind: str, *, minimum_delay: float = 0.0) -> float: + delay = max(minimum_delay, random.uniform(self._ceiling / 2, self._ceiling)) self._ceiling = min(30.0, self._ceiling * 2) summary = _poll_error_summary(error) now = time.monotonic() @@ -280,6 +280,28 @@ def _poll_capacity_delay(error: BaseException, task_kind: str, task_queue: str) return error.poll_capacity_backpressure_delay(task_kind, task_queue) +def _backend_lock_poll_delay( + error: Exception, task_kind: str, task_queue: str, namespace: object, +) -> int | None: + if not isinstance(error, ServerError) or error.status != 503: + return None + body = error.body + if ( + not isinstance(body, dict) or not isinstance(namespace, str) or not namespace + or body.get("reason") != "backend_lock_pressure" + or body.get("poll_status") != "backend_lock_pressure" + or "task" not in body or body["task"] is not None + or body.get("task_kind") != task_kind + "_task" + or body.get("task_queue") != task_queue or body.get("namespace") != namespace + or ("retryable" in body and body["retryable"] is not True) + or ("request_admitted" in body and body["request_admitted"] is not False) + or type(body.get("retry_after_seconds")) is not int + or not 0 < body["retry_after_seconds"] <= 60 + ): + return None + return int(body["retry_after_seconds"]) + + def _command_payload_codec(codec: object) -> str: if codec is None: return serializer.AVRO_CODEC @@ -4468,6 +4490,7 @@ async def _run_until_loop( ) -> WorkflowExecution: deadline = time.monotonic() + timeout next_task_kind = "workflow" + failures = {"workflow": _PollFailureBackoff(), "activity": _PollFailureBackoff()} while not self._stop.is_set(): desc = await self.client.describe_workflow(workflow_id) @@ -4494,9 +4517,14 @@ async def _run_until_loop( task_kinds=self._workflow_poll_task_kinds(), history_page_size=WORKFLOW_HISTORY_PAGE_SIZE, ) - except BaseException: + except BaseException as error: self._release_workflow_capacity() + if isinstance(error, Exception) and await self._retry_run_until_poll_refusal( + error, "workflow", poll_start, deadline, failures["workflow"], + ): + continue raise + failures["workflow"].reset() if self._stop.is_set(): self._release_workflow_capacity() raise asyncio.CancelledError @@ -4543,12 +4571,20 @@ async def _run_until_loop( await asyncio.sleep(poll_interval) continue poll_start = time.perf_counter() - task = await self.client.poll_activity_task( - worker_id=self.worker_id, - task_queue=self.task_queue, - timeout=self._poll_timeout, - build_id=self.build_id, - ) + try: + task = await self.client.poll_activity_task( + worker_id=self.worker_id, + task_queue=self.task_queue, + timeout=self._poll_timeout, + build_id=self.build_id, + ) + except Exception as error: + if await self._retry_run_until_poll_refusal( + error, "activity", poll_start, deadline, failures["activity"], + ): + continue + raise + failures["activity"].reset() if self._stop.is_set(): raise asyncio.CancelledError self._record_poll_metrics( @@ -4566,6 +4602,28 @@ async def _run_until_loop( raise asyncio.CancelledError + async def _retry_run_until_poll_refusal( + self, error: Exception, kind: str, poll_start: float, deadline: float, + failures: _PollFailureBackoff, + ) -> bool: + delay = _backend_lock_poll_delay( + error, kind, self.task_queue, getattr(self.client, "namespace", None), + ) + if delay is None: + return False + self._record_poll_metrics(kind, "error", time.perf_counter() - poll_start) + backoff = failures.retry_delay(error, kind, minimum_delay=delay) + remaining = deadline - time.monotonic() + if remaining <= 0: + raise asyncio.TimeoutError + # Both the public caller timer and this local wait retain their original + # deadline. Stop interrupts the backoff without another poll or dispatch. + with contextlib.suppress(asyncio.TimeoutError): + await asyncio.wait_for(self._stop.wait(), timeout=min(backoff, remaining)) + if time.monotonic() >= deadline: + raise asyncio.TimeoutError + return True + async def _run_until_activity_task(self, task: dict[str, Any]) -> None: task_start = time.perf_counter() outcome = "error" diff --git a/tests/fixtures/replay_regressions/run-until-lock-refusal-cold-history.json b/tests/fixtures/replay_regressions/run-until-lock-refusal-cold-history.json new file mode 100644 index 0000000..d866039 --- /dev/null +++ b/tests/fixtures/replay_regressions/run-until-lock-refusal-cold-history.json @@ -0,0 +1,48 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "run-until-lock-refusal-cold-history", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.poll.failure.recovery", + "input": [ + "0" + ], + "run_id": "run-until-cold-original", + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "ActivityScheduled", + "payload": { + "sequence": 1, + "activity_type": "charge" + } + }, + { + "event_type": "ActivityCompleted", + "payload": { + "sequence": 1, + "activity_type": "charge", + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EDGFtb3VudATGAQ5yZWNlaXB0CghwYWlkAA==" + } + } + } + ], + "expected": { + "command_type": "CompleteWorkflow", + "result": { + "order": "0", + "charged": { + "amount": 99, + "receipt": "paid" + }, + "run": "run-until-cold-original" + } + } +} diff --git a/tests/test_worker.py b/tests/test_worker.py index 7320d63..63fadf6 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -2,9 +2,11 @@ import asyncio import contextlib +import json import logging import sys import threading +from pathlib import Path from unittest.mock import AsyncMock, Mock import httpx @@ -5029,6 +5031,232 @@ def test_process_metrics_memory_bytes_is_current_resident_set( class TestRunUntil: + @pytest.mark.asyncio + async def test_lock_refusal_then_cold_history_preserves_original_activity_result( + self, mock_client: AsyncMock, + ) -> None: + from tests.test_poll_failure_backoff import RecoveredWorkflow + + fixture = json.loads((Path(__file__).parent / "fixtures/replay_regressions/" + "run-until-lock-refusal-cold-history.json").read_text()) + mock_client.namespace = "default" + mock_client.get_cluster_info.return_value = compatible_cluster_info(worker_protocol={ + "version": PROTOCOL_VERSION, + "server_capabilities": {"query_tasks": False, "long_poll_timeout": 30}, + }) + run_id = fixture["workflow"]["run_id"] + completed = False + + async def describe(workflow_id: str) -> WorkflowExecution: + return WorkflowExecution( + workflow_id, run_id, fixture["workflow"]["type"], "completed" if completed else "running", + ) + + async def complete(**kwargs: object) -> dict[str, str]: + nonlocal completed + completed = True + return {"outcome": "completed"} + + mock_client.describe_workflow.side_effect = describe + mock_client.complete_workflow_task.side_effect = complete + mock_client.poll_workflow_task.side_effect = [ServerError(503, { + "task": None, "poll_status": "backend_lock_pressure", "reason": "backend_lock_pressure", + "namespace": "default", "task_queue": "orders", "task_kind": "workflow_task", + "retry_after_seconds": 1, + }), { + "task_id": "original-cold-task", "workflow_task_attempt": 3, + "workflow_type": fixture["workflow"]["type"], "workflow_id": "order-0", + "run_id": run_id, "payload_codec": "avro", + "arguments": serializer.encode(fixture["workflow"]["input"]), "history_events": fixture["history"], + }] + worker = Worker( + mock_client, task_queue="orders", workflows=[RecoveredWorkflow], + max_concurrent_workflow_tasks=1, poll_timeout=0.01, + ) + execution = await worker.run_until(workflow_id="order-0", timeout=3.0, poll_interval=0.01) + assert execution.run_id == run_id + assert execution.status == "completed" + mock_client.complete_workflow_task.assert_awaited_once() + publication = mock_client.complete_workflow_task.call_args.kwargs + assert publication["task_id"] == "original-cold-task" + assert publication["workflow_task_attempt"] == 3 + commands = publication["commands"] + assert len(commands) == 1 + assert commands[0]["type"] == "complete_workflow" + assert serializer.decode_envelope(commands[0]["result"], codec="avro") == fixture["expected"]["result"] + mock_client.complete_activity_task.assert_not_awaited() + mock_client.fail_workflow_task.assert_not_awaited() + assert worker._workflow_reserved == 0 + + def refusal_worker( + self, client: AsyncMock, kind: str, + ) -> tuple[Worker, AsyncMock, dict[str, object]]: + client.namespace = "default" + client.get_cluster_info.return_value = compatible_cluster_info(worker_protocol={ + "version": PROTOCOL_VERSION, + "server_capabilities": {"query_tasks": False, "long_poll_timeout": 30}, + }) + client.describe_workflow.return_value = WorkflowExecution( + "original", "original-run", "test-wf", "running", + ) + body: dict[str, object] = { + "task": None, "poll_status": "backend_lock_pressure", "reason": "backend_lock_pressure", + "namespace": "default", "task_queue": "orders", "task_kind": kind + "_task", + "retry_after_seconds": 1, + } + worker = Worker( + client, worker_id="w1", task_queue="orders", + workflows=[TestWorkflow] if kind == "workflow" else [], + activities=[echo_activity] if kind == "activity" else [], poll_timeout=0.01, + max_concurrent_workflow_tasks=1, + ) + return worker, getattr(client, "poll_" + kind + "_task"), body + + @pytest.mark.asyncio + @pytest.mark.parametrize("kind", ["workflow", "activity"]) + @pytest.mark.parametrize("registration_delay", [0.0, 0.04]) + async def test_run_until_lock_pressure_retains_original_deadline( + self, mock_client: AsyncMock, kind: str, registration_delay: float, + ) -> None: + worker, poll, body = self.refusal_worker(mock_client, kind) + poll.side_effect = ServerError(503, body) + + async def register(**kwargs: object) -> dict[str, object]: + await asyncio.sleep(registration_delay) + return {"worker_id": "w1", "registered": True} + + mock_client.register_worker.side_effect = register + started = asyncio.get_running_loop().time() + with pytest.raises(TimeoutError, match="original"): + await worker.run_until(workflow_id="original", timeout=0.08, poll_interval=0.01) + elapsed = asyncio.get_running_loop().time() - started + + assert 0.06 <= elapsed < 1.0 + poll.assert_awaited_once() + assert worker._stop.is_set() + mock_client.deregister_worker_registration.assert_awaited_once_with("w1", expected_namespace="default") + + @pytest.mark.asyncio + @pytest.mark.parametrize("kind", ["workflow", "activity"]) + async def test_run_until_stop_interrupts_lock_backoff( + self, mock_client: AsyncMock, kind: str, + ) -> None: + worker, poll, body = self.refusal_worker(mock_client, kind) + entered = asyncio.Event() + + async def refuse(**kwargs: object) -> None: + entered.set() + raise ServerError(503, body) + + poll.side_effect = refuse + running = asyncio.create_task(worker.run_until(workflow_id="original", timeout=5.0)) + await asyncio.wait_for(entered.wait(), timeout=1.0) + await asyncio.sleep(0) + await asyncio.wait_for(worker.stop(), timeout=0.5) + with pytest.raises(asyncio.CancelledError): + await running + + poll.assert_awaited_once() + mock_client.complete_workflow_task.assert_not_awaited() + mock_client.complete_activity_task.assert_not_awaited() + mock_client.deregister_worker_registration.assert_awaited_once_with("w1", expected_namespace="default") + + @pytest.mark.asyncio + @pytest.mark.parametrize("kind", ["workflow", "activity"]) + @pytest.mark.parametrize("change", [ + {"task_queue": "foreign"}, {"task_kind": "query_task"}, {"task": {"task_id": "claimed"}}, + {"poll_status": "empty"}, {"retry_after_seconds": True}, {"retry_after_seconds": 61}, + {"retryable": False}, {"request_admitted": True}, + ]) + async def test_run_until_refuses_malformed_or_admitted_poll_response( + self, mock_client: AsyncMock, kind: str, change: dict[str, object], + ) -> None: + worker, poll, body = self.refusal_worker(mock_client, kind) + refusal = ServerError(503, {**body, **change}) + poll.side_effect = refusal + with pytest.raises(ServerError) as observed: + await worker.run_until(workflow_id="original", timeout=2.0) + assert observed.value is refusal + poll.assert_awaited_once() + mock_client.complete_workflow_task.assert_not_awaited() + mock_client.complete_activity_task.assert_not_awaited() + + @pytest.mark.asyncio + @pytest.mark.parametrize("kind", ["workflow", "activity"]) + async def test_run_until_recovers_typed_lock_refusal( + self, mock_client: AsyncMock, kind: str, + ) -> None: + mock_client.namespace = "default" + mock_client.get_cluster_info.return_value = compatible_cluster_info(worker_protocol={ + "version": PROTOCOL_VERSION, + "server_capabilities": {"query_tasks": False, "long_poll_timeout": 30}, + }) + mock_client.describe_workflow.side_effect = [ + WorkflowExecution("original", "original-run", "test-wf", "running"), + WorkflowExecution("original", "original-run", "test-wf", "running"), + WorkflowExecution("original", "original-run", "test-wf", "completed"), + ] + refusal = ServerError(503, { + "task": None, "poll_status": "backend_lock_pressure", "reason": "backend_lock_pressure", + "namespace": "default", "task_queue": "orders", "task_kind": kind + "_task", + "retry_after_seconds": 1, "backend": {"driver": "sqlite", "lock_pressure": True}, + }) + poll = getattr(mock_client, "poll_" + kind + "_task") + poll.side_effect = [refusal, None] + worker = Worker( + mock_client, worker_id="w1", task_queue="orders", + workflows=[TestWorkflow] if kind == "workflow" else [], + activities=[echo_activity] if kind == "activity" else [], + poll_timeout=0.01, + max_concurrent_workflow_tasks=1, + ) + + execution = await worker.run_until(workflow_id="original", timeout=2.0, poll_interval=0.01) + + assert execution.status == "completed" + assert execution.run_id == "original-run" + assert poll.await_count == 2 + assert all(call.kwargs["worker_id"] == "w1" for call in poll.await_args_list) + assert all(call.kwargs["task_queue"] == "orders" for call in poll.await_args_list) + mock_client.register_worker.assert_awaited_once() + mock_client.deregister_worker_registration.assert_awaited_once_with("w1", expected_namespace="default") + mock_client.complete_workflow_task.assert_not_awaited() + mock_client.complete_activity_task.assert_not_awaited() + + @pytest.mark.asyncio + @pytest.mark.parametrize("kind", ["workflow", "activity"]) + @pytest.mark.parametrize("refusal_kind", ["foreign_namespace", "unauthorized"]) + async def test_run_until_preserves_fatal_or_foreign_poll_refusal( + self, mock_client: AsyncMock, kind: str, refusal_kind: str, + ) -> None: + mock_client.namespace = "default" + mock_client.get_cluster_info.return_value = compatible_cluster_info(worker_protocol={ + "version": PROTOCOL_VERSION, + "server_capabilities": {"query_tasks": False, "long_poll_timeout": 30}, + }) + mock_client.describe_workflow.return_value = WorkflowExecution( + "original", "original-run", "test-wf", "running", + ) + refusal = Unauthorized() if refusal_kind == "unauthorized" else ServerError(503, { + "task": None, "poll_status": "backend_lock_pressure", "reason": "backend_lock_pressure", + "namespace": "foreign", "task_queue": "orders", "task_kind": kind + "_task", + "retry_after_seconds": 1, + }) + poll = getattr(mock_client, "poll_" + kind + "_task") + poll.side_effect = refusal + worker = Worker( + mock_client, worker_id="w1", task_queue="orders", + workflows=[TestWorkflow] if kind == "workflow" else [], + activities=[echo_activity] if kind == "activity" else [], poll_timeout=0.01, + ) + + with pytest.raises(type(refusal)) as observed: + await worker.run_until(workflow_id="original", timeout=2.0, poll_interval=0.01) + + assert observed.value is refusal + poll.assert_awaited_once() + mock_client.deregister_worker_registration.assert_awaited_once_with("w1", expected_namespace="default") + @pytest.mark.asyncio @pytest.mark.parametrize("has_workflows,has_activities", [(True, False), (False, True), (False, False)]) async def test_run_until_polls_only_registered_task_families(