feat(queues): add pgqueue primitive — Postgres-native work queue
SKIP LOCKED + heartbeat-based lease + LISTEN/NOTIFY. Lives in pgqueue schema. Idempotent migration, installable in any database. API: enqueue, claim_next, heartbeat, ack, nack, janitor_reset_stale_claims. View: pgqueue.depth for dashboard. Reference: skills/data-stack-conventions.md
This commit is contained in:
Chief of Staff
parent
7b5ffe6b11
commit
13f3e3a7aa
1 changed files with 340 additions and 0 deletions
340
migrations/2026-06-01_pgqueue_primitive.sql
Normal file
340
migrations/2026-06-01_pgqueue_primitive.sql
Normal file
|
|
@ -0,0 +1,340 @@
|
|||
-- pgqueue — Postgres-native queue primitive for Scobleclaw.
|
||||
--
|
||||
-- One engine, three jobs. This is the queue primitive. Lives in `pgqueue`
|
||||
-- schema so it can be installed in any database (pool_manager, homeschool, etc.)
|
||||
-- without polluting the public schema.
|
||||
--
|
||||
-- Design (2026-06-01):
|
||||
-- - SKIP LOCKED for atomic claim (Postgres does the work)
|
||||
-- - Heartbeat-based lease: `lease_until` column, janitor resets stale claims
|
||||
-- - LISTEN/NOTIFY trigger for sub-second wakeup
|
||||
-- - Idempotency key column: dedupes retried enqueues
|
||||
-- - Per-queue isolation: claim from one queue doesn't block another
|
||||
--
|
||||
-- Why not pgmq: it would mean adding a new extension. The visibility-timeout
|
||||
-- primitive pgmq buys you is a 30-line janitor + a `lease_until` column. Same
|
||||
-- semantics, no extension.
|
||||
--
|
||||
-- Reference: skills/data-stack-conventions.md (Postgres as the data stack).
|
||||
|
||||
BEGIN;
|
||||
|
||||
CREATE SCHEMA IF NOT EXISTS pgqueue;
|
||||
|
||||
-- ── jobs table ──────────────────────────────────────────────────────
|
||||
CREATE TABLE IF NOT EXISTS pgqueue.jobs (
|
||||
id bigserial PRIMARY KEY,
|
||||
queue text NOT NULL,
|
||||
payload jsonb NOT NULL,
|
||||
idempotency_key text,
|
||||
state text NOT NULL DEFAULT 'ready'
|
||||
CHECK (state IN ('ready', 'claimed', 'done', 'dead')),
|
||||
priority int NOT NULL DEFAULT 0, -- higher = sooner
|
||||
run_after timestamptz NOT NULL DEFAULT now(),
|
||||
attempts int NOT NULL DEFAULT 0,
|
||||
max_attempts int NOT NULL DEFAULT 5,
|
||||
last_error text,
|
||||
claimed_by text,
|
||||
claimed_at timestamptz,
|
||||
lease_until timestamptz, -- heartbeat; janitor resets if expired
|
||||
created_at timestamptz NOT NULL DEFAULT now(),
|
||||
updated_at timestamptz NOT NULL DEFAULT now(),
|
||||
finished_at timestamptz
|
||||
);
|
||||
|
||||
-- Index for claim query: order by (priority desc, created_at) on ready jobs
|
||||
CREATE INDEX IF NOT EXISTS jobs_claim_idx
|
||||
ON pgqueue.jobs (queue, priority DESC, created_at)
|
||||
WHERE state = 'ready';
|
||||
|
||||
-- Idempotency: same (queue, idempotency_key) returns the same job id
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS jobs_idem_idx
|
||||
ON pgqueue.jobs (queue, idempotency_key)
|
||||
WHERE idempotency_key IS NOT NULL;
|
||||
|
||||
-- Index for janitor: scan claimed jobs whose lease has expired
|
||||
CREATE INDEX IF NOT EXISTS jobs_lease_idx
|
||||
ON pgqueue.jobs (lease_until)
|
||||
WHERE state = 'claimed';
|
||||
|
||||
-- Index for dashboard
|
||||
CREATE INDEX IF NOT EXISTS jobs_state_idx
|
||||
ON pgqueue.jobs (queue, state, created_at DESC);
|
||||
|
||||
-- ── enqueue ─────────────────────────────────────────────────────────
|
||||
CREATE OR REPLACE FUNCTION pgqueue.enqueue(
|
||||
p_queue text,
|
||||
p_payload jsonb,
|
||||
p_idempotency_key text DEFAULT NULL,
|
||||
p_priority int DEFAULT 0,
|
||||
p_run_after timestamptz DEFAULT now()
|
||||
)
|
||||
RETURNS bigint
|
||||
LANGUAGE plpgsql
|
||||
AS $$
|
||||
DECLARE
|
||||
v_id bigint;
|
||||
BEGIN
|
||||
-- If idempotency_key provided and the row already exists, return its id.
|
||||
-- This dedupes retries from upstream (e.g., a cron enqueuing twice).
|
||||
IF p_idempotency_key IS NOT NULL THEN
|
||||
SELECT id INTO v_id
|
||||
FROM pgqueue.jobs
|
||||
WHERE queue = p_queue AND idempotency_key = p_idempotency_key;
|
||||
IF v_id IS NOT NULL THEN
|
||||
RETURN v_id;
|
||||
END IF;
|
||||
END IF;
|
||||
|
||||
INSERT INTO pgqueue.jobs (queue, payload, idempotency_key, priority, run_after)
|
||||
VALUES (p_queue, p_payload, p_idempotency_key, p_priority, p_run_after)
|
||||
RETURNING id INTO v_id;
|
||||
|
||||
-- Wake up any worker LISTENing on this queue
|
||||
PERFORM pg_notify('pgqueue:' || p_queue, v_id::text);
|
||||
|
||||
RETURN v_id;
|
||||
END;
|
||||
$$;
|
||||
|
||||
-- ── claim next job (the heart of SKIP LOCKED) ──────────────────────
|
||||
-- Atomically claim one ready job. Returns NULL if no job available.
|
||||
-- p_worker_id: identifies the worker (hostname, pod name, etc.)
|
||||
-- p_lease_for: how long the lease is valid (e.g., '5 minutes'::interval)
|
||||
-- — worker must heartbeat before this expires or the janitor
|
||||
-- will reset the claim.
|
||||
CREATE OR REPLACE FUNCTION pgqueue.claim_next(
|
||||
p_queue text,
|
||||
p_worker_id text,
|
||||
p_lease_for interval DEFAULT '5 minutes'::interval
|
||||
)
|
||||
RETURNS TABLE (
|
||||
id bigint,
|
||||
queue text,
|
||||
payload jsonb,
|
||||
priority int,
|
||||
attempts int,
|
||||
max_attempts int,
|
||||
claimed_at timestamptz,
|
||||
lease_until timestamptz
|
||||
)
|
||||
LANGUAGE plpgsql
|
||||
AS $$
|
||||
BEGIN
|
||||
RETURN QUERY
|
||||
WITH next_job AS (
|
||||
SELECT j.id
|
||||
FROM pgqueue.jobs j
|
||||
WHERE j.queue = p_queue
|
||||
AND j.state = 'ready'
|
||||
AND j.run_after <= now()
|
||||
ORDER BY j.priority DESC, j.created_at
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT 1
|
||||
)
|
||||
UPDATE pgqueue.jobs j
|
||||
SET state = 'claimed',
|
||||
claimed_by = p_worker_id,
|
||||
claimed_at = now(),
|
||||
lease_until = now() + p_lease_for,
|
||||
attempts = j.attempts + 1,
|
||||
updated_at = now()
|
||||
FROM next_job
|
||||
WHERE j.id = next_job.id
|
||||
RETURNING j.id, j.queue, j.payload, j.priority, j.attempts, j.max_attempts,
|
||||
j.claimed_at, j.lease_until;
|
||||
END;
|
||||
$$;
|
||||
|
||||
-- ── heartbeat: extend the lease on a job you still hold ───────────
|
||||
CREATE OR REPLACE FUNCTION pgqueue.heartbeat(
|
||||
p_job_id bigint,
|
||||
p_worker_id text,
|
||||
p_extend_by interval DEFAULT '5 minutes'::interval
|
||||
)
|
||||
RETURNS boolean
|
||||
LANGUAGE plpgsql
|
||||
AS $$
|
||||
DECLARE
|
||||
v_updated int;
|
||||
BEGIN
|
||||
UPDATE pgqueue.jobs
|
||||
SET lease_until = now() + p_extend_by,
|
||||
updated_at = now()
|
||||
WHERE id = p_job_id
|
||||
AND state = 'claimed'
|
||||
AND claimed_by = p_worker_id;
|
||||
GET DIAGNOSTICS v_updated = ROW_COUNT;
|
||||
RETURN v_updated = 1;
|
||||
END;
|
||||
$$;
|
||||
|
||||
-- ── ack: mark job done ─────────────────────────────────────────────
|
||||
CREATE OR REPLACE FUNCTION pgqueue.ack(
|
||||
p_job_id bigint,
|
||||
p_worker_id text
|
||||
)
|
||||
RETURNS boolean
|
||||
LANGUAGE plpgsql
|
||||
AS $$
|
||||
DECLARE
|
||||
v_updated int;
|
||||
BEGIN
|
||||
UPDATE pgqueue.jobs
|
||||
SET state = 'done',
|
||||
finished_at = now(),
|
||||
updated_at = now()
|
||||
WHERE id = p_job_id
|
||||
AND state = 'claimed'
|
||||
AND claimed_by = p_worker_id;
|
||||
GET DIAGNOSTICS v_updated = ROW_COUNT;
|
||||
RETURN v_updated = 1;
|
||||
END;
|
||||
$$;
|
||||
|
||||
-- ── nack: requeue or dead-letter ──────────────────────────────────
|
||||
-- If attempts < max_attempts: requeue with backoff
|
||||
-- Else: mark dead, requires manual intervention (or admin purges)
|
||||
CREATE OR REPLACE FUNCTION pgqueue.nack(
|
||||
p_job_id bigint,
|
||||
p_worker_id text,
|
||||
p_error text DEFAULT NULL,
|
||||
p_requeue_after interval DEFAULT '30 seconds'::interval
|
||||
)
|
||||
RETURNS text -- 'requeued' | 'dead' | 'lost'
|
||||
LANGUAGE plpgsql
|
||||
AS $$
|
||||
DECLARE
|
||||
v_state text;
|
||||
v_attempts int;
|
||||
v_max int;
|
||||
BEGIN
|
||||
SELECT state, attempts, max_attempts
|
||||
INTO v_state, v_attempts, v_max
|
||||
FROM pgqueue.jobs
|
||||
WHERE id = p_job_id
|
||||
FOR UPDATE;
|
||||
|
||||
IF v_state IS NULL THEN
|
||||
RETURN 'lost'; -- job doesn't exist
|
||||
END IF;
|
||||
IF v_state != 'claimed' THEN
|
||||
RETURN 'lost'; -- not in a state we can nack from
|
||||
END IF;
|
||||
-- Worker id check is enforced by application; we trust here.
|
||||
|
||||
IF v_attempts >= v_max THEN
|
||||
UPDATE pgqueue.jobs
|
||||
SET state = 'dead',
|
||||
last_error = p_error,
|
||||
finished_at = now(),
|
||||
updated_at = now(),
|
||||
claimed_by = NULL,
|
||||
claimed_at = NULL,
|
||||
lease_until = NULL
|
||||
WHERE id = p_job_id;
|
||||
RETURN 'dead';
|
||||
ELSE
|
||||
UPDATE pgqueue.jobs
|
||||
SET state = 'ready',
|
||||
run_after = now() + p_requeue_after,
|
||||
last_error = p_error,
|
||||
claimed_by = NULL,
|
||||
claimed_at = NULL,
|
||||
lease_until = NULL,
|
||||
updated_at = now()
|
||||
WHERE id = p_job_id;
|
||||
|
||||
-- Wake up workers — the job is available again
|
||||
PERFORM pg_notify('pgqueue:' || (
|
||||
SELECT queue FROM pgqueue.jobs WHERE id = p_job_id
|
||||
), p_job_id::text);
|
||||
RETURN 'requeued';
|
||||
END IF;
|
||||
END;
|
||||
$$;
|
||||
|
||||
-- ── janitor: reset stale claims whose lease expired ────────────────
|
||||
-- Run this every 1-5 minutes via cron. Idempotent and safe to run often.
|
||||
CREATE OR REPLACE FUNCTION pgqueue.janitor_reset_stale_claims(
|
||||
p_grace interval DEFAULT '0 seconds'::interval
|
||||
)
|
||||
RETURNS int -- number of jobs reset
|
||||
LANGUAGE plpgsql
|
||||
AS $$
|
||||
DECLARE
|
||||
v_count int;
|
||||
BEGIN
|
||||
WITH stale AS (
|
||||
SELECT id
|
||||
FROM pgqueue.jobs
|
||||
WHERE state = 'claimed'
|
||||
AND lease_until IS NOT NULL
|
||||
AND lease_until + p_grace < now()
|
||||
FOR UPDATE SKIP LOCKED
|
||||
)
|
||||
UPDATE pgqueue.jobs j
|
||||
SET state = 'ready',
|
||||
claimed_by = NULL,
|
||||
claimed_at = NULL,
|
||||
lease_until = NULL,
|
||||
last_error = COALESCE(j.last_error, '') ||
|
||||
CASE WHEN j.last_error IS NOT NULL THEN E'\n' ELSE '' END ||
|
||||
'lease expired at ' || now()::text,
|
||||
updated_at = now()
|
||||
FROM stale
|
||||
WHERE j.id = stale.id;
|
||||
GET DIAGNOSTICS v_count = ROW_COUNT;
|
||||
RETURN v_count;
|
||||
END;
|
||||
$$;
|
||||
|
||||
-- ── view: queue depth per state ───────────────────────────────────
|
||||
CREATE OR REPLACE VIEW pgqueue.depth AS
|
||||
SELECT
|
||||
queue,
|
||||
state,
|
||||
count(*) AS count,
|
||||
min(created_at) AS oldest,
|
||||
max(updated_at) AS newest_update
|
||||
FROM pgqueue.jobs
|
||||
GROUP BY queue, state
|
||||
ORDER BY queue, state;
|
||||
|
||||
-- ── helper: notify on insert (in case enqueue() was bypassed) ─────
|
||||
-- This is a safety net. enqueue() does pg_notify directly.
|
||||
-- If someone INSERTs into pgqueue.jobs directly, the trigger still wakes
|
||||
-- listeners. Belt and suspenders.
|
||||
CREATE OR REPLACE FUNCTION pgqueue.notify_new_job() RETURNS trigger AS $$
|
||||
BEGIN
|
||||
PERFORM pg_notify('pgqueue:' || NEW.queue, NEW.id::text);
|
||||
RETURN NEW;
|
||||
END;
|
||||
$$ LANGUAGE plpgsql;
|
||||
|
||||
DROP TRIGGER IF EXISTS jobs_notify_insert ON pgqueue.jobs;
|
||||
CREATE TRIGGER jobs_notify_insert
|
||||
AFTER INSERT ON pgqueue.jobs
|
||||
FOR EACH ROW EXECUTE FUNCTION pgqueue.notify_new_job();
|
||||
|
||||
-- ── cleanup: drop dead/done jobs older than retention ─────────────
|
||||
-- For testing; production should run this less often with a longer window.
|
||||
CREATE OR REPLACE FUNCTION pgqueue.gc(
|
||||
p_retain_done interval DEFAULT '7 days'::interval,
|
||||
p_retain_dead interval DEFAULT '30 days'::interval
|
||||
)
|
||||
RETURNS TABLE (done_deleted int, dead_deleted int)
|
||||
LANGUAGE plpgsql
|
||||
AS $$
|
||||
BEGIN
|
||||
DELETE FROM pgqueue.jobs WHERE state = 'done' AND finished_at < now() - p_retain_done;
|
||||
GET DIAGNOSTICS done_deleted = ROW_COUNT;
|
||||
|
||||
DELETE FROM pgqueue.jobs WHERE state = 'dead' AND finished_at < now() - p_retain_dead;
|
||||
GET DIAGNOSTICS dead_deleted = ROW_COUNT;
|
||||
|
||||
RETURN NEXT;
|
||||
END;
|
||||
$$;
|
||||
|
||||
COMMIT;
|
||||
Loading…
Add table
Add a link
Reference in a new issue