[world-postgres] Add a producer role that enqueues without a runner or startup recovery

.changeset/postgres-producer-role.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-postgres': minor
+---
+
+Add a `role` option (`'producer' | 'worker'`, default `'worker'`, env fallback `WORKFLOW_POSTGRES_ROLE`) so a process that only enqueues can skip the Graphile Worker runner and startup recovery. A `'producer'` World still ensures the schema on `start()`, so it can enqueue into a fresh database, but never claims jobs it cannot execute and never re-enqueues runs another process owns.

docs/content/docs/v5/configuration/worlds.mdx

@@ -158,6 +158,14 @@ The Postgres World is a self-hosted durable backend for long-running server proc
 - Number of concurrent workers polling for jobs.
 - Also bounds concurrent parent-to-child workflow return-value polls.
 
+### `role`
+
+- Environment variable: `WORKFLOW_POSTGRES_ROLE` (`producer` selects it)
+- Default: `worker`
+- Whether this process both enqueues and executes messages (`worker`) or only enqueues them (`producer`).
+- A `producer` still ensures the schema on `start()`, so it can enqueue into a fresh database, and `queue()` is unchanged. It starts no Graphile Worker runner and runs no startup recovery.
+- Use it for a deployment unit that submits work it cannot execute, such as an API server that did not compile the workflow and step code. Otherwise its runner claims jobs only to fail their HTTP delivery, and every replica re-enqueues every active run on startup.
+
 ### `applicationManagedShutdown`
 
 - Environment variable: `WORKFLOW_POSTGRES_APPLICATION_MANAGED_SHUTDOWN` (`1` enables)

docs/content/worlds/v5/postgres.mdx

@@ -220,6 +220,12 @@ Maximum size of the internal `pg.Pool` used when `createWorld()` constructs the
 
 For higher worker concurrency, Graphile Worker recommends setting `maxPoolSize` to `10` or `queueConcurrency + 2`, whichever is larger.
 
+### `WORKFLOW_POSTGRES_ROLE`
+
+Set to `producer` for a process that only enqueues work. Default: `worker`, which both enqueues and executes.
+
+A producer still ensures the schema on `start()`, so it can enqueue into a fresh database, and enqueueing is unchanged. It starts no graphile-worker runner and re-enqueues no active runs. Use it for a deployment unit that submits work it cannot execute — one that did not compile the `"use workflow"` and `"use step"` code, and so does not serve `.well-known/workflow/v1/*`. Otherwise its runner claims jobs only to fail their HTTP delivery, and every replica re-enqueues every active run on startup, including runs a live peer is executing. Set `role: 'producer'` instead in a programmatic World.
+
 ### `WORKFLOW_POSTGRES_APPLICATION_MANAGED_SHUTDOWN`
 
 Set to `1` when the application or framework coordinates shutdown and awaits `world.close()` before closing its workflow HTTP server and any caller-owned pool. Default: unset (`false`).

packages/world-postgres/README.md

@@ -88,6 +88,23 @@ Closing the world stops the queue from accepting new jobs and waits for active j
 
 An aborted HTTP request does not guarantee that its server-side handler stopped, so workflow and step handlers must continue to tolerate at-least-once execution. Keep the workflow HTTP routes and any caller-owned pool available until `world.close()` resolves.
 
+### Producer-only processes
+
+A deployment unit that submits work but cannot execute it — one that did not compile the `"use workflow"` / `"use step"` code, and so does not serve `.well-known/workflow/v1/*` — should be a producer:
+
+```typescript
+import { createWorld } from '@workflow/world-postgres';
+
+const world = createWorld({
+  connectionString: process.env.DATABASE_URL!,
+  role: 'producer',
+});
+
+await world.start();
+```
+
+`start()` still ensures the schema, so a producer can enqueue into a fresh database, and `queue()` behaves exactly as it does for a worker. What a producer never does is call Graphile Worker's `run()` or `reenqueueActiveRuns()`. Both are unwanted there: a runner in a process with no local executor claims jobs only to fail their HTTP delivery, and startup recovery from every replica of a frequently-rolled process re-enqueues every active run, including runs a live peer is executing. Set `WORKFLOW_POSTGRES_ROLE=producer` instead when the package is selected through `WORKFLOW_TARGET_WORLD` rather than constructed in code.
+
 ## Configuration options
 
 | Option             | Type      | Default                                                                                | Description                                                                                          |
@@ -97,6 +114,7 @@ An aborted HTTP request does not guarantee that its server-side handler stopped,
 | `pool`             | `pg.Pool` | Not applicable                                                                         | Optional. When set, used for Drizzle, Graphile Worker, and stream writes. `world.close()` does not end it. |
 | `jobPrefix`        | `string`  | `process.env.WORKFLOW_POSTGRES_JOB_PREFIX`                                             | Optional prefix for queue job names                                                                  |
 | `queueConcurrency` | `number`  | `50`                                                                                   | Number of concurrent active step executions per process. Must be high enough to cover any parent→child workflow polling in flight because each `Run#returnValue` await holds a worker slot until the child run terminates. |
