From afcfbadc4f3e51349855677b8bd0e72178ad0a99 Mon Sep 17 00:00:00 2001
From: James M Snell <jasnell@gmail.com>
Date: Mon, 16 Dec 2024 10:31:32 -0800
Subject: [PATCH] quic: working more on impl

---
 doc/api/quic.md                               |  60 +--
 lib/internal/quic/quic.js                     | 140 ++++--
 lib/internal/quic/state.js                    |  98 +++--
 lib/internal/quic/stats.js                    |  68 ++-
 src/quic/application.cc                       |  31 +-
 src/quic/application.h                        |   2 +-
 src/quic/data.cc                              |   3 +-
 src/quic/endpoint.cc                          |  31 +-
 src/quic/http3.cc                             |  67 ++-
 src/quic/logstream.cc                         |   2 +-
 src/quic/packet.cc                            |  23 +-
 src/quic/packet.h                             |   9 +-
 src/quic/session.cc                           | 272 ++++++++----
 src/quic/session.h                            |  58 ++-
 src/quic/streams.cc                           | 400 ++++++++++++++----
 src/quic/streams.h                            | 129 +++++-
 src/req_wrap-inl.h                            |   4 +-
 ...test-quic-internal-endpoint-stats-state.js |   8 +-
 18 files changed, 999 insertions(+), 406 deletions(-)

diff --git a/doc/api/quic.md b/doc/api/quic.md
index f23819548af5ee..af6883f61d2f49 100644
--- a/doc/api/quic.md
+++ b/doc/api/quic.md
@@ -480,20 +480,24 @@ added: REPLACEME
 
 * {quic.OnHandshakeCallback}
 
-### `session.openBidirectionalStream()`
+### `session.openBidirectionalStream([options])`
 
 <!-- YAML
 added: REPLACEME
 -->
 
+* `options` {Object}
+  * `headers` {Object}
 * Returns: {quic.QuicStream}
 
-### `session.openUnidirectionalStream()`
+### `session.openUnidirectionalStream([options])`
 
 <!-- YAML
 added: REPLACEME
 -->
 
