Conversation
|
/ai review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| if not arn or not self._sampled_in(arn): | ||
| return | ||
| state = self._ensure_state(arn) | ||
| state = self._open_state(arn) |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_y7ziufzqv3gutnka44y3y6m7zx
[P1] _open_state() releases _lock before operations are adopted and RUNNING is scheduled. An operation-change callback can pass this check, pause while on_invocation_end() closes and emits the terminal record, then resume and emit a trailing RUNNING record; it can also overwrite the final operations snapshot. Make change admission through scheduling atomic with close, such as by tracking in-flight callbacks under a condition and waiting for them before closing, and add a barrier-based test for this interleaving.
| failed_pending, start_error = self._ensure_worker_locked() | ||
| started = not self._disabled | ||
| start_error = self._ensure_worker_locked() | ||
| worker_running = self._worker is not None |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_zbiazsubqqmypvsd55kwgxpyqi
[P1] After worker startup fails, concurrent drains can all observe _worker is None and independently call _pump; a concurrent schedule can also successfully start a worker before pumping begins. These consumers may export concurrently, and one can flush and signal the shared event while another export is still blocked, causing drain() to return before delivery. Claim exclusive inline-pump ownership while holding _condition, prevent worker startup during that ownership, and make other drains wait for the owner's event. Cover this with concurrent drains and a blocked inline export.
Codex AI reviewTwo concurrency races remain in the new terminal-ordering and worker-failure handling. Reviewed commit |
The export scheduler held a single pending slot for the whole plugin. When two executions were in flight on one plugin instance, execution B's RUNNING snapshot could replace execution A's SUCCEEDED snapshot before the worker claimed it. A's drain() then returned once B's record was flushed, and A's terminal record was never exported. Lambda runs one invocation per environment, but the local test runner and the conformance suite drive independent executions concurrently through one shared plugin, where this lost 30-37% of terminal records in a two-thread probe. Key the pending map by executionArn. A snapshot only replaces its own execution's entry, the worker exports entries in first-arrival order, and a flush request is honored only once every pending record is exported, so drain() means everything scheduled by any execution has been delivered. A RUNNING snapshot never replaces a pending terminal snapshot for the same execution. Close the execution state in on_invocation_end before emitting and draining. An operation-change hook from a checkpoint completing during the drain, or arriving after the state was discarded, is dropped instead of recreating state and emitting a trailing RUNNING record. When the worker thread cannot be started, drain() now exports pending records inline on the calling thread instead of disabling export for the plugin's lifetime and dropping the record.
5c64ef4 to
8a59539
Compare
1 participant
Problem / Motivation
The Workflow Insight export scheduler keeps one pending snapshot for the whole plugin. When two executions are in flight on one plugin instance, execution B's
RUNNINGsnapshot can replace execution A'sSUCCEEDEDsnapshot before the worker claims it. A'sdrain()then returns as soon as B's record is flushed. A's terminal record is never exported.Raised in #719 (comment), issue 1.
Why it matters
In
on-completemode the terminal record is the only record, so this is total data loss for that execution. Lambda runs one invocation per environment, so production is mostly safe. The local test runner (aws-durable-execution-sdk-python-testing) drives independent executions concurrently through one sharedworkflow_insight(...)instance, and the conformance suite runs on top of it. Both are affected.Two-thread probe, two executions through one plugin, 200 trials per mode:
main(8742ad9c)What changed
_export_scheduler.py: the pending map is keyed byexecutionArn. A snapshot only replaces its own execution's entry. The worker exports entries in first-arrival order and honors a flush request only once every pending record is exported, sodrain()means everything scheduled by any execution has been delivered. Memory is bounded by executions in flight, the same bound the plugin's_statealready has. ARUNNINGsnapshot never replaces a pending terminal snapshot for the same execution.plugin.py:on_invocation_endcloses the execution state before it emits and drains. Anon_operation_changehook from a checkpoint completing during the drain, or arriving after the state was discarded, is dropped instead of recreating state and emitting a trailingRUNNINGrecord. This is the guard the Java plugin has (closeAndSchedule) and Python lacked.Worker-start failure no longer disables export for the plugin's lifetime and drops the record.
drain()exports pending records inline on the calling thread and flushes, matching Java. The warning logs once.flowchart LR subgraph Before A1[exec A schedules SUCCEEDED]:::ctx --> S1[one pending slot]:::removed B1[exec B schedules RUNNING]:::ctx --> S1 S1 --> W1[worker exports B only]:::removed end subgraph After A2[exec A schedules SUCCEEDED]:::ctx --> SA[pending A]:::added B2[exec B schedules RUNNING]:::ctx --> SB[pending B]:::added SA --> W2[worker exports A, then B, then flush]:::changed SB --> W2 end classDef added fill:#DCFCE7,stroke:#16A34A,color:#14532D,stroke-width:2px classDef changed fill:#FEF3C7,stroke:#D97706,color:#78350F,stroke-width:2px classDef removed fill:#FEE2E2,stroke:#DC2626,color:#7F1D1D,stroke-dasharray:4 3 classDef ctx fill:#E0F2FE,stroke:#0284C7,color:#0C4A6E🟩 added · 🟨 changed · 🟥 removed · 🟦 unchanged
Each execution now owns its pending slot, so one execution's snapshot never displaces another's.
Not in this PR: the coalescing contract for
on-changebursts (issue 2 in the same comment). That is a separate follow-up on top of this change.Tests
test_export_scheduler.py: per-execution keying with a blocked exporter;RUNNINGcannot supersede a pending terminal; two threads x 50 rounds all deliver their terminal record; worker-start failure delivers inline ondrain()(replaces the test that asserted the record was dropped).test_plugin.py: two executions driven concurrently through one plugin for 50 rounds in bothon-completeandon-change, exactly 50SUCCEEDEDper ARN and empty_stateafter; a lateon_operation_changeduring a blocked drain is dropped;on_operation_changeafter invocation end is ignored.Manual verification
Two-thread probe above (
probe_concurrent_arns.py, driveson_invocation_start/on_operation_change/on_invocation_endfor two ARNs through one plugin). Not committed; the equivalent istest_concurrent_executions_on_one_plugin_each_deliver_terminal_record.Screenshots / video
N/A — no user interface changes.
no linked issue: raised as a review comment on #719; #687 tracks broader Workflow Insight follow-up work