Each {@link WorkflowInsightRecord} is a complete snapshot of the execution, so a newer record fully supersedes any - * record still waiting to be exported. While an export is in flight, additional updates are coalesced into a single - * "pending" slot — intermediate records are dropped because the latest one already contains all of their information. - * This prevents overlapping {@code export()} calls when updates arrive faster than the exporters can keep up, and it - * keeps exporter I/O off the SDK threads that deliver plugin hooks. + *
One scheduler serves the whole execution environment — the exporters it fans out to belong to the environment, and + * an environment can host several durable executions at the same time (Lambda Managed Instances makes that routine). + * The work, by contrast, is held on the invocation it belongs to: each {@link InsightPlugin} instance is one + * invocation's plugin and owns that invocation's latest-record slot, drain signal, mid-export marker and drain-waiter + * count. Coalescing happens only within one invocation, because the slot belongs to it. * - *
Exports are otherwise fire-and-forget; {@link #drain()} is called before the invocation returns to guarantee the - * final record is delivered. + *
That shape is not a tidying-up. The same facts once lived in five structures keyed by execution ARN — the plugin's + * state map plus this class's {@code pending}, {@code settled}, {@code exporting} and {@code drainWaiters} — and two of + * them could disagree about one execution. They did: "nothing queued for this ARN" was read as "nothing outstanding", + * which is equally true of a record already taken and being exported, so an exiting pump completed another execution's + * drain signal mid-export and that invocation returned before its record was delivered. Every fact now exists exactly + * once, as a field of the one object the SDK gave that invocation, so the disagreement has no place to happen — and + * since the SDK creates that object and drops it, the scheduler has no execution registry to keep in step with + * anything. + * + *
Each {@link WorkflowInsightRecord} is a complete snapshot of its execution, so a newer record from the same + * invocation fully supersedes one still waiting to be exported. While an export is in flight, additional updates from + * that invocation are coalesced into its slot — intermediate records are dropped because the latest one already + * contains all of their information. A record from a different invocation never displaces another's record. + * + *
The slot takes whichever record is handed to it last and compares nothing, so "newer" has to be established before + * the hand-off. Customer code runs while a record is being built and can re-enter a hook of the same invocation, which + * builds and hands over a newer record first; the build it re-entered from then hands over an older snapshot last. + * {@code InsightPlugin}'s build revision identifies each build and {@link #scheduleIfNotSuperseded} drops a record + * whose build has been overtaken, so the slot only ever advances. + * + *
A single pump exports the queued records one at a time, in the order the invocations first queued work + * ({@link #queue}, which is ordering only — membership in it is the same fact as "this invocation has a record", + * written in one place), so exporters still never see two exports at once and each record keeps its per-exporter + * fan-out. {@link #flush()} requests are served by that same pump, between records, so an exporter never sees a + * {@code flush()} overlap an {@code export()} either. Requests are served as a batch — the cadence is at most one flush + * per requesting invocation end, not exactly one — and a flush is preceded by the queued records a drain is waiting + * for, so a burst of invocation ends is covered by one flush rather than one each. A request made while a flush runs + * waits for the next turn. Exports are otherwise fire-and-forget; {@link #drain(InsightPlugin)} is called before an + * invocation returns and waits for that invocation's own latest record to reach every exporter. + * + *
Because there is one pump, that wait can also cover records other invocations had already queued ahead of this
+ * one: a drain is not isolated from the queue's head-of-line cost. What per-invocation ownership guarantees is that
+ * another invocation's record can never displace this one — records coalesce only within their own invocation,
+ * and a drain cannot return until its own latest record has reached every exporter.
*/
final class ExportScheduler {
private static final AtomicInteger THREAD_NUMBER = new AtomicInteger();
+ /**
+ * Upper bound on the passes {@link #drainAll()} makes over the outstanding invocations. Only reached if new work
+ * keeps arriving for as long as the drain runs; a normal drain settles in two passes.
+ */
+ private static final int MAX_DRAIN_ALL_PASSES = 1_000;
+
+ /** How long one {@link #drainAll()} pass waits for a running pump before taking another pass. */
+ private static final long PUMP_WAIT_MILLIS = 50;
+
/** Shared for the process lifetime; idle daemon workers are reclaimed, so nothing keeps the runtime alive. */
private static final ExecutorService WORKERS = Executors.newCachedThreadPool(runnable -> {
var thread = new Thread(runnable, "workflow-insight-export-" + THREAD_NUMBER.incrementAndGet());
@@ -43,8 +92,62 @@ final class ExportScheduler {
/** Completes when the current pump finishes; {@code null} while idle. Guarded by {@code this}. */
private CompletableFuture Written only by the thread that enters {@link #pump} — a worker, or the caller's own thread on the
+ * rejected-worker fallback, which is a case where the calling thread genuinely is the pump — and cleared
+ * by that same thread on the way out, only if it is still the recorded one. A compare-and-set on the way out rather
+ * than a blind clear: if some anomaly ever did leave two pumps running, the one that finishes first must not clear
+ * the other, and must not leave a stale thread behind that a later, legitimate {@code flush()} from that same
+ * thread would be mistaken for.
+ */
+ private final AtomicReference With a single exporter the fan-out runs on the pump thread and {@link #pumpThread} already recognizes it. With
+ * two or more, {@link #forEachExporterSettled} submits one task per exporter and the pump then joins them all, so
+ * the callback runs on a thread that is not the pump but that the pump cannot outlive: a wait for the pump issued
+ * from there is the same wait-for cycle, two threads wide instead of one. The pump parks in the join, so it never
+ * reaches the point in its loop that would complete the future the worker is parked on.
+ *
+ * An instance field rather than a static: a fan-out worker of one scheduler is not pump-dependent on any other
+ * scheduler, and refusing its waits there would be a false positive. Set and cleared around each callback by the
+ * thread that runs it, restoring whatever was there before rather than blindly removing, so a callback that the
+ * pump ran inline (the rejected-worker fallback, where the fan-out thread is the pump) cannot clear a mark
+ * an enclosing frame still needs.
+ */
+ private final ThreadLocal This is the only collection of per-invocation objects the scheduler has, and it holds an instance for exactly
+ * as long as that instance has a record waiting: nothing here has to be cleaned up at an invocation boundary, and
+ * an instance the SDK has dropped is unreachable from the scheduler the moment its last record is taken.
+ *
+ * Invariant, and the only thing that could still be said twice: an invocation is in here exactly while its
+ * {@link InsightPlugin#record} is non-null. Every record moves through {@link #queueRecord}, {@link #takeRecord} or
+ * {@link #dropRecord}, which write both halves together, and a set makes a double entry impossible by construction.
+ */
+ private final Set A queue of requests rather than a single flag: the pump takes the requests that are queued when its turn
+ * begins and satisfies all of them with one flush, so concurrent invocation ends share a flush; a request enqueued
+ * while that flush runs stays in the queue for the next turn, because a flush already in progress cannot be shown
+ * to have seen the new requester's records.
+ */
+ private final Deque The slot takes whichever record is handed over last and does not compare record ages, so this is the right
+ * entry point only for a record that cannot be superseded. The plugin's RUNNING records go through
+ * {@link #scheduleIfNotSuperseded} and its final record through {@link #closeAndSchedule}; both add the ordering
+ * checks this one omits.
*/
- void schedule(WorkflowInsightRecord record) {
+ void schedule(InsightPlugin execution, WorkflowInsightRecord record) {
CompletableFuture The invocation may already have ended. No RUNNING snapshot may follow the final record, so
+ * {@link InsightPlugin#closed} rejects it.
+ *
+ * A newer build of this same invocation may already have started. Customer code runs inside a build — the
+ * content transforms, an operation result transform, a serializer for a customer type — and can re-enter a hook, so
+ * the build that hands its record over last is not necessarily the build that started last. Without the revision
+ * check the slot would take that older snapshot and the newer one would be lost, or, if a pump had already taken
+ * the newer one, an exporter would see the older snapshot after the newer one.
+ *
+ * The superseded record is dropped rather than queued. Nothing is lost: a record is a complete snapshot of one
+ * execution, so the record that superseded it carries everything it carries. That is the same property that makes
+ * the slot's coalescing sound.
+ *
+ * Both checks and the hand-off are one critical section, on the monitor that owns both fields, so a record
+ * cannot pass the checks and then be queued after the record that supersedes it.
+ *
+ * @param buildRevision the revision the caller took before it started building this record
+ * @return whether the record was queued
+ */
+ boolean scheduleIfNotSuperseded(InsightPlugin execution, WorkflowInsightRecord record, long buildRevision) {
+ CompletableFuture The final record is queued without the build-revision check {@link #scheduleIfNotSuperseded} makes. Customer
+ * code running inside the final record's build can start a newer RUNNING build, which would leave the final
+ * record's revision stale, and a checked hand-off would then drop it and leave a RUNNING snapshot as the
+ * execution's last exported state. Exempting it cannot let a stale record win, because {@code closed} is set in
+ * this same critical section and every RUNNING record handed over afterwards is rejected.
+ */
+ void closeAndSchedule(InsightPlugin execution, WorkflowInsightRecord finalRecord) {
+ if (finalRecord == null && execution.closed) {
+ // The idempotent second call from the hook's `finally`. A volatile read, so the common case of an
+ // invocation end that already scheduled its final record does not take the lock again.
+ return;
+ }
+ CompletableFuture The marking is not a separate step in a separate structure: leaving the slot and becoming "inside the
+ * exporters" are one write of one object, so no reader can see the invocation between the two and conclude it has
+ * nothing outstanding.
*/
- void drain() {
+ private WorkflowInsightRecord takeRecord(InsightPlugin execution) {
+ WorkflowInsightRecord record = execution.record;
+ execution.record = null;
+ queue.remove(execution);
+ execution.exporting = true;
+ return record;
+ }
+
+ /** Drops this invocation's queued record without exporting it. Caller holds the lock. */
+ private void dropRecord(InsightPlugin execution) {
+ execution.record = null;
+ queue.remove(execution);
+ }
+
+ // --- Draining. ---
+
+ /**
+ * Waits until the latest record of one invocation has been handed to every exporter. Safe to call when that
+ * invocation has nothing outstanding. Used before the invocation returns to guarantee the final record is
+ * delivered.
+ *
+ * The wait is for that invocation's own latest record. Another invocation's record can never displace it, so
+ * this always returns having delivered this invocation's latest snapshot; but since one pump exports serially, the
+ * wait can also cover records other invocations had already queued ahead of it.
+ *
+ * While this waits, the invocation counts a drain waiter — on the instance itself, so the count cannot come to
+ * describe a different one. It tells the pump that this record gates an invocation return, so the pump exports it
+ * before spending a flush fan-out. See {@link #exportRecordsADrainIsWaitingFor}.
+ *
+ * Called from the pump thread itself, or from an exporter fan-out worker that pump is waiting for, the wait is
+ * refused and reported instead of made: see {@link #refuseWaitThatWouldBlockThePump}.
+ */
+ void drain(InsightPlugin execution) {
+ // Re-entered from a thread the pump's progress depends on: waiting here would park on a signal only that pump
+ // can settle. Refuse and return; the record stays queued and that same pump exports it once it resumes its
+ // loop.
+ if (refuseWaitThatWouldBlockThePump("drain(execution)")) {
+ return;
+ }
+ synchronized (this) {
+ if (execution.settled == null) {
+ return;
+ }
+ execution.drainWaiters++;
+ }
+ try {
+ drainUntilSettled(execution);
+ } finally {
+ synchronized (this) {
+ execution.drainWaiters--;
+ }
+ }
+ }
+
+ private void drainUntilSettled(InsightPlugin execution) {
while (true) {
+ CompletableFuture Returns once nothing is queued and no pump owns the scheduler, which is exactly "every record scheduled so far
+ * has reached the exporters": a pump only returns to idle with its queue empty and the record it took settled.
+ *
+ * Bounded by the number of passes, not by the set of invocations seen: an invocation that queues new work after
+ * it was already drained must still be waited for (dropping it would silently weaken every assertion made after
+ * this returns), while a producer that never stops cannot keep this spinning forever.
+ */
+ void drainAll() {
+ // Every pass below is a drain, and each one would be refused; without this the loop spends all of its passes
+ // reporting the same refusal.
+ if (refuseWaitThatWouldBlockThePump("drainAll()")) {
+ return;
+ }
+ for (int pass = 0; pass < MAX_DRAIN_ALL_PASSES; pass++) {
+ List {@link #queue} is the only collection of per-invocation objects the scheduler has, so this is the whole of the
+ * per-invocation state the environment retains. Zero means the environment — which outlives every invocation —
+ * holds nothing belonging to any invocation it has served.
+ */
+ synchronized int retainedInvocationCount() {
+ return queue.size();
+ }
+
+ /** Test seam: whether the scheduler still holds a reference to one particular invocation. */
+ synchronized boolean retains(InsightPlugin execution) {
+ return queue.contains(execution);
+ }
+
+ /** Gives up one invocation's outstanding work: drops its queued record and releases every drain waiting on it. */
+ private void abandon(InsightPlugin execution) {
+ CompletableFuture Those records are the last records of invocations that cannot return until they are exported, and their ends
+ * cannot ask for their flush until then. Exporting them first is therefore what lets one flush serve a whole burst
+ * of invocation ends: without it the pump interleaves one record and one flush fan-out, and each end pays for a
+ * flush of its own even though every request is coalesced.
+ *
+ * Bounded by the snapshot taken under the lock, so a producer that keeps scheduling for an invocation someone is
+ * draining cannot hold a flush back indefinitely — and records nobody waits for are not exported here at all, so a
+ * stream of {@code ON_CHANGE} snapshots still cannot starve a flush: it waits at most one ordinary fan-out plus
+ * this pass over the invocations whose return is already blocked on their own record.
+ */
+ private void exportRecordsADrainIsWaitingFor() {
+ List Requests are coalesced: the pump takes every request queued at the start of its turn, exports any queued
+ * record a drain is still waiting for, re-takes the requests those ends make as they are released, and satisfies
+ * them all with one flush. Invocation ends that overlap therefore share a flush instead of paying for one fan-out
+ * each. That is sound because a caller drains its own record before asking, so a flush that starts after
+ * the request was enqueued has that record in the exporter's buffer. A request enqueued while a flush is already
+ * running is never satisfied by it — it waits for the next turn.
+ *
+ * A queue that never runs dry cannot starve a request either: the pump alternates one record and one batch of
+ * requests, so a flush waits at most one export fan-out.
+ *
+ * Called from the pump thread itself — or from an exporter fan-out worker that pump is waiting for, which is
+ * what a callback re-entering the scheduler does when two or more exporters are configured — the request is refused
+ * and reported instead of made: see {@link #refuseWaitThatWouldBlockThePump}.
+ */
+ void flush() {
+ // Re-entered from a thread the pump's progress depends on: the pump is the only thread that could serve the
+ // request, and it cannot while this caller has not returned. Refuse rather than enqueue a request nobody
+ // serves.
+ if (refuseWaitThatWouldBlockThePump("flush()")) {
+ return;
+ }
+ CompletableFuture Private and called only from the pump: routing every flush through the pump is what keeps a {@code flush()}
+ * from overlapping an {@code export()}, so this must not be reachable from outside. The per-exporter fan-out below
+ * is parallelism within one flush, not concurrency with an export.
*/
- void flushAll() {
+ private void flushEveryExporter() {
forEachExporterSettled(InsightExporter::flush);
}
+ // --- Exporting. ---
+
/**
* Exports one record to every exporter, each on its own worker, and waits for all of them to settle. One failing or
* slow exporter never blocks or fails the others, and an export error never propagates into the execution.
@@ -168,24 +811,47 @@ private void exportToAll(WorkflowInsightRecord record) {
/** Runs the action for every exporter concurrently and returns once all have settled, reporting each failure. */
private void forEachExporterSettled(Consumer Invariant: the thread that waits for the pump is never a thread the pump waits for. {@link #flush()} waits for
+ * a request only a pump can complete, and a drain waits for a signal only a pump can complete or for the running
+ * pump's own handle. All three are satisfied by the pump between records.
+ *
+ * Two threads qualify. The pump thread itself: a wait issued from there is a wait-for cycle one thread wide —
+ * the pump parks on the future it would itself have completed, so it never reaches the point in its loop that
+ * completes it, and no other thread may take over because {@code inFlight} is this pump's. And an exporter fan-out
+ * worker: with two or more exporters the pump submits one task per exporter and joins them all, so a wait issued
+ * from a callback running on one of those workers is the same cycle two threads wide — the worker parks on a future
+ * only the pump can complete, and the pump is parked in the join waiting for that worker. Neither is a monitor
+ * deadlock, so the JVM's deadlock detection cannot see either one, and the invocation simply never returns.
+ *
+ * Reachable through anything a fan-out calls synchronously: with a single exporter the fan-out runs on the pump
+ * thread, so a customer exporter's {@code export()} or {@code flush()} that asks the scheduler for a flush, or a
+ * non-conforming {@code exportOne}, is enough; with several it runs on a worker instead, and the same call is
+ * refused for the same reason. A conforming production {@code exportOne} does not re-enter the scheduler, so this
+ * is hardening.
+ *
+ * So the call fails fast instead: the plugin's failure handler is told — it logs — and the caller returns as it
+ * would from any other flush or drain, with nothing propagating into the execution. The queued work itself is not
+ * dropped by refusing a drain: the record stays in the invocation's slot, and the pump that is waiting for this
+ * caller exports it as soon as this caller returns and the fan-out it belongs to settles. Callers that are neither
+ * — every SDK hook thread — never enter this branch and behave exactly as before; the check is a volatile read plus
+ * a thread-local read, so no lock is added to that path.
+ */
+ private boolean refuseWaitThatWouldBlockThePump(String call) {
+ if (pumpThread.get() == Thread.currentThread()) {
+ reportFailure(new IllegalStateException(call
+ + " was called from the export pump thread, the only thread able to serve it; the call was refused"
+ + " rather than deadlocking the invocation"));
+ return true;
+ }
+ if (Boolean.TRUE.equals(exporterFanOutThread.get())) {
+ reportFailure(new IllegalStateException(call
+ + " was called from an exporter fan-out worker the export pump is waiting for, so the pump cannot"
+ + " serve it; the call was refused rather than deadlocking the invocation"));
+ return true;
+ }
+ return false;
+ }
}
diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/InsightExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/InsightExporter.java
index 69ff39739..26c2d2b7a 100644
--- a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/InsightExporter.java
+++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/InsightExporter.java
@@ -11,7 +11,34 @@ public interface InsightExporter {
/** Emits one record to the destination. */
void export(WorkflowInsightRecord record);
- /** Flushes any buffered records; no-op by default. */
+ /**
+ * Flushes any records this exporter has buffered. The default is a no-op; override it only if
+ * {@link #export(WorkflowInsightRecord)} buffers rather than emitting immediately.
+ *
+ * Called at most once per sampled-in invocation end, after that invocation's own record — if it emitted one —
+ * has been handed to every exporter. An end that emits no record still flushes (a non-terminal suspend under
+ * {@code ON_COMPLETE}, a success under {@code ON_FAILURE}), so records buffered by that execution's earlier
+ * emissions are never left behind. Invocation ends that overlap may share a single flush: one flush is enough for
+ * all of them, because it starts only after each of their records has been handed to every exporter. An execution
+ * that is sampled out neither exports nor flushes.
+ *
+ * Never called concurrently with {@link #export(WorkflowInsightRecord)} by the plugins one
+ * {@link WorkflowInsight#workflowInsight} factory creates. That factory owns the scheduler serializing them, so the
+ * guarantee is per factory rather than per environment: an exporter instance handed to two factories is served by
+ * two schedulers, which can call its {@code export} and {@code flush} at the same time. Build the factory once per
+ * handler — which is what a {@code DurableConfig} created once per handler does — and give each factory its own
+ * exporter instances if a single exporter cannot tolerate concurrent calls.
+ *
+ * May cover records belonging to other executions running in the same environment, so it is not a per-execution
+ * barrier.
+ *
+ * Must return promptly. No invocation whose end is waiting on this flush can return until it returns, and since
+ * overlapping ends may share one flush, a slow flush is billed to every one of those invocations — not only to the
+ * one that asked for it.
+ *
+ * Failures are isolated: a {@link Throwable} thrown here is reported through the plugin's failure handler, never
+ * retried, never propagated into the execution, and never prevents another exporter from flushing.
+ */
default void flush() {}
/**
diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/InsightPlugin.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/InsightPlugin.java
new file mode 100644
index 000000000..3223522f8
--- /dev/null
+++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/InsightPlugin.java
@@ -0,0 +1,443 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.insight;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicLong;
+import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
+import software.amazon.lambda.durable.plugin.InvocationEndInfo;
+import software.amazon.lambda.durable.plugin.InvocationInfo;
+import software.amazon.lambda.durable.plugin.OperationChangeInfo;
+import software.amazon.lambda.durable.plugin.OperationChangeItemInfo;
+
+/**
+ * The Workflow Insight plugin instance for one Lambda invocation: both the state that invocation's records are built
+ * from and the slot the {@link ExportScheduler} exports them through.
+ *
+ * The SDK creates one of these per invocation, from the {@link InvocationInfo} it is about to hand the first hook,
+ * and drops it when the invocation returns. So everything about an execution is a plain field here — the parsed ARN,
+ * the stable start time, the one-time sampling decision, the detached input snapshot, the latest queued record, the
+ * drain signal, the mid-export marker and the drain-waiter count. There is nothing to key by execution ARN and nothing
+ * to register or release: an instance is the registration, and its lifetime is the invocation's.
+ *
+ * Two objects outlive the invocation and are shared by every instance the factory creates: the resolved
+ * {@link InsightSettings} and the {@link ExportScheduler}. The scheduler is shared on purpose — serializing exports is
+ * a property of the exporters, which belong to the environment, not to one invocation.
+ *
+ * Ownership. Three groups of fields:
+ *
+ * {@link #closed} is additionally {@code volatile}: the hook threads read it without the lock as a fast pre-check.
+ * That read only ever skips work — the authoritative check is made under the monitor by
+ * {@link ExportScheduler#scheduleIfOpen}. It is written once, from false to true, and never back: an execution that
+ * suspends and resumes gets a new instance rather than a reset one.
+ */
+final class InsightPlugin implements DurableExecutionPlugin {
+
+ /** Resolved configuration, shared by every instance of the environment. */
+ private final InsightSettings settings;
+
+ /**
+ * Shared with every other instance: exports are serialized across the whole environment. Package-private because it
+ * is the monitor this instance's scheduling fields are guarded by, which the tests in this package hold when they
+ * read them.
+ */
+ final ExportScheduler scheduler;
+
+ // --- Identity, from the InvocationInfo the factory was called with. ---
+
+ /** The execution this instance observes. */
+ final String executionArn;
+
+ /** The parsed execution ARN, parsed once for every record this instance builds. */
+ final ArnParser arn;
+
+ /** Stable execution start time, from {@code InvocationInfo.executionStartTime()}. */
+ final Instant startTime;
+
+ /** The one-time sampling decision; deterministic in the ARN, so a resumed invocation decides the same way. */
+ final boolean sampledIn;
+
+ // --- The input snapshot. ---
+
+ /**
+ * Detached snapshot of the execution input, the single source of truth for {@code input} on every emission of this
+ * invocation. Written by {@code onInvocationStart}, read by every later build.
+ */
+ private volatile Object cachedInput;
+
+ // --- Build ordering. ---
+
+ /**
+ * Counts the record builds this invocation has started. The value a build takes identifies that build.
+ *
+ * Customer code runs inside a build, on the hook thread: the input and output content transforms, an operation's
+ * result transform, and any Jackson serializer registered for a customer type. That code can call back into a hook
+ * of this same instance, and it runs before anything is scheduled, so a build can be overtaken by a newer build
+ * that starts and finishes inside it. Two hook threads for one invocation would produce the same overlap.
+ *
+ * The scheduler's slot holds one record per invocation and takes whichever record is handed to it last, with no
+ * comparison of age. An overtaken build would therefore write its older snapshot over the newer one. Every build
+ * takes the next value here before it starts, and the scheduler queues the record only while that value is still
+ * the newest, so an overtaken build's record is dropped instead.
+ *
+ * An {@link AtomicLong} rather than a {@code volatile long}: {@code ++} on a {@code volatile long} is a
+ * read-modify-write, so two concurrent builds can take the same value and each conclude its own record is the
+ * newest. That is the very case the check exists for, so a racy counter would guard nothing.
+ */
+ private final AtomicLong buildRevision = new AtomicLong();
+
+ // --- Scheduling state: guarded by the scheduler's monitor. ---
+
+ /**
+ * The latest record for this invocation that no pump has picked up yet, or {@code null} when none is queued.
+ *
+ * A newer record replaces an older one here — each record is a complete snapshot, so the older one carries
+ * nothing the newer one lacks. That is the whole of coalescing: one slot, on the instance, which no other
+ * invocation can reach.
+ *
+ * Which record is newer is decided by {@link #buildRevision}, not by the order the records reach this slot. The
+ * slot itself takes the last hand-off unconditionally, and the last hand-off is not the newest build when a build
+ * was overtaken by one that customer code started from inside it.
+ */
+ WorkflowInsightRecord record;
+
+ /**
+ * Completes once this invocation's latest record has been handed to every exporter; {@code null} when nothing is
+ * outstanding.
+ */
+ CompletableFuture Set and cleared in the same critical sections that move {@link #record}, so "no queued record" is never
+ * mistaken for "nothing outstanding" while the record is inside the exporters.
+ */
+ boolean exporting;
+
+ /**
+ * How many {@code drain} calls are waiting for this invocation right now.
+ *
+ * A record with a waiter gates an invocation return, so the pump exports it before it spends a flush fan-out.
+ */
+ int drainWaiters;
+
+ /**
+ * Set once invocation end begins; never cleared. Guarded by the scheduler's monitor — the same monitor that queues
+ * the record, so the check and the hand-off are one critical section — and {@code volatile} for the hook-side
+ * pre-check.
+ *
+ * A checkpoint that completes while the end record is being drained still delivers an operation-change hook to
+ * this same instance, and that RUNNING snapshot must not supersede the final record.
+ *
+ * This orders RUNNING records against the final record; {@link #buildRevision} orders RUNNING records against
+ * each other. Neither covers the other's case. A boolean cannot say which of two RUNNING builds is newer, and the
+ * revision cannot reject a RUNNING record that follows the final one, because the final record is queued without a
+ * revision check. See {@link ExportScheduler#closeAndSchedule}.
+ */
+ volatile boolean closed;
+
+ /**
+ * Creates the instance that serves one invocation. Identity comes from {@code info} rather than from the first
+ * hook, so every field a record is keyed by exists before any hook can fire.
+ *
+ * @throws RuntimeException if the invocation has no usable execution ARN; the SDK contains that exactly as it
+ * contains a hook failure, by skipping this plugin for the invocation
+ */
+ InsightPlugin(InsightSettings settings, ExportScheduler scheduler, InvocationInfo info) {
+ this.settings = settings;
+ this.scheduler = scheduler;
+ this.executionArn = info.durableExecutionArn();
+ this.arn = ArnParser.parse(executionArn);
+ this.startTime = info.executionStartTime();
+ this.sampledIn = WorkflowInsight.shouldSample(executionArn, settings.samplingRate);
+ }
+
+ /** Test seam: waits until every scheduled record has been handed to the exporters. */
+ void drainExports() {
+ scheduler.drainAll();
+ }
+
+ @Override
+ public void onInvocationStart(InvocationInfo info) {
+ try {
+ if (!sampledIn) {
+ return;
+ }
+ // Detach the execution input from the live handler value immediately, before the user handler or any
+ // content transform can mutate it. This raw, detached snapshot is the single source of truth for input
+ // on every emission (start / change / end); each build hands transforms a separate defensive copy so a
+ // mutating transform cannot corrupt it. Guard the snapshot: a Throwable here (e.g. a payload whose
+ // serialization overflows the stack) must omit the captured input, never fail the user handler.
+ try {
+ cachedInput = Json.deepCopyContent(info.executionInput());
+ } catch (Throwable t) {
+ WorkflowInsight.logSafely("failed to snapshot execution input; omitting input", t);
+ cachedInput = null;
+ }
+ if (settings.emitMode == WorkflowInsightConfig.EmitMode.ON_CHANGE) {
+ // The revision is taken before the build, never after. Customer code runs inside buildRecord and can
+ // re-enter a hook of this instance, which builds a newer record; a revision read afterwards would
+ // already be that newer build's, and this older record would pass the check and overwrite it.
+ long revision = beginBuild();
+ scheduler.scheduleIfNotSuperseded(
+ this, buildRecord("RUNNING", info.operations(), null, cachedInput, null, null), revision);
+ }
+ } catch (Throwable t) {
+ WorkflowInsight.logSafely("onInvocationStart failed", t);
+ }
+ }
+
+ @Override
+ public void onOperationChange(OperationChangeInfo info) {
+ try {
+ if (settings.emitMode != WorkflowInsightConfig.EmitMode.ON_CHANGE || !sampledIn) {
+ return;
+ }
+ // Lock-free pre-check: this invocation's end may already have begun, in which case no RUNNING snapshot may
+ // follow the final record. The authoritative check is made again under the scheduler's lock below.
+ if (closed) {
+ return;
+ }
+ long revision = beginBuild();
+ scheduler.scheduleIfNotSuperseded(
+ this, buildRecord("RUNNING", info.operations(), null, cachedInput, null, null), revision);
+ } catch (Throwable t) {
+ WorkflowInsight.logSafely("onOperationChange failed", t);
+ }
+ }
+
+ // onInvocationEnd is the hook the SDK awaits, so it is where the export queue is drained before the invocation
+ // returns; this guarantees the final record (scheduled above the drain) is delivered. The drain and flush run
+ // in finally so they also cover the paths where record construction fails.
+ @Override
+ public void onInvocationEnd(InvocationEndInfo info) {
+ try {
+ String status = WorkflowInsight.mapStatus(info.invocationStatus());
+ boolean isTerminal = "SUCCEEDED".equals(status) || "FAILED".equals(status);
+ boolean isFailure = "FAILED".equals(status);
+ boolean shouldEmit;
+ switch (settings.emitMode) {
+ case ON_CHANGE:
+ shouldEmit = true;
+ break;
+ case ON_FAILURE:
+ shouldEmit = isFailure;
+ break;
+ case ON_COMPLETE:
+ default:
+ shouldEmit = isTerminal;
+ break;
+ }
+
+ WorkflowInsightRecord finalRecord = null;
+ if (sampledIn && shouldEmit) {
+ // No build revision is taken here. Customer code running inside this build can start a newer RUNNING
+ // build, which would make a revision taken here stale, and a checked hand-off would then drop the final
+ // record and leave a RUNNING snapshot as this execution's last exported state. The final record is
+ // instead ordered by `closed`, which closeAndSchedule sets in the same critical section that queues it.
+ finalRecord = buildRecord(
+ status,
+ info.operations(),
+ Instant.now(),
+ cachedInput,
+ info.executionResult(),
+ info.executionError());
+ }
+ // Close before the drain below: an operation-change hook arriving from a checkpoint that completes
+ // during the drain is rejected, so no RUNNING snapshot can follow (or replace) the final record.
+ scheduler.closeAndSchedule(this, finalRecord);
+ } catch (Throwable t) {
+ // A plugin failure at end-of-invocation (record construction, transforms, truncation, export/flush,
+ // or optional exporter class linkage) must never disrupt durable execution.
+ WorkflowInsight.logSafely("onInvocationEnd failed", t);
+ } finally {
+ // If record construction failed above, this instance is still open: close it so a late change hook cannot
+ // schedule into the drain. Idempotent when already closed.
+ scheduler.closeAndSchedule(this, null);
+ // Sampled-out invocations never schedule a record, so there is nothing to drain or flush. The drain needs
+ // no lookup: this instance is the thing whose record it waits for.
+ if (sampledIn) {
+ drainAndFlush();
+ }
+ // Nothing is released here. There is no per-execution entry to remove — this instance is the state, the SDK
+ // drops it when the invocation returns, and a suspended execution that resumes in the same container is
+ // served by a new instance built from the resume's own InvocationInfo (same stable start time, same
+ // deterministic sampling decision, its own input snapshot). A plugin failure therefore cannot turn into a
+ // state leak, because there is no place a leak could accumulate.
+ }
+ }
+
+ /**
+ * Waits for this invocation's scheduled record to reach the exporters, then flushes each exporter once. The wait is
+ * per invocation: another execution running in the same environment can never displace this record, so this always
+ * returns having delivered this invocation's latest snapshot. It is not insulated from the queue, though — one pump
+ * exports serially, so records another execution had already queued ahead of this one are exported first and this
+ * drain waits for them too.
+ *
+ * The flush goes through the scheduler's queue and is served by that same pump, between records, so no exporter
+ * ever sees this invocation's {@code flush()} overlap another's {@code export()}. Invocation ends that overlap
+ * share one flush: the cadence the exporter contract promises is at most one flush per sampled-in invocation end,
+ * not exactly one.
+ */
+ private void drainAndFlush() {
+ try {
+ scheduler.drain(this);
+ } catch (Throwable t) {
+ WorkflowInsight.logSafely("failed to drain export scheduler", t);
+ }
+ try {
+ scheduler.flush();
+ } catch (Throwable t) {
+ WorkflowInsight.logSafely("exporter flush failed", t);
+ }
+ }
+
+ // --- Record building. ---
+
+ /** Starts a record build and returns the revision that identifies it. */
+ private long beginBuild() {
+ return buildRevision.incrementAndGet();
+ }
+
+ /**
+ * Whether the identified build is still the newest one this invocation has started.
+ *
+ * Read by the scheduler inside the critical section that queues the record, so a record that passes cannot be
+ * queued after a record that supersedes it. A build that starts after the check passes still supersedes this one:
+ * its record is handed over later and replaces this one in the slot, which is the order the slot should have.
+ */
+ boolean isNewestBuild(long revision) {
+ return buildRevision.get() == revision;
+ }
+
+ private WorkflowInsightRecord buildRecord(
+ String status,
+ Map This belongs to the execution environment, not to an invocation. {@link WorkflowInsight#workflowInsight} resolves
+ * it once and the factory it returns hands the same instance to every {@link InsightPlugin} it creates, so resolving
+ * defaults, validating the sampling rate and indexing the operation overrides happen once per environment rather than
+ * once per invocation.
+ */
+final class InsightSettings {
+
+ /** Sampling rate, clamped to [0, 1]; the per-invocation decision is derived from it and the execution ARN. */
+ final double samplingRate;
+
+ final WorkflowInsightConfig.EmitMode emitMode;
+
+ /** True when nested operations (parallel branches, map items, nested steps) are dropped from the record. */
+ final boolean topLevelOnly;
+
+ final boolean includeErrors;
+
+ /** May be null, which means "every default": include input, output and errors, with no transforms. */
+ final ContentConfig content;
+
+ /** Operation overrides indexed by operation name, in declaration order. */
+ final Map Per-execution state (keyed by execution ARN) holds only the stable start time, the parsed ARN, the one-time
- * sampling decision, and a detached snapshot of the execution input. State is removed on every {@code onInvocationEnd}
- * — including non-terminal PENDING/RETRYING suspends — so a suspended execution never leaks a retained entry for the
- * lifetime of a warm container. Nothing is lost across a resume: the next invocation recreates the same stable start
- * time from {@link InvocationInfo#executionStartTime()}, the same sampling decision deterministically from the ARN, and
- * the input snapshot from {@link InvocationInfo#executionInput()}.
+ * {@link #workflowInsight} returns a {@link DurableExecutionPluginFactory}, so the SDK creates one
+ * {@link InsightPlugin} per Lambda invocation and drops it when the invocation returns. Everything about an execution —
+ * the stable start time, the parsed ARN, the one-time sampling decision, the detached input snapshot, the queued
+ * record, the drain signal — is therefore a plain field of that instance. Nothing is keyed by execution ARN, and there
+ * is no per-execution entry to remove at invocation end, so a suspended execution cannot leak one for the lifetime of a
+ * warm container. Nothing is lost across a resume either: the resume's own {@link InvocationInfo} carries the same
+ * stable start time, the sampling decision is deterministic in the ARN, and the input snapshot is taken again from
+ * {@link InvocationInfo#executionInput()}.
+ *
+ * What belongs to the execution environment rather than to an invocation stays in the factory: the resolved
+ * {@link InsightSettings}, the exporters, and the {@link ExportScheduler} that serializes exports across every
+ * execution the environment hosts.
*/
@Experimental
public final class WorkflowInsight {
@@ -49,369 +46,44 @@ public final class WorkflowInsight {
private WorkflowInsight() {}
- /** Creates a Workflow Insight plugin from the given config. Mirrors the JS {@code workflowInsight(config)}. */
- public static DurableExecutionPlugin workflowInsight(WorkflowInsightConfig config) {
- return new InsightPlugin(config);
- }
-
- /** Per-execution state, keyed by execution ARN, to prevent warm-container bleed and handle resume. */
- private static final class ExecutionState {
- final Instant startTime;
- final ArnParser arn;
- final boolean sampledIn;
- volatile Object cachedInput;
-
- /**
- * Set once invocation end begins; guarded by {@code this}. A checkpoint that completes while the end record is
- * being drained still delivers an operation-change hook, and that RUNNING snapshot must not supersede the final
- * record.
- */
- boolean closed;
-
- ExecutionState(Instant startTime, ArnParser arn, boolean sampledIn) {
- this.startTime = startTime;
- this.arn = arn;
- this.sampledIn = sampledIn;
- }
-
- /** Schedules the record unless the invocation has already ended; the check and the hand-off are atomic. */
- boolean scheduleIfOpen(ExportScheduler scheduler, WorkflowInsightRecord record) {
- synchronized (this) {
- if (closed) {
- return false;
- }
- scheduler.schedule(record);
- return true;
- }
- }
-
- /** Marks the invocation ended and, when given a record, schedules it as the last one for this execution. */
- void closeAndSchedule(ExportScheduler scheduler, WorkflowInsightRecord finalRecord) {
- synchronized (this) {
- closed = true;
- if (finalRecord != null) {
- scheduler.schedule(finalRecord);
- }
- }
- }
+ /**
+ * Creates a Workflow Insight plugin factory from the given config. Mirrors the JS {@code workflowInsight(config)}.
+ *
+ * The configuration is resolved once, here; the exporters and the scheduler that serializes exports across them
+ * are created once, here. The returned factory then builds one plugin instance per invocation, which is what lets
+ * that instance hold its execution's state in plain fields.
+ *
+ * @param config the plugin configuration
+ * @return a factory to hand to {@code DurableConfig.Builder.withPlugins}
+ */
+ public static DurableExecutionPluginFactory workflowInsight(WorkflowInsightConfig config) {
+ InsightSettings settings = new InsightSettings(config);
+ ExportScheduler scheduler = new ExportScheduler(
+ settings.exporters, WorkflowInsight::exportRecord, t -> logSafely("export scheduling failed", t));
+ return info -> new InsightPlugin(settings, scheduler, info);
}
- static final class InsightPlugin implements DurableExecutionPlugin {
- private final double samplingRate;
- private final WorkflowInsightConfig.EmitMode emitMode;
- private final boolean topLevelOnly;
- private final boolean includeErrors;
- private final ContentConfig content;
- private final Map The state is built through the {@link Executor} seam rather than by racing threads: the executor parks the
+ * pump task and reports rejection, which is the same shape the scheduler produces on its own in the window between
+ * the pump loop's return to idle and its {@code finally} — a live pump whose handle is no longer the installed one.
+ */
+ @Test
+ void anExitingPumpDoesNotReleaseADrainWhoseRecordIsStillInsideTheExporter() throws Exception {
+ var executor = new ParkingExecutor();
+ var exporting = new CountDownLatch(1);
+ var release = new CountDownLatch(1);
+ Set The scheduler no longer resolves an execution ARN to anything — an invocation's state is its plugin
+ * instance, created from the {@link InvocationInfo} the SDK is about to hand the first hook — so tests hold the
+ * instance exactly as the SDK does. Everything here goes through the same constructor and the same factory production
+ * uses; nothing is a test-only back door into the scheduler.
+ */
+final class Executions {
+
+ private static final Instant START = Instant.parse("2026-08-05T00:00:00Z");
+
+ private Executions() {}
+
+ /** The invocation description the SDK would build for one execution, with no payload or operation snapshot. */
+ static InvocationInfo info(String executionArn) {
+ return new InvocationInfo("req", executionArn, true, START, null, Map.of(), Map.of());
+ }
+
+ /**
+ * One invocation's plugin instance, bound to this scheduler and configured with the plugin's defaults. For tests
+ * that drive the scheduler directly and do not care how records are shaped.
+ */
+ static InsightPlugin plugin(ExportScheduler scheduler, String executionArn) {
+ return new InsightPlugin(
+ new InsightSettings(WorkflowInsightConfig.builder().build()), scheduler, info(executionArn));
+ }
+
+ /**
+ * One invocation's plugin instance from the factory, for the identity the SDK would have built it with. The
+ * invocation's own {@code InvocationInfo} still goes to {@code onInvocationStart}; this is the same pair of facts
+ * that info carries, which is all an instance's identity is.
+ */
+ static InsightPlugin plugin(DurableExecutionPluginFactory factory, String executionArn, Instant startTime) {
+ return plugin(factory, new InvocationInfo("req", executionArn, true, startTime, null, Map.of(), Map.of()));
+ }
+
+ /**
+ * One invocation's plugin instance, exactly as the SDK creates it: from the factory, with that invocation's info.
+ */
+ static InsightPlugin plugin(DurableExecutionPluginFactory factory, InvocationInfo info) {
+ return (InsightPlugin) factory.createPlugin(info);
+ }
+
+ /**
+ * Whether the scheduler still owes this invocation anything: a queued record, a record inside the exporters, an
+ * uncompleted drain signal, or a drain waiting on it. Read under the monitor those fields are guarded by.
+ *
+ * This is the question the plugin's {@code retainedStateCount()} seam used to answer for a whole registry. There
+ * is no registry to count now — an invocation's state is its plugin instance, and the SDK drops it — so the
+ * property worth asserting is that nothing the environment outlives keeps hold of it.
+ */
+ static boolean outstanding(InsightPlugin plugin) {
+ synchronized (plugin.scheduler) {
+ return plugin.record != null || plugin.exporting || plugin.settled != null || plugin.drainWaiters > 0;
+ }
+ }
+
+ /**
+ * The instance plus its first hook, in the order the SDK dispatches them: the factory is called with the very
+ * {@link InvocationInfo} that {@code onInvocationStart} then receives.
+ */
+ static InsightPlugin started(DurableExecutionPluginFactory factory, InvocationInfo info) {
+ InsightPlugin plugin = plugin(factory, info);
+ plugin.onInvocationStart(info);
+ return plugin;
+ }
+}
diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/ExportSchedulerFanOutReentryTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/ExportSchedulerFanOutReentryTest.java
new file mode 100644
index 000000000..973fe97d3
--- /dev/null
+++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/ExportSchedulerFanOutReentryTest.java
@@ -0,0 +1,199 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.insight;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.api.Test;
+
+/**
+ * A {@code flush()} or {@code drain()} issued from an exporter fan-out worker is refused and reported, not waited on.
+ *
+ * With two or more exporters configured, the pump does not run the exporter callbacks itself: it submits one task
+ * per exporter and then waits for all of them. A callback therefore runs on a worker the pump is blocked on, and a wait
+ * for the pump issued from that worker is a wait-for cycle two threads wide — the worker parks on a future only the
+ * pump can complete, and the pump cannot resume its loop until that worker returns. The single-exporter case runs the
+ * callback on the pump thread itself and is covered by {@link ExportSchedulerReentrantFlushTest}; this covers the
+ * fan-out, which the pump-thread identity check alone does not recognize.
+ */
+class ExportSchedulerFanOutReentryTest {
+
+ /** Longest a call that must return promptly may take before the property under test is considered broken. */
+ private static final long DEADLINE_MILLIS = 5_000;
+
+ private static String arn(int index) {
+ return "arn:aws:lambda:us-west-2:111122223333:function:f:$LATEST/durable-execution/exec-" + index + "/inv-1";
+ }
+
+ private static WorkflowInsightRecord record(String executionArn, String status) {
+ var r = new WorkflowInsightRecord();
+ r.executionArn = executionArn;
+ r.status = status;
+ return r;
+ }
+
+ /** Unbounded thread-per-task executor, like the cached pool the plugin injects in production. */
+ private static Executor sharedWorkers() {
+ return command -> {
+ var thread = new Thread(command, "fan-out-reentry-worker");
+ thread.setDaemon(true);
+ thread.start();
+ };
+ }
+
+ private static final class CountingExporter implements InsightExporter {
+ final AtomicInteger exports = new AtomicInteger();
+ final AtomicInteger flushes = new AtomicInteger();
+
+ @Override
+ public void export(WorkflowInsightRecord record) {
+ exports.incrementAndGet();
+ }
+
+ @Override
+ public void flush() {
+ flushes.incrementAndGet();
+ }
+ }
+
+ @Test
+ void flushReenteredFromAFanOutWorkerIsRefusedReportedAndLosesNoWork() throws Exception {
+ var first = new CountingExporter();
+ var second = new CountingExporter();
+ var failures = new CopyOnWriteArrayList Coalescing is sound because every requester drains its own record before asking, so a flush that starts
+ * after the request was made has that record in the buffer. It is what stops N ends that ask together from paying for N
+ * serialized flush fan-outs — a 60 ms exporter flush cost the slowest of 8 ends ~520 ms before this change.
+ */
+class ExportSchedulerFlushCoalescingTest {
+
+ /** A slow flush must not cost the caller more than a small multiple of the one flush it asked for. */
+ private static final int PROMPTNESS_FACTOR = 3;
+
+ private static String arn(int index) {
+ return "arn:aws:lambda:us-west-2:111122223333:function:f:$LATEST/durable-execution/exec-" + index + "/inv-1";
+ }
+
+ private static WorkflowInsightRecord record(String executionArn, String status) {
+ var r = new WorkflowInsightRecord();
+ r.executionArn = executionArn;
+ r.status = status;
+ return r;
+ }
+
+ private static ExportScheduler scheduler(Executor executor, InsightExporter... exporters) {
+ return new ExportScheduler(List.of(exporters), (rec, exp) -> exp.export(rec), t -> {}, executor);
+ }
+
+ /** Unbounded thread-per-task executor, like the cached pool the plugin injects in production. */
+ private static Executor sharedWorkers() {
+ return command -> {
+ var thread = new Thread(command, "coalescing-worker");
+ thread.setDaemon(true);
+ thread.start();
+ };
+ }
+
+ private static final class SlowFlushExporter implements InsightExporter {
+ private final long flushMillis;
+ final AtomicInteger flushes = new AtomicInteger();
+
+ SlowFlushExporter(long flushMillis) {
+ this.flushMillis = flushMillis;
+ }
+
+ @Override
+ public void export(WorkflowInsightRecord record) {}
+
+ @Override
+ public void flush() {
+ flushes.incrementAndGet();
+ sleep(flushMillis);
+ }
+ }
+
+ /**
+ * N invocation ends whose records have already been delivered ask for their flush together: they share one flush,
+ * so the slowest pays a small multiple of one flush rather than N times one.
+ *
+ * The records are drained before the requests are made on purpose: this is coalescing on its own, with every
+ * request already queued when the pump reaches its flush step. The harder case — ends that are still inside
+ * {@code drain()} when the first request is served, and so cannot have asked yet — is
+ * {@link #simultaneousDrainAndFlushEndsShareAFlushRatherThanOneEach()}.
+ */
+ @Test
+ void invocationEndsAskingForAFlushTogetherShareOneFlush() {
+ long flushMillis = 60;
+ int executions = 8;
+ var exporter = new SlowFlushExporter(flushMillis);
+ var scheduler = scheduler(sharedWorkers(), exporter);
+
+ var durations = Collections.synchronizedList(new java.util.ArrayList The cadence itself — at most one flush per request, requests that overlap sharing one flush — is covered by
+ * {@link ExportSchedulerFlushCoalescingTest}.
+ */
+class ExportSchedulerFlushSerializationTest {
+
+ /** Longest a flush may take to be served before the property under test is considered broken. */
+ private static final long FLUSH_DEADLINE_MILLIS = 2_000;
+
+ private static String arn(int index) {
+ return "arn:aws:lambda:us-west-2:111122223333:function:f:$LATEST/durable-execution/exec-" + index + "/inv-1";
+ }
+
+ private static WorkflowInsightRecord record(String executionArn, String status) {
+ var r = new WorkflowInsightRecord();
+ r.executionArn = executionArn;
+ r.status = status;
+ return r;
+ }
+
+ private static ExportScheduler scheduler(
+ Executor executor, List That thread is the only one able to serve the request it would be making — flush requests are served by the pump,
+ * between records — so waiting for it is a wait-for cycle one thread wide, and the invocation never returns. With a
+ * single exporter the fan-out runs on the pump thread, so anything an exporter's {@code export()} does synchronously is
+ * enough to reach it.
+ */
+class ExportSchedulerReentrantFlushTest {
+
+ /** Longest a call that must return promptly may take before the property under test is considered broken. */
+ private static final long DEADLINE_MILLIS = 5_000;
+
+ private static String arn(int index) {
+ return "arn:aws:lambda:us-west-2:111122223333:function:f:$LATEST/durable-execution/exec-" + index + "/inv-1";
+ }
+
+ private static WorkflowInsightRecord record(String executionArn, String status) {
+ var r = new WorkflowInsightRecord();
+ r.executionArn = executionArn;
+ r.status = status;
+ return r;
+ }
+
+ /** Unbounded thread-per-task executor, like the cached pool the plugin injects in production. */
+ private static Executor sharedWorkers() {
+ return command -> {
+ var thread = new Thread(command, "reentrant-flush-worker");
+ thread.setDaemon(true);
+ thread.start();
+ };
+ }
+
+ private static final class CountingExporter implements InsightExporter {
+ final AtomicInteger exports = new AtomicInteger();
+ final AtomicInteger flushes = new AtomicInteger();
+
+ @Override
+ public void export(WorkflowInsightRecord record) {
+ exports.incrementAndGet();
+ }
+
+ @Override
+ public void flush() {
+ flushes.incrementAndGet();
+ }
+ }
+
+ @Test
+ void flushReenteredFromThePumpThreadIsRefusedReportedAndLeavesTheSchedulerUsable() throws Exception {
+ var exporter = new CountingExporter();
+ var failures = new CopyOnWriteArrayList
+ *
+ *
+ *