From 4a25d523f3ef45a0bb0630f39967b9f431b61bfb Mon Sep 17 00:00:00 2001 From: Damien Brown Date: Tue, 8 Sep 2026 22:13:34 +1000 Subject: [PATCH 1/2] fix: restart dispatcher polling loop after an unexpected crash Since #204, a rejection inside Dispatcher.listen() (e.g. a transient "terminating connection due to administrator command" from Postgres) is caught and logged, but the loop is never restarted. The worker stays alive with a dead dispatcher, so the engine's respawn path is never triggered and every job piles up in the waiting state until the process is manually restarted. Restart the loop from the existing .catch() after sleeping for the polling interval, as long as the dispatcher has not been stopped. Fixes #213 --- .../engine/src/execution/dispatcher.test.ts | 29 +++++++++++++++++++ packages/engine/src/execution/dispatcher.ts | 10 +++++-- 2 files changed, 36 insertions(+), 3 deletions(-) diff --git a/packages/engine/src/execution/dispatcher.test.ts b/packages/engine/src/execution/dispatcher.test.ts index 8e25f5d8..2b00fcfb 100644 --- a/packages/engine/src/execution/dispatcher.test.ts +++ b/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"); diff --git a/packages/engine/src/execution/dispatcher.ts b/packages/engine/src/execution/dispatcher.ts index dffc6bd3..b0fb8cd2 100644 --- a/packages/engine/src/execution/dispatcher.ts +++ b/packages/engine/src/execution/dispatcher.ts @@ -85,13 +85,17 @@ export class Dispatcher { } /** - * Starts the dispatcher loop. + * Starts the dispatcher loop. If the loop crashes unexpectedly it is restarted after a short delay. */ start() { logger("Dispatcher").debug(`Starting dispatcher...`); this.isRunning = true; - void this.listen().catch((error: unknown) => { - logger("Dispatcher").error("Dispatcher polling loop crashed unexpectedly:", error); + void this.listen().catch(async (error: unknown) => { + // A transient backend error (e.g. a dropped DB connection) rejects the loop. Without a restart + // no job would ever be claimed again and they would pile up in waiting state. + logger("Dispatcher").error("Dispatcher polling loop crashed unexpectedly, restarting:", error); + await this.sleep(this.sleepDelay); + if (this.isRunning) this.start(); }); } From e695e3d6558e9ede3bc7b8c012740be25470d574 Mon Sep 17 00:00:00 2001 From: Damien Brown Date: Tue, 8 Sep 2026 22:25:08 +1000 Subject: [PATCH 2/2] fix: catch inside the dispatcher loop instead of restarting it Address review feedback: wrap the body of the while loop in listen() in a try/catch so a failed iteration logs, backs off for the polling interval, and continues. start() is restored to its previous form. --- packages/engine/src/execution/dispatcher.ts | 81 +++++++++++---------- 1 file changed, 42 insertions(+), 39 deletions(-) diff --git a/packages/engine/src/execution/dispatcher.ts b/packages/engine/src/execution/dispatcher.ts index b0fb8cd2..efd31a9d 100644 --- a/packages/engine/src/execution/dispatcher.ts +++ b/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); } } @@ -85,17 +92,13 @@ export class Dispatcher { } /** - * Starts the dispatcher loop. If the loop crashes unexpectedly it is restarted after a short delay. + * Starts the dispatcher loop. */ start() { logger("Dispatcher").debug(`Starting dispatcher...`); this.isRunning = true; - void this.listen().catch(async (error: unknown) => { - // A transient backend error (e.g. a dropped DB connection) rejects the loop. Without a restart - // no job would ever be claimed again and they would pile up in waiting state. - logger("Dispatcher").error("Dispatcher polling loop crashed unexpectedly, restarting:", error); - await this.sleep(this.sleepDelay); - if (this.isRunning) this.start(); + void this.listen().catch((error: unknown) => { + logger("Dispatcher").error("Dispatcher polling loop crashed unexpectedly:", error); }); }