fix: make sure serialized result is deserialize-able

Author: zhongkechen

examples/handler.py

@@ -0,0 +1,39 @@
+from aws_durable_execution_sdk_python.config import Duration
+from aws_durable_execution_sdk_python.context import DurableContext, StepContext, durable_step
+from aws_durable_execution_sdk_python.execution import durable_execution
+from aws_durable_execution_sdk_python.lambda_service import LambdaClient
+import boto3
+
+@durable_step
+def my_step(step_context: StepContext, my_arg: string) -> str:
+    step_context.logger.info("Hello from my_step")
+    return f"from my_step: {my_arg}"
+
+def durable_handler(event, context) -> dict:
+    msg: str = context.step(my_step(raw_event))
+
+    context.wait(Duration.from_seconds(10))
+
+    context.logger.info("Waited for 10 seconds without consuming CPU.")
+
+    return {
+        "statusCode": 200,
+        "body": raw_event,
+    }
+
+raw_event = None
+
+botoclient = boto3.client("lambda",
+                    endpoint_url="https://gamma.lambda.aws.a2z.com",
+                    region_name="us-east-1",
+                )
+
+def lambda_handler(event, context):
+    global raw_event
+    raw_event = str(event)
+    print("event: " + str(event))
+
+    result = durable_execution(boto3_client=botoclient)(durable_handler)(event, context)
+    result["Result"] = raw_event
+    return result
+

examples/handler.zip

@@ -0,0 +1 @@
+{'DurableExecutionArn': 'arn:aws:lambda:us-east-1:858039354069:function:myDurablePythonFunction2:$LATEST/durable-execution/8e89d3bc-455f-426f-97c8-72c565aac75a/cac7f0d9-6dc2-3f9b-bcf9-8666cf6af052', 'CheckpointToken': 'AgGAAXjQr2ytV2m8Vv5j81VX/+CvAAAAAQAHYXdzLWttcwBLYXJuOmF3czprbXM6dXMtZWFzdC0xOjMxMDI3MzUwNTMwNDprZXkvZGM2OWUzOGQtYmFjNy00NjY2LThiY2QtZjliYzBjMjBlYTBiALgBAgEAeFOOxgoIuAmgS5wl8hGyK5/Q6zA+mRvvS6BWlEQhnQGVATQLMQpKFE1fNLy2FWNPuWEAAAB+MHwGCSqGSIb3DQEHBqBvMG0CAQAwaAYJKoZIhvcNAQcBMB4GCWCGSAFlAwQBLjARBAyJ1kwrOrnTlRVAWrkCARCAO0c0v/dWH34T7wD63MS35+5edzXgJxJAPSnD0BmhV/zGq1u8bIjOJOYjo6cAGsf/PG153w7omAOgtQbgAgAAAAAMAAAQAAAAAAAAAAAAAAAAALmdpx86+Llkf5D4pUCe6RX/////AAAAAQAAAAAAAAAAAAAAAQAAAPUbq6CcIk55x6/RjhqokZApQWTYt+kyB1y+fWJhlH9vN4jy36nkefO5gyQJSy8nJQEY8BlmP/ETJhsKXhUoMg4UoloDos9TtKKXfIY1KZzcPGQfoLf0SwvSkCJQVwnM46dkjK9tpki4s4T2BXebYnw37XrPzVTnfnmqU60QjSs5+riKXDu9TDcDc9U0LKXKphaku06RXzg6HevXTumk40w0m5FBCLprphhW94oQgp0RM3ehAzAYuKzNSVPmE0l0Aii8Btie5qCtSA2JnDUT0qVvij08kYDG4XD6T9SPNDXA5mLucMwbzrOun7rI28AEwjMC8I5Lj2hFbu8WkTF7E1hghTBbI/k', 'InitialExecutionState': {'Operations': [{'Id': 'cac7f0d9-6dc2-3f9b-bcf9-8666cf6af052', 'Name': '8e89d3bc-455f-426f-97c8-72c565aac75a', 'Type': 'EXECUTION', 'StartTimestamp': 1781307233431, 'Status': 'STARTED', 'ExecutionDetails': {'InputPayload': '{}'}}]}, 'UpdatedOperationIds': []}

