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
 
@@ -95,3 +106,36 @@ file retention.
 Custom serializers can override `serialize(value, SerDesContext)` and
 `deserialize(data, typeToken, SerDesContext)` to receive stable execution/entity
 identity. Existing implementations inherit defaults that call the original methods.
+
+## Invocation deadline cleanup
+
+The SDK captures the Lambda invocation's remaining time once and uses monotonic
+cutoffs for admission and its wait policy. Deadline exhaustion selects a retryable
+invocation interruption; it is not checkpointed as a business-operation failure.
+Catching an admission rejection cannot turn the stopped invocation into a
+successful fallback. Completed operation results and failures remain available for
+replay.
+
+Cleanup allowances determine when the SDK requests cooperative cancellation and
+stops ordinary admission. They do not make cancellation equivalent to actual task
+exit. Finalization still waits for admitted user work, continuations and in-flight
+checkpoint/poll transport, and runs the matching invocation End on its owner.
+The runtime caller waits for that owner and
+End to finish, including after a deadline or interruption. End is not moved to a
+completion-observer thread or deferred until after the SDK call returns.
+
+Interruption is recorded, remaining required cleanup still runs, and the interrupt
+flag is restored. A terminal execution result acknowledged by the
+service remains authoritative for a handled deadline cutoff. An accepted result
+does not suppress an otherwise-propagating delivery or metadata failure, and an
+available JVM fatal keeps its original identity and priority.
+
+These cutoffs do not guarantee bounded caller return or worker reclamation.
+Already-running application I/O, synchronous initial-state reads, custom clients,
+plugin hooks and other code that ignores interruption can keep the caller waiting
+for actual cleanup beyond the policy budget. An executor that runs the root inline,
+including a saturated `CallerRunsPolicy`, can also block the runtime caller before
+it reaches a timeout-aware wait. Use finite downstream I/O timeouts and executor
+configurations that allow progress. Cancellation cannot undo a request already
+accepted by the service; interrupted side effects still require the documented
+retry and idempotency strategy.

docs/design.md

@@ -145,19 +145,34 @@ 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)                                                                                                                          |
 
 ### Thread Pool Architecture
 
+Each invocation tracks logical task outcomes separately from actual task exit.
+Deadline and cleanup cutoffs fence admission and request cooperative cancellation,
+while finalization still waits for admitted work and in-flight transport to finish.
+The matching invocation End runs on its owner, and the runtime caller waits for
+that owner and End before returning. Code that ignores interruption can therefore
+outlast the policy budget; see [invocation deadline cleanup](advanced/configuration.md#invocation-deadline-cleanup).
+
 The SDK uses two separate thread pools with distinct responsibilities:
 
 **User Executor (`DurableConfig.executorService`):**
-- Runs user-defined operations (the code passed to `ctx.step()` and `ctx.stepAsync()`)
+- 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

@@ -19,6 +19,15 @@ those original snapshots and retain the existing zero-work and +16-thread limits
 Newly discovered request identities invalidate the warm cohort instead of moving
 its original baseline or final measurement to another request.
 
+The current deadline controls fence admission and request cooperative cancellation,
+but the SDK still awaits actual owner cleanup, admitted transport and the matching
+invocation End before returning. Non-cooperative work can therefore keep the
+runtime caller blocked beyond its deadline. This remains a partial change for
+[#726](https://github.com/aws/aws-durable-execution-sdk-java/issues/726), not proof
+that every desired LMI lifecycle bound is satisfied. The acceptance limits below
+remain unchanged; local tests do not replace a deployed cloud acceptance run, and
+eventual cleanup cannot turn an earlier lifecycle violation into a pass.
+
 ## Test design
 
 * Each fixture is a published durable function with LMI invocation concurrency
@@ -30,7 +39,7 @@ its original baseline or final measurement to another request.
   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
@@ -65,17 +74,14 @@ its original baseline or final measurement to another request.
   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
@@ -170,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
@@ -219,7 +226,7 @@ barrier diagnostics never authorize a retry. Every attempted cohort remains in
 the escape ledger, so a later successful cohort cannot hide an earlier pool escape.
 
 `timeouts/*.json` distinguishes server timeout logs from an SDK deadline
-cancellation that returns early with an invocation error. A client HTTP timeout
+cancellation reported as an invocation error after owner cleanup. A client HTTP timeout
 is a collection error. Deadline victims use asynchronous service invocation so a
 longer durable retry does not consume the driver HTTP timeout; invocation
 outcomes come from request-correlated runtime logs. Invocation request failures
@@ -251,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):