Storage — PostgreSQL Schema, Monthly Partitions and the Repository

The state machine decides. PostgreSQL remembers. In this page you write the schema (migrations/V001__init.sql) and the repository (pkg/coord/repository*.go) — every SQL statement the engine runs on the hot path.

#Step 1: Pick the tables

#Step 2: Workflow definitions

Definitions are versioned. The primary key is (id, version), and a deployed flag marks the version new runs use:

-- migrations/V001__init.sql
CREATE TABLE IF NOT EXISTS workflow_definitions(
    id TEXT NOT NULL,
    name TEXT NOT NULL,
    version INT NOT NULL,
    spec_json JSONB NOT NULL,
    created_by TEXT NOT NULL,
    enabled BOOLEAN NOT NULL DEFAULT true,
    webhook_url TEXT,
    deployed BOOLEAN NOT NULL DEFAULT false,
    deployed_at TIMESTAMPTZ,
    deleted_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    PRIMARY KEY (id, version)
);

Definitions are never physically deleted (deleted_at instead), so a run that started on a version can always load it and finish.

#Step 3: Partition run data by run ID

Run data grows forever. A busy engine creates millions of task rows a day. Two problems follow:

• Indexes grow, and every insert gets slower.
• Deleting old runs with DELETE is slow and bloats the table.

The fix is range partitioning by month. The trick is choosing the partition key. Every hot query already filters on run_id. If the run ID itself encodes the creation time, then the run ID alone tells PostgreSQL which month's partition to look in.

That is exactly what a UUIDv7 is: the first 48 bits are a millisecond timestamp, so its text form sorts by creation time.

Declare the run table partitioned by id:

CREATE TABLE IF NOT EXISTS workflow_runs(
    id TEXT COLLATE "C" NOT NULL,
    workflow_definition_id TEXT NOT NULL,
    workflow_version INT NOT NULL,
    status TEXT NOT NULL DEFAULT 'PENDING',
    input_json JSONB,
    context_json JSONB, -- {input, config} only; task outputs live in task_instances
    error_json JSONB,
    deleted_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    completed_at TIMESTAMPTZ,
    PRIMARY KEY (id),
    FOREIGN KEY (workflow_definition_id, workflow_version) REFERENCES workflow_definitions(id, version)
) PARTITION BY RANGE (id);

COLLATE "C" makes text comparison plain byte order, so the range bounds behave exactly like the UUID's time ordering.

#Step 4: The task table

task_instances is the hottest table. Every state change of every task updates a row here. Its partition key is run_id, so a task lives in the same month's partition as its run:

CREATE TABLE IF NOT EXISTS task_instances(
    id TEXT COLLATE "C" NOT NULL,
    run_id TEXT COLLATE "C" NOT NULL,
    node_id TEXT NOT NULL,
    node_type TEXT NOT NULL,
    queue TEXT,                -- worker queue the task is dispatched on
    state TEXT NOT NULL DEFAULT 'PENDING',
    attempt INT NOT NULL DEFAULT 0,
    input_json JSONB,
    output_json JSONB,
    last_error JSONB,
    branches TEXT[],           -- SWITCH: successors it selected
    wake_at TIMESTAMPTZ,       -- WAITING: when the timer fires (WAIT) or times out (WAIT_FOR_EVENT)
    event_name TEXT,           -- WAITING on WAIT_FOR_EVENT: the event that resolves it
    scheduled_at TIMESTAMPTZ,
    started_at TIMESTAMPTZ,
    finished_at TIMESTAMPTZ,
    completed_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    lease_owner TEXT,
    lease_expires_at TIMESTAMPTZ,
    PRIMARY KEY (run_id, id),
    FOREIGN KEY (run_id) REFERENCES workflow_runs(id) ON DELETE CASCADE
) PARTITION BY RANGE (run_id);

Every field of NodeState and every field of CreateTask has a column: branches, wake_at, event_name, attempt. The lease is just two columns.

#Keep the indexes few and partial

