From b8ab8a115fabc04a6f86f45547d8147f09784c1e Mon Sep 17 00:00:00 2001 From: Pooya Paridel Date: Thu, 22 Oct 2026 10:24:03 -0700 Subject: [PATCH 1/6] fix(telemetry): write error shape, plain results and log envelope Write every record error as {name, message} in that key order. The plugin hooks carry only ErrorObject, which has no cause, so Python records never have a cause. Names are unchanged. Decode an opted-in operation result that is the default ExtendedTypeSerDes envelope into plain JSON values. A result that fails to decode keeps its parsed or raw value. LambdaLogExporter now prints the Lambda JSON log envelope {"timestamp","level":"INFO","message":""}. --- .../README.md | 72 +++- .../exporters/lambda_log_exporter.py | 41 +- .../plugin.py | 139 ++++++- .../tests/test_error_and_result_format.py | 352 ++++++++++++++++++ .../tests/test_exporters.py | 34 +- .../tests/test_plugin.py | 10 +- 6 files changed, 621 insertions(+), 27 deletions(-) create mode 100644 packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_error_and_result_format.py 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 759adcb83..235301904 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/README.md +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/README.md @@ -41,8 +41,8 @@ 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 as single JSON lines (the `LambdaLogExporter` default, see +below), carrying the name-keyed `operationsByName` summary. The `S3Exporter` writes the lossless per-occurrence `operations` array, one object per execution (upsert-by-execution-name, so re-emission overwrites rather than appends). @@ -74,6 +74,30 @@ 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, which Lambda sends to the function's own +log group. The line is the Lambda JSON log envelope, with keys in this order: + +```json +{"timestamp":"2026-01-01T00:00:01.234Z","level":"INFO","message":""} +``` + +`timestamp` is UTC with millisecond precision and a `Z` suffix. `message` is +the compact record JSON as a string; parse it to get the record. The exporter +writes the envelope itself because Lambda captures Python `print` output as +plain text, in both the plain text and the JSON log format, and Lambda does not +double-encode output that is already JSON (see +[Log and monitor Python Lambda functions](https://docs.aws.amazon.com/lambda/latest/dg/python-logging.html) +and [Lambda log formats](https://docs.aws.amazon.com/lambda/latest/dg/monitoring-cloudwatchlogs-logformat.html)). + +`max_record_size_bytes` (default 256 KB) limits the record in `message`. The +JSON escaping of `message` can make the line up to about 2x the record; a +CloudWatch Logs event can be up to 1 MB +([PutLogEvents](https://docs.aws.amazon.com/AmazonCloudWatchLogs/latest/APIReference/API_PutLogEvents.html)). +The lines have `level` `INFO`, so Lambda drops them when the function's +application log level is `WARN` or above. + ### CloudWatchLogsExporter Writes one `PutLogEvents` event per record to `log_group_name`, in a stream @@ -223,6 +247,50 @@ Behavior is validated cross-SDK by the `telemetry` conformance suite > drains every pending snapshot and flushes exporters before it returns; > invocations that emit nothing do not start or flush the worker. +## Error names and causes + +Every error in a record (the execution `error`, each `operations[].error`, and +each `operationsByName..error`) has the shape +`{"name": ..., "message": ..., "cause": ...}`, with keys in that order. `name` +is the SDK's error type, unchanged, and `message` is the error message. `cause` +has the same shape, recursively, and is present only when the SDK provides a +cause. + +In Python, the plugin hooks carry only an `ErrorObject` (type, message, data, +stack trace) with no cause, so Python records have `name` and `message` and +never have `cause`. The exporter does not build a cause from other operations +or from `ErrorData`. `include_errors=False` removes operation errors. + +Names are each language's own and differ across SDKs. For example, a step that +throws `PaymentError` can be reported as `StepError` in JS, +`aws_durable_execution_sdk_python.exceptions.StepError` or `PaymentError` in +Python, and `StepFailedException` or `PaymentError` in Java. Query error names +per SDK, or query on `message` and the `cause` chain. + +## Operation results + +An opted-in operation `result` (`OperationOverride(result=...)`) is passed to +the transform as a plain JSON value. When the checkpointed string is the +default serializer's (`ExtendedTypeSerDes`) envelope, a JSON object whose keys +are exactly `t` and `v` with a string `t`, it is decoded with that serializer +and converted to plain JSON: + +- `Decimal`: an integer when it is integral and `adjusted() < 300`; otherwise a + number when the float round-trips exactly; otherwise `str(value)` (for + example `"12345678901234567.89"` or `"1E+5000"`). NaN and Infinity become + `null`. +- `datetime` and `date`: `isoformat()`. `UUID`: string. `bytes`: base64. + `tuple`: list. `BatchResult`: `to_dict()`. + +Any other result is parsed as JSON, or kept as the raw string when it is not +JSON. A result that fails to decode keeps its parsed or raw value, and the rest +of the record is still emitted. + +The hooks do not say which serializer wrote a result, so the envelope is +recognized by shape alone. A result written by `JsonSerDes` or a custom +`SerDes` that has this exact shape is decoded too: `{"t": "s", "v": "x"}` is +reported as `"x"`. + ## Requirements - `aws-durable-execution-sdk-python` with the plugin invocation hooks that 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..c40db253a 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,16 @@ from __future__ import annotations +import base64 import datetime +import decimal import json import math import threading -from typing import Any, Callable +import uuid +from typing import TYPE_CHECKING, Any, Callable +from aws_durable_execution_sdk_python.concurrency.models import BatchResult from aws_durable_execution_sdk_python.plugin import ( DurableInstrumentationPlugin, InvocationEndInfo, @@ -46,6 +50,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, @@ -62,6 +70,9 @@ DurableExecutionTelemetryConfig, ) +if TYPE_CHECKING: + from aws_durable_execution_sdk_python.lambda_service import ErrorObject + # Maps the SDK invocation status onto the record status. A durable execution # suspends (PENDING) while waiting; from the execution's point of view it is @@ -152,16 +163,130 @@ 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 as a ``{"t": ..., "v": + ...}`` envelope. An envelope is decoded with that default serializer and + converted to plain JSON values (see ``_to_json_compatible``). A payload that + is not an envelope is parsed JSON, or the raw string when it is not JSON. An + envelope that fails to decode falls back to the parsed payload. Never raises. + + The hooks do not say which serializer wrote a result, so the envelope is + recognised by shape alone: a result written by ``JsonSerDes`` or a custom + ``SerDes`` that has this exact shape is decoded too. + """ + try: + parsed = json.loads(raw_result) + except (ValueError, TypeError, RecursionError): + return raw_result + try: + if _is_extended_type_envelope(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 + pass + 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 while its leading digit's +# exponent is below this bound; larger values never build a huge int. +_MAX_INT_ADJUSTED = 300 + + +def _to_json_compatible(value: Any) -> Any: + """Convert a decoded result into plain JSON values, without recursion. + + ``Decimal``: an int when integral and ``adjusted() < 300``, else a float + when it round-trips exactly, else ``str(value)``; NaN and Infinity become + ``None``. ``datetime`` / ``date`` -> ``isoformat()``, ``UUID`` -> string, + ``bytes`` -> base64, ``tuple`` -> list, ``BatchResult`` -> ``to_dict()``. + Non-finite floats become ``None``. + """ + root: list[Any] = [None] + # (value, target container, key in target) + 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 enumerate(item) + ) + elif isinstance(item, dict): + converted_dict: dict[str, Any] = {} + target[key] = converted_dict + for name, child in item.items(): + converted_dict[str(name)] = None + pending.append((child, converted_dict, str(name))) + 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 checked first so Decimal("1E+1000000") never builds a + # million-digit int. + if value.adjusted() < _MAX_INT_ADJUSTED and value == value.to_integral_value(): + return int(value) + as_float = float(value) + if math.isfinite(as_float) and 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) + + +def _error_shape(error: ErrorObject) -> dict[str, Any]: + """Return an error as ``{name, message}`` in that key order. + + The plugin hooks carry an ``ErrorObject`` (type, message, data, stack + trace) and no cause, so records written by this SDK never have ``cause``. + The name is the SDK's ErrorType, unchanged. + """ + return {"name": error.type, "message": error.message} + + class _ExecutionState: __slots__ = ("start_time", "parsed_arn", "cached_input", "operations") @@ -369,7 +494,7 @@ 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"] = _error_shape(op.error) 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 +561,7 @@ 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"] = _error_shape(error) record["operations"] = self._build_operations(operations) self._scheduler.schedule(record) diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_error_and_result_format.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_error_and_result_format.py new file mode 100644 index 000000000..f9a1514af --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_error_and_result_format.py @@ -0,0 +1,352 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Error shape and opted-in operation result decoding in emitted records.""" + +from __future__ import annotations + +import base64 +import dataclasses +import datetime +import decimal +import json +import time +import uuid +from typing import Any + +import pytest +from aws_durable_execution_sdk_python.concurrency.models import ( + BatchItem, + BatchItemStatus, + BatchResult, + CompletionReason, +) +from aws_durable_execution_sdk_python.lambda_service import ( + ErrorObject, + OperationStatus, + OperationSubType, +) +from aws_durable_execution_sdk_python.plugin import ( + InvocationEndInfo, + InvocationStartInfo, + InvocationStatus, + OperationEndInfo, + OperationInfo, + OperationType, +) +from aws_durable_execution_sdk_python.serdes import EXTENDED_TYPES_SERDES + +from aws_durable_execution_sdk_python_telemetry_exporter import ( + ContentConfig, + ContentOperations, + DurableExecutionTelemetryConfig, + OperationOverride, + durable_execution_telemetry, +) +from aws_durable_execution_sdk_python_telemetry_exporter.operations_index import ( + with_operations_by_name, +) +from aws_durable_execution_sdk_python_telemetry_exporter.plugin import ( + _decode_checkpoint_result, + _to_json_compatible, +) + +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) +STEP_ERROR = "aws_durable_execution_sdk_python.exceptions.StepError" + + +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]) -> Any: # noqa: PLR6301 + return record + + def export(self, record: dict[str, Any]) -> None: + self.records.append(record) + + def flush(self) -> None: + return None + + +def _op(name: str, *, result=None, error=None, status=OperationStatus.SUCCEEDED): + return OperationEndInfo( + operation_id=name, + operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name=name, + parent_id=None, + start_time=T0, + is_replayed=False, + status=status, + end_time=T1, + result=result, + error=error, + attempt=1, + ) + + +def _emit( + ops: list[OperationInfo], + *, + error: ErrorObject | None = None, + content: ContentConfig | None = None, +) -> dict[str, Any]: + exporter = CaptureExporter() + plugin = durable_execution_telemetry( + DurableExecutionTelemetryConfig(exporters=[exporter], content=content) + ) + 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=None, + operations=operations, + ) + ) + plugin.on_invocation_end( + InvocationEndInfo( + request_id=None, + execution_arn=ARN, + is_first_invocation=True, + execution_start_time=T0, + status=InvocationStatus.FAILED if error else InvocationStatus.SUCCEEDED, + error=error, + execution_result=None, + operations=operations, + ) + ) + assert len(exporter.records) == 1 + return exporter.records[0] + + +def _err(type_name: str | None, message: str | None, data: str | None = None): + return ErrorObject(message=message, type=type_name, data=data, stack_trace=None) + + +# -- error shape --------------------------------------------------------------- + + +def test_hook_errors_carry_no_cause(): + # The plugin hooks expose ErrorObject only; it has no cause field, so a + # Python record can never carry ``cause``. + fields = {f.name for f in dataclasses.fields(ErrorObject)} + assert fields == {"message", "type", "data", "stack_trace"} + assert "error" in {f.name for f in dataclasses.fields(InvocationEndInfo)} + assert "error" in {f.name for f in dataclasses.fields(OperationEndInfo)} + + +def test_error_shape_and_key_order_everywhere(): + record = _emit( + [ + _op( + "charge", + status=OperationStatus.FAILED, + error=_err("PaymentError", "card declined", data='{"code":1}'), + ) + ], + error=_err(STEP_ERROR, "card declined", data="ignored"), + ) + assert list(record["error"]) == ["name", "message"] + # Names are the SDK's ErrorType, unchanged. + assert record["error"] == {"name": STEP_ERROR, "message": "card declined"} + op_error = record["operations"][0]["error"] + assert list(op_error) == ["name", "message"] + assert op_error == {"name": "PaymentError", "message": "card declined"} + by_name = with_operations_by_name(record)["operationsByName"]["charge"]["error"] + assert list(by_name) == ["name", "message"] + assert by_name == op_error + # ErrorData is never turned into a cause. + assert "cause" not in json.dumps(record) + + +def test_error_with_missing_type_and_message_keeps_both_keys(): + record = _emit([], error=_err(None, None)) + assert list(record["error"]) == ["name", "message"] + assert record["error"] == {"name": None, "message": None} + + +def test_include_errors_false_removes_operation_errors_everywhere(): + record = _emit( + [_op("charge", status=OperationStatus.FAILED, error=_err("E", "m"))], + error=_err(STEP_ERROR, "m"), + content=ContentConfig(operations=ContentOperations(include_errors=False)), + ) + assert "error" not in record["operations"][0] + assert "error" not in with_operations_by_name(record)["operationsByName"]["charge"] + assert record["error"] == {"name": STEP_ERROR, "message": "m"} + + +# -- result decoding ----------------------------------------------------------- + + +def _results(*ops: OperationInfo) -> dict[str, Any]: + overrides = [OperationOverride(op.name or "", result=lambda r: r) for op in ops] + record = _emit( + list(ops), + content=ContentConfig(operations=ContentOperations(overrides=overrides)), + ) + return {op["name"]: op.get("result") for op in record["operations"]} + + +def _envelope(value: Any) -> str: + raw = EXTENDED_TYPES_SERDES.serialize(value, None) # type: ignore[arg-type] + assert set(json.loads(raw)) == {"t", "v"} + return raw + + +def test_envelope_result_decodes_to_plain_json(): + uid = uuid.UUID("12345678-1234-5678-1234-567812345678") + value = { + "amount": decimal.Decimal("12.50"), + "count": decimal.Decimal(3), + "when": datetime.datetime(2026, 1, 2, 3, 4, 5, 6000, tzinfo=datetime.UTC), + "day": datetime.date(2026, 1, 2), + "id": uid, + "blob": b"\x00\xffhi", + "pair": (1, "a"), + "nested": [{"x": decimal.Decimal("0.1")}], + } + results = _results(_op("r", result=_envelope(value))) + assert results["r"] == { + "amount": 12.5, + "count": 3, + "when": "2026-01-02T03:04:05.006000+00:00", + "day": "2026-01-02", + "id": str(uid), + "blob": base64.b64encode(b"\x00\xffhi").decode(), + "pair": [1, "a"], + "nested": [{"x": 0.1}], + } + json.dumps(results["r"], allow_nan=False) + + +def test_batch_result_becomes_to_dict(): + batch = BatchResult( + all=[BatchItem(0, BatchItemStatus.SUCCEEDED, result=decimal.Decimal(2))], + completion_reason=CompletionReason.ALL_COMPLETED, + ) + decoded = _decode_checkpoint_result(_envelope(batch)) + expected = batch.to_dict() + expected["all"][0]["result"] = 2 + assert decoded == json.loads(json.dumps(expected)) + + +@pytest.mark.parametrize( + ("text", "expected"), + [ + ("12", 12), + ("-0", 0), + ("1.10", 1.1), + ("0.1", 0.1), + ("1E+299", 10**299), + ("1E+300", 1e300), + ("12345678901234567.89", "12345678901234567.89"), + ("1E+5000", "1E+5000"), + ("1E+1000000", "1E+1000000"), + ("1E-1000000", "1E-1000000"), + ("NaN", None), + ("sNaN", None), + ("Infinity", None), + ("-Infinity", None), + ], +) +def test_decimal_conversion(text, expected): + raw = _envelope({"v": decimal.Decimal(text)}) + started = time.perf_counter() + decoded = _decode_checkpoint_result(raw) + assert time.perf_counter() - started < 0.5 + assert decoded == {"v": expected} + assert type(decoded["v"]) is type(expected) + + +def test_non_finite_plain_json_becomes_null(): + assert _decode_checkpoint_result("[NaN, Infinity, 1.5]") == [None, None, 1.5] + + +def test_600_nested_arrays(): + depth = 600 + raw = "[" * depth + "]" * depth + decoded = _decode_checkpoint_result(raw) + level = 1 + while decoded: + decoded = decoded[0] + level += 1 + assert level == depth + + # Converted without recursion, also for tuples. + value: Any = () + for _ in range(depth - 1): + value = (value, decimal.Decimal("1.5")) + converted = _to_json_compatible(value) + level = 1 + while converted: + assert converted[1] == 1.5 + converted = converted[0] + level += 1 + assert level == depth + + +def test_deep_envelope_the_core_cannot_decode_does_not_raise(): + depth = 600 + raw = '{"t":"l","v":[' * depth + "]}" * depth + decoded = _decode_checkpoint_result(raw) + # json.loads or the core decoder gives up; the raw/parsed value is kept. + assert decoded is not None + + +def test_non_envelope_results_pass_through(): + assert _decode_checkpoint_result("42") == 42 + assert _decode_checkpoint_result('{"a": 1}') == {"a": 1} + assert _decode_checkpoint_result("not json") == "not json" + # Unknown tag: decoding fails, parsed value kept. + assert _decode_checkpoint_result('{"t":"zz","v":1}') == {"t": "zz", "v": 1} + # Extra key: not an envelope. + assert _decode_checkpoint_result('{"t":"s","v":"x","w":1}') == { + "t": "s", + "v": "x", + "w": 1, + } + # Shape collision: a JsonSerDes value that looks like an envelope is decoded. + assert _decode_checkpoint_result('{"t":"s","v":"x"}') == "x" + + +def test_one_failing_result_keeps_the_rest_of_the_record(): + results = _results( + _op("good", result=_envelope({"a": decimal.Decimal("1.5")})), + _op("bad", result='{"t":"d","v":"not-a-decimal"}'), + _op("text", result="plain text"), + ) + assert results == { + "good": {"a": 1.5}, + "bad": {"t": "d", "v": "not-a-decimal"}, + "text": "plain text", + } + + +def test_unsupported_decoded_type_falls_back_to_raw(monkeypatch): + class Odd: + pass + + monkeypatch.setattr( + EXTENDED_TYPES_SERDES, "deserialize", lambda data, ctx=None: {"x": Odd()} + ) + raw = '{"t":"m","v":{}}' + # Decoded value cannot be converted; the parsed payload is kept. + assert _decode_checkpoint_result(raw) == {"t": "m", "v": {}} + + +def test_unsupported_value_after_fallback_returns_raw_string(monkeypatch): + from aws_durable_execution_sdk_python_telemetry_exporter import plugin + + def boom(value): + raise TypeError + + monkeypatch.setattr(plugin, "_to_json_compatible", boom) + assert _decode_checkpoint_result('{"a":1}') == '{"a":1}' 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..8c8647381 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 @@ -10,7 +10,9 @@ from __future__ import annotations +import datetime import json +import re from typing import Any import pytest @@ -96,14 +98,32 @@ 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): - LambdaLogExporter().export(_record()) - out = capsys.readouterr().out.strip() +def test_lambda_log_export_prints_one_envelope_line(capsys): + exporter = LambdaLogExporter() + before = datetime.datetime.now(datetime.UTC) + exporter.export(_record()) + after = datetime.datetime.now(datetime.UTC) + out = capsys.readouterr().out + assert out.endswith("\n") and out.count("\n") == 1 + line = out.rstrip("\n") # 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) + assert list(envelope) == ["timestamp", "level", "message"] + assert envelope["level"] == "INFO" + # message parses back to exactly the rendered record + assert isinstance(envelope["message"], str) + assert json.loads(envelope["message"]) == exporter.render(_record()) + assert envelope["message"] == json.dumps( + exporter.render(_record()), separators=(",", ":"), ensure_ascii=False + ) + # UTC, millisecond precision, Z suffix + ts = envelope["timestamp"] + assert re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z", ts) + parsed_ts = datetime.datetime.fromisoformat(ts.replace("Z", "+00:00")) + assert parsed_ts.utcoffset() == datetime.timedelta(0) + assert before.replace(microsecond=before.microsecond // 1000 * 1000) <= parsed_ts + assert parsed_ts <= after 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..7cb94c1b4 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,6 +13,7 @@ from __future__ import annotations import datetime +import json import threading from typing import Any @@ -591,9 +592,12 @@ 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 + # compact JSON envelope via LambdaLogExporter; the record is its message + envelope = json.loads(out.strip()) + assert envelope["level"] == "INFO" + record = json.loads(envelope["message"]) + assert record["recordType"] == "DurableExecutionTelemetry" + assert "operationsByName" in record # -- terminal timestamps: end time only for SUCCEEDED/FAILED (comment 3) ------ From f4cfeffd5af7e7ea037e3199ffea8c2294b4af13 Mon Sep 17 00:00:00 2001 From: Pooya Paridel Date: Thu, 22 Oct 2026 11:30:54 -0700 Subject: [PATCH 2/6] fix(telemetry): bound numbers and the log line Decode an integral Decimal to an int only when abs(n) <= 2**53 - 1. Otherwise write a float only when it round-trips exactly and is within DynamoDB's number range (38 significant digits, magnitude 1E-130 to 9.99...E+125), else str(value). DynamoDBExporter also writes any number it cannot store, including NaN and Infinity, as a string attribute, so a decoded or plain JSON result no longer makes the item fail. LambdaLogExporter.render returns the printed envelope, so the existing single truncation pass measures the line. U+2028 and U+2029 are escaped in the line, and the size check counts them escaped. Cap every error message at 8,192 characters plus an ellipsis. Truncation first drops every cause and sets droppedCauses; Python records have no cause, so this step does not change them. Plain JSON results skip the conversion walk unless they contain NaN or an infinite number. Replace the wall-clock test assertion with a count of int conversions. --- .../exporters/dynamodb_exporter.py | 33 ++- .../exporters/lambda_log_exporter.py | 28 +- .../plugin.py | 84 +++++- .../truncation.py | 43 ++- .../tests/test_error_and_result_format.py | 72 ++++- .../tests/test_exporters.py | 16 +- .../tests/test_public_path_limits.py | 276 ++++++++++++++++++ .../tests/test_shaping.py | 38 +++ 8 files changed, 543 insertions(+), 47 deletions(-) create mode 100644 packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_public_path_limits.py diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/dynamodb_exporter.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/dynamodb_exporter.py index 1b3c9fe13..cf2d72b8e 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/dynamodb_exporter.py +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/dynamodb_exporter.py @@ -5,8 +5,8 @@ from __future__ import annotations +import decimal import json -from decimal import Decimal from typing import Any from aws_durable_execution_sdk_python_telemetry_exporter.exporters._common import ( @@ -20,6 +20,11 @@ class DynamoDBExporter: """Writes ``operationsByName`` records to a DynamoDB table with PutItem. + Numbers DynamoDB cannot store (more than 38 significant digits, a + magnitude outside 1E-130 to 9.99...E+125, NaN or Infinity) are written as + their JSON text in a string attribute, so such a value never drops the + record. + The partition key holds ``executionArn``. With the default sort key (``sk`` = ``emittedAt``) every export adds a new item, keeping the full history. With ``sort_key=None`` later exports overwrite the item. @@ -51,9 +56,11 @@ def __init__( if region else boto3.client("dynamodb") ) - from boto3.dynamodb.types import TypeSerializer # deferred, same reason + # deferred, same reason + from boto3.dynamodb.types import DYNAMODB_CONTEXT, TypeSerializer self._serializer = TypeSerializer() + self._context = DYNAMODB_CONTEXT def render(self, record: dict[str, Any]) -> dict[str, Any]: return with_operations_by_name(record) @@ -68,8 +75,26 @@ def export(self, record: dict[str, Any]) -> None: def flush(self) -> None: return None + def _number(self, text: str) -> decimal.Decimal | str: + # TypeSerializer checks numbers with DYNAMODB_CONTEXT, which traps any + # inexact, rounded, overflowing or underflowing value. Check the same + # way, plus DynamoDB's own magnitude range (1E-130 to 9.99...E+125), + # and keep a number either would reject as its JSON text. + try: + number = self._context.create_decimal(text) + except decimal.DecimalException: + return text + if number and not -130 <= number.adjusted() <= 125: + return text + return decimal.Decimal(text) + def _marshal(self, item: dict[str, Any]) -> dict[str, Any]: # DynamoDB numbers must be ``Decimal``; a JSON round trip converts every - # float and leaves the rest of the record untouched. - plain = json.loads(compact_dumps(item), parse_float=Decimal) + # number and leaves the rest of the record untouched. + plain = json.loads( + compact_dumps(item), + parse_float=self._number, + parse_int=self._number, + parse_constant=str, + ) return {key: self._serializer.serialize(value) for key, value in plain.items()} 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 f7c9ef57c..40bf0ab5a 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 @@ -31,12 +31,12 @@ class LambdaLogExporter: 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. + ``render`` returns this envelope, so ``max_record_size_bytes`` (default + 256,000) limits the printed line itself (in UTF-8 bytes, without the + trailing newline). Truncation runs once, against the envelope, including + the JSON escaping of ``message``. U+2028 and U+2029 are written as + ``\\u2028`` and ``\\u2029`` so tools that split on them keep one record per + line. CloudWatch Logs accepts events of up to 1 MB. """ def __init__(self, max_record_size_bytes: int | None = None) -> None: @@ -45,22 +45,26 @@ def __init__(self, max_record_size_bytes: int | None = None) -> None: ) def render(self, record: dict[str, Any]) -> dict[str, Any]: - return with_operations_by_name(record) - - def export(self, record: dict[str, Any]) -> None: timestamp = ( datetime.datetime.now(datetime.UTC) .isoformat(timespec="milliseconds") .replace("+00:00", "Z") ) - envelope = { + return { "timestamp": timestamp, "level": "INFO", - "message": compact_dumps(self.render(record)), + "message": compact_dumps(with_operations_by_name(record)), } + + def export(self, record: dict[str, Any]) -> None: + line = ( + compact_dumps(self.render(record)) + .replace("\u2028", "\\u2028") + .replace("\u2029", "\\u2029") + ) # 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 + print(line, 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 c40db253a..28293573e 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 @@ -190,15 +190,36 @@ def _decode_checkpoint_result(raw_result: str) -> Any: primitive lists as plain JSON, and everything else as a ``{"t": ..., "v": ...}`` envelope. An envelope is decoded with that default serializer and converted to plain JSON values (see ``_to_json_compatible``). A payload that - is not an envelope is parsed JSON, or the raw string when it is not JSON. An - envelope that fails to decode falls back to the parsed payload. Never raises. + is not an envelope is parsed JSON, or the raw string when it is not JSON; + its tree is only walked when it holds NaN or an infinite number. An + envelope that fails to decode falls back to the parsed payload. Never + raises. The hooks do not say which serializer wrote a result, so the envelope is recognised by shape alone: a result written by ``JsonSerDes`` or a custom - ``SerDes`` that has this exact shape is decoded too. + ``SerDes`` that has this exact shape is decoded too. Only the core's known + tags are decoded, and the core may coerce ``v`` while decoding (for + example ``{"t": "b", "v": "yes"}`` becomes ``True``). An unknown tag fails + to decode and passes through unchanged. """ + non_finite = False + + def parse_constant(name: str) -> float: + nonlocal non_finite + non_finite = True + return float(name) + + def parse_float(text: str) -> float: + nonlocal non_finite + number = float(text) + if not math.isfinite(number): # for example 1e999 + non_finite = True + return number + try: - parsed = json.loads(raw_result) + parsed = json.loads( + raw_result, parse_constant=parse_constant, parse_float=parse_float + ) except (ValueError, TypeError, RecursionError): return raw_result try: @@ -208,22 +229,31 @@ def _decode_checkpoint_result(raw_result: str) -> Any: return _to_json_compatible(decoded) except Exception: # noqa: BLE001 - fall back to the parsed payload pass - return _to_json_compatible(parsed) + # Plain JSON is already JSON-compatible, except NaN and Infinity. + return _to_json_compatible(parsed) if non_finite else parsed except Exception: # noqa: BLE001 - one bad result never breaks the record return raw_result -# An integral Decimal becomes a JSON integer only while its leading digit's -# exponent is below this bound; larger values never build a huge int. -_MAX_INT_ADJUSTED = 300 +# Largest integer that every JSON reader (including JavaScript) parses exactly. +_MAX_SAFE_INTEGER = 2**53 - 1 +# 2**53 - 1 has 16 digits, so its leading digit's exponent is 15. +_MAX_SAFE_INTEGER_ADJUSTED = 15 +# DynamoDB's number range, the narrowest of the record destinations: at most +# 38 significant digits and a magnitude from 1E-130 to 9.99...E+125. +_MAX_NUMBER_DIGITS = 38 +_MAX_NUMBER_ADJUSTED = 125 +_MIN_NUMBER_ADJUSTED = -130 def _to_json_compatible(value: Any) -> Any: """Convert a decoded result into plain JSON values, without recursion. - ``Decimal``: an int when integral and ``adjusted() < 300``, else a float - when it round-trips exactly, else ``str(value)``; NaN and Infinity become - ``None``. ``datetime`` / ``date`` -> ``isoformat()``, ``UUID`` -> string, + ``Decimal``: an int when integral and ``abs(n) <= 2**53 - 1``, else a float + when it round-trips exactly and is within DynamoDB's number range (at most + 38 significant digits, magnitude 1E-130 to 9.99...E+125), else + ``str(value)``; NaN and Infinity become ``None``. Negative zero becomes + ``0``. ``datetime`` / ``date`` -> ``isoformat()``, ``UUID`` -> string, ``bytes`` -> base64, ``tuple`` -> list, ``BatchResult`` -> ``to_dict()``. Non-finite floats become ``None``. """ @@ -261,10 +291,24 @@ def _scalar_to_json_compatible(value: Any) -> Any: return None # adjusted() is checked first so Decimal("1E+1000000") never builds a # million-digit int. - if value.adjusted() < _MAX_INT_ADJUSTED and value == value.to_integral_value(): - return int(value) + if ( + value.adjusted() <= _MAX_SAFE_INTEGER_ADJUSTED + and value == value.to_integral_value() + ): + as_int = int(value) + if abs(as_int) <= _MAX_SAFE_INTEGER: + return as_int + if value and not ( + _MIN_NUMBER_ADJUSTED <= value.adjusted() <= _MAX_NUMBER_ADJUSTED + ): + return str(value) as_float = float(value) - if math.isfinite(as_float) and decimal.Decimal(repr(as_float)) == value: + written = decimal.Decimal(repr(as_float)) + if ( + math.isfinite(as_float) + and written == value + and len(written.as_tuple().digits) <= _MAX_NUMBER_DIGITS + ): return as_float return str(value) if isinstance(value, (datetime.datetime, datetime.date)): @@ -277,14 +321,22 @@ def _scalar_to_json_compatible(value: Any) -> Any: raise TypeError(msg) +# A longer error message keeps its first 8,192 characters followed by "…". +_MAX_ERROR_MESSAGE_CHARS = 8192 + + def _error_shape(error: ErrorObject) -> dict[str, Any]: """Return an error as ``{name, message}`` in that key order. The plugin hooks carry an ``ErrorObject`` (type, message, data, stack trace) and no cause, so records written by this SDK never have ``cause``. - The name is the SDK's ErrorType, unchanged. + The name is the SDK's ErrorType, unchanged. A message longer than 8,192 + characters is cut to its first 8,192 characters followed by ``…``. """ - return {"name": error.type, "message": error.message} + message = error.message + if isinstance(message, str) and len(message) > _MAX_ERROR_MESSAGE_CHARS: + message = message[:_MAX_ERROR_MESSAGE_CHARS] + "\u2026" + return {"name": error.type, "message": message} class _ExecutionState: diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/truncation.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/truncation.py index d826d2e0d..81c730bc7 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/truncation.py +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/truncation.py @@ -4,6 +4,9 @@ """Best-effort record size limiter. Direct port of the JS ``truncation.ts``. Drop order: + 0. every error ``cause`` (``droppedCauses: true``). Records written by this + SDK never carry a cause, so this step only changes a record whose errors + were given a ``cause`` by other code; 1. operation ``result`` fields, oldest operation first (each dropped op marked ``truncated: true``); 2. whole operations, oldest first (``droppedOperations`` count); @@ -14,7 +17,9 @@ ``render`` maps the record to the exact value the exporter serializes, so the size check measures what is actually emitted. Byte size is measured with JS-compatible compact JSON (no whitespace, non-ASCII preserved) to match -``JSON.stringify`` byte counts. +``JSON.stringify`` byte counts. U+2028 and U+2029 are counted as their 6-byte +``\\uXXXX`` escapes, which ``LambdaLogExporter`` prints; for every other +exporter this overcounts by 3 bytes per such character. """ from __future__ import annotations @@ -25,11 +30,24 @@ def json_byte_size(value: Any) -> int | None: try: - return len( - json.dumps(value, separators=(",", ":"), ensure_ascii=False).encode("utf-8") - ) + text = json.dumps(value, separators=(",", ":"), ensure_ascii=False) except (TypeError, ValueError): return None + # Each U+2028 / U+2029 is 3 bytes raw and 6 bytes escaped. + return len(text.encode("utf-8")) + 3 * (text.count("\u2028") + text.count("\u2029")) + + +def _has_cause(holder: dict[str, Any]) -> bool: + error = holder.get("error") + return isinstance(error, dict) and "cause" in error + + +def _drop_cause(holder: dict[str, Any]) -> None: + # ``holder`` is already a copy; the error is copied too, never mutated. + if _has_cause(holder): + holder["error"] = { + key: value for key, value in holder["error"].items() if key != "cause" + } def truncate_record( @@ -53,6 +71,12 @@ def truncate_record( range(len(ops)), key=lambda i: (ops[i].get("startTime") or "\uffff", i) ) + # Phase 0: drop every cause. The cause chain hangs off each error only, so + # removing the top-level ``cause`` removes the whole chain. + dropped_causes = _has_cause(record) or any(_has_cause(op) for op in ops) + for op in ops: + _drop_cause(op) + any_result = False dropped_ops = 0 dropped_input = False @@ -62,6 +86,9 @@ def candidate() -> dict[str, Any]: out = dict(record) out["operations"] = [op for i, op in enumerate(ops) if kept[i]] out["truncated"] = True + if dropped_causes: + _drop_cause(out) + out["droppedCauses"] = True if dropped_ops > 0: out["droppedOperations"] = dropped_ops if dropped_input: @@ -101,7 +128,13 @@ def fits() -> bool: if not fits() and record.get("output") is not None: dropped_output = True - if not any_result and dropped_ops == 0 and not dropped_input and not dropped_output: + if ( + not dropped_causes + and not any_result + and dropped_ops == 0 + and not dropped_input + and not dropped_output + ): return record return candidate() diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_error_and_result_format.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_error_and_result_format.py index f9a1514af..ce78555c6 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_error_and_result_format.py +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_error_and_result_format.py @@ -10,7 +10,6 @@ import datetime import decimal import json -import time import uuid from typing import Any @@ -243,10 +242,21 @@ def test_batch_result_becomes_to_dict(): [ ("12", 12), ("-0", 0), + ("-0.0", 0), ("1.10", 1.1), ("0.1", 0.1), - ("1E+299", 10**299), - ("1E+300", 1e300), + ("9007199254740991", 9007199254740991), + ("-9007199254740991", -9007199254740991), + ("9007199254740992", 9007199254740992.0), + ("9007199254740993", "9007199254740993"), + ("1E+20", 1e20), + ("1E+125", 1e125), + ("1E+126", "1E+126"), + ("1E-130", 1e-130), + ("1E-131", "1E-131"), + ("1E+299", "1E+299"), + ("1E+300", "1E+300"), + ("1" * 39, "1" * 39), ("12345678901234567.89", "12345678901234567.89"), ("1E+5000", "1E+5000"), ("1E+1000000", "1E+1000000"), @@ -259,15 +269,57 @@ def test_batch_result_becomes_to_dict(): ) def test_decimal_conversion(text, expected): raw = _envelope({"v": decimal.Decimal(text)}) - started = time.perf_counter() decoded = _decode_checkpoint_result(raw) - assert time.perf_counter() - started < 0.5 assert decoded == {"v": expected} assert type(decoded["v"]) is type(expected) +class _CountingDecimal(decimal.Decimal): + int_calls = 0 + + def __int__(self) -> int: + type(self).int_calls += 1 + return super().__int__() + + +@pytest.mark.parametrize( + "text", ["1E+16", "1E+299", "1E+5000", "1E+1000000", "1" * 4000] +) +def test_out_of_range_integral_decimal_never_builds_an_int(text): + # Counts int() conversions instead of timing them: a Decimal past the + # 2**53 - 1 bound is rejected on its exponent before any int is built. + _CountingDecimal.int_calls = 0 + assert _to_json_compatible([_CountingDecimal(text)]) == [ + float(text) if text == "1E+16" else text + ] + assert _CountingDecimal.int_calls == 0 + assert _to_json_compatible([_CountingDecimal("12")]) == [12] + assert _CountingDecimal.int_calls == 1 + + def test_non_finite_plain_json_becomes_null(): assert _decode_checkpoint_result("[NaN, Infinity, 1.5]") == [None, None, 1.5] + assert _decode_checkpoint_result('{"a": [1e999, -1e999]}') == {"a": [None, None]} + + +def test_plain_json_skips_the_tree_walk(monkeypatch): + from aws_durable_execution_sdk_python_telemetry_exporter import plugin + + calls: list[Any] = [] + real = plugin._to_json_compatible + + def spy(value): + calls.append(value) + return real(value) + + monkeypatch.setattr(plugin, "_to_json_compatible", spy) + assert _decode_checkpoint_result('{"a": [1, 2.5, "x", null, true]}') == { + "a": [1, 2.5, "x", None, True] + } + assert _decode_checkpoint_result('{"t": "zz", "v": 1}') == {"t": "zz", "v": 1} + assert calls == [] + assert _decode_checkpoint_result("[NaN]") == [None] + assert len(calls) == 1 def test_600_nested_arrays(): @@ -315,6 +367,14 @@ def test_non_envelope_results_pass_through(): } # Shape collision: a JsonSerDes value that looks like an envelope is decoded. assert _decode_checkpoint_result('{"t":"s","v":"x"}') == "x" + assert _decode_checkpoint_result('{"t":"i","v":5}') == 5 + assert _decode_checkpoint_result('{"t":"m","v":{"a":1}}') == {"a": 1} + # The core coerces ``v`` for a known tag. + assert _decode_checkpoint_result('{"t":"b","v":"yes"}') is True + assert _decode_checkpoint_result('{"t":"temperature","v":21.5}') == { + "t": "temperature", + "v": 21.5, + } def test_one_failing_result_keeps_the_rest_of_the_record(): @@ -349,4 +409,4 @@ def boom(value): raise TypeError monkeypatch.setattr(plugin, "_to_json_compatible", boom) - assert _decode_checkpoint_result('{"a":1}') == '{"a":1}' + assert _decode_checkpoint_result('{"a":NaN}') == '{"a":NaN}' 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 8c8647381..f8373e12a 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 @@ -34,6 +34,9 @@ from aws_durable_execution_sdk_python_telemetry_exporter.exporters.s3_exporter import ( S3Partitioning as S3PartitioningFromModule, ) +from aws_durable_execution_sdk_python_telemetry_exporter.operations_index import ( + with_operations_by_name, +) def _record(**kw: Any) -> dict[str, Any]: @@ -90,10 +93,13 @@ def test_every_exporter_is_re_exported_from_the_package_root(): # -- LambdaLogExporter -------------------------------------------------------- -def test_lambda_log_default_size_and_render_is_operations_by_name(): +def test_lambda_log_default_size_and_render_is_the_envelope(): exporter = LambdaLogExporter() assert exporter.max_record_size_bytes == 256_000 - shaped = exporter.render(_record()) + envelope = exporter.render(_record()) + # render returns the printed envelope, so truncation measures the line. + assert list(envelope) == ["timestamp", "level", "message"] + shaped = json.loads(envelope["message"]) assert "operations" not in shaped assert shaped["operationsByName"]["greet"]["count"] == 1 @@ -113,10 +119,12 @@ def test_lambda_log_export_prints_one_envelope_line(capsys): assert envelope["level"] == "INFO" # message parses back to exactly the rendered record assert isinstance(envelope["message"], str) - assert json.loads(envelope["message"]) == exporter.render(_record()) + expected = with_operations_by_name(_record()) + assert json.loads(envelope["message"]) == expected assert envelope["message"] == json.dumps( - exporter.render(_record()), separators=(",", ":"), ensure_ascii=False + expected, separators=(",", ":"), ensure_ascii=False ) + assert envelope["message"] == exporter.render(_record())["message"] # UTC, millisecond precision, Z suffix ts = envelope["timestamp"] assert re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z", ts) diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_public_path_limits.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_public_path_limits.py new file mode 100644 index 000000000..32aa2dca9 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_public_path_limits.py @@ -0,0 +1,276 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Public plugin path into real exporters: decoded numbers, line size, caps.""" + +from __future__ import annotations + +import datetime +import decimal +import json +import logging +from typing import Any + +import pytest +from aws_durable_execution_sdk_python.lambda_service import ( + ErrorObject, + OperationStatus, + OperationSubType, +) +from aws_durable_execution_sdk_python.plugin import ( + InvocationEndInfo, + InvocationStartInfo, + InvocationStatus, + OperationEndInfo, + OperationInfo, + OperationType, +) +from aws_durable_execution_sdk_python.serdes import EXTENDED_TYPES_SERDES + +from aws_durable_execution_sdk_python_telemetry_exporter import ( + ContentConfig, + ContentOperations, + DurableExecutionTelemetryConfig, + DynamoDBExporter, + LambdaLogExporter, + OperationOverride, + durable_execution_telemetry, +) + +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) +BIG_39_DIGITS = "123456789012345678901234567890123456789" + + +class StubDynamoDBClient: + def __init__(self) -> None: + self.items: list[dict[str, Any]] = [] + + def put_item(self, **kwargs: Any) -> None: + self.items.append(kwargs["Item"]) + + +def _op(name: str, index: int, *, result: str | None = None) -> OperationEndInfo: + start = T0 + datetime.timedelta(milliseconds=index) + return OperationEndInfo( + operation_id=f"{name}-id", + operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name=name, + parent_id=None, + start_time=start, + is_replayed=False, + status=OperationStatus.SUCCEEDED, + end_time=T1, + result=result, + error=None, + attempt=1, + ) + + +def _run( + exporter: Any, + ops: list[OperationEndInfo], + *, + execution_input: str | None = None, + error: ErrorObject | None = None, +) -> None: + overrides = [OperationOverride(op.name or "", result=lambda r: r) for op in ops] + plugin = durable_execution_telemetry( + DurableExecutionTelemetryConfig( + exporters=[exporter], + content=ContentConfig(operations=ContentOperations(overrides=overrides)), + ) + ) + operations: dict[str, OperationInfo] = {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=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 if error else InvocationStatus.SUCCEEDED, + error=error, + execution_result=None, + operations=operations, + ) + ) + + +def _envelope(value: Any) -> str: + return EXTENDED_TYPES_SERDES.serialize(value, None) # type: ignore[arg-type] + + +# -- B1: decoded numbers reach DynamoDB ---------------------------------------- + + +def _dynamodb_results(*results: str, caplog: pytest.LogCaptureFixture) -> Any: + client = StubDynamoDBClient() + exporter = DynamoDBExporter(table_name="t", client=client) + ops = [_op(f"op{i}", i, result=raw) for i, raw in enumerate(results)] + with caplog.at_level(logging.WARNING): + _run(exporter, ops) + assert "failed" not in caplog.text + assert len(client.items) == 1 + by_name = client.items[0]["operationsByName"]["M"] + return [by_name[f"op{i}"]["M"]["result"] for i in range(len(results))] + + +def test_out_of_range_decoded_decimals_keep_the_dynamodb_item(caplog): + results = _dynamodb_results( + _envelope(decimal.Decimal("1E+299")), + _envelope(decimal.Decimal(BIG_39_DIGITS)), + _envelope(decimal.Decimal("9007199254740993")), + _envelope({"amount": decimal.Decimal("12.50"), "n": decimal.Decimal("7")}), + caplog=caplog, + ) + assert results[:3] == [ + {"S": "1E+299"}, + {"S": BIG_39_DIGITS}, + # Exact for JavaScript readers, which would round the number. + {"S": "9007199254740993"}, + ] + assert results[3] == {"M": {"amount": {"N": "12.5"}, "n": {"N": "7"}}} + + +def test_out_of_range_plain_json_numbers_keep_the_dynamodb_item(caplog): + results = _dynamodb_results( + "1e300", + BIG_39_DIGITS, + "[1e-200, 1.5, 3]", + caplog=caplog, + ) + assert results == [ + {"S": "1e+300"}, + {"S": BIG_39_DIGITS}, + {"L": [{"S": "1e-200"}, {"N": "1.5"}, {"N": "3"}]}, + ] + + +def test_dynamodb_marshal_turns_non_finite_numbers_into_strings(): + client = StubDynamoDBClient() + exporter = DynamoDBExporter(table_name="t", client=client) + marshalled = exporter._marshal({"a": float("nan"), "b": float("inf"), "c": 1}) + assert marshalled == {"a": {"S": "NaN"}, "b": {"S": "Infinity"}, "c": {"N": "1"}} + + +def test_dynamodb_marshal_applies_the_service_magnitude_range(): + exporter = DynamoDBExporter(table_name="t", client=StubDynamoDBClient()) + marshalled = exporter._marshal({"a": [1e125, 1e126, 1e-130, 1e-131, 0, -0.0]}) + assert marshalled["a"]["L"] == [ + {"N": "1E+125"}, + {"S": "1e+126"}, + {"N": "1E-130"}, + {"S": "1e-131"}, + {"N": "0"}, + {"N": "-0.0"}, + ] + + +# -- B2: LambdaLogExporter bounds the printed line ------------------------------ + + +def _printed_line(capsys: pytest.CaptureFixture[str]) -> str: + out = capsys.readouterr().out + assert out.endswith("\n") and out.count("\n") == 1 + return out[:-1] + + +def test_quote_heavy_record_line_fits_the_limit(capsys): + limit = 262_144 + exporter = LambdaLogExporter(max_record_size_bytes=limit) + # Each '"' is 2 bytes in the record JSON and 4 bytes in the printed line. + ops = [_op(f"q{i}", i, result=json.dumps('"' * 4000)) for i in range(40)] + _run(exporter, ops, execution_input='"\\' * 10_000) + line = _printed_line(capsys) + assert len(line.encode("utf-8")) <= limit + envelope = json.loads(line) + record = json.loads(envelope["message"]) + assert record["truncated"] is True + # The single pass measured the line, so it still keeps what fits. + kept = [summary for summary in record["operationsByName"].values()] + assert kept + assert any("result" in summary for summary in kept) + + +def test_capped_execution_error_fits_a_64_kb_line(capsys): + limit = 65_536 + exporter = LambdaLogExporter(max_record_size_bytes=limit) + # \x01 is the worst case: 6 bytes in the record, 7 in the printed line. + message = "\x01" * 100_000 + ops = [_op(f"q{i}", i, result=json.dumps('"' * 2000)) for i in range(40)] + _run( + exporter, + ops, + execution_input='"' * 50_000, + error=ErrorObject(message=message, type="E", data=None, stack_trace=None), + ) + line = _printed_line(capsys) + assert len(line.encode("utf-8")) <= limit + record = json.loads(json.loads(line)["message"]) + assert record["error"]["message"] == "\x01" * 8192 + "\u2026" + assert record["droppedInput"] is True + + +def test_line_separators_are_escaped_and_text_round_trips(capsys): + exporter = LambdaLogExporter() + text = 'a\u2028b\u2029c é "q" \\ \n \x01 \U0001f600' + _run(exporter, [_op("s", 0, result=json.dumps(text))]) + out = capsys.readouterr().out + assert "\u2028" not in out and "\u2029" not in out + assert len(out.splitlines()) == 1 + # Non-ASCII stays raw; control characters and quotes are escaped. + assert "é" in out and "\U0001f600" in out and "\x01" not in out + record = json.loads(json.loads(out)["message"]) + assert record["operationsByName"]["s"]["result"] == text + + +def test_line_separator_escaping_is_counted_in_the_limit(capsys): + limit = 4096 + exporter = LambdaLogExporter(max_record_size_bytes=limit) + ops = [_op(f"s{i}", i, result=json.dumps("\u2028" * 300)) for i in range(10)] + _run(exporter, ops) + line = _printed_line(capsys) + assert len(line.encode("utf-8")) <= limit + assert json.loads(json.loads(line)["message"])["truncated"] is True + + +# -- rule 7: message cap -------------------------------------------------------- + + +def test_error_messages_are_capped_at_8192_characters(capsys): + exporter = LambdaLogExporter(max_record_size_bytes=0) + op = OperationEndInfo( + operation_id="f-id", + operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="f", + parent_id=None, + start_time=T0, + is_replayed=False, + status=OperationStatus.FAILED, + end_time=T1, + result=None, + error=ErrorObject(message="m" * 8193, type="E", data=None, stack_trace=None), + attempt=1, + ) + _run( + exporter, + [op], + error=ErrorObject(message="x" * 8192, type="E", data=None, stack_trace=None), + ) + record = json.loads(json.loads(_printed_line(capsys))["message"]) + # Exactly 8,192 characters is kept as is. + assert record["error"]["message"] == "x" * 8192 + assert record["operationsByName"]["f"]["error"]["message"] == "m" * 8192 + "…" diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_shaping.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_shaping.py index 8f2c8f274..671e2e575 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_shaping.py +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_shaping.py @@ -160,3 +160,41 @@ def test_apply_operations_format_both_keeps_array_and_adds_map(): def test_operations_format_enum_values(): assert {f.value for f in OperationsFormat} == {"array", "by-name", "both"} + + +def test_truncation_drops_every_cause_first(): + cause = { + "name": "Inner", + "message": "c" * 400, + "cause": {"name": "E", "message": "x"}, + } + op_error = {"name": "E", "message": "m", "cause": cause} + record = { + "executionArn": "arn", + "error": {"name": "E", "message": "m", "cause": cause}, + "operations": [_op("a", error=op_error, result="r")], + } + out = truncate_record(record, 300, render=lambda r: r) + assert out["droppedCauses"] is True + assert out["error"] == {"name": "E", "message": "m"} + assert out["operations"][0]["error"] == {"name": "E", "message": "m"} + # Dropping causes was enough: later phases did not run. + assert out["operations"][0]["result"] == "r" + # The input record is never mutated. + assert record["error"]["cause"] is cause + assert record["operations"][0]["error"]["cause"] is cause + + +def test_truncation_without_causes_sets_no_dropped_causes(): + record = _record_with_results([2000]) + out = truncate_record(record, 400, render=lambda r: r) + assert out["truncated"] is True + assert "droppedCauses" not in out + + +def test_json_byte_size_counts_line_separators_as_escapes(): + from aws_durable_execution_sdk_python_telemetry_exporter.truncation import ( + json_byte_size, + ) + + assert json_byte_size("\u2028\u2029") == len('"\\u2028\\u2029"') From 26ded9416c3836ac021cbcad01218ebd9fde7391 Mon Sep 17 00:00:00 2001 From: Pooya Paridel Date: Thu, 22 Oct 2026 11:30:54 -0700 Subject: [PATCH 3/6] docs(telemetry): describe number, line size and message limits Document the Decimal rules, including negative zero, the DynamoDB string fallback, the line limit of LambdaLogExporter, the message cap, the cause truncation step, and that Python records have no cause. State that WARN filtering applies only with the JSON log format, and which envelope-shaped results are decoded or passed through. --- .../README.md | 57 ++++++++++++++----- 1 file changed, 42 insertions(+), 15 deletions(-) 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 235301904..133f84c16 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/README.md +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/README.md @@ -91,12 +91,17 @@ double-encode output that is already JSON (see [Log and monitor Python Lambda functions](https://docs.aws.amazon.com/lambda/latest/dg/python-logging.html) and [Lambda log formats](https://docs.aws.amazon.com/lambda/latest/dg/monitoring-cloudwatchlogs-logformat.html)). -`max_record_size_bytes` (default 256 KB) limits the record in `message`. The -JSON escaping of `message` can make the line up to about 2x the record; a -CloudWatch Logs event can be up to 1 MB +`max_record_size_bytes` (default 256,000 bytes) limits the printed line: the +envelope, including the JSON escaping of `message`, in UTF-8 bytes without the +trailing newline. Truncation measures that line in one pass. A CloudWatch Logs +event can be up to 1 MB ([PutLogEvents](https://docs.aws.amazon.com/AmazonCloudWatchLogs/latest/APIReference/API_PutLogEvents.html)). -The lines have `level` `INFO`, so Lambda drops them when the function's -application log level is `WARN` or above. +U+2028 and U+2029 are written as `\u2028` and `\u2029`, so a tool that treats +them as line breaks still sees one line per record. + +The lines have `level` `INFO`. With the JSON log format and an application log +level of `WARN` or above, Lambda drops them. With the Text log format, Lambda +does not filter them. ### CloudWatchLogsExporter @@ -113,6 +118,9 @@ CloudWatchLogsExporter(log_group_name="/custom/durable-execution-telemetry") `PutItem` keyed by `partition_key` (default `pk`) = `executionArn`. With the default `sort_key="sk"` (= `emittedAt`) every export adds an item; pass `sort_key=None` for a key-only table that upserts. IAM: `dynamodb:PutItem`. +A number DynamoDB cannot store (more than 38 significant digits, a magnitude +outside 1E-130 to 9.99...E+125, NaN or Infinity) is written as a string +attribute holding its JSON text, so it never drops the item. ```python DynamoDBExporter(table_name="durable-execution-telemetry") @@ -257,9 +265,20 @@ has the same shape, recursively, and is present only when the SDK provides a cause. In Python, the plugin hooks carry only an `ErrorObject` (type, message, data, -stack trace) with no cause, so Python records have `name` and `message` and -never have `cause`. The exporter does not build a cause from other operations -or from `ErrorData`. `include_errors=False` removes operation errors. +stack trace) with no cause, so no error in a Python record has a `cause`: not +the execution `error`, not an operation error. The exporter does not build a +cause from other operations or from `ErrorData`. `include_errors=False` +removes operation errors. + +Every `message` is capped at 8,192 characters. A longer message keeps its first +8,192 characters followed by `…`. + +When a record is over its exporter's size limit, truncation first removes every +`cause` and sets `droppedCauses: true`, then runs its usual steps (operation +results, then whole operations, then input, then output). Python records have +no cause, so the first step never changes them. With the message cap, a record +whose operations, input and output are dropped fits any limit of 64 KB or +more. Names are each language's own and differ across SDKs. For example, a step that throws `PaymentError` can be reported as `StepError` in JS, @@ -275,21 +294,29 @@ default serializer's (`ExtendedTypeSerDes`) envelope, a JSON object whose keys are exactly `t` and `v` with a string `t`, it is decoded with that serializer and converted to plain JSON: -- `Decimal`: an integer when it is integral and `adjusted() < 300`; otherwise a - number when the float round-trips exactly; otherwise `str(value)` (for - example `"12345678901234567.89"` or `"1E+5000"`). NaN and Infinity become - `null`. +- `Decimal`: an integer when it is integral and its absolute value is at most + `2**53 - 1` (9007199254740991), the largest integer a JavaScript reader parses + exactly. Otherwise a float when the float round-trips exactly and is within + DynamoDB's number range (at most 38 significant digits, magnitude 1E-130 to + 9.99...E+125). Otherwise `str(value)`, for example `"9007199254740993"`, + `"12345678901234567.89"` or `"1E+299"`. NaN and Infinity become `null`. + Negative zero (`Decimal("-0")`, `Decimal("-0.0")`) is integral, so it becomes + `0` and loses its sign. - `datetime` and `date`: `isoformat()`. `UUID`: string. `bytes`: base64. `tuple`: list. `BatchResult`: `to_dict()`. Any other result is parsed as JSON, or kept as the raw string when it is not -JSON. A result that fails to decode keeps its parsed or raw value, and the rest -of the record is still emitted. +JSON. In parsed JSON, NaN and Infinity become `null`; nothing else changes. A +result that fails to decode keeps its parsed or raw value, and the rest of the +record is still emitted. The hooks do not say which serializer wrote a result, so the envelope is recognized by shape alone. A result written by `JsonSerDes` or a custom `SerDes` that has this exact shape is decoded too: `{"t": "s", "v": "x"}` is -reported as `"x"`. +reported as `"x"`. Only the core serializer's known tags are decoded, and the +core may coerce `v` while decoding: `{"t": "b", "v": "yes"}` is reported as +`true`. An unknown tag, such as `{"t": "temperature", "v": 21.5}`, fails to +decode and passes through unchanged. ## Requirements From 0ef77b826fa5add6a7742ea5b6168400c9ca0431 Mon Sep 17 00:00:00 2001 From: Pooya Paridel Date: Thu, 22 Oct 2026 12:35:23 -0700 Subject: [PATCH 4/6] fix(telemetry): keep lone surrogates and measure sizes exactly json_byte_size encodes inside its try again, so a lone surrogate makes the size unknown and the record is kept instead of raising and being dropped by every size-limited exporter. json_byte_size is exact again: it no longer counts U+2028 and U+2029 as escapes. truncate_record takes an optional measure function, and the scheduler passes an exporter's measure when it has one. LambdaLogExporter measures the exact printed line, which escapes U+2028, U+2029 and lone surrogates as \uXXXX, so printing a lone surrogate never raises and other exporters no longer over-count. --- .../_export_scheduler.py | 7 +- .../exporters/lambda_log_exporter.py | 36 ++++++-- .../truncation.py | 26 ++++-- .../types.py | 3 + .../tests/test_public_path_limits.py | 84 +++++++++++++++++++ .../tests/test_shaping.py | 30 ++++++- 6 files changed, 166 insertions(+), 20 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/_export_scheduler.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/_export_scheduler.py index ed702591d..626881a65 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/_export_scheduler.py +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/_export_scheduler.py @@ -146,8 +146,13 @@ def _run(self) -> None: def _export(self, record: dict[str, Any]) -> None: for exporter in self._exporters: try: + # ``measure`` is optional: an exporter that writes its rendered + # value as something other than compact JSON supplies it. shaped = truncate_record( - record, exporter.max_record_size_bytes, exporter.render + record, + exporter.max_record_size_bytes, + exporter.render, + getattr(exporter, "measure", None), ) exporter.export(shaped) except Exception as exc: # noqa: BLE001 - one exporter must not break others 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 40bf0ab5a..e66e0c5e5 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 @@ -6,6 +6,7 @@ from __future__ import annotations import datetime +import re from typing import Any from aws_durable_execution_sdk_python_telemetry_exporter.exporters._common import ( @@ -16,6 +17,20 @@ ) +# U+2028 and U+2029 (line breaks to some tools) and lone surrogates (no UTF-8 +# encoding, so printing one raises). A JSON line holds them only inside +# strings, where a \uXXXX escape is valid and decodes to the same text. +_ESCAPED_IN_LINE = re.compile("[\u2028\u2029\ud800-\udfff]") + + +def _escape(match: re.Match[str]) -> str: + return f"\\u{ord(match.group()):04x}" + + +def _line(rendered: Any) -> str: + return _ESCAPED_IN_LINE.sub(_escape, compact_dumps(rendered)) + + class LambdaLogExporter: """Writes ``operationsByName`` records to the function's own log group. @@ -34,9 +49,11 @@ class LambdaLogExporter: ``render`` returns this envelope, so ``max_record_size_bytes`` (default 256,000) limits the printed line itself (in UTF-8 bytes, without the trailing newline). Truncation runs once, against the envelope, including - the JSON escaping of ``message``. U+2028 and U+2029 are written as - ``\\u2028`` and ``\\u2029`` so tools that split on them keep one record per - line. CloudWatch Logs accepts events of up to 1 MB. + the JSON escaping of ``message``; ``measure`` returns the size of that + exact line. U+2028 and U+2029 are written as ``\\u2028`` and ``\\u2029`` + so tools that split on them keep one record per line, and a lone surrogate + is written as its ``\\uXXXX`` escape so printing it never raises. + CloudWatch Logs accepts events of up to 1 MB. """ def __init__(self, max_record_size_bytes: int | None = None) -> None: @@ -56,12 +73,15 @@ def render(self, record: dict[str, Any]) -> dict[str, Any]: "message": compact_dumps(with_operations_by_name(record)), } + def measure(self, rendered: Any) -> int | None: + """Return the UTF-8 size of the line printed for ``rendered``.""" + try: + return len(_line(rendered).encode("utf-8")) + except (TypeError, ValueError): + return None + def export(self, record: dict[str, Any]) -> None: - line = ( - compact_dumps(self.render(record)) - .replace("\u2028", "\\u2028") - .replace("\u2029", "\\u2029") - ) + line = _line(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(line, flush=True) # noqa: T201 diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/truncation.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/truncation.py index 81c730bc7..0e246a3ed 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/truncation.py +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/truncation.py @@ -15,11 +15,13 @@ Identity/timeline fields are never dropped. The input record is never mutated. ``render`` maps the record to the exact value the exporter serializes, so the -size check measures what is actually emitted. Byte size is measured with -JS-compatible compact JSON (no whitespace, non-ASCII preserved) to match -``JSON.stringify`` byte counts. U+2028 and U+2029 are counted as their 6-byte -``\\uXXXX`` escapes, which ``LambdaLogExporter`` prints; for every other -exporter this overcounts by 3 bytes per such character. +size check measures what is actually emitted. By default the byte size is that +of JS-compatible compact JSON (no whitespace, non-ASCII preserved), matching +``JSON.stringify`` byte counts. An exporter that writes the rendered value +differently passes its own ``measure`` (``LambdaLogExporter`` does, because it +escapes some characters in the printed line). A value that cannot be measured +(for example one holding a lone surrogate, which has no UTF-8 encoding) is +returned unchanged, never raised. """ from __future__ import annotations @@ -29,12 +31,16 @@ def json_byte_size(value: Any) -> int | None: + """Return the UTF-8 size of ``value`` as compact JSON, or ``None``. + + ``None`` when the value is not JSON-serializable or has no UTF-8 encoding + (a lone surrogate raises ``UnicodeEncodeError``, a ``ValueError``). + """ try: text = json.dumps(value, separators=(",", ":"), ensure_ascii=False) + return len(text.encode("utf-8")) except (TypeError, ValueError): return None - # Each U+2028 / U+2029 is 3 bytes raw and 6 bytes escaped. - return len(text.encode("utf-8")) + 3 * (text.count("\u2028") + text.count("\u2029")) def _has_cause(holder: dict[str, Any]) -> bool: @@ -54,12 +60,14 @@ def truncate_record( record: dict[str, Any], max_bytes: int | None, render: Callable[[dict[str, Any]], Any] | None = None, + measure: Callable[[Any], int | None] | None = None, ) -> dict[str, Any]: render = render or (lambda r: r) + measure = measure or json_byte_size if max_bytes is None or max_bytes <= 0: return record - initial = json_byte_size(render(record)) + initial = measure(render(record)) if initial is None or initial <= max_bytes: return record @@ -100,7 +108,7 @@ def candidate() -> dict[str, Any]: return out def fits() -> bool: - size = json_byte_size(render(candidate())) + size = measure(render(candidate())) return size is not None and size <= max_bytes # Phase 1: drop operation results oldest-first. diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/types.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/types.py index 9dfe18eb0..dabce1977 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/types.py +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/types.py @@ -56,6 +56,9 @@ class TelemetryExporter(Protocol): ``render`` maps the canonical record dict to the exact shape the exporter serializes (identity for array exporters, the ``operationsByName`` expansion for point-access exporters). + An exporter may also define ``measure(rendered) -> int | None``, the + byte size of what it writes for a rendered value; without it the size is + that of compact JSON. """ max_record_size_bytes: int | None diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_public_path_limits.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_public_path_limits.py index 32aa2dca9..3398e3a87 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_public_path_limits.py +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_public_path_limits.py @@ -7,8 +7,10 @@ import datetime import decimal +import io import json import logging +import sys from typing import Any import pytest @@ -274,3 +276,85 @@ def test_error_messages_are_capped_at_8192_characters(capsys): # Exactly 8,192 characters is kept as is. assert record["error"]["message"] == "x" * 8192 assert record["operationsByName"]["f"]["error"]["message"] == "m" * 8192 + "…" + + +# -- N-B1: a lone surrogate never drops the record ----------------------------- + + +def test_lone_surrogate_in_error_message_keeps_the_dynamodb_item(caplog): + client = StubDynamoDBClient() + exporter = DynamoDBExporter(table_name="t", client=client) + with caplog.at_level(logging.WARNING): + _run( + exporter, + [], + error=ErrorObject( + message="a\ud800b", type="E", data=None, stack_trace=None + ), + ) + assert "failed" not in caplog.text + assert len(client.items) == 1 + assert client.items[0]["error"]["M"]["message"] == {"S": "a\ud800b"} + + +def test_lone_surrogate_is_escaped_in_the_printed_line(monkeypatch, caplog): + # A strict UTF-8 stdout, as in a real process; printing a raw lone + # surrogate to it raises UnicodeEncodeError. + raw = io.BytesIO() + stdout = io.TextIOWrapper(raw, encoding="utf-8", errors="strict") + monkeypatch.setattr(sys, "stdout", stdout) + with caplog.at_level(logging.WARNING): + _run( + LambdaLogExporter(), + [], + error=ErrorObject( + message="a\ud800b", type="E", data=None, stack_trace=None + ), + ) + stdout.flush() + assert "failed" not in caplog.text + line = raw.getvalue().decode("utf-8") + assert line.count("\n") == 1 + assert "a\\ud800b" in line + record = json.loads(json.loads(line)["message"]) + assert record["error"]["message"] == "a\ud800b" + + +def test_lone_surrogates_are_counted_in_the_line_limit(capsys): + limit = 4096 + exporter = LambdaLogExporter(max_record_size_bytes=limit) + ops = [_op(f"s{i}", i, result=json.dumps("\ud800" * 300)) for i in range(10)] + _run(exporter, ops) + line = _printed_line(capsys) + assert len(line.encode("utf-8")) <= limit + assert json.loads(json.loads(line)["message"])["truncated"] is True + + +# -- non-blocking 1: sizes are exact per exporter ------------------------------- + + +def test_line_separators_are_counted_raw_for_other_exporters(caplog): + client = StubDynamoDBClient() + # 60,000 U+2028 are 180,000 bytes raw and 360,000 bytes escaped. + exporter = DynamoDBExporter( + table_name="t", client=client, max_record_size_bytes=200_000 + ) + with caplog.at_level(logging.WARNING): + _run(exporter, [_op("s", 0, result=json.dumps("\u2028" * 60_000))]) + assert len(client.items) == 1 + item = client.items[0] + assert "truncated" not in item + assert item["operationsByName"]["M"]["s"]["M"]["result"] == {"S": "\u2028" * 60_000} + + +def test_lambda_log_measure_is_the_printed_line_size(capsys): + exporter = LambdaLogExporter() + record: dict[str, Any] = { + "executionArn": ARN, + "operations": [], + "error": {"name": "E", "message": '\u2028\u2029\ud800é\U0001f600"\x01'}, + } + size = exporter.measure(exporter.render(record)) + exporter.export(record) + assert size == len(_printed_line(capsys).encode("utf-8")) + assert exporter.measure({"a": object()}) is None diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_shaping.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_shaping.py index 671e2e575..77f678aa8 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_shaping.py +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_shaping.py @@ -5,6 +5,8 @@ from __future__ import annotations +from typing import Any + from aws_durable_execution_sdk_python_telemetry_exporter.operations_index import ( OperationsFormat, apply_operations_format, @@ -192,9 +194,33 @@ def test_truncation_without_causes_sets_no_dropped_causes(): assert "droppedCauses" not in out -def test_json_byte_size_counts_line_separators_as_escapes(): +def test_json_byte_size_is_exact_for_line_separators(): from aws_durable_execution_sdk_python_telemetry_exporter.truncation import ( json_byte_size, ) - assert json_byte_size("\u2028\u2029") == len('"\\u2028\\u2029"') + # Written raw by every exporter except LambdaLogExporter: 3 bytes each. + assert json_byte_size({"a": "\u2028" * 10}) == 38 + assert json_byte_size("\u2028\u2029") == len('"\u2028\u2029"'.encode()) + + +def test_json_byte_size_returns_none_for_a_lone_surrogate(): + from aws_durable_execution_sdk_python_telemetry_exporter.truncation import ( + json_byte_size, + ) + + assert json_byte_size({"a": "a\ud800b"}) is None + + +def test_truncate_record_uses_the_given_measure(): + record = _record_with_results([100]) + sizes: list[Any] = [] + + def measure(value: Any) -> int: + sizes.append(value) + return 10**6 + + out = truncate_record(record, 1000, render=lambda r: r, measure=measure) + assert sizes and out["truncated"] is True + # The default measure would keep this small record unchanged. + assert truncate_record(record, 1000, render=lambda r: r) is record From 41c2d7ea84c97bcf57fe8391086bb3f8d12a54c5 Mon Sep 17 00:00:00 2001 From: Pooya Paridel Date: Thu, 22 Oct 2026 12:35:23 -0700 Subject: [PATCH 5/6] fix(telemetry): cap error names and bound large integers Cap every error name, like the message, at 8,192 code points followed by an ellipsis. An int result, from an envelope or from plain JSON, whose absolute value is greater than 2**53 - 1 is written as its decimal string, like a Decimal. Plain JSON is only walked when it has a run of 16 or more digits, NaN or an infinite number. Remove the 38-digit check on the float path, which could never fail. DynamoDBExporter stores an integer with more than 38 digits but at most 38 significant digits, such as 10**38, as a number. --- .../exporters/dynamodb_exporter.py | 15 ++- .../plugin.py | 61 ++++++---- .../tests/test_public_path_limits.py | 108 ++++++++++++++++++ 3 files changed, 158 insertions(+), 26 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/dynamodb_exporter.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/dynamodb_exporter.py index cf2d72b8e..aacc7b3bb 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/dynamodb_exporter.py +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/dynamodb_exporter.py @@ -23,7 +23,9 @@ class DynamoDBExporter: Numbers DynamoDB cannot store (more than 38 significant digits, a magnitude outside 1E-130 to 9.99...E+125, NaN or Infinity) are written as their JSON text in a string attribute, so such a value never drops the - record. + record. A field can therefore be a number in one item and a string in + another. An integer with more than 38 digits but at most 38 significant + ones (for example ``10**38``) is stored as a number. The partition key holds ``executionArn``. With the default sort key (``sk`` = ``emittedAt``) every export adds a new item, keeping the full @@ -83,7 +85,16 @@ def _number(self, text: str) -> decimal.Decimal | str: try: number = self._context.create_decimal(text) except decimal.DecimalException: - return text + # An integer with trailing zeros, such as 10**38, has more than 38 + # digits but few significant ones; its normalized form fits. + digits = text.removeprefix("-") + significant = digits.rstrip("0") + if not digits.isdigit() or len(significant) > 38: + return text + sign = text[: len(text) - len(digits)] + exponent = len(digits) - len(significant) + number = decimal.Decimal(f"{sign}{significant}E{exponent}") + return number if number.adjusted() <= 125 else text if number and not -130 <= number.adjusted() <= 125: return text return decimal.Decimal(text) 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 28293573e..d055fc5aa 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 @@ -36,6 +36,7 @@ import decimal import json import math +import re import threading import uuid from typing import TYPE_CHECKING, Any, Callable @@ -191,7 +192,8 @@ def _decode_checkpoint_result(raw_result: str) -> Any: ...}`` envelope. An envelope is decoded with that default serializer and converted to plain JSON values (see ``_to_json_compatible``). A payload that is not an envelope is parsed JSON, or the raw string when it is not JSON; - its tree is only walked when it holds NaN or an infinite number. An + its tree is only walked when it holds NaN, an infinite number, or a run of + 16 or more digits (a possible integer beyond ``2**53 - 1``). An envelope that fails to decode falls back to the parsed payload. Never raises. @@ -229,8 +231,13 @@ def parse_float(text: str) -> float: return _to_json_compatible(decoded) except Exception: # noqa: BLE001 - fall back to the parsed payload pass - # Plain JSON is already JSON-compatible, except NaN and Infinity. - return _to_json_compatible(parsed) if non_finite else parsed + # Plain JSON is already JSON-compatible, except NaN, Infinity and + # integers beyond 2**53 - 1. Such an integer has at least 16 digits; + # the C-speed scan may also match digits in a string or a float, which + # only costs a walk that leaves them unchanged. + if non_finite or _LONG_DIGIT_RUN.search(raw_result): + return _to_json_compatible(parsed) + return parsed except Exception: # noqa: BLE001 - one bad result never breaks the record return raw_result @@ -239,9 +246,10 @@ def parse_float(text: str) -> float: _MAX_SAFE_INTEGER = 2**53 - 1 # 2**53 - 1 has 16 digits, so its leading digit's exponent is 15. _MAX_SAFE_INTEGER_ADJUSTED = 15 -# DynamoDB's number range, the narrowest of the record destinations: at most -# 38 significant digits and a magnitude from 1E-130 to 9.99...E+125. -_MAX_NUMBER_DIGITS = 38 +_LONG_DIGIT_RUN = re.compile(r"\d{16,}") +# DynamoDB's number range, the narrowest of the record destinations: a +# magnitude from 1E-130 to 9.99...E+125. Its 38-significant-digit limit needs +# no check: a float written exactly has at most 17 significant digits. _MAX_NUMBER_ADJUSTED = 125 _MIN_NUMBER_ADJUSTED = -130 @@ -249,11 +257,11 @@ def parse_float(text: str) -> float: def _to_json_compatible(value: Any) -> Any: """Convert a decoded result into plain JSON values, without recursion. + ``int``: unchanged when ``abs(n) <= 2**53 - 1``, else ``str(n)``. ``Decimal``: an int when integral and ``abs(n) <= 2**53 - 1``, else a float - when it round-trips exactly and is within DynamoDB's number range (at most - 38 significant digits, magnitude 1E-130 to 9.99...E+125), else - ``str(value)``; NaN and Infinity become ``None``. Negative zero becomes - ``0``. ``datetime`` / ``date`` -> ``isoformat()``, ``UUID`` -> string, + when it round-trips exactly and is within DynamoDB's number range + (magnitude 1E-130 to 9.99...E+125), else ``str(value)``; NaN and Infinity + become ``None``. Negative zero becomes ``0``. ``datetime`` / ``date`` -> ``isoformat()``, ``UUID`` -> string, ``bytes`` -> base64, ``tuple`` -> list, ``BatchResult`` -> ``to_dict()``. Non-finite floats become ``None``. """ @@ -284,8 +292,11 @@ def _to_json_compatible(value: Any) -> Any: 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)): + if value is None or isinstance(value, (str, bool)): return value + if isinstance(value, int): + # Beyond 2**53 - 1 a JavaScript reader would round it. + return value if abs(value) <= _MAX_SAFE_INTEGER else str(value) if isinstance(value, decimal.Decimal): if not value.is_finite(): return None @@ -304,11 +315,7 @@ def _scalar_to_json_compatible(value: Any) -> Any: return str(value) as_float = float(value) written = decimal.Decimal(repr(as_float)) - if ( - math.isfinite(as_float) - and written == value - and len(written.as_tuple().digits) <= _MAX_NUMBER_DIGITS - ): + if math.isfinite(as_float) and written == value: return as_float return str(value) if isinstance(value, (datetime.datetime, datetime.date)): @@ -321,8 +328,16 @@ def _scalar_to_json_compatible(value: Any) -> Any: raise TypeError(msg) -# A longer error message keeps its first 8,192 characters followed by "…". -_MAX_ERROR_MESSAGE_CHARS = 8192 +# A longer error name or message keeps its first 8,192 code points followed +# by "…". A Python str is a sequence of code points, so len() and slicing +# count code points and never split a surrogate pair. +_MAX_ERROR_TEXT_CHARS = 8192 + + +def _cap(text: Any) -> Any: + if isinstance(text, str) and len(text) > _MAX_ERROR_TEXT_CHARS: + return text[:_MAX_ERROR_TEXT_CHARS] + "\u2026" + return text def _error_shape(error: ErrorObject) -> dict[str, Any]: @@ -330,13 +345,11 @@ def _error_shape(error: ErrorObject) -> dict[str, Any]: The plugin hooks carry an ``ErrorObject`` (type, message, data, stack trace) and no cause, so records written by this SDK never have ``cause``. - The name is the SDK's ErrorType, unchanged. A message longer than 8,192 - characters is cut to its first 8,192 characters followed by ``…``. + The name is the SDK's ErrorType, otherwise unchanged. A name or message + longer than 8,192 code points is cut to its first 8,192 code points + followed by ``…``. """ - message = error.message - if isinstance(message, str) and len(message) > _MAX_ERROR_MESSAGE_CHARS: - message = message[:_MAX_ERROR_MESSAGE_CHARS] + "\u2026" - return {"name": error.type, "message": message} + return {"name": _cap(error.type), "message": _cap(error.message)} class _ExecutionState: diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_public_path_limits.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_public_path_limits.py index 3398e3a87..1e6893738 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_public_path_limits.py +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_public_path_limits.py @@ -278,6 +278,61 @@ def test_error_messages_are_capped_at_8192_characters(capsys): assert record["operationsByName"]["f"]["error"]["message"] == "m" * 8192 + "…" +def test_error_names_are_capped_at_8192_code_points(capsys): + exporter = LambdaLogExporter(max_record_size_bytes=0) + _run( + exporter, + [], + error=ErrorObject( + message="\U0001f600" * 8193, type="N" * 70_000, data=None, stack_trace=None + ), + ) + record = json.loads(json.loads(_printed_line(capsys))["message"]) + assert record["error"]["name"] == "N" * 8192 + "\u2026" + # Code points, not UTF-16 units: 8,192 whole emoji are kept. + assert record["error"]["message"] == "\U0001f600" * 8192 + "\u2026" + + +def test_long_error_name_fits_a_64_kb_line(capsys): + limit = 65_536 + exporter = LambdaLogExporter(max_record_size_bytes=limit) + ops = [_op(f"q{i}", i, result=json.dumps('"' * 2000)) for i in range(20)] + _run( + exporter, + ops, + error=ErrorObject( + message="m" * 100, type="N" * 70_000, data=None, stack_trace=None + ), + ) + line = _printed_line(capsys) + assert len(line.encode("utf-8")) <= limit + record = json.loads(json.loads(line)["message"]) + assert record["error"]["name"] == "N" * 8192 + "\u2026" + + +def test_worst_case_capped_name_and_message_fit_a_128_kb_line(capsys): + limit = 131_072 + exporter = LambdaLogExporter(max_record_size_bytes=limit) + # \x01 is the worst case for both: 7 bytes each in the printed line. + ops = [_op(f"q{i}", i, result=json.dumps('"' * 2000)) for i in range(40)] + _run( + exporter, + ops, + execution_input='"' * 50_000, + error=ErrorObject( + message="\x01" * 70_000, type="\x01" * 70_000, data=None, stack_trace=None + ), + ) + line = _printed_line(capsys) + assert len(line.encode("utf-8")) <= limit + record = json.loads(json.loads(line)["message"]) + assert record["error"] == { + "name": "\x01" * 8192 + "\u2026", + "message": "\x01" * 8192 + "\u2026", + } + assert record["droppedInput"] is True + + # -- N-B1: a lone surrogate never drops the record ----------------------------- @@ -358,3 +413,56 @@ def test_lambda_log_measure_is_the_printed_line_size(capsys): exporter.export(record) assert size == len(_printed_line(capsys).encode("utf-8")) assert exporter.measure({"a": object()}) is None + + +# -- non-blocking 2: integers beyond 2**53 - 1 ---------------------------------- + + +def _results(capsys: pytest.CaptureFixture[str], *results: str) -> list[Any]: + ops = [_op(f"op{i}", i, result=raw) for i, raw in enumerate(results)] + _run(LambdaLogExporter(), ops) + by_name = json.loads(json.loads(_printed_line(capsys))["message"])[ + "operationsByName" + ] + return [by_name[f"op{i}"]["result"] for i in range(len(results))] + + +def test_large_ints_become_strings(capsys): + big = 12345678901234567 + safe = 2**53 - 1 + results = _results( + capsys, + _envelope({"x": big, "y": safe, "z": -big}), + _envelope((1, -safe)), + str(big), + json.dumps([safe, -(2**60), {"k": 2**53}]), + # Long digit runs in strings and floats are left alone. + json.dumps({"s": "12345678901234567", "f": 0.12345678901234568}), + str(safe), + ) + assert results == [ + {"x": "12345678901234567", "y": safe, "z": "-12345678901234567"}, + [1, -safe], + "12345678901234567", + [safe, str(-(2**60)), {"k": str(2**53)}], + {"s": "12345678901234567", "f": 0.12345678901234568}, + safe, + ] + + +# -- non-blocking 7: integers with trailing zeros stay DynamoDB numbers --------- + + +def test_dynamodb_marshal_keeps_integers_with_few_significant_digits(): + exporter = DynamoDBExporter(table_name="t", client=StubDynamoDBClient()) + marshalled = exporter._marshal( + {"a": [10**38, -(10**38), 10**125, 10**126, 10**38 + 1, 1.0e38]} + ) + assert marshalled["a"]["L"] == [ + {"N": "1E+38"}, + {"N": "-1E+38"}, + {"N": "1E+125"}, + {"S": str(10**126)}, + {"S": str(10**38 + 1)}, + {"N": "1E+38"}, + ] From 060c382cc5677ef3e17646215cadcbcb82dd84fd Mon Sep 17 00:00:00 2001 From: Pooya Paridel Date: Thu, 22 Oct 2026 12:35:24 -0700 Subject: [PATCH 6/6] docs(telemetry): describe name cap, int bound and size assumptions State that names and messages are capped at 8,192 code points, the ARN and name length assumptions behind the 64 KB statement, the 128 KB worst case, the 2**53 - 1 bound for int results, lone surrogate escaping in LambdaLogExporter, and that a DynamoDB attribute can be a number in one item and a string in another. --- .../README.md | 39 +++++++++++++------ 1 file changed, 27 insertions(+), 12 deletions(-) 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 133f84c16..500bc8c53 100644 --- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/README.md +++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/README.md @@ -97,7 +97,10 @@ trailing newline. Truncation measures that line in one pass. A CloudWatch Logs event can be up to 1 MB ([PutLogEvents](https://docs.aws.amazon.com/AmazonCloudWatchLogs/latest/APIReference/API_PutLogEvents.html)). U+2028 and U+2029 are written as `\u2028` and `\u2029`, so a tool that treats -them as line breaks still sees one line per record. +them as line breaks still sees one line per record. A lone surrogate (a string +character with no UTF-8 encoding) is written as its `\uXXXX` escape, so +printing it never fails. The limit counts these escapes; other exporters' +limits count the characters as they write them. The lines have `level` `INFO`. With the JSON log format and an application log level of `WARN` or above, Lambda drops them. With the Text log format, Lambda @@ -120,7 +123,11 @@ default `sort_key="sk"` (= `emittedAt`) every export adds an item; pass `sort_key=None` for a key-only table that upserts. IAM: `dynamodb:PutItem`. A number DynamoDB cannot store (more than 38 significant digits, a magnitude outside 1E-130 to 9.99...E+125, NaN or Infinity) is written as a string -attribute holding its JSON text, so it never drops the item. +attribute holding its JSON text, so it never drops the item. The same field can +therefore be a number (`N`) in one item and a string (`S`) in another; readers +that filter or compare on the attribute type should allow for both. An integer +with more than 38 digits but at most 38 significant digits, such as `10**38`, +is stored as a number. ```python DynamoDBExporter(table_name="durable-execution-telemetry") @@ -270,15 +277,20 @@ the execution `error`, not an operation error. The exporter does not build a cause from other operations or from `ErrorData`. `include_errors=False` removes operation errors. -Every `message` is capped at 8,192 characters. A longer message keeps its first -8,192 characters followed by `…`. +Every `name` and `message` is capped at 8,192 characters, counted as Unicode +code points (a Python `str` is a sequence of code points, so a cut never splits +a character). A longer value keeps its first 8,192 code points followed by `…`. When a record is over its exporter's size limit, truncation first removes every `cause` and sets `droppedCauses: true`, then runs its usual steps (operation results, then whole operations, then input, then output). Python records have -no cause, so the first step never changes them. With the message cap, a record -whose operations, input and output are dropped fits any limit of 64 KB or -more. +no cause, so the first step never changes them. With the name and message +caps, a record whose operations, input and output are dropped fits any limit of +64 KB or more, assuming ARN fields of normal length (a few hundred bytes, as in +real Lambda ARNs) and an error `name` of normal length (a type name such as +`module.QualName`). Without the `name` assumption, the worst case (a name and +message of 8,192 control characters, each escaped as `\u00XX`) fits any limit +of 128 KB (131,072 bytes) or more. Names are each language's own and differ across SDKs. For example, a step that throws `PaymentError` can be reported as `StepError` in JS, @@ -294,11 +306,12 @@ default serializer's (`ExtendedTypeSerDes`) envelope, a JSON object whose keys are exactly `t` and `v` with a string `t`, it is decoded with that serializer and converted to plain JSON: +- `int`: unchanged when its absolute value is at most `2**53 - 1` + (9007199254740991), the largest integer a JavaScript reader parses exactly. + Otherwise `str(value)`, for example `"12345678901234567"`. - `Decimal`: an integer when it is integral and its absolute value is at most - `2**53 - 1` (9007199254740991), the largest integer a JavaScript reader parses - exactly. Otherwise a float when the float round-trips exactly and is within - DynamoDB's number range (at most 38 significant digits, magnitude 1E-130 to - 9.99...E+125). Otherwise `str(value)`, for example `"9007199254740993"`, + `2**53 - 1`. Otherwise a float when the float round-trips exactly and is + within DynamoDB's number range (magnitude 1E-130 to 9.99...E+125). Otherwise `str(value)`, for example `"9007199254740993"`, `"12345678901234567.89"` or `"1E+299"`. NaN and Infinity become `null`. Negative zero (`Decimal("-0")`, `Decimal("-0.0")`) is integral, so it becomes `0` and loses its sign. @@ -306,7 +319,9 @@ and converted to plain JSON: `tuple`: list. `BatchResult`: `to_dict()`. Any other result is parsed as JSON, or kept as the raw string when it is not -JSON. In parsed JSON, NaN and Infinity become `null`; nothing else changes. A +JSON. In parsed JSON, NaN and Infinity become `null`, and an integer whose +absolute value is greater than `2**53 - 1` becomes its decimal string, as +above; nothing else changes. A result that fails to decode keeps its parsed or raw value, and the rest of the record is still emitted.