wip: almost fixed now

Just need to check for stream state change after connection receives a packet. Ideally we only check finishing streams.

* Related #10

[ci skip]

Author: tegefaulkes

src/QUICClient.ts

@@ -147,7 +147,7 @@ class QUICClient extends EventTarget {
       },
       config: quicConfig,
       logger: logger.getChild(
-        `${QUICConnection.name} ${scid.toString().slice(32)}`,
+        `${QUICConnection.name} ${scid.toString().slice(32)}-${Math.floor(Math.random() * 100)}`,
       ),
     });
     connection.addEventListener('error', handleConnectionError, { once: true });

src/QUICConnection.ts

@@ -343,6 +343,7 @@ class QUICConnection extends EventTarget {
       try {
         this.logger.debug(`Did a recv ${data.byteLength}`);
         this.conn.recv(data, recvInfo);
+        this.logger.info(`RECEIVED ${data.byteLength} of data`)
       } catch (e) {
         this.logger.error(`recv error ${e.message}`);
         // Depending on the exception, the `this.conn.recv`
@@ -385,14 +386,18 @@ class QUICConnection extends EventTarget {
         !this.conn.isDraining() &&
         (this.conn.isInEarlyData() || this.conn.isEstablished())
       ) {
+        this.logger.info('JKASDFHKSJDHFJKSDHFJK');
+        QUICStream.logStreamState(0 as StreamId, this.conn, this.logger);
         for (const streamId of this.conn.readable() as Iterable<StreamId>) {
           let quicStream = this.streamMap.get(streamId);
+          this.logger.info(`Checking stream readable ${streamId}`);
+          QUICStream.logStreamState(streamId, this.conn, this.logger);
           if (quicStream == null) {
             // The creation will set itself to the stream map
             quicStream = await QUICStream.createQUICStream({
               streamId,
               connection: this,
-              logger: this.logger.getChild(`${QUICStream.name} ${streamId}`),
+              logger: this.logger.getChild(`${QUICStream.name} ${streamId}-${Math.floor(Math.random() * 100)}`),
             });
             this.dispatchEvent(
               new events.QUICConnectionStreamEvent({ detail: quicStream }),
@@ -401,13 +406,15 @@ class QUICConnection extends EventTarget {
           quicStream.read();
         }
         for (const streamId of this.conn.writable() as Iterable<StreamId>) {
+          this.logger.info(`Checking stream writable ${streamId}`);
+          QUICStream.logStreamState(streamId, this.conn, this.logger);
           let quicStream = this.streamMap.get(streamId);
           if (quicStream == null) {
             // The creation will set itself to the stream map
             quicStream = await QUICStream.createQUICStream({
               streamId,
               connection: this,
-              logger: this.logger.getChild(`${QUICStream.name} ${streamId}`),
+              logger: this.logger.getChild(`${QUICStream.name} ${streamId}-${Math.floor(Math.random() * 100)}`),
             });
             this.dispatchEvent(
               new events.QUICConnectionStreamEvent({ detail: quicStream }),
@@ -530,6 +537,7 @@ class QUICConnection extends EventTarget {
             sendInfo.to.port,
             sendInfo.to.host,
           );
+          this.logger.info(`SENT ${sendLength} of data`)
         } catch (e) {
           this.logger.error(`send error ${e.message}`);
           this.dispatchEvent(
@@ -601,7 +609,7 @@ class QUICConnection extends EventTarget {
       const quicStream = await QUICStream.createQUICStream({
         streamId: streamId!,
         connection: this,
-        logger: this.logger.getChild(`${QUICStream.name} ${streamId!}`),
+        logger: this.logger.getChild(`${QUICStream.name} ${streamId!}-${Math.floor(Math.random() * 100)}`),
       });
       const writer = quicStream.writable.getWriter();
 

src/QUICServer.ts

@@ -265,7 +265,7 @@ class QUICServer extends EventTarget {
       remoteInfo,
       config: this.config,
       logger: this.logger.getChild(
-        `${QUICConnection.name} ${scid.toString().slice(32)}`,
+        `${QUICConnection.name} ${scid.toString().slice(32)}-${Math.floor(Math.random() * 100)}`,
       ),
     });
 

src/QUICStream.ts

@@ -48,8 +48,10 @@ class QUICStream
   protected _recvClosed: boolean = false;
   protected _recvPaused: boolean = false;
   protected resolveWritableP?: () => void;
-  public readonly finishedP: Promise<void>;
-  protected resolveFinishedP: () => void;
+  // This resolves when `streamSend` would result in a `StreamStopped(u64)` error indicating sending has ended
+  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>()
 
   /**
    * For `reasonToCode`, return 0 means "unknown reason"
@@ -106,12 +108,6 @@ class QUICStream
     this.streamMap = connection.streamMap;
     this.reasonToCode = reasonToCode;
     this.codeToReason = codeToReason;
-    const {
-      p: finishedP,
-      resolveP: resolveFinishedP
-    } = utils.promise<void>();
-    this.finishedP = finishedP;
-    this.resolveFinishedP = resolveFinishedP;
 
     // Try the BYOB later, it seems more performant
 
@@ -218,12 +214,17 @@ class QUICStream
         await this.closeSend(true, e);
       }
     }
-    await this.streamSend(new Uint8Array(0), true);
+    // await this.streamSend(new Uint8Array(0), true).catch(e => console.error(e));
+    QUICStream.logStreamState(this.streamId, this.conn, this.logger);
+    this.logger.info('Waiting for FINISH')
+    await Promise.all([
+      this.sendFinishedProm.p.then(() => console.log('send finished')),
+      this.recvFinishedProm.p.then(() => console.log('recv finished')),
+    ])
+    this.logger.info('DONE waiting for FINISH')
+    QUICStream.logStreamState(this.streamId, this.conn, this.logger);
     this.streamMap.delete(this.streamId);
-    this.logger.info(`finished ${this.conn.streamFinished(this.streamId)}`);
     // We need to wait for the connection to finish before fully destroying
-    this.logger.info('Waiting for finish')
-    await this.finishedP;
     this.dispatchEvent(new events.QUICStreamDestroyEvent());
     this.logger.info(`Destroyed ${this.constructor.name}`);
   }
@@ -234,13 +235,34 @@ class QUICStream
    */
   @ready(new errors.ErrorQUICStreamDestroyed(), false, ['destroying'])
   public read(): void {
-    // TODO: check if the readable has errored out?
-    this.logger.info(`finished read ${this.conn.streamFinished(this.streamId)}`);
+    // After reading it's possible the writer had a state change.
+    try {
+      this.conn.streamWritable(this.streamId, 0)
+      this.logger.info('writable still good')
+    } catch (e) {
+      this.logger.info(e.message);
+      // const reason = await this.processSendStreamError(e, 'send');
+      // If the writable has ended, we need to close the writable.
+      this.logger.info('writable has ended')
+      this.sendFinishedProm.resolveP();
+      if (!this._sendClosed) {
+        const err = Error('reason');
+        this.writableController.error(err)
+        void this.closeSend(true, err);
+      }
+    }
     if (this.conn.streamFinished(this.streamId)) {
-      this.resolveFinishedP();
+      // Stream has finished
+      this.recvFinishedProm.resolveP();
+      if (!this._recvClosed) {
+        const err = Error('TMP ERROR');
+        this.readableController.error(err)
+        void this.closeRecv(true, err);
+      }
     }
     if (this._recvPaused) {
       // Do nothing if we are paused
+      this.logger.info('Skipping read, paused');
       return;
     }
     void this.streamRecv();
@@ -252,18 +274,28 @@ class QUICStream
    */
   @ready(new errors.ErrorQUICStreamDestroyed(), false, ['destroying'])
   public write(): void {
-    this.logger.info(`finished write ${this.conn.streamFinished(this.streamId)}`);
-    if (this.conn.streamFinished(this.streamId)) {
-      this.resolveFinishedP();
-    }
     try {
       this.conn.streamWritable(this.streamId, 0)
     } catch (e) {
       this.logger.info(e.message);
       // const reason = await this.processSendStreamError(e, 'send');
       // If the writable has ended, we need to close the writable.
-      this.writableController.error(Error('TMP ERROR'))
-      void this.closeSend();
+      this.logger.info('writable has ended')
+      this.sendFinishedProm.resolveP();
+      if (!this._sendClosed) {
+        const err = Error('TMP ERROR');
+        this.writableController.error(err)
+        void this.closeSend(true, err);
+      }
+    }
+    if (this.conn.streamFinished(this.streamId)) {
+      // Stream has finished
+      this.recvFinishedProm.resolveP();
+      if (!this._recvClosed) {
+        const err = Error('TMP ERROR');
+        this.readableController.error(err)
+        void this.closeRecv(true, err);
+      }
     }
     if (this.resolveWritableP != null) {
       this.resolveWritableP();
@@ -273,6 +305,7 @@ class QUICStream
   protected async streamRecv(): Promise<void> {
     const buf = Buffer.alloc(1024);
     let recvLength: number, fin: boolean;
+    this.logger.info('trying receiving');
     try {
       [recvLength, fin] = this.conn.streamRecv(this.streamId, buf);
     } catch (e) {
@@ -285,18 +318,18 @@ class QUICStream
         // or through an exception here where the stream reports an error
         // Since we don't call this method unless it is readable
         // This should never be reported... (this branch should be dead code)
-        this.logger.debug('Stream reported: done');
+        this.logger.info('Stream reported: done');
         return;
       } else {
-        this.logger.debug('Stream reported: error');
-        const match = e.message.match(/StreamReset\((.+)\)/);
-        if (match != null) {
+        this.logger.info('Stream reported: error');
+        // Signal receiving has ended
+        this.recvFinishedProm.resolveP();
+        const reason = await this.processSendStreamError(e, 'recv');
+        if (reason != null) {
           // If it is `StreamReset(u64)` error, then the peer has closed
           // the stream, and we are receiving the error code
-          const code = parseInt(match[1]);
-          const reason = await this.codeToReason('recv', code);
           this.readableController.error(reason);
-          await this.closeRecv();
+          await this.closeRecv(true, reason);
         } else {
           // If it is not a `StreamReset(u64)`, then something else broke
           // and we need to propagate the error up and down the stream
@@ -320,6 +353,8 @@ class QUICStream
       this.logger.info('Stream reported: fin');
       if (!this._recvClosed) this.readableController.close();
       await this.closeRecv();
+      // Signal receiving has ended
+      this.recvFinishedProm.resolveP();
       return;
     }
     // Now we pause receiving if the queue is full
@@ -351,6 +386,8 @@ class QUICStream
         // This ensures that we are always blocked below.
         sentLength = -1;
       } else {
+        // Signal sending has ended
+        this.sendFinishedProm.resolveP();
         // We may receive a `StreamStopped(u64)` exception
         // meaning the peer has signalled for us to stop writing
         // If this occurs, we need to go back to the writable stream
@@ -360,7 +397,7 @@ class QUICStream
         const reason = await this.processSendStreamError(e, 'send');
         if (reason != null) {
           // We have to close the send side (but the stream is already closed)
-          await this.closeSend();
+          await this.closeSend(true, e);
           // Throws the exception back to the writer
           throw reason;
         } else {
@@ -410,13 +447,13 @@ class QUICStream
       if (e.message !== 'Done') throw e;
     }
     await this.connection.send();
+    QUICStream.logStreamState(this.streamId, this.conn, this.logger);
     if (this[status] !== 'destroying' && this._recvClosed && this._sendClosed) {
       // Only destroy if we are not already destroying
       // and that both recv and send is closed
-      await this.destroy();
+      void this.destroy();
     }
     this.logger.info(`Closed Recv`);
-    this.logger.info(`finished ${this.conn.streamFinished(this.streamId)}`);
   }
 
   /**
@@ -448,10 +485,9 @@ class QUICStream
     if (this[status] !== 'destroying' && this._recvClosed && this._sendClosed) {
       // Only destroy if we are not already destroying
       // and that both recv and send is closed
-      await this.destroy();
+      void this.destroy();
     }
     this.logger.info(`Closed Send`);
-    this.logger.info(`finished ${this.conn.streamFinished(this.streamId)}`);
   }
 
   protected async processSendStreamError(e: Error, type: 'recv' | 'send' ): Promise<any | null> {
@@ -464,6 +500,28 @@ class QUICStream
     }
     return null;
   }
+
+  public static logStreamState(streamId: StreamId, conn: Connection, logger: Logger) {
+    let message = 'STATE:';
+    try {
+      conn.streamWritable(streamId, 0);
+      message += 'W'
+    } catch (e) {
+      message += '!W'
+    }
+    try {
+      if (conn.streamReadable(streamId)) message += 'R';
+      else message += '!R';
+    } catch (e) {
+      message += '!R'
+    }
+    if (conn.streamFinished(streamId)) {
+      message += 'F'
+    }else {
+      message += '!F'
+    }
+    logger.info(message);
+  }
 }
 
 export default QUICStream;

tests/QUICStream.test.ts

@@ -207,19 +207,20 @@ describe(QUICStream.name, () => {
         console.log('a');
         await Promise.race([
           streamEndedProm.p,
-          sleep(5000).then(() => {
-            throw Error("Creation timed out");
+          sleep(1000).then(() => {
+            throw Error("Ending timed out");
           }),
         ])
         console.log('a');
         expect(streamCreatedCount).toBe(streamsNum);
         expect(streamEndedCount).toBe(streamsNum);
         console.log('TEST DONE')
       } catch (e) {
-        console.error(e);
+        console.log(e.message);
+        throw e;
       } finally {
         console.log('Clean up');
-        await sleep(5000);
+        await sleep(1000);
         console.log('cleaning client');
         await client?.destroy({ force: true });
         console.log('cleaning server');