n8n/patches/bull@4.16.4.patch

diff --git a/lib/job.js b/lib/job.js
index 1642d3fb35cbd64b553291660fc63bae2e7f2ef5..3301e3fbc7296f545b65331dad1f12c820267e0d 100644
--- a/lib/job.js
+++ b/lib/job.js
@@ -513,9 +513,14 @@ Job.prototype.finished = async function() {
         }
       };
 
-      const onFailed = (jobId, failedReason) => {
+      const onFailed = async (jobId, failedReason) => {
         if (String(jobId) === String(this.id)) {
-          reject(new Error(failedReason));
+          const job = await Job.fromId(this.queue, this.id);
+          const error = new Error(failedReason);
+          if (job && job.stacktrace && job.stacktrace.length > 0) {
+            error.stack = job.stacktrace.join('\n');
+          }
+          reject(new Error(error));
           removeListeners();
         }
       };
diff --git a/lib/queue.js b/lib/queue.js
index 19eefc35d815e324e47c4fe4c152f39c7c547470..d3124c7cac18b0ad26d1e0ddb20b3b76c1b35bc1 100755
--- a/lib/queue.js
+++ b/lib/queue.js
@@ -1112,12 +1112,15 @@ Queue.prototype._processJobOnNextTick = function(
   job
 ) {
   if (!this.closing) {
-    (this.paused || Promise.resolve())
+    // A job fetched by a completion before a local pause still goes to the handler.
+    Promise.resolve(job ? undefined : this.paused)
       .then(() => {
         const gettingNextJob = job ? Promise.resolve(job) : this.getNextJob();
 
         return (this.processing[index] = gettingNextJob
           .then(this.processJob)
+          // Keeps that job in this chain, so whenCurrentJobsFinished waits for it.
+          .then(next => (next && this.paused ? this.processJob(next) : next))
           .then(processJobs, err => {
             if (!(this.closing && err.message === 'Connection is closed.')) {
               utils.emitSafe(this, 'error', err, 'Error processing job');
@@ -1238,29 +1241,66 @@ Queue.prototype.getNextJob = async function() {
   }
 
   if (this.drained) {
-    //
-    // Waiting for new jobs to arrive
-    //
-    try {
-      this.bclient.blocked = true;
-      const jobId = await this.bclient.brpoplpush(
-        this.keys.wait,
-        this.keys.active,
-        this.settings.drainDelay
-      );
-      this.bclient.blocked = false;
+    // n8n patch: the idle slots of this queue take turns to wait for a job, so a
+    // new job wakes one slot in each idle worker, not every idle slot.
+    const turn = (this.n8nIdleTurn || Promise.resolve()).then(() =>
+      this.n8nWaitForJob()
+    );
+    this.n8nIdleTurn = turn.catch(() => {});
+    return turn;
+  } else {
+    return this.moveToActive();
+  }
+};
 
-      if (jobId) {
-        return this.moveToActive(jobId);
-      }
-    } catch (err) {
-      // Swallow error if locally paused since we did force a disconnection
-      if (!(this.paused && err.message === 'Connection is closed.')) {
-        throw err;
+Queue.prototype.n8nWaitForJob = async function() {
+  // A slot whose turn comes after a pause or close must not start a new wait
+  if (this.closing || this.paused) {
+    return;
+  }
+
+  //
+  // Waiting for new jobs to arrive
+  //
+  try {
+    this.bclient.blocked = true;
+    // n8n patch: BRPOPLPUSH moves a job to active before its lock is set. If the
+    // event loop blocks in between, the stall check fails a job that never ran.
+    // BLMOVE to the same list only waits for a job, and moveToActive then claims
+    // and locks it in one script. Redis before 6.2 has no BLMOVE.
+    if (!this.n8nWithoutBlmove) {
+      try {
+        const ready = await this.bclient.blmove(
+          this.keys.wait,
+          this.keys.wait,
+          'RIGHT',
+          'RIGHT',
+          this.settings.drainDelay
+        );
+        this.bclient.blocked = false;
+        return ready ? this.moveToActive() : undefined;
+      } catch (err) {
+        if (!/unknown command/i.test(err.message)) {
+          throw err;
+        }
+        this.n8nWithoutBlmove = true;
       }
     }
-  } else {
-    return this.moveToActive();
+    const jobId = await this.bclient.brpoplpush(
+      this.keys.wait,
+      this.keys.active,
+      this.settings.drainDelay
+    );
+    this.bclient.blocked = false;
+
+    if (jobId) {
+      return this.moveToActive(jobId);
+    }
+  } catch (err) {
+    // Swallow error if locally paused since we did force a disconnection
+    if (!(this.paused && err.message === 'Connection is closed.')) {
+      throw err;
+    }
   }
 };