13f3e3a7aa
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
340 lines
11 KiB
PL/PgSQL
340 lines
11 KiB
PL/PgSQL
-- 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;
|