wip: still fixing the problem

* Related #14

[ci skip]

Author: tegefaulkes

src/QUICConnection.ts

@@ -46,7 +46,6 @@ class QUICConnection extends EventTarget {
   protected codeToReason: StreamCodeToReason;
   protected maxReadableStreamBytes: number | undefined;
   protected maxWritableStreamBytes: number | undefined;
-  protected destroyingMap: Map<StreamId, QUICStream> = new Map();
 
   // This basically allows one to await this promise
   // once resolved, always resolved...
@@ -471,7 +470,6 @@ class QUICConnection extends EventTarget {
             quicStream = await QUICStream.createQUICStream({
               streamId,
               connection: this,
-              destroyingMap: this.destroyingMap,
               codeToReason: this.codeToReason,
               reasonToCode: this.reasonToCode,
               maxReadableStreamBytes: this.maxReadableStreamBytes,
@@ -482,6 +480,7 @@ class QUICConnection extends EventTarget {
               new events.QUICConnectionStreamEvent({ detail: quicStream }),
             );
           }
+          this.logger.info('processing read');
           quicStream.read();
           quicStream.dispatchEvent(new events.QUICStreamReadableEvent());
         }
@@ -494,7 +493,6 @@ class QUICConnection extends EventTarget {
               connection: this,
               codeToReason: this.codeToReason,
               reasonToCode: this.reasonToCode,
-              destroyingMap: this.destroyingMap,
               maxReadableStreamBytes: this.maxReadableStreamBytes,
               logger: this.logger.getChild(`${QUICStream.name} ${streamId}`),
             });
@@ -503,15 +501,15 @@ class QUICConnection extends EventTarget {
             );
           }
           quicStream.dispatchEvent(new events.QUICStreamWritableEvent());
+          this.logger.info('processing write');
           quicStream.write();
         }
