This is an automated email from the ASF dual-hosted git repository. markt-asf pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit cf17fa992319c12eeeb18e29805a80286f65542e Author: Mark Thomas <[email protected]> AuthorDate: Fri Nov 6 19:00:26 2026 +0000 Fix async request leak when dispatch is requested in an error state BZ70290. If the processor was in an error state when the container thread left the initial service call, a pending async dispatch (MUST_DISPATCH) could never be performed because the connection is closed instead. The request was recycled in that state so onComplete() was not called and the in-progress async count was never decremented. - AsyncStateMachine: in an error state, complete the async processing from MUST_DISPATCH (fire onComplete(), decrement the count). - CoyoteAdapter.service(): trigger the early post-processing for a pending dispatch as well as a pending completion. Add tests that drive the state machine directly and tests that exercise the behaviour through CoyoteAdapter. Co-Authored-By: Claude Sonnet 5.5 <[email protected]> --- .../apache/catalina/connector/CoyoteAdapter.java | 7 +- java/org/apache/coyote/AsyncStateMachine.java | 15 +- .../TestCoyoteAdapterAsyncErrorState.java | 286 ++++++++++++++++++++ test/org/apache/coyote/TestAsyncStateMachine.java | 301 +++++++++++++++++++++ webapps/docs/changelog.xml | 11 + 5 files changed, 616 insertions(+), 4 deletions(-) diff --git a/java/org/apache/catalina/connector/CoyoteAdapter.java b/java/org/apache/catalina/connector/CoyoteAdapter.java index 6875a82ac1..07dd9353e5 100644 --- a/java/org/apache/catalina/connector/CoyoteAdapter.java +++ b/java/org/apache/catalina/connector/CoyoteAdapter.java @@ -382,10 +382,11 @@ public class CoyoteAdapter implements Adapter { AtomicBoolean error = new AtomicBoolean(false); res.action(ActionCode.IS_ERROR, error); - if (request.isAsyncCompleting() && error.get()) { + if ((request.isAsyncCompleting() || request.isAsyncDispatching()) && error.get()) { // Connection will be forcibly closed which will prevent - // completion happening at the usual point. Need to trigger - // call to onComplete() here. + // completion or the dispatch requested during error handling + // from happening at the usual point. Need to trigger the call + // to onComplete() here. res.action(ActionCode.ASYNC_POST_PROCESS, null); async = false; } diff --git a/java/org/apache/coyote/AsyncStateMachine.java b/java/org/apache/coyote/AsyncStateMachine.java index f0ef0a3972..e6db49d4d2 100644 --- a/java/org/apache/coyote/AsyncStateMachine.java +++ b/java/org/apache/coyote/AsyncStateMachine.java @@ -303,7 +303,20 @@ class AsyncStateMachine { asyncCtxt.decrementInProgressAsyncCount(); return SocketState.ASYNC_END; } else if (state == AsyncState.MUST_DISPATCH) { - updateState(AsyncState.DISPATCHING); + if (!processor.getErrorState().isError()) { + updateState(AsyncState.DISPATCHING); + return SocketState.ASYNC_END; + } + // The processor is in an error state (including CLOSE_CLEAN which + // still permits IO) so the connection will be closed rather than + // being used for the dispatch. The dispatch requested (e.g. by a + // call to dispatch() from AsyncListener.onError()) can no longer be + // performed. Complete the async processing here so that + // onComplete() is called and the in-progress async count is + // balanced, rather than leaving the request in MUST_DISPATCH. + asyncCtxt.fireOnComplete(); + updateState(AsyncState.DISPATCHED); + asyncCtxt.decrementInProgressAsyncCount(); return SocketState.ASYNC_END; } else if (state == AsyncState.DISPATCHING) { if (processor.getErrorState().isIoAllowed() && processor.flushBufferedWrite()) { diff --git a/test/org/apache/catalina/connector/TestCoyoteAdapterAsyncErrorState.java b/test/org/apache/catalina/connector/TestCoyoteAdapterAsyncErrorState.java new file mode 100644 index 0000000000..db3888665c --- /dev/null +++ b/test/org/apache/catalina/connector/TestCoyoteAdapterAsyncErrorState.java @@ -0,0 +1,286 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.catalina.connector; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.InetAddress; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import jakarta.servlet.AsyncContext; +import jakarta.servlet.AsyncEvent; +import jakarta.servlet.AsyncListener; +import jakarta.servlet.DispatcherType; +import jakarta.servlet.ServletException; +import jakarta.servlet.http.HttpServlet; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletResponse; + +import org.junit.Assert; +import org.junit.Test; + +import org.apache.catalina.Context; +import org.apache.catalina.Wrapper; +import org.apache.catalina.core.StandardContext; +import org.apache.catalina.startup.Tomcat; +import org.apache.catalina.startup.TomcatBaseTest; +import org.apache.catalina.valves.ValveBase; +import org.apache.coyote.ActionCode; + +/** + * Tests the handling in {@link CoyoteAdapter#service} of a request that starts async processing and requests a + * dispatch during the initial service call while the connection is, or goes into, an error state. The error is applied + * by the servlet between {@code startAsync()} and {@code dispatch()} so the state machine passes through the same + * states in the same order every time. No client reset, timing or large response is required. + */ +public class TestCoyoteAdapterAsyncErrorState extends TomcatBaseTest { + + private static final String REQUEST_ATTRIBUTE = TestCoyoteAdapterAsyncErrorState.class.getName() + ".request"; + + private StandardContext context; + + /* + * BZ 70290: CLOSE_NOW prevents any further IO so the dispatch requested after the error can not be performed + * (MUST_ERROR -> MUST_DISPATCH). onComplete() has to be called once and the in-progress async count has to be + * balanced. Previously the request was recycled while still in MUST_DISPATCH and the count leaked. + */ + @Test + public void testBz70290DispatchAfterCloseNowIsCompleted() throws Exception { + TestServlet servlet = new TestServlet(ActionCode.CLOSE_NOW); + startTomcat(servlet); + + sendRequest(); + + assertCompletedOnce(servlet); + Assert.assertEquals("The dispatch can not be performed", 0, servlet.dispatches.get()); + } + + + /* + * BZ 70290: CLOSE_CLEAN is an error state but it still permits IO. Whatever the outcome for the dispatch, the + * request must not be recycled with the async processing neither completed nor dispatched: onComplete() has to be + * called once and the in-progress async count has to be balanced. + */ + @Test + public void testBz70290DispatchAfterCloseCleanIsCompleted() throws Exception { + TestServlet servlet = new TestServlet(ActionCode.DISABLE_SWALLOW_INPUT); + startTomcat(servlet); + + sendRequest(); + + assertCompletedOnce(servlet); + } + + + /* + * BZ 70290: Control. With no error the dispatch has to be performed and the client has to receive the response. + */ + @Test + public void testBz70290DispatchWithoutErrorIsPerformed() throws Exception { + TestServlet servlet = new TestServlet(null); + startTomcat(servlet); + + String response = sendRequest(); + + Assert.assertTrue(response, response.startsWith("HTTP/1.1 200")); + Assert.assertTrue(response, response.contains("DISPATCHED")); + Assert.assertEquals(1, servlet.dispatches.get()); + assertCompletedOnce(servlet); + } + + + /* + * BZ 70290: Parity with the dispatch case. complete() requested after CLOSE_NOW has to result in onComplete() being + * called once and the in-progress async count being balanced. + */ + @Test + public void testBz70290CompleteAfterCloseNowIsCompleted() throws Exception { + TestServlet servlet = new TestServlet(ActionCode.CLOSE_NOW, true); + startTomcat(servlet); + + sendRequest(); + + assertCompletedOnce(servlet); + } + + + /* + * BZ 70290: Parity with the dispatch case. complete() requested after CLOSE_CLEAN has to result in onComplete() + * being called once and the in-progress async count being balanced. + */ + @Test + public void testBz70290CompleteAfterCloseCleanIsCompleted() throws Exception { + TestServlet servlet = new TestServlet(ActionCode.DISABLE_SWALLOW_INPUT, true); + startTomcat(servlet); + + sendRequest(); + + assertCompletedOnce(servlet); + } + + + /* + * BZ 70290: Control. complete() with no error completes normally. + */ + @Test + public void testBz70290CompleteWithoutErrorIsCompleted() throws Exception { + TestServlet servlet = new TestServlet(null, true); + startTomcat(servlet); + + String response = sendRequest(); + + Assert.assertTrue(response, response.startsWith("HTTP/1.1 200")); + assertCompletedOnce(servlet); + } + + + private void startTomcat(TestServlet servlet) throws Exception { + Tomcat tomcat = getTomcatInstance(); + Context ctx = getProgrammaticRootContext(); + context = (StandardContext) ctx; + + Wrapper w = Tomcat.addServlet(ctx, "test", servlet); + w.setAsyncSupported(true); + ctx.addServletMapping("/test", "test"); + + // Makes the Catalina request available to the servlet so it can put the connection into an error state + ctx.getPipeline().addValve(new ValveBase(true) { + @Override + public void invoke(Request request, Response response) throws IOException, ServletException { + request.setAttribute(REQUEST_ATTRIBUTE, request); + getNext().invoke(request, response); + } + }); + + tomcat.start(); + } + + + /* + * Raw socket rather than HttpURLConnection since the latter silently retries a GET if the connection is closed + * without a response. + */ + private String sendRequest() throws IOException { + try (Socket socket = new Socket(InetAddress.getLoopbackAddress(), getPort())) { + socket.setSoTimeout(5000); + socket.getOutputStream().write(("GET /test HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .getBytes(StandardCharsets.US_ASCII)); + socket.getOutputStream().flush(); + + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + InputStream is = socket.getInputStream(); + byte[] buf = new byte[1024]; + try { + int n; + while ((n = is.read(buf)) >= 0) { + baos.write(buf, 0, n); + } + } catch (IOException ioe) { + // Connection reset or timeout. Whatever was received is returned. + } + return baos.toString(StandardCharsets.ISO_8859_1); + } + } + + + private void assertCompletedOnce(TestServlet servlet) throws Exception { + Assert.assertTrue("onComplete() was not called", servlet.completed.await(5, TimeUnit.SECONDS)); + Assert.assertEquals("onComplete() must be called exactly once", 1, servlet.completions.get()); + + // The count is decremented just after onComplete() returns + StandardContext ctx = context; + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (ctx.getInProgressAsyncCount() != 0 && System.nanoTime() < deadline) { + Thread.sleep(10); + } + Assert.assertEquals("In-progress async count must be balanced", 0, ctx.getInProgressAsyncCount()); + } + + + private static class TestServlet extends HttpServlet { + + private static final long serialVersionUID = 1L; + + private final ActionCode errorAction; + private final boolean complete; + + final AtomicInteger dispatches = new AtomicInteger(); + final AtomicInteger completions = new AtomicInteger(); + final CountDownLatch completed = new CountDownLatch(1); + + TestServlet(ActionCode errorAction) { + this(errorAction, false); + } + + TestServlet(ActionCode errorAction, boolean complete) { + this.errorAction = errorAction; + this.complete = complete; + } + + @Override + protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws IOException { + if (req.getDispatcherType() == DispatcherType.ASYNC) { + dispatches.incrementAndGet(); + resp.setContentType("text/plain"); + resp.getWriter().print("DISPATCHED"); + return; + } + + AsyncContext ac = req.startAsync(); + ac.addListener(new AsyncListener() { + @Override + public void onComplete(AsyncEvent event) { + completions.incrementAndGet(); + completed.countDown(); + } + + @Override + public void onTimeout(AsyncEvent event) { + // NO-OP + } + + @Override + public void onError(AsyncEvent event) { + // NO-OP + } + + @Override + public void onStartAsync(AsyncEvent event) { + // NO-OP + } + }); + + if (errorAction != null) { + // The connection fails + Request request = (Request) req.getAttribute(REQUEST_ATTRIBUTE); + request.getCoyoteRequest().action(errorAction, null); + } + + // Request the dispatch (or completion) on the container thread during the initial service call + if (complete) { + ac.complete(); + } else { + ac.dispatch(); + } + } + } +} diff --git a/test/org/apache/coyote/TestAsyncStateMachine.java b/test/org/apache/coyote/TestAsyncStateMachine.java new file mode 100644 index 0000000000..89cfa10124 --- /dev/null +++ b/test/org/apache/coyote/TestAsyncStateMachine.java @@ -0,0 +1,301 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.coyote; + +import java.io.IOException; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import org.apache.coyote.http11.Http11NioProtocol; +import org.apache.coyote.http11.Http11Processor; +import org.apache.tomcat.util.net.AbstractEndpoint.Handler.SocketState; +import org.apache.tomcat.util.net.SocketEvent; + +/** + * Tests the handling of the async state machine when the connection is in an error state at the point the container + * thread that made the initial service() call exits. The tests drive the state machine directly through a processor so + * the interleavings are deterministic and no sockets, threads or timing are required. + */ +public class TestAsyncStateMachine { + + private TestProcessor processor; + private TestAsyncContextCallback asyncCtxt; + + @Before + public void setupTest() { + processor = new TestProcessor(); + asyncCtxt = new TestAsyncContextCallback(); + + // The initial service() call runs on the container (request) thread + processor.getRequest().setRequestThread(); + + // ServletRequest.startAsync(). The caller increments the in-progress count when starting async. + asyncCtxt.incrementInProgressAsyncCount(); + processor.action(ActionCode.ASYNC_START, asyncCtxt); + } + + + /* + * BZ 70290: The client drops the connection during the initial service() call (STARTING -> MUST_ERROR) and + * AsyncListener.onError() then calls AsyncContext.dispatch() (MUST_ERROR -> MUST_DISPATCH). The dispatch can no + * longer be performed so onComplete() must be called and the in-progress count must be decremented. Previously the + * request was left in MUST_DISPATCH, onComplete() was never called and the count leaked. + */ + @Test + public void testBz70290DispatchFromOnErrorAfterConnectionClosed() throws IOException { + processor.setErrorState(ErrorState.CLOSE_CONNECTION_NOW, new IOException()); + Assert.assertTrue(isAsyncError()); + + processor.action(ActionCode.ASYNC_DISPATCH, null); + Assert.assertTrue(isAsyncDispatching()); + + SocketState state = processor.asyncPostProcess(); + + Assert.assertEquals(SocketState.ASYNC_END, state); + assertCompleted(); + } + + + /* + * BZ 70290: As above but the error state is CLOSE_NOW rather than CLOSE_CONNECTION_NOW. Both prevent further IO so + * both must balance the in-progress count. + */ + @Test + public void testBz70290DispatchFromOnErrorAfterStreamReset() throws IOException { + processor.setErrorState(ErrorState.CLOSE_NOW, new IOException()); + + processor.action(ActionCode.ASYNC_DISPATCH, null); + + SocketState state = processor.asyncPostProcess(); + + Assert.assertEquals(SocketState.ASYNC_END, state); + assertCompleted(); + } + + + /* + * BZ 70290: onComplete() must only be called once and the in-progress count must only be decremented once, even if + * the post processing is triggered more than once (CoyoteAdapter triggers it early when the connection is in an + * error state and the processor triggers it again on the normal path). + */ + @Test + public void testBz70290PostProcessTwiceDoesNotDoubleComplete() throws IOException { + processor.setErrorState(ErrorState.CLOSE_CONNECTION_NOW, new IOException()); + processor.action(ActionCode.ASYNC_DISPATCH, null); + + processor.asyncPostProcess(); + assertCompleted(); + + // State is now DISPATCHED. A second call is not valid and must not trigger a second completion. + try { + processor.asyncPostProcess(); + Assert.fail("Expected IllegalStateException"); + } catch (IllegalStateException expected) { + // Expected + } + assertCompleted(); + } + + + /* + * BZ 70290: Control. With no error the dispatch must still be performed: the state machine moves to DISPATCHING, + * onComplete() is not called yet and the in-progress count is only decremented once the dispatch has completed. + */ + @Test + public void testBz70290DispatchWithoutErrorIsStillPerformed() throws IOException { + processor.action(ActionCode.ASYNC_DISPATCH, null); + + SocketState state = processor.asyncPostProcess(); + + Assert.assertEquals(SocketState.ASYNC_END, state); + Assert.assertTrue("Dispatch must still be pending", isAsyncDispatching()); + Assert.assertEquals(0, asyncCtxt.getCompleteCount()); + Assert.assertEquals(0, asyncCtxt.getDecrementCount()); + + // The dispatch is processed + processor.action(ActionCode.ASYNC_DISPATCHED, null); + + Assert.assertFalse(isAsync()); + Assert.assertEquals(1, asyncCtxt.getDecrementCount()); + } + + + /* + * BZ 70290: An error state that still permits IO (CLOSE_CLEAN) still results in the connection being closed, so the + * pending dispatch can not be performed. onComplete() must be called and the in-progress count must be balanced. + */ + @Test + public void testBz70290DispatchWithIoAllowedErrorStateIsCompleted() throws IOException { + processor.setErrorState(ErrorState.CLOSE_CLEAN, null); + + processor.action(ActionCode.ASYNC_DISPATCH, null); + + SocketState state = processor.asyncPostProcess(); + + Assert.assertEquals(SocketState.ASYNC_END, state); + assertCompleted(); + } + + + /* + * BZ 70290: Control. The equivalent complete() handling (MUST_COMPLETE) in an error state already balanced the + * count. Confirm the dispatch and complete cases behave consistently. + */ + @Test + public void testBz70290CompleteAfterConnectionClosedBalancesCount() throws IOException { + processor.setErrorState(ErrorState.CLOSE_CONNECTION_NOW, new IOException()); + + processor.action(ActionCode.ASYNC_COMPLETE, null); + + SocketState state = processor.asyncPostProcess(); + + Assert.assertEquals(SocketState.ASYNC_END, state); + assertCompleted(); + } + + + /* + * BZ 70290: Parity with the dispatch case. complete() after CLOSE_NOW (MUST_ERROR -> MUST_COMPLETE or + * STARTING -> MUST_COMPLETE) must call onComplete() once and balance the in-progress count. + */ + @Test + public void testBz70290CompleteAfterCloseNowBalancesCount() throws IOException { + processor.setErrorState(ErrorState.CLOSE_NOW, new IOException()); + + processor.action(ActionCode.ASYNC_COMPLETE, null); + + SocketState state = processor.asyncPostProcess(); + + Assert.assertEquals(SocketState.ASYNC_END, state); + assertCompleted(); + } + + + /* + * BZ 70290: Parity with the dispatch case. complete() after CLOSE_CLEAN (an error state that still permits IO) must + * call onComplete() once and balance the in-progress count. + */ + @Test + public void testBz70290CompleteAfterCloseCleanBalancesCount() throws IOException { + processor.setErrorState(ErrorState.CLOSE_CLEAN, null); + + processor.action(ActionCode.ASYNC_COMPLETE, null); + + SocketState state = processor.asyncPostProcess(); + + Assert.assertEquals(SocketState.ASYNC_END, state); + assertCompleted(); + } + + + private void assertCompleted() { + Assert.assertFalse("Async processing must have ended", isAsync()); + Assert.assertEquals("onComplete() must be called exactly once", 1, asyncCtxt.getCompleteCount()); + Assert.assertEquals("In-progress async count must be decremented exactly once", 1, + asyncCtxt.getDecrementCount()); + Assert.assertEquals("In-progress async count must be balanced", 0, asyncCtxt.getInProgress()); + } + + + private boolean isAsync() { + AtomicBoolean result = new AtomicBoolean(); + processor.action(ActionCode.ASYNC_IS_ASYNC, result); + return result.get(); + } + + + private boolean isAsyncDispatching() { + AtomicBoolean result = new AtomicBoolean(); + processor.action(ActionCode.ASYNC_IS_DISPATCHING, result); + return result.get(); + } + + + private boolean isAsyncError() { + AtomicBoolean result = new AtomicBoolean(); + processor.action(ActionCode.ASYNC_IS_ERROR, result); + return result.get(); + } + + + private static class TestProcessor extends Http11Processor { + + TestProcessor() { + super(new Http11NioProtocol(), null); + } + + @Override + protected void processSocketEvent(SocketEvent event, boolean dispatch) { + // NO-OP. Tests drive the state machine directly rather than via a socket. + } + + @Override + protected boolean flushBufferedWrite() throws IOException { + // There is no socket so there is never any buffered data to flush + return false; + } + + @Override + protected void setErrorState(ErrorState errorState, Throwable t) { + super.setErrorState(errorState, t); + } + } + + + private static class TestAsyncContextCallback implements AsyncContextCallback { + + private int inProgress; + private int decrementCount; + private int completeCount; + + @Override + public void fireOnComplete() { + completeCount++; + } + + @Override + public boolean isAvailable() { + return true; + } + + @Override + public void incrementInProgressAsyncCount() { + inProgress++; + } + + @Override + public void decrementInProgressAsyncCount() { + inProgress--; + decrementCount++; + } + + int getInProgress() { + return inProgress; + } + + int getDecrementCount() { + return decrementCount; + } + + int getCompleteCount() { + return completeCount; + } + } +} diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml index d8a9a05416..8550f6b78a 100644 --- a/webapps/docs/changelog.xml +++ b/webapps/docs/changelog.xml @@ -522,6 +522,17 @@ <code>ExpiresFilter</code> now takes account of headers set by a previous filter in the filter chain. (markt) </fix> + <fix> + <bug>70290</bug>: Fix a leak of the in-progress async request count + that could prevent a clean shutdown. If the connection was in an error + state when the initial async service call completed (for example, a + client dropped the connection and an + <code>AsyncListener.onError()</code> handler then called + <code>AsyncContext.dispatch()</code>), the request could be recycled + while still in the <code>MUST_DISPATCH</code> state so that + <code>onComplete()</code> was not called and the in-progress async + count was not decremented. (markt) + </fix> </changelog> </subsection> <subsection name="Coyote"> --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
