Files
Pranay Prakash e7ef9d823b perf(core): lazy inline step start (save one world round-trip per step) (#2478)
* perf(core): lazy inline step start to save a world round-trip per step

The owned-inline runtime path used to write step_created (suspension
handler) and then step_started (executeStep) as two separate world
round-trips for a step it already owns and is about to run inline. This
defers the step_created write: executeStep sends a single step_started
carrying the step input, and the world creates the step on the fly
(materializing the step entity plus a synthetic step_created event so
replay still observes it). Mirrors the existing resilient run_started ->
run_created pattern.

Exactly-one ownership is preserved by the world's atomic create-claim:
the loser of a concurrent lazy step_started gets EntityConflictError,
which executeStep maps to `skipped`, so it never runs the body. A lazy
step_started is only ever sent for a brand-new step (the suspension
handler defers only steps with no prior step_created), so crash recovery
still re-runs a `running` step via the normal non-lazy step_started.

Worlds updated: world-local, world-postgres (implicit create + synthetic
step_created event), world-vercel (routes the input as the v4 frame
payload and threads the server's stepCreated flag). @workflow/world adds
optional `input` to step_started and a `stepCreated` EventResult signal.

Rollout: server-first. The matching workflow-server change must deploy
before this ships; the Vercel world targets a single Vercel-operated
backend (server always >= SDK). For local/postgres the world ships in the
same package as the runtime, so there is no version skew.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(core): materialize deferred step before failing unregistered step on lazy inline path

The lazy inline step-start optimization defers a step's step_created write,
expecting executeStep to materialize the step via a lazy step_started carrying
its input. For an UNREGISTERED step, executeStep bails out before sending that
step_started and writes step_failed directly — but the step entity was never
created, so the world's "step must exist" ordering guard rejects the
step_failed and the run wedges (times out).

This regressed the StepNotRegisteredError e2e tests uniformly across every
framework/world (the ghost step never reached `failed`). Fix: on the lazy path,
send the lazy step_started first to materialize the step (entity + synthetic
step_created, keeping replay correct), then write step_failed. The lazy
step_started's atomic create-claim preserves exactly-one-owner: a concurrent
winner makes ours reject with EntityConflictError → skipped, so the failure is
never written twice.

Adds world-level regression tests (world-local, world-postgres) asserting a
lazy step_started followed by step_failed marks the step failed.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-18 05:51:35 +00:00
..
2025-10-23 12:07:52 +03:00
2026-06-17 17:06:19 -07:00
2025-10-23 12:07:52 +03:00
2026-06-17 17:06:19 -07:00
2025-10-23 12:07:52 +03: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"

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

Configuration Options

Option Type Default Description
connectionString string process.env.WORKFLOW_POSTGRES_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.

Environment Variables

Variable Description Default
WORKFLOW_TARGET_WORLD Set to "@workflow/world-postgres" to use this world -
WORKFLOW_POSTGRES_URL PostgreSQL connection string '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

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:

pnpm exec workflow-postgres-setup
# or
npm exec workflow-postgres-setup

The CLI automatically loads .env files and will use 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 the workflow or step 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 and step execution is sent through /.well-known/workflow/v1/flow and /.well-known/workflow/v1/step

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"