Fix BZ 69068 - trigger read timeouts for async reads on HTTP/2

Test case provided by hypnoce

java/org/apache/coyote/http2/LocalStrings.properties

@@ -121,6 +121,7 @@ streamProcessor.error.connection=Connection [{0}], Stream [{1}], An error occurr
 streamProcessor.error.stream=Connection [{0}], Stream [{1}], An error occurred during processing that was fatal to the stream
 streamProcessor.flushBufferedWrite.entry=Connection [{0}], Stream [{1}], Flushing buffered writes
 streamProcessor.service.error=Error during request processing
+streamProcessor.streamReadTimeout=Stream read timeout
 
 streamStateMachine.debug.change=Connection [{0}], Stream [{1}], State changed from [{2}] to [{3}]
 streamStateMachine.invalidFrame=Connection [{0}], Stream [{1}], State [{2}], Frame type [{3}]

java/org/apache/coyote/http2/Stream.java

@@ -1075,6 +1075,8 @@ abstract class StreamInputBuffer implements InputBuffer {
         abstract boolean isRequestBodyFullyRead();
 
         abstract void insertReplayedBody(ByteChunk body);
+
+        protected abstract boolean timeoutRead(long now);
     }
 
 
@@ -1101,6 +1103,8 @@ class StandardStreamInputBuffer extends StreamInputBuffer {
         // 'write mode'.
         private volatile ByteBuffer inBuffer;
         private volatile boolean readInterest;
+        // If readInterest is true, data must be available to read no later than this time.
+        private volatile long readTimeoutExpiry;
         private volatile boolean closed;
         private boolean resetReceived;
 
@@ -1199,6 +1203,12 @@ final boolean isReadyForRead() {
 
                 if (!isRequestBodyFullyRead()) {
                     readInterest = true;
+                    long readTimeout = handler.getProtocol().getStreamReadTimeout();
+                    if (readTimeout > 0) {
+                        readTimeoutExpiry = System.currentTimeMillis() + handler.getProtocol().getStreamReadTimeout();
+                    } else {
+                        readTimeoutExpiry = Long.MAX_VALUE;
+                    }
                 }
 
                 return false;
@@ -1350,6 +1360,12 @@ final void swallowUnread() throws IOException {
                 }
             }
         }
+
+
+        @Override
+        protected boolean timeoutRead(long now) {
+            return readInterest && now > readTimeoutExpiry;
+        }
     }
 
 
@@ -1411,5 +1427,12 @@ boolean isRequestBodyFullyRead() {
         void insertReplayedBody(ByteChunk body) {
             // NO-OP
         }
+
+
+        @Override
+        protected boolean timeoutRead(long now) {
+            // Reading from a saved request. Will never time out.
+            return false;
+        }
     }
 }

java/org/apache/coyote/http2/StreamProcessor.java

@@ -18,13 +18,15 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.net.SocketTimeoutException;
 import java.util.Enumeration;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.Set;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
+import jakarta.servlet.RequestDispatcher;
 import jakarta.servlet.ServletConnection;
 import jakarta.servlet.http.HttpServletResponse;
 
@@ -543,4 +545,22 @@ protected final boolean flushBufferedWrite() throws IOException {
     protected final SocketState dispatchEndRequest() throws IOException {
         return SocketState.CLOSED;
     }
+
+
+    /**
+     * {@inheritDoc}
+     * <p>
+     * First checks for a stream read timeout and processes it if detected. If no stream read timeout is detected then
+     * the superclass is called to check for an asynchronous processing timeout.
+     */
+    @Override
+    public void timeoutAsync(long now) {
+        if (stream.getInputBuffer().timeoutRead(now)) {
+            stream.getCoyoteRequest().setAttribute(RequestDispatcher.ERROR_EXCEPTION,
+                    new SocketTimeoutException(sm.getString("streamProcessor.streamReadTimeout")));
+            processSocketEvent(SocketEvent.ERROR, true);
+        } else {
+            super.timeoutAsync(now);
+        }
+    }
 }

test/org/apache/coyote/http2/Http2TestBase.java

