Skip to content

Commit 7c1e17d

Browse files
hsilanhsilan
committed
Merge branch 'main' into retryable-operation
2 parents 9af4a42 + 27bd968 commit 7c1e17d

16 files changed

Lines changed: 329 additions & 20 deletions

File tree

‎examples/src/main/java/software/amazon/lambda/durable/examples/child/VirtualChildContextExample.java‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
package software.amazon.lambda.durable.examples.child;
4+
5+
import java.time.Duration;
6+
import software.amazon.lambda.durable.DurableContext;
7+
import software.amazon.lambda.durable.DurableFuture;
8+
import software.amazon.lambda.durable.DurableHandler;
9+
import software.amazon.lambda.durable.config.RunInChildContextConfig;
10+
import software.amazon.lambda.durable.examples.types.GreetingRequest;
11+
12+
/**
13+
* Example demonstrating virtual child context workflows with the Durable Execution SDK.
14+
*
15+
* <p>This handler runs three concurrent child contexts using {@code runInChildContextAsync}:
16+
*
17+
* <ol>
18+
* <li><b>Order validation</b> — performs a step then suspends via {@code wait()} before completing
19+
* <li><b>Inventory check</b> — performs a step then suspends via {@code wait()} before completing
20+
* <li><b>Shipping estimate</b> — nests another child context inside it to demonstrate hierarchical contexts
21+
* </ol>
22+
*
23+
* <p>All three child contexts run concurrently. Results are collected with {@link DurableFuture#allOf} and combined
24+
* into a summary string.
25+
*/
26+
public class VirtualChildContextExample extends DurableHandler<GreetingRequest, String> {
27+
28+
@Override
29+
public String handleRequest(GreetingRequest input, DurableContext context) {
30+
var name = input.getName();
31+
context.getLogger().info("Starting child context workflow for {}", name);
32+
33+
// Child context 1: Order validation — step + wait + step
34+
var orderFuture = context.runInChildContextAsync(
35+
"order-validation",
36+
String.class,
37+
child -> {
38+
var prepared = child.step("prepare-order", String.class, stepCtx -> "Order for " + name);
39+
child.getLogger().info("Order prepared, waiting for validation");
40+
41+
child.wait("validation-delay", Duration.ofSeconds(5));
42+
43+
return child.step("validate-order", String.class, stepCtx -> prepared + " [validated]");
44+
},
45+
RunInChildContextConfig.builder().isVirtual(true).build());
46+
47+
// Child context 2: Inventory check — step + wait + step
48+
var inventoryFuture = context.runInChildContextAsync(
49+
"inventory-check",
50+
String.class,
51+
child -> {
52+
var stock = child.step("check-stock", String.class, stepCtx -> "Stock available for " + name);
53+
child.getLogger().info("Stock checked, waiting for confirmation");
54+
55+
child.wait("confirmation-delay", Duration.ofSeconds(3));
56+
57+
return child.step("confirm-inventory", String.class, stepCtx -> stock + " [confirmed]");
58+
},
59+
RunInChildContextConfig.builder().isVirtual(true).build());
60+
61+
// Child context 3: Shipping estimate — nests a child context inside it
62+
var shippingFuture = context.runInChildContextAsync(
63+
"shipping-estimate",
64+
String.class,
65+
child -> {
66+
var baseRate = child.step("calculate-base-rate", String.class, stepCtx -> "Base rate for " + name);
67+
68+
// Nested child context: calculate regional adjustment
69+
var adjustment = child.runInChildContext(
70+
"regional-adjustment",
71+
String.class,
72+
nested -> nested.step(
73+
"lookup-region", String.class, stepCtx -> baseRate + " + regional adjustment"),
74+
RunInChildContextConfig.builder().isVirtual(true).build());
75+
76+
return child.step("finalize-shipping", String.class, stepCtx -> adjustment + " [shipping ready]");
77+
},
78+
RunInChildContextConfig.builder().isVirtual(true).build());
79+
80+
// Collect all results using allOf
81+
context.getLogger().info("Waiting for all child contexts to complete");
82+
var results = DurableFuture.allOf(orderFuture, inventoryFuture, shippingFuture);
83+
84+
// Combine into summary
85+
var summary = String.join(" | ", results);
86+
context.getLogger().info("All child contexts complete: {}", summary);
87+
88+
return summary;
89+
}
90+
}

