feat(map): Fix map replay, wait-inside-map, and concurrency race conditions

Author: ayushiahjolia

sdk/src/main/java/software/amazon/lambda/durable/CompletionConfig.java

@@ -36,16 +36,26 @@ public static CompletionConfig firstSuccessful() {
 
     /** Complete when the specified number of items have succeeded. */
     public static CompletionConfig minSuccessful(int count) {
+        if (count < 1) {
+            throw new IllegalArgumentException("minSuccessful must be at least 1, got: " + count);
+        }
         return new CompletionConfig(count, null, null);
     }
 
     /** Complete when more than the specified number of failures have occurred. */
     public static CompletionConfig toleratedFailureCount(int count) {
+        if (count < 0) {
+            throw new IllegalArgumentException("toleratedFailureCount must be non-negative, got: " + count);
+        }
         return new CompletionConfig(null, count, null);
     }
 
     /** Complete when the failure percentage exceeds the specified threshold (0.0 to 1.0). */
     public static CompletionConfig toleratedFailurePercentage(double percentage) {
+        if (percentage < 0.0 || percentage > 1.0) {
+            throw new IllegalArgumentException(
+                    "toleratedFailurePercentage must be between 0.0 and 1.0, got: " + percentage);
+        }
         return new CompletionConfig(null, null, percentage);
     }
 

sdk/src/main/java/software/amazon/lambda/durable/MapConfig.java

@@ -71,6 +71,9 @@ public Builder serDes(SerDes serDes) {
         }
 
         public MapConfig build() {
+            if (maxConcurrency != null && maxConcurrency < 1) {
+                throw new IllegalArgumentException("maxConcurrency must be at least 1, got: " + maxConcurrency);
+            }
             return new MapConfig(this);
         }
     }

sdk/src/main/java/software/amazon/lambda/durable/model/MapResult.java

@@ -48,9 +48,9 @@ public ErrorObject getError(int index) {
         return items.get(index).error();
     }
 
-    /** Returns true if all items succeeded (no errors). */
+    /** Returns true if all items succeeded (no failures or not-started items). */
     public boolean allSucceeded() {
-        return items.stream().noneMatch(item -> item.error() != null);
+        return items.stream().allMatch(item -> item.status() == MapResultItem.Status.SUCCEEDED);
     }
 
     /** Returns the number of items in this result. */
@@ -64,13 +64,19 @@ public List<T> results() {
                 items.stream().map(MapResultItem::result).toList());
     }
 
-    /** Returns results that succeeded (non-null results). */
+    /** Returns results from items that succeeded (includes null results from successful items). */
     public List<T> succeeded() {
-        return items.stream().map(MapResultItem::result).filter(r -> r != null).toList();
+        return items.stream()
+                .filter(item -> item.status() == MapResultItem.Status.SUCCEEDED)
+                .map(MapResultItem::result)
+                .toList();
     }
 
-    /** Returns errors that occurred (non-null errors). */
+    /** Returns errors from items that failed. */
     public List<ErrorObject> failed() {
-        return items.stream().map(MapResultItem::error).filter(e -> e != null).toList();
+        return items.stream()
+                .filter(item -> item.status() == MapResultItem.Status.FAILED)
+                .map(MapResultItem::error)
+                .toList();
     }
 }

sdk/src/main/java/software/amazon/lambda/durable/operation/BaseConcurrentOperation.java

@@ -370,11 +370,8 @@ public R get() {
             }
         }
 
-        // Race completionFuture against executionExceptionFuture.
-        // If branches suspend, executionExceptionFuture completes with SuspendExecutionException,
-        // which propagates up through the handler to DurableExecutor → returns PENDING.
-        // If branches complete, completionFuture completes and we proceed to read the result.
-        executionManager.runUntilCompleteOrSuspend(completionFuture).join();
+        // Block until operation completes. No-op if the future is already completed.
+        completionFuture.join();
 
         var op = getOperation();
 

sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java

@@ -209,12 +209,8 @@ protected Operation waitForOperationCompletion() {
             }
         }
 
-        // Block until operation completes or execution suspends.
-        // Using runUntilCompleteOrSuspend races completionFuture against executionExceptionFuture,
-        // so when all active threads suspend (e.g., wait inside map branches), the
-        // SuspendExecutionException propagates and this thread is freed — preventing thread leaks
-        // on shared executor pools across invocations.
-        executionManager.runUntilCompleteOrSuspend(completionFuture).join();
+        // Block until operation completes. No-op if the future is already completed.
+        completionFuture.join();
 
         // Get result based on status
         var op = getOperation();

sdk/src/main/java/software/amazon/lambda/durable/validation/ParameterValidator.java

@@ -7,6 +7,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.IdentityHashMap;
+import java.util.LinkedHashSet;
 import java.util.Set;
 import java.util.WeakHashMap;
 import java.util.concurrent.ConcurrentHashMap;
