Bound worker shutdown and fence deregistration recovery

CHANGELOG.md

@@ -6,6 +6,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [2.10.1] - Unreleased
+
+### Fixed
+
+- Keep worker deregistration inside the remaining shutdown deadline. When the
+  runtime advertises generation-fenced receipts, retry matching transient or
+  lost responses with the original registration token. Without advertised
+  fencing, use one legacy attempt. Exhausted or uncertain shutdown keeps
+  registration removal unconfirmed instead of reporting success.
+
 ## [2.10.0] - 2026-10-23
 
 ### Added

docs/reference/worker.md

@@ -65,8 +65,27 @@ worker-stop behavior. Neither uses this generic poll-failure backoff.
 before polling. After registration succeeds, shutdown stops every poller and
 waits for accepted work within the shared `shutdown_timeout` deadline before
 removing that worker-plane registration. Work still running at the deadline is
-cancelled. Calling `stop()` more than once is safe and sends at most one
-deregistration request. No deregistration is sent if registration failed.
+cancelled. Calling `stop()` more than once shares the same shutdown operation
+and result. No deregistration is sent if registration failed.
+
+**Unreleased source behavior, prepared for SDK 2.10.1:** the final HTTP call and any recovery use the
+original remaining shutdown budget. If Server advertises
+`worker_deregistration_fencing`, Worker retains its registration token and uses
+the fenced receipt endpoint. Matching transient responses and transport losses
+can retry that same generation. A completed receipt can replay safely even if
+a replacement now owns the worker ID. Authentication, malformed or mismatched
+receipts, lost authority and expired receipts fail without retrying. Without
+advertised fencing, Worker keeps one attempt through the legacy worker-plane
+DELETE and does not replay an uncertain result. That legacy route has no
+generation fence. Keep worker IDs unique to a process, and do not reuse an ID
+until the prior worker has fully stopped.
+
+If drain consumes the deadline, no late deregistration request is sent. A final
+HTTP timeout or exhausted recovery raises `RuntimeError` and leaves removal
+unconfirmed. The Server might have committed a request whose response was lost;
+an exception is not proof that the registration still exists. Inspect the
+runtime before taking an operator recovery action. These changes are not in a
+published package yet.
 
 Cancelling an await cannot stop a running synchronous activity thread. Its
 physical slot remains occupied until the callback returns, and a late return

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "2.10.0"
+version = "2.10.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.10.0"
-registry-version = "2.10.0"
+product-train = "2.10.1"
+registry-version = "2.10.1"
 supported-server-versions = "2.8.0"
 worker-protocol-version = "1.20"
 control-plane-version = "2"

src/durable_workflow/_metric_routes.py

@@ -12,6 +12,7 @@
       for suffix in ("", "/pass")),
     "/workers", "/workers/{worker_id}",
     "/worker/register", "/worker/heartbeat", "/worker/registrations/{worker_id}",
