diff --git a/examples/basic-app/app/queues/welcome-email.ts b/examples/basic-app/app/queues/welcome-email.ts index b1d9f952..2c494fec 100644 --- a/examples/basic-app/app/queues/welcome-email.ts +++ b/examples/basic-app/app/queues/welcome-email.ts @@ -1,13 +1,28 @@ -import { defineJob } from "@wrnexus/queue"; +import { defineQueue } from "@wrnexus/queue"; export interface WelcomeEmailPayload { userId: number; email: string; } -export default defineJob({ +/* + * A file under app/queues must default-export defineQueue(...) -- the loader + * refuses anything else rather than registering a queue that would never run. + * This example used the older defineJob() shape and so failed to load, which + * took the whole example app down with it. + */ +export default defineQueue({ name: "welcome-email", - async run(job) { - console.log(`Welcome email queued for ${job.data.email}`); + jobs: { + send: { + validate: (data: unknown): data is WelcomeEmailPayload => + typeof data === "object" && + data !== null && + Number.isInteger((data as { userId?: unknown }).userId) && + typeof (data as { email?: unknown }).email === "string", + run: async ({ email }: WelcomeEmailPayload) => { + console.log(`Welcome email queued for ${email}`); + }, + }, }, }); diff --git a/packages/queue/src/durable.ts b/packages/queue/src/durable.ts index 9509ddf3..83e705f6 100644 --- a/packages/queue/src/durable.ts +++ b/packages/queue/src/durable.ts @@ -357,7 +357,18 @@ export function createDurableQueue(options: DurableQueueOptions = {}): DurableQu }, start() { - if (!accepting) throw new Error("WRN-QUEUE-CLOSED: queue is shutting down"); + /* + * Starting is an explicit intent to run, so it reopens a queue that was + * shut down rather than refusing for ever. Without this a queue is + * single-use, which breaks any process that boots more than one app -- + * a test suite closing one harness and opening the next, a hot reload, + * a multi-tenant host. The failure lands far from its cause: the SECOND + * app to boot throws, because an unrelated one shut down earlier. + * + * `add()` keeps its guard, so work offered to a queue that is shutting + * down is still refused until someone deliberately starts it again. + */ + accepting = true; if (timer) return; timer = setInterval(() => { void this.drain().catch(async (error: unknown) => { diff --git a/packages/queue/test/defined.test.ts b/packages/queue/test/defined.test.ts index b6e99663..39cc7a2a 100644 --- a/packages/queue/test/defined.test.ts +++ b/packages/queue/test/defined.test.ts @@ -63,3 +63,32 @@ test("defineQueue validates payloads before persistence", async () => { }); await expect(queue.number.add("no" as never)).rejects.toThrow("WRN-QUEUE-PAYLOAD"); }); + +// `shutdown()` used to close a queue permanently: it set `accepting = false` +// and `start()` refused for ever after. That made a queue single-use, which +// breaks any process that boots more than one app -- a test suite closing one +// harness and opening the next, a hot reload, a multi-tenant host. The failure +// was remote from its cause: the SECOND app to boot threw WRN-QUEUE-CLOSED out +// of the dev server, so a perfectly good app failed because an unrelated one +// had shut down earlier in the same process. +test("a queue that was shut down can be started again", async () => { + const { createDurableQueue, memoryQueueStore } = await import("../src/index.ts"); + const queue = createDurableQueue({ pollMs: 5, store: memoryQueueStore() }); + + queue.start(); + await queue.shutdown(); + expect(queue.health ? (await queue.health()).accepting : false).toBe(false); + + // The explicit intent to run again reopens it. + queue.start(); + expect((await queue.health()).accepting).toBe(true); + + // And it genuinely works, rather than merely reporting that it does. + const done: string[] = []; + queue.process("work", async (job: { data: { id: string } }) => void done.push(job.data.id)); + await queue.add("work", { id: "after-restart" }); + await queue.drain(); + expect(done).toEqual(["after-restart"]); + + await queue.shutdown(); +});