diff --git a/CHANGELOG.md b/CHANGELOG.md index 44d48ae..b948795 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- Back off workflow, activity and query polling independently after repeated + HTTP or transport failures, with jitter and a 30-second cap. Successful polls + reset the delay, and repeated authentication warnings are rate-limited. + ## [2.9.1] - 2026-10-22 ### Security diff --git a/docs/reference/worker.md b/docs/reference/worker.md index 029bfc2..11dca87 100644 --- a/docs/reference/worker.md +++ b/docs/reference/worker.md @@ -29,6 +29,29 @@ isolated worker processes and their resource limits. The separately negotiated cooperative-cancellation capability retains its supervised process execution; selecting a default thread does not enable that protocol. +## Poll failures and recovery + +This backoff is available in unreleased SDK source. Until the next package +release, published workers retain their fixed one-second poll-error wait. + +Workflow, activity and query pollers back off independently after failed polls, +including HTTP errors, rejected credentials and unreachable runtimes. The +retry ceiling starts at 1 second, doubles on each consecutive failure and stops +at 30 seconds. Each delay is chosen randomly between half and all of that +ceiling, so workers do not retry in lockstep. Any successful poll, including an +empty response, resets the ceiling to 1 second. Shutdown interrupts the wait. + +The first failure is logged immediately. An unchanged failure is logged at most +once per minute per poller until polling succeeds. HTTP 401 identifies rejected +authentication; HTTP 403 identifies denied authorization. Check the worker +credentials and their permissions. Once requests succeed again, polling +recovers without a worker restart. Poll-error metrics +still count every failed poll, including failures whose warning is suppressed. + +Typed capacity backpressure follows the Server's advertised retry delay. +Storage-admission failures retain their separate identity-preserving retry and +worker-stop behavior. Neither uses this generic poll-failure backoff. + ## Lifecycle and shutdown `Worker.run()` and `Worker.run_until()` register through the worker protocol diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index bbb4ddd..938d461 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -966,6 +966,20 @@ compatibility manifest. A client configured with only the opposite role's token still fails before transport for actual worker or control operations; use `token` when one shared credential intentionally authorizes both planes. +## Poll failures + +The backoff described here is available in unreleased SDK source. Published +workers retain their fixed one-second poll-error wait until the next release. + +Workflow, activity and query pollers use independent exponential backoff after +HTTP, authentication or transport failures. The retry ceiling doubles from 1 +second to a maximum of 30 seconds; each wait uses random jitter between half +and all of that ceiling. Any successful poll, including an empty result, resets +it. Shutdown interrupts the wait. Identical warnings repeat at most once per +minute per poller, while poll-error metrics count each failure. HTTP 401 and 403 +diagnostics identify credential and permission failures. Typed capacity and +storage-admission handling remain separate. See [Worker recovery](reference/worker.md#poll-failures-and-recovery). + ## Activity heartbeat refusals Use `await activity.context().heartbeat(details)` in an async activity or diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index 7bbc75e..fb911a0 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -23,6 +23,7 @@ import json import logging import pickle +import random import sys import threading import time @@ -79,6 +80,7 @@ QueryFailed, QueryTaskAuthorityLost, ServerError, + Unauthorized, UpdateRejected, UpdateValidationFailed, WorkflowEntryArgumentsMismatchError, @@ -228,6 +230,10 @@ def _is_storage_admission_error(error: BaseException) -> bool: def _poll_error_summary(error: BaseException) -> str: + if isinstance(error, Unauthorized) or isinstance(error, ServerError) and error.status == 401: + return "authentication failed (HTTP 401); check the worker credentials" + if isinstance(error, ServerError) and error.status == 403: + return "authorization failed (HTTP 403); check the worker credentials and permissions" if not isinstance(error, ServerError): return str(error) reason = error.reason() @@ -236,6 +242,29 @@ def _poll_error_summary(error: BaseException) -> str: return f"server returned {error.status} reason={reason[:64]!r}" +class _PollFailureBackoff: + """One failure streak per poller, independent of durable retry policies.""" + + def __init__(self) -> None: + self.reset() + + def reset(self) -> None: + self._ceiling = 1.0 + 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) + self._ceiling = min(30.0, self._ceiling * 2) + summary = _poll_error_summary(error) + now = time.monotonic() + if summary != self._last_summary or now - self._last_warning_at >= 60.0: + log.warning("%s poll error: %s; retrying in %.2f s", task_kind, summary, delay) + self._last_summary = summary + self._last_warning_at = now + return delay + + def _poll_capacity_delay(error: BaseException, task_kind: str, task_queue: str) -> int | None: if not isinstance(error, ServerError): return None @@ -3680,6 +3709,7 @@ def _admit_workflow_work( return dispatched async def _poll_workflow_tasks(self) -> None: + failures = _PollFailureBackoff() while not self._stop.is_set(): await self._reserve_workflow_capacity() if self._stop.is_set(): @@ -3711,9 +3741,9 @@ async def _poll_workflow_tasks(self) -> None: await asyncio.sleep(delay) continue self._record_poll_metrics("workflow", "error", time.perf_counter() - poll_start) - log.warning("workflow poll error: %s", _poll_error_summary(e)) - await asyncio.sleep(1.0) + await asyncio.sleep(failures.retry_delay(e, "workflow")) continue + failures.reset() if task is None: self._release_workflow_capacity() self._record_poll_metrics("workflow", "empty", time.perf_counter() - poll_start) @@ -3786,6 +3816,7 @@ async def _report_unhandled_workflow_task_error( ) async def _poll_activity_tasks(self) -> None: + failures = _PollFailureBackoff() while not self._stop.is_set(): self._validate_cooperative_activity_handlers() if (len(self._remote_activity_threads) + len(self._remote_activity_processes) @@ -3820,9 +3851,9 @@ async def _poll_activity_tasks(self) -> None: await asyncio.sleep(delay) continue self._record_poll_metrics("activity", "error", time.perf_counter() - poll_start) - log.warning("activity poll error: %s", _poll_error_summary(e)) - await asyncio.sleep(1.0) + await asyncio.sleep(failures.retry_delay(e, "activity")) continue + failures.reset() if task is None: self._act_semaphore.release() self._record_poll_metrics("activity", "empty", time.perf_counter() - poll_start) @@ -3850,6 +3881,7 @@ async def _dispatch_activity_task(self, task: dict[str, Any]) -> None: self._act_semaphore.release() async def _poll_query_tasks(self, *, client: Client | None = None, track_tasks: bool = True) -> None: + failures = _PollFailureBackoff() query_thread_stop = self._query_thread_stop if client is not None else None client = client or self.client while not self._stop.is_set() and ( @@ -3877,9 +3909,9 @@ async def _poll_query_tasks(self, *, client: Client | None = None, track_tasks: await asyncio.sleep(delay) continue self._record_poll_metrics("query", "error", time.perf_counter() - poll_start) - log.warning("query poll error: %s", _poll_error_summary(e)) - await asyncio.sleep(1.0) + await asyncio.sleep(failures.retry_delay(e, "query")) continue + failures.reset() if task is None: self._record_poll_metrics("query", "empty", time.perf_counter() - poll_start) if self._stop.is_set() or ( diff --git a/tests/fixtures/replay_regressions/poll-error-recovery-cold-history.json b/tests/fixtures/replay_regressions/poll-error-recovery-cold-history.json new file mode 100644 index 0000000..0c465db --- /dev/null +++ b/tests/fixtures/replay_regressions/poll-error-recovery-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": "poll-error-recovery-cold-history", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.poll.failure.recovery", + "input": [ + "017" + ], + "run_id": "original-run", + "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": "017", + "charged": { + "amount": 99, + "receipt": "paid" + }, + "run": "original-run" + } + } +} diff --git a/tests/test_poll_failure_backoff.py b/tests/test_poll_failure_backoff.py new file mode 100644 index 0000000..f6058ca --- /dev/null +++ b/tests/test_poll_failure_backoff.py @@ -0,0 +1,315 @@ +from __future__ import annotations + +import asyncio +import json +import logging +import random +import time +from collections.abc import Generator +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import httpx +import pytest + +import durable_workflow.worker as worker_module +from durable_workflow import serializer, workflow +from durable_workflow.client import Client +from durable_workflow.errors import ServerError, Unauthorized +from durable_workflow.retry_policy import TransportRetryPolicy +from durable_workflow.worker import Worker +from durable_workflow.workflow import WorkflowContext + +POLL_KINDS = ["workflow", "activity", "query"] + + +@workflow.defn(name="tests.poll.failure.recovery") +class RecoveredWorkflow: + def run(self, ctx: WorkflowContext, order: str) -> Generator[object, object, object]: + charged = yield ctx.schedule_activity("charge", [order]) + return {"order": order, "charged": charged, "run": ctx.run_id} + + +async def test_recovered_poll_replays_recorded_activity_without_rescheduling( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fixture = json.loads((Path(__file__).parent / "fixtures/replay_regressions/poll-error-recovery-cold-history.json") + .read_text()) + client = AsyncMock(spec=Client) + worker = Worker(client, task_queue="orders", workflows=[RecoveredWorkflow]) + real_sleep = asyncio.sleep + completed = asyncio.Event() + calls = 0 + delays: list[float] = [] + task = { + "task_id": "recovered-task", "workflow_task_attempt": 3, + "workflow_type": fixture["workflow"]["type"], "workflow_id": "order-017", + "run_id": fixture["workflow"]["run_id"], "payload_codec": "avro", + "arguments": serializer.encode(fixture["workflow"]["input"]), "history_events": fixture["history"], + } + + async def poll(**_: object) -> dict[str, object] | None: + nonlocal calls + calls += 1 + if calls == 1: + raise Unauthorized() + if calls == 2: + raise ServerError(503, {}) + if calls == 3: + raise httpx.ConnectError("runtime unavailable") + if calls == 4: + return task + await completed.wait() + worker._stop.set() + return None + + async def complete(**_: object) -> dict[str, str]: + completed.set() + return {"outcome": "completed"} + + async def sleep(delay: float) -> None: + delays.append(delay) + + client.poll_workflow_task.side_effect = poll + client.complete_workflow_task.side_effect = complete + monkeypatch.setattr(worker_module.asyncio, "sleep", sleep) + monkeypatch.setattr(random, "uniform", lambda low, high: high) + await asyncio.wait_for(worker._poll_workflow_tasks(), timeout=2) + await asyncio.gather(*worker._in_flight) + await real_sleep(0) # Let task done callbacks return their reservations. + assert delays == [1, 2, 4] + client.complete_workflow_task.assert_awaited_once() + publication = client.complete_workflow_task.call_args.kwargs + assert publication["task_id"] == "recovered-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"] + client.fail_workflow_task.assert_not_awaited() + assert worker._workflow_reserved == 0 + + +@pytest.mark.parametrize("kind", POLL_KINDS) +@pytest.mark.parametrize("successful_task", [False, True], ids=["empty", "task"]) +@pytest.mark.parametrize("error", [ + Unauthorized("Invalid or missing authentication token"), + ServerError(401, {"message": "private response"}), + ServerError(403, {"message": "private response"}), + ServerError(503, {}), + httpx.ConnectError("connection refused"), +]) +async def test_poll_failure_growth_cap_and_empty_success_reset( + kind: str, successful_task: bool, error: Exception, monkeypatch: pytest.MonkeyPatch, +) -> None: + client = AsyncMock(spec=Client) + worker = Worker(client, task_queue="orders") + poll = getattr(client, f"poll_{kind}_task") + calls = 0 + delays: list[float] = [] + + async def respond(**_: object) -> dict[str, object] | None: + nonlocal calls + calls += 1 + if calls == 10: + return {"task_id": "recovered"} if successful_task else None + if calls == 12: + worker._stop.set() + return None + raise error + + async def sleep(delay: float) -> None: + if delay: + delays.append(delay) + + dispatched: list[object] = [] + + async def dispatch(task: object, **_: object) -> None: + dispatched.append(task) + if kind == "activity": + worker._act_semaphore.release() + + poll.side_effect = respond + monkeypatch.setattr(worker, f"_dispatch_{kind}_task", dispatch) + monkeypatch.setattr(worker_module.asyncio, "sleep", sleep) + monkeypatch.setattr(random, "uniform", lambda low, high: high) + await getattr(worker, f"_poll_{kind}_tasks")() + await asyncio.gather(*worker._in_flight) + + assert delays == [1, 2, 4, 8, 16, 30, 30, 30, 30, 1] + assert calls == 12 + assert len(dispatched) == int(successful_task) + # Failed/empty polls must release their reservations before retrying. + assert worker._act_semaphore._value == worker.max_concurrent_activity_tasks + assert worker._workflow_reserved == 0 + + +def test_jitter_uses_a_fresh_sample_with_bounded_delays(monkeypatch: pytest.MonkeyPatch) -> None: + bounds: list[tuple[float, float]] = [] + + def uniform(low: float, high: float) -> float: + bounds.append((low, high)) + return low if len(bounds) % 2 else high + + monkeypatch.setattr(random, "uniform", uniform) + backoff = worker_module._PollFailureBackoff() + delays = [backoff.retry_delay(Unauthorized(), "workflow") for _ in range(8)] + assert delays == [0.5, 2, 2, 8, 8, 30, 15, 30] + assert bounds == [(0.5, 1), (1, 2), (2, 4), (4, 8), (8, 16), (15, 30), (15, 30), (15, 30)] + + +@pytest.mark.parametrize("kind", POLL_KINDS) +async def test_changed_auth_failure_is_reported_immediately( + kind: str, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, +) -> None: + client = AsyncMock(spec=Client) + worker = Worker(client, task_queue="orders") + getattr(client, f"poll_{kind}_task").side_effect = [ + Unauthorized(), Unauthorized(), ServerError(403, {}), ServerError(403, {}), + ] + delays: list[float] = [] + + async def sleep(delay: float) -> None: + delays.append(delay) + if len(delays) == 4: + worker._stop.set() + + monkeypatch.setattr(worker_module.asyncio, "sleep", sleep) + with caplog.at_level(logging.WARNING, logger="durable_workflow.worker"): + await getattr(worker, f"_poll_{kind}_tasks")() + warnings = [r.getMessage() for r in caplog.records if r.name == "durable_workflow.worker"] + assert len(warnings) == 2 + assert "HTTP 401" in warnings[0] + assert "HTTP 403" in warnings[1] + assert delays[2] >= 2 # A changed failure does not reset the request backoff. + + +@pytest.mark.parametrize("kind", POLL_KINDS) +@pytest.mark.parametrize("error,diagnostic", [ + (Unauthorized("private response"), "authentication failed (HTTP 401)"), + (ServerError(403, {"message": "private response"}), "authorization failed (HTTP 403)"), +]) +async def test_identical_auth_warnings_are_rate_limited_and_reset_after_success( + kind: str, error: Exception, diagnostic: str, + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, +) -> None: + client = AsyncMock(spec=Client) + worker = Worker(client, task_queue="orders") + now = 0.0 + calls = 0 + outcomes: list[str] = [] + + async def respond(**_: object) -> None: + nonlocal calls + calls += 1 + if calls in (12, 14): + if calls == 14: + worker._stop.set() + return None + raise error + + async def sleep(delay: float) -> None: + nonlocal now + now += delay + + getattr(client, f"poll_{kind}_task").side_effect = respond + monkeypatch.setattr(worker_module.asyncio, "sleep", sleep) + monkeypatch.setattr(random, "uniform", lambda low, high: high) + monkeypatch.setattr(worker_module, "time", SimpleNamespace( + monotonic=lambda: now, perf_counter=time.perf_counter, + )) + monkeypatch.setattr(worker, "_record_poll_metrics", lambda _kind, outcome, _duration: outcomes.append(outcome)) + with caplog.at_level(logging.WARNING, logger="durable_workflow.worker"): + await getattr(worker, f"_poll_{kind}_tasks")() + + warnings = [r.getMessage() for r in caplog.records if r.name == "durable_workflow.worker"] + # First refusal at 0, reminders at 61, 121 and 181, then the post-success streak. + assert len(warnings) == 5 + assert all(diagnostic in message and "credentials" in message for message in warnings) + assert "private response" not in caplog.text + assert outcomes.count("error") == 12 + + +@pytest.mark.parametrize("kind", POLL_KINDS) +async def test_backoff_does_not_absorb_storage_admission( + kind: str, monkeypatch: pytest.MonkeyPatch, +) -> None: + client = AsyncMock(spec=Client) + worker = Worker(client, task_queue="orders") + refusal = ServerError(503, {"reason": "storage_admission_unavailable"}) + getattr(client, f"poll_{kind}_task").side_effect = refusal + sleep = AsyncMock() + monkeypatch.setattr(worker_module.asyncio, "sleep", sleep) + polling = getattr(worker, f"_poll_{kind}_tasks")() + with pytest.raises(ServerError) as caught: + await polling + assert caught.value is refusal + sleep.assert_not_awaited() + + +@pytest.mark.parametrize("kind", POLL_KINDS) +async def test_shutdown_interrupts_a_capped_backoff(kind: str, monkeypatch: pytest.MonkeyPatch) -> None: + client = AsyncMock(spec=Client) + worker = Worker(client, task_queue="orders", shutdown_timeout=0.5) + getattr(client, f"poll_{kind}_task").side_effect = Unauthorized() + waiting = asyncio.Event() + real_sleep = asyncio.sleep + + async def sleep(delay: float) -> None: + if delay == 30: + waiting.set() + await real_sleep(delay) + + monkeypatch.setattr(worker_module.asyncio, "sleep", sleep) + monkeypatch.setattr(random, "uniform", lambda low, high: high) + task = asyncio.create_task(getattr(worker, f"_poll_{kind}_tasks")()) + worker._poller_tasks.add(task) + try: + await asyncio.wait_for(waiting.wait(), timeout=1) + await asyncio.wait_for(worker.stop(), timeout=1) + assert task.cancelled() + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +@pytest.mark.parametrize("kind", POLL_KINDS) +@pytest.mark.parametrize("status", [401, 403, 503, None], ids=["401", "403", "503", "transport"]) +async def test_actual_client_poll_errors_back_off_then_recover( + kind: str, status: int | None, monkeypatch: pytest.MonkeyPatch, +) -> None: + client = Client( + "https://runtime.example", worker_token="test-worker-token", namespace="orders", + retry_policy=TransportRetryPolicy(max_attempts=1), + ) + worker = Worker(client, task_queue="orders") + calls = 0 + delays: list[float] = [] + + def respond(request: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + assert request.url.path == f"/api/worker/{kind}-tasks/poll" + assert request.headers["Authorization"] == "Bearer test-worker-token" + assert request.headers["X-Namespace"] == "orders" + if calls in (4, 6): + if calls == 6: + worker._stop.set() + return httpx.Response(200, json={"task": None}) + if status is None: + raise httpx.ConnectError("runtime unavailable", request=request) + return httpx.Response(status, json={"message": "poll refused"}) + + async def sleep(delay: float) -> None: + if delay: + delays.append(delay) + + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(respond)) + client._runtime_external_payload_transport_resolved = True + monkeypatch.setattr(worker_module.asyncio, "sleep", sleep) + monkeypatch.setattr(random, "uniform", lambda low, high: high) + async with client: + await getattr(worker, f"_poll_{kind}_tasks")() + assert calls == 6 + assert delays == [1, 2, 4, 1] diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index e2be47c..2619639 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -40,6 +40,7 @@ RepeatedPatchWorkflow, ) from tests.test_pending_operation_replay import PendingOperation +from tests.test_poll_failure_backoff import RecoveredWorkflow from tests.test_replay_clock import PersistedClockWorkflow from tests.test_update_signal_condition_replay import ( PostConditionReceiversWorkflow, @@ -369,6 +370,7 @@ def run(self, ctx: WorkflowContext, cleanup_seconds: int = 0): # type: ignore[n MixedPatchHelpersWorkflow, RepeatedPatchWorkflow, PendingOperation, + RecoveredWorkflow, PagedRecordedSideEffectsWorkflow, PostConditionReceiversWorkflow, RecordedSideEffectWorkflow, diff --git a/tests/test_worker.py b/tests/test_worker.py index c4f6789..0673342 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -957,7 +957,7 @@ async def test_malformed_poll_warning_omits_large_server_response( ) async def stop_after_backoff(seconds: float) -> None: - assert seconds == 1.0 + assert 0.5 <= seconds <= 1.0 worker._stop.set() monkeypatch.setattr(worker_module.asyncio, "sleep", stop_after_backoff) @@ -969,10 +969,11 @@ async def stop_after_backoff(seconds: float) -> None: for record in caplog.records if record.name == "durable_workflow.worker" ] - assert warnings == [ + assert len(warnings) == 1 + assert warnings[0].startswith( f"{task_kind} poll error: server returned 429 " - "reason='long_poll_capacity_exhausted'" - ] + "reason='long_poll_capacity_exhausted'; retrying in " + ) assert "server_capabilities" not in caplog.text def test_poll_error_summary_bounds_reason(self) -> None: