diff --git a/insight-plugin/README.md b/insight-plugin/README.md index f626ad830..dce94cfad 100644 --- a/insight-plugin/README.md +++ b/insight-plugin/README.md @@ -69,6 +69,13 @@ application: - **Emission modes.** `ON_COMPLETE` emits one terminal record; `ON_FAILURE` emits only on terminal failure; `ON_CHANGE` emits at invocation start, on every operation change, and at invocation end (matching JS). Non-terminal statuses map to `RUNNING`. +- **Bounded `ON_CHANGE` delivery.** Start and operation-change hooks enqueue complete snapshots + without waiting for exporter I/O. Each invocation keeps a FIFO of up to 16 waiting snapshots plus + one in-flight export. When that bound is reached, only the oldest waiting `RUNNING` snapshot is + dropped; normal bursts are preserved rather than unconditionally coalesced. Invocation end seals + the queue, reserves the final snapshot, drains records in order, and flushes each exporter once + before Lambda can freeze the environment. A late `RUNNING` snapshot cannot follow or overwrite + the invocation-final record. - **Operation filtering** mirrors JS: the `EXECUTION` pseudo-operation and unnamed operations are dropped; `TOP_LEVEL` detail drops any operation with a `parentId`; an `OperationOverride.exclude` drops by name. Operation `result` is included only when an `OperationOverride.withResult` diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/ExportScheduler.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/ExportScheduler.java new file mode 100644 index 000000000..f18eb2c4e --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/ExportScheduler.java @@ -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 export; + private final Runnable flush; + private final Consumer failureHandler; + private final ArrayDeque queue = new ArrayDeque<>(); + private final CompletableFuture drained = new CompletableFuture<>(); + + private boolean running; + private boolean sealed; + + ExportScheduler(Consumer export, Runnable flush, Consumer failureHandler) { + this(DEFAULT_CAPACITY, WORKERS, export, flush, failureHandler); + } + + ExportScheduler( + int capacity, + Executor executor, + Consumer export, + Runnable flush, + Consumer 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 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); + } + } + } + + 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. + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsight.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsight.java index c8979bb82..e5b4e629e 100644 --- a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsight.java +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsight.java @@ -41,6 +41,11 @@ * 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()}. + * + *

