From d9d9349f207cf49d124acc7cdbe40387e02277e7 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 04:54:28 +0000 Subject: [PATCH 1/2] Read an accepted update's outcome later by update id (#212) Client.update_outcome() and update_outcome_receipt() call GET /workflows/{workflow_id}/updates/{update_id} (server#1131) on a Server that advertises capabilities.update_outcome_reads. They never resubmit the update: without the capability they raise RuntimeCapabilityUnsupported after discovery. A pending read raises the new UpdateStillPending (never a None result) with a read-again-by-id hint; a failed update raises UpdateFailed, a rejected one UpdateRejected, and 404 update_not_found the new UpdateNotFound (an UpdateFailed). WorkflowHandle and the sync client/handle get the same methods. update_workflow's docstring now states that it returns the receipt and that the Server defaults to wait_for=accepted. Part of durable-workflow/.github#212. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 8 + docs/sdk-reference.md | 42 ++++ src/durable_workflow/__init__.py | 4 + src/durable_workflow/_metric_routes.py | 1 + src/durable_workflow/client.py | 218 +++++++++++++++- src/durable_workflow/errors.py | 78 ++++++ src/durable_workflow/sync.py | 70 ++++++ tests/test_metric_routes.py | 1 + tests/test_update_outcome.py | 328 +++++++++++++++++++++++++ 9 files changed, 744 insertions(+), 6 deletions(-) create mode 100644 tests/test_update_outcome.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 8d3f6ae4..3680af42 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- `Client.update_outcome()` and `Client.update_outcome_receipt()` read an accepted update's outcome later by its update id with `GET /api/workflows/{workflow_id}/updates/{update_id}`, and never submit the update again (durable-workflow/.github#212, server#1131). `update_outcome()` waits for the outcome by default and returns the decoded result, including `None`, with optional `result_type` decoding. A still-pending update raises the new `UpdateStillPending`, which carries `workflow_id`, `run_id`, `update_name`, `update_id`, `request_id` and `update_status` and tells the caller to read again by update id; it is never returned as `None`. A failed update raises `UpdateFailed`, a rejected one `UpdateRejected`, and an unknown or pruned id the new `UpdateNotFound` (a subclass of `UpdateFailed`). `update_outcome_receipt()` returns the raw receipt and answers at once by default. Both need a Server that advertises `capabilities.update_outcome_reads` (native 3.6.0 and later); on an older Server they raise `RuntimeCapabilityUnsupported` after discovery, without reading or resubmitting anything. `WorkflowHandle` and the synchronous client and handle have the same methods. + +### Changed + +- The `Client.update_workflow()` and `WorkflowHandle.update()` docstrings now say what they always did: they return the Server receipt, not the decoded result, and the Server answers at `wait_for="accepted"` when it is omitted. A receipt with `update_status: "accepted"` is a pending update even though its `result` is `None`. + ## [2.16.0] - 2026-11-12 ### Added diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index 15e35e64..6d842fc0 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -618,6 +618,48 @@ workers, and unsupported capability paths raise `UpdateValidationFailed` with the Server's typed reason and retryability. `wait_for="completed"` additionally waits for the accepted update handler to reach its terminal outcome. +`Client.update_workflow()` (and `WorkflowHandle.update()`) returns the Server +receipt, not the decoded handler result. When `wait_for` is omitted the Server +answers at `accepted`. Read the receipt's `update_status`: `"completed"` carries +the decoded value in `result`, which may be `None`; `"accepted"` means the +update is still pending and has no result yet, even though `result` is `None`. +`wait_timed_out` is `True` when a `wait_for="completed"` wait ran out first. + +To read the outcome later, keep the receipt's `update_id` and call +`Client.update_outcome()` (or `WorkflowHandle.update_outcome()`). It reads the +update by id and never submits it again, so it cannot admit a second update: + +```python +from durable_workflow import UpdateNotFound, UpdateStillPending + +receipt = await client.update_workflow("order-1", "rename", args=["new name"]) +update_id = receipt["update_id"] + +try: + result = await client.update_outcome("order-1", update_id, wait_timeout_seconds=30) +except UpdateStillPending as pending: + ... # still running: call update_outcome() again with pending.update_id +except UpdateNotFound: + ... # no retained update has this id, for example after pruning +``` + +`update_outcome()` waits for the outcome by default and returns the decoded +result, including `None` (`result_type=` decodes it to an annotation). A +pending update raises `UpdateStillPending` with `workflow_id`, `run_id`, +`update_name`, `update_id`, `request_id` and `update_status`; it is never +returned as `None`. A failed update raises `UpdateFailed`, a rejected one +`UpdateRejected`, and an unknown or pruned id `UpdateNotFound` (a subclass of +`UpdateFailed`; it does not prove the update was never admitted). +`update_outcome_receipt()` returns the raw receipt instead and answers at once +unless you pass `wait_for="completed"`. The read follows the update across +continue-as-new: the receipt's `run_id` names the run that owns it now. + +Both need a Server that advertises `capabilities.update_outcome_reads: true` +in `GET /api/cluster/info` (native Server 3.6.0 and later). On an older Server +they raise `RuntimeCapabilityUnsupported` after discovery, without reading or +resubmitting the update. The synchronous `durable_workflow.sync.Client` and +its workflow handles offer the same methods. + Malformed signal and query payloads are reported as typed client errors with the server's documented reason and status preserved: diff --git a/src/durable_workflow/__init__.py b/src/durable_workflow/__init__.py index 941e659a..81e350a0 100644 --- a/src/durable_workflow/__init__.py +++ b/src/durable_workflow/__init__.py @@ -105,7 +105,9 @@ TypedValueError, Unauthorized, UpdateFailed, + UpdateNotFound, UpdateRejected, + UpdateStillPending, UpdateValidationFailed, WorkflowAlreadyStarted, WorkflowCancelled, @@ -412,7 +414,9 @@ "TransportRetryPolicy", "Unauthorized", "UpdateFailed", + "UpdateNotFound", "UpdateRejected", + "UpdateStillPending", "UpdateValidationFailed", "WorkflowAlreadyStarted", "WorkflowCancelled", diff --git a/src/durable_workflow/_metric_routes.py b/src/durable_workflow/_metric_routes.py index ad2e1d56..578b6397 100644 --- a/src/durable_workflow/_metric_routes.py +++ b/src/durable_workflow/_metric_routes.py @@ -31,6 +31,7 @@ *(f"/workflows/{{workflow_id}}/{action}" for action in ("archive", "cancel", "repair", "terminate", "request-cancellation", "runs")), *(f"/workflows/{{workflow_id}}/{action}/{{name}}" for action in ("signal", "query", "update")), + "/workflows/{workflow_id}/updates/{update_id}", "/workflows/{workflow_id}/message-streams/{name}/messages", "/workflows/{workflow_id}/runs/{run_id}", *(f"/workflows/{{workflow_id}}/runs/{{run_id}}/{action}" diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index 2f38d917..95705b5b 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -75,11 +75,14 @@ ServerError, SignalFailed, UpdateFailed, + UpdateRejected, + UpdateStillPending, WorkflowCancelled, WorkflowFailed, WorkflowTerminated, WorkflowTimedOut, _raise_for_status, + _raise_for_update_outcome_read, ) from .external_storage import ( RUNTIME_EXTERNAL_PAYLOAD_REFERENCE_SCHEMA, @@ -121,6 +124,8 @@ CONTROL_PLANE_REQUEST_CONTRACT_VERSION = 1 _QUERY_TASKS_DISCOVERY_PATH = "worker_protocol.server_capabilities.query_tasks" _COOPERATIVE_CANCELLATION_DISCOVERY_PATH = "worker_protocol.server_capabilities.cooperative_cancellation" +_UPDATE_OUTCOME_READS_DISCOVERY_PATH = "capabilities.update_outcome_reads" +_UPDATE_OUTCOME_WAIT_STAGES = ("accepted", "completed") _UPDATE_WAIT_STAGES_DISCOVERY_PATH = ( "control_plane.request_contract.operations.update.fields.wait_for.canonical_values" ) @@ -304,6 +309,45 @@ def _command_request_failure(exc: Exception, *, command: str, request_id: str) - return exc +def _update_outcome_value( + receipt: dict[str, Any], *, workflow_id: str, update_id: str, result_type: Any = None, +) -> Any: + """Return a completed update's decoded result, or raise for any other receipt. + + A pending receipt raises :class:`UpdateStillPending`, never ``None``. + """ + status = receipt.get("update_status") + if isinstance(receipt.get("result_envelope"), dict) or ( + status == "completed" and receipt.get("wait_timed_out") is not True + ): + value = receipt.get("result") + return value if result_type is None else decode_payload_value(value, result_type, "update result") + if status == "failed": + message = receipt.get("failure_message") + raise UpdateFailed( + message if isinstance(message, str) and message else "workflow update failed", + status=200, + body=receipt, + ) + if status == "rejected": + raise UpdateRejected("update rejected", body=receipt) + + def text(name: str) -> str | None: + value = receipt.get(name) + return value if isinstance(value, str) and value else None + + raise UpdateStillPending( + workflow_id=text("workflow_id") or workflow_id, + update_id=text("update_id") or update_id, + run_id=text("run_id"), + update_name=text("update_name"), + request_id=text("request_id"), + update_status=text("update_status"), + wait_timed_out=receipt.get("wait_timed_out") is True, + body=receipt, + ) + + def _resolve_namespace_name( name: str | None, namespace_alias: str | None, @@ -1633,7 +1677,7 @@ async def update( wait_timeout_seconds: int | None = None, request_id: str | None = None, ) -> Any: - """Send a synchronous update to this workflow and wait for the result. See :meth:`Client.update_workflow`.""" + """Send an update to this workflow and return the Server's receipt. See :meth:`Client.update_workflow`.""" return await self._client.update_workflow( self.workflow_id, update_name, @@ -1643,6 +1687,38 @@ async def update( request_id=request_id, ) + async def update_outcome( + self, + update_id: str, + *, + wait_for: str = "completed", + wait_timeout_seconds: int | None = None, + result_type: Any = None, + ) -> Any: + """Read an accepted update's outcome later by update id. See :meth:`Client.update_outcome`.""" + return await self._client.update_outcome( + self.workflow_id, + update_id, + wait_for=wait_for, + wait_timeout_seconds=wait_timeout_seconds, + result_type=result_type, + ) + + async def update_outcome_receipt( + self, + update_id: str, + *, + wait_for: str = "accepted", + wait_timeout_seconds: int | None = None, + ) -> dict[str, Any]: + """Read an accepted update's receipt by update id. See :meth:`Client.update_outcome_receipt`.""" + return await self._client.update_outcome_receipt( + self.workflow_id, + update_id, + wait_for=wait_for, + wait_timeout_seconds=wait_timeout_seconds, + ) + class StandaloneActivityHandle: """Convenience wrapper for operating on one standalone activity job. @@ -2068,6 +2144,7 @@ async def _request( timeout: float | None = None, context: str = "", retry: bool = True, + raise_for_status: Callable[..., None] = _raise_for_status, ) -> Any: if worker and discovery: raise ValueError("a request cannot be both worker-plane and cluster discovery") @@ -2167,7 +2244,7 @@ async def _do_request() -> httpx.Response: body = exc.response.json() except ValueError: body = exc.response.text - _raise_for_status(exc.response.status_code, body, context=context) + raise_for_status(exc.response.status_code, body, context=context) raise # unreachable, but keeps type checker happy status_code = str(resp.status_code) @@ -3098,6 +3175,27 @@ async def _require_update_wait_stage(self, wait_for: str) -> None: supported_values=supported_stages, ) + async def _require_update_outcome_reads(self, operation: str) -> None: + info = await self._runtime_discovery( + operation=operation, + required_path=_UPDATE_OUTCOME_READS_DISCOVERY_PATH, + ) + capabilities = info.get("capabilities") + supported = capabilities.get("update_outcome_reads") if isinstance(capabilities, dict) else None + if supported is not True: + raise RuntimeCapabilityUnsupported( + operation, + _UPDATE_OUTCOME_READS_DISCOVERY_PATH, + ( + f"{operation} is unsupported by this runtime because GET /api/cluster/info " + f"does not advertise {_UPDATE_OUTCOME_READS_DISCOVERY_PATH}=true (native " + "Server 3.6.0 or later), so an update outcome cannot be read by id. The SDK " + "never resubmits an update in place of reading it. Upgrade Server, or retry " + "update_workflow() with the update's original request_id, which returns the " + "retained receipt instead of admitting another update." + ), + ) + def get_workflow_handle( self, workflow_id: str, *, run_id: str | None = None, workflow_type: str = "" ) -> WorkflowHandle: @@ -4964,18 +5062,28 @@ async def update_workflow( wait_timeout_seconds: int | None = None, request_id: str | None = None, ) -> Any: - """Send a synchronous update to a running workflow and wait for the result. + """Send an update to a running workflow and return the Server's receipt. Updates are request/response calls to a named handler on the workflow; the handler may mutate durable workflow state and return a value. - ``wait_for`` selects how long the server waits before returning — - typically ``completed`` to block until the handler finishes, or - ``accepted`` to return once the request is durably accepted and routed. + ``wait_for`` selects how long the server waits before answering: + ``"accepted"`` (the default when omitted) answers once the update is + durably accepted and routed, and ``"completed"`` waits up to + ``wait_timeout_seconds`` for the handler's outcome. When the workflow contract declares a validator, a capable Server does not return that accepted state until synchronous validation succeeds. Validator rejection and validation-routing failures remain typed and do not silently fall back to post-accept handler failure. + The return value is the receipt ``dict``, not the decoded handler + result. Its ``update_status`` says where the update is: ``"completed"`` + carries the decoded value in ``result`` (which may be ``None``) and the + encoded ``result_envelope``; ``"accepted"`` means the update is still + pending and has no result yet, even though ``result`` is ``None``, and + ``wait_timed_out`` is ``True`` when a completion wait ran out. Keep the + receipt's ``update_id`` to read the outcome later with + :meth:`update_outcome`, which never admits another update. + ``request_id`` is the caller command key (1-255 opaque, case-sensitive bytes) that makes a retry of this update safe. When omitted the client generates ``sdk-python-update-`` once per call, so the transport @@ -5025,6 +5133,104 @@ async def update_workflow( data.setdefault("request_id", key) return data + async def update_outcome( + self, + workflow_id: str, + update_id: str, + *, + wait_for: str = "completed", + wait_timeout_seconds: int | None = None, + result_type: Any = None, + ) -> Any: + """Read an accepted update's outcome later by its update id, without submitting it again. + + By default this waits for the outcome (``wait_for="completed"``) up to + ``wait_timeout_seconds``, or the Server's update wait default. It + returns the decoded result of a completed update, including ``None``. + ``result_type`` opts in to decoding the result to that annotation, as + :meth:`get_result` does. + + A still-pending update raises + :class:`~durable_workflow.errors.UpdateStillPending` and is never + returned as ``None``; call this again with the same ``update_id`` to + keep waiting. A failed update raises + :class:`~durable_workflow.errors.UpdateFailed`, a rejected one + :class:`~durable_workflow.errors.UpdateRejected`, and an id with no + retained update (including one pruned with its run) + :class:`~durable_workflow.errors.UpdateNotFound`. + + The read follows the update across continue-as-new; the receipt's + ``run_id`` names the run that owns it now. It needs a Server that + advertises ``capabilities.update_outcome_reads`` (native 3.6.0 and + later) and otherwise raises + :class:`~durable_workflow.errors.RuntimeCapabilityUnsupported` without + reading or resubmitting anything. + """ + if result_type is not None: + validate_annotation(result_type, "result_type") + receipt = await self._read_update_outcome( + "Client.update_outcome", workflow_id, update_id, wait_for, wait_timeout_seconds, + ) + return _update_outcome_value(receipt, workflow_id=workflow_id, update_id=update_id, result_type=result_type) + + async def update_outcome_receipt( + self, + workflow_id: str, + update_id: str, + *, + wait_for: str = "accepted", + wait_timeout_seconds: int | None = None, + ) -> dict[str, Any]: + """Read an accepted update's Server receipt by its update id. It writes nothing. + + By default this answers at once (``wait_for="accepted"``); pass + ``wait_for="completed"`` to wait for the outcome. The receipt is the + same one :meth:`update_workflow` returns: ``update_status`` is + ``"accepted"`` while the update is pending (``wait_timed_out`` says + whether a completion wait ran out) and ``"completed"`` once it has a + ``result``, which may be ``None``. + + Failed, rejected and unknown updates raise as in + :meth:`update_outcome`. Without the Server's + ``capabilities.update_outcome_reads`` this raises + :class:`~durable_workflow.errors.RuntimeCapabilityUnsupported`; + resubmitting the update is never a substitute for reading it. + """ + return await self._read_update_outcome( + "Client.update_outcome_receipt", workflow_id, update_id, wait_for, wait_timeout_seconds, + ) + + async def _read_update_outcome( + self, + operation: str, + workflow_id: str, + update_id: str, + wait_for: str, + wait_timeout_seconds: int | None, + ) -> dict[str, Any]: + if wait_for not in _UPDATE_OUTCOME_WAIT_STAGES: + raise ValueError(f"{operation} wait_for must be 'accepted' or 'completed', not {wait_for!r}") + if not isinstance(update_id, str) or not update_id.strip(): + raise ValueError(f"{operation} needs the update id") + await self._require_update_outcome_reads(operation) + path = f"/workflows/{quote(workflow_id, safe='')}/updates/{quote(update_id, safe='')}" + query: dict[str, Any] = {} + if wait_for == "completed": + query["wait_for"] = wait_for + if wait_timeout_seconds is not None: + query["wait_timeout_seconds"] = wait_timeout_seconds + if query: + path += "?" + urlencode(query) + data = await self._request( + "GET", path, context=workflow_id, raise_for_status=_raise_for_update_outcome_read, + ) + if not isinstance(data, dict): + raise ServerError( + 200, + {"reason": "invalid_update_receipt", "message": f"expected JSON object, got {type(data).__name__}"}, + ) + return data + async def get_result( self, handle: WorkflowHandle, diff --git a/src/durable_workflow/errors.py b/src/durable_workflow/errors.py index f53d0338..67f77078 100644 --- a/src/durable_workflow/errors.py +++ b/src/durable_workflow/errors.py @@ -636,6 +636,55 @@ def _identity(body: dict[str, Any], name: str) -> str | None: return value if isinstance(value, str) and value.strip() else None +class UpdateNotFound(UpdateFailed): + """No retained update has this id for this workflow (HTTP 404 ``update_not_found``). + + Raised by :meth:`~durable_workflow.Client.update_outcome` and + :meth:`~durable_workflow.Client.update_outcome_receipt`. This includes an + update pruned with its run, so it does not prove the update was never + admitted. It subclasses :class:`UpdateFailed`, so callers that treat every + unreadable outcome alike can catch that. + """ + + +class UpdateStillPending(DurableWorkflowError): + """An accepted update that has not completed, so it has no result yet. + + Raised by :meth:`~durable_workflow.Client.update_outcome` when the Server + answers with an accepted, still-open update: either a plain read, or a + completion wait that ran out (:attr:`wait_timed_out`). This is never a + ``None`` result. Call ``update_outcome()`` again with :attr:`update_id` to + keep waiting; reading never admits another update. + """ + + def __init__( + self, + *, + workflow_id: str, + update_id: str, + run_id: str | None = None, + update_name: str | None = None, + request_id: str | None = None, + update_status: str | None = None, + wait_timed_out: bool = False, + body: dict[str, Any] | None = None, + ) -> None: + label = f"{update_name} ({update_id})" if update_name else update_id + super().__init__( + f"Update {label} on workflow {workflow_id} is still {update_status or 'pending'}; it has no " + f"result yet. Call update_outcome() again with update id {update_id!r} to keep waiting. " + "Reading never admits another update." + ) + self.workflow_id = workflow_id + self.update_id = update_id + self.run_id = run_id + self.update_name = update_name + self.request_id = request_id + self.update_status = update_status + self.wait_timed_out = wait_timed_out + self.body = body + + class UpdateRejected(DurableWorkflowError): """A workflow update was rejected by the workflow's validator.""" @@ -1089,6 +1138,35 @@ def update_validation_failed(default: str) -> UpdateValidationFailed: raise ServerError(status, body) +def _raise_for_update_outcome_read(status: int, body: object, *, context: str = "") -> None: + """Map a refused update outcome read (Server #1131) to the update's typed failure. + + ``404 update_not_found`` raises :class:`UpdateNotFound`, and a rejected + update, which answers with its original rejection status, raises + :class:`UpdateRejected` as submitting it did. Everything else, including + ``422`` failed updates, takes the usual mapping. + """ + if status >= 400 and isinstance(body, dict): + reason = body.get("reason") + message = body.get("message") + if status == 404 and reason == "update_not_found": + raise UpdateNotFound( + message if isinstance(message, str) and message else "update not found", + status=status, + body=body, + ) + if body.get("update_status") == "rejected": + if not isinstance(message, str) or not message: + message = f"update rejected: {reason}" if isinstance(reason, str) and reason else "update rejected" + errors = body.get("validation_errors") + raise UpdateRejected( + message, + validation_errors=errors if isinstance(errors, dict) and errors else None, + body=body, + ) + _raise_for_status(status, body, context=context) + + def _control_plane_operation(body: object) -> str | None: if not isinstance(body, dict): return None diff --git a/src/durable_workflow/sync.py b/src/durable_workflow/sync.py index 9c37e805..22293ec2 100644 --- a/src/durable_workflow/sync.py +++ b/src/durable_workflow/sync.py @@ -171,6 +171,39 @@ def update( ) ) + def update_outcome( + self, + update_id: str, + *, + wait_for: str = "completed", + wait_timeout_seconds: int | None = None, + result_type: Any = None, + ) -> Any: + return self._runner.run( + self._handle.update_outcome( + update_id, + wait_for=wait_for, + wait_timeout_seconds=wait_timeout_seconds, + result_type=result_type, + ) + ) + + def update_outcome_receipt( + self, + update_id: str, + *, + wait_for: str = "accepted", + wait_timeout_seconds: int | None = None, + ) -> dict[str, Any]: + result: dict[str, Any] = self._runner.run( + self._handle.update_outcome_receipt( + update_id, + wait_for=wait_for, + wait_timeout_seconds=wait_timeout_seconds, + ) + ) + return result + class SyncScheduleHandle: """Blocking wrapper around an async schedule handle.""" @@ -592,6 +625,43 @@ def update_workflow( ) ) + def update_outcome( + self, + workflow_id: str, + update_id: str, + *, + wait_for: str = "completed", + wait_timeout_seconds: int | None = None, + result_type: Any = None, + ) -> Any: + return self._runner.run( + self._async.update_outcome( + workflow_id, + update_id, + wait_for=wait_for, + wait_timeout_seconds=wait_timeout_seconds, + result_type=result_type, + ) + ) + + def update_outcome_receipt( + self, + workflow_id: str, + update_id: str, + *, + wait_for: str = "accepted", + wait_timeout_seconds: int | None = None, + ) -> dict[str, Any]: + result: dict[str, Any] = self._runner.run( + self._async.update_outcome_receipt( + workflow_id, + update_id, + wait_for=wait_for, + wait_timeout_seconds=wait_timeout_seconds, + ) + ) + return result + def get_result( self, handle: SyncWorkflowHandle, diff --git a/tests/test_metric_routes.py b/tests/test_metric_routes.py index ac5b2944..f9d9fc5a 100644 --- a/tests/test_metric_routes.py +++ b/tests/test_metric_routes.py @@ -14,6 +14,7 @@ ROUTES = [ ("/workflows/wf-{id}/signal/signal-{id}", "/workflows/{workflow_id}/signal/{name}"), ("/workflows/wf-{id}/runs/run-{id}/history/export", "/workflows/{workflow_id}/runs/{run_id}/history/export"), + ("/workflows/wf-{id}/updates/update-{id}", "/workflows/{workflow_id}/updates/{update_id}"), ("/worker/workflow-tasks/task-{id}/complete", "/worker/workflow-tasks/{task_id}/complete"), ("/worker/activity-tasks/task-{id}/heartbeat", "/worker/activity-tasks/{task_id}/heartbeat"), ("/worker/query-tasks/task-{id}/status", "/worker/query-tasks/{task_id}/status"), diff --git a/tests/test_update_outcome.py b/tests/test_update_outcome.py new file mode 100644 index 00000000..fcedc8e1 --- /dev/null +++ b/tests/test_update_outcome.py @@ -0,0 +1,328 @@ +"""Read an accepted update's outcome later by update id (durable-workflow/.github#212, Server #1131). + +A read is a GET that never admits another update. A pending update raises +``UpdateStillPending`` and is never returned as ``None``; a completed update +returns its decoded result, including ``None``. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from typing import Any +from urllib.parse import parse_qs, urlsplit + +import httpx +import pytest + +from durable_workflow import serializer +from durable_workflow.client import Client, WorkflowHandle +from durable_workflow.errors import ( + RuntimeCapabilityUnsupported, + TypedValueError, + UpdateFailed, + UpdateNotFound, + UpdateRejected, + UpdateStillPending, +) +from durable_workflow.metrics import CLIENT_REQUESTS, InMemoryMetrics +from durable_workflow.retry_policy import TransportRetryPolicy +from durable_workflow.sync import Client as SyncClient +from durable_workflow.sync import SyncWorkflowHandle + +CLUSTER = {"version": "3.6.0", "capabilities": {"update_outcome_reads": True}} + + +def _envelope(value: Any) -> dict[str, str]: + return serializer.envelope(value, codec="avro") + + +def _receipt(**fields: Any) -> dict[str, Any]: + receipt: dict[str, Any] = { + "workflow_id": "order 1", "run_id": "run-9", "update_id": "upd-1", "update_name": "rename", + "request_id": "rename-1", "update_status": "completed", "result": None, "result_envelope": None, + "wait_timed_out": False, + } + receipt.update(fields) + return receipt + + +def _path(request: httpx.Request) -> str: + """The request path as sent, still percent-encoded.""" + return request.url.raw_path.split(b"?", 1)[0].decode() + + +class Server: + """Answers queued (status, body) pairs and records every request.""" + + def __init__(self, *responses: tuple[int, Any]) -> None: + self.responses = list(responses) + self.requests: list[httpx.Request] = [] + + def __call__(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + status, body = self.responses.pop(0) + return httpx.Response(status, json=body) + + @property + def reads(self) -> list[httpx.Request]: + return [request for request in self.requests if request.url.path != "/api/cluster/info"] + + +@pytest.fixture +async def connect() -> AsyncIterator[Any]: + clients: list[Client] = [] + + async def make(server: Server, *, metrics: InMemoryMetrics | None = None) -> Client: + client = Client( + "http://localhost:8080", token="test-token", namespace="ns1", metrics=metrics, + retry_policy=TransportRetryPolicy(max_attempts=1), + ) + await client._http.aclose() + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(server)) + client._runtime_external_payload_transport_resolved = True + clients.append(client) + return client + + yield make + for client in clients: + await client.aclose() + + +async def test_completed_reads_return_the_decoded_result_including_none(connect: Any) -> None: + server = Server( + (200, CLUSTER), + (200, _receipt(result_envelope=_envelope("renamed"))), + (200, _receipt(update_id="upd-2", result_envelope=_envelope(None))), + ) + client = await connect(server) + + assert await client.update_outcome("order 1", "upd-1", wait_timeout_seconds=10) == "renamed" + assert await client.update_outcome("order 1", "upd-2") is None + + # Discovery runs once; each read is a body-less GET on the update, never a POST. + assert [_path(request) for request in server.requests] == [ + "/api/cluster/info", "/api/workflows/order%201/updates/upd-1", "/api/workflows/order%201/updates/upd-2", + ] + assert all(request.method == "GET" and request.content == b"" for request in server.reads) + assert parse_qs(server.reads[0].url.query.decode()) == {"wait_for": ["completed"], "wait_timeout_seconds": ["10"]} + assert parse_qs(server.reads[1].url.query.decode()) == {"wait_for": ["completed"]} + + +async def test_a_result_type_decodes_the_completed_result(connect: Any) -> None: + server = Server((200, CLUSTER), (200, _receipt(result_envelope=_envelope(7))), (200, _receipt( + result_envelope=_envelope("seven"), + ))) + client = await connect(server) + + assert await client.update_outcome("order 1", "upd-1", result_type=int) == 7 + with pytest.raises(TypedValueError): + await client.update_outcome("order 1", "upd-1", result_type=int) + + +async def test_a_pending_read_raises_and_points_at_reading_again(connect: Any) -> None: + pending = _receipt(update_status="accepted", wait_timed_out=True) + server = Server((200, CLUSTER), (202, pending)) + client = await connect(server) + + with pytest.raises(UpdateStillPending) as excinfo: + await client.update_outcome("order 1", "upd-1", wait_timeout_seconds=1) + + error = excinfo.value + assert (error.workflow_id, error.run_id, error.update_name, error.update_id, error.request_id) == ( + "order 1", "run-9", "rename", "upd-1", "rename-1", + ) + assert error.update_status == "accepted" + assert error.wait_timed_out is True + assert error.body is not None and error.body["update_status"] == "accepted" + assert "update_outcome() again with update id 'upd-1'" in str(error) + assert "never admits another update" in str(error) + + +async def test_a_plain_pending_read_is_not_a_none_result(connect: Any) -> None: + server = Server((200, CLUSTER), (202, _receipt(update_status="accepted"))) + client = await connect(server) + + with pytest.raises(UpdateStillPending) as excinfo: + await client.update_outcome("order 1", "upd-1", wait_for="accepted") + + assert excinfo.value.wait_timed_out is False + assert server.reads[0].url.query == b"" + + +async def test_reading_again_after_pending_returns_the_result(connect: Any) -> None: + server = Server( + (200, CLUSTER), + (202, _receipt(update_status="accepted", wait_timed_out=True)), + (200, _receipt(result_envelope=_envelope({"ok": True}))), + ) + client = await connect(server) + + with pytest.raises(UpdateStillPending) as excinfo: + await client.update_outcome("order 1", "upd-1") + assert await client.update_outcome("order 1", excinfo.value.update_id) == {"ok": True} + assert len(server.reads) == 2 + assert all(request.method == "GET" for request in server.requests) + + +async def test_an_accepted_receipt_read_returns_the_receipt_without_waiting(connect: Any) -> None: + pending = _receipt(update_status="accepted") + server = Server((200, CLUSTER), (202, pending)) + client = await connect(server) + + assert await client.update_outcome_receipt("order 1", "upd-1") == pending + assert server.reads[0].method == "GET" + assert _path(server.reads[0]) == "/api/workflows/order%201/updates/upd-1" + assert server.reads[0].url.query == b"" + + +async def test_a_completed_receipt_carries_the_decoded_result(connect: Any) -> None: + server = Server((200, CLUSTER), (200, _receipt(result_envelope=_envelope(None)))) + client = await connect(server) + + receipt = await client.update_outcome_receipt("order 1", "upd-1", wait_for="completed", wait_timeout_seconds=5) + + assert receipt["update_status"] == "completed" + assert receipt["result"] is None + assert parse_qs(urlsplit(str(server.reads[0].url)).query) == { + "wait_for": ["completed"], "wait_timeout_seconds": ["5"], + } + + +async def test_a_failed_update_raises_update_failed(connect: Any) -> None: + body = _receipt(update_status="failed", failure_id="failure-1", failure_message="inventory unavailable") + server = Server((200, CLUSTER), (422, body)) + client = await connect(server) + + with pytest.raises(UpdateFailed) as excinfo: + await client.update_outcome("order 1", "upd-1") + + error = excinfo.value + assert not isinstance(error, UpdateNotFound) + assert str(error) == "inventory unavailable" + assert (error.status, error.update_id, error.run_id, error.failure_id) == (422, "upd-1", "run-9", "failure-1") + + +async def test_an_unknown_or_pruned_update_raises_update_not_found(connect: Any) -> None: + body = {"reason": "update_not_found", "message": "Update [upd-1] is not retained for workflow [workflow-1].", + "workflow_id": "workflow-1", "update_id": "upd-1"} + server = Server((200, CLUSTER), (404, body)) + client = await connect(server) + + # The workflow id contains "workflow"; the read must not be mistaken for WorkflowNotFound. + with pytest.raises(UpdateNotFound) as excinfo: + await client.update_outcome("workflow-1", "upd-1") + + error = excinfo.value + assert isinstance(error, UpdateFailed) + assert (error.status, error.reason, error.update_id) == (404, "update_not_found", "upd-1") + assert str(error) == body["message"] + + +@pytest.mark.parametrize( + "status,reason", + [(409, "structural_limit_exceeded"), (404, "unknown_update"), (422, "invalid_arguments")], +) +async def test_a_rejected_update_raises_update_rejected(connect: Any, status: int, reason: str) -> None: + body = _receipt(update_status="rejected", reason=reason, rejection_reason=reason, accepted=False) + server = Server((200, CLUSTER), (status, body)) + client = await connect(server) + + with pytest.raises(UpdateRejected) as excinfo: + await client.update_outcome("order 1", "upd-1") + + assert excinfo.value.body == body + assert reason in str(excinfo.value) + + +@pytest.mark.parametrize( + "cluster", + [ + {"version": "3.5.0", "capabilities": {}}, + {"version": "3.5.0", "capabilities": {"update_outcome_reads": False}}, + {"version": "2.4.0"}, + ], +) +@pytest.mark.parametrize("method", ["update_outcome", "update_outcome_receipt"]) +async def test_a_server_without_the_capability_is_refused_before_any_read( + connect: Any, cluster: dict[str, Any], method: str, +) -> None: + server = Server((200, cluster)) + client = await connect(server) + + with pytest.raises(RuntimeCapabilityUnsupported) as excinfo: + await getattr(client, method)("order 1", "upd-1") + + assert excinfo.value.capability == "capabilities.update_outcome_reads" + assert excinfo.value.operation == f"Client.{method}" + assert "update_outcome_reads" in str(excinfo.value) + # Only discovery ran: nothing was read, and nothing was posted or resubmitted. + assert [(request.method, request.url.path) for request in server.requests] == [("GET", "/api/cluster/info")] + + +@pytest.mark.parametrize("wait_for", ["validated", "", "COMPLETED"]) +async def test_an_unknown_wait_stage_is_refused_before_any_request(connect: Any, wait_for: str) -> None: + server = Server() + client = await connect(server) + + with pytest.raises(ValueError, match="wait_for"): + await client.update_outcome("order 1", "upd-1", wait_for=wait_for) + with pytest.raises(ValueError, match="wait_for"): + await client.update_outcome_receipt("order 1", "upd-1", wait_for=wait_for) + assert server.requests == [] + + +@pytest.mark.parametrize("update_id", ["", " "]) +async def test_an_empty_update_id_is_refused_before_any_request(connect: Any, update_id: str) -> None: + server = Server() + client = await connect(server) + + with pytest.raises(ValueError, match="update id"): + await client.update_outcome("order 1", update_id) + assert server.requests == [] + + +async def test_reads_use_a_bounded_metrics_route(connect: Any) -> None: + metrics = InMemoryMetrics() + server = Server((200, CLUSTER), (200, _receipt(result_envelope=_envelope(1)))) + client = await connect(server, metrics=metrics) + + await client.update_outcome("order 1", "upd-1") + + tags = {"method": "GET", "route": "/workflows/{workflow_id}/updates/{update_id}", "plane": "control", + "status_code": "200", "outcome": "ok"} + assert metrics.counter_value(CLIENT_REQUESTS, tags) == 1 + + +async def test_handle_reads_forward_to_the_client(connect: Any) -> None: + server = Server( + (200, CLUSTER), + (200, _receipt(result_envelope=_envelope("renamed"))), + (202, _receipt(update_status="accepted")), + ) + client = await connect(server) + handle = WorkflowHandle(client, workflow_id="order 1", run_id="run-1") + + assert await handle.update_outcome("upd-1", wait_timeout_seconds=3) == "renamed" + assert (await handle.update_outcome_receipt("upd-1"))["update_status"] == "accepted" + assert [_path(request) for request in server.reads] == ["/api/workflows/order%201/updates/upd-1"] * 2 + + +def test_sync_client_and_handle_read_outcomes() -> None: + server = Server( + (200, CLUSTER), + (200, _receipt(result_envelope=_envelope(None))), + (202, _receipt(update_status="accepted", wait_timed_out=True)), + (202, _receipt(update_status="accepted")), + ) + with SyncClient("http://localhost:8080", token="test-token") as client: + client._async._http = httpx.AsyncClient( + base_url=client._async.base_url, transport=httpx.MockTransport(server), + ) + client._async._runtime_external_payload_transport_resolved = True + assert client.update_outcome("order 1", "upd-1") is None + handle = SyncWorkflowHandle(WorkflowHandle(client._async, workflow_id="order 1"), client._runner) + with pytest.raises(UpdateStillPending): + handle.update_outcome("upd-1", wait_timeout_seconds=1) + assert client.update_outcome_receipt("order 1", "upd-1")["update_status"] == "accepted" + + assert [request.method for request in server.requests] == ["GET"] * 4 From d655753843c20ace0263cb96ba2fa4d9834d2cc7 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 05:20:49 +0000 Subject: [PATCH 2/2] Check update outcome reads against the request (#212) Refuse an empty, "." or ".." workflow or update id (and an invalid wait_timeout_seconds) before any request, so ".." cannot resolve onto the workflow describe route. A 2xx answer must name the requested update_id, carry a string update_status and match an echoed workflow_id; update_outcome() returns a value only from a completed receipt. Anything else raises the new UnrecognizedUpdateOutcome. A wait_timeout_seconds read extends the HTTP timeout to the wait plus 30 s when that exceeds the client timeout. Part of durable-workflow/.github#212. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 2 + docs/sdk-reference.md | 11 ++++ src/durable_workflow/__init__.py | 2 + src/durable_workflow/client.py | 71 +++++++++++++++++++----- src/durable_workflow/errors.py | 18 ++++++ tests/test_update_outcome.py | 95 +++++++++++++++++++++++++++++++- 6 files changed, 183 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3680af42..e7271844 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `Client.update_outcome()` and `Client.update_outcome_receipt()` read an accepted update's outcome later by its update id with `GET /api/workflows/{workflow_id}/updates/{update_id}`, and never submit the update again (durable-workflow/.github#212, server#1131). `update_outcome()` waits for the outcome by default and returns the decoded result, including `None`, with optional `result_type` decoding. A still-pending update raises the new `UpdateStillPending`, which carries `workflow_id`, `run_id`, `update_name`, `update_id`, `request_id` and `update_status` and tells the caller to read again by update id; it is never returned as `None`. A failed update raises `UpdateFailed`, a rejected one `UpdateRejected`, and an unknown or pruned id the new `UpdateNotFound` (a subclass of `UpdateFailed`). `update_outcome_receipt()` returns the raw receipt and answers at once by default. Both need a Server that advertises `capabilities.update_outcome_reads` (native 3.6.0 and later); on an older Server they raise `RuntimeCapabilityUnsupported` after discovery, without reading or resubmitting anything. `WorkflowHandle` and the synchronous client and handle have the same methods. +- Update outcome reads check the request and answer. An empty, `.` or `..` workflow or update id, an unknown `wait_for` or a non-positive `wait_timeout_seconds` raises `ValueError` before any request, including discovery, so `..` can never resolve onto the workflow's describe route. A successful answer must name the requested `update_id`, carry a string `update_status` and, when it echoes a `workflow_id`, match it. `update_outcome()` returns a value only for a completed receipt. Any other answer raises the new `UnrecognizedUpdateOutcome` with `workflow_id`, `update_id` and `body`, and never returns a value. With `wait_timeout_seconds`, the HTTP timeout becomes that wait plus 30 seconds whenever that is longer than the client's `timeout`, so a long Server-side wait is not cut off. + ### Changed - The `Client.update_workflow()` and `WorkflowHandle.update()` docstrings now say what they always did: they return the Server receipt, not the decoded result, and the Server answers at `wait_for="accepted"` when it is omitted. A receipt with `update_status: "accepted"` is a pending update even though its `result` is `None`. diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index 6d842fc0..470be2ad 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -654,6 +654,17 @@ returned as `None`. A failed update raises `UpdateFailed`, a rejected one unless you pass `wait_for="completed"`. The read follows the update across continue-as-new: the receipt's `run_id` names the run that owns it now. +Both methods refuse an empty, `.` or `..` workflow or update id, an unknown +`wait_for` and a non-positive `wait_timeout_seconds` with `ValueError` before +sending anything. They accept an answer only when it names the requested +`update_id`, carries a string `update_status` and, if it echoes a +`workflow_id`, names the requested workflow. `update_outcome()` returns a value +only from a completed receipt. Any other answer raises `UnrecognizedUpdateOutcome` +(with `workflow_id`, `update_id` and `body`) instead of a value. When you pass +`wait_timeout_seconds`, the HTTP timeout for that read becomes the wait plus 30 +seconds if that is longer than the client's `timeout`, so the client does not +cut off a long Server-side wait. + Both need a Server that advertises `capabilities.update_outcome_reads: true` in `GET /api/cluster/info` (native Server 3.6.0 and later). On an older Server they raise `RuntimeCapabilityUnsupported` after discovery, without reading or diff --git a/src/durable_workflow/__init__.py b/src/durable_workflow/__init__.py index 81e350a0..236324dc 100644 --- a/src/durable_workflow/__init__.py +++ b/src/durable_workflow/__init__.py @@ -104,6 +104,7 @@ SignalFailed, TypedValueError, Unauthorized, + UnrecognizedUpdateOutcome, UpdateFailed, UpdateNotFound, UpdateRejected, @@ -413,6 +414,7 @@ "SignalFailed", "TransportRetryPolicy", "Unauthorized", + "UnrecognizedUpdateOutcome", "UpdateFailed", "UpdateNotFound", "UpdateRejected", diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index 95705b5b..6ff148a5 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -74,6 +74,7 @@ RuntimeDiscoveryUnavailable, ServerError, SignalFailed, + UnrecognizedUpdateOutcome, UpdateFailed, UpdateRejected, UpdateStillPending, @@ -126,6 +127,9 @@ _COOPERATIVE_CANCELLATION_DISCOVERY_PATH = "worker_protocol.server_capabilities.cooperative_cancellation" _UPDATE_OUTCOME_READS_DISCOVERY_PATH = "capabilities.update_outcome_reads" _UPDATE_OUTCOME_WAIT_STAGES = ("accepted", "completed") +# Client-side headroom beyond a requested completion wait, so the HTTP timeout +# never cuts off a Server that is still legitimately waiting. +_UPDATE_OUTCOME_WAIT_HEADROOM_SECONDS = 30.0 _UPDATE_WAIT_STAGES_DISCOVERY_PATH = ( "control_plane.request_contract.operations.update.fields.wait_for.canonical_values" ) @@ -317,9 +321,7 @@ def _update_outcome_value( A pending receipt raises :class:`UpdateStillPending`, never ``None``. """ status = receipt.get("update_status") - if isinstance(receipt.get("result_envelope"), dict) or ( - status == "completed" and receipt.get("wait_timed_out") is not True - ): + if isinstance(receipt.get("result_envelope"), dict) or status == "completed": value = receipt.get("result") return value if result_type is None else decode_payload_value(value, result_type, "update result") if status == "failed": @@ -332,6 +334,15 @@ def _update_outcome_value( if status == "rejected": raise UpdateRejected("update rejected", body=receipt) + if status != "accepted": + raise UnrecognizedUpdateOutcome( + f"update {update_id} on workflow {workflow_id} answered with update_status {status!r}, " + "which is not completed, accepted, failed or rejected; no result was returned", + workflow_id=workflow_id, + update_id=update_id, + body=receipt, + ) + def text(name: str) -> str | None: value = receipt.get(name) return value if isinstance(value, str) and value else None @@ -5157,7 +5168,16 @@ async def update_outcome( :class:`~durable_workflow.errors.UpdateFailed`, a rejected one :class:`~durable_workflow.errors.UpdateRejected`, and an id with no retained update (including one pruned with its run) - :class:`~durable_workflow.errors.UpdateNotFound`. + :class:`~durable_workflow.errors.UpdateNotFound`. A value is returned + only from a completed receipt for the requested update; an answer that + names another update or workflow, or has any other status, raises + :class:`~durable_workflow.errors.UnrecognizedUpdateOutcome`. + + An empty, ``.`` or ``..`` workflow or update id, an unknown + ``wait_for`` or a non-positive ``wait_timeout_seconds`` raises + :class:`ValueError` before any request. With ``wait_timeout_seconds`` + the HTTP timeout becomes that wait plus 30 seconds when that exceeds + the client's ``timeout``, so a long wait is not cut off. The read follows the update across continue-as-new; the receipt's ``run_id`` names the run that owns it now. It needs a Server that @@ -5190,8 +5210,9 @@ async def update_outcome_receipt( whether a completion wait ran out) and ``"completed"`` once it has a ``result``, which may be ``None``. - Failed, rejected and unknown updates raise as in - :meth:`update_outcome`. Without the Server's + Failed, rejected and unknown updates, invalid ids and receipts that do + not match the request raise as in :meth:`update_outcome`, and long + waits extend the HTTP timeout the same way. Without the Server's ``capabilities.update_outcome_reads`` this raises :class:`~durable_workflow.errors.RuntimeCapabilityUnsupported`; resubmitting the update is never a substitute for reading it. @@ -5210,8 +5231,18 @@ async def _read_update_outcome( ) -> dict[str, Any]: if wait_for not in _UPDATE_OUTCOME_WAIT_STAGES: raise ValueError(f"{operation} wait_for must be 'accepted' or 'completed', not {wait_for!r}") - if not isinstance(update_id, str) or not update_id.strip(): - raise ValueError(f"{operation} needs the update id") + # "." and ".." would resolve onto another route (".." reaches the workflow itself). + for name, value in (("workflow id", workflow_id), ("update id", update_id)): + if not isinstance(value, str) or not value.strip() or value in (".", ".."): + raise ValueError(f"{operation} needs a non-empty {name} other than '.' or '..', not {value!r}") + if wait_timeout_seconds is not None and ( + isinstance(wait_timeout_seconds, bool) + or not isinstance(wait_timeout_seconds, int) + or wait_timeout_seconds < 1 + ): + raise ValueError( + f"{operation} wait_timeout_seconds must be a positive integer, not {wait_timeout_seconds!r}" + ) await self._require_update_outcome_reads(operation) path = f"/workflows/{quote(workflow_id, safe='')}/updates/{quote(update_id, safe='')}" query: dict[str, Any] = {} @@ -5221,15 +5252,29 @@ async def _read_update_outcome( query["wait_timeout_seconds"] = wait_timeout_seconds if query: path += "?" + urlencode(query) + timeout = None + if wait_timeout_seconds is not None: + timeout = max(self.timeout, wait_timeout_seconds + _UPDATE_OUTCOME_WAIT_HEADROOM_SECONDS) data = await self._request( - "GET", path, context=workflow_id, raise_for_status=_raise_for_update_outcome_read, + "GET", path, context=workflow_id, timeout=timeout, raise_for_status=_raise_for_update_outcome_read, ) + problem = None if not isinstance(data, dict): - raise ServerError( - 200, - {"reason": "invalid_update_receipt", "message": f"expected JSON object, got {type(data).__name__}"}, + problem = f"is not a JSON object but {type(data).__name__}" + elif data.get("update_id") != update_id: + problem = f"names update {data.get('update_id')!r}" + elif not isinstance(data.get("update_status"), str): + problem = "carries no string update_status" + elif data.get("workflow_id") is not None and data.get("workflow_id") != workflow_id: + problem = f"names workflow {data.get('workflow_id')!r}" + if problem is not None: + raise UnrecognizedUpdateOutcome( + f"{operation} asked for update {update_id} on workflow {workflow_id}, but the answer {problem}", + workflow_id=workflow_id, + update_id=update_id, + body=data, ) - return data + return cast(dict[str, Any], data) async def get_result( self, diff --git a/src/durable_workflow/errors.py b/src/durable_workflow/errors.py index 67f77078..4fb12e24 100644 --- a/src/durable_workflow/errors.py +++ b/src/durable_workflow/errors.py @@ -685,6 +685,24 @@ def __init__( self.body = body +class UnrecognizedUpdateOutcome(DurableWorkflowError): + """An update outcome read answered with a receipt the SDK cannot trust. + + Raised by :meth:`~durable_workflow.Client.update_outcome` and + :meth:`~durable_workflow.Client.update_outcome_receipt` when a successful + answer does not name the requested update, carries no string + ``update_status``, echoes another workflow, or (for ``update_outcome``) + is neither completed, pending, failed nor rejected. No value is returned + from such an answer. :attr:`body` preserves it for diagnosis. + """ + + def __init__(self, message: str, *, workflow_id: str, update_id: str, body: object | None = None) -> None: + super().__init__(message) + self.workflow_id = workflow_id + self.update_id = update_id + self.body = body + + class UpdateRejected(DurableWorkflowError): """A workflow update was rejected by the workflow's validator.""" diff --git a/tests/test_update_outcome.py b/tests/test_update_outcome.py index fcedc8e1..30384be7 100644 --- a/tests/test_update_outcome.py +++ b/tests/test_update_outcome.py @@ -19,6 +19,7 @@ from durable_workflow.errors import ( RuntimeCapabilityUnsupported, TypedValueError, + UnrecognizedUpdateOutcome, UpdateFailed, UpdateNotFound, UpdateRejected, @@ -271,16 +272,104 @@ async def test_an_unknown_wait_stage_is_refused_before_any_request(connect: Any, assert server.requests == [] -@pytest.mark.parametrize("update_id", ["", " "]) -async def test_an_empty_update_id_is_refused_before_any_request(connect: Any, update_id: str) -> None: +@pytest.mark.parametrize("bad_id", ["", " ", ".", ".."]) +@pytest.mark.parametrize("method", ["update_outcome", "update_outcome_receipt"]) +async def test_empty_and_dot_ids_are_refused_before_any_request(connect: Any, bad_id: str, method: str) -> None: + # ".." would otherwise resolve onto /workflows/{id} and read the describe body. server = Server() client = await connect(server) with pytest.raises(ValueError, match="update id"): - await client.update_outcome("order 1", update_id) + await getattr(client, method)("order 1", bad_id) + with pytest.raises(ValueError, match="workflow id"): + await getattr(client, method)(bad_id, "upd-1") assert server.requests == [] +@pytest.mark.parametrize("wait_timeout_seconds", [0, -5, True, 1.5]) +async def test_an_invalid_wait_timeout_is_refused_before_any_request(connect: Any, wait_timeout_seconds: Any) -> None: + server = Server() + client = await connect(server) + + with pytest.raises(ValueError, match="wait_timeout_seconds"): + await client.update_outcome("order 1", "upd-1", wait_timeout_seconds=wait_timeout_seconds) + assert server.requests == [] + + +DESCRIBE_BODY = {"workflow_id": "order 1", "run_id": "run-9", "status": "running", "workflow_type": "orders"} + + +@pytest.mark.parametrize( + "body", + [ + pytest.param(DESCRIBE_BODY, id="describe-like"), + pytest.param(_receipt(update_id="upd-other", result_envelope=_envelope("x")), id="other-update"), + pytest.param(_receipt(workflow_id="order 2", result_envelope=_envelope("x")), id="other-workflow"), + pytest.param({**_receipt(result_envelope=_envelope("x")), "update_status": None}, id="no-status"), + pytest.param(["not", "a", "receipt"], id="not-an-object"), + ], +) +@pytest.mark.parametrize("method", ["update_outcome", "update_outcome_receipt"]) +async def test_a_receipt_that_does_not_match_the_request_is_refused(connect: Any, body: Any, method: str) -> None: + server = Server((200, CLUSTER), (200, body)) + client = await connect(server) + + with pytest.raises(UnrecognizedUpdateOutcome) as excinfo: + await getattr(client, method)("order 1", "upd-1") + + assert (excinfo.value.workflow_id, excinfo.value.update_id) == ("order 1", "upd-1") + received = excinfo.value.body + if isinstance(body, dict): + # The client decodes result_envelope into result before checking the receipt. + assert isinstance(received, dict) + received = {key: value for key, value in received.items() if key != "result"} + body = {key: value for key, value in body.items() if key != "result"} + assert received == body + + +async def test_a_receipt_without_a_workflow_echo_is_accepted(connect: Any) -> None: + body = _receipt(result_envelope=_envelope(3)) + del body["workflow_id"] + server = Server((200, CLUSTER), (200, body)) + client = await connect(server) + + assert await client.update_outcome("order 1", "upd-1") == 3 + + +@pytest.mark.parametrize("status", ["applied", "unknown", ""]) +async def test_a_non_completed_success_receipt_never_returns_a_value(connect: Any, status: str) -> None: + server = Server((200, CLUSTER), (200, _receipt(update_status=status, result="looks like a result"))) + client = await connect(server) + + with pytest.raises(UnrecognizedUpdateOutcome) as excinfo: + await client.update_outcome("order 1", "upd-1") + + assert repr(status) in str(excinfo.value) + + +@pytest.mark.parametrize( + "wait_timeout_seconds,client_timeout,expected", + [ + (None, 60.0, 60.0), + (5, 60.0, 60.0), + (30, 60.0, 60.0), + (60, 60.0, 90.0), + (120, 60.0, 150.0), + (10, 5.0, 40.0), + ], +) +async def test_long_waits_extend_the_http_timeout_and_short_ones_keep_it( + connect: Any, wait_timeout_seconds: int | None, client_timeout: float, expected: float, +) -> None: + server = Server((200, CLUSTER), (200, _receipt(result_envelope=_envelope(1)))) + client = await connect(server) + client.timeout = client_timeout + + await client.update_outcome("order 1", "upd-1", wait_timeout_seconds=wait_timeout_seconds) + + assert server.reads[0].extensions["timeout"]["read"] == expected + + async def test_reads_use_a_bounded_metrics_route(connect: Any) -> None: metrics = InMemoryMetrics() server = Server((200, CLUSTER), (200, _receipt(result_envelope=_envelope(1))))