@@ -118,8 +119,9 @@ public static void validateOperationName(String name, int maxLength) {
     /**
      * Validates that a collection has deterministic iteration order.
      *
-     * <p>Rejects known unordered collection types: {@link HashSet} (and subclasses), and views returned by
-     * {@link HashMap}, {@link IdentityHashMap}, {@link WeakHashMap}, and {@link ConcurrentHashMap}.
+     * <p>Rejects known unordered collection types: {@link HashSet} (but not {@link LinkedHashSet}, which has stable
+     * insertion-order iteration), and views returned by {@link HashMap}, {@link IdentityHashMap}, {@link WeakHashMap},
+     * and {@link ConcurrentHashMap}.
      *
      * @param items the collection to validate
      * @throws IllegalArgumentException if items is null or has non-deterministic iteration order
@@ -128,6 +130,10 @@ public static void validateOrderedCollection(Collection<?> items) {
         if (items == null) {
             throw new IllegalArgumentException("items cannot be null");
         }
+        // LinkedHashSet extends HashSet but has stable insertion-order iteration — allow it
+        if (items instanceof LinkedHashSet) {
+            return;
+        }
         if (items instanceof HashSet || isUnorderedMapView(items)) {
             throw new IllegalArgumentException("items must have deterministic iteration order");
         }

sdk/src/test/java/software/amazon/lambda/durable/CompletionConfigTest.java

@@ -61,4 +61,48 @@ void toleratedFailurePercentage_setsPercentage() {
         assertNull(config.toleratedFailureCount());
         assertEquals(0.25, config.toleratedFailurePercentage());
     }
+
+    @Test
+    void minSuccessful_withZero_shouldThrow() {
+        var exception = assertThrows(IllegalArgumentException.class, () -> CompletionConfig.minSuccessful(0));
+        assertEquals("minSuccessful must be at least 1, got: 0", exception.getMessage());
+    }
+
+    @Test
+    void minSuccessful_withNegative_shouldThrow() {
+        var exception = assertThrows(IllegalArgumentException.class, () -> CompletionConfig.minSuccessful(-1));
+        assertEquals("minSuccessful must be at least 1, got: -1", exception.getMessage());
+    }
+
+    @Test
+    void toleratedFailureCount_withNegative_shouldThrow() {
+        var exception = assertThrows(IllegalArgumentException.class, () -> CompletionConfig.toleratedFailureCount(-1));
+        assertEquals("toleratedFailureCount must be non-negative, got: -1", exception.getMessage());
+    }
+
+    @Test
+    void toleratedFailurePercentage_withNegative_shouldThrow() {
+        var exception =
+                assertThrows(IllegalArgumentException.class, () -> CompletionConfig.toleratedFailurePercentage(-0.1));
+        assertEquals("toleratedFailurePercentage must be between 0.0 and 1.0, got: -0.1", exception.getMessage());
+    }
+
+    @Test
+    void toleratedFailurePercentage_aboveOne_shouldThrow() {
+        var exception =
+                assertThrows(IllegalArgumentException.class, () -> CompletionConfig.toleratedFailurePercentage(1.5));
+        assertEquals("toleratedFailurePercentage must be between 0.0 and 1.0, got: 1.5", exception.getMessage());
+    }
+
+    @Test
+    void toleratedFailurePercentage_atBoundaries_shouldPass() {
+        assertDoesNotThrow(() -> CompletionConfig.toleratedFailurePercentage(0.0));
+        assertDoesNotThrow(() -> CompletionConfig.toleratedFailurePercentage(1.0));
+    }
+
+    @Test
+    void toleratedFailureCount_withZero_shouldPass() {
+        var config = CompletionConfig.toleratedFailureCount(0);
+        assertEquals(0, config.toleratedFailureCount());
+    }
 }

sdk/src/test/java/software/amazon/lambda/durable/MapConfigTest.java

@@ -101,4 +101,26 @@ void toBuilder_canOverrideValues() {
         assertEquals(10, modified.maxConcurrency());
         assertEquals(4, original.maxConcurrency());
     }
+
+    @Test
+    void builderWithZeroMaxConcurrency_shouldThrow() {
+        var exception = assertThrows(
+                IllegalArgumentException.class,
+                () -> MapConfig.builder().maxConcurrency(0).build());
+        assertEquals("maxConcurrency must be at least 1, got: 0", exception.getMessage());
+    }
+
+    @Test
+    void builderWithNegativeMaxConcurrency_shouldThrow() {
+        var exception = assertThrows(
+                IllegalArgumentException.class,
+                () -> MapConfig.builder().maxConcurrency(-1).build());
+        assertEquals("maxConcurrency must be at least 1, got: -1", exception.getMessage());
+    }
+
+    @Test
+    void builderWithNullMaxConcurrency_shouldPass() {
+        var config = MapConfig.builder().maxConcurrency(null).build();
+        assertNull(config.maxConcurrency());
+    }
 }

sdk/src/test/java/software/amazon/lambda/durable/validation/ParameterValidatorTest.java

@@ -331,11 +331,9 @@ void validateOrderedCollection_withOrderedSet_shouldPass() {
     }
 
     @Test
-    void validateOrderedCollection_withLinkedHashSet_shouldThrow() {
-        // LinkedHashSet extends HashSet, so it's rejected even though it has deterministic order
-        assertThrows(
-                IllegalArgumentException.class,
-                () -> ParameterValidator.validateOrderedCollection(new LinkedHashSet<>(List.of("a", "b"))));
+    void validateOrderedCollection_withLinkedHashSet_shouldPass() {
+        // LinkedHashSet extends HashSet but has stable insertion-order iteration — allowed
+        assertDoesNotThrow(() -> ParameterValidator.validateOrderedCollection(new LinkedHashSet<>(List.of("a", "b"))));
     }
 
     @Test