fix(world-postgres): serialize keyed deliveries across workers

.changeset/postgres-keyed-delivery-serialization.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-postgres': patch
+---
+
+Serialize keyed queue deliveries across Postgres worker processes.

packages/world-postgres/HOW_IT_WORKS.md

@@ -36,6 +36,8 @@ Call `world.start()` to initialize graphile-worker workers. When `.start()` is c
 
 When the runtime returns `{ timeoutSeconds }`, the worker schedules a new Graphile job with a future `runAt` time before finishing the current task.
 
+Messages carrying an idempotency key are deterministically assigned to one of 2,048 Graphile named queues per configured job prefix. Graphile holds that queue while the handler runs, so a same-key replacement remains pending rather than executing on another worker. The default replacement behavior stays enabled so delayed reschedules and crash recovery retain a durable successor. Because named queues persist after their jobs complete, the bounded mapping avoids an ever-growing queue table; unrelated keys that hash to the same bucket are serialized.
+
 The worker sends workflow orchestration and queued step messages to the combined `.well-known/workflow/v1/flow` endpoint.
 
 In **Next.js**, the `world.start()` call needs to be added to `instrumentation.ts|js` to ensure workers start before request handling. Use `workflow/runtime` for `getWorld` (same as the testing server and other framework plugins):

packages/world-postgres/README.md

@@ -187,6 +187,9 @@ and its token can be reused. If the token is never reused, the expired
 - Graphile jobs are acknowledged only after execution finishes, or after the worker durably schedules a delayed follow-up job
 - Backlog stays in PostgreSQL when all execution slots are busy
 - Retry and sleep-style delays use Graphile `runAt` scheduling
+- Messages with an idempotency key are assigned to one of 2,048 Graphile named queues per `jobPrefix`. This prevents the same keyed delivery from executing concurrently across worker processes while keeping delayed successors durable. Hash collisions can serialize unrelated keyed messages.
+- Existing jobs from a version without named queues do not participate in this lock. Drain keyed jobs and old producers during the first rollout or rollback when a fully protected handoff is required.
+- An abrupt process death can hold its named queue until Graphile Worker's four-hour stale-lock recovery. Graceful shutdown releases it normally.
 - Workflow orchestration and queued step execution are both sent through `/.well-known/workflow/v1/flow`
 
 ## Development

packages/world-postgres/src/queue.test.ts

@@ -15,6 +15,10 @@ import { MessageData } from './message.js';
 import { createQueue } from './queue.js';
 
 const transport = new JsonTransport();
+const DEFAULT_IDEMPOTENCY_QUEUE =
+  'workflow_idempotency_9d70a9b47b31ee4a598d79d3949545c491ca35b9fb2aed6b7ac8bdc277f09df3_1a5';
+const PREFIXED_IDEMPOTENCY_QUEUE =
+  'workflow_idempotency_4ed3ab71b053166eab6696d23bd4b98bba01573451cf7fbad1c531a9460e716b_1a5';
 const createdQueues: Array<ReturnType<typeof createQueue>> = [];
 const createdServers: Server[] = [];
 
@@ -489,6 +493,7 @@ describe('postgres queue http execution', () => {
         expect.objectContaining({
           jobKey: 'step_01ABC',
           maxAttempts: 49,
+          queueName: DEFAULT_IDEMPOTENCY_QUEUE,
           runAt: new Date('2024-01-01T00:00:05.000Z'),
         })
       );
@@ -497,6 +502,153 @@ describe('postgres queue http execution', () => {
     }
   });
 
