Merge remote-tracking branch 'origin/main' into child-context

Author: zhongkechen

.github/workflows/e2e-tests.yml

@@ -30,7 +30,6 @@ permissions:
 
 jobs:
   e2e-tests:
-    if: github.event_name == 'pull_request'
     env:
       AWS_REGION: us-west-2
     runs-on: ubuntu-latest
@@ -74,5 +73,5 @@ jobs:
       - name: Build locally
         run: mvn -B -q -Dmaven.test.skip=true install --file pom.xml
       - name: Cloud Based Integration Tests
-        run: mvn clean test -B -Dtest.cloud.enabled=true -Dtest=CloudBasedIntegrationTest
-        working-directory: ./examples
+        run: mvn clean test -B -Dtest.cloud.enabled=true -Dtest=CloudBasedIntegrationTest -Dtest.function.name.suffix='-java${{ matrix.java }}'
+        working-directory: ./examples

docs/adr/002-phaser-based-coordination.md

@@ -1,6 +1,6 @@
 # ADR-002: Phaser-Based Operation Coordination
 
-**Status:** Accepted  
+**Status:** Superseded by ADR-003 CompletableFuture-Based Operation Coordination
 **Date:** 2025-12-29
 
 

docs/adr/003-completable-future-based-coordination.md

@@ -0,0 +1,4 @@
+# ADR-003: CompletableFuture-Based Operation Coordination
+
+**Status:** Todo
+**Date:** 2026-02-18

docs/adr/003-child-context-execution.md docs/adr/004-child-context-execution.mddocs/adr/003-child-context-execution.md renamed to docs/adr/004-child-context-execution.md

@@ -1,4 +1,4 @@
-# ADR-003: Child Context Execution (`runInChildContext`)
+# ADR-004: Child Context Execution (`runInChildContext`)
 
 **Status:** Accepted  
 **Date:** 2026-02-16

docs/design.md

@@ -99,7 +99,7 @@ The SDK uses two separate thread pools with distinct responsibilities:
 - Default: cached daemon thread pool
 
 **Internal Executor (`InternalExecutor.INSTANCE`):**
-- Runs SDK coordination tasks: checkpoint batching, polling for wait completion, phaser management
+- Runs SDK coordination tasks: checkpoint batching, polling for wait completion
 - Dedicated cached thread pool with daemon threads named `durable-sdk-internal-*`
 - Not configurable by users
 
