fix: better handling for stream cancellation

Author: tegefaulkes

src/QUICConnection.ts

@@ -895,8 +895,10 @@ class QUICConnection extends EventTarget {
         this.dispatchEvent(
           new events.QUICConnectionStreamEvent({ detail: quicStream }),
         );
+        // No need to read after creation, doing so will throw during early cancellation
+      } else {
+        quicStream.read();
       }
-      quicStream.read();
     }
     for (const streamId of this.conn.writable() as Iterable<StreamId>) {
       const quicStream = this.streamMap.get(streamId);

src/QUICStream.ts

@@ -88,8 +88,10 @@ class QUICStream
     //  create Peer state.
     try {
       connection.conn.streamSend(streamId, new Uint8Array(0), false);
-    } catch {
-      // FIXME: If there is an error here then stream will not create? Maybe we should abort?
+    } catch (e) {
+      // We ignore any errors here, if this is a server side stream then state already exists.
+      // But it's possible for the stream to already be closed or have an error here.
+      // These errors will be handled by the QUICStream and not here.
     }
     const stream = new this({
       streamId,
@@ -163,6 +165,18 @@ class QUICStream
                 // and we need to propagate the error up and down the stream
                 controller.error(reason);
                 await this.closeRecv(true, reason);
+                // It is possible the stream was cancelled, let's check the writable state;
+                try {
+                  this.conn.streamWritable(this.streamId, 0);
+                } catch (e) {
+                  const match = e.message.match(/InvalidStreamState\((.+)\)/);
+                  if (match == null) {
+                    return never(
+                      'Errors besides [InvalidStreamState(StreamId)] are not expected here',
+                    );
+                  }
+                  this.writableController.error(reason);
+                }
               }
               break;
             }
@@ -319,8 +333,9 @@ class QUICStream
           this.readableController.close();
         }
       } catch (e) {
-        // Ignore if done, not normally meant to happen but possible in rare cases
-        if (e.message !== 'Done') {
+        if (e.message === 'Done') {
+          never();
+        } else {
           this.logger.debug(`Stream recv reported: error ${e.message}`);
           if (!this._recvClosed) {
             // Close stream in background
@@ -329,6 +344,18 @@ class QUICStream
                 (await this.processSendStreamError(e, 'recv')) ?? e;
               this.readableController.error(reason);
               await this.closeRecv(true, reason);
+              // It is possible the stream was cancelled, let's check the writable state;
+              try {
+                this.conn.streamWritable(this.streamId, 0);
+              } catch (e) {
+                const match = e.message.match(/InvalidStreamState\((.+)\)/);
+                if (match == null) {
+                  return never(
+                    'Errors besides [InvalidStreamState(StreamId)] are not expected here',
+                  );
+                }
+                this.writableController.error(reason);
+              }
             })();
           }
         }
@@ -488,7 +515,7 @@ class QUICStream
     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 never('Should never reach an [InvalidState(StreamId)] error');
     }
     return null;
   }

src/utils.ts

@@ -335,8 +335,8 @@ function decodeConnectionId(connIdString: ConnectionIdString): ConnectionId {
   return Buffer.from(connIdString, 'hex') as ConnectionId;
 }
 