Every index is rewritten on every update — unless PostgreSQL can do a HOT update (heap-only tuple), which skips index maintenance. HOT works only when no indexed column changes. state, lease_owner and lease_expires_at change constantly, so indexing them fully would kill HOT updates.

So use partial indexes that only contain the few rows each background loop needs:

-- Keep task_instances indexes minimal: every state change rewrites each index,
-- and indexing the state/lease columns would prevent HOT updates.
-- (run_id, node_id) also serves every lookup by run_id.
CREATE UNIQUE INDEX IF NOT EXISTS idx_task_instances_run_node ON task_instances(run_id, node_id);
-- Partial index for counting a run's active tasks: only rows in active states,
-- which are a small fraction of the table once workflows complete.
CREATE INDEX IF NOT EXISTS idx_task_instances_run_active
    ON task_instances (run_id)
    WHERE state IN ('PENDING', 'RUNNING', 'QUEUED', 'WAITING');
-- Lease sweeper: find RUNNING tasks whose lease expired.
CREATE INDEX IF NOT EXISTS idx_task_instances_running_lease
    ON task_instances (lease_expires_at)
    WHERE state = 'RUNNING';
-- Timer: find WAITING tasks that are due.
CREATE INDEX IF NOT EXISTS idx_task_instances_wake
    ON task_instances (wake_at)
    WHERE state = 'WAITING' AND wake_at IS NOT NULL;

Index	Used by
(run_id, node_id) unique	"one task per node per run", and loading a run's tasks
run_active	counting a run's active tasks
running_lease	the lease sweeper (recovery page)
wake	the timer loop

The unique index on (run_id, node_id) is a safety net. Even if two decisions somehow raced, the database would refuse a second task for the same node.

#Buffered signals

-- Events signalled to a run before a WAIT_FOR_EVENT node was waiting for them.
CREATE TABLE IF NOT EXISTS workflow_events(
    id TEXT COLLATE "C" NOT NULL,
    run_id TEXT COLLATE "C" NOT NULL,
    name TEXT NOT NULL,
    payload JSONB,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    consumed_at TIMESTAMPTZ,
    PRIMARY KEY (run_id, id),
    FOREIGN KEY (run_id) REFERENCES workflow_runs(id) ON DELETE CASCADE
) PARTITION BY RANGE (run_id);

CREATE INDEX IF NOT EXISTS idx_workflow_events_pending
    ON workflow_events (run_id, name, created_at)
    WHERE consumed_at IS NULL;

#Step 5: Create and drop partitions

Partitions must exist before a run lands in them. First, a function that turns a timestamp into the smallest UUIDv7 string at that time — the partition boundary:

-- composer_uuid7_bound returns the smallest UUIDv7 text with a timestamp >= ts:
-- the partition boundary for runs created from ts onwards.
CREATE OR REPLACE FUNCTION composer_uuid7_bound(ts TIMESTAMPTZ) RETURNS TEXT
LANGUAGE sql IMMUTABLE AS $$
    SELECT substr(h, 1, 8) || '-' || substr(h, 9, 4) || '-0000-0000-000000000000'
      FROM (SELECT lpad(to_hex(floor(extract(epoch FROM ts) * 1000)::BIGINT), 12, '0') AS h) b
$$;

Then create this month plus a few ahead, for all three run-scoped tables:

CREATE OR REPLACE FUNCTION composer_ensure_partitions(months_ahead INT DEFAULT 3) RETURNS INT
LANGUAGE plpgsql AS $$
DECLARE
    month_start TIMESTAMPTZ;
    tbl TEXT;
    created INT := 0;
BEGIN
    PERFORM pg_advisory_xact_lock(hashtext('composer_partitions'));
    FOR i IN 0..months_ahead LOOP
        month_start := date_trunc('month', NOW() AT TIME ZONE 'UTC') AT TIME ZONE 'UTC'
                       + make_interval(months => i);
        FOREACH tbl IN ARRAY ARRAY['workflow_runs', 'task_instances', 'workflow_events'] LOOP
            IF to_regclass(format('%I_p%s', tbl, to_char(month_start AT TIME ZONE 'UTC', 'YYYYMM'))) IS NULL THEN
                EXECUTE format('CREATE TABLE %I PARTITION OF %I FOR VALUES FROM (%L) TO (%L)',
                    tbl || '_p' || to_char(month_start AT TIME ZONE 'UTC', 'YYYYMM'), tbl,
                    composer_uuid7_bound(month_start),
                    composer_uuid7_bound(month_start + INTERVAL '1 month'));
                created := created + 1;
            END IF;
        END LOOP;
    END LOOP;
    RETURN created;
