Fail the run on a typed oversized-completion 413 (.github#228)

CHANGELOG.md

@@ -16,6 +16,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - 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`.
 
+### Fixed
+
+- A workflow-task completion the Server refuses as too large (Server 3.6.0's typed `413 payload_too_large` with `limit_kind: workflow_task_completion_bytes`, over 16 payload limits, 32 MiB by default) now raises `WorkflowTaskCompletionTooLarge` with the Server's `limit` and the encoded request `size` (durable-workflow/.github#228). The worker no longer reports this as a workflow-task failure, which made the Server retry the same oversized completion or block replay. It never resends the refused completion: it completes the same task once with a non-retryable `fail_workflow` command whose message starts with `Structural limit exceeded: workflow_task_completion_bytes:`, so the run fails with the `structural_limit` failure category and guidance to split the fan-out, use child workflows or continue-as-new. If that follow-up is rejected too, the worker logs it and never calls `/fail`. Other 413 responses and 5xx retries are unchanged.
+
 ## [2.16.0] - 2026-11-12
 
 ### Added

docs/sdk-reference.md

@@ -1035,6 +1035,31 @@ quiet_client = Client(
 )
 ```
 
+## Workflow-task completion size
+
+Each command's arguments must fit the server payload limit (2 MiB by
+default). Server 3.6.0 and later also bound the whole workflow-task
+completion, all of the commands one task produces, at 16 payload limits
+(32 MiB by default). A larger completion is refused with HTTP 413
+`payload_too_large` and `limit_kind: workflow_task_completion_bytes`.
+
+The SDK raises `WorkflowTaskCompletionTooLarge` for that response. It is a
+`ServerError` with `limit` (the Server's aggregate limit in bytes) and `size`
+(the encoded request body the SDK sent). Replaying the task would produce the
+same commands, so the worker never resends the refused completion and does not
+report a workflow-task failure. It completes the same task once with a
+non-retryable `fail_workflow` command whose message starts with
+`Structural limit exceeded: workflow_task_completion_bytes:`, which the Server
+records with the `structural_limit` failure category. If that small completion
+also fails, the worker logs it and stops: it resends neither completion and
+does not report a workflow-task failure, so the task is left to its lease.
+Other 413 responses keep the previous handling.
+
+To stay under the aggregate, schedule a large fan-out in bounded batches
+across workflow tasks, move groups of work into child workflows, or
+continue-as-new. Enable external payload storage so large arguments travel as
+references.
+
 ## Avro payload type boundaries
 
 The default Avro codec uses the fixed recursive

src/durable_workflow/__init__.py

@@ -116,6 +116,7 @@
     WorkflowFailed,
     WorkflowNotFound,
     WorkflowPayloadDecodeError,
+    WorkflowTaskCompletionTooLarge,
     WorkflowTerminated,
     WorkflowTimedOut,
 )
@@ -424,6 +425,7 @@
     "WorkflowCancelled",
     "WorkflowFailed",
     "WorkflowNotFound",
+    "WorkflowTaskCompletionTooLarge",
     "WorkflowTerminated",
     "WorkflowTimedOut",
     "EXTERNAL_TASK_INPUT_CONTRACT_SCHEMA",

src/durable_workflow/client.py

@@ -1877,6 +1877,14 @@ def iter_history(
         )
 
 