+| `role`             | `'producer' \| 'worker'` | `'worker'`; `WORKFLOW_POSTGRES_ROLE=producer` selects `'producer'`                | Whether this process both enqueues and executes messages (`'worker'`) or only enqueues them (`'producer'`). A producer ensures the schema on `start()` but starts no Graphile Worker runner and runs no startup recovery. |
 | `applicationManagedShutdown` | `boolean` | `false`; `WORKFLOW_POSTGRES_APPLICATION_MANAGED_SHUTDOWN=1` enables it for the default package configuration | Whether the application coordinates shutdown and awaits `world.close()` instead of Graphile Worker responding automatically. |
 
 ## Environment variables
@@ -108,6 +126,7 @@ An aborted HTTP request does not guarantee that its server-side handler stopped,
 | `WORKFLOW_POSTGRES_JOB_PREFIX`         | Prefix for queue job names                                   | -                                               |
 | `WORKFLOW_POSTGRES_WORKER_CONCURRENCY` | Number of concurrent workers                                 | `50`                                            |
 | `WORKFLOW_POSTGRES_MAX_POOL_SIZE`      | Internal `pg.Pool` max size                                  | `10`                                            |
+| `WORKFLOW_POSTGRES_ROLE`               | Set to `producer` for a process that only enqueues              | `worker`                                        |
 | `WORKFLOW_POSTGRES_APPLICATION_MANAGED_SHUTDOWN` | Set to `1` when the application coordinates shutdown and awaits `world.close()` | unset (`false`) |
 | `WORKFLOW_POSTGRES_HOOK_RETENTION_LIMIT_DAYS` | Maximum Hook minimum retention in days | `30` |
 

packages/world-postgres/src/config.ts

@@ -1,9 +1,14 @@
 import type { Pool } from 'pg';
+import { z } from 'zod/v4';
 
 type PgConnectionConfig =
   | { connectionString: string; maxPoolSize?: number; pool?: undefined }
   | { pool: Pool; connectionString?: undefined; maxPoolSize?: undefined };
 
