Fix missing notifyRemoteAsyncErrors http config wiring

jetty-core/jetty-http2/jetty-http2-common/src/main/java/org/eclipse/jetty/http2/HTTP2Channel.java

@@ -63,7 +63,7 @@ public interface Server
         // TODO: can it be simplified? The callback seems to only be succeeded, which
         //  means it can be converted into a Runnable which may just be the return type
         //  so we can get rid of the Callback parameter.
-        public Runnable onFailure(Throwable failure, Callback callback);
+        public Runnable onFailure(Throwable failure, boolean remote, Callback callback);
 
         // TODO: is this needed?
         public boolean isIdle();

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

@@ -156,20 +156,15 @@ public void onDataAvailable(Stream stream)
         public void onReset(Stream stream, ResetFrame frame, Callback callback)
         {
             EofException failure = new EofException("Reset " + ErrorCode.toString(frame.getError(), null));
-            onFailure(stream, failure, callback);
+            getConnection().onStreamFailure(stream, failure, true, callback);
         }
 
         @Override
         public void onFailure(Stream stream, int error, String reason, Throwable failure, Callback callback)
         {
             if (!(failure instanceof QuietException))
                 failure = new EofException(failure);
-            onFailure(stream, failure, callback);
-        }
-
-        private void onFailure(Stream stream, Throwable failure, Callback callback)
-        {
-            getConnection().onStreamFailure(stream, failure, callback);
+            getConnection().onStreamFailure(stream, failure, false, callback);
         }
 
         @Override

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

@@ -194,14 +194,14 @@ public void onStreamTimeout(Stream stream, TimeoutException timeout, Promise<Boo
         }
     }
 
