quic: add support for HTTP/3 datagrams

Signed-off-by: Tim Perry <pimterry@gmail.com>

Author: pimterry

doc/api/quic.md

@@ -358,6 +358,29 @@ delivered out of order. The [`session.ondatagramstatus`][] callback reports
 whether each sent datagram was `'acknowledged'`, `'lost'`, or `'abandoned'`
 (never sent on the wire).
 
+#### HTTP/3 datagrams
+
+On HTTP/3 sessions, datagrams are associated with an individual request stream
+rather than with the session as a whole. Send and receive them per-stream with
+[`stream.sendDatagram()`][] and [`stream.ondatagram`][], passing and receiving
+your application payload directly; the protocol framing that binds a datagram to
+its stream is handled internally and never visible to your code. Both peers must
+set [`application.enableDatagrams`][] to `true`.
+
+The session-level datagram API does not apply to HTTP/3 sessions:
+[`session.sendDatagram()`][] throws `ERR_INVALID_STATE` and
+[`session.ondatagram`][] never fires. (Only non-HTTP/3 ALPNs use the raw,
+session-level datagrams described above.)
+
+Because datagrams are unordered with respect to stream data, a datagram is only
+delivered while its associated stream still exists on the receiver; one that
+arrives for an unknown or already-closed stream is dropped.
+
+Under the hood, each HTTP/3 datagram ([RFC 9297][]) is carried in a QUIC
+`DATAGRAM` frame prefixed with a _Quarter Stream ID_ (the request stream's id
+divided by four) that binds it to the stream. This prefix is added and stripped
+automatically; most applications never need to think about it.
+
 ### 0-RTT early data and session resumption
 
 QUIC supports 0-RTT early data, allowing a client that has previously connected
@@ -1114,7 +1137,12 @@ added: v23.8.0
 
 * Type: {quic.OnDatagramCallback}
 
-The callback to invoke when a new datagram is received from a remote peer. Read/write.
+The callback to invoke when a new raw datagram is received from a remote peer.
+Read/write.
+
+This is only used for non-HTTP/3 sessions. On HTTP/3 sessions datagrams are
+bound to a request stream and delivered via [`stream.ondatagram`][] instead;
+this callback never fires. See [Datagrams][].
 
 ### `session.ondatagramstatus`
 
@@ -1384,9 +1412,13 @@ added: v23.8.0
   **Default:** `'utf8'`.
 * Returns: {Promise} for a {bigint} datagram ID.
 
-Sends an unreliable datagram to the remote peer, returning a promise for
+Sends an unreliable raw datagram to the remote peer, returning a promise for
 the datagram ID.
 
+This method is for non-HTTP/3 sessions only. On HTTP/3 sessions datagrams are
+per-stream, so this method throws `ERR_INVALID_STATE`; use
+[`stream.sendDatagram()`][] instead. See [Datagrams][].
+
 If `datagram` is a string, it will be encoded using the specified `encoding`.
 
 If `datagram` is an `ArrayBufferView`, the bytes are copied into an
@@ -1403,12 +1435,6 @@ inherently unreliable).
 If the datagram payload is zero-length (empty string after encoding, detached
 buffer, or zero-length view), `0n` is returned and no datagram is sent.
 
-For HTTP/3 sessions, the peer must advertise `SETTINGS_H3_DATAGRAM=1`
-(via `application: { enableDatagrams: true }`) for datagrams to be sent.
-If the peer's setting is `0`, `sendDatagram()` returns `0n` (per RFC 9297
-§3, an endpoint MUST NOT send HTTP Datagrams unless the peer indicated
-support).
-
 Datagrams cannot be fragmented — each must fit within a single QUIC packet.
 The maximum datagram size is determined by the peer's
 `maxDatagramFrameSize` transport parameter (which the peer advertises during
@@ -2012,6 +2038,20 @@ continue using the still-active direction on a bidirectional stream),
 abort the other direction with [`writer.fail()`][], or tear down the
 whole stream with [`stream.destroy()`][]. Read/write.
 
+### `stream.ondatagram`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {quic.OnStreamDatagramCallback}
+
+The callback invoked when a datagram associated with this stream is received.
+It receives the datagram payload as a `Uint8Array` and a `boolean` indicating
+whether it arrived as 0-RTT early data. Read/write.
+
+Only applies to HTTP/3 sessions. See [Datagrams][].
+
 ### `stream.headers`
 
 <!-- YAML
