fix(insight): schedule on-change exports #699
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,146 @@ | ||
| // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
| package software.amazon.lambda.durable.insight; | ||
|
|
||
| import java.util.ArrayDeque; | ||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.concurrent.Executor; | ||
| import java.util.concurrent.ExecutorService; | ||
| import java.util.concurrent.Executors; | ||
| import java.util.concurrent.atomic.AtomicInteger; | ||
| import java.util.function.Consumer; | ||
|
|
||
| /** Serializes ON_CHANGE exports through a bounded, invocation-scoped FIFO. */ | ||
| final class ExportScheduler { | ||
| static final int DEFAULT_CAPACITY = 16; | ||
|
|
||
| private static final AtomicInteger THREAD_NUMBER = new AtomicInteger(); | ||
| // Shared for the Lambda process lifetime. Cached daemon workers are reclaimed after idle periods, while each | ||
| // invocation retains its own serial scheduler, bounded queue, and flush barrier. | ||
| private static final ExecutorService WORKERS = Executors.newCachedThreadPool(runnable -> { | ||
| var thread = new Thread(runnable, "workflow-insight-export-" + THREAD_NUMBER.incrementAndGet()); | ||
| thread.setDaemon(true); | ||
| return thread; | ||
| }); | ||
|
|
||
| private final int capacity; | ||
| private final Executor executor; | ||
| private final Consumer<WorkflowInsightRecord> export; | ||
| private final Runnable flush; | ||
| private final Consumer<Throwable> failureHandler; | ||
| private final ArrayDeque<WorkflowInsightRecord> queue = new ArrayDeque<>(); | ||
| private final CompletableFuture<Void> drained = new CompletableFuture<>(); | ||
|
|
||
| private boolean running; | ||
| private boolean sealed; | ||
|
|
||
| ExportScheduler(Consumer<WorkflowInsightRecord> export, Runnable flush, Consumer<Throwable> failureHandler) { | ||
| this(DEFAULT_CAPACITY, WORKERS, export, flush, failureHandler); | ||
| } | ||
|
|
||
| ExportScheduler( | ||
| int capacity, | ||
| Executor executor, | ||
| Consumer<WorkflowInsightRecord> export, | ||
| Runnable flush, | ||
| Consumer<Throwable> failureHandler) { | ||
| if (capacity <= 0) { | ||
| throw new IllegalArgumentException("capacity must be positive"); | ||
| } | ||
| this.capacity = capacity; | ||
| this.executor = executor; | ||
| this.export = export; | ||
| this.flush = flush; | ||
| this.failureHandler = failureHandler; | ||
| } | ||
|
|
||
| /** Queues a complete RUNNING snapshot, dropping the oldest waiting snapshot only under backpressure. */ | ||
| synchronized boolean schedule(WorkflowInsightRecord record) { | ||
| if (sealed) { | ||
| return false; | ||
| } | ||
| enqueue(record); | ||
| startWorker(); | ||
| return true; | ||
| } | ||
|
|
||
| /** Seals this invocation, queues its final snapshot, and returns a future for the ordered flush barrier. */ | ||
| synchronized CompletableFuture<Void> sealAndDrain(WorkflowInsightRecord finalRecord) { | ||
| if (!sealed) { | ||
| sealed = true; | ||
| if (finalRecord != null) { | ||
| enqueue(finalRecord); | ||
| } | ||
| startWorker(); | ||
| } | ||
| return drained; | ||
| } | ||
|
|
||
| private void enqueue(WorkflowInsightRecord record) { | ||
| if (queue.size() == capacity) { | ||
| queue.removeFirst(); | ||
| } | ||
| queue.addLast(record); | ||
| } | ||
|
|
||
| private void startWorker() { | ||
| if (running) { | ||
| return; | ||
| } | ||
| running = true; | ||
| try { | ||
| executor.execute(this::pump); | ||
| } catch (Throwable t) { | ||
| running = false; | ||
| reportFailure(t); | ||
| if (sealed) { | ||
| queue.clear(); | ||
| drained.complete(null); | ||
|
Comment on lines
+97
to
+98
Contributor
There was a problem hiding this comment. Codex AI review · Finding [P2] Do not report a sealed queue as drained when worker submission fails. If the worker went idle and |
||
| } | ||
| } | ||
| } | ||
|
|
||
| private void pump() { | ||
| while (true) { | ||
| WorkflowInsightRecord record; | ||
| synchronized (this) { | ||
| if (!queue.isEmpty()) { | ||
| record = queue.removeFirst(); | ||
| } else if (sealed) { | ||
| record = null; | ||
| } else { | ||
| running = false; | ||
| return; | ||
| } | ||
| } | ||
|
|
||
| if (record != null) { | ||
| runSafely(() -> export.accept(record)); | ||
| continue; | ||
| } | ||
|
|
||
| runSafely(flush); | ||
| synchronized (this) { | ||
| running = false; | ||
| drained.complete(null); | ||
| } | ||
| return; | ||
| } | ||
| } | ||
|
|
||
| private void runSafely(Runnable action) { | ||
| try { | ||
| action.run(); | ||
| } catch (Throwable t) { | ||
| reportFailure(t); | ||
| } | ||
| } | ||
|
|
||
| private void reportFailure(Throwable t) { | ||
| try { | ||
| failureHandler.accept(t); | ||
| } catch (Throwable ignored) { | ||
| // A scheduler diagnostic must never disrupt durable execution. | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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
executionResultor objects returned by content/result transforms, whiledeepCopy()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.