END;
$$;

The advisory lock makes it safe for several coordinators to call this at once. There is deliberately no DEFAULT partition: a default partition holding rows would block creating future partitions.

Retention drops whole months — instant, no bloat — but never a month that still has active runs:

CREATE OR REPLACE FUNCTION composer_drop_partitions(older_than TIMESTAMPTZ) RETURNS SETOF TEXT
LANGUAGE plpgsql AS $$
DECLARE
    part RECORD;
    suffix TEXT;
    active BOOLEAN;
BEGIN
    PERFORM pg_advisory_xact_lock(hashtext('composer_partitions'));
    FOR part IN
        SELECT c.relname
          FROM pg_inherits i
          JOIN pg_class c ON c.oid = i.inhrelid
         WHERE i.inhparent = 'workflow_runs'::regclass
         ORDER BY c.relname
    LOOP
        suffix := substring(part.relname FROM '_p(\d{6})$');
        CONTINUE WHEN suffix IS NULL
            OR to_timestamp(suffix, 'YYYYMM') AT TIME ZONE 'UTC' + INTERVAL '1 month' > older_than;

        EXECUTE format('SELECT EXISTS (SELECT 1 FROM %I WHERE status IN (''RUNNING'', ''PENDING''))',
                       part.relname) INTO active;
        IF active THEN
            RAISE NOTICE 'keeping partition %: it still has active runs', part.relname;
            CONTINUE;
        END IF;

        -- Referencing tables first, so their foreign keys never see orphans.
        EXECUTE format('DROP TABLE IF EXISTS %I', 'workflow_events_p' || suffix);
        EXECUTE format('DROP TABLE IF EXISTS %I', 'task_instances_p' || suffix);
        EXECUTE format('ALTER TABLE workflow_runs DETACH PARTITION %I', part.relname);
        EXECUTE format('DROP TABLE %I', part.relname);
        RETURN NEXT suffix;
    END LOOP;
END;
$$;

SELECT composer_ensure_partitions(3);

Because the partition comes from the run ID, a client-supplied run ID must be a recent UUIDv7 — otherwise it would point at a partition that does not exist. You will validate that in the coordinator.

#Step 6: The DB interfaces

Repository methods must work both on the pool and inside a transaction. Define the smallest interface that covers both in pkg/db/interface.go:

// pkg/db/interface.go

// Querier is a subset of DB interface for query operations.
// This is useful for accepting both DB and Tx types.
type Querier interface {
    Query(ctx context.Context, sql string, args ...interface{}) (pgx.Rows, error)
    QueryRow(ctx context.Context, sql string, args ...interface{}) pgx.Row
    Exec(ctx context.Context, sql string, args ...interface{}) (pgconn.CommandTag, error)
}

Both *pgxpool.Pool and pgx.Tx satisfy it.

#Step 7: The repository

Create pkg/coord/repository.go. Its rules are written right at the top:

// pkg/coord/repository.go

// repository holds the SQL for workflow runs, task instances, waits and the
// outbox. Every method takes the Querier to run on, so callers decide the
// transaction boundary; methods never log — they return errors for the caller.
//
// Run-scoped tables are partitioned by run ID, so every query here filters on
// run_id to touch a single partition.
type repository struct {
    // streamShards is the number of streams per task queue.
    streamShards int
}

// taskRef identifies one execution attempt of a task. Results are applied only
// when all fields match, which fences off reports from stale attempts.
type taskRef struct {
    TaskID  string
    RunID   string
    NodeID  string
    Attempt int
}

#Lock and load a run

