Files
Nathan Rajlich b4ba79ebc5 feat: publish each run's X25519 public key on the run entity (#3095)
* feat: publish each run's X25519 public key on the run entity

A cross-run writer needs the recipient run's public key to seal a payload
to it. Derive that key at `start()` and stamp it on the run, so a hook
resumption or a forwarded-stream writer can find it on a run fetch it was
already making instead of spending ~350ms on `run-key`.

The key is derived from the per-run key material `getEncryptionKeyForRun()`
already returns, so nothing about key acquisition changes. It is not
secret: the matching private scalar is never stored anywhere, only
re-derived on demand from the deployment's own env seed. Storing it beside
run metadata therefore does not weaken the run's confidentiality.

**Presence is the writer-side gate for sealed envelopes.** A run only
carries a public key if the runtime that created it could also open one —
which holds by construction, since derivation and `encp` dispatch both
live in `@workflow/core`, so any core that can stamp can also open. Runs
are pinned to their creating deployment, so the capability this attests to
is still accurate at resume time. Writers seal iff the field is set and
otherwise fall back to the symmetric path, which makes version skew
degrade gracefully instead of wedging a run.

The field rides on `run_created`, and is mirrored onto the queued
`runInput` so the resilient-start path (server recreates the run from the
queue message when the `run_created` write failed) doesn't silently
produce a run that can't receive sealed writes.

world-vercel's compile-time wire-contract guard caught the new field
before it could be silently dropped on the v4 path, exactly as designed —
routed into the frame meta block as plaintext metadata.

Also adds browser- and VM-safe base64 helpers to `sealed-box.ts`, since
neither `Buffer` nor `btoa` can be assumed in every context that module
runs in. `base64ToBytes` returns undefined on malformed input rather than
throwing, so a corrupt stored key degrades to "no usable public key" and
falls back to the symmetric path instead of crashing a resumption. Both
are cross-validated against `Buffer` in tests.

* review: fix public-key loss on resilient start and lifecycle updates

Two real bugs found in review, both in the local worlds. Neither surfaces
as an error — a run just silently stops accepting sealed cross-run writes
and falls back to the slow symmetric path forever.

**Resilient start dropped the key.** When a `run_started` arrives for a
run that was never created, world-local and world-postgres rebuild the run
from the queued message. Neither copied `encryptionPublicKey` onto the run
row or the synthetic `run_created` event they write. That is precisely the
scenario this field exists to survive. (The equivalent server-side path was
already handled.)

**world-local also wiped the key on every lifecycle transition.** Its
run_started / run_completed / run_failed / run_cancelled handlers rewrite
the whole run document field-by-field, so any field not explicitly listed
is dropped — meaning the key was lost on the *first* `run_started`, not
just on the resilient path. All four rebuild sites now carry it.
world-postgres is safe here by construction because it issues
column-scoped SQL UPDATEs rather than rewriting the row.

**base64 decoding is now strict.** The decoder accepted shapes that
cannot describe a whole number of bytes (`length % 4 === 1`) and ignored
anything after a mid-string `=`, returning a short array instead of
`undefined`. That is worse than throwing: a corrupt stored key looked
*present*, so callers sealed to garbage rather than taking the symmetric
fallback. Now rejects out-of-alphabet characters, bad lengths, misplaced
padding, and non-zero trailing bits — with a round-trip test over every
length 0–48 to make sure the strictness does not overshoot.

* fix: send encryptionPublicKey in the v4 POST frame meta

`splitEventDataForV4` lifted the run's public key into the frame meta and
`events.ts` spread that meta into `CreateEventV4Input`, but
`buildPostFrameMeta` — which copies meta onto the wire field by field — never
forwarded `encryptionPublicKey`, and the field was missing from
`CreateEventV4Input` entirely. Because the meta is applied with a spread,
TypeScript's excess-property check doesn't fire, so the key was computed, put
in the meta, and then silently dropped before the request was sent.

The server therefore never received the key, never stored it on the run
entity, and every cross-run writer fell back to the symmetric envelope. Every
symptom pointed away from the SDK: a deliberately oversized key was accepted
rather than rejected (the field never arrived), the key was absent from the run
row, and `resumeHook()` always chose `encr`.

