quic: resolve various small further issues & omissions in HTTP/3 API

Author: pimterry

doc/api/quic.md

@@ -2392,10 +2392,11 @@ added: REPLACEME
 
 Attaches HTTP/3 to an existing `node:quic` session.
 
-The HTTP/3 session must be attached before the QUIC session becomes active and
-begins emitting events: for servers that means synchronously inside a server's
-`onsession` callback, or for clients before the handshake completes. Throws
-`ERR_INVALID_STATE` if the QUIC session is already attached or active.
+The HTTP/3 session must be attached before any streams are created locally, and
+before the QUIC session becomes active and begins emitting events. For servers
+that means synchronously inside a server's `onsession` callback, or for clients
+before the handshake completes. Throws `ERR_INVALID_STATE` if the QUIC session
+is already attached, already active, or already has streams.
 
 ### `http3session.request(headers[, options])`
 
@@ -2766,6 +2767,16 @@ added: REPLACEME
 
 The session that owns this stream. Read only.
 
+### `http3stream.stream`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {quic.QuicStream}
+
+The underlying QUIC stream. Read only.
+
 ### `http3stream.onheaders`
 
 <!-- YAML
@@ -3161,7 +3172,7 @@ added:
 
 The ALPN (Application-Layer Protocol Negotiation) identifier(s). This option is
 **required**: `node:quic` is transport-only and assumes no application protocol,
-so a session created without an ALPN is rejected with an `ERR_INVALID_ARG_VALUE`
+so a session created without an ALPN is rejected with an `ERR_MISSING_OPTION`
 error. Consumers that layer a protocol on top set it themselves (for example
 [`connectHttp3()`][] negotiates `'h3'`).
 

lib/internal/quic/http3.js

@@ -3,7 +3,9 @@
 const {
   ArrayIsArray,
   ArrayPrototypePush,
-  ObjectHasOwn,
+  ArrayPrototypeSome,
+  ObjectKeys,
+  StringPrototypeToLowerCase,
   Symbol,
   SymbolAsyncIterator,
 } = primordials;
