fix: restart dispatcher polling loop after an unexpected crash

packages/engine/src/execution/dispatcher.test.ts

@@ -74,6 +74,35 @@ describe("Dispatcher", () => {
       await dispatcher.stop();
     });
 
+    sidequestTest("keeps polling after a transient backend error", async ({ backend }) => {
+      expect(await backend.listJobs({ state: "waiting" })).toHaveLength(1);
+
+      const mockClaim = vi
+        .spyOn(backend, "claimPendingJob")
+        .mockRejectedValueOnce(new Error("Connection terminated unexpectedly"));
+
+      const dispatcher = new Dispatcher(
+        backend,
+        new QueueManager(backend, config.queues!),
+        new ExecutorManager(backend, config as NonNullableEngineConfig),
+        100,
+      );
+      dispatcher.start();
+
+      runMock.mockImplementationOnce(() => {
+        return { type: "completed", result: "foo", __is_job_transition__: true } as CompletedResult;
+      });
+
+      await vi.waitUntil(async () => {
+        const jobs = await backend.listJobs({ state: "waiting" });
+        return jobs.length === 0;
+      });
+
+      expect(mockClaim.mock.calls.length).toBeGreaterThan(1);
+
+      await dispatcher.stop();
+    });
+
     sidequestTest("does not claim job when there is no available slot for the queue", async ({ backend }) => {
       await createJob(backend, "default");
 

packages/engine/src/execution/dispatcher.ts

@@ -29,46 +29,53 @@ export class Dispatcher {
    */
   private async listen() {
     while (this.isRunning) {
-      const queues = await this.queueManager.getActiveQueuesWithRunnableJobs();
+      try {
+        const queues = await this.queueManager.getActiveQueuesWithRunnableJobs();
 
-      let shouldSleep = true;
+        let shouldSleep = true;
 
-      for (const queue of queues) {
-        const availableSlots = this.executorManager.availableSlotsByQueue(queue);
-        if (availableSlots <= 0) {
-          logger("Dispatcher").debug(`Queue ${queue.name} limit reached!`);
-          await this.sleep(this.sleepDelay);
-          continue;
-        }
+        for (const queue of queues) {
+          const availableSlots = this.executorManager.availableSlotsByQueue(queue);
+          if (availableSlots <= 0) {
+            logger("Dispatcher").debug(`Queue ${queue.name} limit reached!`);
+            await this.sleep(this.sleepDelay);
+            continue;
+          }
 
-        const globalSlots = this.executorManager.availableSlotsGlobal();
-        if (globalSlots <= 0) {
-          logger("Dispatcher").debug(`Global concurrency limit reached!`);
-          await this.sleep(this.sleepDelay);
-          continue;
-        }
+          const globalSlots = this.executorManager.availableSlotsGlobal();
+          if (globalSlots <= 0) {
+            logger("Dispatcher").debug(`Global concurrency limit reached!`);
+            await this.sleep(this.sleepDelay);
+            continue;
+          }
 
-        const jobs: JobData[] = await this.backend.claimPendingJob(queue.name, Math.min(availableSlots, globalSlots));
+          const jobs: JobData[] = await this.backend.claimPendingJob(queue.name, Math.min(availableSlots, globalSlots));
 
-        if (jobs.length > 0) {
-          // if a job was found on any queue do not sleep
-          shouldSleep = false;
-        }
+          if (jobs.length > 0) {
+            // if a job was found on any queue do not sleep
+            shouldSleep = false;
+          }
 
-        for (const job of jobs) {
-          // adds jobs to active sets before execution to avoid race conditions
-          // because the execution is not awaited. This way we ensure that available slots
-          // are correctly calculated.
-          this.executorManager.queueJob(queue, job);
-          // does not await for job execution. Guard against any unexpected rejection so a single
-          // job can never crash the engine with an unhandled promise rejection.
-          void this.executorManager.execute(queue, job).catch((error: unknown) => {
-            logger("Dispatcher").error(`Unexpected error executing job ${job.id}:`, error);
-          });
+          for (const job of jobs) {
+            // adds jobs to active sets before execution to avoid race conditions
+            // because the execution is not awaited. This way we ensure that available slots
+            // are correctly calculated.
+            this.executorManager.queueJob(queue, job);
+            // does not await for job execution. Guard against any unexpected rejection so a single
+            // job can never crash the engine with an unhandled promise rejection.
+            void this.executorManager.execute(queue, job).catch((error: unknown) => {
+              logger("Dispatcher").error(`Unexpected error executing job ${job.id}:`, error);
+            });
+          }
         }
-      }
 
-      if (shouldSleep) {
+        if (shouldSleep) {
+          await this.sleep(this.sleepDelay);
+        }
+      } catch (error) {
+        // A transient backend error (e.g. a dropped DB connection) must not end the loop, otherwise
+        // no job would ever be claimed again and they would pile up in waiting state.
+        logger("Dispatcher").error("Dispatcher polling iteration failed, retrying:", error);
         await this.sleep(this.sleepDelay);
       }
     }