quic: complete datagram support

Author: jasnell

doc/api/quic.md

@@ -548,18 +548,56 @@ added: v23.8.0
 
 The local and remote socket addresses associated with the session. Read only.
 
-### `session.sendDatagram(datagram)`
+### `session.sendDatagram(datagram[, encoding])`
 
 <!-- YAML
 added: v23.8.0
 -->
 
-* `datagram` {string|ArrayBufferView}
-* Returns: {bigint}
+* `datagram` {string|ArrayBufferView|Promise}
+* `encoding` {string} The encoding to use if `datagram` is a string.
+  **Default:** `'utf8'`.
+* Returns: {Promise} for a {bigint} datagram ID.
 
-Sends an unreliable datagram to the remote peer, returning the datagram ID.
-If the datagram payload is specified as an `ArrayBufferView`, then ownership of
-that view will be transferred to the underlying stream.
+Sends an unreliable datagram to the remote peer, returning a promise for
+the datagram ID.
+
+If `datagram` is a string, it will be encoded using the specified `encoding`.
+
+If `datagram` is an `ArrayBufferView`, the underlying `ArrayBuffer` will be
+transferred if possible (taking ownership to prevent mutation after send).
+If the buffer is not transferable (e.g., a `SharedArrayBuffer` or a view
+over a subset of a larger buffer such as a pooled `Buffer`), the data will
+be copied instead.
+
+If `datagram` is a `Promise`, it will be awaited before sending. If the
+session closes while awaiting, `0n` is returned silently (datagrams are
+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.
+
+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
+the handshake). If the peer sets this to `0`, datagrams are not supported
+and `0n` will be returned. If the datagram exceeds the peer's limit, it
+will be silently dropped and `0n` returned. The local
+`maxDatagramFrameSize` transport parameter (default: `1200` bytes) controls
+what this endpoint advertises to the peer as its own maximum.
+
+### `session.maxDatagramSize`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {bigint}
+
+The maximum datagram payload size in bytes that the peer will accept,
+as advertised in the peer's `maxDatagramFrameSize` transport parameter.
+Returns `0n` if the peer does not support datagrams or if the handshake
+has not yet completed. Datagrams larger than this value will not be sent.
 
 ### `session.stats`
 
@@ -1679,6 +1717,13 @@ added: v23.8.0
 -->
 
 * Type: {bigint|number}
+* **Default:** `1200`
+
+The maximum size in bytes of a DATAGRAM frame payload that this endpoint
+is willing to receive. Set to `0` to disable datagram support. The peer
+will not send datagrams larger than this value. The actual maximum size of
+a datagram that can be _sent_ is determined by the peer's
+`maxDatagramFrameSize`, not this endpoint's value.
 
 ## Callbacks
 

lib/internal/quic/quic.js

@@ -5,15 +5,23 @@
 /* c8 ignore start */
 
 const {
+  ArrayBufferPrototypeGetByteLength,
   ArrayBufferPrototypeTransfer,
   ArrayIsArray,
   ArrayPrototypePush,
   BigInt,
+  DataViewPrototypeGetBuffer,
+  DataViewPrototypeGetByteLength,
+  DataViewPrototypeGetByteOffset,
   ObjectDefineProperties,
   ObjectKeys,
   PromiseWithResolvers,
   SafeSet,
   SymbolAsyncDispose,
+  TypedArrayPrototypeGetBuffer,
+  TypedArrayPrototypeGetByteLength,
+  TypedArrayPrototypeGetByteOffset,
+  TypedArrayPrototypeSlice,
   Uint8Array,
 } = primordials;
 
