diff --git a/apps/desktop/src/lib/trpc/workers/WorkerTaskRunner.test.ts b/apps/desktop/src/lib/trpc/workers/WorkerTaskRunner.test.ts index c458f8bcfbb..4b69612ee4a 100644 --- a/apps/desktop/src/lib/trpc/workers/WorkerTaskRunner.test.ts +++ b/apps/desktop/src/lib/trpc/workers/WorkerTaskRunner.test.ts @@ -262,4 +262,34 @@ describe("WorkerTaskRunner failure handling", () => { expect(workers[0]?.terminated).toBe(true); await runner.dispose(); }); + + test("a pre-aborted queued task does not strand the task behind it", async () => { + const runner = new WorkerTaskRunner({ + workerScriptPath: "/mock/worker.js", + concurrency: 1, + createWorker, + }); + + const first = runner.runTask("status", { id: "first" }); + const aborted = runner.runTask( + "status", + { id: "aborted" }, + { signal: AbortSignal.abort() }, + ); + let third: unknown = "pending"; + void runner.runTask("status", { id: "third" }).then( + (result) => { + third = result; + }, + (error) => { + third = error; + }, + ); + + await first; + await expect(aborted).rejects.toThrow(); + await flushMicrotasks(10); + expect(third).toEqual({ workerId: 1 }); + await runner.dispose(); + }); }); diff --git a/apps/desktop/src/lib/trpc/workers/WorkerTaskRunner.ts b/apps/desktop/src/lib/trpc/workers/WorkerTaskRunner.ts index a8b018a45f6..f9f8e4d8905 100644 --- a/apps/desktop/src/lib/trpc/workers/WorkerTaskRunner.ts +++ b/apps/desktop/src/lib/trpc/workers/WorkerTaskRunner.ts @@ -256,29 +256,31 @@ export class WorkerTaskRunner { } if (this.queue.length === 0) break; - const nextTaskId = this.queue.shift(); - if (!nextTaskId) continue; - const task = this.tasks.get(nextTaskId); - if (!task) continue; - - if (task.abortSignal?.aborted) { - this.rejectTask(task.taskId, new WorkerTaskAbortedError("cancelled")); - continue; + while (this.queue.length > 0 && !slot.activeTaskId) { + const nextTaskId = this.queue.shift(); + if (!nextTaskId) continue; + const task = this.tasks.get(nextTaskId); + if (!task) continue; + + if (task.abortSignal?.aborted) { + this.rejectTask(task.taskId, new WorkerTaskAbortedError("cancelled")); + continue; + } + + slot.activeTaskId = task.taskId; + task.slotId = slot.id; + task.timeoutHandle = setTimeout(() => { + this.handleTaskTimeout(task.taskId); + }, task.timeoutMs); + + const request: WorkerTaskRequestMessage = { + kind: "task", + taskId: task.taskId, + taskType: task.taskType, + payload: task.payload, + }; + slot.worker.postMessage(request); } - - slot.activeTaskId = task.taskId; - task.slotId = slot.id; - task.timeoutHandle = setTimeout(() => { - this.handleTaskTimeout(task.taskId); - }, task.timeoutMs); - - const request: WorkerTaskRequestMessage = { - kind: "task", - taskId: task.taskId, - taskType: task.taskType, - payload: task.payload, - }; - slot.worker.postMessage(request); } }