wip: still trying to fix

* Related #4

[ci skip]

Author: tegefaulkes

src/QUICConnection.ts

@@ -436,10 +436,7 @@ class QUICConnection extends EventTarget {
         if (this.resolveCloseP != null) this.resolveCloseP();
         return;
       }
-      if (
-        !this.conn.isDraining() &&
-        (this.conn.isInEarlyData() || this.conn.isEstablished())
-      ) {
+      if (this.conn.isInEarlyData() || this.conn.isEstablished()) {
         for (const streamId of this.conn.readable() as Iterable<StreamId>) {
           let quicStream = this.streamMap.get(streamId);
           if (quicStream == null) {
@@ -457,7 +454,6 @@ class QUICConnection extends EventTarget {
               new events.QUICConnectionStreamEvent({ detail: quicStream }),
             );
           }
-          this.logger.info('processing read');
           quicStream.read();
         }
         for (const streamId of this.conn.writable() as Iterable<StreamId>) {
@@ -476,18 +472,19 @@ class QUICConnection extends EventTarget {
               new events.QUICConnectionStreamEvent({ detail: quicStream }),
             );
           }
-          this.logger.info('processing write');
           quicStream.write();
         }
-        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 {
+      const nums: Array<number> = [];
+      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.
+        nums.push(streamId);
+        quicStream.read();
+      }
+      this.logger.info(`checking read finally recv for ${nums}`);
       this.logger.debug('RECV FINALLY');
       // Set the timeout
       this.checkTimeout();
@@ -500,7 +497,7 @@ class QUICConnection extends EventTarget {
       ) {
         this.logger.debug('CALLING DESTROY 2');
         // Destroy in the background, we still need to process packets
-        void this.destroy();
+        void this.destroy().catch(() => {});
       }
     }
   }
@@ -596,6 +593,7 @@ class QUICConnection extends EventTarget {
           );
           this.logger.debug(`SENT ${sendLength} of data`);
         } catch (e) {
+          console.error(e);
           this.logger.error(`send error ${e.message}`);
           this.dispatchEvent(
             new events.QUICConnectionErrorEvent({ detail: e }),
@@ -604,13 +602,13 @@ class QUICConnection extends EventTarget {
         }
       }
     } 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();
+      const nums: Array<number> = [];
+      for (const [streamId, quicStream] of this.streamMap) {
+        // Stream sending can finish after a packet is sent
+        nums.push(streamId);
+        quicStream.read();
       }
+      this.logger.info(`checking read finally send for ${nums}`);
       this.logger.debug('SEND FINALLY');
       this.checkTimeout();
       if (
@@ -700,7 +698,7 @@ class QUICConnection extends EventTarget {
     ) {
       this.logger.debug('CALLING DESTROY 3');
       // Destroy in the background, we still need to process packets
-      void this.destroy();
+      void this.destroy().catch(() => {});
     }
     this.checkTimeout();
   };

src/QUICStream.ts

@@ -161,6 +161,7 @@ class QUICStream
         },
         write: async (chunk: Uint8Array) => {
           await this.streamSend(chunk);
+          void this.connection.send().catch(() => {});
         },
         close: async () => {
           // This gracefully closes, by sending a message at the end
@@ -169,11 +170,13 @@ class QUICStream
           // If this itself results in an error, we can continue
           // But continue to do the below
           this.logger.info('sending fin frame');
+          // This.sendFinishedProm.resolveP();
           await this.streamSend(new Uint8Array(0), true).catch((e) => {
             // Ignore send error if stream is already closed
             if (e.message !== 'send') throw e;
           });
           await this.closeSend();
+          void this.connection.send().catch(() => {});
         },
         abort: async (reason?: any) => {
           // Abort can be called even if there are writes are queued up
@@ -221,7 +224,7 @@ class QUICStream
       this.writableController.error(e);
       await this.closeSend(true, e);
     }
-    await this.connection.send();
+    void this.connection.send().catch(() => {});
     this.logger.debug('waiting for underlying streams to finish');
     this.isFinished();
     // We need to wait for the connection to finish before fully destroying
@@ -241,10 +244,11 @@ class QUICStream
     // After reading it's possible the writer had a state change.
     this.isSendFinished();
     if (this._recvPaused) {
+      console.log('SKIPPING!');
       // Do nothing if we are paused
       return;
     }
