fix: finalize invocation hooks on the handler thread

.github/scripts/verify_otel_api_compatibility.py

@@ -142,11 +142,14 @@ def run_matrix(args: argparse.Namespace) -> int:
             for view in ("otel-invocation", "otel-execution"):
                 name = f"{label}-api{version}-{view}"
                 negative = label == "old-old" and version == "1.49.0"
+                incompatible_core = label == "old-new"
                 case: dict[str, object] = {"name": name, "expected_negative_control": negative,
+                                          "expected_core_rejection": incompatible_core,
                                           "api": jar_facts(api), "context": jar_facts(context)}
                 command = [args.java, "-cp", os.pathsep.join(map(str, [classes, core, plugin, *dependencies])),
                            PROBE, str(core), str(plugin), str(api), str(context), view,
-                           str(negative).lower(), str(version == "1.66.0").lower()]
+                           str(negative).lower(), str(version == "1.66.0").lower(),
+                           str(incompatible_core).lower()]
                 try:
                     log = output / f"{name}.log"
                     execute(command, log, env=probe_environment(view), timeout=90)
@@ -155,6 +158,8 @@ def run_matrix(args: argparse.Namespace) -> int:
                         raise RuntimeError("Probe did not report successful completion")
                     if negative and "NEGATIVE_CONTROL_REPRODUCED" not in contents:
                         raise RuntimeError("Released negative control did not reproduce the reported failure")
+                    if incompatible_core and "CORE_LIFECYCLE_REJECTION_CONFIRMED" not in contents:
+                        raise RuntimeError("Old core must reject the new plugin before invocation startup")
                     case["passed"] = True
                 except RuntimeError as error:
                     failures += 1

.github/workflows/otel-conformance-tests.yml

@@ -68,8 +68,9 @@ jobs:
       actions: write
       contents: read
       id-token: write
-    uses: aws/aws-durable-execution-conformance-tests/.github/workflows/opentelemetry-orchestrator.yml@a66037abbbfa55fde97f714e30f0bc262edefd63
+    uses: aws/aws-durable-execution-conformance-tests/.github/workflows/opentelemetry-orchestrator.yml@1768f32d61d958b974ce9e66748df72e23bc19b2
     with:
+      case_count: 20
       runs_on: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository && 'ubuntu-latest' || format('codebuild-github-actions-runner-{0}-{1}', github.run_id, github.run_attempt) }}
       language: java
       resource_prefix: j

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

@@ -21,6 +21,7 @@
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.lambda.LambdaClient;
 import software.amazon.awssdk.services.lambda.model.ErrorObject;
+import software.amazon.awssdk.services.lambda.model.Event;
 import software.amazon.awssdk.services.lambda.model.OperationStatus;
 import software.amazon.awssdk.services.sts.StsClient;
 import software.amazon.lambda.durable.TypeToken;
@@ -818,7 +819,20 @@ void testConcurrentWaitForConditionExample() {
                 lambdaClient);
         var result = runner.run(new ConcurrentWaitForConditionExample.Input(3, 100, 50));
 
-        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+        try {
+            assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+        } catch (RuntimeException | AssertionError failure) {
+            try {
+                // Preserve the already-fetched service timeline without payloads, callback IDs, or checkpoint tokens.
+                var history = result.getHistoryEvents().stream()
+                        .map(CloudBasedIntegrationTest::historyMetadata)
+                        .toList();
+                System.err.println("Concurrent wait-for-condition history: " + new JacksonSerDes().serialize(history));
+            } catch (RuntimeException diagnosticsFailure) {
+                failure.addSuppressed(diagnosticsFailure);
+            }
+            throw failure;
+        }
 
         // Verify each operation finished with 3 attempts
         var allOperationsOutput = result.getResult();
@@ -840,6 +854,30 @@ void testConcurrentWaitForConditionExample() {
         }
     }
 
