fix(insight): schedule on-change exports - #699
Conversation
Move exporter I/O off checkpoint callbacks through a bounded per-invocation FIFO. Preserve normal bursts, drop oldest waiting snapshots only under backpressure, and drain the final record plus one flush barrier at invocation end.
|
/ai review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| if (queue.size() == capacity) { | ||
| queue.removeFirst(); | ||
| } | ||
| queue.addLast(record); |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_zd7gncmrres5u3dmq4nazbmvno
[P2] Freeze each record before placing it on the asynchronous queue. Records can contain live mutable values, notably executionResult or objects returned by content/result transforms, while deepCopy() currently occurs only when the worker eventually exports them. Under backpressure, later mutation can therefore change an earlier snapshot and make Workflow Insight disagree with the already-serialized durable result. Deep-copy or otherwise detach mutable record content during enqueue, and add a test that mutates content before a delayed worker runs.
| queue.clear(); | ||
| drained.complete(null); |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_yk6awyvhew5xhws3cqn4dgeagn
[P2] Do not report a sealed queue as drained when worker submission fails. If the worker went idle and sealAndDrain() encounters a rejection or thread-creation failure, this branch discards the reserved final record and every queued update, skips exporter flushing, and lets onInvocationEnd() return successfully. Preserve the queue and synchronously drain it at the invocation boundary, or retry submission while leaving drained incomplete; cover rejection specifically on the final submission.
Codex AI reviewFound two correctness issues in the new Reviewed commit |
1 participant
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
Issue Link, if available
Related to #679.
Addresses #661 (comment).
Description
ON_CHANGEpreviously performed exporter I/O andflush()synchronously from checkpoint callbacks. Slow S3 or CloudWatch calls could delay checkpoint coordination, and a delayedRUNNINGwrite could finish after the terminal S3 write.This PR moves
ON_CHANGEdelivery to an invocation-scoped bounded FIFO. The queue preserves normal bursts, drops the oldest waitingRUNNINGsnapshot only after reaching its depth of 16, reserves the invocation-final record, and rejects late updates after sealing.onInvocationEnd()drains the ordered queue and flushes each exporter once before Lambda can freeze.The implementation deliberately avoids the unconditional latest-slot coalescing pattern identified in aws/aws-durable-execution-sdk-python#702 (comment). A deterministic no-delay test verifies that eleven
RUNNINGsnapshots plus one final snapshot all arrive.Demo/Screenshots
Not applicable. This is backend scheduling and lifecycle behavior with no UI changes.
Checklist
Testing
git diff --checkmvn -B -q spotless:check --file pom.xmlmvn -q -pl insight-plugin -am clean verifymvn -q clean verifyExportSchedulerTestandWorkflowInsightHookTestruns to check deterministic orderingUnit Tests
Yes.
ExportSchedulerTestcovers no-delay burst preservation, bounded oldest-drop behavior, final-record retention, late-update rejection, idle-worker restart, exporter failure isolation, flush ordering, and transient executor rejection recovery.Integration Tests
Yes.
WorkflowInsightHookTestdrives the real plugin hooks and verifies that blocked exporter I/O does not blockonOperationChange(), invocation end waits for ordered draining, normal bursts are preserved, the terminal record is last, and exporters flush once.The complete
insight-pluginreactor and full Maven reactor pass.Examples
No example change is needed. The public Workflow Insight configuration API is unchanged.