stream: fix multiple Writable.destroy() calls.

lib/internal/streams/writable.js

@@ -57,7 +57,6 @@ const {
   getHighWaterMark,
   getDefaultHighWaterMark
 } = require('internal/streams/state');
-const assert = require('internal/assert');
 const {
   ERR_INVALID_ARG_TYPE,
   ERR_METHOD_NOT_IMPLEMENTED,
@@ -854,11 +853,9 @@ Writable.prototype.destroy = function(err, cb) {
   const state = this._writableState;
 
   // Invoke pending callbacks.
-  if (
-    state.bufferedIndex < state.buffered.length ||
-    state[kOnFinished].length
-  ) {
-    assert(!state.destroyed);
+  if (!state.destroyed &&
+    (state.bufferedIndex < state.buffered.length ||
+      state[kOnFinished].length)) {
     process.nextTick(errorBuffer, state);
   }
 

test/parallel/test-http-many-requests.js

@@ -0,0 +1,31 @@
+'use strict';
+
+const common = require('../common');
+const http = require('http');
+const Countdown = require('../common/countdown');
+
+const NUM_REQ = 128
+
+const agent = new http.Agent({ keepAlive: true });
+const countdown = new Countdown(NUM_REQ, () => server.close());
+
+const server = http.createServer(common.mustCall(function(req, res) {
+  res.setHeader('content-type', 'application/json; charset=utf-8')
+  res.end(JSON.stringify({ hello: 'world' }))
+}, NUM_REQ)).listen(0, function() {
+  for (let i = 0; i < NUM_REQ; ++i) {
+    const req = http.request({
+      port: server.address().port,
+      agent,
+      method: 'GET'
+    }, function(res) {
+      res.resume();
+      res.on('end', () => {
+        countdown.dec();
+      })
+    })
+    .end();
+  }
+});
+
+process.on('warning', common.mustNotCall());

test/parallel/test-stream-writable-destroy.js

@@ -460,3 +460,14 @@ const assert = require('assert');
     assert.strictEqual(write.destroyed, true);
   }));
 }
+
+{
+  // Destroy twice
+  const write = new Writable({
+    write(chunk, enc, cb) { cb(); }
+  });
+
+  write.end(common.mustCall());
+  write.destroy();
+  write.destroy();
+}