From 1456d129abf037fe1a4e33db128959e1284a284d Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Mon, 10 Aug 2026 23:38:54 +0000 Subject: [PATCH 1/4] fix(plugin): surface missing state fields on hook info records Plugin hook infos dropped state that the Python and JS SDKs carry, so a Java plugin could not see per-operation replay status at the attempt and change hooks, nor operation state at the invocation hooks. Attempt and change-item fields: - UserFunctionStartInfo / UserFunctionEndInfo gain `isReplay`, the operation-level indicator for whether THIS operation was observed via checkpointed state. This is distinct from the existing `isReplayingChildren`, which describes the child operations of a context body and does not substitute for it. - OperationChangeItemInfo was a reduced record; it now carries the full operation surface (`status`, `attempt`, `isReplay`, `error`), ordered to match OperationEndInfo, so an operation seen through a change delta exposes the same fields as through the per-operation hooks. Invocation-info enrichment: - InvocationInfo gains `operations` and `updatedOperations`. - InvocationEndInfo gains `operations` and `executionStartTime`; the latter was present on the start info but dropped from the end record, forcing plugins to correlate back to the start hook. - `updatedOperations` derives from the input's UpdatedOperationIds intersected with the tracked operations, so it is empty on the first invocation and names the externally-completed operations on a replay. - The end-info snapshot is taken at end time, so unlike the start info it also includes operations created during the invocation. ExecutionManager now snapshots the operation ids delivered in the initial state and exposes non-mutating accessors for them. The attempt hook needs the replay indicator at its firing site, but getOperation() delegates to getOperationAndUpdateReplayState, which flips REPLAY to EXECUTION mode as a side effect; reading it from a plugin-hook site would mutate execution state. wasObservedAtInvocationStart is a pure containment check instead, and gives one consistent definition of `isReplay` across the attempt, change and invocation hooks. Both invocation maps are keyed by operation id and valued with OperationChangeItemInfo, now the richest operation snapshot record the SDK has, so a single conversion path feeds the change hook and both invocation hooks. The record name is a wart in this role; renaming it is left as a follow-up. These are positional records, so the added components surface every constructor call site. No defaulting overloads were introduced: a convenience constructor is exactly how a future internal call site would silently ship empty maps to plugins, which is the failure mode this change fixes. Call sites in the OpenTelemetry plugin tests were updated mechanically. Payload surfaces stay out of scope: no `result` and no execution input/result on any info. Verified with the full module build, unit tests and spotless:check. The conformance handlers that assert these field shapes, and the live suite results, are in the stacked follow-up PR. Refs: #604 --- .../durable/execution/DurableExecutor.java | 24 ++++++- .../durable/execution/ExecutionManager.java | 54 +++++++++++++- .../operation/BaseDurableOperation.java | 6 +- .../durable/plugin/InvocationEndInfo.java | 70 ++++++++++++------- .../lambda/durable/plugin/InvocationInfo.java | 46 ++++++------ .../plugin/OperationChangeItemInfo.java | 21 +++++- .../durable/plugin/PluginInfoConverter.java | 50 ++++++++++--- .../durable/plugin/UserFunctionEndInfo.java | 4 ++ .../durable/plugin/UserFunctionStartInfo.java | 4 ++ .../lambda/durable/DurableConfigTest.java | 4 +- .../plugin/PluginInfoConverterTest.java | 11 +-- 11 files changed, 224 insertions(+), 70 deletions(-) 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..8b2ae7a62 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 @@ -91,7 +91,13 @@ public static DurableExecutionOutput execute( executionArn, isFirstInvocation, executionManager.getExecutionOperation().startTimestamp(), - userInput)); + userInput, + PluginInfoConverter.toOperationItemMap( + executionManager.getOperationsSnapshot(), + executionManager.getInitialOperationIds()), + PluginInfoConverter.toOperationItemMap( + executionManager.getUpdatedOperationsSnapshot(), + executionManager.getInitialOperationIds()))); if (inputFailure != null) { ExceptionHelper.sneakyThrow(inputFailure); } @@ -120,6 +126,7 @@ public static DurableExecutionOutput execute( if (cause instanceof SuspendExecutionException) { fireOnInvocationEnd( pluginRunner, + executionManager, requestId, executionArn, isFirstInvocation, @@ -138,6 +145,7 @@ public static DurableExecutionOutput execute( && unrecoverableDurableExecutionException.isRetryable()) { fireOnInvocationEnd( pluginRunner, + executionManager, requestId, executionArn, isFirstInvocation, @@ -152,6 +160,7 @@ public static DurableExecutionOutput execute( logger.debug("Execution failed: {}", cause.getMessage()); fireOnInvocationEnd( pluginRunner, + executionManager, requestId, executionArn, isFirstInvocation, @@ -168,6 +177,7 @@ public static DurableExecutionOutput execute( DurableExecutionOutput.success(handleLargePayload(executionManager, outputPayload)); fireOnInvocationEnd( pluginRunner, + executionManager, requestId, executionArn, isFirstInvocation, @@ -188,6 +198,7 @@ public static DurableExecutionOutput execute( private static void fireOnInvocationEnd( PluginRunner pluginRunner, + ExecutionManager executionManager, String requestId, String executionArn, boolean isFirstInvocation, @@ -196,7 +207,16 @@ private static void fireOnInvocationEnd( Object executionInput, Object executionResult) { 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..a798105ee 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,11 @@ 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()), + durableContext.isReplaying(), + 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..8a90e21ef 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; /** @@ -17,42 +18,45 @@ * 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 operations a snapshot of the checkpointed operations delivered at the start of this invocation, keyed by + * operation ID. Includes the initial EXECUTION operation. Empty-but-never-null. + * @param updatedOperations the subset of {@code operations} that changed externally between the previous invocation and + * this one (a wait timer expired, a callback was received, a chained invoke completed), keyed by operation ID. + * Sourced from the {@code UpdatedOperationIds} field of the durable invocation input, so it is empty on the first + * invocation. Empty-but-never-null. */ public record InvocationInfo( String requestId, String durableExecutionArn, boolean isFirstInvocation, Instant executionStartTime, - @Experimental Object executionInput) { + @Experimental Object executionInput, + Map operations, + 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()); + } + + /** 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..18e494b02 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,17 @@ 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 already present in the checkpointed state when it started + * @param isReplayingChildren true if the child operations of this context body are replaying from checkpoints * @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, + boolean isReplayingChildren, + Integer attempt) { return new UserFunctionStartInfo( identifier.operationId(), identifier.name(), @@ -117,6 +124,7 @@ public static UserFunctionStartInfo toUserFunctionStartInfo( identifier.subType() != null ? identifier.subType().getValue() : null, parentId, Instant.now(), + isReplay, isReplayingChildren, attempt); } @@ -139,6 +147,7 @@ public static UserFunctionEndInfo toUserFunctionEndInfo( startInfo.parentId(), startInfo.startTimestamp(), Instant.now(), + startInfo.isReplay(), startInfo.isReplayingChildren(), startInfo.attempt(), outcome, @@ -153,25 +162,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 replayedOperationIds 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 replayedOperationIds) { 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, replayedOperationIds), + toOperationItemMap(allOperations, replayedOperationIds)); + } + + /** + * Converts durable operations to an unmodifiable map of {@link OperationChangeItemInfo}, keyed by operation ID. + * + * @param operations the durable operations to convert + * @param replayedOperationIds 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 replayedOperationIds) { + return operations.stream() + .collect(Collectors.toUnmodifiableMap( + Operation::id, + operation -> + toOperationChangeItemInfo(operation, replayedOperationIds.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 +205,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..9e0dd58ca 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,6 +17,9 @@ * @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 isReplay true if THIS operation was already present in the execution's checkpointed state when it started + * (i.e. observed via replay rather than created fresh in this invocation). Distinct from + * {@code isReplayingChildren}, which is about the child operations of a context body. * @param isReplayingChildren true if child operations within this context are being replayed from checkpoints * @param attempt 1-based attempt number for steps/waitForCondition, null for context operations * @param outcome the user function outcome @@ -30,6 +33,7 @@ public record UserFunctionEndInfo( String parentId, Instant startTimestamp, Instant endTimestamp, + boolean isReplay, boolean isReplayingChildren, Integer attempt, UserFunctionOutcome outcome, 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..4e956b16f 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,6 +16,9 @@ * @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 isReplay true if THIS operation was already present in the execution's checkpointed state when it started + * (i.e. observed via replay rather than created fresh in this invocation). Distinct from + * {@code isReplayingChildren}, which is about the child operations of a context body. * @param isReplayingChildren true if child operations within this context are being replayed from checkpoints * @param attempt 1-based attempt number for steps/waitForCondition, null for context operations */ @@ -26,5 +29,6 @@ public record UserFunctionStartInfo( String subType, String parentId, Instant startTimestamp, + boolean isReplay, boolean isReplayingChildren, 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..c90daa020 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 @@ -232,7 +232,7 @@ void toOperationEndInfo_resultIsNull_whenOperationHasNoResult() { @Test void toUserFunctionStartInfo_stepAttempt() { - var info = PluginInfoConverter.toUserFunctionStartInfo(STEP_IDENTIFIER, PARENT_ID, false, 3); + var info = PluginInfoConverter.toUserFunctionStartInfo(STEP_IDENTIFIER, PARENT_ID, true, false, 3); assertEquals(OPERATION_ID, info.id()); assertEquals(OPERATION_NAME, info.name()); @@ -240,16 +240,18 @@ void toUserFunctionStartInfo_stepAttempt() { assertEquals("Step", info.subType()); assertEquals(PARENT_ID, info.parentId()); assertNotNull(info.startTimestamp()); + assertTrue(info.isReplay()); assertFalse(info.isReplayingChildren()); 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, true, null); assertEquals("CONTEXT", info.type()); assertEquals("Map", info.subType()); + assertFalse(info.isReplay()); assertTrue(info.isReplayingChildren()); assertNull(info.attempt()); } @@ -258,7 +260,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, false, 1); var endInfo = PluginInfoConverter.toUserFunctionEndInfo(startInfo, UserFunctionOutcome.SUCCEEDED, null); @@ -266,6 +268,7 @@ void toUserFunctionEndInfo_succeeded() { assertEquals(OPERATION_NAME, endInfo.name()); assertEquals(startInfo.startTimestamp(), endInfo.startTimestamp()); assertNotNull(endInfo.endTimestamp()); + assertTrue(endInfo.isReplay()); assertFalse(endInfo.isReplayingChildren()); assertEquals(1, endInfo.attempt()); assertEquals(UserFunctionOutcome.SUCCEEDED, endInfo.outcome()); @@ -275,7 +278,7 @@ void toUserFunctionEndInfo_succeeded() { @Test void toUserFunctionEndInfo_failed() { var error = new RuntimeException("step failed"); - var startInfo = PluginInfoConverter.toUserFunctionStartInfo(STEP_IDENTIFIER, null, false, 2); + var startInfo = PluginInfoConverter.toUserFunctionStartInfo(STEP_IDENTIFIER, null, false, false, 2); var endInfo = PluginInfoConverter.toUserFunctionEndInfo(startInfo, UserFunctionOutcome.FAILED, error); From 78d9aa5a74b2a8b5b80b5e73fd4a7e22b68237e5 Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Mon, 24 Aug 2026 22:10:31 +0000 Subject: [PATCH 2/4] fix(plugin): complete mainline rebase integration --- .../durable/execution/DurableExecutor.java | 1 + .../lambda/durable/plugin/InvocationInfo.java | 29 ++++++--- .../durable/plugin/UserFunctionEndInfo.java | 31 ++++++++- .../durable/plugin/UserFunctionStartInfo.java | 16 ++++- .../plugin/PluginInfoConverterTest.java | 64 ++++++++++++++++++- .../durable/plugin/PluginRunnerTest.java | 3 +- 6 files changed, 132 insertions(+), 12 deletions(-) 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 8b2ae7a62..ff39fd3ab 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; 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 8a90e21ef..2196556b4 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 @@ -16,14 +16,9 @@ * @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 operations a snapshot of the checkpointed operations delivered at the start of this invocation, keyed by - * operation ID. Includes the initial EXECUTION operation. Empty-but-never-null. - * @param updatedOperations the subset of {@code operations} that changed externally between the previous invocation and - * this one (a wait timer expired, a callback was received, a chained invoke completed), keyed by operation ID. - * Sourced from the {@code UpdatedOperationIds} field of the durable invocation input, so it is empty on the first - * invocation. Empty-but-never-null. + * @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 + * @param updatedOperations operations changed externally since the previous invocation, keyed by operation ID */ public record InvocationInfo( String requestId, @@ -56,6 +51,24 @@ public InvocationInfo( 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() { 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 9e0dd58ca..9b6b91607 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 @@ -37,4 +37,33 @@ public record UserFunctionEndInfo( boolean isReplayingChildren, Integer attempt, UserFunctionOutcome outcome, - @Experimental Throwable error) {} + @Experimental Throwable error) { + + /** Creates user-function end information for an operation not marked as replayed. */ + public UserFunctionEndInfo( + String id, + String name, + String type, + String subType, + String parentId, + Instant startTimestamp, + Instant endTimestamp, + boolean isReplayingChildren, + Integer attempt, + UserFunctionOutcome outcome, + Throwable error) { + this( + id, + name, + type, + subType, + parentId, + startTimestamp, + endTimestamp, + false, + isReplayingChildren, + attempt, + outcome, + 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 4e956b16f..0b1247554 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 @@ -31,4 +31,18 @@ public record UserFunctionStartInfo( Instant startTimestamp, boolean isReplay, boolean isReplayingChildren, - Integer attempt) {} + Integer attempt) { + + /** Creates user-function start information for an operation not marked as replayed. */ + public UserFunctionStartInfo( + String id, + String name, + String type, + String subType, + String parentId, + Instant startTimestamp, + boolean isReplayingChildren, + Integer attempt) { + this(id, name, type, subType, parentId, startTimestamp, false, isReplayingChildren, attempt); + } +} 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 c90daa020..d82cf58a2 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,6 +230,66 @@ 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 @@ -290,7 +352,7 @@ void toUserFunctionEndInfo_failed() { @Test void toUserFunctionEndInfo_incomplete() { var error = new SuspendExecutionException(); - var startInfo = PluginInfoConverter.toUserFunctionStartInfo(MAP_IDENTIFIER, null, false, null); + var startInfo = PluginInfoConverter.toUserFunctionStartInfo(MAP_IDENTIFIER, null, false, false, null); var endInfo = PluginInfoConverter.toUserFunctionEndInfo(startInfo, UserFunctionOutcome.INCOMPLETE, error); diff --git a/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginRunnerTest.java b/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginRunnerTest.java index dc21c9e8b..82f432fa7 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginRunnerTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginRunnerTest.java @@ -238,7 +238,7 @@ private static OperationChangeInfo operationChangeInfo() { } private static UserFunctionStartInfo attemptInfo() { - return new UserFunctionStartInfo("op-1", "test-step", "STEP", null, null, Instant.now(), false, 1); + return new UserFunctionStartInfo("op-1", "test-step", "STEP", null, null, Instant.now(), false, false, 1); } private static UserFunctionEndInfo attemptEndInfo() { @@ -251,6 +251,7 @@ private static UserFunctionEndInfo attemptEndInfo() { Instant.now(), Instant.now(), false, + false, 1, UserFunctionOutcome.SUCCEEDED, null); From ecc58560473dc896f7934d0aee320391c73a6753 Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Mon, 24 Aug 2026 23:39:46 +0000 Subject: [PATCH 3/4] fix(plugin): harden invocation operation snapshots --- .../lambda/durable/PluginIntegrationTest.java | 62 +++++++++++++++++++ .../durable/execution/DurableExecutor.java | 29 +++++---- .../lambda/durable/plugin/InvocationInfo.java | 10 +-- 3 files changed, 85 insertions(+), 16 deletions(-) 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 ff39fd3ab..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 @@ -87,18 +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, - PluginInfoConverter.toOperationItemMap( - executionManager.getOperationsSnapshot(), - executionManager.getInitialOperationIds()), - PluginInfoConverter.toOperationItemMap( - executionManager.getUpdatedOperationsSnapshot(), - executionManager.getInitialOperationIds()))); + 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); } @@ -207,6 +209,9 @@ private static void fireOnInvocationEnd( Throwable error, Object executionInput, Object executionResult) { + if (pluginRunner.isEmpty()) { + return; + } pluginRunner.onInvocationEnd(new InvocationEndInfo( requestId, executionArn, 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 2196556b4..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 @@ -17,8 +17,10 @@ * @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 unavailable - * @param operations checkpointed operations delivered at invocation start, keyed by operation ID - * @param updatedOperations operations changed externally since the previous invocation, keyed by operation ID + * @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, @@ -26,8 +28,8 @@ public record InvocationInfo( boolean isFirstInvocation, Instant executionStartTime, @Experimental Object executionInput, - Map operations, - Map updatedOperations) { + @Experimental Map operations, + @Experimental Map updatedOperations) { public InvocationInfo { requireNonNull(executionStartTime, "executionStartTime"); From 726603271f06e136c1602bab0177420fef8cf6e6 Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Tue, 25 Aug 2026 22:04:38 +0000 Subject: [PATCH 4/4] fix(plugin): simplify replay metadata --- .../examples/general/PluginExample.java | 4 +- .../operation/BaseDurableOperation.java | 1 - .../durable/plugin/PluginInfoConverter.java | 25 +++++-------- .../durable/plugin/UserFunctionEndInfo.java | 37 +------------------ .../durable/plugin/UserFunctionStartInfo.java | 22 +---------- .../plugin/PluginInfoConverterTest.java | 13 +++---- .../durable/plugin/PluginRunnerTest.java | 3 +- 7 files changed, 21 insertions(+), 84 deletions(-) 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/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java index a798105ee..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 @@ -354,7 +354,6 @@ protected T runUserFunction(Integer attempt, Supplier userFunction) { operationIdentifier, durableContext.getParentId(), executionManager.wasObservedAtInvocationStart(getOperationId()), - durableContext.isReplaying(), attempt); pluginRunner.onUserFunctionStart(startInfo); try { 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 18e494b02..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 @@ -106,17 +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 this operation was already present in the checkpointed state when it started - * @param isReplayingChildren true if the child operations of this context body are replaying from checkpoints + * @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 isReplay, - boolean isReplayingChildren, - Integer attempt) { + OperationIdentifier identifier, String parentId, boolean isReplay, Integer attempt) { return new UserFunctionStartInfo( identifier.operationId(), identifier.name(), @@ -125,7 +120,6 @@ public static UserFunctionStartInfo toUserFunctionStartInfo( parentId, Instant.now(), isReplay, - isReplayingChildren, attempt); } @@ -148,7 +142,6 @@ public static UserFunctionEndInfo toUserFunctionEndInfo( startInfo.startTimestamp(), Instant.now(), startInfo.isReplay(), - startInfo.isReplayingChildren(), startInfo.attempt(), outcome, error); @@ -162,7 +155,7 @@ 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 replayedOperationIds ids of the operations delivered in this invocation's initial state, used to populate + * @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 */ @@ -171,29 +164,29 @@ public static OperationChangeInfo toOperationChangeInfo( String durableExecutionArn, Collection updatedOperations, Collection allOperations, - Set replayedOperationIds) { + Set initialOperationIds) { return new OperationChangeInfo( requestId, durableExecutionArn, - toOperationItemMap(updatedOperations, replayedOperationIds), - toOperationItemMap(allOperations, replayedOperationIds)); + 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 replayedOperationIds ids of the operations delivered in this invocation's initial state, used to populate + * @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 replayedOperationIds) { + Collection operations, Set initialOperationIds) { return operations.stream() .collect(Collectors.toUnmodifiableMap( Operation::id, operation -> - toOperationChangeItemInfo(operation, replayedOperationIds.contains(operation.id())))); + toOperationChangeItemInfo(operation, initialOperationIds.contains(operation.id())))); } private static OperationChangeItemInfo toOperationChangeItemInfo(Operation operation, boolean isReplay) { 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 9b6b91607..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,10 +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 isReplay true if THIS operation was already present in the execution's checkpointed state when it started - * (i.e. observed via replay rather than created fresh in this invocation). Distinct from - * {@code isReplayingChildren}, which is about the child operations of a context body. - * @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 @@ -34,36 +31,6 @@ public record UserFunctionEndInfo( Instant startTimestamp, Instant endTimestamp, boolean isReplay, - boolean isReplayingChildren, Integer attempt, UserFunctionOutcome outcome, - @Experimental Throwable error) { - - /** Creates user-function end information for an operation not marked as replayed. */ - public UserFunctionEndInfo( - String id, - String name, - String type, - String subType, - String parentId, - Instant startTimestamp, - Instant endTimestamp, - boolean isReplayingChildren, - Integer attempt, - UserFunctionOutcome outcome, - Throwable error) { - this( - id, - name, - type, - subType, - parentId, - startTimestamp, - endTimestamp, - false, - isReplayingChildren, - attempt, - outcome, - error); - } -} + @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 0b1247554..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,10 +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 isReplay true if THIS operation was already present in the execution's checkpointed state when it started - * (i.e. observed via replay rather than created fresh in this invocation). Distinct from - * {@code isReplayingChildren}, which is about the child operations of a context body. - * @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( @@ -30,19 +27,4 @@ public record UserFunctionStartInfo( String parentId, Instant startTimestamp, boolean isReplay, - boolean isReplayingChildren, - Integer attempt) { - - /** Creates user-function start information for an operation not marked as replayed. */ - public UserFunctionStartInfo( - String id, - String name, - String type, - String subType, - String parentId, - Instant startTimestamp, - boolean isReplayingChildren, - Integer attempt) { - this(id, name, type, subType, parentId, startTimestamp, false, isReplayingChildren, attempt); - } -} + Integer attempt) {} 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 d82cf58a2..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 @@ -294,7 +294,7 @@ void operationChangeItemInfo_toString_omitsResult() { @Test void toUserFunctionStartInfo_stepAttempt() { - var info = PluginInfoConverter.toUserFunctionStartInfo(STEP_IDENTIFIER, PARENT_ID, true, false, 3); + var info = PluginInfoConverter.toUserFunctionStartInfo(STEP_IDENTIFIER, PARENT_ID, true, 3); assertEquals(OPERATION_ID, info.id()); assertEquals(OPERATION_NAME, info.name()); @@ -303,18 +303,16 @@ void toUserFunctionStartInfo_stepAttempt() { assertEquals(PARENT_ID, info.parentId()); assertNotNull(info.startTimestamp()); assertTrue(info.isReplay()); - assertFalse(info.isReplayingChildren()); assertEquals(3, info.attempt()); } @Test void toUserFunctionStartInfo_contextOperation() { - var info = PluginInfoConverter.toUserFunctionStartInfo(MAP_IDENTIFIER, PARENT_ID, false, true, null); + var info = PluginInfoConverter.toUserFunctionStartInfo(MAP_IDENTIFIER, PARENT_ID, false, null); assertEquals("CONTEXT", info.type()); assertEquals("Map", info.subType()); assertFalse(info.isReplay()); - assertTrue(info.isReplayingChildren()); assertNull(info.attempt()); } @@ -322,7 +320,7 @@ void toUserFunctionStartInfo_contextOperation() { @Test void toUserFunctionEndInfo_succeeded() { - var startInfo = PluginInfoConverter.toUserFunctionStartInfo(STEP_IDENTIFIER, PARENT_ID, true, false, 1); + var startInfo = PluginInfoConverter.toUserFunctionStartInfo(STEP_IDENTIFIER, PARENT_ID, true, 1); var endInfo = PluginInfoConverter.toUserFunctionEndInfo(startInfo, UserFunctionOutcome.SUCCEEDED, null); @@ -331,7 +329,6 @@ void toUserFunctionEndInfo_succeeded() { assertEquals(startInfo.startTimestamp(), endInfo.startTimestamp()); assertNotNull(endInfo.endTimestamp()); assertTrue(endInfo.isReplay()); - assertFalse(endInfo.isReplayingChildren()); assertEquals(1, endInfo.attempt()); assertEquals(UserFunctionOutcome.SUCCEEDED, endInfo.outcome()); assertNull(endInfo.error()); @@ -340,7 +337,7 @@ void toUserFunctionEndInfo_succeeded() { @Test void toUserFunctionEndInfo_failed() { var error = new RuntimeException("step failed"); - var startInfo = PluginInfoConverter.toUserFunctionStartInfo(STEP_IDENTIFIER, null, false, false, 2); + var startInfo = PluginInfoConverter.toUserFunctionStartInfo(STEP_IDENTIFIER, null, false, 2); var endInfo = PluginInfoConverter.toUserFunctionEndInfo(startInfo, UserFunctionOutcome.FAILED, error); @@ -352,7 +349,7 @@ void toUserFunctionEndInfo_failed() { @Test void toUserFunctionEndInfo_incomplete() { var error = new SuspendExecutionException(); - var startInfo = PluginInfoConverter.toUserFunctionStartInfo(MAP_IDENTIFIER, null, false, false, null); + var startInfo = PluginInfoConverter.toUserFunctionStartInfo(MAP_IDENTIFIER, null, false, null); var endInfo = PluginInfoConverter.toUserFunctionEndInfo(startInfo, UserFunctionOutcome.INCOMPLETE, error); diff --git a/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginRunnerTest.java b/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginRunnerTest.java index 82f432fa7..dc21c9e8b 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginRunnerTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginRunnerTest.java @@ -238,7 +238,7 @@ private static OperationChangeInfo operationChangeInfo() { } private static UserFunctionStartInfo attemptInfo() { - return new UserFunctionStartInfo("op-1", "test-step", "STEP", null, null, Instant.now(), false, false, 1); + return new UserFunctionStartInfo("op-1", "test-step", "STEP", null, null, Instant.now(), false, 1); } private static UserFunctionEndInfo attemptEndInfo() { @@ -251,7 +251,6 @@ private static UserFunctionEndInfo attemptEndInfo() { Instant.now(), Instant.now(), false, - false, 1, UserFunctionOutcome.SUCCEEDED, null);