diff --git a/.changeset/fix-recovery-queue-namespace.md b/.changeset/fix-recovery-queue-namespace.md new file mode 100644 index 0000000000..7ae929cc08 --- /dev/null +++ b/.changeset/fix-recovery-queue-namespace.md @@ -0,0 +1,7 @@ +--- +'@workflow/world': patch +'@workflow/world-local': patch +'@workflow/world-postgres': patch +--- + +Use the active queue namespace when re-enqueuing workflow runs during world startup recovery. diff --git a/packages/world-local/src/reenqueue.test.ts b/packages/world-local/src/reenqueue.test.ts index ac38bfa9e9..97eeea00fa 100644 --- a/packages/world-local/src/reenqueue.test.ts +++ b/packages/world-local/src/reenqueue.test.ts @@ -1,6 +1,7 @@ import { rm } from 'node:fs/promises'; import os from 'node:os'; import path from 'node:path'; +import { WorkflowInvokePayloadSchema } from '@workflow/world'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { createWorld } from './index.js'; import { createRun, updateRun } from './test-helpers.js'; @@ -14,10 +15,12 @@ describe('re-enqueue active runs on start', () => { let dataDir: string; beforeEach(() => { + vi.stubEnv('WORKFLOW_QUEUE_NAMESPACE', undefined); dataDir = path.join(os.tmpdir(), `wf-reenqueue-${Date.now()}`); }); afterEach(async () => { + vi.unstubAllEnvs(); await rm(dataDir, { recursive: true, force: true }); }); @@ -86,6 +89,49 @@ describe('re-enqueue active runs on start', () => { await world2.close(); }); + it('re-enqueues runs to the active queue namespace', async () => { + vi.stubEnv('WORKFLOW_QUEUE_NAMESPACE', 'custom'); + + const world1 = createWorld({ dataDir }); + await world1.start(); + + const pendingRun = await createRun(world1, { + deploymentId: 'dpl_1', + workflowName: 'myWorkflow', + input: new Uint8Array([1]), + }); + + await world1.close(); + + const world2 = createWorld({ dataDir }); + const namespacedRunIds: string[] = []; + const unnamespacedRunIds: string[] = []; + const namespacedHandler = world2.createQueueHandler( + '__custom_wkf_workflow_', + async (message) => { + const body = WorkflowInvokePayloadSchema.parse(message); + namespacedRunIds.push(body.runId); + } + ); + world2.registerHandler('__custom_wkf_workflow_', namespacedHandler); + // Capture an incorrectly reconstructed queue without allowing it to enter + // the local queue's retry loop. + world2.registerHandler('__wkf_workflow_', async (req) => { + const body = await req.json(); + unnamespacedRunIds.push(body.runId); + return Response.json({ ok: true }); + }); + + await world2.start(); + + await vi.waitFor(() => { + expect(namespacedRunIds).toEqual([pendingRun.runId]); + }); + expect(unnamespacedRunIds).toHaveLength(0); + + await world2.close(); + }); + it('does nothing when there are no active runs', async () => { // Create a world with only completed runs const world1 = createWorld({ dataDir }); diff --git a/packages/world-postgres/src/index.ts b/packages/world-postgres/src/index.ts index 95c66e522c..d9ab8280f3 100644 --- a/packages/world-postgres/src/index.ts +++ b/packages/world-postgres/src/index.ts @@ -70,7 +70,12 @@ export function createWorld( }), async start() { await queue.start(); - await reenqueueActiveRuns(storage.runs, queue.queue, 'world-postgres'); + await reenqueueActiveRuns( + storage.runs, + queue.queue, + 'world-postgres', + config.namespace + ); }, async close() { await streamer.close(); diff --git a/packages/world/src/recovery.test.ts b/packages/world/src/recovery.test.ts new file mode 100644 index 0000000000..64bf9d961b --- /dev/null +++ b/packages/world/src/recovery.test.ts @@ -0,0 +1,51 @@ +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 createRuns(): Storage['runs'] { + return { + list: vi.fn(async ({ status }) => ({ + data: + status === 'pending' + ? [ + { + runId: 'wrun_AAA', + workflowName: 'myWorkflow', + status, + }, + ] + : [], + hasMore: false, + cursor: null, + })), + } as unknown as Storage['runs']; +} + +describe('reenqueueActiveRuns', () => { + afterEach(() => { + vi.unstubAllEnvs(); + }); + + it('uses WORKFLOW_QUEUE_NAMESPACE for recovered runs', async () => { + vi.stubEnv('WORKFLOW_QUEUE_NAMESPACE', 'custom'); + const enqueue = vi.fn(); + + await reenqueueActiveRuns(createRuns(), enqueue, 'test'); + + expect(enqueue).toHaveBeenCalledWith('__custom_wkf_workflow_myWorkflow', { + runId: 'wrun_AAA', + }); + }); + + it('prefers an explicit namespace over WORKFLOW_QUEUE_NAMESPACE', async () => { + vi.stubEnv('WORKFLOW_QUEUE_NAMESPACE', 'environment'); + const enqueue = vi.fn(); + + await reenqueueActiveRuns(createRuns(), enqueue, 'test', 'explicit'); + + expect(enqueue).toHaveBeenCalledWith('__explicit_wkf_workflow_myWorkflow', { + runId: 'wrun_AAA', + }); + }); +}); diff --git a/packages/world/src/recovery.ts b/packages/world/src/recovery.ts index d20cfa33c1..ec13face94 100644 --- a/packages/world/src/recovery.ts +++ b/packages/world/src/recovery.ts @@ -1,6 +1,10 @@ -import type { Queue } from './queue.js'; import type { Storage } from './interfaces.js'; import type { ValidQueueName } from './queue.js'; +import { + getQueueTopicPrefix, + type Queue, + resolveQueueNamespace, +} from './queue.js'; /** * Re-enqueue all active (pending/running) workflow runs so they resume @@ -10,12 +14,18 @@ import type { ValidQueueName } from './queue.js'; * @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 workflowQueuePrefix = getQueueTopicPrefix( + 'workflow', + resolveQueueNamespace(namespace) + ); let reenqueued = 0; for (const status of ['pending', 'running'] as const) { let cursor: string | undefined; @@ -28,7 +38,7 @@ export async function reenqueueActiveRuns( }); for (const run of page.data) { try { - const queueName: ValidQueueName = `__wkf_workflow_${run.workflowName}`; + const queueName: ValidQueueName = `${workflowQueuePrefix}${run.workflowName}`; await enqueue(queueName, { runId: run.runId }); reenqueued++; } catch (err) {