quic: refactor qlog handling

Because of the timing of qlog events emitted by ngtcp2, it
becomes difficult to handle those as events on the QuicSession
object because the final qlog entry is not emitted until the
ngtcp2_conn is freed, which can occur when the object is being
garbage collected (meaning, we a: can't call out to javascript
and b: don't have an object we can use to emit the event).

This refactors it into a QLogStream object that allows the
qlog data to be piped out using a separate Readable stream.

PR-URL: #34160
Reviewed-By: Anna Henningsen <anna@addaleax.net>

Author: jasnell

doc/api/quic.md

@@ -274,7 +274,7 @@ added: REPLACEME
   * `maxStatelessResetsPerHost` {number} The maximum number of stateless
     resets that the `QuicSocket` is permitted to send per remote host.
     Default: `10`.
-  * `qlog` {boolean} Whether to emit ['qlog'][] events for incoming sessions.
+  * `qlog` {boolean} Whether to enable ['qlog'][] for incoming sessions.
     (For outgoing client sessions, set `client.qlog`.) Default: `false`.
   * `retryTokenTimeout` {number} The maximum number of *seconds* for retry token
     validation. Default: `10` seconds.
@@ -633,20 +633,6 @@ The callback will be invoked with three arguments:
 
 The `'pathValidation'` event will be emitted multiple times.
 
-#### Event: `'qlog'`
-<!-- YAML
-added: REPLACEME
--->
-
-* `jsonChunk` {string} A JSON fragment.
-
-Emitted if the `qlog: true` option was passed to `quicsocket.connect()` or
-`net.createQuicSocket()` functions.
-
-The argument is a JSON fragment according to the [qlog standard][].
-
-The `'qlog'` event will be emitted multiple times.
-
 #### Event: `'secure'`
 <!-- YAML
 added: REPLACEME
@@ -1044,6 +1030,19 @@ added: REPLACEME
 A `BigInt` representing the total number of `QuicStreams` initiated by the
 connected peer.
 
+#### quicsession.qlog
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {stream.Readable}
+
+If `qlog` support is enabled for `QuicSession`, the `quicsession.qlog` property
+provides a [`stream.Readable`][] that may be used to access the `qlog` event
+data according to the [qlog standard][]. For client `QuicSessions`, the
+`quicsession.qlog` property will be `undefined` untilt the `'qlog'` event
+is emitted.
+
 #### quicsession.remoteAddress
 <!-- YAML
 added: REPLACEME
@@ -1183,6 +1182,17 @@ The `sessionTicket` and `remoteTransportParams` are useful when creating a new
 
 The `'sessionTicket'` event may be emitted multiple times.
 
+#### Event: `'qlog'`
+<!-- YAML
+added: REPLACEME
+-->
+
+The `'qlog'` event is emitted when the `QuicClientSession` is ready to begin
+providing `qlog` event data. The callback is invoked with a single argument:
+
+* `qlog` {stream.Readable} A [`stream.Readable`][] that is also available using
+  the `quicsession.qlog` property.
+
 #### Event: `'usePreferredAddress'`
 <!-- YAML
 added: REPLACEME
@@ -1573,7 +1583,7 @@ added: REPLACEME
     transport parameters from a previously established session. These would
     have been provided as part of the `'sessionTicket'` event on a previous
     `QuicClientSession` object.
-  * `qlog` {boolean} Whether to emit ['qlog'][] events for this session.
+  * `qlog` {boolean} Whether to enable ['qlog'][] for this session.
     Default: `false`.
   * `requestOCSP` {boolean} If `true`, specifies that the OCSP status request
     extension will be added to the client hello and an `'OCSPResponse'` event
@@ -2278,6 +2288,7 @@ added: REPLACEME
 Set to `true` if the `QuicStream` is unidirectional.
 
 [`crypto.getCurves()`]: crypto.html#crypto_crypto_getcurves
+[`stream.Readable`]: #stream_class_stream_readable
 [`tls.DEFAULT_ECDH_CURVE`]: #tls_tls_default_ecdh_curve
 [`tls.getCiphers()`]: tls.html#tls_tls_getciphers
 [ALPN]: https://tools.ietf.org/html/rfc7301
@@ -2286,5 +2297,5 @@ Set to `true` if the `QuicStream` is unidirectional.
 [modifying the default cipher suite]: tls.html#tls_modifying_the_default_tls_cipher_suite
 [OpenSSL Options]: crypto.html#crypto_openssl_options
 [Perfect Forward Secrecy]: #tls_perfect_forward_secrecy
-['qlog']: #quic_event_qlog
+['qlog']: #quic_quicsession_qlog
 [qlog standard]: https://tools.ietf.org/id/draft-marx-qlog-event-definitions-quic-h3-00.html

lib/internal/quic/core.js

@@ -40,6 +40,7 @@ const {
   validateCreateSecureContextOptions,
   validateQuicSocketConnectOptions,
   QuicSessionSharedState,
+  QLogStream,
 } = require('internal/quic/util');
 const util = require('util');
 const assert = require('internal/assert');
@@ -236,6 +237,7 @@ const kRemoveSession = Symbol('kRemove');
 const kRemoveStream = Symbol('kRemoveStream');
 const kServerBusy = Symbol('kServerBusy');
 const kSetHandle = Symbol('kSetHandle');
+const kSetQLogStream = Symbol('kSetQLogStream');
 const kSetSocket = Symbol('kSetSocket');
 const kSetSocketAfterBind = Symbol('kSetSocketAfterBind');
 const kStartFilePipe = Symbol('kStartFilePipe');
@@ -423,37 +425,17 @@ function onSessionUsePreferredAddress(address, port, family) {
   });
 }
 