+def _request_body_bytes(request: httpx.Request) -> int | None:
+    """The encoded body length of a sent request, when it was buffered."""
+    try:
+        return len(request.content)
+    except httpx.RequestNotRead:
+        return None
+
+
 class Client:
     """Async HTTP client for Durable Workflow control-plane and worker APIs.
 
@@ -2255,6 +2263,13 @@ async def _do_request() -> httpx.Response:
                     body = exc.response.json()
                 except ValueError:
                     body = exc.response.text
+                # The typed completion-size refusal carries the limit; the
+                # size is the encoded request body this client sent.
+                if raise_for_status is _raise_for_status:
+                    raise_for_status(
+                        exc.response.status_code, body, context=context,
+                        request_bytes=_request_body_bytes(exc.request),
+                    )
                 raise_for_status(exc.response.status_code, body, context=context)
                 raise  # unreachable, but keeps type checker happy
 

src/durable_workflow/errors.py

@@ -301,6 +301,60 @@
         )
 
 
+WORKFLOW_TASK_COMPLETION_BYTES = "workflow_task_completion_bytes"
+
+
+class WorkflowTaskCompletionTooLarge(ServerError):
+    """The Server refused a workflow-task completion as too large (typed 413).
+
+    Server 3.6.0 and later accept a workflow-task completion of up to 16
+    payload limits (32 MiB by default) and refuse a larger one with ``413
+    payload_too_large`` and ``limit_kind: workflow_task_completion_bytes``.
+    Replaying the task produces the same commands, so resending it cannot
+    succeed: this error is not retryable. The worker fails the run with a
+    ``structural_limit`` failure instead of resending the completion.
+
+    :attr:`limit` is the Server's aggregate limit in bytes, when it reported
+    one. :attr:`size` is the encoded request body the SDK sent, in bytes.
+    """
+
+    non_retryable = True
+    limit_kind = WORKFLOW_TASK_COMPLETION_BYTES
+
+    def __init__(self, status: int, body: object, *, size: int | None = None) -> None:
+        super().__init__(status, body)
+        limit = body.get("limit") if isinstance(body, dict) else None
+        message = body.get("message") if isinstance(body, dict) else None
+        self.limit: int | None = limit if type(limit) is int else None
+        self.size = size
+        self.server_message: str | None = message if isinstance(message, str) and message else None
+        self.args = (self.structural_limit_message(),)
+
+    @staticmethod
+    def matches(status: int, body: object) -> bool:
+        """Whether a response is the typed completion-size refusal."""
+        return (
+            status == 413
+            and isinstance(body, dict)
+            and body.get("reason") == "payload_too_large"
+            and body.get("limit_kind") == WORKFLOW_TASK_COMPLETION_BYTES
+        )
+
+    def structural_limit_message(self) -> str:
+        """The failure message the worker records on the run.
+
+        It starts with ``Structural limit exceeded:``, which the Server
+        classifies as the ``structural_limit`` failure category.
+        """
+        size = f"{self.size} bytes" if self.size is not None else "larger than the limit"
+        limit = f"the server limit of {self.limit} bytes" if self.limit is not None else "the server limit"
+        return (
+            f"Structural limit exceeded: {WORKFLOW_TASK_COMPLETION_BYTES}: the workflow task completion is "
+            f"{size}, over {limit}. Split the fan-out into smaller batches, move work into child workflows, "
+            "or continue-as-new; offload large payloads with external payload storage."
+        )
+
+
 class NexusOperationFailed(DurableWorkflowError):
     """A Nexus service operation completed with a typed service failure."""
 
@@ -979,10 +1033,15 @@
     """Raised when the core ``avro`` runtime dependency is unavailable."""
 
 
-def _raise_for_status(status: int, body: object, *, context: str = "") -> None:
+def _raise_for_status(
+    status: int, body: object, *, context: str = "", request_bytes: int | None = None,
+) -> None:
     if status < 400:
         return
 
+    if WorkflowTaskCompletionTooLarge.matches(status, body):
+        raise WorkflowTaskCompletionTooLarge(status, body, size=request_bytes)
+
     reason = body.get("reason") if isinstance(body, dict) else None
     message = body.get("message", "") if isinstance(body, dict) else str(body)
     operation = _control_plane_operation(body)

src/durable_workflow/worker.py

@@ -91,6 +91,7 @@
     UpdateRejected,
     UpdateValidationFailed,
     WorkflowEntryArgumentsMismatchError,
+    WorkflowTaskCompletionTooLarge,
 )
 from .external_storage import ExternalPayloadCache, ExternalStorageDriver, store_external_payload
 from .interceptors import (
@@ -377,6 +378,10 @@ def _is_final_query_task_rejection(error: BaseException) -> bool:
 
 
 def _should_fail_workflow_task_after_completion_error(error: BaseException) -> bool:
+    # The run already failed with a structural-limit failure. A workflow-task
+    # failure would retry the same oversized completion or block replay.
+    if isinstance(error, WorkflowTaskCompletionTooLarge):
+        return False
     if isinstance(error, ExternalPayloadError):
         return False
     if isinstance(error, InvalidArgument):
@@ -2865,7 +2870,7 @@ async def _run_workflow_task_with_history(
                 )
             except Exception as e:
                 log.warning("failed to complete workflow update task %s: %s", task_id, e)
-                if _is_storage_admission_error(e):
+                if _is_storage_admission_error(e) or isinstance(e, WorkflowTaskCompletionTooLarge):
                     return None
                 if _should_fail_workflow_task_after_completion_error(e):
                     await self._report_workflow_task_after_completion_error(task_id, attempt, e)
@@ -3038,7 +3043,7 @@ def serialize_commands(source: list[Command]) -> list[dict[str, Any]]:
             )
         except Exception as e:
             log.warning("failed to complete workflow task %s: %s", task_id, e)
-            if _is_storage_admission_error(e):
+            if _is_storage_admission_error(e) or isinstance(e, WorkflowTaskCompletionTooLarge):
                 return None
             if _should_fail_workflow_task_after_completion_error(e):
                 await self._report_workflow_task_after_completion_error(task_id, attempt, e)
@@ -3054,6 +3059,71 @@ async def _complete_workflow_task_with_retry(
         message_stream_cursors: list[dict[str, Any]] | None = None,
         message_stream_waits: list[dict[str, Any]] | None = None,
         sticky_cache: dict[str, Any] | None = None,
+    ) -> None:
+        try:
+            await self._send_workflow_task_completion(
+                task_id=task_id,
+                attempt=attempt,
+                commands=commands,
+                message_stream_cursors=message_stream_cursors,
+                message_stream_waits=message_stream_waits,
+                sticky_cache=sticky_cache,
+            )
+        except WorkflowTaskCompletionTooLarge as error:
+            await self._fail_run_for_oversized_completion(task_id, attempt, error)
+            raise
+
+    async def _fail_run_for_oversized_completion(
+        self,
+        task_id: str,
+        attempt: int,
+        error: WorkflowTaskCompletionTooLarge,
+    ) -> None:
+        """Fail the run once with a structural-limit failure (.github#228).
+
+        The refused completion is never resent: a replay produces the same
+        commands. The same leased task completes with one small
+        ``fail_workflow`` command instead, which the Server records with the
+        ``structural_limit`` failure category. If that completion also fails,
+        the typed error is raised again with the follow-up's error as its
+        cause, so the caller never reports a workflow-task failure (``/fail``)
+        on this path: the Server would retry the task into the same 413 or
+        block replay.
+        """
+        message = error.structural_limit_message()
+        log.error(
+            "workflow task %s completion refused by the server: %s (limit_kind=%s, size=%s bytes, "
+            "limit=%s bytes); failing the run instead of resending it",
+            task_id,
+            message,
+            error.limit_kind,
+            error.size,
+            error.limit,
+        )
+        command = FailWorkflow(
+            message,
+            exception_type=type(error).__name__,
+            non_retryable=True,
+        ).to_server_command(self.task_queue)
+        try:
+            await self._send_workflow_task_completion(task_id=task_id, attempt=attempt, commands=[command])
+        except Exception as fail_error:
+            log.warning(
+                "failed to record the structural-limit failure for workflow task %s: %s",
+                task_id,
+                fail_error,
+            )
+            raise error from fail_error
+
+    async def _send_workflow_task_completion(
+        self,
+        *,
+        task_id: str,
+        attempt: int,
+        commands: list[dict[str, Any]],
+        message_stream_cursors: list[dict[str, Any]] | None = None,
+        message_stream_waits: list[dict[str, Any]] | None = None,
+        sticky_cache: dict[str, Any] | None = None,
     ) -> None:
         for completion_attempt in range(1, _WORKFLOW_TASK_COMPLETION_MAX_ATTEMPTS + 1):
             try:

tests/fixtures/replay_regressions/completion-size-limit-fails-run-once.json

@@ -0,0 +1,45 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "python-completion-size-limit-fails-run-once",
+  "protocol_version": "1.20",
+  "bindings": ["python"],
+  "workflow": {
+    "type": "sdk.completion-size-limit",
+    "input": [{"activities": 4, "argument_bytes": 2048, "limit": 4096}]
+  },
+  "command_sequence": [
+    {
+      "type": "workflow_task_request",
+      "route": "complete",
+      "command_types": ["schedule_activity", "schedule_activity", "schedule_activity", "schedule_activity"],
+      "non_retryable": [false, false, false, false],
+      "structural_limit": [false, false, false, false]
+    },
+    {
+      "type": "workflow_task_request",
+      "route": "complete",
+      "command_types": ["fail_workflow"],
+      "non_retryable": [true],
+      "structural_limit": [true]
+    }
+  ],
+  "expected": {
+    "command_sequence": [
+      {
+        "type": "workflow_task_request",
+        "route": "complete",
+        "command_types": ["schedule_activity", "schedule_activity", "schedule_activity", "schedule_activity"],
+        "non_retryable": [false, false, false, false],
+        "structural_limit": [false, false, false, false]
+      },
+      {
+        "type": "workflow_task_request",
+        "route": "complete",
+        "command_types": ["fail_workflow"],
+        "non_retryable": [true],
+        "structural_limit": [true]
+      }
+    ]
+  }
+}