fixup! quic: add support for HTTP/3 datagrams

Queue datagrams for pending streams

Author: pimterry

lib/internal/quic/quic.js

@@ -2474,13 +2474,17 @@ class QuicStream {
    * 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.
+   *
+   * If the stream is still pending (no id yet) the datagram is buffered and
+   * sending attempted once the stream opens. `0n` is still returned if it
+   * cannot even be buffered.
    * @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.
+   * @returns {bigint} The datagram id, or `0n` if it was not queued.
    */
   sendDatagram(datagram, encoding = 'utf8') {
     assertIsQuicStream(this);
-    if (this.destroyed || this.pending) return kNilDatagramId;
+    if (this.destroyed) 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,

src/quic/application.h

@@ -224,10 +224,12 @@ class Session::Application : public MemoryRetainer {
                                size_t datalen,
                                const Session::DatagramReceivedFlags& flags);
 
-  // Sends a datagram associated with the given stream. Applications that do
-  // not support stream-bound datagrams return 0. Returns the underlying QUIC
-  // datagram id, or 0 if the datagram was not queued.
-  virtual datagram_id SendDatagram(Stream* stream, Store&& payload) {
+  // Sends a datagram associated with the given stream, using the caller's
+  // reserved id. Applications that do not support stream-bound datagrams
+  // return 0. Returns the id on success, or 0 if the datagram was not queued.
+  virtual datagram_id SendDatagram(Stream* stream,
+                                   Store&& payload,
+                                   datagram_id id) {
     return 0;
   }
 

src/quic/http3.cc

@@ -405,15 +405,17 @@ class Http3ApplicationImpl final : public Session::Application {
     nghttp3_conn_unblock_stream(*this, stream->id());
   }
 
-  datagram_id SendDatagram(Stream* stream, Store&& payload) override {
+  datagram_id SendDatagram(Stream* stream,
+                           Store&& payload,
+                           datagram_id id) override {
     if (!SupportsDatagrams()) return 0;
 
     // HTTP/3 datagrams can only ever be associated with client-initiated
     // bidi streams (requests). Those ids are always divisible by 4, and
     // sent /4 in the framing itself.
-    stream_id id = stream->id();
-    if (id < 0 || id % 4 != 0) return 0;
-    uint64_t qsid = static_cast<uint64_t>(id) / 4;
+    stream_id sid = stream->id();
+    if (sid < 0 || sid % 4 != 0) return 0;
+    uint64_t qsid = static_cast<uint64_t>(sid) / 4;
 
     size_t prefix_len = nghttp3_put_uvarintlen(qsid);
     uv_buf_t buf = payload;
@@ -433,7 +435,7 @@ class Http3ApplicationImpl final : public Session::Application {
     nghttp3_put_uvarint(dest, qsid);
     // N.b. HTTP/3 Datagram payloads can be empty (RFC 9297).
     if (buf.len > 0) memcpy(dest + prefix_len, buf.base, buf.len);
-    return session().SendDatagram(Store(std::move(backing), total));
+    return session().SendDatagram(Store(std::move(backing), total), id);
   }
 
   void ReceiveDatagram(const uint8_t* data,

src/quic/session.cc

@@ -1170,8 +1170,9 @@ struct Session::Impl final : public MemoryRetainer {
       return;
     }
 
-    args.GetReturnValue().Set(
-        BigInt::New(env->isolate(), session->SendDatagram(std::move(store))));
+    datagram_id id = session->ReserveDatagramId();
+    args.GetReturnValue().Set(BigInt::New(
+        env->isolate(), session->SendDatagram(std::move(store), id)));
   }
 
   JS_METHOD(LocalTransportParams) {
@@ -2542,7 +2543,12 @@ void Session::Send(Packet::Ptr packet, const PathStorage& path) {
   Send(std::move(packet));
 }
 
-datagram_id Session::SendDatagram(Store&& data) {
+datagram_id Session::ReserveDatagramId() {
+  DCHECK(!is_destroyed());
+  return ++impl_->state()->last_datagram_id;
+}
+
+datagram_id Session::SendDatagram(Store&& data, datagram_id id) {
   DCHECK(!is_destroyed());
 
   // Sending a datagram is best effort. If we cannot send it for any reason,
@@ -2577,8 +2583,8 @@ datagram_id Session::SendDatagram(Store&& data) {
     return 0;
   }
 
-  // Assign the datagram ID.
-  datagram_id did = ++impl_->state()->last_datagram_id;
+  // Use the id reserved by the caller.
+  datagram_id did = id;
 
   // Check queue capacity. Apply the drop policy when full.
   auto max_pending = impl_->state()->max_pending_datagrams;

src/quic/session.h

@@ -426,7 +426,16 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
 
   void Send(Packet::Ptr packet);
   void Send(Packet::Ptr packet, const PathStorage& path);
-  datagram_id SendDatagram(Store&& data);
+  // Reserves the next datagram id from the session-wide counter without
+  // queueing anything. Every send path mints an id here up front and passes
+  // it to SendDatagram. The id is only "exposed" (returned to JS) once the
+  // datagram is committed (queued now, or buffered for a pending stream); a
+  // reserved id that is never committed is simply discarded.
+  datagram_id ReserveDatagramId();
+  // Queues a datagram for sending using a previously reserved id. Returns the
+  // id on success, or 0 if it could not be queued, in which case the caller
+  // discards the id and emits no status.
+  datagram_id SendDatagram(Store&& data, datagram_id id);
 
   // Pending datagram accessors for use by SendPendingData.
   struct PendingDatagram {

src/quic/streams.cc

@@ -617,21 +617,29 @@ struct Stream::Impl {
     ASSIGN_OR_RETURN_UNWRAP(&stream, args.This());
     DCHECK(args[0]->IsArrayBufferView());
 
-    // A datagram can only be associated with a stream that has a real id.
-    // A pending stream has no id yet, so there is nothing to bind to.
-    if (stream->is_pending()) {
-      return args.GetReturnValue().Set(BigInt::New(env->isolate(), 0));
-    }
-
     Store store;
     if (!Store::From(args[0].As<ArrayBufferView>()).To(&store)) {
       return;
     }
 
+    // Mint the id up front. It is exposed (returned non-zero to JS) only
+    // if the datagram is committed - queued now, or buffered for a pending
+    // stream. Sync rejection returns 0 and discards (no subsequent status).
+    datagram_id id = stream->session().ReserveDatagramId();
+
+    // A pending stream has no id yet, so the datagram cannot be framed and
+    // bound to a Quarter Stream ID. Buffer it and flush when the stream opens.
+    // If it cannot be buffered (queue full) the id is discarded.
+    if (stream->is_pending()) {
+      bool buffered = stream->EnqueuePendingDatagram(id, std::move(store));
+      return args.GetReturnValue().Set(
+          BigInt::New(env->isolate(), buffered ? id : 0));
+    }
+
     Session::SendPendingDataScope send_scope(&stream->session());
-    datagram_id id =
-        stream->session().application().SendDatagram(stream, std::move(store));
-    args.GetReturnValue().Set(BigInt::New(env->isolate(), id));
+    datagram_id result = stream->session().application().SendDatagram(
+        stream, std::move(store), id);
+    args.GetReturnValue().Set(BigInt::New(env->isolate(), result));
   }
 };
 
@@ -1313,6 +1321,21 @@ void Stream::NotifyStreamOpened(stream_id id) {
           headers->flags);
     }
   }
+  if (!pending_datagram_queue_.empty()) {
+    // Like the headers flush above, this runs inside an ngtcp2 callback, so
+    // the queued datagrams ride the session's deferred flush; no send scope.
+    std::deque<PendingDatagram> queue;
+    pending_datagram_queue_.swap(queue);
+    pending_datagram_bytes_ = 0;
+    for (auto& dgram : queue) {
+      // The id was already returned to JS. If the send is rejected now (e.g.
+      // the framed size exceeds the peer's max) we report ABANDONED.
+      if (session().application().SendDatagram(
+              this, std::move(dgram.data), dgram.id) == 0) {
+        session().DatagramStatus(dgram.id, DatagramStatus::ABANDONED);
+      }
+    }
+  }
   // If the stream is not a local undirectional stream and is_readable is
   // false, then we should shutdown the streams readable side now.
   if (!is_local_unidirectional() && !is_readable()) {
@@ -1350,6 +1373,23 @@ void Stream::EnqueuePendingHeaders(HeadersKind kind,
       kind, Global<Array>(env()->isolate(), headers), flags));
 }
 
+bool Stream::EnqueuePendingDatagram(datagram_id id, Store&& store) {
+  // Bound the pending queue by the stream's outbound high water mark. When
+  // full, we refuse to send the datagram (its id is discarded by the caller).
+  // A high water mark of 0  means unbounded, matching the stream-write path.
+  uint64_t hwm = state()->high_water_mark;
+  size_t incoming = store.length();
+  if (hwm > 0 && !pending_datagram_queue_.empty() &&
+      pending_datagram_bytes_ + incoming > hwm) {
+    Debug(this, "Pending datagram buffer full, refusing datagram %" PRIu64, id);
+    return false;
+  }
+  Debug(this, "Buffering datagram %" PRIu64 " for pending stream", id);
+  pending_datagram_bytes_ += incoming;
+  pending_datagram_queue_.push_back({id, std::move(store)});
+  return true;
+}
+
 bool Stream::is_pending() const {
   return state()->pending;
 }
@@ -1683,6 +1723,19 @@ void Stream::Destroy(QuicError error) {
   }
   state()->pending = 0;
 
+  // Datagrams still buffered for a pending stream had their ids returned to
+  // JS but will never reach the wire, so abandon them here to close each
+  // handle. If the session is already gone there is nobody to notify.
+  if (!pending_datagram_queue_.empty()) {
+    if (!session_->is_destroyed()) {
+      for (auto& dgram : pending_datagram_queue_) {
+        session_->DatagramStatus(dgram.id, DatagramStatus::ABANDONED);
+      }
+    }
+    pending_datagram_queue_.clear();
+    pending_datagram_bytes_ = 0;
+  }
+
   maybe_pending_stream_.reset();
 
   // End the writable before marking as destroyed.

src/quic/streams.h

@@ -15,6 +15,7 @@
 #include "bindingdata.h"
 #include "data.h"
 
+#include <deque>
 #include <vector>
 
 namespace node::quic {
@@ -435,6 +436,11 @@ class Stream final : public AsyncWrap,
                              v8::Local<v8::Array> headers,
                              HeadersFlags flags);
 
+  // Buffers a datagram (with its already-reserved id) sent on a still-pending
+  // stream, bounded by the stream's high water mark. Returns false without
+  // queueing when the buffer is full, in which case the id is discarded.
+  bool EnqueuePendingDatagram(datagram_id id, Store&& store);
+
   ArenaSlotBase stats_slot_;
   ArenaSlotBase state_slot_;
   BaseObjectWeakPtr<Session> session_;
@@ -452,6 +458,13 @@ class Stream final : public AsyncWrap,
   error_code pending_close_read_code_ = 0;
   error_code pending_close_write_code_ = 0;
 
+  struct PendingDatagram {
+    datagram_id id;
+    Store data;
+  };
+  std::deque<PendingDatagram> pending_datagram_queue_;
+  size_t pending_datagram_bytes_ = 0;
+
   struct StoredPriority {
     StreamPriority priority = StreamPriority::DEFAULT;
     StreamPriorityFlags flags = StreamPriorityFlags::NON_INCREMENTAL;

test/parallel/test-quic-h3-datagram.mjs

@@ -330,3 +330,85 @@ const cert = readKey('agent1-cert.pem');
   await clientSession.close();
   await serverEndpoint.close();
 }
+
+// Test 6: a datagram sent on a still-pending stream gets a real, trackable id,
+// is delivered once the stream opens, and reports a terminal status via
+// ondatagramstatus like any other datagram.
+{
+  const serverGotDatagram = Promise.withResolvers();
+  const serverDone = Promise.withResolvers();
+
+  const serverEndpoint = await listen(mustCall(async (ss) => {
+    ss.onstream = mustCall((stream) => {
+      stream.ondatagram = (data) => {
+        deepStrictEqual([...data], [7, 8, 9]);
+        serverGotDatagram.resolve();
+      };
+    }, 2);
+    await serverDone.promise;
+    ss.close();
+  }), {
+    sni: { '*': { keys: [key], certs: [cert] } },
+    application: { enableDatagrams: true },
+    // Only one concurrent client bidi stream, so the second request stays
+    // pending until the first closes.
+    transportParams: { maxDatagramFrameSize: 100, initialMaxStreamsBidi: 1 },
+    onheaders: mustCall(function() {
+      this.sendHeaders({ ':status': '200' });
+      this.writer.endSync();
+    }, 2),
+  });
+
+  const datagramStatus = Promise.withResolvers();
+  const clientSession = await connect(serverEndpoint.address, {
+    servername: 'localhost',
+    verifyPeer: 'manual',
+    application: { enableDatagrams: true },
+    transportParams: { maxDatagramFrameSize: 100 },
+    ondatagramstatus: (id, status) => datagramStatus.resolve({ id, status }),
+  });
+  await clientSession.opened;
+
+  // First request holds the single available bidi slot.
+  const stream1 = await clientSession.createBidirectionalStream({
+    headers: {
+      ':method': 'GET', ':path': '/first',
+      ':scheme': 'https', ':authority': 'localhost',
+    },
+    onheaders: mustCall(function(headers) {
+      strictEqual(headers[':status'], '200');
+    }),
+  });
+
+  // Second stream is pending; the datagram is buffered with a real id.
+  const secondResponse = Promise.withResolvers();
+  const stream2 = await clientSession.createBidirectionalStream({
+    headers: {
+      ':method': 'GET', ':path': '/second',
+      ':scheme': 'https', ':authority': 'localhost',
+    },
+    onheaders: mustCall(() => secondResponse.resolve()),
+  });
+  ok(stream2.pending);
+  const pendingId = stream2.sendDatagram(new Uint8Array([7, 8, 9]));
+  ok(pendingId > 0n);
+
+  // Free the slot; stream2 opens and the buffered datagram flushes.
+  stream1.writer.endSync();
+  for await (const _ of stream1) { /* drain */ } // eslint-disable-line no-unused-vars
+  await stream1.closed;
+
+  await Promise.all([secondResponse.promise, serverGotDatagram.promise]);
+
+  // The buffered datagram is tracked: its id reports exactly one terminal
+  // status (acknowledged on loopback, but lost is also valid).
+  const { id, status } = await datagramStatus.promise;
+  strictEqual(id, pendingId);
+  ok(status === 'acknowledged' || status === 'lost');
+
+  for await (const _ of stream2) { /* drain */ } // eslint-disable-line no-unused-vars
+  await stream2.closed;
+  serverDone.resolve();
+  await clientSession.close();
+  await serverEndpoint.close();
+}