diff --git a/CHANGELOG.md b/CHANGELOG.md index e727184..d716202 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index 470be2a..3d15019 100644 --- a/docs/sdk-reference.md +++ b/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 diff --git a/src/durable_workflow/__init__.py b/src/durable_workflow/__init__.py index 236324d..1736695 100644 --- a/src/durable_workflow/__init__.py +++ b/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", diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index 6ff148a..1d9aa5c 100644 --- a/src/durable_workflow/client.py +++ b/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 diff --git a/src/durable_workflow/errors.py b/src/durable_workflow/errors.py index 4fb12e2..9450902 100644 --- a/src/durable_workflow/errors.py +++ b/src/durable_workflow/errors.py @@ -301,6 +301,60 @@ def is_storage_admission_failure(self, poll_request_id: str | None = None) -> bo ) +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 @@ class AvroNotInstalledError(DurableWorkflowError, ImportError): """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) diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index 797a441..3ceb98d 100644 --- a/src/durable_workflow/worker.py +++ b/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: diff --git a/tests/fixtures/replay_regressions/completion-size-limit-fails-run-once.json b/tests/fixtures/replay_regressions/completion-size-limit-fails-run-once.json new file mode 100644 index 0000000..1e86768 --- /dev/null +++ b/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] + } + ] + } +} diff --git a/tests/test_completion_size_limit.py b/tests/test_completion_size_limit.py new file mode 100644 index 0000000..8645b88 --- /dev/null +++ b/tests/test_completion_size_limit.py @@ -0,0 +1,201 @@ +"""Typed 413 for an oversized workflow-task completion (.github#228). + +Server 3.6.0 refuses a workflow-task completion over its aggregate limit with +``413 payload_too_large`` and ``limit_kind: workflow_task_completion_bytes``. +The worker must not resend that completion or report a workflow-task failure +(which would retry the same completion or block replay). It fails the run once +with a ``structural_limit`` failure instead. +""" +from __future__ import annotations + +import json +from typing import Any +from unittest.mock import AsyncMock, patch + +import httpx +import pytest + +from durable_workflow import Client, Worker, WorkflowTaskCompletionTooLarge, serializer, workflow +from durable_workflow.errors import InvalidArgument, ServerError, _raise_for_status +from durable_workflow.retry_policy import TransportRetryPolicy +from durable_workflow.worker import _should_fail_workflow_task_after_completion_error + +LIMIT = 4096 +TYPED_413 = { + "message": ( + f"Workflow task completion exceeds the maximum allowed size of {LIMIT} bytes. Each command's " + "arguments may be up to 256 bytes; offload larger payloads or split the work across workflow tasks." + ), + "reason": "payload_too_large", + "limit": LIMIT, + "limit_kind": "workflow_task_completion_bytes", + "protocol_version": "1.0", + "server_capabilities": {}, +} + + +@workflow.defn(name="completion-fan-out-wf") +class FanOutWorkflow: + def run(self, ctx): # type: ignore[no-untyped-def] + yield [ctx.schedule_activity("act", ["x" * 2048]) for _ in range(4)] + + +def task() -> dict[str, Any]: + return { + "task_id": "task-413", + "workflow_id": "wf-413", + "run_id": "run-413", + "workflow_type": "completion-fan-out-wf", + "workflow_task_attempt": 3, + "history_events": [], + "arguments": serializer.encode([], codec="avro"), + "payload_codec": "avro", + } + + +def worker_for(responses: list[tuple[int, Any]], requests: list[httpx.Request]) -> tuple[Client, Worker]: + """A real Client over a mocked HTTP transport; completions answer in order.""" + + def reply(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/complete") or request.url.path.endswith("/fail"): + requests.append(request) + status, body = responses.pop(0) if responses else (200, {"outcome": "completed"}) + return httpx.Response(status, json=body) + if request.url.path == "/api/cluster/info": + return httpx.Response(200, json={"worker_protocol": {"server_capabilities": {}}}) + return httpx.Response(404, json={"reason": "not_found"}) + + client = Client("https://runtime.example", worker_token="synthetic-worker", namespace="default", + retry_policy=TransportRetryPolicy(max_attempts=1)) + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(reply)) + worker = Worker(client, task_queue="q", workflows=[FanOutWorkflow], activities=[], worker_id="worker-413") + return client, worker + + +def body(request: httpx.Request) -> dict[str, Any]: + decoded = json.loads(request.content) + assert isinstance(decoded, dict) + return decoded + + +def test_typed_413_maps_to_typed_non_retryable_error_with_limit_and_size() -> None: + with pytest.raises(WorkflowTaskCompletionTooLarge) as caught: + _raise_for_status(413, TYPED_413, request_bytes=9000) + error = caught.value + assert isinstance(error, ServerError) + assert (error.status, error.reason(), error.limit, error.size) == (413, "payload_too_large", LIMIT, 9000) + assert error.limit_kind == "workflow_task_completion_bytes" + assert error.non_retryable is True + assert error.server_message == TYPED_413["message"] + message = str(error) + assert message.startswith("Structural limit exceeded: workflow_task_completion_bytes:") + assert "9000 bytes" in message and f"{LIMIT} bytes" in message + assert "child workflows" in message and "continue-as-new" in message + assert _should_fail_workflow_task_after_completion_error(error) is False + + +@pytest.mark.parametrize("payload", [ + {"reason": "payload_too_large", "limit": LIMIT}, + {"reason": "payload_too_large", "limit": LIMIT, "limit_kind": "something_else"}, + {"reason": "other", "limit_kind": "workflow_task_completion_bytes"}, + "Request Entity Too Large", +]) +def test_other_413_bodies_stay_generic_server_errors(payload: Any) -> None: + with pytest.raises(ServerError) as caught: + _raise_for_status(413, payload, request_bytes=9000) + assert type(caught.value) is ServerError + + +@pytest.mark.asyncio +async def test_typed_413_fails_the_run_once_without_resending_the_completion() -> None: + requests: list[httpx.Request] = [] + client, worker = worker_for([(413, TYPED_413)], requests) + async with client: + result = await worker._run_workflow_task(task()) + + assert result is None + assert [request.url.path for request in requests] == [ + "/api/worker/workflow-tasks/task-413/complete", + "/api/worker/workflow-tasks/task-413/complete", + ] + oversized, failure = (body(request) for request in requests) + assert [command["type"] for command in oversized["commands"]] == ["schedule_activity"] * 4 + assert (failure["lease_owner"], failure["workflow_task_attempt"]) == ("worker-413", 3) + assert len(failure["commands"]) == 1 + command = failure["commands"][0] + assert command["type"] == "fail_workflow" + assert command["non_retryable"] is True + assert command["exception_type"] == "WorkflowTaskCompletionTooLarge" + assert command["message"].startswith("Structural limit exceeded: workflow_task_completion_bytes:") + assert f"is {len(requests[0].content)} bytes" in command["message"] + assert f"limit of {LIMIT} bytes" in command["message"] + assert "sticky_cache" not in failure + + +@pytest.mark.parametrize("status, payload", [ + (409, {"reason": "invalid_commands"}), + (409, {}), + (422, {"reason": "validation_failed"}), + (401, {"reason": "unauthorized"}), +]) +@pytest.mark.asyncio +async def test_rejected_structural_limit_completion_never_calls_fail(status: int, payload: Any) -> None: + requests: list[httpx.Request] = [] + client, worker = worker_for([(413, TYPED_413), (status, payload)], requests) + async with client: + result = await worker._run_workflow_task(task()) + + assert result is None + # The refused completion is never resent, and a rejected follow-up is not + # reported through /fail, which would retry into the same 413 or block + # replay on the Server. + assert [request.url.path for request in requests] == [ + "/api/worker/workflow-tasks/task-413/complete", + "/api/worker/workflow-tasks/task-413/complete", + ] + assert body(requests[1])["commands"][0]["type"] == "fail_workflow" + + +@pytest.mark.asyncio +async def test_rejected_structural_limit_completion_reraises_the_typed_error() -> None: + requests: list[httpx.Request] = [] + client, worker = worker_for([(413, TYPED_413), (422, {"reason": "validation_failed"})], requests) + async with client: + with pytest.raises(WorkflowTaskCompletionTooLarge) as caught: + await worker._complete_workflow_task_with_retry( + task_id="task-413", attempt=3, commands=[{"type": "complete_workflow"}], + ) + + assert (caught.value.limit, caught.value.size) == (LIMIT, len(requests[0].content)) + assert isinstance(caught.value.__cause__, InvalidArgument) + assert [request.url.path for request in requests] == ["/api/worker/workflow-tasks/task-413/complete"] * 2 + + +@pytest.mark.asyncio +async def test_untyped_413_keeps_the_existing_task_failure() -> None: + requests: list[httpx.Request] = [] + client, worker = worker_for([(413, {"reason": "payload_too_large", "limit": LIMIT})], requests) + async with client: + result = await worker._run_workflow_task(task()) + + assert result is None + assert [request.url.path for request in requests] == [ + "/api/worker/workflow-tasks/task-413/complete", + "/api/worker/workflow-tasks/task-413/fail", + ] + failure = body(requests[1])["failure"] + assert failure["type"] == "ServerError" + assert failure["message"].startswith("workflow task completion failed after commands were produced") + + +@pytest.mark.asyncio +async def test_5xx_completion_errors_still_retry_the_same_completion() -> None: + requests: list[httpx.Request] = [] + client, worker = worker_for([(503, {"reason": "server_busy"}), (200, {"outcome": "completed"})], requests) + async with client: + with patch("durable_workflow.worker.asyncio.sleep", new_callable=AsyncMock): + result = await worker._run_workflow_task(task()) + + assert result is not None + assert [request.url.path for request in requests] == ["/api/worker/workflow-tasks/task-413/complete"] * 2 + assert requests[0].content == requests[1].content diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index e16c59d..25a4667 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -8,6 +8,7 @@ from typing import Any from unittest.mock import AsyncMock +import httpx import pytest from durable_workflow import Replayer, Worker, activity, serializer, workflow @@ -19,6 +20,7 @@ WorkflowPayloadDecodeError, ) from durable_workflow.replay_conformance import compose_report +from durable_workflow.retry_policy import TransportRetryPolicy from durable_workflow.workflow import ( LocalActivityExecutionAborted, WorkflowContext, @@ -583,6 +585,60 @@ async def run(scenario: dict[str, Any]) -> dict[str, Any]: return [asyncio.run(run(scenario)) for scenario in inputs] +@workflow.defn(name="sdk.completion-size-limit.fan-out") +class CompletionSizeFanOutWorkflow: + def run(self, ctx, activities, argument_bytes): # type: ignore[no-untyped-def] + yield [ctx.schedule_activity("act", ["x" * argument_bytes]) for _ in range(activities)] + + +def _execute_completion_size_limit(inputs: Any) -> list[dict[str, Any]]: + """Answer a fan-out completion with Server 3.6.0's typed 413 (.github#228) and report + every workflow-task request the real worker sends afterwards.""" + assert isinstance(inputs, list) and len(inputs) == 1 and isinstance(inputs[0], dict) + scenario = inputs[0] + refusal = {"message": "Workflow task completion exceeds the maximum allowed size.", "reason": "payload_too_large", + "limit": scenario["limit"], "limit_kind": "workflow_task_completion_bytes"} + requests: list[httpx.Request] = [] + + def reply(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/cluster/info": + return httpx.Response(200, json={"worker_protocol": {"server_capabilities": {}}}) + requests.append(request) + if len(requests) == 1: + return httpx.Response(413, json=refusal) + return httpx.Response(200, json={"outcome": "completed"}) + + async def run() -> list[dict[str, Any]]: + client = Client("https://corpus.example", worker_token="corpus", namespace="default", + retry_policy=TransportRetryPolicy(max_attempts=1)) + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(reply)) + async with client: + worker = Worker(client, task_queue="queue", worker_id="holder", workflows=[CompletionSizeFanOutWorkflow]) + await worker._run_workflow_task({ + "task_id": "task", "workflow_id": "workflow", "run_id": "run", "workflow_task_attempt": 1, + "workflow_type": "sdk.completion-size-limit.fan-out", "payload_codec": "avro", + "arguments": serializer.encode([scenario["activities"], scenario["argument_bytes"]], codec="avro"), + "history_events": [{"event_type": "WorkflowStarted", "payload": {}}], + }) + observed = [] + for request in requests: + body = json.loads(request.content) + commands = body.get("commands", []) + observed.append({ + "type": "workflow_task_request", + "route": request.url.path.rsplit("/", 1)[-1], + "command_types": [command["type"] for command in commands], + "non_retryable": [command.get("non_retryable", False) for command in commands], + "structural_limit": [ + command.get("message", "").startswith("Structural limit exceeded: workflow_task_completion_bytes:") + for command in commands + ], + }) + return observed + + return asyncio.run(run()) + + def _execute_fixture(fixture: dict[str, Any]) -> list[dict[str, Any]]: assert fixture.get("fixture_schema") == FIXTURE_SCHEMA assert "python" in fixture.get("bindings", []) @@ -598,6 +654,11 @@ def _execute_fixture(fixture: dict[str, Any]) -> list[dict[str, Any]]: _assert_matches(fixture["command_sequence"], commands, f"{fixture['id']}.command_sequence") _assert_matches(fixture["expected"], {"command_sequence": commands}, f"{fixture['id']}.expected") return commands + if workflow_type == "sdk.completion-size-limit": + commands = _execute_completion_size_limit(workflow.get("input")) + _assert_matches(fixture["command_sequence"], commands, f"{fixture['id']}.command_sequence") + _assert_matches(fixture["expected"], {"command_sequence": commands}, f"{fixture['id']}.expected") + return commands if workflow_type == "sdk.local-activity-backoff": commands = [_execute_local_activity_backoff(workflow.get("input"))] _assert_matches(fixture["command_sequence"], commands, f"{fixture['id']}.command_sequence") @@ -827,7 +888,7 @@ def test_checked_in_replay_regression_corpus_uses_official_replayer( if workflow.get("type") in { "sdk.replay-conformance-report", "sdk.query-task-dispatch", "sdk.generator-query-task-dispatch", "sdk.generator-side-effect-query-task-dispatch", "sdk.observed-worker-replay", "sdk.local-activity-backoff", - "sdk.sticky-suffix", + "sdk.sticky-suffix", "sdk.completion-size-limit", }: _execute_fixture(fixture) return