From 63d6d6fb1e0b4453d7370e60806c43f1180669c6 Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Wed, 26 Aug 2026 20:56:01 +0000 Subject: [PATCH 1/3] Fix checkpoint completion suspension race --- .../durable/execution/ExecutionManager.java | 17 +++--- .../operation/BaseDurableOperation.java | 21 +++++++- .../execution/ExecutionManagerTest.java | 52 +++++++++++++++++++ 3 files changed, 80 insertions(+), 10 deletions(-) diff --git a/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java b/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java index 1c45cb0d6..4150c81c8 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java @@ -139,7 +139,7 @@ public void registerOperation(BaseDurableOperation operation) { // ===== Checkpoint Completion Handler ===== /** Called by CheckpointManager when a checkpoint completes. Updates operationStorage and notify operations . */ - private void onCheckpointComplete(List newOperations) { + void onCheckpointComplete(List newOperations) { var updatedOperations = new ArrayList(); newOperations.forEach(op -> { // Detect a status change against the previously stored operation @@ -147,13 +147,14 @@ private void onCheckpointComplete(List newOperations) { if (previous == null || previous.status() != op.status()) { updatedOperations.add(op); } - // Update operation storage - operationStorage.put(op.id(), op); - // call registered operation's onCheckpointComplete method for completed operations - registeredOperations.computeIfPresent(op.id(), (id, operation) -> { - operation.onCheckpointComplete(op); - return operation; - }); + // Publish the updated state and notify its waiter atomically. Otherwise, a waiter can observe the terminal + // state before its completion future is completed and attempt to suspend with no pending operations. + var registeredOperation = registeredOperations.get(op.id()); + if (registeredOperation == null) { + operationStorage.put(op.id(), op); + } else { + registeredOperation.processCheckpointUpdate(op, () -> operationStorage.put(op.id(), op)); + } }); // Fire onOperationChange when a checkpoint response changed one or more operations diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java index 35a71f0da..ebc0d87bb 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java @@ -234,7 +234,7 @@ protected Operation waitForOperationCompletion() { // is between `isOperationCompleted` and `thenRun`. // If this operation is a branch/iteration of a ConcurrencyOperation (map or parallel), the branches/iterations // must be completed sequentially to avoid race conditions. - synchronized (parentOperation == null ? completionFuture : parentOperation.completionFuture) { + synchronized (completionLock()) { if (!isOperationCompleted()) { // Add a completion stage to completionFuture so that when the completionFuture is completed, // it will register the current Context thread synchronously to make sure it is always registered @@ -393,6 +393,19 @@ public void onCheckpointComplete(Operation operation) { } } + /** + * Publishes a checkpoint update and notifies this operation under the same lock used by operation waiters. + * + * @param operation the updated operation state + * @param publishUpdate publishes the operation to execution state + */ + public final void processCheckpointUpdate(Operation operation, Runnable publishUpdate) { + synchronized (completionLock()) { + publishUpdate.run(); + onCheckpointComplete(operation); + } + } + /** Marks the operation as already completed (in replay). */ protected void markAlreadyCompleted() { // When the operation is already completed in a replay, we complete completionFuture immediately @@ -404,7 +417,7 @@ protected void markAlreadyCompleted() { private void markCompletionFutureCompleted() { // It's important that we synchronize access to the future, otherwise the processing could happen // on someone else's thread and cause a race condition. - synchronized (parentOperation == null ? completionFuture : parentOperation.completionFuture) { + synchronized (completionLock()) { // Completing the future here will also run any other completion stages that have been attached // to the future. In our case, other contexts may have attached a function to reactivate themselves, // so they will definitely have a chance to reactivate before we finish completing and deactivating @@ -413,6 +426,10 @@ private void markCompletionFutureCompleted() { } } + private CompletableFuture completionLock() { + return parentOperation == null ? completionFuture : parentOperation.completionFuture; + } + /** * Terminates the execution with the given exception. * diff --git a/sdk/src/test/java/software/amazon/lambda/durable/execution/ExecutionManagerTest.java b/sdk/src/test/java/software/amazon/lambda/durable/execution/ExecutionManagerTest.java index dea26d90b..13529391b 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/execution/ExecutionManagerTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/execution/ExecutionManagerTest.java @@ -8,6 +8,10 @@ import static org.mockito.Mockito.when; import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import org.junit.jupiter.api.Test; import software.amazon.awssdk.services.lambda.model.CheckpointUpdatedExecutionState; import software.amazon.awssdk.services.lambda.model.GetDurableExecutionStateResponse; @@ -17,7 +21,11 @@ import software.amazon.lambda.durable.DurableConfig; import software.amazon.lambda.durable.TestUtils; import software.amazon.lambda.durable.client.DurableExecutionClient; +import software.amazon.lambda.durable.context.DurableContextImpl; import software.amazon.lambda.durable.model.DurableExecutionInput; +import software.amazon.lambda.durable.model.OperationIdentifier; +import software.amazon.lambda.durable.model.OperationSubType; +import software.amazon.lambda.durable.operation.BaseDurableOperation; class ExecutionManagerTest { private static final String EXECUTION_OP_ID = "01234567-0123-0123-0123-012345678901"; @@ -199,4 +207,48 @@ void isOperationUpdatedSinceLastInvocation_handlesMultipleIds() { assertTrue(manager.isOperationUpdatedSinceLastInvocation("3")); assertFalse(manager.isOperationUpdatedSinceLastInvocation("4")); } + + @Test + void checkpointStateIsNotPublishedBeforeOperationCompletion() throws Exception { + var manager = createManager(List.of(executionOp(), stepOp("step", OperationStatus.PENDING))); + var durableContext = mock(DurableContextImpl.class); + when(durableContext.getExecutionManager()).thenReturn(manager); + + class TestOperation extends BaseDurableOperation { + TestOperation() { + super(OperationIdentifier.of("step", "step", OperationSubType.STEP), durableContext, null); + } + + @Override + protected void start() {} + + @Override + protected void replay(Operation existing) {} + + CompletableFuture completionLock() { + return completionFuture; + } + } + + var operation = new TestOperation(); + var checkpointStarted = new CountDownLatch(1); + CompletableFuture checkpoint; + synchronized (operation.completionLock()) { + checkpoint = CompletableFuture.runAsync(() -> { + checkpointStarted.countDown(); + manager.onCheckpointComplete(List.of(stepOp("step", OperationStatus.SUCCEEDED))); + }); + assertTrue(checkpointStarted.await(5, TimeUnit.SECONDS)); + assertThrows(TimeoutException.class, () -> checkpoint.get(500, TimeUnit.MILLISECONDS)); + assertEquals( + OperationStatus.PENDING, + manager.getOperationAndUpdateReplayState("step").status()); + } + + checkpoint.get(5, TimeUnit.SECONDS); + assertTrue(operation.getCompletionFuture().isDone()); + assertEquals( + OperationStatus.SUCCEEDED, + manager.getOperationAndUpdateReplayState("step").status()); + } } From d1070f02e85a981e6c961a03e70b4b0575e2792e Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Wed, 26 Aug 2026 23:35:55 +0000 Subject: [PATCH 2/3] Fix checkpoint waiter synchronization races --- .../durable/execution/CheckpointManager.java | 24 +++++ .../durable/execution/ExecutionManager.java | 71 ++++++++++--- .../execution/CheckpointManagerTest.java | 17 ++++ .../execution/DurableExecutionTest.java | 99 +++++++++++++++++++ .../execution/ExecutionManagerTest.java | 64 ++++++++++++ 5 files changed, 260 insertions(+), 15 deletions(-) diff --git a/sdk/src/main/java/software/amazon/lambda/durable/execution/CheckpointManager.java b/sdk/src/main/java/software/amazon/lambda/durable/execution/CheckpointManager.java index c4f7af03e..011378256 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/execution/CheckpointManager.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/execution/CheckpointManager.java @@ -11,6 +11,7 @@ import java.util.Objects; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.function.BooleanSupplier; import java.util.function.Consumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -40,6 +41,8 @@ class CheckpointManager { private final Map>> pollingFutures = new ConcurrentHashMap<>(); private final ApiRequestDelayedBatcher checkpointApiRequestDelayedBatcher; private final DurableConfig config; + private final BooleanSupplier tryStartCheckpointProcessing; + private final Runnable finishCheckpointProcessing; private String checkpointToken; CheckpointManager( @@ -47,10 +50,22 @@ class CheckpointManager { String durableExecutionArn, String checkpointToken, Consumer> callback) { + this(config, durableExecutionArn, checkpointToken, callback, () -> true, () -> {}); + } + + CheckpointManager( + DurableConfig config, + String durableExecutionArn, + String checkpointToken, + Consumer> callback, + BooleanSupplier tryStartCheckpointProcessing, + Runnable finishCheckpointProcessing) { this.config = config; this.durableExecutionArn = durableExecutionArn; this.callback = callback; this.checkpointToken = checkpointToken; + this.tryStartCheckpointProcessing = tryStartCheckpointProcessing; + this.finishCheckpointProcessing = finishCheckpointProcessing; this.checkpointApiRequestDelayedBatcher = new ApiRequestDelayedBatcher<>( MAX_ITEM_COUNT, MAX_BATCH_SIZE_BYTES, CheckpointManager::estimateSize, this::checkpointBatch); } @@ -191,6 +206,13 @@ private void checkpointBatch(List updates) { return; } + // Starting the backend request is coordinated with the last-thread suspension decision. Once suspension + // wins that race, no later poll/checkpoint may advance backend state behind the PENDING response. + if (!tryStartCheckpointProcessing.getAsBoolean()) { + logger.debug("Skipping checkpoint API call because execution has already completed"); + return; + } + var startTime = System.nanoTime(); logger.debug("Calling durable checkpoint API with {} updates: {}", updates.size(), request); try { @@ -228,6 +250,8 @@ private void checkpointBatch(List updates) { } } catch (AwsServiceException e) { throw DurableApiErrorClassifier.classifyException(e); + } finally { + finishCheckpointProcessing.run(); } } } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java b/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java index 4150c81c8..53d80522a 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java @@ -69,6 +69,8 @@ public class ExecutionManager implements SafeCloseable { private final Set activeThreads = Collections.synchronizedSet(new HashSet<>()); private static final ThreadLocal currentThreadContext = new ThreadLocal<>(); private final CompletableFuture executionExceptionFuture = new CompletableFuture<>(); + // Guarded by activeThreads so starting a checkpoint request is atomic with the last-thread suspension decision. + private int checkpointRequestsInFlight; // ===== Checkpoint Batching ===== private final CheckpointManager checkpointManager; @@ -83,8 +85,13 @@ public ExecutionManager(DurableExecutionInput input, DurableConfig config, Conte input.updatedOperationIds() != null ? Set.copyOf(input.updatedOperationIds()) : Collections.emptySet(); // Create checkpoint batcher for internal coordination - this.checkpointManager = - new CheckpointManager(config, durableExecutionArn, input.checkpointToken(), this::onCheckpointComplete); + this.checkpointManager = new CheckpointManager( + config, + durableExecutionArn, + input.checkpointToken(), + this::onCheckpointComplete, + this::tryStartCheckpointProcessing, + this::finishCheckpointProcessing); this.operationStorage = checkpointManager.fetchAllPages(input.initialExecutionState()).stream() .collect(Collectors.toConcurrentMap(Operation::id, op -> op)); @@ -149,12 +156,14 @@ void onCheckpointComplete(List newOperations) { } // Publish the updated state and notify its waiter atomically. Otherwise, a waiter can observe the terminal // state before its completion future is completed and attempt to suspend with no pending operations. - var registeredOperation = registeredOperations.get(op.id()); - if (registeredOperation == null) { - operationStorage.put(op.id(), op); - } else { - registeredOperation.processCheckpointUpdate(op, () -> operationStorage.put(op.id(), op)); - } + registeredOperations.compute(op.id(), (id, registeredOperation) -> { + if (registeredOperation == null) { + operationStorage.put(op.id(), op); + } else { + registeredOperation.processCheckpointUpdate(op, () -> operationStorage.put(op.id(), op)); + } + return registeredOperation; + }); }); // Fire onOperationChange when a checkpoint response changed one or more operations @@ -250,14 +259,14 @@ public void registerActiveThread(String threadId) { * @param threadId the thread ID to deregister */ public void deregisterActiveThread(String threadId) { - // Skip if already suspended - if (executionExceptionFuture.isDone()) { - return; - } - // Add synchronized block to avoid remove then check race condition and make sure that // the suspendExecution is called only once synchronized (activeThreads) { + // Skip if already suspended + if (executionExceptionFuture.isDone()) { + return; + } + boolean removed = activeThreads.remove(threadId); if (removed) { logger.trace("Deregistered thread '{}' Active threads: {}", threadId, activeThreads.size()); @@ -265,7 +274,7 @@ public void deregisterActiveThread(String threadId) { logger.warn("Thread '{}' not active, cannot deregister", threadId); } - if (activeThreads.isEmpty()) { + if (shouldSuspendExecution()) { logger.info("No active threads remaining - suspending execution"); preSuspendCheck(); suspendExecution(); @@ -273,6 +282,34 @@ public void deregisterActiveThread(String threadId) { } } + boolean tryStartCheckpointProcessing() { + synchronized (activeThreads) { + if (executionExceptionFuture.isDone()) { + return false; + } + checkpointRequestsInFlight++; + return true; + } + } + + void finishCheckpointProcessing() { + synchronized (activeThreads) { + if (checkpointRequestsInFlight == 0) { + throw new IllegalStateException("No checkpoint request is in flight"); + } + checkpointRequestsInFlight--; + if (shouldSuspendExecution()) { + logger.info("Checkpoint processing completed with no active threads - suspending execution"); + preSuspendCheck(); + signalSuspension(); + } + } + } + + private boolean shouldSuspendExecution() { + return activeThreads.isEmpty() && checkpointRequestsInFlight == 0 && !executionExceptionFuture.isDone(); + } + private void preSuspendCheck() { var hasAnyPendingOperation = operationStorage.values().stream().anyMatch(o -> switch (o.type()) { case STEP -> o.status() == OperationStatus.PENDING; @@ -377,10 +414,14 @@ public void terminateExecution(UnrecoverableDurableExecutionException exception) /** Suspends the execution by completing the execution exception future with a {@link SuspendExecutionException}. */ public void suspendExecution() { + throw signalSuspension(); + } + + private SuspendExecutionException signalSuspension() { var ex = new SuspendExecutionException(); stopAllOperations(ex); executionExceptionFuture.completeExceptionally(ex); - throw ex; + return ex; } /** diff --git a/sdk/src/test/java/software/amazon/lambda/durable/execution/CheckpointManagerTest.java b/sdk/src/test/java/software/amazon/lambda/durable/execution/CheckpointManagerTest.java index 3c4885ddb..61f534e26 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/execution/CheckpointManagerTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/execution/CheckpointManagerTest.java @@ -75,6 +75,23 @@ void checkpoint_sendsUpdateAndReturnsCompletedFuture() throws Exception { assertTrue(future.isDone()); } + @Test + void checkpoint_skipsApiCallWhenExecutionAlreadyCompleted() throws Exception { + var finishCheckpointProcessing = mock(Runnable.class); + var guardedBatcher = new CheckpointManager( + config, "arn:test", "token-1", callbackOperations::addAll, () -> false, finishCheckpointProcessing); + var update = OperationUpdate.builder() + .id("op-1") + .type(OperationType.STEP) + .action(OperationAction.START) + .build(); + + guardedBatcher.checkpoint(update).get(200, TimeUnit.MILLISECONDS); + + verifyNoInteractions(client); + verify(finishCheckpointProcessing, never()).run(); + } + @Test void pollForUpdate_completesWhenOperationReturned() throws Exception { var operation = Operation.builder() diff --git a/sdk/src/test/java/software/amazon/lambda/durable/execution/DurableExecutionTest.java b/sdk/src/test/java/software/amazon/lambda/durable/execution/DurableExecutionTest.java index 5612194ac..23f677e05 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/execution/DurableExecutionTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/execution/DurableExecutionTest.java @@ -12,7 +12,11 @@ import java.time.Instant; import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Test; import software.amazon.awssdk.services.lambda.model.CheckpointUpdatedExecutionState; import software.amazon.awssdk.services.lambda.model.ErrorObject; @@ -23,10 +27,13 @@ import software.amazon.awssdk.services.lambda.model.StepDetails; import software.amazon.lambda.durable.DurableConfig; import software.amazon.lambda.durable.TestUtils; +import software.amazon.lambda.durable.context.DurableContextImpl; import software.amazon.lambda.durable.exception.UnrecoverableDurableExecutionException; import software.amazon.lambda.durable.model.DurableExecutionInput; import software.amazon.lambda.durable.model.ExecutionStatus; +import software.amazon.lambda.durable.model.OperationIdentifier; import software.amazon.lambda.durable.model.OperationSubType; +import software.amazon.lambda.durable.operation.BaseDurableOperation; class DurableExecutionTest { @@ -108,6 +115,86 @@ void testExecutePending() { assertNull(output.result()); } + @Test + void waiterFirstTerminalCheckpointReturnsSuccessfulOutput() { + var pendingStep = Operation.builder() + .id("step") + .name("step") + .type(OperationType.STEP) + .subType(OperationSubType.STEP.getValue()) + .status(OperationStatus.PENDING) + .build(); + var input = new DurableExecutionInput( + EXECUTION_ARN, + "token1", + CheckpointUpdatedExecutionState.builder() + .operations(List.of(executionOp(), pendingStep)) + .build()); + var checkpointAttempted = new CountDownLatch(1); + var checkpointFuture = new AtomicReference>(); + + var output = DurableExecutor.execute( + input, + null, + get(String.class), + (userInput, ctx) -> { + var durableContext = (DurableContextImpl) ctx; + var manager = durableContext.getExecutionManager(); + + class TestOperation extends BaseDurableOperation { + TestOperation() { + super(OperationIdentifier.of("step", "step", OperationSubType.STEP), durableContext, null); + } + + @Override + protected void start() {} + + @Override + protected void replay(Operation existing) {} + + Operation awaitCompletion() { + return waitForOperationCompletion(); + } + + @Override + protected void deregisterActiveThread(String threadId) { + checkpointFuture.set(CompletableFuture.runAsync(() -> { + checkpointAttempted.countDown(); + try { + manager.onCheckpointComplete(List.of(Operation.builder() + .id("step") + .name("step") + .type(OperationType.STEP) + .subType(OperationSubType.STEP.getValue()) + .status(OperationStatus.SUCCEEDED) + .build())); + } finally { + manager.finishCheckpointProcessing(); + } + })); + try { + assertTrue(checkpointAttempted.await(5, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + super.deregisterActiveThread(threadId); + } + } + + var operation = new TestOperation(); + operation.execute(); + assertTrue(manager.tryStartCheckpointProcessing()); + var completed = operation.awaitCompletion(); + checkpointFuture.get().join(); + return completed.statusAsString(); + }, + configWithMockClient()); + + assertEquals(ExecutionStatus.SUCCEEDED, output.status()); + assertTrue(output.result().contains(OperationStatus.SUCCEEDED.toString())); + } + @Test void testExecuteFailure() { var executionOp = Operation.builder() @@ -351,4 +438,16 @@ void testExecutorNotShutdownAfterMultipleHandlerInvocations() { assertTrue(output1.result().contains("Result 1: test-input-1")); assertTrue(output2.result().contains("Result 2: test-input-2")); } + + private Operation executionOp() { + return Operation.builder() + .id(EXECUTION_OP_ID) + .type(OperationType.EXECUTION) + .status(OperationStatus.STARTED) + .startTimestamp(EXECUTION_START_TIME) + .executionDetails(ExecutionDetails.builder() + .inputPayload("\"test-input\"") + .build()) + .build(); + } } diff --git a/sdk/src/test/java/software/amazon/lambda/durable/execution/ExecutionManagerTest.java b/sdk/src/test/java/software/amazon/lambda/durable/execution/ExecutionManagerTest.java index 13529391b..056c80e1a 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/execution/ExecutionManagerTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/execution/ExecutionManagerTest.java @@ -12,6 +12,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; import software.amazon.awssdk.services.lambda.model.CheckpointUpdatedExecutionState; import software.amazon.awssdk.services.lambda.model.GetDurableExecutionStateResponse; @@ -251,4 +252,67 @@ CompletableFuture completionLock() { OperationStatus.SUCCEEDED, manager.getOperationAndUpdateReplayState("step").status()); } + + @Test + void deferredSuspensionOccursWhenCheckpointFinishesWithoutReactivatingThread() { + var manager = createManager(List.of(executionOp(), stepOp("step", OperationStatus.PENDING))); + manager.registerActiveThread("root"); + assertTrue(manager.tryStartCheckpointProcessing()); + + manager.deregisterActiveThread("root"); + assertFalse(manager.isExecutionCompletedExceptionally()); + + manager.finishCheckpointProcessing(); + assertTrue(manager.isExecutionCompletedExceptionally()); + } + + @Test + void checkpointDeliveryIsAtomicWithOperationRegistration() throws Exception { + var manager = createManager(List.of(executionOp(), stepOp("step", OperationStatus.PENDING))); + var durableContext = mock(DurableContextImpl.class); + when(durableContext.getExecutionManager()).thenReturn(manager); + var publicationReached = new CountDownLatch(1); + var allowPublication = new CountDownLatch(1); + var idCalls = new AtomicInteger(); + var terminalOperation = mock(Operation.class); + when(terminalOperation.id()).thenAnswer(invocation -> { + if (idCalls.incrementAndGet() == 3) { + publicationReached.countDown(); + assertTrue(allowPublication.await(5, TimeUnit.SECONDS)); + } + return "step"; + }); + when(terminalOperation.name()).thenReturn("step"); + when(terminalOperation.type()).thenReturn(OperationType.STEP); + when(terminalOperation.subType()).thenReturn(OperationSubType.STEP.getValue()); + when(terminalOperation.status()).thenReturn(OperationStatus.SUCCEEDED); + + class TestOperation extends BaseDurableOperation { + TestOperation() { + super(OperationIdentifier.of("step", "step", OperationSubType.STEP), durableContext, null); + } + + @Override + protected void start() {} + + @Override + protected void replay(Operation existing) { + markAlreadyCompleted(); + } + } + + var checkpoint = CompletableFuture.runAsync(() -> manager.onCheckpointComplete(List.of(terminalOperation))); + assertTrue(publicationReached.await(5, TimeUnit.SECONDS)); + var registration = CompletableFuture.supplyAsync(TestOperation::new); + try { + assertThrows(TimeoutException.class, () -> registration.get(500, TimeUnit.MILLISECONDS)); + } finally { + allowPublication.countDown(); + } + + checkpoint.get(5, TimeUnit.SECONDS); + var operation = registration.get(5, TimeUnit.SECONDS); + operation.execute(); + assertTrue(operation.getCompletionFuture().isDone()); + } } From dd69599464a2a2576e9f47558deec1e1a2024564 Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Fri, 28 Aug 2026 19:08:59 +0000 Subject: [PATCH 3/3] Validate skipped checkpoint batches in E2E logs --- .github/workflows/e2e-tests.yml | 41 +++++++++++++++++++ .../durable/execution/CheckpointManager.java | 6 +++ 2 files changed, 47 insertions(+) diff --git a/.github/workflows/e2e-tests.yml b/.github/workflows/e2e-tests.yml index 8d9f9d4cd..2c176dcb2 100644 --- a/.github/workflows/e2e-tests.yml +++ b/.github/workflows/e2e-tests.yml @@ -89,6 +89,8 @@ jobs: --resolve-image-repos --resolve-s3 --parameter-overrides \ 'ParameterKey=Architecture,ParameterValue=x86_64 ParameterKey=JavaVersion,ParameterValue=java${{ matrix.java }} ParameterKey=FunctionNamePrefix,ParameterValue=Java${{ matrix.java }}- ParameterKey=RoleArn,ParameterValue=${{ secrets.TEST_LAMBDA_EXECUTION_ROLE_ARN }}' working-directory: ./examples + - name: Record E2E log start time + run: echo "E2E_LOG_START_TIME_MS=$(date +%s%3N)" >> "$GITHUB_ENV" - name: Cloud Based Integration Tests run: | mvn clean test -B \ @@ -102,6 +104,45 @@ jobs: -Djunit.jupiter.execution.parallel.config.strategy=fixed \ -Djunit.jupiter.execution.parallel.config.fixed.parallelism=${{ env.E2E_TEST_PARALLELISM }} working-directory: ./examples + - name: Check checkpoint invariant logs + env: + E2E_STACK_NAME: Java${{ matrix.java }}-JavaSDKCloudBasedIntegrationTestStack + run: | + set -euo pipefail + + # Allow the final Lambda invocation logs to reach CloudWatch before checking every + # managed function log group created by the E2E stack. + sleep 10 + + log_groups=$(aws cloudformation list-stack-resources \ + --stack-name "$E2E_STACK_NAME" \ + --query "StackResourceSummaries[?ResourceType=='AWS::Logs::LogGroup'].PhysicalResourceId" \ + --output text) + + if [[ -z "$log_groups" ]]; then + echo "::error::No managed Lambda log groups found for $E2E_STACK_NAME" + exit 1 + fi + + invariant_violated=false + for log_group in $log_groups; do + messages=$(aws logs filter-log-events \ + --log-group-name "$log_group" \ + --start-time "$E2E_LOG_START_TIME_MS" \ + --filter-pattern '"Checkpoint invariant violation"' \ + --query "events[].message" \ + --output text) + + if [[ -n "$messages" ]]; then + invariant_violated=true + echo "::error title=Checkpoint invariant violation::Found in $log_group" + echo "$messages" + fi + done + + if [[ "$invariant_violated" == true ]]; then + exit 1 + fi - name: Publish test case summary if: always() env: diff --git a/sdk/src/main/java/software/amazon/lambda/durable/execution/CheckpointManager.java b/sdk/src/main/java/software/amazon/lambda/durable/execution/CheckpointManager.java index 011378256..34f44134c 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/execution/CheckpointManager.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/execution/CheckpointManager.java @@ -209,6 +209,12 @@ private void checkpointBatch(List updates) { // Starting the backend request is coordinated with the last-thread suspension decision. Once suspension // wins that race, no later poll/checkpoint may advance backend state behind the PENDING response. if (!tryStartCheckpointProcessing.getAsBoolean()) { + if (!request.isEmpty()) { + logger.error( + "Checkpoint invariant violation: skipping {} operation updates because execution has already" + + " completed", + request.size()); + } logger.debug("Skipping checkpoint API call because execution has already completed"); return; }