feat(insight): add async export scheduling - #702
Conversation
0bbf504 to
6b73e82
Compare
|
/ai review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| if not barrier.wait(remaining): | ||
| barrier.canceled = True |
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
|
/ai review |
Codex AI reviewNo actionable findings. Residual risk remains around real Lambda freeze/thaw behavior with exporters that stay blocked beyond the configured timeout. Reviewed commit |
This comment has been minimized.
This comment has been minimized.
GitHub Actions
Inactive
GitHub Actions
Failure
| # must omit endTime/durationMs. Passing end_time=None makes _emit | ||
| # drop both fields. Output and error likewise belong only to a | ||
| # terminal record. | ||
| self._emit( | ||
| self._schedule_record( |
There was a problem hiding this comment.
Claude AI review · Finding arf_v1_4qdr5mczjf6yvbzm4jlpoh2z7y
The comment above this call still says "Passing end_time=None makes _emit drop both fields," but _emit was split/renamed into _build_record/_schedule_record earlier in this same PR. A future maintainer grepping for _emit (or trusting this comment while modifying the schedule path) will be misled since that method no longer exists. Update the comment to reference the current method name.
| # must omit endTime/durationMs. Passing end_time=None makes _emit | |
| # drop both fields. Output and error likewise belong only to a | |
| # terminal record. | |
| self._emit( | |
| self._schedule_record( | |
| # must omit endTime/durationMs. Passing end_time=None makes | |
| # _schedule_record drop both fields. Output and error likewise | |
| # belong only to a terminal record. | |
| self._schedule_record( |
Claude AI reviewThis PR adds an async, coalescing per-exporter export scheduler ( I traced the full lane/queue/worker state machine (coalescing via Only a minor documentation nit remains (see inline comment). Residual, already-documented risk: Reviewed commit |
Move all exporter work off the SDK checkpoint thread. A new private _ExportScheduler owns one lazy daemon worker per exporter lane; per-exporter copy, render, truncation, export() and flush() now run there, so a slow exporter never blocks workflow progress. Per lane: at most one in-flight record and one latest pending record per execution ARN. Cumulative snapshots for the same ARN coalesce (the in-flight record is never cancelled); updating a pending ARN moves it to the back for FIFO fairness across ARNs; pending ARNs are capped with oldest-eviction. A blocked worker is retained and never replaced, and idle workers exit after the drain, so threads cannot grow unbounded. on_operation_change returns immediately unless emit mode is on-change. Invocation end schedules the final record, then drains and flushes the touched lanes under one shared deadline; on timeout the workflow response is returned and delivery degrades to best-effort. Exceptions in render/export/flush are isolated and logged. Add WorkflowInsightConfig.export_timeout_seconds (default 5.0), validated as a finite number greater than zero (rejects bool, NaN, infinity, and non-positive values). Add scheduler, plugin-async, and config unit tests plus updated on-change coalescing coverage; refresh the README note. No core SDK changes.
Skip a lane record when copy.deepcopy fails instead of aliasing the shared canonical record. The alias let this lane's truncation mutate the object other lanes still read, breaking workflow isolation. A copy failure is now logged through the module logger and the lane keeps draining, matching render/truncation failure handling. Also remove the dead _inflight_arn lane field (written, never read). Tests: deepcopy-failure skips the record, does not call the exporter, logs the failure, and the lane continues to export a later valid record; a non-aliasing regression guards in-place mutation; a warm-container cross-invocation test proves bounded invocation-end waits, no A/B merge, and FIFO drain + flush after unblock.
Make the two shared-timeout tests wait deterministically for their released lane workers to stop before returning, so their daemon workers cannot exit between a later test's baseline capture and its assertion. Replace the fragile process-global thread-count delta in test_blocked_worker_is_not_replaced with lane-local worker identity, aliveness, and a lane-scoped worker count. This proves the blocked lane never spawns a replacement without depending on global thread state. Product code is unchanged.
Cancelled flush barriers no longer pile up behind a blocked exporter. end_invocation now pairs each barrier with its lane and, on timeout, calls _ExporterLane.cancel_flush(barrier): under the lane lock it marks the barrier cancelled and pulls its still-queued _FLUSH marker out, completing it there. If the worker already popped the marker the flush is left to the worker; an in-flight synchronous flush is not killed. This keeps queue and barrier state bounded across many warm invocations while preserving record ordering, normal flush, the shared deadline, blocked-worker retention, and bounded pending state. Also switch the lane Condition from the default RLock to an explicit non-reentrant Lock; the lane never re-acquires _cond while holding it. Tests: deterministic repeated-timeout test (blocked exporter across many warm invocations) plus queued-vs-already-popped cancellation race tests.
Reject the same exporter instance appearing more than once in WorkflowInsightConfig.exporters with a clear ValueError, compared by object identity (not equality/hash) during config normalization. Two distinct instances of the same class stay valid and each keeps its own lane; the default exporter is unaffected. Preserves the one-thread-per-distinct-instance safety and avoids duplicate, timing-dependent scheduling. Route _ExecutionState.scheduled mutation and read through the plugin _lock via _mark_scheduled/_was_scheduled, consistent with the other state fields. The lock is released before any scheduler/end_invocation or exporter work, so no new lock ordering or deadlock is introduced. Add tests: same instance twice raises; two distinct same-class instances each get a lane; default exporter unaffected; scheduled flag tracks scheduling.
9e0e717 to
2e7049c
Compare
GitHub Actions
In progress
GitHub Actions
In progress
1 participant
Summary
Tracks #687.
Design
export_timeout_secondsdefaults to 5 seconds and bounds drain plus flushValidation
Review decisions
_inflight_arnwas dead stateRLockLock; the lane never re-acquires_condwhile holding it, so recursion support is unneeded and misuse now fails loudlyWorkflowInsightConfignow rejects a duplicate exporter instance (by object identity, not equality/hash) with a clearValueError; preserves one-thread-per-distinct-instance safety and avoids duplicate, timing-dependent scheduling. Distinct same-class instances and the default exporter are unaffected_ExecutionState.scheduledwas read/written without the plugin lock_lockvia_mark_scheduled/_was_scheduled, consistent with the other state fields; the lock is released before any scheduler/end_invocationor exporter work, so no new lock ordering or deadlock. SDK serializes hooks, so scope is minimalReviewed three times with
commit-code-reviewer, plus two post-public-review passes addressing the rows above. No actionable findings remain.