fix(scheduler): remove multi when updating a job scheduler (#3108)

roggervalf · web-flow · commit 4b619cab9a6b · 2025-03-11T07:59:11.000-06:00
diff --git a/src/classes/job-scheduler.ts b/src/classes/job-scheduler.ts
@@ -137,20 +137,19 @@ export class JobScheduler extends QueueBase {
             }
           }
 
-          const multi = (await this.client).multi();
+          const mergedOpts = this.getNextJobOpts(
+            nextMillis,
+            jobSchedulerId,
+            {
+              ...opts,
+              repeat: filteredRepeatOpts,
+              telemetry,
+            },
+            iterationCount,
+            newOffset,
+          );
 
           if (override) {
-            const mergedOpts = this.getNextJobOpts(
-              nextMillis,
-              jobSchedulerId,
-              {
-                ...opts,
-                repeat: filteredRepeatOpts,
-                telemetry,
-              },
-              iterationCount,
-              newOffset,
-            );
             const jobId = await this.scripts.addJobScheduler(
               jobSchedulerId,
               nextMillis,
@@ -185,49 +184,33 @@ export class JobScheduler extends QueueBase {
 
             return job;
           } else {
-            this.scripts.updateJobSchedulerNextMillis(
-              (<unknown>multi) as RedisClient,
+            const jobId = await this.scripts.updateJobSchedulerNextMillis(
               jobSchedulerId,
               nextMillis,
+              JSON.stringify(typeof jobData === 'undefined' ? {} : jobData),
+              Job.optsAsJSON(mergedOpts),
+              producerId,
             );
-          }
-
-          const job = this.createNextJob<T, R, N>(
-            (<unknown>multi) as RedisClient,
-            jobName,
-            nextMillis,
-            newOffset,
-            jobSchedulerId,
-            {
-              ...opts,
-              repeat: { ...filteredRepeatOpts, offset: newOffset },
-              telemetry,
-            },
-            jobData,
-            iterationCount,
-            producerId,
-          );
-
-          const results = await multi.exec(); // multi.exec returns an array of results [ err, result ][]
 
-          // Check if there are any errors
-          const erroredResult = results.find(result => result[0]);
-          if (erroredResult) {
-            throw new Error(
-              `Error upserting job scheduler ${jobSchedulerId} - ${erroredResult[0]}`,
-            );
-          }
+            if (jobId) {
+              const job = new this.Job<T, R, N>(
+                this,
+                jobName,
+                jobData,
+                mergedOpts,
+                jobId,
+              );
 
-          // Get last result with the job id
-          const lastResult = results.pop();
-          job.id = lastResult[1] as string;
+              job.id = jobId;
 
-          span?.setAttributes({
-            [TelemetryAttributes.JobSchedulerId]: jobSchedulerId,
-            [TelemetryAttributes.JobId]: job.id,
-          });
+              span?.setAttributes({
+                [TelemetryAttributes.JobSchedulerId]: jobSchedulerId,
+                [TelemetryAttributes.JobId]: job.id,
+              });
 
-          return job;
+              return job;
+            }
+          }
         },
       );
     }