-    public void onStreamFailure(Stream stream, Throwable failure, Callback callback)
+    public void onStreamFailure(Stream stream, Throwable failure, boolean remote, Callback callback)
     {
         if (LOG.isDebugEnabled())
-            LOG.debug("Processing stream failure on {}", stream, failure);
+            LOG.debug("Processing {}stream failure on {}", remote ? "remote " : "", stream, failure);
         HTTP2Channel.Server channel = (HTTP2Channel.Server)((HTTP2Stream)stream).getAttachment();
         if (channel != null)
         {
-            Runnable task = channel.onFailure(failure, callback);
+            Runnable task = channel.onFailure(failure, remote, callback);
             if (task != null)
             {
                 // We must dispatch to another thread because the task

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

@@ -585,9 +585,9 @@ public void onTimeout(TimeoutException timeout, BiConsumer<Runnable, Boolean> co
     }
 
     @Override
-    public Runnable onFailure(Throwable failure, Callback callback)
+    public Runnable onFailure(Throwable failure, boolean remote, Callback callback)
     {
-        Runnable runnable = _httpChannel.onFailure(failure);
+        Runnable runnable = remote ? _httpChannel.onRemoteFailure(failure) : _httpChannel.onFailure(failure);
         return () ->
         {
             if (runnable != null)

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

@@ -67,10 +67,10 @@ public void onTimeout(TimeoutException timeout, BiConsumer<Runnable, Boolean> co
     }
 
     @Override
-    public Runnable onFailure(Throwable failure, Callback callback)
+    public Runnable onFailure(Throwable failure, boolean remote, Callback callback)
     {
         if (LOG.isDebugEnabled())
-            LOG.debug("failure on {}", this, failure);
+            LOG.debug("{}failure on {}", remote ? "remote " : "", this, failure);
         processFailure(failure);
         close(failure);
         return callback::succeeded;

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-client-transport/src/main/java/org/eclipse/jetty/http3/client/transport/internal/HttpReceiverOverHTTP3.java

@@ -154,7 +154,7 @@ public void onIdleTimeout(Stream.Client stream, Throwable failure, Promise<Boole
     }
 
     @Override
-    public void onFailure(Stream.Client stream, long error, Throwable failure)
+    public void onFailure(Stream.Client stream, boolean remote, long error, Throwable failure)
     {
         responseFailure(failure, Promise.noop());
     }

jetty-core/jetty-http3/jetty-http3-client-transport/src/main/java/org/eclipse/jetty/http3/client/transport/internal/SessionClientListener.java

@@ -77,11 +77,11 @@ public boolean onIdleTimeout(Session session)
     @Override
     public void onDisconnect(Session session, long error, String reason)
     {
-        onFailure(session, error, reason, new ClosedChannelException());
+        onFailure(session, false, error, reason, new ClosedChannelException());
     }
 
     @Override
-    public void onFailure(Session session, long error, String reason, Throwable failure)
+    public void onFailure(Session session, boolean remote, long error, String reason, Throwable failure)
     {
         if (failConnectionPromise(failure))
             return;

jetty-core/jetty-http3/jetty-http3-client/src/main/java/org/eclipse/jetty/http3/client/ClientHTTP3Session.java

@@ -204,7 +204,7 @@ else if (key == SettingsFrame.MAX_BLOCKED_STREAMS)
     private void failControlStream(Throwable failure)
     {
         long error = HTTP3ErrorCode.CLOSED_CRITICAL_STREAM_ERROR.code();
-        onFailure(error, "control_stream_failure", failure);
+        onFailure(false, error, "control_stream_failure", failure);
     }
 
     @Override
@@ -254,9 +254,9 @@ protected boolean onIdleTimeout()
     }
 
     @Override
-    protected void onFailure(long error, String reason, Throwable failure)
+    protected void onFailure(boolean remote, long error, String reason, Throwable failure)
     {
-        session.onSessionFailure(error, reason, failure);
+        session.onSessionFailure(error, remote, reason, failure);
     }
 
     @Override

jetty-core/jetty-http3/jetty-http3-client/src/main/java/org/eclipse/jetty/http3/client/HTTP3StreamClient.java

@@ -148,13 +148,13 @@ protected void notifyIdleTimeout(TimeoutException timeout, Promise<Boolean> prom
     }
 
     @Override
-    protected void notifyFailure(long error, Throwable failure)
+    protected void notifyFailure(boolean remote, long error, Throwable failure)
     {
         Listener listener = getListener();
         try
         {
             if (listener != null)
-                listener.onFailure(this, error, failure);
+                listener.onFailure(this, remote, error, failure);
         }
         catch (Throwable x)
         {

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

@@ -439,7 +439,7 @@ public void onData(long streamId, DataFrame frame)
         if (stream != null)
             stream.onData(frame);
         else
-            onSessionFailure(HTTP3ErrorCode.FRAME_UNEXPECTED_ERROR.code(), "invalid_frame_sequence", new IllegalStateException("invalid frame sequence"));
+            onSessionFailure(HTTP3ErrorCode.FRAME_UNEXPECTED_ERROR.code(), false, "invalid_frame_sequence", new IllegalStateException("invalid frame sequence"));
     }
 
     @Override
@@ -682,7 +682,7 @@ private void failStreams(Predicate<HTTP3Stream> predicate, long error, String re
                     stream.reset(error, failure);
                 // Since the stream failure was generated
                 // by a GOAWAY, notify the application.
-                stream.onFailure(error, failure);
+                stream.onFailure(false, error, failure);
             });
     }
 
@@ -796,7 +796,7 @@ public void onClose(long error, String reason)
         failStreams(stream -> true, error, reason, false, failure);
 
         if (notifyFailure)
-            onSessionFailure(error, reason, failure);
+            onSessionFailure(error, false, reason, failure);
 
         notifyDisconnect(error, reason);
     }
@@ -814,27 +814,27 @@ private void notifyDisconnect(long error, String reason)
     }
 
     @Override
-    public void onStreamFailure(long streamId, long error, Throwable failure)
+    public void onStreamFailure(long streamId, boolean remote, long error, Throwable failure)
     {
         if (LOG.isDebugEnabled())
             LOG.debug("stream failure 0x{}/{} for stream #{} on {}", Long.toHexString(error), failure.getMessage(), streamId, this);
         HTTP3Stream stream = getStream(streamId);
         if (stream != null)
-            stream.onFailure(error, failure);
+            stream.onFailure(remote, error, failure);
     }
 
     @Override
-    public void onSessionFailure(long error, String reason, Throwable failure)
+    public void onSessionFailure(long error, boolean remote, String reason, Throwable failure)
     {
-        notifyFailure(error, reason, failure);
+        notifyFailure(error, remote, reason, failure);
         inwardClose(error, reason);
     }
 
-    private void notifyFailure(long error, String reason, Throwable failure)
+    private void notifyFailure(long error, boolean remote, String reason, Throwable failure)
     {
         try
         {
-            listener.onFailure(this, error, reason, failure);
+            listener.onFailure(this, remote, error, reason, failure);
         }
         catch (Throwable x)
         {

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

@@ -370,13 +370,13 @@ public void onTrailer(HeadersFrame frame)
 
     protected abstract void notifyIdleTimeout(TimeoutException timeout, Promise<Boolean> promise);
 
-    public void onFailure(long error, Throwable failure)
+    public void onFailure(boolean remote, long error, Throwable failure)
     {
-        notifyFailure(error, failure);
+        notifyFailure(remote, error, failure);
         session.removeStream(this, failure);
     }
 
-    protected abstract void notifyFailure(long error, Throwable failure);
+    protected abstract void notifyFailure(boolean remote, long error, Throwable failure);
 
     protected boolean validateAndUpdate(EnumSet<FrameState> allowed, FrameState target)
     {
@@ -392,7 +392,7 @@ protected boolean validateAndUpdate(EnumSet<FrameState> allowed, FrameState targ
             if (frameState == FrameState.FAILED)
                 return false;
             frameState = FrameState.FAILED;
-            session.onSessionFailure(HTTP3ErrorCode.FRAME_UNEXPECTED_ERROR.code(), "invalid_frame_sequence", new IllegalStateException("invalid frame sequence"));
+            session.onSessionFailure(HTTP3ErrorCode.FRAME_UNEXPECTED_ERROR.code(), false, "invalid_frame_sequence", new IllegalStateException("invalid frame sequence"));
             return false;
         }
     }