Add the field to `CreateEventV4Input`, forward it in `buildPostFrameMeta`, and
cover it for both `run_created` and resilient-start `run_started`. Also add a
generic guard asserting that every field the splitter puts in the meta reaches
the wire, so the next omission in this hand-maintained mapping fails a test
instead of silently degrading encryption.
2026-07-27 14:50:42 -07:00
..
2025-10-23 12:07:52 +03:00
2026-07-22 09:56:48 -07:00
2025-10-23 12:07:52 +03:00
2026-07-22 09:56:48 -07:00
2025-10-23 12:07:52 +03:00

@workflow/world-postgres

An embedded worker/workflow system backed by PostgreSQL for multi-host self-hosted solutions. This is a reference implementation - a production-ready solution might run workers in separate processes with a more robust queuing system.

Installation

npm install @workflow/world-postgres
# or
pnpm add @workflow/world-postgres
# or
yarn add @workflow/world-postgres

Usage

Basic Setup

The postgres world can be configured by setting the WORKFLOW_TARGET_WORLD environment variable to the package name:

export WORKFLOW_TARGET_WORLD="@workflow/world-postgres"

Configuration

Configure the PostgreSQL world using environment variables:

# Required: PostgreSQL connection string
export WORKFLOW_POSTGRES_URL="postgres://username:password@localhost:5432/database"

# Optional: Job prefix for queue operations
export WORKFLOW_POSTGRES_JOB_PREFIX="myapp"

# Optional: Worker concurrency (default: 10)
export WORKFLOW_POSTGRES_WORKER_CONCURRENCY="10"

# Optional: Internal pg.Pool max size (default: 10)
export WORKFLOW_POSTGRES_MAX_POOL_SIZE="10"

# Optional: Let the application coordinate shutdown (default: false)
export WORKFLOW_POSTGRES_APPLICATION_MANAGED_SHUTDOWN="1"

Programmatic Usage

You can also create a PostgreSQL world directly in your code:

import { createWorld } from "@workflow/world-postgres";

const world = createWorld({
  connectionString: "postgres://username:password@localhost:5432/database",
  jobPrefix: "myapp", // optional
  queueConcurrency: 50, // optional
  maxPoolSize: 10, // optional, overrides WORKFLOW_POSTGRES_MAX_POOL_SIZE when `pool` is omitted
});

// Or pass an existing pg.Pool (shared with your app Drizzle, etc.); `world.close()` will not end it.
import { Pool } from "pg";
const pool = new Pool({ connectionString: process.env.DATABASE_URL });
const worldFromPool = createWorld({ pool });

Application-managed Shutdown

By default, Graphile Worker responds automatically when the application is asked to shut down. If your application already coordinates shutdown, set WORKFLOW_POSTGRES_APPLICATION_MANAGED_SHUTDOWN=1 when selecting the package with WORKFLOW_TARGET_WORLD, or set applicationManagedShutdown: true when calling createWorld() directly. Await world.close() from your shutdown path so Graphile Worker cannot terminate the process as soon as its queue stops, before your application finishes closing dependent resources:

import { createWorld } from '@workflow/world-postgres';

const world = createWorld({
  connectionString: process.env.DATABASE_URL!,
  applicationManagedShutdown: true,
});

await world.start();

Use this option only when your application or framework has its own shutdown hook. Handle cleanup errors there and await world.close() first, then close the workflow HTTP server and any caller-owned pg.Pool.

Closing the world stops the queue from accepting new jobs and waits for active jobs. After Graphile Worker's graceful-shutdown timeout (5 seconds by default), it aborts any workflow HTTP request that is still pending. Graphile Worker then unlocks the same row through its normal failure handling. Graphile counts a delivery attempt when it claims the row, so the aborted delivery consumes that attempt and is retried only if its Graphile attempt budget remains. A one-attempt or final-attempt job is unlocked but not retried. The shutdown handler does not create a replacement row.

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.

Configuration Options

