diff --git a/packages/engine/src/execution/dispatcher.test.ts b/packages/engine/src/execution/dispatcher.test.ts index 8e25f5d..2b00fcf 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 dffc6bd..efd31a9 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); } }