Skip to content

Commit 9d1e06d

Browse files
ParidelPooyaParidelPooya
committed
Merge branch 'fix/checkpoint-wrapped-error' into fix/telemetry-align-with-js
2 parents 0d12f6c + d030d52 commit 9d1e06d

2 files changed

Lines changed: 369 additions & 3 deletions

File tree

‎sdk-integration-tests/src/test/java/software/amazon/lambda/durable/ChildContextWrappedErrorIntegrationTest.java‎

Lines changed: 303 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,303 @@
1+
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
package software.amazon.lambda.durable;
4+
5+
import static org.junit.jupiter.api.Assertions.assertEquals;
6+
import static org.junit.jupiter.api.Assertions.assertNotNull;
7+
8+
import java.time.Duration;
9+
import java.util.ArrayList;
10+
import java.util.Collections;
11+
import java.util.List;
12+
import java.util.function.Consumer;
13+
import java.util.function.Function;
14+
import org.junit.jupiter.params.ParameterizedTest;
15+
import org.junit.jupiter.params.provider.EnumSource;
16+
import software.amazon.awssdk.services.lambda.model.ErrorObject;
17+
import software.amazon.lambda.durable.config.StepConfig;
18+
import software.amazon.lambda.durable.exception.CallbackFailedException;
19+
import software.amazon.lambda.durable.exception.CallbackTimeoutException;
20+
import software.amazon.lambda.durable.exception.ChildContextFailedException;
21+
import software.amazon.lambda.durable.exception.InvokeStoppedException;
22+
import software.amazon.lambda.durable.exception.InvokeTimedOutException;
23+
import software.amazon.lambda.durable.execution.SuspendExecutionException;
24+
import software.amazon.lambda.durable.model.ExecutionStatus;
25+
import software.amazon.lambda.durable.retry.RetryStrategies;
26+
import software.amazon.lambda.durable.testing.LocalDurableTestRunner;
27+
import software.amazon.lambda.durable.testing.TestResult;
28+
29+
/**
30+
* A failed context checkpoints the wrapped (original) error, never the SDK wrapper, and the live and replayed failure
31+
* that user code catches has the same type and error object.
32+
*/
33+
class ChildContextWrappedErrorIntegrationTest {
34+
35+
static final String MSG = "card declined";
36+
37+
public static class PaymentError extends RuntimeException {
38+
public PaymentError(String message) {
39+
super(message);
40+
}
41+
}
42+
43+
private static final StepConfig NO_RETRY =
44+
StepConfig.builder().retryStrategy(RetryStrategies.Presets.NO_RETRY).build();
45+
46+
/** What user code caught: the exception class and the type and message of its error object. */
47+
private record Caught(String exception, String errorType, String errorMessage) {}
48+
49+
private static Caught caught(Throwable t) {
50+
var eo = t instanceof software.amazon.lambda.durable.exception.DurableOperationException d
51+
? d.getErrorObject()
52+
: null;
53+
return new Caught(
54+
t.getClass().getName(),
55+
eo == null ? null : eo.errorType(),
56+
eo == null ? t.getMessage() : eo.errorMessage());
57+
}
58+
59+
private record Outcome(Caught live, Caught replayed, TestResult<String> result) {}
60+
61+
/**
62+
* Runs {@code failing}, records what it threw, waits so that a new invocation starts, and records what the replay
63+
* of {@code failing} throws.
64+
*/
65+
private static Outcome run(
66+
Function<DurableContext, Object> failing, Consumer<LocalDurableTestRunner<String, String>> drive) {
67+
List<Caught> seen = Collections.synchronizedList(new ArrayList<>());
68+
var runner = LocalDurableTestRunner.create(String.class, (String in, DurableContext ctx) -> {
69+
try {
70+
failing.apply(ctx);
71+
throw new IllegalStateException("did not fail");
72+
} catch (SuspendExecutionException e) {
73+
throw e;
74+
} catch (RuntimeException e) {
75+
seen.add(caught(e));
76+
}
77+
ctx.wait("pause", Duration.ofSeconds(1));
78+
return "done";
79+
});
80+
if (drive != null) {
81+
runner.run("in");
82+
drive.accept(runner);
83+
}
84+
var result = runner.runUntilComplete("in");
85+
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
86+
assertEquals(2, seen.size(), "one live and one replayed failure: " + seen);
87+
return new Outcome(seen.get(0), seen.get(1), result);
88+
}
89+
90+
private static ErrorObject contextError(TestResult<String> result, String name) {
91+
var details = result.getOperation(name).getContextDetails();
92+
assertNotNull(details, name + " context details");
93+
return details.error();
94+
}
95+
96+
private static ErrorObject error(String type) {
97+
return ErrorObject.builder().errorType(type).errorMessage(MSG).build();
98+
}
99+
100+
private static Object invokeInChild(DurableContext ctx) {
101+
return ctx.runInChildContext("child", String.class, cc -> cc.invoke("callee", "fn", "x", String.class));
102+
}
103+
104+
private static Object callbackInChild(DurableContext ctx) {
105+
return ctx.runInChildContext(
106+
"child",
107+
String.class,
108+
cc -> cc.createCallback("cb", String.class).get());
109+
}
110+
111+
enum Scenario {
112+
STEP_ERROR,
113+
INVOKE_FAILED,
114+
INVOKE_STOPPED,
115+
INVOKE_STOPPED_USER_TYPE,
116+
INVOKE_TIMED_OUT,
117+
CALLBACK_FAILED,
118+
CALLBACK_FAILED_NO_TYPE,
119+
CALLBACK_FAILED_TYPE_ERROR,
120+
CALLBACK_TIMED_OUT
121+
}
122+
123+
@ParameterizedTest
124+
@EnumSource(Scenario.class)
125+
void contextCheckpointsTheWrappedErrorAndReplaysTheSameShape(Scenario scenario) {
126+
Outcome outcome;
127+
String type;
128+
String message = MSG;
129+
switch (scenario) {
130+
case STEP_ERROR -> {
131+
outcome = run(
132+
c -> c.runInChildContext(
133+
"child",
134+
String.class,
135+
cc -> cc.step(
136+
"charge",
137+
String.class,
138+
sc -> {
139+
throw new PaymentError(MSG);
140+
},
141+
NO_RETRY)),
142+
null);
143+
type = PaymentError.class.getName();
144+
}
145+
case INVOKE_FAILED -> {
146+
outcome = run(
147+
ChildContextWrappedErrorIntegrationTest::invokeInChild,
148+
r -> r.failChainedInvoke("callee", error("com.acme.PaymentError")));
149+
type = "com.acme.PaymentError";
150+
}
151+
case INVOKE_STOPPED -> {
152+
outcome = run(
153+
ChildContextWrappedErrorIntegrationTest::invokeInChild,
154+
r -> r.stopChainedInvoke("callee", error("States.Stopped")));
155+
type = InvokeStoppedException.class.getName();
156+
}
157+
case INVOKE_STOPPED_USER_TYPE -> {
158+
outcome = run(
159+
ChildContextWrappedErrorIntegrationTest::invokeInChild,
160+
r -> r.stopChainedInvoke("callee", error("com.acme.PaymentError")));
161+
type = InvokeStoppedException.class.getName();
162+
}
163+
case INVOKE_TIMED_OUT -> {
164+
outcome = run(
165+
ChildContextWrappedErrorIntegrationTest::invokeInChild, r -> r.timeoutChainedInvoke("callee"));
166+
type = InvokeTimedOutException.class.getName();
167+
message = null;
168+
}
169+
case CALLBACK_FAILED -> {
170+
outcome = run(
171+
ChildContextWrappedErrorIntegrationTest::callbackInChild,
172+
r -> r.failCallback(r.getCallbackId("cb"), error("com.acme.PaymentError")));
173+
type = "com.acme.PaymentError";
174+
}
175+
case CALLBACK_FAILED_NO_TYPE -> {
176+
outcome = run(
177+
ChildContextWrappedErrorIntegrationTest::callbackInChild,
178+
r -> r.failCallback(r.getCallbackId("cb"), error(null)));
179+
type = CallbackFailedException.class.getName();
180+
}
181+
case CALLBACK_FAILED_TYPE_ERROR -> {
182+
// The explicit type "Error" is a type: it is kept, unlike a missing one.
183+
outcome = run(
184+
ChildContextWrappedErrorIntegrationTest::callbackInChild,
185+
r -> r.failCallback(r.getCallbackId("cb"), error("Error")));
186+
type = "Error";
187+
}
188+
case CALLBACK_TIMED_OUT -> {
189+
outcome = run(
190+
ChildContextWrappedErrorIntegrationTest::callbackInChild,
191+
r -> r.timeoutCallback(r.getCallbackId("cb")));
192+
type = CallbackTimeoutException.class.getName();
193+
message = null; // SDK message with the callback id, checked below
194+
}
195+
default -> throw new IllegalArgumentException(scenario.name());
196+
}
197+
198+
var checkpoint = contextError(outcome.result(), "child");
199+
assertNotNull(checkpoint, "context checkpoint error");
200+
assertEquals(type, checkpoint.errorType(), "context checkpoint errorType");
201+
if (scenario == Scenario.CALLBACK_TIMED_OUT) {
202+
assertEquals(
203+
"Callback timed out: "
204+
+ outcome.result()
205+
.getOperation("cb")
206+
.getCallbackDetails()
207+
.callbackId(),
208+
checkpoint.errorMessage());
209+
} else {
210+
assertEquals(message, checkpoint.errorMessage(), "context checkpoint errorMessage");
211+
}
212+
213+
assertEquals(outcome.live(), outcome.replayed(), "live and replayed failures have the same shape");
214+
if (scenario == Scenario.STEP_ERROR) {
215+
assertEquals(PaymentError.class.getName(), outcome.live().exception());
216+
} else {
217+
assertEquals(
218+
ChildContextFailedException.class.getName(), outcome.live().exception());
219+
assertEquals(type, outcome.live().errorType());
220+
}
221+
}
222+
223+
@ParameterizedTest
224+
@EnumSource(
225+
value = Scenario.class,
226+
names = {"INVOKE_STOPPED", "INVOKE_TIMED_OUT"})
227+
void nestedContextsKeepTheSdkErrorIdentity(Scenario scenario) {
228+
var outcome = run(
229+
c -> c.runInChildContext(
230+
"outer", String.class, oc -> invokeInChild(oc).toString()),
231+
r -> {
232+
if (scenario == Scenario.INVOKE_STOPPED) {
233+
r.stopChainedInvoke("callee", error("States.Stopped"));
234+
} else {
235+
r.timeoutChainedInvoke("callee");
236+
}
237+
});
238+
String type = scenario == Scenario.INVOKE_STOPPED
239+
? InvokeStoppedException.class.getName()
240+
: InvokeTimedOutException.class.getName();
241+
assertEquals(type, contextError(outcome.result(), "child").errorType(), "inner context");
242+
assertEquals(type, contextError(outcome.result(), "outer").errorType(), "outer context");
243+
assertEquals(outcome.live(), outcome.replayed());
244+
assertEquals(type, outcome.live().errorType());
245+
}
246+
247+
@ParameterizedTest
248+
@EnumSource(
249+
value = Scenario.class,
250+
names = {"INVOKE_STOPPED", "CALLBACK_TIMED_OUT"})
251+
void mapItemCheckpointCarriesTheSdkErrorType(Scenario scenario) {
252+
var runner = LocalDurableTestRunner.create(String.class, (String in, DurableContext ctx) -> {
253+
var result = ctx.map("fanout", List.of("a"), String.class, (item, i, mc) -> {
254+
if (scenario == Scenario.INVOKE_STOPPED) {
255+
return mc.invoke("callee", "fn", "x", String.class);
256+
}
257+
return mc.createCallback("cb", String.class).get();
258+
});
259+
return result.getError(0) == null ? "no error" : "failed";
260+
});
261+
runner.run("in");
262+
if (scenario == Scenario.INVOKE_STOPPED) {
263+
runner.stopChainedInvoke("callee", error("States.Stopped"));
264+
} else {
265+
runner.timeoutCallback(runner.getCallbackId("cb"));
266+
}
267+
var result = runner.runUntilComplete("in");
268+
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
269+
String type = scenario == Scenario.INVOKE_STOPPED
270+
? InvokeStoppedException.class.getName()
271+
: CallbackTimeoutException.class.getName();
272+
assertEquals("failed", result.getResult(String.class));
273+
assertEquals(type, contextError(result, "fanout-iteration-0").errorType());
274+
}
275+
276+
/**
277+
* {@code waitForCallback} rethrows the callback's own exception live, so its context keeps the callback's own error
278+
* object: a replay that can read only the completed context then reports the same error object.
279+
*/
280+
@ParameterizedTest
281+
@EnumSource(
282+
value = Scenario.class,
283+
names = {"CALLBACK_FAILED", "CALLBACK_FAILED_NO_TYPE", "CALLBACK_TIMED_OUT"})
284+
void waitForCallbackContextKeepsTheCallbackErrorObject(Scenario scenario) {
285+
var outcome = run(c -> c.waitForCallback("wfc", String.class, (id, sc) -> {}), r -> {
286+
switch (scenario) {
287+
case CALLBACK_TIMED_OUT -> r.timeoutCallback(r.getCallbackId("wfc-callback"));
288+
case CALLBACK_FAILED -> r.failCallback(r.getCallbackId("wfc-callback"), error("com.acme.PaymentError"));
289+
default -> r.failCallback(r.getCallbackId("wfc-callback"), error(null));
290+
}
291+
});
292+
var callbackError = outcome.result()
293+
.getOperation("wfc-callback")
294+
.getCallbackDetails()
295+
.error();
296+
assertEquals(callbackError, contextError(outcome.result(), "wfc"));
297+
assertEquals(outcome.live().errorType(), outcome.replayed().errorType());
298+
String live = scenario == Scenario.CALLBACK_TIMED_OUT
299+
? CallbackTimeoutException.class.getName()
300+
: CallbackFailedException.class.getName();
301+
assertEquals(live, outcome.live().exception());
302+
}
303+
}

0 commit comments

Comments
 (0)