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-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;
         }
     }

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

@@ -150,7 +150,7 @@ private void processDataFrames(boolean setFillInterest)
             long error = HTTP3ErrorCode.REQUEST_CANCELLED_ERROR.code();
             getEndPoint().close(error, x);
             // Notify the application that a failure happened.
-            parser.getListener().onStreamFailure(getEndPoint().getStreamId(), error, x);
+            parser.getListener().onStreamFailure(getEndPoint().getStreamId(), false, error, x);
         }
     }
 
@@ -238,7 +238,7 @@ private void processNonDataFrames()
             long error = HTTP3ErrorCode.REQUEST_CANCELLED_ERROR.code();
             getEndPoint().close(error, x);
             // Notify the application that a failure happened.
-            parser.getListener().onStreamFailure(getEndPoint().getStreamId(), error, x);
+            parser.getListener().onStreamFailure(getEndPoint().getStreamId(), false, error, x);
         }
     }
 

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

@@ -109,28 +109,28 @@ else if (filled < 0)
         }
         catch (QpackException.SessionException x)
         {
-            fail(x.getErrorCode(), x.getMessage(), x);
+            fail(x.getErrorCode(), false, x.getMessage(), x);
         }
         catch (Throwable x)
         {
-            fail(HTTP3ErrorCode.INTERNAL_ERROR.code(), "internal_error", x);
+            fail(HTTP3ErrorCode.INTERNAL_ERROR.code(), false, "internal_error", x);
         }
     }
 
-    private void fail(long errorCode, String message, Throwable failure)
+    private void fail(long errorCode, boolean remote, String message, Throwable failure)
     {
         buffer.release();
         buffer = null;
         if (LOG.isDebugEnabled())
             LOG.debug("could not process instruction stream {}", getEndPoint(), failure);
-        notifySessionFailure(errorCode, message, failure);
+        notifySessionFailure(errorCode, remote, message, failure);
     }
 