+    private static Map<String, Object> historyMetadata(Event event) {
+        var row = new HashMap<String, Object>();
+        row.put("eventId", event.eventId());
+        row.put("type", event.eventTypeAsString());
+        row.put("operationId", event.id());
+        row.put("parentId", event.parentId());
+        row.put("name", event.name());
+        row.put("at", String.valueOf(event.eventTimestamp()));
+        var retries = event.stepSucceededDetails() != null
+                ? event.stepSucceededDetails().retryDetails()
+                : event.stepFailedDetails() != null ? event.stepFailedDetails().retryDetails() : null;
+        if (retries != null) {
+            row.put("attempt", retries.currentAttempt());
+            row.put("nextAttemptDelaySeconds", retries.nextAttemptDelaySeconds());
+        }
+        var invocation = event.invocationCompletedDetails();
+        if (invocation != null) {
+            row.put("invocationStart", String.valueOf(invocation.startTimestamp()));
+            row.put("invocationEnd", String.valueOf(invocation.endTimestamp()));
+            row.put("invocationFailed", invocation.error() != null);
+        }
+        return row;
+    }
+
     @Test
     void testPluginExample() {
         var runner =

examples/src/test/java/software/amazon/lambda/durable/examples/parallel/ParallelFailureToleranceExampleTest.java

@@ -5,8 +5,27 @@
 import static org.junit.jupiter.api.Assertions.*;
 
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import software.amazon.awssdk.services.lambda.model.OperationStatus;
+import software.amazon.awssdk.services.lambda.model.OperationType;
+import software.amazon.lambda.durable.DurableConfig;
+import software.amazon.lambda.durable.TypeToken;
+import software.amazon.lambda.durable.model.ConcurrencyCompletionStatus;
 import software.amazon.lambda.durable.model.ExecutionStatus;
+import software.amazon.lambda.durable.model.ParallelResult;
+import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
+import software.amazon.lambda.durable.plugin.OperationEndInfo;
+import software.amazon.lambda.durable.plugin.UserFunctionEndInfo;
+import software.amazon.lambda.durable.plugin.UserFunctionStartInfo;
+import software.amazon.lambda.durable.serde.JacksonSerDes;
 import software.amazon.lambda.durable.testing.LocalDurableTestRunner;
 
 class ParallelFailureToleranceExampleTest {
@@ -41,19 +60,104 @@ void succeedsWhenAllBranchesSucceed() {
         assertEquals(3, output.succeeded());
     }
 
-    @Test
-    void failsWhenFailuresExceedTolerance() {
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void failsWhenFailuresExceedTolerance(boolean holdHealthyBranch) throws Exception {
         var handler = new ParallelFailureToleranceExample();
-        var runner = LocalDurableTestRunner.create(ParallelFailureToleranceExample.Input.class, handler);
+        var probe = new CompletionProbe(holdHealthyBranch);
+        var config = DurableConfig.builder().withPlugins(probe.newPlugin()).build();
+        var runner = LocalDurableTestRunner.create(ParallelFailureToleranceExample.Input.class, handler)
+                .withDurableConfig(config);
+        var caller = Executors.newSingleThreadExecutor();
+        try {
+            var input = new ParallelFailureToleranceExample.Input(List.of("svc-a", "bad-svc-b", "bad-svc-c"), 1, 2);
+            var invocation = caller.submit(() -> runner.runUntilComplete(input));
+            if (holdHealthyBranch) {
+                await(probe.parallelStored);
+                probe.releaseHealthy.countDown();
+            }
+            var result = invocation.get(5, TimeUnit.SECONDS);
+            assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+            var output = result.getResult(ParallelFailureToleranceExample.Output.class);
+            assertEquals(2, output.failed());
 
-        // 2 bad services, toleratedFailureCount=1 — second failure exceeds tolerance
-        var input = new ParallelFailureToleranceExample.Input(List.of("svc-a", "bad-svc-b", "bad-svc-c"), 1, 2);
-        var result = runner.runUntilComplete(input);
+            var operation = result.getOperation("call-services");
+            assertEquals(OperationStatus.SUCCEEDED, operation.getStatus());
+            var stored = new JacksonSerDes()
+                    .deserialize(operation.getContextDetails().result(), TypeToken.get(ParallelResult.class));
+            assertEquals(ConcurrencyCompletionStatus.FAILURE_TOLERANCE_EXCEEDED, stored.completionStatus());
+            assertFalse(stored.completionStatus().isSucceeded());
+            assertEquals(3, stored.size());
+            assertEquals(
+                    List.of(ParallelResult.Status.FAILED, ParallelResult.Status.FAILED),
+                    stored.statuses().subList(1, 3));
+            assertTrue(List.of(ParallelResult.Status.SUCCEEDED, ParallelResult.Status.SKIPPED)
+                    .contains(stored.statuses().get(0)));
+            assertEquals(output.succeeded(), stored.succeeded());
+            assertEquals(output.failed(), stored.failed());
+            assertEquals(1 - output.succeeded(), stored.skipped());
+            if (holdHealthyBranch) assertEquals(0, output.succeeded());
 
-        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+            var completedCalls = result.getOperations().stream()
+                    .filter(op -> op.getType() == OperationType.STEP && op.isCompleted())
+                    .collect(Collectors.toMap(op -> op.getName(), op -> probe.completedCalls.get(op.getName())));
+            assertEquals(1, completedCalls.get("invoke-bad-svc-b"));
+            assertEquals(1, completedCalls.get("invoke-bad-svc-c"));
+            var replay = runner.run(input);
+            assertEquals(ExecutionStatus.SUCCEEDED, replay.getStatus());
+            assertEquals(output, replay.getResult(ParallelFailureToleranceExample.Output.class));
+            completedCalls.forEach((name, count) -> assertEquals(
+                    count,
+                    probe.completedCalls.get(name),
+                    "Completed step bodies must not run again on replay: " + name));
+        } finally {
+            probe.releaseHealthy.countDown();
+            caller.shutdownNow();
+            assertTrue(caller.awaitTermination(5, TimeUnit.SECONDS));
+        }
+    }
 
-        var output = result.getResult(ParallelFailureToleranceExample.Output.class);
-        assertEquals(2, output.failed());
-        assertEquals(1, output.succeeded());
+    private static void await(CountDownLatch latch) {
+        try {
+            assertTrue(latch.await(5, TimeUnit.SECONDS), "Controlled branch scheduling was not released");
+        } catch (InterruptedException failure) {
+            Thread.currentThread().interrupt();
+            throw new AssertionError(failure);
+        }
+    }
+
+    private static final class CompletionProbe {
+        final boolean holdHealthy;
+        final CountDownLatch healthyStarted = new CountDownLatch(1);
+        final CountDownLatch releaseHealthy = new CountDownLatch(1);
+        final CountDownLatch parallelStored = new CountDownLatch(1);
+        final Map<String, Integer> completedCalls = new ConcurrentHashMap<>();
+
+        CompletionProbe(boolean holdHealthy) {
+            this.holdHealthy = holdHealthy;
+        }
+
+        DurableExecutionPlugin newPlugin() {
+            return new DurableExecutionPlugin() {
+                @Override
+                public void onUserFunctionStart(UserFunctionStartInfo info) {
+                    if (!holdHealthy || !"STEP".equals(info.type())) return;
+                    if ("invoke-svc-a".equals(info.name())) {
+                        healthyStarted.countDown();
+                        await(releaseHealthy);
+                    } else if (info.name().startsWith("invoke-bad-")) await(healthyStarted);
+                }
+
+                @Override
+                public void onUserFunctionEnd(UserFunctionEndInfo info) {
+                    if ("STEP".equals(info.type())) completedCalls.merge(info.name(), 1, Integer::sum);
+                }
+
+                @Override
+                public void onOperationEnd(OperationEndInfo info) {
+                    if ("call-services".equals(info.name())) parallelStored.countDown();
+                }
+            };
+        }
     }
 }