-    void this.streamRecv();
+    void this.streamRecv().catch(() => {});
   }
 
   /**
@@ -254,7 +258,7 @@ class QUICStream
   @ready(new errors.ErrorQUICStreamDestroyed(), false, ['destroying'])
   public write(): void {
     // Checking if writable has ended
-    void this.isSendFinished();
+    this.isSendFinished();
     if (this.resolveWritableP != null) {
       this.resolveWritableP();
     }
@@ -283,7 +287,7 @@ class QUICStream
           'Readable stream closed early with no reason',
         );
         this.readableController.error(err);
-        void this.closeRecv(true, err);
+        void this.closeRecv(true, err).catch(() => {});
       }
     }
     return recvFinished;
@@ -301,6 +305,7 @@ class QUICStream
     } catch (e) {
       // If the writable has ended, we need to close the writable.
       // We need to do this in the background to keep this synchronous.
+      this.sendFinishedProm.resolveP();
       void this.processSendStreamError(e, 'send').then((reason) => {
         if (!this._sendClosed) {
           const err =
@@ -309,10 +314,10 @@ class QUICStream
               'Writable stream closed early with no reason',
             );
           this.writableController.error(err);
-          void this.closeSend(true, err);
+          void this.closeSend(true, err).catch(() => {});
         }
-        this.sendFinishedProm.resolveP();
       });
+      this.logger.info('send FINISHED');
       return true;
     }
   }
@@ -321,63 +326,63 @@ class QUICStream
     const buf = Buffer.alloc(1024);
     let recvLength: number, fin: boolean;
     this.logger.debug('trying receiving');
-    try {
-      [recvLength, fin] = this.conn.streamRecv(this.streamId, buf);
-    } catch (e) {
-      if (e.message === 'Done') {
-        // When it is reported to be `Done`, it just means that there is no data to read
-        // it does not mean that the stream is closed or finished
-        // In such a case, we just ignore and continue
-        // However after the stream is closed, then it would continue to return `Done`
-        // This can only occur in 2 ways, either via the `fin`
-        // 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)
-        return;
-      } else {
-        this.logger.info('Stream reported: error');
-        this.logger.error(`Stream reported: error ${e.message}`);
-        // 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
-          this.readableController.error(reason);
-          await this.closeRecv(true, reason);
+    while (true) {
+      try {
+        [recvLength, fin] = this.conn.streamRecv(this.streamId, buf);
+      } catch (e) {
+        if (e.message === 'Done') {
+          // When it is reported to be `Done`, it just means that there is no data to read
+          // it does not mean that the stream is closed or finished
+          // In such a case, we just ignore and continue
+          // However after the stream is closed, then it would continue to return `Done`
+          // This can only occur in 2 ways, either via the `fin`
+          // 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)
+          return;
         } else {
-          // If it is not a `StreamReset(u64)`, then something else broke
-          // and we need to propagate the error up and down the stream
-          this.readableController.error(e);
-          await this.closeRecv(true, e);
+          this.logger.info('Stream reported: error');
+          this.logger.error(`Stream reported: error ${e.message}`);
+          // Signal receiving has ended
+          this.recvFinishedProm.resolveP();
+          if (!this._recvClosed) {
+            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
+              this.readableController.error(reason);
+              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
+              this.readableController.error(e);
+              await this.closeRecv(true, e);
+            }
+          }
+          return;
         }
-        return;
       }
-    } finally {
-      // Let's check if sending side has finished
-      await this.connection.send();
-    }
 
-    // If fin is true, then that means, the stream is CLOSED
-    if (fin) {
-      // This will render `stream.cancel` a noop
-      this.logger.info('Stream reported: fin');
-      if (!this._recvClosed) this.readableController.close();
-      await this.closeRecv();
-      // Signal receiving has ended
-      this.recvFinishedProm.resolveP();
-      return;
-    }
-    // Only fin packets are 0 length, so we enqueue after checking fin
-    if (!this._recvClosed) {
-      this.readableController.enqueue(buf.subarray(0, recvLength));
-    }
-    // Now we pause receiving if the queue is full
-    if (
-      this.readableController.desiredSize != null &&
-      this.readableController.desiredSize <= 0
-    ) {
-      this._recvPaused = true;
+      // If fin is true, then that means, the stream is CLOSED
+      if (fin) {
+        // This will render `stream.cancel` a noop
+        this.logger.info('Stream reported: fin');
+        if (!this._recvClosed) this.readableController.close();
+        await this.closeRecv();
+        // Signal receiving has ended
+        this.recvFinishedProm.resolveP();
+        return;
+      }
+      if (!this._recvClosed) {
+        this.readableController.enqueue(buf.subarray(0, recvLength));
+      }
+      // Now we pause receiving if the queue is full
+      if (
+        this.readableController.desiredSize != null &&
+        this.readableController.desiredSize <= 0
+      ) {
+        this._recvPaused = true;
+      }
     }
   }
 
@@ -423,8 +428,6 @@ class QUICStream
           throw e;
         }
       }
-    } finally {
-      await this.connection.send();
     }
     if (sentLength < chunk.length) {
       const { p: writableP, resolveP: resolveWritableP } = utils.promise();
@@ -447,6 +450,7 @@ class QUICStream
     isError: boolean = false,
     reason?: any,
   ): Promise<void> {
+    if (isError) this.logger.error(reason.message);
     // Further closes are NOPs
     if (this._recvClosed) return;
     this.logger.info(`Close Recv`);
@@ -481,6 +485,7 @@ class QUICStream
     isError: boolean = false,
     reason?: any,
   ): Promise<void> {
+    if (isError) this.logger.error(reason.message);
     // Further closes are NOPs
     if (this._sendClosed) return;
     this.logger.info(`Close Send`);
@@ -530,6 +535,39 @@ class QUICStream
     }
     return null;
   }
+
+  static checkStreamStates(
+    conn: Connection,
+    streamId: number,
+    message: string,
+    logger: Logger,
+  ) {
+    const fin = conn.streamFinished(streamId);
+    const read = conn.streamReadable(streamId);
+    let write: boolean | string;
+    try {
+      write = conn.streamWritable(streamId, 0);
+    } catch (e) {
+      write = e.message;
+    }
+    let cap: number | string;
+    try {
+      cap = conn.streamCapacity(streamId);
+    } catch (e) {
+      cap = e.message;
+    }
+    let readIter = false;
+    for (const id of conn.readable()) {
+      if (streamId === id) readIter = true;
+    }
+    let writeIter = false;
+    for (const id of conn.writable()) {
+      if (streamId === id) writeIter = true;
+    }
+    logger.info(
+      `Stream states (${message}) iterRW(${readIter}, ${writeIter}),finished(${fin}), read(${read}), write(${write}), capacity(${cap})`,
+    );
+  }
 }
 
 export default QUICStream;