feat: add invocation-owned executor isolation

docs/advanced/configuration.md

@@ -34,11 +34,22 @@ public class OrderProcessor extends DurableHandler<Order, OrderResult> {
 | `withLambdaClientBuilder()` | Custom AWS Lambda client                | Auto-configured Lambda client |
 | `withSerDes()`              | Serializer for step results             | Jackson with default settings |
 | `withExecutorService()`     | Thread pool for user-defined operations | Cached daemon thread pool     |
+| `withInvocationExecutorFactory()` | Factory for isolated per-invocation user executors | Disabled |
 | `withLoggerConfig()`        | Logger behavior configuration           | Suppress logs during replay   |
 | `withPollingStrategy()`     | Backend polling strategy                | Exponential backoff: 1s base, 2x rate, FULL jitter, 10s max |
 | `withCheckpointDelay()`     | How often the SDK checkpoints updates   | `Duration.ofSeconds(0)` (as soon as possible) |
 
-The `withExecutorService()` option configures the thread pool used for running user-defined operations. Internal SDK coordination (checkpoint batching, polling) runs on an SDK-managed thread pool.
+The `withExecutorService()` option configures a caller-owned executor shared by every invocation that uses the configuration. The SDK never shuts it down. To isolate concurrent Lambda Managed Instances invocations, configure a factory instead:
+
+```java
+DurableConfig.builder()
+    .withInvocationExecutorFactory(() -> Executors.newFixedThreadPool(4))
+    .build();
+```
+
+The factory is called once per invocation and may be called concurrently on Lambda Managed Instances. It must return a fresh executor. The invocation caller waits for actual admitted work and End before shutting that executor down, including after cancellation. Code that ignores interruption can delay this shutdown beyond the policy budget; executor termination is then observed with any cleanup time that remains. Returning an executor already leased to an active invocation is rejected. The factory and `withExecutorService()` are mutually exclusive. Both options run all user code, including plugin-enabled root handlers, steps, and child contexts. Internal SDK coordination such as checkpoint batching and polling remains on the SDK-managed executor. `toBuilder()` and the local test runner preserve the factory, including across replay/resume.
+
+A bounded executor supplied through `withExecutorService()` remains shared. On Lambda Managed Instances, concurrent roots can occupy all of its workers while waiting for queued steps. Use the invocation factory when capacity must be isolated between native invocations. Isolation does not define the capacity required by arbitrarily nested synchronous child contexts; size or design those separately.
 
 ### Dynamic plugin loading
 

docs/design.md

@@ -145,6 +145,7 @@ public class MyHandler extends DurableHandler<Input, Output> {
 | `lambdaClientBuilder` | Auto-created `LambdaClient` for current region, primed for performance (see [`DurableConfig.java`](../sdk/src/main/java/software/amazon/lambda/durable/DurableConfig.java))       |
 | `serDes`              | `JacksonSerDes`                                                                                                                                                                   |
 | `executorService`     | `Executors.newCachedThreadPool()` (for user-defined operations only)                                                                                                              |
+| `invocationExecutorFactory` | Disabled; when configured, creates an SDK-owned executor for each invocation                                                                                               |
 | `loggerConfig`        | `LoggerConfig.defaults()` (suppress replay logs)                                                                                                                                  |
 | `pollingStrategy`     | Exponential backoff: 1s base, 2x rate, FULL jitter, 10s max                                                                                                                      |
 | `checkpointDelay`     | `Duration.ofSeconds(0)` (checkpoint as soon as possible)                                                                                                                          |
@@ -163,8 +164,15 @@ The SDK uses two separate thread pools with distinct responsibilities:
 **User Executor (`DurableConfig.executorService`):**
 - Runs the root handler, steps, child-context user code, and map/parallel coordination
 - Configurable via `DurableConfig.builder().withExecutorService()`
+- Caller-owned and shared across invocations; the SDK never shuts it down
 - Default: cached daemon thread pool
 
+**Invocation Executor Factory (`DurableConfig.invocationExecutorFactory`):**
+- Optional alternative to the shared user executor
+- Creates one executor per invocation for all root, step, child-context user code, and map/parallel coordination
+- SDK-owned: the invocation caller waits for actual admitted work and End before requesting executor shutdown, including after cancellation. Non-cooperative work can delay that shutdown beyond the policy budget. Executor termination is observed with any cleanup time that remains.
+- Must return a fresh executor and cannot be combined with `withExecutorService()`
+
 **Internal Executor (`InternalExecutor.INSTANCE`):**
 - Runs SDK coordination tasks: checkpoint batching, polling for wait completion
 - Dedicated cached thread pool with daemon threads named `durable-sdk-internal-*`

lmi-tests/README.md

@@ -39,7 +39,7 @@ eventual cleanup cannot turn an earlier lifecycle violation into a pass.
   SDK's minimum JDK prevents fixture-only changes from introducing newer JDK APIs
   into the shared Maven reactor. On manual dispatch, a separate
   serialized cloud job builds the selected commit, deploys all five
-  fixture functions, runs all 13 cases, then collects evidence. A persistent
+  fixture functions, runs all 12 gating cases, then collects evidence. A persistent
   CloudFormation stack owns all five
   functions and log groups. The stack, functions and bucket remain after both
   successful and failed test runs; later deployments update the same resources. Each uses 2 GiB / 1 vCPU and a single
@@ -74,17 +74,14 @@ eventual cleanup cannot turn an earlier lifecycle violation into a pass.
   Each real checkpoint/poll call has a request-local call ID paired with its
   `finally` exit. Missing or mismatched exits prevent cleanup from reusing the
   fixture, and an exit after wrapper return remains a lifecycle failure.
-* A bounded same-JVM root barrier establishes the fixed-pool reproduction.
-  `fixed2` uses a shared two-thread pool so two roots starve their first steps.
-  `nested2` uses four threads and a separate cohort-scoped child barrier. Both
-  child handlers must enter before either schedules nested map/parallel work.
-  The child barrier has an independent 8-second setup budget; the nested progress
-  budget starts when it releases, not when the roots passed their earlier barrier.
-  A broken child barrier fails setup without growing the pool; an escape after
-  successful child admission still fails the starvation assertion.
-  Other cases hold steps with private S3 control objects. The driver requires
-  distinct request IDs active in the same JVM. Placement has its own deadline
-  and failure category. Environment replacement is never worker recovery.
+* A bounded same-JVM root barrier verifies that `fixed2` gives each invocation
+  its own two-thread executor. `nested2` is retained in the persistent stack for
+  baseline diagnostics but does not produce a gating LMI case. It uses four
+  shared threads and a separate cohort-scoped child barrier before scheduling
+  nested map/parallel work. Other cases hold steps with private S3 control
+  objects. The driver requires distinct request IDs active in the same JVM.
+  Placement has its own deadline and failure category. Environment replacement
+  is never worker recovery.
 * Timeout victims start before healthy peers, leaving the peers time to hold
   the other runtime slots during recovery. Diagnostics capture the real context
   deadline; no fake clock, context, checkpoint backend, or time-skipping runner
@@ -179,8 +176,9 @@ python3 lmi-tests/cloud_suite.py collect
 ```
 
 The deploy command creates or updates and retains `default1`, `default2`, `default8`,
-`fixed2`, and `nested2` together. The test command requires all five and runs the
-full suite. CI publishes one combined artifact, named `lmi-e2e-RUN-ATTEMPT`, with
+`fixed2`, and `nested2` together. The test command requires all five and runs 12
+gating cases; `nested2` is retained but excluded. CI publishes one combined
+artifact, named `lmi-e2e-RUN-ATTEMPT`, with
 the single `template.json` and per-function configuration snapshots. The
 CloudFormation execution role needs permission to manage the function scaling
 configuration; the driver calls `lambda:GetFunctionScalingConfig` to verify it
@@ -260,9 +258,10 @@ a normally ordered retry cannot replace an admitted participant.
 After stopping durable executions, cleanup allows any pre-stop stubborn retry its
 configured residual hold (capped at 120 seconds), plus the existing 30-second
 collection margin and 20-second quiet period. This cleanup allowance does not alter
-the original invocation deadline + 5s or progress/recovery acceptance limits. No virtual-thread executor variant is deployed
-until its executor contract is defined; default cached and shared fixed pools
-are covered separately.
+the original invocation deadline + 5s or progress/recovery acceptance limits.
+No virtual-thread executor variant is deployed until its executor contract is defined.
+The cloud suite gates default cached and invocation-local fixed pools; the shared
+fixed nesting fixture remains available as a non-gating baseline diagnostic.
 
 CloudFormation's native [FunctionScalingConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/TemplateReference/aws-properties-lambda-function-functionscalingconfig.html)
 sets the limits in the resource declaration. LMI provisioning and version behavior

lmi-tests/cloud_suite.py

@@ -25,7 +25,7 @@
 OWNER = "java-sdk-lmi-e2e"
 DEFAULT_STACK = "java-lmi-e2e"
 FIXTURES = {"default1": (1, "default"), "default2": (2, "default"),
-            "default8": (8, "default"), "fixed2": (2, "fixed"), "nested2": (2, "nested")}
+            "default8": (8, "default"), "fixed2": (2, "invocation-fixed"), "nested2": (2, "nested")}
 PEER_LOG_OBSERVATION_SECONDS = 20
 PEER_PLACEMENT_SECONDS = 60
 
@@ -201,7 +201,7 @@ def deploy_fixtures(manifest, tags, seconds=1800):
         wait_stack(manifest["stack"], expected, remaining_budget(deadline, seconds))
     stack = existing_stack(manifest["stack"])
     outputs = {output["OutputKey"]: output["OutputValue"] for output in stack["Outputs"]}
-    require(set(outputs) == set(FIXTURES), "The stack must expose all five LMI fixtures")
+    require(set(outputs) == set(FIXTURES), "The stack must expose all required LMI fixtures")
     for fixture in FIXTURES:
         arn = outputs[fixture]
         record_fixture(manifest, fixture, arn)
@@ -893,10 +893,10 @@ def collect_warm_snapshots(cloud, items, seconds=30):
 def cases_for_fixture(cloud, fixture):
     if fixture == "default1":
         return [("baseline-concurrency1", lambda: replay_case(cloud, fixture, "baseline"))]
-    if fixture in {"fixed2", "nested2"}:
-        scenario = "fixed" if fixture == "fixed2" else "nested"
-        return [("fixed-two-roots" if fixture == "fixed2" else "fixed-nested-map-parallel",
-                 lambda: fixed_case(cloud, fixture, scenario))]
+    if fixture == "fixed2":
+        return [("isolated-fixed-two-roots", lambda: fixed_case(cloud, fixture, "fixed"))]
+    if fixture == "nested2":
+        return []
     cases = [(fixture + "-isolation", lambda: overlap_case(cloud, fixture)),
              (fixture + "-suspend-cleanup-replay", lambda: replay_case(cloud, fixture, "suspend")),
              (fixture + "-timeout-recovery", lambda: timeout_case(cloud, fixture))]
@@ -1022,7 +1022,7 @@ def revalidate_cases(cloud, records):
 def run_tests():
     manifest = json.loads(MANIFEST.read_text())
     if set(manifest["functions"]) != set(FIXTURES):
-        raise PreconditionError("All five functions must be deployed before running the cloud suite")
+        raise PreconditionError("All required functions must be deployed before running the cloud suite")
     cloud = Cloud(manifest, ARTIFACTS)
     suite = ET.Element("testsuite", name="LMI cloud lifecycle")
     records = []

lmi-tests/cloud_support.py

@@ -213,7 +213,7 @@ def assert_fixed(events, markers, progress_start="BARRIER_PASSED"):
     require(len({e["requestId"] for e in entered}) == len(markers), "Missing distinct runtime invocations")
     require(max(e["sequence"] for e in entered) < min(e["sequence"] for e in passed),
             "Root invocations did not overlap at the barrier")
-    require(not selected(events, "ESCAPE"), "Shared fixed executor starved its own queued work")
+    require(not selected(events, "ESCAPE"), "Fixed executor starved its own queued work")
     # Child admission is one cohort-wide phase. A participant scheduled later must
     # not get a fresh progress budget after its peer has already left the barrier.
     cohort_start = min((e["nanos"] for e in selected(events, progress_start)), default=None)
@@ -222,7 +222,7 @@ def assert_fixed(events, markers, progress_start="BARRIER_PASSED"):
         start = selected(events, progress_start, marker)
         started = cohort_start if progress_start == "CHILD_BARRIER_PASSED" else (start[0]["nanos"] if start else None)
         require(start and progress and 0 <= (progress[-1]["nanos"] - started) < 8_000_000_000,
-                "Shared executor did not progress within budget")
+                "Fixed executor did not progress within budget")
 
 
 def assert_nested_reached(events, markers):

lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/LifecycleHandler.java

@@ -25,6 +25,7 @@
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
 import software.amazon.awssdk.services.lambda.model.OperationType;
 import software.amazon.lambda.durable.DurableConfig;
 import software.amazon.lambda.durable.DurableContext;
@@ -75,27 +76,42 @@ public void handleRequest(InputStream source, OutputStream destination, Context
         var input = WIRE.deserialize(
                 new String(source.readAllBytes(), StandardCharsets.UTF_8), TypeToken.get(DurableExecutionInput.class));
         var trace = new InvocationTrace(readUserInput(input), runtime, input.durableExecutionArn());
+        var invocationExecutor = new AtomicReference<ThreadPoolExecutor>();
         trace.wrapperEnter();
         var status = "THREW";
         try {
-            var config = DurableConfig.builder()
-                    .withExecutorService(shared.getExecutorService())
-                    .withDurableExecutionClient(new ObservedClient(shared.getDurableExecutionClient(), trace))
-                    .build();
+            var configBuilder = DurableConfig.builder()
+                    .withDurableExecutionClient(new ObservedClient(shared.getDurableExecutionClient(), trace));
+            if ("invocation-fixed".equals(System.getenv("LMI_EXECUTOR"))) {
+                configBuilder.withInvocationExecutorFactory(() -> {
+                    var executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(2);
+                    invocationExecutor.set(executor);
+                    return executor;
+                });
+            } else {
+                configBuilder.withExecutorService(shared.getExecutorService());
+            }
+            var config = configBuilder.build();
             var response = DurableExecutor.execute(
                     input,
                     runtime,
                     TypeToken.get(FixtureInput.class),
-                    (value, context) -> handle(value, context, trace),
+                    (value, context) -> handle(value, context, trace, executor(invocationExecutor)),
                     config);
             destination.write(WIRE.serialize(response).getBytes(StandardCharsets.UTF_8));
             status = response.status().name();
         } finally {
-            trace.snapshot((ThreadPoolExecutor) shared.getExecutorService());
+            trace.snapshot(executor(invocationExecutor));
             trace.wrapperExit(status);
         }
     }
 
+    private ThreadPoolExecutor executor(AtomicReference<ThreadPoolExecutor> invocationExecutor) {
+        return invocationExecutor.get() != null
+                ? invocationExecutor.get()
+                : (ThreadPoolExecutor) shared.getExecutorService();
+    }
+
     private static FixtureInput readUserInput(DurableExecutionInput input) {
         var operation = input.initialExecutionState().operations().stream()
                 .filter(op -> op.type() == OperationType.EXECUTION)
@@ -104,7 +120,8 @@ private static FixtureInput readUserInput(DurableExecutionInput input) {
         return JSON.deserialize(operation.executionDetails().inputPayload(), TypeToken.get(FixtureInput.class));
     }
 
-    String handle(FixtureInput input, DurableContext context, InvocationTrace trace) {
+    String handle(
+            FixtureInput input, DurableContext context, InvocationTrace trace, ThreadPoolExecutor invocationExecutor) {
         trace.rootEnter();
         try {
             if (!context.isReplaying()
@@ -113,7 +130,7 @@ String handle(FixtureInput input, DurableContext context, InvocationTrace trace)
                 trace.event("PLACEMENT_MISS");
                 return "PLACEMENT_MISS";
             }
-            return runScenario(input, context, trace);
+            return runScenario(input, context, trace, invocationExecutor);
         } finally {
             if ("suspend".equals(input.scenario())) {
                 trace.event("CLEANUP_ENTER");
@@ -124,13 +141,14 @@ String handle(FixtureInput input, DurableContext context, InvocationTrace trace)
         }
     }
 
-    private String runScenario(FixtureInput input, DurableContext context, InvocationTrace trace) {
+    private String runScenario(
+            FixtureInput input, DurableContext context, InvocationTrace trace, ThreadPoolExecutor invocationExecutor) {
         return switch (input.scenario()) {
             case "baseline", "replay", "suspend" -> replay(input, context, trace);
             case "hold", "probe" -> context.step("held-step", String.class, step -> hold(input, trace), NO_RETRY);
             case "timeout", "failure-inflight", "return-inflight" -> inFlight(input, context, trace);
             case "stubborn" -> stubborn(input, context, trace);
-            case "fixed", "nested" -> fixed(input, context, trace);
+            case "fixed", "nested" -> fixed(input, context, trace, invocationExecutor);
             case "success" -> context.step("success", String.class, step -> body(trace, "success", input.marker()));
             case "failure" -> context.step("failure", String.class, step -> fail(trace, input.marker()), NO_RETRY);
             default -> throw new IllegalArgumentException("Unknown fixture scenario");
@@ -226,8 +244,7 @@ private String stubborn(FixtureInput input, DurableContext context, InvocationTr
         return input.marker();
     }
 
-    private String fixed(FixtureInput input, DurableContext context, InvocationTrace trace) {
-        var pool = (ThreadPoolExecutor) shared.getExecutorService();
+    private String fixed(FixtureInput input, DurableContext context, InvocationTrace trace, ThreadPoolExecutor pool) {
         awaitRootBarrier(input, trace);
         if ("nested".equals(input.scenario())) {
             var value = context.step("success", String.class, step -> body(trace, "success", input.marker()));