From d030d5278d2ad61afc01444614ea1a1bd3394918 Mon Sep 17 00:00:00 2001 From: Pooya Paridel Date: Thu, 22 Oct 2026 09:13:34 -0700 Subject: [PATCH 1/2] fix(sdk): checkpoint the wrapped error for failed contexts A context that failed because an SDK error escaped it checkpointed the inner operation's raw ErrorObject. For a timed-out callback or invoke that object is null, and for a stopped invoke it is the stop error (for example States.Stopped), so the context checkpoint lost which error failed it. The context now checkpoints the wrapped error: the original ErrorObject of a wrapper such as StepFailedException when it has a type, otherwise an ErrorObject typed with the SDK error's class name (CallbackTimeoutException, InvokeTimedOutException, InvokeStoppedException, or the exception class when the error has no type), keeping message, data and stack trace. SDK operation exception types are not rebuilt from error data on translation. waitForCallback keeps the callback's own ErrorObject, because it rethrows the callback's exception live. --- ...ildContextWrappedErrorIntegrationTest.java | 303 ++++++++++++++++++ .../operation/ChildContextOperation.java | 69 +++- 2 files changed, 369 insertions(+), 3 deletions(-) create mode 100644 sdk-integration-tests/src/test/java/software/amazon/lambda/durable/ChildContextWrappedErrorIntegrationTest.java 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..0352fc69f 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,8 @@ 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.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 +37,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 +203,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 +251,66 @@ 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}; when that object has a type, it is checkpointed as it is. An SDK error that names the + * failure itself (a timed-out callback, a timed-out or stopped invoke), or an exception whose {@link ErrorObject} + * is missing or has no type, is checkpointed under its own class name, keeping the original message, data and stack + * trace when there are any. The context checkpoint then always says which error failed it, so a reader of the + * checkpoint never has to look at the context's children, which the backend can prune once the context completes. + */ + static ErrorObject wrappedError(DurableOperationException exception) { + var original = exception.getErrorObject(); + boolean namedBySdk = exception instanceof CallbackTimeoutException + || exception instanceof InvokeTimedOutException + || exception instanceof InvokeStoppedException; + if (!namedBySdk && original != null && !isBlankType(original.errorType())) { + 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; } From 8dde1e75606b29601b642f6884d82794498757bb Mon Sep 17 00:00:00 2001 From: Pooya Paridel Date: Thu, 22 Oct 2026 09:23:27 -0700 Subject: [PATCH 2/2] fix(sdk): keep other operation exceptions' error objects as they are Only an SDK error that names the failure itself (timed-out callback, timed-out or stopped invoke) and a CallbackFailedException or InvokeFailedException whose error has no type are checkpointed under their class name. Any other DurableOperationException that escapes a context, such as one a user throws with an empty or missing error object, keeps that object, as the OpenTelemetry plugin's VirtualOperationErrorStatusTest requires. --- .../operation/ChildContextOperation.java | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) 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 0352fc69f..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,7 @@ 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; @@ -256,18 +257,22 @@ public T get() { * wrapped (original) error, never the wrapper. * *

A wrapper such as {@code StepFailedException} or {@code CallbackFailedException} carries the original error's - * {@link ErrorObject}; when that object has a type, it is checkpointed as it is. An SDK error that names the - * failure itself (a timed-out callback, a timed-out or stopped invoke), or an exception whose {@link ErrorObject} - * is missing or has no type, is checkpointed under its own class name, keeping the original message, data and stack - * trace when there are any. The context checkpoint then always says which error failed it, so a reader of the - * checkpoint never has to look at the context's children, which the backend can prune once the context completes. + * {@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; - if (!namedBySdk && original != null && !isBlankType(original.errorType())) { + boolean typelessFailure = (exception.getClass() == CallbackFailedException.class + || exception.getClass() == InvokeFailedException.class) + && (original == null || isBlankType(original.errorType())); + if (!namedBySdk && !typelessFailure) { return original; } var message =