Preserve chain exception policy across asynchronous continuations

README.md

@@ -253,16 +253,82 @@ does not re-enter routing or write shared timeout fields. The streaming
 ExternalServiceWorkerDispatchTest forces read-listener continuation and verifies
 that an unrelated request progresses while the upstream is blocked; it also
 checks body preservation, one rewrite, direct GETs, and unchanged unmatched routing.
-The worker continuation currently constructs a new default ExceptionHandler per
-dispatched request. It returns the default structured error response (ERR10011 /
-HTTP 400 for an uncaught IOException) after the original exception-handler frame
-has returned. This does not reuse the chain's configured handler or its enabled
-flag: the dispatched path applies default exception handling even when the chain
-intentionally disables it. Other worker-entry paths still use their surrounding
-chain policy. This is an accepted limitation of this fix, tracked in
-[issue #2820](https://github.com/networknt/light-4j/issues/2820) for separate
-exception-module continuation support and review. The current implementation
-retains ingress-proxy's compile dependency on the exception module.
+HandlerContinuation in the handler module captures the active chain exception
+boundary before asynchronous body reads and matched-request worker dispatch.
+ExceptionHandler supplies its existing policy without rebuilding middleware or
+changing its next handler. Named handlers are retained; when exception handling
+is disabled and omitted from the chain, failures remain under Undertow's default
+policy. Nested boundaries restore their enclosing scope, and state is isolated
+per exchange. Capturing a boundary does not snapshot thread-local MDC. At resume,
+one scope reads the directly captured request holder (or an empty context when
+absent), then restores the executing thread's previous MDC on every exit. Nested
+frames reuse that scope. Policy identity and exchange identity share one frame
+tracker; proxy callbacks do not maintain a second thread-local guard. Framework
+MDC updates use one stable RequestMdcContext holder with an atomic immutable-map
+reference. The first boundary execution or capture initializes that holder
+before its asynchronous hand-off; addHandlerMDCContext can also initialize it on
+first serial use. Later updates replace only the holder's snapshot, never its
+attachment. RequestState stores only MDC in a concurrent weak-identity index; it never stores
+an exception policy. Entries are removed through bounded ReferenceQueue cleanup
+after their exchange is collected; initialization never registers a completion listener.
+Captured scopes read the holder directly, avoiding index
+lookups on callback resumes. An explicitly retained HandlerContinuation.Scope carries
+the lexical policy across hand-offs; a new capture outside its frame has no policy.
+This prevents an earlier sub-pipeline boundary from shadowing a later one. Native
+Undertow HandlerWrapper policies are applied at execution, not capture time.
+Custom asynchronous code must initialize/capture before hand-off.
+The Map-typed attachment key remains compatible with read-only consumers; direct
+attachment replacement or mutation is unsupported after initialization.
+Proxy callbacks use this scope rather than the unscoped
+LightThreadExecutor. ExceptionHandler retains its original response-channel
+availability policy after commitment; it does not forcibly close general-service
+responses. Its initial worker frame still clears request MDC at chain exit.
+BodyHandler, DumpHandler and McpHandler redispatch, HMAC replay/admin completion,
+and ProxyHandler acquisition, retries, timeout cancellation and client send/response
+callbacks carry the selected boundary. The body-read listener itself uses its
+pre-suspension scope and scopes interceptor
+execution as well as downstream handling, and preserves pending-read lifecycle.
+Proxy actions reuse one captured policy; internal synchronous calls do not add
+boundary frames or reset MDC again. Unexpected proxy failures cancel only the
+upstream request's channels on multiplexed connections. Before response commitment
+an unhandled failure returns 500; after commitment it closes the failed response
+stream without appending an error. Undertow's HTTP/2 ServerConnection.close()
+sends RST_STREAM for that stream, preserving healthy siblings. Late failures on
+completed exchanges do not rewrite a response or close a reused connection.
+The neutral Runnable adapter attempts a 500 before commitment or a stream reset
+after commitment before propagating the original Error. Failed writes/resets are
+attached as suppressed failures and logged without replacing the original Error;
+selected exception policies keep
+their existing handling behavior. Ordinary exceptions use the same guarded cleanup.
+Proxy callbacks execute on the downstream exchange's I/O thread, without a
+callback-wide monitor or shared mutable callback slot. A per-request FIFO queues
+both foreign-thread and owner-thread arrivals, preserving submission order. The queue
+is private to proxy-handler. Fatal cleanup runs on its I/O owner before
+the queue stops. Executor rejection closes an immutable resource snapshot
+published through a volatile field
+and reports the failure through the selected policy once; it never reads or mutates
+the owner's request/connection/attempt/stopping fields. Later stopped submissions
+only close their own incoming resources. Accepted work that fails replays only its
+failure through that policy. The public `CONNECTION`
+attachment is populated when a connection is accepted. Retry generations discard
+late acquisition/send/response callbacks and release the previous attempt's resources
+before accepting a new connection. A proxy-generated terminal response is marked
+before sending, so benign late callbacks leave it draining. Pushed exchanges inherit
+the selected policy and an independent MDC snapshot, with their own cancellation
+state. Unexpected committed-response callbacks reset only the affected HTTP/2 stream.
+Local allocation probes are separate qualification artifacts, not repository source
+or CI qualification; gateway performance remains an explicit review gate.
+The review-11 ThreadMXBean probe includes request setup and four callbacks: both
+branches measured 1.25-1.31 KB with empty MDC and 2.97-3.40 KB with three fields
+(20k warm-up, three 20k samples). A separate real HTTP/1.1 loopback probe includes
+ExceptionHandler, ProxyHandler, upstream transport and the Java HTTP client; its
+129-133 KB/request totals are not attributable solely to continuation overhead.
+Neither probe is a production throughput qualification.
+Custom asynchronous handlers must explicitly capture before registering their
+callbacks; this contract cannot intercept arbitrary third-party hand-offs.
+Body-reading requests capture lazily only when a read suspends. ingress-proxy uses
+exception only for tests. Interrupted upstream sends restore the interrupt flag
+and stop retrying immediately.
 Regressions compare
 worker-entry and streamed-continuation failures, cover 405 rejection and metrics
 injection, and check that concurrent initialization shares one HTTP client.

body/src/main/java/com/networknt/body/BodyHandler.java

@@ -19,6 +19,7 @@
 import tools.jackson.core.type.TypeReference;
 import com.networknt.config.Config;
 import com.networknt.handler.Handler;
+import com.networknt.handler.HandlerContinuation;
 import com.networknt.handler.MiddlewareHandler;
 import com.networknt.httpstring.AttachmentConstants;
 import com.networknt.utility.StringUtils;
@@ -105,7 +106,7 @@ public void handleRequest(final HttpServerExchange exchange) throws Exception {
         String contentType = exchange.getRequestHeaders().getFirst(Headers.CONTENT_TYPE);
         if (contentType != null) {
             if (exchange.isInIoThread()) {
-                exchange.dispatch(this);
+                exchange.dispatch(HandlerContinuation.capture(exchange, this));
                 return;
             }
             exchange.startBlocking();

body/src/test/java/com/networknt/body/BodyContinuationTest.java

@@ -0,0 +1,38 @@
+package com.networknt.body;
+
+import com.networknt.handler.HandlerContinuation;
+import io.undertow.Undertow;
+import io.undertow.server.HttpHandler;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.net.http.*;
+import java.time.Duration;
+import static org.junit.jupiter.api.Assertions.*;
+
+class BodyContinuationTest {
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    void redispatchRetainsBoundary(boolean enabled) throws Exception {
+        BodyHandler body = new BodyHandler();
+        body.setNext(exchange -> {
+            assertFalse(exchange.isInIoThread());
+            throw new IllegalStateException("downstream failure");
+        });
+        HttpHandler root = enabled ? exchange -> HandlerContinuation.execute(exchange, task -> ex -> {
+            try { task.handleRequest(ex); }
+            catch (IllegalStateException _) { ex.setStatusCode(409).getResponseSender().send("selected"); }
+        }, body) : body;
+        Undertow server = Undertow.builder().addHttpListener(0, "127.0.0.1").setHandler(root).build();
+        try (HttpClient client = HttpClient.newHttpClient()) {
+            server.start();
+            int port = ((InetSocketAddress) server.getListenerInfo().getFirst().getAddress()).getPort();
+            var response = client.send(HttpRequest.newBuilder(URI.create("http://127.0.0.1:" + port + "/"))
+                    .timeout(Duration.ofSeconds(5)).header("Content-Type", "text/plain")
+                    .POST(HttpRequest.BodyPublishers.ofString("body")).build(), HttpResponse.BodyHandlers.ofString());
+            assertEquals(enabled ? 409 : 500, response.statusCode(), response.body());
+            assertEquals(enabled, response.body().contains("selected"));
+        } finally { server.stop(); }
+    }
+}

dump/src/main/java/com/networknt/dump/DumpHandler.java

@@ -17,6 +17,7 @@
 package com.networknt.dump;
 
 import com.networknt.handler.Handler;
+import com.networknt.handler.HandlerContinuation;
 import com.networknt.handler.MiddlewareHandler;
 import io.undertow.Handlers;
 import io.undertow.server.HttpHandler;
@@ -77,7 +78,7 @@ public boolean isEnabled() {
     @Override
     public void handleRequest(final HttpServerExchange exchange) throws Exception {
         if (exchange.isInIoThread()) {
-            exchange.dispatch(this);
+            exchange.dispatch(HandlerContinuation.capture(exchange, this));
             return;
         }
         if(isEnabled()) {

dump/src/test/java/com/networknt/dump/DumpContinuationTest.java

@@ -0,0 +1,39 @@
+package com.networknt.dump;
+
+import com.networknt.handler.HandlerContinuation;
+import io.undertow.Undertow;
+import io.undertow.server.*;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.net.http.*;
+import java.time.Duration;
+import static org.junit.jupiter.api.Assertions.*;
+
+class DumpContinuationTest {
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    void redispatchRetainsSelectedBoundary(boolean enabled) throws Exception {
+        DumpHandler handler = new DumpHandler() {
+            @Override public void handleRequest(HttpServerExchange exchange) throws Exception {
+                if (!exchange.isInIoThread()) throw new IllegalStateException("resumed failure");
+                super.handleRequest(exchange);
+            }
+        };
+        HttpHandler root = enabled ? exchange -> HandlerContinuation.execute(exchange, task -> ex -> {
+            try { task.handleRequest(ex); }
+            catch (IllegalStateException _) { ex.setStatusCode(409).getResponseSender().send("selected"); }
+        }, handler) : handler;
+        Undertow server = Undertow.builder().addHttpListener(0, "127.0.0.1").setHandler(root).build();
+        try (HttpClient client = HttpClient.newHttpClient()) {
+            server.start();
+            int port = ((InetSocketAddress) server.getListenerInfo().getFirst().getAddress()).getPort();
+            var response = client.send(HttpRequest.newBuilder(URI.create("http://127.0.0.1:" + port + "/mcp"))
+                    .timeout(Duration.ofSeconds(5)).POST(HttpRequest.BodyPublishers.ofString("{}"))
+                    .build(), HttpResponse.BodyHandlers.ofString());
+            assertEquals(enabled ? 409 : 500, response.statusCode(), response.body());
+            assertEquals(enabled, response.body().contains("selected"));
+        } finally { server.stop(); }
+    }
+}

exception/src/main/java/com/networknt/exception/ExceptionHandler.java

@@ -17,7 +17,9 @@
 package com.networknt.exception;
 
 import com.networknt.handler.Handler;
+import com.networknt.handler.HandlerContinuation;
 import com.networknt.handler.MiddlewareHandler;
+import com.networknt.status.Status;
 import io.undertow.Handlers;
 import io.undertow.server.HttpHandler;
 import io.undertow.server.HttpServerExchange;
@@ -79,58 +81,52 @@ public void handleRequest(final HttpServerExchange exchange) throws Exception {
             return;
         }
 
-        // These two lines of code should be placed here to avoid the log being printed twice
-        // when switching from the IO thread to the worker thread.
+        try {
+            HandlerContinuation.execute(exchange, continuation -> ex -> handleContinuation(ex, continuation),
+                    ex -> Handler.next(ex, next));
+        } finally {
+            // Initial chain execution owns request-thread cleanup. Resumed frames
+            // scope and restore the executing thread's MDC themselves.
+            MDC.clear();
+        }
+    }
+
+    private void handleContinuation(HttpServerExchange exchange, HttpHandler continuation) throws Exception {
+        // Log each executing frame, including asynchronous continuations.
         String path = exchange.getRequestPath();
-        if(logger.isDebugEnabled()) logger.debug("ExceptionHandler starting - path: {}", path);
+        if(logger.isDebugEnabled()) logger.debug("ExceptionHandler frame starting - path: {}", path);
 
         try {
-            Handler.next(exchange, next);
+            continuation.handleRequest(exchange);
         } catch (Throwable e) {
             logger.error("ExceptionHandler error processing - path: {}", path, e);
-            if(exchange.isResponseChannelAvailable()) {
-                //handle exceptions
-                if(e instanceof RuntimeException) {
-                    // check if it is FrameworkException which is subclass of RuntimeException.
-                    if(e instanceof FrameworkException) {
-                        FrameworkException fe = (FrameworkException)e;
-                        exchange.setStatusCode(fe.getStatus().getStatusCode());
-                        exchange.getResponseHeaders().put(Headers.CONTENT_TYPE, "application/json");
-                        exchange.getResponseSender().send(fe.getStatus().toString());
-                        logger.error(fe.getStatus().toString(), e);
-                    } else {
-                        setExchangeStatus(exchange, STATUS_RUNTIME_EXCEPTION);
-                    }
-                } else {
-                    if(e instanceof ApiException) {
-                        ApiException ae = (ApiException)e;
-                        exchange.setStatusCode(ae.getStatus().getStatusCode());
-                        exchange.getResponseHeaders().put(Headers.CONTENT_TYPE, "application/json");
-                        exchange.getResponseSender().send(ae.getStatus().toString());
-                        logger.error(ae.getStatus().toString(), e);
-                    } else if(e instanceof ClientException){
-                        ClientException ce = (ClientException)e;
-                        if(ce.getStatus().getStatusCode() == 0){
-                            setExchangeStatus(exchange, STATUS_UNCAUGHT_EXCEPTION);
-                        } else {
-                            exchange.setStatusCode(ce.getStatus().getStatusCode());
-                            exchange.getResponseHeaders().put(Headers.CONTENT_TYPE, "application/json");
-                            exchange.getResponseSender().send(ce.getStatus().toString());
-                        }
-
-                    } else {
-                        setExchangeStatus(exchange, STATUS_UNCAUGHT_EXCEPTION);
-                    }
-                }
-            }
+            if (exchange.isResponseChannelAvailable()) writeFailure(exchange, e);
         } finally {
-            if(logger.isDebugEnabled()) logger.debug("ExceptionHandler completed - path: {}", path);
-            // at last, clean the MDC. Most likely, correlationId in side.
-            //logger.debug("Clear MDC");
-            MDC.clear();
+            if(logger.isDebugEnabled()) logger.debug("ExceptionHandler frame completed - path: {}", path);
         }
     }
 
+    private void writeFailure(HttpServerExchange exchange, Throwable failure) {
+        if (failure instanceof FrameworkException framework) {
+            sendTypedStatus(exchange, framework.getStatus(), failure, true);
+        } else if (failure instanceof RuntimeException) {
+            setExchangeStatus(exchange, STATUS_RUNTIME_EXCEPTION);
+        } else if (failure instanceof ApiException api) {
+            sendTypedStatus(exchange, api.getStatus(), failure, true);
+        } else if (failure instanceof ClientException) {
+            Status status = ClientException.getStatus();
+            if (status.getStatusCode() == 0) setExchangeStatus(exchange, STATUS_UNCAUGHT_EXCEPTION);
+            else sendTypedStatus(exchange, status, failure, false);
+        } else setExchangeStatus(exchange, STATUS_UNCAUGHT_EXCEPTION);
+    }
+
+    private static void sendTypedStatus(HttpServerExchange exchange, Status status, Throwable failure, boolean log) {
+        exchange.setStatusCode(status.getStatusCode());
+        exchange.getResponseHeaders().put(Headers.CONTENT_TYPE, "application/json");
+        exchange.getResponseSender().send(status.toString());
+        if (log && logger.isErrorEnabled()) logger.error(status.toString(), failure);
+    }
+
     @Override
     public HttpHandler getNext() {
         return next;