diff --git a/examples/src/main/java/software/amazon/lambda/durable/examples/general/CustomConfigExample.java b/examples/src/main/java/software/amazon/lambda/durable/examples/general/CustomConfigExample.java index 28d248395..5bbd902b0 100644 --- a/examples/src/main/java/software/amazon/lambda/durable/examples/general/CustomConfigExample.java +++ b/examples/src/main/java/software/amazon/lambda/durable/examples/general/CustomConfigExample.java @@ -31,6 +31,7 @@ *
  • Automatic region detection with fallback to us-east-1 for testing environments *
  • Environment variable credentials provider *
  • Custom SerDes with snake_case property naming + *
  • Optional round-trip validation toggle for performance-sensitive workloads * */ public class CustomConfigExample extends DurableHandler { @@ -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(); } 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 a21251e19..f3610082e 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/DurableConfig.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/DurableConfig.java @@ -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. + * + *

    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 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. + * + *

    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. * diff --git a/sdk/src/main/java/software/amazon/lambda/durable/DurableHandler.java b/sdk/src/main/java/software/amazon/lambda/durable/DurableHandler.java index a31c94d00..6fceee077 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/DurableHandler.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/DurableHandler.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(); * } * } 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 acdda283b..f2fa1a561 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 @@ -149,6 +149,8 @@ private void executeChildContext() { } private void handleChildContextSuccess(T result) { + var serializedResult = serializeResultWithDeserializedValue(result); + if (replayChildren.get() || isVirtual || parentOperation != null && parentOperation.isOperationCompleted()) { // Skip checkpointing if // - parent ConcurrencyOperation has already completed, preventing race conditions where a child finishes @@ -156,23 +158,22 @@ private void handleChildContextSuccess(T result) { // - replaying a SUCCEEDED child with replayChildren=true — skip checkpointing. // - nestingType is FLAT // Mark the completableFuture completed so get() doesn't block waiting for a checkpoint response. - cachedOperationResult.set(DeserializedOperationResult.succeeded(result)); + cachedOperationResult.set(DeserializedOperationResult.succeeded(serializedResult.deserialized())); markAlreadyCompleted(); } else { - checkpointSuccess(result); + checkpointSuccess(serializedResult); } } - private void checkpointSuccess(T result) { - var serialized = serializeResult(result); - + private void checkpointSuccess(SerializedResult serializedResult) { + var serialized = serializedResult.serialized(); if (serialized == null || serialized.getBytes(StandardCharsets.UTF_8).length < LARGE_RESULT_THRESHOLD) { sendOperationUpdate( OperationUpdate.builder().action(OperationAction.SUCCEED).payload(serialized)); } else { // Large result: checkpoint with empty payload + ReplayChildren flag. // Store the result so get() can return it directly without deserializing the empty payload. - cachedOperationResult.set(DeserializedOperationResult.succeeded(result)); + cachedOperationResult.set(DeserializedOperationResult.succeeded(serializedResult.deserialized())); sendOperationUpdate(OperationUpdate.builder() .action(OperationAction.SUCCEED) .payload("") diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/MapOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/MapOperation.java index c2afb1ab7..75a706cf2 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/MapOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/MapOperation.java @@ -177,22 +177,22 @@ protected void replay(Operation existing) { @Override protected void handleCompletion(ConcurrencyCompletionStatus concurrencyCompletionStatus) { - this.cachedResult = constructMapResult(concurrencyCompletionStatus); - var serialized = serializeResult(cachedResult); - var serializedBytes = serialized.getBytes(StandardCharsets.UTF_8); + var serializedResult = serializeResultWithDeserializedValue(constructMapResult(concurrencyCompletionStatus)); + this.cachedResult = serializedResult.deserialized(); + var serializedBytes = serializedResult.serialized().getBytes(StandardCharsets.UTF_8); if (serializedBytes.length < LARGE_RESULT_THRESHOLD) { sendOperationUpdate(OperationUpdate.builder() .action(OperationAction.SUCCEED) .subType(getSubType().getValue()) - .payload(serialized)); + .payload(serializedResult.serialized())); } else { // Large result: checkpoint with stripped payload + replayChildren flag - var strippedResult = serializeResult(stripMapResult(cachedResult)); + var strippedResult = serializeResultWithDeserializedValue(stripMapResult(cachedResult)); sendOperationUpdate(OperationUpdate.builder() .action(OperationAction.SUCCEED) .subType(getSubType().getValue()) - .payload(strippedResult) + .payload(strippedResult.serialized()) .contextOptions( ContextOptions.builder().replayChildren(true).build())); } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/ParallelOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/ParallelOperation.java index 4ae7b8386..b095d6443 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/ParallelOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/ParallelOperation.java @@ -76,15 +76,17 @@ protected void handleCompletion(ConcurrencyCompletionStatus concurrencyCompletio int failedCount = Math.toIntExact( statuses.stream().filter(s -> s == ParallelResult.Status.FAILED).count()); int skippedCount = items.size() - succeededCount - failedCount; - cachedResult = new ParallelResult( + var result = new ParallelResult( items.size(), succeededCount, failedCount, skippedCount, concurrencyCompletionStatus, statuses); + var serializedResult = serializeResultWithDeserializedValue(result); + cachedResult = serializedResult.deserialized(); // Branches added after checkpoint will not exist in the checkpointed result, but they'll be in the returned // value from get() method. sendOperationUpdate(OperationUpdate.builder() .action(OperationAction.SUCCEED) .subType(getSubType().getValue()) - .payload(serializeResult(cachedResult)) + .payload(serializedResult.serialized()) .contextOptions(ContextOptions.builder().replayChildren(true).build())); } 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 6ccf24e0f..de9384656 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 @@ -34,6 +34,8 @@ public abstract class SerializableDurableOperation extends BaseDurableOperation implements DurableFuture { private static final Logger logger = LoggerFactory.getLogger(SerializableDurableOperation.class); + protected record SerializedResult(String serialized, T deserialized) {} + private final TypeToken resultTypeToken; private final SerDes resultSerDes; @@ -94,14 +96,13 @@ protected T deserializeResult(String result) { } } - /** - * Serializes the result to a string. - * - * @param result the result to serialize - * @return the serialized string - */ - protected String serializeResult(T result) { - return resultSerDes.serialize(result); + protected SerializedResult serializeResultWithDeserializedValue(T result) { + var serialized = resultSerDes.serialize(result); + T deserialized = result; + if (shouldValidateSerializationRoundTrip()) { + deserialized = deserializeResult(serialized); + } + return new SerializedResult<>(serialized, deserialized); } /** @@ -110,8 +111,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(); } /** 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 6018a0ccf..6b4ed0629 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 @@ -145,8 +145,9 @@ private void checkpointStarted() { private void handleStepSucceeded(T result) { // Send SUCCEED + var serializedResult = serializeResultWithDeserializedValue(result); var successUpdate = - OperationUpdate.builder().action(OperationAction.SUCCEED).payload(serializeResult(result)); + OperationUpdate.builder().action(OperationAction.SUCCEED).payload(serializedResult.serialized()); // sendOperationUpdate must be synchronous here. When waiting for the return of this call, // the context threads waiting for the result of this step operation will be wakened up and registered. 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 8404e1d68..a4d20b778 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 @@ -128,15 +128,15 @@ private void executeCheckLogic(T currentState, int attempt) { // Execute check function in user executor WaitForConditionResult result = checkFunc.apply(currentState, stepContext); - // Serialize/deserialize round-trip on the value to ensure state is checkpoint-safe - var serializedState = serializeResult(result.value()); - T deserializedValue = deserializeResult(serializedState); + // Continue with the checkpoint-shaped state so first execution matches replay. + var serializedState = serializeResultWithDeserializedValue(result.value()); + T deserializedValue = serializedState.deserialized(); if (result.isDone()) { // Condition met — checkpoint SUCCEED var successUpdate = OperationUpdate.builder() .action(OperationAction.SUCCEED) - .payload(serializedState); + .payload(serializedState.serialized()); sendOperationUpdate(successUpdate); } else { // Compute delay from strategy @@ -145,7 +145,7 @@ private void executeCheckLogic(T currentState, int attempt) { // Checkpoint RETRY with delay var retryUpdate = OperationUpdate.builder() .action(OperationAction.RETRY) - .payload(serializedState) + .payload(serializedState.serialized()) .stepOptions(StepOptions.builder() .nextAttemptDelaySeconds(Math.toIntExact(delay.toSeconds())) .build()); diff --git a/sdk/src/test/java/software/amazon/lambda/durable/DurableConfigTest.java b/sdk/src/test/java/software/amazon/lambda/durable/DurableConfigTest.java index 94d13f5ef..9f1dceddf 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/DurableConfigTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/DurableConfigTest.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 diff --git a/sdk/src/test/java/software/amazon/lambda/durable/operation/ChildContextOperationTest.java b/sdk/src/test/java/software/amazon/lambda/durable/operation/ChildContextOperationTest.java index ac56262a4..e9c67e7c1 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/operation/ChildContextOperationTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/operation/ChildContextOperationTest.java @@ -24,16 +24,43 @@ import software.amazon.lambda.durable.context.DurableContextImpl; import software.amazon.lambda.durable.exception.ChildContextFailedException; import software.amazon.lambda.durable.exception.NonDeterministicExecutionException; +import software.amazon.lambda.durable.exception.SerDesException; import software.amazon.lambda.durable.execution.ExecutionManager; import software.amazon.lambda.durable.execution.ThreadContext; import software.amazon.lambda.durable.execution.ThreadType; import software.amazon.lambda.durable.model.OperationIdentifier; import software.amazon.lambda.durable.model.OperationSubType; import software.amazon.lambda.durable.serde.JacksonSerDes; +import software.amazon.lambda.durable.serde.SerDes; /** Unit tests for ChildContextOperation. */ class ChildContextOperationTest { + private static final class SerializationOnlySerDes implements SerDes { + @Override + public String serialize(Object value) { + return "\"serialized\""; + } + + @Override + public T deserialize(String data, TypeToken typeToken) { + throw new SerDesException("cannot deserialize"); + } + } + + private static final class TransformingSerDes implements SerDes { + @Override + public String serialize(Object value) { + return "\"deserialized\""; + } + + @Override + @SuppressWarnings("unchecked") + public T deserialize(String data, TypeToken typeToken) { + return (T) "deserialized"; + } + } + private static final JacksonSerDes SERDES = new JacksonSerDes(); private DurableContextImpl durableContext; @@ -49,8 +76,13 @@ void setUp() { } private DurableConfig createConfig() { + return createConfig(true); + } + + private DurableConfig createConfig(boolean validateSerializationRoundTrip) { return DurableConfig.builder() .withExecutorService(Executors.newCachedThreadPool()) + .withSerializationRoundTripValidation(validateSerializationRoundTrip) .build(); } @@ -58,20 +90,28 @@ private DurableConfig createConfig() { OperationIdentifier.of("1", "test-context", OperationSubType.RUN_IN_CHILD_CONTEXT); private ChildContextOperation createOperation(Function func) { + return createOperation(func, SERDES); + } + + private ChildContextOperation createOperation(Function func, SerDes serDes) { return new ChildContextOperation<>( OPERATION_IDENTIFIER, func, TypeToken.get(String.class), - RunInChildContextConfig.builder().serDes(SERDES).build(), + RunInChildContextConfig.builder().serDes(serDes).build(), durableContext); } private ChildContextOperation createVirtualOperation(Function func) { + return createVirtualOperation(func, SERDES); + } + + private ChildContextOperation createVirtualOperation(Function func, SerDes serDes) { return new ChildContextOperation<>( OPERATION_IDENTIFIER, func, TypeToken.get(String.class), - RunInChildContextConfig.builder().serDes(SERDES).isVirtual(true).build(), + RunInChildContextConfig.builder().serDes(serDes).isVirtual(true).build(), durableContext); } @@ -132,6 +172,16 @@ void executeVirtualContext() { assertTrue(functionCalled.get(), "Function should be called during SUCCEEDED replay"); } + @Test + void virtualChildReturnsDeserializedResult() { + var operation = createVirtualOperation(ctx -> "raw", new TransformingSerDes()); + + operation.execute(); + var result = operation.get(); + + assertEquals("deserialized", result); + } + // ===== FAILED replay ===== /** FAILED replay throws the original exception without re-executing. */ @@ -311,6 +361,40 @@ void childSkipsSuccessCheckpointWhenParentAlreadyCompleted() throws Exception { .sendOperationUpdate(argThat(update -> update.action() == OperationAction.SUCCEED)); } + /** Virtual child still validates result round-trip before skipping a success checkpoint. */ + @Test + void virtualChildFailsWhenResultCannotBeDeserialized() throws Exception { + when(executionManager.getOperationAndUpdateReplayState("1")).thenReturn(null); + + var operation = createVirtualOperation(ctx -> "result", new SerializationOnlySerDes()); + operation.execute(); + Thread.sleep(200); + + var thrown = assertThrows(ChildContextFailedException.class, operation::get); + assertTrue(thrown.getMessage().contains(SerDesException.class.getName())); + verify(executionManager, never()) + .sendOperationUpdate(argThat(update -> update.action() == OperationAction.SUCCEED)); + verify(executionManager, never()) + .sendOperationUpdate(argThat(update -> update.action() == OperationAction.FAIL)); + } + + /** Virtual child skips deserialization and returns the original result when round-trip validation is disabled. */ + @Test + void virtualChildReturnsOriginalResultWhenValidationDisabled() throws Exception { + when(executionManager.getOperationAndUpdateReplayState("1")).thenReturn(null); + when(durableContext.getDurableConfig()).thenReturn(createConfig(false)); + + var operation = createVirtualOperation(ctx -> "result", new SerializationOnlySerDes()); + operation.execute(); + Thread.sleep(200); + + assertEquals("result", operation.get()); + verify(executionManager, never()) + .sendOperationUpdate(argThat(update -> update.action() == OperationAction.SUCCEED)); + verify(executionManager, never()) + .sendOperationUpdate(argThat(update -> update.action() == OperationAction.FAIL)); + } + /** Child skips failure checkpoint when parent operation has already completed. */ @Test void childSkipsFailureCheckpointWhenParentAlreadyCompleted() throws Exception { diff --git a/sdk/src/test/java/software/amazon/lambda/durable/operation/SerializableDurableOperationTest.java b/sdk/src/test/java/software/amazon/lambda/durable/operation/SerializableDurableOperationTest.java index c544eb235..a88c57f6f 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/operation/SerializableDurableOperationTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/operation/SerializableDurableOperationTest.java @@ -20,6 +20,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import software.amazon.awssdk.services.lambda.model.ErrorObject; @@ -27,7 +28,9 @@ import software.amazon.awssdk.services.lambda.model.OperationStatus; import software.amazon.awssdk.services.lambda.model.OperationType; import software.amazon.awssdk.services.lambda.model.OperationUpdate; +import software.amazon.lambda.durable.DurableConfig; import software.amazon.lambda.durable.TypeToken; +import software.amazon.lambda.durable.client.DurableExecutionClient; import software.amazon.lambda.durable.context.DurableContextImpl; import software.amazon.lambda.durable.exception.IllegalDurableOperationException; import software.amazon.lambda.durable.exception.NonDeterministicExecutionException; @@ -42,6 +45,45 @@ class SerializableDurableOperationTest { + private static final class TrackingSerDes extends JacksonSerDes { + private final AtomicInteger deserializeCount = new AtomicInteger(0); + + @Override + public T deserialize(String data, TypeToken typeToken) { + deserializeCount.incrementAndGet(); + return super.deserialize(data, typeToken); + } + + int getDeserializeCount() { + return deserializeCount.get(); + } + } + + private static final class SerializationOnlySerDes implements SerDes { + @Override + public String serialize(Object value) { + return "\"serialized\""; + } + + @Override + public T deserialize(String data, TypeToken typeToken) { + throw new SerDesException("cannot deserialize"); + } + } + + private static final class TransformingSerDes implements SerDes { + @Override + public String serialize(Object value) { + return "\"deserialized\""; + } + + @Override + @SuppressWarnings("unchecked") + public T deserialize(String data, TypeToken typeToken) { + return (T) "deserialized"; + } + } + private static final String OPERATION_ID = "1"; private static final String CONTEXT_ID = "1-step"; private static final String OPERATION_NAME = "name"; @@ -331,7 +373,7 @@ protected void replay(Operation existing) {} @Override public String get() { - assertEquals("abc", deserializeResult(serializeResult("abc"))); + assertEquals("abc", deserializeResult("\"abc\"")); assertEquals("", deserializeResult("\"\"")); assertThrows(SerDesException.class, () -> deserializeResult("x")); return RESULT; @@ -340,6 +382,101 @@ public String get() { op.get(); } + @Test + void serializeResultWithDeserializedValueValidatesRoundTrip() { + var serDes = new TrackingSerDes(); + SerializableDurableOperation op = + new SerializableDurableOperation<>(OPERATION_IDENTIFIER, RESULT_TYPE, serDes, durableContext) { + @Override + protected void start() {} + + @Override + protected void replay(Operation existing) {} + + @Override + public String get() { + var result = serializeResultWithDeserializedValue("abc"); + assertEquals("\"abc\"", result.serialized()); + assertEquals("abc", result.deserialized()); + assertEquals(1, serDes.getDeserializeCount()); + return RESULT; + } + }; + + op.get(); + } + + @Test + void serializeResultWithDeserializedValueReturnsRoundTrippedResult() { + var serDes = new TransformingSerDes(); + SerializableDurableOperation op = + new SerializableDurableOperation<>(OPERATION_IDENTIFIER, RESULT_TYPE, serDes, durableContext) { + @Override + protected void start() {} + + @Override + protected void replay(Operation existing) {} + + @Override + public String get() { + var result = serializeResultWithDeserializedValue("raw"); + assertEquals("\"deserialized\"", result.serialized()); + assertEquals("deserialized", result.deserialized()); + return RESULT; + } + }; + + op.get(); + } + + @Test + void serializeResultWithDeserializedValueThrowsWhenRoundTripFails() { + var serDes = new SerializationOnlySerDes(); + SerializableDurableOperation op = + new SerializableDurableOperation<>(OPERATION_IDENTIFIER, RESULT_TYPE, serDes, durableContext) { + @Override + protected void start() {} + + @Override + protected void replay(Operation existing) {} + + @Override + public String get() { + var thrown = + assertThrows(SerDesException.class, () -> serializeResultWithDeserializedValue("abc")); + assertEquals("cannot deserialize", thrown.getMessage()); + return RESULT; + } + }; + + op.get(); + } + + @Test + void serializeResultWithDeserializedValueSkipsRoundTripValidationWhenDisabled() { + when(durableContext.getDurableConfig()).thenReturn(configWithSerializationValidation(false)); + + var serDes = new SerializationOnlySerDes(); + SerializableDurableOperation op = + new SerializableDurableOperation<>(OPERATION_IDENTIFIER, RESULT_TYPE, serDes, durableContext) { + @Override + protected void start() {} + + @Override + protected void replay(Operation existing) {} + + @Override + public String get() { + var result = serializeResultWithDeserializedValue("abc"); + assertEquals("\"serialized\"", result.serialized()); + assertEquals("abc", result.deserialized()); + return RESULT; + } + }; + + op.get(); + } + @Test void deserializeException() { SerializableDurableOperation op = @@ -366,6 +503,53 @@ public String get() { op.get(); } + @Test + void serializeExceptionValidatesRoundTrip() { + var serDes = new TrackingSerDes(); + SerializableDurableOperation op = + new SerializableDurableOperation<>(OPERATION_IDENTIFIER, RESULT_TYPE, serDes, durableContext) { + @Override + protected void start() {} + + @Override + protected void replay(Operation existing) {} + + @Override + public String get() { + var error = serializeException(new RuntimeException("test exception")); + assertEquals(RuntimeException.class.getName(), error.errorType()); + assertEquals(1, serDes.getDeserializeCount()); + return RESULT; + } + }; + + op.get(); + } + + @Test + void serializeExceptionSkipsRoundTripValidationWhenDisabled() { + when(durableContext.getDurableConfig()).thenReturn(configWithSerializationValidation(false)); + + var serDes = new SerializationOnlySerDes(); + SerializableDurableOperation op = + new SerializableDurableOperation<>(OPERATION_IDENTIFIER, RESULT_TYPE, serDes, durableContext) { + @Override + protected void start() {} + + @Override + protected void replay(Operation existing) {} + + @Override + public String get() { + var error = serializeException(new RuntimeException("test exception")); + assertEquals(RuntimeException.class.getName(), error.errorType()); + return RESULT; + } + }; + + op.get(); + } + @Test void polling() { SerializableDurableOperation op = @@ -411,4 +595,11 @@ public String get() { op.execute(); verify(executionManager, times(1)).sendOperationUpdate(update.build()); } + + private DurableConfig configWithSerializationValidation(boolean validateSerializationRoundTrip) { + return DurableConfig.builder() + .withDurableExecutionClient(mock(DurableExecutionClient.class)) + .withSerializationRoundTripValidation(validateSerializationRoundTrip) + .build(); + } }