Fix transaction error handling (#973)

* fix exec error handling

* lint

---------

Co-authored-by: Evan Tahler <evan@evantahler.com>

src/core/queue.ts

@@ -86,12 +86,18 @@ export class Queue extends EventEmitter {
     const toRun = await RunPlugins(this, "beforeEnqueue", func, q, job, args);
     if (toRun === false) return toRun;
 
-    await this.connection.redis
+    const response = await this.connection.redis
       .multi()
       .sadd(this.connection.key("queues"), q)
       .rpush(this.connection.key("queue", q), this.encode(q, func, args))
       .exec();
 
+    response.forEach((res) => {
+      if (res[0] !== null) {
+        throw res[0];
+      }
+    });
+
     await RunPlugins(this, "afterEnqueue", func, q, job, args);
     return toRun;
   }
@@ -129,7 +135,7 @@ export class Queue extends EventEmitter {
       }
     }
 
-    await this.connection.redis
+    const response = await this.connection.redis
       .multi()
       .rpush(this.connection.key("delayed:" + rTimestamp), item)
       .sadd(this.connection.key("timestamps:" + item), "delayed:" + rTimestamp)
@@ -139,6 +145,12 @@ export class Queue extends EventEmitter {
         rTimestamp.toString(),
       )
       .exec();
+
+    response.forEach((res) => {
+      if (res[0] !== null) {
+        throw res[0];
+      }
+    });
   }
   /**
    * - In ms, the number of ms to delay before this job is able to start being worked on.
@@ -168,11 +180,17 @@ export class Queue extends EventEmitter {
    */
   async delQueue(q: string) {
     const { redis } = this.connection;
-    await redis
+    const response = await redis
       .multi()
       .del(this.connection.key("queue", q))
       .srem(this.connection.key("queues"), q)
       .exec();
+
+    response.forEach((res) => {
+      if (res[0] !== null) {
+        throw res[0];
+      }
+    });
   }
 
   /**
@@ -230,7 +248,14 @@ export class Queue extends EventEmitter {
       }
     }
 
-    await pipeline.exec();
+    const response = await pipeline.exec();
+
+    response.forEach((res) => {
+      if (res[0] !== null) {
+        throw res[0];
+      }
+    });
+
     return numJobsDeleted;
   }
 
@@ -258,7 +283,14 @@ export class Queue extends EventEmitter {
       }
     }
 
-    await pipeline.exec();
+    const response = await pipeline.exec();
+
+    response.forEach((res) => {
+      if (res[0] !== null) {
+        throw res[0];
+      }
+    });
+
     return timestamps.map((t) => parseInt(t, 10));
   }
 
@@ -501,7 +533,13 @@ export class Queue extends EventEmitter {
       );
     }
 
-    await pipeline.exec();
+    const response = await pipeline.exec();
+
+    response.forEach((res) => {
+      if (res[0] !== null) {
+        throw res[0];
+      }
+    });
 
     return errorPayload;
   }

src/core/scheduler.ts

@@ -254,11 +254,17 @@ export class Scheduler extends EventEmitter {
     await this.watchIfPossible(this.connection.key("delayed_queue_schedule"));
     const length = await this.connection.redis.llen(key);
     if (length === 0) {
-      await this.connection.redis
+      const response = await this.connection.redis
         .multi()
         .del(key)
         .zrem(this.connection.key("delayed_queue_schedule"), timestamp)
         .exec();
+
+      response.forEach((res) => {
+        if (res[0] !== null) {
+          throw res[0];
+        }
+      });
     }
     await this.unwatchIfPossible();
   }

src/core/worker.ts

@@ -404,16 +404,23 @@ export class Worker extends EventEmitter {
   }
 
   private async succeed(job: ParsedJob, duration: number) {
-    await this.connection.redis
+    const response = await this.connection.redis
       .multi()
       .incr(this.connection.key("stat", "processed"))
       .incr(this.connection.key("stat", "processed", this.name))
       .exec();
+
+    response.forEach((res) => {
+      if (res[0] !== null) {
+        throw res[0];
+      }
+    });
+
     this.emit("success", this.queue, job, this.result, duration);
   }
 
   private async fail(err: Error, duration: number) {
-    await this.connection.redis
+    const response = await this.connection.redis
       .multi()
       .incr(this.connection.key("stat", "failed"))
       .incr(this.connection.key("stat", "failed", this.name))
@@ -422,6 +429,13 @@ export class Worker extends EventEmitter {
         JSON.stringify(this.failurePayload(err, this.job)),
       )
       .exec();
+
+    response.forEach((res) => {
+      if (res[0] !== null) {
+        throw res[0];
+      }
+    });
+
     this.emit("failure", this.queue, this.job, err, duration);
   }
 
@@ -512,7 +526,7 @@ export class Worker extends EventEmitter {
       return;
     }
 
-    await this.connection.redis
+    const response = await this.connection.redis
       .multi()
       .srem(this.connection.key("workers"), name + ":" + queues)
       .del(this.connection.key("worker", "ping", name))
@@ -521,6 +535,12 @@ export class Worker extends EventEmitter {
       .del(this.connection.key("stat", "failed", name))
       .del(this.connection.key("stat", "processed", name))
       .exec();
+
+    response.forEach((res) => {
+      if (res[0] !== null) {
+        throw res[0];
+      }
+    });
   }
 
   async checkQueues() {