quic: add promise to QuicStream for pending strms

before this PR, it was necessary to poll, if a
stream can not be created immediately due to
flow control.
This PR adds a promise to QuicStream, that fulfills,
when a stream is available and ready.

Signed-off-by: Marten Richter <marten.richter@freenet.de>
PR-URL: #65862
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Tim Perry <pimterry@gmail.com>

Author: martenrichter

doc/api/quic.md

@@ -1861,6 +1861,19 @@ Either `'application'` or `'transport'`. Indicates the namespace of
 added: v23.8.0
 -->
 
+### `stream.opened`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {Promise}
+
+A promise that is immediately fulfilled, if the stream fits within
+flow control limits or fulfilled when the pending stream is created.
+It rejects, if a pending stream is closed with an error before being
+created.
+
 ### `stream.closed`
 
 <!-- YAML

lib/internal/quic/quic.js

@@ -184,6 +184,7 @@ const kEmptyObject = { __proto__: null };
 
 const {
   kAttachFileHandle,
+  kAvailable,
   kBlocked,
   kConnect,
   kDatagram,
@@ -948,6 +949,11 @@ setCallbacks({
   },
 
   // QuicStream callbacks
+  onStreamAvailable() {
+    debug('stream available callback', this[kOwner]);
+    this[kOwner][kAvailable]();
+  },
+
   onStreamBlocked() {
     debug('stream blocked callback', this[kOwner]);
     // Called when the stream C++ handle has been blocked by flow control.
@@ -1602,6 +1608,7 @@ class QuicStream {
     fileHandle: undefined,
     headers: undefined,
     pendingTrailers: undefined,
+    pendingStream: PromiseWithResolvers(),
     onerror: undefined,
     onblocked: undefined,
     onreset: undefined,
@@ -1655,6 +1662,10 @@ class QuicStream {
     inner.state = new QuicStreamState(
       kPrivateConstructor, handle.state, handle.stateByteOffset);
 
+    if (!inner.state.pending) {
+      inner.pendingStream.resolve();
+    }
+
     if (hasObserver('quic')) {
       startPerf(this, kPerfEntry, { type: 'quic', name: 'QuicStream' });
     }
@@ -1719,6 +1730,15 @@ class QuicStream {
     return this.#inner.state.pending;
   }
 
+  /**
+   * Promise that resolves once the stream is available and not pending.
+   * @type {Promise<void>}
+   */
+  get opened() {
+    assertIsQuicStream(this);
+    return this.#inner.pendingStream.promise;
+  }
+
   /**
    * True if any data on this stream was received as 0-RTT (early data)
    * before the TLS handshake completed. Early data is less secure and
@@ -2597,6 +2617,13 @@ class QuicStream {
     } else {
       inner.pendingClose.resolve();
     }
+    if (inner.state.pending) {
+      if (error !== undefined) {
+        inner.pendingStream.reject(error);
+      } else {
+        inner.pendingStream.resolve(error);
+      }
+    }
     debug('stream closed');
     if (onStreamClosedChannel.hasSubscribers) {
       onStreamClosedChannel.publish({
@@ -2642,6 +2669,12 @@ class QuicStream {
     }
   }
 
+  [kAvailable]() {
+    // The formerly pending stream is now available
+    const inner = this.#inner;
+    inner.pendingStream.resolve();
+  }
+
   [kBlocked]() {
     const inner = this.#inner;
     // The blocked event should only be called if the stream was created with

lib/internal/quic/symbols.js

@@ -28,6 +28,7 @@ const {
 // public API.
 
 const kAttachFileHandle = Symbol('kAttachFileHandle');
+const kAvailable = Symbol('kAvailable');
 const kBlocked = Symbol('kBlocked');
 const kConnect = Symbol('kConnect');
 const kDrain = Symbol('kDrain');
@@ -63,6 +64,7 @@ const kVersionNegotiation = Symbol('kVersionNegotiation');
 
 module.exports = {
   kAttachFileHandle,
+  kAvailable,
   kBlocked,
   kConnect,
   kDatagram,

src/quic/bindingdata.h

@@ -55,6 +55,7 @@ class SessionManager;
   V(session_path_validation, SessionPathValidation)                            \
   V(session_ticket, SessionTicket)                                             \
   V(session_version_negotiation, SessionVersionNegotiation)                    \
+  V(stream_available, StreamAvailable)                                         \
   V(stream_blocked, StreamBlocked)                                             \
   V(stream_close, StreamClose)                                                 \
   V(stream_created, StreamCreated)                                             \

src/quic/streams.cc

@@ -1195,10 +1195,13 @@ void Stream::NotifyStreamOpened(stream_id id) {
   // sending can eventually call into JavaScript and destroy the stream.
   BaseObjectPtr<Stream> self(this);
   auto& application = session().application();
-  error_code internal_error = application.GetInternalErrorCode();
   if (!application.StreamOpened(*this) && !is_destroyed()) {
+    error_code internal_error = application.GetInternalErrorCode();
     Destroy(QuicError::ForApplication(internal_error));
   }
+
+  // We inform, the js side that the pending stream is now available
+  if (!is_destroyed()) EmitStreamAvailable();
 }
 
 void Stream::NotifyReadableEnded(error_code code) {
@@ -1779,6 +1782,14 @@ void Stream::SendStopSending(error_code code) {
 
 // ============================================================================
 
+void Stream::EmitStreamAvailable() {
+  if (!env()->can_call_into_js()) {
+    return;
+  }
+  CallbackScope<Stream> cb_scope(this);
+  MakeCallback(BindingData::Get(env()).stream_available_callback(), 0, nullptr);
+}
+
 void Stream::EmitBlocked() {
   // state()->wants_block will be set from the javascript side if the
   // stream object has a handler for the blocked event.

src/quic/streams.h

@@ -433,6 +433,10 @@ class Stream final : public AsyncWrap,
 
   // JavaScript callouts
 
+  // Notifies the JavaScript side that a previously pending stream
+  // is now available.
+  void EmitStreamAvailable();
+
   // Notifies the JavaScript side that the stream has been destroyed.
   void EmitClose(const QuicError& error);
 

test/parallel/test-quic-internal-setcallbacks.mjs

@@ -26,6 +26,7 @@ const callbacks = {
   onSessionOrigin() {},
   onSessionGoaway() {},
   onSessionVersionNegotiation() {},
+  onStreamAvailable() {},
   onStreamCreated() {},
   onStreamBlocked() {},
   onStreamClose() {},

test/parallel/test-quic-stream-limits-pending.mjs

@@ -38,28 +38,55 @@ const serverEndpoint = await listen(mustCall((serverSession) => {
 const clientSession = await connect(serverEndpoint.address);
 await clientSession.opened;
 
+let opened = 0;
+
 // First stream opens immediately (within the limit).
 const s1 = await clientSession.createBidirectionalStream({
   body: encoder.encode('stream 1'),
 });
 
+// eslint-disable-next-line node-core/must-call-assert
+s1.opened.then(() => {
+  opened++;
+});
+
 // Second stream is created but queued as pending because the
 // server only allows 1 concurrent bidi stream.
 const s2 = await clientSession.createBidirectionalStream({
   body: encoder.encode('stream 2'),
 });
 
+// eslint-disable-next-line node-core/must-call-assert
+s2.opened.then(() => {
+  opened++;
+});
+
+// Third stream is created but queued as pending because the
+// server only allows 1 concurrent bidi stream.
+const s3 = await clientSession.createBidirectionalStream({
+  body: encoder.encode('stream 3'),
+});
+
+
 // s2 should be pending until s1 closes and the server grants
 // more stream credits.
 assert.strictEqual(s2.pending, true);
+assert.strictEqual(opened, 1);
 
 // Drain and close the first stream.
 for await (const _ of s1) { /* drain */ } // eslint-disable-line no-unused-vars
 await s1.closed;
 
+const err = new Error('Test error');
+s3.destroy(err);
+
+await Promise.all([assert.rejects(s3.opened, err), assert.rejects(s3.closed, err)]);
+
+
 // After s1 closes, the server sends MAX_STREAMS which opens s2.
 // Wait for the server to receive both streams.
 await allDone.promise;
+assert.strictEqual(opened, 2);
 
 // s2 should no longer be pending.
 for await (const _ of s2) { /* drain */ } // eslint-disable-line no-unused-vars

test/parallel/test-quic-stream-limits-uni.mjs

@@ -34,20 +34,35 @@ const serverEndpoint = await listen(mustCall((serverSession) => {
 const clientSession = await connect(serverEndpoint.address);
 await clientSession.opened;
 
+let opened = 0;
+
 // First uni stream opens immediately.
 const s1 = await clientSession.createUnidirectionalStream({
   body: encoder.encode('uni 1'),
 });
 
+// eslint-disable-next-line node-core/must-call-assert
+s1.opened.then(() => {
+  opened++;
+});
+
 // Second uni stream is pending (limit = 1).
 const s2 = await clientSession.createUnidirectionalStream({
   body: encoder.encode('uni 2'),
 });
+
+// eslint-disable-next-line node-core/must-call-assert
+s2.opened.then(() => {
+  opened++;
+});
+assert.strictEqual(opened, 1);
+
 assert.strictEqual(s2.pending, true);
 
 // Wait for both to complete.
 await s1.closed;
 await allDone.promise;
+assert.strictEqual(opened, 2);
 await s2.closed;
 
 await clientSession.close();

typings/internalBinding/quic.d.ts

@@ -19,6 +19,7 @@ interface QuicCallbacks {
                                 versions: number[],
                                 supports: number[]) => void;
   onStreamCreated: (stream: Stream) => void;
+  onStreamAvailable: () => void;
   onStreamBlocked: () => void;
   onStreamClose: (error: [number,bigint,string]) => void;
   onStreamReset: (error: [number,bigint,string]) => void;