{@code ON_CHANGE} hooks enqueue complete snapshots into a bounded per-invocation FIFO; they never perform exporter + * I/O on the checkpoint callback thread. The invocation-end hook seals the FIFO with the final snapshot, drains it in + * order, and flushes exporters before removing state. Under backpressure only the oldest waiting snapshot is dropped, + * while the invocation-final snapshot is always retained and exported last. */ @Experimental public final class WorkflowInsight { @@ -59,12 +64,14 @@ private static final class ExecutionState { final Instant startTime; final ArnParser arn; final boolean sampledIn; + final ExportScheduler scheduler; volatile Object cachedInput; - ExecutionState(Instant startTime, ArnParser arn, boolean sampledIn) { + ExecutionState(Instant startTime, ArnParser arn, boolean sampledIn, ExportScheduler scheduler) { this.startTime = startTime; this.arn = arn; this.sampledIn = sampledIn; + this.scheduler = scheduler; } } @@ -100,8 +107,14 @@ int retainedStateCount() { } private ExecutionState getState(String arn, Instant startTime) { - return byArn.computeIfAbsent( - arn, a -> new ExecutionState(startTime, ArnParser.parse(a), shouldSample(a, samplingRate))); + return byArn.computeIfAbsent(arn, a -> { + boolean sampledIn = shouldSample(a, samplingRate); + ExportScheduler scheduler = sampledIn && emitMode == WorkflowInsightConfig.EmitMode.ON_CHANGE + ? new ExportScheduler( + this::exportRecord, this::flushExporters, t -> logSafely("export scheduler failed", t)) + : null; + return new ExecutionState(startTime, ArnParser.parse(a), sampledIn, scheduler); + }); } @Override @@ -122,8 +135,8 @@ public void onInvocationStart(InvocationInfo info) { logSafely("failed to snapshot execution input; omitting input", t); state.cachedInput = null; } - if (emitMode == WorkflowInsightConfig.EmitMode.ON_CHANGE) { - emit(buildRecord( + if (state.scheduler != null) { + state.scheduler.schedule(buildRecord( state, info.durableExecutionArn(), "RUNNING", @@ -148,7 +161,7 @@ public void onOperationChange(OperationChangeInfo info) { if (state == null || !state.sampledIn) { return; } - emit(buildRecord( + state.scheduler.schedule(buildRecord( state, info.durableExecutionArn(), "RUNNING", @@ -164,8 +177,9 @@ public void onOperationChange(OperationChangeInfo info) { @Override public void onInvocationEnd(InvocationEndInfo info) { + ExecutionState state = null; try { - ExecutionState state = getState(info.durableExecutionArn(), info.executionStartTime()); + state = getState(info.durableExecutionArn(), info.executionStartTime()); String status = mapStatus(info.invocationStatus()); boolean isTerminal = "SUCCEEDED".equals(status) || "FAILED".equals(status); boolean isFailure = "FAILED".equals(status); @@ -184,7 +198,7 @@ public void onInvocationEnd(InvocationEndInfo info) { } if (state.sampledIn && shouldEmit) { - emit(buildRecord( + WorkflowInsightRecord record = buildRecord( state, info.durableExecutionArn(), status, @@ -192,44 +206,78 @@ public void onInvocationEnd(InvocationEndInfo info) { Instant.now(), state.cachedInput, info.executionResult(), - info.executionError())); + info.executionError()); + if (state.scheduler != null) { + state.scheduler.sealAndDrain(record).join(); + } else { + emitAndFlush(record); + } } } 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. logSafely("onInvocationEnd failed", t); } finally { + // If final record construction failed, still seal and drain any ON_CHANGE records already queued. + if (state != null && state.scheduler != null) { + try { + state.scheduler.sealAndDrain(null).join(); + } catch (Throwable t) { + logSafely("failed to drain export scheduler", t); + } + } // Remove per-execution state on EVERY invocation end, including non-terminal PENDING/RETRYING suspends, // once any emission work above is done. Nothing durable is lost: the next invocation's onInvocation // start recreates the stable startTime from InvocationInfo.executionStartTime() (stable across - // resumes), - // the one-time sampling decision deterministically from the ARN, and the input snapshot from - // InvocationInfo.executionInput(). Retaining state instead leaked one entry per suspended execution for - // the lifetime of the warm container. This runs even if emission above threw, so a plugin failure can - // never turn into a state leak. + // resumes), the one-time sampling decision deterministically from the ARN, and the input snapshot from + // InvocationInfo.executionInput(). byArn.remove(info.durableExecutionArn()); } } + private void emitAndFlush(WorkflowInsightRecord record) { + for (InsightExporter exporter : exporters) { + exportRecord(record, exporter); + flushExporter(exporter); + } + } + /** Serializes each record to every exporter, isolating failures so one exporter never blocks the others. */ - private void emit(WorkflowInsightRecord record) { + private void exportRecord(WorkflowInsightRecord record) { for (InsightExporter exporter : exporters) { - try { - // Give each exporter its own deep copy: truncation returns the original record when it already - // fits, so without this a custom exporter that mutates operations or nested content would corrupt - // every exporter that runs after it. - WorkflowInsightRecord isolated = record.deepCopy(); - WorkflowInsightRecord shaped = - Truncation.truncateRecord(isolated, exporter.maxRecordSizeBytes(), exporter::render); - exporter.export(shaped); - exporter.flush(); - } catch (Throwable t) { - // Catch Throwable, not just RuntimeException: deep copy, truncation, an exporter's render/export/ - // flush, or the linkage of an optional exporter class (a NoClassDefFoundError when the S3 / - // CloudWatch SDK is absent) can each fail with an Error. Isolating every Throwable here guarantees - // one failing exporter cannot block the exporters that run after it, nor disrupt the execution. - logSafely("exporter failed", t); - } + exportRecord(record, exporter); + } + } + + private void exportRecord(WorkflowInsightRecord record, InsightExporter exporter) { + try { + // Give each exporter its own deep copy: truncation returns the original record when it already fits, + // so without this a custom exporter that mutates operations or nested content would corrupt every + // exporter that runs after it. + WorkflowInsightRecord isolated = record.deepCopy(); + WorkflowInsightRecord shaped = + Truncation.truncateRecord(isolated, exporter.maxRecordSizeBytes(), exporter::render); + exporter.export(shaped); + } catch (Throwable t) { + // Catch Throwable, not just RuntimeException: deep copy, truncation, an exporter's render/export, or + // optional exporter class linkage can each fail with an Error. Isolating every Throwable here + // guarantees one failing exporter cannot block the exporters after it or disrupt the execution. + logSafely("exporter failed", t); + } + } + + /** Flushes each exporter once at the invocation boundary after all scheduled records have been exported. */ + private void flushExporters() { + for (InsightExporter exporter : exporters) { + flushExporter(exporter); + } + } + + private void flushExporter(InsightExporter exporter) { + try { + exporter.flush(); + } catch (Throwable t) { + logSafely("exporter flush failed", t); } } diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/ExportSchedulerTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/ExportSchedulerTest.java new file mode 100644 index 000000000..cb2b70075 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/ExportSchedulerTest.java @@ -0,0 +1,179 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; + +class ExportSchedulerTest { + + private static final class ManualExecutor implements Executor { + private final ArrayDeque tasks = new ArrayDeque<>(); + + @Override + public void execute(Runnable command) { + tasks.addLast(command); + } + + void runNext() { + tasks.removeFirst().run(); + } + } + + private static final class RejectOnceExecutor implements Executor { + private Runnable task; + private boolean reject = true; + + @Override + public void execute(Runnable command) { + if (reject) { + reject = false; + throw new RejectedExecutionException("transient rejection"); + } + task = command; + } + + void run() { + task.run(); + } + } + + @Test + void preservesNoDelayBurstWithinCapacityAndFlushesAfterFinal() { + var executor = new ManualExecutor(); + var calls = new ArrayList(); + var scheduler = scheduler(16, executor, calls, new AtomicInteger()); + + for (var i = 1; i <= 11; i++) { + assertTrue(scheduler.schedule(record("running-" + i))); + } + var drained = scheduler.sealAndDrain(record("final")); + + assertFalse(drained.isDone()); + executor.runNext(); + drained.join(); + + assertEquals(13, calls.size()); + for (var i = 1; i <= 11; i++) { + assertEquals("export:running-" + i, calls.get(i - 1)); + } + assertEquals("export:final", calls.get(11)); + assertEquals("flush", calls.get(12)); + } + + @Test + void retriesQueuedRecordsAfterTransientExecutorRejection() { + var executor = new RejectOnceExecutor(); + var calls = new ArrayList(); + var failures = new AtomicInteger(); + var scheduler = scheduler(4, executor, calls, failures); + + scheduler.schedule(record("running-1")); + assertEquals(1, failures.get()); + scheduler.schedule(record("running-2")); + var drained = scheduler.sealAndDrain(record("final")); + executor.run(); + drained.join(); + + assertEquals(List.of("export:running-1", "export:running-2", "export:final", "flush"), calls); + } + + @Test + void dropsOldestWaitingRecordsOnlyAfterCapacityIsReached() { + var executor = new ManualExecutor(); + var calls = new ArrayList(); + var scheduler = scheduler(3, executor, calls, new AtomicInteger()); + + for (var i = 1; i <= 5; i++) { + scheduler.schedule(record("running-" + i)); + } + var drained = scheduler.sealAndDrain(record("final")); + executor.runNext(); + drained.join(); + + assertEquals(List.of("export:running-4", "export:running-5", "export:final", "flush"), calls); + } + + @Test + void rejectsRecordsScheduledAfterFinalBarrier() { + var executor = new ManualExecutor(); + var calls = new ArrayList(); + var scheduler = scheduler(3, executor, calls, new AtomicInteger()); + + scheduler.schedule(record("running")); + var drained = scheduler.sealAndDrain(record("final")); + assertFalse(scheduler.schedule(record("late-running"))); + executor.runNext(); + drained.join(); + + assertEquals(List.of("export:running", "export:final", "flush"), calls); + } + + @Test + void restartsWorkerWhenFinalRecordArrivesAfterQueueWentIdle() { + var executor = new ManualExecutor(); + var calls = new ArrayList(); + var scheduler = scheduler(3, executor, calls, new AtomicInteger()); + + scheduler.schedule(record("running")); + executor.runNext(); + assertEquals(List.of("export:running"), calls); + + var drained = scheduler.sealAndDrain(record("final")); + assertFalse(drained.isDone()); + executor.runNext(); + drained.join(); + + assertEquals(List.of("export:running", "export:final", "flush"), calls); + } + + @Test + void exporterFailureDoesNotSkipLaterRecordsOrFlush() { + var executor = new ManualExecutor(); + var calls = new ArrayList(); + var failures = new AtomicInteger(); + var scheduler = new ExportScheduler( + 3, + executor, + record -> { + calls.add("export:" + record.executionName); + if ("broken".equals(record.executionName)) { + throw new AssertionError("boom"); + } + }, + () -> calls.add("flush"), + ignored -> failures.incrementAndGet()); + + scheduler.schedule(record("broken")); + var drained = scheduler.sealAndDrain(record("final")); + executor.runNext(); + drained.join(); + + assertEquals(List.of("export:broken", "export:final", "flush"), calls); + assertEquals(1, failures.get()); + } + + private ExportScheduler scheduler(int capacity, Executor executor, List calls, AtomicInteger failures) { + return new ExportScheduler( + capacity, + executor, + record -> calls.add("export:" + record.executionName), + () -> calls.add("flush"), + ignored -> failures.incrementAndGet()); + } + + private WorkflowInsightRecord record(String executionName) { + var record = new WorkflowInsightRecord(); + record.executionName = executionName; + return record; + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/ExporterIsolationTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/ExporterIsolationTest.java index b9590d9a1..721d401a7 100644 --- a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/ExporterIsolationTest.java +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/ExporterIsolationTest.java @@ -14,7 +14,9 @@ import org.junit.jupiter.api.Test; import software.amazon.awssdk.services.lambda.model.OperationStatus; 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.InvocationStatus; import software.amazon.lambda.durable.plugin.OperationChangeItemInfo; /** @@ -71,7 +73,6 @@ void firstExporterMutationsDoNotLeakIntoLaterExporter() { var mutating = new MutatingExporter(); var good = new CapturingExporter(); DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight(WorkflowInsightConfig.builder() - .emitMode(WorkflowInsightConfig.EmitMode.ON_CHANGE) .content(ContentConfig.builder() .addOverride(OperationOverride.withResult("compute", r -> r)) .build()) @@ -99,6 +100,8 @@ void firstExporterMutationsDoNotLeakIntoLaterExporter() { null, "{\"x\":1}")); plugin.onInvocationStart(new InvocationInfo("req", ARN, true, START, input, ops, Map.of())); + plugin.onInvocationEnd( + new InvocationEndInfo("req", ARN, true, START, ops, InvocationStatus.SUCCEEDED, null, input, null)); assertEquals(1, good.records.size()); var rec = good.records.get(0); diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/JsonJavaTimeTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/JsonJavaTimeTest.java index 331a26a14..9c657a7dd 100644 --- a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/JsonJavaTimeTest.java +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/JsonJavaTimeTest.java @@ -14,7 +14,9 @@ import org.junit.jupiter.api.Test; import software.amazon.awssdk.services.lambda.model.OperationStatus; 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.InvocationStatus; import software.amazon.lambda.durable.plugin.OperationChangeItemInfo; /** @@ -56,10 +58,8 @@ public void export(WorkflowInsightRecord record) { @Test void pluginOutputWithInstantInInputSerializesInsteadOfDropping() { var exporter = new CapturingExporter(); - DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight(WorkflowInsightConfig.builder() - .emitMode(WorkflowInsightConfig.EmitMode.ON_CHANGE) - .addExporter(exporter) - .build()); + DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight( + WorkflowInsightConfig.builder().addExporter(exporter).build()); Map input = new LinkedHashMap<>(); input.put("startedAt", TS); @@ -80,6 +80,8 @@ void pluginOutputWithInstantInInputSerializesInsteadOfDropping() { null, null)); plugin.onInvocationStart(new InvocationInfo("req", ARN, true, START, input, ops, Map.of())); + plugin.onInvocationEnd( + new InvocationEndInfo("req", ARN, true, START, ops, InvocationStatus.SUCCEEDED, null, input, null)); assertEquals(1, exporter.records.size()); var rec = exporter.records.get(0); diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/MutableNumberIsolationTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/MutableNumberIsolationTest.java index 31b0413ce..d2bc61c3a 100644 --- a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/MutableNumberIsolationTest.java +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/MutableNumberIsolationTest.java @@ -18,7 +18,9 @@ import org.junit.jupiter.api.Test; import software.amazon.awssdk.services.lambda.model.OperationStatus; 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.InvocationStatus; import software.amazon.lambda.durable.plugin.OperationChangeItemInfo; /** @@ -115,7 +117,6 @@ public void export(WorkflowInsightRecord record) { }; var good = new CapturingExporter(); DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight(WorkflowInsightConfig.builder() - .emitMode(WorkflowInsightConfig.EmitMode.ON_CHANGE) .addExporter(mutating) .addExporter(good) .build()); @@ -129,9 +130,11 @@ public void export(WorkflowInsightRecord record) { input.put("custom", new MutableNumber(99)); plugin.onInvocationStart(new InvocationInfo("req", ARN, true, START, input, ops(), Map.of())); - // Mutate all originals after emission; isolated copies must be unaffected. + // Mutate all originals after snapshotting; the later terminal export must use the detached values. topLevel.set(-1); ((AtomicLong) list.get(0)).set(-1L); + plugin.onInvocationEnd( + new InvocationEndInfo("req", ARN, true, START, ops(), InvocationStatus.SUCCEEDED, null, input, null)); assertEquals(1, good.records.size()); var emitted = assertInstanceOf(Map.class, good.records.get(0).input); diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OperationOrderingTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OperationOrderingTest.java index 6c9d121fb..90dcbb856 100644 --- a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OperationOrderingTest.java +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OperationOrderingTest.java @@ -13,7 +13,9 @@ import org.junit.jupiter.api.Test; import software.amazon.awssdk.services.lambda.model.OperationStatus; 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.InvocationStatus; import software.amazon.lambda.durable.plugin.OperationChangeItemInfo; /** @@ -45,11 +47,11 @@ private static OperationChangeItemInfo item( private WorkflowInsightRecord emitStart(Map ops) { var exporter = new CapturingExporter(); - DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight(WorkflowInsightConfig.builder() - .emitMode(WorkflowInsightConfig.EmitMode.ON_CHANGE) - .addExporter(exporter) - .build()); + DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight( + WorkflowInsightConfig.builder().addExporter(exporter).build()); plugin.onInvocationStart(new InvocationInfo("req", ARN, true, START, "in", ops, Map.of())); + plugin.onInvocationEnd( + new InvocationEndInfo("req", ARN, true, START, ops, InvocationStatus.SUCCEEDED, null, "in", null)); return exporter.records.get(0); } diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/WorkflowInsightHookTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/WorkflowInsightHookTest.java index 3eaebf3d6..cb9534502 100644 --- a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/WorkflowInsightHookTest.java +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/WorkflowInsightHookTest.java @@ -4,13 +4,19 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.time.Duration; import java.time.Instant; import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; import software.amazon.awssdk.services.lambda.model.OperationStatus; import software.amazon.lambda.durable.plugin.DurableExecutionPlugin; @@ -46,6 +52,39 @@ public void flush() { } } + private static final class BlockingExporter implements InsightExporter { + final List records = new ArrayList<>(); + final CountDownLatch firstExportStarted = new CountDownLatch(1); + final CountDownLatch releaseFirstExport = new CountDownLatch(1); + final AtomicInteger exports = new AtomicInteger(); + int flushes; + + @Override + public void export(WorkflowInsightRecord record) { + if (exports.incrementAndGet() == 1) { + firstExportStarted.countDown(); + await(releaseFirstExport); + } + records.add(record); + } + + @Override + public void flush() { + flushes++; + } + + private static void await(CountDownLatch latch) { + try { + if (!latch.await(5, TimeUnit.SECONDS)) { + throw new AssertionError("timed out waiting for test latch"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError("interrupted while waiting for test latch", e); + } + } + } + private Map ops(String name, OperationStatus status) { Map m = new LinkedHashMap<>(); m.put( @@ -81,6 +120,56 @@ void onChangeEmitsAtStartChangeAndEnd() { assertEquals("RUNNING", exporter.records.get(0).status()); assertEquals("RUNNING", exporter.records.get(1).status()); assertEquals("SUCCEEDED", exporter.records.get(2).status()); + assertEquals(1, exporter.flushes, "ON_CHANGE flushes once after the invocation-final record"); + } + + @Test + void onChangePreservesNoDelayBurstWithinQueueBound() { + var exporter = new CapturingExporter(); + DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight(WorkflowInsightConfig.builder() + .emitMode(WorkflowInsightConfig.EmitMode.ON_CHANGE) + .addExporter(exporter) + .build()); + + plugin.onInvocationStart(start(true)); + for (var i = 1; i <= 10; i++) { + var operations = ops("change-" + i, OperationStatus.SUCCEEDED); + plugin.onOperationChange(new OperationChangeInfo("req", ARN, operations, operations)); + } + plugin.onInvocationEnd(end(InvocationStatus.SUCCEEDED, "out", null)); + + assertEquals(12, exporter.records.size(), "start + ten changes + final are preserved without backpressure"); + assertTrue(exporter.records.subList(0, 11).stream().allMatch(r -> "RUNNING".equals(r.status()))); + assertEquals("SUCCEEDED", exporter.records.get(11).status()); + assertEquals(1, exporter.flushes); + } + + @Test + void onChangeCheckpointHookDoesNotWaitForExporterAndInvocationEndDrainsInOrder() throws Exception { + var exporter = new BlockingExporter(); + DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight(WorkflowInsightConfig.builder() + .emitMode(WorkflowInsightConfig.EmitMode.ON_CHANGE) + .addExporter(exporter) + .build()); + + plugin.onInvocationStart(start(true)); + assertTrue(exporter.firstExportStarted.await(5, TimeUnit.SECONDS), "initial export reached worker"); + + var change = new OperationChangeInfo( + "req", ARN, ops("greet", OperationStatus.SUCCEEDED), ops("greet", OperationStatus.SUCCEEDED)); + assertTimeoutPreemptively(Duration.ofSeconds(1), () -> plugin.onOperationChange(change)); + + var invocationEnd = + CompletableFuture.runAsync(() -> plugin.onInvocationEnd(end(InvocationStatus.SUCCEEDED, "out", null))); + assertFalse(invocationEnd.isDone(), "invocation end waits for the blocked export lane"); + + exporter.releaseFirstExport.countDown(); + invocationEnd.get(5, TimeUnit.SECONDS); + + assertEquals( + List.of("RUNNING", "RUNNING", "SUCCEEDED"), + exporter.records.stream().map(WorkflowInsightRecord::status).toList()); + assertEquals(1, exporter.flushes); } @Test