@@ -2146,6 +2186,29 @@ the [`stream.onwanttrailers`][] callback, or set ahead of time via
 [`stream.pendingTrailers`][]. Throws `ERR_INVALID_STATE` if the session
 does not support headers.
 
+### `stream.sendDatagram(datagram[, encoding])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* `datagram` {string|ArrayBufferView}
+* `encoding` {string} The encoding to use if `datagram` is a string.
+  **Default:** `'utf8'`.
+* Returns: {bigint} The datagram id, or `0n` if the datagram was not sent.
+
+Sends an unreliable datagram associated with this stream to the peer, where it
+is delivered to the corresponding stream's [`stream.ondatagram`][] callback. The
+payload is sent as-is; the framing that binds it to the stream is handled
+internally.
+
+Delivery is best-effort: datagrams may be lost, reordered, or dropped. `0n` is
+returned (and the payload silently discarded) if the datagram cannot be sent.
+The delivery status of a sent datagram is reported via the session's
+[`session.ondatagramstatus`][] callback.
+
+Only applies to HTTP/3 sessions. See [Datagrams][].
+
 ### `stream.priority`
 
 <!-- YAML
@@ -3745,6 +3808,16 @@ added: v23.8.0
 * `this` {quic.QuicStream}
 * `error` {any}
 
+### Callback: `OnStreamDatagramCallback`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* `this` {quic.QuicStream}
+* `datagram` {Uint8Array} The datagram payload.
+* `early` {boolean} `true` if the datagram arrived as 0-RTT early data.
+
 ### Callback: `OnHeadersCallback`
 
 <!-- YAML
@@ -4424,6 +4497,7 @@ throughput issues caused by flow control.
 
 [Aborting a stream]: #aborting-a-stream
 [Callback error handling]: #callback-error-handling
+[Datagrams]: #datagrams
 [JSON-SEQ]: https://www.rfc-editor.org/rfc/rfc7464
 [NSS Key Log Format]: https://udn.realityripple.com/docs/Mozilla/Projects/NSS/Key_Log_Format
 [Permission Model]: permissions.md#permission-model
@@ -4504,10 +4578,12 @@ throughput issues caused by flow control.
 [`sessionOptions.token`]: #sessionoptionstoken-client-only
 [`stream.destroy()`]: #streamdestroyerror-options
 [`stream.headers`]: #streamheaders
+[`stream.ondatagram`]: #streamondatagram
 [`stream.onerror`]: #streamonerror
 [`stream.onwanttrailers`]: #streamonwanttrailers
 [`stream.pendingTrailers`]: #streampendingtrailers
 [`stream.priority`]: #streampriority
+[`stream.sendDatagram()`]: #streamsenddatagramdatagram-encoding
 [`stream.sendHeaders()`]: #streamsendheadersheaders-options
 [`stream.sendInformationalHeaders()`]: #streamsendinformationalheadersheaders
 [`stream.sendTrailers()`]: #streamsendtrailersheaders

lib/internal/quic/quic.js