-// Called by the C++ internals to emit a QLog record.
-function onSessionQlog(str) {
-  if (this.qlogBuffer === undefined) this.qlogBuffer = '';
-
+// Called by the C++ internals to emit a QLog record. This can
+// be called before the QuicSession has been fully initialized,
+// in which case we store a reference and defer emitting the
+// qlog event until after we're initialized.
+function onSessionQlog(handle) {
   const session = this[owner_symbol];
-
-  if (session && session.listenerCount('qlog') > 0) {
-    // Emit this chunk along with any previously buffered data.
-    str = this.qlogBuffer + str;
-    this.qlogBuffer = '';
-    if (str === '') return;
-    session.emit('qlog', str);
-  } else {
-    // Buffer the data until both the JS session object and a listener
-    // become available.
-    this.qlogBuffer += str;
-
-    if (!session || this.waitingForQlogListener) return;
-    this.waitingForQlogListener = true;
-
-    function onNewListener(ev) {
-      if (ev === 'qlog') {
-        session.removeListener('newListener', onNewListener);
-        process.nextTick(() => {
-          onSessionQlog.call(this, '');
-        });
-      }
-    }
-
-    session.on('newListener', onNewListener);
-  }
+  const stream = new QLogStream(handle);
+  if (session)
+    session[kSetQLogStream](stream);
+  else
+    this.qlogStream = stream;
 }
 
 // Called by the C++ internals when a client QuicSession receives
@@ -1631,6 +1613,7 @@ class QuicSession extends EventEmitter {
     highWaterMark: undefined,
     defaultEncoding: undefined,
     state: undefined,
+    qlogStream: undefined,
   };
 
   constructor(socket, options) {
@@ -1662,6 +1645,14 @@ class QuicSession extends EventEmitter {
     };
   }
 
+  [kSetQLogStream](stream) {
+    const state = this[kInternalState];
+    state.qlogStream = stream;
+    process.nextTick(() => {
+      this.emit('qlog', state.qlogStream);
+    });
+  }
+
   // Sets the internal handle for the QuicSession instance. For
   // server QuicSessions, this is called immediately as the
   // handle is created before the QuicServerSession JS object.
@@ -1676,6 +1667,10 @@ class QuicSession extends EventEmitter {
       state.state = new QuicSessionSharedState(handle.state);
       state.handshakeAckHistogram = new Histogram(handle.ack);
       state.handshakeContinuationHistogram = new Histogram(handle.rate);
+      if (handle.qlogStream !== undefined) {
+        this[kSetQLogStream](handle.qlogStream);
+        handle.qlogStream = undefined;
+      }
     } else {
       if (state.handshakeAckHistogram)
         state.handshakeAckHistogram[kDestroyHistogram]();
@@ -1934,6 +1929,10 @@ class QuicSession extends EventEmitter {
     return { bidi, uni };
   }
 
+  get qlog() {
+    return this[kInternalState].qlogStream;
+  }
+
   get address() {
     return this[kInternalState].socket?.address || {};
   }

lib/internal/quic/util.js

@@ -15,12 +15,24 @@ const {
   },
 } = require('internal/errors');
 
+const {
+  symbols: {
+    owner_symbol,
+  },
+} = require('internal/async_hooks');
+
 const {
   kHandle,
 } = require('internal/stream_base_commons');
 
 const endianness = require('os').endianness();
 
+const { Readable } = require('stream');
+const {
+  kUpdateTimer,
+  onStreamRead,
+} = require('internal/stream_base_commons');
+
 const assert = require('internal/assert');
 assert(process.versions.ngtcp2 !== undefined);
 
@@ -885,6 +897,31 @@ class QuicSessionSharedState {
   }
 }
 
+class QLogStream extends Readable {
+  constructor(handle) {
+    super({ autoDestroy: true });
+    this[kHandle] = handle;
+    handle[owner_symbol] = this;
+    handle.onread = onStreamRead;
+  }
+
+  _read() {
+    if (this[kHandle])
+      this[kHandle].readStart();
+  }
+
+  _destroy() {
+    // Release the references on the handle so that
+    // it can be garbage collected.
+    this[kHandle][owner_symbol] = undefined;
+    this[kHandle] = undefined;
+  }
+
+  [kUpdateTimer]() {
+    // Does nothing
+  }
+}
+
 module.exports = {
   getAllowUnauthorized,
   getSocketType,
@@ -903,4 +940,5 @@ module.exports = {
   validateCreateSecureContextOptions,
   validateQuicSocketConnectOptions,
   QuicSessionSharedState,
+  QLogStream,
 };

src/async_wrap.h

@@ -59,6 +59,7 @@ namespace node {
   V(PROCESSWRAP)                                                              \
   V(PROMISE)                                                                  \
   V(QUERYWRAP)                                                                \
+  V(QLOGSTREAM)                                                               \
   V(QUICCLIENTSESSION)                                                        \
   V(QUICSERVERSESSION)                                                        \
   V(QUICSENDWRAP)                                                             \

src/env.h

@@ -441,6 +441,7 @@ constexpr size_t kFsStatsBufferLength =
   V(secure_context_constructor_template, v8::FunctionTemplate)                 \
   V(shutdown_wrap_template, v8::ObjectTemplate)                                \
   V(streambaseoutputstream_constructor_template, v8::ObjectTemplate)           \
+  V(qlogoutputstream_constructor_template, v8::ObjectTemplate)                 \
   V(tcp_constructor_template, v8::FunctionTemplate)                            \
   V(tty_constructor_template, v8::FunctionTemplate)                            \
   V(write_wrap_template, v8::ObjectTemplate)                                   \