This is the first statement of every event's transaction. FOR UPDATE holds the run's row lock until commit:

// LockRun takes the run's row lock for the rest of the transaction. Every
// state transition of a run happens under this lock, which serializes
// concurrent completions so join decisions always see each other's writes.
func (repository) LockRun(ctx context.Context, q db.Querier, runID string) (*runRow, error) {
    run := &runRow{ID: runID}
    err := q.QueryRow(ctx, `
        SELECT workflow_definition_id, workflow_version, status
          FROM workflow_runs
         WHERE id = $1 AND deleted_at IS NULL
           FOR UPDATE`,
        runID).Scan(&run.WorkflowID, &run.WorkflowVersion, &run.Status)
    if errors.Is(err, pgx.ErrNoRows) {
        return nil, fmt.Errorf("%w: workflow run %s", ErrNotFound, runID)
    }
    if err != nil {
        return nil, fmt.Errorf("lock workflow run: %w", err)
    }
    return run, nil
}

LoadRun locks, then builds the statemachine.Run snapshot:

// LoadRun locks the run and loads the snapshot the state machine decides on:
// the run row, its context and the state of every reached node.
func (r repository) LoadRun(ctx context.Context, q db.Querier, runID string) (*statemachine.Run, error) {
    row, err := r.LockRun(ctx, q, runID)
    if err != nil {
        return nil, err
    }
    rc, err := r.LoadRunContext(ctx, q, runID)
    if err != nil {
        return nil, err
    }
    nodes, err := r.LoadTasks(ctx, q, runID)
    if err != nil {
        return nil, err
    }
    return &statemachine.Run{
        ID:              runID,
        WorkflowID:      row.WorkflowID,
        WorkflowVersion: row.WorkflowVersion,
        Status:          row.Status,
        Context:         rc,
        Nodes:           nodes,
    }, nil
}

// LoadTasks returns the state of every task of the run, keyed by node.
func (repository) LoadTasks(ctx context.Context, q db.Querier, runID string) (map[string]*statemachine.NodeState, error) {
    rows, err := q.Query(ctx, `
        SELECT node_id, id, state, attempt, branches, COALESCE(event_name, '')
          FROM task_instances WHERE run_id = $1`, runID)
    if err != nil {
        return nil, fmt.Errorf("load tasks: %w", err)
    }
    defer rows.Close()

    nodes := map[string]*statemachine.NodeState{}
    for rows.Next() {
        var nodeID string
        n := &statemachine.NodeState{}
        if err := rows.Scan(&nodeID, &n.TaskID, &n.Status, &n.Attempt, &n.Branches, &n.AwaitedSignal); err != nil {
            return nil, fmt.Errorf("scan task: %w", err)
        }
        if n.Status != model.TaskInstanceStatusWaiting {
            n.AwaitedSignal = ""
        }
        nodes[nodeID] = n
    }
    return nodes, rows.Err()
}

LoadTasks reads six small columns — no inputs, no outputs. Outputs come only through LoadOutputs, which the machine calls via its Store:

// LoadOutputs returns the outputs of the given completed nodes of a run.
func (repository) LoadOutputs(ctx context.Context, q db.Querier, runID string, nodeIDs []string) (map[string]json.RawMessage, error) {
    rows, err := q.Query(ctx, `
        SELECT node_id, output_json FROM task_instances
         WHERE run_id = $1 AND node_id = ANY($2) AND state IN ('COMPLETED', 'SUCCEEDED')`,
        runID, nodeIDs)
    if err != nil {
        return nil, fmt.Errorf("load task outputs: %w", err)
    }
    defer rows.Close()

    outputs := make(map[string]json.RawMessage, len(nodeIDs))
    for rows.Next() {
        var nodeID string
        var output json.RawMessage
        if err := rows.Scan(&nodeID, &output); err != nil {
            return nil, fmt.Errorf("scan task output: %w", err)
        }
        outputs[nodeID] = output
    }
    return outputs, rows.Err()
}

#Insert a run

