withSerDes(SerDes serDes) {
return new CloudDurableTestRunner<>(
- functionArn, inputType, outputType, lambdaClient, pollInterval, timeout, invocationType, serDes);
+ functionArn,
+ inputType,
+ outputType,
+ lambdaClient,
+ pollInterval,
+ timeout,
+ invocationType,
+ inputSerDesOverride,
+ serDes);
+ }
+
+ /**
+ * Returns a new runner with a separate value codec for the initial Lambda invocation payload.
+ *
+ * The input codec is independent of the SerDes used for persisted execution payloads and must not be a
+ * {@link ComposableSerDes}. By default, the configured persisted SerDes is used when it is a value codec; for a
+ * composable pipeline, its root value codec is used. The deployed handler must configure the same codec with
+ * {@link software.amazon.lambda.durable.DurableConfig.Builder#withInputSerDes(SerDes)}.
+ */
+ public CloudDurableTestRunner withInputSerDes(SerDes inputSerDes) {
+ return new CloudDurableTestRunner<>(
+ functionArn,
+ inputType,
+ outputType,
+ lambdaClient,
+ pollInterval,
+ timeout,
+ invocationType,
+ Objects.requireNonNull(inputSerDes, "inputSerDes cannot be null"),
+ serDes);
}
/** Invokes the Lambda function, polls execution history until completion, and returns the result. */
@@ -138,7 +209,7 @@ public TestResult runUntilComplete(I input) {
public TestResult run(I input) {
try {
// Serialize input
- var inputJson = serDes.serialize(input);
+ var inputJson = serializeInput(input);
// Invoke function
var invokeRequest = InvokeRequest.builder()
@@ -161,7 +232,7 @@ public TestResult run(I input) {
// Process events into TestResult
var processor = new HistoryEventProcessor();
- var result = processor.processEvents(events, outputType, serDes);
+ var result = processor.processEvents(events, outputType, serDes, new SerDesRunner(null), executionArn);
this.lastResult = result;
return result;
} catch (Exception e) {
@@ -179,7 +250,7 @@ public TestResult run(I input) {
public AsyncExecution startAsync(I input) {
try {
// Serialize input
- var inputJson = serDes.serialize(input);
+ var inputJson = serializeInput(input);
// Invoke function with EVENT type (async)
var invokeRequest = InvokeRequest.builder()
@@ -216,4 +287,21 @@ public TestOperation getOperation(String name) {
}
return lastResult.getOperation(name);
}
+
+ private String serializeInput(I input) {
+ return inputSerDes.serialize(input);
+ }
+
+ private static SerDes inputValueCodec(SerDes inputSerDes, SerDes persistedSerDes) {
+ var codec = inputSerDes;
+ if (codec == null) {
+ codec = persistedSerDes instanceof ComposableSerDes composable
+ ? composable.getValueCodec()
+ : persistedSerDes;
+ }
+ if (codec instanceof ComposableSerDes) {
+ throw new IllegalArgumentException("inputSerDes must be a value codec, not a composable pipeline");
+ }
+ return codec;
+ }
}
diff --git a/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/LocalDurableTestRunner.java b/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/LocalDurableTestRunner.java
index f04e1c27b..a7d3196eb 100644
--- a/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/LocalDurableTestRunner.java
+++ b/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/LocalDurableTestRunner.java
@@ -6,6 +6,7 @@
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
+import java.util.Objects;
import java.util.UUID;
import java.util.function.BiFunction;
import software.amazon.awssdk.services.lambda.model.CheckpointUpdatedExecutionState;
@@ -23,6 +24,7 @@
import software.amazon.lambda.durable.model.ExecutionStatus;
import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
import software.amazon.lambda.durable.serde.SerDes;
+import software.amazon.lambda.durable.serde.SerDesRunner;
import software.amazon.lambda.durable.testing.local.LocalMemoryExecutionClient;
import software.amazon.lambda.durable.testing.local.OperationResult;
@@ -40,24 +42,33 @@ public class LocalDurableTestRunner {
private final TypeToken outputType;
private final BiFunction handler;
private final LocalMemoryExecutionClient storage;
+ private final SerDes inputSerDes;
+ private final SerDes inputSerDesOverride;
private final SerDes serDes;
private final DurableConfig customerConfig;
private final Instant executionStartTime = Instant.now();
+ private final String executionName = UUID.randomUUID().toString();
+ private final String invocationId = UUID.randomUUID().toString();
+ private final String executionArn = String.format(
+ "arn:aws:lambda:us-east-1:123456789012:function:test:$LATEST/durable-execution/%s/%s",
+ executionName, invocationId);
private LocalDurableTestRunner(
TypeToken inputType,
TypeToken outputType,
BiFunction handlerFn,
- DurableConfig customerConfig) {
+ DurableConfig customerConfig,
+ SerDes inputSerDes) {
this.inputType = inputType;
this.outputType = outputType;
this.handler = handlerFn;
+ this.inputSerDesOverride = inputSerDes;
this.storage = new LocalMemoryExecutionClient();
// Create config that uses customer's configuration but overrides the client with in-memory storage
if (customerConfig != null) {
// Use customer's config but override the client with our in-memory implementation
- this.customerConfig = DurableConfig.builder()
+ var configBuilder = DurableConfig.builder()
.withDurableExecutionClient(storage)
.withSerDes(customerConfig.getSerDes())
.withExecutorService(customerConfig.getExecutorService())
@@ -66,14 +77,23 @@ private LocalDurableTestRunner(
.withLoggerConfig(customerConfig.getLoggerConfig())
// Temporary: remove along with the checkpointEmptyMap flag in a future major version.
.withCheckpointEmptyMap(customerConfig.shouldCheckpointEmptyMap())
- .withPlugins(customerConfig.getPluginRunner().getPlugins().toArray(new DurableExecutionPlugin[0]))
- .build();
+ .withDeserializeAfterSerialization(customerConfig.shouldDeserializeAfterSerialization())
+ .withPlugins(customerConfig.getPluginRunner().getPlugins().toArray(new DurableExecutionPlugin[0]));
+ configBuilder.withInputSerDes(inputSerDes != null ? inputSerDes : customerConfig.getInputSerDes());
+ if (customerConfig.getSerDesExecutorService() != null) {
+ configBuilder.withSerDesExecutorService(customerConfig.getSerDesExecutorService());
+ }
+ this.customerConfig = configBuilder.build();
} else {
// Fallback to default config with in-memory client
- this.customerConfig =
- DurableConfig.builder().withDurableExecutionClient(storage).build();
+ var configBuilder = DurableConfig.builder().withDurableExecutionClient(storage);
+ if (inputSerDes != null) {
+ configBuilder.withInputSerDes(inputSerDes);
+ }
+ this.customerConfig = configBuilder.build();
}
this.serDes = this.customerConfig.getSerDes();
+ this.inputSerDes = this.customerConfig.getInputSerDes();
}
/**
@@ -88,7 +108,7 @@ private LocalDurableTestRunner(
*/
public static LocalDurableTestRunner create(
Class inputType, BiFunction handlerFn) {
- return new LocalDurableTestRunner<>(TypeToken.get(inputType), null, handlerFn, null);
+ return new LocalDurableTestRunner<>(TypeToken.get(inputType), null, handlerFn, null, null);
}
/**
@@ -109,7 +129,7 @@ public static LocalDurableTestRunner create(
*/
public static LocalDurableTestRunner create(
TypeToken inputType, BiFunction handlerFn) {
- return new LocalDurableTestRunner<>(inputType, null, handlerFn, null);
+ return new LocalDurableTestRunner<>(inputType, null, handlerFn, null, null);
}
/**
@@ -125,7 +145,7 @@ public static LocalDurableTestRunner create(
*/
public static LocalDurableTestRunner create(
Class inputType, BiFunction handlerFn, DurableConfig config) {
- return new LocalDurableTestRunner<>(TypeToken.get(inputType), null, handlerFn, config);
+ return new LocalDurableTestRunner<>(TypeToken.get(inputType), null, handlerFn, config, null);
}
/**
* Creates a LocalDurableTestRunner that uses a custom configuration. This allows the test runner to use custom
@@ -163,7 +183,7 @@ public static LocalDurableTestRunner create(
*/
public static LocalDurableTestRunner create(
TypeToken inputType, BiFunction handlerFn, DurableConfig config) {
- return new LocalDurableTestRunner<>(inputType, null, handlerFn, config);
+ return new LocalDurableTestRunner<>(inputType, null, handlerFn, config, null);
}
/**
@@ -179,7 +199,7 @@ public static LocalDurableTestRunner create(
*/
public static LocalDurableTestRunner create(Class inputType, DurableHandler handler) {
return new LocalDurableTestRunner<>(
- TypeToken.get(inputType), null, handler::handleRequest, handler.getConfiguration());
+ TypeToken.get(inputType), null, handler::handleRequest, handler.getConfiguration(), null);
}
/**
@@ -187,17 +207,33 @@ public static LocalDurableTestRunner create(Class inputType, Dur
* a new runner instance.
*/
public LocalDurableTestRunner withDurableConfig(DurableConfig config) {
- return new LocalDurableTestRunner<>(inputType, outputType, handler, config);
+ return new LocalDurableTestRunner<>(inputType, outputType, handler, config, inputSerDesOverride);
}
/** Overrides the output type for this test runner. */
public LocalDurableTestRunner withOutputType(TypeToken outputType) {
- return new LocalDurableTestRunner<>(inputType, outputType, handler, customerConfig);
+ return new LocalDurableTestRunner<>(inputType, outputType, handler, customerConfig, inputSerDesOverride);
}
/** Overrides the output type for this test runner. */
public LocalDurableTestRunner withOutputType(Class outputType) {
- return new LocalDurableTestRunner<>(inputType, TypeToken.get(outputType), handler, customerConfig);
+ return new LocalDurableTestRunner<>(
+ inputType, TypeToken.get(outputType), handler, customerConfig, inputSerDesOverride);
+ }
+
+ /**
+ * Returns a new runner with a separate value codec for the initial Lambda invocation payload.
+ *
+ * The returned runner uses this codec both to serialize the external payload and to configure
+ * {@link DurableExecutor} to deserialize it. Persisted pipeline stages are not invoked at this boundary.
+ */
+ public LocalDurableTestRunner withInputSerDes(SerDes inputSerDes) {
+ return new LocalDurableTestRunner<>(
+ inputType,
+ outputType,
+ handler,
+ customerConfig,
+ Objects.requireNonNull(inputSerDes, "inputSerDes cannot be null"));
}
/**
@@ -233,16 +269,18 @@ public LocalDurableTestRunner withOutputType(Class outputType) {
* @return LocalDurableTestRunner configured with the handler's settings
*/
public static LocalDurableTestRunner create(TypeToken inputType, DurableHandler handler) {
- return new LocalDurableTestRunner<>(inputType, null, handler::handleRequest, handler.getConfiguration());
+ return new LocalDurableTestRunner<>(inputType, null, handler::handleRequest, handler.getConfiguration(), null);
}
/** Run a single invocation (may return PENDING if waiting/retrying). */
public TestResult run(I input) {
+ var serDesRunner = new SerDesRunner(customerConfig.getSerDesExecutorService());
var durableInput = createDurableInput(input);
var output = DurableExecutor.execute(durableInput, mockLambdaContext(), inputType, handler, customerConfig);
- return storage.toTestResult(output, outputType, serDes);
+ return storage.toTestResult(
+ output, outputType, serDes, serDesRunner, executionArn, invocationId, executionName);
}
/**
@@ -281,7 +319,14 @@ public void simulateFireAndForgetCheckpointLoss(String stepName) {
/** Returns the {@link TestOperation} for the given operation name, or null if not found. */
public TestOperation getOperation(String name) {
var op = storage.getOperationByName(name);
- return op != null ? new TestOperation(op, serDes) : null;
+ return op != null
+ ? new TestOperation(
+ op,
+ List.of(),
+ serDes,
+ new SerDesRunner(customerConfig.getSerDesExecutorService()),
+ executionArn)
+ : null;
}
/** Get callback ID for a named callback operation. */
@@ -330,12 +375,7 @@ public void stopChainedInvoke(String name, ErrorObject error) {
}
private DurableExecutionInput createDurableInput(I input) {
- var executionName = UUID.randomUUID().toString();
- var invocationId = UUID.randomUUID().toString();
- var executionArn = String.format(
- "arn:aws:lambda:us-east-1:123456789012:function:test:$LATEST/durable-execution/%s/%s",
- executionName, invocationId);
- var inputJson = serDes.serialize(input);
+ var inputJson = serializeInput(input);
var executionOp = Operation.builder()
.id(invocationId)
.name(executionName)
@@ -347,7 +387,9 @@ private DurableExecutionInput createDurableInput(I input) {
.build();
// Load previous operations and include them in InitialExecutionState
- var existingOps = storage.getAllOperations();
+ var existingOps = storage.getAllOperations().stream()
+ .filter(op -> op.type() != OperationType.EXECUTION)
+ .toList();
var allOps = new ArrayList<>(List.of(executionOp));
allOps.addAll(existingOps);
@@ -362,6 +404,10 @@ private DurableExecutionInput createDurableInput(I input) {
updatedOperationIds);
}
+ private String serializeInput(I input) {
+ return inputSerDes.serialize(input);
+ }
+
private Context mockLambdaContext() {
return null; // Minimal - tests don't need real Lambda context
}
diff --git a/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/TestOperation.java b/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/TestOperation.java
index 31a28b988..308b7e397 100644
--- a/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/TestOperation.java
+++ b/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/TestOperation.java
@@ -18,22 +18,39 @@
import software.amazon.awssdk.services.lambda.model.WaitDetails;
import software.amazon.lambda.durable.TypeToken;
import software.amazon.lambda.durable.execution.ExecutionManager;
+import software.amazon.lambda.durable.model.OperationSubType;
import software.amazon.lambda.durable.serde.SerDes;
+import software.amazon.lambda.durable.serde.SerDesContext;
+import software.amazon.lambda.durable.serde.SerDesPayloadKind;
+import software.amazon.lambda.durable.serde.SerDesRunner;
/** Wrapper for AWS SDK Operation providing convenient access methods. */
public class TestOperation {
private final Operation operation;
private final List events;
private final SerDes serDes;
+ private final SerDesRunner serDesRunner;
+ private final String durableExecutionArn;
public TestOperation(Operation operation, SerDes serDes) {
this(operation, List.of(), serDes);
}
public TestOperation(Operation operation, List events, SerDes serDes) {
+ this(operation, events, serDes, null, null);
+ }
+
+ public TestOperation(
+ Operation operation,
+ List events,
+ SerDes serDes,
+ SerDesRunner serDesRunner,
+ String durableExecutionArn) {
this.operation = operation;
this.events = events;
this.serDes = serDes;
+ this.serDesRunner = serDesRunner;
+ this.durableExecutionArn = durableExecutionArn;
}
/** Returns the raw history events associated with this operation. */
@@ -119,7 +136,40 @@ public T getStepResult(TypeToken type) {
if (details == null || details.result() == null) {
return null;
}
- return serDes.deserialize(details.result(), type);
+ if (serDesRunner == null) {
+ return serDes.deserialize(details.result(), type);
+ }
+ var subType = java.util.Arrays.stream(OperationSubType.values())
+ .filter(value -> value.getValue().equals(operation.subType()))
+ .findFirst()
+ .orElse(OperationSubType.STEP);
+ var payloadKind =
+ subType == OperationSubType.WAIT_FOR_CONDITION ? SerDesPayloadKind.STATE : SerDesPayloadKind.RESULT;
+ var resultAttempt = resultAttempt(details, subType);
+ return serDesRunner.deserialize(
+ serDes,
+ details.result(),
+ type,
+ SerDesContext.forOperation(
+ durableExecutionArn,
+ operation.id(),
+ operation.name(),
+ operation.parentId(),
+ operation.type(),
+ subType,
+ payloadKind,
+ resultAttempt));
+ }
+
+ private Integer resultAttempt(StepDetails details, OperationSubType subType) {
+ var attempt = details.attempt();
+ if (subType == OperationSubType.WAIT_FOR_CONDITION
+ && operation.status() == OperationStatus.FAILED
+ && attempt != null
+ && attempt > 1) {
+ return attempt - 1;
+ }
+ return attempt;
}
/** Returns the step error, or null if the step succeeded or this is not a step operation. */
diff --git a/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/TestResult.java b/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/TestResult.java
index 7de85beef..f78fb80b7 100644
--- a/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/TestResult.java
+++ b/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/TestResult.java
@@ -15,6 +15,9 @@
import software.amazon.lambda.durable.TypeToken;
import software.amazon.lambda.durable.model.ExecutionStatus;
import software.amazon.lambda.durable.serde.SerDes;
+import software.amazon.lambda.durable.serde.SerDesContext;
+import software.amazon.lambda.durable.serde.SerDesPayloadKind;
+import software.amazon.lambda.durable.serde.SerDesRunner;
/**
* Represents the result of a durable execution, providing access to the execution status, output, operations, and
@@ -33,6 +36,8 @@ public class TestResult {
private final List allEvents;
private final SerDes serDes;
private final TypeToken resultType;
+ private final SerDesRunner serDesRunner;
+ private final SerDesContext outputContext;
public TestResult(
ExecutionStatus status,
@@ -42,6 +47,21 @@ public TestResult(
List allEvents,
TypeToken resultType,
SerDes serDes) {
+ this(status, resultPayload, error, operations, allEvents, resultType, serDes, null, null, null, null);
+ }
+
+ public TestResult(
+ ExecutionStatus status,
+ String resultPayload,
+ ErrorObject error,
+ List operations,
+ List allEvents,
+ TypeToken resultType,
+ SerDes serDes,
+ SerDesRunner serDesRunner,
+ String durableExecutionArn,
+ String executionOperationId,
+ String executionOperationName) {
this.status = status;
this.resultPayload = resultPayload;
this.error = error;
@@ -51,6 +71,11 @@ public TestResult(
this.allEvents = List.copyOf(allEvents);
this.serDes = serDes;
this.resultType = resultType;
+ this.serDesRunner = serDesRunner;
+ this.outputContext = serDesRunner == null
+ ? null
+ : SerDesContext.forExecution(
+ durableExecutionArn, executionOperationId, executionOperationName, SerDesPayloadKind.OUTPUT);
}
/** Returns the execution status (SUCCEEDED, FAILED, or PENDING). */
@@ -75,12 +100,18 @@ public T getResult(TypeToken resultType) {
if (resultPayload == null || resultPayload.isEmpty()) {
var lastEvent = allEvents.get(allEvents.size() - 1);
if (lastEvent.eventType() == EventType.EXECUTION_SUCCEEDED) {
- return serDes.deserialize(
+ return deserialize(
lastEvent.executionSucceededDetails().result().payload(), resultType);
}
return null;
}
- return serDes.deserialize(resultPayload, resultType);
+ return deserialize(resultPayload, resultType);
+ }
+
+ private T deserialize(String payload, TypeToken type) {
+ return serDesRunner == null
+ ? serDes.deserialize(payload, type)
+ : serDesRunner.deserialize(serDes, payload, type, outputContext);
}
/** Deserializes and returns the execution output if the result type is known. */
diff --git a/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/cloud/HistoryEventProcessor.java b/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/cloud/HistoryEventProcessor.java
index 4a3b8f1b2..445ee335c 100644
--- a/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/cloud/HistoryEventProcessor.java
+++ b/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/cloud/HistoryEventProcessor.java
@@ -5,19 +5,23 @@
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
+import java.util.Objects;
import software.amazon.awssdk.services.lambda.model.CallbackDetails;
import software.amazon.awssdk.services.lambda.model.ChainedInvokeDetails;
import software.amazon.awssdk.services.lambda.model.ContextDetails;
import software.amazon.awssdk.services.lambda.model.ErrorObject;
import software.amazon.awssdk.services.lambda.model.Event;
+import software.amazon.awssdk.services.lambda.model.EventType;
import software.amazon.awssdk.services.lambda.model.Operation;
import software.amazon.awssdk.services.lambda.model.OperationStatus;
import software.amazon.awssdk.services.lambda.model.OperationType;
import software.amazon.awssdk.services.lambda.model.StepDetails;
import software.amazon.awssdk.services.lambda.model.WaitDetails;
import software.amazon.lambda.durable.TypeToken;
+import software.amazon.lambda.durable.execution.ExecutionManager;
import software.amazon.lambda.durable.model.ExecutionStatus;
import software.amazon.lambda.durable.serde.SerDes;
+import software.amazon.lambda.durable.serde.SerDesRunner;
import software.amazon.lambda.durable.testing.AsyncExecution;
import software.amazon.lambda.durable.testing.CloudDurableTestRunner;
import software.amazon.lambda.durable.testing.TestOperation;
@@ -37,11 +41,33 @@ public class HistoryEventProcessor {
* @return a TestResult containing the execution status, output, and operation details
*/
public TestResult processEvents(List events, TypeToken outputType, SerDes serDes) {
+ return processEvents(events, outputType, serDes, null, null);
+ }
+
+ /**
+ * Processes execution history using SDK-managed SerDes context.
+ *
+ * @param events the raw history events from the GetDurableExecutionHistory API
+ * @param outputType the expected output type for deserialization
+ * @param serDes the SerDes used by the durable function
+ * @param serDesRunner invocation-scoped SerDes runner, or {@code null} for legacy direct calls
+ * @param durableExecutionArn durable execution ARN, required when {@code serDesRunner} is supplied
+ * @param the handler output type
+ * @return a TestResult containing the execution status, output, and operation details
+ */
+ public TestResult processEvents(
+ List events,
+ TypeToken outputType,
+ SerDes serDes,
+ SerDesRunner serDesRunner,
+ String durableExecutionArn) {
var operations = new HashMap();
var operationEvents = new HashMap>();
var status = ExecutionStatus.PENDING;
String result = null;
ErrorObject error = null;
+ String executionOperationId = null;
+ String executionOperationName = null;
for (var event : events) {
var eventType = event.eventType();
@@ -56,7 +82,8 @@ public TestResult processEvents(List events, TypeToken outputTy
switch (eventType) {
case EXECUTION_STARTED -> {
- // Execution started - no action needed, just track the event
+ executionOperationId = operationId;
+ executionOperationName = event.name();
}
case INVOCATION_COMPLETED -> {
var details = event.invocationCompletedDetails();
@@ -111,7 +138,14 @@ public TestResult processEvents(List events, TypeToken outputTy
if (operationId != null) {
operations.putIfAbsent(
operationId,
- createStepOperation(operationId, event.name(), null, OperationStatus.STARTED, 1));
+ createStepOperation(
+ operationId,
+ event.name(),
+ event.parentId(),
+ event.subType(),
+ null,
+ OperationStatus.STARTED,
+ 1));
}
}
case STEP_SUCCEEDED -> {
@@ -126,7 +160,13 @@ public TestResult processEvents(List events, TypeToken outputTy
operations.put(
operationId,
createStepOperation(
- operationId, event.name(), stepResult, OperationStatus.SUCCEEDED, attempt));
+ operationId,
+ event.name(),
+ event.parentId(),
+ event.subType(),
+ stepResult,
+ OperationStatus.SUCCEEDED,
+ attempt));
}
}
case STEP_FAILED -> {
@@ -137,7 +177,14 @@ public TestResult processEvents(List events, TypeToken outputTy
: 1;
operations.put(
operationId,
- createStepOperation(operationId, event.name(), null, OperationStatus.FAILED, attempt));
+ createStepOperation(
+ operationId,
+ event.name(),
+ event.parentId(),
+ event.subType(),
+ null,
+ OperationStatus.FAILED,
+ attempt));
}
}
@@ -224,7 +271,11 @@ public TestResult processEvents(List events, TypeToken outputTy
CHAINED_INVOKE_TIMED_OUT,
CHAINED_INVOKE_STOPPED -> {
if (operationId != null) {
- operations.putIfAbsent(operationId, createInvokeOperation(operationId, event));
+ if (eventType == EventType.CHAINED_INVOKE_STARTED) {
+ operations.putIfAbsent(operationId, createInvokeOperation(operationId, event));
+ } else {
+ operations.put(operationId, createInvokeOperation(operationId, event));
+ }
}
}
@@ -236,14 +287,60 @@ public TestResult processEvents(List events, TypeToken outputTy
var testOperations = new ArrayList();
for (var entry : operations.entrySet()) {
var opEvents = operationEvents.getOrDefault(entry.getKey(), List.of());
- testOperations.add(new TestOperation(entry.getValue(), opEvents, serDes));
+ var operation = withEventTimestamps(entry.getValue(), opEvents);
+ testOperations.add(
+ serDesRunner == null
+ ? new TestOperation(operation, opEvents, serDes)
+ : new TestOperation(operation, opEvents, serDes, serDesRunner, durableExecutionArn));
}
- return new TestResult<>(status, result, error, testOperations, events, outputType, serDes);
+ if (executionOperationId == null && durableExecutionArn != null) {
+ var parts = durableExecutionArn.split("/", -1);
+ executionOperationId = parts[parts.length - 1];
+ }
+ return serDesRunner == null
+ ? new TestResult<>(status, result, error, testOperations, events, outputType, serDes)
+ : new TestResult<>(
+ status,
+ result,
+ error,
+ testOperations,
+ events,
+ outputType,
+ serDes,
+ serDesRunner,
+ durableExecutionArn,
+ executionOperationId,
+ executionOperationName);
+ }
+
+ private Operation withEventTimestamps(Operation operation, List events) {
+ var startTimestamp = events.stream()
+ .map(Event::eventTimestamp)
+ .filter(Objects::nonNull)
+ .min(java.time.Instant::compareTo)
+ .orElse(operation.startTimestamp());
+ var endTimestamp = ExecutionManager.isTerminalStatus(operation.status())
+ ? events.stream()
+ .map(Event::eventTimestamp)
+ .filter(Objects::nonNull)
+ .max(java.time.Instant::compareTo)
+ .orElse(operation.endTimestamp())
+ : operation.endTimestamp();
+ return operation.toBuilder()
+ .startTimestamp(startTimestamp)
+ .endTimestamp(endTimestamp)
+ .build();
}
private Operation createStepOperation(
- String id, String name, String stepResult, OperationStatus status, Integer attempt) {
+ String id,
+ String name,
+ String parentId,
+ String subType,
+ String stepResult,
+ OperationStatus status,
+ Integer attempt) {
var stepDetails = StepDetails.builder()
.result(stepResult)
.attempt(attempt != null ? attempt : 1)
@@ -252,8 +349,10 @@ private Operation createStepOperation(
return Operation.builder()
.id(id)
.name(name)
+ .parentId(parentId)
.status(status)
.type(OperationType.STEP)
+ .subType(subType)
.stepDetails(stepDetails)
.build();
}
@@ -267,8 +366,10 @@ private Operation createWaitOperation(String id, String name, OperationStatus st
return Operation.builder()
.id(id)
.name(name)
+ .parentId(event.parentId())
.status(status)
.type(OperationType.WAIT)
+ .subType(event.subType())
.waitDetails(builder.build())
.build();
}
@@ -302,8 +403,10 @@ private Operation createCallbackOperation(String id, String name, OperationStatu
return Operation.builder()
.id(id)
.name(name)
+ .parentId(event.parentId())
.status(status)
.type(OperationType.CALLBACK)
+ .subType(event.subType())
.callbackDetails(builder.build())
.build();
}
@@ -315,7 +418,7 @@ private Operation createInvokeOperation(String id, Event event) {
switch (event.eventType()) {
case CHAINED_INVOKE_STARTED -> OperationStatus.STARTED;
case CHAINED_INVOKE_SUCCEEDED -> {
- var details = event.callbackSucceededDetails();
+ var details = event.chainedInvokeSucceededDetails();
if (details != null
&& details.result() != null
&& details.result().payload() != null) {
@@ -324,7 +427,7 @@ private Operation createInvokeOperation(String id, Event event) {
yield OperationStatus.SUCCEEDED;
}
case CHAINED_INVOKE_FAILED -> {
- var details = event.callbackFailedDetails();
+ var details = event.chainedInvokeFailedDetails();
if (details != null
&& details.error() != null
&& details.error().payload() != null) {
@@ -359,8 +462,10 @@ private Operation createInvokeOperation(String id, Event event) {
return Operation.builder()
.id(id)
.name(event.name())
+ .parentId(event.parentId())
.status(status)
.type(OperationType.CHAINED_INVOKE)
+ .subType(event.subType())
.chainedInvokeDetails(builder.build())
.build();
}
@@ -383,6 +488,7 @@ private Operation createContextOperation(String id, String name, OperationStatus
return Operation.builder()
.id(id)
.name(name)
+ .parentId(event.parentId())
.status(status)
.type(OperationType.CONTEXT)
.subType(event.subType())
diff --git a/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/local/LocalMemoryExecutionClient.java b/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/local/LocalMemoryExecutionClient.java
index 25f016cd9..ca9cd0861 100644
--- a/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/local/LocalMemoryExecutionClient.java
+++ b/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/local/LocalMemoryExecutionClient.java
@@ -24,6 +24,7 @@
import software.amazon.lambda.durable.client.DurableExecutionClient;
import software.amazon.lambda.durable.model.DurableExecutionOutput;
import software.amazon.lambda.durable.serde.SerDes;
+import software.amazon.lambda.durable.serde.SerDesRunner;
import software.amazon.lambda.durable.testing.TestOperation;
import software.amazon.lambda.durable.testing.TestResult;
@@ -131,9 +132,22 @@ public List getUpdatedOperationIdsSinceLastInvocation() {
/** Build TestResult from current state. */
public TestResult toTestResult(DurableExecutionOutput output, TypeToken resultType, SerDes serDes) {
+ return toTestResult(output, resultType, serDes, null, null, null, null);
+ }
+
+ /** Build a context-aware TestResult from current state. */
+ public TestResult toTestResult(
+ DurableExecutionOutput output,
+ TypeToken resultType,
+ SerDes serDes,
+ SerDesRunner serDesRunner,
+ String durableExecutionArn,
+ String executionOperationId,
+ String executionOperationName) {
var testOperations = existingOperations.values().stream()
.filter(op -> op.type() != OperationType.EXECUTION)
- .map(op -> new TestOperation(op, eventProcessor.getEventsForOperation(op.id()), serDes))
+ .map(op -> new TestOperation(
+ op, eventProcessor.getEventsForOperation(op.id()), serDes, serDesRunner, durableExecutionArn))
.toList();
return new TestResult<>(
output.status(),
@@ -142,7 +156,11 @@ public TestResult toTestResult(DurableExecutionOutput output, TypeToken T deserialize(String data, TypeToken typeToken) {
+ deserializations.incrementAndGet();
+ return (T) data;
+ }
+ };
+ var execution = new AsyncExecution<>(
+ EXECUTION_ARN, lambdaClient, TypeToken.get(String.class), serDes, Duration.ZERO, Duration.ofSeconds(1));
+ var snapshots = new AtomicInteger();
+
+ execution.pollUntil(current -> {
+ assertEquals("step-result", current.getOperation("step").getStepResult(String.class));
+ assertEquals("step-result", current.getOperation("step").getStepResult(String.class));
+ return snapshots.incrementAndGet() == 2;
+ });
+
+ assertEquals(2, deserializations.get());
+ }
+
+ private static List stepEvents() {
+ var startedAt = Instant.parse("2026-08-25T00:00:00Z");
+ return List.of(
+ Event.builder()
+ .id("step-id")
+ .name("step")
+ .subType("Step")
+ .eventType(EventType.STEP_STARTED)
+ .eventTimestamp(startedAt)
+ .stepStartedDetails(StepStartedDetails.builder().build())
+ .build(),
+ Event.builder()
+ .id("step-id")
+ .name("step")
+ .subType("Step")
+ .eventType(EventType.STEP_SUCCEEDED)
+ .eventTimestamp(startedAt.plusSeconds(1))
+ .stepSucceededDetails(StepSucceededDetails.builder()
+ .result(EventResult.builder()
+ .payload("step-result")
+ .build())
+ .retryDetails(
+ RetryDetails.builder().currentAttempt(1).build())
+ .build())
+ .build());
+ }
+}
diff --git a/sdk-testing/src/test/java/software/amazon/lambda/durable/testing/CloudDurableTestRunnerTest.java b/sdk-testing/src/test/java/software/amazon/lambda/durable/testing/CloudDurableTestRunnerTest.java
index b1fde42e9..a3d418a9d 100644
--- a/sdk-testing/src/test/java/software/amazon/lambda/durable/testing/CloudDurableTestRunnerTest.java
+++ b/sdk-testing/src/test/java/software/amazon/lambda/durable/testing/CloudDurableTestRunnerTest.java
@@ -5,10 +5,24 @@
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.Mockito.*;
+import java.nio.file.Path;
import java.time.Duration;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.ArgumentCaptor;
import software.amazon.awssdk.services.lambda.LambdaClient;
import software.amazon.awssdk.services.lambda.model.InvocationType;
+import software.amazon.awssdk.services.lambda.model.InvokeRequest;
+import software.amazon.awssdk.services.lambda.model.InvokeResponse;
+import software.amazon.lambda.durable.TypeToken;
+import software.amazon.lambda.durable.exception.SerDesException;
+import software.amazon.lambda.durable.serde.JacksonSerDes;
+import software.amazon.lambda.durable.serde.SerDes;
+import software.amazon.lambda.durable.serde.SerDesContext;
+import software.amazon.lambda.durable.serde.SerDesPayloadKind;
+import software.amazon.lambda.durable.serde.SerDesRunner;
+import software.amazon.lambda.durable.serde.SerDesStage;
+import software.amazon.lambda.durable.serde.filesystem.FileSystemSerDesStage;
class CloudDurableTestRunnerTest {
@@ -31,4 +45,153 @@ void testPlaceholderMethods() {
assertThrows(IllegalStateException.class, () -> runner.getOperation("test"));
}
+
+ @Test
+ void rejectsComposableInputSerDes() {
+ var mockClient = mock(LambdaClient.class);
+ var runner = CloudDurableTestRunner.create(
+ "arn:aws:lambda:us-east-2:123:function:test", String.class, String.class, mockClient);
+
+ var failure = assertThrows(
+ IllegalArgumentException.class,
+ () -> runner.withInputSerDes(new JacksonSerDes().then(wrappingStage())));
+
+ assertTrue(failure.getMessage().contains("value codec"));
+ }
+
+ @Test
+ void persistedComposableSerDesUsesRootValueCodec() {
+ var mockClient = mock(LambdaClient.class);
+ when(mockClient.invoke(any(InvokeRequest.class)))
+ .thenReturn(InvokeResponse.builder()
+ .durableExecutionArn("arn:aws:lambda:us-east-2:123:function:test:1/durable-execution/e/i")
+ .build());
+ var runner = CloudDurableTestRunner.create(
+ "arn:aws:lambda:us-east-2:123:function:test", String.class, String.class, mockClient)
+ .withSerDes(wrappingValueCodec().then(wrappingStage()));
+
+ runner.startAsync("value");
+
+ var request = ArgumentCaptor.forClass(InvokeRequest.class);
+ verify(mockClient).invoke(request.capture());
+ assertEquals("<\"value\">", request.getValue().payload().asUtf8String());
+ }
+
+ @Test
+ void plainPersistedSerDesIsUsedAsDefaultInputCodec() {
+ var mockClient = mock(LambdaClient.class);
+ when(mockClient.invoke(any(InvokeRequest.class)))
+ .thenReturn(InvokeResponse.builder()
+ .durableExecutionArn("arn:aws:lambda:us-east-2:123:function:test:1/durable-execution/e/i")
+ .build());
+ var runner = CloudDurableTestRunner.create(
+ "arn:aws:lambda:us-east-2:123:function:test", String.class, String.class, mockClient)
+ .withSerDes(wrappingValueCodec());
+
+ runner.startAsync("value");
+
+ var request = ArgumentCaptor.forClass(InvokeRequest.class);
+ verify(mockClient).invoke(request.capture());
+ assertEquals("<\"value\">", request.getValue().payload().asUtf8String());
+ }
+
+ @Test
+ void explicitInputValueCodecIsUsed() {
+ var mockClient = mock(LambdaClient.class);
+ when(mockClient.invoke(any(InvokeRequest.class)))
+ .thenReturn(InvokeResponse.builder()
+ .durableExecutionArn("arn:aws:lambda:us-east-2:123:function:test:1/durable-execution/e/i")
+ .build());
+ var runner = CloudDurableTestRunner.create(
+ "arn:aws:lambda:us-east-2:123:function:test", String.class, String.class, mockClient)
+ .withInputSerDes(wrappingValueCodec());
+
+ runner.startAsync("value");
+
+ var request = ArgumentCaptor.forClass(InvokeRequest.class);
+ verify(mockClient).invoke(request.capture());
+ assertEquals("<\"value\">", request.getValue().payload().asUtf8String());
+ }
+
+ @Test
+ void valueCodecInputRoundTripsThroughFileSystemPipeline(@TempDir Path basePath) {
+ var mockClient = mock(LambdaClient.class);
+ var executionArn = "arn:aws:lambda:us-east-2:123:function:test:1/durable-execution/e/i";
+ when(mockClient.invoke(any(InvokeRequest.class)))
+ .thenReturn(InvokeResponse.builder()
+ .durableExecutionArn(executionArn)
+ .build());
+ var persistedSerDes =
+ new JacksonSerDes().then(FileSystemSerDesStage.builder(basePath).build());
+ var runner = CloudDurableTestRunner.create(
+ "arn:aws:lambda:us-east-2:123:function:test", String.class, String.class, mockClient)
+ .withSerDes(persistedSerDes);
+
+ runner.startAsync("value");
+
+ var request = ArgumentCaptor.forClass(InvokeRequest.class);
+ verify(mockClient).invoke(request.capture());
+ assertEquals(
+ "value",
+ new SerDesRunner(null)
+ .deserialize(
+ persistedSerDes,
+ request.getValue().payload().asUtf8String(),
+ TypeToken.get(String.class),
+ SerDesContext.forExecution(executionArn, "i", "execution", SerDesPayloadKind.INPUT)));
+ }
+
+ @Test
+ void replacingPersistedSerDesPreservesExplicitInputSerDes() {
+ var mockClient = mock(LambdaClient.class);
+ when(mockClient.invoke(any(InvokeRequest.class)))
+ .thenReturn(InvokeResponse.builder()
+ .durableExecutionArn("arn:aws:lambda:us-east-2:123:function:test:1/durable-execution/e/i")
+ .build());
+ var runner = CloudDurableTestRunner.create(
+ "arn:aws:lambda:us-east-2:123:function:test", String.class, String.class, mockClient)
+ .withInputSerDes(wrappingValueCodec())
+ .withSerDes(new JacksonSerDes());
+
+ runner.startAsync("value");
+
+ var request = ArgumentCaptor.forClass(InvokeRequest.class);
+ verify(mockClient).invoke(request.capture());
+ assertEquals("<\"value\">", request.getValue().payload().asUtf8String());
+ }
+
+ private static SerDesStage wrappingStage() {
+ return new SerDesStage() {
+ @Override
+ public String serialize(String value, SerDesContext context) {
+ return "<" + value + ">";
+ }
+
+ @Override
+ public String deserialize(String data, SerDesContext context) {
+ if (!data.startsWith("<")) {
+ return data;
+ }
+ if (!data.endsWith(">")) {
+ throw new SerDesException("Malformed wrapping stage value");
+ }
+ return data.substring(1, data.length() - 1);
+ }
+ };
+ }
+
+ private static SerDes wrappingValueCodec() {
+ var delegate = new JacksonSerDes();
+ return new SerDes() {
+ @Override
+ public String serialize(Object value) {
+ return "<" + delegate.serialize(value) + ">";
+ }
+
+ @Override
+ public T deserialize(String data, TypeToken typeToken) {
+ return delegate.deserialize(data.substring(1, data.length() - 1), typeToken);
+ }
+ };
+ }
}
diff --git a/sdk-testing/src/test/java/software/amazon/lambda/durable/testing/LocalDurableTestRunnerTest.java b/sdk-testing/src/test/java/software/amazon/lambda/durable/testing/LocalDurableTestRunnerTest.java
index 36f1bbced..e6b6d58ac 100644
--- a/sdk-testing/src/test/java/software/amazon/lambda/durable/testing/LocalDurableTestRunnerTest.java
+++ b/sdk-testing/src/test/java/software/amazon/lambda/durable/testing/LocalDurableTestRunnerTest.java
@@ -5,17 +5,30 @@
import static org.junit.jupiter.api.Assertions.*;
import static software.amazon.lambda.durable.TypeToken.get;
+import java.nio.file.Path;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import software.amazon.lambda.durable.DurableConfig;
import software.amazon.lambda.durable.TypeToken;
+import software.amazon.lambda.durable.exception.SerDesException;
import software.amazon.lambda.durable.model.ExecutionStatus;
import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
import software.amazon.lambda.durable.plugin.InvocationInfo;
+import software.amazon.lambda.durable.serde.Base64StringBinaryCodec;
+import software.amazon.lambda.durable.serde.BinarySerDesStage;
+import software.amazon.lambda.durable.serde.ComposableBinarySerDesStage;
+import software.amazon.lambda.durable.serde.JacksonSerDes;
+import software.amazon.lambda.durable.serde.SerDes;
+import software.amazon.lambda.durable.serde.SerDesContext;
+import software.amazon.lambda.durable.serde.SerDesStage;
+import software.amazon.lambda.durable.serde.Utf8StringBinaryCodec;
+import software.amazon.lambda.durable.serde.filesystem.FileSystemSerDesStage;
class LocalDurableTestRunnerTest {
@@ -114,4 +127,189 @@ public void onInvocationStart(InvocationInfo info) {
assertNotNull(executionStartTimes.get(0));
assertEquals(executionStartTimes.get(0), executionStartTimes.get(1));
}
+
+ @Test
+ void checkpointedLargeOutputReplaysWithoutDuplicateExecutionOperation() {
+ var stepExecutions = new AtomicInteger();
+ var largeResult = "x".repeat(7 * 1024 * 1024);
+ var runner = LocalDurableTestRunner.create(String.class, (input, context) -> {
+ context.step("once", Void.class, step -> {
+ stepExecutions.incrementAndGet();
+ return null;
+ });
+ return largeResult;
+ })
+ .withOutputType(String.class);
+
+ var firstResult = runner.run("test");
+ var replayResult = runner.run("test");
+
+ assertEquals(ExecutionStatus.SUCCEEDED, firstResult.getStatus());
+ assertEquals(largeResult, firstResult.getResult());
+ assertEquals(ExecutionStatus.SUCCEEDED, replayResult.getStatus());
+ assertEquals(largeResult, replayResult.getResult());
+ assertEquals(1, stepExecutions.get());
+ }
+
+ @Test
+ void filesystemPersistedSerDesUsesDefaultJacksonInputCodec(@TempDir Path basePath) {
+ var config = DurableConfig.builder()
+ .withSerDes(new JacksonSerDes()
+ .then(FileSystemSerDesStage.builder(basePath).build()))
+ .build();
+ var runner = LocalDurableTestRunner.create(String.class, (input, context) -> input, config)
+ .withOutputType(String.class);
+
+ var result = runner.run("value");
+
+ assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+ assertEquals("value", result.getResult());
+ }
+
+ @Test
+ void plainPersistedSerDesIsUsedAsDefaultInputCodec() {
+ var config = DurableConfig.builder().withSerDes(prefixedStringSerDes()).build();
+ var runner = LocalDurableTestRunner.create(String.class, (input, context) -> input, config)
+ .withOutputType(String.class);
+
+ var result = runner.run("value");
+
+ assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+ assertEquals("value", result.getResult());
+ }
+
+ @Test
+ void explicitInputSerDesIsUsedByRunnerAndRuntime() {
+ var config = DurableConfig.builder().withSerDes(new JacksonSerDes()).build();
+ var runner = LocalDurableTestRunner.create(String.class, (input, context) -> input, config)
+ .withInputSerDes(prefixedStringSerDes())
+ .withOutputType(String.class);
+
+ var result = runner.run("value");
+
+ assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+ assertEquals("value", result.getResult());
+ }
+
+ @Test
+ void initialInputBypassesPersistedPipelineStages() {
+ var deserializeCalls = new AtomicInteger();
+ var persistedSerDes = new JacksonSerDes().then(new SerDesStage() {
+ @Override
+ public String serialize(String value, SerDesContext context) {
+ return "persisted:" + value;
+ }
+
+ @Override
+ public String deserialize(String data, SerDesContext context) {
+ deserializeCalls.incrementAndGet();
+ if (data.startsWith("custom:")) {
+ throw new SerDesException("Initial input collided with a persisted stage frame");
+ }
+ return data.startsWith("persisted:") ? data.substring("persisted:".length()) : data;
+ }
+ });
+ var config = DurableConfig.builder()
+ .withSerDes(persistedSerDes)
+ .withInputSerDes(prefixedStringSerDes())
+ .build();
+ var runner = LocalDurableTestRunner.create(
+ String.class, (input, context) -> input + ":" + deserializeCalls.get(), config)
+ .withOutputType(String.class);
+
+ var result = runner.run("value");
+
+ assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+ assertEquals("value:0", result.getResult());
+ }
+
+ @Test
+ void rejectsComposableInputSerDes(@TempDir Path basePath) {
+ var config = DurableConfig.builder()
+ .withSerDes(new JacksonSerDes()
+ .then(FileSystemSerDesStage.builder(basePath).build()))
+ .build();
+ var runner = LocalDurableTestRunner.create(String.class, (input, context) -> input, config);
+
+ var failure = assertThrows(
+ IllegalArgumentException.class,
+ () -> runner.withInputSerDes(new JacksonSerDes().then(wrappingStage(new AtomicInteger()))));
+
+ assertTrue(failure.getMessage().contains("value codec"));
+ }
+
+ @Test
+ void rawInputPassesThroughPersistedStagesBeforeFileSystemSerDesStage(@TempDir Path basePath) {
+ var deserializeCalls = new AtomicInteger();
+ var persistedSerDes = new JacksonSerDes()
+ .then(bytesStage(deserializeCalls))
+ .then(FileSystemSerDesStage.builder(basePath).build());
+ var config = DurableConfig.builder().withSerDes(persistedSerDes).build();
+ var runner = LocalDurableTestRunner.create(
+ String.class, (input, context) -> input + ":" + deserializeCalls.get(), config)
+ .withOutputType(String.class);
+
+ var result = runner.run("value");
+
+ assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+ assertEquals("value:0", result.getResult());
+ }
+
+ private static SerDesStage wrappingStage(AtomicInteger deserializeCalls) {
+ return new SerDesStage() {
+ @Override
+ public String serialize(String value, SerDesContext context) {
+ return "<" + value + ">";
+ }
+
+ @Override
+ public String deserialize(String data, SerDesContext context) {
+ deserializeCalls.incrementAndGet();
+ if (!data.startsWith("<")) {
+ return data;
+ }
+ if (!data.endsWith(">")) {
+ throw new SerDesException("Malformed wrapping stage value");
+ }
+ return data.substring(1, data.length() - 1);
+ }
+ };
+ }
+
+ private static SerDesStage bytesStage(AtomicInteger deserializeCalls) {
+ return ComposableBinarySerDesStage.builder()
+ .startWith(Utf8StringBinaryCodec.INSTANCE)
+ .then(new BinarySerDesStage() {
+ @Override
+ public byte[] serialize(byte[] value, SerDesContext context) {
+ return value;
+ }
+
+ @Override
+ public byte[] deserialize(byte[] data, SerDesContext context) {
+ deserializeCalls.incrementAndGet();
+ return data;
+ }
+ })
+ .endWith(Base64StringBinaryCodec.INSTANCE)
+ .build();
+ }
+
+ private static SerDes prefixedStringSerDes() {
+ return new SerDes() {
+ @Override
+ public String serialize(Object value) {
+ return "custom:" + value;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public T deserialize(String data, TypeToken typeToken) {
+ if (!TypeToken.get(String.class).equals(typeToken) || !data.startsWith("custom:")) {
+ throw new SerDesException("Invalid custom string payload");
+ }
+ return (T) data.substring("custom:".length());
+ }
+ };
+ }
}
diff --git a/sdk-testing/src/test/java/software/amazon/lambda/durable/testing/TestOperationTest.java b/sdk-testing/src/test/java/software/amazon/lambda/durable/testing/TestOperationTest.java
new file mode 100644
index 000000000..ba0d29a3e
--- /dev/null
+++ b/sdk-testing/src/test/java/software/amazon/lambda/durable/testing/TestOperationTest.java
@@ -0,0 +1,70 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.testing;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.lambda.model.Operation;
+import software.amazon.awssdk.services.lambda.model.OperationStatus;
+import software.amazon.awssdk.services.lambda.model.StepDetails;
+import software.amazon.lambda.durable.TypeToken;
+import software.amazon.lambda.durable.model.OperationSubType;
+import software.amazon.lambda.durable.serde.SerDes;
+import software.amazon.lambda.durable.serde.SerDesContext;
+import software.amazon.lambda.durable.serde.SerDesRunner;
+import software.amazon.lambda.durable.serde.SerDesStage;
+
+class TestOperationTest {
+ private static final String EXECUTION_ARN = "arn:aws:lambda:us-east-1:123456789012:function:test:$LATEST"
+ + "/durable-execution/execution-id/invocation-id";
+
+ @Test
+ void failedWaitForConditionReadsStateFromPreviousAttempt() {
+ var observedContext = new AtomicReference();
+ var valueCodec = new SerDes() {
+ @Override
+ public String serialize(Object value) {
+ return value.toString();
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public T deserialize(String data, TypeToken typeToken) {
+ return (T) data;
+ }
+ };
+ var serDes = valueCodec.then(new SerDesStage() {
+ @Override
+ public String serialize(String value, SerDesContext context) {
+ return value;
+ }
+
+ @Override
+ public String deserialize(String data, SerDesContext context) {
+ observedContext.set(context);
+ return data;
+ }
+ });
+ var operation = Operation.builder()
+ .id("wait-id")
+ .name("wait-condition")
+ .type(OperationSubType.WAIT_FOR_CONDITION.getOperationType())
+ .subType(OperationSubType.WAIT_FOR_CONDITION.getValue())
+ .status(OperationStatus.FAILED)
+ .stepDetails(StepDetails.builder()
+ .attempt(3)
+ .result("retained-state")
+ .build())
+ .build();
+ var testOperation = new TestOperation(operation, List.of(), serDes, new SerDesRunner(null), EXECUTION_ARN);
+
+ assertEquals("retained-state", testOperation.getStepResult(String.class));
+ assertEquals(2, observedContext.get().attempt());
+ assertEquals("operation/wait-id/state/attempt-2", observedContext.get().entityId());
+ assertNull(observedContext.get().originalValue());
+ }
+}
diff --git a/sdk-testing/src/test/java/software/amazon/lambda/durable/testing/cloud/HistoryEventProcessorTest.java b/sdk-testing/src/test/java/software/amazon/lambda/durable/testing/cloud/HistoryEventProcessorTest.java
new file mode 100644
index 000000000..326be662c
--- /dev/null
+++ b/sdk-testing/src/test/java/software/amazon/lambda/durable/testing/cloud/HistoryEventProcessorTest.java
@@ -0,0 +1,155 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.testing.cloud;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.lambda.model.ChainedInvokeStartedDetails;
+import software.amazon.awssdk.services.lambda.model.ChainedInvokeSucceededDetails;
+import software.amazon.awssdk.services.lambda.model.Event;
+import software.amazon.awssdk.services.lambda.model.EventResult;
+import software.amazon.awssdk.services.lambda.model.EventType;
+import software.amazon.awssdk.services.lambda.model.ExecutionStartedDetails;
+import software.amazon.awssdk.services.lambda.model.ExecutionSucceededDetails;
+import software.amazon.awssdk.services.lambda.model.OperationStatus;
+import software.amazon.awssdk.services.lambda.model.OperationType;
+import software.amazon.awssdk.services.lambda.model.RetryDetails;
+import software.amazon.awssdk.services.lambda.model.StepStartedDetails;
+import software.amazon.awssdk.services.lambda.model.StepSucceededDetails;
+import software.amazon.lambda.durable.TypeToken;
+import software.amazon.lambda.durable.serde.SerDes;
+import software.amazon.lambda.durable.serde.SerDesContext;
+import software.amazon.lambda.durable.serde.SerDesPayloadKind;
+import software.amazon.lambda.durable.serde.SerDesRunner;
+import software.amazon.lambda.durable.serde.SerDesStage;
+
+class HistoryEventProcessorTest {
+ private static final String EXECUTION_ARN = "arn:aws:lambda:us-east-1:123456789012:function:test:$LATEST"
+ + "/durable-execution/execution-id/invocation-id";
+
+ @Test
+ void deserializesCloudResultsWithDurablePayloadContext() {
+ var observedContexts = new ArrayList();
+ var serDes = recordingStringSerDes(observedContexts);
+ var startedAt = Instant.parse("2026-08-24T00:00:00Z");
+ var events = List.of(
+ Event.builder()
+ .id("invocation-id")
+ .name("execution")
+ .eventType(EventType.EXECUTION_STARTED)
+ .eventTimestamp(startedAt)
+ .executionStartedDetails(
+ ExecutionStartedDetails.builder().build())
+ .build(),
+ Event.builder()
+ .id("step-id")
+ .name("step")
+ .subType("Step")
+ .eventType(EventType.STEP_STARTED)
+ .eventTimestamp(startedAt.plusSeconds(1))
+ .stepStartedDetails(StepStartedDetails.builder().build())
+ .build(),
+ Event.builder()
+ .id("step-id")
+ .name("step")
+ .subType("Step")
+ .eventType(EventType.STEP_SUCCEEDED)
+ .eventTimestamp(startedAt.plusSeconds(3))
+ .stepSucceededDetails(StepSucceededDetails.builder()
+ .result(EventResult.builder()
+ .payload("step-result")
+ .build())
+ .retryDetails(
+ RetryDetails.builder().currentAttempt(2).build())
+ .build())
+ .build(),
+ Event.builder()
+ .id("invoke-id")
+ .name("invoke")
+ .eventType(EventType.CHAINED_INVOKE_STARTED)
+ .eventTimestamp(startedAt.plusSeconds(4))
+ .chainedInvokeStartedDetails(ChainedInvokeStartedDetails.builder()
+ .functionName("target")
+ .build())
+ .build(),
+ Event.builder()
+ .id("invoke-id")
+ .name("invoke")
+ .eventType(EventType.CHAINED_INVOKE_SUCCEEDED)
+ .eventTimestamp(startedAt.plusSeconds(5))
+ .chainedInvokeSucceededDetails(ChainedInvokeSucceededDetails.builder()
+ .result(EventResult.builder()
+ .payload("invoke-result")
+ .build())
+ .build())
+ .build(),
+ Event.builder()
+ .id("invocation-id")
+ .name("execution")
+ .eventType(EventType.EXECUTION_SUCCEEDED)
+ .eventTimestamp(startedAt.plusSeconds(6))
+ .executionSucceededDetails(ExecutionSucceededDetails.builder()
+ .result(EventResult.builder()
+ .payload("execution-result")
+ .build())
+ .build())
+ .build());
+
+ var result = new HistoryEventProcessor()
+ .processEvents(events, TypeToken.get(String.class), serDes, new SerDesRunner(null), EXECUTION_ARN);
+
+ assertEquals("execution-result", result.getResult());
+ assertEquals("step-result", result.getOperation("step").getStepResult(String.class));
+ assertEquals(Duration.ofSeconds(2), result.getOperation("step").getDuration());
+ assertEquals(OperationStatus.SUCCEEDED, result.getOperation("invoke").getStatus());
+ assertEquals(
+ "invoke-result",
+ result.getOperation("invoke").getChainedInvokeDetails().result());
+ assertEquals(2, observedContexts.size());
+
+ var outputContext = observedContexts.get(0);
+ assertEquals(OperationType.EXECUTION, outputContext.operationType());
+ assertEquals(SerDesPayloadKind.OUTPUT, outputContext.payloadKind());
+ assertEquals("execution/invocation-id/output", outputContext.entityId());
+ assertNull(outputContext.originalValue());
+
+ var stepContext = observedContexts.get(1);
+ assertEquals(OperationType.STEP, stepContext.operationType());
+ assertEquals(SerDesPayloadKind.RESULT, stepContext.payloadKind());
+ assertEquals("operation/step-id/result/attempt-2", stepContext.entityId());
+ assertEquals(2, stepContext.attempt());
+ }
+
+ private static SerDes recordingStringSerDes(List observedContexts) {
+ SerDes valueCodec = new SerDes() {
+ @Override
+ public String serialize(Object value) {
+ return value.toString();
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public T deserialize(String data, TypeToken typeToken) {
+ return (T) data;
+ }
+ };
+ return valueCodec.then(new SerDesStage() {
+ @Override
+ public String serialize(String value, SerDesContext context) {
+ return value;
+ }
+
+ @Override
+ public String deserialize(String data, SerDesContext context) {
+ observedContexts.add(context);
+ return data;
+ }
+ });
+ }
+}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/DurableConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/DurableConfig.java
index 5101b9fda..e5aa1711d 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/DurableConfig.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/DurableConfig.java
@@ -27,6 +27,7 @@
import software.amazon.lambda.durable.plugin.PluginRunner;
import software.amazon.lambda.durable.retry.PollingStrategies;
import software.amazon.lambda.durable.retry.PollingStrategy;
+import software.amazon.lambda.durable.serde.ComposableSerDes;
import software.amazon.lambda.durable.serde.JacksonSerDes;
import software.amazon.lambda.durable.serde.SerDes;
@@ -94,7 +95,9 @@ public final class DurableConfig {
private final DurableExecutionClient durableExecutionClient;
private final SerDes serDes;
+ private final SerDes inputSerDes;
private final ExecutorService executorService;
+ private final ExecutorService serDesExecutorService;
private final LoggerConfig loggerConfig;
private final PollingStrategy pollingStrategy;
private final Duration checkpointDelay;
@@ -107,8 +110,10 @@ private DurableConfig(Builder builder) {
this.durableExecutionClient = Objects.requireNonNullElseGet(
builder.durableExecutionClient, DurableConfig::createDefaultDurableExecutionClient);
this.serDes = Objects.requireNonNullElseGet(builder.serDes, JacksonSerDes::new);
+ this.inputSerDes = inputValueCodec(builder.inputSerDes, this.serDes);
this.executorService =
Objects.requireNonNullElseGet(builder.executorService, DurableConfig::createDefaultExecutor);
+ this.serDesExecutorService = builder.serDesExecutorService;
this.loggerConfig = Objects.requireNonNullElseGet(builder.loggerConfig, LoggerConfig::defaults);
this.pollingStrategy = Objects.requireNonNullElse(builder.pollingStrategy, PollingStrategies.Presets.DEFAULT);
this.checkpointDelay = Objects.requireNonNullElseGet(builder.checkpointDelay, () -> Duration.ofSeconds(0));
@@ -155,6 +160,19 @@ public SerDes getSerDes() {
return serDes;
}
+ /**
+ * Gets the context-free value codec used to deserialize the initial Lambda invocation payload.
+ *
+ * This codec is separate from the persisted SerDes pipeline because the initial payload is received before a
+ * durable execution context exists. If it is not explicitly configured, a plain persisted SerDes is reused, while a
+ * composable persisted SerDes contributes only its root value codec.
+ *
+ * @return the initial invocation value codec
+ */
+ public SerDes getInputSerDes() {
+ return inputSerDes;
+ }
+
/**
* Gets the configured ExecutorService.
*
@@ -164,6 +182,15 @@ public ExecutorService getExecutorService() {
return executorService;
}
+ /**
+ * Gets the executor used for customer SerDes calls and payload storage I/O.
+ *
+ * @return the configured executor, or {@code null} when SerDes calls execute inline
+ */
+ public ExecutorService getSerDesExecutorService() {
+ return serDesExecutorService;
+ }
+
/**
* Gets the configured LoggerConfig.
*
@@ -232,9 +259,16 @@ public void validateConfiguration() {
if (getSerDes() == null) {
throw new IllegalStateException("SerDes configuration failed");
}
+ if (getInputSerDes() == null) {
+ throw new IllegalStateException("Input SerDes configuration failed");
+ }
if (getExecutorService() == null) {
throw new IllegalStateException("ExecutorService configuration failed");
}
+ if (getSerDesExecutorService() != null && getSerDesExecutorService() == getExecutorService()) {
+ throw new IllegalStateException(
+ "SerDes ExecutorService must be different from the user operation ExecutorService");
+ }
}
/**
@@ -311,11 +345,26 @@ private static ExecutorService createDefaultExecutor() {
return DEFAULT_USER_THREAD_POOL;
}
+ private static SerDes inputValueCodec(SerDes inputSerDes, SerDes persistedSerDes) {
+ var codec = inputSerDes;
+ if (codec == null) {
+ codec = persistedSerDes instanceof ComposableSerDes composable
+ ? composable.getValueCodec()
+ : persistedSerDes;
+ }
+ if (codec instanceof ComposableSerDes) {
+ throw new IllegalArgumentException("inputSerDes must be a value codec, not a composable pipeline");
+ }
+ return codec;
+ }
+
/** Builder for DurableConfig. Provides fluent API for configuring SDK components. */
public static final class Builder {
private DurableExecutionClient durableExecutionClient;
private SerDes serDes;
+ private SerDes inputSerDes;
private ExecutorService executorService;
+ private ExecutorService serDesExecutorService;
private LoggerConfig loggerConfig;
private PollingStrategy pollingStrategy;
private Duration checkpointDelay;
@@ -381,6 +430,27 @@ public Builder withSerDes(SerDes serDes) {
return this;
}
+ /**
+ * Sets the context-free value codec used to deserialize the initial Lambda invocation payload.
+ *
+ *
The initial input codec is independent of the SerDes used for persisted execution payloads and must not be
+ * a {@link ComposableSerDes}. If not set, a plain persisted SerDes is reused, while a composable persisted
+ * SerDes contributes only its root value codec.
+ *
+ * @param inputSerDes initial invocation value codec
+ * @return this builder
+ * @throws NullPointerException if inputSerDes is null
+ * @throws IllegalArgumentException if inputSerDes is a composable pipeline
+ */
+ public Builder withInputSerDes(SerDes inputSerDes) {
+ inputSerDes = Objects.requireNonNull(inputSerDes, "inputSerDes cannot be null");
+ if (inputSerDes instanceof ComposableSerDes) {
+ throw new IllegalArgumentException("inputSerDes must be a value codec, not a composable pipeline");
+ }
+ this.inputSerDes = inputSerDes;
+ return this;
+ }
+
/**
* Sets a custom ExecutorService for running user-defined operations. If not set, a default cached thread pool
* will be created.
@@ -396,6 +466,22 @@ public Builder withExecutorService(ExecutorService executorService) {
return this;
}
+ /**
+ * Sets the executor used for customer SerDes calls and blocking payload storage I/O. If not set, SerDes calls
+ * execute inline on the calling thread.
+ *
+ *
This executor must be different from the user operation executor to prevent synchronous SerDes dispatch
+ * from deadlocking a saturated operation pool.
+ *
+ * @param executorService the dedicated SerDes executor
+ * @return this builder
+ */
+ public Builder withSerDesExecutorService(ExecutorService executorService) {
+ this.serDesExecutorService =
+ Objects.requireNonNull(executorService, "SerDes ExecutorService cannot be null");
+ return this;
+ }
+
/**
* Sets a custom LoggerConfig. If not set, defaults to suppressing replay logs.
*
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/InvokeConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/InvokeConfig.java
index e9dc7af24..5eeaf9ee4 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/config/InvokeConfig.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/config/InvokeConfig.java
@@ -13,11 +13,13 @@ public class InvokeConfig {
private final SerDes payloadSerDes;
private final SerDes resultSerDes;
private final String tenantId;
+ private final boolean usePersistedSerDesForPayload;
public InvokeConfig(Builder builder) {
this.payloadSerDes = builder.payloadSerDes;
this.resultSerDes = builder.resultSerDes;
this.tenantId = builder.tenantId;
+ this.usePersistedSerDesForPayload = builder.usePersistedSerDesForPayload;
}
public SerDes payloadSerDes() {
@@ -32,12 +34,17 @@ public String tenantId() {
return tenantId;
}
+ /** Returns whether the target should decode this invoke payload with its persisted SerDes pipeline. */
+ public boolean usePersistedSerDesForPayload() {
+ return usePersistedSerDesForPayload;
+ }
+
public static Builder builder() {
- return new Builder(null, null, null);
+ return new Builder(null, null, null, false);
}
public Builder toBuilder() {
- return new Builder(payloadSerDes, resultSerDes, tenantId);
+ return new Builder(payloadSerDes, resultSerDes, tenantId, usePersistedSerDesForPayload);
}
/** Builder for creating InvokeConfig instances. */
@@ -45,11 +52,14 @@ public static class Builder {
private SerDes payloadSerDes;
private SerDes resultSerDes;
private String tenantId;
+ private boolean usePersistedSerDesForPayload;
- private Builder(SerDes payloadSerDes, SerDes resultSerDes, String tenantId) {
+ private Builder(
+ SerDes payloadSerDes, SerDes resultSerDes, String tenantId, boolean usePersistedSerDesForPayload) {
this.payloadSerDes = payloadSerDes;
this.resultSerDes = resultSerDes;
this.tenantId = tenantId;
+ this.usePersistedSerDesForPayload = usePersistedSerDesForPayload;
}
/**
@@ -69,9 +79,13 @@ public Builder tenantId(String tenantId) {
/**
* Sets a custom serializer for the invoke operation payload.
*
- *
If not specified, the invoke operation will use the default SerDes configured for the handler. This allows
- * per-invoke customization of serialization behavior, useful for invoke operations that need special handling
- * (e.g., custom date formats, encryption, compression).
+ *
If not specified, the invoke operation uses the handler's context-free input codec by default, or its
+ * persisted SerDes when {@link #usePersistedSerDesForPayload(boolean)} is enabled. This method allows
+ * per-invoke customization of serialization behavior, useful for invoke operations that need special handling.
+ *
+ *
By default, the serialized value is sent unchanged and must match the target function's ordinary input
+ * wire format. When {@link #usePersistedSerDesForPayload(boolean)} is enabled, it must instead be compatible
+ * with the target durable handler's persisted SerDes.
*
* @param payloadSerDes the custom serializer to use, or null to use the default
* @return this builder for method chaining
@@ -81,6 +95,26 @@ public Builder payloadSerDes(SerDes payloadSerDes) {
return this;
}
+ /**
+ * Selects whether a compatible durable target should deserialize the invoke payload with its persisted SerDes
+ * pipeline.
+ *
+ *
This is disabled by default so standard Lambda functions, non-Java durable functions, and older Java SDK
+ * versions continue to receive the configured serialized payload unchanged. Enable it only when the target is a
+ * Java durable handler that supports the SDK's chained-invoke payload frame and configures a compatible
+ * persisted SerDes pipeline.
+ *
+ *
When enabled without an explicit {@link #payloadSerDes(SerDes)}, the caller's persisted SerDes is used.
+ * Otherwise, the caller's context-free input codec is the default payload serializer.
+ *
+ * @param enabled whether the compatible target should use its persisted SerDes pipeline
+ * @return this builder for method chaining
+ */
+ public Builder usePersistedSerDesForPayload(boolean enabled) {
+ this.usePersistedSerDesForPayload = enabled;
+ return this;
+ }
+
/**
* Sets a custom serializer for the invoke result.
*
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/context/DurableContextImpl.java b/sdk/src/main/java/software/amazon/lambda/durable/context/DurableContextImpl.java
index 0c79165ec..a5b19480f 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/context/DurableContextImpl.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/context/DurableContextImpl.java
@@ -177,9 +177,10 @@ public DurableFuture invokeAsync(
config = config.toBuilder().serDes(getDurableConfig().getSerDes()).build();
}
if (config.payloadSerDes() == null) {
- config = config.toBuilder()
- .payloadSerDes(getDurableConfig().getSerDes())
- .build();
+ var payloadSerDes = config.usePersistedSerDesForPayload()
+ ? getDurableConfig().getSerDes()
+ : getDurableConfig().getInputSerDes();
+ config = config.toBuilder().payloadSerDes(payloadSerDes).build();
}
var operationId = nextOperationId();
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/exception/CallbackException.java b/sdk/src/main/java/software/amazon/lambda/durable/exception/CallbackException.java
index a8fb6a011..92d745e0f 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/exception/CallbackException.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/exception/CallbackException.java
@@ -3,6 +3,7 @@
package software.amazon.lambda.durable.exception;
import software.amazon.awssdk.services.lambda.model.Operation;
+import software.amazon.lambda.durable.util.ExceptionHelper;
/** Thrown when a callback operation encounters an error. */
public class CallbackException extends DurableOperationException {
@@ -13,7 +14,20 @@ public CallbackException(Operation operation, String message) {
}
public CallbackException(Operation operation, String message, Throwable cause) {
- super(operation, operation.callbackDetails().error(), message, cause);
+ this(operation, message, cause, null);
+ }
+
+ protected CallbackException(Operation operation, String message, Throwable cause, Throwable deserializedError) {
+ super(
+ operation,
+ operation.callbackDetails().error(),
+ message,
+ operation.callbackDetails().error() != null
+ ? ExceptionHelper.deserializeStackTrace(
+ operation.callbackDetails().error().stackTrace())
+ : null,
+ cause,
+ deserializedError);
this.callbackId = operation.callbackDetails().callbackId();
}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/exception/CallbackFailedException.java b/sdk/src/main/java/software/amazon/lambda/durable/exception/CallbackFailedException.java
index e3fb9e177..d966bbc3f 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/exception/CallbackFailedException.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/exception/CallbackFailedException.java
@@ -8,7 +8,11 @@
/** Exception thrown when a callback fails due to an error from the external system. */
public class CallbackFailedException extends CallbackException {
public CallbackFailedException(Operation operation) {
- super(operation, buildMessage(operation.callbackDetails().error()));
+ this(operation, null);
+ }
+
+ public CallbackFailedException(Operation operation, Throwable deserializedError) {
+ super(operation, buildMessage(operation.callbackDetails().error()), deserializedError, deserializedError);
}
private static String buildMessage(ErrorObject error) {
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/exception/DurableOperationException.java b/sdk/src/main/java/software/amazon/lambda/durable/exception/DurableOperationException.java
index 73078ea1d..5eec194db 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/exception/DurableOperationException.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/exception/DurableOperationException.java
@@ -11,6 +11,7 @@
public class DurableOperationException extends DurableExecutionException {
private final Operation operation;
private final ErrorObject errorObject;
+ private final transient Throwable deserializedError;
public DurableOperationException(Operation operation, ErrorObject errorObject) {
this(operation, errorObject, errorObject != null ? errorObject.errorMessage() : null);
@@ -36,9 +37,20 @@ public DurableOperationException(
String errorMessage,
StackTraceElement[] stackTrace,
Throwable cause) {
+ this(operation, errorObject, errorMessage, stackTrace, cause, null);
+ }
+
+ protected DurableOperationException(
+ Operation operation,
+ ErrorObject errorObject,
+ String errorMessage,
+ StackTraceElement[] stackTrace,
+ Throwable cause,
+ Throwable deserializedError) {
super(errorMessage, cause, stackTrace);
this.operation = operation;
this.errorObject = errorObject;
+ this.deserializedError = deserializedError;
}
/** Returns the error details from the failed operation. */
@@ -60,4 +72,13 @@ public OperationStatus getOperationStatus() {
public String getOperationId() {
return operation.id();
}
+
+ /**
+ * Returns the original error reconstructed by the operation that produced this exception, when available.
+ *
+ * This is used internally when a child context forwards an operation failure through a different SerDes.
+ */
+ public Throwable deserializedError() {
+ return deserializedError;
+ }
}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/exception/InvokeException.java b/sdk/src/main/java/software/amazon/lambda/durable/exception/InvokeException.java
index 37bbf2ff9..e88a3a34e 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/exception/InvokeException.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/exception/InvokeException.java
@@ -3,14 +3,30 @@
package software.amazon.lambda.durable.exception;
import software.amazon.awssdk.services.lambda.model.Operation;
+import software.amazon.lambda.durable.util.ExceptionHelper;
/** Base exception for chained invoke operation failures. */
public class InvokeException extends DurableOperationException {
public InvokeException(Operation operation) {
+ this(operation, null);
+ }
+
+ protected InvokeException(Operation operation, Throwable deserializedError) {
super(
operation,
operation.chainedInvokeDetails() != null
? operation.chainedInvokeDetails().error()
- : null);
+ : null,
+ operation.chainedInvokeDetails() != null
+ && operation.chainedInvokeDetails().error() != null
+ ? operation.chainedInvokeDetails().error().errorMessage()
+ : null,
+ operation.chainedInvokeDetails() != null
+ && operation.chainedInvokeDetails().error() != null
+ ? ExceptionHelper.deserializeStackTrace(
+ operation.chainedInvokeDetails().error().stackTrace())
+ : null,
+ deserializedError,
+ deserializedError);
}
}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/exception/InvokeFailedException.java b/sdk/src/main/java/software/amazon/lambda/durable/exception/InvokeFailedException.java
index 45f84c341..9bd5b5e01 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/exception/InvokeFailedException.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/exception/InvokeFailedException.java
@@ -8,6 +8,10 @@
public class InvokeFailedException extends InvokeException {
public InvokeFailedException(Operation operation) {
- super(operation);
+ this(operation, null);
+ }
+
+ public InvokeFailedException(Operation operation, Throwable deserializedError) {
+ super(operation, deserializedError);
}
}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/exception/InvokeStoppedException.java b/sdk/src/main/java/software/amazon/lambda/durable/exception/InvokeStoppedException.java
index 01dd3e22c..7786e00f7 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/exception/InvokeStoppedException.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/exception/InvokeStoppedException.java
@@ -10,4 +10,8 @@ public class InvokeStoppedException extends InvokeException {
public InvokeStoppedException(Operation operation) {
super(operation);
}
+
+ public InvokeStoppedException(Operation operation, Throwable deserializedError) {
+ super(operation, deserializedError);
+ }
}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/exception/InvokeTimedOutException.java b/sdk/src/main/java/software/amazon/lambda/durable/exception/InvokeTimedOutException.java
index a0c36c623..df2241924 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/exception/InvokeTimedOutException.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/exception/InvokeTimedOutException.java
@@ -10,4 +10,8 @@ public class InvokeTimedOutException extends InvokeException {
public InvokeTimedOutException(Operation operation) {
super(operation);
}
+
+ public InvokeTimedOutException(Operation operation, Throwable deserializedError) {
+ super(operation, deserializedError);
+ }
}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/exception/RetryableSerDesException.java b/sdk/src/main/java/software/amazon/lambda/durable/exception/RetryableSerDesException.java
new file mode 100644
index 000000000..d40ad067a
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/exception/RetryableSerDesException.java
@@ -0,0 +1,20 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.exception;
+
+/**
+ * Indicates a transient serialization or deserialization failure that may succeed when retried.
+ *
+ *
{@link software.amazon.lambda.durable.serde.RetrySerDesStage} and
+ * {@link software.amazon.lambda.durable.serde.RetryBinarySerDesStage} retry only this exception type. Other
+ * {@link SerDesException} instances are treated as permanent failures.
+ */
+public class RetryableSerDesException extends SerDesException {
+ public RetryableSerDesException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public RetryableSerDesException(String message) {
+ super(message);
+ }
+}
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..f7928942b 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
@@ -12,7 +12,6 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.services.lambda.model.ErrorObject;
-import software.amazon.awssdk.services.lambda.model.Operation;
import software.amazon.awssdk.services.lambda.model.OperationAction;
import software.amazon.awssdk.services.lambda.model.OperationType;
import software.amazon.awssdk.services.lambda.model.OperationUpdate;
@@ -31,6 +30,9 @@
import software.amazon.lambda.durable.plugin.InvocationStatus;
import software.amazon.lambda.durable.plugin.PluginRunner;
import software.amazon.lambda.durable.serde.SerDes;
+import software.amazon.lambda.durable.serde.SerDesContext;
+import software.amazon.lambda.durable.serde.SerDesPayloadKind;
+import software.amazon.lambda.durable.serde.internal.ChainedInvokePayloadFrame;
import software.amazon.lambda.durable.util.ExceptionHelper;
/**
@@ -76,8 +78,7 @@ public static DurableExecutionOutput execute(
I userInput = null;
Throwable inputFailure = null;
try {
- userInput = extractUserInput(
- executionManager.getExecutionOperation(), config.getSerDes(), inputType);
+ userInput = extractUserInput(executionManager, config, inputType);
} catch (Throwable t) {
inputFailure = t;
}
@@ -159,11 +160,17 @@ public static DurableExecutionOutput execute(
cause,
pluginExecutionInput.get(),
null);
- return DurableExecutionOutput.failure(buildErrorObject(cause, config.getSerDes()));
+ return DurableExecutionOutput.failure(
+ buildErrorObject(cause, executionManager, config.getSerDes()));
}
// user handler complete successfully
logger.debug("Execution completed");
- var outputPayload = config.getSerDes().serialize(result);
+ var outputPayload = executionManager
+ .getSerDesRunner()
+ .serialize(
+ config.getSerDes(),
+ result,
+ executionContext(executionManager, SerDesPayloadKind.OUTPUT));
var output =
DurableExecutionOutput.success(handleLargePayload(executionManager, outputPayload));
fireOnInvocationEnd(
@@ -228,7 +235,7 @@ private static String handleLargePayload(ExecutionManager executionManager, Stri
return outputPayload;
}
- private static ErrorObject buildErrorObject(Throwable e, SerDes serDes) {
+ private static ErrorObject buildErrorObject(Throwable e, ExecutionManager executionManager, SerDes serDes) {
// exceptions thrown from operations, e.g. Step
if (e instanceof DurableOperationException durableOperationException) {
return durableOperationException.getErrorObject();
@@ -237,16 +244,39 @@ private static ErrorObject buildErrorObject(Throwable e, SerDes serDes) {
return unrecoverableDurableExecutionException.getErrorObject();
}
// exceptions thrown from non-operation code
- return ExceptionHelper.buildErrorObject(e, serDes);
+ return ErrorObject.builder()
+ .errorType(e.getClass().getName())
+ .errorMessage(e.getMessage())
+ .errorData(executionManager
+ .getSerDesRunner()
+ .serialize(serDes, e, executionContext(executionManager, SerDesPayloadKind.EXCEPTION)))
+ .stackTrace(ExceptionHelper.serializeStackTrace(e.getStackTrace()))
+ .build();
}
- private static I extractUserInput(Operation executionOp, SerDes serDes, TypeToken inputType) {
+ private static I extractUserInput(
+ ExecutionManager executionManager, DurableConfig config, TypeToken inputType) {
+ var executionOp = executionManager.getExecutionOperation();
if (executionOp.executionDetails() == null) {
throw new IllegalDurableOperationException("EXECUTION operation missing executionDetails");
}
var inputPayload = executionOp.executionDetails().inputPayload();
- return serDes.deserialize(inputPayload, inputType);
+ var serDes = config.getInputSerDes();
+ if (ChainedInvokePayloadFrame.isFramed(inputPayload)) {
+ inputPayload = ChainedInvokePayloadFrame.decode(inputPayload);
+ serDes = config.getSerDes();
+ }
+ return executionManager
+ .getSerDesRunner()
+ .deserialize(
+ serDes, inputPayload, inputType, executionContext(executionManager, SerDesPayloadKind.INPUT));
+ }
+
+ private static SerDesContext executionContext(ExecutionManager executionManager, SerDesPayloadKind payloadKind) {
+ var operation = executionManager.getExecutionOperation();
+ return SerDesContext.forExecution(
+ executionManager.getDurableExecutionArn(), operation.id(), operation.name(), payloadKind);
}
/**
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..6888ef931 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
@@ -29,6 +29,7 @@
import software.amazon.lambda.durable.model.SafeCloseable;
import software.amazon.lambda.durable.operation.BaseDurableOperation;
import software.amazon.lambda.durable.plugin.PluginInfoConverter;
+import software.amazon.lambda.durable.serde.SerDesRunner;
/**
* Central manager for durable execution coordination.
@@ -63,6 +64,7 @@ public class ExecutionManager implements SafeCloseable {
private final AtomicReference executionMode;
private final DurableConfig durableConfig;
private final Set updatedOperationIdsSinceLastInvocation;
+ private final SerDesRunner serDesRunner;
// ===== Thread Coordination =====
private final Map registeredOperations = new ConcurrentHashMap<>();
@@ -77,6 +79,7 @@ public ExecutionManager(DurableExecutionInput input, DurableConfig config, Conte
durableConfig = config;
this.durableExecutionArn = input.durableExecutionArn();
this.lambdaContext = lambdaContext;
+ this.serDesRunner = new SerDesRunner(config.getSerDesExecutorService());
// Store the set of operation IDs updated since the last successful invocation
this.updatedOperationIdsSinceLastInvocation =
@@ -115,6 +118,11 @@ public String getDurableExecutionArn() {
return durableExecutionArn;
}
+ /** Returns the invocation-scoped SerDes runner. */
+ public SerDesRunner getSerDesRunner() {
+ return serDesRunner;
+ }
+
/** Returns {@code true} if the execution is currently replaying completed operations. */
public boolean isReplaying() {
return executionMode.get() == ExecutionMode.REPLAY;
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..19e271fb9 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
@@ -30,6 +30,9 @@
import software.amazon.lambda.durable.plugin.PluginInfoConverter;
import software.amazon.lambda.durable.plugin.PluginRunner;
import software.amazon.lambda.durable.plugin.UserFunctionOutcome;
+import software.amazon.lambda.durable.serde.SerDesContext;
+import software.amazon.lambda.durable.serde.SerDesPayloadKind;
+import software.amazon.lambda.durable.serde.SerDesRunner;
import software.amazon.lambda.durable.util.ExceptionHelper;
/**
@@ -60,6 +63,7 @@ public abstract class BaseDurableOperation {
protected final boolean isVirtual;
protected final AtomicBoolean replayCompletedOperation = new AtomicBoolean(false);
private final DurableContextImpl durableContext;
+ private final SerDesRunner serDesRunner;
private final AtomicReference> runningUserHandler = new AtomicReference<>(null);
protected BaseDurableOperation(
@@ -86,6 +90,8 @@ protected BaseDurableOperation(
this.parentOperation = parentOperation;
this.durableContext = durableContext;
this.executionManager = durableContext.getExecutionManager();
+ var invocationSerDesRunner = executionManager.getSerDesRunner();
+ this.serDesRunner = invocationSerDesRunner != null ? invocationSerDesRunner : new SerDesRunner(null);
this.isVirtual = isVirtual;
this.completionFuture = new CompletableFuture<>();
@@ -118,6 +124,24 @@ protected DurableContextImpl getContext() {
return durableContext;
}
+ /** Builds the SerDes context for a payload owned by this operation. */
+ protected SerDesContext createSerDesContext(SerDesPayloadKind payloadKind, Integer attempt) {
+ return SerDesContext.forOperation(
+ executionManager.getDurableExecutionArn(),
+ getOperationId(),
+ getName(),
+ durableContext.getParentId(),
+ getType(),
+ getSubType(),
+ payloadKind,
+ attempt);
+ }
+
+ /** Returns the invocation-scoped SerDes runner. */
+ protected SerDesRunner getSerDesRunner() {
+ return serDesRunner;
+ }
+
/** Gets the operation type. */
public OperationType getType() {
return operationIdentifier.operationType();
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/CallbackOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/CallbackOperation.java
index 9d9481fb9..305b04fd9 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/operation/CallbackOperation.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/CallbackOperation.java
@@ -77,7 +77,9 @@ public T get() {
return switch (op.status()) {
case SUCCEEDED -> deserializeResult(op.callbackDetails().result());
- case FAILED -> throw new CallbackFailedException(op);
+ case FAILED ->
+ throw new CallbackFailedException(
+ op, deserializeException(op.callbackDetails().error()));
case TIMED_OUT -> throw new CallbackTimeoutException(op);
default ->
throw terminateExecutionWithIllegalDurableOperationException(
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/ChildContextOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/ChildContextOperation.java
index 8c299cfa4..d8d6b91db 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/operation/ChildContextOperation.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/ChildContextOperation.java
@@ -199,7 +199,7 @@ private void handleChildContextFailure(Throwable exception) {
final ErrorObject errorObject;
if (exception instanceof DurableOperationException opEx) {
- errorObject = opEx.getErrorObject();
+ errorObject = rebindForwardedException(opEx);
} else {
errorObject = serializeException(exception);
}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/InvokeOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/InvokeOperation.java
index 9e2c54ace..9d60609e9 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/operation/InvokeOperation.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/InvokeOperation.java
@@ -15,6 +15,8 @@
import software.amazon.lambda.durable.exception.InvokeTimedOutException;
import software.amazon.lambda.durable.model.OperationIdentifier;
import software.amazon.lambda.durable.serde.SerDes;
+import software.amazon.lambda.durable.serde.SerDesPayloadKind;
+import software.amazon.lambda.durable.serde.internal.ChainedInvokePayloadFrame;
/**
* Durable operation that invokes another Lambda function and waits for its result.
@@ -64,13 +66,18 @@ protected void replay(Operation existing) {
}
private void startInvocation() {
+ var serializedPayload = getSerDesRunner()
+ .serialize(payloadSerDes, this.payload, createSerDesContext(SerDesPayloadKind.INVOKE_PAYLOAD, null));
var update = OperationUpdate.builder()
.action(OperationAction.START)
.chainedInvokeOptions(ChainedInvokeOptions.builder()
.functionName(functionName)
.tenantId(invokeConfig.tenantId())
.build())
- .payload(payloadSerDes.serialize(this.payload));
+ .payload(
+ invokeConfig.usePersistedSerDesForPayload()
+ ? ChainedInvokePayloadFrame.encode(serializedPayload)
+ : serializedPayload);
sendOperationUpdate(update);
}
@@ -85,11 +92,12 @@ public T get() {
var op = waitForOperationCompletion();
var invokeDetails = op.chainedInvokeDetails();
var result = invokeDetails != null ? invokeDetails.result() : null;
+ var error = invokeDetails != null ? invokeDetails.error() : null;
return switch (op.status()) {
case SUCCEEDED -> deserializeResult(result);
- case FAILED -> throw new InvokeFailedException(op);
- case TIMED_OUT -> throw new InvokeTimedOutException(op);
- case STOPPED -> throw new InvokeStoppedException(op);
+ case FAILED -> throw new InvokeFailedException(op, deserializeException(error));
+ case TIMED_OUT -> throw new InvokeTimedOutException(op, deserializeException(error));
+ case STOPPED -> throw new InvokeStoppedException(op, deserializeException(error));
// Unexpected status which should not happen. This is added for forward-compatibility.
default -> throw new InvokeException(op);
};
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/SerializableDurableOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/SerializableDurableOperation.java
index 6457c996d..5c3552bcf 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/operation/SerializableDurableOperation.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/SerializableDurableOperation.java
@@ -8,9 +8,13 @@
import software.amazon.lambda.durable.DurableFuture;
import software.amazon.lambda.durable.TypeToken;
import software.amazon.lambda.durable.context.DurableContextImpl;
+import software.amazon.lambda.durable.exception.DurableOperationException;
+import software.amazon.lambda.durable.exception.RetryableSerDesException;
import software.amazon.lambda.durable.exception.SerDesException;
import software.amazon.lambda.durable.model.OperationIdentifier;
import software.amazon.lambda.durable.serde.SerDes;
+import software.amazon.lambda.durable.serde.SerDesContext;
+import software.amazon.lambda.durable.serde.SerDesPayloadKind;
import software.amazon.lambda.durable.util.ExceptionHelper;
/**
@@ -85,8 +89,14 @@ protected SerializableDurableOperation(
* @throws SerDesException if deserialization fails
*/
protected T deserializeResult(String result) {
+ return deserializeResult(result, SerDesPayloadKind.RESULT, null);
+ }
+
+ /** Deserializes a result with explicit payload kind and attempt metadata. */
+ protected T deserializeResult(String result, SerDesPayloadKind payloadKind, Integer attempt) {
try {
- return resultSerDes.deserialize(result, resultTypeToken);
+ return getSerDesRunner()
+ .deserialize(resultSerDes, result, resultTypeToken, createSerDesContext(payloadKind, attempt));
} catch (SerDesException e) {
logger.warn(
"Failed to deserialize {} result for operation name '{}'. Ensure the result is properly encoded.",
@@ -106,8 +116,17 @@ protected T deserializeResult(String result) {
* @return the serialized string and the deserialized result
*/
protected SerializedResult serializeAndDeserializeResult(T result) {
- var serialized = resultSerDes.serialize(result);
- var deserialized = shouldDeserializeAfterSerialization() ? deserializeResult(serialized) : result;
+ return serializeAndDeserializeResult(result, SerDesPayloadKind.RESULT, null);
+ }
+
+ /** Serializes a result with explicit payload kind and attempt metadata. */
+ protected SerializedResult serializeAndDeserializeResult(
+ T result, SerDesPayloadKind payloadKind, Integer attempt) {
+ var context = createSerDesContext(payloadKind, attempt);
+ var serialized = getSerDesRunner().serialize(resultSerDes, result, context);
+ var deserialized = shouldDeserializeAfterSerialization()
+ ? getSerDesRunner().deserialize(resultSerDes, serialized, resultTypeToken, context)
+ : result;
return new SerializedResult<>(serialized, deserialized);
}
@@ -119,13 +138,52 @@ protected SerializedResult serializeAndDeserializeResult(T result) {
*/
@SuppressWarnings("ThrowableNotThrown")
protected ErrorObject serializeException(Throwable throwable) {
- var error = ExceptionHelper.buildErrorObject(throwable, resultSerDes);
+ return serializeException(throwable, null);
+ }
+
+ /** Serializes a throwable with attempt metadata. */
+ protected ErrorObject serializeException(Throwable throwable, Integer attempt) {
+ var context = createSerDesContext(SerDesPayloadKind.EXCEPTION, attempt);
+ var error = ErrorObject.builder()
+ .errorType(throwable.getClass().getName())
+ .errorMessage(throwable.getMessage())
+ .errorData(getSerDesRunner().serialize(resultSerDes, throwable, context))
+ .stackTrace(ExceptionHelper.serializeStackTrace(throwable.getStackTrace()))
+ .build();
if (shouldDeserializeAfterSerialization()) {
- deserializeException(error);
+ deserializeException(error, attempt);
}
return error;
}
+ /**
+ * Re-serializes an exception forwarded from another durable operation under this operation's context.
+ *
+ * Context-dependent SerDes implementations may store the source error data under the producing operation or
+ * invoked execution. Rebinding reconstructable exceptions prevents a parent checkpoint from later trying to read
+ * that data using the parent's unrelated entity identity.
+ */
+ protected ErrorObject rebindForwardedException(DurableOperationException exception) {
+ return rebindForwardedException(exception, null);
+ }
+
+ /**
+ * Re-serializes an exception forwarded from another durable operation under this operation's attempt context.
+ *
+ * @param exception the forwarded durable operation exception
+ * @param attempt the receiving operation's attempt, or {@code null} when attempts do not apply
+ * @return error data owned by this operation when the original exception can be reconstructed; otherwise the
+ * forwarded error data
+ */
+ protected ErrorObject rebindForwardedException(DurableOperationException exception, Integer attempt) {
+ var error = exception.getErrorObject();
+ if (error == null || exception.getOperation() == null) {
+ return error;
+ }
+ var original = exception.deserializedError();
+ return original != null ? serializeException(original, attempt) : error;
+ }
+
private boolean shouldDeserializeAfterSerialization() {
var config = getContext().getDurableConfig();
return config == null || config.shouldDeserializeAfterSerialization();
@@ -139,6 +197,15 @@ private boolean shouldDeserializeAfterSerialization() {
* @return the reconstructed throwable, or null if reconstruction is not possible
*/
protected Throwable deserializeException(ErrorObject errorObject) {
+ return deserializeException(errorObject, null);
+ }
+
+ /** Deserializes a throwable with attempt metadata. */
+ protected Throwable deserializeException(ErrorObject errorObject, Integer attempt) {
+ return deserializeExceptionWithContext(errorObject, createSerDesContext(SerDesPayloadKind.EXCEPTION, attempt));
+ }
+
+ private Throwable deserializeExceptionWithContext(ErrorObject errorObject, SerDesContext context) {
Throwable original = null;
if (errorObject == null) {
return original;
@@ -153,8 +220,12 @@ protected Throwable deserializeException(ErrorObject errorObject) {
Class exceptionClass = Class.forName(errorType);
if (Throwable.class.isAssignableFrom(exceptionClass)) {
- original =
- resultSerDes.deserialize(errorData, TypeToken.get(exceptionClass.asSubclass(Throwable.class)));
+ original = getSerDesRunner()
+ .deserialize(
+ resultSerDes,
+ errorData,
+ TypeToken.get(exceptionClass.asSubclass(Throwable.class)),
+ context);
if (original != null) {
original.setStackTrace(ExceptionHelper.deserializeStackTrace(errorObject.stackTrace()));
@@ -162,6 +233,8 @@ protected Throwable deserializeException(ErrorObject errorObject) {
}
} catch (ClassNotFoundException e) {
logger.warn("Cannot re-construct original exception type. Falling back to generic StepFailedException.");
+ } catch (RetryableSerDesException e) {
+ throw e;
} catch (SerDesException e) {
logger.warn("Cannot deserialize original exception data. Falling back to generic StepFailedException.", e);
}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/StepOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/StepOperation.java
index 467a87b94..52cfdb98d 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/operation/StepOperation.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/StepOperation.java
@@ -25,6 +25,7 @@
import software.amazon.lambda.durable.execution.ThreadType;
import software.amazon.lambda.durable.logging.DurableLogger;
import software.amazon.lambda.durable.model.OperationIdentifier;
+import software.amazon.lambda.durable.serde.SerDesPayloadKind;
import software.amazon.lambda.durable.util.ExceptionHelper;
/**
@@ -117,7 +118,7 @@ private void executeStepLogic(int attempt) {
// through onUserFunctionEnd; retry/checkpoint handling stays outside the boundary.
T result = runUserFunction(attempt, () -> function.apply(stepContext));
- handleStepSucceeded(result);
+ handleStepSucceeded(result, attempt);
} catch (Throwable e) {
handleStepFailure(e, attempt);
}
@@ -144,8 +145,8 @@ private void checkpointStarted() {
}
}
- private void handleStepSucceeded(T result) {
- var serializedResult = serializeAndDeserializeResult(result);
+ private void handleStepSucceeded(T result, int attempt) {
+ var serializedResult = serializeAndDeserializeResult(result, SerDesPayloadKind.RESULT, attempt);
// Send SUCCEED
var successUpdate =
@@ -168,9 +169,9 @@ private void handleStepFailure(Throwable exception, int attempt) {
final ErrorObject errorObject;
if (exception instanceof DurableOperationException durableOperationException) {
- errorObject = durableOperationException.getErrorObject();
+ errorObject = rebindForwardedException(durableOperationException, attempt);
} else {
- errorObject = serializeException(exception);
+ errorObject = serializeException(exception, attempt);
}
var retryDecision = config.retryStrategy().makeRetryDecision(exception, attempt);
@@ -205,8 +206,9 @@ public T get() {
if (op.status() == OperationStatus.SUCCEEDED) {
var stepDetails = op.stepDetails();
var result = (stepDetails != null) ? stepDetails.result() : null;
+ var attempt = stepDetails != null ? stepDetails.attempt() : null;
- return deserializeResult(result);
+ return deserializeResult(result, SerDesPayloadKind.RESULT, attempt);
} else {
var errorObject = op.stepDetails().error();
@@ -216,7 +218,8 @@ public T get() {
}
// Attempt to reconstruct and throw the original exception
- Throwable original = deserializeException(errorObject);
+ var attempt = op.stepDetails() != null ? op.stepDetails().attempt() : null;
+ Throwable original = deserializeException(errorObject, attempt);
if (original != null) {
ExceptionHelper.sneakyThrow(original);
}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/WaitForConditionOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/WaitForConditionOperation.java
index 6a653c37b..b3a51c7c7 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/operation/WaitForConditionOperation.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/WaitForConditionOperation.java
@@ -23,6 +23,7 @@
import software.amazon.lambda.durable.logging.DurableLogger;
import software.amazon.lambda.durable.model.OperationIdentifier;
import software.amazon.lambda.durable.model.WaitForConditionResult;
+import software.amazon.lambda.durable.serde.SerDesPayloadKind;
import software.amazon.lambda.durable.util.ExceptionHelper;
/**
@@ -76,12 +77,14 @@ public T get() {
if (op.status() == OperationStatus.SUCCEEDED) {
var stepDetails = op.stepDetails();
var result = (stepDetails != null) ? stepDetails.result() : null;
- return deserializeResult(result);
+ var attempt = stepDetails != null ? stepDetails.attempt() : null;
+ return deserializeResult(result, SerDesPayloadKind.STATE, attempt);
} else {
var errorObject = op.stepDetails().error();
// Attempt to reconstruct and throw the original exception
- Throwable original = deserializeException(errorObject);
+ var attempt = op.stepDetails() != null ? op.stepDetails().attempt() : null;
+ Throwable original = deserializeException(errorObject, attempt);
if (original != null) {
ExceptionHelper.sneakyThrow(original);
}
@@ -97,7 +100,7 @@ private void resumeCheckLoop(Operation existing) {
var checkpointData = stepDetails != null ? stepDetails.result() : null;
T currentState; // Get current state
if (checkpointData != null) {
- currentState = deserializeResult(checkpointData);
+ currentState = deserializeResult(checkpointData, SerDesPayloadKind.STATE, attempt - 1);
} else {
currentState = config.initialState();
}
@@ -131,7 +134,8 @@ private void executeCheckLogic(T currentState, int attempt) {
runUserFunction(attempt, () -> checkFunc.apply(currentState, stepContext));
// Normalize the value through SerDes so first execution matches replay.
- var serializedState = serializeAndDeserializeResult(result.value());
+ var serializedState =
+ serializeAndDeserializeResult(result.value(), SerDesPayloadKind.STATE, attempt);
T deserializedValue = serializedState.deserialized();
if (result.isDone()) {
@@ -161,7 +165,7 @@ private void executeCheckLogic(T currentState, int attempt) {
.thenRun(() -> executeCheckLogic(deserializedValue, attempt + 1));
}
} catch (Throwable e) {
- handleCheckFailure(e);
+ handleCheckFailure(e, attempt);
}
}
};
@@ -169,7 +173,7 @@ private void executeCheckLogic(T currentState, int attempt) {
runUserHandler(userHandler, ThreadType.STEP);
}
- private void handleCheckFailure(Throwable exception) {
+ private void handleCheckFailure(Throwable exception, int attempt) {
exception = ExceptionHelper.unwrapCompletableFuture(exception);
if (exception instanceof SuspendExecutionException suspendExecutionException) {
throw suspendExecutionException;
@@ -179,8 +183,8 @@ private void handleCheckFailure(Throwable exception) {
}
final var errorObject = (exception instanceof DurableOperationException durableOpEx)
- ? durableOpEx.getErrorObject()
- : serializeException(exception);
+ ? rebindForwardedException(durableOpEx, attempt)
+ : serializeException(exception, attempt);
// Checkpoint FAIL
var failUpdate = OperationUpdate.builder().action(OperationAction.FAIL).error(errorObject);
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/serde/Base64StringBinaryCodec.java b/sdk/src/main/java/software/amazon/lambda/durable/serde/Base64StringBinaryCodec.java
new file mode 100644
index 000000000..6b4812c84
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/serde/Base64StringBinaryCodec.java
@@ -0,0 +1,22 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.serde;
+
+import java.util.Base64;
+
+/** Converts bytes to and from standard Base64 strings. */
+public final class Base64StringBinaryCodec implements StringBinaryCodec {
+ public static final Base64StringBinaryCodec INSTANCE = new Base64StringBinaryCodec();
+
+ private Base64StringBinaryCodec() {}
+
+ @Override
+ public byte[] toBytes(String value) {
+ return Base64.getDecoder().decode(value);
+ }
+
+ @Override
+ public String fromBytes(byte[] data) {
+ return Base64.getEncoder().encodeToString(data);
+ }
+}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/serde/BinarySerDesStage.java b/sdk/src/main/java/software/amazon/lambda/durable/serde/BinarySerDesStage.java
new file mode 100644
index 000000000..4baac5d31
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/serde/BinarySerDesStage.java
@@ -0,0 +1,34 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.serde;
+
+/**
+ * A reversible binary stage used inside a {@link ComposableBinarySerDesStage}.
+ *
+ *
Implementations must include any metadata needed for deserialization, such as format versions or encryption
+ * initialization vectors, in the returned bytes.
+ *
+ *
The enclosing composable stage passes the same durable payload context to each binary stage. During serialization,
+ * {@link SerDesContext#originalValue()} is the object supplied to the root value codec. During deserialization it is
+ * {@code null}. Stages must treat the original value as read-only. The context itself may be {@code null} only when the
+ * stage is invoked outside an SDK-managed SerDes call.
+ */
+public interface BinarySerDesStage {
+ /**
+ * Applies this transformation during forward serialization.
+ *
+ * @param value the non-null input bytes
+ * @param context the current durable payload context, or {@code null} outside SDK-managed calls
+ * @return the non-null transformed bytes
+ */
+ byte[] serialize(byte[] value, SerDesContext context);
+
+ /**
+ * Reverses this transformation during deserialization.
+ *
+ * @param data the non-null bytes produced by this transformation
+ * @param context the current durable payload context, or {@code null} outside SDK-managed calls
+ * @return the non-null bytes expected by the preceding transformation
+ */
+ byte[] deserialize(byte[] data, SerDesContext context);
+}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/serde/ComposableBinarySerDesStage.java b/sdk/src/main/java/software/amazon/lambda/durable/serde/ComposableBinarySerDesStage.java
new file mode 100644
index 000000000..bd4ef868f
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/serde/ComposableBinarySerDesStage.java
@@ -0,0 +1,183 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.serde;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import software.amazon.lambda.durable.exception.RetryableSerDesException;
+import software.amazon.lambda.durable.exception.SerDesException;
+
+/**
+ * A string SerDes stage containing an ordered chain of binary transformations.
+ *
+ *
Serialization converts the input string with the starting codec, applies binary stages in declaration order,
+ * converts the final bytes to a string with the ending codec, and adds a versioned frame. Deserialization reverses the
+ * complete process when that frame is present and passes unrecognized input through unchanged. The context supplied to
+ * this string stage is forwarded unchanged to every binary stage.
+ */
+public final class ComposableBinarySerDesStage implements SerDesStage {
+ private static final String FRAME_MARKER = "__durable_execution_composable_binary_serdes:";
+ private static final String FRAME_PREFIX = FRAME_MARKER + "1:";
+
+ private final StringBinaryCodec startingCodec;
+ private final List binaryStages;
+ private final StringBinaryCodec endingCodec;
+
+ private ComposableBinarySerDesStage(
+ StringBinaryCodec startingCodec, List binaryStages, StringBinaryCodec endingCodec) {
+ this.startingCodec = startingCodec;
+ this.binaryStages = List.copyOf(binaryStages);
+ this.endingCodec = endingCodec;
+ }
+
+ /** Creates a builder whose methods follow forward serialization order. */
+ public static StartBuilder builder() {
+ return new Builder();
+ }
+
+ @Override
+ public String serialize(String value, SerDesContext context) {
+ Objects.requireNonNull(value, "value cannot be null");
+ var current = invokeToBytes(startingCodec, value, "starting codec");
+ for (int index = 0; index < binaryStages.size(); index++) {
+ current = invokeSerialize(binaryStages.get(index), current, context, index);
+ }
+ return FRAME_PREFIX + invokeFromBytes(endingCodec, current, "ending codec");
+ }
+
+ @Override
+ public String deserialize(String data, SerDesContext context) {
+ Objects.requireNonNull(data, "data cannot be null");
+ if (!data.startsWith(FRAME_MARKER)) {
+ return data;
+ }
+ if (!data.startsWith(FRAME_PREFIX)) {
+ throw new SerDesException("Unsupported or malformed composable binary SerDes frame");
+ }
+ var current = invokeToBytes(endingCodec, data.substring(FRAME_PREFIX.length()), "ending codec");
+ for (int index = binaryStages.size() - 1; index >= 0; index--) {
+ current = invokeDeserialize(binaryStages.get(index), current, context, index);
+ }
+ return invokeFromBytes(startingCodec, current, "starting codec");
+ }
+
+ private static byte[] invokeToBytes(StringBinaryCodec codec, String value, String name) {
+ try {
+ return requireResult(codec.toBytes(value), name);
+ } catch (Throwable failure) {
+ throw componentFailure(name, "convert string to bytes", failure);
+ }
+ }
+
+ private static String invokeFromBytes(StringBinaryCodec codec, byte[] data, String name) {
+ try {
+ return requireResult(codec.fromBytes(data), name);
+ } catch (Throwable failure) {
+ throw componentFailure(name, "convert bytes to string", failure);
+ }
+ }
+
+ private static byte[] invokeSerialize(BinarySerDesStage stage, byte[] value, SerDesContext context, int index) {
+ try {
+ return requireResult(stage.serialize(value, context), binaryStageName(index, stage));
+ } catch (Throwable failure) {
+ throw componentFailure(binaryStageName(index, stage), "serialize", failure);
+ }
+ }
+
+ private static byte[] invokeDeserialize(BinarySerDesStage stage, byte[] data, SerDesContext context, int index) {
+ try {
+ return requireResult(stage.deserialize(data, context), binaryStageName(index, stage));
+ } catch (Throwable failure) {
+ throw componentFailure(binaryStageName(index, stage), "deserialize", failure);
+ }
+ }
+
+ private static T requireResult(T result, String component) {
+ if (result == null) {
+ throw new SerDesException(component + " returned null for non-null input");
+ }
+ return result;
+ }
+
+ private static String binaryStageName(int index, BinarySerDesStage stage) {
+ return String.format("binary stage %d (%s)", index, stage.getClass().getName());
+ }
+
+ private static RuntimeException componentFailure(String component, String action, Throwable failure) {
+ if (failure instanceof Error error) {
+ throw error;
+ }
+ var message = String.format("Composable binary SerDes stage %s failed to %s", component, action);
+ if (failure instanceof RetryableSerDesException) {
+ return new RetryableSerDesException(message, failure);
+ }
+ return new SerDesException(message, failure);
+ }
+
+ /** Builder stage that requires the starting string/binary codec. */
+ public interface StartBuilder {
+ /**
+ * Sets the codec that converts the input string to bytes during serialization.
+ *
+ * @param codec the starting boundary codec
+ * @return the binary-stage builder
+ */
+ BinaryStagesBuilder startWith(StringBinaryCodec codec);
+ }
+
+ /** Builder stage that accepts binary stages in processing order. */
+ public interface BinaryStagesBuilder {
+ /**
+ * Appends a binary transformation.
+ *
+ * @param stage the binary stage
+ * @return this builder stage
+ */
+ BinaryStagesBuilder then(BinarySerDesStage stage);
+
+ /**
+ * Sets the codec that converts the final bytes to a string during serialization.
+ *
+ * @param codec the ending boundary codec
+ * @return the completed builder
+ */
+ CompletedBuilder endWith(StringBinaryCodec codec);
+ }
+
+ /** Builder stage that permits only construction of the completed binary pipeline. */
+ public interface CompletedBuilder {
+ /** Returns the immutable string stage. */
+ ComposableBinarySerDesStage build();
+ }
+
+ private static final class Builder implements StartBuilder, BinaryStagesBuilder, CompletedBuilder {
+ private StringBinaryCodec startingCodec;
+ private final List binaryStages = new ArrayList<>();
+ private StringBinaryCodec endingCodec;
+
+ @Override
+ public BinaryStagesBuilder startWith(StringBinaryCodec codec) {
+ startingCodec = Objects.requireNonNull(codec, "starting codec cannot be null");
+ return this;
+ }
+
+ @Override
+ public BinaryStagesBuilder then(BinarySerDesStage stage) {
+ binaryStages.add(Objects.requireNonNull(stage, "binary stage cannot be null"));
+ return this;
+ }
+
+ @Override
+ public CompletedBuilder endWith(StringBinaryCodec codec) {
+ endingCodec = Objects.requireNonNull(codec, "ending codec cannot be null");
+ return this;
+ }
+
+ @Override
+ public ComposableBinarySerDesStage build() {
+ return new ComposableBinarySerDesStage(startingCodec, binaryStages, endingCodec);
+ }
+ }
+}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/serde/ComposableSerDes.java b/sdk/src/main/java/software/amazon/lambda/durable/serde/ComposableSerDes.java
new file mode 100644
index 000000000..385ae8bd5
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/serde/ComposableSerDes.java
@@ -0,0 +1,190 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.serde;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import software.amazon.lambda.durable.TypeToken;
+import software.amazon.lambda.durable.exception.RetryableSerDesException;
+import software.amazon.lambda.durable.exception.SerDesException;
+
+/**
+ * An immutable SerDes processing pipeline.
+ *
+ * The first component is the value codec. Every later component is a {@link SerDesStage} that consumes and produces
+ * a string. Serialization runs from first to last; deserialization runs from last to first. Each stage returns
+ * unrecognized input unchanged, allowing raw values to pass through to the root value codec. SDK-managed calls pass the
+ * same {@link SerDesContext} explicitly to every stage. During serialization that context also exposes the original
+ * object supplied to the value codec.
+ */
+public final class ComposableSerDes implements SerDes {
+ private final SerDes valueCodec;
+ private final List stages;
+
+ private ComposableSerDes(SerDes valueCodec, List stages) {
+ this.valueCodec = Objects.requireNonNull(valueCodec, "valueCodec cannot be null");
+ this.stages = List.copyOf(stages);
+ }
+
+ /**
+ * Creates a pipeline with a value codec followed by zero or more string stages.
+ *
+ * @param valueCodec the value codec
+ * @param remaining reversible string stages
+ * @return an immutable pipeline
+ */
+ public static ComposableSerDes of(SerDes valueCodec, SerDesStage... remaining) {
+ Objects.requireNonNull(remaining, "remaining stages cannot be null");
+ valueCodec = Objects.requireNonNull(valueCodec, "valueCodec cannot be null");
+ var stages = new ArrayList();
+ if (valueCodec instanceof ComposableSerDes composable) {
+ valueCodec = composable.valueCodec;
+ stages.addAll(composable.stages);
+ }
+ for (var stage : remaining) {
+ stages.add(Objects.requireNonNull(stage, "pipeline stage cannot be null"));
+ }
+ return new ComposableSerDes(valueCodec, stages);
+ }
+
+ /**
+ * Creates a pipeline builder.
+ *
+ * @param valueCodec the value codec which converts values to and from strings
+ * @return a new builder
+ */
+ public static Builder builder(SerDes valueCodec) {
+ return new Builder(valueCodec);
+ }
+
+ /** Returns the value codec at the start of this pipeline. */
+ public SerDes getValueCodec() {
+ return valueCodec;
+ }
+
+ /** Returns a new pipeline with the supplied string stage appended. */
+ @Override
+ public SerDes then(SerDesStage stage) {
+ var combined = new ArrayList<>(stages);
+ combined.add(Objects.requireNonNull(stage, "stage cannot be null"));
+ return new ComposableSerDes(valueCodec, combined);
+ }
+
+ @Override
+ public String serialize(Object value) {
+ return serialize(value, null);
+ }
+
+ String serialize(Object value, SerDesContext context) {
+ if (value == null) {
+ return null;
+ }
+ var stageContext = context == null ? null : context.withOriginalValue(value);
+ String current = invokeValueCodecSerialize(valueCodec, value);
+ for (int index = 0; index < stages.size(); index++) {
+ current = invokeStageSerialize(stages.get(index), current, stageContext, index + 1);
+ }
+ return current;
+ }
+
+ @Override
+ public T deserialize(String data, TypeToken typeToken) {
+ return deserialize(data, typeToken, null);
+ }
+
+ T deserialize(String data, TypeToken typeToken, SerDesContext context) {
+ if (data == null) {
+ return null;
+ }
+ Objects.requireNonNull(typeToken, "typeToken cannot be null");
+ var stageContext = context == null ? null : context.withOriginalValue(null);
+ String current = data;
+ for (int index = stages.size() - 1; index >= 0; index--) {
+ current = invokeStageDeserialize(stages.get(index), current, stageContext, index + 1);
+ }
+ return invokeValueCodecDeserialize(valueCodec, current, typeToken);
+ }
+
+ private static String invokeStageSerialize(SerDesStage stage, String value, SerDesContext context, int index) {
+ try {
+ var result = stage.serialize(value, context);
+ if (result == null) {
+ throw new SerDesException("Stage returned null for a non-null value");
+ }
+ return result;
+ } catch (Throwable failure) {
+ throw stageFailure(index, stage, "serialize", failure);
+ }
+ }
+
+ private static String invokeStageDeserialize(SerDesStage stage, String data, SerDesContext context, int index) {
+ try {
+ var result = stage.deserialize(data, context);
+ if (result == null) {
+ throw new SerDesException("Stage returned null for non-null input");
+ }
+ return result;
+ } catch (Throwable failure) {
+ throw stageFailure(index, stage, "deserialize", failure);
+ }
+ }
+
+ private static String invokeValueCodecSerialize(SerDes valueCodec, Object value) {
+ try {
+ var result = valueCodec.serialize(value);
+ if (result == null) {
+ throw new SerDesException("Value codec returned null for a non-null value");
+ }
+ return result;
+ } catch (Throwable failure) {
+ throw stageFailure(0, valueCodec, "serialize", failure);
+ }
+ }
+
+ private static T invokeValueCodecDeserialize(SerDes valueCodec, String data, TypeToken typeToken) {
+ try {
+ return valueCodec.deserialize(data, typeToken);
+ } catch (Throwable failure) {
+ throw stageFailure(0, valueCodec, "deserialize", failure);
+ }
+ }
+
+ private static RuntimeException stageFailure(int index, Object stage, String action, Throwable failure) {
+ if (failure instanceof Error error) {
+ throw error;
+ }
+ var message = String.format(
+ "SerDes pipeline stage %d (%s) failed to %s",
+ index, stage.getClass().getName(), action);
+ if (failure instanceof RetryableSerDesException) {
+ return new RetryableSerDesException(message, failure);
+ }
+ return new SerDesException(message, failure);
+ }
+
+ /** Builder for an immutable {@link ComposableSerDes}. */
+ public static final class Builder {
+ private SerDes valueCodec;
+ private final List stages = new ArrayList<>();
+
+ private Builder(SerDes valueCodec) {
+ this.valueCodec = Objects.requireNonNull(valueCodec, "valueCodec cannot be null");
+ if (valueCodec instanceof ComposableSerDes composable) {
+ this.valueCodec = composable.valueCodec;
+ stages.addAll(composable.stages);
+ }
+ }
+
+ /** Appends a reversible string stage. */
+ public Builder then(SerDesStage stage) {
+ stages.add(Objects.requireNonNull(stage, "stage cannot be null"));
+ return this;
+ }
+
+ /** Returns the immutable pipeline. */
+ public ComposableSerDes build() {
+ return new ComposableSerDes(valueCodec, stages);
+ }
+ }
+}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/serde/RetryBinarySerDesStage.java b/sdk/src/main/java/software/amazon/lambda/durable/serde/RetryBinarySerDesStage.java
new file mode 100644
index 000000000..301791ce9
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/serde/RetryBinarySerDesStage.java
@@ -0,0 +1,48 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.serde;
+
+import java.util.Objects;
+import software.amazon.lambda.durable.exception.RetryableSerDesException;
+import software.amazon.lambda.durable.retry.RetryStrategy;
+
+/**
+ * A binary-stage decorator that retries transient failures from another {@link BinarySerDesStage}.
+ *
+ * Only {@link RetryableSerDesException} is retried. Other failures are propagated immediately. Retry delays block
+ * the thread executing the SerDes call: the caller by default or the configured SerDes executor thread. Every attempt
+ * receives the same {@link SerDesContext} supplied to this decorator and a fresh copy of the original input bytes.
+ */
+public final class RetryBinarySerDesStage implements BinarySerDesStage {
+ private final BinarySerDesStage delegate;
+ private final SerDesRetryExecutor retryExecutor;
+
+ /**
+ * Creates a retrying binary-stage decorator.
+ *
+ * @param delegate the stage to invoke
+ * @param retryStrategy strategy that controls attempts and delays
+ */
+ public RetryBinarySerDesStage(BinarySerDesStage delegate, RetryStrategy retryStrategy) {
+ this(delegate, retryStrategy, SerDesRetryExecutor.DEFAULT_SLEEPER);
+ }
+
+ RetryBinarySerDesStage(
+ BinarySerDesStage delegate, RetryStrategy retryStrategy, SerDesRetryExecutor.Sleeper sleeper) {
+ this.delegate = Objects.requireNonNull(delegate, "delegate cannot be null");
+ retryExecutor = new SerDesRetryExecutor(retryStrategy, sleeper);
+ }
+
+ @Override
+ public byte[] serialize(byte[] value, SerDesContext context) {
+ var snapshot = Objects.requireNonNull(value, "value cannot be null").clone();
+ return retryExecutor.execute("binary stage serialization", () -> delegate.serialize(snapshot.clone(), context));
+ }
+
+ @Override
+ public byte[] deserialize(byte[] data, SerDesContext context) {
+ var snapshot = Objects.requireNonNull(data, "data cannot be null").clone();
+ return retryExecutor.execute(
+ "binary stage deserialization", () -> delegate.deserialize(snapshot.clone(), context));
+ }
+}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/serde/RetrySerDesStage.java b/sdk/src/main/java/software/amazon/lambda/durable/serde/RetrySerDesStage.java
new file mode 100644
index 000000000..82698e901
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/serde/RetrySerDesStage.java
@@ -0,0 +1,44 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.serde;
+
+import java.util.Objects;
+import software.amazon.lambda.durable.exception.RetryableSerDesException;
+import software.amazon.lambda.durable.retry.RetryStrategy;
+
+/**
+ * A string-stage decorator that retries transient failures from another {@link SerDesStage}.
+ *
+ *
Only {@link RetryableSerDesException} is retried. Other failures are propagated immediately. Retry delays block
+ * the thread executing the SerDes call: the caller by default or the configured SerDes executor thread. Every attempt
+ * receives the same {@link SerDesContext} supplied to this decorator.
+ */
+public final class RetrySerDesStage implements SerDesStage {
+ private final SerDesStage delegate;
+ private final SerDesRetryExecutor retryExecutor;
+
+ /**
+ * Creates a retrying string-stage decorator.
+ *
+ * @param delegate the stage to invoke
+ * @param retryStrategy strategy that controls attempts and delays
+ */
+ public RetrySerDesStage(SerDesStage delegate, RetryStrategy retryStrategy) {
+ this(delegate, retryStrategy, SerDesRetryExecutor.DEFAULT_SLEEPER);
+ }
+
+ RetrySerDesStage(SerDesStage delegate, RetryStrategy retryStrategy, SerDesRetryExecutor.Sleeper sleeper) {
+ this.delegate = Objects.requireNonNull(delegate, "delegate cannot be null");
+ retryExecutor = new SerDesRetryExecutor(retryStrategy, sleeper);
+ }
+
+ @Override
+ public String serialize(String value, SerDesContext context) {
+ return retryExecutor.execute("string stage serialization", () -> delegate.serialize(value, context));
+ }
+
+ @Override
+ public String deserialize(String data, SerDesContext context) {
+ return retryExecutor.execute("string stage deserialization", () -> delegate.deserialize(data, context));
+ }
+}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/serde/SerDes.java b/sdk/src/main/java/software/amazon/lambda/durable/serde/SerDes.java
index b8f39e1c1..4d7c9f14b 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/serde/SerDes.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/serde/SerDes.java
@@ -5,9 +5,10 @@
import software.amazon.lambda.durable.TypeToken;
/**
- * Interface for serialization and deserialization of objects.
+ * Interface for serialization and deserialization of objects at the persisted string boundary.
*
- *
Implementations must support both simple types via {@link Class} and complex generic types via {@link TypeToken}.
+ *
A {@link ComposableSerDes} starts with one SerDes value codec and may be followed by {@link SerDesStage}
+ * transformations that consume and produce strings.
*/
public interface SerDes {
/**
@@ -36,4 +37,14 @@ public interface SerDes {
* @return the deserialized object, or null if data is null
*/
T deserialize(String data, TypeToken typeToken);
+
+ /**
+ * Returns an immutable processing pipeline with a string stage appended.
+ *
+ * @param nextStage the reversible string stage to append
+ * @return a SerDes backed by an immutable processing pipeline
+ */
+ default SerDes then(SerDesStage nextStage) {
+ return ComposableSerDes.builder(this).then(nextStage).build();
+ }
}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/serde/SerDesContext.java b/sdk/src/main/java/software/amazon/lambda/durable/serde/SerDesContext.java
new file mode 100644
index 000000000..fdb371b3f
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/serde/SerDesContext.java
@@ -0,0 +1,89 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.serde;
+
+import software.amazon.awssdk.services.lambda.model.OperationType;
+import software.amazon.lambda.durable.model.OperationSubType;
+
+/**
+ * Describes the durable payload currently being processed by a {@link SerDesStage} or {@link BinarySerDesStage}.
+ *
+ * The SDK passes this context explicitly only to pipeline stages. Root {@link SerDes} value codecs remain
+ * context-free. During serialization, {@link #originalValue()} contains the object supplied to the root value codec;
+ * during deserialization it is {@code null}. Stages must treat the original value as read-only.
+ */
+public record SerDesContext(
+ String durableExecutionArn,
+ String entityId,
+ SerDesPayloadKind payloadKind,
+ String operationId,
+ String operationName,
+ String parentId,
+ OperationType operationType,
+ OperationSubType operationSubType,
+ Integer attempt,
+ Object originalValue) {
+
+ /** Creates context for a root execution payload. */
+ public static SerDesContext forExecution(
+ String durableExecutionArn,
+ String executionOperationId,
+ String executionOperationName,
+ SerDesPayloadKind payloadKind) {
+ return new SerDesContext(
+ durableExecutionArn,
+ "execution/" + executionOperationId + "/" + payloadKind.getEntitySuffix(),
+ payloadKind,
+ executionOperationId,
+ executionOperationName,
+ null,
+ OperationType.EXECUTION,
+ null,
+ null,
+ null);
+ }
+
+ /** Creates context for an operation payload. */
+ public static SerDesContext forOperation(
+ String durableExecutionArn,
+ String operationId,
+ String operationName,
+ String parentId,
+ OperationType operationType,
+ OperationSubType operationSubType,
+ SerDesPayloadKind payloadKind,
+ Integer attempt) {
+ var entityId = "operation/" + operationId + "/" + payloadKind.getEntitySuffix();
+ if (attempt != null) {
+ entityId += "/attempt-" + attempt;
+ }
+ return new SerDesContext(
+ durableExecutionArn,
+ entityId,
+ payloadKind,
+ operationId,
+ operationName,
+ parentId,
+ operationType,
+ operationSubType,
+ attempt,
+ null);
+ }
+
+ SerDesContext withOriginalValue(Object originalValue) {
+ if (this.originalValue == originalValue) {
+ return this;
+ }
+ return new SerDesContext(
+ durableExecutionArn,
+ entityId,
+ payloadKind,
+ operationId,
+ operationName,
+ parentId,
+ operationType,
+ operationSubType,
+ attempt,
+ originalValue);
+ }
+}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/serde/SerDesPayloadKind.java b/sdk/src/main/java/software/amazon/lambda/durable/serde/SerDesPayloadKind.java
new file mode 100644
index 000000000..098e2b565
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/serde/SerDesPayloadKind.java
@@ -0,0 +1,24 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.serde;
+
+/** Identifies the durable payload being serialized or deserialized. */
+public enum SerDesPayloadKind {
+ INPUT("input"),
+ OUTPUT("output"),
+ RESULT("result"),
+ INVOKE_PAYLOAD("invoke-payload"),
+ STATE("state"),
+ EXCEPTION("exception");
+
+ private final String entitySuffix;
+
+ SerDesPayloadKind(String entitySuffix) {
+ this.entitySuffix = entitySuffix;
+ }
+
+ /** Returns the stable suffix used in external payload entity identifiers. */
+ public String getEntitySuffix() {
+ return entitySuffix;
+ }
+}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/serde/SerDesRetryExecutor.java b/sdk/src/main/java/software/amazon/lambda/durable/serde/SerDesRetryExecutor.java
new file mode 100644
index 000000000..aaf014ff3
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/serde/SerDesRetryExecutor.java
@@ -0,0 +1,87 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.serde;
+
+import java.time.Duration;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import software.amazon.lambda.durable.exception.RetryableSerDesException;
+import software.amazon.lambda.durable.exception.SerDesException;
+import software.amazon.lambda.durable.retry.RetryDecision;
+import software.amazon.lambda.durable.retry.RetryStrategy;
+
+final class SerDesRetryExecutor {
+ static final Sleeper DEFAULT_SLEEPER = delay -> {
+ if (delay.getSeconds() > 0) {
+ TimeUnit.SECONDS.sleep(delay.getSeconds());
+ }
+ if (delay.getNano() > 0) {
+ TimeUnit.NANOSECONDS.sleep(delay.getNano());
+ }
+ };
+
+ private final RetryStrategy retryStrategy;
+ private final Sleeper sleeper;
+
+ SerDesRetryExecutor(RetryStrategy retryStrategy, Sleeper sleeper) {
+ this.retryStrategy = Objects.requireNonNull(retryStrategy, "retryStrategy cannot be null");
+ this.sleeper = Objects.requireNonNull(sleeper, "sleeper cannot be null");
+ }
+
+ T execute(String action, Supplier operation) {
+ int attempt = 1;
+ while (true) {
+ try {
+ return operation.get();
+ } catch (RetryableSerDesException failure) {
+ var decision = makeRetryDecision(action, failure, attempt);
+ if (!decision.shouldRetry()) {
+ throw failure;
+ }
+ waitForRetry(action, failure, attempt, decision.delay());
+ attempt++;
+ }
+ }
+ }
+
+ private RetryDecision makeRetryDecision(String action, RetryableSerDesException failure, int attempt) {
+ try {
+ var decision = retryStrategy.makeRetryDecision(failure, attempt);
+ if (decision == null) {
+ throw new SerDesException(
+ String.format("Retry strategy returned null for SerDes %s attempt %d", action, attempt));
+ }
+ return decision;
+ } catch (SerDesException e) {
+ throw e;
+ } catch (RuntimeException e) {
+ throw new SerDesException(
+ String.format("Retry strategy failed for SerDes %s attempt %d", action, attempt), e);
+ }
+ }
+
+ private void waitForRetry(String action, RetryableSerDesException failure, int attempt, Duration delay) {
+ if (delay == null || delay.isNegative()) {
+ throw new SerDesException(String.format(
+ "Retry strategy returned an invalid delay for SerDes %s attempt %d", action, attempt));
+ }
+ if (delay.isZero()) {
+ return;
+ }
+ try {
+ sleeper.sleep(delay);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ var interrupted = new SerDesException(
+ String.format("Interrupted while waiting to retry SerDes %s after attempt %d", action, attempt), e);
+ interrupted.addSuppressed(failure);
+ throw interrupted;
+ }
+ }
+
+ @FunctionalInterface
+ interface Sleeper {
+ void sleep(Duration delay) throws InterruptedException;
+ }
+}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/serde/SerDesRunner.java b/sdk/src/main/java/software/amazon/lambda/durable/serde/SerDesRunner.java
new file mode 100644
index 000000000..c1f55b959
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/serde/SerDesRunner.java
@@ -0,0 +1,224 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.serde;
+
+import java.lang.ref.WeakReference;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Collections;
+import java.util.HexFormat;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
+import software.amazon.lambda.durable.TypeToken;
+import software.amazon.lambda.durable.exception.RetryableSerDesException;
+import software.amazon.lambda.durable.exception.SerDesException;
+import software.amazon.lambda.durable.util.ExceptionHelper;
+
+/**
+ * Runs customer SerDes calls and passes {@link SerDesContext} explicitly to composable pipeline stages.
+ *
+ * Calls execute inline unless an executor is configured. Instances are invocation-scoped so successful
+ * deserialization results are cached only for one Lambda invocation. Completed values use a bounded weak-reference
+ * cache, while concurrent calls for the same value share one in-flight deserialization.
+ */
+public final class SerDesRunner {
+ static final int MAX_COMPLETED_DESERIALIZATIONS = 256;
+ private static final Object CACHE_MISS = new Object();
+ private static final Object NULL_VALUE = new Object();
+
+ private final ExecutorService executorService;
+ private final Map> inFlightDeserializations = new ConcurrentHashMap<>();
+ private final Map> completedDeserializations =
+ Collections.synchronizedMap(new LinkedHashMap<>(16, 0.75f, true) {
+ @Override
+ protected boolean removeEldestEntry(Map.Entry> eldest) {
+ return size() > MAX_COMPLETED_DESERIALIZATIONS;
+ }
+ });
+
+ /**
+ * Creates an invocation-scoped runner.
+ *
+ * @param executorService executor for SerDes calls, or {@code null} to execute inline
+ */
+ public SerDesRunner(ExecutorService executorService) {
+ this.executorService = executorService;
+ }
+
+ /** Serializes a value with the supplied durable payload context. */
+ public String serialize(SerDes serDes, Object value, SerDesContext context) {
+ Objects.requireNonNull(serDes, "serDes cannot be null");
+ return run("serialize", context, () -> serializeWithContext(serDes, value, context));
+ }
+
+ /** Deserializes a value with invocation-scoped caching. */
+ @SuppressWarnings("unchecked")
+ public T deserialize(SerDes serDes, String data, TypeToken typeToken, SerDesContext context) {
+ Objects.requireNonNull(serDes, "serDes cannot be null");
+ Objects.requireNonNull(typeToken, "typeToken cannot be null");
+ Objects.requireNonNull(context, "SerDesContext cannot be null");
+ var key = new CacheKey(
+ serDes,
+ context.durableExecutionArn(),
+ context.entityId(),
+ context.payloadKind(),
+ context.attempt(),
+ typeToken,
+ hash(data));
+ var cached = getCompleted(key);
+ if (cached != CACHE_MISS) {
+ return (T) unmaskNull(cached);
+ }
+
+ var pending = new CompletableFuture