fix: looptask and cron implementation

Author: climba03003

packages/cronjob/lib/adapter/mongodb.ts

@@ -49,15 +49,15 @@ export class MongoDBAdapter extends Adapter {
         uid: task.uid,
         once: task.once ?? false,
         delay: task.delay,
-        executeAt
+        executeAt,
+        isDeleted: false
       })
     } else {
-      let $set: any = null
+      const $set: any = { isDeleted: false }
       if (_task.delay !== task.delay) {
-        $set = { delay: task.delay }
+        $set.delay = task.delay
       }
       if (_task.executeAt !== executeAt) {
-        $set ??= {}
         $set.executeAt = executeAt
       }
       if ($set !== null) {

packages/cronjob/lib/cronjob.ts

@@ -32,7 +32,8 @@ export interface CronJobOptions {
 export class CronJob<RootContext = unknown> extends EventEmitter {
   // state
   application: string
-  tasks: Record<string, _TaskExecutor>
+  #tasks: Record<string, _TaskExecutor>
+  #deleted: Record<string, boolean>
   nextTick: null | NodeJS.Timeout
   isDestroyed: boolean
   isLocked: boolean
@@ -47,7 +48,8 @@ export class CronJob<RootContext = unknown> extends EventEmitter {
   constructor (options: CronJobOptions) {
     super()
     this.application = options.application
-    this.tasks = {}
+    this.#tasks = {}
+    this.#deleted = {}
     this.nextTick = null
     this.isDestroyed = false
     this.isLocked = false
@@ -69,7 +71,7 @@ export class CronJob<RootContext = unknown> extends EventEmitter {
     if (this.isDestroyed) return ''
     // we need to prefix uid
     !uid.startsWith('interval-') && (uid = `interval-${uid}`)
-    this.tasks[uid] = async () => {
+    this.#tasks[uid] = async () => {
       try {
         await executor((context ?? this.#context) as Context)
       } catch (err) {
@@ -93,7 +95,7 @@ export class CronJob<RootContext = unknown> extends EventEmitter {
     if (this.isDestroyed) return ''
     // we need to prefix uid
     !uid.startsWith('timeout-') && (uid = `timeout-${uid}`)
-    this.tasks[uid] = async () => {
+    this.#tasks[uid] = async () => {
       try {
         await executor((context ?? this.#context) as Context)
       } catch (err) {
@@ -116,7 +118,7 @@ export class CronJob<RootContext = unknown> extends EventEmitter {
     if (this.isDestroyed) return ''
     // we need to prefix uid
     !uid.startsWith('immediate-') && (uid = `immediate-${uid}`)
-    this.tasks[uid] = async () => {
+    this.#tasks[uid] = async () => {
       try {
         await executor((context ?? this.#context) as Context)
       } catch (err) {
@@ -136,43 +138,71 @@ export class CronJob<RootContext = unknown> extends EventEmitter {
     cron: string,
     uid: string,
     context?: Context
+  ): Promise<string> {
+    return await this.#setCronJob(executor, cron, uid, true, context)
+  }
+
+  async #setCronJob<Context = RootContext>(
+    executor: TaskExecutor<Context>,
+    cron: string,
+    uid: string,
+    fresh: boolean,
+    context?: Context
   ): Promise<string> {
     const nextExecuteAt = Number(parseExpression(cron).next().toDate())
     const ms = nextExecuteAt - Date.now()
+    const _uid = `timeout-cron-${uid}`
+    if (fresh) this.#deleted[_uid] = false
+    if (this.#deleted[_uid]) return ''
 
     return await this.setTimeout(async (context) => {
+      if (this.#deleted[_uid]) return
       setImmediate(() => {
         // we execute immediately for the next task
         Promise.race([
           executor(context),
-          this.setCronJob(executor, cron, uid, context)
+          this.#setCronJob(executor, cron, uid, false, context)
         ]).catch((err) => {
           this.emit('error', err)
         })
       })
-    }, ms, `timeout-cron-${uid}`, context)
+    }, ms, _uid, context)
   }
 
   async setLoopTask<Context = RootContext>(
     executor: TaskExecutor<Context>,
     uid: string,
     context?: Context
   ): Promise<string> {
+    return await this.#setLoopTask(executor, uid, true, context)
+  }
+
+  async #setLoopTask<Context = RootContext>(
+    executor: TaskExecutor<Context>,
+    uid: string,
+    fresh: boolean,
+    context?: Context
+  ): Promise<string> {
+    const _uid = `immediate-loop-${uid}`
+    if (fresh) this.#deleted[_uid] = false
+    if (this.#deleted[_uid]) return ''
     return await this.setImmediate(async (context) => {
+      if (this.#deleted[_uid]) return
       try {
         await executor(context)
       } finally {
-        await this.setLoopTask(executor, uid, context)
+        await this.#setLoopTask(executor, uid, false, context)
       }
-    }, `timeout-loop-${uid}`, context)
+    }, _uid, context)
   }
 
   clearInterval (uid: string): void {
+    this.#deleted[uid] = true
     this.#deleteTask(uid)
   }
 
   clearTimeout (uid: string): void {
-    this.#deleteTask(uid)
+    this.clearInterval(uid)
   }
 
   async #_deleteTask (uid: string): Promise<void> {
@@ -228,7 +258,7 @@ export class CronJob<RootContext = unknown> extends EventEmitter {
       await this.adapter.updateTask(task.uid, executeAt, task.once)
     }
 
-    const executor = this.tasks[task.uid]
+    const executor = this.#tasks[task.uid]
     if (typeof executor !== 'function') {
       // when we missing runtime
       // we delay the task to maxExecutionMS

packages/cronjob/test/adapter/mongodb.test.ts

@@ -1,3 +1,4 @@
+import { parseExpression } from 'cron-parser'
 import Fastify from 'fastify'
 import { MongoClient } from 'mongodb'
 import assert from 'node:assert/strict'
@@ -19,6 +20,10 @@ function createDeferredPromise (): { promise: Promise<void>, resolve: () => void
   return promise
 }
 
+function isJob (uid: unknown, expected: string): boolean {
+  return String(uid).includes(expected)
+}
+
 test('MongoDBAdapter', async function (t) {
   const client = new MongoClient(MONGODB_URI)
   await client.connect()
@@ -49,15 +54,18 @@ test('MongoDBAdapter', async function (t) {
     const now = Date.now()
 
     if (
-      (String(task.uid).includes('interval') || String(task.uid).includes('timeout')) &&
-      timestamps[task.uid].length === 2
+      (
+        isJob(task.uid, 'interval') ||
+        isJob(task.uid, 'timeout') ||
+        isJob(task.uid, 'immediate')
+      ) && timestamps[task.uid].length === 2
     ) {
       const expected = timestamps[task.uid][0]
       const _from = expected - RANDOM_GAP
       const _to = expected + RANDOM_GAP
       const diff = now - expected
 
-      if (String(task.uid).includes('interval')) {
+      if (isJob(task.uid, 'interval')) {
         if (ticks[task.uid] >= 2) {
           fastify.cronjob.clearInterval(task.uid as string)
           assert.equal(ticks[task.uid], 2)
@@ -66,12 +74,29 @@ test('MongoDBAdapter', async function (t) {
           timestamps[task.uid][0] = now + task.delay
           assert.equal(ticks[task.uid], 1)
         }
+      } else if (isJob(task.uid, 'cron')) {
+        if (ticks[task.uid] >= 2) {
+          fastify.cronjob.clearInterval(task.uid as string)
+          assert.equal(ticks[task.uid] >= 2, true)
+          dones[task.uid]()
+        } else {
+          timestamps[task.uid][0] = now + task.delay
+          assert.equal(ticks[task.uid], 1)
+        }
+      } else if (isJob(task.uid, 'loop')) {
+        fastify.cronjob.clearInterval(task.uid as string)
+        assert.equal(ticks[task.uid] >= 1, true)
+        dones[task.uid]()
       } else {
+        fastify.cronjob.clearInterval(task.uid as string)
         assert.equal(ticks[task.uid], 1)
         dones[task.uid]()
       }
-      assert.equal(_from < now && now < _to, true)
-      assert.equal(diff < RANDOM_GAP, true)
+
+      if (!isJob(task.uid, 'loop') && !isJob(task.uid, 'cron')) {
+        assert.equal(_from < now && now < _to, true)
+        assert.equal(diff < RANDOM_GAP, true)
+      }
     }
   })
 
@@ -97,6 +122,25 @@ test('MongoDBAdapter', async function (t) {
     await promise.promise
   }
 
+  const checkCronJob = async function (cron: string): Promise<void> {
+    const next = parseExpression(cron).next().toDate()
+    const promise = createDeferredPromise()
+    const uid = await fastify.cronjob.setCronJob(async () => {}, cron, cron)
+    dones[uid] = promise.resolve
+    timestamps[uid] = [+next]
+    ticks[uid] = 0
+    await promise.promise
+  }
+
+  const checkLoopTask = async function (name: number): Promise<void> {
+    const promise = createDeferredPromise()
+    const uid = await fastify.cronjob.setLoopTask(async () => {}, '' + name)
+    dones[uid] = promise.resolve
+    timestamps[uid] = [Date.now()]
+    ticks[uid] = 0
+    await promise.promise
+  }
+
   t.after(async function () {
     await fastify.close()
     await db.dropDatabase()
@@ -142,4 +186,22 @@ test('MongoDBAdapter', async function (t) {
     ]
     await Promise.allSettled(promises)
   })
+
+  await t.test('cronjob', async function () {
+    const promises = [
+      checkCronJob('* * * * * *'),
+      checkCronJob('*/2 * * * * *'),
+      checkCronJob('*/3 * * * * *'),
+      checkCronJob('*/4 * * * * *'),
+      checkCronJob('*/5 * * * * *')
+    ]
+    await Promise.allSettled(promises)
+  })
+
+  await t.test('looptask', async function () {
+    const promises = [
+      checkLoopTask(384)
+    ]
+    await Promise.allSettled(promises)
+  })
 })