diff --git a/packages/aws-durable-execution-sdk-python-insight/README.md b/packages/aws-durable-execution-sdk-python-insight/README.md index 17b5617b..0febae3a 100644 --- a/packages/aws-durable-execution-sdk-python-insight/README.md +++ b/packages/aws-durable-execution-sdk-python-insight/README.md @@ -61,16 +61,22 @@ Behavior is validated cross-SDK by the `insight` conformance suite > workflow progress. Because each configured exporter is driven by its own > single background worker, each exporter object may belong to only one live > `WorkflowInsightPlugin`: listing it twice or sharing it across plugin instances -> raises `ValueError`. Separate instances of the same exporter class (e.g. two -> `S3Exporter`s for different buckets) are fine. A blocked lane retains at most -> 1,024 pending executions and 16 MB of estimated retained memory; it drops the -> oldest pending snapshot when either bound is reached. Rapid cumulative -> snapshots for one execution are coalesced, so a lane may skip intermediate -> `on-change` records; the terminal record is always delivered under normal -> completion. At invocation end the plugin drains and flushes the touched -> exporters under a single shared deadline +> raises `ValueError`. Separate instances of the same exporter class are fine. +> Each lane keeps up to 16 pending snapshots per execution, 1,024 records total, +> and 16 MB of estimated retained memory. When a bound fills, it drops the oldest +> pending snapshot so recent progress and terminal snapshots are retained. At +> invocation end the plugin drains and flushes the touched exporters under a +> single shared deadline > (`WorkflowInsightConfig.export_timeout_seconds`, default `5.0`); on timeout the > workflow response is returned and record delivery degrades to best-effort. +> +> `flush()` is lane-wide, not execution-scoped: it applies to the configured +> exporter instance's entire buffer. A barrier may therefore publish records +> from another execution that were already buffered, while a record scheduled +> after that barrier is exported after the flush and waits for a later barrier. +> A custom batching exporter that requires execution-level isolation should key +> its buffer by `executionArn` or use a distinct exporter instance per isolated +> stream. ## Requirements diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/_export_scheduler.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/_export_scheduler.py index 5e0d4298..f1de78b5 100644 --- a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/_export_scheduler.py +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/_export_scheduler.py @@ -9,10 +9,9 @@ Each exporter lane: -* Keeps the latest pending snapshot per execution ARN and processes ARNs - round-robin. -* Drops the oldest pending snapshot when the execution-count or byte budget is - reached, keeping memory bounded. +* Keeps a bounded FIFO per execution ARN and processes ARNs round-robin. +* Drops the oldest pending snapshots when the per-execution, lane-wide record, + or byte limit is reached, favoring recent progress and terminal snapshots. * Uses a lane-wide flush barrier at invocation end. All barriers share one timeout; timed-out barriers are removed without replacing a blocked worker. * Stops its worker after an invocation has drained and the lane becomes idle. @@ -30,7 +29,8 @@ import time import types import uuid -from collections import OrderedDict, deque +from collections import deque +from dataclasses import dataclass from typing import Any from aws_durable_execution_sdk_python_insight.truncation import truncate_record @@ -45,9 +45,19 @@ # bounded regardless of how long a worker stays blocked. _DEFAULT_MAX_PENDING_EXECUTIONS = 1024 -# Estimated Python object memory retained by one blocked lane. Lambda functions -# can be configured with 128 MiB, so keep the instrumentation backlog well below -# that floor. This is a conservative budget signal, not an exact heap measurement. +# Upper bound on all pending records in a lane. Keeping this equal to the +# original distinct-execution cap preserves the scheduler's previous worst-case +# record count even though one execution can now retain a short burst. +_DEFAULT_MAX_PENDING_RECORDS = 1024 + +# Upper bound on records waiting for one execution in one lane. This is large +# enough to preserve the known 11-progress-plus-terminal burst without relying +# on daemon-thread scheduling, while still bounding memory behind a blocked +# exporter. The in-flight record is not included in this count. +_DEFAULT_MAX_PENDING_RECORDS_PER_EXECUTION = 16 + +# Estimated Python object memory retained by one blocked lane. This keeps the +# instrumentation backlog well below Lambda's 128 MiB memory floor. _DEFAULT_MAX_PENDING_BYTES = 16_000_000 @@ -242,6 +252,16 @@ def _copy_record_containers(record: dict[str, Any]) -> dict[str, Any]: _RECORD = "record" _FLUSH = "flush" +_RecordToken = tuple[str, int] + + +@dataclass(slots=True) +class _PendingRecord: + sequence: int + generation: int + value: dict[str, Any] + size: int + class _FlushBarrier: """A one-shot flush marker the invocation-end thread waits on. @@ -280,10 +300,16 @@ def __init__( exporter: InsightExporter, *, max_pending_executions: int = _DEFAULT_MAX_PENDING_EXECUTIONS, + max_pending_records: int = _DEFAULT_MAX_PENDING_RECORDS, + max_pending_records_per_execution: int = ( + _DEFAULT_MAX_PENDING_RECORDS_PER_EXECUTION + ), max_pending_bytes: int = _DEFAULT_MAX_PENDING_BYTES, ) -> None: self._exporter = exporter self._max_pending = max(1, max_pending_executions) + self._max_pending_records = max(1, max_pending_records) + self._max_pending_per_execution = max(1, max_pending_records_per_execution) self._max_pending_bytes = max(1, max_pending_bytes) self._pending_bytes = 0 self._inflight_bytes = 0 @@ -294,12 +320,15 @@ def __init__( # makes any accidental recursive acquisition fail loudly instead of # silently succeeding. self._cond = threading.Condition(threading.Lock()) - # Ordered work list: entries are (_RECORD, arn) or (_FLUSH, barrier). + # Ordered work list: entries are (_RECORD, (arn, generation)) or + # (_FLUSH, barrier). A generation changes whenever a flush is queued, so + # one ARN can have independent tokens on both sides of a barrier. self._queue: deque[tuple[str, Any]] = deque() - # arn -> (latest pending record, retained-memory estimate). Insertion - # order is both record age and fairness order because replacing an ARN - # moves it to the back. - self._pending: OrderedDict[str, tuple[dict[str, Any], int]] = OrderedDict() + # arn -> FIFO ordered by record sequence. Queue-token order, not this + # mapping, controls round-robin fairness. + self._pending: dict[str, deque[_PendingRecord]] = {} + self._generation = 0 + self._next_sequence = 0 self._stop_when_idle = False self._worker: threading.Thread | None = None self._disabled = False @@ -318,37 +347,45 @@ def schedule( self._stop_when_idle = False size = max(0, record_size) if size > self._max_pending_bytes: - superseded = self._pending.pop(execution_arn, None) - if superseded is not None: - _, superseded_size = superseded - self._pending_bytes -= superseded_size - self._remove_record_token(execution_arn) _logger.warning( "workflow-insight: pending record for %s on %s exceeds the " - "byte budget (%d > %d); dropping this record%s", + "byte budget (%d > %d); dropping this record", execution_arn, type(self._exporter).__name__, size, self._max_pending_bytes, - " and its superseded pending snapshot" - if superseded is not None - else "", ) return - if execution_arn in self._pending: - # Coalesce: replace the pending record and move it to the back so - # a busy execution cannot starve the others. - _, old_size = self._pending[execution_arn] - self._pending_bytes -= old_size - self._pending[execution_arn] = (record, size) - self._pending_bytes += size - self._pending.move_to_end(execution_arn) - self._move_record_token_to_back(execution_arn) + pending = self._pending.get(execution_arn) + if pending is None: + pending = deque() + self._pending[execution_arn] = pending + elif len(pending) >= self._max_pending_per_execution: + self._drop_oldest_pending_record(execution_arn) + pending = self._pending.setdefault(execution_arn, deque()) + _logger.warning( + "workflow-insight: pending export FIFO for %s on %s is " + "full (cap=%d); dropping oldest pending record", + execution_arn, + type(self._exporter).__name__, + self._max_pending_per_execution, + ) + + generation = self._generation + has_generation = bool(pending and pending[-1].generation == generation) + pending.append( + _PendingRecord(self._next_sequence, generation, record, size) + ) + self._pending_bytes += size + self._next_sequence += 1 + token = (execution_arn, generation) + if has_generation: + self._move_record_token_to_back(token) else: - self._pending[execution_arn] = (record, size) - self._pending_bytes += size - self._queue.append((_RECORD, execution_arn)) - self._enforce_pending_caps() + self._queue.append((_RECORD, token)) + self._enforce_pending_execution_cap() + self._enforce_pending_record_cap() + self._enforce_pending_byte_cap() self._ensure_worker_locked() self._cond.notify() @@ -361,6 +398,7 @@ def enqueue_flush(self) -> _FlushBarrier: barrier.complete() return barrier self._queue.append((_FLUSH, barrier)) + self._generation += 1 self._ensure_worker_locked() self._cond.notify() return barrier @@ -395,31 +433,66 @@ def cancel_flush(self, barrier: _FlushBarrier) -> None: # -- queue bookkeeping (must hold ``_cond``) ------------------------------ - def _move_record_token_to_back(self, execution_arn: str) -> None: + def _move_record_token_to_back(self, token: _RecordToken) -> None: for index, (kind, payload) in enumerate(self._queue): - if kind == _RECORD and payload == execution_arn: + if kind == _RECORD and payload == token: del self._queue[index] - self._queue.append((_RECORD, execution_arn)) + self._queue.append((_RECORD, token)) return - # No token means the arn is currently in flight; a fresh token will be - # appended when it leaves flight (the next schedule sees it absent from - # ``_pending``), which yields the "export A then latest" behavior. - def _enforce_pending_caps(self) -> None: + def _requeue_record_before_flush(self, token: _RecordToken) -> None: + """Requeue behind peers without crossing this ARN's next generation.""" + execution_arn, generation = token + for index, (kind, payload) in enumerate(self._queue): + later_same_arn = ( + kind == _RECORD + and payload[0] == execution_arn + and payload[1] > generation + ) + if kind == _FLUSH or later_same_arn: + self._queue.insert(index, (_RECORD, token)) + return + self._queue.append((_RECORD, token)) + + def _oldest_pending_arn(self) -> str: + return min( + self._pending, + key=lambda arn: self._pending[arn][0].sequence, + ) + + def _enforce_pending_execution_cap(self) -> None: while len(self._pending) > self._max_pending: - old_arn, _ = self._drop_oldest_pending() + old_arn = self._oldest_pending_arn() + self._drop_pending_execution(old_arn) _logger.warning( - "workflow-insight: export lane for %s reached its execution cap " - "(%d); dropping pending record for %s", + "workflow-insight: export lane for %s is full " + "(cap=%d); dropping pending records for %s", type(self._exporter).__name__, self._max_pending, old_arn, ) + + def _enforce_pending_record_cap(self) -> None: + while sum(len(records) for records in self._pending.values()) > ( + self._max_pending_records + ): + old_arn = self._oldest_pending_arn() + self._drop_oldest_pending_record(old_arn) + _logger.warning( + "workflow-insight: export lane for %s reached its pending " + "record cap (%d); dropping oldest pending record for %s", + type(self._exporter).__name__, + self._max_pending_records, + old_arn, + ) + + def _enforce_pending_byte_cap(self) -> None: while ( self._pending_bytes + self._inflight_bytes > self._max_pending_bytes and self._pending ): - old_arn, dropped_size = self._drop_oldest_pending() + old_arn = self._oldest_pending_arn() + dropped_size = self._drop_oldest_pending_record(old_arn) _logger.warning( "workflow-insight: export lane for %s reached its pending byte " "budget (%d); dropping %d-byte pending record for %s", @@ -429,15 +502,28 @@ def _enforce_pending_caps(self) -> None: old_arn, ) - def _drop_oldest_pending(self) -> tuple[str, int]: - old_arn, (_, old_size) = self._pending.popitem(last=False) - self._pending_bytes -= old_size - self._remove_record_token(old_arn) - return old_arn, old_size + def _drop_oldest_pending_record(self, execution_arn: str) -> int: + records = self._pending[execution_arn] + dropped = records.popleft() + self._pending_bytes -= dropped.size + token = (execution_arn, dropped.generation) + if not records or records[0].generation != dropped.generation: + self._remove_record_token(token) + if not records: + del self._pending[execution_arn] + return dropped.size + + def _drop_pending_execution(self, execution_arn: str) -> None: + records = self._pending.pop(execution_arn) + self._pending_bytes -= sum(record.size for record in records) + for index in range(len(self._queue) - 1, -1, -1): + kind, payload = self._queue[index] + if kind == _RECORD and payload[0] == execution_arn: + del self._queue[index] - def _remove_record_token(self, execution_arn: str) -> None: + def _remove_record_token(self, token: _RecordToken) -> None: for index, (kind, payload) in enumerate(self._queue): - if kind == _RECORD and payload == execution_arn: + if kind == _RECORD and payload == token: del self._queue[index] return @@ -497,12 +583,21 @@ def _run_worker(self) -> None: record: dict[str, Any] | None = None record_size = 0 if kind == _RECORD: - pending = self._pending.pop(payload, None) - if pending is None: + token: _RecordToken = payload + execution_arn, generation = token + pending = self._pending.get(execution_arn) + if not pending or pending[0].generation != generation: continue - record, record_size = pending + pending_record = pending.popleft() + record_size = pending_record.size self._pending_bytes -= record_size self._inflight_bytes += record_size + record = pending_record.value + if pending and pending[0].generation == generation: + # One record per ARN turn within this barrier generation. + self._requeue_record_before_flush(token) + elif not pending: + del self._pending[execution_arn] if kind == _RECORD and record is not None: try: @@ -572,6 +667,10 @@ def _pending_count(self) -> int: with self._cond: return len(self._pending) + def _pending_record_count(self) -> int: + with self._cond: + return sum(len(records) for records in self._pending.values()) + def _pending_bytes_count(self) -> int: with self._cond: return self._pending_bytes @@ -597,6 +696,10 @@ def __init__( exporters: list[InsightExporter], *, max_pending_executions: int = _DEFAULT_MAX_PENDING_EXECUTIONS, + max_pending_records: int = _DEFAULT_MAX_PENDING_RECORDS, + max_pending_records_per_execution: int = ( + _DEFAULT_MAX_PENDING_RECORDS_PER_EXECUTION + ), max_pending_bytes: int = _DEFAULT_MAX_PENDING_BYTES, ) -> None: self._max_pending_bytes = max(1, max_pending_bytes) @@ -604,6 +707,8 @@ def __init__( _ExporterLane( exporter, max_pending_executions=max_pending_executions, + max_pending_records=max_pending_records, + max_pending_records_per_execution=max_pending_records_per_execution, max_pending_bytes=self._max_pending_bytes, ) for exporter in exporters diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/e2e/parallel_fan_in_int_test.py b/packages/aws-durable-execution-sdk-python-insight/tests/e2e/parallel_fan_in_int_test.py new file mode 100644 index 00000000..950f58e8 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/e2e/parallel_fan_in_int_test.py @@ -0,0 +1,124 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""End-to-end test for on-change export bursts during parallel fan-in.""" + +from __future__ import annotations + +import threading +import time +from typing import Any + +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.execution import ( + InvocationStatus, + durable_execution, +) + +from aws_durable_execution_sdk_python_insight import ( + WorkflowInsightConfig, + workflow_insight, +) +from aws_durable_execution_sdk_python_testing.runner import ( + DurableFunctionTestResult, + DurableFunctionTestRunner, +) + + +_BRANCH_COUNT = 6 + + +class _BlockingCaptureExporter: + """Blocks the first export so the real hook burst queues deterministically.""" + + def __init__(self) -> None: + self.max_record_size_bytes: int | None = None + self.started = threading.Event() + self.release = threading.Event() + self.records: list[dict[str, Any]] = [] + self._lock = threading.Lock() + + def render(self, record: dict[str, Any]) -> dict[str, Any]: + return record + + def export(self, record: dict[str, Any]) -> None: + if not self.started.is_set(): + self.started.set() + self.release.wait(10.0) + with self._lock: + self.records.append(record) + + def flush(self) -> None: + return None + + def snapshots(self) -> list[dict[str, Any]]: + with self._lock: + return list(self.records) + + +def _branch(index: int): + def run(context: DurableContext) -> int: + return context.step(lambda _step_context: index, name=f"step-{index}") + + return run + + +def _parallel_handler(event: Any, context: DurableContext) -> list[int]: # noqa: ARG001 + return context.parallel( + [_branch(index) for index in range(_BRANCH_COUNT)], + name="fan-in", + ).get_results() + + +def _wait_until(predicate, timeout: float = 10.0) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.005) + return predicate() + + +def test_parallel_fan_in_preserves_every_on_change_snapshot() -> None: + capture = _BlockingCaptureExporter() + plugin = workflow_insight( + WorkflowInsightConfig( + exporters=[capture], + emit_mode="on-change", + operation_detail="full-tree", + ) + ) + handler = durable_execution(_parallel_handler, plugins=[plugin]) + results: list[DurableFunctionTestResult] = [] + + with DurableFunctionTestRunner(handler=handler, execution_timeout=15) as runner: + run_thread = threading.Thread( + target=lambda: results.append(runner.run(input="{}")), + daemon=True, + ) + run_thread.start() + try: + assert capture.started.wait(5.0) + lane = plugin._scheduler._lanes[0] + # Invocation start is in flight. Three real PluginExecutor changes + # and the terminal record must queue behind it before release. + assert _wait_until(lambda: lane._pending_record_count() == 4) + finally: + capture.release.set() + run_thread.join(10.0) + + assert not run_thread.is_alive() + assert len(results) == 1 + assert results[0].status is InvocationStatus.SUCCEEDED + + records = capture.snapshots() + assert [record["status"] for record in records] == [ + "RUNNING", + "RUNNING", + "RUNNING", + "RUNNING", + "SUCCEEDED", + ] + final_names = {operation["name"] for operation in records[-1]["operations"]} + assert "fan-in" in final_names + assert {f"step-{index}" for index in range(_BRANCH_COUNT)} <= final_names diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_export_scheduler.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_export_scheduler.py index aea400f5..1ae5b5cd 100644 --- a/packages/aws-durable-execution-sdk-python-insight/tests/test_export_scheduler.py +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_export_scheduler.py @@ -5,7 +5,7 @@ These drive the scheduler directly with plain record dicts and purpose-built exporter doubles. Synchronization uses events/predicates (not sleeps) so the -coalescing, fairness, drain, flush-ordering, timeout and thread-lifecycle +bounded FIFO, fairness, drain, flush-ordering, timeout and thread-lifecycle invariants are asserted deterministically rather than by timing luck. """ @@ -138,6 +138,24 @@ def release_flush(self) -> None: self._flush_release.set() +class FirstExportBlockingRecorder(RecordingExporter): + """Records call order but blocks the first export until released.""" + + def __init__(self) -> None: + super().__init__() + self.started = threading.Event() + self._release = threading.Event() + + def export(self, record: dict[str, Any]) -> None: + if not self.started.is_set(): + self.started.set() + self._release.wait(5.0) + super().export(record) + + def release(self) -> None: + self._release.set() + + class BlockingBufferedExporter: """Blocks export and publishes buffered records only when flush runs.""" @@ -298,19 +316,19 @@ def test_repeated_scheduling_does_not_grow_threads(): assert _wait_until(lambda: not lane._worker_alive()) -# -- coalescing / fairness / isolation --------------------------------------- +# -- bounded FIFO / fairness / isolation ------------------------------------- -def test_same_execution_coalescing_exports_inflight_then_latest(): +def test_same_execution_fifo_exports_all_pending_records(): exporter = BlockingExporter() scheduler = _ExportScheduler([exporter]) scheduler.schedule(ARN_A, _rec(ARN_A, "a1")) assert _wait_until(exporter.started.is_set) # a1 is in flight - # While a1 is in flight, a2 and a3 arrive and coalesce to the latest (a3). + # While a1 is in flight, a2 and a3 remain ordered in the pending FIFO. scheduler.schedule(ARN_A, _rec(ARN_A, "a2")) scheduler.schedule(ARN_A, _rec(ARN_A, "a3")) exporter.release() - assert _wait_until(lambda: exporter.exported_values() == ["a1", "a3"]) + assert _wait_until(lambda: exporter.exported_values() == ["a1", "a2", "a3"]) scheduler.end_invocation(5.0) @@ -320,16 +338,15 @@ def test_different_executions_isolated_and_fair(): scheduler.schedule(ARN_A, _rec(ARN_A, "a1")) assert _wait_until(exporter.started.is_set) # a1 in flight scheduler.schedule(ARN_B, _rec(ARN_B, "b1")) # queued: [B] - scheduler.schedule(ARN_B, _rec(ARN_B, "b2")) # coalesce B -> b2 + scheduler.schedule(ARN_B, _rec(ARN_B, "b2")) # B FIFO: [b1, b2] scheduler.schedule(ARN_A, _rec(ARN_A, "a2")) # queued: [B, A] exporter.release() - # a1 (in flight) first, then FIFO fairness B before the re-added A, each - # carrying its latest coalesced value. - assert _wait_until(lambda: exporter.exported_values() == ["a1", "b2", "a2"]) + # a1 (in flight) first, then one record per ARN turn: B, A, then B again. + assert _wait_until(lambda: exporter.exported_values() == ["a1", "b1", "a2", "b2"]) scheduler.end_invocation(5.0) -def test_terminal_record_supersedes_pending_running(): +def test_terminal_record_follows_pending_running_records(): exporter = BlockingExporter() scheduler = _ExportScheduler([exporter]) scheduler.schedule(ARN_A, _rec(ARN_A, "r1", status="RUNNING")) @@ -337,7 +354,22 @@ def test_terminal_record_supersedes_pending_running(): scheduler.schedule(ARN_A, _rec(ARN_A, "r2", status="RUNNING")) scheduler.schedule(ARN_A, _rec(ARN_A, "final", status="SUCCEEDED")) exporter.release() - assert _wait_until(lambda: exporter.exported_values() == ["r1", "final"]) + assert _wait_until(lambda: exporter.exported_values() == ["r1", "r2", "final"]) + scheduler.end_invocation(5.0) + + +def test_pending_fifo_cap_drops_oldest_and_retains_terminal(): + exporter = BlockingExporter() + scheduler = _ExportScheduler([exporter], max_pending_records_per_execution=2) + scheduler.schedule(ARN_A, _rec(ARN_A, "inflight")) + assert _wait_until(exporter.started.is_set) + scheduler.schedule(ARN_A, _rec(ARN_A, "oldest")) + scheduler.schedule(ARN_A, _rec(ARN_A, "newer")) + scheduler.schedule(ARN_A, _rec(ARN_A, "final", status="SUCCEEDED")) + exporter.release() + assert _wait_until( + lambda: exporter.exported_values() == ["inflight", "newer", "final"] + ) scheduler.end_invocation(5.0) @@ -439,6 +471,93 @@ def test_flush_happens_after_export(): assert kinds == ["export", "flush"] +def test_flush_barrier_waits_for_entire_pending_fifo(): + exporter = FirstExportBlockingRecorder() + scheduler = _ExportScheduler([exporter]) + lane = scheduler._lanes[0] + scheduler.schedule(ARN_A, _rec(ARN_A, "a1")) + assert _wait_until(exporter.started.is_set) + scheduler.schedule(ARN_A, _rec(ARN_A, "a2")) + scheduler.schedule(ARN_A, _rec(ARN_A, "a3")) + barrier = lane.enqueue_flush() + exporter.release() + + assert barrier.wait(5.0) + lane.request_stop_when_idle() + assert _wait_until(lambda: not lane._worker_alive()) + assert exporter.calls == [ + ("export", "a1"), + ("export", "a2"), + ("export", "a3"), + ("flush", None), + ] + + +def test_flush_barrier_splits_same_execution_fifo_by_generation(): + exporter = FirstExportBlockingRecorder() + scheduler = _ExportScheduler([exporter]) + lane = scheduler._lanes[0] + scheduler.schedule(ARN_A, _rec(ARN_A, "a1")) + assert _wait_until(exporter.started.is_set) + + scheduler.schedule(ARN_A, _rec(ARN_A, "a2")) + barrier = lane.enqueue_flush() + scheduler.schedule(ARN_A, _rec(ARN_A, "a3")) + exporter.release() + + assert barrier.wait(5.0) + lane.request_stop_when_idle() + assert _wait_until(lambda: not lane._worker_alive()) + assert exporter.calls == [ + ("export", "a1"), + ("export", "a2"), + ("flush", None), + ("export", "a3"), + ] + + +def test_record_scheduled_after_barrier_waits_for_later_flush(): + exporter = FirstExportBlockingRecorder() + scheduler = _ExportScheduler([exporter]) + lane = scheduler._lanes[0] + scheduler.schedule(ARN_A, _rec(ARN_A, "a1")) + assert _wait_until(exporter.started.is_set) + + barrier = lane.enqueue_flush() + scheduler.schedule(ARN_A, _rec(ARN_A, "a2")) + exporter.release() + + assert barrier.wait(5.0) + lane.request_stop_when_idle() + assert _wait_until(lambda: not lane._worker_alive()) + assert exporter.calls == [ + ("export", "a1"), + ("flush", None), + ("export", "a2"), + ] + + +def test_lane_flush_includes_other_execution_records_already_queued(): + exporter = FirstExportBlockingRecorder() + scheduler = _ExportScheduler([exporter]) + lane = scheduler._lanes[0] + scheduler.schedule(ARN_A, _rec(ARN_A, "a1")) + assert _wait_until(exporter.started.is_set) + + scheduler.schedule(ARN_B, _rec(ARN_B, "b1")) + barrier = lane.enqueue_flush() + exporter.release() + + assert barrier.wait(5.0) + lane.request_stop_when_idle() + assert _wait_until(lambda: not lane._worker_alive()) + assert exporter.calls == [ + ("export", "a1"), + ("export", "b1"), + ("flush", None), + ] + + def test_export_and_flush_exceptions_are_isolated(): failing = FailingExporter() good = RecordingExporter() @@ -564,23 +683,20 @@ def test_pending_execution_cap_evicts_oldest(): scheduler.end_invocation(5.0) -def test_pending_byte_budget_evicts_oldest_large_record(): +def test_pending_record_cap_preserves_original_lane_memory_bound(): exporter = BlockingExporter() - scheduler = _ExportScheduler([exporter], max_pending_bytes=3_000) + scheduler = _ExportScheduler([exporter], max_pending_records=3) lane = scheduler._lanes[0] - scheduler.schedule(ARN_A, _rec(ARN_A, "inflight")) - assert _wait_until(exporter.started.is_set) - - scheduler.schedule(ARN_B, _rec(ARN_B, "b" * 1_500)) - scheduler.schedule(ARN_C, _rec(ARN_C, "c" * 1_500)) - - assert lane._pending_count() == 1 - assert lane._pending_bytes_count() <= 3_000 + scheduler.schedule(ARN_A, _rec(ARN_A, "a1")) + assert _wait_until(exporter.started.is_set) # a1 in flight (not pending) + scheduler.schedule(ARN_B, _rec(ARN_B, "b1")) + scheduler.schedule(ARN_C, _rec(ARN_C, "c1")) + scheduler.schedule(ARN_D, _rec(ARN_D, "d1")) + scheduler.schedule(ARN_A, _rec(ARN_A, "a2")) + assert lane._pending_record_count() == 3 exporter.release() scheduler.end_invocation(5.0) - exported = exporter.exported_values() - assert exported[0] == "inflight" - assert exported[1] == "c" * 1_500 + assert exporter.exported_values() == ["a1", "c1", "d1", "a2"] def test_non_json_record_reaches_exporter_without_evicting_backlog(): @@ -621,24 +737,25 @@ def test_individually_over_budget_record_does_not_evict_existing_backlog(): assert exported[1:] == ["b" * 700, "c" * 700] -def test_over_budget_replacement_removes_superseded_same_arn_only(): +def test_over_budget_record_preserves_accepted_same_arn_fifo(): exporter = BlockingExporter() scheduler = _ExportScheduler([exporter], max_pending_bytes=3_500) lane = scheduler._lanes[0] scheduler.schedule(ARN_A, _rec(ARN_A, "inflight")) assert _wait_until(exporter.started.is_set) - scheduler.schedule(ARN_A, _rec(ARN_A, "stale-running")) + scheduler.schedule(ARN_A, _rec(ARN_A, "accepted-running")) scheduler.schedule(ARN_B, _rec(ARN_B, "unrelated")) scheduler.schedule( ARN_A, _rec(ARN_A, "terminal" * 500, status="SUCCEEDED"), ) - assert lane._pending_count() == 1 + assert lane._pending_count() == 2 + assert lane._pending_record_count() == 2 assert lane._pending_bytes_count() <= 3_500 exporter.release() scheduler.end_invocation(5.0) - assert exporter.exported_values() == ["inflight", "unrelated"] + assert exporter.exported_values() == ["inflight", "accepted-running", "unrelated"] def test_retained_size_traverses_slots_after_shallow_size_failure(): @@ -714,6 +831,68 @@ def test_record_sizing_exception_does_not_escape_schedule(): assert exporter.exported_values() == ["inflight", "custom-sized"] +def test_pending_record_cap_evicts_true_oldest_across_arns(): + exporter = BlockingExporter() + scheduler = _ExportScheduler([exporter], max_pending_records=2) + scheduler.schedule(ARN_C, _rec(ARN_C, "inflight")) + assert _wait_until(exporter.started.is_set) + + scheduler.schedule(ARN_A, _rec(ARN_A, "a1")) + scheduler.schedule(ARN_B, _rec(ARN_B, "b1", status="SUCCEEDED")) + scheduler.schedule(ARN_A, _rec(ARN_A, "a2")) + + exporter.release() + scheduler.end_invocation(5.0) + # A1 is globally oldest. B1 remains even though scheduling A2 moved A's + # fairness token behind B's token. + assert exporter.exported_values() == ["inflight", "b1", "a2"] + + +def test_pending_byte_budget_evicts_oldest_large_record(): + exporter = BlockingExporter() + scheduler = _ExportScheduler([exporter], max_pending_bytes=3_000) + lane = scheduler._lanes[0] + scheduler.schedule(ARN_A, _rec(ARN_A, "inflight")) + assert _wait_until(exporter.started.is_set) + + scheduler.schedule(ARN_B, _rec(ARN_B, "b" * 1_500)) + scheduler.schedule(ARN_C, _rec(ARN_C, "c" * 1_500)) + + assert lane._pending_record_count() == 1 + assert lane._pending_bytes_count() <= 3_000 + exporter.release() + scheduler.end_invocation(5.0) + exported = exporter.exported_values() + assert exported[0] == "inflight" + assert exported[1] == "c" * 1_500 + + +def test_cancelled_barrier_preserves_generation_order_and_terminal(): + exporter = BlockingExporter() + scheduler = _ExportScheduler([exporter]) + lane = scheduler._lanes[0] + scheduler.schedule(ARN_A, _rec(ARN_A, "inflight")) + assert _wait_until(exporter.started.is_set) + + scheduler.schedule(ARN_A, _rec(ARN_A, "pre-1")) + scheduler.schedule(ARN_A, _rec(ARN_A, "pre-2")) + barrier = lane.enqueue_flush() + scheduler.schedule(ARN_A, _rec(ARN_A, "terminal", status="SUCCEEDED")) + lane.cancel_flush(barrier) + lane.request_stop_when_idle() + exporter.release() + + assert _wait_until(lambda: not lane._worker_alive()) + assert exporter.exported_values() == [ + "inflight", + "pre-1", + "pre-2", + "terminal", + ] + assert lane._pending_record_count() == 0 + assert lane._queue_len() == 0 + + def test_retained_size_traverses_filtered_opaque_referents(): class HiddenList(list[Any]): def __init__(self, value: Any) -> None: @@ -866,12 +1045,14 @@ def test_repeated_timeouts_behind_blocked_exporter_stay_bounded(): for i in range(50): scheduler.schedule(ARN_A, _rec(ARN_A, f"a{i + 2}")) assert scheduler.end_invocation(0.02) is False - # One latest detached flush plus at most one coalesced record token. + # At most 16 generation tokens plus one detached eventual flush. assert lane._queued_flush_count() == 1 - assert lane._queue_len() <= 2 + assert lane._queue_len() <= 17 + assert lane._pending_record_count() <= 16 - assert lane._queue_len() <= 2 + assert lane._queue_len() <= 17 assert lane._pending_count() <= 1 + assert lane._pending_record_count() <= 16 assert lane._queued_flush_count() == 1 assert lane._worker is worker assert worker.is_alive() @@ -882,7 +1063,8 @@ def test_repeated_timeouts_behind_blocked_exporter_stay_bounded(): assert _wait_until(lambda: not lane._worker_alive()) exported = exporter.exported_values() assert exported[0] == "a1" - assert len(exported) <= 2 + assert len(exported) <= 17 + assert exported[-1] == "a51" assert exporter.flushed == 1 diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_plugin.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_plugin.py index 3cefcd32..6a6f2b0e 100644 --- a/packages/aws-durable-execution-sdk-python-insight/tests/test_plugin.py +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_plugin.py @@ -365,15 +365,13 @@ def test_cold_resume_reports_prior_terminal_ops_with_fresh_plugin(): assert rec["durationMs"] is not None and rec["durationMs"] >= 0 -# -- on-change schedules RUNNING records, coalescing intermediates (comment 2) -- +# -- on-change preserves ordinary RUNNING-record bursts (comment 2) ----------- -def test_on_change_schedules_running_and_delivers_terminal(): - # Under the async scheduler, rapid cumulative RUNNING snapshots for one - # execution may coalesce (the design explicitly allows a lane to observe only - # a subset of intermediate records). The invariants that always hold: the - # terminal record is delivered last, every earlier record is RUNNING, and the - # terminal record carries the full, de-duplicated operation set. +def test_on_change_schedules_running_on_each_change_and_delivers_terminal(): + # The bounded FIFO preserves this ordinary burst without relying on the + # daemon worker receiving a turn between hooks: invocation start, both + # operation changes, then the terminal snapshot must all be observable. exporter = CaptureExporter() plugin = workflow_insight( WorkflowInsightConfig(exporters=[exporter], emit_mode="on-change") @@ -395,9 +393,7 @@ def test_on_change_schedules_running_and_delivers_terminal(): plugin.on_invocation_end(_end(operations=_ops(op1, op2))) # SUCCEEDED (terminal) statuses = [r["status"] for r in exporter.records] - assert statuses, "at least the terminal record must be delivered" - assert statuses[-1] == "SUCCEEDED" - assert set(statuses[:-1]) <= {"RUNNING"} + assert statuses == ["RUNNING", "RUNNING", "RUNNING", "SUCCEEDED"] final = exporter.records[-1] assert [op["name"] for op in final["operations"]] == ["s1", "s2"] # No duplicate operation entries within a record (no end/change double-count). diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_plugin_async.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_plugin_async.py index e0a9c432..5cc278c2 100644 --- a/packages/aws-durable-execution-sdk-python-insight/tests/test_plugin_async.py +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_plugin_async.py @@ -237,6 +237,38 @@ def test_operation_change_returns_immediately_with_blocked_exporter(): plugin.on_invocation_end(_end(_ops(op))) +def test_tight_on_change_burst_preserves_all_records(): + """Fast consecutive hooks do not depend on daemon-thread scheduling. + + This mirrors the reported parallel/map fan-in shape: invocation start, ten + back-to-back operation changes, then terminal completion. The default FIFO + depth preserves all twelve records even if the worker gets no turn until + after the caller has scheduled the entire burst. + """ + exporter = _BufferedExporter() + plugin = workflow_insight( + WorkflowInsightConfig(exporters=[exporter], emit_mode="on-change") + ) + plugin.on_invocation_start(_start({})) + operations: dict[str, OperationInfo] = {} + for index in range(10): + operation = _step(f"step-{index}", str(index)) + operations[operation.operation_id] = operation + plugin.on_operation_change( + OperationChangeInfo( + execution_arn=ARN, + updated_operations=_ops(operation), + operations=dict(operations), + ) + ) + plugin.on_invocation_end(_end(operations)) + + assert [record["status"] for record in exporter.published] == [ + *("RUNNING" for _ in range(11)), + "SUCCEEDED", + ] + + # -- invocation-end drain is bounded by export_timeout_seconds ----------------