stream: make .destroy() interact better with write queue

Make sure that it is safe to call the callback for `_write()` even in the presence of `.destroy()` calls during that write. In particular, letting the write queue continue processing would previously have thrown an exception, because processing writes after calling `.destroy()` is forbidden. One test had to be modified to account for the fact that callbacks for writes will now always be called, even when the stream is destroyed during the process.
nodejs · addaleax · Nov 3, 2018 · Nov 3, 2018 · Nov 3, 2018 · af56ec26f9f5eece026ebe2bbffbf4ede0193d59
commit af56ec26f9f5eece026ebe2bbffbf4ede0193d59
diff --git a/lib/_stream_writable.js b/lib/_stream_writable.js
@@ -461,7 +461,7 @@ function onwrite(stream, er) {
     onwriteError(stream, state, sync, er, cb);
   else {
     // Check if we're actually ready to finish, but don't emit yet
-    var finished = needFinish(state);
+    var finished = needFinish(state) || stream.destroyed;
 
     if (!finished &&
         !state.corked &&

diff --git a/test/parallel/test-stream-write-destroy.js b/test/parallel/test-stream-write-destroy.js
@@ -0,0 +1,59 @@
+'use strict';
+require('../common');
+const assert = require('assert');
+const { Writable } = require('stream');
+
+// Test interaction between calling .destroy() on a writable and pending
+// writes.
+
+for (const withPendingData of [ false, true ]) {
+  for (const useEnd of [ false, true ]) {
+    const callbacks = [];
+
+    const w = new Writable({
+      write(data, enc, cb) {
+        callbacks.push(cb);
+      },
+      // Effectively disable the HWM to observe 'drain' events more easily.
+      highWaterMark: 1
+    });
+
+    let chunksWritten = 0;
+    let drains = 0;
+    let finished = false;
+    w.on('drain', () => drains++);
+    w.on('finish', () => finished = true);
+
+    w.write('abc', () => chunksWritten++);
+    assert.strictEqual(chunksWritten, 0);
+    assert.strictEqual(drains, 0);
+    callbacks.shift()();
+    assert.strictEqual(chunksWritten, 1);
+    assert.strictEqual(drains, 1);
+
+    if (withPendingData) {
+      // Test 2 cases: There either is or is not data still in the write queue.
+      // (The second write will never actually get executed either way.)
+      w.write('def', () => chunksWritten++);
+    }
+    if (useEnd) {
+      // Again, test 2 cases: Either we indicate that we want to end the
+      // writable or not.
+      w.end('ghi', () => chunksWritten++);
+    } else {
+      w.write('ghi', () => chunksWritten++);
+    }
+
+    assert.strictEqual(chunksWritten, 1);
+    w.destroy();
+    assert.strictEqual(chunksWritten, 1);
+    callbacks.shift()();
+    assert.strictEqual(chunksWritten, 2);
+    assert.strictEqual(callbacks.length, 0);
+    assert.strictEqual(drains, 1);
+
+    // When we used `.end()`, we see the 'finished' event if and only if
+    // we actually finished processing the write queue.
+    assert.strictEqual(finished, !withPendingData && useEnd);
+  }
+}
diff --git a/test/parallel/test-tls-invoke-queued.js b/test/parallel/test-tls-invoke-queued.js
@@ -40,8 +40,8 @@ const server = tls.createServer({
     c.write('world!', null, common.mustCall(function() {
       c.destroy();
     }));
-    // Data on next _write() will be written but callback will not be invoked
-    c.write(' gosh', null, common.mustNotCall());
+    // Data on next _write() will be written, and the cb will still be invoked
+    c.write(' gosh', null, common.mustCall());
   }));
 
   server.close();