Keep run_until polling through matching backend-lock refusals

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
 

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

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"

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"

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