@@ -113,7 +113,7 @@ public static Collection<Object[]> data() {
     protected static final String TRAILER_HEADER_VALUE = "test";
 
     // Client
-    private Socket s;
+    protected Socket s;
     protected HpackEncoder hpackEncoder;
     protected Input input;
     protected TestOutput output;

test/org/apache/coyote/http2/TestAsyncReadListener.java

@@ -0,0 +1,145 @@
+/*
+ * 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.http2;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import jakarta.servlet.AsyncContext;
+import jakarta.servlet.AsyncEvent;
+import jakarta.servlet.AsyncListener;
+import jakarta.servlet.ReadListener;
+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.Before;
+import org.junit.Test;
+
+import org.apache.catalina.Context;
+import org.apache.catalina.LifecycleException;
+import org.apache.catalina.Wrapper;
+import org.apache.catalina.startup.Tomcat;
+
+public class TestAsyncReadListener extends Http2TestBase {
+
+    private AsyncServlet asyncServlet;
+
+    @Before
+    public void before() {
+        asyncServlet = new AsyncServlet();
+    }
+
+    @Test
+    public void testEmptyWindow() throws Exception {
+        http2Connect();
+
+        byte[] headersFrameHeader = new byte[9];
+        ByteBuffer headersPayload = ByteBuffer.allocate(128);
+        byte[] dataFrameHeader = new byte[9];
+        ByteBuffer dataPayload = ByteBuffer.allocate(256);
+        byte[] trailerFrameHeader = new byte[9];
+        ByteBuffer trailerPayload = ByteBuffer.allocate(256);
+
+
+        buildPostRequest(headersFrameHeader, headersPayload, false, null, -1, "/async", dataFrameHeader, dataPayload,
+            null, trailerFrameHeader, trailerPayload, 3);
+
+        synchronized (asyncServlet) {
+            // Write the headers
+            writeFrame(headersFrameHeader, headersPayload);
+            asyncServlet.wait(4000);
+            s.close();
+            asyncServlet.wait(4000);
+        }
+        Assert.assertNotNull(asyncServlet.t);
+    }
+
+    @Override
+    protected void configureAndStartWebApplication() throws LifecycleException {
+        Tomcat tomcat = getTomcatInstance();
+
+        Context ctxt = getProgrammaticRootContext();
+        Tomcat.addServlet(ctxt, "simple", new SimpleServlet());
+        ctxt.addServletMappingDecoded("/simple", "simple");
+        Wrapper w = Tomcat.addServlet(ctxt, "async", asyncServlet);
+        w.setAsyncSupported(true);
+        ctxt.addServletMappingDecoded("/async", "async");
+        tomcat.start();
+    }
+
+    public static class AsyncServlet extends HttpServlet {
+        private static final long serialVersionUID = 1L;
+        private volatile Throwable t;
+        @Override
+        protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
+            final AsyncContext asyncContext = req.startAsync();
+            asyncContext.getRequest().getInputStream().setReadListener(new ReadListener() {
+                @Override
+                public void onDataAvailable() throws IOException {
+                    System.out.println("RL-onDataAvailable");
+                    synchronized (AsyncServlet.this) {
+                        AsyncServlet.this.notifyAll();
+                    }
+                }
+
+                @Override
+                public void onAllDataRead() throws IOException {
+                    System.out.println("RL-onAllDataRead");
+                }
+
+                @Override
+                public void onError(Throwable throwable) {
+                    System.out.println("RL-onError ");
+                    System.out.println(throwable);
+                    synchronized (AsyncServlet.this) {
+                        t = throwable;
+                        AsyncServlet.this.notifyAll();
+                        asyncContext.complete();
+                    }
+                }
+            });
+            System.out.println("setReadListener");
+            asyncContext.addListener(new AsyncListener() {
+
+                @Override
+                public void onComplete(AsyncEvent event) throws IOException {
+                    System.out.println("AL-onComplete");
+                }
+
+                @Override
+                public void onTimeout(AsyncEvent event) throws IOException {
+                    System.out.println("AL-onTimeout");
+                }
+
+                @Override
+                public void onError(AsyncEvent event) throws IOException {
+                    System.out.println("AL-onError");
+                }
+
+                @Override
+                public void onStartAsync(AsyncEvent event) throws IOException {
+                    System.out.println("AL-onStartAsync");
+                }
+            });
+            System.out.println("setAsyncListener");
+        }
+    }
+}

webapps/docs/changelog.xml

@@ -130,6 +130,10 @@
         Fix a crash on Windows setting CA certificate on null path.
         (remm)
       </fix>
+      <fix>
+        <bug>69068</bug>: Ensure read timouts are triggered for asynchronous,
+        non-blocking reads when using HTTP/2. (markt)
+      </fix>
     </changelog>
   </subsection>
   <subsection name="Other">