@@ -276,6 +276,8 @@ const {
 
 const kNilDatagramId = 0n;
 
+const kApplicationTypeHttp3 = 2;
+
 // Module-level registry of all live QuicEndpoint instances. Used by
 // connect() and listen() to find existing endpoints for reuse instead
 // of creating a new one per session.
@@ -945,6 +947,16 @@ setCallbacks({
     this[kOwner][kBlocked]();
   },
 
+  /**
+   * Called when an HTTP/3 datagram (RFC 9297) is received for this stream.
+   * @param {Uint8Array} uint8Array The datagram payload
+   * @param {boolean} early True if received as 0-RTT early data.
+   */
+  onStreamDatagram(uint8Array, early) {
+    debug('stream datagram callback', this[kOwner]);
+    this[kOwner][kDatagram](uint8Array, early);
+  },
+
   onStreamDrain() {
     // Called when the stream's outbound buffer has capacity for more data.
     debug('stream drain callback', this[kOwner]);
@@ -1571,6 +1583,7 @@ class QuicStream {
     onerror: undefined,
     onblocked: undefined,
     onreset: undefined,
+    ondatagram: undefined,
     onheaders: undefined,
     ontrailers: undefined,
     oninfo: undefined,
@@ -1776,6 +1789,29 @@ class QuicStream {
     }
   }
 
+  /**
+   * The callback invoked when an HTTP/3 datagram (RFC 9297) associated with
+   * this stream is received. Read/write.
+   * @type {OnStreamDatagramCallback}
+   */
+  get ondatagram() {
+    assertIsQuicStream(this);
+    return this.#inner.ondatagram;
+  }
+
+  set ondatagram(fn) {
+    assertIsQuicStream(this);
+    const inner = this.#inner;
+    if (fn === undefined) {
+      inner.ondatagram = undefined;
+      inner.state.wantsDatagram = false;
+    } else {
+      validateFunction(fn, 'ondatagram');
+      inner.ondatagram = FunctionPrototypeBind(fn, this);
+      inner.state.wantsDatagram = true;
+    }
+  }
+
   /** @type {OnHeadersCallback} */
   get onheaders() {
     assertIsQuicStream(this);
@@ -2429,6 +2465,48 @@ class QuicStream {
     this.#handle.resetStream(BigInt(code));
   }
 
+  /**
+   * Sends an HTTP/3 datagram (RFC 9297) associated with this stream. The
+   * payload is framed with the stream's Quarter Stream ID by the C++ layer
+   * before being queued for transmission.
+   *
+   * Datagrams are unreliable: they may be lost, reordered, or dropped if too
+   * large or if datagrams were not negotiated. In any such case `0n` is
+   * returned and the payload is silently discarded. Otherwise the underlying
+   * QUIC datagram id is returned.
+   * @param {ArrayBufferView|string} datagram The datagram payload.
+   * @param {string} [encoding] The encoding to use if datagram is a string.
+   * @returns {bigint} The datagram id, or `0n` if it was not sent.
+   */
+  sendDatagram(datagram, encoding = 'utf8') {
+    assertIsQuicStream(this);
+    if (this.destroyed || this.pending) return kNilDatagramId;
+
+    // Datagrams are gated by the session's negotiated max datagram size.
+    // A value of 0 means datagrams are disabled (transport or, for HTTP/3,
+    // the peer's SETTINGS_H3_DATAGRAM). The C++ layer enforces the exact
+    // limit including the Quarter Stream ID prefix.
+    if (getQuicSessionState(this.#inner.session).maxDatagramSize === 0) {
+      return kNilDatagramId;
+    }
+
+    if (typeof datagram === 'string') {
+      datagram = new Uint8Array(Buffer.from(datagram, encoding));
+    } else if (!isArrayBufferView(datagram)) {
+      throw new ERR_INVALID_ARG_TYPE('datagram',
+                                     ['ArrayBufferView', 'string'],
+                                     datagram);
+    }
+
+    const length = isDataView(datagram) ?
+      DataViewPrototypeGetByteLength(datagram) :
+      TypedArrayPrototypeGetByteLength(datagram);
+
+    const id = this.#handle.sendDatagram(datagram);
+    debug(`stream datagram ${id} sent with ${length} bytes`);
+    return id;
+  }
+
   /**
    * The priority of the stream. If the stream is destroyed or if
    * the session does not support priority, `null` will be
@@ -2538,6 +2616,7 @@ class QuicStream {
     inner.pendingClose.resolve = undefined;
     inner.onblocked = undefined;
     inner.onreset = undefined;
+    inner.ondatagram = undefined;
     inner.onheaders = undefined;
     inner.onerror = undefined;
     inner.ontrailers = undefined;
@@ -2591,6 +2670,14 @@ class QuicStream {
     safeCallbackInvoke(inner.onreset, this, error);
   }
 
+  [kDatagram](u8, early) {
+    const inner = this.#inner;
+    assert(typeof inner.ondatagram === 'function',
+           'Unexpected stream datagram event');
+    if (inner.session === undefined) return;
+    safeCallbackInvoke(inner.ondatagram, this, u8, early);
+  }
+
   [kHeaders](headers, kind) {
     const block = parseHeaderPairs(headers);
     const kindName = kHeadersKindName[kind] ?? kind;
@@ -3359,6 +3446,13 @@ class QuicSession {
       throw new ERR_INVALID_STATE('Session is closed');
     }
 
+    // HTTP/3 sessions must use HTTP/3 datagrams
+    if (this.#inner.state.applicationType === kApplicationTypeHttp3) {
+      throw new ERR_INVALID_STATE(
+        'Raw datagrams are not supported on HTTP/3 sessions; ' +
+        'use stream.sendDatagram() instead');
+    }
+
     const maxDatagramSize = this.#inner.state.maxDatagramSize;
 
     // The peer max datagram size is either unknown or they have explicitly

lib/internal/quic/state.js

@@ -102,6 +102,7 @@ const {
   IDX_STATE_STREAM_WANTS_HEADERS,
   IDX_STATE_STREAM_WANTS_RESET,
   IDX_STATE_STREAM_WANTS_TRAILERS,
+  IDX_STATE_STREAM_WANTS_DATAGRAM,
   IDX_STATE_STREAM_RECEIVED_EARLY_DATA,
   IDX_STATE_STREAM_WRITE_DESIRED_SIZE,
   IDX_STATE_STREAM_HIGH_WATER_MARK,
@@ -143,6 +144,7 @@ assert(IDX_STATE_STREAM_WANTS_BLOCK !== undefined);
 assert(IDX_STATE_STREAM_WANTS_HEADERS !== undefined);
 assert(IDX_STATE_STREAM_WANTS_RESET !== undefined);
 assert(IDX_STATE_STREAM_WANTS_TRAILERS !== undefined);
+assert(IDX_STATE_STREAM_WANTS_DATAGRAM !== undefined);
 assert(IDX_STATE_STREAM_WRITE_DESIRED_SIZE !== undefined);
 assert(IDX_STATE_STREAM_RESET_CODE !== undefined);
 
@@ -798,6 +800,20 @@ class QuicStreamState {
     DataViewPrototypeSetUint8(handle, this.#offset + IDX_STATE_STREAM_WANTS_BLOCK, val ? 1 : 0);
   }
 
+  /** @type {boolean} */
+  get wantsDatagram() {
+    const handle = this.#handle;
+    if (handle === undefined) return undefined;
+    return DataViewPrototypeGetUint8(handle, this.#offset + IDX_STATE_STREAM_WANTS_DATAGRAM) !== 0;
+  }
+
+  /** @type {boolean} */
+  set wantsDatagram(val) {
+    const handle = this.#handle;
+    if (handle === undefined) return;
+    DataViewPrototypeSetUint8(handle, this.#offset + IDX_STATE_STREAM_WANTS_DATAGRAM, val ? 1 : 0);
+  }
+
   /** @type {boolean} */
   get wantsHeaders() {
     const handle = this.#handle;
@@ -905,6 +921,7 @@ class QuicStreamState {
       wantsReset,
       wantsHeaders,
       wantsTrailers,
+      wantsDatagram,
       early,
       resetCode,
       writeDesiredSize,
@@ -925,6 +942,7 @@ class QuicStreamState {
       wantsReset,
       wantsHeaders,
       wantsTrailers,
+      wantsDatagram,
       early,
       resetCode,
       writeDesiredSize,
@@ -961,6 +979,7 @@ class QuicStreamState {
       wantsReset,
       wantsHeaders,
       wantsTrailers,
+      wantsDatagram,
       early,
       resetCode,
       writeDesiredSize,
@@ -981,6 +1000,7 @@ class QuicStreamState {
       wantsReset,
       wantsHeaders,
       wantsTrailers,
+      wantsDatagram,
       early,
       resetCode,
       writeDesiredSize,

src/quic/application.cc

@@ -203,6 +203,15 @@ bool Session::Application::AcknowledgeStreamData(stream_id id, size_t datalen) {
   return true;
 }
 
+void Session::Application::ReceiveDatagram(
+    const uint8_t* data,
+    size_t datalen,
+    const Session::DatagramReceivedFlags& flags) {
+  // Raw QUIC applications have no datagram framing: deliver the payload
+  // verbatim to the session-level ondatagram handler.
+  session().DeliverRawDatagram(data, datalen, flags);
+}
+
 void Session::Application::CollectSessionTicketAppData(
     SessionTicket::AppData* app_data) const {
   // By default, write just the application type byte.