quic: complete the javascript implementation of QUIC

Signed-off-by: James M Snell <[email protected]>
Assisted-by: Opencode:Opus 4.6

Author: jasnell

lib/internal/blob.js

@@ -1,6 +1,7 @@
 'use strict';
 
 const {
+  ArrayPrototypePush,
   MathMax,
   MathMin,
   ObjectDefineProperties,
@@ -527,13 +528,82 @@ function createBlobReaderStream(reader) {
   }, { highWaterMark: 0 });
 }
 
+// Maximum number of chunks to collect in a single batch to prevent
+// unbounded memory growth when the DataQueue has a large burst of data.
+const kMaxBatchChunks = 16;
+
+async function* createBlobReaderIterable(reader, options = {}) {
+  const { getReadError } = options;
+  let wakeup = PromiseWithResolvers();
+  reader.setWakeup(wakeup.resolve);
+
+  try {
+    while (true) {
+      const batch = [];
+      let blocked = false;
+      let eos = false;
+      let error = null;
+
+      // Pull as many chunks as available synchronously.
+      // reader.pull(callback) calls the callback synchronously via
+      // MakeCallback, so we can collect multiple chunks per iteration
+      // step without any async overhead.
+      while (true) {
+        let pullResult;
+        reader.pull((status, buffer) => {
+          pullResult = { status, buffer };
+        });
+
+        if (pullResult.status === 0) {
+          eos = true;
+          break;
+        }
+        if (pullResult.status < 0) {
+          error = typeof getReadError === 'function' ?
+            getReadError(pullResult.status) :
+            new ERR_INVALID_STATE('The reader is not readable');
+          break;
+        }
+        if (pullResult.status === 2) {
+          blocked = true;
+          break;
+        }
+        ArrayPrototypePush(batch, new Uint8Array(pullResult.buffer));
+        if (batch.length >= kMaxBatchChunks) break;
+      }
+
+      if (batch.length > 0) {
+        yield batch;
+      }
+
+      if (eos) return;
+      if (error) throw error;
+
+      if (blocked) {
+        const fin = await wakeup.promise;
+        wakeup = PromiseWithResolvers();
+        reader.setWakeup(wakeup.resolve);
+        // If the wakeup was triggered by FIN (EndReadable), the DataQueue
+        // is capped. Continue the loop to pull again -- the next pull will
+        // return EOS. Without this, a race between the data notification
+        // and the FIN notification can leave the iterator waiting for a
+        // wakeup that will never come.
+        if (fin) continue;
+      }
+    }
+  } finally {
+    reader.setWakeup(undefined);
+  }
+}
+
 module.exports = {
   Blob,
   createBlob,
   createBlobFromFilePath,
+  createBlobReaderIterable,
+  createBlobReaderStream,
   isBlob,
   kHandle,
   resolveObjectURL,
   TransferableBlob,
-  createBlobReaderStream,
 };

lib/internal/errors.js