@@ -66,6 +74,8 @@ const {
 const {
   isArrayBuffer,
   isArrayBufferView,
+  isDataView,
+  isPromise,
   isSharedArrayBuffer,
 } = require('util/types');
 
@@ -190,6 +200,8 @@ const onSessionVersionNegotiationChannel = dc.channel('quic.session.version.nego
 const onSessionOriginChannel = dc.channel('quic.session.receive.origin');
 const onSessionHandshakeChannel = dc.channel('quic.session.handshake');
 
+const kNilDatagramId = 0n;
+
 /**
  * @typedef {import('../socketaddress.js').SocketAddress} SocketAddress
  * @typedef {import('../crypto/keys.js').KeyObject} KeyObject
@@ -427,7 +439,7 @@ setCallbacks({
    * @param {boolean} early
    */
   onSessionDatagram(uint8Array, early) {
-    debug('session datagram callback', uint8Array.byteLength, early);
+    debug('session datagram callback', TypedArrayPrototypeGetByteLength(uint8Array), early);
     this[kOwner][kDatagram](uint8Array, early);
   },
 
@@ -1100,6 +1112,8 @@ class QuicSession {
   #onstream = undefined;
   /** @type {OnDatagramCallback|undefined} */
   #ondatagram = undefined;
+  /** @type {OnDatagramStatusCallback|undefined} */
+  #ondatagramstatus = undefined;
   /** @type {object|undefined} */
   #sessionticket = undefined;
   /** @type {object|undefined} */
@@ -1200,6 +1214,38 @@ class QuicSession {
     }
   }
 
+  /**
+   * The ondatagramstatus callback is called when the status of a sent datagram
+   * is received. This is best-effort only.
+   * @type {OnDatagramStatusCallback}
+   */
+  get ondatagramstatus() {
+    QuicSession.#assertIsQuicSession(this);
+    return this.#ondatagramstatus;
+  }
+
+  set ondatagramstatus(fn) {
+    QuicSession.#assertIsQuicSession(this);
+    if (fn === undefined) {
+      this.#ondatagramstatus = undefined;
+      this.#state.hasDatagramStatusListener = false;
+    } else {
+      validateFunction(fn, 'ondatagramstatus');
+      this.#ondatagramstatus = fn.bind(this);
+      this.#state.hasDatagramStatusListener = true;
+    }
+  }
+
+  /**
+   * The maximum datagram size the peer will accept, or 0 if datagrams
+   * are not supported or the handshake has not yet completed.
+   * @type {bigint}
+   */
+  get maxDatagramSize() {
+    QuicSession.#assertIsQuicSession(this);
+    return this.#state.maxDatagramSize;
+  }
+
   /**
    * The statistics collected for this session.
    * @type {QuicSessionStats}
@@ -1312,42 +1358,93 @@ class QuicSession {
    * of the sent datagram will be reported via the datagram-status event if
    * possible.
    *
-   * If a string is given it will be encoded as UTF-8.
+   * If a string is given it will be encoded using the specified encoding.
    *
-   * If an ArrayBufferView is given, the view will be copied.
-   * @param {ArrayBufferView|string} datagram The datagram payload
-   * @returns {Promise<void>}
+   * If an ArrayBufferView is given, the underlying ArrayBuffer will be
+   * transferred if possible, otherwise the data will be copied.
+   *
+   * If a Promise is given, it will be awaited before sending. If the
+   * session closes while awaiting, 0n is returned silently.
+   * @param {ArrayBufferView|string|Promise} datagram The datagram payload
+   * @param {string} [encoding] The encoding to use if datagram is a string
+   * @returns {Promise<bigint>} The datagram ID
    */
-  async sendDatagram(datagram) {
+  async sendDatagram(datagram, encoding = 'utf8') {
     QuicSession.#assertIsQuicSession(this);
     if (this.#isClosedOrClosing) {
       throw new ERR_INVALID_STATE('Session is closed');
     }
+    let offset, length, buffer;
+
+    const maxDatagramSize = this.#state.maxDatagramSize;
+
+    // The peer max datagram size is either unknown or they have explicitly
+    // indicated that they do not support datagrams by setting it to 0. In
+    // either case, we do not send the datagram.
+    if (maxDatagramSize === 0n) return kNilDatagramId;
+
+    if (isPromise(datagram)) {
+      datagram = await datagram;
+      // Session may have closed while awaiting. Since datagrams are
+      // inherently unreliable, silently return rather than throwing.
+      if (this.#isClosedOrClosing) return kNilDatagramId;
+    }
+
     if (typeof datagram === 'string') {
-      datagram = Buffer.from(datagram, 'utf8');
+      datagram = Buffer.from(datagram, encoding);
+      length = TypedArrayPrototypeGetByteLength(datagram);
+      if (length === 0) return kNilDatagramId;
     } else {
       if (!isArrayBufferView(datagram)) {
         throw new ERR_INVALID_ARG_TYPE('datagram',
                                        ['ArrayBufferView', 'string'],
                                        datagram);
       }
-      const length = datagram.byteLength;
-      const offset = datagram.byteOffset;
-      datagram = new Uint8Array(ArrayBufferPrototypeTransfer(datagram.buffer),
-                                length, offset);
+      if (isDataView(datagram)) {
+        offset = DataViewPrototypeGetByteOffset(datagram);
+        length = DataViewPrototypeGetByteLength(datagram);
+        buffer = DataViewPrototypeGetBuffer(datagram);
+      } else {
+        offset = TypedArrayPrototypeGetByteOffset(datagram);
+        length = TypedArrayPrototypeGetByteLength(datagram);
+        buffer = TypedArrayPrototypeGetBuffer(datagram);
+      }
+
+      // If the view has zero length (e.g. detached buffer), there's
+      // nothing to send.
+      if (length === 0) return kNilDatagramId;
+
+      if (isSharedArrayBuffer(buffer) ||
+          offset !== 0 ||
+          length !== ArrayBufferPrototypeGetByteLength(buffer)) {
+        // Copy if the buffer is not transferable (SharedArrayBuffer)
+        // or if the view is over a subset of the buffer (e.g. a
+        // Node.js Buffer from the pool).
+        datagram = TypedArrayPrototypeSlice(
+          new Uint8Array(buffer), offset, offset + length);
+      } else {
+        datagram = new Uint8Array(
+          ArrayBufferPrototypeTransfer(buffer), offset, length);
+      }
     }
 
-    debug(`sending datagram with ${datagram.byteLength} bytes`);
+    // The peer max datagram size is less than the datagram we want to send,
+    // so... don't send it.
+    if (length > maxDatagramSize) return kNilDatagramId;
 
     const id = this.#handle.sendDatagram(datagram);
 
-    if (onSessionSendDatagramChannel.hasSubscribers) {
+    if (id !== kNilDatagramId && onSessionSendDatagramChannel.hasSubscribers) {
       onSessionSendDatagramChannel.publish({
+        __proto__: null,
         id,
-        length: datagram.byteLength,
+        length,
         session: this,
       });
     }
+
+    debug(`datagram ${id} sent with ${length} bytes`);
+    return id;
   }
 
   /**
@@ -1472,6 +1569,7 @@ class QuicSession {
 
     this.#onstream = undefined;
     this.#ondatagram = undefined;
+    this.#ondatagramstatus = undefined;
     this.#sessionticket = undefined;
     this.#token = undefined;
 
@@ -1531,19 +1629,20 @@ class QuicSession {
   }
 
   /**
-   * @param {Uint8Array} u8
-   * @param {boolean} early
+   * @param {Uint8Array} u8 The datagram payload
+   * @param {boolean} early A boolean indicating whether this datagram was received before the handshake completed
    */
   [kDatagram](u8, early) {
-    // The datagram event should only be called if the session was created with
+    // The datagram event should only be called if the session has
     // an ondatagram callback. The callback should always exist here.
-    assert(this.#ondatagram, 'Unexpected datagram event');
+    assert(typeof this.#ondatagram === 'function', 'Unexpected datagram event');
     if (this.destroyed) return;
-    const length = u8.byteLength;
+    const length = TypedArrayPrototypeGetByteLength(u8);
     this.#ondatagram(u8, early);
 
     if (onSessionReceiveDatagramChannel.hasSubscribers) {
       onSessionReceiveDatagramChannel.publish({
+        __proto__: null,
         length,
         early,
         session: this,
@@ -1556,9 +1655,15 @@ class QuicSession {
    * @param {'lost'|'acknowledged'} status
    */
   [kDatagramStatus](id, status) {
+    // The datagram status event should only be called if the session has
+    // an ondatagramstatus callback. The callback should always exist here.
+    assert(typeof this.#ondatagramstatus === 'function', 'Unexpected datagram status event');
     if (this.destroyed) return;
+    this.#ondatagramstatus(id, status);
+
     if (onSessionReceiveDatagramStatusChannel.hasSubscribers) {
       onSessionReceiveDatagramStatusChannel.publish({
+        __proto__: null,
         id,
         status,
         session: this,