diff --git a/src/classes/scripts.ts b/src/classes/scripts.ts
@@ -371,11 +371,42 @@ export class Scripts {
   }
 
   async updateJobSchedulerNextMillis(
-    client: RedisClient,
     jobSchedulerId: string,
     nextMillis: number,
-  ): Promise<number> {
-    return client.zadd(this.queue.keys.repeat, nextMillis, jobSchedulerId);
+    templateData: string,
+    delayedJobOpts: JobsOptions,
+    // The job id of the job that produced this next iteration
+    producerId?: string,
+  ): Promise<string | null> {
+    const client = await this.queue.client;
+
+    const queueKeys = this.queue.keys;
+
+    const keys: (string | number | Buffer)[] = [
+      queueKeys.repeat,
+      queueKeys.delayed,
+      queueKeys.wait,
+      queueKeys.paused,
+      queueKeys.meta,
+      queueKeys.prioritized,
+      queueKeys.marker,
+      queueKeys.id,
+      queueKeys.events,
+      queueKeys.pc,
+      producerId ? this.queue.toKey(producerId) : '',
+    ];
+
+    const args = [
+      nextMillis,
+      jobSchedulerId,
+      templateData,
+      pack(delayedJobOpts),
+      Date.now(),
+      queueKeys[''],
+      producerId,
+    ];
+
+    return this.execCommand(client, 'updateJobScheduler', keys.concat(args));
   }
 
   private removeRepeatableArgs(
diff --git a/src/commands/addJobScheduler-10.lua b/src/commands/addJobScheduler-10.lua
@@ -34,6 +34,9 @@
 local rcall = redis.call
 local repeatKey = KEYS[1]
 local delayedKey = KEYS[2]
+local waitKey = KEYS[3]
+local pausedKey = KEYS[4]
+local metaKey = KEYS[5]
 local prioritizedKey = KEYS[6]
 
 local nextMillis = ARGV[1]
@@ -42,12 +45,10 @@ local templateOpts = cmsgpack.unpack(ARGV[5])
 local prefixKey = ARGV[8]
 
 -- Includes
---- @include "includes/addDelayedJob"
---- @include "includes/addJobWithPriority"
+--- @include "includes/addJobFromScheduler"
 --- @include "includes/getOrSetMaxEvents"
 --- @include "includes/isQueuePaused"
 --- @include "includes/removeJob"
---- @include "includes/storeJob"
 --- @include "includes/storeJobScheduler"
 
 -- If we are overriding a repeatable job we must delete the delayed job for
@@ -68,12 +69,12 @@ if prevMillis ~= false then
             removeJob(currentJobId, true, prefixKey, true --[[remove debounce key]] )
             rcall("ZREM", prioritizedKey, currentJobId)
         else
-            if isQueuePaused(KEYS[5]) then
-                if rcall("LREM", KEYS[4], 1, currentJobId) > 0 then
+            if isQueuePaused(metaKey) then
+                if rcall("LREM", pausedKey, 1, currentJobId) > 0 then
                     removeJob(currentJobId, true, prefixKey, true --[[remove debounce key]] )
                 end
             else
-                if rcall("LREM", KEYS[3], 1, currentJobId) > 0 then
+                if rcall("LREM", waitKey, 1, currentJobId) > 0 then
                     removeJob(currentJobId, true, prefixKey, true --[[remove debounce key]] )
                 end
             end
@@ -86,40 +87,12 @@ storeJobScheduler(jobSchedulerId, schedulerKey, repeatKey, nextMillis, scheduler
 
 if rcall("EXISTS", nextDelayedJobKey) ~= 1 then
     local eventsKey = KEYS[9]
-    local metaKey = KEYS[5]
     local maxEvents = getOrSetMaxEvents(metaKey)
 
     rcall("INCR", KEYS[8])
 
-    local delayedOpts = cmsgpack.unpack(ARGV[6])
-
-    local delay, priority = storeJob(eventsKey, nextDelayedJobKey, nextDelayedJobId, schedulerOpts['name'], ARGV[4],
-        delayedOpts, ARGV[7], nil, nil, jobSchedulerId)
-
-    if delay ~= 0 then
-        addDelayedJob(nextDelayedJobId, delayedKey, eventsKey,
-            ARGV[7], maxEvents, KEYS[7], delay)
-    else
-        local isPaused = isQueuePaused(KEYS[5])
-      
-        -- Standard or priority add
-        if priority == 0 then
-            if isPaused then
-                -- LIFO or FIFO
-                local pushCmd = delayedOpts['lifo'] and 'RPUSH' or 'LPUSH'
-                rcall(pushCmd, KEYS[4], nextDelayedJobId)
-            else
-                -- LIFO or FIFO
-                local pushCmd = delayedOpts['lifo'] and 'RPUSH' or 'LPUSH'
-                rcall(pushCmd, KEYS[3], nextDelayedJobId)
-            end
-        else
-            -- Priority add
-            addJobWithPriority(KEYS[7], KEYS[6], priority, nextDelayedJobId, KEYS[10], isPaused)
-        end
-        -- Emit waiting event
-        rcall("XADD", eventsKey, "MAXLEN", "~", maxEvents,  "*", "event", "waiting", "jobId", nextDelayedJobId)
-    end
+    addJobFromScheduler(nextDelayedJobKey, nextDelayedJobId, ARGV[6], waitKey, pausedKey, metaKey, prioritizedKey,
+      KEYS[10], delayedKey, KEYS[7], eventsKey, schedulerOpts['name'], maxEvents, ARGV[7], ARGV[4], jobSchedulerId)
 
     if ARGV[9] ~= "" then
         rcall("HSET", ARGV[9], "nrjid", nextDelayedJobId)
diff --git a/src/commands/includes/addJobFromScheduler.lua b/src/commands/includes/addJobFromScheduler.lua
@@ -0,0 +1,41 @@
+--[[
+  Add delay marker if needed.
+]]
+
+-- Includes
+--- @include "addDelayedJob"
+--- @include "addJobWithPriority"
+--- @include "isQueuePaused"
+--- @include "storeJob"
+
+local function addJobFromScheduler(jobKey, jobId, rawOpts, waitKey, pausedKey, metaKey, prioritizedKey,
+  priorityCounter, delayedKey, markerKey, eventsKey, name, maxEvents, timestamp, data, jobSchedulerId)
+  local opts = cmsgpack.unpack(rawOpts)
+
+  local delay, priority = storeJob(eventsKey, jobKey, jobId, name, data,
+    opts, timestamp, nil, nil, jobSchedulerId)
+
+  if delay ~= 0 then
+    addDelayedJob(jobId, delayedKey, eventsKey, timestamp, maxEvents, markerKey, delay)
+  else
+    local isPaused = isQueuePaused(metaKey)
+  
+    -- Standard or priority add
+    if priority == 0 then
+      if isPaused then
+        -- LIFO or FIFO
+        local pushCmd = opts['lifo'] and 'RPUSH' or 'LPUSH'
+        rcall(pushCmd, pausedKey, jobId)
+      else
+        -- LIFO or FIFO
+        local pushCmd = opts['lifo'] and 'RPUSH' or 'LPUSH'
+        rcall(pushCmd, waitKey, jobId)
+      end
+    else
+      -- Priority add
+      addJobWithPriority(markerKey, prioritizedKey, priority, jobId, priorityCounter, isPaused)
+    end
+    -- Emit waiting event
+    rcall("XADD", eventsKey, "MAXLEN", "~", maxEvents,  "*", "event", "waiting", "jobId", jobId)
+  end
+end
diff --git a/src/commands/updateJobScheduler-11.lua b/src/commands/updateJobScheduler-11.lua
@@ -2,13 +2,17 @@
   Updates a job scheduler and adds next delayed job
 
   Input:
-    KEYS[1] 'marker',
-    KEYS[2] 'meta'
-    KEYS[3] 'id'
-    KEYS[4] 'delayed'
-    KEYS[5] events stream key
-    KEYS[6] 'repeat' key
-    KEYS[7] producer key
+    KEYS[1]  'repeat' key
+    KEYS[2]  'delayed'
+    KEYS[3]  'wait' key
+    KEYS[4]  'paused' key
+    KEYS[5]  'meta'
+    KEYS[6]  'prioritized' key
+    KEYS[7]  'marker',
+    KEYS[8]  'id'
+    KEYS[9]  events stream key
+    KEYS[10] 'pc' priority counter
+    KEYS[11] producer key
 
     ARGV[1] next milliseconds
     ARGV[2] jobs scheduler id
@@ -22,16 +26,20 @@
       next delayed job id  - OK
 ]]
 local rcall = redis.call
