Fix missing notifyRemoteAsyncErrors http config wiring

jetty-core/jetty-http2/jetty-http2-server/src/main/java/org/eclipse/jetty/http2/server/HTTP2ServerConnectionFactory.java

@@ -13,6 +13,7 @@
 
 package org.eclipse.jetty.http2.server;
 
+import java.io.EOFException;
 import java.util.Map;
 import java.util.concurrent.TimeoutException;
 
@@ -155,7 +156,7 @@ public void onDataAvailable(Stream stream)
         @Override
         public void onReset(Stream stream, ResetFrame frame, Callback callback)
         {
-            EofException failure = new EofException("Reset " + ErrorCode.toString(frame.getError(), null));
+            EOFException failure = new EOFException("Reset " + ErrorCode.toString(frame.getError(), null));
             onFailure(stream, failure, callback);
         }
 

jetty-core/jetty-http2/jetty-http2-server/src/main/java/org/eclipse/jetty/http2/server/internal/HttpStreamOverHTTP2.java

@@ -13,6 +13,7 @@
 
 package org.eclipse.jetty.http2.server.internal;
 
+import java.io.EOFException;
 import java.nio.ByteBuffer;
 import java.util.concurrent.TimeoutException;
 import java.util.function.BiConsumer;
@@ -38,6 +39,7 @@
 import org.eclipse.jetty.io.Connection;
 import org.eclipse.jetty.io.Content;
 import org.eclipse.jetty.io.EndPoint;
+import org.eclipse.jetty.io.EofException;
 import org.eclipse.jetty.server.HttpChannel;
 import org.eclipse.jetty.server.HttpStream;
 import org.eclipse.jetty.server.Request;
