Repository: websockets/ws
Pull request #1464 (2018-10-23 09:00:08)

Do not waste time compressing when socket is closed
Addresses #1226 - I agree with @lpinca 's approach in general.  This has been tested with the offending code listed in the issue and it correctly short circuits while still providing a relevant error to each pending request.  I've also added a unit test that fails without the added code as it tries to write/compress too much.

Not sure how I feel about having to add the extra `on: () => {}` to the Sender unit tests for the socket object, but these are mocks anyway.  Open to suggestions if this isn't preferred.

lpinca: Thank you for the PR. Some thoughts in addition to the comments:

1. Each callback should ideally be called before the `'close'` event is emitted on the `WebSocket`. The current approach relies to the fact that the `'close'` event listener is added to the `net.Socket` before the [`'close'` listener](https://github.com/websockets/ws/blob/d2317b1d8dda7d36f1ba952f533ff45e2464ae52/lib/websocket.js#L750) that actually emit `'close'` on the `WebSocket`. This is good and should work but there might be a chunk that is already being compressed when the `'close'` event is emitted on the `net.Socket` and this is not caught because it is no longer in the queue. As a result the callback for this message could be invoked after the `'close'` event is emitted on the `WebSocket` and after all other callback that are short circuited.
1. Each callback should find the ready state to `CLOSING`. If I'm not wrong when the queue is cleared the `WebSocket` ready state is still `OPEN` so the user might surprised to get the error and have the ready state set to `OPEN`.

The second point should be easy to fix, it should be sufficient to set the ready state to `CLOSING` in the `'close'` listener added in the `Sender` before clearing the queue.

The first point is harder and I'm not sure how to fix that.
Evertras: For the second point, my testing is showing the ready state as "closed".  Is "closing" a better state here?
lpinca: > I'm not sure how to reproduce the hypothetical you listed, is there an example you could provide?

I'll try to create a test case when I get a chance. Thank you.
lpinca: @Evertras here is the test case

```js
const WebSocket = require('ws');
const EventEmitter = require('events');
const { randomBytes } = require('crypto');

const ee = new EventEmitter();

const wss = new WebSocket.Server({
  perMessageDeflate: { threshold: 0 },
  port: 0
}, function () {
  const ws = new WebSocket(`ws://localhost:${wss.address().port}`, {
    perMessageDeflate: { threshold: 0 }
  });

  ws.on('open', function () {
    const pmd = ws._extensions['permessage-deflate'];
    const compress = pmd.compress;

    // Monkey patch compress to show that deflate ends after the `'close'` event
    // is emitted.
    pmd.compress = function (data, fin, callback) {
      compress.call(pmd, data, fin, function (err, buf) {
        console.log('deflate done');
        callback(err, buf);
      });
    };

    ws.send(randomBytes(1024 * 1024), function (err) {
      if (err) console.log(err);
    });

    // Make the other peer destory the socket while data is being compressed.
    ee.emit('destroy');
  });

  ws.on('close', function () {
    console.log('close');
  });
});

