// The SCHEMA the Postgres driver installs, apart from the statements it runs against it // (`driver-pg-sql.ts`): one constant an operator can read, apply and diff on its own. // // `SQL_JOBS_TABLE` is the queue's ONE install point — every durable table this package owns is // declared in it, so boot applies one constant and `x dev` and production get the same schema. // A table that shipped is extended with `alter table ... add column if not exists`, never by // editing its `create`: `create table if not exists` is a no-op against a database that already // has the table, so a new column added only there reaches new installs and nothing else. // // `dev-queue.ts` splits this constant on `;` and applies it statement by statement, so comments // inside it carry NO semicolons and NO apostrophes — a `;` in prose yields a comment-only chunk // and an odd quote count means a `;` sits inside an open literal. `driver-pg-sql.test.ts` pins // both, plus the shape of every chunk the split yields. export const SQL_JOBS_TABLE = ` create table if not exists x_jobs ( id uuid primary key, name text not null, queue text not null default 'default', input jsonb not null, idempotency_key text not null, run_id uuid not null, attempt int not null default 0, max_attempts int not null default 3, state text not null default 'ready', run_at timestamptz not null default now(), visible_at timestamptz, claimed_by text, last_error text, tenant_id text, created_at timestamptz not null default now(), updated_at timestamptz not null default now() ); -- The enqueuing requests trace and actor, carried onto the row so the job span is a CHILD of -- the request that queued it and an audit trail can say who asked. Added by alter because -- x_jobs shipped without them. alter table x_jobs add column if not exists traceparent text; alter table x_jobs add column if not exists enqueued_by text; -- Partial unique index: one LIVE job per (name, tenant, idempotency key). Completed rows stay for -- history, so re-running the same work tomorrow is allowed and re-delivering it today is not. -- -- The NAME is in the key, and its absence was silent data loss: two jobs that happened to derive -- the same natural key from the same input ("user:42") shared one namespace, so the second -- enqueue deduped against the FIRST jobs row and returned its id. The work never ran, no error -- was raised, and the queue showed one healthy job. -- -- The TENANT is in the key for that same argument with one word substituted, and it is the worse -- half. Every natural key an app writes is unique only WITHIN a tenant, invoice:1001 and -- order:5540 being the shapes the docs suggest, so tenant B enqueuing while tenant A held that -- key deduped into tenant A row: tenant B work never ran, and tenant B caller received tenant A -- job id, which is valid on every id-addressed surface. Cancel takes an id with no tenant -- predicate. -- -- coalesce rather than the bare column, because a null tenant_id compares unequal to every other -- null under a unique index and a tenantless queue would lose its dedupe entirely. All tenantless -- rows share one namespace instead, which is exactly what they had before tenancy existed. -- -- Each superseded index is dropped rather than left beside the new one. Each is strictly narrower, -- so keeping it would keep enforcing exactly the collision this fixes. drop index if exists x_jobs_idempotency_live_idx; drop index if exists x_jobs_name_idempotency_live_idx; create unique index if not exists x_jobs_name_tenant_idempotency_live_idx on x_jobs (name, (coalesce(tenant_id, '')), idempotency_key) where state in ('ready', 'delayed', 'running', 'suspended'); create index if not exists x_jobs_claim_idx on x_jobs (queue, run_at) where state in ('ready', 'delayed', 'suspended'); create table if not exists x_job_steps ( run_id uuid not null, name text not null, status text not null, output jsonb, started_at timestamptz not null default now(), completed_at timestamptz, wake_at timestamptz, event text, correlation_key text, attempts int not null default 1, error text, primary key (run_id, name) ); -- The backfill ledger. Keyed by RUN, not by name: a completed name blocks a re-run, and a forced -- one writes a new row, so what each pass swept survives the rerun that followed it. create table if not exists x_backfills ( run_id uuid primary key, name text not null, checksum text not null, status text not null default 'running', app_version text not null, rows_processed bigint not null default 0, last_cursor text, started_at timestamptz not null default now(), completed_at timestamptz ); create index if not exists x_backfills_name_idx on x_backfills (name, started_at desc); -- The transactional outbox. Staged by the SAME connection as the callers business rows, so it -- commits or vanishes with them. The relay publishes what committed. create table if not exists x_outbox ( id uuid primary key, job text not null, queue text not null default 'default', input jsonb not null, idempotency_key text not null, max_attempts int not null default 3, run_at timestamptz not null default now(), staged_at timestamptz not null default now(), tenant_id text, traceparent text, enqueued_by text, published_at timestamptz ); create index if not exists x_outbox_unpublished_idx on x_outbox (staged_at) where published_at is null; -- What the retention sweep reads, the other half of the table: published rows, oldest first. create index if not exists x_outbox_published_idx on x_outbox (published_at) where published_at is not null; -- The relays claim lease. A claim stamped in the same statement that locks the row is what stops -- two relays publishing one row twice: for update skip locked holds its locks only until that -- statement ends, which under autocommit is before claim returns. claimed_at is also what gives -- back the rows of a relay that died mid-batch, since a claim nothing can expire strands them. -- Added by alter because x_outbox shipped without them. alter table x_outbox add column if not exists claimed_at timestamptz; alter table x_outbox add column if not exists claimed_by text; -- The run the published job carries, allocated when the row is STAGED so the enqueue that staged -- it can answer which run it started. The row id is the published job id for the same reason. -- Added by alter because x_outbox shipped without it. A row staged before this column existed is -- given a run id by the default, which is what the driver would have minted at its publish. alter table x_outbox add column if not exists run_id uuid not null default gen_random_uuid(); -- The scheduler watermark. Without a durable one a redeployed scheduler has no idea what the -- pod it replaced already fired, so runRound takes the arming branch and every occurrence -- between the two processes is dropped with nothing logged. create table if not exists x_scheduler_state ( task_name text primary key, last_fired_at timestamptz not null, updated_at timestamptz not null default now() ); -- The last occurrence that DISPATCHED, and when. Apart from the watermark, which arming a task or -- skipping missed occurrences also moves: an operator asking when a task last ran is asking about -- jobs that were queued. Added by alter because x_scheduler_state shipped without them. alter table x_scheduler_state add column if not exists fired_occurrence_at timestamptz; alter table x_scheduler_state add column if not exists fired_at timestamptz; -- Leader election as an EXPIRING LEASE rather than a session advisory lock: the executor this -- package is handed is a pool, and a session-level pg_try_advisory_lock is released the moment -- that connection goes back to it. A row with an expiry needs no connection affinity at all. create table if not exists x_scheduler_leader ( lock_key text primary key, holder text not null, expires_at timestamptz not null ); -- Fleet-wide concurrency. One row per HELD SLOT, so the primary key is what serialises two -- workers reaching for the same slot — job.concurrency was documented, in the manifest, and -- enforced by nothing before this table existed. create table if not exists x_job_leases ( lease_key text not null, slot int not null, holder text not null, expires_at timestamptz not null, primary key (lease_key, slot) ); create index if not exists x_job_leases_expiry_idx on x_job_leases (expires_at); -- Events step.waitForEvent consumes. Stored and not broadcast: the publisher is a web pod and -- the resumer is a worker pod, so an in-heap bus strands every waiting run in a real deployment. create table if not exists x_job_events ( id uuid primary key, name text not null, payload jsonb not null, correlation_key text, published_at timestamptz not null default now(), expires_at timestamptz not null ); create index if not exists x_job_events_lookup_idx on x_job_events (name, published_at); -- What the body last reported through progress, as one value: done, total, note and the instant. -- A column and not a table, because it is read with the row and dies with it. Added by alter -- because x_jobs shipped without it. alter table x_jobs add column if not exists progress jsonb; -- The stack of the thrown value last_error describes, bounded by the driver before it is bound. alter table x_jobs add column if not exists last_error_stack text; -- How many times the row has been claimed, moved by the claim itself and never reset. A settle is -- fenced on it as well as on claimed_by: a worker that takes back its own lapsed job has the same -- id as the body still unwinding from the first claim, and only the ordinal tells the two apart. alter table x_jobs add column if not exists claims int not null default 0; -- The keyset a paged listing seeks by. Newest first, id as the tiebreak, so a page is an index -- range and never a sort of the whole table. create index if not exists x_jobs_created_idx on x_jobs (created_at desc, id desc); -- What the retention sweep reads. Partial over the four terminal states, so a live row never pays -- for it and the oldest finished rows are an index range rather than a scan of the whole queue. create index if not exists x_jobs_finished_idx on x_jobs (updated_at) where state in ('done', 'failed', 'dead', 'cancelled'); -- What the queue depth read (SQL_STATS) reads, every 15 seconds from every worker. Partial over -- the six states it counts, so the read is a range over the counted rows and never a pass over the -- finished ones the retention window keeps. Its predicate is the same list SQL_STATS filters on, -- which is what lets the planner prove the index applies. Queue, state and run_at are every column -- the read uses, so it can be answered from the index alone. create index if not exists x_jobs_depth_idx on x_jobs (queue, state, run_at) where state in ('ready', 'delayed', 'running', 'suspended', 'failed', 'dead'); -- The rows of one run, by run id: the retention sweep asks, per deleted row, whether another row of -- the run still stands before it deletes the step records of that run, and its follow-up and its -- orphan pass ask again. Without it each question is a scan of the whole queue. Not partial: run_id -- is not null, so a "where run_id is not null" predicate would index the same rows for nothing. create index if not exists x_jobs_run_idx on x_jobs (run_id); -- A paused queue or task, one row each. A queue pause is a fact about the QUEUE and not a column -- on every one of its rows: the claim reads this table with not exists, and a pause that had to -- rewrite a million rows would not take effect within one poll interval. create table if not exists x_job_pauses ( kind text not null, name text not null, paused_at timestamptz not null default now(), primary key (kind, name) ); -- The worker registry. One row per live worker, rewritten on its heartbeat and expired by time -- like a lease: a killed worker announces nothing and its row stops being read. Never cleaned up -- by anything that has to be running. create table if not exists x_job_workers ( id text primary key, host text not null, queues jsonb not null, concurrency int not null, in_flight jsonb not null, started_at timestamptz not null, heartbeat_at timestamptz not null default now(), expires_at timestamptz not null ); -- When the worker began draining (a retire, a SIGTERM, a stop) - set by its first draining -- announce and kept by every later one, null while it serves. A queue-depth autoscaler leaves the -- jobs such a worker still holds out of its count (SQL_SCALER_DEPTH): its replacement cannot take -- them. Added by alter because x_job_workers shipped without it. alter table x_job_workers add column if not exists draining_at timestamptz; -- Per-job history, in buckets. A settle adds to the one-minute bucket in the SAME statement that -- moves the row, and the scheduler leader folds old buckets into wider ones, so the key count per -- job name is fixed whatever the throughput. create table if not exists x_job_counters ( job text not null, bucket_ms int not null, bucket_start timestamptz not null, done bigint not null default 0, retried bigint not null default 0, failed bigint not null default 0, dead bigint not null default 0, duration_ms bigint not null default 0, primary key (job, bucket_ms, bucket_start) ); create index if not exists x_job_counters_age_idx on x_job_counters (bucket_ms, bucket_start); `.trim();