diff --git a/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/ChildContextWrappedErrorIntegrationTest.java b/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/ChildContextWrappedErrorIntegrationTest.java new file mode 100644 index 000000000..c31c7277e --- /dev/null +++ b/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/ChildContextWrappedErrorIntegrationTest.java @@ -0,0 +1,303 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.function.Consumer; +import java.util.function.Function; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; +import software.amazon.awssdk.services.lambda.model.ErrorObject; +import software.amazon.lambda.durable.config.StepConfig; +import software.amazon.lambda.durable.exception.CallbackFailedException; +import software.amazon.lambda.durable.exception.CallbackTimeoutException; +import software.amazon.lambda.durable.exception.ChildContextFailedException; +import software.amazon.lambda.durable.exception.InvokeStoppedException; +import software.amazon.lambda.durable.exception.InvokeTimedOutException; +import software.amazon.lambda.durable.execution.SuspendExecutionException; +import software.amazon.lambda.durable.model.ExecutionStatus; +import software.amazon.lambda.durable.retry.RetryStrategies; +import software.amazon.lambda.durable.testing.LocalDurableTestRunner; +import software.amazon.lambda.durable.testing.TestResult; + +/** + * A failed context checkpoints the wrapped (original) error, never the SDK wrapper, and the live and replayed failure + * that user code catches has the same type and error object. + */ +class ChildContextWrappedErrorIntegrationTest { + + static final String MSG = "card declined"; + + public static class PaymentError extends RuntimeException { + public PaymentError(String message) { + super(message); + } + } + + private static final StepConfig NO_RETRY = + StepConfig.builder().retryStrategy(RetryStrategies.Presets.NO_RETRY).build(); + + /** What user code caught: the exception class and the type and message of its error object. */ + private record Caught(String exception, String errorType, String errorMessage) {} + + private static Caught caught(Throwable t) { + var eo = t instanceof software.amazon.lambda.durable.exception.DurableOperationException d + ? d.getErrorObject() + : null; + return new Caught( + t.getClass().getName(), + eo == null ? null : eo.errorType(), + eo == null ? t.getMessage() : eo.errorMessage()); + } + + private record Outcome(Caught live, Caught replayed, TestResult result) {} + + /** + * Runs {@code failing}, records what it threw, waits so that a new invocation starts, and records what the replay + * of {@code failing} throws. + */ + private static Outcome run( + Function failing, Consumer> drive) { + List seen = Collections.synchronizedList(new ArrayList<>()); + var runner = LocalDurableTestRunner.create(String.class, (String in, DurableContext ctx) -> { + try { + failing.apply(ctx); + throw new IllegalStateException("did not fail"); + } catch (SuspendExecutionException e) { + throw e; + } catch (RuntimeException e) { + seen.add(caught(e)); + } + ctx.wait("pause", Duration.ofSeconds(1)); + return "done"; + }); + if (drive != null) { + runner.run("in"); + drive.accept(runner); + } + var result = runner.runUntilComplete("in"); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + assertEquals(2, seen.size(), "one live and one replayed failure: " + seen); + return new Outcome(seen.get(0), seen.get(1), result); + } + + private static ErrorObject contextError(TestResult result, String name) { + var details = result.getOperation(name).getContextDetails(); + assertNotNull(details, name + " context details"); + return details.error(); + } + + private static ErrorObject error(String type) { + return ErrorObject.builder().errorType(type).errorMessage(MSG).build(); + } + + private static Object invokeInChild(DurableContext ctx) { + return ctx.runInChildContext("child", String.class, cc -> cc.invoke("callee", "fn", "x", String.class)); + } + + private static Object callbackInChild(DurableContext ctx) { + return ctx.runInChildContext( + "child", + String.class, + cc -> cc.createCallback("cb", String.class).get()); + } + + enum Scenario { + STEP_ERROR, + INVOKE_FAILED, + INVOKE_STOPPED, + INVOKE_STOPPED_USER_TYPE, + INVOKE_TIMED_OUT, + CALLBACK_FAILED, + CALLBACK_FAILED_NO_TYPE, + CALLBACK_FAILED_TYPE_ERROR, + CALLBACK_TIMED_OUT + } + + @ParameterizedTest + @EnumSource(Scenario.class) + void contextCheckpointsTheWrappedErrorAndReplaysTheSameShape(Scenario scenario) { + Outcome outcome; + String type; + String message = MSG; + switch (scenario) { + case STEP_ERROR -> { + outcome = run( + c -> c.runInChildContext( + "child", + String.class, + cc -> cc.step( + "charge", + String.class, + sc -> { + throw new PaymentError(MSG); + }, + NO_RETRY)), + null); + type = PaymentError.class.getName(); + } + case INVOKE_FAILED -> { + outcome = run( + ChildContextWrappedErrorIntegrationTest::invokeInChild, + r -> r.failChainedInvoke("callee", error("com.acme.PaymentError"))); + type = "com.acme.PaymentError"; + } + case INVOKE_STOPPED -> { + outcome = run( + ChildContextWrappedErrorIntegrationTest::invokeInChild, + r -> r.stopChainedInvoke("callee", error("States.Stopped"))); + type = InvokeStoppedException.class.getName(); + } + case INVOKE_STOPPED_USER_TYPE -> { + outcome = run( + ChildContextWrappedErrorIntegrationTest::invokeInChild, + r -> r.stopChainedInvoke("callee", error("com.acme.PaymentError"))); + type = InvokeStoppedException.class.getName(); + } + case INVOKE_TIMED_OUT -> { + outcome = run( + ChildContextWrappedErrorIntegrationTest::invokeInChild, r -> r.timeoutChainedInvoke("callee")); + type = InvokeTimedOutException.class.getName(); + message = null; + } + case CALLBACK_FAILED -> { + outcome = run( + ChildContextWrappedErrorIntegrationTest::callbackInChild, + r -> r.failCallback(r.getCallbackId("cb"), error("com.acme.PaymentError"))); + type = "com.acme.PaymentError"; + } + case CALLBACK_FAILED_NO_TYPE -> { + outcome = run( + ChildContextWrappedErrorIntegrationTest::callbackInChild, + r -> r.failCallback(r.getCallbackId("cb"), error(null))); + type = CallbackFailedException.class.getName(); + } + case CALLBACK_FAILED_TYPE_ERROR -> { + // The explicit type "Error" is a type: it is kept, unlike a missing one. + outcome = run( + ChildContextWrappedErrorIntegrationTest::callbackInChild, + r -> r.failCallback(r.getCallbackId("cb"), error("Error"))); + type = "Error"; + } + case CALLBACK_TIMED_OUT -> { + outcome = run( + ChildContextWrappedErrorIntegrationTest::callbackInChild, + r -> r.timeoutCallback(r.getCallbackId("cb"))); + type = CallbackTimeoutException.class.getName(); + message = null; // SDK message with the callback id, checked below + } + default -> throw new IllegalArgumentException(scenario.name()); + } + + var checkpoint = contextError(outcome.result(), "child"); + assertNotNull(checkpoint, "context checkpoint error"); + assertEquals(type, checkpoint.errorType(), "context checkpoint errorType"); + if (scenario == Scenario.CALLBACK_TIMED_OUT) { + assertEquals( + "Callback timed out: " + + outcome.result() + .getOperation("cb") + .getCallbackDetails() + .callbackId(), + checkpoint.errorMessage()); + } else { + assertEquals(message, checkpoint.errorMessage(), "context checkpoint errorMessage"); + } + + assertEquals(outcome.live(), outcome.replayed(), "live and replayed failures have the same shape"); + if (scenario == Scenario.STEP_ERROR) { + assertEquals(PaymentError.class.getName(), outcome.live().exception()); + } else { + assertEquals( + ChildContextFailedException.class.getName(), outcome.live().exception()); + assertEquals(type, outcome.live().errorType()); + } + } + + @ParameterizedTest + @EnumSource( + value = Scenario.class, + names = {"INVOKE_STOPPED", "INVOKE_TIMED_OUT"}) + void nestedContextsKeepTheSdkErrorIdentity(Scenario scenario) { + var outcome = run( + c -> c.runInChildContext( + "outer", String.class, oc -> invokeInChild(oc).toString()), + r -> { + if (scenario == Scenario.INVOKE_STOPPED) { + r.stopChainedInvoke("callee", error("States.Stopped")); + } else { + r.timeoutChainedInvoke("callee"); + } + }); + String type = scenario == Scenario.INVOKE_STOPPED + ? InvokeStoppedException.class.getName() + : InvokeTimedOutException.class.getName(); + assertEquals(type, contextError(outcome.result(), "child").errorType(), "inner context"); + assertEquals(type, contextError(outcome.result(), "outer").errorType(), "outer context"); + assertEquals(outcome.live(), outcome.replayed()); + assertEquals(type, outcome.live().errorType()); + } + + @ParameterizedTest + @EnumSource( + value = Scenario.class, + names = {"INVOKE_STOPPED", "CALLBACK_TIMED_OUT"}) + void mapItemCheckpointCarriesTheSdkErrorType(Scenario scenario) { + var runner = LocalDurableTestRunner.create(String.class, (String in, DurableContext ctx) -> { + var result = ctx.map("fanout", List.of("a"), String.class, (item, i, mc) -> { + if (scenario == Scenario.INVOKE_STOPPED) { + return mc.invoke("callee", "fn", "x", String.class); + } + return mc.createCallback("cb", String.class).get(); + }); + return result.getError(0) == null ? "no error" : "failed"; + }); + runner.run("in"); + if (scenario == Scenario.INVOKE_STOPPED) { + runner.stopChainedInvoke("callee", error("States.Stopped")); + } else { + runner.timeoutCallback(runner.getCallbackId("cb")); + } + var result = runner.runUntilComplete("in"); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + String type = scenario == Scenario.INVOKE_STOPPED + ? InvokeStoppedException.class.getName() + : CallbackTimeoutException.class.getName(); + assertEquals("failed", result.getResult(String.class)); + assertEquals(type, contextError(result, "fanout-iteration-0").errorType()); + } + + /** + * {@code waitForCallback} rethrows the callback's own exception live, so its context keeps the callback's own error + * object: a replay that can read only the completed context then reports the same error object. + */ + @ParameterizedTest + @EnumSource( + value = Scenario.class, + names = {"CALLBACK_FAILED", "CALLBACK_FAILED_NO_TYPE", "CALLBACK_TIMED_OUT"}) + void waitForCallbackContextKeepsTheCallbackErrorObject(Scenario scenario) { + var outcome = run(c -> c.waitForCallback("wfc", String.class, (id, sc) -> {}), r -> { + switch (scenario) { + case CALLBACK_TIMED_OUT -> r.timeoutCallback(r.getCallbackId("wfc-callback")); + case CALLBACK_FAILED -> r.failCallback(r.getCallbackId("wfc-callback"), error("com.acme.PaymentError")); + default -> r.failCallback(r.getCallbackId("wfc-callback"), error(null)); + } + }); + var callbackError = outcome.result() + .getOperation("wfc-callback") + .getCallbackDetails() + .error(); + assertEquals(callbackError, contextError(outcome.result(), "wfc")); + assertEquals(outcome.live().errorType(), outcome.replayed().errorType()); + String live = scenario == Scenario.CALLBACK_TIMED_OUT + ? CallbackTimeoutException.class.getName() + : CallbackFailedException.class.getName(); + assertEquals(live, outcome.live().exception()); + } +} 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 062079ed8..245f66bc0 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 @@ -25,6 +25,9 @@ import software.amazon.lambda.durable.exception.CallbackTimeoutException; import software.amazon.lambda.durable.exception.ChildContextFailedException; import software.amazon.lambda.durable.exception.DurableOperationException; +import software.amazon.lambda.durable.exception.InvokeFailedException; +import software.amazon.lambda.durable.exception.InvokeStoppedException; +import software.amazon.lambda.durable.exception.InvokeTimedOutException; import software.amazon.lambda.durable.exception.MapIterationFailedException; import software.amazon.lambda.durable.exception.ParallelBranchFailedException; import software.amazon.lambda.durable.exception.StepFailedException; @@ -35,6 +38,7 @@ import software.amazon.lambda.durable.logging.DurableLogger; import software.amazon.lambda.durable.model.DeserializedOperationResult; import software.amazon.lambda.durable.model.OperationIdentifier; +import software.amazon.lambda.durable.model.OperationSubType; import software.amazon.lambda.durable.util.ExceptionHelper; /** @@ -200,7 +204,10 @@ private void handleChildContextFailure(Throwable exception) { final ErrorObject errorObject; if (exception instanceof DurableOperationException opEx) { - errorObject = opEx.getErrorObject(); + // waitForCallback rethrows the callback's own exception, whose error object is the callback's; its context + // keeps that same object so a replay that can only read the context reports the same error object. + errorObject = + getSubType() == OperationSubType.WAIT_FOR_CALLBACK ? opEx.getErrorObject() : wrappedError(opEx); } else { errorObject = serializeException(exception); } @@ -245,9 +252,70 @@ public T get() { } } + /** + * Returns the error to checkpoint for a context that failed because an SDK operation exception escaped it: the + * wrapped (original) error, never the wrapper. + * + *

A wrapper such as {@code StepFailedException} or {@code CallbackFailedException} carries the original error's + * {@link ErrorObject}, which is checkpointed as it is. An SDK error that names the failure itself (a timed-out + * callback, a timed-out or stopped invoke), and a callback or invoke failure whose error has no type, is + * checkpointed under its own class name instead, keeping the original message, data and stack trace when there are + * any: its {@link ErrorObject} is missing or is the stop error, so it does not say which error failed the context. + * The context checkpoint then says which error failed it without its children, which the backend can prune once the + * context completes. Any other operation exception keeps its error object as it is, including a missing one. + */ + static ErrorObject wrappedError(DurableOperationException exception) { + var original = exception.getErrorObject(); + boolean namedBySdk = exception instanceof CallbackTimeoutException + || exception instanceof InvokeTimedOutException + || exception instanceof InvokeStoppedException; + boolean typelessFailure = (exception.getClass() == CallbackFailedException.class + || exception.getClass() == InvokeFailedException.class) + && (original == null || isBlankType(original.errorType())); + if (!namedBySdk && !typelessFailure) { + return original; + } + var message = + original != null && original.errorMessage() != null ? original.errorMessage() : exception.getMessage(); + var stackTrace = original != null + && original.stackTrace() != null + && !original.stackTrace().isEmpty() + ? original.stackTrace() + : ExceptionHelper.serializeStackTrace(exception.getStackTrace()); + return ErrorObject.builder() + .errorType(exception.getClass().getName()) + .errorMessage(message) + .errorData(original != null ? original.errorData() : null) + .stackTrace(stackTrace) + .build(); + } + + /** + * {@code true} for a missing type or one made only of ASCII whitespace (space, tab, line feed, carriage return, + * form feed, vertical tab). + */ + private static boolean isBlankType(String type) { + return type == null || type.chars().allMatch(c -> c == ' ' || (c >= '\t' && c <= '\r')); + } + + /** {@code true} when {@code errorType} names an SDK operation exception, which is never rebuilt from error data. */ + private static boolean isSdkOperationException(String errorType) { + if (errorType == null || !errorType.startsWith(DurableOperationException.class.getPackageName() + ".")) { + return false; + } + try { + return DurableOperationException.class.isAssignableFrom(Class.forName(errorType)); + } catch (ClassNotFoundException e) { + return false; + } + } + private Throwable translateException(Operation op, ErrorObject errorObject) { - // Attempt to reconstruct and throw the original exception - Throwable original = deserializeException(errorObject); + // Attempt to reconstruct and throw the original exception. An SDK operation exception recorded by + // wrappedError is not rebuilt from its data: the context reports it through its own failure type below. + Throwable original = errorObject != null && isSdkOperationException(errorObject.errorType()) + ? null + : deserializeException(errorObject); if (original != null) { return original; }