+export const PostgresWorldRoleSchema = z.enum(['producer', 'worker']);
+
+export type PostgresWorldRole = z.infer<typeof PostgresWorldRoleSchema>;
+
 export type PostgresWorldConfig = PgConnectionConfig & {
   jobPrefix?: string;
   /**
@@ -12,6 +17,18 @@ export type PostgresWorldConfig = PgConnectionConfig & {
    */
   namespace?: string;
   queueConcurrency?: number;
+  /**
+   * Whether this process both enqueues and executes messages (`worker`, the
+   * default) or only enqueues them (`producer`). A producer's start() still
+   * ensures the schema, so it can enqueue into a fresh database, but it never
+   * starts a Graphile Worker runner and never re-enqueues active runs. Use it
+   * for a deployment unit that submits work it cannot execute — one that did
+   * not compile the workflow and step code, and so does not serve
+   * `.well-known/workflow/v1/*` — so it neither claims jobs it can only fail
+   * nor replays runs another process owns. The `WORKFLOW_POSTGRES_ROLE`
+   * environment variable is used as a fallback when this option is unset.
+   */
+  role?: PostgresWorldRole;
   /**
    * Whether the application coordinates shutdown instead of Graphile Worker
    * responding automatically. The application must await world.close().

packages/world-postgres/src/index.ts

@@ -1,7 +1,11 @@
 import type { Storage, World } from '@workflow/world';
 import { mintedSpecVersion, reenqueueActiveRuns } from '@workflow/world';
 import { Pool } from 'pg';
-import type { PostgresWorldConfig } from './config.js';
+import {
+  type PostgresWorldConfig,
+  type PostgresWorldRole,
+  PostgresWorldRoleSchema,
+} from './config.js';
 import { createClient, type Drizzle } from './drizzle/index.js';
 import { createQueue } from './queue.js';
 import {
@@ -64,8 +68,19 @@ export function createWorld(
       ...(maxPoolSize !== undefined ? { max: maxPoolSize } : {}),
     });
 
+  // Role resolution order: the explicit option, then WORKFLOW_POSTGRES_ROLE,
+  // then the worker default that every release before the option existed had.
+  // An unrecognized environment value falls back to that default: the variable
+  // is an escape hatch, not a hard requirement.
+  const envRole = PostgresWorldRoleSchema.safeParse(
+    process.env.WORKFLOW_POSTGRES_ROLE?.toLowerCase()
+  );
+  const role: PostgresWorldRole =
+    config.role ??
+    (envRole.success ? envRole.data : PostgresWorldRoleSchema.enum.worker);
+
   const drizzle = createClient(pool);
-  const queue = createQueue(config, pool);
+  const queue = createQueue(config, pool, role);
   // Opens its `LISTEN` connection lazily, on the first `waitForTerminalStatus`
   // call, so a deployment that never awaits a run never pays for it.
   const runStatusListener = createRunStatusListener(pool);
@@ -85,6 +100,12 @@ export function createWorld(
     }),
     async start() {
       await queue.start();
+      // A producer executes nothing, so recovering a run here would only hand
+      // it to whichever process claims it next — including a peer that is
+      // already replaying it.
+      if (role === PostgresWorldRoleSchema.enum.producer) {
+        return;
+      }
       await reenqueueActiveRuns(
         storage.runs,
         queue.queue,
@@ -104,5 +125,6 @@ export function createWorld(
 }
 
 // Re-export schema for users who want to extend or inspect the database schema
-export type { PostgresWorldConfig } from './config.js';
+export { PostgresWorldRoleSchema } from './config.js';
+export type { PostgresWorldConfig, PostgresWorldRole } from './config.js';
 export * from './drizzle/schema.js';

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

@@ -11,6 +11,7 @@ import {
   type WorkerUtils,
 } from 'graphile-worker';
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { PostgresWorldRoleSchema } from './config.js';
 import { MessageData } from './message.js';
 import { createQueue } from './queue.js';
 
@@ -529,13 +530,49 @@ describe('postgres queue http execution', () => {
       })
     );
   });
+
+  it('never starts a runner in the producer role, even when a local executor is reachable', async () => {
+    const requests: Array<{
+      method: string | undefined;
+      url: string | undefined;
+      headers: Record<string, string | string[] | undefined>;
+      body: string;
+    }> = [];
+    const port = await getUnusedLoopbackPort();
+    await startWorkflowHttpServer(requests, port);
+    process.env.PORT = String(port);
+
+    const queue = buildQueue(
+      { connectionString: 'postgres://test' },
+      pool,
+      PostgresWorldRoleSchema.enum.producer
+    );
+    await queue.start();
+
+    expect(run).not.toHaveBeenCalled();
+    expect(workerUtilsMock.migrate).toHaveBeenCalledOnce();
+
+    await queue.queue('__wkf_workflow_test-step', {
+      runId: 'run_01ABC',
+      stepId: 'step_01ABC',
+      stepName: 'test-step',
+    });
+
+    expect(workerUtilsMock.addJob).toHaveBeenCalledWith(
+      'workflow_flows',
+      expect.objectContaining({ id: 'test-step' }),
+      expect.anything()
+    );
+    expect(requests).toEqual([]);
+  });
 });
 
 function buildQueue(
   config: Parameters<typeof createQueue>[0],
-  pgPool: Parameters<typeof createQueue>[1]
+  pgPool: Parameters<typeof createQueue>[1],
+  role: Parameters<typeof createQueue>[2] = PostgresWorldRoleSchema.enum.worker
 ) {
-  const queue = createQueue(config, pgPool);
+  const queue = createQueue(config, pgPool, role);
   createdQueues.push(queue);
   return queue;
 }

packages/world-postgres/src/queue.ts

@@ -30,7 +30,11 @@ import {
 import type { Pool } from 'pg';
 import { monotonicFactory } from 'ulid';
 import { z } from 'zod/v4';
-import type { PostgresWorldConfig } from './config.js';
+import {
+  type PostgresWorldConfig,
+  type PostgresWorldRole,
+  PostgresWorldRoleSchema,
+} from './config.js';
 import { MessageData } from './message.js';
 
 function createGraphileLogger() {
@@ -85,7 +89,8 @@ export type PostgresQueue = Queue & {
 
 export function createQueue(
   config: PostgresWorldConfig,
-  pool: Pool
+  pool: Pool,
+  role: PostgresWorldRole
 ): PostgresQueue {
   const port = process.env.PORT ? Number(process.env.PORT) : undefined;
   const localWorld = createWorld({ dataDir: undefined, port });
@@ -471,6 +476,11 @@ export function createQueue(
       return;
     }
 
+    // A producer enqueues but never claims, so it still ensures the schema and
+    // then stops short of the runner. `queue()` awaits this same `start()`, so
+    // the check has to live here rather than at the call sites.
+    const isProducer = role === PostgresWorldRoleSchema.enum.producer;
+
     if (!startPromise) {
       startPromise = (async () => {
         try {
@@ -480,15 +490,17 @@ export function createQueue(
           });
           await workerUtils.migrate();
           await migratePgBossJobs(workerUtils);
-          await startRunnerWhenExecutorIsReady();
+          if (!isProducer) {
+            await startRunnerWhenExecutorIsReady();
+          }
         } catch (err) {
           startPromise = null;
           throw err;
         }
       })();
     }
     await startPromise;
-    if (!closing && !runner && !runnerStart) {
+    if (!isProducer && !closing && !runner && !runnerStart) {
       await startRunnerWhenExecutorIsReady();
     }
   }