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

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,26 @@ 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 => {
+                    console.log(err);
+                    assert.ok(err instanceof MultiErrorReply);
+                    assert.equal(err.replies.length, 2);
+                    assert.deepEqual(err.errorIndexes, [1]);
+                    assert.ok(err.replies[1] instanceof ErrorReply);
+                    return true;
+                }
+            );
+        }, {
+            ...GLOBAL.SERVERS.OPEN,
+            minimumDockerVersion: [6, 2] // CLIENT INFO
+        });
     });
 
     testUtils.testWithClient('scripts', async client => {

packages/client/lib/errors.ts

@@ -63,3 +63,14 @@ export class ErrorReply extends Error {
         this.stack = undefined;
     }
 }
+
+export class MultiErrorReply extends ErrorReply {
+    replies: Array<ErrorReply>;
+    errorIndexes: Array<number>;
+
+    constructor(replies: Array<ErrorReply>, errorIndexes: Array<number>) {
+        super(`${errorIndexes.length} commands failed, see .replies and .errorIndexes for more information`);
+        this.replies = replies;
+        this.errorIndexes = errorIndexes;
+    }
+}

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;
@@ -79,9 +79,16 @@ export default class RedisMultiCommand {
     }
 
     transformReplies(rawReplies: Array<RedisCommandRawReply>): Array<RedisCommandRawReply> {
-        return rawReplies.map((reply, i) => {
-            const { transformReply, args } = this.queue[i];
-            return transformReply ? transformReply(reply, args.preserve) : reply;
-        });
+        const errorIndexes: Array<number> = [],
+            replies = rawReplies.map((reply, i) => {
+                if (reply instanceof ErrorReply) {
+                    errorIndexes.push(i);
+                }
+                const { transformReply, args } = this.queue[i];
+                return transformReply ? transformReply(reply, args.preserve) : reply;
+            });
+
+        if (errorIndexes.length) throw new MultiErrorReply(replies, errorIndexes);
+        return replies;
     }
 }