Merge pull request #91 from oliver-oloughlin/patch/catch-cron-errors

catch errors that occur within exitOn and setInterval callbacks + upd…

Author: oliver-oloughlin

README.md

@@ -704,11 +704,11 @@ db.listenQueue(async (data) => {
 ### cron()
 
 Create a cron job that will run on interval, either indefinitely or until an
-exit condition is met. Interval defaults to 1 second if not set. Like with queue
-listeners, there can be multiple cron jobs defined.
+exit condition is met. Interval defaults to 1 hour if not set. Like with queue
+listeners, multiple cron jobs can be created.
 
 ```ts
-// Will repeat indefinitely with 1 second interval
+// Will repeat indefinitely with 1 hour interval
 db.cron(() => console.log("Hello World!"))
 
 // First job starts with a 10 second delay, after that there is a 5 second delay between jobs
@@ -722,8 +722,8 @@ db.cron(() => console.log("I terminate after running 10 times"), {
   // If this is set it will override the fixed interval
   setInterval: ({ count }) => count * 500
 
-  // Count starts at 0 and is given before the current job is run
-  exit: ({ count }) => count === 10,
+  // Count starts at 0, exitOn is run before the current job
+  exitOn: ({ count }) => count === 10,
 })
 ```
 

src/kvdex.ts

@@ -93,14 +93,17 @@ export function kvdex<const T extends SchemaDefinition>(
 
     // Add queue listener
     kv.listenQueue(async (msg) => {
+      // Parse queue message
       const parsed = parseQueueMessage(msg)
       if (!parsed.ok) {
         return
       }
 
+      // Find correct queue handlers
       const { __data__, __handlerId__ } = parsed.msg
       const handlers = queueHandlers.get(__handlerId__)
 
+      // Run queue handlers
       await allFulfilled(handlers?.map((handler) => handler(__data__)) ?? [])
     })
   }
@@ -343,6 +346,7 @@ export class KvDex<const T extends Schema<SchemaDefinition>> {
    *
    * @param job - Work that will be run for each job interval.
    * @param options - Cron options.
+   * @returns Cron job ID.
    */
   async cron(
     job: (msg: CronMessage) => unknown,
@@ -379,16 +383,33 @@ export class KvDex<const T extends Schema<SchemaDefinition>> {
     // Add cron job listener
     this.listenQueue<CronMessage>(async (msg) => {
       // Check if exit criteria is met, end repeating cron job if true
-      const exit = await options?.exitOn?.(msg) ?? false
+      let exit = false
+      try {
+        exit = await options?.exitOn?.(msg) ?? false
+      } catch (e) {
+        console.error(
+          `An error was caught while running exitOn callback for cron job {ID = ${id}}`,
+          e,
+        )
+      }
+
       if (exit) {
         await options?.onExit?.(msg)
         return
       }
 
       // Set the next interval
-      const interval = options?.setInterval
-        ? await options?.setInterval!(msg)
-        : options?.interval ?? DEFAULT_CRON_INTERVAL
+      let interval = DEFAULT_CRON_INTERVAL
+      try {
+        interval = options?.setInterval
+          ? await options?.setInterval!(msg)
+          : options?.interval ?? DEFAULT_CRON_INTERVAL
+      } catch (e) {
+        console.error(
+          `An error was caught while running setInterval callback for cron job {ID = ${id}}`,
+          e,
+        )
+      }
 
       await allFulfilled([
         // Enqueue next job
@@ -411,6 +432,9 @@ export class KvDex<const T extends Schema<SchemaDefinition>> {
       isFirstJob: true,
       enqueueTimestamp: new Date(),
     }, options?.startDelay)
+
+    // Return the cron job id
+    return id
   }
 }