From de0c4541da506fc419dacfac373b395a7f5f700c Mon Sep 17 00:00:00 2001 From: yaythomas Date: Fri, 16 Oct 2026 18:04:28 +0000 Subject: [PATCH] fix(sdk): skip plugin hook work without plugins When a checkpoint response changed an operation's status, onCheckpointComplete converted every stored operation for the operation change hook, even when no plugin was registered. So the cost of each checkpoint grew with the number of operations, and long sequential executions slowed down as they ran. onCheckpointComplete now builds the change info and calls onOperationChange only when a plugin is registered. Plugins receive the same information as before. Fixes #795 --- .../durable/execution/ExecutionManager.java | 16 ++-- .../execution/ExecutionManagerTest.java | 88 ++++++++++++++++++- 2 files changed, 92 insertions(+), 12 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 007df8665..1743aefaa 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 @@ -260,17 +260,13 @@ void onCheckpointComplete(List newOperations) { }); }); - // Fire onOperationChange when a checkpoint response changed one or more operations - if (!updatedOperations.isEmpty()) { + // Fire onOperationChange when a checkpoint response changed one or more operations. Only plugins read + // the change info, and building it converts every stored operation. So skip it when no plugin is registered. + var pluginRunner = durableConfig.getPluginRunner(); + if (!updatedOperations.isEmpty() && !pluginRunner.isEmpty()) { var requestId = lambdaContext != null ? lambdaContext.getAwsRequestId() : null; - durableConfig - .getPluginRunner() - .onOperationChange(PluginInfoConverter.toOperationChangeInfo( - requestId, - durableExecutionArn, - updatedOperations, - operationStorage.values(), - initialOperationIds)); + pluginRunner.onOperationChange(PluginInfoConverter.toOperationChangeInfo( + requestId, durableExecutionArn, updatedOperations, operationStorage.values(), initialOperationIds)); } } 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 7bb872911..52e93353f 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 @@ -4,7 +4,10 @@ import static org.junit.jupiter.api.Assertions.*; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.CALLS_REAL_METHODS; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.when; import java.time.Instant; @@ -12,6 +15,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.ConcurrentHashMap; @@ -24,6 +28,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; @@ -46,6 +51,9 @@ import software.amazon.lambda.durable.model.OperationIdentifier; import software.amazon.lambda.durable.model.OperationSubType; import software.amazon.lambda.durable.operation.BaseDurableOperation; +import software.amazon.lambda.durable.plugin.DurableExecutionPlugin; +import software.amazon.lambda.durable.plugin.OperationChangeInfo; +import software.amazon.lambda.durable.plugin.PluginInfoConverter; class ExecutionManagerTest { private static final String EXECUTION_OP_ID = "01234567-0123-0123-0123-012345678901"; @@ -55,13 +63,16 @@ class ExecutionManagerTest { private DurableExecutionClient client; - private ExecutionManager createManager(List operations) { + private ExecutionManager createManager(List operations, DurableExecutionPlugin... plugins) { client = TestUtils.createMockClient(); var initialState = CheckpointUpdatedExecutionState.builder().operations(operations).build(); return new ExecutionManager( new DurableExecutionInput(EXECUTION_ARN, "test-token", initialState), - DurableConfig.builder().withDurableExecutionClient(client).build(), + DurableConfig.builder() + .withDurableExecutionClient(client) + .withPlugins(plugins) + .build(), null); } @@ -506,6 +517,79 @@ protected void replay(Operation existing) {} manager.getOperationAndUpdateReplayState("step").status()); } + @Test + void checkpointWithoutPluginsSkipsOperationChangeInfo() { + try (var manager = createManager(List.of(executionOp(), stepOp("step", OperationStatus.PENDING)))) { + var operation = registerStepOperation(manager); + try (var converter = mockStatic(PluginInfoConverter.class, CALLS_REAL_METHODS)) { + manager.onCheckpointComplete( + List.of(stepOp("step", OperationStatus.SUCCEEDED), stepOp("other", OperationStatus.SUCCEEDED))); + + converter.verify( + () -> PluginInfoConverter.toOperationChangeInfo(any(), any(), any(), any(), any()), never()); + } + + // The checkpoint still stores every operation and completes the waiting operation. + assertTrue(operation.getCompletionFuture().isDone()); + assertEquals( + Map.of( + EXECUTION_OP_ID, + OperationStatus.STARTED, + "step", + OperationStatus.SUCCEEDED, + "other", + OperationStatus.SUCCEEDED), + statusesById(manager)); + } + } + + @Test + void checkpointWithPluginReportsEachStatusChangeOnce() { + var changes = new ArrayList(); + var plugin = new DurableExecutionPlugin() { + @Override + public void onOperationChange(OperationChangeInfo info) { + changes.add(info); + } + }; + try (var manager = createManager(List.of(executionOp(), stepOp("step", OperationStatus.PENDING)), plugin)) { + var response = List.of(stepOp("step", OperationStatus.SUCCEEDED), stepOp("other", OperationStatus.STARTED)); + manager.onCheckpointComplete(response); + // The same response again changes no status, so it reports no change. + manager.onCheckpointComplete(response); + + assertEquals(1, changes.size()); + var change = changes.get(0); + assertEquals(Set.of("step", "other"), change.updatedOperations().keySet()); + assertEquals( + OperationStatus.SUCCEEDED, + change.updatedOperations().get("step").status()); + assertEquals( + Set.of(EXECUTION_OP_ID, "step", "other"), + change.operations().keySet()); + assertEquals( + OperationStatus.STARTED, change.operations().get("other").status()); + } + } + + /** Registers a step operation with the manager, so a checkpoint that reports "step" notifies it. */ + private BaseDurableOperation registerStepOperation(ExecutionManager manager) { + var durableContext = mock(DurableContextImpl.class); + when(durableContext.getExecutionManager()).thenReturn(manager); + return new BaseDurableOperation( + OperationIdentifier.of("step", "step", OperationSubType.STEP), durableContext, null) { + @Override + protected void start() {} + + @Override + protected void replay(Operation existing) {} + }; + } + + private static Map statusesById(ExecutionManager manager) { + return manager.getOperationsSnapshot().stream().collect(Collectors.toMap(Operation::id, Operation::status)); + } + @Test void operationProcessCheckpointUpdateDelegatesThroughManagerLock() throws Exception { var manager = createManager(List.of(executionOp(), stepOp("step", OperationStatus.PENDING)));