‎examples/src/test/java/software/amazon/lambda/durable/examples/CloudBasedIntegrationTest.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -530,6 +530,23 @@ void testChildContextExample() {
530530
assertNotNull(runner.getOperation("shipping-estimate"));
531531
}
532532

533+
@Test
534+
void testVirtualChildContextExample() {
535+
var runner = CloudDurableTestRunner.create(
536+
arn("virtual-child-context-example"), GreetingRequest.class, String.class, lambdaClient);
537+
var result = runner.run(new GreetingRequest("Alice"));
538+
539+
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
540+
assertEquals(
541+
"Order for Alice [validated] | Stock available for Alice [confirmed] | Base rate for Alice + regional adjustment [shipping ready]",
542+
result.getResult());
543+
544+
// Verify step operations in child context were tracked
545+
assertNotNull(runner.getOperation("validate-order"));
546+
assertNotNull(runner.getOperation("confirm-inventory"));
547+
assertNotNull(runner.getOperation("finalize-shipping"));
548+
}
549+
533550
@ParameterizedTest
534551
@CsvSource({"100, 1000, 20", "500, 2000, 30", "1000, 3000, 50"})
535552
void testManyAsyncStepsExample(int steps, long maxExecutionTime, long maxReplayTime) {

‎examples/src/test/java/software/amazon/lambda/durable/examples/child/VirtualChildContextExampleTest.java‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
package software.amazon.lambda.durable.examples.child;
4+
5+
import static org.junit.jupiter.api.Assertions.assertEquals;
6+
7+
import org.junit.jupiter.api.Test;
8+
import software.amazon.lambda.durable.examples.types.GreetingRequest;
9+
import software.amazon.lambda.durable.model.ExecutionStatus;
10+
import software.amazon.lambda.durable.testing.LocalDurableTestRunner;
11+
12+
class VirtualChildContextExampleTest {
13+
14+
@Test
15+
void testVirtualChildContextExampleRunsToCompletion() {
16+
var handler = new VirtualChildContextExample();
17+
var runner = LocalDurableTestRunner.create(GreetingRequest.class, handler);
18+
19+
var input = new GreetingRequest("Alice");
20+
var result = runner.runUntilComplete(input);
21+
22+
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
23+
assertEquals(
24+
"Order for Alice [validated] | Stock available for Alice [confirmed] | Base rate for Alice + regional adjustment [shipping ready]",
25+
result.getResult(String.class));
26+
}
27+
28+
@Test
29+
void testVirtualChildContextExampleSuspendsOnFirstRun() {
30+
var handler = new VirtualChildContextExample();
31+
var runner = LocalDurableTestRunner.create(GreetingRequest.class, handler);
32+
33+
var input = new GreetingRequest("Bob");
34+
35+
// First run should suspend due to wait operations inside child contexts
36+
var result = runner.run(input);
37+
assertEquals(ExecutionStatus.PENDING, result.getStatus());
38+
}
39+
40+
@Test
41+
void testVirtualChildContextExampleReplay() {
42+
var handler = new VirtualChildContextExample();
43+
var runner = LocalDurableTestRunner.create(GreetingRequest.class, handler);
44+
45+
var input = new GreetingRequest("Alice");
46+
47+
// First full execution
48+
var result1 = runner.runUntilComplete(input);
49+
assertEquals(ExecutionStatus.SUCCEEDED, result1.getStatus());
50+
51+
// Replay — should return cached results
52+
var result2 = runner.run(input);
53+
assertEquals(ExecutionStatus.SUCCEEDED, result2.getStatus());
54+
assertEquals(result1.getResult(String.class), result2.getResult(String.class));
55+
}
56+
}

‎examples/template.yaml‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -296,6 +296,23 @@ Resources:
296296
- lambda:GetDurableExecutionState
297297
Resource: !Sub "arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:child-context-example-${JavaVersion}-runtime"
298298

299+
VirtualChildContextExampleFunction:
300+
Type: AWS::Serverless::Function
301+
Properties:
302+
FunctionName: !Join
303+
- '-'
304+
- - 'virtual-child-context-example'
305+
- !Ref JavaVersion
306+
- runtime
307+
Handler: "software.amazon.lambda.durable.examples.child.VirtualChildContextExample"
308+
Policies:
309+
- Statement:
310+
- Effect: Allow
311+
Action:
312+
- lambda:CheckpointDurableExecutions
313+
- lambda:GetDurableExecutionState
314+
Resource: !Sub "arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:virtual-child-context-example-${JavaVersion}-runtime"
315+
299316
WaitAsyncExampleFunction:
300317
Type: AWS::Serverless::Function
301318
Properties:

‎pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@
5050
<maven.compiler.source>17</maven.compiler.source>
5151
<maven.compiler.target>17</maven.compiler.target>
5252
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
53-
<aws.sdk.version>2.43.0</aws.sdk.version>
53+
<aws.sdk.version>2.43.1</aws.sdk.version>
5454
<jackson.version>2.21.3</jackson.version>
5555
<junit.version>6.0.3</junit.version>
5656
<mockito.version>5.23.0</mockito.version>

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

Lines changed: 60 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
import java.util.concurrent.atomic.AtomicInteger;
99
import org.junit.jupiter.api.Test;
1010
import software.amazon.awssdk.services.lambda.model.OperationType;
11+
import software.amazon.lambda.durable.config.RunInChildContextConfig;
1112
import software.amazon.lambda.durable.model.ExecutionStatus;
1213
import software.amazon.lambda.durable.testing.LocalDurableTestRunner;
1314

@@ -42,6 +43,34 @@ void childContextResultSurvivesReplay() {
4243
assertEquals(1, childExecutionCount.get(), "Child function should not re-execute on replay");
4344
}
4445

46+
@Test
47+
void virtualChildContextResultSurvivesReplay() {
48+
var childExecutionCount = new AtomicInteger(0);
49+
50+
var runner = LocalDurableTestRunner.create(
51+
String.class,
52+
(input, ctx) -> ctx.runInChildContext(
53+
"compute",
54+
TypeToken.get(String.class),
55+
child -> {
56+
childExecutionCount.incrementAndGet();
57+
return child.step("work", String.class, stepCtx -> "result-" + input);
58+
},
59+
RunInChildContextConfig.builder().isVirtual(true).build()));
60+
61+
// First run - executes child context
62+
var result = runner.runUntilComplete("test");
63+
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
64+
assertEquals("result-test", result.getResult(String.class));
65+
assertEquals(1, childExecutionCount.get());
66+
67+
// Second run - replays, should return cached result without re-executing
68+
result = runner.run("test");
69+
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
70+
assertEquals("result-test", result.getResult(String.class));
71+
assertEquals(2, childExecutionCount.get(), "Child function should re-execute on replay");
72+
}
73+
4574
/**
4675
* A child context that fails with a reconstructable exception SHALL preserve the exception type, message, and error
4776
* details through the checkpoint-and-replay cycle.
@@ -62,7 +91,7 @@ void childContextExceptionPreservedOnReplay() {
6291
assertEquals(ExecutionStatus.FAILED, result.getStatus());
6392
assertEquals(1, childExecutionCount.get());
6493

65-
// Second run - replays, should throw same exception without re-executing
94+
// Second run - replays, should throw same exception with re-executing
6695
result = runner.run("test");
6796
assertEquals(ExecutionStatus.FAILED, result.getStatus());
6897
assertTrue(result.getError().isPresent());
@@ -72,6 +101,36 @@ void childContextExceptionPreservedOnReplay() {
72101
assertEquals(1, childExecutionCount.get(), "Child function should not re-execute on failed replay");
73102
}
74103

104+
@Test
105+
void virtualChildContextExceptionPreservedOnReplay() {
106+
var childExecutionCount = new AtomicInteger(0);
107+
108+
var runner = LocalDurableTestRunner.create(
109+
String.class,
110+
(input, ctx) -> ctx.runInChildContext(
111+
"failing",
112+
String.class,
113+
child -> {
114+
childExecutionCount.incrementAndGet();
115+
throw new IllegalArgumentException("bad input: " + input);
116+
},
117+
RunInChildContextConfig.builder().isVirtual(true).build()));
118+
119+
// First run - child context fails
120+
var result = runner.run("test");
121+
assertEquals(ExecutionStatus.FAILED, result.getStatus());
122+
assertEquals(1, childExecutionCount.get());
123+
124+
// Second run - replays, should throw same exception with re-executing
125+
result = runner.run("test");
126+
assertEquals(ExecutionStatus.FAILED, result.getStatus());
127+
assertTrue(result.getError().isPresent());
128+
var error = result.getError().get();
129+
assertEquals("java.lang.IllegalArgumentException", error.errorType());
130+
assertEquals("bad input: test", error.errorMessage());
131+
assertEquals(2, childExecutionCount.get(), "Child function should re-execute on failed replay");
132+
}
133+
75134
/** Operations checkpointed from within a child context SHALL have the child context's ID as their parentId. */
76135
@Test
77136
void operationsInChildContextHaveCorrectParentId() {

‎sdk/src/main/java/software/amazon/lambda/durable/config/RunInChildContextConfig.java‎

Lines changed: 23 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
// SPDX-License-Identifier: Apache-2.0
33
package software.amazon.lambda.durable.config;
44

5+
import java.util.Objects;
56
import software.amazon.lambda.durable.serde.SerDes;
67

78
/**
@@ -11,9 +12,11 @@
1112
*/
1213
public class RunInChildContextConfig {
1314
private final SerDes serDes;
15+
private final Boolean isVirtual;
1416

1517
private RunInChildContextConfig(Builder builder) {
1618
this.serDes = builder.serDes;
19+
this.isVirtual = Objects.requireNonNullElse(builder.isVirtual, false);
1720
}
1821

1922
/**
@@ -24,8 +27,13 @@ public SerDes serDes() {
2427
return serDes;
2528
}
2629

30+
/** Returns true if the context operation will not be checkpointed, false otherwise. */
31+
public Boolean isVirtual() {
32+
return isVirtual;
33+
}
34+
2735
public Builder toBuilder() {
28-
return new Builder(serDes);
36+
return new Builder().serDes(serDes).isVirtual(isVirtual);
2937
}
3038

3139
/**
@@ -34,16 +42,15 @@ public Builder toBuilder() {
3442
* @return a new Builder instance
3543
*/
3644
public static Builder builder() {
37-
return new Builder(null);
45+
return new Builder();
3846
}
3947

4048
/** Builder for creating StepConfig instances. */
4149
public static class Builder {
4250
private SerDes serDes;
51+
private Boolean isVirtual;
4352

44-
public Builder(SerDes serDes) {
45-
this.serDes = serDes;
46-
}
53+
private Builder() {}
4754

4855
/**
4956
* Sets a custom serializer for the step.
@@ -60,6 +67,17 @@ public Builder serDes(SerDes serDes) {
6067
return this;
6168
}
6269

70+
/**
71+
* Sets whether the context is virtual (not checkpointed) or not.
72+
*
73+
* @param isVirtual true if the context is virtual (no checkpointing), false otherwise
74+
* @return this builder for method chaining
75+
*/
76+
public Builder isVirtual(Boolean isVirtual) {
77+
this.isVirtual = isVirtual;
78+
return this;
79+
}
80+
6381
/**
6482
* Builds the RunInChildContextConfig instance.
6583
*

‎sdk/src/main/java/software/amazon/lambda/durable/context/DurableContextImpl.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -515,6 +515,6 @@ private String nextOperationId() {
515515
* @return the parent of this context if virtual, otherwise this context id
516516
*/
517517
public String getParentId() {
518-
return isVirtual ? parentContext.getContextId() : getContextId();
518+
return isVirtual ? parentContext.getParentId() : getContextId();
519519
}
520520
}

‎sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,14 +5,14 @@
55
import java.time.Instant;
66
import java.util.ArrayList;
77
import java.util.Collections;
8-
import java.util.HashMap;
98
import java.util.HashSet;
109
import java.util.List;
1110
import java.util.Map;
1211
import java.util.Objects;
1312
import java.util.Set;
1413
import java.util.concurrent.CancellationException;
1514
import java.util.concurrent.CompletableFuture;
15+
import java.util.concurrent.ConcurrentHashMap;
1616
import java.util.concurrent.ThreadPoolExecutor;
1717
import java.util.concurrent.atomic.AtomicReference;
1818
import java.util.stream.Collectors;
@@ -59,7 +59,7 @@ public class ExecutionManager implements AutoCloseable {
5959
private final DurableConfig durableConfig;
6060

6161
// ===== Thread Coordination =====
62-
private final Map<String, BaseDurableOperation> registeredOperations = Collections.synchronizedMap(new HashMap<>());
62+
private final Map<String, BaseDurableOperation> registeredOperations = new ConcurrentHashMap<>();
6363
private final Set<String> activeThreads = Collections.synchronizedSet(new HashSet<>());
6464
private static final ThreadLocal<ThreadContext> currentThreadContext = new ThreadLocal<>();
6565
private final CompletableFuture<Void> executionExceptionFuture = new CompletableFuture<>();

0 commit comments

Comments
 (0)