Skip to content

Commit 6c0ffdb

Browse files
Generate and reuse durable request ids for signals and updates
Part of durable-workflow/.github#216 (safe signal/update retries after a lost response). Work in progress; see the draft PR body for the remaining steps. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1 parent 7fe39db commit 6c0ffdb

10 files changed

Lines changed: 674 additions & 41 deletions

‎src/durable_workflow/client.py‎

Lines changed: 125 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
from __future__ import annotations
2020

2121
import asyncio
22+
import contextlib
2223
import copy
2324
import hashlib
2425
import json as json_module
@@ -47,6 +48,7 @@
4748
from ._cooperative_cancellation import CancellationDelivery, CancellationRequest
4849
from ._query_task_authority import QueryTaskAuthority, current_query_authority
4950
from .errors import (
51+
REQUEST_ID_CONFLICT,
5052
ExternalPayloadIntegrityMismatch,
5153
ExternalPayloadOversized,
5254
ExternalPayloadUnavailable,
@@ -55,6 +57,8 @@
5557
RuntimeCapabilityUnsupported,
5658
RuntimeDiscoveryUnavailable,
5759
ServerError,
60+
SignalFailed,
61+
UpdateFailed,
5862
WorkflowCancelled,
5963
WorkflowFailed,
6064
WorkflowTerminated,
@@ -239,6 +243,49 @@ def _normalize_base_url(base_url: str) -> str:
239243
return base_url.rstrip("/")
240244

241245

246+
def _command_request_id(request_id: str | None, command: str) -> str:
247+
"""Return the caller command key for one logical signal or update call.
248+
249+
The key is the body field ``request_id`` (1-255 opaque, case-sensitive
250+
bytes). A caller-supplied key is sent verbatim; otherwise one
251+
``sdk-python-<command>-<uuid4 hex>`` key is generated per logical call so
252+
the transport retry policy resends the identical key.
253+
"""
254+
return request_id or f"sdk-python-{command}-{uuid.uuid4().hex}"
255+
256+
257+
def _attach_request_id(exc: BaseException, request_id: str) -> None:
258+
"""Record the sent command key on an exception as ``exc.request_id``.
259+
260+
Typed SDK failures declare the attribute; transport errors raised by
261+
``httpx`` receive it dynamically so a manual retry can reuse the key.
262+
"""
263+
with contextlib.suppress(AttributeError, TypeError):
264+
exc.request_id = request_id # type: ignore[attr-defined]
265+
266+
267+
def _command_request_failure(exc: Exception, *, command: str, request_id: str) -> Exception:
268+
"""Return the exception a failed signal or update call should raise.
269+
270+
A ``request_id_conflict`` refusal that reached the caller as a generic
271+
:class:`ServerError` becomes the command's typed failure. Every other
272+
exception is returned unchanged, annotated with the key that was sent.
273+
"""
274+
if isinstance(exc, ServerError) and exc.reason() == REQUEST_ID_CONFLICT:
275+
message = exc.body.get("message") if isinstance(exc.body, dict) else None
276+
if not isinstance(message, str) or not message:
277+
message = f"{command} request id already retained with a different intent"
278+
if command == "update" and isinstance(exc.body, dict):
279+
return UpdateFailed(
280+
message, status=exc.status, body=exc.body, reason=REQUEST_ID_CONFLICT, request_id=request_id,
281+
)
282+
return SignalFailed(
283+
message, reason=REQUEST_ID_CONFLICT, status=exc.status, body=exc.body, request_id=request_id,
284+
)
285+
_attach_request_id(exc, request_id)
286+
return exc
287+
288+
242289
def _route_for_metrics(path: str) -> str:
243290
clean_path = path.split("?", 1)[0]
244291
parts = [part for part in clean_path.strip("/").split("/") if part]
@@ -1346,9 +1393,17 @@ async def describe_run(self, run_id: str | None = None) -> WorkflowRun:
13461393
raise ValueError("run_id is required to describe a workflow run from a handle")
13471394
return await self._client.describe_workflow_run(self.workflow_id, selected_run_id)
13481395

1349-
async def signal(self, signal_name: str, args: list[Any] | None = None) -> None:
1396+
async def signal(
1397+
self,
1398+
signal_name: str,
1399+
args: list[Any] | None = None,
1400+
*,
1401+
request_id: str | None = None,
1402+
) -> dict[str, Any]:
13501403
"""Deliver an external signal to this workflow. See :meth:`Client.signal_workflow`."""
1351-
await self._client.signal_workflow(self.workflow_id, signal_name, args=args)
1404+
return await self._client.signal_workflow(
1405+
self.workflow_id, signal_name, args=args, request_id=request_id,
1406+
)
13521407

13531408
async def append_message(
13541409
self,
@@ -4303,15 +4358,38 @@ async def send_webhook_bridge_event(
43034358
return BridgeAdapterOutcome.from_dict(data)
43044359

43054360
async def signal_workflow(
4306-
self, workflow_id: str, signal_name: str, *, args: list[Any] | None = None
4307-
) -> None:
4361+
self,
4362+
workflow_id: str,
4363+
signal_name: str,
4364+
*,
4365+
args: list[Any] | None = None,
4366+
request_id: str | None = None,
4367+
) -> dict[str, Any]:
43084368
"""Deliver an external signal to a running workflow.
43094369
43104370
Signals are fire-and-forget: the server records the signal in durable
43114371
history and returns immediately. They do not wait for the workflow to
43124372
observe the signal. See the main docs for how to declare the allowed
43134373
signal names on a workflow type.
4374+
4375+
``request_id`` is the caller command key (1-255 opaque, case-sensitive
4376+
bytes) that makes a retry of this signal safe. When omitted the client
4377+
generates ``sdk-python-signal-<uuid>`` once per call, so the transport
4378+
retry policy resends the same key. Resending the same key with the same
4379+
signal name, arguments and run target returns the original receipt
4380+
instead of recording a second signal.
4381+
4382+
Returns the Server's admission receipt. ``request_id`` is always
4383+
present (the key that was sent); a Server that retains command keys
4384+
also reports ``retained`` (``True`` when an earlier admission was
4385+
returned) and ``intent_verified``. Raises
4386+
:class:`~durable_workflow.errors.SignalFailed` with
4387+
``reason == "request_id_conflict"`` when the key was already retained
4388+
for a signal with a different intent; the retained command is described
4389+
in its ``details``. Every exception raised after the key was chosen,
4390+
including transport errors, carries it as ``request_id``.
43144391
"""
4392+
key = _command_request_id(request_id, "signal")
43154393
body: dict[str, Any] = {}
43164394
if args:
43174395
body["input"] = self._payload_envelope(
@@ -4320,7 +4398,19 @@ async def signal_workflow(
43204398
workflow_id=workflow_id,
43214399
signal_name=signal_name,
43224400
)
4323-
await self._request("POST", f"/workflows/{workflow_id}/signal/{signal_name}", json=body, context=workflow_id)
4401+
body["request_id"] = key
4402+
try:
4403+
data = await self._request(
4404+
"POST", f"/workflows/{workflow_id}/signal/{signal_name}", json=body, context=workflow_id,
4405+
)
4406+
except Exception as exc:
4407+
failure = _command_request_failure(exc, command="signal", request_id=key)
4408+
if failure is exc:
4409+
raise
4410+
raise failure from exc
4411+
receipt: dict[str, Any] = data if isinstance(data, dict) else {}
4412+
receipt.setdefault("request_id", key)
4413+
return receipt
43244414

43254415
async def append_message_stream(
43264416
self,
@@ -4496,15 +4586,28 @@ async def update_workflow(
44964586
Validator rejection and validation-routing failures remain typed and
44974587
do not silently fall back to post-accept handler failure.
44984588
4499-
``request_id`` lets the caller deduplicate retries. Raises
4500-
:class:`~durable_workflow.errors.UpdateRejected` when the workflow's
4501-
validator rejects the update, or
4589+
``request_id`` is the caller command key (1-255 opaque, case-sensitive
4590+
bytes) that makes a retry of this update safe. When omitted the client
4591+
generates ``sdk-python-update-<uuid>`` once per call, so the transport
4592+
retry policy resends the same key. Resending the same key with the same
4593+
update name, arguments and run target returns the original receipt
4594+
(``retained`` is ``True`` on a Server that reports it) instead of
4595+
admitting a second update. The returned response always carries the
4596+
``request_id`` that was sent.
4597+
4598+
Raises :class:`~durable_workflow.errors.UpdateRejected` when the
4599+
workflow's validator rejects the update, or
45024600
:class:`~durable_workflow.errors.UpdateValidationFailed` when the
45034601
declared validation boundary cannot be enforced. An accepted update
45044602
whose handler fails raises :class:`~durable_workflow.errors.UpdateFailed`
4505-
with the failure message, Server response and durable identities.
4603+
with the failure message, Server response and durable identities; the
4604+
same type with ``reason == "request_id_conflict"`` reports a key that
4605+
was already retained for an update with a different intent, described
4606+
in its ``details``. Every exception raised after the key was chosen,
4607+
including transport errors, carries it as ``request_id``.
45064608
"""
45074609
await self._require_update_wait_stage(wait_for or "accepted")
4610+
key = _command_request_id(request_id, "update")
45084611
body: dict[str, Any] = {}
45094612
if args:
45104613
body["input"] = self._payload_envelope(
@@ -4517,10 +4620,19 @@ async def update_workflow(
45174620
body["wait_for"] = wait_for
45184621
if wait_timeout_seconds is not None:
45194622
body["wait_timeout_seconds"] = wait_timeout_seconds
4520-
body["request_id"] = request_id or f"sdk-python-update-{uuid.uuid4().hex}"
4521-
return await self._request(
4522-
"POST", f"/workflows/{workflow_id}/update/{update_name}", json=body, context=workflow_id
4523-
)
4623+
body["request_id"] = key
4624+
try:
4625+
data = await self._request(
4626+
"POST", f"/workflows/{workflow_id}/update/{update_name}", json=body, context=workflow_id
4627+
)
4628+
except Exception as exc:
4629+
failure = _command_request_failure(exc, command="update", request_id=key)
4630+
if failure is exc:
4631+
raise
4632+
raise failure from exc
4633+
if isinstance(data, dict):
4634+
data.setdefault("request_id", key)
4635+
return data
45244636

45254637
async def get_result(
45264638
self,

‎src/durable_workflow/errors.py‎

Lines changed: 76 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -412,8 +412,45 @@ def validation_errors(self) -> dict[str, Any] | None:
412412
return None
413413

414414

415+
REQUEST_ID_CONFLICT = "request_id_conflict"
416+
"""Reason reported when a command key is reused with a different intent."""
417+
418+
_REQUEST_ID_CONFLICT_FIELDS = (
419+
"request_id",
420+
"retained_command_id",
421+
"retained_command_type",
422+
"retained_outcome",
423+
"conflict",
424+
)
425+
426+
427+
def _body_string(body: object, name: str) -> str | None:
428+
if isinstance(body, dict):
429+
value = body.get(name)
430+
if isinstance(value, str) and value:
431+
return value
432+
return None
433+
434+
435+
def _request_id_conflict_details(body: object) -> dict[str, Any]:
436+
"""Return the retained-command fields of a ``request_id_conflict`` response body."""
437+
if not isinstance(body, dict) or body.get("reason") != REQUEST_ID_CONFLICT:
438+
return {}
439+
return {name: body[name] for name in _REQUEST_ID_CONFLICT_FIELDS if name in body}
440+
441+
415442
class SignalFailed(DurableWorkflowError):
416-
"""A workflow signal was rejected before it could be delivered."""
443+
"""A workflow signal was rejected before it could be delivered.
444+
445+
:attr:`request_id` is the caller command key the client sent with the
446+
signal (caller-supplied or SDK-generated), so the application can retry
447+
the same logical signal with the same key. :attr:`reason` is the Server's
448+
machine-readable refusal reason; ``request_id_conflict`` means the key was
449+
already retained for a signal with a different intent, and
450+
:attr:`details` then carries ``retained_command_id``,
451+
``retained_command_type``, ``retained_outcome`` and the mismatched
452+
``conflict`` parts.
453+
"""
417454

418455
def __init__(
419456
self,
@@ -422,11 +459,15 @@ def __init__(
422459
reason: str | None = None,
423460
status: int | None = None,
424461
body: object | None = None,
462+
request_id: str | None = None,
463+
details: dict[str, Any] | None = None,
425464
) -> None:
426465
super().__init__(message)
427466
self.reason = reason
428467
self.status = status
429468
self.body = body
469+
self.request_id = request_id if request_id is not None else _body_string(body, "request_id")
470+
self.details: dict[str, Any] = dict(details) if details is not None else _request_id_conflict_details(body)
430471

431472
@property
432473
def validation_errors(self) -> dict[str, Any] | None:
@@ -508,17 +549,37 @@ def __init__(
508549

509550

510551
class UpdateFailed(InvalidArgument):
511-
"""An accepted workflow update failed during handler execution.
552+
"""An accepted workflow update failed, or its command key was refused.
512553
513554
``body`` preserves the Server response, including its durable failure
514555
identity. This inherits from :class:`InvalidArgument` to preserve catches
515556
written for earlier SDK versions that mapped all HTTP 422 responses there.
557+
558+
:attr:`request_id` is the caller command key the client sent with the
559+
update (caller-supplied or SDK-generated). :attr:`reason` is the Server's
560+
machine-readable reason when it returned one; ``request_id_conflict``
561+
means the key was already retained for an update with a different intent,
562+
and :attr:`details` then carries ``retained_command_id``,
563+
``retained_command_type``, ``retained_outcome`` and the mismatched
564+
``conflict`` parts.
516565
"""
517566

518-
def __init__(self, message: str, *, status: int, body: dict[str, Any]) -> None:
567+
def __init__(
568+
self,
569+
message: str,
570+
*,
571+
status: int,
572+
body: dict[str, Any],
573+
reason: str | None = None,
574+
request_id: str | None = None,
575+
details: dict[str, Any] | None = None,
576+
) -> None:
519577
super().__init__(message)
520578
self.status = status
521579
self.body = body
580+
self.reason = reason if reason is not None else self._identity(body, "reason")
581+
self.request_id = request_id if request_id is not None else self._identity(body, "request_id")
582+
self.details: dict[str, Any] = dict(details) if details is not None else _request_id_conflict_details(body)
522583
self.workflow_id = self._identity(body, "workflow_id")
523584
self.run_id = self._identity(body, "run_id")
524585
self.update_id = self._identity(body, "update_id")
@@ -909,6 +970,18 @@ def update_validation_failed(default: str) -> UpdateValidationFailed:
909970
raise UpdateRejected(message or "update rejected", body=body)
910971
if reason in update_validation_failure_reasons:
911972
raise update_validation_failed("update validation failed")
973+
if reason == REQUEST_ID_CONFLICT:
974+
# The retained command type names the key scope; the Server also
975+
# labels the control-plane operation. Either selects the typed failure.
976+
command_type = _body_string(body, "retained_command_type") or operation
977+
if command_type == "signal":
978+
raise signal_failed("signal request id already retained with a different intent")
979+
if command_type == "update" and isinstance(body, dict):
980+
raise UpdateFailed(
981+
message or "update request id already retained with a different intent",
982+
status=status,
983+
body=body,
984+
)
912985
raise ServerError(status, body)
913986

914987
if status == 504:

‎src/durable_workflow/retry_policy.py‎

Lines changed: 18 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,8 @@
33
.. warning::
44
55
:class:`TransportRetryPolicy` covers **only client-side HTTP retries** for
6-
transient transport errors (connection failures, timeouts, 5xx responses,
6+
transient transport errors (connection failures, timeouts, a server or
7+
proxy that disconnects before sending a response, 5xx responses,
78
429 rate-limiting). It is **not** the activity retry policy. Activity-level
89
retry and timeout configuration lives on
910
:class:`durable_workflow.workflow.ActivityRetryPolicy` and is passed to
@@ -201,9 +202,16 @@ class TransportRetryPolicy:
201202
"""
202203
Retry policy for transient HTTP transport errors.
203204
204-
Retries requests that fail with transient errors (connection errors,
205-
timeouts, 5xx server errors, 429 rate limit). Does not retry client
206-
errors (4xx except 429).
205+
Retries requests that fail with transient errors: ``httpx.NetworkError``
206+
(including ``ConnectError``), ``httpx.TimeoutException``,
207+
``httpx.RemoteProtocolError`` (the server or a proxy closed the connection
208+
before sending a response), 5xx server errors and 429 rate limit. Does not
209+
retry client errors (4xx except 429) or ``httpx.LocalProtocolError``.
210+
211+
A retried request is resent byte-identical. Signals and updates carry a
212+
durable ``request_id`` per logical call, so a retry after a lost response
213+
is answered from the retained receipt instead of admitting a second
214+
command.
207215
208216
This policy runs inside :class:`~durable_workflow.Client` around HTTP
209217
requests. It does not retry workflow runs, workflow tasks, activity
@@ -228,8 +236,12 @@ def should_retry(self, exc: Exception, attempt: int) -> bool:
228236
if attempt >= self.max_attempts:
229237
return False
230238

231-
# Retry connection errors and timeouts
232-
if isinstance(exc, httpx.ConnectError | httpx.TimeoutException | httpx.NetworkError):
239+
# Retry connection errors, timeouts and a peer that disconnected before
240+
# answering (a lost response is recovered through the request's identity).
241+
if isinstance(
242+
exc,
243+
httpx.ConnectError | httpx.TimeoutException | httpx.NetworkError | httpx.RemoteProtocolError,
244+
):
233245
return True
234246

235247
# Retry 5xx server errors and 429 rate limit

0 commit comments

Comments
 (0)