-        // Checking shortlist if streams have finished.
-        for (const [streamId, stream] of this.destroyingMap) {
-          if (stream.isFinished()) {
-            // If it has finished, it will trigger its own clean up.
-            // Remove the stream from the shortlist.
-            this.destroyingMap.delete(streamId);
-          }
+        for (const [streamId, quicStream] of this.streamMap) {
+          // Checking if state has changed, we need to check all streams here for two reasons
+          //  1. Reading can end, but we won't know since it won't be listed as readable.
+          //  2. Writing can end, but we won't know unless we check here or attempt a write.
+          if (quicStream[status] === 'destroying') quicStream.isRecvFinished();
+          quicStream.isSendFinished();
         }
       }
     } finally {
@@ -632,6 +630,13 @@ class QUICConnection extends EventTarget {
         this.dispatchEvent(new events.QUICConnectionSendEvent());
       }
     } finally {
+      for (const [, quicStream] of this.streamMap) {
+        // Checking if state has changed, we need to check all streams here for two reasons
+        //  1. Reading can end, but we won't know since it won't be listed as readable.
+        //  2. Writing can end, but we won't know unless we check here or attempt a write.
+        if (quicStream[status] === 'destroying') quicStream.isRecvFinished();
+        quicStream.isSendFinished();
+      }
       this.logger.debug('SEND FINALLY');
       this.checkTimeout();
       if (
@@ -689,7 +694,6 @@ class QUICConnection extends EventTarget {
         connection: this,
         codeToReason: this.codeToReason,
         reasonToCode: this.reasonToCode,
-        destroyingMap: this.destroyingMap,
         maxReadableStreamBytes: this.maxReadableStreamBytes,
         maxWritableStreamBytes: this.maxWritableStreamBytes,
         logger: this.logger.getChild(`${QUICStream.name} ${streamId!}`),

src/QUICStream.ts

@@ -53,7 +53,6 @@ class QUICStream
   protected sendFinishedProm = utils.promise<void>();
   // This resolves when `streamRecv` results in a `StreamReset(u64)` or a fin flag indicating receiving has ended
   protected recvFinishedProm = utils.promise<void>();
-  protected destroyingMap: Map<StreamId, QUICStream>;
 
   /**
    * For `reasonToCode`, return 0 means "unknown reason"
@@ -66,7 +65,6 @@ class QUICStream
   public static async createQUICStream({
     streamId,
     connection,
-    destroyingMap,
     reasonToCode = () => 0,
     codeToReason = (type, code) =>
       new Error(`${type.toString()} ${code.toString()}`),
@@ -76,7 +74,6 @@ class QUICStream
   }: {
     streamId: StreamId;
     connection: QUICConnection;
-    destroyingMap: Map<StreamId, QUICStream>;
     reasonToCode?: StreamReasonToCode;
     codeToReason?: StreamCodeToReason;
     maxReadableStreamBytes?: number;
@@ -96,7 +93,6 @@ class QUICStream
       connection,
       reasonToCode,
       codeToReason,
-      destroyingMap,
       maxReadableStreamBytes,
       maxWritableStreamBytes,
       logger,
@@ -111,7 +107,6 @@ class QUICStream
     connection,
     reasonToCode,
     codeToReason,
-    destroyingMap,
     maxReadableStreamBytes,
     maxWritableStreamBytes,
     logger,
@@ -120,7 +115,6 @@ class QUICStream
     connection: QUICConnection;
     reasonToCode: StreamReasonToCode;
     codeToReason: StreamCodeToReason;
-    destroyingMap: Map<StreamId, QUICStream>;
     maxReadableStreamBytes: number;
     maxWritableStreamBytes: number;
     logger: Logger;
@@ -133,7 +127,13 @@ class QUICStream
     this.streamMap = connection.streamMap;
     this.reasonToCode = reasonToCode;
     this.codeToReason = codeToReason;
-    this.destroyingMap = destroyingMap;
+
+    this.sendFinishedProm.p.finally(() =>
+      logger.info('sendFinishedProm resolved'),
+    );
+    this.recvFinishedProm.p.finally(() =>
+      logger.info('recvFinishedProm resolved'),
+    );
 
     this.readable = new ReadableStream(
       {
@@ -238,16 +238,12 @@ class QUICStream
       await this.closeSend(true, e);
     }
     await this.connection.send();
-    // Await this.streamSend(new Uint8Array(0), true).catch(e => console.error(e));
     this.logger.debug('waiting for underlying streams to finish');
-    this.destroyingMap.set(this.streamId, this);
     this.isFinished();
+    // We need to wait for the connection to finish before fully destroying
     await Promise.all([this.sendFinishedProm.p, this.recvFinishedProm.p]);
     this.logger.debug('done waiting for underlying streams to finish');
     this.streamMap.delete(this.streamId);
-    // Remove from the shortlist, just in case
-    this.destroyingMap.delete(this.streamId);
-    // We need to wait for the connection to finish before fully destroying
     this.dispatchEvent(new events.QUICStreamDestroyEvent());
     this.logger.info(`Destroyed ${this.constructor.name}`);
   }
@@ -370,7 +366,7 @@ class QUICStream
         // This should never be reported... (this branch should be dead code)
         return;
       } else {
-        this.logger.debug('Stream reported: error');
+        this.logger.info('Stream reported: error');
         this.logger.error(`Stream reported: error ${e.message}`);
         // Signal receiving has ended
         this.recvFinishedProm.resolveP();
@@ -396,7 +392,7 @@ class QUICStream
     // If fin is true, then that means, the stream is CLOSED
     if (fin) {
       // This will render `stream.cancel` a noop
-      this.logger.debug('Stream reported: fin');
+      this.logger.info('Stream reported: fin');
       if (!this._recvClosed) this.readableController.close();
       await this.closeRecv();
       // Signal receiving has ended
@@ -484,7 +480,7 @@ class QUICStream
   ): Promise<void> {
     // Further closes are NOPs
     if (this._recvClosed) return;
-    this.logger.debug(`Close Recv`);
+    this.logger.info(`Close Recv`);
     // Indicate that the receiving side is closed
     this._recvClosed = true;
     const code = isError ? await this.reasonToCode('send', reason) : 0;
@@ -518,7 +514,7 @@ class QUICStream
   ): Promise<void> {
     // Further closes are NOPs
     if (this._sendClosed) return;
-    this.logger.debug(`Close Send`);
+    this.logger.info(`Close Send`);
     // Indicate that the sending side is closed
     this._sendClosed = true;
     // If the QUIC stream is already closed
@@ -551,14 +547,18 @@ class QUICStream
     e: Error,
     type: 'recv' | 'send',
   ): Promise<any | null> {
-    const match =
+    let match =
       e.message.match(/StreamStopped\((.+)\)/) ??
-      e.message.match(/InvalidStreamState\((.+)\)/) ??
       e.message.match(/StreamReset\((.+)\)/);
     if (match != null) {
       const code = parseInt(match[1]);
       return await this.codeToReason(type, code);
     }
+    match = e.message.match(/InvalidStreamState\((.+)\)/);
+    if (match != null) {
+      // `InvalidStreamState()` returns the stream Id and not any actual error code
+      return await this.codeToReason(type, 0);
+    }
     return null;
   }
 }

tests/concurrency.test.ts

@@ -4,6 +4,7 @@ import type { Crypto, Host, Port } from '@';
 import type { TlsConfig } from '@/config';
 import { fc, testProp } from '@fast-check/jest';
 import Logger, { formatting, LogLevel, StreamHandler } from '@matrixai/logger';
+import { status } from '@matrixai/async-init/dist/CreateDestroy';
 import QUICServer from '@/QUICServer';
 import { promise } from '@/utils';
 import QUICClient from '@/QUICClient';
@@ -24,6 +25,69 @@ describe('Concurrency tests', () => {
     ops: Crypto;
   };
 
+  // Tracking resources
+  let sockets: Array<QUICSocket>;
+
+  const clearResourcesList = () => {
+    sockets = [];
+  };
+
+  const listResources = () => {
+    logger.info(`RESOURCE REPORT`);
+    for (const [socketNum, socket] of sockets.entries()) {
+      logger.info(`Socket-${socketNum} status ${socket[status]}`);
+      const serverConnections = socket.connectionMap.serverConnections;
+      logger.info(`severConnections ${serverConnections.size}`);
+      for (const [connId, connection] of serverConnections) {
+        logger.info(`connectionId-${connId}`);
+        const serverStreams = connection.streamMap;
+        for (const [streamId, stream] of serverStreams) {
+          logger.info(
+            `streamId-${streamId} ${stream.recvClosed},${stream.sendClosed}`,
+          );
+        }
+      }
+
+      const clientConnections = socket.connectionMap.clientConnections;
+      logger.info(`clientConnections ${clientConnections.size}`);
+      for (const [connId, connection] of clientConnections) {
+        logger.info(`connectionId-${connId}`);
+        const serverStreams = connection.streamMap;
+        for (const [streamId, stream] of serverStreams) {
+          logger.info(
+            `streamId-${streamId} ${stream.recvClosed},${stream.sendClosed}`,
+          );
+        }
+      }
+    }
+  };
+
+  const destroyResources = async () => {
+    const socketDestroyProms: Array<Promise<void>> = [];
+    for (const [, socket] of sockets.entries()) {
+      const connectionDestroyProms: Array<Promise<void>> = [];
+      for (const [, connection] of socket.connectionMap) {
+        const streamDestroyProms: Array<Promise<void>> = [];
+        for (const [, stream] of connection.streamMap) {
+          streamDestroyProms.push(stream.destroy());
+        }
+        connectionDestroyProms.push(
+          (async () => {
+            await Promise.allSettled(streamDestroyProms);
+            await connection.destroy({ force: true });
+          })(),
+        );
+      }
+      socketDestroyProms.push(
+        (async () => {
+          await Promise.allSettled(connectionDestroyProms);
+          await socket.stop();
+        })(),
+      );
+    }
+    return await Promise.allSettled(socketDestroyProms);
+  };
+
   beforeEach(async () => {
     crypto = {
       key: await testsUtils.generateKey(),
@@ -33,6 +97,13 @@ describe('Concurrency tests', () => {
         randomBytes: testsUtils.randomBytes,
       },
     };
+    clearResourcesList();
+  });
+
+  afterEach(async () => {
+    console.log('AFTER EACH');
+    listResources();
+    await destroyResources();
   });
 
   /**
@@ -455,7 +526,6 @@ describe('Concurrency tests', () => {
       );
     }
   };
-  jest.setTimeout(100000);
   testProp.only(
     'Multiple clients sharing a socket with a server',
     [
@@ -474,6 +544,11 @@ describe('Concurrency tests', () => {
       serverStreams1,
       serverStreams2,
     ) => {
+      const clientsInfosA = clientDatas1.map((v) => v.streams.length);
+      const clientsInfosB = clientDatas2.map((v) => v.streams.length);
+      logger.info(`clientsA: ${clientsInfosA}`);
+      logger.info(`clientsB: ${clientsInfosB}`);
+
       const tlsConfig1 = await tlsConfigProm1;
       const tlsConfig2 = await tlsConfigProm2;
       const cleanUpHoldProm = promise<void>();
@@ -486,6 +561,8 @@ describe('Concurrency tests', () => {
         crypto,
         logger: logger.getChild('socket'),
       });
+      sockets.push(socket1);
+      sockets.push(socket2);
       await socket1.start({
         host: '127.0.0.1' as Host,
       });
@@ -556,10 +633,12 @@ describe('Concurrency tests', () => {
       logger.info('STARTING TEST');
       try {
         await (async () => {
+          console.log('waiting for client proms');
           await Promise.all([
             Promise.all(clientProms1),
             Promise.all(clientProms2),
           ]);
+          console.log('DONE waiting for client proms');
           // Allow for streams to be negotiated
           await sleep(200);
           cleanUpHoldProm.resolveP();
@@ -583,6 +662,7 @@ describe('Concurrency tests', () => {
           )}`,
         );
       }
+      logger.info('CLOSING SOCKETS');
       await socket1.stop();
       await socket2.stop();
       logger.info('TEST FULLY DONE!');