Fix namespaced active run recovery (#2888)

Signed-off-by: Casey Gowrie <ctgowrie@gmail.com>

Author: ctgowrie

.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.

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 });

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();

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<Queue['queue']>();
+
+    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<Queue['queue']>();
+
+    await reenqueueActiveRuns(createRuns(), enqueue, 'test', 'explicit');
+
+    expect(enqueue).toHaveBeenCalledWith('__explicit_wkf_workflow_myWorkflow', {
+      runId: 'wrun_AAA',
+    });
+  });
+});

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<void> {
+  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) {