-function never(): never {
-  throw new errors.ErrorQUICUndefinedBehaviour();
+function never(message?: string): never {
+  throw new errors.ErrorQUICUndefinedBehaviour(message);
 }
 
 function certificateDERToPEM(der: Uint8Array): string {

tests/QUICStream.test.ts

@@ -791,12 +791,13 @@ describe(QUICStream.name, () => {
     await client.destroy({ force: true });
     await server.stop({ force: true });
   });
-  test('streams can be cancelled', async () => {
+  test('streams can be cancelled after data sent', async () => {
     const cancelReason = Symbol('CancelReason');
     const connectionEventProm =
       utils.promise<events.QUICServerConnectionEvent>();
     const tlsConfig1 = await generateConfig(defaultType);
     const tlsConfig2 = await generateConfig(defaultType);
+    const reasonConverters = testsUtils.createReasonConverters();
     const server = new QUICServer({
       crypto: {
         key,
@@ -809,6 +810,7 @@ describe(QUICStream.name, () => {
         verifyPeer: true,
         ca: tlsConfig2.ca,
       },
+      ...reasonConverters,
     });
     testsUtils.extractSocket(server, sockets);
     server.addEventListener(
@@ -831,6 +833,7 @@ describe(QUICStream.name, () => {
         key: tlsConfig2.key,
         cert: tlsConfig2.cert,
       },
+      ...reasonConverters,
     });
     testsUtils.extractSocket(client, sockets);
     const conn = (await connectionEventProm.p).detail;
@@ -848,19 +851,28 @@ describe(QUICStream.name, () => {
     const writer = clientStream.writable.getWriter();
     await writer.write(message);
     writer.releaseLock();
-    await serverStreamProm.p;
     clientStream.cancel(cancelReason);
     await expect(clientStream.readable.getReader().read()).rejects.toBe(
       cancelReason,
     );
     await expect(clientStream.writable.getWriter().write()).rejects.toBe(
       cancelReason,
     );
+
     // Let's check that the server side ended
     const serverStream = await serverStreamProm.p;
-    await expect(
-      serverStream.readable.pipeTo(serverStream.writable),
-    ).rejects.toThrow();
+    const serverReadProm = (async () => {
+      for await (const _ of serverStream.readable) {
+        // Just consume until stream throws
+      }
+    })();
+    await expect(serverReadProm).rejects.toBe(cancelReason);
+    const serverWriter = serverStream.writable.getWriter();
+    // Should throw
+    await expect(serverWriter.write(Buffer.from('hello'))).rejects.toBe(
+      cancelReason,
+    );
+
     // And client stream should've cleaned up
     await testsUtils.sleep(100);
     expect(clientStream[destroyed]).toBeTrue();
@@ -873,6 +885,7 @@ describe(QUICStream.name, () => {
       utils.promise<events.QUICServerConnectionEvent>();
     const tlsConfig1 = await generateConfig(defaultType);
     const tlsConfig2 = await generateConfig(defaultType);
+    const reasonConverters = testsUtils.createReasonConverters();
     const server = new QUICServer({
       crypto: {
         key,
@@ -885,6 +898,7 @@ describe(QUICStream.name, () => {
         verifyPeer: true,
         ca: tlsConfig2.ca,
       },
+      ...reasonConverters,
     });
     testsUtils.extractSocket(server, sockets);
     server.addEventListener(
@@ -907,6 +921,7 @@ describe(QUICStream.name, () => {
         key: tlsConfig2.key,
         cert: tlsConfig2.cert,
       },
+      ...reasonConverters,
     });
     testsUtils.extractSocket(client, sockets);
     const conn = (await connectionEventProm.p).detail;
@@ -921,25 +936,34 @@ describe(QUICStream.name, () => {
     // Let's make a new streams.
     const clientStream = await client.connection.streamNew();
     clientStream.cancel(cancelReason);
-
     await expect(clientStream.readable.getReader().read()).rejects.toBe(
       cancelReason,
     );
     await expect(clientStream.writable.getWriter().write()).rejects.toBe(
       cancelReason,
     );
+
     // Let's check that the server side ended
     const serverStream = await serverStreamProm.p;
-    await expect(
-      serverStream.readable.pipeTo(serverStream.writable),
-    ).rejects.toThrow('recv 0');
+    const serverReadProm = (async () => {
+      for await (const _ of serverStream.readable) {
+        // Just consume until stream throws
+      }
+    })();
+    await expect(serverReadProm).rejects.toBe(cancelReason);
+    const serverWriter = serverStream.writable.getWriter();
+    // Should throw
+    await expect(serverWriter.write(Buffer.from('hello'))).rejects.toBe(
+      cancelReason,
+    );
+
     // And client stream should've cleaned up
     await testsUtils.sleep(100);
     expect(clientStream[destroyed]).toBeTrue();
     await client.destroy({ force: true });
     await server.stop({ force: true });
   });
-  test('Stream will end when waiting for more data', async () => {
+  test('stream will end when waiting for more data', async () => {
     // Needed to check that the pull based reading of data doesn't break when we
     // temporarily run out of data to read
     const connectionEventProm =
@@ -1008,7 +1032,7 @@ describe(QUICStream.name, () => {
     await client.destroy({ force: true });
     await server.stop({ force: true });
   });
-  test('Stream can error when blocked on data', async () => {
+  test('stream can error when blocked on data', async () => {
     // This checks that if the readable web-stream is full and not pulling data,
     // we will still respond to an error in the readable stream
 

tests/utils.ts

@@ -4,6 +4,7 @@ import type QUICSocket from '@/QUICSocket';
 import type QUICClient from '@/QUICClient';
 import type QUICServer from '@/QUICServer';
 import type QUICStream from '@/QUICStream';
+import type { StreamCodeToReason, StreamReasonToCode } from '@';
 import { Crypto } from '@peculiar/webcrypto';
 import * as x509 from '@peculiar/x509';
 import { never } from '@/utils';
@@ -740,6 +741,35 @@ async function generateConfig(type: KeyTypes): Promise<TLSConfigs> {
   };
 }
 
+/**
+ * This will create a `reasonToCode` and `CodeToReason` function that will
+ * allow errors to "jump" the network boundary. It does this by mapping the
+ * errors to an incrementing code and returning them on the other end of the
+ * connection.
+ *
+ * Note: this should ONLY be used for testing as it requires the client and
+ * server to share the same instance of `reasonToCode` and `codeToReason`.
+ */
+function createReasonConverters() {
+  const reasonMap = new Map<number, any>();
+  let code = 0;
+
+  const reasonToCode: StreamReasonToCode = (_type, reason) => {
+    code++;
+    reasonMap.set(code, reason);
+    return code;
+  };
+
+  const codeToReason: StreamCodeToReason = (_type, code) => {
+    return reasonMap.get(code) ?? new Error('Reason not found');
+  };
+
+  return {
+    reasonToCode,
+    codeToReason,
+  };
+}
+
 export {
   sleep,
   randomBytes,
@@ -760,6 +790,7 @@ export {
   waitForTimeoutNull,
   connStats,
   generateConfig,
+  createReasonConverters,
 };
 
 export type { Messages, StreamData, KeyTypes, TLSConfigs };