Do better checks when searching for jobs to resume on restart + add more unit tests

Author: b0dea

src/job/index.ts

@@ -186,7 +186,7 @@ class Job<T extends JobAttributesData = JobAttributesData> {
   attrs: JobAttributes<T>;
 
   constructor(options: Modify<JobAttributes<T>, { _id?: mongodb.ObjectId }>) {
-    const { pulse, type, nextRunAt, ...args } = options ?? {};
+    const { pulse, type, nextRunAt, repeatAt, repeatInterval, lastFinishedAt, ...args } = options ?? {};
 
     // Save Pulse instance
     this.pulse = pulse;
@@ -213,7 +213,10 @@ class Job<T extends JobAttributesData = JobAttributesData> {
       name: attrs.name || '',
       priority: attrs.priority,
       type: type || 'once',
-      nextRunAt: nextRunAt || new Date(),
+      // if a job that's non-recurring has a lastFinishedAt (finished the job), do not default nextRunAt to now
+      // only if it will be defaulted either by explicitly setting it or by computing it computeNextRunAt
+      nextRunAt:
+        repeatAt || repeatInterval ? nextRunAt || new Date() : !lastFinishedAt ? nextRunAt || new Date() : nextRunAt,
     };
   }
 

src/pulse/resume-on-restart.ts

@@ -28,11 +28,14 @@ export const resumeOnRestart: ResumeOnRestartMethod = function (this: Pulse, res
             {
               lockedAt: { $exists: true },
               nextRunAt: { $ne: null },
-              $or: [{ $expr: { $eq: ['$runCount', '$finishedCount'] } }, { lastFinishedAt: { $exists: false } }],
+              $or: [
+                { $expr: { $eq: ['$runCount', '$finishedCount'] } },
+                { $or: [{ lastFinishedAt: { $exists: false } }, { lastFinishedAt: null }] },
+              ],
             },
             {
               lockedAt: { $exists: false },
-              lastFinishedAt: { $exists: false },
+              $or: [{ lastFinishedAt: { $exists: false } }, { lastFinishedAt: null }],
               nextRunAt: { $lte: now, $ne: null },
             },
           ],
@@ -51,8 +54,10 @@ export const resumeOnRestart: ResumeOnRestartMethod = function (this: Pulse, res
     // Handling for recurring jobs using repeatInterval or repeatAt
     this._collection
       .find({
-        $or: [{ repeatInterval: { $exists: true } }, { repeatAt: { $exists: true } }],
-        nextRunAt: { $exists: false },
+        $and: [
+          { $or: [{ repeatInterval: { $exists: true } }, { repeatAt: { $exists: true } }] },
+          { $or: [{ nextRunAt: { $lte: now } }, { nextRunAt: { $exists: false } }, { nextRunAt: null }] },
+        ],
       })
       .toArray()
       .then((jobs) => {

test/unit/pulse.spec.ts

@@ -215,19 +215,30 @@ describe('Test Pulse', () => {
         expect(globalPulseInstance._resumeOnRestart).toBeFalsy();
       });
 
+      test('should not reschedule successfully finished non-recurring jobs', async () => {
+        const job = globalPulseInstance.create('sendEmail', { to: 'user@example.com' });
+        job.attrs.lastFinishedAt = new Date();
+        job.attrs.nextRunAt = null;
+        await job.save();
+
+        await globalPulseInstance.resumeOnRestart();
+
+        const updatedJob = (await globalPulseInstance.jobs({ name: 'sendEmail' }))[0];
+        expect(updatedJob.attrs.nextRunAt).toBeNull();
+      });
+
       test('should resume non-recurring jobs on restart', async () => {
         const job = globalPulseInstance.create('sendEmail', { to: 'user@example.com' });
-        job.attrs.nextRunAt = new Date(Date.now() - 1000); // Set nextRunAt in the past
+        job.attrs.nextRunAt = new Date(Date.now() - 1000);
         await job.save();
 
         await globalPulseInstance.resumeOnRestart();
 
         const updatedJob = (await globalPulseInstance.jobs({ name: 'sendEmail' }))[0];
-        const now = Date.now();
-        expect(updatedJob.attrs.nextRunAt?.getTime()).toBeGreaterThan(now - 100); // Allow a 100ms buffer
+        expect(updatedJob.attrs.nextRunAt?.getTime()).toBeGreaterThan(Date.now() - 100);
       });
 
-      test('should resume recurring jobs on restart', async () => {
+      test('should resume recurring jobs on restart - interval', async () => {
         const job = globalPulseInstance.create('sendEmail', { to: 'user@example.com' });
         job.attrs.repeatInterval = '5 minutes';
         job.attrs.nextRunAt = null;
@@ -236,9 +247,31 @@ describe('Test Pulse', () => {
         await globalPulseInstance.resumeOnRestart();
 
         const updatedJob = (await globalPulseInstance.jobs({ name: 'sendEmail' }))[0];
-        const now = Date.now();
         expect(updatedJob.attrs.nextRunAt).not.toBeNull();
-        expect(updatedJob.attrs.nextRunAt?.getTime()).toBeGreaterThan(now - 100); // Allow a 100ms buffer
+      });
+
+      test('should resume recurring jobs on restart - cron', async () => {
+        const job = globalPulseInstance.create('sendEmail', { to: 'user@example.com' });
+        job.attrs.repeatInterval = '*/5 * * * *';
+        job.attrs.nextRunAt = null;
+        await job.save();
+
+        await globalPulseInstance.resumeOnRestart();
+
+        const updatedJob = (await globalPulseInstance.jobs({ name: 'sendEmail' }))[0];
+        expect(updatedJob.attrs.nextRunAt).not.toBeNull();
+      });
+
+      test('should resume recurring jobs on restart - repeatAt', async () => {
+        const job = globalPulseInstance.create('sendEmail', { to: 'user@example.com' });
+        job.attrs.repeatAt = '1:00 am';
+        job.attrs.nextRunAt = null;
+        await job.save();
+
+        await globalPulseInstance.resumeOnRestart();
+
+        const updatedJob = (await globalPulseInstance.jobs({ name: 'sendEmail' }))[0];
+        expect(updatedJob.attrs.nextRunAt).not.toBeNull();
       });
 
       test('should not modify jobs with existing nextRunAt', async () => {
@@ -252,6 +285,61 @@ describe('Test Pulse', () => {
         const updatedJob = (await globalPulseInstance.jobs({ name: 'sendEmail' }))[0];
         expect(updatedJob.attrs.nextRunAt?.getTime()).toEqual(futureDate.getTime());
       });
+
+      test('should handle jobs that started but have not finished (non-recurring)', async () => {
+        const job = globalPulseInstance.create('processData', { data: 'sample' });
+        job.attrs.nextRunAt = null;
+        job.attrs.lockedAt = new Date();
+        await job.save();
+
+        await globalPulseInstance.resumeOnRestart();
+
+        const updatedJob = (await globalPulseInstance.jobs({ name: 'processData' }))[0];
+
+        const now = Date.now();
+        expect(updatedJob.attrs.nextRunAt).not.toBeNull();
+        expect(updatedJob.attrs.nextRunAt?.getTime()).toBeGreaterThan(now - 100);
+      });
+
+      test('should handle recurring jobs that started but have not finished', async () => {
+        const job = globalPulseInstance.create('processData', { data: 'sample' });
+        job.attrs.repeatInterval = '10 minutes';
+        job.attrs.lockedAt = new Date();
+        job.attrs.nextRunAt = new Date(Date.now() + 10000); // Next run in 10 seconds
+        await job.save();
+
+        await globalPulseInstance.resumeOnRestart();
+
+        const updatedJob = (await globalPulseInstance.jobs({ name: 'processData' }))[0];
+        expect(updatedJob.attrs.lockedAt).not.toBeNull(); // Job remains locked
+        expect(updatedJob.attrs.nextRunAt).not.toBeNull(); // Scheduling intact
+      });
+
+      test('should handle interrupted recurring jobs after server recovery', async () => {
+        const job = globalPulseInstance.create('processData', { data: 'sample' });
+        job.attrs.repeatInterval = '5 minutes';
+        job.attrs.lastModifiedBy = 'server_crash';
+        job.attrs.nextRunAt = null;
+        await job.save();
+
+        await globalPulseInstance.resumeOnRestart();
+
+        const updatedJob = (await globalPulseInstance.jobs({ name: 'processData' }))[0];
+        expect(updatedJob.attrs.nextRunAt).not.toBeNull();
+        expect(updatedJob.attrs.lastModifiedBy).not.toEqual('server_crash');
+      });
+
+      test('should not modify non-recurring jobs with lastFinishedAt in the past', async () => {
+        const job = globalPulseInstance.create('sendEmail', { to: 'user@example.com' });
+        job.attrs.lastFinishedAt = new Date(Date.now() - 10000); // Finished 10 seconds ago
+        job.attrs.nextRunAt = null;
+        await job.save();
+
+        await globalPulseInstance.resumeOnRestart();
+
+        const updatedJob = (await globalPulseInstance.jobs({ name: 'sendEmail' }))[0];
+        expect(updatedJob.attrs.nextRunAt).toBeNull(); // Job remains finished
+      });
     });
   });