quic: convert openStream to Promise

Although most of the time openStream will be able to create the stream
immediately, when a stream is opened before the handshake is complete
we have to wait for the handshake to be complete before continuing.

PR-URL: #34351
Reviewed-By: Robert Nagy <[email protected]>
Reviewed-By: Anna Henningsen <[email protected]>

Author: jasnell

lib/internal/quic/core.js

@@ -19,6 +19,7 @@ const {
   Promise,
   PromiseAll,
   PromiseReject,
+  PromiseResolve,
   RegExp,
   Set,
   Symbol,
@@ -213,13 +214,13 @@ const kDestroy = Symbol('kDestroy');
 const kEndpointBound = Symbol('kEndpointBound');
 const kEndpointClose = Symbol('kEndpointClose');
 const kHandshake = Symbol('kHandshake');
+const kHandshakeComplete = Symbol('kHandshakeComplete');
 const kHandshakePost = Symbol('kHandshakePost');
 const kHeaders = Symbol('kHeaders');
 const kInternalState = Symbol('kInternalState');
 const kInternalClientState = Symbol('kInternalClientState');
 const kInternalServerState = Symbol('kInternalServerState');
 const kListen = Symbol('kListen');
-const kMakeStream = Symbol('kMakeStream');
 const kMaybeBind = Symbol('kMaybeBind');
 const kOnFileOpened = Symbol('kOnFileOpened');
 const kOnFileUnpipe = Symbol('kOnFileUnpipe');
@@ -1651,6 +1652,9 @@ class QuicSession extends EventEmitter {
     destroyed: false,
     earlyData: false,
     handshakeComplete: false,
+    handshakeCompletePromise: undefined,
+    handshakeCompletePromiseResolve: undefined,
+    handshakeCompletePromiseReject: undefined,
     idleTimeout: false,
     maxPacketLength: NGTCP2_DEFAULT_MAX_PKTLEN,
     servername: undefined,
@@ -1715,6 +1719,26 @@ class QuicSession extends EventEmitter {
     });
   }
 
+  [kHandshakeComplete]() {
+    const state = this[kInternalState];
+    if (state.handshakeComplete)
+      return PromiseResolve();
+
+    if (state.handshakeCompletePromise !== undefined)
+      return state.handshakeCompletePromise;
+
+    state.handshakeCompletePromise = new Promise((resolve, reject) => {
+      state.handshakeCompletePromiseResolve = resolve;
+      state.handshakeCompletePromiseReject = reject;
+    }).finally(() => {
+      state.handshakeCompletePromise = undefined;
+      state.handshakeCompletePromiseReject = undefined;
+      state.handshakeCompletePromiseResolve = undefined;
+    });
+
+    return state.handshakeCompletePromise;
+  }
+
   // 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.
