fix: Fixed bug with concurrent stream cancellation

Author: tegefaulkes

src/QUICConnection.ts

@@ -907,9 +907,22 @@ class QUICConnection extends EventTarget {
         //  The QUICStream will always exist before processing it's writable.
         //  1. First time it is seen in the readable iterator
         //  2. created using `streamNew()`
-        never();
+
+        // There is one condition where this can happen. That is when both sides of the stream cancel concurrently.
+        // Local state is cleaned up while the remote side still sends a closing frame.
+        try {
+          this.conn.streamWritable(streamId, 0);
+          this.logger.debug(
+            `streamId ${streamId} was writable without an existing stream`,
+          );
+        } catch (e) {
+          this.logger.debug(
+            `streamId ${streamId} was writable without an existing stream and error ${e.message}`,
+          );
+        }
+      } else {
+        quicStream.write();
       }
-      quicStream.write();
     }
   }
 

src/QUICStream.ts

@@ -327,9 +327,7 @@ class QUICStream
           this.readableController.close();
         }
       } catch (e) {
-        if (e.message === 'Done') {
-          never();
-        } else {
+        if (e.message !== 'Done') {
           this.logger.debug(`Stream recv reported: error ${e.message}`);
           if (!this._recvClosed) {
             // Close stream in background

tests/QUICStream.test.ts

@@ -963,6 +963,90 @@ describe(QUICStream.name, () => {
     await client.destroy({ force: true });
     await server.stop({ force: true });
   });
+  test('streams can be cancelled concurrently 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,
+        ops: serverCrypto,
+      },
+      logger: logger.getChild(QUICServer.name),
+      config: {
+        key: tlsConfig1.key,
+        cert: tlsConfig1.cert,
+        verifyPeer: true,
+        ca: tlsConfig2.ca,
+      },
+      ...reasonConverters,
+    });
+    testsUtils.extractSocket(server, sockets);
+    server.addEventListener(
+      'serverConnection',
+      (e: events.QUICServerConnectionEvent) => 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,
+        key: tlsConfig2.key,
+        cert: tlsConfig2.cert,
+      },
+      ...reasonConverters,
+    });
+    testsUtils.extractSocket(client, sockets);
+    const conn = (await connectionEventProm.p).detail;
+    // Do the test
+    const serverStreamProm = utils.promise<QUICStream>();
+    conn.addEventListener(
+      'connectionStream',
+      (event: events.QUICConnectionStreamEvent) => {
+        serverStreamProm.resolveP(event.detail);
+      },
+    );
+    // Let's make a new streams.
+    const message = Buffer.from('Hello!');
+    const clientStream = await client.connection.streamNew();
+    const writer = clientStream.writable.getWriter();
+    await writer.write(message);
+    writer.releaseLock();
+    const serverStream = await serverStreamProm.p;
+    serverStream.cancel(cancelReason);
+    clientStream.cancel(cancelReason);
+
+    // Checking stream states
+    await expect(clientStream.readable.getReader().read()).rejects.toBe(
+      cancelReason,
+    );
+    await expect(clientStream.writable.getWriter().write()).rejects.toBe(
+      cancelReason,
+    );
+    await expect(serverStream.readable.getReader().read()).rejects.toBe(
+      cancelReason,
+    );
+    await expect(serverStream.writable.getWriter().write()).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 () => {
     // Needed to check that the pull based reading of data doesn't break when we
     // temporarily run out of data to read