quic: use http3Binding to drop header & priority stream logic from QUIC

Author: pimterry

lib/internal/quic/http3.js

@@ -3,6 +3,7 @@
 const {
   ArrayIsArray,
   ArrayPrototypePush,
+  ObjectHasOwn,
   Symbol,
   SymbolAsyncIterator,
 } = primordials;
@@ -40,6 +41,7 @@ const {
 
 const {
   setHttp3Callbacks,
+  createHttp3Handle,
 
   QUIC_STREAM_HEADERS_KIND_INITIAL: kHeadersKindInitial,
   QUIC_STREAM_HEADERS_KIND_HINTS: kHeadersKindHints,
@@ -65,8 +67,10 @@ const {
 } = require('internal/quic/diagnostics');
 
 const {
+  validateBoolean,
   validateFunction,
   validateObject,
+  validateOneOf,
 } = require('internal/validators');
 
 const {
@@ -128,14 +132,29 @@ function parseHeaderPairs(pairs) {
  *   will be sent after these headers.
  */
 
+const kGetHttp3Handle = Symbol('kGetHttp3Handle');
+const kSubmitInitialHeaders = Symbol('kSubmitInitialHeaders');
+
+function priorityFieldValue(level, incremental) {
+  const urgency = level === 'high' ? 0 : level === 'low' ? 7 : 3;
+  if (urgency === 3 && !incremental) return undefined;
+  return incremental ? `u=${urgency}, i` : `u=${urgency}`;
+}
+
 class Http3Stream {
   #stream;
   #session;
+  #h3handle;
   #headers = undefined;
   #onheaders = undefined;
   #oninfo = undefined;
   #ontrailers = undefined;
   #onwanttrailers = undefined;
+  // Client-side stored priority (the value this side requested). The server
+  // side reads the peer's requested priority from nghttp3 instead.
+  #priority = { __proto__: null, level: 'default', incremental: false };
+  // Have we submitted an initial header block (request/response) to nghttp3?
+  #headersSubmitted = false;
 
   /**
    * @param {QuicStream} stream the underlying QUIC stream
@@ -148,6 +167,7 @@ class Http3Stream {
     }
     this.#stream = stream;
     this.#session = session;
+    this.#h3handle = session[kGetHttp3Handle]();
     const handle = stream[kStreamHandle];
     if (handle !== undefined) {
       handle[kApplicationOwner] = this;
@@ -241,9 +261,57 @@ class Http3Stream {
   // True when the stream's data was received as 0-RTT early data.
   get early() { return this.#stream.early; }
 
-  get priority() { return this.#stream.priority; }
+  get #isServer() {
+    return getQuicSessionState(this.#session.session).isServer;
+  }
 
-  setPriority(options) { return this.#stream.setPriority(options); }
+  get #knownToNgHttp3() {
+    return this.#isServer || this.#headersSubmitted;
+  }
+
+  // The stream's priority. On a client this is the value we requested; on a
+  // server it is the peer's requested priority read from nghttp3.
+  get priority() {
+    const stream = this.#stream;
+    if (stream.destroyed || this.#h3handle === undefined) return null;
+    if (!this.#isServer) {
+      return { level: this.#priority.level,
+               incremental: this.#priority.incremental };
+    }
+    const packed = this.#h3handle.getPriority(stream[kStreamHandle]);
+    if (packed === undefined) return null;
+    const urgency = packed >> 1;
+    const incremental = !!(packed & 1);
+    const level = urgency < 3 ? 'high' : urgency > 3 ? 'low' : 'default';
+    return { level, incremental };
+  }
+
+  setPriority(options = kEmptyObject) {
+    const stream = this.#stream;
+    if (stream.destroyed) return;
+    validateObject(options, 'options');
+    const { level = 'default', incremental = false } = options;
+    validateOneOf(level, 'options.level', ['default', 'low', 'high']);
+    validateBoolean(incremental, 'options.incremental');
+    this.#priority = { __proto__: null, level, incremental };
+
+    // Before the stream is known to nghttp3 (client, pre-submit) the requested
+    // priority is carried in the initial header block by sendHeaders;
+    // afterwards a change is signaled with a PRIORITY_UPDATE frame.
+    if (this.#knownToNgHttp3 && this.#h3handle !== undefined) {
+      const urgency = level === 'high' ? 0 : level === 'low' ? 7 : 3;
+      this.#h3handle.setPriority(
+        stream[kStreamHandle], (urgency << 1) | (incremental ? 1 : 0));
+    }
+  }
+
+  [kSubmitInitialHeaders](headerString, flags) {
+    const stream = this.#stream;
+    if (stream.destroyed || this.#h3handle === undefined) return false;
+    this.#headersSubmitted = true;
+    return this.#h3handle.sendHeaders(
+      stream[kStreamHandle], headerString, flags);
+  }
 
   #updateHeaderInterest() {
     getQuicStreamState(this.#stream).wantsHeaders =
@@ -303,14 +371,29 @@ class Http3Stream {
    */
   sendHeaders(headers, options = kEmptyObject) {
     const stream = this.#stream;
-    if (stream.destroyed) return false;
+    if (stream.destroyed || this.#h3handle === undefined) return false;
     validateObject(headers, 'headers');
     const { terminal = false } = options;
+
+    // A client request carries its requested priority as a priority header
+    // when set - server responses signal via setPriority instead.
+    let toSend = headers;
+    if (!this.#isServer) {
+      const pri = priorityFieldValue(
+        this.#priority.level, this.#priority.incremental);
+      if (pri !== undefined && !ObjectHasOwn(headers, 'priority')) {
+        toSend = { __proto__: null, ...headers, priority: pri };
+      }
+    }
+
     const headerString = buildNgHeaderString(
-      headers, assertValidPseudoHeader, true /* strictSingleValueFields */);
+      toSend, assertValidPseudoHeader, true /* strictSingleValueFields */);
     const flags = terminal ? kHeadersFlagsTerminal : kHeadersFlagsNone;
-    return stream[kStreamHandle].sendHeaders(
-      kHeadersKindInitial, headerString, flags);
+    // The stream now exists in nghttp3; later priority changes use
+    // PRIORITY_UPDATE rather than the header block.
+    this.#headersSubmitted = true;
+    return this.#h3handle.sendHeaders(
+      stream[kStreamHandle], headerString, flags);
   }
 
   /**
@@ -321,11 +404,12 @@ class Http3Stream {
   sendInformationalHeaders(headers) {
     const stream = this.#stream;
     if (stream.destroyed) return false;
+    if (this.#h3handle === undefined) return false;
     validateObject(headers, 'headers');
     const headerString = buildNgHeaderString(
       headers, assertValidPseudoHeader, true);
-    return stream[kStreamHandle].sendHeaders(
-      kHeadersKindHints, headerString, kHeadersFlagsNone);
+    return this.#h3handle.sendInformationalHeaders(
+      stream[kStreamHandle], headerString);
   }
 
   /**
@@ -337,11 +421,12 @@ class Http3Stream {
   sendTrailers(headers) {
     const stream = this.#stream;
     if (stream.destroyed) return false;
+    if (this.#h3handle === undefined) return false;
     validateObject(headers, 'headers');
     const headerString =
       buildNgHeaderString(headers, assertValidPseudoHeaderTrailer);
-    return stream[kStreamHandle].sendHeaders(
-      kHeadersKindTrailing, headerString, kHeadersFlagsNone);
+    return this.#h3handle.sendTrailers(
+      stream[kStreamHandle], headerString);
   }
 
   // Outbound body writer (stream/iter push writer).
@@ -366,6 +451,7 @@ class Http3Stream {
 
 class Http3Session {
   #session;
+  #h3handle;
   #onstream = undefined;
   #ongoaway = undefined;
   #onorigin = undefined;
@@ -390,6 +476,11 @@ class Http3Session {
     // routed back here.
     const handle = session[kSessionHandle];
     if (handle !== undefined) {
+      if (handle[kApplicationOwner] !== undefined) {
+        throw new ERR_INVALID_STATE(
+          'The QUIC session already has an application attached');
+      }
+      this.#h3handle = createHttp3Handle(handle);
       handle[kApplicationOwner] = this;
     }
     // Claim the session's incoming streams. This takes the single
@@ -412,6 +503,9 @@ class Http3Session {
     if (onsettings !== undefined) this.onsettings = onsettings;
   }
 
+  // Internal: hands the per-session HTTP/3 handle to this session's streams.
+  [kGetHttp3Handle]() { return this.#h3handle; }
+
   /**
    * The peer initiated a graceful shutdown of the session (HTTP/3
    * GOAWAY). The session stops allowing new streams.
@@ -579,14 +673,22 @@ class Http3Session {
       onwanttrailers,
       onreset,
       onerror,
+      priority,
+      incremental,
       ...quicOptions
     } = options;
 
     let headerString;
     if (headers !== undefined) {
+      let toSend = headers;
+      const pri = priorityFieldValue(priority ?? 'default', incremental ?? false);
+      if (pri !== undefined && !ObjectHasOwn(headers, 'priority')) {
+        toSend = { __proto__: null, ...headers, priority: pri };
+      }
       headerString = buildNgHeaderString(
-        headers, assertValidPseudoHeader, true /* strictSingleValueFields */);
+        toSend, assertValidPseudoHeader, true /* strictSingleValueFields */);
     }
+
     const stream = await this.#session.createBidirectionalStream(quicOptions);
     const wrapped = new Http3Stream(stream, this, {
       __proto__: null,
@@ -597,6 +699,11 @@ class Http3Session {
       onreset,
       onerror,
     });
+
+    if (priority !== undefined || incremental !== undefined) {
+      wrapped.setPriority({ __proto__: null, level: priority, incremental });
+    }
+
     // Submit the request headers only after the callbacks above are
     // attached. Nothing for this stream can hit the wire before the
     // headers are submitted: the HTTP/3 application only learns about the
@@ -606,8 +713,7 @@ class Http3Session {
     if (headerString !== undefined) {
       const flags = quicOptions.body === undefined ?
         kHeadersFlagsTerminal : kHeadersFlagsNone;
-      if (!stream[kStreamHandle].sendHeaders(
-        kHeadersKindInitial, headerString, flags)) {
+      if (!wrapped[kSubmitInitialHeaders](headerString, flags)) {
         wrapped.destroy();
         throw new ERR_QUIC_OPEN_STREAM_FAILED();
       }

lib/internal/quic/quic.js

@@ -281,8 +281,6 @@ const endpointRegistry = new SafeSet();
  *   FileHandle|AsyncIterable|Iterable|Promise|null} [body] The outbound
  *   body source. See the public docs for `stream.setBody()` for details
  *   on supported types. When omitted, the stream is closed immediately.
- * @property {'high'|'default'|'low'} [priority] The priority level of the stream.
- * @property {boolean} [incremental] Whether to interleave data with same-priority streams.
  * @property {number} [highWaterMark] The high water mark for write
  *   backpressure, in bytes. **Default:** `65536`.
  */
@@ -472,12 +470,6 @@ const endpointRegistry = new SafeSet();
  *   frames do not themselves carry a reason field over the wire.
  */
 
-/**
- * @typedef {object} StreamPriority
- * @property {'default' | 'low' | 'high'} level The priority level of the stream.
- * @property {boolean} incremental Whether to interleave data with same-priority streams.
- */
-
 /**
  * @typedef {object} QuicSessionPath
  * @property {SocketAddress} local The local address for this path
@@ -2079,45 +2071,6 @@ class QuicStream {
     this.#handle.resetStream(BigInt(code));
   }
 
-  /**
-   * The priority of the stream. If the stream is destroyed or if
-   * the session does not support priority, `null` will be
-   * returned.
-   * @type {StreamPriority | null}
-   */
-  get priority() {
-    assertIsQuicStream(this);
-    if (this.destroyed ||
-        !getQuicSessionState(this.#inner.session).isPrioritySupported) return null;
-    const packed = this.#handle.getPriority();
-    const urgency = packed >> 1;
-    const incremental = !!(packed & 1);
-    const level = urgency < 3 ? 'high' : urgency > 3 ? 'low' : 'default';
-    return { level, incremental };
-  }
-
-  /**
-   * Sets the priority of the stream.
-   * @param {StreamPriority} [options]
-   */
-  setPriority(options = kEmptyObject) {
-    assertIsQuicStream(this);
-    if (this.destroyed) return;
-    if (!getQuicSessionState(this.#inner.session).isPrioritySupported) {
-      throw new ERR_INVALID_STATE(
-        'The session does not support stream priority');
-    }
-    validateObject(options, 'options');
-    const {
-      level = 'default',
-      incremental = false,
-    } = options;
-    validateOneOf(level, 'options.level', ['default', 'low', 'high']);
-    validateBoolean(incremental, 'options.incremental');
-    const urgency = level === 'high' ? 0 : level === 'low' ? 7 : 3;
-    this.#handle.setPriority((urgency << 1) | (incremental ? 1 : 0));
-  }
-
   [kFinishClose](error) {
     const inner = this.#inner;
     inner.pendingClose ??= PromiseWithResolvers();
@@ -2761,26 +2714,16 @@ class QuicSession {
     validateObject(options, 'options');
     const {
       body,
-      priority = 'default',
-      incremental = false,
       highWaterMark = kDefaultHighWaterMark,
     } = options;
 
-    validateOneOf(priority, 'options.priority', ['default', 'low', 'high']);
-    validateBoolean(incremental, 'options.incremental');
-
     const validatedBody = validateBody(body);
 
     const handle = this.#handle.openStream(direction, validatedBody);
     if (handle === undefined) {
       throw new ERR_QUIC_OPEN_STREAM_FAILED();
     }
 
-    if (inner.state.isPrioritySupported) {
-      const urgency = priority === 'high' ? 0 : priority === 'low' ? 7 : 3;
-      handle.setPriority((urgency << 1) | (incremental ? 1 : 0));
-    }
-
     const stream = new QuicStream(
       kPrivateConstructor, handle, this, direction, true /* isLocal */);
     inner.streams.add(stream);