diff --git a/migrations/2026-06-01_pgqueue_primitive.sql b/migrations/2026-06-01_pgqueue_primitive.sql new file mode 100644 index 0000000..2d05aa3 --- /dev/null +++ b/migrations/2026-06-01_pgqueue_primitive.sql @@ -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;