Diagnostics / bug fix for WF queues (#654)

dbos-inc · Oct 29, 2024 · 944dffe · 944dffe
1 parent 9f293e5
commit 944dffe
Show file tree

Hide file tree

Showing 4 changed files with 32 additions and 9 deletions.
diff --git a/src/eventreceiver.ts b/src/eventreceiver.ts
@@ -104,4 +104,4 @@ export interface DBOSEventReceiverQuery
   endTime?: number;
   startSeq?: bigint;
   endSeq?: bigint;
-}
+}
diff --git a/src/system_database.ts b/src/system_database.ts
@@ -1034,10 +1034,12 @@ export class PostgresSystemDatabase implements SystemDatabase {
         if (queue.rateLimit && numRecentQueries >= queue.rateLimit.limitPerPeriod) {
           break;
         }
+
         const res = await trx<workflow_status>(`${DBOSExecutor.systemDBSchemaName}.workflow_status`)
           .where('workflow_uuid', id)
           .andWhere('status', StatusString.ENQUEUED)
           .update('status', StatusString.PENDING);
+
         if (res > 0) {
           claimedIDs.push(id);
           await trx<workflow_queue>(`${DBOSExecutor.systemDBSchemaName}.workflow_queue`)

diff --git a/src/testing/testing_runtime.ts b/src/testing/testing_runtime.ts
@@ -78,6 +78,7 @@ export interface TestingRuntime {
 
   destroy(): Promise<void>; // Release resources after tests.
   deactivateEventReceivers(): Promise<void>; // Deactivate event receivers.
+  initEventReceivers(): Promise<void>; // Init / reactivate event receivers.
 }
 
 /**
@@ -113,9 +114,6 @@ export class TestingRuntimeImpl implements TestingRuntime {
     await this.#dbosExec.init(userClasses);
     this.#server = new DBOSHttpServer(this.#dbosExec);
     await this.initEventReceivers();
-    this.#scheduler = new DBOSScheduler(this.#dbosExec);
-    this.#scheduler.initScheduler();
-    this.#wfQueueRunner = wfQueueRunner.dispatchLoop(this.#dbosExec);
     this.#applicationConfig = this.#dbosExec.config.application ?? {};
     this.#isInitialized = true;
   }
@@ -124,6 +122,9 @@ export class TestingRuntimeImpl implements TestingRuntime {
     for (const evtRcvr of this.#dbosExec!.eventReceivers) {
       await evtRcvr.initialize(this.#dbosExec!);
     }
+    this.#scheduler = new DBOSScheduler(this.#dbosExec!);
+    this.#scheduler.initScheduler();
+    this.#wfQueueRunner = wfQueueRunner.dispatchLoop(this.#dbosExec!);
   }
 
   /**
@@ -140,7 +141,13 @@ export class TestingRuntimeImpl implements TestingRuntime {
 
   async deactivateEventReceivers() {
     for (const evtRcvr of this.#server?.dbosExec?.eventReceivers || []) {
-      await evtRcvr.destroy();
+      try {
+        await evtRcvr.destroy();
+      }
+      catch (err) {
+        const e = err as Error;
+        this.#server?.dbosExec?.logger.warn(`Error destroying event receiver: ${e.message}`);
+      }
     }
     await this.#scheduler?.destroyScheduler();
     try {
@@ -150,8 +157,7 @@ export class TestingRuntimeImpl implements TestingRuntime {
     catch (err) {
       const e = err as Error;
       this.#server?.dbosExec?.logger.warn(`Error destroying workflow queue runner: ${e.message}`);
-    }  
-
+    }
   }
 
   /**

diff --git a/src/wfqueue.ts b/src/wfqueue.ts
@@ -61,14 +61,29 @@ class WFQueueRunner
 
             // Check queues
             for (const [_qn, q] of this.wfQueuesByName) {
-                const wfids = await exec.systemDatabase.findAndMarkStartableWorkflows(q);
+                let wfids: string[];
+                try {
+                    wfids = await exec.systemDatabase.findAndMarkStartableWorkflows(q);
+                }
+                catch (e) {
+                    const err = e as Error;
+                    exec.logger.warn(`Error getting startable workflows: ${err.message}`);
+                    // On the premise that this was a transaction conflict error, just try again later.
+                    wfids = [];
+                }
 
                 if (wfids.length > 0) {
                     await debugTriggerPoint(DEBUG_TRIGGER_WORKFLOW_QUEUE_START);
                 }
 
                 for (const wfid of wfids) {
-                    const _wfh = await exec.executeWorkflowUUID(wfid);
+                    try {
+                        const _wfh = await exec.executeWorkflowUUID(wfid);
+                    }
+                    catch (e) {
+                        exec.logger.warn(`Could not execute workflow with id ${wfid}`);
+                        exec.logger.warn(e);
+                    }
                 }
             }
         }