[bugfix] replace phaser with future (#103)

* replace phaser with future

* update design doc for completionFuture

* update docs

Author: zhongkechen

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/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.

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

@@ -9,8 +9,8 @@ public interface DurableFuture<T> {
     /**
      * Blocks until the operation completes and returns the result.
      *
-     * <p>This delegates to operation.get() which handles: - Phaser blocking (arriveAndAwaitAdvance) - Thread
-     * deregistration (allows suspension) - Thread reactivation (resumes execution) - Result retrieval
+     * <p>This delegates to operation.get() which handles: - Thread deregistration (allows suspension) - Thread
+     * reactivation (resumes execution) - Result retrieval
      *
      * @return the operation result
      */

sdk/src/main/java/com/amazonaws/lambda/durable/execution/CheckpointBatcher.java

@@ -139,10 +139,6 @@ private void checkpointBatch(List<OperationUpdate> updates) {
             logger.debug("Durable checkpoint API called (latency={}ns): {}.", System.nanoTime() - startTime, response);
 
             // Notify callback of completion
-            // TODO: sam local backend returns no new execution state when called with zero
-            // updates. WHY?
-            // This means the polling will never receive an operation update and complete
-            // the Phaser.
             checkpointToken = response.checkpointToken();
             if (response.newExecutionState() != null) {
                 // fetch all pages of operations

sdk/src/main/java/com/amazonaws/lambda/durable/execution/ExecutionManager.java

@@ -4,13 +4,13 @@
 
 import com.amazonaws.lambda.durable.DurableConfig;
 import com.amazonaws.lambda.durable.exception.UnrecoverableDurableExecutionException;
+import com.amazonaws.lambda.durable.operation.BaseDurableOperation;
 import java.time.Duration;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Phaser;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import org.slf4j.Logger;
@@ -28,8 +28,8 @@
  * <ul>
  *   <li>Execution state (operations, checkpoint token)
  *   <li>Thread lifecycle (registration/deregistration)
- *   <li>Phaser management (coordination)
- *   <li>Checkpoint batching (via CheckpointManager)
+ *   <li>Checkpoint batching (via CheckpointBatcher)
+ *   <li>Checkpoint result handling (CheckpointBatcher callback)
  *   <li>Polling (for waits and retries)
  * </ul>
  *
@@ -43,15 +43,16 @@ public class ExecutionManager {
     private static final Logger logger = LoggerFactory.getLogger(ExecutionManager.class);
 
     // ===== Execution State =====
-    private final Map<String, Operation> operations;
+    private final Map<String, Operation> operationStorage;
     private final String executionOperationId;
     private final String durableExecutionArn;
     private final AtomicReference<ExecutionMode> executionMode;
 
     // ===== Thread Coordination =====
+    private final Map<String, BaseDurableOperation<?>> registeredOperations =
+            Collections.synchronizedMap(new HashMap<>());
     private final Map<String, ThreadType> activeThreads = Collections.synchronizedMap(new HashMap<>());
     private static final ThreadLocal<OperationContext> currentContext = new ThreadLocal<>();
-    private final Map<String, Phaser> openPhasers = Collections.synchronizedMap(new HashMap<>());
     private final CompletableFuture<Void> executionExceptionFuture = new CompletableFuture<>();
 
     // ===== Checkpoint Batching =====
@@ -69,12 +70,12 @@ public ExecutionManager(
         this.checkpointBatcher =
                 new CheckpointBatcher(config, durableExecutionArn, checkpointToken, this::onCheckpointComplete);
 
-        this.operations = checkpointBatcher.fetchAllPages(initialExecutionState).stream()
+        this.operationStorage = checkpointBatcher.fetchAllPages(initialExecutionState).stream()
                 .collect(Collectors.toConcurrentMap(Operation::id, op -> op));
 
         // Start in REPLAY mode if we have more than just the initial EXECUTION operation
         this.executionMode =
-                new AtomicReference<>(operations.size() > 1 ? ExecutionMode.REPLAY : ExecutionMode.EXECUTION);
+                new AtomicReference<>(operationStorage.size() > 1 ? ExecutionMode.REPLAY : ExecutionMode.EXECUTION);
     }
 
     // ===== State Management =====
@@ -87,24 +88,22 @@ public boolean isReplaying() {
         return executionMode.get() == ExecutionMode.REPLAY;
     }
 
+    public void registerOperation(BaseDurableOperation<?> operation) {
+        registeredOperations.put(operation.getOperationId(), operation);
+    }
+
     // ===== Checkpoint Completion Handler =====
-    /** Called by CheckpointManager when a checkpoint completes. Updates state and advances phasers. */
+    /** Called by CheckpointManager when a checkpoint completes. Updates operationStorage and notify operations . */
     private void onCheckpointComplete(List<Operation> newOperations) {
-        // Update operation storage
-        newOperations.forEach(op -> operations.put(op.id(), op));
-
-        // Advance phasers for completed operations
-        for (Operation operation : newOperations) {
-            if (openPhasers.containsKey(operation.id()) && isTerminalStatus(operation.status())) {
-                var phaser = openPhasers.get(operation.id());
-
-                // Two-phase completion
-                logger.trace("Advancing phaser 0 -> 1: {}", phaser);
-                phaser.arriveAndAwaitAdvance();
-                logger.trace("Advancing phaser 1 -> 2: {}", phaser);
-                phaser.arriveAndAwaitAdvance();
-            }
-        }
+        newOperations.forEach(op -> {
+            // Update operation storage
+            operationStorage.put(op.id(), op);
+            // call registered operation's onCheckpointComplete method for completed operations
+            registeredOperations.computeIfPresent(op.id(), (id, operation) -> {
+                operation.onCheckpointComplete(op);
+                return operation;
+            });
+        });
     }
 
     /**
@@ -115,7 +114,7 @@ private void onCheckpointComplete(List<Operation> newOperations) {
      * @return the existing operation, or null if not found (first execution)
      */
     public Operation getOperationAndUpdateReplayState(String operationId) {
-        var existing = operations.get(operationId);
+        var existing = operationStorage.get(operationId);
         if (executionMode.get() == ExecutionMode.REPLAY) {
             if (existing == null || !isTerminalStatus(existing.status())) {
                 if (executionMode.compareAndSet(ExecutionMode.REPLAY, ExecutionMode.EXECUTION)) {
@@ -127,7 +126,7 @@ public Operation getOperationAndUpdateReplayState(String operationId) {
     }
 
     public Operation getExecutionOperation() {
-        return operations.get(executionOperationId);
+        return operationStorage.get(executionOperationId);
     }
 
     // ===== Thread Coordination =====
@@ -185,15 +184,6 @@ public void deregisterActiveThreadAndUnsetCurrentContext(String threadId) {
         }
     }
 
-    // ===== Phaser Management =====
-
-    public Phaser startPhaser(String operationId) {
-        var phaser = new Phaser(1);
-        openPhasers.put(operationId, phaser);
-        logger.trace("Started phaser for operation '{}'", operationId);
-        return phaser;
-    }
-
     // ===== Checkpointing =====
 
     // This method will checkpoint the operation updates to the durable backend and return a future which completes