Option Type Default Description
connectionString string process.env.WORKFLOW_POSTGRES_URL, process.env.DATABASE_URL, or 'postgres://world:world@localhost:5432/world' Used only when pool is omitted, to construct an internal pool
maxPoolSize number process.env.WORKFLOW_POSTGRES_MAX_POOL_SIZE or pg.Pool default (10) Optional. Sets the internal pg.Pool max size when createWorld() creates the pool
pool pg.Pool — 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 — each Run#returnValue await holds a worker slot until the child run terminates.
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

Variable Description Default
WORKFLOW_TARGET_WORLD Set to "@workflow/world-postgres" to use this world -
WORKFLOW_POSTGRES_URL PostgreSQL connection string DATABASE_URL or 'postgres://world:world@localhost:5432/world'
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_APPLICATION_MANAGED_SHUTDOWN Set to 1 when the application coordinates shutdown and awaits world.close() unset (false)

When pool is omitted, maxPoolSize precedence is: createWorld({ maxPoolSize }), then WORKFLOW_POSTGRES_MAX_POOL_SIZE, then the pg.Pool default.

For higher worker concurrency, Graphile Worker recommends setting maxPoolSize to 10 or queueConcurrency + 2, whichever is larger.

Database Setup

This package uses PostgreSQL with the following components:

  • graphile-worker: For queue processing and job management
  • Drizzle ORM: For database operations and schema management
  • pg (node-postgres): For PostgreSQL client connections. Drizzle and Graphile Worker share a pg.Pool, while LISTEN uses a dedicated pg.Client created from the same connection options.

Quick Setup with CLI

The easiest way to set up your database is using the included CLI tool:

# npm
npx --package=@workflow/world-postgres bootstrap

# pnpm
pnpm dlx --package @workflow/world-postgres bootstrap

# Yarn
yarn dlx --package @workflow/world-postgres bootstrap

# Bun
bunx --package @workflow/world-postgres bootstrap

The CLI and runtime World automatically load the connection string from:

  1. WORKFLOW_POSTGRES_URL environment variable
  2. DATABASE_URL environment variable
  3. Default: postgres://world:world@localhost:5432/world

Database Schema

The setup creates the following tables:

  • workflow_runs - Stores workflow execution runs
  • workflow_events - Stores workflow events
  • workflow_steps - Stores individual workflow steps
  • workflow_hooks - Stores webhook hooks
  • workflow_stream_chunks - Stores streaming data chunks

You can also access the schema programmatically:

import { runs, events, steps, hooks, streams } from '@workflow/world-postgres';
// or
import * as schema from '@workflow/world-postgres/schema';

Make sure your PostgreSQL database is accessible and the user has sufficient permissions to create tables and manage jobs.

Features

  • Durable Storage: Stores workflow runs, events, steps, hooks, and webhooks in PostgreSQL
  • Queue Processing: Uses graphile-worker as the durable queue and executes jobs over the workflow HTTP routes
  • Durable Delays: Re-schedules waits and retries in PostgreSQL
  • Streaming: Real-time event streaming capabilities
  • Health Checks: Built-in connection health monitoring
  • Configurable Concurrency: Adjustable worker concurrency for queue processing

Queue Behavior

  • 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
  • Workflow orchestration and queued step execution are both sent through /.well-known/workflow/v1/flow

Development

For local development, you can use the included Docker Compose configuration:

# Start PostgreSQL database
docker-compose up -d

# Create and run migrations
pnpm drizzle-kit generate
pnpm drizzle-kit migrate

# Set environment variables for local development
export WORKFLOW_POSTGRES_URL="postgres://world:world@localhost:5432/world"
export WORKFLOW_TARGET_WORLD="@workflow/world-postgres"

Testing

Integration tests use Testcontainers to start a PostgreSQL container. Docker must be installed and running before you run tests.

  • Linux/macOS: Start the Docker daemon (e.g. sudo systemctl start docker or Docker Desktop).
  • WSL2: Use Docker Desktop with WSL2 integration, or run the Docker engine inside WSL and ensure the daemon is started. Verify with docker info.

Then from the package directory:

pnpm build
pnpm test

World Selection

To use the PostgreSQL world, set the WORKFLOW_TARGET_WORLD environment variable to the package name:

export WORKFLOW_TARGET_WORLD="@workflow/world-postgres"