Improve error handling

Arguments are now passed to an command error in case they exist
An error is only emitted if that very same error is not already handled in a callback

index.js

@@ -5,6 +5,7 @@ var tls = require('tls');
 var util = require('util');
 var utils = require('./lib/utils');
 var Queue = require('double-ended-queue');
+var CommandError = require('./lib/customError');
 var Command = require('./lib/command').Command;
 var OfflineCommand = require('./lib/command').OfflineCommand;
 var EventEmitter = require('events');
@@ -264,11 +265,11 @@ RedisClient.prototype.create_stream = function () {
     });
 
     this.stream.once('close', function (hadError) {
-        self.connection_gone('close', new Error('Stream connection closed' + (hadError ? ' because of a transmission error' : '')));
+        self.connection_gone('close', hadError ? new Error('Stream connection closed with a transmission error') : null);
     });
 
     this.stream.once('end', function () {
-        self.connection_gone('end', new Error('Stream connection ended'));
+        self.connection_gone('end', null);
     });
 
     this.stream.on('drain', function () {
@@ -320,16 +321,29 @@ RedisClient.prototype.warn = function (msg) {
 
 // Flush provided queues, erroring any items with a callback first
 RedisClient.prototype.flush_and_error = function (error, queue_names) {
+    var callbacks_not_called = [];
     queue_names = queue_names || ['offline_queue', 'command_queue'];
     for (var i = 0; i < queue_names.length; i++) {
         for (var command_obj = this[queue_names[i]].shift(); command_obj; command_obj = this[queue_names[i]].shift()) {
+            var err = new CommandError(error);
+            err.command = command_obj.command.toUpperCase();
+            if (command_obj.args.length) {
+                err.args = command_obj.args;
+            }
             if (typeof command_obj.callback === 'function') {
-                error.command = command_obj.command.toUpperCase();
+                command_obj.callback(err);
-                command_obj.callback(error);
+            } else {
+                callbacks_not_called.push(err);
             }
         }
         this[queue_names[i]] = new Queue();
     }
+    // Mutate the original error that will be emitted
+    // This is fine, as we don't manipulate any user errors
+    if (callbacks_not_called.length !== 0) {
+        error.errors = callbacks_not_called;
+    }
+    return callbacks_not_called.length === 0;
 };
 
 RedisClient.prototype.on_error = function (err) {
@@ -546,8 +560,10 @@ RedisClient.prototype.connection_gone = function (why, error) {
 
     // If this is a requested shutdown, then don't retry
     if (this.closing) {
-        debug('Connection ended from quit command, not retrying.');
+        debug('Connection ended by quit / end command, not retrying.');
-        this.flush_and_error(new Error('Redis connection gone from ' + why + ' event.'));
+        error = new Error('Stream connection ended and running command aborted. It might have been processed.');
+        error.code = 'NR_OFFLINE';
+        this.flush_and_error(error);
         return;
     }
 
@@ -567,10 +583,18 @@ RedisClient.prototype.connection_gone = function (why, error) {
         if (typeof this.retry_delay !== 'number') {
             // Pass individual error through
             if (this.retry_delay instanceof Error) {
-                error = this.retry_delay;
+                error = new CommandError(this.retry_delay);
+            }
+            // Attention: there might be the case where there's no error!
+            if (!error) {
+                error = new Error('Stream connection ended and running command aborted. It might have been processed.');
+                error.code = 'NR_OFFLINE';
+            }
+            // Only emit an error in case that a running command had no callback
+            if (!this.flush_and_error(error)) {
+                error.message = 'Stream connection ended and all running commands aborted. They might have been processed.';
+                this.emit('error', error);
             }
-            this.flush_and_error(error);
-            this.emit('error', error);
             this.end(false);
             return;
         }
@@ -595,11 +619,11 @@ RedisClient.prototype.connection_gone = function (why, error) {
     } else if (this.command_queue.length !== 0) {
         error = new Error('Redis connection lost and command aborted in uncertain state. It might have been processed.');
         error.code = 'UNCERTAIN_STATE';
-        this.flush_and_error(error, ['command_queue']);
+        if (!this.flush_and_error(error, ['command_queue'])) {
-        error.message = 'Redis connection lost and commands aborted in uncertain state. They might have been processed.';
+            // Only emit if not all commands had a callback that already handled the error
-        // TODO: Reconsider emitting this always, as each running command is handled anyway
+            error.message = 'Redis connection lost and commands aborted in uncertain state. They might have been processed.';
-        // This should likely be removed in v.3. This is different to the broken connection as we'll reconnect here
+            this.emit('error', error);
-        this.emit('error', error);
+        }
     }
 
     if (this.retry_max_delay !== null && this.retry_delay > this.retry_max_delay) {
@@ -618,6 +642,9 @@ RedisClient.prototype.return_error = function (err) {
     var command_obj = this.command_queue.shift();
     if (command_obj && command_obj.command && command_obj.command.toUpperCase) {
         err.command = command_obj.command.toUpperCase();
+        if (command_obj.args.length) {
+            err.args = command_obj.args;
+        }
     }
 
     var match = err.message.match(utils.err_code);
@@ -786,6 +813,9 @@ function handle_offline_command (self, command_obj) {
         }
         err = new Error(command + " can't be processed. " + msg);
         err.command = command;
+        if (command_obj.args.length) {
+            err.args = command_obj.args;
+        }
         err.code = 'NR_OFFLINE';
         utils.reply_in_order(self, callback, err);
     } else {

lib/customError.js

@@ -0,0 +1,16 @@
+'use strict';
+
+var util = require('util');
+
+function CommandError (error) {
+    Error.captureStackTrace(this, this.constructor);
+    this.name = this.constructor.name;
+    this.message = error.message;
+    for (var keys = Object.keys(error), key = keys.pop(); key; key = keys.pop()) {
+        this[key] = error[key];
+    }
+}
+
+util.inherits(CommandError, Error);
+
+module.exports = CommandError;

lib/extendedApi.js

@@ -47,7 +47,10 @@ RedisClient.prototype.send_command = RedisClient.prototype.sendCommand = functio
 RedisClient.prototype.end = function (flush) {
     // Flush queue if wanted
     if (flush) {
-        this.flush_and_error(new Error("The command can't be processed. The connection has already been closed."));
+        var err = new Error("The command can't be processed. The connection has already been closed.");
+        err.code = 'NR_OFFLINE';
+        this.flush_and_error(err);
+        // TODO: Emit an error in case a command did not have a callback
     } else if (arguments.length === 0) {
         this.warn(
             'Using .end() without the flush parameter is deprecated and throws from v.3.0.0 on.\n' +

lib/multi.js

@@ -23,7 +23,8 @@ function Multi (client, args) {
 function pipeline_transaction_command (self, command, args, index, cb, call_on_write) {
     // Queueing is done first, then the commands are executed
     self._client.send_command(command, args, function (err, reply) {
-        if (err) {
+        // Ignore the multi command. This is applied by node_redis and the user does not benefit by it
+        if (err && index !== -1) {
             if (cb) {
                 cb(err);
             }
@@ -44,13 +45,12 @@ function multi_callback (self, err, replies) {
     var i = 0, args;
 
     if (err) {
-        // The errors would be circular
+        err.errors = self.errors;
-        var connection_error = ['CONNECTION_BROKEN', 'UNCERTAIN_STATE'].indexOf(err.code) !== -1;
+        err.command = 'EXEC';
-        err.errors = connection_error ? [] : self.errors;
         if (self.callback) {
             self.callback(err);
             // Exclude connection errors so that those errors won't be emitted twice
-        } else if (!connection_error) {
+        } else if (err.code !== 'CONNECTION_BROKEN') {
             self._client.emit('error', err);
         }
         return;
@@ -91,7 +91,7 @@ Multi.prototype.exec_transaction = function exec_transaction (callback) {
     self.callback = callback;
     self._client.cork();
     self.wants_buffers = new Array(len);
-    pipeline_transaction_command(self, 'multi', []);
+    pipeline_transaction_command(self, 'multi', [], -1);
     // Drain queue, callback will catch 'QUEUED' or error
     for (var index = 0; index < len; index++) {
         // The commands may not be shifted off, since they are needed in the result handler

test/auth.spec.js

@@ -183,7 +183,7 @@ describe('client authentication', function () {
                     }
                 });
                 client.on('reconnecting', function (params) {
-                    assert.strictEqual(params.error.message, 'Stream connection closed');
+                    assert.strictEqual(params.error, null);
                 });
             });
 

test/connection.spec.js

@@ -53,7 +53,7 @@ describe('connection tests', function () {
                 }
             });
             client.set('foo', 'bar', function (err, res) {
-                assert.strictEqual(err.message, 'Redis connection gone from close event.');
+                assert.strictEqual(err.message, 'Stream connection ended and running command aborted. It might have been processed.');
                 called = -1;
             });
         });
@@ -62,7 +62,7 @@ describe('connection tests', function () {
             var called = false;
             client = redis.createClient(9999);
             client.set('foo', 'bar', function (err, res) {
-                assert.strictEqual(err.message, 'Redis connection gone from close event.');
+                assert.strictEqual(err.message, 'Stream connection ended and running command aborted. It might have been processed.');
                 called = true;
             });
             var bool = client.quit(function (err, res) {
@@ -277,13 +277,12 @@ describe('connection tests', function () {
                         text += data;
                         return '';
                     });
-                    var end = helper.callFuncAfter(done, 2);
                     client = redis.createClient({
                         retryStrategy: function (options) {
                             if (options.totalRetryTime > 150) {
                                 client.set('foo', 'bar', function (err, res) {
                                     assert.strictEqual(err.message, 'Connection timeout');
-                                    end();
+                                    done();
                                 });
                                 // Pass a individual error message to the error handler
                                 return new Error('Connection timeout');
@@ -294,28 +293,23 @@ describe('connection tests', function () {
                         retryMaxDelay: 123,
                         port: 9999
                     });
-
+                    process.nextTick(function () {
-                    client.on('error', function (err) {
-                        unhookIntercept();
                         assert.strictEqual(
                             text,
                             'node_redis: WARNING: You activated the retry_strategy and max_attempts at the same time. This is not possible and max_attempts will be ignored.\n' +
                             'node_redis: WARNING: You activated the retry_strategy and retry_max_delay at the same time. This is not possible and retry_max_delay will be ignored.\n'
                         );
-                        assert.strictEqual(err.message, 'Connection timeout');
+                        unhookIntercept();
-                        assert(!err.code);
-                        end();
                     });
                 });
 
                 it('retry_strategy used to reconnect', function (done) {
-                    var end = helper.callFuncAfter(done, 2);
                     client = redis.createClient({
                         retry_strategy: function (options) {
                             if (options.total_retry_time > 150) {
                                 client.set('foo', 'bar', function (err, res) {
                                     assert.strictEqual(err.code, 'ECONNREFUSED');
-                                    end();
+                                    done();
                                 });
                                 return false;
                             }
@@ -323,11 +317,6 @@ describe('connection tests', function () {
                         },
                         port: 9999
                     });
-
-                    client.on('error', function (err) {
-                        assert.strictEqual(err.code, 'ECONNREFUSED');
-                        end();
-                    });
                 });
             });
 

test/multi.spec.js

@@ -180,7 +180,8 @@ describe("The 'multi' method", function () {
 
                     client.multi([['set', 'foo', 'bar'], ['get', 'foo']]).exec(function (err, res) {
                         assert(/Redis connection in broken state/.test(err.message));
-                        assert.strictEqual(err.errors.length, 0);
+                        assert.strictEqual(err.errors.length, 2);
+                        assert.strictEqual(err.errors[0].args.length, 2);
                     });
                 });
 

test/node_redis.spec.js

@@ -366,7 +366,7 @@ describe('The node_redis client', function () {
                         client = redis.createClient();
                         client.quit(function () {
                             client.get('foo', function (err, res) {
-                                assert(err.message.indexOf('Redis connection gone') !== -1);
+                                assert.strictEqual(err.message, 'Stream connection ended and running command aborted. It might have been processed.');
                                 assert.strictEqual(client.offline_queue.length, 0);
                                 done();
                             });
@@ -1024,14 +1024,25 @@ describe('The node_redis client', function () {
                             helper.killConnection(client);
                         });
 
+                        var end = helper.callFuncAfter(done, 3);
                         client.on('error', function (err) {
-                            if (/uncertain state/.test(err.message)) {
+                            if (err.command === 'EXEC') {
-                                assert.equal(client.command_queue.length, 0);
+                                assert.strictEqual(client.command_queue.length, 0);
-                                done();
+                                assert.strictEqual(err.errors.length, 9);
+                                assert.strictEqual(err.errors[1].command, 'SET');
+                                assert.deepEqual(err.errors[1].args, ['foo1', 'bar1']);
+                                end();
+                            } else if (err.code === 'UNCERTAIN_STATE') {
+                                assert.strictEqual(client.command_queue.length, 0);
+                                assert.strictEqual(err.errors.length, 4);
+                                assert.strictEqual(err.errors[0].command, 'SET');
+                                assert.deepEqual(err.errors[0].args, ['foo0', 'bar0']);
+                                end();
                             } else {
                                 assert.equal(err.code, 'ECONNREFUSED');
                                 assert.equal(err.errno, 'ECONNREFUSED');
                                 assert.equal(err.syscall, 'connect');
+                                end();
                             }
                         });
                     });
@@ -1101,14 +1112,21 @@ describe('The node_redis client', function () {
                             helper.killConnection(client);
                         });
 
+                        var end = helper.callFuncAfter(done, 3);
                         client.on('error', function (err) {
-                            if (err.code === 'UNCERTAIN_STATE') {
+                            if (err.command === 'EXEC') {
                                 assert.equal(client.command_queue.length, 0);
-                                done();
+                                assert.equal(err.errors.length, 9);
+                                end();
+                            } else if (err.code === 'UNCERTAIN_STATE') {
+                                assert.equal(client.command_queue.length, 0);
+                                assert.equal(err.errors.length, 4);
+                                end();
                             } else {
                                 assert.equal(err.code, 'ECONNREFUSED');
                                 assert.equal(err.errno, 'ECONNREFUSED');
                                 assert.equal(err.syscall, 'connect');
+                                end();
                             }
                         });
                     });