@@ -587,7 +589,8 @@ public void onTimeout(TimeoutException timeout, BiConsumer<Runnable, Boolean> co
     @Override
     public Runnable onFailure(Throwable failure, Callback callback)
     {
-        Runnable runnable = _httpChannel.onFailure(failure);
+        boolean remote = failure instanceof EOFException;
+        Runnable runnable = remote ? _httpChannel.onRemoteFailure(new EofException(failure)) : _httpChannel.onFailure(failure);
         return () ->
         {
             if (runnable != null)

jetty-core/jetty-http2/jetty-http2-tests/src/test/java/org/eclipse/jetty/http2/tests/AbstractTest.java

@@ -58,7 +58,12 @@ public class AbstractTest
 
     protected void start(Handler handler) throws Exception
     {
-        HTTP2CServerConnectionFactory connectionFactory = new HTTP2CServerConnectionFactory(new HttpConfiguration());
+        start(handler, new HttpConfiguration());
+    }
+
+    protected void start(Handler handler, HttpConfiguration httpConfiguration) throws Exception
+    {
+        HTTP2CServerConnectionFactory connectionFactory = new HTTP2CServerConnectionFactory(httpConfiguration);
         connectionFactory.setInitialSessionRecvWindow(FlowControlStrategy.DEFAULT_WINDOW_SIZE);
         connectionFactory.setInitialStreamRecvWindow(FlowControlStrategy.DEFAULT_WINDOW_SIZE);
         prepareServer(connectionFactory);

jetty-core/jetty-http2/jetty-http2-tests/src/test/java/org/eclipse/jetty/http2/tests/AsyncIOTest.java

@@ -17,22 +17,33 @@
 import java.nio.ByteBuffer;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.eclipse.jetty.http.HttpFields;
 import org.eclipse.jetty.http.MetaData;
+import org.eclipse.jetty.http2.ErrorCode;
 import org.eclipse.jetty.http2.api.Session;
 import org.eclipse.jetty.http2.api.Stream;
 import org.eclipse.jetty.http2.frames.DataFrame;
 import org.eclipse.jetty.http2.frames.HeadersFrame;
+import org.eclipse.jetty.http2.frames.ResetFrame;
 import org.eclipse.jetty.io.Content;
+import org.eclipse.jetty.io.EofException;
 import org.eclipse.jetty.server.Handler;
+import org.eclipse.jetty.server.HttpConfiguration;
 import org.eclipse.jetty.server.Request;
 import org.eclipse.jetty.server.Response;
+import org.eclipse.jetty.util.BufferUtil;
 import org.eclipse.jetty.util.Callback;
 import org.eclipse.jetty.util.FuturePromise;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.nullValue;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
@@ -241,6 +252,67 @@ public void onClosed(Stream stream)
 */
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void testClientResetRemoteErrorNotification(boolean notify) throws Exception
+    {
+        CountDownLatch latch = new CountDownLatch(1);
+        AtomicReference<Response> responseRef = new AtomicReference<>();
+        AtomicReference<Throwable> failureRef = new AtomicReference<>();
+        HttpConfiguration httpConfiguration = new HttpConfiguration();
+        httpConfiguration.setNotifyRemoteAsyncErrors(notify);
+        start(new Handler.Abstract()
+        {
+            @Override
+            public boolean handle(Request request, Response response, Callback callback)
+            {
+                request.addFailureListener(failureRef::set);
+                responseRef.set(response);
+                latch.countDown();
+                return true;
+            }
+        }, httpConfiguration);
+
+        Session session = newClientSession(new Session.Listener() {});
+        MetaData.Request metaData = newRequest("GET", HttpFields.EMPTY);
+        HeadersFrame frame = new HeadersFrame(metaData, null, true);
+        FuturePromise<Stream> promise = new FuturePromise<>();
+        session.newStream(frame, promise, null);
+        Stream stream = promise.get(5, TimeUnit.SECONDS);
+
+        // Wait for the server to be idle.
+        assertTrue(latch.await(5, TimeUnit.SECONDS));
+        sleep(500);
+
+        stream.reset(new ResetFrame(stream.getId(), ErrorCode.CANCEL_STREAM_ERROR.code), Callback.NOOP);
+
+        if (notify)
+            // Wait for the reset to be notified to the failure listener.
+            await().atMost(5, TimeUnit.SECONDS).until(failureRef::get, instanceOf(EofException.class));
+        else
+            // Wait for the reset to NOT be notified to the failure listener.
+            await().atMost(5, TimeUnit.SECONDS).during(1, TimeUnit.SECONDS).until(failureRef::get, nullValue());
+
+        // Assert that writing to the response fails.
+        var cb = new Callback()
+        {
+            private Throwable failure = null;
+
+            @Override
+            public void failed(Throwable x)
+            {
+                failure = x;
+            }
+
+            Throwable failure()
+            {
+                return failure;
+            }
+        };
+        responseRef.get().write(true, BufferUtil.EMPTY_BUFFER, cb);
+        await().atMost(5, TimeUnit.SECONDS).until(cb::failure, instanceOf(EofException.class));
+    }
+
     private static void sleep(long ms) throws InterruptedIOException
     {
         try

jetty-core/jetty-http2/jetty-http2-tests/src/test/java/org/eclipse/jetty/http2/tests/AsyncServletTest.java

@@ -154,58 +154,6 @@ public void test()
 //        assertTrue(clientLatch.await(2 * idleTimeout, TimeUnit.MILLISECONDS));
 //    }
 //
-//    @Test
-//    public void testStartAsyncThenClientResetWithoutRemoteErrorNotification() throws Exception
-//    {
-//        HttpConfiguration httpConfiguration = new HttpConfiguration();
-//        httpConfiguration.setNotifyRemoteAsyncErrors(false);
-//        prepareServer(new HTTP2ServerConnectionFactory(httpConfiguration));
-//        ServletContextHandler context = new ServletContextHandler(server, "/");
-//        AtomicReference<AsyncContext> asyncContextRef = new AtomicReference<>();
-//        CountDownLatch latch = new CountDownLatch(1);
-//        context.addServlet(new ServletHolder(new HttpServlet()
-//        {
-//            @Override
-//            protected void service(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException
-//            {
-//                AsyncContext asyncContext = request.startAsync();
-//                asyncContext.setTimeout(0);
-//                asyncContextRef.set(asyncContext);
-//                latch.countDown();
-//            }
-//        }), servletPath + "/*");
-//        server.start();
-//
-//        prepareClient();
-//        client.start();
-//        Session session = newClient(new Session.Listener() {});
-//        MetaData.Request metaData = newRequest("GET", HttpFields.EMPTY);
-//        HeadersFrame frame = new HeadersFrame(metaData, null, true);
-//        FuturePromise<Stream> promise = new FuturePromise<>();
-//        session.newStream(frame, promise, null);
-//        Stream stream = promise.get(5, TimeUnit.SECONDS);
-//
-//        // Wait for the server to be in ASYNC_WAIT.
-//        assertTrue(latch.await(5, TimeUnit.SECONDS));
-//        sleep(500);
-//
-//        stream.reset(new ResetFrame(stream.getId(), ErrorCode.CANCEL_STREAM_ERROR.code), Callback.NOOP);
-//
-//        // Wait for the reset to be processed by the server.
-//        sleep(500);
-//
-//        AsyncContext asyncContext = asyncContextRef.get();
-//        ServletResponse response = asyncContext.getResponse();
-//        ServletOutputStream output = response.getOutputStream();
-//
-//        assertThrows(IOException.class,
-//            () ->
-//            {
-//                // Large writes or explicit flush() must
-//                // fail because the stream has been reset.
-//                output.flush();
-//            });
-//    }
 //
 //    @Test
 //    public void testStartAsyncThenServerSessionIdleTimeout() throws Exception

jetty-core/jetty-http3/jetty-http3-common/src/main/java/org/eclipse/jetty/http3/HTTP3StreamConnection.java

@@ -14,7 +14,6 @@
 package org.eclipse.jetty.http3;
 
 import java.io.IOException;
-import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeoutException;
@@ -275,7 +274,7 @@ private void tryReleaseInputBuffer(boolean force)
         }
     }
 
-    private MessageParser.Result parseAndFill(boolean setFillInterest)
+    private MessageParser.Result parseAndFill(boolean setFillInterest) throws IOException
     {
         try
         {
@@ -336,16 +335,9 @@ private MessageParser.Result parseAndFill(boolean setFillInterest)
         }
     }
 
-    private int fill(ByteBuffer byteBuffer)
+    private int fill(ByteBuffer byteBuffer) throws IOException
     {
-        try
-        {
-            return getEndPoint().fill(byteBuffer);
-        }
-        catch (IOException x)
-        {
-            throw new UncheckedIOException(x.getMessage(), x);
-        }
+        return getEndPoint().fill(byteBuffer);
     }
 
     private void processHeaders(HeadersFrame frame, boolean wasBlocked, Runnable delegate)

jetty-core/jetty-http3/jetty-http3-server/src/main/java/org/eclipse/jetty/http3/server/internal/HttpStreamOverHTTP3.java

@@ -13,6 +13,7 @@
 
 package org.eclipse.jetty.http3.server.internal;
 
+import java.io.EOFException;
 import java.nio.ByteBuffer;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeoutException;
@@ -33,6 +34,7 @@
 import org.eclipse.jetty.http3.frames.DataFrame;
 import org.eclipse.jetty.http3.frames.HeadersFrame;
 import org.eclipse.jetty.io.Content;
+import org.eclipse.jetty.io.EofException;
 import org.eclipse.jetty.server.HttpChannel;
 import org.eclipse.jetty.server.HttpStream;
 import org.eclipse.jetty.server.Request;
@@ -536,6 +538,8 @@ public Runnable onFailure(Throwable failure)
             chunk = Content.Chunk.from(failure, true);
         }
         connection.onFailure(failure);
-        return httpChannel.onFailure(failure);
+
+        boolean remote = failure instanceof EOFException;
+        return remote ? httpChannel.onRemoteFailure(new EofException(failure)) : httpChannel.onFailure(failure);
     }
 }

