fix #2665 - handle errors in multi/pipeline replies (#2666)

* fix #2665 - handle errors in multi/pipeline replies

* fix MultiErrorReply replies type

* run tests on all versions, remove console.log, fix bug

* add errors iterator helper

* test `.errors()` as well

Author: leibale

packages/client/lib/client/index.spec.ts

@@ -3,7 +3,7 @@ import testUtils, { GLOBAL, waitTillBeenCalled } from '../test-utils';
 import RedisClient, { RedisClientType } from '.';
 import { RedisClientMultiCommandType } from './multi-command';
 import { RedisCommandRawReply, RedisModules, RedisFunctions, RedisScripts } from '../commands';
-import { AbortError, ClientClosedError, ClientOfflineError, ConnectionTimeoutError, DisconnectsClientError, SocketClosedUnexpectedlyError, WatchError } from '../errors';
+import { AbortError, ClientClosedError, ClientOfflineError, ConnectionTimeoutError, DisconnectsClientError, ErrorReply, MultiErrorReply, SocketClosedUnexpectedlyError, WatchError } from '../errors';
 import { defineScript } from '../lua-script';
 import { spy } from 'sinon';
 import { once } from 'events';
@@ -602,6 +602,23 @@ describe('Client', () => {
             ...GLOBAL.SERVERS.OPEN,
             minimumDockerVersion: [6, 2] // CLIENT INFO
         });
+
+        testUtils.testWithClient('should handle error replies (#2665)', async client => {
+            await assert.rejects(
+                client.multi()
+                    .set('key', 'value')
+                    .hGetAll('key')
+                    .exec(),
+                err => {
+                    assert.ok(err instanceof MultiErrorReply);
+                    assert.equal(err.replies.length, 2);
+                    assert.deepEqual(err.errorIndexes, [1]);
+                    assert.ok(err.replies[1] instanceof ErrorReply);
+                    assert.deepEqual([...err.errors()], [err.replies[1]]);
+                    return true;
+                }
+            );
+        }, GLOBAL.SERVERS.OPEN);
     });
 
     testUtils.testWithClient('scripts', async client => {

packages/client/lib/errors.ts

@@ -1,3 +1,5 @@
+import { RedisCommandRawReply } from './commands';
+
 export class AbortError extends Error {
     constructor() {
         super('The command was aborted');
@@ -63,3 +65,20 @@ export class ErrorReply extends Error {
         this.stack = undefined;
     }
 }
+
+export class MultiErrorReply extends ErrorReply {
+    replies;
+    errorIndexes;
+
+    constructor(replies: Array<RedisCommandRawReply | ErrorReply>, errorIndexes: Array<number>) {
+        super(`${errorIndexes.length} commands failed, see .replies and .errorIndexes for more information`);
+        this.replies = replies;
+        this.errorIndexes = errorIndexes;
+    }
+
+    *errors() {
+        for (const index of this.errorIndexes) {
+            yield this.replies[index];
+        }
+    }
+}

packages/client/lib/multi-command.ts

@@ -1,6 +1,6 @@
 import { fCallArguments } from './commander';
 import { RedisCommand, RedisCommandArguments, RedisCommandRawReply, RedisFunction, RedisScript } from './commands';
-import { WatchError } from './errors';
+import { ErrorReply, MultiErrorReply, WatchError } from './errors';
 
 export interface RedisMultiQueuedCommand {
     args: RedisCommandArguments;
@@ -69,7 +69,7 @@ export default class RedisMultiCommand {
         return transformedArguments;
     }
 
-    handleExecReplies(rawReplies: Array<RedisCommandRawReply>): Array<RedisCommandRawReply> {
+    handleExecReplies(rawReplies: Array<RedisCommandRawReply | ErrorReply>): Array<RedisCommandRawReply> {
         const execReply = rawReplies[rawReplies.length - 1] as (null | Array<RedisCommandRawReply>);
         if (execReply === null) {
             throw new WatchError();
@@ -78,10 +78,18 @@ export default class RedisMultiCommand {
         return this.transformReplies(execReply);
     }
 
-    transformReplies(rawReplies: Array<RedisCommandRawReply>): Array<RedisCommandRawReply> {
-        return rawReplies.map((reply, i) => {
-            const { transformReply, args } = this.queue[i];
-            return transformReply ? transformReply(reply, args.preserve) : reply;
-        });
+    transformReplies(rawReplies: Array<RedisCommandRawReply | ErrorReply>): Array<RedisCommandRawReply> {
+        const errorIndexes: Array<number> = [],
+            replies = rawReplies.map((reply, i) => {
+                if (reply instanceof ErrorReply) {
+                    errorIndexes.push(i);
+                    return reply;
+                }
+                const { transformReply, args } = this.queue[i];
+                return transformReply ? transformReply(reply, args.preserve) : reply;
+            });
+
+        if (errorIndexes.length) throw new MultiErrorReply(replies, errorIndexes);
+        return replies;
     }
 }