[fix] Do not swallow errors emitted while destroying (#1672)

Refs: nodejs/node#29197

Author: lpinca

lib/stream.js

@@ -78,6 +78,8 @@ function createWebSocketStream(ws, options) {
   });
 
   ws.once('error', function error(err) {
+    if (duplex.destroyed) return;
+
     duplex.destroy(err);
   });
 
@@ -94,8 +96,15 @@ function createWebSocketStream(ws, options) {
       return;
     }
 
-    ws.once('close', function close() {
+    let called = false;
+
+    ws.once('error', function error(err) {
+      called = true;
       callback(err);
+    });
+
+    ws.once('close', function close() {
+      if (!called) callback(err);
       process.nextTick(emitClose, duplex);
     });
     ws.terminate();

test/create-websocket-stream.test.js

@@ -225,6 +225,52 @@ describe('createWebSocketStream', () => {
       });
     });
 
+    it('does not swallow errors that may occur while destroying', (done) => {
+      const frame = Buffer.concat(
+        Sender.frame(Buffer.from([0x22, 0xfa, 0xec, 0x78]), {
+          fin: true,
+          rsv1: true,
+          opcode: 0x02,
+          mask: false,
+          readOnly: false
+        })
+      );
+
+      const wss = new WebSocket.Server(
+        {
+          perMessageDeflate: true,
+          port: 0
+        },
+        () => {
+          const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
+          const duplex = createWebSocketStream(ws);
+
+          duplex.on('error', (err) => {
+            assert.ok(err instanceof Error);
+            assert.strictEqual(err.code, 'Z_DATA_ERROR');
+            assert.strictEqual(err.errno, -3);
+
+            duplex.on('close', () => {
+              wss.close(done);
+            });
+          });
+
+          let bytesRead = 0;
+
+          ws.on('open', () => {
+            ws._socket.on('data', (chunk) => {
+              bytesRead += chunk.length;
+              if (bytesRead === frame.length) duplex.destroy();
+            });
+          });
+        }
+      );
+
+      wss.on('connection', (ws) => {
+        ws._socket.write(frame);
+      });
+    });
+
     it("does not suppress the throwing behavior of 'error' events", (done) => {
       const wss = new WebSocket.Server({ port: 0 }, () => {
         const ws = new WebSocket(`ws://localhost:${wss.address().port}`);

test/permessage-deflate.test.js

@@ -589,6 +589,7 @@ describe('PerMessageDeflate', () => {
       perMessageDeflate.accept([{}]);
       perMessageDeflate.decompress(data, true, (err) => {
         assert.ok(err instanceof Error);
+        assert.strictEqual(err.code, 'Z_DATA_ERROR');
         assert.strictEqual(err.errno, -3);
         done();
       });