wss.on('connection', function (ws) {
  ee.on('destroy', function () {
    ws.terminate();
  });
});
```
Evertras: I see.  The deflate has already started by the time the close event from the socket is received, so by the time we're even aware that the socket has closed then zlib has already started to do its work.  The closest I could get to an interruption is in [this callback](https://github.com/websockets/ws/blob/master/lib/permessage-deflate.js#L434) and at that point all the expensive work is done.

I'm not sure if it's even possible to completely remove this scenario.  Once it's passed to zlib, some compression may take place while the socket is closing since it's all batched up in the `flush` call.  I'm also not sure if it's worth trying to chase.
lpinca: Yes, I think there is not much we can do about it, we could check if the socket is closed [here]( https://github.com/websockets/ws/blob/d2317b1d8dda7d36f1ba952f533ff45e2464ae52/lib/sender.js#L335) so at least we avoid framing that chunk but I'm not sure if it's worth the effort, it's really and edge case.

The only thing that bugs me is that when this happens the order of callbacks is no longer respected and that is why in the original issue I thought it was better to clear the send queue [here](https://github.com/websockets/ws/blob/d2317b1d8dda7d36f1ba952f533ff45e2464ae52/lib/sender.js#L335) but if we do that, callbacks are invoked after the `'close'` event is emitted.


lpinca: It could be addressed by refactoring the already complex close procedure, for example by emitting the `'close'` event only when both the `Receiver` and `Sender` complete their processing.

That said, I think this PR is already a big improvement and could land as is, just give me a few days to review it again (sorry I'm kinda busy).

Sometimes I think adding permessage-deflate was a mistake, the amount of complexity it added is huge.

Thank you.
Evertras: Agreed there are ways to improve it, but yeah, it'll be complicated.  One step at a time I think.  No rush on approval, thanks for all the input!

Patch:
diff --git a/.travis.yml b/.travis.yml
index 4b1d501b..ebb063d6 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,6 +1,7 @@
 language: node_js
 sudo: false
 node_js:
+  - "11"
   - "10"
   - "8"
   - "6"
diff --git a/appveyor.yml b/appveyor.yml
index eba87832..917169ad 100644
--- a/appveyor.yml
+++ b/appveyor.yml
@@ -1,5 +1,6 @@
 environment:
   matrix:
+    - nodejs_version: "11"
     - nodejs_version: "10"
     - nodejs_version: "8"
     - nodejs_version: "6"
diff --git a/lib/extension.js b/lib/extension.js
index 3f48d751..a9b84728 100644
--- a/lib/extension.js
+++ b/lib/extension.js
@@ -64,7 +64,7 @@ function parse (header) {
     if (extensionName === undefined) {
       if (end === -1 && tokenChars[code] === 1) {
         if (start === -1) start = i;
-      } else if (code === 0x20/* ' ' */|| code === 0x09/* '\t' */) {
+      } else if (code === 0x20/* ' ' */ || code === 0x09/* '\t' */) {
         if (end === -1 && start !== -1) end = i;
       } else if (code === 0x3b/* ';' */ || code === 0x2c/* ',' */) {
         if (start === -1) {
@@ -103,7 +103,7 @@ function parse (header) {
         }
 
         start = end = -1;
-      } else if (code === 0x3d/* '=' */&& start !== -1 && end === -1) {
+      } else if (code === 0x3d/* '=' */ && start !== -1 && end === -1) {
         paramName = header.slice(start, i);
         start = end = -1;
       } else {
diff --git a/lib/sender.js b/lib/sender.js
index 060e5539..500bd98c 100644
--- a/lib/sender.js
+++ b/lib/sender.js
@@ -27,6 +27,20 @@ class Sender {
     this._bufferedBytes = 0;
     this._deflating = false;
     this._queue = [];
+
+    if (this._extensions[PerMessageDeflate.extensionName]) {
+      this._socket.once('close', () => {
+        const err = new Error('WebSocket is not open: readyState 2 (CLOSING)');
+
+        while (this._queue.length) {
+          const params = this._queue.shift();
+          const cb = params[params.length - 1];
+
+          this._bufferedBytes -= params[1].length;
+          if (typeof cb === 'function') cb(err);
+        }
+      });
+    }
   }
 
   /**
diff --git a/lib/websocket.js b/lib/websocket.js
index 5b8497d6..79cb219f 100644
--- a/lib/websocket.js
+++ b/lib/websocket.js
@@ -123,6 +123,11 @@ class WebSocket extends EventEmitter {
       maxPayload
     );
 
+    //
+    // `Sender` must be instantiated before adding the `socketOnClose` listener.
+    // This allows the sender queue, when used, to be emptied before
+    // `socketOnClose` is called.
+    //
     this._sender = new Sender(socket, this._extensions);
     this._receiver = receiver;
     this._socket = socket;
diff --git a/package.json b/package.json
index 5a4bc74a..884b7419 100644
--- a/package.json
+++ b/package.json
@@ -33,14 +33,14 @@
   "devDependencies": {
     "benchmark": "~2.1.2",
     "bufferutil": "~4.0.0",
-    "eslint": "~5.6.1",
+    "eslint": "~5.8.0",
     "eslint-config-standard": "~12.0.0",
     "eslint-plugin-import": "~2.14.0",
-    "eslint-plugin-node": "~7.0.0",
+    "eslint-plugin-node": "~8.0.0",
     "eslint-plugin-promise": "~4.0.0",
     "eslint-plugin-standard": "~4.0.0",
     "mocha": "~5.2.0",
-    "nyc": "~13.0.1",
+    "nyc": "~13.1.0",
     "utf-8-validate": "~5.0.0"
   }
 }


Test patch:
diff --git a/test/sender.test.js b/test/sender.test.js
index 623ce595..e7c7dd10 100644
--- a/test/sender.test.js
+++ b/test/sender.test.js
@@ -5,6 +5,16 @@ const assert = require('assert');
 const PerMessageDeflate = require('../lib/permessage-deflate');
 const Sender = require('../lib/sender');
 
+class MockSocket {
+  constructor ({ write, once } = {}) {
+    if (write) this.write = write;
+    if (once) this.once = once;
+  }
+
+  write () {}
+  once () {}
+}
+
 describe('Sender', function () {
   describe('.frame', function () {
     it('does not mutate the input buffer if data is `readOnly`', function () {
@@ -38,12 +48,13 @@ describe('Sender', function () {
     it('compresses data if compress option is enabled', function (done) {
       const perMessageDeflate = new PerMessageDeflate({ threshold: 0 });
       let count = 0;
-      const sender = new Sender({
+      const mockSocket = new MockSocket({
         write: (data) => {
           assert.strictEqual(data[0] & 0x40, 0x40);
           if (++count === 3) done();
         }
-      }, {
+      });
+      const sender = new Sender(mockSocket, {
         'permessage-deflate': perMessageDeflate
       });
 
@@ -57,14 +68,53 @@ describe('Sender', function () {
       sender.send('hi', options);
     });
 
+    it('does not compress enqueued messages after socket closes', function (done) {
+      const numMessages = 100;
+      let numErrors = 0;
+
+      const mockSocket = new MockSocket({
+        write: (data) => {
+          // Test that `PerMessageDeflate#compress()` and `Socket#write()` is
+          // called only once (for the first message that is not queued).
+          assert.strictEqual(numErrors, numMessages);
+          done();
+        },
+        once: (ev, cb) => {
+          if (ev === 'close') process.nextTick(cb);
+        }
+      });
+
+      const perMessageDeflate = new PerMessageDeflate({ threshold: 0 });
+      perMessageDeflate.accept([{}]);
+
+      const sender = new Sender(mockSocket, {
+        'permessage-deflate': perMessageDeflate
+      });
+
+      const options = { compress: true, fin: true };
+      sender.send('hi', options);
+
+      for (let i = 0; i < numMessages; i++) {
+        sender.send('hi', options, (err) => {
+          assert.ok(err instanceof Error);
+          assert.strictEqual(
+            err.message,
+            'WebSocket is not open: readyState 2 (CLOSING)'
+          );
+          numErrors++;
+        });
+      }
+    });
+
     it('does not compress data for small payloads', function (done) {
       const perMessageDeflate = new PerMessageDeflate();
-      const sender = new Sender({
+      const mockSocket = new MockSocket({
         write: (data) => {
           assert.notStrictEqual(data[0] & 0x40, 0x40);
           done();
         }
-      }, {
+      });
+      const sender = new Sender(mockSocket, {
         'permessage-deflate': perMessageDeflate
       });
 
@@ -76,7 +126,7 @@ describe('Sender', function () {
     it('compresses all frames in a fragmented message', function (done) {
       const fragments = [];
       const perMessageDeflate = new PerMessageDeflate({ threshold: 3 });
-      const sender = new Sender({
+      const mockSocket = new MockSocket({
         write: (data) => {
           fragments.push(data);
           if (fragments.length !== 2) return;
@@ -87,7 +137,8 @@ describe('Sender', function () {
           assert.strictEqual(fragments[1].length, 6);
           done();
         }
-      }, {
+      });
+      const sender = new Sender(mockSocket, {
         'permessage-deflate': perMessageDeflate
       });
 
@@ -100,7 +151,7 @@ describe('Sender', function () {
     it('compresses no frames in a fragmented message', function (done) {
       const fragments = [];
       const perMessageDeflate = new PerMessageDeflate({ threshold: 3 });
-      const sender = new Sender({
+      const mockSocket = new MockSocket({
         write: (data) => {
           fragments.push(data);
           if (fragments.length !== 2) return;
@@ -111,7 +162,8 @@ describe('Sender', function () {
           assert.strictEqual(fragments[1].length, 5);
           done();
         }
-      }, {
+      });
+      const sender = new Sender(mockSocket, {
         'permessage-deflate': perMessageDeflate
       });
 
@@ -124,7 +176,7 @@ describe('Sender', function () {
     it('compresses empty buffer as first fragment', function (done) {
       const fragments = [];
       const perMessageDeflate = new PerMessageDeflate({ threshold: 0 });
-      const sender = new Sender({
+      const mockSocket = new MockSocket({
         write: (data) => {
           fragments.push(data);
           if (fragments.length !== 2) return;
@@ -135,7 +187,8 @@ describe('Sender', function () {
           assert.strictEqual(fragments[1].length, 8);
           done();
         }
-      }, {
+      });
+      const sender = new Sender(mockSocket, {
         'permessage-deflate': perMessageDeflate
       });
 
@@ -148,7 +201,7 @@ describe('Sender', function () {
     it('compresses empty buffer as last fragment', function (done) {
       const fragments = [];
       const perMessageDeflate = new PerMessageDeflate({ threshold: 0 });
-      const sender = new Sender({
+      const mockSocket = new MockSocket({
         write: (data) => {
           fragments.push(data);
           if (fragments.length !== 2) return;
@@ -159,7 +212,8 @@ describe('Sender', function () {
           assert.strictEqual(fragments[1].length, 3);
           done();
         }
-      }, {
+      });
+      const sender = new Sender(mockSocket, {
         'permessage-deflate': perMessageDeflate
       });
 
@@ -172,11 +226,12 @@ describe('Sender', function () {
     it('handles many send calls while processing without crashing on flush', function (done) {
       let count = 0;
       const perMessageDeflate = new PerMessageDeflate();
-      const sender = new Sender({
+      const mockSocket = new MockSocket({
         write: () => {
           if (++count > 1e4) done();
         }
-      }, {
+      });
+      const sender = new Sender(mockSocket, {
         'permessage-deflate': perMessageDeflate
       });
 
@@ -196,14 +251,15 @@ describe('Sender', function () {
     it('works with multiple types of data', function (done) {
       const perMessageDeflate = new PerMessageDeflate({ threshold: 0 });
       let count = 0;
-      const sender = new Sender({
+      const mockSocket = new MockSocket({
         write: (data) => {
           if (++count === 1) return;
 
           assert.ok(data.equals(Buffer.from([0x89, 0x02, 0x68, 0x69])));
           if (count === 4) done();
         }
-      }, {
+      });
+      const sender = new Sender(mockSocket, {
         'permessage-deflate': perMessageDeflate
       });
 
@@ -222,14 +278,15 @@ describe('Sender', function () {
     it('works with multiple types of data', function (done) {
       const perMessageDeflate = new PerMessageDeflate({ threshold: 0 });
       let count = 0;
-      const sender = new Sender({
+      const mockSocket = new MockSocket({
         write: (data) => {
           if (++count === 1) return;
 
           assert.ok(data.equals(Buffer.from([0x8a, 0x02, 0x68, 0x69])));
           if (count === 4) done();
         }
-      }, {
+      });
+      const sender = new Sender(mockSocket, {
         'permessage-deflate': perMessageDeflate
       });
 
@@ -249,12 +306,13 @@ describe('Sender', function () {
       const perMessageDeflate = new PerMessageDeflate({ threshold: 0 });
 
       let count = 0;
-      const sender = new Sender({
+      const mockSocket = new MockSocket({
         write: (data, cb) => {
           count++;
           if (cb) cb();
         }
-      }, {
+      });
+      const sender = new Sender(mockSocket, {
         'permessage-deflate': perMessageDeflate
       });
 
diff --git a/test/websocket.test.js b/test/websocket.test.js
index 7383e410..dc945865 100644
--- a/test/websocket.test.js
+++ b/test/websocket.test.js
@@ -2041,6 +2041,40 @@ describe('WebSocket', function () {
       });
     });
 
+    it('reports the state as `CLOSING` in callbacks of discarded queued messages', function (done) {
+      const wss = new WebSocket.Server({
+        perMessageDeflate: { threshold: 0 },
+        port: 0
+      }, () => {
+        const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
+      });
+
+      wss.on('connection', (ws) => {
+        const map = () => crypto.randomBytes(16);
+        let count = 100;
+
+        Array.from({ length: count }).map(map).forEach((data, i) => {
+          ws.send(data, (err) => {
+            assert.ok(err instanceof Error);
+
+            if (i > 0) {
+              // The first message is not queued and compression completes after
+              // the `'close'` event is emitted on the `net.Socket`.
+              assert.strictEqual(ws.readyState, WebSocket.CLOSING);
+              assert.strictEqual(
+                err.message,
+                'WebSocket is not open: readyState 2 (CLOSING)'
+              );
+            }
+
+            if (--count === 0) done();
+          });
+        });
+
+        wss.close();
+      });
+    });
+
     describe('#send', function () {
       it('ignores the `compress` option if the extension is disabled', function (done) {
         const wss = new WebSocket.Server({ port: 0 }, () => {