quic: implement QuicSession close as promise

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

Author: jasnell

doc/api/quic.md

@@ -352,7 +352,7 @@ Binds the `QuicEndpoint` if it has not already been bound. User code will
 not typically be responsible for binding a `QuicEndpoint` as the owning
 `QuicSocket` will do that automatically.
 
-* `options` {object}
+* `options` {Object}
   * `signal` {AbortSignal} Optionally allows the `bind()` to be canceled
     using an `AbortController`.
 * Returns: {Promise}
@@ -821,17 +821,16 @@ added: REPLACEME
 
 Information about the cipher algorithm selected for the session.
 
-#### quicsession.close(\[callback\])
+#### quicsession.close()
 <!-- YAML
 added: REPLACEME
 -->
 
-* `callback` {Function} Callback invoked when the close operation is completed
-
 Begins a graceful close of the `QuicSession`. Existing `QuicStream` instances
 will be permitted to close naturally. New `QuicStream` instances will not be
 permitted. Once all `QuicStream` instances have closed, the `QuicSession`
-instance will be destroyed.
+instance will be destroyed. Returns a `Promise` that is resolved once the
+`QuicSession` instance is destroyed.
 
 #### quicsession.closeCode
 <!-- YAML

lib/internal/quic/core.js

@@ -18,6 +18,7 @@ const {
   Number,
   Promise,
   PromiseAll,
+  PromiseReject,
   RegExp,
   Set,
   Symbol,
@@ -98,7 +99,6 @@ const {
 const {
   codes: {
     ERR_INVALID_ARG_TYPE,
-    ERR_INVALID_CALLBACK,
     ERR_INVALID_STATE,
     ERR_OPERATION_FAILED,
     ERR_QUIC_FAILED_TO_CREATE_SESSION,
@@ -795,7 +795,7 @@ class QuicEndpoint {
 
   [kClose]() {
     if (this.destroyed) {
-      return Promise.reject(
+      return PromiseReject(
         new ERR_INVALID_STATE('QuicEndpoint is already destroyed'));
     }
     const promise = deferredClosePromise(this[kInternalState]);
@@ -1392,7 +1392,7 @@ class QuicSocket extends EventEmitter {
 
   [kClose]() {
     if (this.destroyed) {
-      return Promise.reject(
+      return PromiseReject(
         new ERR_INVALID_STATE('QuicSocket is already destroyed'));
     }
     const state = this[kInternalState];
@@ -1413,14 +1413,13 @@ class QuicSocket extends EventEmitter {
       return promise;
     }
 
-    // Otherwise, loop through each of the known sessions
-    // and close them.
-    // TODO(@jasnell): These will be promises soon, but we
-    // do not want to await them.
-    for (const session of state.sessions)
-      session.close();
-
-    return promise;
+    // Otherwise, loop through each of the known sessions and close them.
+    const reqs = [promise];
+    for (const session of state.sessions) {
+      reqs.push(session.close()
+                       .catch((error) => this.destroy(error)));
+    }
+    return PromiseAll(reqs);
   }
 
   // Initiate an abrupt close and destruction of the QuicSocket.
@@ -1656,7 +1655,9 @@ class QuicSession extends EventEmitter {
     cipherVersion: undefined,
     closeCode: NGTCP2_NO_ERROR,
     closeFamily: QUIC_ERROR_APPLICATION,
-    closing: false,
+    closePromise: undefined,
+    closePromiseResolve: undefined,
+    closePromiseReject: undefined,
     destroyed: false,
     earlyData: false,
     handshakeComplete: false,
@@ -1768,17 +1769,6 @@ class QuicSession extends EventEmitter {
     this.destroy(err);
   }
 
-  // Causes the QuicSession to be immediately destroyed, but with
-  // additional metadata set.
-  [kDestroy](code, family, silent, statelessReset) {
-    const state = this[kInternalState];
-    state.closeCode = code;
-    state.closeFamily = family;
-    state.silentClose = silent;
-    state.statelessReset = statelessReset;
-    this.destroy();
-  }
-
   // Closes the specified stream with the given code. The
   // QuicStream object will be destroyed.
   [kStreamClose](id, code) {
@@ -1789,23 +1779,23 @@ class QuicSession extends EventEmitter {
     stream.destroy();
   }
 
-  // Delivers a block of headers to the appropriate QuicStream
-  // instance. This will only be called if the ALPN selected
-  // is known to support headers.
-  [kHeaders](id, headers, kind, push_id) {
+  [kStreamReset](id, code) {
     const stream = this[kInternalState].streams.get(id);
     if (stream === undefined)
       return;
 
-    stream[kHeaders](headers, kind, push_id);
+    stream[kStreamReset](code);
   }
 
-  [kStreamReset](id, code) {
+  // Delivers a block of headers to the appropriate QuicStream
+  // instance. This will only be called if the ALPN selected
+  // is known to support headers.
+  [kHeaders](id, headers, kind, push_id) {
     const stream = this[kInternalState].streams.get(id);
     if (stream === undefined)
       return;
 
-    stream[kStreamReset](code);
+    stream[kHeaders](headers, kind, push_id);
   }
 
   [kInspect](depth, options) {
@@ -1850,8 +1840,13 @@ class QuicSession extends EventEmitter {
     if (!this[kHandshakePost]())
       return;
 
-    process.nextTick(
-      emit.bind(this, 'secure', servername, alpn, this.cipher));
+    process.nextTick(() => {
+      try {
+        this.emit('secure', servername, alpn, this.cipher);
+      } catch (error) {
+        this.destroy(error);
+      }
+    });
   }
 
   // Non-op for the default case. QuicClientSession
@@ -1863,10 +1858,10 @@ class QuicSession extends EventEmitter {
 
   [kRemoveStream](stream) {
     this[kInternalState].streams.delete(stream.id);
+    this[kMaybeDestroy]();
   }
 
   [kAddStream](id, stream) {
-    stream.once('close', this[kMaybeDestroy].bind(this));
     this[kInternalState].streams.set(id, stream);
   }
 
@@ -1875,49 +1870,55 @@ class QuicSession extends EventEmitter {
   // informationational notification. It is not called on
   // server QuicSession instances.
   [kUsePreferredAddress](address) {
-    process.nextTick(
-      emit.bind(this, 'usePreferredAddress', address));
+    process.nextTick(() => {
+      try {
+        this.emit('usePreferredAddress', address);
+      } catch (error) {
+        this.destroy(error);
+      }
+    });
+  }
+
+  close() {
+    return this[kInternalState].closePromise || this[kClose]();
+  }
+
+  [kClose]() {
+    if (this.destroyed) {
+      return PromiseReject(
+        new ERR_INVALID_STATE('QuicSession is already destroyed'));
+    }
+    const promise = deferredClosePromise(this[kInternalState]);
+    if (!this[kMaybeDestroy]()) {
+      this[kHandle].gracefulClose();
+    }
+    return promise;
+  }
+
+  get closing() {
+    return this[kInternalState].closePromise !== undefined;
   }
 
   // The QuicSession will be destroyed if close() has been
   // called and there are no remaining streams
   [kMaybeDestroy]() {
     const state = this[kInternalState];
-    if (state.closing && state.streams.size === 0) {
+    if (this.closing && state.streams.size === 0) {
       this.destroy();
       return true;
     }
     return false;
   }
 
-  // Closing allows any existing QuicStream's to gracefully
-  // complete while disallowing any new QuicStreams from being
-  // opened (in either direction). Calls to openStream() will
-  // fail, and new streams from the peer will be rejected/ignored.
-  close(callback) {
+  // Causes the QuicSession to be immediately destroyed, but with
+  // additional metadata set.
+  [kDestroy](code, family, silent, statelessReset) {
     const state = this[kInternalState];
-    if (state.destroyed) {
-      throw new ERR_INVALID_STATE(
-        `${this.constructor.name} is already destroyed`);
-    }
-
-    if (callback) {
-      if (typeof callback !== 'function')
-        throw new ERR_INVALID_CALLBACK();
-      this.once('close', callback);
-    }
-
-    // If we're already closing, do nothing else.
-    // Callback will be invoked once the session
-    // has been destroyed
-    if (state.closing)
-      return;
-    state.closing = true;
-
-    // If there are no active streams, we can close immediately,
-    // otherwise set the graceful close flag.
-    if (!this[kMaybeDestroy]())
-      this[kHandle].gracefulClose();
+    state.closeCode = code;
+    state.closeFamily = family;
+    state.silentClose = silent;
+    state.statelessReset = statelessReset;
+    this.destroy();
   }
 
   // Destroying synchronously shuts down and frees the
@@ -1939,7 +1940,6 @@ class QuicSession extends EventEmitter {
     if (state.destroyed)
       return;
     state.destroyed = true;
-    state.closing = false;
 
     // Destroy any pending streams immediately. These
     // are streams that have been created but have not
@@ -1974,7 +1974,13 @@ class QuicSession extends EventEmitter {
 
     // If we are destroying with an error, schedule the
     // error to be emitted on process.nextTick.
-    if (error) process.nextTick(emit.bind(this, 'error', error));
+    if (error) {
+      if (typeof state.closePromiseReject === 'function')
+        state.closePromiseReject(error);
+      process.nextTick(emit.bind(this, 'error', error));
+    } else if (typeof state.closePromiseResolve === 'function')
+      state.closePromiseResolve();
+
     process.nextTick(emit.bind(this, 'close'));
   }
 
@@ -2100,10 +2106,6 @@ class QuicSession extends EventEmitter {
     return this[kInternalState].destroyed;
   }
 
-  get closing() {
-    return this[kInternalState].closing;
-  }
-
   get closeCode() {
     const state = this[kInternalState];
     return {
@@ -2123,11 +2125,11 @@ class QuicSession extends EventEmitter {
 
   openStream(options) {
     const state = this[kInternalState];
-    if (state.destroyed) {
+    if (this.destroyed) {
       throw new ERR_INVALID_STATE(
         `${this.constructor.name} is already destroyed`);
     }
-    if (state.closing) {
+    if (this.closing) {
       throw new ERR_INVALID_STATE(
         `${this.constructor.name} is closing`);
     }
@@ -2255,11 +2257,11 @@ class QuicSession extends EventEmitter {
   updateKey() {
     const state = this[kInternalState];
     // Initiates a key update for the connection.
-    if (state.destroyed) {
+    if (this.destroyed) {
       throw new ERR_INVALID_STATE(
         `${this.constructor.name} is already destroyed`);
     }
-    if (state.closing) {
+    if (this.closing) {
       throw new ERR_INVALID_STATE(
         `${this.constructor.name} is closing`);
     }
@@ -2282,7 +2284,6 @@ class QuicSession extends EventEmitter {
     return this[kHandle].removeFromSocket();
   }
 }
-
 class QuicServerSession extends QuicSession {
   [kInternalServerState] = {
     contexts: []
@@ -2914,7 +2915,6 @@ class QuicStream extends Duplex {
 
   _destroy(error, callback) {
     const state = this[kInternalState];
-    state.session[kRemoveStream](this);
     const handle = this[kHandle];
     // Do not use handle after this point as the underlying C++
     // object has been destroyed. Any attempt to use the object
@@ -2925,6 +2925,7 @@ class QuicStream extends Duplex {
       state.stats = new BigInt64Array(handle.stats);
       handle.destroy();
     }
+    state.session[kRemoveStream](this);
     // The destroy callback must be invoked in a nextTick
     process.nextTick(() => callback(error));
   }