examples/output

@@ -31,6 +31,7 @@
  *   <li>Automatic region detection with fallback to us-east-1 for testing environments
  *   <li>Environment variable credentials provider
  *   <li>Custom SerDes with snake_case property naming
+ *   <li>Optional round-trip validation toggle for performance-sensitive workloads
  * </ul>
  */
 public class CustomConfigExample extends DurableHandler<String, String> {
@@ -68,6 +69,8 @@ protected DurableConfig createConfiguration() {
         return DurableConfig.builder()
                 .withDurableExecutionClient(durableClient)
                 .withSerDes(customSerDes)
+                // Disable the extra deserialize pass if your workload is sensitive to the added validation cost.
+                .withSerializationRoundTripValidation(false)
                 .build();
     }
 

examples/output.json

@@ -98,6 +98,7 @@ public final class DurableConfig {
     private final LoggerConfig loggerConfig;
     private final PollingStrategy pollingStrategy;
     private final Duration checkpointDelay;
+    private final boolean validateSerializationRoundTrip;
     private final PluginRunner pluginRunner;
 
     private DurableConfig(Builder builder) {
@@ -109,6 +110,7 @@ private DurableConfig(Builder builder) {
         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));
+        this.validateSerializationRoundTrip = builder.validateSerializationRoundTrip;
         this.pluginRunner = builder.plugins.isEmpty() ? PluginRunner.noOp() : new PluginRunner(builder.plugins);
 
         validateConfiguration();
@@ -186,6 +188,19 @@ public Duration getCheckpointDelay() {
         return checkpointDelay;
     }
 
+    /**
+     * Gets whether serialized operation data should be immediately deserialized to verify round-trip compatibility.
+     *
+     * <p>When enabled, the SDK validates serialized operation results and exceptions before checkpointing them. This
+     * catches incompatible SerDes behavior early at the cost of an extra deserialize pass. Defaults to true, and custom
+     * SerDes implementations are still expected to be round-trip safe even if this validation is disabled.
+     *
+     * @return true when round-trip serialization validation is enabled
+     */
+    public boolean shouldValidateSerializationRoundTrip() {
+        return validateSerializationRoundTrip;
+    }
+
     /**
      * Gets the plugin runner that dispatches lifecycle events to registered plugins.
      *
@@ -293,6 +308,7 @@ public static final class Builder {
         private LoggerConfig loggerConfig;
         private PollingStrategy pollingStrategy;
         private Duration checkpointDelay;
+        private boolean validateSerializationRoundTrip = true;
         private List<DurableExecutionPlugin> plugins = new ArrayList<>();
 
         public Builder() {}
@@ -403,6 +419,22 @@ public Builder withCheckpointDelay(Duration duration) {
             return this;
         }
 
+        /**
+         * Controls whether the SDK immediately deserializes serialized results and exceptions to verify they can be
+         * read back before checkpointing.
+         *
+         * <p>This validation is enabled by default. Disable it only to avoid the extra deserialize pass when the
+         * additional safety check is too expensive for your workload. Custom SerDes implementations are still expected
+         * to round-trip SDK-managed values correctly.
+         *
+         * @param validateSerializationRoundTrip true to validate serialized data with an immediate deserialize pass
+         * @return This builder
+         */
+        public Builder withSerializationRoundTripValidation(boolean validateSerializationRoundTrip) {
+            this.validateSerializationRoundTrip = validateSerializationRoundTrip;
+            return this;
+        }
+
         /**
          * Registers one or more plugins for lifecycle event instrumentation.
          *

examples/src/main/java/software/amazon/lambda/durable/examples/general/CustomConfigExample.java

@@ -117,6 +117,7 @@ public DurableConfig getConfiguration() {
      *         .withDurableExecutionClient(durableClient)
      *         .withSerDes(customSerDes)  // Optional: custom SerDes for user data
      *         .withExecutorService(customExecutor)  // Optional: custom thread pool
+     *         .withSerializationRoundTripValidation(false)  // Optional: skip extra validation deserialize pass
      *         .build();
      * }
      * }</pre>

sdk/src/main/java/software/amazon/lambda/durable/DurableConfig.java

@@ -149,6 +149,8 @@ private void executeChildContext() {
     }
 
     private void handleChildContextSuccess(T result) {
+        var serialized = serializeResult(result);
+
         if (replayChildren.get() || isVirtual || parentOperation != null && parentOperation.isOperationCompleted()) {
             // Skip checkpointing if
             // - parent ConcurrencyOperation has already completed, preventing race conditions where a child finishes
@@ -159,13 +161,11 @@ private void handleChildContextSuccess(T result) {
             cachedOperationResult.set(DeserializedOperationResult.succeeded(result));
             markAlreadyCompleted();
         } else {
-            checkpointSuccess(result);
+            checkpointSuccess(result, serialized);
         }
     }
 
-    private void checkpointSuccess(T result) {
-        var serialized = serializeResult(result);
-
+    private void checkpointSuccess(T result, String serialized) {
         if (serialized == null || serialized.getBytes(StandardCharsets.UTF_8).length < LARGE_RESULT_THRESHOLD) {
             sendOperationUpdate(
                     OperationUpdate.builder().action(OperationAction.SUCCEED).payload(serialized));

sdk/src/main/java/software/amazon/lambda/durable/DurableHandler.java

@@ -101,7 +101,11 @@ protected T deserializeResult(String result) {
      * @return the serialized string
      */
     protected String serializeResult(T result) {
-        return resultSerDes.serialize(result);
+        var serialized = resultSerDes.serialize(result);
+        if (shouldValidateSerializationRoundTrip()) {
+            deserializeResult(serialized);
+        }
+        return serialized;
     }
 
     /**
@@ -110,8 +114,18 @@ protected String serializeResult(T result) {
      * @param throwable the exception to serialize
      * @return the serialized error object
      */
+    @SuppressWarnings("ThrowableNotThrown")
     protected ErrorObject serializeException(Throwable throwable) {
-        return ExceptionHelper.buildErrorObject(throwable, resultSerDes);
+        var error = ExceptionHelper.buildErrorObject(throwable, resultSerDes);
+        if (shouldValidateSerializationRoundTrip()) {
+            deserializeException(error);
+        }
+        return error;
+    }
+
+    private boolean shouldValidateSerializationRoundTrip() {
+        var config = getContext().getDurableConfig();
+        return config == null || config.shouldValidateSerializationRoundTrip();
     }
 
     /**

sdk/src/main/java/software/amazon/lambda/durable/operation/ChildContextOperation.java

@@ -87,6 +87,24 @@ void testBuilder_WithCustomExecutorService() {
         assertNotNull(config.getSerDes());
     }
 
+    @Test
+    void testBuilder_SerializationRoundTripValidationDefaultsToTrue() {
+        var config =
+                DurableConfig.builder().withDurableExecutionClient(mockClient).build();
+
+        assertTrue(config.shouldValidateSerializationRoundTrip());
+    }
+
+    @Test
+    void testBuilder_WithSerializationRoundTripValidationDisabled() {
+        var config = DurableConfig.builder()
+                .withDurableExecutionClient(mockClient)
+                .withSerializationRoundTripValidation(false)
+                .build();
+
+        assertFalse(config.shouldValidateSerializationRoundTrip());
+    }
+
     @Test
     void testBuilder_WithAllCustomComponents() {
         var config = DurableConfig.builder()
@@ -131,6 +149,7 @@ void testBuilder_FluentAPI() {
         assertSame(builder, builder.withDurableExecutionClient(mockClient));
         assertSame(builder, builder.withSerDes(mockSerDes));
         assertSame(builder, builder.withExecutorService(mockExecutor));
+        assertSame(builder, builder.withSerializationRoundTripValidation(false));
     }
 
     @Test