perf(scheduler): save offset value when every is provided (#3142)

Author: roggervalf

src/classes/job-scheduler.ts

@@ -93,19 +93,17 @@ export class JobScheduler extends QueueBase {
     }
 
     let nextMillis: number;
-    let newOffset = offset || 0;
-
+    let newOffset: number | null = null;
     if (every) {
       const prevSlot = Math.floor(startMillis / every) * every;
+
+      newOffset = typeof offset === 'number' ? offset : startMillis - prevSlot;
+
       const nextSlot = prevSlot + every;
       if (prevMillis || offset) {
         nextMillis = nextSlot;
       } else {
         nextMillis = prevSlot;
-        newOffset = startMillis - prevSlot;
-
-        // newOffset should always be positive, but we do an extra safety check
-        newOffset = newOffset < 0 ? 0 : newOffset;
       }
     } else if (pattern) {
       nextMillis = await this.repeatStrategy(now, repeatOpts, jobName);
@@ -162,6 +160,7 @@ export class JobScheduler extends QueueBase {
                 pattern,
                 every,
                 limit,
+                offset: newOffset,
               },
               Job.optsAsJSON(mergedOpts),
               producerId,
@@ -245,8 +244,8 @@ export class JobScheduler extends QueueBase {
 
     mergedOpts.repeat = {
       ...opts.repeat,
-      count: currentCount,
       offset,
+      count: currentCount,
       endDate: opts.repeat?.endDate
         ? new Date(opts.repeat.endDate).getTime()
         : undefined,
@@ -255,51 +254,6 @@ export class JobScheduler extends QueueBase {
     return mergedOpts;
   }
 
-  private createNextJob<T = any, R = any, N extends string = string>(
-    client: RedisClient,
-    name: N,
-    nextMillis: number,
-    offset: number,
-    jobSchedulerId: string,
-    opts: JobsOptions,
-    data: T,
-    currentCount: number,
-    // The job id of the job that produced this next iteration
-    producerId?: string,
-  ) {
-    //
-    // Generate unique job id for this iteration.
-    //
-    const jobId = this.getSchedulerNextJobId({
-      jobSchedulerId,
-      nextMillis,
-    });
-
-    const now = Date.now();
-    const delay = nextMillis + offset - now;
-
-    const mergedOpts = {
-      ...opts,
-      jobId,
-      delay: delay < 0 ? 0 : delay,
-      timestamp: now,
-      prevMillis: nextMillis,
-      repeatJobKey: jobSchedulerId,
-    };
-
-    mergedOpts.repeat = { ...opts.repeat, count: currentCount };
-
-    const job = new this.Job<T, R, N>(this, name, data, mergedOpts, jobId);
-    job.addJob(client);
-
-    if (producerId) {
-      const producerJobKey = this.toKey(producerId);
-      client.hset(producerJobKey, 'nrjid', job.id);
-    }
-
-    return job;
-  }
-
   async removeJobScheduler(jobSchedulerId: string): Promise<number> {
     return this.scripts.removeJobScheduler(jobSchedulerId);
   }
@@ -347,7 +301,11 @@ export class JobScheduler extends QueueBase {
       }
 
       if (jobData.every) {
-        jobSchedulerData.every = jobData.every;
+        jobSchedulerData.every = parseInt(jobData.every);
+      }
+
+      if (jobData.offset) {
+        jobSchedulerData.offset = parseInt(jobData.offset);
       }
 
       if (jobData.data || jobData.opts) {

src/classes/scripts.ts

@@ -419,7 +419,7 @@ export class Scripts {
     nextMillis: number,
     templateData: string,
     delayedJobOpts: JobsOptions,
-    // The job id of the job that produced this next iteration
+    // The job id of the job that produced this next iteration - TODO: remove in next breaking change
     producerId?: string,
   ): Promise<string | null> {
     const client = await this.queue.client;

src/commands/includes/storeJobScheduler.lua

@@ -31,6 +31,11 @@ local function storeJobScheduler(schedulerId, schedulerKey, repeatKey, nextMilli
     table.insert(optionalValues, opts['every'])
   end
 
+  if opts['offset'] then
+    table.insert(optionalValues, "offset")
+    table.insert(optionalValues, opts['offset'])
+  end
+
   local jsonTemplateOpts = cjson.encode(templateOpts)
   if jsonTemplateOpts and jsonTemplateOpts ~= '{}' then
     table.insert(optionalValues, "opts")

src/interfaces/job-scheduler-json.ts

@@ -14,7 +14,8 @@ export interface JobSchedulerJson<D = any> {
   endDate?: number;
   tz?: string;
   pattern?: string;
-  every?: string;
+  every?: number;
   next?: number;
+  offset?: number;
   template?: JobSchedulerTemplateJson<D>;
 }

src/interfaces/repeat-options.ts

@@ -43,7 +43,6 @@ export interface RepeatOptions extends Omit<ParserOptions, 'iterator'> {
 
   /**
    * Offset in milliseconds to affect the next iteration time
-   *
    * */
   offset?: number;
 

src/interfaces/repeatable-options.ts

@@ -5,4 +5,5 @@ export type RepeatableOptions = {
   limit?: number;
   pattern?: string;
   every?: number;
+  offset?: number;
 };

tests/test_job_scheduler.ts

@@ -288,7 +288,8 @@ describe('Job Scheduler', function () {
         name: 'test',
         next: 1486439520000,
         iterationCount: 1,
-        every: '240000',
+        every: 240000,
+        offset: 120003,
       });
 
       await this.clock.tickAsync(ONE_MINUTE);
@@ -334,7 +335,8 @@ describe('Job Scheduler', function () {
           name: 'test',
           next: 1486439700000,
           iterationCount: 2,
-          every: '60000',
+          every: 60000,
+          offset: 0,
         });
 
         const count = await queue.getJobCountByTypes('delayed');
@@ -1103,9 +1105,16 @@ describe('Job Scheduler', function () {
       this.clock.setSystemTime(date);
       const nextTick = 2 * ONE_SECOND;
 
+      let iterationCount = 0;
       const worker = new Worker(
         queueName,
-        async () => {
+        async job => {
+          if (iterationCount === 0) {
+            expect(job.opts.delay).to.be.eq(0);
+          } else {
+            expect(job.opts.delay).to.be.eq(2000);
+          }
+          iterationCount++;
           this.clock.tick(nextTick);
         },
         { autorun: false, connection, prefix },

tests/test_job_scheduler_stress.ts

@@ -4,7 +4,7 @@ import { default as IORedis } from 'ioredis';
 import { beforeEach, describe, it, before, after as afterAll } from 'mocha';
 
 import { v4 } from 'uuid';
-import { Queue, QueueEvents, Repeat, Worker } from '../src/classes';
+import { Job, Queue, QueueEvents, Repeat, Worker } from '../src/classes';
 import { delay, removeAllQueueData } from '../src/utils';
 
 const ONE_SECOND = 1000;
@@ -215,4 +215,67 @@ describe('Job Scheduler Stress', function () {
     expect(completedJobs).to.be.eql(1);
     await queue.close();
   });
+
+  describe("when using 'every' option and jobs are moved to active some time after delay", function () {
+    it('should repeat every 2 seconds and start immediately', async function () {
+      let iterationCount = 0;
+      const worker = new Worker(
+        queueName,
+        async job => {
+          if (iterationCount === 0) {
+            expect(job.opts.delay).to.be.eq(0);
+          } else {
+            expect(job.opts.delay).to.be.gte(1850);
+          }
+          iterationCount++;
+        },
+        { autorun: false, connection, prefix },
+      );
+
+      let prev: Job;
+      let counter = 0;
+
+      const completing = new Promise<void>((resolve, reject) => {
+        worker.on('completed', async job => {
+          try {
+            if (prev) {
+              expect(prev.timestamp).to.be.lte(job.timestamp);
+              expect(job.processedOn! - prev.processedOn!).to.be.gte(1900);
+            }
+            prev = job;
+            counter++;
+            if (counter === 5) {
+              resolve();
+            }
+          } catch (err) {
+            console.log(err);
+            reject(err);
+          }
+        });
+      });
+
+      await queue.upsertJobScheduler(
+        'repeat',
+        {
+          every: 2000,
+        },
+        { data: { foo: 'bar' } },
+      );
+
+      const waitingCountBefore = await queue.getWaitingCount();
+      expect(waitingCountBefore).to.be.eq(1);
+
+      worker.run();
+
+      await completing;
+
+      const waitingCount = await queue.getWaitingCount();
+      expect(waitingCount).to.be.eq(0);
+
+      const delayedCountAfter = await queue.getDelayedCount();
+      expect(delayedCountAfter).to.be.eq(1);
+
+      await worker.close();
+    });
+  });
 });