@@ -171,14 +171,14 @@ context.step("name", Type.class, supplier,
                                     │
                     ┌───────────────┴───────────────┐
                     ▼                               ▼
-┌──────────────────────────────┐    ┌──────────────────────────────┐
-│  DurableContext              │    │  ExecutionManager            │
-│  - User-facing API           │    │  - State (ops, token)        │
-│  - step(), stepAsync()       │    │  - Thread coordination       │
-│  - wait()                    │    │  - Phaser management         │
-│  - Operation ID counter      │    │  - Checkpoint batching       │
-└──────────────────────────────┘    │  - Polling                   │
-            │                       └──────────────────────────────┘
+┌──────────────────────────────┐    ┌─────────────────────────────────┐
+│  DurableContext              │    │  ExecutionManager               │
+│  - User-facing API           │    │  - State (ops, token)           │
+│  - step(), stepAsync(), etc  │    │  - Thread coordination          │
+│  - wait()                    │    │  - Checkpoint batching          │
+│  - Operation ID counter      │    │  - Checkpoint response handling │
+└──────────────────────────────┘    │  - Polling                      │
+            │                       └─────────────────────────────────┘
             │                                       │
             ▼                                       ▼
 ┌──────────────────────────────┐    ┌──────────────────────────────┐
@@ -213,13 +213,14 @@ com.amazonaws.lambda.durable
 │   ├── CheckpointBatcher    # Batching (package-private)
 │   ├── CheckpointCallback   # Callback interface
 │   ├── SuspendExecutionException
-│   ├── ThreadType           # CONTEXT, STEP
-│   └── ExecutionPhase       # RUNNING(0), COMPLETING(1), DONE(2)
+│   └── ThreadType           # CONTEXT, STEP
 │
 ├── operation/
-│   ├── DurableOperation<T>  # Interface
-│   ├── StepOperation<T>     # Step logic
-│   └── WaitOperation        # Wait logic
+│   ├── BaseDurableOperation<T>  # Common operation logic
+│   ├── StepOperation<T>         # Step logic
+│   ├── InvokeOperation<T>       # Invoke logic
+│   ├── CallbackOperation<T>     # Callback logic
+│   └── WaitOperation            # Wait logic
 │
 ├── logging/
 │   ├── DurableLogger        # Context-aware logger wrapper (MDC-based)
@@ -328,7 +329,7 @@ sequenceDiagram
     WO->>EM: deregisterActiveThread("Root")
     
     Note over EM: No active threads!
-    EM->>EM: suspendExecutionFuture.complete()
+    EM->>EM: executionExceptionFuture.completeExceptionally(SuspendExecutionException)
     EM-->>WO: throw SuspendExecutionException
     
     Note over UC: Execution suspended, returns PENDING
@@ -562,36 +563,29 @@ This approach ensures suspension happens precisely when no thread can make progr
 #### Advanced Feature: In-Process Completion
 In scenarios where waits or step retries would normally suspend execution, but other active threads prevent suspension, the SDK automatically switches to in-process completion by polling the backend until timing conditions are met. This allows complex concurrent workflows to complete efficiently without unnecessary Lambda re-invocations or extended waiting periods.
 
-### From Thread Tracking to Phaser Coordination
-
-Thread counting handles simple cases, but complex scenarios require sophisticated coordination:
-
-**Simple case - Wait operations:**
-```java
-context.wait(Duration.ofMinutes(5)); // Root deregisters → immediate suspension
-```
-
-**Complex case - Blocking on retrying operations:**
-```java
-var future1 = context.stepAsync("step1", () -> failsAndRetries());
-var result = context.step("step2", () -> future1.get() + "-processed");
-```
+### Active Thread Tracking and Operation Completion Coordination
 
-**Without phasers:** Simple thread counting fails because step2's thread would stay registered while blocked on `future1.get()`, preventing `activeThreads.isEmpty()` from triggering suspension → Lambda stays active during step1's retry delay instead of suspending.
+Each piece of user code - main function body, step body or child context body - runs in its own thread. Execution manager tracks active running threads. When a new step or child context is created, a new thread will be created and registered in execution manager. When the user code is blocked on `get()` or synchronous durable operations, the thread will be deregistered from execution manager. When there is no active running thread, the function execution will be suspended.
 
-**What should happen instead:** step2's root thread must deregister when blocked, allow suspension during step1's retry, then coordinate re-registration when step1 completes with checkpointed results.
+These user threads and the system thread use CompletableFuture to communicate the completion of operations. When a context executes a step, the communication happens as shown below
 
-**The problem:** When step1 retries, step2's root thread must:
-1. Deregister (to allow suspension during retry delay)
-2. Block until step1 either completes successfully or wants to suspend for another retry
-3. Re-register when step1 finishes or when resuming from suspension
-4. Ensure step1's result is checkpointed before proceeding
+| Sequence  | Context thread                                                                             | Step Thread                                                         | System Thread                                                                                                                                                                   |
+|-----------|--------------------------------------------------------------------------------------------|---------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 1         | create StepOperation, create CompletableFuture                                             | (not created)                                                       | (idle)                                                                                                                                                                          |
+| 2         | checkpoint START event (synchronously or asynchronously)                                   | (not created)                                                       | call checkpoint API                                                                                                                                                             |
+| 3         | create and register the Step thread                                                        | execute user code for the step                                      | (idle)                                                                                                                                                                          |
+| 4         | call `get()`, deregister the context thread and wait for the CompletableFuture to complete | (continue)                                                          | (idle)                                                                                                                                                                          |
+| 5         | (blocked)                                                                                  | checkpoint the step result and wait for checkpoint call to complete | call checkpoint API, and handle the API response. If it is a terminal response, it will complete the Step operation CompletableFuture, register and unblock the context thread. |
+| 6         | retrieve the result of the step                                                            | deregister and terminate the Step thread                            | (idle)                                                                                                                                                                          |        
 
-**Additional complex scenarios:**
-- **Nested blocking:** Multiple threads blocking on each other's results
-- **Future operations:** `runInChildContext` with multiple child threads coordinating
-- **Race conditions:** Ensuring checkpoint completion before thread lifecycle changes
+If the user code completes quickly, an alternative scenario could happen as follows
 
-These scenarios are why we chose **phasers** - a multi-party synchronization primitive that coordinates checkpoint-driven completion.
+| Sequence | Context thread                                                                | Step Thread                                                         | System Thread                                                                                                                          |
+|----------|-------------------------------------------------------------------------------|---------------------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------|
+| 1        | create StepOperation, create CompletableFuture                                | (not created)                                                       | (idle)                                                                                                                                 |
+| 2        | checkpoint START event (synchronously or asynchronously)                      | (not created)                                                       | call checkpoint API                                                                                                                    |
+| 3        | create and register the Step thread                                           | execute user code for the step and complete quickly                 | (idle)                                                                                                                                 |
+| 5        | (do something else or just get starved)                                       | checkpoint the step result and wait for checkpoint call to complete | call checkpoint API, and handle the API response. If it is a terminal response, it will complete the Step operation CompletableFuture. |
+| 4        | call `get()`. It's not blocked because CompletableFuture is already completed | deregister and terminate the Step thread                            | (idle)                                                                                                                                 |
+| 6        | retrieve the result of the step                                               | (ended)                                                             | (idle)                                                                                                                                 |        
 
-See [ADR-002: Phaser-Based Operation Coordination](adr/002-phaser-based-coordination.md) for detailed implementation and usage patterns.

examples/src/main/java/com/amazonaws/lambda/durable/examples/SimpleInvokeExample.java

@@ -18,13 +18,13 @@ public String handleRequest(GreetingRequest input, DurableContext context) {
         // invoke `simple-step-example` function
         var future = context.invokeAsync(
                 "call-greeting1",
-                "simple-step-example:$LATEST",
+                "simple-step-example" + input.getName() + ":$LATEST",
                 input,
                 String.class,
                 InvokeConfig.builder().build());
         var result2 = context.invoke(
                 "call-greeting2",
-                "simple-step-example:$LATEST",
+                "simple-step-example" + input.getName() + ":$LATEST",
                 input,
                 String.class,
                 InvokeConfig.builder().build());

examples/src/test/java/com/amazonaws/lambda/durable/examples/CloudBasedIntegrationTest.java

@@ -21,6 +21,7 @@ class CloudBasedIntegrationTest {
 
     private static String account;
     private static String region;
+    private static String functionNameSuffix;
 
     static boolean isEnabled() {
         var enabled = "true".equals(System.getProperty("test.cloud.enabled"));
@@ -40,6 +41,7 @@ static void setup() {
 
         account = System.getProperty("test.aws.account");
         region = System.getProperty("test.aws.region");
+        functionNameSuffix = System.getProperty("test.function.name.suffix", "");
 
         if (account == null || region == null) {
             var sts = StsClient.create();
@@ -52,7 +54,8 @@ static void setup() {
     }
 
     private static String arn(String functionName) {
-        return "arn:aws:lambda:" + region + ":" + account + ":function:" + functionName + ":$LATEST";
+        return "arn:aws:lambda:" + region + ":" + account + ":function:" + functionName + functionNameSuffix
+                + ":$LATEST";
     }
 
     @Test
@@ -71,7 +74,7 @@ void testSimpleStepExample() {
     @Test
     void testSimpleInvokeExample() {
         var runner = CloudDurableTestRunner.create(arn("simple-invoke-example"), Map.class, String.class);
-        var result = runner.run(Map.of("message", "test"));
+        var result = runner.run(Map.of("name", functionNameSuffix));
 
         assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
         assertNotNull(result.getResult(String.class));

pom.xml

@@ -23,9 +23,9 @@
         <maven.compiler.source>17</maven.compiler.source>
         <maven.compiler.target>17</maven.compiler.target>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <aws.sdk.version>2.41.28</aws.sdk.version>
+        <aws.sdk.version>2.41.32</aws.sdk.version>
         <jackson.version>2.21.0</jackson.version>
-        <junit.version>6.0.2</junit.version>
+        <junit.version>6.0.3</junit.version>
         <mockito.version>5.21.0</mockito.version>
         <slf4j.version>2.0.17</slf4j.version>
         <jacoco.version>0.8.14</jacoco.version>

sdk/src/main/java/com/amazonaws/lambda/durable/CallbackConfig.java

@@ -3,6 +3,7 @@
 package com.amazonaws.lambda.durable;
 
 import com.amazonaws.lambda.durable.serde.SerDes;
+import com.amazonaws.lambda.durable.validation.ParameterValidator;
 import java.time.Duration;
 
 /** Configuration for callback operations. */
@@ -60,11 +61,13 @@ private Builder(Duration timeout, Duration heartbeatTimeout, SerDes serDes) {
         }
 
         public Builder timeout(Duration timeout) {
+            ParameterValidator.validateOptionalDuration(timeout, "Callback timeout");
             this.timeout = timeout;
             return this;
         }
 
         public Builder heartbeatTimeout(Duration heartbeatTimeout) {
+            ParameterValidator.validateOptionalDuration(heartbeatTimeout, "Heartbeat timeout");
             this.heartbeatTimeout = heartbeatTimeout;
             return this;
         }

sdk/src/main/java/com/amazonaws/lambda/durable/DurableConfig.java

@@ -312,6 +312,7 @@ public Builder withLoggerConfig(LoggerConfig loggerConfig) {
          * @return This builder
          */
         public Builder withPollingInterval(Duration duration) {
+            // No validation - polling intervals can be less than 1 second (e.g., 200ms with backoff)
             this.pollingInterval = duration;
             return this;
         }