@@ -1827,8 +1851,18 @@ class QuicSession extends EventEmitter {
     state.verifyErrorReason = verifyErrorReason;
     state.verifyErrorCode = verifyErrorCode;
     state.earlyData = earlyData;
-    if (!this[kHandshakePost]())
+
+    if (!this[kHandshakePost]()) {
+      if (typeof state.handshakeCompletePromiseReject === 'function') {
+        // TODO(@jasnell): Proper error
+        state.handshakeCompletePromiseReject(
+          new ERR_OPERATION_FAILED('Handshake failed'));
+      }
       return;
+    }
+
+    if (typeof state.handshakeCompletePromiseResolve === 'function')
+      state.handshakeCompletePromiseResolve();
 
     process.nextTick(() => {
       try {
@@ -1971,6 +2005,12 @@ class QuicSession extends EventEmitter {
     } else if (typeof state.closePromiseResolve === 'function')
       state.closePromiseResolve();
 
+    if (typeof state.handshakeCompletePromiseReject === 'function') {
+      // TODO(@jasnell): Proper error
+      state.handshakeCompletePromiseReject(
+        new ERR_OPERATION_FAILED('Handshake failed'));
+    }
+
     process.nextTick(emit.bind(this, 'close'));
   }
 
@@ -2113,8 +2153,7 @@ class QuicSession extends EventEmitter {
     return this[kInternalState].statelessReset;
   }
 
-  openStream(options) {
-    const state = this[kInternalState];
+  async openStream(options) {
     if (this.destroyed) {
       throw new ERR_INVALID_STATE(
         `${this.constructor.name} is already destroyed`);
@@ -2123,51 +2162,42 @@ class QuicSession extends EventEmitter {
       throw new ERR_INVALID_STATE(
         `${this.constructor.name} is closing`);
     }
+
     const {
       halfOpen,  // Unidirectional or Bidirectional
       highWaterMark,
       defaultEncoding,
     } = validateQuicStreamOptions(options);
 
-    const stream = new QuicStream({
-      highWaterMark,
-      defaultEncoding,
-      readable: !halfOpen
-    }, this);
+    await this[kHandshakeComplete]();
 
-    state.pendingStreams.add(stream);
-
-    // If early data is being used, we can create the internal QuicStream on the
-    // ready event, that is immediately after the internal QuicSession handle
-    // has been created. Otherwise, we have to wait until the secure event
-    // signaling the completion of the TLS handshake.
-    const makeStream = QuicSession[kMakeStream].bind(this, stream, halfOpen);
-    let deferred = false;
-    if (!this.handshakeComplete) {
-      deferred = true;
-      this.once('secure', makeStream);
+    if (this.destroyed) {
+      throw new ERR_INVALID_STATE(
+        `${this.constructor.name} is already destroyed`);
+    }
+    if (this.closing) {
+      throw new ERR_INVALID_STATE(
+        `${this.constructor.name} is closing`);
     }
 
-    if (!deferred)
-      makeStream(stream, halfOpen);
-
-    return stream;
-  }
-
-  static [kMakeStream](stream, halfOpen) {
-    this[kInternalState].pendingStreams.delete(stream);
     const handle =
       halfOpen ?
         _openUnidirectionalStream(this[kHandle]) :
         _openBidirectionalStream(this[kHandle]);
 
-    if (handle === undefined) {
-      stream.destroy(new ERR_OPERATION_FAILED('Unable to create QuicStream'));
-      return;
-    }
+    if (handle === undefined)
+      throw new ERR_OPERATION_FAILED('Unable to create QuicStream');
+
+    const stream = new QuicStream({
+      highWaterMark,
+      defaultEncoding,
+      readable: !halfOpen
+    }, this);
 
     stream[kSetHandle](handle);
     this[kAddStream](stream.id, stream);
+
+    return stream;
   }
 
   get duration() {

test/parallel/test-quic-client-connect-multiple-parallel.js

@@ -44,11 +44,9 @@ async function connect(server, client) {
   for (let i = 0; i < kCount; i++) {
     const server = createQuicSocket({ server: options });
 
-    server.on('session', common.mustCall((session) => {
-      session.on('secure', common.mustCall(() => {
-        const stream = session.openStream({ halfOpen: true });
-        stream.end('Hi!');
-      }));
+    server.on('session', common.mustCall(async (session) => {
+      const stream = await session.openStream({ halfOpen: true });
+      stream.end('Hi!');
     }));
 
     server.on('close', common.mustCall());

test/parallel/test-quic-client-connect-multiple-sequential.js

@@ -45,11 +45,9 @@ async function connect(server, client) {
   for (let i = 0; i < kCount; i++) {
     const server = createQuicSocket({ server: options });
 
-    server.on('session', common.mustCall((session) => {
-      session.on('secure', common.mustCall(() => {
-        const stream = session.openStream({ halfOpen: true });
-        stream.end('Hi!');
-      }));
+    server.on('session', common.mustCall(async (session) => {
+      const stream = await session.openStream({ halfOpen: true });
+      stream.end('Hi!');
     }));
 
     server.on('close', common.mustCall());

test/parallel/test-quic-client-empty-preferred-address.js

@@ -42,7 +42,7 @@ const options = { key, cert, ca, alpn: 'zzz' };
     preferredAddressPolicy: 'accept',
   });
 
-  const stream = clientSession.openStream();
+  const stream = await clientSession.openStream();
   stream.end('hello');
 
   await Promise.all([

test/parallel/test-quic-client-server.js

@@ -129,7 +129,7 @@ client.on('close', common.mustCall(onSocketClose.bind(client)));
         });
       }));
 
-    session.on('secure', common.mustCall((servername, alpn, cipher) => {
+    session.on('secure', common.mustCall(async (servername, alpn, cipher) => {
       debug('QuicServerSession TLS Handshake Complete');
       debug('  Server name: %s', servername);
       debug('  ALPN: %s', alpn);
@@ -143,7 +143,7 @@ client.on('close', common.mustCall(onSocketClose.bind(client)));
       assert(session.authenticated);
       assert.strictEqual(session.authenticationError, undefined);
 
-      const uni = session.openStream({ halfOpen: true });
+      const uni = await session.openStream({ halfOpen: true });
       assert(uni.unidirectional);
       assert(!uni.bidirectional);
       assert(uni.serverInitiated);
@@ -221,8 +221,8 @@ client.on('close', common.mustCall(onSocketClose.bind(client)));
         name: 'Error'
       };
       assert.throws(() => session.ping(), err);
-      assert.throws(() => session.openStream(), err);
       assert.throws(() => session.updateKey(), err);
+      assert.rejects(() => session.openStream(), err);
     }));
   }));
 
@@ -264,7 +264,7 @@ client.on('close', common.mustCall(onSocketClose.bind(client)));
     debug('  Params: %s', params.toString('hex'));
   }, 2));
 
-  req.on('secure', common.mustCall((servername, alpn, cipher) => {
+  req.on('secure', common.mustCall(async (servername, alpn, cipher) => {
     debug('QuicClientSession TLS Handshake Complete');
     debug('  Server name: %s', servername);
     debug('  ALPN: %s', alpn);
@@ -308,7 +308,7 @@ client.on('close', common.mustCall(onSocketClose.bind(client)));
     }
 
     const file = fs.createReadStream(__filename);
-    const stream = req.openStream();
+    const stream = await req.openStream();
     file.pipe(stream);
     let data = '';
     stream.resume();

test/parallel/test-quic-errors-quicsession-openstream.js

@@ -7,6 +7,7 @@ if (!common.hasQuic)
 // Test errors thrown when openStream is called incorrectly
 // or is not permitted
 
+const { once } = require('events');
 const { createHook } = require('async_hooks');
 const assert = require('assert');
 const { createQuicSocket } = require('net');
@@ -18,18 +19,12 @@ createHook({
   }
 }).enable();
 
-const Countdown = require('../common/countdown');
 const { key, cert, ca } = require('../common/quic');
 
 const options = { key, cert, ca, alpn: 'zzz', maxStreamsUni: 0 };
 const server = createQuicSocket({ server: options });
 const client = createQuicSocket({ client: options });
 
-const countdown = new Countdown(1, () => {
-  server.close();
-  client.close();
-});
-
 server.on('close', common.mustCall());
 client.on('close', common.mustCall());
 
@@ -44,40 +39,48 @@ client.on('close', common.mustCall());
     port: server.endpoints[0].address.port
   });
 
-  ['z', 1, {}, [], null, Infinity, 1n].forEach((i) => {
-    assert.throws(
-      () => req.openStream({ halfOpen: i }),
-      { code: 'ERR_INVALID_ARG_TYPE' }
-    );
-  });
+  for (const halfOpen of ['z', 1, {}, [], null, Infinity, 1n]) {
+    await assert.rejects(req.openStream({ halfOpen }), {
+      code: 'ERR_INVALID_ARG_TYPE'
+    });
+  }
 
-  ['', 1n, {}, [], false, 'zebra'].forEach((defaultEncoding) => {
-    assert.throws(() => req.openStream({ defaultEncoding }), {
+  for (const defaultEncoding of ['', 1n, {}, [], false, 'zebra']) {
+    await assert.rejects(req.openStream({ defaultEncoding }), {
       code: 'ERR_INVALID_ARG_VALUE'
     });
-  });
+  }
 
-  [-1, Number.MAX_SAFE_INTEGER + 1].forEach((highWaterMark) => {
-    assert.throws(() => req.openStream({ highWaterMark }), {
+  for (const highWaterMark of [-1, Number.MAX_SAFE_INTEGER + 1]) {
+    await assert.rejects(req.openStream({ highWaterMark }), {
       code: 'ERR_OUT_OF_RANGE'
     });
-  });
+  }
 
-  ['a', 1n, [], {}, false].forEach((highWaterMark) => {
-    assert.throws(() => req.openStream({ highWaterMark }), {
+  for (const highWaterMark of ['a', 1n, [], {}, false]) {
+    await assert.rejects(req.openStream({ highWaterMark }), {
       code: 'ERR_INVALID_ARG_TYPE'
     });
-  });
+  }
 
   // Unidirectional streams are not allowed. openStream will succeeed
   // but the stream will be destroyed immediately. The underlying
   // QuicStream C++ handle will not be created.
-  req.openStream({
-    halfOpen: true,
-    highWaterMark: 10,
-    defaultEncoding: 'utf16le'
-  }).on('error', common.expectsError({
-    code: 'ERR_OPERATION_FAILED'
-  })).on('error', common.mustCall(() => countdown.dec()));
+  await assert.rejects(
+    req.openStream({
+      halfOpen: true,
+      highWaterMark: 10,
+      defaultEncoding: 'utf16le'
+    }), {
+      code: 'ERR_OPERATION_FAILED'
+    });
+
+  server.close();
+  client.close();
+
+  await Promise.all([
+    once(server, 'close'),
+    once(client, 'close')
+  ]);
 
 })().then(common.mustCall());