jetty-core/jetty-quic/jetty-quic-common/src/main/java/org/eclipse/jetty/quic/common/QuicStreamEndPoint.java

@@ -13,6 +13,7 @@
 
 package org.eclipse.jetty.quic.common;
 
+import java.io.EOFException;
 import java.io.IOException;
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
@@ -41,6 +42,7 @@ public class QuicStreamEndPoint extends AbstractEndPoint
 {
     private static final Logger LOG = LoggerFactory.getLogger(QuicStreamEndPoint.class);
     private static final ByteBuffer LAST_FLAG = ByteBuffer.allocate(0);
+    private static final ByteBuffer EMPTY_WRITABLE_BUFFER = ByteBuffer.allocate(0);
 
     private final QuicSession session;
     private final long streamId;
@@ -265,12 +267,25 @@ public boolean onReadable()
         }
         else
         {
-            QuicStreamEndPoint streamEndPoint = getQuicSession().getStreamEndPoint(streamId);
-            if (streamEndPoint.isStreamFinished())
+            if (isStreamFinished())
             {
-                EofException e = new EofException();
-                streamEndPoint.getFillInterest().onFail(e);
-                streamEndPoint.getQuicSession().onFailure(e);
+                // Check if the stream was finished normally.
+                try
+                {
+                    fill(EMPTY_WRITABLE_BUFFER);
+                }
+                catch (EOFException x)
+                {
+                    // Got reset.
+                    getFillInterest().onFail(x);
+                    getQuicSession().onFailure(x);
+                }
+                catch (Throwable x)
+                {
+                    EofException e = new EofException(x);
+                    getFillInterest().onFail(e);
+                    getQuicSession().onFailure(e);
+                }
             }
         }
         return interested;

