diff --git a/.changeset/postgres-producer-role.md b/.changeset/postgres-producer-role.md new file mode 100644 index 0000000000..794874f00c --- /dev/null +++ b/.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. diff --git a/docs/content/docs/v5/configuration/worlds.mdx b/docs/content/docs/v5/configuration/worlds.mdx index ed49ee51f9..33f9bcbfc4 100644 --- a/docs/content/docs/v5/configuration/worlds.mdx +++ b/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) diff --git a/docs/content/worlds/v5/postgres.mdx b/docs/content/worlds/v5/postgres.mdx index 2d6160163e..50a89aa0ae 100644 --- a/docs/content/worlds/v5/postgres.mdx +++ b/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`). diff --git a/packages/world-postgres/README.md b/packages/world-postgres/README.md index a782e135ab..b5c2e42cf7 100644 --- a/packages/world-postgres/README.md +++ b/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` | diff --git a/packages/world-postgres/src/config.ts b/packages/world-postgres/src/config.ts index 057a11f4fb..3c485c1e92 100644 --- a/packages/world-postgres/src/config.ts +++ b/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; + 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(). diff --git a/packages/world-postgres/src/index.ts b/packages/world-postgres/src/index.ts index 336daad58e..e586f7ae1a 100644 --- a/packages/world-postgres/src/index.ts +++ b/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'; diff --git a/packages/world-postgres/src/queue.test.ts b/packages/world-postgres/src/queue.test.ts index 84937fac17..0c47fdd377 100644 --- a/packages/world-postgres/src/queue.test.ts +++ b/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; + 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[0], - pgPool: Parameters[1] + pgPool: Parameters[1], + role: Parameters[2] = PostgresWorldRoleSchema.enum.worker ) { - const queue = createQueue(config, pgPool); + const queue = createQueue(config, pgPool, role); createdQueues.push(queue); return queue; } diff --git a/packages/world-postgres/src/queue.ts b/packages/world-postgres/src/queue.ts index 38655120e4..6eec72010d 100644 --- a/packages/world-postgres/src/queue.ts +++ b/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,7 +490,9 @@ export function createQueue( }); await workerUtils.migrate(); await migratePgBossJobs(workerUtils); - await startRunnerWhenExecutorIsReady(); + if (!isProducer) { + await startRunnerWhenExecutorIsReady(); + } } catch (err) { startPromise = null; throw err; @@ -488,7 +500,7 @@ export function createQueue( })(); } await startPromise; - if (!closing && !runner && !runnerStart) { + if (!isProducer && !closing && !runner && !runnerStart) { await startRunnerWhenExecutorIsReady(); } } diff --git a/packages/world-postgres/src/reenqueue.test.ts b/packages/world-postgres/src/reenqueue.test.ts index 0ded6200dc..0a8b87a8a7 100644 --- a/packages/world-postgres/src/reenqueue.test.ts +++ b/packages/world-postgres/src/reenqueue.test.ts @@ -3,6 +3,7 @@ import { createWorld as createLocalTestWorld } from '@workflow/world-local'; import { makeWorkerUtils, run, type WorkerUtils } from 'graphile-worker'; import { Pool } from 'pg'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { PostgresWorldRoleSchema } from './config.js'; import { createWorld } from './index.js'; import { createEventsStorage, @@ -122,6 +123,7 @@ describe('re-enqueue active runs on start', () => { afterEach(async () => { delete process.env.WORKFLOW_LOCAL_BASE_URL; delete process.env.WORKFLOW_POSTGRES_APPLICATION_MANAGED_SHUTDOWN; + delete process.env.WORKFLOW_POSTGRES_ROLE; delete process.env.WORKFLOW_POSTGRES_URL; delete process.env.DATABASE_URL; delete process.env.PORT; @@ -359,6 +361,117 @@ describe('re-enqueue active runs on start', () => { expect(internalPool?.end).toHaveBeenCalledOnce(); }); + it('starts a graphile runner and recovers active runs in the default role', async () => { + mockRunsList({ + pending: [{ runId: 'wrun_AAA', workflowName: 'wfA' }], + }); + + const world = createWorld({ connectionString: 'postgres://test', pool }); + await world.start(); + + expect(run).toHaveBeenCalledOnce(); + expect(workerUtilsMock.addJob).toHaveBeenCalledOnce(); + + await world.close(); + }); + + it('does not start a graphile runner in the producer role', async () => { + const world = createWorld({ + connectionString: 'postgres://test', + pool, + role: PostgresWorldRoleSchema.enum.producer, + }); + await world.start(); + + expect(run).not.toHaveBeenCalled(); + expect(workerUtilsMock.migrate).toHaveBeenCalledOnce(); + + await world.close(); + }); + + it('does not re-enqueue active runs in the producer role', async () => { + mockRunsList({ + pending: [{ runId: 'wrun_AAA', workflowName: 'wfA' }], + running: [{ runId: 'wrun_BBB', workflowName: 'wfB' }], + }); + + const world = createWorld({ + connectionString: 'postgres://test', + pool, + role: PostgresWorldRoleSchema.enum.producer, + }); + await world.start(); + + expect(workerUtilsMock.addJob).not.toHaveBeenCalled(); + + await world.close(); + }); + + it('still enqueues messages in the producer role', async () => { + const world = createWorld({ + connectionString: 'postgres://test', + pool, + role: PostgresWorldRoleSchema.enum.producer, + }); + await world.start(); + + await world.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(run).not.toHaveBeenCalled(); + + await world.close(); + }); + + it('reads the producer role from WORKFLOW_POSTGRES_ROLE', async () => { + process.env.WORKFLOW_POSTGRES_ROLE = 'producer'; + mockRunsList({ + running: [{ runId: 'wrun_BBB', workflowName: 'wfB' }], + }); + + const world = createWorld({ connectionString: 'postgres://test', pool }); + await world.start(); + + expect(run).not.toHaveBeenCalled(); + expect(workerUtilsMock.addJob).not.toHaveBeenCalled(); + + await world.close(); + }); + + it('ignores an unrecognized WORKFLOW_POSTGRES_ROLE', async () => { + process.env.WORKFLOW_POSTGRES_ROLE = 'producers'; + + const world = createWorld({ connectionString: 'postgres://test', pool }); + await world.start(); + + expect(run).toHaveBeenCalledOnce(); + + await world.close(); + }); + + it('closes only what a producer opened', async () => { + const world = createWorld({ + connectionString: 'postgres://test', + pool, + role: PostgresWorldRoleSchema.enum.producer, + }); + await world.start(); + + await expect(world.close()).resolves.toBeUndefined(); + + expect(runnerMock.stop).not.toHaveBeenCalled(); + expect(workerUtilsMock.release).toHaveBeenCalledOnce(); + expect(localWorldClose).toHaveBeenCalledOnce(); + }); + it('does not close a caller-owned pool', async () => { const world = createWorld({ pool }); await world.start();