+* `options` {Object}
+  * `headers` {Object
 * Returns: {quic.QuicStream}
 
 ### `session.path`
@@ -1008,15 +1012,15 @@ added: REPLACEME
 added: REPLACEME
 -->
 
-### `streamState.id`
+### `streamState.destroyed`
 
 <!-- YAML
 added: REPLACEME
 -->
 
-* {bigint}
+* {boolean}
 
-### `streamState.finSent`
+### `streamState.finReceived`
 
 <!-- YAML
 added: REPLACEME
@@ -1024,7 +1028,7 @@ added: REPLACEME
 
 * {boolean}
 
-### `streamState.finReceived`
+### `streamState.finSent`
 
 <!-- YAML
 added: REPLACEME
@@ -1032,7 +1036,7 @@ added: REPLACEME
 
 * {boolean}
 
-### `streamState.readEnded`
+### `streamState.hasReader`
 
 <!-- YAML
 added: REPLACEME
@@ -1040,15 +1044,15 @@ added: REPLACEME
 
 * {boolean}
 
-### `streamState.writeEnded`
+### `streamState.id`
 
 <!-- YAML
 added: REPLACEME
 -->
 
-* {boolean}
+* {bigint}
 
-### `streamState.destroyed`
+### `streamState.paused`
 
 <!-- YAML
 added: REPLACEME
@@ -1056,7 +1060,7 @@ added: REPLACEME
 
 * {boolean}
 
-### `streamState.paused`
+### `streamState.pending`
 
 <!-- YAML
 added: REPLACEME
@@ -1064,7 +1068,7 @@ added: REPLACEME
 
 * {boolean}
 
-### `streamState.reset`
+### `streamState.readEnded`
 
 <!-- YAML
 added: REPLACEME
@@ -1072,7 +1076,7 @@ added: REPLACEME
 
 * {boolean}
 
-### `streamState.hasReader`
+### `streamState.reset`
 
 <!-- YAML
 added: REPLACEME
@@ -1112,21 +1116,21 @@ added: REPLACEME
 
 * {boolean}
 
-## Class: `QuicStreamStats`
+### `streamState.writeEnded`
 
 <!-- YAML
 added: REPLACEME
 -->
 
-### `streamStats.isConnected`
+* {boolean}
+
+## Class: `QuicStreamStats`
 
 <!-- YAML
 added: REPLACEME
 -->
 
-* {bigint}
-
-### `streamStats.createdAt`
+### `streamStats.ackedAt`
 
 <!-- YAML
 added: REPLACEME
@@ -1134,7 +1138,7 @@ added: REPLACEME
 
 * {bigint}
 
-### `streamStats.receivedAt`
+### `streamStats.bytesReceived`
 
 <!-- YAML
 added: REPLACEME
@@ -1142,7 +1146,7 @@ added: REPLACEME
 
 * {bigint}
 
-### `streamStats.ackedAt`
+### `streamStats.bytesSent`
 
 <!-- YAML
 added: REPLACEME
@@ -1150,7 +1154,7 @@ added: REPLACEME
 
 * {bigint}
 
-### `streamStats.closingAt`
+### `streamStats.createdAt`
 
 <!-- YAML
 added: REPLACEME
@@ -1166,7 +1170,7 @@ added: REPLACEME
 
 * {bigint}
 
-### `streamStats.bytesReceived`
+### `streamStats.finalSize`
 
 <!-- YAML
 added: REPLACEME
@@ -1174,7 +1178,7 @@ added: REPLACEME
 
 * {bigint}
 
-### `streamStats.bytesSent`
+### `streamStats.isConnected`
 
 <!-- YAML
 added: REPLACEME
@@ -1206,7 +1210,15 @@ added: REPLACEME
 
 * {bigint}
 
-### `streamStats.finalSize`
+### `streamStats.openedAt`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* {bigint}
+
+### `streamStats.receivedAt`
 
 <!-- YAML
 added: REPLACEME
diff --git a/lib/internal/quic/quic.js b/lib/internal/quic/quic.js
index 2a485b81bc2f42..0a047f4e2cf034 100644
--- a/lib/internal/quic/quic.js
+++ b/lib/internal/quic/quic.js
@@ -86,6 +86,10 @@ const {
   kHandle: kSocketAddressHandle,
 } = require('internal/socketaddress');
 
+const {
+  isBlob,
+} = require('internal/blob');
+
 const {
   isKeyObject,
   isCryptoKey,
@@ -407,7 +411,7 @@ setCallbacks({
    * @param {number} status If context indicates an error, provides the error code.
    */
   onEndpointClose(context, status) {
-    debug('endpoint close callback', this[kOwner], status);
+    debug('endpoint close callback', status);
     this[kOwner][kFinishClose](context, status);
   },
   /**
@@ -429,7 +433,7 @@ setCallbacks({
    * @param {string} [reason]
    */
   onSessionClose(errorType, code, reason) {
-    debug('session close callback', this[kOwner], errorType, code, reason);
+    debug('session close callback', errorType, code, reason);
     this[kOwner][kFinishClose](errorType, code, reason);
   },
 
@@ -542,7 +546,7 @@ setCallbacks({
   onStreamClose(error) {
     // Called when the stream C++ handle has been closed.
     debug(`stream ${this[kOwner].id} closed callback with error: ${error}`);
-    this[kOwner].destroy(error);
+    this[kOwner][kFinishClose](error);
   },
 
   onStreamReset(error) {
@@ -606,9 +610,16 @@ class QuicStream {
     this.#state = new QuicStreamState(kPrivateConstructor, this.#handle.state);
     this.#reader = this.#handle.getReader();
 
-    debug(`stream ${this.id} created [${this.direction}]`);
+    if (this.pending) {
+      debug(`pending ${this.direction} stream created`);
+    } else {
+      debug(`${this.direction} stream ${this.id} created`);
+    }
   }
 
+  /** @type {boolean} */
+  get pending() { return this.#state.pending; }
+
   /** @type {OnBlockedCallback} */
   get onblocked() { return this.#onblocked; }
 
@@ -674,15 +685,19 @@ class QuicStream {
   /** @type {QuicSession} */
   get session() { return this.#session; }
 
-  /** @type {bigint} */
+  /**
+   * Returns the id for this stream. If the stream is destroyed or still pending,
+   * `undefined` will be returned.
+   * @type {bigint}
+   */
   get id() {
-    if (this.destroyed) return undefined;
+    if (this.destroyed || this.pending) return undefined;
     return this.#state.id;
   }
 
   /** @type {'bidi'|'uni'} */
   get direction() {
-    return this.#direction === 0 ? 'bidi' : 'uni';
+    return this.#direction === STREAM_DIRECTION_BIDIRECTIONAL ? 'bidi' : 'uni';
   }
 
   /** @returns {boolean} */
@@ -709,37 +724,43 @@ class QuicStream {
    */
   sendHeaders(headers) {
     validateObject(headers, 'headers');
-    debug('stream sending headers', headers);
+    if (this.pending) {
+      debug('pending stream enqueing headers', headers);
+    } else {
+      debug(`stream ${this.id} sending headers`, headers);
+    }
     // TODO(@jasnell): Support differentiating between early headers, primary headers, etc
     return this.#handle.sendHeaders(1, mapToHeaders(headers), 1);
   }
 
-  /**
-   * @param {*} [error]
-   * @returns {Promise<void>}
-   */
-  destroy(error) {
+  [kFinishClose](error) {
     if (this.destroyed) return this.#pendingClose.promise;
     if (error !== undefined) {
-      debug(`destroying stream ${this.id} with error: ${error}`);
+      if (this.pending) {
+        debug(`destroying pending stream with error: ${error}`);
+      } else {
+        debug(`destroying stream ${this.id} with error: ${error}`);
+      }
       this.#pendingClose.reject(error);
     } else {
-      debug(`destroying stream ${this.id} with no error`);
+      if (this.pending) {
+        debug('destroying pending stream with no error');
+      } else {
+        debug(`destroying stream ${this.id} with no error`);
+      }
       this.#pendingClose.resolve();
     }
     this.#stats[kFinishClose]();
     this.#state[kFinishClose]();
+    this.#session[kRemoveStream](this);
+    this.#session = undefined;
     this.#pendingClose.reject = undefined;
     this.#pendingClose.resolve = undefined;
     this.#onblocked = undefined;
     this.#onreset = undefined;
     this.#onheaders = undefined;
     this.#ontrailers = undefined;
-    this.#session[kRemoveStream](this);
-    this.#session = undefined;
-    this.#handle.destroy();
     this.#handle = undefined;
-    return this.#pendingClose.promise;
   }
 
   [kBlocked]() {
@@ -795,6 +816,7 @@ class QuicStream {
     return `Stream ${inspect({
       id: this.id,
       direction: this.direction,
+      pending: this.pending,
       stats: this.stats,
       state: this.state,
       session: this.session,
@@ -960,6 +982,9 @@ class QuicSession {
   /** @type {QuicEndpoint} */
   get endpoint() { return this.#endpoint; }
 
+  /** @type {boolean} */
+  get pending() { return !this.#state.isStreamOpenAllowed; }
+
   /**
    * The path is the local and remote addresses of the session.
    * @type {Path}
@@ -981,56 +1006,108 @@ class QuicSession {
   /**
    * @returns {QuicStream}
    */
-  openBidirectionalStream() {
+  openBidirectionalStream(options = kEmptyObject) {
     if (this.#isClosedOrClosing) {
-      throw new ERR_INVALID_STATE('Session is closed');
+      throw new ERR_INVALID_STATE('Session is closed. New streams cannot be opened.');
     }
-    if (!this.state.isStreamOpenAllowed) {
-      throw new ERR_QUIC_OPEN_STREAM_FAILED();
+    if (this.pending) {
+      debug('opening new pending bidirectional stream');
+    } else {
+      debug('opening new bidirectional stream');
     }
 
-    debug('opening new bidirectional stream');
+    validateObject(options, 'options');
+    const {
+      body,
+      headers,
+    } = options;
+    if (headers !== undefined) {
+      validateObject(headers, 'options.headers');
+    }
 
     const handle = this.#handle.openStream(STREAM_DIRECTION_BIDIRECTIONAL);
     if (handle === undefined) {
       throw new ERR_QUIC_OPEN_STREAM_FAILED();
     }
-    const stream = new QuicStream(kPrivateConstructor, handle, this, 0 /* Bidirectional */);
+
+    if (headers !== undefined) {
+      handle.sendHeaders(1, mapToHeaders(headers), 1);
+    }
+
+    const stream = new QuicStream(kPrivateConstructor, handle, this,
+                                  STREAM_DIRECTION_BIDIRECTIONAL);
     this.#streams.add(stream);
 
     if (onSessionOpenStreamChannel.hasSubscribers) {
       onSessionOpenStreamChannel.publish({
         stream,
         session: this,
+        options: { __proto__: null, ...options },
       });
     }
     return stream;
   }
 
+  #validateBody(body) {
+    // TODO(@jasnell): Support streaming sources
+    if (body === undefined ||
+        isArrayBuffer(body) ||
+        isArrayBufferView(body) ||
+        isBlob(body)) {
+      return body;
+    }
+
+    throw new ERR_INVALID_ARG_TYPE('options.body', [
+      'ArrayBuffer',
+      'ArrayBufferView',
+      'Blob',
+    ], body);
+  }
+
   /**
    * @returns {QuicStream}
    */
-  openUnidirectionalStream() {
+  openUnidirectionalStream(options = kEmptyObject) {
     if (this.#isClosedOrClosing) {
-      throw new ERR_INVALID_STATE('Session is closed');
+      throw new ERR_INVALID_STATE('Session is closed. New streams cannot be opened.');
     }
-    if (!this.state.isStreamOpenAllowed) {
-      throw new ERR_QUIC_OPEN_STREAM_FAILED();
+    if (this.pending) {
+      debug('opening new pending unidirectional stream');
+    } else {
+      debug('opening new unidirectional stream');
     }
 
-    debug('opening new unidirectional stream');
+    validateObject(options, 'options');
+    const {
+      body,
+      headers,
+    } = options;
+    if (headers !== undefined) {
+      validateObject(headers, 'options.headers');
+    }
+    const validatedBody = this.#validateBody(body);
 
     const handle = this.#handle.openStream(STREAM_DIRECTION_UNIDIRECTIONAL);
     if (handle === undefined) {
       throw new ERR_QUIC_OPEN_STREAM_FAILED();
     }
-    const stream = new QuicStream(kPrivateConstructor, handle, this, 1 /* Unidirectional */);
+
+    if (headers !== undefined) {
+      handle.sendHeaders(1, mapToHeaders(headers), 1);
+    }
+    if (validatedBody !== undefined) {
+      headers.attachSource(validatedBody);
+    }
+
+    const stream = new QuicStream(kPrivateConstructor, handle, this,
+                                  STREAM_DIRECTION_UNIDIRECTIONAL);
     this.#streams.add(stream);
 
     if (onSessionOpenStreamChannel.hasSubscribers) {
       onSessionOpenStreamChannel.publish({
         stream,
         session: this,
+        options: { __proto__: null, ...options },
       });
     }
 
@@ -2007,6 +2084,7 @@ class QuicEndpoint {
 
   [kFinishClose](context, status) {
     if (this.#handle === undefined) return;
+    debug('endpoint is finishing close', context, status);
     this.#handle = undefined;
     this.#stats[kFinishClose]();
     this.#state[kFinishClose]();
diff --git a/lib/internal/quic/state.js b/lib/internal/quic/state.js
index a46bcf53c601c6..053fd01ba922c9 100644
--- a/lib/internal/quic/state.js
+++ b/lib/internal/quic/state.js
@@ -14,7 +14,6 @@ const {
   codes: {
     ERR_ILLEGAL_CONSTRUCTOR,
     ERR_INVALID_ARG_TYPE,
-    ERR_INVALID_STATE,
   },
 } = require('internal/errors');
 
@@ -64,11 +63,11 @@ const {
   IDX_STATE_ENDPOINT_PENDING_CALLBACKS,
 
   IDX_STATE_STREAM_ID,
+  IDX_STATE_STREAM_PENDING,
   IDX_STATE_STREAM_FIN_SENT,
   IDX_STATE_STREAM_FIN_RECEIVED,
   IDX_STATE_STREAM_READ_ENDED,
   IDX_STATE_STREAM_WRITE_ENDED,
-  IDX_STATE_STREAM_DESTROYED,
   IDX_STATE_STREAM_PAUSED,
   IDX_STATE_STREAM_RESET,
   IDX_STATE_STREAM_HAS_READER,
@@ -100,11 +99,11 @@ assert(IDX_STATE_ENDPOINT_CLOSING !== undefined);
 assert(IDX_STATE_ENDPOINT_BUSY !== undefined);
 assert(IDX_STATE_ENDPOINT_PENDING_CALLBACKS !== undefined);
 assert(IDX_STATE_STREAM_ID !== undefined);
+assert(IDX_STATE_STREAM_PENDING !== undefined);
 assert(IDX_STATE_STREAM_FIN_SENT !== undefined);
 assert(IDX_STATE_STREAM_FIN_RECEIVED !== undefined);
 assert(IDX_STATE_STREAM_READ_ENDED !== undefined);
 assert(IDX_STATE_STREAM_WRITE_ENDED !== undefined);
-assert(IDX_STATE_STREAM_DESTROYED !== undefined);
 assert(IDX_STATE_STREAM_PAUSED !== undefined);
 assert(IDX_STATE_STREAM_RESET !== undefined);
 assert(IDX_STATE_STREAM_HAS_READER !== undefined);
@@ -131,39 +130,33 @@ class QuicEndpointState {
     this.#handle = new DataView(buffer);
   }
 
-  #assertNotClosed() {
-    if (this.#handle.byteLength === 0) {
-      throw new ERR_INVALID_STATE('Endpoint is closed');
-    }
-  }
-
   /** @type {boolean} */
   get isBound() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_ENDPOINT_BOUND);
   }
 
   /** @type {boolean} */
   get isReceiving() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_ENDPOINT_RECEIVING);
   }
 
   /** @type {boolean} */
   get isListening() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_ENDPOINT_LISTENING);
   }
 
   /** @type {boolean} */
   get isClosing() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_ENDPOINT_CLOSING);
   }
 
   /** @type {boolean} */
   get isBusy() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_ENDPOINT_BUSY);
   }
 
@@ -174,7 +167,7 @@ class QuicEndpointState {
    * @type {bigint}
    */
   get pendingCallbacks() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return DataViewPrototypeGetBigUint64(this.#handle, IDX_STATE_ENDPOINT_PENDING_CALLBACKS);
   }
 
@@ -244,123 +237,117 @@ class QuicSessionState {
     this.#handle = new DataView(buffer);
   }
 
-  #assertNotClosed() {
-    if (this.#handle.byteLength === 0) {
-      throw new ERR_INVALID_STATE('Session is closed');
-    }
-  }
-
   /** @type {boolean} */
   get hasPathValidationListener() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_SESSION_PATH_VALIDATION);
   }
 
   /** @type {boolean} */
   set hasPathValidationListener(val) {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return;
     DataViewPrototypeSetUint8(this.#handle, IDX_STATE_SESSION_PATH_VALIDATION, val ? 1 : 0);
   }
 
   /** @type {boolean} */
   get hasVersionNegotiationListener() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_SESSION_VERSION_NEGOTIATION);
   }
 
   /** @type {boolean} */
   set hasVersionNegotiationListener(val) {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return;
     DataViewPrototypeSetUint8(this.#handle, IDX_STATE_SESSION_VERSION_NEGOTIATION, val ? 1 : 0);
   }
 
   /** @type {boolean} */
   get hasDatagramListener() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_SESSION_DATAGRAM);
   }
 
   /** @type {boolean} */
   set hasDatagramListener(val) {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return;
     DataViewPrototypeSetUint8(this.#handle, IDX_STATE_SESSION_DATAGRAM, val ? 1 : 0);
   }
 
   /** @type {boolean} */
   get hasSessionTicketListener() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_SESSION_SESSION_TICKET);
   }
 
   /** @type {boolean} */
   set hasSessionTicketListener(val) {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return;
     DataViewPrototypeSetUint8(this.#handle, IDX_STATE_SESSION_SESSION_TICKET, val ? 1 : 0);
   }
 
   /** @type {boolean} */
   get isClosing() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_SESSION_CLOSING);
   }
 
   /** @type {boolean} */
   get isGracefulClose() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_SESSION_GRACEFUL_CLOSE);
   }
 
   /** @type {boolean} */
   get isSilentClose() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_SESSION_SILENT_CLOSE);
   }
 
   /** @type {boolean} */
   get isStatelessReset() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_SESSION_STATELESS_RESET);
   }
 
   /** @type {boolean} */
   get isDestroyed() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_SESSION_DESTROYED);
   }
 
   /** @type {boolean} */
   get isHandshakeCompleted() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_SESSION_HANDSHAKE_COMPLETED);
   }
 
   /** @type {boolean} */
   get isHandshakeConfirmed() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_SESSION_HANDSHAKE_CONFIRMED);
   }
 
   /** @type {boolean} */
   get isStreamOpenAllowed() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_SESSION_STREAM_OPEN_ALLOWED);
   }
 
   /** @type {boolean} */
   get isPrioritySupported() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_SESSION_PRIORITY_SUPPORTED);
   }
 
   /** @type {boolean} */
   get isWrapped() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_SESSION_WRAPPED);
   }
 
   /** @type {bigint} */
   get lastDatagramId() {
-    this.#assertNotClosed();
+    if (this.#handle.byteLength === 0) return undefined;
     return DataViewPrototypeGetBigUint64(this.#handle, IDX_STATE_SESSION_LAST_DATAGRAM_ID);
   }
 
@@ -450,86 +437,103 @@ class QuicStreamState {
 
   /** @type {bigint} */
   get id() {
+    if (this.#handle.byteLength === 0) return undefined;
     return DataViewPrototypeGetBigInt64(this.#handle, IDX_STATE_STREAM_ID);
   }
 
+  /** @type {boolean} */
+  get pending() {
+    if (this.#handle.byteLength === 0) return undefined;
+    return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_STREAM_PENDING);
+  }
+
   /** @type {boolean} */
   get finSent() {
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_STREAM_FIN_SENT);
   }
 
   /** @type {boolean} */
   get finReceived() {
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_STREAM_FIN_RECEIVED);
   }
 
   /** @type {boolean} */
   get readEnded() {
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_STREAM_READ_ENDED);
   }
 
   /** @type {boolean} */
   get writeEnded() {
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_STREAM_WRITE_ENDED);
   }
 
-  /** @type {boolean} */
-  get destroyed() {
-    return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_STREAM_DESTROYED);
-  }
-
   /** @type {boolean} */
   get paused() {
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_STREAM_PAUSED);
   }
 
   /** @type {boolean} */
   get reset() {
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_STREAM_RESET);
   }
 
   /** @type {boolean} */
   get hasReader() {
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_STREAM_HAS_READER);
   }
 
   /** @type {boolean} */
   get wantsBlock() {
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_STREAM_WANTS_BLOCK);
   }
 
   /** @type {boolean} */
   set wantsBlock(val) {
+    if (this.#handle.byteLength === 0) return;
     DataViewPrototypeSetUint8(this.#handle, IDX_STATE_STREAM_WANTS_BLOCK, val ? 1 : 0);
   }
 
   /** @type {boolean} */
   get wantsHeaders() {
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_STREAM_WANTS_HEADERS);
   }
 
   /** @type {boolean} */
   set wantsHeaders(val) {
+    if (this.#handle.byteLength === 0) return;
     DataViewPrototypeSetUint8(this.#handle, IDX_STATE_STREAM_WANTS_HEADERS, val ? 1 : 0);
   }
 
   /** @type {boolean} */
   get wantsReset() {
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_STREAM_WANTS_RESET);
   }
 
   /** @type {boolean} */
   set wantsReset(val) {
+    if (this.#handle.byteLength === 0) return;
     DataViewPrototypeSetUint8(this.#handle, IDX_STATE_STREAM_WANTS_RESET, val ? 1 : 0);
   }
 
   /** @type {boolean} */
   get wantsTrailers() {
+    if (this.#handle.byteLength === 0) return undefined;
     return !!DataViewPrototypeGetUint8(this.#handle, IDX_STATE_STREAM_WANTS_TRAILERS);
   }
 
   /** @type {boolean} */
   set wantsTrailers(val) {
+    if (this.#handle.byteLength === 0) return;
     DataViewPrototypeSetUint8(this.#handle, IDX_STATE_STREAM_WANTS_TRAILERS, val ? 1 : 0);
   }
 
@@ -542,11 +546,11 @@ class QuicStreamState {
     return {
       __proto__: null,
       id: `${this.id}`,
+      pending: this.pending,
       finSent: this.finSent,
       finReceived: this.finReceived,
       readEnded: this.readEnded,
       writeEnded: this.writeEnded,
-      destroyed: this.destroyed,
       paused: this.paused,
       reset: this.reset,
       hasReader: this.hasReader,
@@ -572,11 +576,11 @@ class QuicStreamState {
 
     return `QuicStreamState ${inspect({
       id: this.id,
+      pending: this.pending,
       finSent: this.finSent,
       finReceived: this.finReceived,
       readEnded: this.readEnded,
       writeEnded: this.writeEnded,
-      destroyed: this.destroyed,
       paused: this.paused,
       reset: this.reset,
       hasReader: this.hasReader,
diff --git a/lib/internal/quic/stats.js b/lib/internal/quic/stats.js
index 3ac9523d9aeca4..425b1c62841020 100644
--- a/lib/internal/quic/stats.js
+++ b/lib/internal/quic/stats.js
@@ -17,6 +17,7 @@ const {
 } = require('internal/errors');
 
 const { inspect } = require('internal/util/inspect');
+const assert = require('internal/assert');
 
 const {
   kFinishClose,
@@ -76,9 +77,9 @@ const {
   IDX_STATS_SESSION_DATAGRAMS_LOST,
 
   IDX_STATS_STREAM_CREATED_AT,
+  IDX_STATS_STREAM_OPENED_AT,
   IDX_STATS_STREAM_RECEIVED_AT,
   IDX_STATS_STREAM_ACKED_AT,
-  IDX_STATS_STREAM_CLOSING_AT,
   IDX_STATS_STREAM_DESTROYED_AT,
   IDX_STATS_STREAM_BYTES_RECEIVED,
   IDX_STATS_STREAM_BYTES_SENT,
@@ -88,6 +89,57 @@ const {
   IDX_STATS_STREAM_FINAL_SIZE,
 } = internalBinding('quic');
 
+assert(IDX_STATS_ENDPOINT_CREATED_AT !== undefined);
+assert(IDX_STATS_ENDPOINT_DESTROYED_AT !== undefined);
+assert(IDX_STATS_ENDPOINT_BYTES_RECEIVED !== undefined);
+assert(IDX_STATS_ENDPOINT_BYTES_SENT !== undefined);
+assert(IDX_STATS_ENDPOINT_PACKETS_RECEIVED !== undefined);
+assert(IDX_STATS_ENDPOINT_PACKETS_SENT !== undefined);
+assert(IDX_STATS_ENDPOINT_SERVER_SESSIONS !== undefined);
+assert(IDX_STATS_ENDPOINT_CLIENT_SESSIONS !== undefined);
+assert(IDX_STATS_ENDPOINT_SERVER_BUSY_COUNT !== undefined);
+assert(IDX_STATS_ENDPOINT_RETRY_COUNT !== undefined);
+assert(IDX_STATS_ENDPOINT_VERSION_NEGOTIATION_COUNT !== undefined);
+assert(IDX_STATS_ENDPOINT_STATELESS_RESET_COUNT !== undefined);
+assert(IDX_STATS_ENDPOINT_IMMEDIATE_CLOSE_COUNT !== undefined);
+assert(IDX_STATS_SESSION_CREATED_AT !== undefined);
+assert(IDX_STATS_SESSION_CLOSING_AT !== undefined);
+assert(IDX_STATS_SESSION_DESTROYED_AT !== undefined);
+assert(IDX_STATS_SESSION_HANDSHAKE_COMPLETED_AT !== undefined);
+assert(IDX_STATS_SESSION_HANDSHAKE_CONFIRMED_AT !== undefined);
+assert(IDX_STATS_SESSION_GRACEFUL_CLOSING_AT !== undefined);
+assert(IDX_STATS_SESSION_BYTES_RECEIVED !== undefined);
+assert(IDX_STATS_SESSION_BYTES_SENT !== undefined);
+assert(IDX_STATS_SESSION_BIDI_IN_STREAM_COUNT !== undefined);
+assert(IDX_STATS_SESSION_BIDI_OUT_STREAM_COUNT !== undefined);
+assert(IDX_STATS_SESSION_UNI_IN_STREAM_COUNT !== undefined);
+assert(IDX_STATS_SESSION_UNI_OUT_STREAM_COUNT !== undefined);
+assert(IDX_STATS_SESSION_LOSS_RETRANSMIT_COUNT !== undefined);
+assert(IDX_STATS_SESSION_MAX_BYTES_IN_FLIGHT !== undefined);
+assert(IDX_STATS_SESSION_BYTES_IN_FLIGHT !== undefined);
+assert(IDX_STATS_SESSION_BLOCK_COUNT !== undefined);
+assert(IDX_STATS_SESSION_CWND !== undefined);
+assert(IDX_STATS_SESSION_LATEST_RTT !== undefined);
+assert(IDX_STATS_SESSION_MIN_RTT !== undefined);
+assert(IDX_STATS_SESSION_RTTVAR !== undefined);
+assert(IDX_STATS_SESSION_SMOOTHED_RTT !== undefined);
+assert(IDX_STATS_SESSION_SSTHRESH !== undefined);
+assert(IDX_STATS_SESSION_DATAGRAMS_RECEIVED !== undefined);
+assert(IDX_STATS_SESSION_DATAGRAMS_SENT !== undefined);
+assert(IDX_STATS_SESSION_DATAGRAMS_ACKNOWLEDGED !== undefined);
+assert(IDX_STATS_SESSION_DATAGRAMS_LOST !== undefined);
+assert(IDX_STATS_STREAM_CREATED_AT !== undefined);
+assert(IDX_STATS_STREAM_OPENED_AT !== undefined);
+assert(IDX_STATS_STREAM_RECEIVED_AT !== undefined);
+assert(IDX_STATS_STREAM_ACKED_AT !== undefined);
+assert(IDX_STATS_STREAM_DESTROYED_AT !== undefined);
+assert(IDX_STATS_STREAM_BYTES_RECEIVED !== undefined);
+assert(IDX_STATS_STREAM_BYTES_SENT !== undefined);
+assert(IDX_STATS_STREAM_MAX_OFFSET !== undefined);
+assert(IDX_STATS_STREAM_MAX_OFFSET_ACK !== undefined);
+assert(IDX_STATS_STREAM_MAX_OFFSET_RECV !== undefined);
+assert(IDX_STATS_STREAM_FINAL_SIZE !== undefined);
+
 class QuicEndpointStats {
   /** @type {BigUint64Array} */
   #handle;
@@ -522,6 +574,11 @@ class QuicStreamStats {
     return this.#handle[IDX_STATS_STREAM_CREATED_AT];
   }
 
+  /** @type {bigint} */
+  get openedAt() {
+    return this.#handle[IDX_STATS_STREAM_OPENED_AT];
+  }
+
   /** @type {bigint} */
   get receivedAt() {
     return this.#handle[IDX_STATS_STREAM_RECEIVED_AT];
@@ -532,11 +589,6 @@ class QuicStreamStats {
     return this.#handle[IDX_STATS_STREAM_ACKED_AT];
   }
 
-  /** @type {bigint} */
-  get closingAt() {
-    return this.#handle[IDX_STATS_STREAM_CLOSING_AT];
-  }
-
   /** @type {bigint} */
   get destroyedAt() {
     return this.#handle[IDX_STATS_STREAM_DESTROYED_AT];
@@ -583,9 +635,9 @@ class QuicStreamStats {
       // We need to convert the values to strings because JSON does not
       // support BigInts.
       createdAt: `${this.createdAt}`,
+      openedAt: `${this.openedAt}`,
       receivedAt: `${this.receivedAt}`,
       ackedAt: `${this.ackedAt}`,
-      closingAt: `${this.closingAt}`,
       destroyedAt: `${this.destroyedAt}`,
       bytesReceived: `${this.bytesReceived}`,
       bytesSent: `${this.bytesSent}`,
@@ -608,9 +660,9 @@ class QuicStreamStats {
     return `StreamStats ${inspect({
       connected: this.isConnected,
       createdAt: this.createdAt,
+      openedAt: this.openedAt,
       receivedAt: this.receivedAt,
       ackedAt: this.ackedAt,
-      closingAt: this.closingAt,
       destroyedAt: this.destroyedAt,
       bytesReceived: this.bytesReceived,
       bytesSent: this.bytesSent,
diff --git a/src/quic/application.cc b/src/quic/application.cc
index 9ac8f9b7f91175..01bba5090fea07 100644
--- a/src/quic/application.cc
+++ b/src/quic/application.cc
@@ -468,19 +468,17 @@ class DefaultApplication final : public Session::Application {
                          void* stream_user_data) override {
     Debug(&session(), "Default application receiving stream data");
 
-    BaseObjectPtr<Stream> stream;
+    BaseObjectWeakPtr<Stream> stream;
     if (stream_user_data == nullptr) {
       // This is the first time we're seeing this stream. Implicitly create it.
       stream = session().CreateStream(stream_id);
-      if (!stream) {
+      if (!stream) [[unlikely]] {
         // We couldn't actually create the stream for whatever reason.
         Debug(&session(), "Default application failed to create new stream");
         return false;
       }
-      // Let the JavaScript side know about the stream before we emit any data.
-      session().EmitStream(stream);
     } else {
-      stream = BaseObjectPtr<Stream>(Stream::From(stream_user_data));
+      stream = BaseObjectWeakPtr<Stream>(Stream::From(stream_user_data));
       if (!stream) {
         Debug(&session(),
               "Default application failed to get existing stream "
@@ -489,16 +487,7 @@ class DefaultApplication final : public Session::Application {
       }
     }
 
-    DCHECK(stream);
-
-    // If the stream is destroyed, we are going to silently ignore the
-    // data here.
-    if (stream->is_destroyed()) {
-      Debug(&session(),
-            "Data received for a stream that is already "
-            "destroyed. Ignoring.");
-      return true;
-    }
+    CHECK(stream);
 
     // Now we can actually receive the data! Woo!
     stream->ReceiveData(data, datalen, flags);
@@ -592,17 +581,17 @@ class DefaultApplication final : public Session::Application {
 
  private:
   void ScheduleStream(int64_t id) {
-    Debug(&session(), "Default application scheduling stream %" PRIi64, id);
-    auto stream = session().FindStream(id);
-    if (stream && !stream->is_destroyed()) {
+    if (auto stream = session().FindStream(id)) [[likely]] {
+      Debug(&session(), "Default application scheduling stream %" PRIi64, id);
       stream->Schedule(&stream_queue_);
     }
   }
 
   void UnscheduleStream(int64_t id) {
-    Debug(&session(), "Default application unscheduling stream %" PRIi64, id);
-    auto stream = session().FindStream(id);
-    if (stream && !stream->is_destroyed()) stream->Unschedule();
+    if (auto stream = session().FindStream(id)) [[likely]] {
+      Debug(&session(), "Default application unscheduling stream %" PRIi64, id);
+      stream->Unschedule();
+    }
   }
 
   Stream::Queue stream_queue_;
diff --git a/src/quic/application.h b/src/quic/application.h
index 0e39cd65f1820f..682a8e0be9ec1b 100644
--- a/src/quic/application.h
+++ b/src/quic/application.h
@@ -146,7 +146,7 @@ struct Session::Application::StreamData final {
   int64_t id = -1;
   int fin = 0;
   ngtcp2_vec data[kMaxVectorCount]{};
-  BaseObjectPtr<Stream> stream;
+  BaseObjectWeakPtr<Stream> stream;
 
   inline operator nghttp3_vec*() {
     return reinterpret_cast<nghttp3_vec*>(data);
diff --git a/src/quic/data.cc b/src/quic/data.cc
index 80f55a075ea23e..06120dd69591b1 100644
--- a/src/quic/data.cc
+++ b/src/quic/data.cc
@@ -257,8 +257,7 @@ std::optional<int> QuicError::crypto_error() const {
 }
 
 MaybeLocal<Value> QuicError::ToV8Value(Environment* env) const {
-  if ((type() == QuicError::Type::TRANSPORT &&
-       code() == NGTCP2_NO_ERROR) ||
+  if ((type() == QuicError::Type::TRANSPORT && code() == NGTCP2_NO_ERROR) ||
       (type() == QuicError::Type::APPLICATION &&
        code() == NGTCP2_APP_NOERROR) ||
       (type() == QuicError::Type::APPLICATION &&
diff --git a/src/quic/endpoint.cc b/src/quic/endpoint.cc
index a3b8dd13fdeacf..da2441b2c4a9ab 100644
--- a/src/quic/endpoint.cc
+++ b/src/quic/endpoint.cc
@@ -567,18 +567,17 @@ int Endpoint::UDP::Send(BaseObjectPtr<Packet> packet) {
   // which we don't want.
   packet->ClearWeak();
   packet->Dispatched();
-  int err = uv_udp_send(
-      packet->req(),
-      &impl_->handle_,
-      &buf,
-      1,
-      packet->destination().data(),
-      uv_udp_send_cb{[](uv_udp_send_t* req, int status) {
-        auto ptr = BaseObjectPtr<Packet>(
-            static_cast<Packet*>(ReqWrap<uv_udp_send_t>::from_req(req)));
-        ptr->env()->DecreaseWaitingRequestCounter();
-        ptr->Done(status);
-      }});
+  int err = uv_udp_send(packet->req(),
+                        &impl_->handle_,
+                        &buf,
+                        1,
+                        packet->destination().data(),
+                        uv_udp_send_cb{[](uv_udp_send_t* req, int status) {
+                          auto ptr = BaseObjectPtr<Packet>(static_cast<Packet*>(
+                              ReqWrap<uv_udp_send_t>::from_req(req)));
+                          ptr->env()->DecreaseWaitingRequestCounter();
+                          ptr->Done(status);
+                        }});
   if (err < 0) {
     // The packet failed.
     packet->Done(err);
@@ -1025,7 +1024,7 @@ BaseObjectPtr<Session> Endpoint::Connect(
     const Session::Options& options,
     std::optional<SessionTicket> session_ticket) {
   // If starting fails, the endpoint will be destroyed.
-  if (!Start()) return BaseObjectPtr<Session>();
+  if (!Start()) return {};
 
   Session::Config config(*this, options, local_address(), remote_address);
 
@@ -1044,7 +1043,7 @@ BaseObjectPtr<Session> Endpoint::Connect(
     THROW_ERR_INVALID_STATE(env(),
                             "Failed to create TLS context: %s",
                             tls_context->validation_error());
-    return BaseObjectPtr<Session>();
+    return {};
   }
   auto session =
       Session::Create(this, config, tls_context.get(), session_ticket);
@@ -1052,9 +1051,9 @@ BaseObjectPtr<Session> Endpoint::Connect(
     THROW_ERR_INVALID_STATE(env(),
                             "Failed to create TLS session: %s",
                             session->tls_session().validation_error());
-    return BaseObjectPtr<Session>();
+    return {};
   }
-  if (!session) return BaseObjectPtr<Session>();
+  if (!session) return {};
   session->set_wrapped();
 
   // Calling SendPendingData here triggers the session to send the initial
diff --git a/src/quic/http3.cc b/src/quic/http3.cc
index 4e53101f46d7b4..4ffbd26b8b1935 100644
--- a/src/quic/http3.cc
+++ b/src/quic/http3.cc
@@ -152,9 +152,12 @@ class Http3Application final : public Session::Application {
                          size_t datalen,
                          const Stream::ReceiveDataFlags& flags,
                          void* unused) override {
-    Debug(&session(), "HTTP/3 application received %zu bytes of data "
+    Debug(&session(),
+          "HTTP/3 application received %zu bytes of data "
           "on stream %" PRIi64 ". Is final? %d",
-          datalen, stream_id, flags.fin);
+          datalen,
+          stream_id,
+          flags.fin);
 
     ssize_t nread = nghttp3_conn_read_stream(
         *this, stream_id, data, datalen, flags.fin ? 1 : 0);
@@ -402,7 +405,6 @@ class Http3Application final : public Session::Application {
       }
 
       data->count = static_cast<size_t>(ret);
-
       if (data->id > 0 && data->id != control_stream_id_ &&
           data->id != qpack_dec_stream_id_ &&
           data->id != qpack_enc_stream_id_) {
@@ -464,11 +466,12 @@ class Http3Application final : public Session::Application {
   }
 
   void OnStreamClose(Stream* stream, uint64_t app_error_code) {
-    if (stream->is_destroyed()) return;
     if (app_error_code != NGHTTP3_H3_NO_ERROR) {
       Debug(&session(),
             "HTTP/3 application received stream close for stream %" PRIi64
-            " with code %" PRIu64, stream->id(), app_error_code);
+            " with code %" PRIu64,
+            stream->id(),
+            app_error_code);
     }
     auto direction = stream->direction();
     stream->Destroy(QuicError::ForApplication(app_error_code));
@@ -478,9 +481,8 @@ class Http3Application final : public Session::Application {
   void OnBeginHeaders(int64_t stream_id) {
     auto stream = session().FindStream(stream_id);
     // If the stream does not exist or is destroyed, ignore!
-    if (!stream || stream->is_destroyed()) [[unlikely]] {
+    if (!stream) [[unlikely]]
       return;
-    }
     Debug(&session(),
           "HTTP/3 application beginning initial block of headers for stream "
           "%" PRIi64,
@@ -491,9 +493,8 @@ class Http3Application final : public Session::Application {
   void OnReceiveHeader(int64_t stream_id, Http3Header&& header) {
     auto stream = session().FindStream(stream_id);
 
-    if (!stream || stream->is_destroyed()) [[unlikely]] {
+    if (!stream) [[unlikely]]
       return;
-    }
     if (header.name() == ":status" && header.value()[0] == '1') {
       Debug(&session(),
             "HTTP/3 application switching to hints headers for stream %" PRIi64,
@@ -511,9 +512,8 @@ class Http3Application final : public Session::Application {
 
   void OnEndHeaders(int64_t stream_id, int fin) {
     auto stream = session().FindStream(stream_id);
-    if (!stream || stream->is_destroyed()) [[unlikely]] {
+    if (!stream) [[unlikely]]
       return;
-    }
     Debug(&session(),
           "HTTP/3 application received end of headers for stream %" PRIi64,
           stream_id);
@@ -531,9 +531,8 @@ class Http3Application final : public Session::Application {
 
   void OnBeginTrailers(int64_t stream_id) {
     auto stream = session().FindStream(stream_id);
-    if (!stream || stream->is_destroyed()) [[unlikely]] {
+    if (!stream) [[unlikely]]
       return;
-    }
     Debug(&session(),
           "HTTP/3 application beginning block of trailers for stream %" PRIi64,
           stream_id);
@@ -542,9 +541,8 @@ class Http3Application final : public Session::Application {
 
   void OnReceiveTrailer(int64_t stream_id, Http3Header&& header) {
     auto stream = session().FindStream(stream_id);
-    if (!stream || stream->is_destroyed()) [[unlikely]] {
+    if (!stream) [[unlikely]]
       return;
-    }
     IF_QUIC_DEBUG(env()) {
       Debug(&session(),
             "Received header \"%s: %s\"",
@@ -556,9 +554,8 @@ class Http3Application final : public Session::Application {
 
   void OnEndTrailers(int64_t stream_id, int fin) {
     auto stream = session().FindStream(stream_id);
-    if (!stream || stream->is_destroyed()) [[unlikely]] {
+    if (!stream) [[unlikely]]
       return;
-    }
     Debug(&session(),
           "HTTP/3 application received end of trailers for stream %" PRIi64,
           stream_id);
@@ -575,9 +572,8 @@ class Http3Application final : public Session::Application {
 
   void OnEndStream(int64_t stream_id) {
     auto stream = session().FindStream(stream_id);
-    if (!stream || stream->is_destroyed()) [[unlikely]] {
+    if (!stream) [[unlikely]]
       return;
-    }
     Debug(&session(),
           "HTTP/3 application received end of stream for stream %" PRIi64,
           stream_id);
@@ -590,9 +586,8 @@ class Http3Application final : public Session::Application {
 
   void OnStopSending(int64_t stream_id, uint64_t app_error_code) {
     auto stream = session().FindStream(stream_id);
-    if (!stream || stream->is_destroyed()) [[unlikely]] {
+    if (!stream) [[unlikely]]
       return;
-    }
     Debug(&session(),
           "HTTP/3 application received stop sending for stream %" PRIi64,
           stream_id);
@@ -601,9 +596,8 @@ class Http3Application final : public Session::Application {
 
   void OnResetStream(int64_t stream_id, uint64_t app_error_code) {
     auto stream = session().FindStream(stream_id);
-    if (!stream || stream->is_destroyed()) [[unlikely]] {
+    if (!stream) [[unlikely]]
       return;
-    }
     Debug(&session(),
           "HTTP/3 application received reset stream for stream %" PRIi64,
           stream_id);
@@ -659,20 +653,16 @@ class Http3Application final : public Session::Application {
     return app;
   }
 
-  static BaseObjectPtr<Stream> FindOrCreateStream(nghttp3_conn* conn,
-                                                  Session* session,
-                                                  int64_t stream_id) {
-    BaseObjectPtr<Stream> stream = session->FindStream(stream_id);
-    if (stream) {
+  static BaseObjectWeakPtr<Stream> FindOrCreateStream(nghttp3_conn* conn,
+                                                      Session* session,
+                                                      int64_t stream_id) {
+    if (auto stream = session->FindStream(stream_id)) {
       return stream;
     }
-
-    stream = session->CreateStream(stream_id);
-    if (stream) {
+    if (auto stream = session->CreateStream(stream_id)) {
       return stream;
     }
-
-    return BaseObjectPtr<Stream>();
+    return {};
   }
 
 #define NGHTTP3_CALLBACK_SCOPE(name)                                           \
@@ -731,15 +721,12 @@ class Http3Application final : public Session::Application {
       return NGHTTP3_ERR_CALLBACK_FAILURE;
     }
     auto& session = app.session();
-    auto stream = FindOrCreateStream(conn, &session, stream_id);
-    if (!stream) [[unlikely]] {
-      return NGHTTP3_ERR_CALLBACK_FAILURE;
-    }
-    if (stream->is_destroyed()) [[unlikely]] {
+    if (auto stream = FindOrCreateStream(conn, &session, stream_id))
+        [[likely]] {
+      stream->ReceiveData(data, datalen, Stream::ReceiveDataFlags{});
       return NGTCP2_SUCCESS;
     }
-    stream->ReceiveData(data, datalen, Stream::ReceiveDataFlags{});
-    return NGTCP2_SUCCESS;
+    return NGHTTP3_ERR_CALLBACK_FAILURE;
   }
 
   static int on_deferred_consume(nghttp3_conn* conn,
diff --git a/src/quic/logstream.cc b/src/quic/logstream.cc
index cf8fd5fef347a5..ed84cad15ec950 100644
--- a/src/quic/logstream.cc
+++ b/src/quic/logstream.cc
@@ -40,7 +40,7 @@ BaseObjectPtr<LogStream> LogStream::Create(Environment* env) {
            ->InstanceTemplate()
            ->NewInstance(env->context())
            .ToLocal(&obj)) {
-    return BaseObjectPtr<LogStream>();
+    return {};
   }
   return MakeDetachedBaseObject<LogStream>(env, obj);
 }
diff --git a/src/quic/packet.cc b/src/quic/packet.cc
index 542fbdbf91eabd..3b03dc25fceac0 100644
--- a/src/quic/packet.cc
+++ b/src/quic/packet.cc
@@ -110,12 +110,11 @@ Local<FunctionTemplate> Packet::GetConstructorTemplate(Environment* env) {
   return tmpl;
 }
 
-BaseObjectPtr<Packet> Packet::Create(
-    Environment* env,
-    Listener* listener,
-    const SocketAddress& destination,
-    size_t length,
-    const char* diagnostic_label) {
+BaseObjectPtr<Packet> Packet::Create(Environment* env,
+                                     Listener* listener,
+                                     const SocketAddress& destination,
+                                     size_t length,
+                                     const char* diagnostic_label) {
   if (BindingData::Get(env).packet_freelist.empty()) {
     Local<Object> obj;
     if (!GetConstructorTemplate(env)
@@ -146,18 +145,16 @@ BaseObjectPtr<Packet> Packet::Clone() const {
       return {};
     }
 
-    return MakeBaseObject<Packet>(
-        env(), listener_, obj, destination_, data_);
+    return MakeBaseObject<Packet>(env(), listener_, obj, destination_, data_);
   }
 
   return FromFreeList(env(), data_, listener_, destination_);
 }
 
-BaseObjectPtr<Packet> Packet::FromFreeList(
-    Environment* env,
-    std::shared_ptr<Data> data,
-    Listener* listener,
-    const SocketAddress& destination) {
+BaseObjectPtr<Packet> Packet::FromFreeList(Environment* env,
+                                           std::shared_ptr<Data> data,
+                                           Listener* listener,
+                                           const SocketAddress& destination) {
   auto& binding = BindingData::Get(env);
   if (binding.packet_freelist.empty()) return {};
   auto obj = binding.packet_freelist.back();
diff --git a/src/quic/packet.h b/src/quic/packet.h
index c33a953b0ad779..ae6f76272e0156 100644
--- a/src/quic/packet.h
+++ b/src/quic/packet.h
@@ -139,11 +139,10 @@ class Packet final : public ReqWrap<uv_udp_send_t> {
   void Done(int status);
 
  private:
-  static BaseObjectPtr<Packet> FromFreeList(
-      Environment* env,
-      std::shared_ptr<Data> data,
-      Listener* listener,
-      const SocketAddress& destination);
+  static BaseObjectPtr<Packet> FromFreeList(Environment* env,
+                                            std::shared_ptr<Data> data,
+                                            Listener* listener,
+                                            const SocketAddress& destination);
 
   Listener* listener_;
   SocketAddress destination_;
diff --git a/src/quic/session.cc b/src/quic/session.cc
index e334cd894470e2..b1f18f834db19c 100644
--- a/src/quic/session.cc
+++ b/src/quic/session.cc
@@ -46,6 +46,7 @@ using v8::Integer;
 using v8::Just;
 using v8::Local;
 using v8::Maybe;
+using v8::MaybeLocal;
 using v8::Nothing;
 using v8::Object;
 using v8::PropertyAttribute;
@@ -115,7 +116,7 @@ namespace quic {
   V(GracefulClose, gracefulClose, false)                                       \
   V(SilentClose, silentClose, false)                                           \
   V(UpdateKey, updateKey, false)                                               \
-  V(DoOpenStream, openStream, false)                                           \
+  V(OpenStream, openStream, false)                                             \
   V(DoSendDatagram, sendDatagram, false)
 
 struct Session::State {
@@ -482,7 +483,7 @@ BaseObjectPtr<Session> Session::Create(
            ->InstanceTemplate()
            ->NewInstance(endpoint->env()->context())
            .ToLocal(&obj)) {
-    return BaseObjectPtr<Session>();
+    return {};
   }
 
   return MakeDetachedBaseObject<Session>(
@@ -662,12 +663,12 @@ void Session::HandleQlog(uint32_t flags, const void* data, size_t len) {
   }
 }
 
-TransportParams Session::GetLocalTransportParams() const {
+const TransportParams Session::GetLocalTransportParams() const {
   DCHECK(!is_destroyed());
   return TransportParams(ngtcp2_conn_get_local_transport_params(*this));
 }
 
-TransportParams Session::GetRemoteTransportParams() const {
+const TransportParams Session::GetRemoteTransportParams() const {
   DCHECK(!is_destroyed());
   return TransportParams(ngtcp2_conn_get_remote_transport_params(*this));
 }
@@ -718,6 +719,14 @@ void Session::Destroy() {
   // invalidated.
   auto streams = streams_;
   for (auto& stream : streams) stream.second->Destroy(last_error_);
+
+  while (!pending_bidi_stream_queue_.IsEmpty()) {
+    pending_bidi_stream_queue_.PopFront()->reject();
+  }
+  while (!pending_uni_stream_queue_.IsEmpty()) {
+    pending_uni_stream_queue_.PopFront()->reject();
+  }
+
   DCHECK(streams_.empty());
 
   STAT_RECORD_TIMESTAMP(Stats, destroyed_at);
@@ -1014,99 +1023,129 @@ void Session::UpdatePath(const PathStorage& storage) {
         remote_address_);
 }
 
-BaseObjectPtr<Stream> Session::FindStream(int64_t id) const {
+BaseObjectWeakPtr<Stream> Session::FindStream(int64_t id) const {
   auto it = streams_.find(id);
-  return it == std::end(streams_) ? BaseObjectPtr<Stream>() : it->second;
+  if (it == std::end(streams_)) return {};
+  return it->second;
 }
 
-BaseObjectPtr<Stream> Session::CreateStream(int64_t id,
-    CreateStreamOption option) {
-  if (!can_create_streams()) return BaseObjectPtr<Stream>();
-  auto stream = Stream::Create(this, id);
-  if (stream) AddStream(stream, option);
-  return stream;
+BaseObjectWeakPtr<Stream> Session::CreateStream(int64_t id,
+                                                CreateStreamOption option) {
+  if (!can_create_streams()) return {};
+  if (auto stream = Stream::Create(this, id)) {
+    AddStream(stream, option);
+    return stream;
+  }
+  return {};
 }
 
-BaseObjectPtr<Stream> Session::OpenStream(Direction direction) {
-  if (!can_create_streams()) return BaseObjectPtr<Stream>();
-  int64_t id;
-  switch (direction) {
-    case Direction::BIDIRECTIONAL: {
-      Debug(this, "Opening bidirectional stream");
-      if (ngtcp2_conn_open_bidi_stream(*this, &id, nullptr) == 0)
-        return CreateStream(id, CreateStreamOption::DOT_NOT_NOTIFY);
-      break;
+MaybeLocal<Object> Session::OpenStream(Direction direction) {
+  // If can_create_streams() returns false, we are not able to open a stream
+  // at all now, even in a pending state. The implication is that that session
+  // is destroyed or closing.
+  if (!can_create_streams()) return MaybeLocal<Object>();
+
+  // If can_open_streams() returns false, we are able to create streams but
+  // they will remain in a pending state. The implication is that the session
+  // TLS handshake is still progressing. Note that when a pending stream is
+  // created, it will not be listed in the streams list.
+  if (!can_open_streams()) {
+    if (auto stream = Stream::Create(this, direction)) [[likely]] {
+      return stream->object();
     }
-    case Direction::UNIDIRECTIONAL: {
-      Debug(this, "Opening uni-directional stream");
-      if (ngtcp2_conn_open_uni_stream(*this, &id, nullptr) == 0)
-        return CreateStream(id, CreateStreamOption::DOT_NOT_NOTIFY);
-      break;
+    return MaybeLocal<Object>();
+  }
+
+  int64_t id = -1;
+  auto open = [&] {
+    switch (direction) {
+      case Direction::BIDIRECTIONAL: {
+        Debug(this, "Opening bidirectional stream");
+        return ngtcp2_conn_open_bidi_stream(*this, &id, nullptr);
+      }
+      case Direction::UNIDIRECTIONAL: {
+        Debug(this, "Opening uni-directional stream");
+        return ngtcp2_conn_open_uni_stream(*this, &id, nullptr);
+      }
+    }
+  };
+
+  switch (open()) {
+    case 0: {
+      // Woo! Our stream was created.
+      CHECK_GE(id, 0);
+      if (auto stream = CreateStream(id, CreateStreamOption::DO_NOT_NOTIFY)) {
+        return stream->object();
+      }
+      return MaybeLocal<Object>();
+    }
+    case NGTCP2_ERR_STREAM_ID_BLOCKED: {
+      // The stream cannot yet be opened.
+      // This is typically caused by the application exceeding the allowed max
+      // number of concurrent streams. We will allow the stream to be created
+      // in a pending state.
+      if (auto stream = Stream::Create(this, direction)) {
+        return stream->object();
+      }
+      return MaybeLocal<Object>();
+    }
+    default: {
+      // The stream could not be opened. Return nothing to signal error.
+      return MaybeLocal<Object>();
     }
   }
-  return BaseObjectPtr<Stream>();
+  UNREACHABLE();
 }
 
 void Session::AddStream(const BaseObjectPtr<Stream>& stream,
                         CreateStreamOption option) {
-  Debug(this, "Adding stream %" PRIi64 " to session", stream->id());
-  ngtcp2_conn_set_stream_user_data(*this, stream->id(), stream.get());
+  CHECK(stream);
+
+  auto id = stream->id();
+  auto direction = stream->direction();
+
   // Let's double check that a stream with the given id does not already
   // exist. If it does, that means we've got a bug somewhere.
-  CHECK_EQ(streams_.find(stream->id()), streams_.end());
+  CHECK_EQ(streams_.find(id), streams_.end());
+
+  Debug(this, "Adding stream %" PRIi64 " to session", id);
 
-  streams_[stream->id()] = stream;
+  // The streams_ map becomes the sole owner of the Stream instance.
+  // We mark the stream detached so that when it is removed from
+  // the session, or is the session is destroyed, the stream will
+  // also be destroyed.
+  auto& inserted = streams_[id] = std::move(stream);
+  inserted->Detach();
+
+  ngtcp2_conn_set_stream_user_data(*this, id, inserted.get());
 
   if (option == CreateStreamOption::NOTIFY) {
-    EmitStream(stream);
+    EmitStream(inserted);
   }
 
   // Update tracking statistics for the number of streams associated with this
   // session.
-  switch (stream->origin()) {
-    case Side::CLIENT: {
-      if (is_server()) {
-        switch (stream->direction()) {
-          case Direction::BIDIRECTIONAL:
-            STAT_INCREMENT(Stats, bidi_in_stream_count);
-            break;
-          case Direction::UNIDIRECTIONAL:
-            STAT_INCREMENT(Stats, uni_in_stream_count);
-            break;
-        }
-      } else {
-        switch (stream->direction()) {
-          case Direction::BIDIRECTIONAL:
-            STAT_INCREMENT(Stats, bidi_out_stream_count);
-            break;
-          case Direction::UNIDIRECTIONAL:
-            STAT_INCREMENT(Stats, uni_out_stream_count);
-            break;
-        }
+  if (ngtcp2_conn_is_local_stream(*this, id)) {
+    switch (direction) {
+      case Direction::BIDIRECTIONAL: {
+        STAT_INCREMENT(Stats, bidi_out_stream_count);
+        break;
+      }
+      case Direction::UNIDIRECTIONAL: {
+        STAT_INCREMENT(Stats, uni_out_stream_count);
+        break;
       }
-      break;
     }
-    case Side::SERVER: {
-      if (is_server()) {
-        switch (stream->direction()) {
-          case Direction::BIDIRECTIONAL:
-            STAT_INCREMENT(Stats, bidi_out_stream_count);
-            break;
-          case Direction::UNIDIRECTIONAL:
-            STAT_INCREMENT(Stats, uni_out_stream_count);
-            break;
-        }
-      } else {
-        switch (stream->direction()) {
-          case Direction::BIDIRECTIONAL:
-            STAT_INCREMENT(Stats, bidi_in_stream_count);
-            break;
-          case Direction::UNIDIRECTIONAL:
-            STAT_INCREMENT(Stats, uni_in_stream_count);
-            break;
-        }
+  } else {
+    switch (direction) {
+      case Direction::BIDIRECTIONAL: {
+        STAT_INCREMENT(Stats, bidi_in_stream_count);
+        break;
+      }
+      case Direction::UNIDIRECTIONAL: {
+        STAT_INCREMENT(Stats, uni_in_stream_count);
+        break;
       }
-      break;
     }
   }
 }
@@ -1219,6 +1258,10 @@ bool Session::can_create_streams() const {
          !is_in_closing_period() && !is_in_draining_period();
 }
 
+bool Session::can_open_streams() const {
+  return state_->stream_open_allowed;
+}
+
 uint64_t Session::max_data_left() const {
   return ngtcp2_conn_get_max_data_left(*this);
 }
@@ -1529,6 +1572,72 @@ CID Session::new_cid(size_t len) const {
   return config_.options.cid_factory->Generate(len);
 }
 
+void Session::ProcessPendingBidiStreams() {
+  // It shouldn't be possible to get here if can_create_streams() is false.
+  CHECK(can_create_streams());
+
+  int64_t id;
+
+  while (!pending_bidi_stream_queue_.IsEmpty()) {
+    if (ngtcp2_conn_get_streams_bidi_left(*this) == 0) {
+      return;
+    }
+
+    switch (ngtcp2_conn_open_bidi_stream(*this, &id, nullptr)) {
+      case 0: {
+        pending_bidi_stream_queue_.PopFront()->fulfill(id);
+        continue;
+      }
+      case NGTCP2_ERR_STREAM_ID_BLOCKED: {
+        // This case really should not happen since we've checked the number
+        // of bidi streams left above. However, if it does happen we'll treat
+        // it the same as if the get_streams_bidi_left call returned zero.
+        return;
+      }
+      default: {
+        // We failed to open the stream for some reason other than being
+        // blocked. Report the failure.
+        pending_bidi_stream_queue_.PopFront()->reject(
+            QuicError::ForTransport(NGTCP2_STREAM_LIMIT_ERROR));
+        continue;
+      }
+    }
+  }
+}
+
+void Session::ProcessPendingUniStreams() {
+  // It shouldn't be possible to get here if can_create_streams() is false.
+  CHECK(can_create_streams());
+
+  int64_t id;
+
+  while (!pending_uni_stream_queue_.IsEmpty()) {
+    if (ngtcp2_conn_get_streams_uni_left(*this) == 0) {
+      return;
+    }
+
+    switch (ngtcp2_conn_open_uni_stream(*this, &id, nullptr)) {
+      case 0: {
+        pending_uni_stream_queue_.PopFront()->fulfill(id);
+        continue;
+      }
+      case NGTCP2_ERR_STREAM_ID_BLOCKED: {
+        // This case really should not happen since we've checked the number
+        // of bidi streams left above. However, if it does happen we'll treat
+        // it the same as if the get_streams_bidi_left call returned zero.
+        return;
+      }
+      default: {
+        // We failed to open the stream for some reason other than being
+        // blocked. Report the failure.
+        pending_uni_stream_queue_.PopFront()->reject(
+            QuicError::ForTransport(NGTCP2_STREAM_LIMIT_ERROR));
+        continue;
+      }
+    }
+  }
+}
+
 // JavaScript callouts
 
 void Session::EmitClose(const QuicError& error) {
@@ -1703,7 +1812,7 @@ void Session::EmitSessionTicket(Store&& ticket) {
   }
 }
 
-void Session::EmitStream(BaseObjectPtr<Stream> stream) {
+void Session::EmitStream(const BaseObjectPtr<Stream>& stream) {
   if (is_destroyed()) return;
   if (!env()->can_call_into_js()) return;
   CallbackScope<Session> cb_scope(this);
@@ -1853,14 +1962,15 @@ struct Session::Impl {
     args.GetReturnValue().Set(session->tls_session().InitiateKeyUpdate());
   }
 
-  static void DoOpenStream(const FunctionCallbackInfo<Value>& args) {
+  static void OpenStream(const FunctionCallbackInfo<Value>& args) {
     Session* session;
     ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
     DCHECK(args[0]->IsUint32());
     auto direction = static_cast<Direction>(args[0].As<Uint32>()->Value());
-    BaseObjectPtr<Stream> stream = session->OpenStream(direction);
-
-    if (stream) args.GetReturnValue().Set(stream->object());
+    Local<Object> stream;
+    if (session->OpenStream(direction).ToLocal(&stream)) [[likely]] {
+      args.GetReturnValue().Set(stream);
+    }
   }
 
   static void DoSendDatagram(const FunctionCallbackInfo<Value>& args) {
@@ -1943,8 +2053,7 @@ struct Session::Impl {
                                         uint64_t max_streams,
                                         void* user_data) {
     NGTCP2_CALLBACK_SCOPE(session)
-    session->application().ExtendMaxStreams(
-        EndpointLabel::LOCAL, Direction::BIDIRECTIONAL, max_streams);
+    session->ProcessPendingBidiStreams();
     return NGTCP2_SUCCESS;
   }
 
@@ -1952,8 +2061,7 @@ struct Session::Impl {
                                        uint64_t max_streams,
                                        void* user_data) {
     NGTCP2_CALLBACK_SCOPE(session)
-    session->application().ExtendMaxStreams(
-        EndpointLabel::LOCAL, Direction::UNIDIRECTIONAL, max_streams);
+    session->ProcessPendingUniStreams();
     return NGTCP2_SUCCESS;
   }
 
diff --git a/src/quic/session.h b/src/quic/session.h
index 49ddd9f962eafd..f84a125321c9da 100644
--- a/src/quic/session.h
+++ b/src/quic/session.h
@@ -117,7 +117,7 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
     const CID::Factory* cid_factory = &CID::Factory::random();
     // If the CID::Factory is a base object, we keep a reference to it
     // so that it cannot be garbage collected.
-    BaseObjectPtr<BaseObject> cid_factory_ref = BaseObjectPtr<BaseObject>();
+    BaseObjectPtr<BaseObject> cid_factory_ref = {};
 
     // When true, QLog output will be enabled for the session.
     bool qlog = false;
@@ -216,6 +216,7 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
           const Config& config,
           TLSContext* tls_context,
           const std::optional<SessionTicket>& ticket);
+  DISALLOW_COPY_AND_MOVE(Session)
   ~Session() override;
 
   uint32_t version() const;
@@ -244,8 +245,8 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
 
   void HandleQlog(uint32_t flags, const void* data, size_t len);
 
-  TransportParams GetLocalTransportParams() const;
-  TransportParams GetRemoteTransportParams() const;
+  const TransportParams GetLocalTransportParams() const;
+  const TransportParams GetRemoteTransportParams() const;
   void UpdatePacketTxTime();
 
   void MemoryInfo(MemoryTracker* tracker) const override;
@@ -257,16 +258,31 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
 
   operator ngtcp2_conn*() const;
 
-  BaseObjectPtr<Stream> FindStream(int64_t id) const;
+  // Note that we are returning a BaseObjectWeakPtr here. The Stream instance
+  // itself is owned strongly by the Session in a detached state. This means
+  // that the only reference to the Stream keeping it alive is held by the
+  // Session itself. That strong reference will be destroyed either when the
+  // stream is removed from session or the session is destroyed. All other
+  // references to the stream should be held weakly.
+  BaseObjectWeakPtr<Stream> FindStream(int64_t id) const;
 
   enum class CreateStreamOption {
     NOTIFY,
-    DOT_NOT_NOTIFY,
+    DO_NOT_NOTIFY,
   };
 
-  BaseObjectPtr<Stream> CreateStream(int64_t id,
-      CreateStreamOption option = CreateStreamOption::NOTIFY);
-  BaseObjectPtr<Stream> OpenStream(Direction direction);
+  // Creates the stream, adding it to the sessions stream map if successful.
+  // A weak reference to the stream is returned.
+  BaseObjectWeakPtr<Stream> CreateStream(
+      int64_t id, CreateStreamOption option = CreateStreamOption::NOTIFY);
+
+  // Open a new locally-initialized stream with the specified directionality.
+  // If the session is not yet in a state where the stream can be openen --
+  // such as when the handshake is not yet sufficiently far along and ORTT
+  // session resumption is not being used -- then the stream will be created
+  // in a pending state where actually opening the stream will be deferred.
+  v8::MaybeLocal<v8::Object> OpenStream(Direction direction);
+
   void ExtendStreamOffset(int64_t id, size_t amount);
   void ExtendOffset(size_t amount);
   void SetLastError(QuicError&& error);
@@ -300,6 +316,14 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
     ~SendPendingDataScope();
   };
 
+  inline PendingStream::PendingStreamQueue& pending_bidi_stream_queue() {
+    return pending_bidi_stream_queue_;
+  }
+
+  inline PendingStream::PendingStreamQueue& pending_uni_stream_queue() {
+    return pending_uni_stream_queue_;
+  }
+
  private:
   struct Impl;
   struct MaybeCloseConnectionScope;
@@ -357,8 +381,17 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
   bool can_send_packets() const;
 
   // Returns false if the Session is currently in a state where it cannot create
-  // new streams.
+  // new streams. Specifically, a stream is not in a state to create streams if
+  // it has been destroyed or is closing.
   bool can_create_streams() const;
+
+  // Returns false if the Session is currently in a state where it cannot open
+  // a new locally-initiated stream. When using 0RTT session resumption, this
+  // will become true immediately after the session ticket and transport params
+  // have been configured. Otherwise, it becomes true after the remote transport
+  // params and tx keys have been installed.
+  bool can_open_streams() const;
+
   uint64_t max_local_streams_uni() const;
   uint64_t max_local_streams_bidi() const;
 
@@ -377,6 +410,9 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
   void UpdateTimer();
   bool StartClosingPeriod();
 
+  void ProcessPendingBidiStreams();
+  void ProcessPendingUniStreams();
+
   // JavaScript callouts
 
   void EmitClose(const QuicError& error = QuicError());
@@ -395,7 +431,7 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
                           const ValidatedPath& newPath,
                           const std::optional<ValidatedPath>& oldPath);
   void EmitSessionTicket(Store&& ticket);
-  void EmitStream(BaseObjectPtr<Stream> stream);
+  void EmitStream(const BaseObjectPtr<Stream>& stream);
   void EmitVersionNegotiation(const ngtcp2_pkt_hd& hd,
                               const uint32_t* sv,
                               size_t nsv);
@@ -432,6 +468,8 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
   BaseObjectPtr<Packet> conn_closebuf_;
   BaseObjectPtr<LogStream> qlog_stream_;
   BaseObjectPtr<LogStream> keylog_stream_;
+  PendingStream::PendingStreamQueue pending_bidi_stream_queue_;
+  PendingStream::PendingStreamQueue pending_uni_stream_queue_;
 
   friend class Application;
   friend class DefaultApplication;
diff --git a/src/quic/streams.cc b/src/quic/streams.cc
index bd2baf37765357..15ee9e3101ed0e 100644
--- a/src/quic/streams.cc
+++ b/src/quic/streams.cc
@@ -21,6 +21,7 @@ using v8::ArrayBufferView;
 using v8::BigInt;
 using v8::FunctionCallbackInfo;
 using v8::FunctionTemplate;
+using v8::Global;
 using v8::Integer;
 using v8::Just;
 using v8::Local;
@@ -37,12 +38,11 @@ namespace quic {
 
 #define STREAM_STATE(V)                                                        \
   V(ID, id, int64_t)                                                           \
+  V(PENDING, pending, uint8_t)                                                 \
   V(FIN_SENT, fin_sent, uint8_t)                                               \
   V(FIN_RECEIVED, fin_received, uint8_t)                                       \
   V(READ_ENDED, read_ended, uint8_t)                                           \
   V(WRITE_ENDED, write_ended, uint8_t)                                         \
-  V(DESTROYING, destroying, uint8_t)                                           \
-  V(DESTROYED, destroyed, uint8_t)                                             \
   V(PAUSED, paused, uint8_t)                                                   \
   V(RESET, reset, uint8_t)                                                     \
   V(HAS_READER, has_reader, uint8_t)                                           \
@@ -56,12 +56,20 @@ namespace quic {
   V(WANTS_TRAILERS, wants_trailers, uint8_t)
 
 #define STREAM_STATS(V)                                                        \
+  /* Marks the timestamp when the stream object was created. */                \
   V(CREATED_AT, created_at)                                                    \
+  /* Marks the timestamp when the stream was opened. This can be different */  \
+  /* from the created_at timestamp if the stream was created in as pending */  \
+  V(OPENED_AT, opened_at)                                                      \
+  /* Marks the timestamp when the stream last received data */                 \
   V(RECEIVED_AT, received_at)                                                  \
+  /* Marks the timestamp when the stream last received an acknowledgement */   \
   V(ACKED_AT, acked_at)                                                        \
-  V(CLOSING_AT, closing_at)                                                    \
+  /* Marks the timestamp when the stream was destroyed */                      \
   V(DESTROYED_AT, destroyed_at)                                                \
+  /* Records the total number of bytes receied by the stream */                \
   V(BYTES_RECEIVED, bytes_received)                                            \
+  /* Records the total number of bytes sent by the stream */                   \
   V(BYTES_SENT, bytes_sent)                                                    \
   V(MAX_OFFSET, max_offset)                                                    \
   V(MAX_OFFSET_ACK, max_offset_ack)                                            \
@@ -78,6 +86,53 @@ namespace quic {
   V(GetPriority, getPriority, true)                                            \
   V(GetReader, getReader, false)
 
+// ============================================================================
+
+PendingStream::PendingStream(Direction direction,
+                             Stream* stream,
+                             BaseObjectWeakPtr<Session> session)
+    : direction_(direction), stream_(stream), session_(session) {
+  if (session_) {
+    if (direction == Direction::BIDIRECTIONAL) {
+      session_->pending_bidi_stream_queue().PushBack(this);
+    } else {
+      session_->pending_uni_stream_queue().PushBack(this);
+    }
+  }
+}
+
+PendingStream::~PendingStream() {
+  pending_stream_queue_.Remove();
+  if (!waiting_) {
+    Debug(stream_, "A pending stream was canceled");
+  }
+}
+
+void PendingStream::fulfill(int64_t id) {
+  CHECK(waiting_);
+  waiting_ = false;
+  stream_->NotifyStreamOpened(id);
+}
+
+void PendingStream::reject(QuicError error) {
+  CHECK(waiting_);
+  waiting_ = false;
+  stream_->Destroy(error);
+}
+
+struct Stream::PendingHeaders {
+  HeadersKind kind;
+  v8::Global<v8::Array> headers;
+  HeadersFlags flags;
+  PendingHeaders(HeadersKind kind_,
+                 v8::Global<v8::Array> headers_,
+                 HeadersFlags flags_)
+      : kind(kind_), headers(std::move(headers_)), flags(flags_) {}
+  DISALLOW_COPY_AND_MOVE(PendingHeaders)
+};
+
+// ============================================================================
+
 struct Stream::State {
 #define V(_, name, type) type name;
   STREAM_STATE(V)
@@ -92,22 +147,20 @@ namespace {
 Maybe<std::shared_ptr<DataQueue>> GetDataQueueFromSource(Environment* env,
                                                          Local<Value> value) {
   DCHECK_IMPLIES(!value->IsUndefined(), value->IsObject());
+  std::vector<std::unique_ptr<DataQueue::Entry>> entries(1);
   if (value->IsUndefined()) {
     return Just(std::shared_ptr<DataQueue>());
   } else if (value->IsArrayBuffer()) {
     auto buffer = value.As<ArrayBuffer>();
-    std::vector<std::unique_ptr<DataQueue::Entry>> entries(1);
     entries.push_back(DataQueue::CreateInMemoryEntryFromBackingStore(
         buffer->GetBackingStore(), 0, buffer->ByteLength()));
     return Just(DataQueue::CreateIdempotent(std::move(entries)));
   } else if (value->IsSharedArrayBuffer()) {
     auto buffer = value.As<SharedArrayBuffer>();
-    std::vector<std::unique_ptr<DataQueue::Entry>> entries(1);
     entries.push_back(DataQueue::CreateInMemoryEntryFromBackingStore(
         buffer->GetBackingStore(), 0, buffer->ByteLength()));
     return Just(DataQueue::CreateIdempotent(std::move(entries)));
   } else if (value->IsArrayBufferView()) {
-    std::vector<std::unique_ptr<DataQueue::Entry>> entries(1);
     entries.push_back(
         DataQueue::CreateInMemoryEntryFromView(value.As<ArrayBufferView>()));
     return Just(DataQueue::CreateIdempotent(std::move(entries)));
@@ -123,7 +176,10 @@ Maybe<std::shared_ptr<DataQueue>> GetDataQueueFromSource(Environment* env,
 }
 }  // namespace
 
+// Provides the implementation of the various JavaScript APIs for the
+// Stream object.
 struct Stream::Impl {
+  // Attaches an outbound data source to the stream.
   static void AttachSource(const FunctionCallbackInfo<Value>& args) {
     Environment* env = Environment::GetCurrent(args);
 
@@ -159,7 +215,14 @@ struct Stream::Impl {
     Local<Array> headers = args[1].As<Array>();
     HeadersFlags flags =
         static_cast<HeadersFlags>(args[2].As<Uint32>()->Value());
-    if (stream->is_destroyed()) return args.GetReturnValue().Set(false);
+
+    // If the stream is pending, the headers will be queued until the
+    // stream is opened, at which time the queued header block will be
+    // immediately sent when the stream is opened.
+    if (stream->is_pending()) {
+      stream->EnqueuePendingHeaders(kind, headers, flags);
+      return args.GetReturnValue().Set(true);
+    }
 
     args.GetReturnValue().Set(stream->session().application().SendHeaders(
         *stream, kind, headers, flags));
@@ -174,14 +237,19 @@ struct Stream::Impl {
     uint64_t code = NGTCP2_APP_NOERROR;
     CHECK_IMPLIES(!args[0]->IsUndefined(), args[0]->IsBigInt());
     if (!args[0]->IsUndefined()) {
-      bool lossless = false;  // not used but still necessary.
-      code = args[0].As<BigInt>()->Uint64Value(&lossless);
+      bool unused = false;  // not used but still necessary.
+      code = args[0].As<BigInt>()->Uint64Value(&unused);
     }
 
-    if (stream->is_destroyed()) return;
     stream->EndReadable();
-    Session::SendPendingDataScope send_scope(&stream->session());
-    ngtcp2_conn_shutdown_stream_read(stream->session(), 0, stream->id(), code);
+
+    if (!stream->is_pending()) {
+      // If the stream is a local unidirectional there's nothing to do here.
+      if (stream->is_local_unidirectional()) return;
+      stream->NotifyReadableEnded(code);
+    } else {
+      stream->pending_close_read_code_ = code;
+    }
   }
 
   // Sends a reset stream to the peer to tell it we will not be sending any
@@ -198,15 +266,21 @@ struct Stream::Impl {
       code = args[0].As<BigInt>()->Uint64Value(&lossless);
     }
 
-    if (stream->is_destroyed() || stream->state_->reset == 1) return;
+    if (stream->state_->reset == 1) return;
+
     stream->EndWritable();
     // We can release our outbound here now. Since the stream is being reset
     // on the ngtcp2 side, we do not need to keep any of the data around
     // waiting for acknowledgement that will never come.
     stream->outbound_.reset();
     stream->state_->reset = 1;
-    Session::SendPendingDataScope send_scope(&stream->session());
-    ngtcp2_conn_shutdown_stream_write(stream->session(), 0, stream->id(), code);
+
+    if (!stream->is_pending()) {
+      if (stream->is_remote_unidirectional()) return;
+      stream->NotifyWritableEnded(code);
+    } else {
+      stream->pending_close_write_code_ = code;
+    }
   }
 
   static void SetPriority(const FunctionCallbackInfo<Value>& args) {
@@ -220,12 +294,26 @@ struct Stream::Impl {
     StreamPriorityFlags flags =
         static_cast<StreamPriorityFlags>(args[1].As<Uint32>()->Value());
 
-    stream->session().application().SetStreamPriority(*stream, priority, flags);
+    if (stream->is_pending()) {
+      stream->pending_priority_ = Stream::PendingPriority{
+          .priority = priority,
+          .flags = flags,
+      };
+    } else {
+      stream->session().application().SetStreamPriority(
+          *stream, priority, flags);
+    }
   }
 
   static void GetPriority(const FunctionCallbackInfo<Value>& args) {
     Stream* stream;
     ASSIGN_OR_RETURN_UNWRAP(&stream, args.This());
+
+    if (stream->is_pending()) {
+      return args.GetReturnValue().Set(
+          static_cast<uint32_t>(StreamPriority::DEFAULT));
+    }
+
     auto priority = stream->session().application().GetStreamPriority(*stream);
     args.GetReturnValue().Set(static_cast<uint32_t>(priority));
   }
@@ -392,7 +480,7 @@ class Stream::Outbound final : public MemoryRetainer {
             // Here, there is no more data to read, but we will might have data
             // in the uncommitted queue. We'll resume the stream so that the
             // session will try to read from it again.
-            if (next_pending_ && !stream_->is_destroyed()) {
+            if (next_pending_) {
               stream_->session().ResumeStream(stream_->id());
             }
             return;
@@ -416,7 +504,7 @@ class Stream::Outbound final : public MemoryRetainer {
           // being asynchronous, our stream is blocking waiting for the data.
           // Now that we have data, let's resume the stream so the session will
           // pull from it again.
-          if (next_pending_ && !stream_->is_destroyed()) {
+          if (next_pending_) {
             stream_->session().ResumeStream(stream_->id());
           }
         },
@@ -697,13 +785,29 @@ BaseObjectPtr<Stream> Stream::Create(Session* session,
            ->InstanceTemplate()
            ->NewInstance(session->env()->context())
            .ToLocal(&obj)) {
-    return BaseObjectPtr<Stream>();
+    return {};
   }
 
   return MakeDetachedBaseObject<Stream>(
       BaseObjectWeakPtr<Session>(session), obj, id, std::move(source));
 }
 
+BaseObjectPtr<Stream> Stream::Create(Session* session,
+                                     Direction direction,
+                                     std::shared_ptr<DataQueue> source) {
+  DCHECK_NOT_NULL(session);
+  Local<Object> obj;
+  if (!GetConstructorTemplate(session->env())
+           ->InstanceTemplate()
+           ->NewInstance(session->env()->context())
+           .ToLocal(&obj)) {
+    return {};
+  }
+
+  return MakeBaseObject<Stream>(
+      BaseObjectWeakPtr<Session>(session), obj, direction, std::move(source));
+}
+
 Stream::Stream(BaseObjectWeakPtr<Session> session,
                v8::Local<v8::Object> object,
                int64_t id,
@@ -712,12 +816,45 @@ Stream::Stream(BaseObjectWeakPtr<Session> session,
       stats_(env()->isolate()),
       state_(env()->isolate()),
       session_(std::move(session)),
-      origin_(id & 0b01 ? Side::SERVER : Side::CLIENT),
-      direction_(id & 0b10 ? Direction::UNIDIRECTIONAL
-                           : Direction::BIDIRECTIONAL),
       inbound_(DataQueue::Create()) {
   MakeWeak();
   state_->id = id;
+  state_->pending = 0;
+  // Allows us to be notified when data is actually read from the
+  // inbound queue so that we can update the stream flow control.
+  inbound_->addBackpressureListener(this);
+
+  const auto defineProperty = [&](auto name, auto value) {
+    object
+        ->DefineOwnProperty(
+            env()->context(), name, value, PropertyAttribute::ReadOnly)
+        .Check();
+  };
+
+  defineProperty(env()->state_string(), state_.GetArrayBuffer());
+  defineProperty(env()->stats_string(), stats_.GetArrayBuffer());
+
+  set_outbound(std::move(source));
+
+  auto params = ngtcp2_conn_get_local_transport_params(this->session());
+  STAT_SET(Stats, max_offset, params->initial_max_data);
+  STAT_SET(Stats, opened_at, stats_->created_at);
+}
+
+Stream::Stream(BaseObjectWeakPtr<Session> session,
+               v8::Local<v8::Object> object,
+               Direction direction,
+               std::shared_ptr<DataQueue> source)
+    : AsyncWrap(session->env(), object, AsyncWrap::PROVIDER_QUIC_STREAM),
+      stats_(env()->isolate()),
+      state_(env()->isolate()),
+      session_(std::move(session)),
+      inbound_(DataQueue::Create()),
+      maybe_pending_stream_(
+          std::make_unique<PendingStream>(direction, this, session_)) {
+  MakeWeak();
+  state_->id = -1;
+  state_->pending = 1;
 
   // Allows us to be notified when data is actually read from the
   // inbound queue so that we can update the stream flow control.
@@ -740,8 +877,77 @@ Stream::Stream(BaseObjectWeakPtr<Session> session,
 }
 
 Stream::~Stream() {
-  // Make sure that Destroy() was called before Stream is destructed.
-  DCHECK(is_destroyed());
+  // Make sure that Destroy() was called before Stream is actually destructed.
+  DCHECK_NE(stats_->destroyed_at, 0);
+}
+
+void Stream::NotifyStreamOpened(int64_t id) {
+  CHECK(is_pending());
+  Debug(this, "Pending stream opened with id %" PRIi64, id);
+  state_->pending = 0;
+  state_->id = id;
+  STAT_RECORD_TIMESTAMP(Stats, opened_at);
+  // Now that the stream is actually opened, add it to the sessions
+  // list of known open streams.
+  session().AddStream(BaseObjectPtr<Stream>(this),
+                      Session::CreateStreamOption::DO_NOT_NOTIFY);
+
+  CHECK_EQ(ngtcp2_conn_set_stream_user_data(this->session(), id, this), 0);
+  maybe_pending_stream_.reset();
+
+  if (pending_priority_) {
+    auto& priority = pending_priority_.value();
+    session().application().SetStreamPriority(
+        *this, priority.priority, priority.flags);
+    pending_priority_ = std::nullopt;
+  }
+  decltype(pending_headers_queue_) queue;
+  pending_headers_queue_.swap(queue);
+  for (auto& headers : queue) {
+    // TODO(@jasnell): What if the application does not support headers?
+    session().application().SendHeaders(*this,
+                                        headers->kind,
+                                        headers->headers.Get(env()->isolate()),
+                                        headers->flags);
+  }
+  // If the stream is not a local undirectional stream and is_readable is
+  // false, then we should shutdown the streams readable side now.
+  if (!is_local_unidirectional() && !is_readable()) {
+    NotifyReadableEnded(pending_close_read_code_);
+  }
+  if (!is_remote_unidirectional() && !is_writable()) {
+    NotifyWritableEnded(pending_close_write_code_);
+  }
+
+  // Finally, if we have an outbound data source attached already, make
+  // sure our stream is scheduled. This is likely a bit superfluous
+  // since the stream likely hasn't had any opporunity to get blocked
+  // yet, but just for completeness, let's make sure.
+  if (outbound_) session().ResumeStream(id);
+}
+
+void Stream::NotifyReadableEnded(uint64_t code) {
+  CHECK(!is_pending());
+  Session::SendPendingDataScope send_scope(&session());
+  ngtcp2_conn_shutdown_stream_read(session(), 0, id(), code);
+}
+
+void Stream::NotifyWritableEnded(uint64_t code) {
+  CHECK(!is_pending());
+  Session::SendPendingDataScope send_scope(&session());
+  ngtcp2_conn_shutdown_stream_write(session(), 0, id(), code);
+}
+
+void Stream::EnqueuePendingHeaders(HeadersKind kind,
+                                   Local<Array> headers,
+                                   HeadersFlags flags) {
+  Debug(this, "Enqueuing headers for pending stream");
+  pending_headers_queue_.push_back(std::make_unique<PendingHeaders>(
+      kind, Global<Array>(env()->isolate(), headers), flags));
+}
+
+bool Stream::is_pending() const {
+  return state_->pending;
 }
 
 int64_t Stream::id() const {
@@ -749,19 +955,32 @@ int64_t Stream::id() const {
 }
 
 Side Stream::origin() const {
-  return origin_;
+  CHECK(!is_pending());
+  return (state_->id & 0b01) ? Side::SERVER : Side::CLIENT;
 }
 
 Direction Stream::direction() const {
-  return direction_;
+  if (state_->pending) {
+    CHECK(maybe_pending_stream_.has_value());
+    auto& val = maybe_pending_stream_.value();
+    return val->direction();
+  }
+  return (state_->id & 0b10) ? Direction::UNIDIRECTIONAL
+                             : Direction::BIDIRECTIONAL;
 }
 
 Session& Stream::session() const {
   return *session_;
 }
 
-bool Stream::is_destroyed() const {
-  return state_->destroyed;
+bool Stream::is_local_unidirectional() const {
+  return direction() == Direction::UNIDIRECTIONAL &&
+         ngtcp2_conn_is_local_stream(*session_, id());
+}
+
+bool Stream::is_remote_unidirectional() const {
+  return direction() == Direction::UNIDIRECTIONAL &&
+         !ngtcp2_conn_is_local_stream(*session_, id());
 }
 
 bool Stream::is_eos() const {
@@ -769,40 +988,27 @@ bool Stream::is_eos() const {
 }
 
 bool Stream::is_writable() const {
-  if (direction() == Direction::UNIDIRECTIONAL) {
-    switch (origin()) {
-      case Side::CLIENT: {
-        if (session_->is_server()) return false;
-        break;
-      }
-      case Side::SERVER: {
-        if (!session_->is_server()) return false;
-        break;
-      }
-    }
+  // Remote unidirectional streams are never writable, and remote streams can
+  // never be pending.
+  if (!is_pending() && direction() == Direction::UNIDIRECTIONAL &&
+      !ngtcp2_conn_is_local_stream(session(), id())) {
+    return false;
   }
   return state_->write_ended == 0;
 }
 
 bool Stream::is_readable() const {
-  if (direction() == Direction::UNIDIRECTIONAL) {
-    switch (origin()) {
-      case Side::CLIENT: {
-        if (!session_->is_server()) return false;
-        break;
-      }
-      case Side::SERVER: {
-        if (session_->is_server()) return false;
-        break;
-      }
-    }
+  // Local unidirectional streams are never readable, and remote streams can
+  // never be pending.
+  if (!is_pending() && direction() == Direction::UNIDIRECTIONAL &&
+      ngtcp2_conn_is_local_stream(session(), id())) {
+    return false;
   }
   return state_->read_ended == 0;
 }
 
 BaseObjectPtr<Blob::Reader> Stream::get_reader() {
-  if (!is_readable() || state_->has_reader)
-    return BaseObjectPtr<Blob::Reader>();
+  if (!is_readable() || state_->has_reader) return {};
   state_->has_reader = 1;
   return Blob::Reader::Create(env(), Blob::Create(env(), inbound_));
 }
@@ -815,17 +1021,17 @@ void Stream::set_final_size(uint64_t final_size) {
 }
 
 void Stream::set_outbound(std::shared_ptr<DataQueue> source) {
-  if (!source || is_destroyed() || !is_writable()) return;
+  if (!source || !is_writable()) return;
   DCHECK_NULL(outbound_);
   outbound_ = std::make_unique<Outbound>(this, std::move(source));
-  session_->ResumeStream(id());
+  if (!is_pending()) session_->ResumeStream(id());
 }
 
 void Stream::EntryRead(size_t amount) {
-  // Tells us that amount bytes were read from inbound_
+  // Tells us that amount bytes we're reading from inbound_
   // We use this as a signal to extend the flow control
   // window to receive more bytes.
-  if (!is_destroyed() && session_) session_->ExtendStreamOffset(id(), amount);
+  session().ExtendStreamOffset(id(), amount);
 }
 
 int Stream::DoPull(bob::Next<ngtcp2_vec> next,
@@ -833,7 +1039,7 @@ int Stream::DoPull(bob::Next<ngtcp2_vec> next,
                    ngtcp2_vec* data,
                    size_t count,
                    size_t max_count_hint) {
-  if (is_destroyed() || is_eos()) {
+  if (is_eos()) {
     std::move(next)(bob::Status::STATUS_EOS, nullptr, 0, [](int) {});
     return bob::Status::STATUS_EOS;
   }
@@ -853,7 +1059,6 @@ int Stream::DoPull(bob::Next<ngtcp2_vec> next,
 }
 
 void Stream::BeginHeaders(HeadersKind kind) {
-  if (is_destroyed()) return;
   headers_length_ = 0;
   headers_.clear();
   set_headers_kind(kind);
@@ -865,8 +1070,8 @@ void Stream::set_headers_kind(HeadersKind kind) {
 
 bool Stream::AddHeader(const Header& header) {
   size_t len = header.length();
-  if (is_destroyed() || !session_->application().CanAddHeader(
-                            headers_.size(), headers_length_, len)) {
+  if (!session_->application().CanAddHeader(
+          headers_.size(), headers_length_, len)) {
     return false;
   }
 
@@ -887,7 +1092,7 @@ bool Stream::AddHeader(const Header& header) {
 }
 
 void Stream::Acknowledge(size_t datalen) {
-  if (is_destroyed() || outbound_ == nullptr) return;
+  if (outbound_ == nullptr) return;
 
   // ngtcp2 guarantees that offset must always be greater than the previously
   // received offset.
@@ -899,11 +1104,12 @@ void Stream::Acknowledge(size_t datalen) {
 }
 
 void Stream::Commit(size_t datalen) {
-  if (!is_destroyed() && outbound_) outbound_->Commit(datalen);
+  STAT_RECORD_TIMESTAMP(Stats, acked_at);
+  if (outbound_) outbound_->Commit(datalen);
 }
 
 void Stream::EndWritable() {
-  if (is_destroyed() || !is_writable()) return;
+  if (!is_writable()) return;
   // If an outbound_ has been attached, we want to mark it as being ended.
   // If the outbound_ is wrapping an idempotent DataQueue, then capping
   // will be a non-op since we're not going to be writing any more data
@@ -913,17 +1119,29 @@ void Stream::EndWritable() {
 }
 
 void Stream::EndReadable(std::optional<uint64_t> maybe_final_size) {
-  if (is_destroyed() || !is_readable()) return;
+  if (!is_readable()) return;
   state_->read_ended = 1;
   set_final_size(maybe_final_size.value_or(STAT_GET(Stats, bytes_received)));
   inbound_->cap(STAT_GET(Stats, final_size));
 }
 
 void Stream::Destroy(QuicError error) {
-  if (is_destroyed() || state_->destroying) return;
-  state_->destroying = 1;
+  if (stats_->destroyed_at != 0) return;
+  // Record the destroyed at timestamp before notifying the JavaScript side
+  // that the stream is being destroyed.
+  STAT_RECORD_TIMESTAMP(Stats, destroyed_at);
+
   DCHECK_NOT_NULL(session_.get());
-  Debug(this, "Stream %" PRIi64 " being destroyed with error %s", id(), error);
+
+  if (!state_->pending) {
+    Debug(
+        this, "Stream %" PRIi64 " being destroyed with error %s", id(), error);
+  } else {
+    Debug(this, "Pending stream being destroyed with error %s", error);
+  }
+  state_->pending = 0;
+
+  maybe_pending_stream_.reset();
 
   // End the writable before marking as destroyed.
   EndWritable();
@@ -947,25 +1165,21 @@ void Stream::Destroy(QuicError error) {
   // handle.
   EmitClose(error);
 
-  // The state_->destroyed state is checked by the EmitClose so it is
-  // important to set this after calling EmitClose.
-  state_->destroying = 0;
-  state_->destroyed = 1;
+  auto session = session_;
+  session_.reset();
+  session->RemoveStream(id());
 
-  if (session_) {
-    // Finally, remove the stream from the session and clear our reference
-    // to the session.
-    BaseObjectPtr<Stream> self(this);
-    session_->RemoveStream(id());
-    session_.reset();
-  }
+  // Critically, make sure that the RemoveStream call is the last thing
+  // trying to use this stream object. Once that call is made, the stream
+  // object is no longer valid and should not be accessed.
+  // Specifically, the session object's streams map holds the its
+  // BaseObjectPtr<Stream> instances in a detached state, meaning that
+  // once that BaseObjectPtr is deleted the Stream will be freed as well.
 }
 
 void Stream::ReceiveData(const uint8_t* data,
                          size_t len,
                          ReceiveDataFlags flags) {
-  if (is_destroyed()) return;
-
   // If reading has ended, or there is no data, there's nothing to do but maybe
   // end the readable side if this is the last bit of data we've received.
   if (state_->read_ended == 1 || len == 0) {
@@ -974,6 +1188,7 @@ void Stream::ReceiveData(const uint8_t* data,
   }
 
   STAT_INCREMENT_N(Stats, bytes_received, len);
+  STAT_RECORD_TIMESTAMP(Stats, received_at);
   auto backing = ArrayBuffer::NewBackingStore(env()->isolate(), len);
   memcpy(backing->Data(), data, len);
   inbound_->append(DataQueue::CreateInMemoryEntryFromBackingStore(
@@ -985,7 +1200,7 @@ void Stream::ReceiveData(const uint8_t* data,
 void Stream::ReceiveStopSending(QuicError error) {
   // Note that this comes from *this* endpoint, not the other side. We handle it
   // if we haven't already shutdown our *receiving* side of the stream.
-  if (is_destroyed() || state_->read_ended) return;
+  if (state_->read_ended) return;
   ngtcp2_conn_shutdown_stream_read(session(), 0, id(), error.code());
   EndReadable();
 }
@@ -1005,8 +1220,7 @@ void Stream::ReceiveStreamReset(uint64_t final_size, QuicError error) {
 void Stream::EmitBlocked() {
   // state_->wants_block will be set from the javascript side if the
   // stream object has a handler for the blocked event.
-  if (is_destroyed() || !env()->can_call_into_js() ||
-      state_->wants_block == 0) {
+  if (!env()->can_call_into_js() || !state_->wants_block) {
     return;
   }
   CallbackScope<Stream> cb_scope(this);
@@ -1014,7 +1228,7 @@ void Stream::EmitBlocked() {
 }
 
 void Stream::EmitClose(const QuicError& error) {
-  if (is_destroyed() || !env()->can_call_into_js()) return;
+  if (!env()->can_call_into_js()) return;
   CallbackScope<Stream> cb_scope(this);
   Local<Value> err;
   if (!error.ToV8Value(env()).ToLocal(&err)) return;
@@ -1022,8 +1236,9 @@ void Stream::EmitClose(const QuicError& error) {
 }
 
 void Stream::EmitHeaders() {
-  if (is_destroyed() || !env()->can_call_into_js() ||
-      state_->wants_headers == 0) {
+  // state_->wants_headers will be set from the javascript side if the
+  // stream object has a handler for the headers event.
+  if (!env()->can_call_into_js() || !state_->wants_headers) {
     return;
   }
   CallbackScope<Stream> cb_scope(this);
@@ -1040,8 +1255,9 @@ void Stream::EmitHeaders() {
 }
 
 void Stream::EmitReset(const QuicError& error) {
-  if (is_destroyed() || !env()->can_call_into_js() ||
-      state_->wants_reset == 0) {
+  // state_->wants_reset will be set from the javascript side if the
+  // stream object has a handler for the reset event.
+  if (!env()->can_call_into_js() || !state_->wants_reset) {
     return;
   }
   CallbackScope<Stream> cb_scope(this);
@@ -1052,8 +1268,9 @@ void Stream::EmitReset(const QuicError& error) {
 }
 
 void Stream::EmitWantTrailers() {
-  if (is_destroyed() || !env()->can_call_into_js() ||
-      state_->wants_trailers == 0) {
+  // state_->wants_trailers will be set from the javascript side if the
+  // stream object has a handler for the trailers event.
+  if (!env()->can_call_into_js() || !state_->wants_trailers) {
     return;
   }
   CallbackScope<Stream> cb_scope(this);
@@ -1064,8 +1281,7 @@ void Stream::EmitWantTrailers() {
 
 void Stream::Schedule(Stream::Queue* queue) {
   // If this stream is not already in the queue to send data, add it.
-  if (!is_destroyed() && outbound_ && stream_queue_.IsEmpty())
-    queue->PushBack(this);
+  if (outbound_ && stream_queue_.IsEmpty()) queue->PushBack(this);
 }
 
 void Stream::Unschedule() {
diff --git a/src/quic/streams.h b/src/quic/streams.h
index 0fb85680a2b271..7461ef5b66f5fe 100644
--- a/src/quic/streams.h
+++ b/src/quic/streams.h
@@ -12,15 +12,61 @@
 #include <node_blob.h>
 #include <node_bob.h>
 #include <node_http_common.h>
+#include <util.h>
 #include "bindingdata.h"
 #include "data.h"
 
 namespace node::quic {
 
 class Session;
+class Stream;
 
 using Ngtcp2Source = bob::SourceImpl<ngtcp2_vec>;
 
+// When a request to open a stream is made before a Session is able to actually
+// open a stream (either because the handshake is not yet sufficiently complete
+// or concurrency limits are temporarily reached) then the request to open the
+// stream is represented as a queued PendingStream.
+//
+// The PendingStream instance itself is held by the stream but sits in a linked
+// list in the session.
+//
+// The PendingStream request can be canceled by dropping the PendingStream
+// instance before it can be fulfilled, at which point it is removed from the
+// pending stream queue.
+//
+// Note that only locally initiated streams can be created in a pending state.
+class PendingStream final {
+ public:
+  PendingStream(Direction direction,
+                Stream* stream,
+                BaseObjectWeakPtr<Session> session);
+  DISALLOW_COPY_AND_MOVE(PendingStream)
+  ~PendingStream();
+
+  // Called when the stream has been opened. Transitions the stream from a
+  // pending state to an opened state.
+  void fulfill(int64_t id);
+
+  // Called when opening the stream fails or is canceled. Transitions the
+  // stream into a closed/destroyed state.
+  void reject(QuicError error = QuicError());
+
+  inline Direction direction() const { return direction_; }
+
+ private:
+  Direction direction_;
+  Stream* stream_;
+  BaseObjectWeakPtr<Session> session_;
+  bool waiting_ = true;
+
+  ListNode<PendingStream> pending_stream_queue_;
+
+ public:
+  using PendingStreamQueue =
+      ListHead<PendingStream, &PendingStream::pending_stream_queue_>;
+};
+
 // QUIC Stream's are simple data flows that may be:
 //
 // * Bidirectional (both sides can send) or Unidirectional (one side can send)
@@ -63,7 +109,7 @@ using Ngtcp2Source = bob::SourceImpl<ngtcp2_vec>;
 // the right thing.
 //
 // A Stream may be in a fully closed state (No longer readable nor writable)
-// state but still have unacknowledged data in it's inbound and outbound
+// state but still have unacknowledged data in both the inbound and outbound
 // queues.
 //
 // A Stream is gracefully closed when (a) both read and write states are closed,
@@ -78,9 +124,14 @@ using Ngtcp2Source = bob::SourceImpl<ngtcp2_vec>;
 //
 // QUIC streams in general do not have headers. Some QUIC applications, however,
 // may associate headers with the stream (HTTP/3 for instance).
-class Stream : public AsyncWrap,
-               public Ngtcp2Source,
-               public DataQueue::BackpressureListener {
+//
+// Streams may be created in a pending state. This means that while the Stream
+// object is created, it has not yet been opened in ngtcp2 and therefore has
+// no official status yet. Certain operations can still be performed on the
+// stream object such as providing data and headers, and destroying the stream.
+class Stream final : public AsyncWrap,
+                     public Ngtcp2Source,
+                     public DataQueue::BackpressureListener {
  public:
   using Header = NgHeaderBase<BindingData>;
 
@@ -94,30 +145,56 @@ class Stream : public AsyncWrap,
   static void InitPerContext(Realm* realm, v8::Local<v8::Object> target);
   static void RegisterExternalReferences(ExternalReferenceRegistry* registry);
 
+  // Creates a new non-pending stream.
   static BaseObjectPtr<Stream> Create(
       Session* session,
       int64_t id,
       std::shared_ptr<DataQueue> source = nullptr);
 
+  // Creates a new pending stream.
+  static BaseObjectPtr<Stream> Create(
+      Session* session,
+      Direction direction,
+      std::shared_ptr<DataQueue> source = nullptr);
+
   // The constructor is only public to be visible by MakeDetachedBaseObject.
   // Call Create to create new instances of Stream.
   Stream(BaseObjectWeakPtr<Session> session,
          v8::Local<v8::Object> obj,
          int64_t id,
          std::shared_ptr<DataQueue> source);
+
+  // Creates the stream in a pending state. The constructor is only public
+  // to be visible to MakeDetachedBaseObject. Call Create to create new
+  // instances of Stream.
+  Stream(BaseObjectWeakPtr<Session> session,
+         v8::Local<v8::Object> obj,
+         Direction direction,
+         std::shared_ptr<DataQueue> source);
+  DISALLOW_COPY_AND_MOVE(Stream)
   ~Stream() override;
 
+  // While the stream is still pending, the id will be -1.
   int64_t id() const;
+
+  // While the stream is still pending, the origin will be invalid.
   Side origin() const;
+
   Direction direction() const;
+
   Session& session() const;
 
-  bool is_destroyed() const;
+  // True if this stream was created in a pending state and is still waiting
+  // to be created.
+  bool is_pending() const;
 
   // True if we've completely sent all outbound data for this stream.
   bool is_eos() const;
 
+  // True if this stream is still in a readable state.
   bool is_readable() const;
+
+  // True if this stream is still in a writable state.
   bool is_writable() const;
 
   // Called by the session/application to indicate that the specified number
@@ -178,6 +255,8 @@ class Stream : public AsyncWrap,
 
  private:
   struct Impl;
+  struct PendingHeaders;
+
   class Outbound;
 
   // Gets a reader for the data received for this stream from the peer,
@@ -186,6 +265,9 @@ class Stream : public AsyncWrap,
   void set_final_size(uint64_t amount);
   void set_outbound(std::shared_ptr<DataQueue> source);
 
+  bool is_local_unidirectional() const;
+  bool is_remote_unidirectional() const;
+
   // JavaScript callouts
 
   // Notifies the JavaScript side that the stream has been destroyed.
@@ -198,19 +280,52 @@ class Stream : public AsyncWrap,
   // trailing headers.
   void EmitWantTrailers();
 
+  void NotifyReadableEnded(uint64_t code);
+  void NotifyWritableEnded(uint64_t code);
+
+  // When a pending stream is finally opened, the NotifyStreamOpened method
+  // will be called and the id will be assigned.
+  void NotifyStreamOpened(int64_t id);
+  void EnqueuePendingHeaders(HeadersKind kind,
+                             v8::Local<v8::Array> headers,
+                             HeadersFlags flags);
+
   AliasedStruct<Stats> stats_;
   AliasedStruct<State> state_;
   BaseObjectWeakPtr<Session> session_;
-  const Side origin_;
-  const Direction direction_;
   std::unique_ptr<Outbound> outbound_;
   std::shared_ptr<DataQueue> inbound_;
 
+  // If the stream cannot be opened yet, it will be created in a pending state.
+  // Once the owning session is able to, it will complete opening of the stream
+  // and the stream id will be assigned.
+  std::optional<std::unique_ptr<PendingStream>> maybe_pending_stream_ =
+      std::nullopt;
+  std::vector<std::unique_ptr<PendingHeaders>> pending_headers_queue_;
+  uint64_t pending_close_read_code_ = NGTCP2_APP_NOERROR;
+  uint64_t pending_close_write_code_ = NGTCP2_APP_NOERROR;
+
+  struct PendingPriority {
+    StreamPriority priority;
+    StreamPriorityFlags flags;
+  };
+  std::optional<PendingPriority> pending_priority_ = std::nullopt;
+
+  // The headers_ field holds a block of headers that have been received and
+  // are being buffered for delivery to the JavaScript side.
+  // TODO(@jasnell): Use v8::Global instead of v8::Local here.
   std::vector<v8::Local<v8::Value>> headers_;
+
+  // The headers_kind_ field indicates the kind of headers that are being
+  // buffered.
   HeadersKind headers_kind_ = HeadersKind::INITIAL;
+
+  // The headers_length_ field holds the total length of the headers that have
+  // been buffered.
   size_t headers_length_ = 0;
 
   friend struct Impl;
+  friend class PendingStream;
 
  public:
   // The Queue/Schedule/Unschedule here are part of the mechanism used to
diff --git a/src/req_wrap-inl.h b/src/req_wrap-inl.h
index 5666df17c2a041..bfcb13b9036310 100644
--- a/src/req_wrap-inl.h
+++ b/src/req_wrap-inl.h
@@ -50,7 +50,9 @@ void ReqWrap<T>::Cancel() {
 }
 
 template <typename T>
-bool ReqWrap<T>::IsDispatched() { return req_.data != nullptr; }
+bool ReqWrap<T>::IsDispatched() {
+  return req_.data != nullptr;
+}
 
 template <typename T>
 AsyncWrap* ReqWrap<T>::GetAsyncWrap() {
diff --git a/test/parallel/test-quic-internal-endpoint-stats-state.js b/test/parallel/test-quic-internal-endpoint-stats-state.js
index 2a350588f2123d..62b42169909d64 100644
--- a/test/parallel/test-quic-internal-endpoint-stats-state.js
+++ b/test/parallel/test-quic-internal-endpoint-stats-state.js
@@ -64,9 +64,7 @@ describe('quic internal endpoint stats and state', { skip: !hasQuic }, () => {
   it('state is not readable after close', () => {
     const endpoint = new QuicEndpoint();
     endpoint.state[kFinishClose]();
-    throws(() => endpoint.state.isBound, {
-      name: 'Error',
-    });
+    strictEqual(endpoint.state.isBound, undefined);
   });
 
   it('state constructor argument is ArrayBuffer', () => {
@@ -142,11 +140,11 @@ describe('quic internal endpoint stats and state', { skip: !hasQuic }, () => {
     const streamState = new QuicStreamState(kPrivateConstructor, new ArrayBuffer(1024));
     const sessionState = new QuicSessionState(kPrivateConstructor, new ArrayBuffer(1024));
 
+    strictEqual(streamState.pending, false);
     strictEqual(streamState.finSent, false);
     strictEqual(streamState.finReceived, false);
     strictEqual(streamState.readEnded, false);
     strictEqual(streamState.writeEnded, false);
-    strictEqual(streamState.destroyed, false);
     strictEqual(streamState.paused, false);
     strictEqual(streamState.reset, false);
     strictEqual(streamState.hasReader, false);
@@ -181,9 +179,9 @@ describe('quic internal endpoint stats and state', { skip: !hasQuic }, () => {
     const streamStats = new QuicStreamStats(kPrivateConstructor, new ArrayBuffer(1024));
     const sessionStats = new QuicSessionStats(kPrivateConstructor, new ArrayBuffer(1024));
     strictEqual(streamStats.createdAt, 0n);
+    strictEqual(streamStats.openedAt, 0n);
     strictEqual(streamStats.receivedAt, 0n);
     strictEqual(streamStats.ackedAt, 0n);
-    strictEqual(streamStats.closingAt, 0n);
     strictEqual(streamStats.destroyedAt, 0n);
     strictEqual(streamStats.bytesReceived, 0n);
     strictEqual(streamStats.bytesSent, 0n);