jetty-core/jetty-quic/jetty-quic-quiche/jetty-quic-quiche-foreign/src/main/java/org/eclipse/jetty/quic/quiche/foreign/ForeignQuicheConnection.java

@@ -14,6 +14,7 @@
 package org.eclipse.jetty.quic.quiche.foreign;
 
 import java.io.ByteArrayOutputStream;
+import java.io.EOFException;
 import java.io.IOException;
 import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
@@ -918,14 +919,19 @@ public int drainClearBytesForStream(long streamId, ByteBuffer buffer) throws IOE
                     MemorySegment fin = scope.allocate(NativeHelper.C_CHAR);
                     read = quiche_h.quiche_conn_stream_recv(quicheConn, streamId, bufferSegment, buffer.remaining(), fin);
 
-                    int prevPosition = buffer.position();
-                    buffer.put(bufferSegment.asByteBuffer().limit((int)read));
-                    buffer.position(prevPosition);
+                    if (read > 0)
+                    {
+                        int prevPosition = buffer.position();
+                        buffer.put(bufferSegment.asByteBuffer().limit((int)read));
+                        buffer.position(prevPosition);
+                    }
                 }
             }
 
             if (read == quiche_error.QUICHE_ERR_DONE)
                 return isStreamFinished(streamId) ? -1 : 0;
+            if (read == quiche_error.QUICHE_ERR_STREAM_RESET)
+                throw new EOFException("failed to read from stream " + streamId + "; quiche_err=" + quiche_error.errToString(read));
             if (read < 0L)
                 throw new IOException("failed to read from stream " + streamId + "; quiche_err=" + quiche_error.errToString(read));
             buffer.position((int)(buffer.position() + read));

jetty-core/jetty-quic/jetty-quic-quiche/jetty-quic-quiche-jna/src/main/java/org/eclipse/jetty/quic/quiche/jna/JnaQuicheConnection.java

@@ -14,6 +14,7 @@
 package org.eclipse.jetty.quic.quiche.jna;
 
 import java.io.ByteArrayOutputStream;
+import java.io.EOFException;
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
@@ -747,6 +748,8 @@ public int drainClearBytesForStream(long streamId, ByteBuffer buffer) throws IOE
             int read = LibQuiche.INSTANCE.quiche_conn_stream_recv(quicheConn, new uint64_t(streamId), buffer, new size_t(buffer.remaining()), fin).intValue();
             if (read == quiche_error.QUICHE_ERR_DONE)
                 return isStreamFinished(streamId) ? -1 : 0;
+            if (read == quiche_error.QUICHE_ERR_STREAM_RESET)
+                throw new EOFException("failed to read from stream " + streamId + "; quiche_err=" + quiche_error.errToString(read));
             if (read < 0L)
                 throw new IOException("failed to read from stream " + streamId + "; quiche_err=" + quiche_error.errToString(read));
             buffer.position(buffer.position() + read);