Prevent zalgo bomb on bad socket while forwarding request

lib/request-proxy/send.js

@@ -25,6 +25,11 @@ var rawHead = require('./util.js').rawHead;
 var strHead = require('./util.js').strHead;
 var TypedError = require('error/typed');
 
+var ChannelDestroyedError = TypedError({
+    type: 'ringpop.request-proxy.channel-destroyed',
+    message: 'Channel was destroyed before forwarding attempt'
+});
+
 var MaxRetriesExceeded = TypedError({
     type: 'ringpop.request-proxy.max-retries-exceeded',
     message: 'Max number of retries exceeded. {maxRetries} were attempted',
@@ -223,12 +228,19 @@ RequestProxySend.prototype.scheduleRetry = function scheduleRetry(callback) {
 RequestProxySend.prototype.send = function send(channelOpts, callback) {
     var self = this;
 
+    if (this.ringpop.channel.destroyed) {
+        process.nextTick(function onTick() {
+            callback(ChannelDestroyedError());
+        });
+        return;
+    }
+
     this.ringpop.channel.send(
         channelOpts,
         channelOpts.endpoint,
         this.getStrHead(),
         this.request.body,
-        this.maxRetries === 0 ? callback : onSend
+        this.maxRetries === 0 ? onSendNoRetries : onSend
     );
 
     this.ringpop.emit('requestProxy.requestProxied');
@@ -248,6 +260,12 @@ RequestProxySend.prototype.send = function send(channelOpts, callback) {
 
         self.scheduleRetry(callback);
     }
+
+    function onSendNoRetries(err, res1, res2) {
+        process.nextTick(function onTick() {
+            callback(err, res1, res2);
+        });
+    }
 };
 
 module.exports = function send(opts, callback) {

test/integration/proxy-test.js

@@ -964,3 +964,53 @@ test('handles checksum failures', function t(assert) {
         });
     });
 });
+
+test('does not crash when forwarding request on closed socket', function t(assert) {
+    var ringpopOpts = {
+        requestProxyMaxRetries: 0
+    };
+
+    var cluster = allocCluster(ringpopOpts, function onReady() {
+        destroyAllSockets();
+
+        cluster.request({
+            host: 'one', key: cluster.keys.two
+        }, function onResponse(err, resp) {
+            assert.ifError(err);
+
+            cluster.destroy();
+            assert.end();
+        });
+    });
+
+    function destroyAllSockets() {
+        var allPeers = cluster.one.channel.peers;
+
+        Object.keys(allPeers).forEach(function each(key) {
+            allPeers[key][0].socket.destroy();
+        });
+    }
+});
+
+test('send on destroyed channel not allowed', function t(assert) {
+    var ringpopOpts = {
+        requestProxyMaxRetries: 0
+    };
+
+    var cluster = allocCluster(ringpopOpts, function onReady() {
+        cluster.one.channel.quit();
+
+        cluster.request({
+            host: 'one', key: cluster.keys.two
+        }, function onResponse(err, resp) {
+            assert.ifError(err);
+
+            assert.equal(resp.statusCode, 500);
+            assert.ok(resp.body.indexOf(
+                'Channel was destroyed before forwarding attempt') >= 0);
+
+            cluster.destroy();
+            assert.end();
+        });
+    });
+});