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