diff --git a/CHANGELOG.md b/CHANGELOG.md index 32623b7..0cddff1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [2.11.0] - Unreleased + +### Fixed + +- Bound failure messages, stacks and diagnostic payloads before reporting them. + The complete encoded request defaults to 1 MiB, including JSON escaping and + markers, and respects lower configured or discovered limits. Preserve failure + type, retry policy and claim identity; refuse an identity that cannot fit. + ## [2.10.1] - Unreleased ### Fixed diff --git a/docs/reference/worker.md b/docs/reference/worker.md index 9e78421..acd1554 100644 --- a/docs/reference/worker.md +++ b/docs/reference/worker.md @@ -29,6 +29,48 @@ isolated worker processes and their resource limits. The separately negotiated cooperative-cancellation capability retains its supervised process execution; selecting a default thread does not enable that protocol. +## Bounded failure reports + +This behavior is available in unreleased SDK source. Published packages still +need the compatible release before these limits apply. + +Failure reports retain at most 62 KiB of UTF-8 message text, 128 KiB of stack +text and 64 KiB of serialized details and diagnostics. These budgets include +truncation markers. The complete encoded JSON request body defaults to at most +1 MiB, counting JSON escaping, claim identities, command batches and metadata. +The message budget reserves 2 KiB below the original 64 KiB ceiling for +runtime-derived summaries. +A lower discovered Server `limits.max_payload_bytes` also applies. Configure +`Client(..., failure_report_limit_bytes=...)` when a proxy or transport has a +lower request limit. This option can lower the default cap; it cannot raise it. +Query-client clones inherit it. Payload warning thresholds remain advisory. + +Oversized diagnostics produce a visible `[truncated ...]` message marker with +structured omitted-field names and original byte sizes. Activity failures also +carry this information in `failure.diagnostics.failure_report`; terminal +workflow failures use `exception.diagnostics.failure_report`. Other existing +report shapes keep the diagnostic JSON in the message marker. Unknown sizes +are `null` when optional details could not be encoded. Avro details are kept as +an intact valid envelope or omitted whole, before an unnecessary payload upload. + +The original failure type, retry/non-retryable choice, authored retry delay and +claim identity remain exact. Activity, workflow-task, query and update rejection +reports, terminal workflow/update failures and local-activity attempt reports +share the complete-request bound. A workflow-task replay failure still leaves +the run open for repair; it does not become a terminal workflow failure. + +A local activity's failure is bounded before workflow code observes it. First +execution and cold replay therefore see the same persisted message. If a later +command batch or newly discovered lower limit cannot fit that observed value, +reporting fails explicitly rather than changing a value already used by the +workflow. Prepared local activities observe the runtime's canonical history. + +If essential identity or semantic fields cannot fit even without optional +text, the client raises `FailureReportTooLarge` before sending the report. +This reporting refusal is separate from the application failure. The worker's +existing reporting-error log identifies it; shortening a task ID or replacing +the application type would change which failure the runtime records. + ## Poll failures and recovery This backoff requires Python SDK 2.10.0 or later. Earlier packages use a fixed diff --git a/pyproject.toml b/pyproject.toml index e118a7f..a00fdb5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "durable-workflow" -version = "2.10.1" +version = "2.11.0" description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server" readme = "README.md" requires-python = ">=3.10" @@ -76,8 +76,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main" durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main" [tool.durable-workflow] -product-train = "2.10.1" -registry-version = "2.10.1" +product-train = "2.11.0" +registry-version = "2.11.0" supported-server-versions = "2.8.0" worker-protocol-version = "1.20" control-plane-version = "2" diff --git a/src/durable_workflow/__init__.py b/src/durable_workflow/__init__.py index dc9bf20..10ee58e 100644 --- a/src/durable_workflow/__init__.py +++ b/src/durable_workflow/__init__.py @@ -81,6 +81,7 @@ ExternalPayloadUnauthorized, ExternalPayloadUnavailable, ExternalPayloadUnsupported, + FailureReportTooLarge, InvalidArgument, InvalidWorkflowDefinition, NamespaceNotFound, @@ -334,6 +335,7 @@ "testing", "workflow", "DurableWorkflowError", + "FailureReportTooLarge", "EXTERNAL_PAYLOAD_REFERENCE_SCHEMA", "RUNTIME_EXTERNAL_PAYLOAD_REFERENCE_SCHEMA", "AzureBlobExternalStorage", diff --git a/src/durable_workflow/_failure_reporting.py b/src/durable_workflow/_failure_reporting.py new file mode 100644 index 0000000..ae66304 --- /dev/null +++ b/src/durable_workflow/_failure_reporting.py @@ -0,0 +1,299 @@ +"""Bound diagnostic text using the same JSON encoder as the HTTP transport. + +Only failure diagnostics may be reduced. Claim identity, classification and +retry policy remain exact, even when that makes the complete report unfit. +""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +import httpx + +from .errors import FailureReportTooLarge + +MESSAGE_BYTES = 62 * 1024 +STACK_BYTES = 128 * 1024 +DETAIL_BYTES = 64 * 1024 +REQUEST_BYTES = 1024 * 1024 +UNENCODABLE_DETAILS = object() +_STACK_MARKER = "\n[truncated]" +_OPTIONAL = ("details", "diagnostics", "runtime_diagnostics", "validation_errors") +_EXCEPTION_IDENTITY = frozenset({ + "type", "class", "code", "reason", "non_retryable", "retryable", "retry_after_seconds", + "next_retry_delay", "activity_type", "activity_attempt_id", "activity_execution_id", + "failed_step_sequence", "failed_activity_execution_id", +}) + + +def encoded_json(value: Any) -> bytes: + """Use HTTPX's actual JSON settings, including escaping and NaN refusal.""" + return httpx.Request("POST", "https://failure.invalid", json=value).content + + +def _size(value: Any) -> int | None: + try: + return len(encoded_json(value)) + except (TypeError, ValueError, RecursionError): + return None + + +def _utf8(value: str) -> bytes: + return value.encode("utf-8", errors="replace") + + +def _prefix(value: str, size: int) -> str: + return _utf8(value)[:max(0, size)].decode("utf-8", errors="ignore") + + +def _marker(omitted: dict[str, int | None]) -> str: + return "\n[truncated " + encoded_json({"omitted_fields": omitted}).decode("utf-8") + "]" + + +class _ObservedLocalFailure(str): + """A local message already delivered to workflow code. + + JSON serializes this as ordinary text. Keep the in-memory distinction until + encoding so final batch fitting cannot change first-run/replay behavior. + """ + + +@dataclass +class _Report: + original: dict[str, Any] + terminal: bool = False + structured: bool = False + + def _copy(self) -> dict[str, Any]: + result = dict(self.original) + if self.terminal and isinstance(result.get("exception"), dict): + result["exception"] = dict(result["exception"]) + return result + + def prepare(self) -> dict[str, Any]: + result = self._copy() + omitted: dict[str, int | None] = {} + self._bound_metadata(result, omitted, DETAIL_BYTES) + if omitted: + self._attach_diagnostics(result, omitted) + return result + + def render(self, budget: int) -> dict[str, Any]: + result = self._copy() + if isinstance(result.get("message"), _ObservedLocalFailure): + return result + previous: dict[str, int | None] = {} + if self.structured: + self._previous_sizes(previous) + omitted = dict(previous) + message = result.get("message", "") + message_limit = min(MESSAGE_BYTES, budget) + if len(_utf8(message)) > message_limit or _prefix(message, len(_utf8(message))) != message: + omitted.setdefault("message", len(_utf8(message))) + self._bound_stack(result, omitted, min(STACK_BYTES, budget)) + self._bound_metadata(result, omitted, min(DETAIL_BYTES, budget)) + if not omitted: + return result + if self._unchanged(previous, omitted, budget): + return result + # Inserting the marker can itself shorten an otherwise small message, + # and inserting diagnostics can require omitting another optional field. + # Recompute until the finite set of omitted fields is stable. + while True: + before = dict(omitted) + self._attach_diagnostics(result, omitted) + marker = _marker(omitted) + if len(_utf8(message)) + len(_utf8(marker)) > message_limit: + omitted.setdefault("message", len(_utf8(message))) + if omitted == before: + break + retained = max(0, message_limit - len(_utf8(marker))) + result["message"] = _prefix(message, retained) + marker + return result + + @staticmethod + def _bound_stack(result: dict[str, Any], omitted: dict[str, int | None], stack_limit: int) -> None: + stack = result.get("stack_trace") + if isinstance(stack, str) and len(_utf8(stack)) > stack_limit: + omitted.setdefault("stack_trace", len(_utf8(stack))) + if stack_limit >= len(_STACK_MARKER): + result["stack_trace"] = _prefix(stack, stack_limit - len(_STACK_MARKER)) + _STACK_MARKER + else: + result.pop("stack_trace", None) + + def _unchanged( + self, previous: dict[str, int | None], omitted: dict[str, int | None], budget: int, + ) -> bool: + message = self.original.get("message", "") + stack = self.original.get("stack_trace") + return bool( + previous and previous == omitted and message.endswith(_marker(previous)) + and len(_utf8(message)) <= min(MESSAGE_BYTES, budget) + and (not isinstance(stack, str) or len(_utf8(stack)) <= min(STACK_BYTES, budget)) + ) + + def _bound_metadata(self, result: dict[str, Any], omitted: dict[str, int | None], budget: int) -> None: + fields = ("exception",) if self.terminal else _OPTIONAL + values = {key: result[key] for key in fields if key in result} + size = _size(values) + if size is not None and size <= budget: + return + for key, value in values.items(): + omitted.setdefault(key, _size(value)) + if key == "exception" and isinstance(value, dict): + result[key] = {k: v for k, v in value.items() if k in _EXCEPTION_IDENTITY} + else: + result.pop(key, None) + + def _previous_sizes(self, omitted: dict[str, int | None]) -> None: + original = self.original.get("exception", {}) if self.terminal else self.original + diagnostics = original.get("diagnostics") if isinstance(original, dict) else None + previous = diagnostics.get("failure_report") if isinstance(diagnostics, dict) else None + sizes = previous.get("omitted_fields") if isinstance(previous, dict) else None + if not isinstance(sizes, dict): + return + for key, size in sizes.items(): + if (key in {*_OPTIONAL, "exception", "message", "stack_trace"} + and (size is None or type(size) is int and size >= 0)): + omitted[key] = size + + def _attach_diagnostics(self, result: dict[str, Any], omitted: dict[str, int | None]) -> None: + if not self.structured: + return + container = result.setdefault("exception", {}) if self.terminal else result + self._previous_sizes(omitted) + original = self.original.get("exception", {}) if self.terminal else self.original + existing = original.get("diagnostics") if isinstance(original, dict) else None + diagnostics = dict(existing) if isinstance(existing, dict) and "diagnostics" not in omitted else {} + diagnostics["failure_report"] = {"omitted_fields": omitted} + container["diagnostics"] = diagnostics + optional = {key: container[key] for key in _OPTIONAL if key in container} + size = _size(container if self.terminal else optional) + if size is not None and size <= DETAIL_BYTES: + return + self._omit_optional_metadata(container, omitted, existing) + + def _omit_optional_metadata( + self, container: dict[str, Any], omitted: dict[str, int | None], existing: Any, + ) -> None: + # Count inserted diagnostics too. Omit optional fields whole so an + # Avro envelope is either valid and intact or explicitly absent. + if existing is not None: + omitted.setdefault("diagnostics", _size(existing)) + # Snapshot the keys because the loop removes entries from this dictionary. + for key in list(container): + removable = key not in _EXCEPTION_IDENTITY if self.terminal else key in _OPTIONAL + if removable and key != "diagnostics": + omitted.setdefault(key if key in _OPTIONAL else "exception", _size(container.pop(key))) + container["diagnostics"] = {"failure_report": {"omitted_fields": omitted}} + + +def _command_reports(command: dict[str, Any]) -> list[tuple[tuple[str | int, ...], _Report]]: + kind = command.get("type") + terminal = kind == "fail_workflow" + result: list[tuple[tuple[str | int, ...], _Report]] = [] + if kind in {"fail_workflow", "fail_update"} or kind == "record_local_activity" and "message" in command: + result.append(((), _Report(command, terminal=terminal, structured=terminal))) + if kind == "record_local_activity": + for index, attempt in enumerate(command.get("attempts", [])): + if isinstance(attempt, dict) and "message" in attempt: + result.append((("attempts", index), _Report(attempt))) + return result + + +def _reports(path: str, body: dict[str, Any]) -> list[tuple[tuple[str | int, ...], _Report]]: + failure = body.get("failure") + if isinstance(failure, dict) and (path.endswith(("/fail", "/reject"))): + return [(("failure",), _Report(failure, structured="/activity-tasks/" in path))] + if "/local-activities/" in path and path.endswith("/outcome"): + report = body.get("report") + if isinstance(report, dict) and "message" in report: + return [(("report",), _Report(report))] + commands = body.get("commands") + if not isinstance(commands, list) or not path.endswith("/complete"): + return [] + return [ + (("commands", index, *location), report) + for index, command in enumerate(commands) if isinstance(command, dict) + for location, report in _command_reports(command) + ] + + +def _replace_report(body: Any, location: tuple[str | int, ...], report: dict[str, Any]) -> Any: + if not location: + return report + result: Any = dict(body) if isinstance(body, dict) else list(body) + key = location[0] + result[key] = _replace_report(body[key], location[1:], report) + return result + + +def prepare_failure_request(path: str, body: Any) -> Any: + """Omit oversized diagnostic envelopes before optional payload uploads. + + Keep original message/stack text until the final fit so its original sizes + survive normal externalization of unrelated business payloads in a batch. + """ + if not path.startswith("/worker/") or not isinstance(body, dict): + return body + result = body + for location, report in _reports(path, body): + if report.structured: + result = _replace_report(result, location, report.prepare()) + return result + + +def _render_reports( + body: dict[str, Any], reports: list[tuple[tuple[str | int, ...], _Report]], budget: int, limit: int, +) -> dict[str, Any]: + result = body + for location, report in reports: + rendered = report.render(budget) + metadata = rendered.get("exception", {}) if report.terminal else { + key: rendered[key] for key in _OPTIONAL if key in rendered + } + metadata_size = _size(metadata) + if metadata_size is None or metadata_size > DETAIL_BYTES: + raise FailureReportTooLarge(limit) + result = _replace_report(result, location, rendered) + return result + + +def bound_failure_request(path: str, body: Any, limit: int = REQUEST_BYTES) -> Any: + if not path.startswith("/worker/") or not isinstance(body, dict): + return body + reports = _reports(path, body) + if not reports: + return body + limit = min(limit, REQUEST_BYTES) + budget = STACK_BYTES + while True: + result = _render_reports(body, reports, budget, limit) + size = _size(result) + if size is not None and size <= limit: + return result + if budget == 0: + raise FailureReportTooLarge(limit) + budget //= 2 + + +def normalize_local_failure( + command: dict[str, Any], *, lease_owner: str, workflow_task_attempt: int, limit: int = REQUEST_BYTES, +) -> dict[str, Any]: + """Fit local diagnostics before the workflow sees its durable failure. + + Business arguments can still be externalized by Client. Include all claim, + retry and attempt data here; the final complete request is measured again. + An already observed message is then immutable, including on a lower limit + discovered later or when another command makes the batch unfit. + """ + frame = {key: value for key, value in command.items() if key != "arguments"} + body = {"lease_owner": lease_owner, "workflow_task_attempt": workflow_task_attempt, "commands": [frame]} + normalized = bound_failure_request("/worker/workflow-tasks/local/complete", body, limit)["commands"][0] + normalized = {**command, **normalized} + if isinstance(normalized.get("message"), str): + normalized["message"] = _ObservedLocalFailure(normalized["message"]) + for attempt in normalized.get("attempts", []): + if isinstance(attempt, dict) and isinstance(attempt.get("message"), str): + attempt["message"] = _ObservedLocalFailure(attempt["message"]) + return normalized diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index 36bb07b..01d41e7 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -48,6 +48,14 @@ scope_timestamp, ) from ._cooperative_cancellation import CancellationDelivery, CancellationRequest +from ._failure_reporting import ( + DETAIL_BYTES, + REQUEST_BYTES, + UNENCODABLE_DETAILS, + bound_failure_request, + encoded_json, + prepare_failure_request, +) from ._metric_routes import route_for_metrics as _route_for_metrics from ._query_task_authority import QueryTaskAuthority, current_query_authority from .errors import ( @@ -1622,6 +1630,7 @@ def __init__( payload_size_limit_bytes: int = serializer.DEFAULT_PAYLOAD_SIZE_BYTES, payload_size_warning_threshold_percent: int = serializer.DEFAULT_WARNING_THRESHOLD_PERCENT, payload_size_warnings: bool = True, + failure_report_limit_bytes: int = REQUEST_BYTES, external_storage: ExternalStorageDriver | None = None, external_storage_threshold_bytes: int | None = None, external_storage_cache: ExternalPayloadCache | None = None, @@ -1643,6 +1652,9 @@ def __init__( self._tls_context = tls_context self.retry_policy = retry_policy or TransportRetryPolicy() self.metrics = metrics if metrics is not None else NOOP_METRICS + if type(failure_report_limit_bytes) is not int or failure_report_limit_bytes < 1: + raise ValueError("failure_report_limit_bytes must be a positive integer") + self.failure_report_limit_bytes = min(REQUEST_BYTES, failure_report_limit_bytes) self.payload_size_warning_config = ( serializer.PayloadSizeWarningConfig( limit_bytes=payload_size_limit_bytes, @@ -1843,6 +1855,14 @@ def _warn_json_payload_size( ), ) + def _failure_report_limit(self) -> int: + limit = self.failure_report_limit_bytes + limits = (self._cluster_info or {}).get("limits") + advertised = limits.get("max_payload_bytes") if isinstance(limits, dict) else None + if type(advertised) is int and advertised > 0: + limit = min(limit, advertised) + return limit + async def _request( self, method: str, @@ -1869,22 +1889,22 @@ async def _request( activity_authority = current_activity_authority.get() if worker and activity_authority is not None and activity_authority.governs(self, self.namespace, path): activity_authority.remaining() - request_json = json + request_json = prepare_failure_request(path, json) if ( not discovery and json is not None - and _contains_direct_external_storage_envelope(json) + and _contains_direct_external_storage_envelope(request_json) ): await self._require_direct_external_storage_support() if ( not discovery and json is not None - and _contains_inline_payload_envelope(json) + and _contains_inline_payload_envelope(request_json) ): transport = await self._runtime_external_payload_transport() if transport is not None: request_json = await self._externalize_runtime_payloads( - json, + request_json, worker=worker, transport=transport, uploaded={}, @@ -1894,6 +1914,9 @@ async def _request( ), ) + # Discovery/external payload references may change the known limit or + # final envelope. Measure again after those transformations. + request_json = bound_failure_request(path, request_json, self._failure_report_limit()) start = time.perf_counter() route = _route_for_metrics(path) discovery_uses_worker_token = discovery and bool(self.worker_token) @@ -6208,11 +6231,20 @@ async def fail_activity_task( if non_retryable: failure["non_retryable"] = True if details is not None: - failure["details"] = self._payload_envelope( - details, - codec=codec, - kind="activity_failure_details", - activity_name=activity_name, + # Inspect the official inline encoding before an optional storage + # driver uploads diagnostics that the failure budget will omit. + try: + inline_details: Any = serializer.envelope(details, codec=codec, size_warning=None) + except (TypeError, ValueError, RecursionError): + if codec != serializer.AVRO_CODEC: + raise + inline_details = UNENCODABLE_DETAILS + failure["details"] = ( + self._payload_envelope( + details, codec=codec, kind="activity_failure_details", activity_name=activity_name, + ) + if isinstance(inline_details, dict) and len(encoded_json(inline_details)) <= DETAIL_BYTES + else inline_details ) body: dict[str, Any] = { "activity_attempt_id": activity_attempt_id, diff --git a/src/durable_workflow/errors.py b/src/durable_workflow/errors.py index 9550403..86916cd 100644 --- a/src/durable_workflow/errors.py +++ b/src/durable_workflow/errors.py @@ -27,6 +27,20 @@ class DurableWorkflowError(Exception): """Base class for every exception raised by the SDK.""" +class FailureReportTooLarge(DurableWorkflowError, ValueError): + """Required failure identity or semantics cannot fit the reporting budget. + + This is a local reporting refusal, not the application's failure. No claim + identity, failure type or retry decision is shortened to make it fit. + """ + + def __init__(self, limit_bytes: int) -> None: + self.limit_bytes = limit_bytes + super().__init__( + f"failure report identity or semantics cannot fit the {limit_bytes}-byte encoded request budget" + ) + + class QueryTaskAuthorityLost(DurableWorkflowError): """A query delivery cannot safely load or invoke its handler.""" diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index 83f440c..6e5a837 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -44,6 +44,7 @@ from ._activity_process import CallbackFailure, CallbackInvocation, CallbackProcessLost, SupervisedCallback from ._cancellation_scope_history import CancellationScopeBudget, CancellationScopeDeliveryReceipt from ._cooperative_cancellation import CancellationRequest, read_cancellation_history +from ._failure_reporting import REQUEST_BYTES, normalize_local_failure from ._prepared_local_activity import PreparedAttempt, PreparedCancellationObserved, PreparedLocalRunner from ._query_task_authority import QueryTaskAuthority, current_query_authority from ._sticky_workflow_cache import CacheKey, StickyWorkflowCache, complete_history @@ -76,6 +77,7 @@ DuplicateRegistrationError, DurableWorkflowError, ExternalPayloadError, + FailureReportTooLarge, InvalidArgument, NonDeterministicReplayError, NonRetryableError, @@ -2328,8 +2330,22 @@ def terminal_failure( } if timeout_kind is not None: command.outcome["timeout_kind"] = timeout_kind + known_limit = getattr(self.client, "_failure_report_limit", None) + limit = known_limit() if callable(known_limit) else REQUEST_BYTES + if type(limit) is not int: + limit = REQUEST_BYTES + try: + normalized = normalize_local_failure( + command.to_server_command(self.task_queue, payload_codec=payload_codec), + lease_owner=self.worker_id, workflow_task_attempt=task["workflow_task_attempt"], limit=limit, + ) + except FailureReportTooLarge as error: + raise LocalActivityExecutionAborted( + "local failure report cannot fit before workflow observation", + ) from error + command.outcome = {key: normalized[key] for key in command.outcome} return ActivityFailed( - message, + command.outcome["message"], activity_type=command.activity_type, exception_type=exception_type, non_retryable=non_retryable, @@ -2469,18 +2485,15 @@ async def heartbeat( cancelled = isinstance(exc, ActivityCancelled) exception_type = type(exc).__name__ message = str(exc).strip() or f"local activity failed with {exception_type}" - invalid_report = len(exception_type.encode("utf-8")) > 255 try: - json.dumps({"message": message, "exception_type": exception_type}, allow_nan=False).encode("utf-8") - except (TypeError, ValueError, UnicodeError): - invalid_report = True - if invalid_report: - exception_type = "InvalidLocalActivityReport" - message = "local activity failure metadata cannot cross the HTTP JSON boundary" - timed_out = False - cancelled = False + if len(exception_type.encode("utf-8")) > 255: + raise ValueError("local activity failure type exceeds the wire limit") + except ValueError as error: + raise LocalActivityExecutionAborted( + "local activity failure type cannot cross the HTTP boundary", + ) from error non_retryable = ( - cancelled or invalid_report or isinstance(exc, NonRetryableError) + cancelled or isinstance(exc, NonRetryableError) or exception_type in non_retryable_types ) retry = not non_retryable and attempt_number < max_attempts @@ -4159,6 +4172,7 @@ def _clone_client_for_query_tasks(self) -> Client: else serializer.DEFAULT_WARNING_THRESHOLD_PERCENT ), payload_size_warnings=warning_config is not None, + failure_report_limit_bytes=self.client.failure_report_limit_bytes, external_storage=self.external_storage, external_storage_threshold_bytes=self.external_storage_threshold_bytes, external_storage_cache=self.external_storage_cache, diff --git a/tests/failure_reporting_cases.py b/tests/failure_reporting_cases.py new file mode 100644 index 0000000..020406d --- /dev/null +++ b/tests/failure_reporting_cases.py @@ -0,0 +1,33 @@ +"""Workflows shared by bounded-failure replay and published-runtime controls.""" +from __future__ import annotations + +import hashlib +from typing import Any + +from durable_workflow import ActivityFailed, NonRetryableError, activity, workflow + + +@workflow.defn(name="tests.local.failure-observation") +class LocalFailureObservation: + def run(self, ctx: Any, non_retryable: bool = False) -> Any: + try: + yield ctx.local_activity("tests.local.large-failure", [non_retryable], retry_policy={"max_attempts": 1}) + except ActivityFailed as error: + message = str(error) + return { + "bytes": len(message.encode()), "digest": hashlib.sha256(message.encode()).hexdigest(), + "marked": "[truncated" in message, "type": error.exception_type, + "non_retryable": error.non_retryable, + } + raise AssertionError("local activity unexpectedly succeeded") + + +class OriginalFailure(Exception): + pass + + +@activity.defn(name="tests.local.large-failure") +async def large_local_failure(non_retryable: bool) -> None: + error_type = NonRetryableError if non_retryable else OriginalFailure + raise error_type("šŸŒˆ" * (3 * 1024 * 1024 // 4)) + diff --git a/tests/fixtures/replay_regressions/local-failure-first-and-cold-observation.json b/tests/fixtures/replay_regressions/local-failure-first-and-cold-observation.json new file mode 100644 index 0000000..bf60027 --- /dev/null +++ b/tests/fixtures/replay_regressions/local-failure-first-and-cold-observation.json @@ -0,0 +1,57 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "local-failure-first-and-cold-observation", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "sdk.local-failure-observation", + "input": [ + { + "limit": 3500, + "non_retryable": false + } + ], + "payload_codec": "avro" + }, + "expected": { + "command_sequence": [ + { + "type": "local_failure_observation", + "first_and_cold": { + "bytes": 1023, + "digest": "8dbf34111991fa7eefa92751fe8f096bb4aebbb332857f55652040ace0ca3fe9", + "marked": true, + "type": "OriginalFailure", + "non_retryable": false + } + } + ] + }, + "provenance": { + "fixture_form": "Input reproducer and expected observation derived from fresh published-runtime qualification. The runner exercises first execution, captured JSON report and cold replay. Not a raw history export.", + "original_message_utf8_bytes": 3145728, + "server_version": "2.11.1", + "core_version": "2.10.2", + "mysql_version": "8.0.46", + "image_digest": "sha256:bf8bed13f5f809ee3425233180a581353eac4683adbee5c4ecbf63bb18bbd102", + "workflow_id": "py-bounded-local-43dd91f310e9", + "run_id": "01m5mtkcdn779b6qqbk89fnswm", + "history_json_sha256": "3ace2024928c5aef8dc2267b8f0fe5f195f7c428d1778ee6e8cb60cf36e2fa40", + "observation_sha256": "8ddccec720b28fc449172d4b6649b78f5ca7a8b07f738a86b61aa1e83b79bab3" + }, + "command_sequence": [ + { + "type": "local_failure_observation", + "first_and_cold": { + "bytes": 1023, + "digest": "8dbf34111991fa7eefa92751fe8f096bb4aebbb332857f55652040ace0ca3fe9", + "marked": true, + "type": "OriginalFailure", + "non_retryable": false + } + } + ] +} diff --git a/tests/integration/test_failure_reporting.py b/tests/integration/test_failure_reporting.py new file mode 100644 index 0000000..893fc7a --- /dev/null +++ b/tests/integration/test_failure_reporting.py @@ -0,0 +1,267 @@ +"""Deliver oversized failures through an unmodified Server and durable replay.""" +from __future__ import annotations + +import asyncio +import json +import os +import uuid +from typing import Any + +import httpx +import pytest + +from durable_workflow import ActivityFailed, Client, workflow +from durable_workflow.workflow import replay + +from .manual_worker import MANUAL_WORKER_CAPABILITY_MANIFEST + + +@workflow.defn(name="tests.bounded-failure-consumer") +class FailureConsumer: + def run(self, ctx: Any) -> Any: + try: + yield ctx.schedule_activity("tests.bounded-failure", [], retry_policy={"max_attempts": 1}) + except ActivityFailed as error: + return { + "type": error.exception_type, "class": error.exception_class, + "message": str(error), "non_retryable": error.non_retryable, "code": error.code, + "diagnostics": (error.exception_payload or {}).get("diagnostics"), + } + raise AssertionError("activity unexpectedly succeeded") + + +def _client(server_url: str, server_token: str) -> Client: + return Client( + server_url, worker_token=server_token, + control_token=os.environ.get("DURABLE_WORKFLOW_CONTROL_TOKEN", server_token), timeout=5, + ) + + +async def _claim(client: Client, identity: str) -> tuple[Any, dict[str, Any]]: + await client.register_worker( + worker_id=identity, task_queue=identity, + supported_workflow_types=["tests.bounded-failure-consumer"], + supported_activity_types=["tests.bounded-failure"], + capability_manifest=MANUAL_WORKER_CAPABILITY_MANIFEST, + ) + handle = await client.start_workflow( + workflow_type="tests.bounded-failure-consumer", task_queue=identity, workflow_id=identity, input=[], + ) + task = await client.poll_workflow_task(worker_id=identity, task_queue=identity, timeout=2) + assert isinstance(task, dict) + return handle, task + + +@pytest.mark.parametrize("scenario", ["message", "stack", "combined", "small"]) +async def test_failure_reaches_original_workflow( + server_url: str, server_token: str, scenario: str, +) -> None: + async def exercise() -> None: + identity = "py-bounded-failure-" + uuid.uuid4().hex[:12] + message = "M" * (3 * 1024 * 1024) if scenario == "message" else "original failure" + stack = "S" * (3 * 1024 * 1024) if scenario == "stack" else "short stack" + details: Any = {"small": True} + if scenario == "combined": + message, stack = "\x00šŸŒˆ\"\\" * 700000, "\x01" * (3 * 1024 * 1024) + details = {"cause": {"nested": "D" * (3 * 1024 * 1024)}} + non_retryable = scenario in {"message", "stack"} + requests: list[httpx.Request] = [] + async with _client(server_url, server_token) as client: + async def record(request: httpx.Request) -> None: + if request.url.path.endswith("/fail"): + requests.append(request) + + client._http.event_hooks["request"].append(record) + handle, task = await _claim(client, identity) + try: + outcome = replay(FailureConsumer, task["history_events"], [], run_id=task["run_id"]) + await client.complete_workflow_task( + task_id=task["task_id"], lease_owner=identity, + workflow_task_attempt=task["workflow_task_attempt"], + commands=[command.to_server_command(identity) for command in outcome.commands], + ) + activity = await client.poll_activity_task(worker_id=identity, task_queue=identity, timeout=2) + assert isinstance(activity, dict) + reply = await client.fail_activity_task( + task_id=activity["task_id"], activity_attempt_id=activity["activity_attempt_id"], + lease_owner=identity, message=message, stack_trace=stack, details=details, + failure_type="OriginalFailure", failure_class="app.OriginalFailure", failure_code=42, + non_retryable=non_retryable, + ) + assert reply["recorded"] is True + assert len(requests) == 1 + assert len(requests[0].content) <= 1024 * 1024 + submitted = json.loads(requests[0].content) + assert submitted["activity_attempt_id"] == activity["activity_attempt_id"] + assert submitted["lease_owner"] == identity + second = await client.poll_workflow_task(worker_id=identity, task_queue=identity, timeout=2) + assert isinstance(second, dict) + completed = replay(FailureConsumer, second["history_events"], [], run_id=second["run_id"]) + await client.complete_workflow_task( + task_id=second["task_id"], lease_owner=identity, + workflow_task_attempt=second["workflow_task_attempt"], + commands=[command.to_server_command(identity) for command in completed.commands], + ) + observed = await handle.result(timeout=3, poll_interval=.05) + assert observed["type"] == "OriginalFailure" + assert observed["class"] == "app.OriginalFailure" + assert observed["code"] == 42 + assert observed["non_retryable"] is non_retryable + assert ("[truncated" in observed["message"]) is (scenario != "small") + if scenario != "small": + assert observed["diagnostics"]["failure_report"]["omitted_fields"] + else: + assert observed["message"] == message + assert observed["diagnostics"] is None + finally: + await client.deregister_worker_registration(identity) + + await asyncio.wait_for(exercise(), timeout=30) + + +@pytest.mark.parametrize("field", ["message", "stack_trace"]) +async def test_workflow_task_failure_remains_a_recorded_replay_failure( + server_url: str, server_token: str, field: str, +) -> None: + async def exercise() -> None: + identity = "py-bounded-task-failure-" + uuid.uuid4().hex[:12] + requests: list[httpx.Request] = [] + async with _client(server_url, server_token) as client: + async def record(request: httpx.Request) -> None: + if request.url.path.endswith("/fail"): + requests.append(request) + + client._http.event_hooks["request"].append(record) + _handle, task = await _claim(client, identity) + try: + values = {"message": "replay failed", "stack_trace": "short stack"} + values[field] = "Z" * (3 * 1024 * 1024) + reply = await client.fail_workflow_task( + task_id=task["task_id"], lease_owner=identity, + workflow_task_attempt=task["workflow_task_attempt"], + failure_type="NonDeterministicReplayError", failure_reason="activity_identity_mismatch", **values, + ) + assert reply["recorded"] is True + assert reply["outcome"] == "failed" + assert reply["next_task_id"] is None + assert len(requests) == 1 + assert len(requests[0].content) <= 1024 * 1024 + body = json.loads(requests[0].content) + assert body["lease_owner"] == identity + assert body["workflow_task_attempt"] == task["workflow_task_attempt"] + assert body["failure"]["type"] == "NonDeterministicReplayError" + assert body["failure"]["reason"] == "activity_identity_mismatch" + assert "[truncated" in body["failure"]["message"] + finally: + await client.deregister_worker_registration(identity) + + await asyncio.wait_for(exercise(), timeout=30) + + +async def test_worker_reports_original_three_mib_exception(server_url: str, server_token: str) -> None: + from durable_workflow import Worker, activity + + class OriginalFailure(Exception): + pass + + @activity.defn(name="tests.bounded-failure") + async def fail() -> None: + raise OriginalFailure("W" * (3 * 1024 * 1024)) + + async def exercise() -> None: + identity = "py-bounded-worker-failure-" + uuid.uuid4().hex[:12] + async with _client(server_url, server_token) as client: + handle = await client.start_workflow( + workflow_type="tests.bounded-failure-consumer", task_queue=identity, workflow_id=identity, input=[], + ) + worker = Worker( + client, task_queue=identity, worker_id=identity, workflows=[FailureConsumer], activities=[fail], + ) + execution = await worker.run_until(workflow_id=handle.workflow_id, timeout=20, poll_interval=.1) + assert execution.status.lower() == "completed" + observed = await handle.result(timeout=3, poll_interval=.05) + assert observed["type"] == "OriginalFailure" + assert observed["non_retryable"] is False + assert "[truncated" in observed["message"] + assert observed["diagnostics"]["failure_report"]["omitted_fields"]["message"] == 3 * 1024 * 1024 + + await asyncio.wait_for(exercise(), timeout=30) + + +@pytest.mark.parametrize("prepared", [False, True], ids=["legacy", "prepared"]) +@pytest.mark.parametrize("non_retryable", [False, True]) +@pytest.mark.parametrize("limit", [1024 * 1024, 3500]) +async def test_local_workflow_observation_matches_recorded_failure_and_cold_replay( + server_url: str, server_token: str, prepared: bool, non_retryable: bool, limit: int, tmp_path: Any, +) -> None: + import hashlib + + from durable_workflow import Worker + from tests.failure_reporting_cases import LocalFailureObservation, large_local_failure + + async def exercise() -> None: + identity = "py-bounded-local-" + uuid.uuid4().hex[:12] + recorded: list[tuple[str, int, bytes]] = [] + async with Client( + server_url, worker_token=server_token, + control_token=os.environ.get("DURABLE_WORKFLOW_CONTROL_TOKEN", server_token), + timeout=5, failure_report_limit_bytes=limit, + ) as client: + async def capture(response: httpx.Response) -> None: + await response.aread() + path = response.request.url.path + if path.endswith(("/complete", "/outcome", "/export")): + index = len(recorded) + body = response.request.content + recorded.append((path, response.status_code, body)) + (tmp_path / f"request-{index}.json").write_bytes(body) + (tmp_path / f"response-{index}.json").write_bytes(response.content) + + client._http.event_hooks["response"].append(capture) + worker = Worker( + client, task_queue=identity, worker_id=identity, workflows=[LocalFailureObservation], + activities=[large_local_failure], + capabilities=["cooperative_cancellation", "prepared_local_activities"] if prepared else [], + ) + await worker._register() + try: + handle = await client.start_workflow( + workflow_type="tests.local.failure-observation", task_queue=identity, + workflow_id=identity, input=[non_retryable], + ) + task = await client.poll_workflow_task(worker_id=identity, task_queue=identity, timeout=2) + assert isinstance(task, dict) + commands = await worker._run_workflow_task(task) + assert commands is not None + first = await handle.result(timeout=3, poll_interval=.05) + history = await handle.get_history() + (tmp_path / "history.json").write_text(json.dumps(history, separators=(",", ":"))) + events = history.get("events", history.get("history_events", [])) + failure = next(event["payload"] for event in events if event["event_type"] == "ActivityFailed") + message = failure["message"] + assert first["bytes"] == len(message.encode()) <= 62 * 1024 + assert first["digest"] == hashlib.sha256(message.encode()).hexdigest() + assert first["marked"] is True + assert first["type"] == ("NonRetryableError" if non_retryable else "OriginalFailure") + assert first["non_retryable"] is non_retryable + cold = replay( + LocalFailureObservation, events, [non_retryable], workflow_id=identity, run_id=handle.run_id or "", + prepare_local_activities=prepared, + local_activity_executor=lambda _command: pytest.fail("cold replay reran local callback"), + ) + assert len(cold.commands) == 1 + assert cold.commands[0].result == first + await handle.export_history() + reports = [(path, status, body) for path, status, body in recorded if not path.endswith("/export")] + assert reports + assert all(status == 200 and len(body) <= limit for _, status, body in reports) + (tmp_path / "observation.json").write_text(json.dumps({ + "workflow_id": identity, "run_id": handle.run_id, "prepared": prepared, + "first_and_cold": first, "limit": limit, + "requests": [{"path": path, "status": status, "bytes": len(body), + "sha256": hashlib.sha256(body).hexdigest()} for path, status, body in recorded], + }, indent=2)) + finally: + await worker.stop() + + await asyncio.wait_for(exercise(), timeout=30) diff --git a/tests/test_failure_reporting.py b/tests/test_failure_reporting.py new file mode 100644 index 0000000..d610650 --- /dev/null +++ b/tests/test_failure_reporting.py @@ -0,0 +1,325 @@ +from __future__ import annotations + +import json +from typing import Any + +import httpx +import pytest + +from durable_workflow import serializer +from durable_workflow.client import Client + +MIB = 1024 * 1024 +MESSAGE_LIMIT = 62 * 1024 +STACK_LIMIT = 128 * 1024 + + +async def capture_failure(kind: str, *, client_kwargs: dict[str, Any] | None = None, **kwargs: Any) -> httpx.Request: + requests: list[httpx.Request] = [] + + def receive(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, json={"recorded": True}) + + async with Client("http://test", worker_token="worker", **(client_kwargs or {})) as client: + client._cluster_info = {} + client._http._transport = httpx.MockTransport(receive) + if kind == "activity": + await client.fail_activity_task( + task_id="task", activity_attempt_id="attempt", lease_owner="owner", **kwargs, + ) + elif kind == "workflow": + await client.fail_workflow_task(task_id="task", workflow_task_attempt=7, lease_owner="owner", **kwargs) + elif kind == "query": + await client.fail_query_task(query_task_id="task", query_task_attempt=7, lease_owner="owner", **kwargs) + else: + await client.reject_update_validation_task( + update_validation_task_id="task", update_validation_attempt=7, lease_owner="owner", + reason="validation_failed", **kwargs, + ) + assert len(requests) == 1 + return requests[0] + + +@pytest.mark.parametrize("kind", ["activity", "workflow", "query", "validation"]) +@pytest.mark.parametrize("oversized", ["message", "stack", "combined"]) +async def test_large_reports_keep_semantics_and_bound_actual_body(kind: str, oversized: str) -> None: + message = "application failure" if oversized == "stack" else "x" * (3 * MIB) + stack = "short stack" if oversized == "message" else "s" * (3 * MIB) + kwargs: dict[str, Any] = {"message": message, "stack_trace": stack, "failure_type": "OriginalFailure"} + if kind == "activity": + kwargs.update(non_retryable=True, failure_class="app.OriginalFailure", failure_code=42) + request = await capture_failure(kind, **kwargs) + body = json.loads(request.content) + failure = body["failure"] + assert len(request.content) <= MIB + assert int(request.headers["content-length"]) == len(request.content) + assert len(failure["message"].encode("utf-8")) <= MESSAGE_LIMIT + assert len(failure["stack_trace"].encode("utf-8")) <= STACK_LIMIT + assert "[truncated" in failure["message"] + assert failure["type"] == "OriginalFailure" + assert body["lease_owner"] == "owner" + if kind == "activity": + assert body["activity_attempt_id"] == "attempt" + assert failure["class"] == "app.OriginalFailure" + assert failure["non_retryable"] is True + assert failure["code"] == 42 + + +async def test_escaping_after_marker_insertion_fits_complete_request_and_lower_limit() -> None: + request = await capture_failure( + "activity", message="\x00\"\\\nšŸŒˆ" * (MIB // 4), stack_trace="\x00" * (3 * MIB), + failure_type="EscapedFailure", details={"nested": ["\x00" * (3 * MIB)]}, + client_kwargs={"failure_report_limit_bytes": 2200, "payload_size_warnings": False}, + ) + failure = json.loads(request.content)["failure"] + assert len(request.content) <= 2200 + assert failure["type"] == "EscapedFailure" + assert "[truncated" in failure["message"] + assert "details" not in failure + assert failure["diagnostics"]["failure_report"]["omitted_fields"]["details"] > 3 * MIB + + +async def test_small_report_is_unchanged_including_valid_avro_details() -> None: + request = await capture_failure("activity", message="small", stack_trace="trace", details={"a": [1.0, -0.0]}) + failure = json.loads(request.content)["failure"] + assert failure == { + "message": "small", "stack_trace": "trace", "details": serializer.envelope({"a": [1.0, -0.0]}), + } + assert serializer.decode_envelope(failure["details"]) == {"a": [1.0, -0.0]} + + +async def test_oversized_details_are_omitted_whole_and_reported() -> None: + request = await capture_failure("activity", message="original", details={"cause": {"nested": "z" * (3 * MIB)}}) + failure = json.loads(request.content)["failure"] + assert "details" not in failure + assert "[truncated" in failure["message"] + assert failure["diagnostics"]["failure_report"]["omitted_fields"]["details"] > 3 * MIB + + +async def test_unfit_claim_identity_is_explicit_refusal_without_http() -> None: + requests: list[httpx.Request] = [] + async with Client("http://test", failure_report_limit_bytes=512) as client: + client._cluster_info = {} + client._http._transport = httpx.MockTransport(lambda request: requests.append(request) or httpx.Response(200)) + with pytest.raises(Exception, match="failure report.*identity|failure report.*semantics") as caught: + await client.fail_activity_task( + task_id="task", activity_attempt_id="a" * 700, lease_owner="owner", + message="x" * (3 * MIB), failure_type="OriginalFailure", + ) + assert type(caught.value).__name__ == "FailureReportTooLarge" + assert requests == [] + + +async def test_terminal_workflow_failure_fits_whole_command_batch_without_mutation() -> None: + commands = [{ + "type": "fail_workflow", "message": "x" * (3 * MIB), "exception_type": "SagaCompensationFailed", + "exception_class": "app.SagaCompensationFailed", "non_retryable": True, + "failed_step_sequence": 3, "failed_activity_execution_id": "activity-id", + "exception": {"type": "SagaCompensationFailed", "class": "app.SagaCompensationFailed", + "cause": {"message": "z" * (3 * MIB)}}, + }] + requests: list[httpx.Request] = [] + async with Client("http://test", failure_report_limit_bytes=4000) as client: + client._cluster_info = {} + client._http._transport = httpx.MockTransport( + lambda request: requests.append(request) or httpx.Response(200, json={"recorded": True}), + ) + await client.complete_workflow_task( + task_id="task", lease_owner="owner", workflow_task_attempt=7, commands=commands, + ) + request = requests[0] + cmd = json.loads(request.content)["commands"][0] + assert len(request.content) <= 4000 + assert cmd["exception_type"] == "SagaCompensationFailed" + assert cmd["exception"]["type"] == "SagaCompensationFailed" + assert cmd["exception_class"] == "app.SagaCompensationFailed" + assert cmd["non_retryable"] is True + assert cmd["failed_step_sequence"] == 3 + assert cmd["failed_activity_execution_id"] == "activity-id" + assert "[truncated" in cmd["message"] + assert "cause" not in cmd["exception"] + assert len(commands[0]["message"]) == 3 * MIB + assert "cause" in commands[0]["exception"] + + +async def test_default_cap_counts_json_escaping_after_all_insertions() -> None: + request = await capture_failure("activity", message="\x00" * (3 * MIB), stack_trace="\x01" * (3 * MIB)) + body = json.loads(request.content) + assert len(request.content) <= MIB + assert "[truncated" in body["failure"]["message"] + assert len(body["failure"]["message"].encode()) <= MESSAGE_LIMIT + assert len(body["failure"]["stack_trace"].encode()) <= STACK_LIMIT + + +async def test_discovered_lower_limit_controls_reporting() -> None: + requests: list[httpx.Request] = [] + async with Client("http://test") as client: + client._cluster_info = {"limits": {"max_payload_bytes": 1700}} + client._http._transport = httpx.MockTransport( + lambda request: requests.append(request) or httpx.Response(200, json={"recorded": True}), + ) + await client.fail_workflow_task( + task_id="task", lease_owner="owner", workflow_task_attempt=2, + message="šŸŒˆ" * MIB, stack_trace="\x00" * MIB, + failure_type="ReplayFailure", failure_reason="nondeterminism", + ) + assert len(requests[0].content) <= 1700 + assert json.loads(requests[0].content)["failure"]["reason"] == "nondeterminism" + + +def test_details_budget_includes_new_diagnostics_and_preserves_retry_delay() -> None: + from durable_workflow._failure_reporting import bound_failure_request, encoded_json + + failure = {"message": "x" * (3 * MIB), "details": serializer.envelope("z" * 49120), + "non_retryable": False, "retry_after_seconds": 19, "type": "ApplicationError"} + original_details = dict(failure["details"]) + result = bound_failure_request("/worker/activity-tasks/task/fail", {"failure": failure})["failure"] + optional = {key: result[key] for key in ("details", "diagnostics") if key in result} + assert len(encoded_json(optional)) <= 64 * 1024 + assert result["non_retryable"] is False + assert result["retry_after_seconds"] == 19 + assert result["type"] == "ApplicationError" + assert failure["details"] == original_details + if "details" in result: + assert serializer.decode_envelope(result["details"]) == "z" * 49120 + else: + assert result["diagnostics"]["failure_report"]["omitted_fields"]["details"] > 0 + + +def test_local_attempts_and_update_failures_share_complete_body_budget() -> None: + from durable_workflow._failure_reporting import bound_failure_request, encoded_json + + attempts = [{"attempt_id": "first", "attempt_number": 1, "message": "a" * (3 * MIB), + "exception_type": "FirstFailure", "non_retryable": False, "backoff_seconds": 17}] + body = {"lease_owner": "owner", "workflow_task_attempt": 3, "commands": [ + {"type": "record_local_activity", "outcome": "failed", "message": "x" * (3 * MIB), + "exception_type": "LocalFailure", "non_retryable": True, "attempts": attempts}, + {"type": "fail_update", "update_id": "exact-update", "exception_type": "UpdateFailure", + "non_retryable": True, "message": "u" * (3 * MIB)}, + ]} + result = bound_failure_request("/worker/workflow-tasks/task/complete", body, 3500) + assert len(encoded_json(result)) <= 3500 + local, update = result["commands"] + assert local["exception_type"] == "LocalFailure" + assert "[truncated" in local["message"] + assert local["attempts"][0]["attempt_id"] == "first" + assert local["attempts"][0]["backoff_seconds"] == 17 + assert local["attempts"][0]["exception_type"] == "FirstFailure" + assert "[truncated" in local["attempts"][0]["message"] + assert update["update_id"] == "exact-update" + assert update["exception_type"] == "UpdateFailure" + assert "[truncated" in update["message"] + assert len(attempts[0]["message"]) == 3 * MIB + + +def test_prepared_local_outcome_uses_unchanged_attempt_and_bounded_failure() -> None: + from durable_workflow._failure_reporting import bound_failure_request, encoded_json + + body = {"lease_owner": "owner", "workflow_task_attempt": 5, + "report": {"outcome": "failed", "message": "\x00" * (3 * MIB), + "exception_type": "OriginalFailure", "non_retryable": True}} + result = bound_failure_request("/worker/workflow-tasks/task/local-activities/exact-attempt/outcome", body, 1800) + assert len(encoded_json(result)) <= 1800 + assert result["workflow_task_attempt"] == 5 + assert result["report"]["exception_type"] == "OriginalFailure" + assert result["report"]["non_retryable"] is True + assert "[truncated" in result["report"]["message"] + + +async def test_unencodable_optional_details_do_not_replace_application_failure() -> None: + details: dict[str, Any] = {} + details["cycle"] = details + request = await capture_failure("activity", message="original", failure_type="OriginalFailure", details=details) + failure = json.loads(request.content)["failure"] + assert failure["type"] == "OriginalFailure" + assert "details" not in failure + assert failure["diagnostics"]["failure_report"]["omitted_fields"]["details"] is None + assert "[truncated" in failure["message"] + + +async def test_advisory_warning_limit_does_not_become_a_transport_limit() -> None: + request = await capture_failure( + "activity", message="small unchanged failure", + client_kwargs={"payload_size_limit_bytes": 1, "payload_size_warnings": False}, + ) + assert json.loads(request.content)["failure"] == {"message": "small unchanged failure"} + + +async def test_query_client_clone_retains_explicit_reporting_limit() -> None: + from durable_workflow import Worker + + async with Client("http://test", failure_report_limit_bytes=1500, payload_size_warnings=False) as client: + worker = Worker(client, task_queue="q") + async with worker._clone_client_for_query_tasks() as cloned: + assert cloned.failure_report_limit_bytes == 1500 + + +@pytest.mark.parametrize("limit", [0, -1, True, 1.5]) +def test_invalid_reporting_limit_is_rejected(limit: Any) -> None: + with pytest.raises(ValueError, match="failure_report_limit_bytes"): + Client("http://test", failure_report_limit_bytes=limit) + + +async def test_final_budget_follows_runtime_externalization_of_other_commands() -> None: + from types import SimpleNamespace + from unittest.mock import AsyncMock, patch + + requests: list[httpx.Request] = [] + async with Client("http://test") as client: + client._cluster_info = {} + client._http._transport = httpx.MockTransport( + lambda request: requests.append(request) or httpx.Response(200, json={"recorded": True}), + ) + commands = [ + {"type": "upsert_memo", "memo": serializer.envelope({"large": "m" * (2 * MIB)})}, + {"type": "fail_workflow", "message": "f" * (3 * MIB), "exception_type": "OriginalFailure"}, + ] + + async def externalize(body: dict[str, Any], **_kwargs: Any) -> dict[str, Any]: + client._cluster_info = {"limits": {"max_payload_bytes": 2100}} + return {**body, "commands": [ + {"type": "upsert_memo", "memo": serializer.envelope({"small": "transformed payload"})}, + body["commands"][1], + ]} + + transport = SimpleNamespace(completion_context=False) + with ( + patch.object(client, "_runtime_external_payload_transport", AsyncMock(return_value=transport)), + patch.object(client, "_externalize_runtime_payloads", side_effect=externalize) as transformed, + ): + await client.complete_workflow_task( + task_id="task", lease_owner="exact-owner", workflow_task_attempt=7, commands=commands, + ) + transformed.assert_awaited_once() + assert len(requests) == 1 + assert len(requests[0].content) <= 2100 + body = json.loads(requests[0].content) + assert body["lease_owner"] == "exact-owner" + assert body["workflow_task_attempt"] == 7 + failure = body["commands"][1] + assert failure["exception_type"] == "OriginalFailure" + assert "[truncated" in failure["message"] + assert failure["exception"]["diagnostics"]["failure_report"]["omitted_fields"]["message"] == 3 * MIB + assert len(commands[1]["message"]) == 3 * MIB + + +def test_prepared_metadata_and_refitting_retain_original_sizes() -> None: + from durable_workflow._failure_reporting import bound_failure_request, prepare_failure_request + + body = {"lease_owner": "owner", "activity_attempt_id": "attempt", "failure": { + "message": "m" * (3 * MIB), "stack_trace": "s" * (3 * MIB), + "type": "OriginalFailure", "details": serializer.envelope({"nested": "d" * (3 * MIB)}), + }} + path = "/worker/activity-tasks/task/fail" + prepared = prepare_failure_request(path, body) + assert len(prepared["failure"]["message"]) == 3 * MIB + assert "details" not in prepared["failure"] + bounded = bound_failure_request(path, prepared) + assert bound_failure_request(path, bounded) == bounded + smaller = bound_failure_request(path, bounded, 2000) + sizes = smaller["failure"]["diagnostics"]["failure_report"]["omitted_fields"] + assert sizes["message"] == 3 * MIB + assert sizes["stack_trace"] == 3 * MIB + assert sizes["details"] > 3 * MIB + assert len(body["failure"]["message"]) == 3 * MIB diff --git a/tests/test_local_failure_reporting.py b/tests/test_local_failure_reporting.py new file mode 100644 index 0000000..98aab72 --- /dev/null +++ b/tests/test_local_failure_reporting.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import hashlib +import json +from unittest.mock import AsyncMock + +import httpx +import pytest + +from durable_workflow import Client, NonRetryableError, Worker, activity, serializer +from durable_workflow.workflow import replay +from tests.failure_reporting_cases import LocalFailureObservation + + +async def observe_local_failure(non_retryable: bool, limit: int) -> dict[str, object]: + class OriginalFailure(Exception): + pass + + error_type = NonRetryableError if non_retryable else OriginalFailure + + @activity.defn(name="tests.local.large-failure") + async def fail(_non_retryable: bool) -> None: + raise error_type("šŸŒˆ" * (3 * 1024 * 1024 // 4)) + + requests: list[httpx.Request] = [] + async with Client("http://test", worker_token="worker", failure_report_limit_bytes=limit) as client: + client._cluster_info = {} + client._http._transport = httpx.MockTransport( + lambda request: requests.append(request) or httpx.Response(200, json={"recorded": True}), + ) + worker = Worker( + client, task_queue="queue", worker_id="original-worker", + workflows=[LocalFailureObservation], activities=[fail], + ) + worker._renew_local_workflow_lease = AsyncMock() # type: ignore[method-assign] + await worker._run_workflow_task({ + "task_id": "original-task", "workflow_task_attempt": 7, "workflow_id": "workflow", "run_id": "run", + "workflow_type": "tests.local.failure-observation", "history_events": [], + "arguments": serializer.envelope([]), "payload_codec": "avro", + }) + complete = next(request for request in requests if request.url.path.endswith("/complete")) + assert len(complete.content) <= limit + body = json.loads(complete.content) + local, terminal = body["commands"] + assert local["type"] == "record_local_activity" + assert terminal["type"] == "complete_workflow" + first = serializer.decode_envelope(terminal["result"]) + assert first["bytes"] <= 62 * 1024 + assert first["marked"] is True + assert first["type"] == error_type.__name__ + assert first["non_retryable"] is non_retryable + assert first["digest"] == hashlib.sha256(local["message"].encode()).hexdigest() + assert local["exception_type"] == error_type.__name__ + assert local["non_retryable"] is non_retryable + assert body["lease_owner"] == "original-worker" + assert body["workflow_task_attempt"] == 7 + history = [{"event_type": "ActivityFailed", "payload": { + "sequence": 1, "activity_type": "tests.local.large-failure", "execution_mode": "local", + "message": local["message"], "exception_type": local["exception_type"], + "non_retryable": local["non_retryable"], "attempts": local["attempts"], + }}] + cold = replay( + LocalFailureObservation, history, [], run_id="run", + local_activity_executor=lambda _command: pytest.fail("cold replay reran the callback"), + ) + assert len(cold.commands) == 1 + assert serializer.decode_envelope(cold.commands[0].to_server_command("queue")["result"]) == first + return first + + +@pytest.mark.parametrize("non_retryable", [False, True]) +@pytest.mark.parametrize("limit", [1024 * 1024, 3500]) +async def test_local_failure_observation_matches_reported_and_cold_message(non_retryable: bool, limit: int) -> None: + await observe_local_failure(non_retryable, limit) + + +def test_final_batch_refuses_to_change_a_workflow_observed_local_message() -> None: + from durable_workflow import FailureReportTooLarge + from durable_workflow._failure_reporting import bound_failure_request, normalize_local_failure + + local = normalize_local_failure( + {"type": "record_local_activity", "activity_type": "original", "outcome": "failed", + "message": "šŸŒˆ" * (3 * 1024 * 1024 // 4), "exception_type": "OriginalFailure", + "non_retryable": False, "attempts": []}, + lease_owner="original-worker", workflow_task_attempt=7, + ) + observed = local["message"] + body = {"lease_owner": "original-worker", "workflow_task_attempt": 7, "commands": [local]} + with pytest.raises(FailureReportTooLarge): + bound_failure_request("/worker/workflow-tasks/original/complete", body, 2000) + assert local["message"] == observed + assert local["exception_type"] == "OriginalFailure" + assert local["non_retryable"] is False + + +async def test_unrepresentable_local_failure_type_is_a_reporting_refusal() -> None: + from durable_workflow.workflow import LocalActivityExecutionAborted, RecordLocalActivity + + original_type = type("OriginalFailure" * 30, (Exception,), {}) + + @activity.defn(name="tests.local.large-failure") + async def fail() -> None: + raise original_type("small original message") + + async with Client("http://test", worker_token="worker") as client: + worker = Worker(client, task_queue="queue", activities=[fail]) + worker._renew_local_workflow_lease = AsyncMock() # type: ignore[method-assign] + command = RecordLocalActivity("tests.local.large-failure", []) + with pytest.raises(LocalActivityExecutionAborted, match="failure type cannot cross"): + await worker._execute_local_activity( + {"task_id": "original-task", "workflow_task_attempt": 7}, command, payload_codec="avro", + ) + assert command.outcome is None diff --git a/tests/test_query_task_authority.py b/tests/test_query_task_authority.py index 5b45cfc..324c2e4 100644 --- a/tests/test_query_task_authority.py +++ b/tests/test_query_task_authority.py @@ -182,11 +182,14 @@ async def test_query_transport_discovery_uses_original_loading_budget() -> None: server = FakeRuntimePayloadServer() task = delivery(workflow_arguments=runtime_envelope(server, ["large"])) - task["lease_expires_at"] = timestamp(datetime.now(timezone.utc) + timedelta(seconds=0.1)) server.responses[("POST", "/api/worker/query-tasks/poll")] = {"task": task} cancelled = asyncio.Event() async def handle(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/query-tasks/poll"): + # Issue the original 100ms lease when the runtime delivers the claim. + # Client/TLS construction happens before the task exists in reality. + task["lease_expires_at"] = timestamp(datetime.now(timezone.utc) + timedelta(seconds=0.1)) if request.url.path.endswith("/cluster/info"): try: await asyncio.sleep(60) diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 3a7c201..98578a7 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -491,6 +491,21 @@ def _execute_fixture(fixture: dict[str, Any]) -> list[dict[str, Any]]: _assert_matches(fixture["expected"], {"command_sequence": commands}, f"{fixture['id']}.expected") assert report["implementation_identity"]["version"] == metadata.version("durable-workflow") return commands + if workflow_type == "sdk.local-failure-observation": + # Exercise first execution and cold replay at the real JSON boundary. + # The fixture holds a reproducer input, not an invented Server history. + from tests.test_local_failure_reporting import observe_local_failure + + assert "history" not in fixture + inputs = workflow.get("input") + assert isinstance(inputs, list) + assert len(inputs) == 1 + assert isinstance(inputs[0], dict) + control = inputs[0] + first = asyncio.run(observe_local_failure(control["non_retryable"], control["limit"])) + commands = [{"type": "local_failure_observation", "first_and_cold": first}] + _assert_matches(fixture["expected"], {"command_sequence": commands}, f"{fixture['id']}.expected") + return commands assert workflow_type in WORKFLOW_TYPES, ( f"replay fixture workflow {workflow_type!r} has no Python implementation; " "register its reproducer workflow in WORKFLOWS"