-    protected void notifySessionFailure(long error, String reason, Throwable failure)
+    protected void notifySessionFailure(long error, boolean remote, String reason, Throwable failure)
     {
         try
         {
-            listener.onSessionFailure(error, reason, failure);
+            listener.onSessionFailure(error, remote, reason, failure);
         }
         catch (Throwable x)
         {

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

@@ -243,7 +243,7 @@ public default void onDisconnect(Session session, long error, String reason)
          * @param reason the failure reason
          * @param failure the failure
          */
-        public default void onFailure(Session session, long error, String reason, Throwable failure)
+        public default void onFailure(Session session, boolean remote, long error, String reason, Throwable failure)
         {
         }
     }

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

@@ -253,10 +253,11 @@ public default void onIdleTimeout(Client stream, Throwable failure, Promise<Bool
              * the stream has been reset.</p>
              *
              * @param stream the stream
+             * @param remote true if the error comes from a remote notification
              * @param error the failure error
              * @param failure the cause of the failure
              */
-            public default void onFailure(Stream.Client stream, long error, Throwable failure)
+            public default void onFailure(Stream.Client stream, boolean remote, long error, Throwable failure)
             {
             }
         }
@@ -369,10 +370,11 @@ public default void onIdleTimeout(Server stream, TimeoutException failure, Promi
              * the stream has been reset.</p>
              *
              * @param stream the stream
+             * @param remote true if the error comes from a remote notification
              * @param error the failure error
              * @param failure the cause of the failure
              */
-            public default void onFailure(Stream.Server stream, long error, Throwable failure)
+            public default void onFailure(Server stream, boolean remote, long error, Throwable failure)
             {
             }
         }

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

@@ -66,32 +66,32 @@ protected long getBodyLength()
 
     protected void emptyBody(ByteBuffer buffer)
     {
-        sessionFailure(buffer, HTTP3ErrorCode.PROTOCOL_ERROR.code(), "invalid_frame", new IOException("invalid empty body frame"));
+        sessionFailure(buffer, HTTP3ErrorCode.PROTOCOL_ERROR.code(), false, "invalid_frame", new IOException("invalid empty body frame"));
     }
 
-    protected void sessionFailure(ByteBuffer buffer, long error, String reason, Throwable failure)
+    protected void sessionFailure(ByteBuffer buffer, long error, boolean remote, String reason, Throwable failure)
     {
         BufferUtil.clear(buffer);
-        notifySessionFailure(error, reason, failure);
+        notifySessionFailure(error, remote, reason, failure);
     }
 
-    protected void notifySessionFailure(long error, String reason, Throwable failure)
+    protected void notifySessionFailure(long error, boolean remote, String reason, Throwable failure)
     {
         try
         {
-            listener.onSessionFailure(error, reason, failure);
+            listener.onSessionFailure(error, remote, reason, failure);
         }
         catch (Throwable x)
         {
             LOG.info("failure while notifying listener {}", listener, x);
         }
     }
 
-    protected void notifyStreamFailure(long streamId, long error, Throwable failure)
+    protected void notifyStreamFailure(long streamId, boolean remote, long error, Throwable failure)
     {
         try
         {
-            listener.onStreamFailure(streamId, error, failure);
+            listener.onStreamFailure(streamId, remote, error, failure);
         }
         catch (Throwable x)
         {

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

@@ -88,7 +88,7 @@ public void parse(ByteBuffer buffer)
                                 // SPEC: message frames on the control stream are invalid.
                                 if (LOG.isDebugEnabled())
                                     LOG.debug("invalid message frame type {} on control stream", Long.toHexString(frameType));
-                                sessionFailure(buffer, HTTP3ErrorCode.FRAME_UNEXPECTED_ERROR.code(), "invalid_frame_type", new IOException("invalid message frame on control stream"));
+                                sessionFailure(buffer, false, HTTP3ErrorCode.FRAME_UNEXPECTED_ERROR.code(), "invalid_frame_type", new IOException("invalid message frame on control stream"));
                                 return;
                             }
 
@@ -135,13 +135,13 @@ public void parse(ByteBuffer buffer)
         {
             if (LOG.isDebugEnabled())
                 LOG.debug("parse failed", x);
-            sessionFailure(buffer, HTTP3ErrorCode.INTERNAL_ERROR.code(), "parser_error", x);
+            sessionFailure(buffer, false, HTTP3ErrorCode.INTERNAL_ERROR.code(), "parser_error", x);
         }
     }
 
-    private void sessionFailure(ByteBuffer buffer, long error, String reason, Throwable failure)
+    private void sessionFailure(ByteBuffer buffer, boolean remote, long error, String reason, Throwable failure)
     {
-        unknownBodyParser.sessionFailure(buffer, error, reason, failure);
+        unknownBodyParser.sessionFailure(buffer, error, remote, reason, failure);
     }
 
     private enum State

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

@@ -127,19 +127,19 @@ private boolean decode(ByteBuffer encoded, boolean last)
         {
             if (LOG.isDebugEnabled())
                 LOG.debug("decode failure", x);
-            notifyStreamFailure(streamId, x.getErrorCode(), x);
+            notifyStreamFailure(streamId, false, x.getErrorCode(), x);
         }
         catch (QpackException.SessionException x)
         {
             if (LOG.isDebugEnabled())
                 LOG.debug("decode failure", x);
-            notifySessionFailure(x.getErrorCode(), x.getMessage(), x);
+            notifySessionFailure(x.getErrorCode(), false, x.getMessage(), x);
         }
         catch (Throwable x)
         {
             if (LOG.isDebugEnabled())
                 LOG.debug("decode failure", x);
-            notifySessionFailure(HTTP3ErrorCode.INTERNAL_ERROR.code(), "internal_error", x);
+            notifySessionFailure(HTTP3ErrorCode.INTERNAL_ERROR.code(), false, "internal_error", x);
         }
         return false;
     }