diff --git a/.changeset/namespace-startup-reenqueue.md b/.changeset/namespace-startup-reenqueue.md new file mode 100644 index 0000000000..c46433b6ca --- /dev/null +++ b/.changeset/namespace-startup-reenqueue.md @@ -0,0 +1,5 @@ +--- +'@workflow/world': patch +--- + +`reenqueueActiveRuns` now builds queue names with the active queue namespace (`WORKFLOW_QUEUE_NAMESPACE`, or a new optional `namespace` argument). Startup recovery previously always enqueued to the un-namespaced `__wkf_workflow_` queue; a handler registered under a namespaced prefix rejects those deliveries with `Unhandled queue`, leaving every recovered run in an endless retry loop after a world restart. diff --git a/packages/world/src/recovery.test.ts b/packages/world/src/recovery.test.ts new file mode 100644 index 0000000000..c493ae2310 --- /dev/null +++ b/packages/world/src/recovery.test.ts @@ -0,0 +1,83 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import type { Storage } from './interfaces.js'; +import type { Queue } from './queue.js'; +import { reenqueueActiveRuns } from './recovery.js'; + +function createRunsStub(workflowNames: string[]): Storage['runs'] { + return { + list: vi.fn(async (params: { status?: string }) => ({ + data: + params.status === 'pending' + ? workflowNames.map((workflowName, i) => ({ + runId: `wrun_${i}`, + workflowName, + })) + : [], + hasMore: false, + cursor: null, + })), + } as unknown as Storage['runs']; +} + +function createEnqueueSpy() { + const calls: Array<{ queueName: string; runId: string }> = []; + const enqueue = vi.fn(async (queueName: string, payload: unknown) => { + calls.push({ + queueName, + runId: (payload as { runId: string }).runId, + }); + return { messageId: `msg_${calls.length}` }; + }) as unknown as Queue['queue']; + return { enqueue, calls }; +} + +describe('reenqueueActiveRuns', () => { + afterEach(() => { + vi.unstubAllEnvs(); + }); + + it('enqueues to the default un-namespaced queue when no namespace is set', async () => { + const { enqueue, calls } = createEnqueueSpy(); + await reenqueueActiveRuns( + createRunsStub(['workflow//eve//workflowEntry']), + enqueue, + 'world-test' + ); + expect(calls).toEqual([ + { + queueName: '__wkf_workflow_workflow//eve//workflowEntry', + runId: 'wrun_0', + }, + ]); + }); + + it('honors WORKFLOW_QUEUE_NAMESPACE so recovered runs target the same queues as live enqueues', async () => { + vi.stubEnv('WORKFLOW_QUEUE_NAMESPACE', 'eve657665'); + const { enqueue, calls } = createEnqueueSpy(); + await reenqueueActiveRuns( + createRunsStub(['workflow//eve//workflowEntry']), + enqueue, + 'world-test' + ); + expect(calls).toEqual([ + { + queueName: '__eve657665_wkf_workflow_workflow//eve//workflowEntry', + runId: 'wrun_0', + }, + ]); + }); + + it('prefers an explicit namespace argument over the environment', async () => { + vi.stubEnv('WORKFLOW_QUEUE_NAMESPACE', 'fromenv'); + const { enqueue, calls } = createEnqueueSpy(); + await reenqueueActiveRuns( + createRunsStub(['my-workflow']), + enqueue, + 'world-test', + 'explicit' + ); + expect(calls).toEqual([ + { queueName: '__explicit_wkf_workflow_my-workflow', runId: 'wrun_0' }, + ]); + }); +}); diff --git a/packages/world/src/recovery.ts b/packages/world/src/recovery.ts index d20cfa33c1..5b7e2b0921 100644 --- a/packages/world/src/recovery.ts +++ b/packages/world/src/recovery.ts @@ -1,21 +1,32 @@ import type { Queue } from './queue.js'; import type { Storage } from './interfaces.js'; import type { ValidQueueName } from './queue.js'; +import { getQueueTopicPrefix, resolveQueueNamespace } from './queue.js'; /** * Re-enqueue all active (pending/running) workflow runs so they resume * processing after a world restart. The workflow handler is idempotent * (event-log replay), so duplicate enqueues are safe. * + * Queue names are built with the active queue namespace so recovered runs + * target the same queues as live enqueues; a handler registered under a + * namespaced prefix would otherwise reject them with `Unhandled queue`. + * * @param runs - Storage runs interface for listing active runs * @param enqueue - Queue's enqueue method * @param label - Log prefix for identifying the world implementation (e.g. "world-local") + * @param namespace - Optional queue namespace; defaults to `WORKFLOW_QUEUE_NAMESPACE` */ export async function reenqueueActiveRuns( runs: Storage['runs'], enqueue: Queue['queue'], - label: string + label: string, + namespace?: string ): Promise { + const prefix = getQueueTopicPrefix( + 'workflow', + resolveQueueNamespace(namespace) + ); let reenqueued = 0; for (const status of ['pending', 'running'] as const) { let cursor: string | undefined; @@ -28,7 +39,7 @@ export async function reenqueueActiveRuns( }); for (const run of page.data) { try { - const queueName: ValidQueueName = `__wkf_workflow_${run.workflowName}`; + const queueName: ValidQueueName = `${prefix}${run.workflowName}`; await enqueue(queueName, { runId: run.runId }); reenqueued++; } catch (err) {