Back off persistent worker poll failures

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

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

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

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 (

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"
+    }
+  }
+}