Issue #6728 - QUIC and HTTP/3

- Fixed end of stream detection.

Signed-off-by: Simone Bordet <simone.bordet@gmail.com>

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

@@ -54,7 +54,7 @@ public abstract class HTTP3Session extends ContainerLifeCycle implements Session
     private static final Logger LOG = LoggerFactory.getLogger(HTTP3Session.class);
 
     private final AutoLock lock = new AutoLock();
-    private final AtomicLong lastId = new AtomicLong();
+    private final AtomicLong lastStreamId = new AtomicLong(0);
     private final Map<Long, HTTP3Stream> streams = new ConcurrentHashMap<>();
     private final ProtocolSession session;
     private final Session.Listener listener;
@@ -225,7 +225,7 @@ public abstract class HTTP3Session extends ContainerLifeCycle implements Session
 
     protected GoAwayFrame newGoAwayFrame(boolean graceful)
     {
-        return new GoAwayFrame(lastId.get());
+        return new GoAwayFrame(lastStreamId.get());
     }
 
     public CompletableFuture<Void> shutdown()
@@ -241,9 +241,9 @@ public abstract class HTTP3Session extends ContainerLifeCycle implements Session
         return result;
     }
 
-    protected void updateLastId(long id)
+    private void updateLastStreamId(long id)
     {
-        Atomics.updateMax(lastId, id);
+        Atomics.updateMax(lastStreamId, id);
     }
 
     public long getIdleTimeout()
@@ -340,6 +340,8 @@ public abstract class HTTP3Session extends ContainerLifeCycle implements Session
             long idleTimeout = getStreamIdleTimeout();
             if (idleTimeout > 0)
                 stream.setIdleTimeout(idleTimeout);
+            if (!local)
+                updateLastStreamId(stream.getId());
             if (LOG.isDebugEnabled())
                 LOG.debug("created {}", stream);
             return stream;

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

@@ -396,20 +396,24 @@ public abstract class HTTP3StreamConnection extends AbstractConnection
                 if (filled > 0)
                     continue;
 
-                if (!remotelyClosed && getEndPoint().isStreamFinished())
-                {
-                    if (LOG.isDebugEnabled())
-                        LOG.debug("detected end of stream on {}", this);
-                    parser.parse(EMPTY_DATA_FRAME.slice());
-                    return MessageParser.Result.FRAME;
-                }
-
                 if (filled == 0)
                 {
+                    // Workaround for a Quiche glitch, that sometimes reports
+                    // an HTTP/3 frame with last=false, but a subsequent read
+                    // of zero bytes reports that the stream is finished.
+                    if (!remotelyClosed && getEndPoint().isStreamFinished())
+                    {
+                        if (LOG.isDebugEnabled())
+                            LOG.debug("detected end of stream on {}", this);
+                        parser.parse(EMPTY_DATA_FRAME.slice());
+                        return MessageParser.Result.FRAME;
+                    }
+
                     setNoData(true);
                     if (setFillInterest)
                         fillInterested();
                 }
+
                 return MessageParser.Result.NO_FRAME;
             }
         }

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

@@ -62,10 +62,7 @@ public class HTTP3SessionServer extends HTTP3Session implements Session.Server
             if (LOG.isDebugEnabled())
                 LOG.debug("received request {} on {}", frame, stream);
             if (stream != null)
-            {
-                updateLastId(streamId);
                 stream.onRequest(frame);
-            }
         }
         else
         {

jetty-quic/quic-server/src/main/java/org/eclipse/jetty/quic/server/ServerProtocolSession.java

@@ -26,7 +26,7 @@ public class ServerProtocolSession extends ProtocolSession
     private static final Logger LOG = LoggerFactory.getLogger(ServerProtocolSession.class);
 
     private final Runnable producer = Invocable.from(Invocable.InvocationType.BLOCKING, this::produce);
-    private final Consumer<QuicStreamEndPoint> configureProtocolEndPoint = this::openProtocolEndPoint;
+    private final Consumer<QuicStreamEndPoint> openProtocolEndPoint = this::openProtocolEndPoint;
 
     public ServerProtocolSession(ServerQuicSession session)
     {
@@ -71,7 +71,7 @@ public class ServerProtocolSession extends ProtocolSession
     protected boolean onReadable(long readableStreamId)
     {
         // On the server, we need a get-or-create semantic in case of reads.
-        QuicStreamEndPoint streamEndPoint = getOrCreateStreamEndPoint(readableStreamId, configureProtocolEndPoint);
+        QuicStreamEndPoint streamEndPoint = getOrCreateStreamEndPoint(readableStreamId, openProtocolEndPoint);
         if (LOG.isDebugEnabled())
             LOG.debug("stream #{} selected for read: {}", readableStreamId, streamEndPoint);
         return streamEndPoint.onReadable();