fix the result size in ParallelResult (#339)

* fix the result size in ParallelResult

* fix test cases for nesting mode

Author: zhongkechen

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

@@ -1142,17 +1142,21 @@ void testParallelWithFirstSuccessful_earlyTermination(NestingType nestingType, i
             var result = parallel.get();
             assertEquals(ConcurrencyCompletionStatus.MIN_SUCCESSFUL_REACHED, result.completionStatus());
             assertTrue(result.completionStatus().isSucceeded());
-            // todo: the result is constructed when handling parallel completion,
-            // which might be earlier than the last branch is added.
-            assertTrue(result.size() <= 3);
-            assertTrue(result.succeeded() <= result.size());
+            assertEquals(3, result.size());
+            assertTrue(result.succeeded() <= 3);
             assertTrue(1 <= result.succeeded());
 
             return "done";
         });
 
         var result = runner.runUntilComplete("test");
         assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
-        assertEquals(events, result.getHistoryEvents().size());
+        if (nestingType == NestingType.FLAT) {
+            assertEquals(2, result.getHistoryEvents().size());
+        } else {
+            // might 4 if only 1 branch completed and at most 8 if all branches completed
+            assertTrue(4 <= result.getHistoryEvents().size());
+            assertTrue(result.getHistoryEvents().size() <= events);
+        }
     }
 }

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

@@ -127,7 +127,11 @@ protected void replay(Operation existing) {
     @Override
     public ParallelResult get() {
         join();
-        return cachedResult;
+        return new ParallelResult(
+                getBranches().size(), // size might be updated after cached result is built
+                cachedResult.succeeded(),
+                cachedResult.failed(),
+                cachedResult.completionStatus());
     }
 
     /** Calls {@link #get()} if not already called. Guarantees that the context is closed. */