diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/README.md b/packages/aws-durable-execution-sdk-python-telemetry-exporter/README.md index 035f2ac5a..dccdd65c1 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/README.md +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/README.md @@ -44,11 +44,132 @@ def handler(event, context): ``` With no exporter configured, records are written to the function's own -CloudWatch log group as single JSON lines (the `LambdaLogExporter` default), -carrying the name-keyed `operationsByName` summary. The `S3Exporter` writes the +CloudWatch log group (the `LambdaLogExporter` default), one Lambda JSON log +line per record whose `message` is the compact record JSON carrying the +name-keyed `operationsByName` summary (see +[LambdaLogExporter](#lambdalogexporter)). The `S3Exporter` writes the lossless per-occurrence `operations` array, one object per execution (upsert-by-execution-name, so re-emission overwrites rather than appends). +## Record fields shared with the JS exporter + +Records follow the JS exporter's shape, which the record readers rely on for +every SDK: + +- `error.name`: see [Error names](#error-names) below. +- An opted-in operation `result` (`operations[].result`, + `operationsByName..result`, and the value passed to an override's + `result` transform) is the plain JSON value, for example `{"amount": 99.99}`. + Results checkpointed by the SDK's default serializer as a `{"t": ..., "v": + ...}` envelope are decoded with that serializer and converted to JSON values: + `datetime`/`date` become an ISO-8601 string from `isoformat()`, `UUID` a + string, `bytes` a base64 string, `tuple` a list, and a map/parallel + `BatchResult` its `to_dict()` form. A `Decimal` keeps its numeric value, but + not its scale or trailing zeros (`Decimal("1.10")` is `1.1`, `Decimal("1E+2")` + is `100`): an integral one of at most 300 digits is a JSON integer, any other + is a JSON number only when that float converts back to exactly the same + `Decimal`, and otherwise the string `str(value)` (for example + `"12345678901234567.89"`, or `"1E+5000"`). A non-finite `Decimal` + or `float` (NaN, Infinity) becomes `null`, so the record is strict JSON. A + payload that is not an envelope is passed through as parsed JSON (non-finite + numbers again become `null`), or as the raw string when it is not JSON; an + envelope that cannot be decoded is passed through as parsed JSON. +- The plugin cannot tell which serializer wrote a result, because the hooks do + not carry it, so an envelope is recognised by its shape alone: a JSON object + with exactly the keys `t` and `v`, where `t` is a string. A result written by + `JsonSerDes`, a custom `SerDes` (which the plugin never runs) or a chained + invoke that has this exact shape, and whose `t` is one of the default + serializer's type tags, is decoded as if the default serializer wrote it. For + example `{"t": "s", "v": "x"}` is reported as `"x"`, `{"t": "b", "v": 0}` as + `false`, and `{"t": "l", "v": [...]}` as the decoded list. A shape-matching + object whose `t` is not a default tag, or whose `v` does not decode, is + passed through unchanged. Avoid a root object of exactly `{"t", "v"}` in + results that you opt in when they do not use the default serializer. +- Every record built at invocation end carries `endTime` (the emission time) + and `durationMs` (`endTime - startTime`), including the `RUNNING` record of + an invocation that suspends or retries (`on-change` mode). Records built at + invocation start or on an operation change carry neither. Do not use the + presence of `endTime` as a completion signal; use `status`. + +## Error names + +The rule for the `error.name` field is the same in the JS, Python and Java +exporters. `error.message` is never changed by this rule. The rule applies to +the execution-level `error`, every `operations[].error`, and every +`operationsByName..error`. + +1. A record names the error that caused the failure. A wrapper error that the + SDK added is never the name, except in the bounded cases of rules 3 and 9. +2. The SDK unwraps an error only when the error's type is one of the SDK's own + wrapper types. A user error is never unwrapped, even if it has a cause. +3. Unwrapping follows the chain of SDK wrappers. The depth counts unwrap steps, + one per wrapper passed, not levels in the operation tree. At most 16 unwraps + happen. If the error reached after 16 unwraps is still a wrapper, its name + is reported. If the chain reaches an error it has already visited, the walk + stops and the last error visited before the repeat is reported. +4. If the chain ends at a user error, the name is that error's type name. +5. If the chain ends at an SDK error that is not a wrapper, the name is that SDK + error's simple class name. A callback timeout is an example. +6. A type name that is a qualified string is reduced to its last segment after + any `.` or `$`. If that segment is empty, the string is kept as it is. A + missing type, or a type made only of blank characters, is `Error`. The blank + characters are the six ASCII whitespace characters: space, `\t`, `\n`, + `\r`, `\f` and `\v`. No other character counts as blank, and names are + never trimmed. +7. A failure gets the same name whether the record is built live or from a + replayed checkpoint. +8. When the execution fails because an operation's error propagated, the + execution `error.name` equals that operation's `error.name`. +9. If a checkpoint holds only a wrapper type, which an older SDK can write, the + name is that wrapper's name. The exporter never guesses a cause from other + operations. + +A callback that is failed externally with an explicit `ErrorType` is named +after that type, after the reduction of rule 6. This includes the type +`"Error"`, which reports `Error`. Only a missing or blank type reports the SDK +callback-failure class. + +User error types have the same name in every SDK, because the user names them. +SDK error types keep each language's own class name. For example, a callback +timeout is `CallbackTimeoutError` in JS and Python and +`CallbackTimeoutException` in Java. Every SDK name for a callback timeout +contains `Timeout`. An invoke timeout does not always: Java names it +`InvokeTimedOutException`. + +How Python applies the rule: + +- The core SDK does the unwrapping (rules 2 to 5) when it checkpoints an + error: a failed step, child context, map item, parallel branch, + `wait_for_callback` context and the execution itself record the type of the + error that a wrapper wraps, not the wrapper. The SDK's wrapper types are + `StepError`, `ChildContextError` (also used for map items and parallel + branches), `InvokeError`, `WaitForConditionError`, `CallbackSubmitterError`, + `CallbackExternalError` and the base `DurableOperationError`. Every other + type is never unwrapped: user errors (including a user subclass of a + wrapper), and SDK errors such as `CallbackTimeoutError`, `SerDesError`, + `ValidationError` and `BatchCompletionError`. +- The exporter names each failed operation by its own `ErrorType`, reduced by + rule 6, and the execution by the execution's own `ErrorType`, reduced. It + does not match errors by message, stack trace or position in the tree. The + depth of the operation tree does not matter: a step in 18 nested child + contexts is `PaymentError` at every level. +- When two map items or parallel branches fail with different types and + `throw_if_error()` raises, each item or branch keeps its own name, and the + execution is named after the error that `throw_if_error()` raised. +- A checkpoint written by an older SDK can hold a wrapper type such as + `aws_durable_execution_sdk_python.exceptions.StepError`. Its name is then the + wrapper's simple class name, for example `StepError` (rule 9). +- A callback that times out is `CallbackTimeoutError` on the callback + operation (the service type is `Callback.Timeout`), on the `wait_for_callback` + context, and on the execution. A callback that an external system fails with + an error type is named by that type's simple name, for example + `PaymentError` for `com.acme.PaymentError`, and `Error` for `Error`. With + no error type, or a blank one, it is `CallbackExternalError`, the class the + SDK raises. A `wait_for_callback` submitter that fails with an empty message + is reported with the SDK's message `Callback submitter failed` on the + execution and is still named after the submitter's error. +- A user error class with no explicit name is named by its class name. + ## Exporters All exporters live in `aws_durable_execution_sdk_python_telemetry_exporter.exporters` and @@ -77,6 +198,36 @@ the Lambda runtime's boto3 otherwise; none adds a required dependency. size limit; omitting it keeps the default. `HttpExporter` and `FileExporter` have no default and do not truncate unless a limit is set. +### LambdaLogExporter + +Prints one line per record to stdout, so it lands in the function's own log +group with no extra IAM. The line is the Lambda JSON log envelope, with keys +in this order: + +```json +{"timestamp":"2026-01-01T00:00:00.000Z","level":"INFO","message":"{\"recordType\":\"DurableExecutionTelemetry\",...}"} +``` + +`timestamp` is the UTC write time with millisecond precision, `level` is +always `INFO`, and `message` is the compact record JSON as a string. This is +the shape the Node.js runtime gives the JS exporter's `console.log` line in the +JSON log format, and record readers for this destination filter on +`level = "INFO" and message like /DurableExecutionTelemetry/` and parse +`message`. Lambda captures Python `print` output as written in both the Text +and JSON log formats, and does not re-encode lines that are already JSON, so +the envelope is never wrapped twice. With the JSON log format and an +application log level above `INFO` (`WARN`, `ERROR`, `FATAL`), Lambda filters +these lines out, as it does for the JS exporter's lines. `requestId` is not +included, because exporters do not receive the request id. +`max_record_size_bytes` (default 256 KB) is measured on the rendered record, +that is the `message` string. The envelope keys and the JSON escaping of +`message` add bytes on top of it, so the printed line can be up to about twice +the record size: escaping turns each `"` and `\` in the record into two bytes. +A typical record grows by about 20%. CloudWatch Logs accepts log events of up +to 1 MB (see the `PutLogEvents` API reference), so a line written for a record +at the default limit fits. A larger `max_record_size_bytes` should stay at or +below about 500 KB for this exporter to keep that margin. + ### CloudWatchLogsExporter Writes one `PutLogEvents` event per record to `log_group_name`, in a stream diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/lambda_log_exporter.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/lambda_log_exporter.py index 5ddb8dfc1..f7c9ef57c 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/lambda_log_exporter.py +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/lambda_log_exporter.py @@ -5,6 +5,7 @@ from __future__ import annotations +import datetime from typing import Any from aws_durable_execution_sdk_python_telemetry_exporter.exporters._common import ( @@ -16,11 +17,26 @@ class LambdaLogExporter: - """Writes ``operationsByName`` records to the function's own log group via ``print``. + """Writes ``operationsByName`` records to the function's own log group. - Port of the JS ``LambdaLogExporter``: ``console.log(JSON.stringify( - withOperationsByName(record)))``. Requires no extra IAM. Emits the name-keyed - summary map (``OPERATIONS_BY_NAME``). + Port of the JS ``LambdaLogExporter``, which calls ``console.log(JSON.stringify( + withOperationsByName(record)))``. In the Lambda JSON log format the Node.js + runtime wraps that line in a ``{timestamp, level, requestId, message}`` + envelope, and the record readers for this destination query + ``level = "INFO"`` and parse ``message``. The Python runtime captures + ``print`` output as plain text in both log formats, so this exporter writes + the envelope itself: one stdout line of compact JSON with the keys + ``timestamp`` (UTC ISO-8601, millisecond precision, ``Z``), ``level`` + (``"INFO"``) and ``message`` (the compact record JSON, as a string). The + request id is not available to exporters, so ``requestId`` is omitted. + Requires no extra IAM. + + ``max_record_size_bytes`` limits the rendered record (the ``message`` + string), as for every exporter; the envelope keys and the JSON escaping of + ``message`` add bytes on top of it. The line can be up to about twice the + record (escaping doubles each ``"`` and ``\\``), about 20% more for a + typical record. CloudWatch Logs accepts events of up to 1 MB, so a record + at the default 256 KB limit fits. """ def __init__(self, max_record_size_bytes: int | None = None) -> None: @@ -32,10 +48,19 @@ def render(self, record: dict[str, Any]) -> dict[str, Any]: return with_operations_by_name(record) def export(self, record: dict[str, Any]) -> None: - # Raw JSON line to stdout -> the function's CloudWatch log group. The - # conformance CloudWatch sink json.loads each line (and unwraps the - # Lambda structured-log envelope when present). - print(compact_dumps(self.render(record)), flush=True) # noqa: T201 + timestamp = ( + datetime.datetime.now(datetime.UTC) + .isoformat(timespec="milliseconds") + .replace("+00:00", "Z") + ) + envelope = { + "timestamp": timestamp, + "level": "INFO", + "message": compact_dumps(self.render(record)), + } + # print (not logging): Lambda leaves stdout lines as written in both + # the Text and JSON log formats, so the envelope is never wrapped twice. + print(compact_dumps(envelope), flush=True) # noqa: T201 def flush(self) -> None: return None diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/plugin.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/plugin.py index c374a5099..af97f4c60 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/plugin.py +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/plugin.py @@ -31,12 +31,24 @@ from __future__ import annotations +import base64 import datetime +import decimal import json import math import threading +import uuid from typing import Any, Callable +from aws_durable_execution_sdk_python.concurrency.models import BatchResult +from aws_durable_execution_sdk_python.exceptions import ( + CallbackExternalError, + CallbackTimeoutError, +) +from aws_durable_execution_sdk_python.lambda_service import ( + ErrorObject, + OperationStatus, +) from aws_durable_execution_sdk_python.plugin import ( DurableInstrumentationPlugin, InvocationEndInfo, @@ -46,6 +58,10 @@ OperationInfo, OperationType, ) +from aws_durable_execution_sdk_python.serdes import ( + EXTENDED_TYPES_SERDES, + SerDesContext, +) from aws_durable_execution_sdk_python_telemetry_exporter._export_scheduler import ( _ExportScheduler, @@ -152,16 +168,231 @@ def _apply_result_override( ) -> Any: if raw_result is None: return None - try: - parsed = json.loads(raw_result) - except (json.JSONDecodeError, TypeError): - parsed = raw_result + parsed = _decode_checkpoint_result(raw_result) try: return transform(parsed) except Exception: # noqa: BLE001 - untrusted transform must never break emission return None +# -- operation results ---------------------------------------------------------- + + +def _is_extended_type_envelope(value: Any) -> bool: + # ExtendedTypeSerDes wraps every non-primitive value in exactly + # {"t": , "v": } at the root. + return ( + isinstance(value, dict) + and value.keys() == {"t", "v"} + and isinstance(value["t"], str) + ) + + +def _decode_checkpoint_result(raw_result: str) -> Any: + """Return an operation's checkpointed result as a plain JSON value. + + The SDK's default serializer (``ExtendedTypeSerDes``) writes primitives and + primitive lists as plain JSON, and everything else (dicts, nested lists, + ``Decimal``, ``datetime``, ...) as a ``{"t": ..., "v": ...}`` envelope. The + JS and Java exporters emit the user's plain JSON value, so an envelope is + decoded with that default serializer and converted to JSON-compatible + values (see ``_to_json_compatible``). A user's custom ``SerDes`` is never + run: a payload that is not an envelope keeps the previous behaviour (parsed + JSON, or the raw string when it is not JSON), and so does an envelope that + fails to decode. Never raises. + + The plugin does not know which serializer wrote a result (the hooks do not + carry it), so the envelope is recognised by shape alone: a JSON object + with exactly the keys ``t`` and ``v`` where ``t`` is a string. A result + written by another serializer (``JsonSerDes``, a custom ``SerDes``, or a + chained invoke's JSON payload) that has this exact shape, and whose ``t`` + is one of the default serializer's tags, is decoded as if the default + serializer wrote it; for example ``{"t": "s", "v": "x"}`` is reported as + ``"x"``. Any other ``t`` fails to decode and is passed through unchanged. + """ + try: + parsed = json.loads(raw_result) + except (json.JSONDecodeError, TypeError, RecursionError): + return raw_result + try: + if not _is_extended_type_envelope(parsed): + # Plain JSON already; only non-finite numbers (json.loads accepts + # NaN and Infinity) need mapping to null. + return _to_json_compatible(parsed) + try: + decoded = EXTENDED_TYPES_SERDES.deserialize(raw_result, SerDesContext()) + return _to_json_compatible(decoded) + except Exception: # noqa: BLE001 - fall back to the parsed payload + return _to_json_compatible(parsed) + except Exception: # noqa: BLE001 - one bad result never breaks the record + return raw_result + + +# An integral Decimal becomes a JSON integer only up to this many digits; a +# larger one keeps its exact string form (see _to_json_compatible). +_MAX_INT_DIGITS = 300 + + +def _to_json_compatible(value: Any) -> Any: + """Convert a value decoded by ``ExtendedTypeSerDes`` into plain JSON values. + + Non-JSON types follow what the Java SDK's default Jackson serializer writes + for the equivalent Java types, keeping the numeric value (not the scale: + ``Decimal("1.10")`` -> ``1.1``): an integral ``Decimal`` of at most 300 + digits -> JSON integer; any other finite ``Decimal`` -> JSON number when + that float converts back to exactly the same ``Decimal``, otherwise its exact + string form (``str(value)``, e.g. ``"12345678901234567.89"`` or + ``"1E+5000"``); ``datetime`` + / ``date`` -> ISO-8601 string (``isoformat()``, offset and precision kept), + ``UUID`` -> string, ``bytes`` -> base64 string, ``tuple`` -> list. A + ``BatchResult`` becomes its ``to_dict()`` form. A non-finite ``Decimal`` or + ``float`` (NaN, Infinity) becomes ``None`` (JSON ``null``), like + ``JSON.stringify(NaN)`` in JS, so the record stays strict JSON. + + Containers are converted with an explicit stack, so a deeply nested result + does not hit the interpreter's recursion limit. + """ + root: list[Any] = [None] + # (value, target container, key in target), processed in source order. + pending: list[tuple[Any, Any, Any]] = [(value, root, 0)] + while pending: + item, target, key = pending.pop() + if isinstance(item, BatchResult): + item = item.to_dict() + if isinstance(item, (list, tuple)): + converted_list: list[Any] = [None] * len(item) + target[key] = converted_list + pending.extend( + (child, converted_list, index) + for index, child in reversed(list(enumerate(item))) + ) + elif isinstance(item, dict): + converted_dict: dict[str, Any] = {} + target[key] = converted_dict + entries = [(str(k), child) for k, child in item.items()] + for name, _child in entries: + converted_dict[name] = None + pending.extend( + (child, converted_dict, name) for name, child in reversed(entries) + ) + else: + target[key] = _scalar_to_json_compatible(item) + return root[0] + + +def _scalar_to_json_compatible(value: Any) -> Any: + if isinstance(value, float): + return value if math.isfinite(value) else None + if value is None or isinstance(value, (str, bool, int)): + return value + if isinstance(value, decimal.Decimal): + if not value.is_finite(): + return None + # adjusted() is the exponent of the leading digit, so the bound keeps + # int() to at most _MAX_INT_DIGITS digits: Decimal("1E+1000000") is 46 + # bytes on the wire but a million-digit int, which takes tens of seconds + # to build and cannot be written by json (the int-to-str digit limit). + if ( + value.adjusted() < _MAX_INT_DIGITS + and value == value.to_integral_value() + and not (value.is_zero() and value.is_signed()) + ): + return int(value) + as_float = float(value) + if decimal.Decimal(repr(as_float)) == value: + return as_float + return str(value) + if isinstance(value, (datetime.datetime, datetime.date)): + return value.isoformat() + if isinstance(value, uuid.UUID): + return str(value) + if isinstance(value, (bytes, bytearray, memoryview)): + return base64.b64encode(bytes(value)).decode("ascii") + msg = f"Unsupported decoded result type: {type(value)!r}" + raise TypeError(msg) + + +# -- error names ---------------------------------------------------------------- +# +# The cross-SDK rule for ``error.name`` (execution ``error``, ``operations[].error`` +# and ``operationsByName..error``), spec version 3. The core SDK +# checkpoints the error a wrapper wraps, never the wrapper (it unwraps its own +# wrappers, at most 16, stopping at a repeated cause), so every failed +# operation and the execution carry the user's type. The exporter only reduces +# that type: +# +# * An operation is named by its own ErrorType, reduced (Table A). A callback +# operation that timed out is ``CallbackTimeoutError``; one failed externally +# with a missing or blank type is ``CallbackExternalError``. An explicit type, +# ``"Error"`` included, is the sender's. +# * The execution is named by the reduced type of its own ErrorType. A +# checkpoint written by an older core can still hold an SDK wrapper type; the +# wrapper's own name is then reported (rule 9). The exporter never infers a +# cause from other operations. +# +# Reduction: the text after the last ``.`` or ``$``; an empty segment keeps the +# string as it is. A missing type, or one made only of the six ASCII whitespace +# characters, is ``Error``. Names are never trimmed. + + +def _qualified(cls: type) -> str: + return f"{cls.__module__}.{cls.__qualname__}" + + +_CALLBACK_TIMEOUT_TYPE = _qualified(CallbackTimeoutError) +_CALLBACK_EXTERNAL_TYPE = _qualified(CallbackExternalError) + +_DEFAULT_ERROR_NAME = "Error" +# Only these six characters are blank (not str.isspace() or strip()). +_BLANK_CHARACTERS = frozenset(" \t\n\r\f\v") + + +def _is_blank(type_name: str | None) -> bool: + return type_name is None or all(ch in _BLANK_CHARACTERS for ch in type_name) + + +def _simple_type_name(type_name: str | None) -> str: + """Reduce a type string to the reported ``error.name`` (Table A). + + The text after the last ``.`` or ``$``: ``"__main__.PaymentError"`` and + ``"pkg.Outer$PaymentError"`` -> ``"PaymentError"``. When that segment is + empty (``"trailing."``, ``"Outer$"``, ``"."``) the string is kept as it is. + A missing or empty type, or one made only of space, ``\\t``, ``\\n``, + ``\\r``, ``\\f`` and ``\\v``, is ``"Error"``. Nothing is trimmed, and other + whitespace (U+00A0, U+2003, U+FEFF, U+001C, ...) is kept. + """ + if type_name is None or _is_blank(type_name): + return _DEFAULT_ERROR_NAME + segment = type_name[max(type_name.rfind("."), type_name.rfind("$")) + 1 :] + return segment or type_name + + +def _operation_error_name(op: OperationInfo) -> str: + """``error.name`` of a failed operation: its own ErrorType, reduced. + + A callback operation's error comes from the service or the external sender: + a timeout is named by the SDK's CallbackTimeoutError (the service type + would reduce to ``Timeout``), and an external failure without a type + (missing or blank) by the SDK's CallbackExternalError, which is what the + SDK raises for both. + """ + error_type = op.error.type if op.error is not None else None + if op.operation_type == OperationType.CALLBACK: + if op.status == OperationStatus.TIMED_OUT: + error_type = _CALLBACK_TIMEOUT_TYPE + elif _is_blank(error_type): + error_type = _CALLBACK_EXTERNAL_TYPE + return _simple_type_name(error_type) + + +def _execution_error_name(error: ErrorObject) -> str: + """``error.name`` of the execution: its own ErrorType, reduced. + + A wrapper type (from an older core) reduces to the wrapper's name. + """ + return _simple_type_name(error.type) + + class _ExecutionState: __slots__ = ("start_time", "parsed_arn", "cached_input", "operations") @@ -312,16 +543,17 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None: should_emit = is_terminal if should_emit: - # Only terminal (SUCCEEDED/FAILED) records carry an end time; a - # PENDING/RETRY invocation end maps to RUNNING (still in flight) and - # must omit endTime/durationMs. Passing end_time=None makes _emit - # drop both fields. Output and error likewise belong only to a - # terminal record. + # Every record built at invocation end carries endTime (the emission + # time) and durationMs, including a PENDING/RETRY end that maps to + # RUNNING -- the same rule as the JS plugin's onInvocationEnd + # (``endTime: new Date()`` for every status). Records built at + # invocation start or on an operation change carry neither. Output + # and error still belong only to a terminal record. self._emit( arn, state, status=status, - end_time=datetime.datetime.now(datetime.UTC) if is_terminal else None, + end_time=datetime.datetime.now(datetime.UTC), output_raw=info.execution_result if is_terminal else None, error=info.error if is_terminal else None, ) @@ -369,7 +601,10 @@ def _build_operations( if op.attempt is not None: entry["attempt"] = op.attempt if self._include_errors and op.error is not None: - entry["error"] = {"name": op.error.type, "message": op.error.message} + entry["error"] = { + "name": _operation_error_name(op), + "message": op.error.message, + } if override is not None and override.result is not None: value = _apply_result_override(override.result, op.result) if value is not None: @@ -436,7 +671,10 @@ def _emit( if output_value is not None: record["output"] = output_value if error is not None: - record["error"] = {"name": error.type, "message": error.message} + record["error"] = { + "name": _execution_error_name(error), + "message": error.message, + } record["operations"] = self._build_operations(operations) self._scheduler.schedule(record) diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/e2e/error_name_int_test.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/e2e/error_name_int_test.py new file mode 100644 index 000000000..3ff522004 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/e2e/error_name_int_test.py @@ -0,0 +1,567 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""End-to-end test: the cross-SDK ``error.name`` rule (Table B). + +Each scenario runs through the LOCAL durable runner (``DurableFunctionTestRunner``) +twice: live, and replayed. "Replayed" means that the handler catches the error, +waits so that a new invocation starts, and rethrows the error, so the record is +built from the replayed checkpoint. Every row asserts: + +* the operation and execution ``error.name`` values of the table; +* ``error.message`` is the original message; +* the live and replayed names are equal; +* each ``operationsByName`` entry has the same name as ``operations[]``. + +Row B15 (a JS ``errorType`` field or a Java SDK-exception subclass) runs in its +Python form: a user error class that subclasses an SDK wrapper. B18 uses a class +created with ``type()`` whose name contains ``$``. +""" + +from __future__ import annotations + +import threading +import time +from dataclasses import dataclass +from typing import Any, Callable + +import pytest +from aws_durable_execution_sdk_python.config import ( + CompletionConfig, + Duration, + MapConfig, + ParallelConfig, + StepConfig, + WaitForCallbackConfig, +) +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.exceptions import StepError, ValidationError +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.lambda_service import ErrorObject +from aws_durable_execution_sdk_python.retries import ( + JitterStrategy, + RetryStrategyConfig, + create_retry_strategy, +) +from aws_durable_execution_sdk_python.waits import ( + WaitForConditionConfig, + WaitForConditionDecision, +) + +from aws_durable_execution_sdk_python_telemetry_exporter import ( + DurableExecutionTelemetryConfig, + LambdaLogExporter, + durable_execution_telemetry, +) +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner + + +MESSAGE = "card declined" + + +class PaymentError(Exception): + """The user error type of Table B.""" + + +class FraudError(Exception): + """B20/B21: a second user error type with the same message.""" + + +class WrappedPaymentError(StepError): + """B15 (Python form): a user error that subclasses an SDK wrapper.""" + + +# B18: a class whose name holds a separator (legal through ``type()``). +DollarError: type[Exception] = type("Payment$Error", (Exception,), {}) + +NO_RETRY = StepConfig( + retry_strategy=create_retry_strategy(RetryStrategyConfig(max_attempts=1)) +) +RETRY_ONCE = StepConfig( + retry_strategy=create_retry_strategy( + RetryStrategyConfig( + max_attempts=2, + initial_delay=Duration.from_seconds(1), + jitter_strategy=JitterStrategy.NONE, + ) + ) +) +NO_RETRY_SUBMITTER = WaitForCallbackConfig( + retry_strategy=create_retry_strategy(RetryStrategyConfig(max_attempts=1)) +) + + +def _raise(exc: BaseException) -> Callable[..., Any]: + def fail(*_args: Any) -> Any: + raise exc + + return fail + + +class _Capture: + def __init__(self) -> None: + self.max_record_size_bytes: int | None = None + self.records: list[dict[str, Any]] = [] + self.lock = threading.Lock() + + def render(self, record: dict[str, Any]) -> dict[str, Any]: + return record + + def export(self, record: dict[str, Any]) -> None: + with self.lock: + self.records.append(record) + + def flush(self) -> None: + return None + + +@dataclass +class Scenario: + row: str + body: Callable[[DurableContext], Any] + # Expected error.name per operation name (only failed operations listed). + operations: dict[str, str] + execution: str + message: str = MESSAGE + setup: Callable[[DurableFunctionTestRunner], None] | None = None + # Drives an external event (a callback failure) while the execution runs. + drive: Callable[[DurableFunctionTestRunner, str], None] | None = None + fresh: Callable[[], Callable[[DurableContext], Any]] | None = None + # Operation messages that differ from the execution's, by operation name. + operation_messages: dict[str, str] | None = None + + +# -- scenario bodies --------------------------------------------------------------- + + +def _b1(ctx: DurableContext) -> Any: + return ctx.step(_raise(PaymentError(MESSAGE)), name="charge", config=NO_RETRY) + + +def _b2() -> Callable[[DurableContext], Any]: + attempts = {"n": 0} + + def flaky(_s: Any) -> Any: + attempts["n"] += 1 + if attempts["n"] == 1: + raise TypeError("transient") + raise PaymentError(MESSAGE) + + def body(ctx: DurableContext) -> Any: + return ctx.step(flaky, name="charge", config=RETRY_ONCE) + + return body + + +def _b3(ctx: DurableContext) -> Any: + return ctx.run_in_child_context(_raise(PaymentError(MESSAGE)), name="checkout") + + +def _b4(ctx: DurableContext) -> Any: + def child(c: DurableContext) -> Any: + return c.step(_raise(PaymentError(MESSAGE)), name="charge", config=NO_RETRY) + + return ctx.run_in_child_context(child, name="checkout") + + +def _b5(ctx: DurableContext) -> Any: + ctx.map(["a"], _raise(PaymentError(MESSAGE)), name="fanout").throw_if_error() + + +def _b6(ctx: DurableContext) -> Any: + def item(c: DurableContext, *_args: Any) -> Any: + return c.step(_raise(PaymentError(MESSAGE)), name="charge", config=NO_RETRY) + + ctx.map(["a"], item, name="fanout").throw_if_error() + + +def _b7(ctx: DurableContext) -> Any: + ctx.parallel( + [lambda _c: "ok", _raise(PaymentError(MESSAGE))], name="branches" + ).throw_if_error() + + +def _b8(ctx: DurableContext) -> Any: + return ctx.wait_for_condition( + _raise(PaymentError(MESSAGE)), + WaitForConditionConfig( + wait_strategy=lambda _s, _n: WaitForConditionDecision.stop_polling(), + initial_state=0, + ), + name="poll", + ) + + +@durable_execution +def _payments_handler(event: Any, context: DurableContext) -> Any: + raise PaymentError(MESSAGE) + + +def _b9_setup(runner: DurableFunctionTestRunner) -> None: + runner.register_durable_function("payments", _payments_handler) + + +def _b9(ctx: DurableContext) -> Any: + return ctx.invoke("payments", {}, name="pay") + + +def _b10(ctx: DurableContext) -> Any: + return ctx.wait_for_callback( + _raise(PaymentError(MESSAGE)), name="approval", config=NO_RETRY_SUBMITTER + ) + + +def _b11(ctx: DurableContext) -> Any: + return ctx.wait_for_callback( + lambda _id, _c: None, + name="approval", + config=WaitForCallbackConfig(timeout=Duration.from_seconds(1)), + ) + + +def _b12_b13(ctx: DurableContext) -> Any: + return ctx.wait_for_callback(lambda _id, _c: None, name="approval") + + +def _fail_callback( + error_type: str | None, message: str | None = MESSAGE +) -> Callable[[DurableFunctionTestRunner, str], None]: + def drive(runner: DurableFunctionTestRunner, arn: str) -> None: + callback_id = runner.wait_for_callback(arn, timeout=30) + runner.send_callback_failure( + callback_id, + ErrorObject(message=message, type=error_type, data=None, stack_trace=None), + ) + + return drive + + +def _b10_empty_message(ctx: DurableContext) -> Any: + # The core re-raises the submitter's failure with its default message. + return ctx.wait_for_callback( + _raise(PaymentError()), name="approval", config=NO_RETRY_SUBMITTER + ) + + +def _b14(ctx: DurableContext) -> Any: + try: + raise TypeError("root cause") + except TypeError as cause: + raise PaymentError(MESSAGE) from cause + + +def _b15(ctx: DurableContext) -> Any: + raise WrappedPaymentError(MESSAGE) + + +def _b16(ctx: DurableContext) -> Any: + try: + ctx.step(_raise(ValueError("x")), name="lookup", config=NO_RETRY) + except StepError: + pass + ctx.run_in_child_context(_raise(ValidationError("x")), name="work") + + +def _b17(ctx: DurableContext) -> Any: + # The step result cannot be serialized: an SDK-only failure. + return ctx.step(lambda _s: object(), name="charge", config=NO_RETRY) + + +def _b18(ctx: DurableContext) -> Any: + return ctx.step(_raise(DollarError(MESSAGE)), name="charge", config=NO_RETRY) + + +def _b20(ctx: DurableContext) -> Any: + def item(_c: DurableContext, value: str, *_args: Any) -> Any: + # One raise site, so both items carry the same message and stack trace. + raise (PaymentError if value == "a" else FraudError)(MESSAGE) + + # Both items run, one at a time, so item 0's error is the one raised. + config: MapConfig[Any] = MapConfig( + max_concurrency=1, completion_config=CompletionConfig.all_completed() + ) + ctx.map(["a", "b"], item, name="fanout", config=config).throw_if_error() + + +def _b21(ctx: DurableContext) -> Any: + config = ParallelConfig( + max_concurrency=1, completion_config=CompletionConfig.all_completed() + ) + ctx.parallel( + [_raise(PaymentError(MESSAGE)), _raise(FraudError(MESSAGE))], + name="branches", + config=config, + ).throw_if_error() + + +def _handled_in_subtree(ctx: DurableContext) -> Any: + # Review case 5b: a failure handled inside the failing context's subtree. + def inner(c: DurableContext) -> Any: + try: + c.step(_raise(ValueError(MESSAGE)), name="lookup", config=NO_RETRY) + except StepError: + return "handled" + return "unreachable" # pragma: no cover + + def outer(c: DurableContext) -> Any: + c.run_in_child_context(inner, name="inner") + return c.step(_raise(PaymentError(MESSAGE)), name="charge", config=NO_RETRY) + + return ctx.run_in_child_context(outer, name="outer") + + +def _nested(depth: int) -> Callable[[DurableContext], Any]: + """A step that throws PaymentError inside ``depth`` nested child contexts.""" + + def level(i: int) -> Callable[[DurableContext], Any]: + def body(c: DurableContext) -> Any: + if i > depth: + return c.step( + _raise(PaymentError(MESSAGE)), name="charge", config=NO_RETRY + ) + return c.run_in_child_context(level(i + 1), name=f"ctx-{i:02d}") + + return body + + return level(1) + + +def _depth_names(depth: int) -> dict[str, str]: + names = {f"ctx-{i:02d}": "PaymentError" for i in range(1, depth + 1)} + names["charge"] = "PaymentError" + return names + + +def _map_in_map_in_child(ctx: DurableContext) -> Any: + def inner_item(_c: DurableContext, *_args: Any) -> Any: + raise PaymentError(MESSAGE) + + def outer_item(c: DurableContext, *_args: Any) -> Any: + c.map(["b"], inner_item, name="inner").throw_if_error() + + def child(c: DurableContext) -> Any: + c.map(["a"], outer_item, name="outer").throw_if_error() + + return ctx.run_in_child_context(child, name="checkout") + + +SCENARIOS = [ + Scenario("B1", _b1, {"charge": "PaymentError"}, "PaymentError"), + Scenario("B2", _b1, {"charge": "PaymentError"}, "PaymentError", fresh=_b2), + Scenario("B3", _b3, {"checkout": "PaymentError"}, "PaymentError"), + Scenario( + "B4", + _b4, + {"checkout": "PaymentError", "charge": "PaymentError"}, + "PaymentError", + ), + Scenario("B5", _b5, {"map-item-0": "PaymentError"}, "PaymentError"), + Scenario( + "B6", + _b6, + {"map-item-0": "PaymentError", "charge": "PaymentError"}, + "PaymentError", + ), + Scenario("B7", _b7, {"parallel-branch-1": "PaymentError"}, "PaymentError"), + Scenario("B8", _b8, {"poll": "PaymentError"}, "PaymentError"), + Scenario("B9", _b9, {"pay": "PaymentError"}, "PaymentError", setup=_b9_setup), + Scenario( + "B10", + _b10, + {"approval": "PaymentError", "approval submitter": "PaymentError"}, + "PaymentError", + ), + Scenario( + "B10-empty-message", + _b10_empty_message, + {"approval": "PaymentError", "approval submitter": "PaymentError"}, + "PaymentError", + message="Callback submitter failed", + # The context and the step keep the submitter's empty message. + operation_messages={"approval": "", "approval submitter": ""}, + ), + Scenario( + "B11", + _b11, + { + "approval": "CallbackTimeoutError", + "approval create callback id": "CallbackTimeoutError", + }, + "CallbackTimeoutError", + message="Callback timed out", + ), + Scenario( + "B12", + _b12_b13, + {"approval": "PaymentError", "approval create callback id": "PaymentError"}, + "PaymentError", + drive=_fail_callback("com.acme.PaymentError"), + ), + Scenario( + "B13", + _b12_b13, + { + "approval": "CallbackExternalError", + "approval create callback id": "CallbackExternalError", + }, + "CallbackExternalError", + drive=_fail_callback(None), + ), + Scenario( + "B13b", + _b12_b13, + {"approval": "Error", "approval create callback id": "Error"}, + "Error", + drive=_fail_callback("Error"), + ), + Scenario( + "B12-empty-message", + _b12_b13, + {"approval": "PaymentError", "approval create callback id": "PaymentError"}, + "PaymentError", + # The core fills in its default message on the context and the + # execution; the callback operation keeps the sender's empty one. + message="Callback failed", + operation_messages={"approval create callback id": ""}, + drive=_fail_callback("com.acme.PaymentError", ""), + ), + Scenario("B14", _b14, {}, "PaymentError"), + Scenario("B15", _b15, {}, "WrappedPaymentError"), + Scenario( + "B16", + _b16, + {"lookup": "ValueError", "work": "ValidationError"}, + "ValidationError", + message="x", + ), + Scenario( + "B17", + _b17, + {"charge": "SerDesError"}, + "SerDesError", + message="Unsupported type: ", + ), + Scenario("B18", _b18, {"charge": "Error"}, "Error"), + Scenario( + "B20", + _b20, + {"map-item-0": "PaymentError", "map-item-1": "FraudError"}, + "PaymentError", + ), + Scenario( + "B21", + _b21, + {"parallel-branch-0": "PaymentError", "parallel-branch-1": "FraudError"}, + "PaymentError", + ), + Scenario( + "handled-in-subtree", + _handled_in_subtree, + {"outer": "PaymentError", "lookup": "ValueError", "charge": "PaymentError"}, + "PaymentError", + ), + # The core records the user's type at every level, so the depth of the + # operation tree is not an unwrap depth (rule 3 counts wrappers passed). + Scenario("depth-18", _nested(18), _depth_names(18), "PaymentError"), + Scenario( + "map-in-map-in-child", + _map_in_map_in_child, + {"checkout": "PaymentError", "map-item-0": "PaymentError"}, + "PaymentError", + ), +] + + +# -- runner ----------------------------------------------------------------------- + + +def _handler(scenario: Scenario, replay: bool) -> Callable[[Any, DurableContext], Any]: + body = scenario.fresh() if scenario.fresh is not None else scenario.body + + def handler(event: Any, ctx: DurableContext) -> Any: + if not replay: + return body(ctx) + try: + body(ctx) + except Exception as exc: # noqa: BLE001 - rethrown after the wait + saved = exc + else: # pragma: no cover - every scenario fails + return "no failure" + # A new invocation starts after the wait; the failure above replays from + # its checkpoint there and is rethrown. + ctx.wait(Duration.from_seconds(1), name="pause") + raise saved + + return handler + + +_TERMINAL = frozenset({"SUCCEEDED", "FAILED", "TIMED_OUT", "STOPPED"}) + + +def run_scenario(scenario: Scenario, replay: bool) -> dict[str, Any]: + capture = _Capture() + plugin = durable_execution_telemetry( + DurableExecutionTelemetryConfig( + exporters=[capture], operation_detail="full-tree" + ) + ) + handler = durable_execution(_handler(scenario, replay), plugins=[plugin]) + with DurableFunctionTestRunner(handler=handler, execution_timeout=60) as runner: + if scenario.setup is not None: + scenario.setup(runner) + arn = runner.run_async(input="{}", execution_timeout=60) + if scenario.drive is not None: + scenario.drive(runner, arn) + runner.wait_for_result(arn, timeout=60) + # Wait for the terminal record, not just the first one: in on-change mode + # earlier RUNNING records may arrive first. + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + with capture.lock: + terminal = [r for r in capture.records if r["status"] in _TERMINAL] + if terminal: + break + time.sleep(0.01) # pragma: no cover + assert terminal, [r["status"] for r in capture.records] + record = terminal[-1] + assert record["status"] == "FAILED" + return record + + +def failed_operations(record: dict[str, Any]) -> dict[str, dict[str, Any]]: + return { + op["name"]: op["error"] + for op in record["operations"] + if "error" in op and op["name"] != "pause" + } + + +def _assert_record(scenario: Scenario, record: dict[str, Any]) -> None: + assert record["error"]["name"] == scenario.execution + errors = failed_operations(record) + names = {name: err["name"] for name, err in errors.items()} + for name, expected in scenario.operations.items(): + assert names.get(name) == expected, (name, names) + assert record["error"]["message"] == scenario.message + messages = scenario.operation_messages or {} + for name in scenario.operations: + assert errors[name]["message"] == messages.get(name, scenario.message), name + # operationsByName carries the same name as operations[]. + # (A name used by several operations, such as nested map items, is skipped.) + by_name = LambdaLogExporter().render(record)["operationsByName"] + used = [op["name"] for op in record["operations"]] + for op in record["operations"]: + if "error" in op and used.count(op["name"]) == 1: + assert by_name[op["name"]]["error"]["name"] == op["error"]["name"] + + +@pytest.mark.parametrize("scenario", SCENARIOS, ids=[s.row for s in SCENARIOS]) +def test_table_b_live_and_replayed(scenario: Scenario) -> None: + live = run_scenario(scenario, replay=False) + replayed = run_scenario(scenario, replay=True) + _assert_record(scenario, live) + _assert_record(scenario, replayed) + assert replayed["error"] == live["error"] + live_names = {n: e["name"] for n, e in failed_operations(live).items()} + replayed_names = {n: e["name"] for n, e in failed_operations(replayed).items()} + assert replayed_names == live_names diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/e2e/record_alignment_int_test.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/e2e/record_alignment_int_test.py new file mode 100644 index 000000000..c1d2f9e91 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/e2e/record_alignment_int_test.py @@ -0,0 +1,154 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""End-to-end test: record fields that are aligned with the JS exporter. + +Drives the plugin through the LOCAL durable runner (``DurableFunctionTestRunner``) +and the real ``@durable_execution`` lifecycle, so the checkpointed values the +plugin decodes (step results, ErrorObjects) are produced by the SDK itself: + +* an opted-in step result is the plain JSON value, not the serializer envelope; +* ``error.name`` is the simple name of the user's error type, for a failure on + the first invocation and for one reached after a retry (a later invocation + that replays the checkpointed failure); +* a RUNNING record built at invocation end (retry suspension) carries + ``endTime``/``durationMs``, while start and operation-change records do not. +""" + +from __future__ import annotations + +from typing import Any + +from aws_durable_execution_sdk_python.config import Duration, StepConfig +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.execution import ( + InvocationStatus, + durable_execution, +) +from aws_durable_execution_sdk_python.retries import ( + JitterStrategy, + RetryStrategyConfig, + create_retry_strategy, +) + +from aws_durable_execution_sdk_python_telemetry_exporter import ( + ContentConfig, + ContentOperations, + DurableExecutionTelemetryConfig, + EmitMode, + OperationOverride, + durable_execution_telemetry, +) +from aws_durable_execution_sdk_python_testing.runner import ( + DurableFunctionTestResult, + DurableFunctionTestRunner, +) + + +class PaymentError(Exception): + """A user-defined error raised from step code.""" + + +class _CaptureExporter: + def __init__(self) -> None: + self.max_record_size_bytes: int | None = None + self.records: list[dict[str, Any]] = [] + + def render(self, record: dict[str, Any]) -> dict[str, Any]: + return record + + def export(self, record: dict[str, Any]) -> None: + self.records.append(record) + + def flush(self) -> None: + return None + + +def _run(handler_fn: Any, config: DurableExecutionTelemetryConfig) -> None: + handler = durable_execution( + handler_fn, plugins=[durable_execution_telemetry(config)] + ) + with DurableFunctionTestRunner(handler=handler, execution_timeout=30) as runner: + result: DurableFunctionTestResult = runner.run(input="{}") + assert result.status in (InvocationStatus.SUCCEEDED, InvocationStatus.FAILED) + + +def test_opted_in_result_is_plain_json() -> None: + def handler(event: Any, context: DurableContext) -> str: # noqa: ARG001 + context.step( + lambda _s: {"amount": 99.99, "items": [{"sku": "a"}]}, name="charge" + ) + return "done" + + capture = _CaptureExporter() + _run( + handler, + DurableExecutionTelemetryConfig( + exporters=[capture], + content=ContentConfig( + operations=ContentOperations( + overrides=[OperationOverride("charge", result=lambda r: r)] + ) + ), + ), + ) + (record,) = capture.records + (charge,) = [op for op in record["operations"] if op["name"] == "charge"] + assert charge["result"] == {"amount": 99.99, "items": [{"sku": "a"}]} + + +def _failing_handler(attempts: int) -> Any: + config = StepConfig( + retry_strategy=create_retry_strategy( + RetryStrategyConfig( + max_attempts=attempts, + initial_delay=Duration.from_seconds(1), + jitter_strategy=JitterStrategy.NONE, + ) + ) + ) + + def handler(event: Any, context: DurableContext) -> str: # noqa: ARG001 + def charge(_s: Any) -> str: + msg = "card declined" + raise PaymentError(msg) + + context.step(charge, name="charge", config=config) + return "unreachable" + + return handler + + +def _terminal_failure( + attempts: int, emit_mode: EmitMode = EmitMode.ON_COMPLETE +) -> tuple[dict[str, Any], list[dict[str, Any]]]: + capture = _CaptureExporter() + _run( + _failing_handler(attempts), + DurableExecutionTelemetryConfig(exporters=[capture], emit_mode=emit_mode), + ) + terminal = capture.records[-1] + assert terminal["status"] == "FAILED" + return terminal, capture.records + + +def test_error_name_is_user_type_on_first_run_and_after_replay() -> None: + first_run, _ = _terminal_failure(attempts=1) + after_retry, _ = _terminal_failure(attempts=2) + for record in (first_run, after_retry): + assert record["error"] == {"name": "PaymentError", "message": "card declined"} + (charge,) = record["operations"] + assert charge["error"] == {"name": "PaymentError", "message": "card declined"} + + +def test_running_record_at_retry_suspension_has_end_time() -> None: + _, records = _terminal_failure(attempts=2, emit_mode=EmitMode.ON_CHANGE) + running_with_end = [ + r for r in records if r["status"] == "RUNNING" and "endTime" in r + ] + # The first invocation ends PENDING while the step waits to retry; that + # invocation-end record is RUNNING with endTime/durationMs, like JS. + assert len(running_with_end) >= 1 + for record in running_with_end: + assert isinstance(record["durationMs"], int) + assert "output" not in record and "error" not in record diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/fixtures/error-name-vectors.json b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/fixtures/error-name-vectors.json new file mode 100644 index 000000000..bf3b77001 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/fixtures/error-name-vectors.json @@ -0,0 +1,106 @@ +{ + "description": "Durable Execution Telemetry error.name reduction vectors (spec version 2). Input is a type string; expected is the reported error.name. Blank means only the six ASCII whitespace characters (space, \\t, \\n, \\r, \\f, \\v); names are never trimmed. Shared byte-for-byte across the JS, Python and Java exporters. Do not edit in one repo only.", + "version": 2, + "reduce": [ + { + "id": "A1", + "input": "PaymentError", + "expected": "PaymentError" + }, + { + "id": "A2", + "input": "com.acme.PaymentError", + "expected": "PaymentError" + }, + { + "id": "A3", + "input": "com.acme.Outer$PaymentError", + "expected": "PaymentError" + }, + { + "id": "A4", + "input": "a.b.Outer$Inner$PaymentError", + "expected": "PaymentError" + }, + { + "id": "A5", + "input": "__main__.PaymentError", + "expected": "PaymentError" + }, + { + "id": "A6", + "input": "States.Timeout", + "expected": "Timeout" + }, + { + "id": "A7", + "input": "trailing.", + "expected": "trailing." + }, + { + "id": "A8", + "input": "Outer$", + "expected": "Outer$" + }, + { + "id": "A9a", + "input": ".", + "expected": "." + }, + { + "id": "A9b", + "input": "$", + "expected": "$" + }, + { + "id": "A10", + "input": "pkg.Foo$1Local", + "expected": "1Local" + }, + { + "id": "A11", + "input": " PaymentError ", + "expected": " PaymentError " + }, + { + "id": "A12a", + "input": "", + "expected": "Error" + }, + { + "id": "A12b", + "input": " ", + "expected": "Error" + }, + { + "id": "A12c", + "input": null, + "expected": "Error" + }, + { + "id": "A13", + "input": " \t\n\r\f\u000b", + "expected": "Error" + }, + { + "id": "A14", + "input": "\u00a0", + "expected": "\u00a0" + }, + { + "id": "A15", + "input": "\ufeff", + "expected": "\ufeff" + }, + { + "id": "A16", + "input": "\u001c", + "expected": "\u001c" + }, + { + "id": "A17", + "input": "\u2003", + "expected": "\u2003" + } + ] +} diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_error_name.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_error_name.py new file mode 100644 index 000000000..47c2d2294 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_error_name.py @@ -0,0 +1,400 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Unit tests for the cross-SDK ``error.name`` rule. + +Table A (type-string reduction) is driven from ``fixtures/error-name-vectors.json``, +which has the same content in the JS, Python and Java exporter repositories. +The core SDK checkpoints the error a wrapper wraps, so each name is the +reduction of that record's own ErrorType. The snapshot cases below build the +ErrorObjects the core writes (``ErrorObject.from_exception``), plus wrapper +types an older core wrote; ``tests/e2e/error_name_int_test.py`` runs Table B +through the local runner. +""" + +from __future__ import annotations + +import datetime +import json +import pathlib +from typing import Any + +import pytest +from aws_durable_execution_sdk_python.exceptions import ( + CallbackExternalError, + CallbackSubmitterError, + CallbackTimeoutError, + ChildContextError, + DurableOperationError, + InvokeError, + StepError, + WaitForConditionError, +) +from aws_durable_execution_sdk_python.lambda_service import ( + ErrorObject, + OperationStatus, + OperationSubType, +) +from aws_durable_execution_sdk_python.plugin import ( + InvocationEndInfo, + InvocationStartInfo, + InvocationStatus, + OperationInfo, + OperationType, +) + +from aws_durable_execution_sdk_python_telemetry_exporter import ( + ContentConfig, + ContentOperations, + DurableExecutionTelemetryConfig, + LambdaLogExporter, + durable_execution_telemetry, +) +from aws_durable_execution_sdk_python_telemetry_exporter.plugin import ( + _simple_type_name, +) + +ARN = "arn:aws:lambda:us-west-2:123456789012:function:my-fn:$LATEST/durable-execution/exec-1/inv-1" +T0 = datetime.datetime(2026, 1, 1, 0, 0, 0, tzinfo=datetime.UTC) +T1 = datetime.datetime(2026, 1, 1, 0, 0, 1, tzinfo=datetime.UTC) + +_VECTORS = json.loads( + (pathlib.Path(__file__).parent / "fixtures" / "error-name-vectors.json").read_text( + encoding="utf-8" + ) +) + + +# -- Table A --------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "vector", _VECTORS["reduce"], ids=[v["id"] for v in _VECTORS["reduce"]] +) +def test_table_a_reduction(vector: dict[str, Any]) -> None: + assert _simple_type_name(vector["input"]) == vector["expected"] + + +@pytest.mark.parametrize( + "vector", _VECTORS["reduce"], ids=[v["id"] for v in _VECTORS["reduce"]] +) +def test_table_a_applies_to_execution_and_operation_records( + vector: dict[str, Any], +) -> None: + err = ErrorObject(message="m", type=vector["input"], data=None, stack_trace=None) + rec, rendered = _record([_op("op", error=err)], err) + assert rec["error"] == {"name": vector["expected"], "message": "m"} + assert rec["operations"][0]["error"] == {"name": vector["expected"], "message": "m"} + assert rendered["operationsByName"]["op"]["error"]["name"] == vector["expected"] + + +def test_vectors_file_is_the_shared_version() -> None: + assert _VECTORS["version"] == 2 + assert [v["id"] for v in _VECTORS["reduce"]][:2] == ["A1", "A2"] + + +# -- helpers --------------------------------------------------------------------- + + +class _Capture: + def __init__(self) -> None: + self.max_record_size_bytes: int | None = None + self.records: list[dict[str, Any]] = [] + + def render(self, record: dict[str, Any]) -> Any: + return record + + def export(self, record: dict[str, Any]) -> None: + self.records.append(record) + + def flush(self) -> None: + return None + + +def _op( + name: str, + *, + error: ErrorObject | None = None, + parent: str | None = None, + op_type: OperationType = OperationType.STEP, + status: OperationStatus | None = None, + sub_type: OperationSubType | None = None, +) -> OperationInfo: + return OperationInfo( + operation_id=name, + operation_type=op_type, + sub_type=sub_type, + name=name, + parent_id=parent, + start_time=T0, + is_replayed=False, + status=status + or (OperationStatus.FAILED if error is not None else OperationStatus.SUCCEEDED), + end_time=T1, + error=error, + ) + + +def _ctx(name: str, **kwargs: Any) -> OperationInfo: + return _op(name, op_type=OperationType.CONTEXT, **kwargs) + + +def _err(type_name: str | None, message: str | None = "card declined") -> ErrorObject: + return ErrorObject(message=message, type=type_name, data=None, stack_trace=None) + + +def _record( + ops: list[OperationInfo], + execution_error: ErrorObject | None, + **config: Any, +) -> tuple[dict[str, Any], dict[str, Any]]: + capture = _Capture() + plugin = durable_execution_telemetry( + DurableExecutionTelemetryConfig( + exporters=[capture], operation_detail="full-tree", **config + ) + ) + operations = {op.operation_id: op for op in ops} + plugin.on_invocation_start( + InvocationStartInfo( + request_id=None, + execution_arn=ARN, + is_first_invocation=True, + execution_start_time=T0, + execution_input="{}", + operations=operations, + ) + ) + plugin.on_invocation_end( + InvocationEndInfo( + request_id=None, + execution_arn=ARN, + is_first_invocation=True, + execution_start_time=T0, + status=InvocationStatus.FAILED, + error=execution_error, + execution_result=None, + operations=operations, + ) + ) + (rec,) = capture.records + return rec, LambdaLogExporter().render(rec) + + +def _names(rec: dict[str, Any]) -> dict[str, Any]: + return {o["name"]: o["error"]["name"] for o in rec["operations"] if "error" in o} + + +PAYMENT = _err("__main__.PaymentError") + + +def _wrapper_type(cls: type) -> str: + return f"{cls.__module__}.{cls.__qualname__}" + + +def _recorded(exc: Exception) -> ErrorObject: + """The ErrorObject the core checkpoints for ``exc``.""" + return ErrorObject.from_exception(exc) + + +def _wrapped(error: ErrorObject, wrapper: type[DurableOperationError]) -> Exception: + """The wrapper the core raises when ``error`` escapes an operation.""" + try: + error.raise_as_operation_error(wrapper) + except DurableOperationError as exc: + return exc + raise AssertionError("raise_as_operation_error must raise") # pragma: no cover + + +# -- which error is named ------------------------------------------------------ + + +def test_step_user_error_names_operation_and_execution() -> None: + execution = _recorded(_wrapped(PAYMENT, StepError)) + rec, rendered = _record([_op("charge", error=PAYMENT)], execution) + assert rec["error"] == {"name": "PaymentError", "message": "card declined"} + assert _names(rec) == {"charge": "PaymentError"} + assert rendered["operationsByName"]["charge"]["error"]["name"] == "PaymentError" + + +@pytest.mark.parametrize( + "wrapper", + [ + StepError, + ChildContextError, + InvokeError, + WaitForConditionError, + CallbackSubmitterError, + CallbackExternalError, + DurableOperationError, + ], +) +def test_the_core_records_the_wrapped_type( + wrapper: type[DurableOperationError], +) -> None: + # A context the wrapper escaped, and the execution, carry the user's type. + recorded = _recorded(_wrapped(PAYMENT, wrapper)) + ops = [ + _ctx("checkout", error=recorded), + _op("charge", parent="checkout", error=PAYMENT), + ] + rec, _ = _record(ops, recorded) + assert rec["error"]["name"] == "PaymentError" + assert _names(rec) == {"checkout": "PaymentError", "charge": "PaymentError"} + + +def test_each_operation_is_named_by_its_own_type() -> None: + # B16: no matching by message; each record reduces its own type. + ops = [ + _op("lookup", error=_err("ValueError", "x")), + _ctx( + "work", + error=_err( + "aws_durable_execution_sdk_python.exceptions.ValidationError", "x" + ), + ), + ] + rec, _ = _record(ops, ops[1].error) + assert _names(rec) == {"lookup": "ValueError", "work": "ValidationError"} + assert rec["error"]["name"] == "ValidationError" + + +@pytest.mark.parametrize("sub_type", [OperationSubType.MAP, OperationSubType.PARALLEL]) +def test_b20_b21_items_with_the_same_message_keep_their_types( + sub_type: OperationSubType, +) -> None: + fraud = _err("__main__.FraudError") + ops = [ + _ctx("fanout", sub_type=sub_type), + _ctx("item-0", parent="fanout", error=PAYMENT), + _ctx("item-1", parent="fanout", error=fraud), + ] + # The execution carries the error throw_if_error raised. + execution = _recorded(_wrapped(PAYMENT, ChildContextError)) + rec, _ = _record(ops, execution) + assert _names(rec) == {"item-0": "PaymentError", "item-1": "FraudError"} + assert rec["error"]["name"] == "PaymentError" + + +def test_user_subclass_of_a_wrapper_is_the_users_name() -> None: + # B15: the core does not unwrap a user subclass; neither does the exporter. + err = _err("__main__.WrappedPaymentError") + rec, _ = _record([], err) + assert rec["error"]["name"] == "WrappedPaymentError" + + +# -- checkpoints written by an older core (rule 9) --------------------------------- + + +@pytest.mark.parametrize( + "wrapper", + [StepError, ChildContextError, InvokeError, CallbackSubmitterError], +) +def test_an_old_wrapper_type_reports_the_wrapper( + wrapper: type[DurableOperationError], +) -> None: + old = _err(_wrapper_type(wrapper)) + # A child that holds the user's type is never used to guess the cause. + ops = [_ctx("checkout", error=old), _op("charge", parent="checkout", error=PAYMENT)] + rec, _ = _record(ops, old) + assert rec["error"]["name"] == wrapper.__name__ + assert _names(rec) == {"checkout": wrapper.__name__, "charge": "PaymentError"} + + +# -- callbacks ------------------------------------------------------------------- + + +def _callback_ops( + callback_error: ErrorObject | None, + status: OperationStatus, + context_error: ErrorObject, +) -> list[OperationInfo]: + return [ + _ctx("approval", error=context_error), + _op( + "approval create callback id", + op_type=OperationType.CALLBACK, + parent="approval", + status=status, + error=callback_error, + ), + ] + + +def _external(error_type: str | None) -> ErrorObject: + """What the core records for a callback failed externally with ``error_type``.""" + return _recorded(CallbackExternalError("card declined", error_type=error_type)) + + +def test_b11_callback_timeout_is_named_consistently() -> None: + service = _err("Callback.Timeout", "Callback timed out") + timeout = _recorded(CallbackTimeoutError("Callback timed out")) + ops = _callback_ops(service, OperationStatus.TIMED_OUT, timeout) + rec, _ = _record(ops, timeout) + assert rec["error"]["name"] == "CallbackTimeoutError" + assert _names(rec) == { + "approval": "CallbackTimeoutError", + "approval create callback id": "CallbackTimeoutError", + } + + +def test_b12_external_failure_with_type_is_the_senders_type() -> None: + sender = _err("com.acme.PaymentError") + recorded = _external("com.acme.PaymentError") + rec, _ = _record(_callback_ops(sender, OperationStatus.FAILED, recorded), recorded) + assert rec["error"]["name"] == "PaymentError" + assert _names(rec) == { + "approval": "PaymentError", + "approval create callback id": "PaymentError", + } + + +def test_b13_external_failure_without_type_is_the_sdk_callback_class() -> None: + recorded = _external(None) + rec, _ = _record( + _callback_ops(_err(None), OperationStatus.FAILED, recorded), recorded + ) + assert rec["error"]["name"] == "CallbackExternalError" + assert _names(rec) == { + "approval": "CallbackExternalError", + "approval create callback id": "CallbackExternalError", + } + + +def test_b13b_external_failure_with_type_error_is_error() -> None: + recorded = _external("Error") + rec, _ = _record( + _callback_ops(_err("Error"), OperationStatus.FAILED, recorded), recorded + ) + assert rec["error"]["name"] == "Error" + assert _names(rec) == {"approval": "Error", "approval create callback id": "Error"} + + +@pytest.mark.parametrize("blank", ["", " ", " \t\n\r\f\v"]) +def test_external_failure_with_blank_type_is_the_sdk_callback_class(blank: str) -> None: + recorded = _external(None) + ops = _callback_ops(_err(blank), OperationStatus.FAILED, recorded) + rec, _ = _record(ops, recorded) + assert rec["error"]["name"] == "CallbackExternalError" + assert _names(rec)["approval create callback id"] == "CallbackExternalError" + + +def test_external_failure_with_non_ascii_space_type_is_kept() -> None: + recorded = _external("\u00a0") + ops = _callback_ops(_err("\u00a0"), OperationStatus.FAILED, recorded) + rec, _ = _record(ops, recorded) + assert rec["error"]["name"] == "\u00a0" + assert _names(rec)["approval create callback id"] == "\u00a0" + + +# -- options ----------------------------------------------------------------------- + + +def test_execution_name_ignores_include_errors_false() -> None: + rec, _ = _record( + [_op("charge", error=PAYMENT)], + _recorded(_wrapped(PAYMENT, StepError)), + content=ContentConfig(operations=ContentOperations(include_errors=False)), + ) + assert rec["error"]["name"] == "PaymentError" + assert "error" not in rec["operations"][0] diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_exporters.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_exporters.py index 87024608f..9539a555b 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_exporters.py +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_exporters.py @@ -11,6 +11,7 @@ from __future__ import annotations import json +import re from typing import Any import pytest @@ -20,6 +21,12 @@ S3Exporter, S3Partitioning, ) +from aws_durable_execution_sdk_python_telemetry_exporter.exporters._common import ( + compact_dumps, +) +from aws_durable_execution_sdk_python_telemetry_exporter.truncation import ( + truncate_record, +) from aws_durable_execution_sdk_python_telemetry_exporter.exporters import ( LambdaLogExporter as LambdaLogExporterFromPkg, ) @@ -96,14 +103,62 @@ def test_lambda_log_default_size_and_render_is_operations_by_name(): assert shaped["operationsByName"]["greet"]["count"] == 1 -def test_lambda_log_export_prints_compact_json_line(capsys): +def test_lambda_log_export_prints_lambda_json_log_envelope(capsys): LambdaLogExporter().export(_record()) - out = capsys.readouterr().out.strip() + out = capsys.readouterr().out + lines = out.splitlines() + assert len(lines) == 1 + line = lines[0] # single compact JSON line, no whitespace separators - assert ", " not in out and '": ' not in out - parsed = json.loads(out) - assert parsed["recordType"] == "DurableExecutionTelemetry" - assert "operationsByName" in parsed + assert ", " not in line and '": ' not in line + envelope = json.loads(line) + # Same keys, same order as the Lambda runtime's JSON envelope (and the Java + # exporter): timestamp, level, [requestId], message. + assert list(envelope) == ["timestamp", "level", "message"] + assert re.fullmatch( + r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z", envelope["timestamp"] + ) + assert envelope["level"] == "INFO" + # message is the compact record JSON as a string, and parses to the record. + message = envelope["message"] + assert isinstance(message, str) + assert ", " not in message and '": ' not in message + exporter = LambdaLogExporter() + assert json.loads(message) == exporter.render(_record()) + assert json.loads(message)["recordType"] == "DurableExecutionTelemetry" + assert "operationsByName" in json.loads(message) + + +def test_lambda_log_line_matches_the_reader_filter(capsys): + # The lambda-log-exporter reader always starts with: + # filter level = "INFO" and message like /DurableExecutionTelemetry/ + # then parses fields out of the message string. + LambdaLogExporter().export( + _record(status="FAILED", error={"name": "E", "message": "m"}) + ) + envelope = json.loads(capsys.readouterr().out) + assert envelope["level"] == "INFO" + assert re.search(r"DurableExecutionTelemetry", envelope["message"]) + assert '"status":"FAILED"' in envelope["message"] + assert '"error":{"name":"E"' in envelope["message"] + + +def test_lambda_log_size_limit_applies_to_the_record_not_the_envelope(capsys): + record = _record(input={"blob": "x" * 500}) + # A limit exactly the size of the rendered record: the record fits as-is. + limit = len(compact_dumps(LambdaLogExporter().render(record)).encode("utf-8")) + exporter = LambdaLogExporter(max_record_size_bytes=limit) + shaped = truncate_record(record, exporter.max_record_size_bytes, exporter.render) + assert "truncated" not in shaped + exporter.export(shaped) + line = capsys.readouterr().out.rstrip("\n") + message = json.loads(line)["message"] + assert len(message.encode("utf-8")) == limit + # The envelope keys and the escaping of message add bytes on top. + assert len(line.encode("utf-8")) > limit + # One byte less and the same record is truncated (the record is measured). + smaller = LambdaLogExporter(max_record_size_bytes=limit - 1) + assert truncate_record(record, limit - 1, smaller.render).get("truncated") is True def test_lambda_log_custom_size_and_flush_noop(): diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_plugin.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_plugin.py index 3e5086031..e80a79e60 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_plugin.py +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_plugin.py @@ -13,10 +13,21 @@ from __future__ import annotations import datetime +import decimal +import json +import math import threading +import time +import uuid from typing import Any import pytest +from aws_durable_execution_sdk_python.concurrency.models import ( + BatchItem, + BatchResult, + CompletionReason, +) +from aws_durable_execution_sdk_python.config import BatchItemStatus from aws_durable_execution_sdk_python.lambda_service import ( ErrorObject, OperationStatus, @@ -31,6 +42,10 @@ OperationInfo, OperationType, ) +from aws_durable_execution_sdk_python.serdes import ( + EXTENDED_TYPES_SERDES, + SerDesContext, +) from aws_durable_execution_sdk_python_telemetry_exporter import ( ContentConfig, @@ -40,6 +55,9 @@ DurableExecutionTelemetryConfig, durable_execution_telemetry, ) +from aws_durable_execution_sdk_python_telemetry_exporter import ( + plugin as plugin_module, +) from aws_durable_execution_sdk_python_telemetry_exporter.plugin import ( _resolve_sampling_rate, ) @@ -591,12 +609,14 @@ def test_default_exporter_actually_emits_to_stdout(capsys): plugin = durable_execution_telemetry(DurableExecutionTelemetryConfig()) _run(plugin, ops=[_step("greet")]) out = capsys.readouterr().out - # compact JSON via LambdaLogExporter - assert '"recordType":"DurableExecutionTelemetry"' in out - assert '"operationsByName"' in out + # Lambda JSON log envelope whose message is the compact record JSON. + line = json.loads(out.strip()) + record = json.loads(line["message"]) + assert record["recordType"] == "DurableExecutionTelemetry" + assert "operationsByName" in record -# -- terminal timestamps: end time only for SUCCEEDED/FAILED (comment 3) ------ +# -- endTime/durationMs: every invocation-end record carries them (JS parity) - def _last_record_for_end_status(status, *, emit_mode="on-change", result=None): @@ -616,18 +636,46 @@ def _last_record_for_end_status(status, *, emit_mode="on-change", result=None): return exporter.records[-1] -def test_pending_end_maps_to_running_and_omits_end_time(): +def test_pending_end_maps_to_running_with_end_time_and_duration(): + # JS onInvocationEnd sets endTime for every status, so a suspended + # invocation's RUNNING record carries endTime/durationMs up to the suspend. rec = _last_record_for_end_status(InvocationStatus.PENDING) assert rec["status"] == "RUNNING" - assert "endTime" not in rec - assert "durationMs" not in rec + assert rec["endTime"].endswith("Z") + assert isinstance(rec["durationMs"], int) and rec["durationMs"] >= 0 + assert "output" not in rec and "error" not in rec -def test_retry_end_maps_to_running_and_omits_end_time(): +def test_retry_end_maps_to_running_with_end_time_and_duration(): rec = _last_record_for_end_status(InvocationStatus.RETRY) assert rec["status"] == "RUNNING" - assert "endTime" not in rec - assert "durationMs" not in rec + assert rec["endTime"].endswith("Z") + assert isinstance(rec["durationMs"], int) and rec["durationMs"] >= 0 + assert "output" not in rec and "error" not in rec + + +def test_start_and_operation_change_records_have_no_end_time(): + exporter = CaptureExporter() + plugin = durable_execution_telemetry( + DurableExecutionTelemetryConfig(exporters=[exporter], emit_mode="on-change") + ) + op = _step("s", op_id="1") + plugin.on_invocation_start(_start(operations={})) + plugin._scheduler.drain() + plugin.on_operation_change( + OperationChangeInfo( + execution_arn=ARN, updated_operations=_ops(op), operations=_ops(op) + ) + ) + plugin._scheduler.drain() + plugin.on_invocation_end( + _end(operations=_ops(op), status=InvocationStatus.PENDING, result=None) + ) + assert [r["status"] for r in exporter.records] == ["RUNNING"] * 3 + start_rec, change_rec, end_rec = exporter.records + for rec in (start_rec, change_rec): + assert "endTime" not in rec and "durationMs" not in rec + assert "endTime" in end_rec and "durationMs" in end_rec def test_succeeded_end_is_terminal_with_end_time_and_duration(): @@ -661,3 +709,252 @@ def test_failed_end_is_terminal_with_end_time_and_duration(): assert rec["endTime"] is not None assert rec["durationMs"] is not None assert rec["error"]["name"] == "StepError" + + +# -- opted-in operation results are the plain JSON value (JS parity) --------- + + +def _serialize(value: Any) -> str: + """Checkpoint a step result exactly as the SDK's default serializer does.""" + return EXTENDED_TYPES_SERDES.serialize(value, SerDesContext("1", ARN)) + + +def _override_plugin(exporter, transform=lambda r: r, name="charge"): + return durable_execution_telemetry( + DurableExecutionTelemetryConfig( + exporters=[exporter], + content=ContentConfig( + operations=ContentOperations( + overrides=[OperationOverride(name, result=transform)] + ) + ), + ) + ) + + +def _charge_result(raw: str | None, transform=lambda r: r) -> Any: + exporter = CaptureExporter() + _run( + _override_plugin(exporter, transform), + ops=[_step("charge", result=raw)], + ) + op = exporter.records[0]["operations"][0] + return op.get("result", "") + + +def test_result_envelope_is_decoded_to_plain_json(): + raw = _serialize({"amount": 99.99}) + assert json.loads(raw) == {"t": "m", "v": {"amount": {"t": "f", "v": 99.99}}} + assert _charge_result(raw) == {"amount": 99.99} + + +def test_override_transform_receives_the_plain_value(): + seen: list[Any] = [] + + def transform(value: Any) -> Any: + seen.append(value) + return {"masked": value["amount"] > 50} + + result = _charge_result(_serialize({"amount": 99.99}), transform) + assert seen == [{"amount": 99.99}] + assert result == {"masked": True} + + +def test_result_non_json_types_become_json_compatible_values(): + value = { + "price": decimal.Decimal("99.99"), + "qty": decimal.Decimal("3"), + "nan": decimal.Decimal("NaN"), + "at": datetime.datetime(2026, 1, 2, 3, 4, 5, 123456, tzinfo=datetime.UTC), + "day": datetime.date(2026, 1, 2), + "id": uuid.UUID("12345678-1234-5678-1234-567812345678"), + "blob": b"\x00\x01", + "pair": (1, "a"), + "nested": [{"k": [1, {"deep": True}]}, None], + } + assert _charge_result(_serialize(value)) == { + "price": 99.99, + "qty": 3, + "nan": None, + "at": "2026-01-02T03:04:05.123456+00:00", + "day": "2026-01-02", + "id": "12345678-1234-5678-1234-567812345678", + "blob": "AAE=", + "pair": [1, "a"], + "nested": [{"k": [1, {"deep": True}]}, None], + } + + +def test_result_top_level_non_json_type_is_converted(): + assert _charge_result(_serialize(decimal.Decimal("12.5"))) == 12.5 + assert _charge_result(_serialize(("x", 2))) == ["x", 2] + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + # A float is used only when it converts back to exactly the same Decimal. + (decimal.Decimal("99.99"), 99.99), + (decimal.Decimal("-0.5"), -0.5), + (decimal.Decimal("3"), 3), + (decimal.Decimal("12345678901234567890"), 12345678901234567890), + # Otherwise the exact string form, so precision and sign are kept. + ( + decimal.Decimal("3.14159265358979323846264338327950288"), + "3.14159265358979323846264338327950288", + ), + (decimal.Decimal("12345678901234567.89"), "12345678901234567.89"), + (decimal.Decimal("1E-400"), "1E-400"), + (decimal.Decimal("-0.0"), -0.0), + (decimal.Decimal("-Infinity"), None), + (decimal.Decimal("sNaN"), None), + # The numeric value is kept, not the scale. + (decimal.Decimal("1.10"), 1.1), + (decimal.Decimal("1E+2"), 100), + # An integral Decimal of up to 300 digits is an integer; a longer one + # keeps its string form. + (decimal.Decimal("9" * 300), int("9" * 300)), + (decimal.Decimal("1" + "0" * 300), 1e300), + (decimal.Decimal("1" * 301), "1" * 301), + ], +) +def test_result_decimal_keeps_its_exact_value(value, expected): + result = _charge_result(_serialize({"d": value}))["d"] + assert result == expected + assert type(result) is type(expected) + if isinstance(expected, float): + assert math.copysign(1, result) == math.copysign(1, expected) + + +@pytest.mark.parametrize( + "raw", + [ + _serialize({"x": float("nan"), "y": float("inf"), "z": [float("-inf"), 1.5]}), + '{"x": NaN, "y": Infinity, "z": [-Infinity, 1.5]}', # plain JSON payload + ], +) +def test_result_non_finite_floats_become_null(raw): + exporter = CaptureExporter(render=LambdaLogExporter().render) + _run(_override_plugin(exporter), ops=[_step("charge", result=raw)]) + result = exporter.records[0]["operations"][0]["result"] + assert result == {"x": None, "y": None, "z": [None, 1.5]} + # The emitted line is strict JSON. + line = json.dumps(exporter.render(exporter.records[0]), allow_nan=False) + assert "NaN" not in line and "Infinity" not in line + + +@pytest.mark.parametrize("exponent", [5000, 1_000_000]) +def test_result_decimal_with_huge_exponent_is_fast_and_emitted(exponent): + value = decimal.Decimal(f"1E+{exponent}") + exporter = CaptureExporter(render=LambdaLogExporter().render) + start = time.perf_counter() + _run( + _override_plugin(exporter), + ops=[_step("charge", result=_serialize({"x": value}))], + ) + line = json.dumps(exporter.render(exporter.records[0]), allow_nan=False) + elapsed = time.perf_counter() - start + assert exporter.records[0]["operations"][0]["result"] == {"x": f"1E+{exponent}"} + assert json.loads(line) + # Converting to int would take tens of seconds for the larger value. + assert elapsed < 1 + + +def test_result_batch_result_becomes_its_dict_form(): + batch = BatchResult( + all=[BatchItem(index=0, status=BatchItemStatus.SUCCEEDED, result={"a": 1})], + completion_reason=CompletionReason.ALL_COMPLETED, + ) + assert _charge_result(_serialize(batch)) == { + "all": [{"index": 0, "status": "SUCCEEDED", "result": {"a": 1}, "error": None}], + "completionReason": "ALL_COMPLETED", + } + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ("42", 42), # primitives are plain JSON already + ('"hi"', "hi"), + ("[1,2,3]", [1, 2, 3]), + ('{"amount":1}', {"amount": 1}), # e.g. a custom JSON SerDes + ('{"t":"m","v":{},"x":1}', {"t": "m", "v": {}, "x": 1}), # not an envelope + ("not json", "not json"), # raw string fallback + ], +) +def test_result_plain_payloads_keep_previous_behaviour(raw, expected): + assert _charge_result(raw) == expected + + +@pytest.mark.parametrize( + "raw", + [ + '{"t":"zz","v":1}', # unknown tag + '{"t":"d","v":"not-a-decimal"}', # decode failure + '{"t":"m","v":[1,2]}', # wrong payload for the tag + '{"t":1,"v":2}', # non-string tag + ], +) +def test_result_undecodable_envelope_falls_back_to_parsed_json(raw): + assert _charge_result(raw) == json.loads(raw) + + +def test_result_unsupported_decoded_type_falls_back_to_parsed_json(monkeypatch): + class _OpaqueSerDes: + def deserialize(self, data: str, context: Any) -> Any: + return object() + + monkeypatch.setattr(plugin_module, "EXTENDED_TYPES_SERDES", _OpaqueSerDes()) + raw = '{"t":"m","v":{}}' + assert _charge_result(raw) == {"t": "m", "v": {}} + + +def test_result_deeply_nested_plain_json_is_converted(): + # 600 nested arrays parse fine but used to exceed the recursion limit on + # conversion; the conversion is iterative. + raw = "[" * 600 + "]" * 600 + assert json.dumps(_charge_result(raw)) == raw + + +def test_json_conversion_does_not_recurse(): + value: Any = 1.5 + for _ in range(20_000): + value = [value] + converted = plugin_module._to_json_compatible(value) # noqa: SLF001 + for _ in range(20_000): + converted = converted[0] + assert converted == 1.5 + + +def test_json_conversion_keeps_dict_order_and_last_duplicate_key(): + assert plugin_module._to_json_compatible( # noqa: SLF001 + {"b": (1, 2), 1: "first", "1": "second"} + ) == {"b": [1, 2], "1": "second"} + assert list(plugin_module._to_json_compatible({"b": 1, "a": 2})) == ["b", "a"] # noqa: SLF001 + + +def test_result_normalization_failure_is_isolated_to_that_result(monkeypatch): + def fail(_value: Any) -> Any: + raise RecursionError("too deep") + + monkeypatch.setattr(plugin_module, "_to_json_compatible", fail) + exporter = CaptureExporter() + _run( + _override_plugin(exporter), + ops=[_step("charge", result="[[1]]"), _step("other", result="[2]")], + ) + (record,) = exporter.records + by_name = {op["name"]: op for op in record["operations"]} + # The failing result falls back to its raw string; the record is emitted. + assert by_name["charge"]["result"] == "[[1]]" + assert set(by_name) == {"charge", "other"} + + +def test_result_in_operations_by_name_is_plain(): + exporter = CaptureExporter(render=LambdaLogExporter().render) + _run( + _override_plugin(exporter), + ops=[_step("charge", result=_serialize({"amount": 99.99}))], + ) + rendered = exporter.render(exporter.records[0]) + assert rendered["operationsByName"]["charge"]["result"] == {"amount": 99.99}