diff --git a/examples/src/main/java/software/amazon/lambda/durable/examples/general/PluginExample.java b/examples/src/main/java/software/amazon/lambda/durable/examples/general/PluginExample.java index 1de876701..28c6675d4 100644 --- a/examples/src/main/java/software/amazon/lambda/durable/examples/general/PluginExample.java +++ b/examples/src/main/java/software/amazon/lambda/durable/examples/general/PluginExample.java @@ -84,8 +84,8 @@ public void onOperationEnd(OperationEndInfo info) { @Override public void onUserFunctionStart(UserFunctionStartInfo info) { System.out.printf( - "[PLUGIN] onUserFunctionStart: name=%s, type=%s, attempt=%s, isReplayingChildren=%s%n", - info.name(), info.type(), info.attempt(), info.isReplayingChildren()); + "[PLUGIN] onUserFunctionStart: name=%s, type=%s, attempt=%s, isReplay=%s%n", + info.name(), info.type(), info.attempt(), info.isReplay()); } @Override diff --git a/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/PluginIntegrationTest.java b/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/PluginIntegrationTest.java index dfe99b9d7..f8e3bceb6 100644 --- a/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/PluginIntegrationTest.java +++ b/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/PluginIntegrationTest.java @@ -103,6 +103,68 @@ void plugin_receivesInvocationEnd_withPendingStatus_onSuspension() { assertEquals(InvocationStatus.PENDING, plugin.invocationEnds.get(0).invocationStatus()); } + @Test + void plugin_invocationSnapshots_trackReplayAcrossSuspension() { + var plugin = new RecordingPlugin(); + var config = DurableConfig.builder().withPlugins(plugin).build(); + + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> context.runInChildContext("child", String.class, child -> { + child.wait("pause", Duration.ofMinutes(1)); + return "done"; + }), + config); + + var firstResult = runner.run("input"); + assertEquals(ExecutionStatus.PENDING, firstResult.getStatus()); + + assertEquals(1, plugin.invocationStarts.size()); + assertEquals(1, plugin.invocationEnds.size()); + var firstStart = plugin.invocationStarts.get(0); + assertTrue(firstStart.isFirstInvocation()); + assertEquals(1, firstStart.operations().size(), "first invocation starts with only the execution operation"); + assertTrue(firstStart.updatedOperations().isEmpty()); + + var firstEnd = plugin.invocationEnds.get(0); + assertEquals(InvocationStatus.PENDING, firstEnd.invocationStatus()); + assertTrue(firstEnd.operations().values().stream().anyMatch(op -> "child".equals(op.name()))); + assertTrue(firstEnd.operations().values().stream().anyMatch(op -> "pause".equals(op.name()))); + + runner.advanceTime(); + var secondResult = runner.run("input"); + assertEquals(ExecutionStatus.SUCCEEDED, secondResult.getStatus()); + + assertEquals(2, plugin.invocationStarts.size()); + assertEquals(2, plugin.invocationEnds.size()); + var secondStart = plugin.invocationStarts.get(1); + assertFalse(secondStart.isFirstInvocation()); + var replayedChild = secondStart.operations().values().stream() + .filter(op -> "child".equals(op.name())) + .findFirst() + .orElseThrow(); + assertTrue(replayedChild.isReplay()); + assertEquals(1, secondStart.updatedOperations().size()); + var updatedPause = secondStart.updatedOperations().values().iterator().next(); + assertEquals("pause", updatedPause.name()); + assertEquals(OperationStatus.SUCCEEDED, updatedPause.status()); + assertTrue(updatedPause.isReplay()); + + var secondEnd = plugin.invocationEnds.get(1); + assertEquals(InvocationStatus.SUCCEEDED, secondEnd.invocationStatus()); + assertTrue(secondEnd.operations().values().stream() + .anyMatch(op -> "child".equals(op.name()) && op.status() == OperationStatus.SUCCEEDED)); + assertTrue(secondEnd.operations().values().stream() + .anyMatch(op -> "pause".equals(op.name()) && op.status() == OperationStatus.SUCCEEDED)); + + var childStarts = plugin.userFunctionStarts.stream() + .filter(info -> "child".equals(info.name())) + .toList(); + assertEquals(2, childStarts.size()); + assertFalse(childStarts.get(0).isReplay()); + assertTrue(childStarts.get(1).isReplay()); + } + @Test void plugin_receivesInvocationEnd_withFailedStatus_onError() { var plugin = new RecordingPlugin(); diff --git a/sdk/src/main/java/software/amazon/lambda/durable/execution/DurableExecutor.java b/sdk/src/main/java/software/amazon/lambda/durable/execution/DurableExecutor.java index 649c7a600..d8db91326 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/execution/DurableExecutor.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/execution/DurableExecutor.java @@ -29,6 +29,7 @@ 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.PluginInfoConverter; import software.amazon.lambda.durable.plugin.PluginRunner; import software.amazon.lambda.durable.serde.SerDes; import software.amazon.lambda.durable.util.ExceptionHelper; @@ -86,12 +87,20 @@ public static DurableExecutionOutput execute( // onInvocationStart runs on the user thread so plugins can // inject ThreadLocal objects, update MDC, etc. // executionStartTime comes from the initial EXECUTION operation in the first backend event. - pluginRunner.onInvocationStart(new InvocationInfo( - requestId, - executionArn, - isFirstInvocation, - executionManager.getExecutionOperation().startTimestamp(), - userInput)); + if (!pluginRunner.isEmpty()) { + pluginRunner.onInvocationStart(new InvocationInfo( + requestId, + executionArn, + isFirstInvocation, + executionManager.getExecutionOperation().startTimestamp(), + userInput, + PluginInfoConverter.toOperationItemMap( + executionManager.getOperationsSnapshot(), + executionManager.getInitialOperationIds()), + PluginInfoConverter.toOperationItemMap( + executionManager.getUpdatedOperationsSnapshot(), + executionManager.getInitialOperationIds()))); + } if (inputFailure != null) { ExceptionHelper.sneakyThrow(inputFailure); } @@ -120,6 +129,7 @@ public static DurableExecutionOutput execute( if (cause instanceof SuspendExecutionException) { fireOnInvocationEnd( pluginRunner, + executionManager, requestId, executionArn, isFirstInvocation, @@ -138,6 +148,7 @@ public static DurableExecutionOutput execute( && unrecoverableDurableExecutionException.isRetryable()) { fireOnInvocationEnd( pluginRunner, + executionManager, requestId, executionArn, isFirstInvocation, @@ -152,6 +163,7 @@ public static DurableExecutionOutput execute( logger.debug("Execution failed: {}", cause.getMessage()); fireOnInvocationEnd( pluginRunner, + executionManager, requestId, executionArn, isFirstInvocation, @@ -168,6 +180,7 @@ public static DurableExecutionOutput execute( DurableExecutionOutput.success(handleLargePayload(executionManager, outputPayload)); fireOnInvocationEnd( pluginRunner, + executionManager, requestId, executionArn, isFirstInvocation, @@ -188,6 +201,7 @@ public static DurableExecutionOutput execute( private static void fireOnInvocationEnd( PluginRunner pluginRunner, + ExecutionManager executionManager, String requestId, String executionArn, boolean isFirstInvocation, @@ -195,8 +209,20 @@ private static void fireOnInvocationEnd( Throwable error, Object executionInput, Object executionResult) { + if (pluginRunner.isEmpty()) { + return; + } pluginRunner.onInvocationEnd(new InvocationEndInfo( - requestId, executionArn, isFirstInvocation, status, error, executionInput, executionResult)); + requestId, + executionArn, + isFirstInvocation, + executionManager.getExecutionOperation().startTimestamp(), + PluginInfoConverter.toOperationItemMap( + executionManager.getOperationsSnapshot(), executionManager.getInitialOperationIds()), + status, + error, + executionInput, + executionResult)); } private static String handleLargePayload(ExecutionManager executionManager, String outputPayload) { 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..b52216d99 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 @@ -5,6 +5,7 @@ import com.amazonaws.services.lambda.runtime.Context; import java.time.Instant; import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; import java.util.HashSet; import java.util.List; @@ -63,6 +64,7 @@ public class ExecutionManager implements SafeCloseable { private final AtomicReference executionMode; private final DurableConfig durableConfig; private final Set updatedOperationIdsSinceLastInvocation; + private final Set initialOperationIds; // ===== Thread Coordination ===== private final Map registeredOperations = new ConcurrentHashMap<>(); @@ -89,6 +91,11 @@ public ExecutionManager(DurableExecutionInput input, DurableConfig config, Conte this.operationStorage = checkpointManager.fetchAllPages(input.initialExecutionState()).stream() .collect(Collectors.toConcurrentMap(Operation::id, op -> op)); + // The ids delivered in this invocation's initial state. Everything else in operationStorage is created during + // this invocation, so this set is what distinguishes replayed operations from freshly-started ones for the + // plugin hooks' isReplay indicators. + this.initialOperationIds = Set.copyOf(operationStorage.keySet()); + // Start in REPLAY mode if we have more than just the initial EXECUTION operation this.executionMode = new AtomicReference<>(operationStorage.size() > 1 ? ExecutionMode.REPLAY : ExecutionMode.EXECUTION); @@ -132,6 +139,47 @@ public boolean isOperationUpdatedSinceLastInvocation(String operationId) { return updatedOperationIdsSinceLastInvocation.contains(operationId); } + /** + * Returns {@code true} if the given operation was present in the checkpointed state delivered at the start of this + * invocation, i.e. it predates this invocation and is being replayed rather than started fresh. Unlike + * {@link #getOperationAndUpdateReplayState(String)} this does not mutate the execution's replay mode, so it is safe + * to call from plugin-hook firing sites. + * + * @param operationId the operation ID to check + * @return true if the operation was delivered in this invocation's initial state + */ + public boolean wasObservedAtInvocationStart(String operationId) { + return initialOperationIds.contains(operationId); + } + + /** Returns the ids of the operations delivered in this invocation's initial state. */ + public Set getInitialOperationIds() { + return initialOperationIds; + } + + /** + * Returns an immutable snapshot of the operations currently tracked for this execution, including the initial + * EXECUTION operation. Non-mutating; intended for the invocation-level plugin hooks. + * + * @return a snapshot of the tracked operations + */ + public Collection getOperationsSnapshot() { + return List.copyOf(operationStorage.values()); + } + + /** + * Returns the subset of {@link #getOperationsSnapshot()} whose ids the backend reported as updated since the last + * successful invocation. Empty on the first invocation. Ids without a corresponding tracked operation are skipped. + * + * @return a snapshot of the externally-updated operations + */ + public Collection getUpdatedOperationsSnapshot() { + return updatedOperationIdsSinceLastInvocation.stream() + .map(operationStorage::get) + .filter(Objects::nonNull) + .toList(); + } + /** Registers an operation so it can receive checkpoint completion notifications. */ public void registerOperation(BaseDurableOperation operation) { registeredOperations.put(operation.getOperationId(), operation); @@ -162,7 +210,11 @@ private void onCheckpointComplete(List newOperations) { durableConfig .getPluginRunner() .onOperationChange(PluginInfoConverter.toOperationChangeInfo( - requestId, durableExecutionArn, updatedOperations, operationStorage.values())); + requestId, + durableExecutionArn, + updatedOperations, + operationStorage.values(), + initialOperationIds)); } } 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..6b337db0c 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 @@ -351,7 +351,10 @@ protected void runUserHandler(Runnable runnable, ThreadType threadType) { protected T runUserFunction(Integer attempt, Supplier userFunction) { var pluginRunner = getPluginRunner(); var startInfo = PluginInfoConverter.toUserFunctionStartInfo( - operationIdentifier, durableContext.getParentId(), durableContext.isReplaying(), attempt); + operationIdentifier, + durableContext.getParentId(), + executionManager.wasObservedAtInvocationStart(getOperationId()), + attempt); pluginRunner.onUserFunctionStart(startInfo); try { T result = userFunction.get(); diff --git a/sdk/src/main/java/software/amazon/lambda/durable/plugin/InvocationEndInfo.java b/sdk/src/main/java/software/amazon/lambda/durable/plugin/InvocationEndInfo.java index c187700f6..145f48e65 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/plugin/InvocationEndInfo.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/plugin/InvocationEndInfo.java @@ -2,58 +2,78 @@ // SPDX-License-Identifier: Apache-2.0 package software.amazon.lambda.durable.plugin; +import java.time.Instant; +import java.util.Map; import software.amazon.lambda.durable.annotations.Experimental; /** * Information provided at the end of a Lambda invocation. * + *

Carries the same invocation-identity surface as {@link InvocationInfo} so an invocation-end hook does not have to + * correlate back to the start hook to learn the execution start time or operation state. + * * @param requestId the Lambda request ID for this invocation * @param durableExecutionArn the durable execution ARN * @param isFirstInvocation true if this is the first invocation of the execution - * @param invocationStatus the invocation outcome (SUCCEEDED, FAILED, or PENDING) + * @param executionStartTime the stable start timestamp of the durable execution + * @param operations a snapshot of operations known when the invocation ended, keyed by operation ID + * @param invocationStatus the invocation outcome * @param executionError non-null if the execution failed; this component is experimental - * @param executionInput the deserialized execution input passed to the user handler, or null when no plugins are - * registered or the input could not be deserialized; this component is experimental - * @param executionResult the value the user handler returned, or null unless the invocation completed the execution - * successfully; this component is experimental + * @param executionInput the deserialized execution input; this component is experimental + * @param executionResult the value returned by a successful handler; this component is experimental */ public record InvocationEndInfo( String requestId, String durableExecutionArn, boolean isFirstInvocation, + Instant executionStartTime, + Map operations, InvocationStatus invocationStatus, @Experimental Throwable executionError, @Experimental Object executionInput, @Experimental Object executionResult) { - /** - * Creates invocation-end information without the execution input or result. - * - *

Retained so callers written before {@code executionInput} and {@code executionResult} were added keep - * compiling; both resolve to null. - * - * @param requestId the Lambda request ID for this invocation - * @param durableExecutionArn the durable execution ARN - * @param isFirstInvocation true if this is the first invocation of the execution - * @param invocationStatus the invocation outcome - * @param executionError non-null if the execution failed - */ + /** Creates invocation-end information without execution payloads or operation state. */ public InvocationEndInfo( String requestId, String durableExecutionArn, boolean isFirstInvocation, InvocationStatus invocationStatus, Throwable executionError) { - this(requestId, durableExecutionArn, isFirstInvocation, invocationStatus, executionError, null, null); + this( + requestId, + durableExecutionArn, + isFirstInvocation, + null, + Map.of(), + invocationStatus, + executionError, + null, + null); + } + + /** Creates invocation-end information without execution start time or operation state. */ + public InvocationEndInfo( + String requestId, + String durableExecutionArn, + boolean isFirstInvocation, + InvocationStatus invocationStatus, + Throwable executionError, + Object executionInput, + Object executionResult) { + this( + requestId, + durableExecutionArn, + isFirstInvocation, + null, + Map.of(), + invocationStatus, + executionError, + executionInput, + executionResult); } - /** - * Returns a representation that omits {@code executionInput} and {@code executionResult}. - * - *

The generated representation would render both payloads, so plugins that log this object whole would start - * emitting customer inputs and results, potentially including secrets or personal data. Read the components - * explicitly to record them. - */ + /** Returns a representation that omits execution payloads and operation snapshots. */ @Override public String toString() { return "InvocationEndInfo[requestId=" + requestId + ", durableExecutionArn=" + durableExecutionArn diff --git a/sdk/src/main/java/software/amazon/lambda/durable/plugin/InvocationInfo.java b/sdk/src/main/java/software/amazon/lambda/durable/plugin/InvocationInfo.java index d786b12c7..d6c67195d 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/plugin/InvocationInfo.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/plugin/InvocationInfo.java @@ -5,6 +5,7 @@ import static java.util.Objects.requireNonNull; import java.time.Instant; +import java.util.Map; import software.amazon.lambda.durable.annotations.Experimental; /** @@ -15,44 +16,62 @@ * @param isFirstInvocation true if this is the first invocation of the execution (not a replay invocation) * @param executionStartTime the start timestamp of the durable execution, taken from the initial EXECUTION operation in * the first event delivered by the backend. Never null and stable across all invocations of the same execution. - * @param executionInput the deserialized execution input passed to the user handler, or null when no plugins are - * registered or the input could not be deserialized; this component is experimental + * @param executionInput the deserialized execution input passed to the user handler, or null when unavailable + * @param operations checkpointed operations delivered at invocation start, keyed by operation ID; this component is + * experimental + * @param updatedOperations operations changed externally since the previous invocation, keyed by operation ID; this + * component is experimental */ public record InvocationInfo( String requestId, String durableExecutionArn, boolean isFirstInvocation, Instant executionStartTime, - @Experimental Object executionInput) { + @Experimental Object executionInput, + @Experimental Map operations, + @Experimental Map updatedOperations) { public InvocationInfo { requireNonNull(executionStartTime, "executionStartTime"); + requireNonNull(operations, "operations"); + requireNonNull(updatedOperations, "updatedOperations"); } - /** - * Creates invocation information without an execution input. - * - *

Retained so callers written before {@code executionInput} was added keep compiling; {@code executionInput} - * resolves to null. - * - * @param requestId the Lambda request ID for this invocation - * @param durableExecutionArn the durable execution ARN - * @param isFirstInvocation true if this is the first invocation of the execution - * @param executionStartTime the start timestamp of the durable execution - * @throws NullPointerException if {@code executionStartTime} is null - */ + /** Creates invocation information without payload or operation snapshots. */ public InvocationInfo( String requestId, String durableExecutionArn, boolean isFirstInvocation, Instant executionStartTime) { - this(requestId, durableExecutionArn, isFirstInvocation, executionStartTime, null); + this(requestId, durableExecutionArn, isFirstInvocation, executionStartTime, null, Map.of(), Map.of()); } - /** - * Returns a representation that omits {@code executionInput}. - * - *

The generated representation would render the execution input, so plugins that log this object whole would - * start emitting customer payloads, potentially including secrets or personal data. Read the component explicitly - * to record it. - */ + /** Creates invocation information without operation snapshots. */ + public InvocationInfo( + String requestId, + String durableExecutionArn, + boolean isFirstInvocation, + Instant executionStartTime, + Object executionInput) { + this(requestId, durableExecutionArn, isFirstInvocation, executionStartTime, executionInput, Map.of(), Map.of()); + } + + /** Creates invocation information without an execution input. */ + public InvocationInfo( + String requestId, + String durableExecutionArn, + boolean isFirstInvocation, + Instant executionStartTime, + Map operations, + Map updatedOperations) { + this( + requestId, + durableExecutionArn, + isFirstInvocation, + executionStartTime, + null, + operations, + updatedOperations); + } + + /** Returns a representation that omits execution payloads and operation snapshots. */ @Override public String toString() { return "InvocationInfo[requestId=" + requestId + ", durableExecutionArn=" + durableExecutionArn diff --git a/sdk/src/main/java/software/amazon/lambda/durable/plugin/OperationChangeItemInfo.java b/sdk/src/main/java/software/amazon/lambda/durable/plugin/OperationChangeItemInfo.java index e3acafac5..2d5a0f9c5 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/plugin/OperationChangeItemInfo.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/plugin/OperationChangeItemInfo.java @@ -7,7 +7,7 @@ import software.amazon.lambda.durable.annotations.Experimental; /** - * Operation-level information for a single operation within an {@link OperationChangeInfo}. + * Operation-level information for an {@link OperationChangeInfo} and invocation operation snapshot. * * @param id operation ID * @param name human-readable operation name (may be null) @@ -16,8 +16,11 @@ * @param parentId parent operation ID (null for root-level operations) * @param startTimestamp when the operation started * @param endTimestamp when the operation ended - * @param error non-null if the operation failed; this component is experimental * @param status operation status + * @param attempt attempt number for retriable operations, null for others + * @param isReplay true if this operation was present in state delivered at invocation start + * @param error non-null if the operation failed; this component is experimental + * @param result checkpointed serialized result, or null if unavailable; this component is experimental */ public record OperationChangeItemInfo( String id, @@ -27,5 +30,17 @@ public record OperationChangeItemInfo( String parentId, Instant startTimestamp, Instant endTimestamp, + OperationStatus status, + Integer attempt, + boolean isReplay, @Experimental Throwable error, - OperationStatus status) {} + @Experimental String result) { + + /** Returns a representation that omits the operation result payload. */ + @Override + public String toString() { + return "OperationChangeItemInfo[id=" + id + ", name=" + name + ", type=" + type + ", subType=" + subType + + ", parentId=" + parentId + ", startTimestamp=" + startTimestamp + ", endTimestamp=" + endTimestamp + + ", status=" + status + ", attempt=" + attempt + ", isReplay=" + isReplay + ", error=" + error + "]"; + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginInfoConverter.java b/sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginInfoConverter.java index 131420f39..10b5b3905 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginInfoConverter.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginInfoConverter.java @@ -4,6 +4,8 @@ import java.time.Instant; import java.util.Collection; +import java.util.Map; +import java.util.Set; import java.util.stream.Collectors; import software.amazon.awssdk.services.lambda.model.Operation; import software.amazon.awssdk.services.lambda.model.OperationStatus; @@ -104,12 +106,12 @@ private static String extractResult(Operation operation) { * * @param identifier the operation identifier containing id, name, type, and subType * @param parentId the parent operation ID (may be null) - * @param isReplay true if the user function is called during replay (context operations) + * @param isReplay true if this operation was present in the checkpointed state delivered at invocation start * @param attempt the 1-based attempt number (null for context operations) * @return a UserFunctionStartInfo record */ public static UserFunctionStartInfo toUserFunctionStartInfo( - OperationIdentifier identifier, String parentId, boolean isReplayingChildren, Integer attempt) { + OperationIdentifier identifier, String parentId, boolean isReplay, Integer attempt) { return new UserFunctionStartInfo( identifier.operationId(), identifier.name(), @@ -117,7 +119,7 @@ public static UserFunctionStartInfo toUserFunctionStartInfo( identifier.subType() != null ? identifier.subType().getValue() : null, parentId, Instant.now(), - isReplayingChildren, + isReplay, attempt); } @@ -139,7 +141,7 @@ public static UserFunctionEndInfo toUserFunctionEndInfo( startInfo.parentId(), startInfo.startTimestamp(), Instant.now(), - startInfo.isReplayingChildren(), + startInfo.isReplay(), startInfo.attempt(), outcome, error); @@ -153,25 +155,41 @@ public static UserFunctionEndInfo toUserFunctionEndInfo( * @param durableExecutionArn the durable execution ARN * @param updatedOperations the durable operations whose status changed in this checkpoint response * @param allOperations all durable operations tracked for the execution after this response + * @param initialOperationIds ids of the operations delivered in this invocation's initial state, used to populate + * each item's {@code isReplay} indicator * @return an OperationChangeInfo record */ public static OperationChangeInfo toOperationChangeInfo( String requestId, String durableExecutionArn, Collection updatedOperations, - Collection allOperations) { + Collection allOperations, + Set initialOperationIds) { return new OperationChangeInfo( requestId, durableExecutionArn, - updatedOperations.stream() - .collect(Collectors.toUnmodifiableMap( - Operation::id, PluginInfoConverter::toOperationChangeItemInfo)), - allOperations.stream() - .collect(Collectors.toUnmodifiableMap( - Operation::id, PluginInfoConverter::toOperationChangeItemInfo))); + toOperationItemMap(updatedOperations, initialOperationIds), + toOperationItemMap(allOperations, initialOperationIds)); + } + + /** + * Converts durable operations to an unmodifiable map of {@link OperationChangeItemInfo}, keyed by operation ID. + * + * @param operations the durable operations to convert + * @param initialOperationIds ids of the operations delivered in this invocation's initial state, used to populate + * each item's {@code isReplay} indicator + * @return an unmodifiable map of operation ID to item info + */ + public static Map toOperationItemMap( + Collection operations, Set initialOperationIds) { + return operations.stream() + .collect(Collectors.toUnmodifiableMap( + Operation::id, + operation -> + toOperationChangeItemInfo(operation, initialOperationIds.contains(operation.id())))); } - private static OperationChangeItemInfo toOperationChangeItemInfo(Operation operation) { + private static OperationChangeItemInfo toOperationChangeItemInfo(Operation operation, boolean isReplay) { return new OperationChangeItemInfo( operation.id(), operation.name(), @@ -180,7 +198,10 @@ private static OperationChangeItemInfo toOperationChangeItemInfo(Operation opera operation.parentId(), operation.startTimestamp(), operation.endTimestamp(), + operation.status(), + operation.stepDetails() != null ? operation.stepDetails().attempt() : null, + isReplay, BaseDurableOperation.extractErrorFromOperation(operation), - operation.status()); + extractResult(operation)); } } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/plugin/UserFunctionEndInfo.java b/sdk/src/main/java/software/amazon/lambda/durable/plugin/UserFunctionEndInfo.java index cc8b41ac7..7eee99545 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/plugin/UserFunctionEndInfo.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/plugin/UserFunctionEndInfo.java @@ -17,7 +17,7 @@ * @param parentId parent operation ID (null for root-level operations) * @param startTimestamp when the user function started * @param endTimestamp when the user function ended - * @param isReplayingChildren true if child operations within this context are being replayed from checkpoints + * @param isReplay true if this operation was present in the checkpointed state delivered at invocation start * @param attempt 1-based attempt number for steps/waitForCondition, null for context operations * @param outcome the user function outcome * @param error non-null if the user function failed or exited incompletely; this component is experimental @@ -30,7 +30,7 @@ public record UserFunctionEndInfo( String parentId, Instant startTimestamp, Instant endTimestamp, - boolean isReplayingChildren, + boolean isReplay, Integer attempt, UserFunctionOutcome outcome, @Experimental Throwable error) {} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/plugin/UserFunctionStartInfo.java b/sdk/src/main/java/software/amazon/lambda/durable/plugin/UserFunctionStartInfo.java index 459e4ae4d..3e5f87842 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/plugin/UserFunctionStartInfo.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/plugin/UserFunctionStartInfo.java @@ -16,7 +16,7 @@ * @param subType operation sub-type (Map, Parallel, WaitForCondition, etc.) — may be null * @param parentId parent operation ID (null for root-level operations) * @param startTimestamp when the user function started - * @param isReplayingChildren true if child operations within this context are being replayed from checkpoints + * @param isReplay true if this operation was present in the checkpointed state delivered at invocation start * @param attempt 1-based attempt number for steps/waitForCondition, null for context operations */ public record UserFunctionStartInfo( @@ -26,5 +26,5 @@ public record UserFunctionStartInfo( String subType, String parentId, Instant startTimestamp, - boolean isReplayingChildren, + boolean isReplay, Integer attempt) {} diff --git a/sdk/src/test/java/software/amazon/lambda/durable/DurableConfigTest.java b/sdk/src/test/java/software/amazon/lambda/durable/DurableConfigTest.java index c0bd54147..266e43a6e 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/DurableConfigTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/DurableConfigTest.java @@ -547,7 +547,7 @@ void testBuilder_WithMultiplePlugins_AllRegistered() { config.getPluginRunner() .onInvocationStart(new software.amazon.lambda.durable.plugin.InvocationInfo( - "req-1", "arn:test", true, java.time.Instant.now())); + "req-1", "arn:test", true, java.time.Instant.now(), java.util.Map.of(), java.util.Map.of())); assertEquals(List.of("p1:onInvocationStart", "p2:onInvocationStart"), calls); } @@ -567,7 +567,7 @@ void testBuilder_WithPlugins_CalledMultipleTimes_Replaces() { config.getPluginRunner() .onInvocationStart(new software.amazon.lambda.durable.plugin.InvocationInfo( - "req-1", "arn:test", true, java.time.Instant.now())); + "req-1", "arn:test", true, java.time.Instant.now(), java.util.Map.of(), java.util.Map.of())); assertEquals(List.of("p2:onInvocationStart", "p3:onInvocationStart"), calls); } diff --git a/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginInfoConverterTest.java b/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginInfoConverterTest.java index 152ef0a4d..2f05f90c3 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginInfoConverterTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginInfoConverterTest.java @@ -5,6 +5,8 @@ import static org.junit.jupiter.api.Assertions.*; import java.time.Instant; +import java.util.List; +import java.util.Set; import org.junit.jupiter.api.Test; import software.amazon.awssdk.services.lambda.model.CallbackDetails; import software.amazon.awssdk.services.lambda.model.ChainedInvokeDetails; @@ -228,11 +230,71 @@ void toOperationEndInfo_resultIsNull_whenOperationHasNoResult() { assertNull(info.result()); } + // ─── toOperationItemMap ────────────────────────────────────────────── + + @Test + void toOperationItemMap_extractsResult_fromSucceededStep() { + var operation = Operation.builder() + .id(OPERATION_ID) + .name(OPERATION_NAME) + .type(OperationType.STEP) + .status(OperationStatus.SUCCEEDED) + .stepDetails(StepDetails.builder() + .attempt(2) + .result("{\"value\":42}") + .build()) + .build(); + + var info = PluginInfoConverter.toOperationItemMap(List.of(operation), Set.of()) + .get(OPERATION_ID); + + assertEquals("{\"value\":42}", info.result()); + assertEquals(2, info.attempt()); + } + + @Test + void toOperationItemMap_omitsResult_fromFailedStep() { + var operation = Operation.builder() + .id(OPERATION_ID) + .name(OPERATION_NAME) + .type(OperationType.STEP) + .status(OperationStatus.FAILED) + .stepDetails(StepDetails.builder() + .attempt(2) + .result("{\"intermediate\":true}") + .build()) + .build(); + + var info = PluginInfoConverter.toOperationItemMap(List.of(operation), Set.of()) + .get(OPERATION_ID); + + assertNull(info.result()); + } + + @Test + void operationChangeItemInfo_toString_omitsResult() { + var info = new OperationChangeItemInfo( + OPERATION_ID, + OPERATION_NAME, + "STEP", + "Step", + PARENT_ID, + START, + END, + OperationStatus.SUCCEEDED, + 1, + false, + null, + "s3cret-result"); + + assertFalse(info.toString().contains("s3cret-result"), "operation result must not leak into logs"); + } + // ─── toUserFunctionStartInfo ──────────────────────────────────────── @Test void toUserFunctionStartInfo_stepAttempt() { - var info = PluginInfoConverter.toUserFunctionStartInfo(STEP_IDENTIFIER, PARENT_ID, false, 3); + var info = PluginInfoConverter.toUserFunctionStartInfo(STEP_IDENTIFIER, PARENT_ID, true, 3); assertEquals(OPERATION_ID, info.id()); assertEquals(OPERATION_NAME, info.name()); @@ -240,17 +302,17 @@ void toUserFunctionStartInfo_stepAttempt() { assertEquals("Step", info.subType()); assertEquals(PARENT_ID, info.parentId()); assertNotNull(info.startTimestamp()); - assertFalse(info.isReplayingChildren()); + assertTrue(info.isReplay()); assertEquals(3, info.attempt()); } @Test void toUserFunctionStartInfo_contextOperation() { - var info = PluginInfoConverter.toUserFunctionStartInfo(MAP_IDENTIFIER, PARENT_ID, true, null); + var info = PluginInfoConverter.toUserFunctionStartInfo(MAP_IDENTIFIER, PARENT_ID, false, null); assertEquals("CONTEXT", info.type()); assertEquals("Map", info.subType()); - assertTrue(info.isReplayingChildren()); + assertFalse(info.isReplay()); assertNull(info.attempt()); } @@ -258,7 +320,7 @@ void toUserFunctionStartInfo_contextOperation() { @Test void toUserFunctionEndInfo_succeeded() { - var startInfo = PluginInfoConverter.toUserFunctionStartInfo(STEP_IDENTIFIER, PARENT_ID, false, 1); + var startInfo = PluginInfoConverter.toUserFunctionStartInfo(STEP_IDENTIFIER, PARENT_ID, true, 1); var endInfo = PluginInfoConverter.toUserFunctionEndInfo(startInfo, UserFunctionOutcome.SUCCEEDED, null); @@ -266,7 +328,7 @@ void toUserFunctionEndInfo_succeeded() { assertEquals(OPERATION_NAME, endInfo.name()); assertEquals(startInfo.startTimestamp(), endInfo.startTimestamp()); assertNotNull(endInfo.endTimestamp()); - assertFalse(endInfo.isReplayingChildren()); + assertTrue(endInfo.isReplay()); assertEquals(1, endInfo.attempt()); assertEquals(UserFunctionOutcome.SUCCEEDED, endInfo.outcome()); assertNull(endInfo.error());