// InsertRun creates a RUNNING workflow run. It fails if the definition is
// missing or deleted, or if the run ID is already taken.
func (repository) InsertRun(ctx context.Context, q db.Querier, p *model.StartWorkflowRequest, rc *model.RunContext) error {
    var deletedAt *time.Time
    err := q.QueryRow(ctx,
        `SELECT deleted_at FROM workflow_definitions WHERE id = $1 AND version = $2`,
        p.WorkflowID, p.WorkflowVersion).Scan(&deletedAt)
    if errors.Is(err, pgx.ErrNoRows) {
        return fmt.Errorf("%w: workflow definition %s v%d", ErrNotFound, p.WorkflowID, p.WorkflowVersion)
    }
    if err != nil {
        return fmt.Errorf("check workflow definition: %w", err)
    }
    if deletedAt != nil {
        return fmt.Errorf("%w: workflow %s v%d has been deleted", ErrInvalid, p.WorkflowID, p.WorkflowVersion)
    }

    contextJSON, err := json.Marshal(rc.Persisted())
    if err != nil {
        return fmt.Errorf("marshal run context: %w", err)
    }
    tag, err := q.Exec(ctx, `
        INSERT INTO workflow_runs (id, workflow_definition_id, workflow_version, status, input_json, context_json)
        VALUES ($1, $2, $3, 'RUNNING', $4::jsonb, $5::jsonb)
        ON CONFLICT (id) DO NOTHING`,
        p.RunID, p.WorkflowID, p.WorkflowVersion, nullJSON(p.Input), contextJSON)
    if err != nil {
        return fmt.Errorf("insert workflow run: %w", err)
    }
    if tag.RowsAffected() == 0 {
        return fmt.Errorf("%w: workflow run %s already exists", ErrConflict, p.RunID)
    }
    return nil
}

ON CONFLICT DO NOTHING plus the rows-affected check makes a client-supplied run ID idempotent: starting the same run twice returns a conflict instead of creating a duplicate.

#Insert a task

// InsertTask inserts the task a CreateTask action describes. It reports
// false, without error, when the (run, node) already has a task.
func (repository) InsertTask(ctx context.Context, q db.Querier, runID string, t statemachine.CreateTask, scheduledAt time.Time) (bool, error) {
    var lastError []byte
    if t.Error != "" {
        lastError, _ = json.Marshal(map[string]string{"error": t.Error})
    }
    tag, err := q.Exec(ctx, `
        INSERT INTO task_instances
               (id, run_id, node_id, node_type, queue, state, input_json, output_json, last_error,
                branches, wake_at, event_name, scheduled_at, finished_at, completed_at)
        VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8::jsonb, $9::jsonb, $10, $11, $12, $13,
                CASE WHEN $6 IN ('COMPLETED', 'FAILED', 'SKIPPED') THEN NOW() END,
                CASE WHEN $6 = 'COMPLETED' THEN NOW() END)
        ON CONFLICT (run_id, node_id) DO NOTHING`,
        t.TaskID, runID, t.Node.NodeID, t.Node.GetNodeTypeForRouting(), t.Node.Queue(), t.Status,
        nullJSON(t.Input), nullJSON(t.Output), lastError,
        t.Branches, t.WakeAt, nullString(t.AwaitedSignal), scheduledAt)
    if err != nil {
        return false, fmt.Errorf("insert task instance for node %s: %w", t.Node.NodeID, err)
    }
    return tag.RowsAffected() == 1, nil
}

#Fenced updates

Look at the WHERE clause of every task update. Each one repeats the fencing check from the state machine — id, node_id, attempt, and an active state:

// CompleteTask marks the referenced attempt COMPLETED and releases its lease.
// It reports false when the attempt is stale or the task already finished.
func (repository) CompleteTask(ctx context.Context, q db.Querier, ref taskRef, output json.RawMessage) (*finishedTask, bool, error) {
    return scanFinishedTask(q.QueryRow(ctx, `
        UPDATE task_instances
           SET state = 'COMPLETED', output_json = $1::jsonb,
               completed_at = NOW(), finished_at = NOW(), updated_at = NOW(),
               lease_owner = NULL, lease_expires_at = NULL
         WHERE run_id = $2 AND id = $3 AND node_id = $4 AND attempt = $5
           AND state IN ('PENDING', 'QUEUED', 'RUNNING')
        RETURNING node_type, started_at`,
        nullJSON(output), ref.RunID, ref.TaskID, ref.NodeID, ref.Attempt))
}