@@ -140,6 +142,12 @@ function parseHeaderPairs(pairs) {
 const kGetHttp3Handle = Symbol('kGetHttp3Handle');
 const kSubmitInitialHeaders = Symbol('kSubmitInitialHeaders');
 
+function hasPriorityHeader(headers) {
+  return ArrayPrototypeSome(
+    ObjectKeys(headers),
+    (name) => StringPrototypeToLowerCase(name) === 'priority');
+}
+
 function priorityFieldValue(level, incremental) {
   const urgency = level === 'high' ? 0 : level === 'low' ? 7 : 3;
   if (urgency === 3 && !incremental) return undefined;
@@ -254,6 +262,9 @@ class Http3Stream {
   /** @type {Http3Session} */
   get session() { return this.#session; }
 
+  /** @type {QuicStream} */
+  get stream() { return this.#stream; }
+
   /** @type {bigint} */
   get id() { return this.#stream.id; }
 
@@ -386,7 +397,7 @@ class Http3Stream {
     if (!this.#isServer) {
       const pri = priorityFieldValue(
         this.#priority.level, this.#priority.incremental);
-      if (pri !== undefined && !ObjectHasOwn(headers, 'priority')) {
+      if (pri !== undefined && !hasPriorityHeader(headers)) {
         toSend = { __proto__: null, ...headers, priority: pri };
       }
     }
@@ -683,11 +694,18 @@ class Http3Session {
       ...quicOptions
     } = options;
 
+    if (priority !== undefined) {
+      validateOneOf(priority, 'options.priority', ['default', 'low', 'high']);
+    }
+    if (incremental !== undefined) {
+      validateBoolean(incremental, 'options.incremental');
+    }
+
     let headerString;
     if (headers !== undefined) {
       let toSend = headers;
       const pri = priorityFieldValue(priority ?? 'default', incremental ?? false);
-      if (pri !== undefined && !ObjectHasOwn(headers, 'priority')) {
+      if (pri !== undefined && !hasPriorityHeader(headers)) {
         toSend = { __proto__: null, ...headers, priority: pri };
       }
       headerString = buildNgHeaderString(

lib/internal/quic/quic.js

@@ -98,6 +98,7 @@ const {
     ERR_INVALID_STATE,
     ERR_INVALID_THIS,
     ERR_MISSING_ARGS,
+    ERR_MISSING_OPTION,
     ERR_OUT_OF_RANGE,
     ERR_QUIC_CONNECTION_FAILED,
     ERR_QUIC_ENDPOINT_CLOSED,
@@ -4433,9 +4434,7 @@ function processTlsOptions(tls, forServer) {
 
   // Encode the ALPN option to wire format (length-prefixed protocol names).
   if (alpn === undefined) {
-    throw new ERR_INVALID_ARG_VALUE(
-      'options.alpn', alpn,
-      'is required');
+    throw new ERR_MISSING_OPTION('options.alpn');
   }
   const protocols = forServer ?
     (ArrayIsArray(alpn) ? alpn : [alpn]) :

src/quic/http3.cc

@@ -251,8 +251,8 @@ MaybeLocal<Object> Http3Settings::ToObject(Environment* env) const {
   static constexpr std::string_view names[] = {"maxHeaderPairs",
                                                "maxHeaderLength",
                                                "maxFieldSectionSize",
-                                               "qpackMaxDtableCapacity",
-                                               "qpackEncoderMaxDtableCapacity",
+                                               "qpackMaxDTableCapacity",
+                                               "qpackEncoderMaxDTableCapacity",
                                                "qpackBlockedStreams",
                                                "enableConnectProtocol"};
   if (tmpl.IsEmpty()) {
@@ -1119,6 +1119,7 @@ class Http3Application final : public Session::Application {
   }
 
   void EmitHeaders(Stream* stream) {
+    stream->RecordReceiveActivity();
     auto it = header_state_.find(stream->id());
     if (it == header_state_.end()) return;
     auto& hs = it->second;
@@ -1826,11 +1827,12 @@ void CreateHttp3Handle(const FunctionCallbackInfo<Value>& args) {
   ASSIGN_OR_RETURN_UNWRAP(&session, args[0]);
 
   if (!session->has_application()) {
-    if (session->is_active()) {
+    if (session->is_active() || session->has_streams()) {
       THROW_ERR_INVALID_STATE(
           session->env(),
           "An application can only be attached to a QUIC session before it "
-          "becomes active (begins emitting events)");
+          "becomes active (begins emitting events) and before any streams "
+          "are created");
       return;
     }
     session->SetApplication(CreateHttp3Application(session));

src/quic/session.cc

@@ -3375,6 +3375,12 @@ BaseObjectPtr<Stream> Session::FindStream(stream_id id) const {
   return it->second;
 }
 
+bool Session::has_streams() const {
+  if (is_destroyed()) return false;
+  return !impl_->streams_.empty() || !pending_bidi_stream_queue().IsEmpty() ||
+         !pending_uni_stream_queue().IsEmpty();
+}
+
 Session::StreamsMap Session::streams() const {
   if (is_destroyed()) return {};
   return impl_->streams_;

src/quic/session.h

@@ -327,6 +327,8 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
   bool has_application() const;
   Application& application() const;
 
+  bool has_streams() const;
+
   // True once the session has started delivering events to JS, which we use
   // as a gate for attaching an Application - it has to happen before this.
   bool is_active() const { return active_; }

src/quic/streams.cc

@@ -1636,6 +1636,10 @@ void Stream::ReceiveStopSending(QuicError error) {
   EmitStopSending(error);
 }
 
+void Stream::RecordReceiveActivity() {
+  STAT_RECORD_TIMESTAMP(Stats, received_at);
+}
+
 void Stream::ReceiveStreamReset(uint64_t final_size, QuicError error) {
   // Importantly, reset stream only impacts the inbound data flow. It has no
   // impact on the outbound data flow. It is essentially a signal that the peer
@@ -1780,7 +1784,6 @@ void Stream::EmitStopSending(const QuicError& error) {
   MakeCallback(BindingData::Get(env()).stream_stop_sending_callback(), 1, &err);
 }
 
-
 // ============================================================================
 
 void Stream::Schedule(Queue* queue) {

src/quic/streams.h

@@ -343,6 +343,11 @@ class Stream final : public AsyncWrap,
   void ReceiveStopSending(QuicError error);
   void ReceiveStreamReset(uint64_t final_size, QuicError error);
 
+  // Records receive-side activity on the stream (the received_at stat, which
+  // also feeds the idle-timeout clock). ReceiveData updates this for DATA
+  // payloads; applications must call it for non-DATA traffic they deliver.
+  void RecordReceiveActivity();
+
   // Sends a reset stream to the peer to tell it we will not be sending any
   // more data for this stream. This has the effect of shutting down the
   // writable side of the stream for this peer. Any data that is held in the
@@ -408,7 +413,6 @@ class Stream final : public AsyncWrap,
   // Notifies the JavaScript side that the peer asked it to stop sending.
   void EmitStopSending(const QuicError& error);
 
-
   // Notifies the JavaScript side that sending data on the stream has been
   // blocked because of flow control restriction.
   void EmitBlocked();

test/parallel/test-quic-alpn.mjs

@@ -49,9 +49,9 @@ await clientSession.close();
 // rather than silently defaulting.
 await assert.rejects(listen(mustNotCall(), {
   sni: { '*': { keys: [key], certs: [cert] } },
-}), { code: 'ERR_INVALID_ARG_VALUE', message: /options\.alpn/ });
+}), { code: 'ERR_MISSING_OPTION', message: /options\.alpn/ });
 await assert.rejects(connect(serverEndpoint.address, { verifyPeer: 'manual' }), {
-  code: 'ERR_INVALID_ARG_VALUE', message: /options\.alpn/,
+  code: 'ERR_MISSING_OPTION', message: /options\.alpn/,
 });
 
 await serverEndpoint.close();

test/parallel/test-quic-h3-no-application.mjs

@@ -103,3 +103,31 @@ const tooLate = { code: 'ERR_INVALID_STATE', message: /before it becomes active/
   await client.close();
   await endpoint.close();
 }
+
+// Client: a wrap after the session has created streams is rejected, even
+// pre-handshake: an application installed then would strand the pending
+// native streams' queued data. The pending stream itself must still be
+// delivered over the native path once the handshake completes.
+{
+  const serverGot = Promise.withResolvers();
+  const endpoint = await quicListen(mustCall((quicSession) => {
+    quicSession.onstream = mustCall(async (stream) => {
+      assert.strictEqual(dec.decode(await bytes(stream)), 'x');
+      quicSession.close();
+      serverGot.resolve();
+    });
+  }), serverOpts);
+  const client = await quicConnect(endpoint.address, clientOpts);
+  // Create a raw (native-path) stream before the handshake completes;
+  // it is pending until the handshake, but already owns queued data.
+  const raw = await client.createUnidirectionalStream({ body: enc.encode('x') });
+  assert.throws(() => new Http3Session(client), {
+    code: 'ERR_INVALID_STATE',
+    message: /before any streams are created/,
+  });
+  await client.opened;
+  await serverGot.promise;
+  await raw.closed;
+  await client.close();
+  await endpoint.close();
+}