+  it('keeps delayed retries on the same idempotency queue', async () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date('2024-01-01T00:00:00.000Z'));
+    const fetchMock = vi.fn(async () => Response.json({ timeoutSeconds: 300 }));
+    vi.stubGlobal('fetch', fetchMock);
+    process.env.WORKFLOW_LOCAL_BASE_URL = 'https://workflow.example.test';
+
+    try {
+      const queue = buildQueue({ connectionString: 'postgres://test' }, pool);
+      await queue.start();
+
+      const task = getTaskHandler('workflow_flows');
+      await task(
+        buildMessageData(
+          '__wkf_workflow_test-step',
+          {
+            runId: 'run_01ABC',
+            stepId: 'step_01ABC',
+            stepName: 'test-step',
+          },
+          { idempotencyKey: 'step_01ABC' }
+        ),
+        { job: { attempts: 1 } }
+      );
+
+      expect(workerUtilsMock.addJob).toHaveBeenCalledWith(
+        'workflow_flows',
+        expect.objectContaining({
+          attempt: 2,
+          idempotencyKey: 'step_01ABC',
+        }),
+        expect.objectContaining({
+          jobKey: 'step_01ABC',
+          maxAttempts: 49,
+          queueName: DEFAULT_IDEMPOTENCY_QUEUE,
+          runAt: new Date('2024-01-01T00:05:00.000Z'),
+        })
+      );
+    } finally {
+      vi.unstubAllGlobals();
+      vi.useRealTimers();
+    }
+  });
+
+  it('does not assign an idempotency queue to unkeyed messages', async () => {
+    const queue = buildQueue({ connectionString: 'postgres://test' }, pool);
+    await queue.start();
+
+    await queue.queue('__wkf_workflow_test-workflow', {
+      runId: 'run_01ABC',
+    });
+
+    expect(workerUtilsMock.addJob).toHaveBeenCalledWith(
+      'workflow_flows',
+      expect.anything(),
+      expect.not.objectContaining({ queueName: expect.anything() })
+    );
+  });
+
+  it('scopes idempotency queues to the configured job prefix', async () => {
+    const queue = buildQueue(
+      { connectionString: 'postgres://test', jobPrefix: 'billing_' },
+      pool
+    );
+    await queue.start();
+
+    await queue.queue(
+      '__wkf_workflow_test-step',
+      {
+        runId: 'run_01ABC',
+        stepId: 'step_01ABC',
+        stepName: 'test-step',
+      },
+      { idempotencyKey: 'step_01ABC' }
+    );
+
+    expect(workerUtilsMock.addJob).toHaveBeenCalledWith(
+      'billing_flows',
+      expect.anything(),
+      expect.objectContaining({
+        queueName: PREFIXED_IDEMPOTENCY_QUEUE,
+      })
+    );
+  });
+
+  it('assigns idempotency queues to migrated keyed jobs', async () => {
+    const migrationPool = {
+      query: vi
+        .fn()
+        .mockResolvedValueOnce({ rows: [{ exists: true }] })
+        .mockResolvedValueOnce({
+          rows: [
+            {
+              name: 'workflow_flows',
+              data: buildMessageData(
+                '__wkf_workflow_test-step',
+                {
+                  runId: 'run_01ABC',
+                  stepId: 'step_01ABC',
+                  stepName: 'test-step',
+                },
+                { idempotencyKey: 'step_01ABC' }
+              ),
+              singleton_key: 'step_01ABC',
+              retry_limit: 3,
+            },
+            {
+              name: 'workflow_flows',
+              data: buildMessageData('__wkf_workflow_test-workflow', {
+                runId: 'run_01ABD',
+              }),
+              singleton_key: 'msg_01ABC',
+              retry_limit: 3,
+            },
+          ],
+        })
+        .mockResolvedValueOnce({ rows: [] }),
+    } as unknown as Parameters<typeof createQueue>[1];
+    const queue = buildQueue(
+      { connectionString: 'postgres://test' },
+      migrationPool
+    );
+
+    await queue.start();
+
+    expect(workerUtilsMock.addJob).toHaveBeenCalledWith(
+      'workflow_flows',
+      expect.objectContaining({
+        id: 'test-step',
+        idempotencyKey: 'step_01ABC',
+      }),
+      {
+        jobKey: 'step_01ABC',
+        maxAttempts: 49,
+        queueName: DEFAULT_IDEMPOTENCY_QUEUE,
+      }
+    );
+    expect(workerUtilsMock.addJob).toHaveBeenCalledWith(
+      'workflow_flows',
+      expect.objectContaining({
+        id: 'test-workflow',
+        idempotencyKey: undefined,
+      }),
+      expect.not.objectContaining({ queueName: expect.anything() })
+    );
+  });
+
   it('queues namespaced producer messages in graphile job metadata', async () => {
     const queue = buildQueue(
       { connectionString: 'postgres://test', namespace: 'custom' },
@@ -526,6 +678,7 @@ describe('postgres queue http execution', () => {
       expect.objectContaining({
         jobKey: 'step_01ABC',
         maxAttempts: 49,
+        queueName: DEFAULT_IDEMPOTENCY_QUEUE,
       })
     );
   });

packages/world-postgres/src/queue.ts

@@ -1,3 +1,4 @@
+import { createHash } from 'node:crypto';
 import { connect } from 'node:net';
 import * as Stream from 'node:stream';
 import { setTimeout as sleep } from 'node:timers/promises';
@@ -53,6 +54,11 @@ function createGraphileLogger() {
 
 const graphileLogger = createGraphileLogger();
 const COMPLETED_IDEMPOTENCY_CACHE_LIMIT = 10_000;
+// Graphile retains named queues after their jobs complete. Keep this bounded
+// at 2,048 idempotency queues per configured job prefix.
+// Keep this mapping stable across releases; changing it requires a drained
+// rollout.
+const IDEMPOTENCY_QUEUE_BUCKET_COUNT = 2_048;
 // Core records MAX_DELIVERIES_EXCEEDED on delivery 49.
 const MAX_GRAPHILE_JOB_ATTEMPTS = 49;
 const GraphileHelpers = z.object({
@@ -74,6 +80,12 @@ type HttpExecutionResult =
 
 type RunnerStart = { controller: AbortController; promise: Promise<void> };
 type LoopbackTarget = { hosts: string[]; port: number };
+type MigratedPgBossJob = {
+  name: string;
+  data: unknown;
+  singleton_key: string | null;
+  retry_limit: number | null;
+};
 
 /**
  * The Postgres queue stores messages under one graphile-worker flow task.
@@ -130,6 +142,42 @@ export function createQueue(
     return `${jobPrefix}flows`;
   }
 
+  const idempotencyQueueScope = createHash('sha256')
+    .update(getJobQueueName())
+    .digest('hex');
+
+  function getIdempotencyQueueName(idempotencyKey: string): string {
+    const bucket = (
+      createHash('sha256').update(idempotencyKey).digest().readUInt16BE(0) %
+      IDEMPOTENCY_QUEUE_BUCKET_COUNT
+    )
+      .toString(16)
+      .padStart(3, '0');
+    return `workflow_idempotency_${idempotencyQueueScope}_${bucket}`;
+  }
+
+  function getMigratedJobIdempotencyKey(data: unknown): string | undefined {
+    const messageData = MessageData.safeParse(data);
+    return messageData.success ? messageData.data.idempotencyKey : undefined;
+  }
+
+  async function addMigratedPgBossJobs(
+    utils: WorkerUtils,
+    jobs: MigratedPgBossJob[]
+  ): Promise<void> {
+    for (const job of jobs) {
+      const jobKey = job.singleton_key ?? undefined;
+      const idempotencyKey = getMigratedJobIdempotencyKey(job.data);
+      await utils.addJob(job.name, job.data as Record<string, unknown>, {
+        jobKey,
+        ...(idempotencyKey
+          ? { queueName: getIdempotencyQueueName(idempotencyKey) }
+          : {}),
+        maxAttempts: Math.max(job.retry_limit ?? 0, MAX_GRAPHILE_JOB_ATTEMPTS),
+      });
+    }
+  }
+
   const createQueueHandler = localWorld.createQueueHandler;
 
   const getDeploymentId: Queue['getDeploymentId'] = async () => {
@@ -200,6 +248,9 @@ export function createQueue(
       }),
       {
         ...(jobKey ? { jobKey } : {}),
+        ...(idempotencyKey
+          ? { queueName: getIdempotencyQueueName(idempotencyKey) }
+          : {}),
         ...(runAt ? { runAt } : {}),
         maxAttempts: MAX_GRAPHILE_JOB_ATTEMPTS,
       }
@@ -390,19 +441,11 @@ export function createQueue(
       ) AS exists`
     );
     if (hasStaging.rows[0]?.exists) {
-      const jobs = await pool.query(
+      const jobs = await pool.query<MigratedPgBossJob>(
         `SELECT name, data, singleton_key, retry_limit
         FROM "workflow"."_pgboss_pending_jobs"`
       );
-      for (const job of jobs.rows) {
-        await utils.addJob(job.name, job.data as Record<string, unknown>, {
-          jobKey: job.singleton_key ?? undefined,
-          maxAttempts: Math.max(
-            job.retry_limit ?? 0,
-            MAX_GRAPHILE_JOB_ATTEMPTS
-          ),
-        });
-      }
+      await addMigratedPgBossJobs(utils, jobs.rows);
       await pool.query(`DROP TABLE "workflow"."_pgboss_pending_jobs"`);
       return;
     }
@@ -415,20 +458,12 @@ export function createQueue(
       ) AS exists`
     );
     if (hasPgBoss.rows[0]?.exists) {
-      const jobs = await pool.query(
+      const jobs = await pool.query<MigratedPgBossJob>(
         `SELECT name, data, singleton_key, retry_limit
         FROM pgboss.job
         WHERE state IN ('created', 'retry')`
       );
-      for (const job of jobs.rows) {
-        await utils.addJob(job.name, job.data as Record<string, unknown>, {
-          jobKey: job.singleton_key ?? undefined,
-          maxAttempts: Math.max(
-            job.retry_limit ?? 0,
-            MAX_GRAPHILE_JOB_ATTEMPTS
-          ),
-        });
-      }
+      await addMigratedPgBossJobs(utils, jobs.rows);
       await pool.query(`DROP SCHEMA pgboss CASCADE`);
     }
   }