// FailTask marks the referenced attempt FAILED and releases its lease.
func (repository) FailTask(ctx context.Context, q db.Querier, ref taskRef, reason string) (*finishedTask, bool, error) {
    lastError, _ := json.Marshal(map[string]string{"error": reason})
    return scanFinishedTask(q.QueryRow(ctx, `
        UPDATE task_instances
           SET state = 'FAILED', last_error = $1::jsonb,
               finished_at = NOW(), updated_at = NOW(),
               lease_owner = NULL, lease_expires_at = NULL
         WHERE run_id = $2 AND id = $3 AND node_id = $4 AND attempt = $5
           AND state IN ('PENDING', 'QUEUED', 'RUNNING')
        RETURNING node_type, started_at`,
        lastError, ref.RunID, ref.TaskID, ref.NodeID, ref.Attempt))
}

The state machine already checked, under the lock, that the attempt matches. So why check again in SQL? Defense in depth. If anything ever writes to a task outside the engine, the update matches zero rows, the shell sees it, and the transaction rolls back instead of saving a decision based on a wrong snapshot.

A retry puts the task back to PENDING, bumps the attempt, and returns everything needed for the next dispatch — including the stored input:

// RequeueTask schedules the next attempt of a failed execution: the task goes
// back to PENDING with attempt+1. It returns the dispatch for the new attempt.
func (repository) RequeueTask(ctx context.Context, q db.Querier, ref taskRef, reason string) (*taskDispatch, bool, error) {
    lastError, _ := json.Marshal(map[string]string{"error": reason})
    d := &taskDispatch{TaskID: ref.TaskID, RunID: ref.RunID, NodeID: ref.NodeID}
    err := q.QueryRow(ctx, `
        UPDATE task_instances
           SET state = 'PENDING', attempt = attempt + 1, last_error = $1::jsonb,
               started_at = NULL, updated_at = NOW(),
               lease_owner = NULL, lease_expires_at = NULL
         WHERE run_id = $2 AND id = $3 AND node_id = $4 AND attempt = $5
           AND state IN ('PENDING', 'QUEUED', 'RUNNING')
        RETURNING node_type, COALESCE(queue, lower(node_type)), attempt, input_json`,
        lastError, ref.RunID, ref.TaskID, ref.NodeID, ref.Attempt).
        Scan(&d.NodeType, &d.Queue, &d.Attempt, &d.Input)
    if errors.Is(err, pgx.ErrNoRows) {
        return nil, false, nil
    }
    if err != nil {
        return nil, false, fmt.Errorf("requeue task: %w", err)
    }
    return d, true, nil
}

#Finish a run

// FinishRun moves a non-terminal run to SUCCEEDED or FAILED. It returns the
// run's duration and whether this call made the transition.
func (repository) FinishRun(ctx context.Context, q db.Querier, runID string, status model.WorkflowInstanceStatus, reason string) (time.Duration, bool, error) {
    var errorJSON []byte
    if reason != "" {
        errorJSON, _ = json.Marshal(map[string]string{"error": reason})
    }
    var createdAt time.Time
    err := q.QueryRow(ctx, `
        UPDATE workflow_runs
           SET status = $1, error_json = $2::jsonb, completed_at = NOW(), updated_at = NOW()
         WHERE id = $3 AND status IN ('RUNNING', 'PENDING')
        RETURNING created_at`,
        status, errorJSON, runID).Scan(&createdAt)
    if errors.Is(err, pgx.ErrNoRows) {
        return 0, false, nil
    }
    if err != nil {
        return 0, false, fmt.Errorf("finish workflow run: %w", err)
    }
    return time.Since(createdAt), true, nil
}

#Waits and signals

Create pkg/coord/repository_waits.go. The timer loop finds due waits with the partial wake index:

// DueWaits returns up to limit WAITING tasks whose wake_at has passed, oldest first.
func (repository) DueWaits(ctx context.Context, q db.Querier, limit int) ([]dueWait, error) {
    rows, err := q.Query(ctx, `
        SELECT run_id, id, node_id
          FROM task_instances
         WHERE state = 'WAITING' AND wake_at <= NOW()
         ORDER BY wake_at
         LIMIT $1`, limit)
    if err != nil {
        return nil, fmt.Errorf("query due waits: %w", err)
    }
    return pgx.CollectRows(rows, func(row pgx.CollectableRow) (dueWait, error) {
        var w dueWait
        err := row.Scan(&w.RunID, &w.TaskID, &w.NodeID)
        return w, err
    })
}

// ResolveWait moves a WAITING task to COMPLETED (with output) or FAILED (with
// reason). It reports false if the task was no longer waiting.
func (repository) ResolveWait(ctx context.Context, q db.Querier, runID string, a statemachine.ResolveWait) (bool, error) {
    var lastError []byte
    if a.Reason != "" {
        lastError, _ = json.Marshal(map[string]string{"error": a.Reason})
    }
    tag, err := q.Exec(ctx, `
        UPDATE task_instances
           SET state = $1, output_json = $2::jsonb, last_error = $3::jsonb,
               finished_at = NOW(), updated_at = NOW(),
               completed_at = CASE WHEN $1 = 'COMPLETED' THEN NOW() END
         WHERE run_id = $4 AND id = $5 AND state = 'WAITING'`,
        a.Status, nullJSON(a.Output), lastError, runID, a.TaskID)
    if err != nil {
        return false, fmt.Errorf("resolve wait: %w", err)
    }
    return tag.RowsAffected() == 1, nil
}

Signals are buffered, found oldest-first, and marked consumed:

// BufferSignal stores a signal until a WAIT_FOR_EVENT node consumes it.
func (repository) BufferSignal(ctx context.Context, q db.Querier, runID string, a statemachine.BufferSignal) error {
    id, err := uuid.NewV7()
    if err != nil {
        return err
    }
    if _, err := q.Exec(ctx,
        `INSERT INTO workflow_events (id, run_id, name, payload) VALUES ($1, $2, $3, $4::jsonb)`,
        id.String(), runID, a.Name, nullJSON(a.Payload)); err != nil {
        return fmt.Errorf("buffer signal: %w", err)
    }
    return nil
}

// PendingSignal returns the oldest unconsumed signal of that name, or nil.
func (repository) PendingSignal(ctx context.Context, q db.Querier, runID, name string) (*statemachine.Signal, error) {
    s := &statemachine.Signal{}
    err := q.QueryRow(ctx, `
        SELECT id, payload FROM workflow_events
         WHERE run_id = $1 AND name = $2 AND consumed_at IS NULL
         ORDER BY created_at
         LIMIT 1`,
        runID, name).Scan(&s.ID, &s.Payload)
    if errors.Is(err, pgx.ErrNoRows) {
        return nil, nil
    }
    if err != nil {
        return nil, fmt.Errorf("query pending signal: %w", err)
    }
    return s, nil
}

// ConsumeSignal marks a buffered signal as used.
func (repository) ConsumeSignal(ctx context.Context, q db.Querier, runID, signalID string) error {
    if _, err := q.Exec(ctx,
        `UPDATE workflow_events SET consumed_at = NOW() WHERE run_id = $1 AND id = $2`,
        runID, signalID); err != nil {
        return fmt.Errorf("consume signal: %w", err)
    }
    return nil
}

Notice that every query — even the ones for signals — filters on run_id. That one habit keeps every hot-path query inside a single partition.

#Step 8: Apply the schema

psql "$DATABASE_URL" -f migrations/V001__init.sql
psql "$DATABASE_URL" -c '\d+ task_instances'     # see the partitions

You should see task_instances_p<YYYYMM> partitions for this month and three ahead. The engine now has a brain (state machine) and a memory (repository). The next page connects them.