@@ -1691,6 +1691,8 @@ E('ERR_QUIC_APPLICATION_ERROR', 'A QUIC application error occurred. %d [%s]', Er
 E('ERR_QUIC_CONNECTION_FAILED', 'QUIC connection failed', Error);
 E('ERR_QUIC_ENDPOINT_CLOSED', 'QUIC endpoint closed: %s (%d)', Error);
 E('ERR_QUIC_OPEN_STREAM_FAILED', 'Failed to open QUIC stream', Error);
+E('ERR_QUIC_STREAM_RESET',
+  'The QUIC stream was reset by the peer with error code %d', Error);
 E('ERR_QUIC_TRANSPORT_ERROR', 'A QUIC transport error occurred. %d [%s]', Error);
 E('ERR_QUIC_VERSION_NEGOTIATION_ERROR', 'The QUIC session requires version negotiation', Error);
 E('ERR_REQUIRE_ASYNC_MODULE', function(filename, parentFilename) {

lib/internal/fs/promises.js

@@ -1994,6 +1994,8 @@ module.exports = {
   },
 
   FileHandle,
+  kHandle,
+  kLocked,
   kRef,
   kUnref,
 };

lib/internal/perf/observe.js

@@ -27,6 +27,7 @@ const {
     NODE_PERFORMANCE_ENTRY_TYPE_HTTP,
     NODE_PERFORMANCE_ENTRY_TYPE_NET,
     NODE_PERFORMANCE_ENTRY_TYPE_DNS,
+    NODE_PERFORMANCE_ENTRY_TYPE_QUIC,
   },
   installGarbageCollectionTracking,
   observerCounts,
@@ -87,6 +88,7 @@ const kSupportedEntryTypes = ObjectFreeze([
   'mark',
   'measure',
   'net',
+  'quic',
   'resource',
 ]);
 
@@ -131,6 +133,7 @@ function getObserverType(type) {
     case 'http': return NODE_PERFORMANCE_ENTRY_TYPE_HTTP;
     case 'net': return NODE_PERFORMANCE_ENTRY_TYPE_NET;
     case 'dns': return NODE_PERFORMANCE_ENTRY_TYPE_DNS;
+    case 'quic': return NODE_PERFORMANCE_ENTRY_TYPE_QUIC;
   }
 }
 

lib/internal/quic/diagnostics.js

@@ -0,0 +1,71 @@
+'use strict';
+
+const dc = require('diagnostics_channel');
+
+const onEndpointCreatedChannel = dc.channel('quic.endpoint.created');
+const onEndpointListeningChannel = dc.channel('quic.endpoint.listen');
+const onEndpointClosingChannel = dc.channel('quic.endpoint.closing');
+const onEndpointClosedChannel = dc.channel('quic.endpoint.closed');
+const onEndpointErrorChannel = dc.channel('quic.endpoint.error');
+const onEndpointBusyChangeChannel = dc.channel('quic.endpoint.busy.change');
+const onEndpointClientSessionChannel = dc.channel('quic.session.created.client');
+const onEndpointServerSessionChannel = dc.channel('quic.session.created.server');
+const onSessionOpenStreamChannel = dc.channel('quic.session.open.stream');
+const onSessionReceivedStreamChannel = dc.channel('quic.session.received.stream');
+const onSessionSendDatagramChannel = dc.channel('quic.session.send.datagram');
+const onSessionUpdateKeyChannel = dc.channel('quic.session.update.key');
+const onSessionClosingChannel = dc.channel('quic.session.closing');
+const onSessionClosedChannel = dc.channel('quic.session.closed');
+const onSessionReceiveDatagramChannel = dc.channel('quic.session.receive.datagram');
+const onSessionReceiveDatagramStatusChannel = dc.channel('quic.session.receive.datagram.status');
+const onSessionPathValidationChannel = dc.channel('quic.session.path.validation');
+const onSessionNewTokenChannel = dc.channel('quic.session.new.token');
+const onSessionTicketChannel = dc.channel('quic.session.ticket');
+const onSessionVersionNegotiationChannel = dc.channel('quic.session.version.negotiation');
+const onSessionOriginChannel = dc.channel('quic.session.receive.origin');
+const onSessionHandshakeChannel = dc.channel('quic.session.handshake');
+const onSessionGoawayChannel = dc.channel('quic.session.goaway');
+const onSessionEarlyRejectedChannel = dc.channel('quic.session.early.rejected');
+const onStreamClosedChannel = dc.channel('quic.stream.closed');
+const onStreamHeadersChannel = dc.channel('quic.stream.headers');
+const onStreamTrailersChannel = dc.channel('quic.stream.trailers');
+const onStreamInfoChannel = dc.channel('quic.stream.info');
+const onStreamResetChannel = dc.channel('quic.stream.reset');
+const onStreamBlockedChannel = dc.channel('quic.stream.blocked');
+const onSessionErrorChannel = dc.channel('quic.session.error');
+const onEndpointConnectChannel = dc.channel('quic.endpoint.connect');
+
+module.exports = {
+  onEndpointCreatedChannel,
+  onEndpointListeningChannel,
+  onEndpointClosingChannel,
+  onEndpointClosedChannel,
+  onEndpointErrorChannel,
+  onEndpointBusyChangeChannel,
+  onEndpointClientSessionChannel,
+  onEndpointServerSessionChannel,
+  onSessionOpenStreamChannel,
+  onSessionReceivedStreamChannel,
+  onSessionSendDatagramChannel,
+  onSessionUpdateKeyChannel,
+  onSessionClosingChannel,
+  onSessionClosedChannel,
+  onSessionReceiveDatagramChannel,
+  onSessionReceiveDatagramStatusChannel,
+  onSessionPathValidationChannel,
+  onSessionNewTokenChannel,
+  onSessionTicketChannel,
+  onSessionVersionNegotiationChannel,
+  onSessionOriginChannel,
+  onSessionHandshakeChannel,
+  onSessionGoawayChannel,
+  onSessionEarlyRejectedChannel,
+  onStreamClosedChannel,
+  onStreamHeadersChannel,
+  onStreamTrailersChannel,
+  onStreamInfoChannel,
+  onStreamResetChannel,
+  onStreamBlockedChannel,
+  onSessionErrorChannel,
+  onEndpointConnectChannel,
+};