-local repeatKey = KEYS[6]
-local delayedKey = KEYS[4]
+local repeatKey = KEYS[1]
+local delayedKey = KEYS[2]
+local waitKey = KEYS[3]
+local pausedKey = KEYS[4]
+local metaKey = KEYS[5]
+local prioritizedKey = KEYS[6]
 local nextMillis = ARGV[1]
 local jobSchedulerId = ARGV[2]
 local timestamp = ARGV[5]
 local prefixKey = ARGV[6]
 local producerId = ARGV[7]
 
 -- Includes
---- @include "includes/addDelayedJob"
+--- @include "includes/addJobFromScheduler"
 --- @include "includes/getOrSetMaxEvents"
 
 local schedulerKey = repeatKey .. ":" .. jobSchedulerId
@@ -43,19 +51,16 @@ local prevMillis = rcall("ZSCORE", repeatKey, jobSchedulerId)
 if prevMillis ~= false then
   local currentDelayedJobId =  "repeat:" .. jobSchedulerId .. ":" .. prevMillis
 
-  if producerId == currentDelayedJobId then
+  if producerId == currentDelayedJobId and rcall("EXISTS", nextDelayedJobKey) ~= 1 then
     local schedulerAttributes = rcall("HMGET", schedulerKey, "name", "data")
 
     rcall("ZADD", repeatKey, nextMillis, jobSchedulerId)
     rcall("HINCRBY", schedulerKey, "ic", 1)
 
-    local eventsKey = KEYS[5]
-    local metaKey = KEYS[2]
+    local eventsKey = KEYS[9]
     local maxEvents = getOrSetMaxEvents(metaKey)
 
-    rcall("INCR", KEYS[3])
-
-    local delayedOpts = cmsgpack.unpack(ARGV[4])
+    rcall("INCR", KEYS[8])
 
     -- TODO: remove this workaround in next breaking change,
     -- all job-schedulers must save job data
@@ -65,11 +70,13 @@ if prevMillis ~= false then
       rcall("HSET", schedulerKey, "data", templateData)
     end
 
-    addDelayedJob(nextDelayedJobKey, nextDelayedJobId, delayedKey, eventsKey, schedulerAttributes[1],
-      templateData or '{}', delayedOpts, timestamp, jobSchedulerId, maxEvents, KEYS[1], nil, nil)
-  
-    if KEYS[7] ~= "" then
-      rcall("HSET", KEYS[7], "nrjid", nextDelayedJobId)
+    addJobFromScheduler(nextDelayedJobKey, nextDelayedJobId, ARGV[4], waitKey, pausedKey, metaKey, prioritizedKey,
+      KEYS[10], delayedKey, KEYS[7], eventsKey, schedulerAttributes[1], maxEvents, ARGV[5],
+      templateData or '{}', jobSchedulerId)
+
+    -- TODO: remove this workaround in next breaking change
+    if KEYS[11] ~= "" then
+      rcall("HSET", KEYS[11], "nrjid", nextDelayedJobId)
     end
 
     return nextDelayedJobId .. "" -- convert to string
