feat: added forceDestroyStreams to connections, server and client to allow force closing streams

Allows us to destroy a connection with `force: false` for force the streams to end later if needed.

[ci skip]

Author: tegefaulkes

src/QUICConnection.ts

@@ -31,7 +31,6 @@ import { Shutdown } from './native/types';
 import * as utils from './utils';
 import * as events from './events';
 import * as errors from './errors';
-import { sleep } from '../tests/utils';
 
 interface QUICConnection extends StartStop {}
 @StartStop({
@@ -621,7 +620,7 @@ class QUICConnection {
 
     // Yield to allow any background processing to settle before proceeding.
     // This will allow any streams to process buffers before continuing
-    await sleep(0);
+    await utils.yieldMicro();
 
     // Destroy all streams
     const streamsDestroyP: Array<Promise<void>> = [];
@@ -1155,6 +1154,21 @@ class QUICConnection {
     return quicStream;
   }
 
+  /**
+   * Destroys all active streams without closing the connection.
+   *
+   * If there are no active streams then it will do nothing.
+   * If the connection is stopped with `force: false` then this can be used
+   * to force close any streams `stop` is waiting for to end.
+   *
+   * Destruction will occur in the background.
+   */
+  public destroyStreams(reason?: any) {
+    for (const quicStream of this.streamMap.values()) {
+      quicStream.cancel(reason);
+    }
+  }
+
   /**
    * Starts the keep alive interval timer.
    *

src/utils.ts

@@ -17,6 +17,13 @@ import * as errors from './errors';
 const textEncoder = new TextEncoder();
 const textDecoder = new TextDecoder('utf-8');
 
+/**
+ * Used to yield to the event loop to allow other micro tasks to process
+ */
+async function yieldMicro(): Promise<void> {
+  return await new Promise<void>((r) => queueMicrotask(r));
+}
+
 /**
  * Convert callback-style to promise-style
  * If this is applied to overloaded function
@@ -550,6 +557,7 @@ function isStreamReset(e: Error): number | false {
 export {
   textEncoder,
   textDecoder,
+  yieldMicro,
   promisify,
   promise,
   bufferWrap,

tests/QUICStream.test.ts

@@ -1313,6 +1313,8 @@ describe(QUICStream.name, () => {
         })(),
       );
     }
+    // Yield to allow streams to propagate
+    await sleep(0);
 
     // Start unforced close of client
     const clientDestroyP = client.destroy({ force: false });
@@ -1402,6 +1404,8 @@ describe(QUICStream.name, () => {
         })(),
       );
     }
+    // Yield to allow streams to propagate
+    await sleep(0);
 
     // Start unforced close of server
     const serverStopP = server.stop({ force: false });
@@ -1489,7 +1493,87 @@ describe(QUICStream.name, () => {
     await expect(asd).rejects.toThrow('read 1');
 
     waitResolveP();
+    await Promise.all(activeServerStreams);
+    await clientDestroyP;
+    await server.stop({ force: true });
+  });
+  test('connection can be forced closed after unforced destroy', async () => {
+    const message = Buffer.from('The Quick Brown Fox Jumped Over The Lazy Dog');
+    const connectionEventProm =
+      utils.promise<events.EventQUICServerConnection>();
+    const tlsConfig = await generateTLSConfig(defaultType);
+    const server = new QUICServer({
+      crypto: {
+        key,
+        ops: serverCrypto,
+      },
+      logger: logger.getChild(QUICServer.name),
+      config: {
+        key: tlsConfig.leafKeyPairPEM.privateKey,
+        cert: tlsConfig.leafCertPEM,
+        verifyPeer: false,
+      },
+    });
+    socketCleanMethods.extractSocket(server);
+    server.addEventListener(
+      events.EventQUICServerConnection.name,
+      (e: events.EventQUICServerConnection) => connectionEventProm.resolveP(e),
+    );
+    await server.start({
+      host: localhost,
+    });
+    const client = await QUICClient.createQUICClient({
+      host: localhost,
+      port: server.port,
+      localHost: localhost,
+      crypto: {
+        ops: clientCrypto,
+      },
+      logger: logger.getChild(QUICClient.name),
+      config: {
+        verifyPeer: false,
+      },
+    });
+    socketCleanMethods.extractSocket(client);
+    const conn = (await connectionEventProm.p).detail;
+
+    // Do the test
+    const { p: waitP, resolveP: waitResolveP } = utils.promise();
+    const activeServerStreams: Array<Promise<void>> = [];
+    conn.addEventListener(
+      events.EventQUICConnectionStream.name,
+      async (streamEvent: events.EventQUICConnectionStream) => {
+        const stream = streamEvent.detail;
+        await waitP;
+        const streamProm = stream.readable
+          .pipeTo(stream.writable)
+          .catch(() => {});
+        activeServerStreams.push(streamProm);
+      },
+    );
+
+    const stream = client.connection.newStream();
+    const writer = stream.writable.getWriter();
+    await writer.write(message);
+    await writer.close();
+
+    // Start unforced close of client
+    const clientDestroyP = client.destroy({ force: false });
+
+    const result = await Promise.race([
+      clientDestroyP.then(() => true),
+      sleep(500).then(() => false),
+    ]);
+
+    expect(result).toBe(false);
+
+    // We can force close the streams causing client destruction to end
+    client.connection.destroyStreams();
     await clientDestroyP;
+    await Promise.allSettled(activeServerStreams);
+
     await server.stop({ force: true });
+    waitResolveP();
+    await waitP;
   });
 });

tests/utils.ts

@@ -23,10 +23,6 @@ async function sleep(ms: number): Promise<void> {
   return await new Promise<void>((r) => setTimeout(r, ms));
 }
 
-async function yieldMicro(): Promise<void> {
-  return await new Promise<void>((r) => queueMicrotask(r));
-}
-
 async function randomBytes(data: ArrayBuffer) {
   webcrypto.getRandomValues(new Uint8Array(data));
 }
@@ -853,7 +849,6 @@ function createReasonConverters() {
 
 export {
   sleep,
-  yieldMicro,
   randomBytes,
   generateKeyPairRSA,
   generateKeyPairECDSA,