+    "/worker/registrations/{worker_id}/deregister",
     "/worker/sessions", "/worker/sessions/{session_id}", "/worker/sessions/{session_id}/heartbeat",
     *(f"/worker/{kind}-tasks/poll" for kind in ("workflow", "activity", "query", "update-validation")),
     *(f"/worker/{kind}-tasks/{{task_id}}/{action}" for kind, actions in (

src/durable_workflow/client.py

@@ -1853,6 +1853,7 @@ async def _request(
         json: Any = None,
         timeout: float | None = None,
         context: str = "",
+        retry: bool = True,
     ) -> Any:
         if worker and discovery:
             raise ValueError("a request cannot be both worker-plane and cluster discovery")
@@ -1920,7 +1921,7 @@ async def _do_request() -> httpx.Response:
 
         try:
             try:
-                resp = await self.retry_policy.execute(_do_request)
+                resp = await self.retry_policy.execute(_do_request) if retry else await _do_request()
             except httpx.HTTPStatusError as exc:
                 status_code = str(exc.response.status_code)
                 outcome = "http_error"
@@ -5245,39 +5246,69 @@ async def close_worker_session(
             json={"worker_id": worker_id, "reason": reason},
         ))
 
-    async def deregister_worker_registration(self, worker_id: str) -> dict[str, Any]:
+    async def deregister_worker_registration(
+        self, worker_id: str, *, registration_token: str | None = None, timeout: float | None = None,
+        expected_namespace: str | None = None,
+    ) -> dict[str, Any]:
         """Remove this runtime's successful worker-plane registration.
 
         Called by :class:`~durable_workflow.Worker` after its pollers and
         in-flight tasks have drained. This is distinct from the operator-only
         :meth:`deregister_worker` control-plane management operation. Returns
         the worker deregistration envelope, including recovered workflow-task
-        count.
+        count. With a registration token, use the runtime's generation-fenced
+        receipt endpoint. Each call makes one attempt: Worker owns safe retries
+        and the remaining shutdown budget. An unfenced DELETE cannot safely be
+        repeated after an uncertain response because the worker ID may be reused.
+        ``expected_namespace`` is a local registration-scope precondition, not
+        a namespace override.
         """
+        if expected_namespace is not None and self.namespace != expected_namespace:
+            raise RuntimeError("worker registration namespace changed; registration removal is unconfirmed")
+        if registration_token is not None and (
+            not isinstance(registration_token, str) or re.fullmatch(r"[a-f0-9]{32}", registration_token) is None
+        ):
+            raise ValueError("registration_token must contain exactly 32 lowercase hexadecimal characters")
+        fenced = registration_token is not None
         try:
             data = await self._request(
-                "DELETE",
-                f"/worker/registrations/{quote(worker_id, safe='')}",
+                "POST" if fenced else "DELETE",
+                f"/worker/registrations/{quote(worker_id, safe='')}" + ("/deregister" if fenced else ""),
                 worker=True,
                 context=worker_id,
+                json={"registration_token": registration_token} if fenced else None,
+                timeout=timeout,
+                retry=False,
             )
         except ServerError as error:
-            if error.status == 404 and error.reason() == "worker_not_found":
+            if not fenced and error.status == 404 and error.reason() == "worker_not_found":
                 return {
                     "worker_id": worker_id,
                     "outcome": "already_deregistered",
                     "recovered_workflow_task_count": 0,
                 }
             raise
-        if not isinstance(data, dict):
+        if not self._valid_worker_deregistration_response(data, worker_id, registration_token):
             raise ServerError(
                 200,
                 {
                     "reason": "invalid_worker_deregistration_response",
-                    "message": f"expected JSON object, got {type(data).__name__}",
+                    "message": "expected a deregistration receipt for this worker registration",
                 },
             )
-        return data
+        return cast(dict[str, Any], data)
+
+    @staticmethod
+    def _valid_worker_deregistration_response(data: Any, worker_id: str, token: str | None) -> bool:
+        if not isinstance(data, dict):
+            return False
+        count = data.get("recovered_workflow_task_count")
+        return (
+            data.get("worker_id") == worker_id
+            and data.get("outcome") == "deregistered"
+            and type(count) is int and count >= 0
+            and (token is None or data.get("registration_token") == token)
+        )
 
     async def heartbeat_worker(
         self,

src/durable_workflow/errors.py

@@ -219,6 +219,21 @@ def reason(self) -> str | None:
             return self.body.get("reason")
         return None
 
+    def worker_deregistration_retry_delay(self, worker_id: str, registration_token: str) -> int | None:
+        """Return a delay only for an identity-preserving fenced transient."""
+        body = self.body
+        if (
+            self.status != 503 or not isinstance(body, dict)
+            or body.get("reason") not in ("backend_lock_pressure", "backend_unavailable")
+            or body.get("operation") != "deregister_worker" or body.get("outcome") != "unknown"
+            or body.get("worker_id") != worker_id or body.get("registration_token") != registration_token
+            or body.get("retryable") is not True
+            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 poll_capacity_backpressure_delay(self, task_kind: str, task_queue: str) -> int | None:
         """Return the advertised wait only for a complete poll-capacity refusal."""
         body = self.body

src/durable_workflow/worker.py

@@ -37,6 +37,8 @@
 from types import FunctionType
 from typing import Annotated, Any, Concatenate, Literal, ParamSpec, TypeVar, Union, get_args, get_origin, get_type_hints
 
+import httpx
+
 from . import serializer
 from ._activity_lease import activity_lease_renewal_supported, current_activity_authority
 from ._activity_process import CallbackFailure, CallbackInvocation, CallbackProcessLost, SupervisedCallback
@@ -1169,6 +1171,10 @@ def __init__(
         self._in_flight: set[asyncio.Task[Any]] = set()
         self._poller_tasks: set[asyncio.Task[Any]] = set()
         self._registered = False
+        self._deregistration_fenced = False
+        self._registration_token: str | None = None
+        self._registration_worker_id: str | None = None
+        self._registration_namespace: object = None
         self._run_started = False
         self._registration_done = asyncio.Event()
         self._shutdown_task: asyncio.Task[None] | None = None
@@ -1295,6 +1301,18 @@ async def _register(self) -> None:
         _validate_server_compatibility(info)
         protocol = info.get("worker_protocol")
         server_capabilities = protocol.get("server_capabilities") if isinstance(protocol, Mapping) else None
+        deregistration = (
+            server_capabilities.get("worker_deregistration_fencing")
+            if isinstance(server_capabilities, Mapping) else None
+        )
+        self._deregistration_fenced = isinstance(deregistration, Mapping) and deregistration.get("supported") is True
+        if isinstance(deregistration, Mapping) and self._deregistration_fenced and (
+            deregistration.get("schema") != "durable-workflow.v2.worker-deregistration.v1"
+            or deregistration.get("endpoint") != "/worker/registrations/{workerId}/deregister"
+            or type(deregistration.get("receipt_retention_seconds")) is not int
+            or deregistration["receipt_retention_seconds"] <= 0
+        ):
+            raise RuntimeError("Server compatibility error: invalid worker_deregistration_fencing contract")
         sticky_support = (
             server_capabilities.get("sticky_execution") if isinstance(server_capabilities, Mapping) else None
         )
@@ -1449,6 +1467,20 @@ async def _register(self) -> None:
             task_slots=self._current_task_slots(),
             process_metrics=self._current_process_metrics(),
         )
+        if self._deregistration_fenced:
+            token = ack.get("registration_token") if isinstance(ack, dict) else None
+            if (
+                not isinstance(token, str) or len(token) != 32
+                or any(character not in "0123456789abcdef" for character in token)
+                or ack.get("worker_id") != self.worker_id or ack.get("registered") is not True
+            ):
+                raise RuntimeError(
+                    "Server returned an invalid registration_token acknowledgement; "
+                    "registration may remain active and unfenced cleanup is unsafe"
+                )
+            self._registration_token = token
+        self._registration_worker_id = self.worker_id
+        self._registration_namespace = getattr(self.client, "namespace", None)
         self._registered = True
         # Adapt to the server-advertised cadence when present so a cluster
         # can pin the worker fleet's heartbeat beat without each worker
@@ -4567,8 +4599,9 @@ def _track_worker_session_from_task(self, task: dict[str, Any]) -> None:
     async def stop(self) -> None:
         """Stop polling, drain tasks, and remove the worker registration.
 
-        Repeated calls share one shutdown result. A failed deregistration is
-        therefore propagated to every caller without sending another DELETE.
+        Repeated calls share one shutdown result. Generation-fenced recovery
+        stays within the original shutdown deadline. Unfenced deregistration
+        makes one attempt and does not retry an uncertain outcome.
         """
         self._stop.set()
         if self._shutdown_task is None:
@@ -4664,8 +4697,9 @@ async def _shutdown(self) -> None:
                 log.warning("worker session %s close failed: %s", session.options.session_id, error)
 
         if self._registered:
-            await self.client.deregister_worker_registration(self.worker_id)
+            await self._deregister_before_deadline(deadline)
             self._registered = False
+            self._registration_token = None
             log.info("worker %s deregistered", self.worker_id)
         self._sticky_cache.clear()
         payload_observers = getattr(self.client, "_workflow_payload_observers", None)
@@ -4676,3 +4710,62 @@ async def _shutdown(self) -> None:
     @staticmethod
     def _remaining_shutdown_time(deadline: float) -> float:
         return max(0.0, deadline - asyncio.get_running_loop().time())
+
+    def _check_deregistration_identity(self) -> None:
+        if self._deregistration_fenced and self._registration_token is None:
+            raise RuntimeError("worker registration token is missing; registration removal is unconfirmed")
+        if self._registration_worker_id is not None and (
+            self.worker_id != self._registration_worker_id
+            or getattr(self.client, "namespace", None) != self._registration_namespace
+        ):
+            raise RuntimeError("worker registration identity changed; registration removal is unconfirmed")
+
+    async def _deregistration_attempt(self, remaining: float) -> tuple[float | None, Exception] | None:
+        kwargs: dict[str, Any] = {}
+        if self._deregistration_fenced:
+            kwargs = {"registration_token": self._registration_token, "timeout": remaining}
+        if isinstance(self._registration_namespace, str):
+            kwargs["expected_namespace"] = self._registration_namespace
+        try:
+            await asyncio.wait_for(
+                self.client.deregister_worker_registration(self.worker_id, **kwargs),
+                timeout=remaining,
+            )
+        except asyncio.TimeoutError as error:
+            raise RuntimeError(
+                "worker shutdown timed out during deregistration; registration removal is unconfirmed"
+            ) from error
+        except ServerError as error:
+            retry_delay = (
+                error.worker_deregistration_retry_delay(self.worker_id, self._registration_token)
+                if self._deregistration_fenced and self._registration_token is not None else None
+            )
+            if retry_delay is None:
+                raise
+            return float(retry_delay), error
+        except (httpx.NetworkError, httpx.TimeoutException, httpx.RemoteProtocolError) as error:
+            if not self._deregistration_fenced:
+                raise
+            return None, error
+        return None
+
+    async def _deregister_before_deadline(self, deadline: float) -> None:
+        attempt = 0
+        while True:
+            self._check_deregistration_identity()
+            remaining = self._remaining_shutdown_time(deadline)
+            if remaining <= 0:
+                raise RuntimeError("worker shutdown deadline exhausted; registration removal is unconfirmed")
+            retry = await self._deregistration_attempt(remaining)
+            if retry is None:
+                return
+            retry_delay, failure = retry
+            delay = min(1.0, 0.1 * 2 ** min(attempt, 4)) if retry_delay is None else retry_delay
+            if delay >= self._remaining_shutdown_time(deadline):
+                raise RuntimeError(
+                    "worker shutdown cannot retry within its deadline; registration removal is unconfirmed"
+                ) from failure
+            if attempt == 0:
+                log.warning("worker deregistration is unconfirmed; retrying the fenced request within its deadline")
+            attempt += 1
+            await asyncio.sleep(delay)

tests/deregistration_recovery.py

@@ -0,0 +1,51 @@
+"""Shared workflow and committed-response fault for deregistration controls."""
+from __future__ import annotations
+
+from typing import Any
+
+import httpx
+
+from durable_workflow import Worker, activity, workflow
+
+
+@activity.defn(name="tests.deregistration.echo")
+def echo(value: str) -> str:
+    return value
+
+
+@workflow.defn(name="tests.deregistration.cold-recovery")
+class RecoveredWorkflow:
+    def run(self, ctx: Any, value: str) -> Any:
+        returned = yield ctx.schedule_activity("tests.deregistration.echo", [value], retry_policy={"max_attempts": 1})
+        return {"value": returned, "run": ctx.run_id}
+
+
+class DiscardCompletedResponse(httpx.AsyncBaseTransport):
+    def __init__(self, replacement: Worker) -> None:
+        self.inner = httpx.AsyncHTTPTransport()
+        self.replacement = replacement
+        self.discarded = False
+        self.deregistration_bodies: list[bytes] = []
+        self.recovered_workflow_task_count: int | None = None
+
+    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+        response = await self.inner.handle_async_request(request)
+        if request.method != "POST" or not request.url.path.endswith("/deregister"):
+            return response
+        self.deregistration_bodies.append(request.content)
+        if self.discarded:
+            return response
+        # The real Server commits before only the caller's response is lost.
+        await response.aread()
+        assert response.status_code == 200
+        receipt = response.json()
+        assert receipt["outcome"] == "deregistered"
+        self.recovered_workflow_task_count = receipt["recovered_workflow_task_count"]
+        await response.aclose()
+        await self.replacement._register()
+        assert self.replacement._registration_token != receipt["registration_token"]
+        self.discarded = True
+        raise httpx.RemoteProtocolError("test discarded a committed response", request=request)
+
+    async def aclose(self) -> None:
+        await self.inner.aclose()