Skip to content
AnvilBase

Queues (PGMQ)

AnvilBase provides durable message queues through PGMQ, a Postgres extension. Because queues live in your project database, enqueuing a message can participate in the same transaction as your data writes — no separate broker, no two-system consistency problems, and queues are backed up with everything else.

Use them for background jobs, deferred work, retryable side effects, and producer/consumer pipelines.

There are two ways to drive queues:

  • The Queues data-plane API (recommended for apps) — a REST + SDK surface over HTTP, the Upstash-QStash / Supabase-Queues equivalent. Server-side only: it requires a project service_role key.
  • Raw SQL inside the project DB (pgmq.*) — best when enqueuing must share a transaction with your data writes.

Queues data-plane API (HTTP + SDK)

The data-plane API lets a backend holding only a project service_role key create queues and send/read/ack/archive messages over HTTP — no SQL, no separate broker. It is gated on service_role by design (a queue is a server-side primitive); an anon or authenticated key gets 403. Tenant scoping is enforced from the verified key, so a key for project A can never touch project B’s queues.

Base URL & shapes

Both URL shapes are mounted, mirroring REST/storage:

  • Native: POST /v1/queue/{project_id}/queues/{name}/send
  • SDK-alias: POST /v1/{project_id}/queue/v1/queues/{name}/send

Endpoints

Method & pathBodyReturns
POST .../queues{ "name": "jobs" }{ "queue": "jobs" } (201)
GET .../queues—[{ queue_name, is_partitioned, is_unlogged, created_at }]
DELETE .../queues/{name}—204
POST .../queues/{name}/send{ message, delay? }{ "msg_id": 1 }
POST .../queues/{name}/send_batch{ messages: [...] }{ "msg_ids": [...] }
POST .../queues/{name}/read{ vt, qty? }[{ msg_id, read_ct, enqueued_at, vt, message }]
POST .../queues/{name}/pop—one message, or null
POST .../queues/{name}/ack{ msg_id }{ "acked": true } (delete)
POST .../queues/{name}/archive{ msg_id }{ "archived": true }
GET .../queues/{name}/archived—the archive (DLQ view)
POST .../queues/{name}/replay{ msg_id }re-enqueues an archived message
GET .../queues/{name}/metrics—{ queue_length, newest/oldest_msg_age_sec, total_messages }
Terminal window
# Send with the service_role key
curl -X POST "$ANVILBASE_URL/v1/queue/$PROJECT_ID/queues/jobs/send" \
  -H "Authorization: Bearer $SERVICE_ROLE_KEY" \
  -H "Content-Type: application/json" \
  -d '{"message":{"task":"email","to":"ada@example.com"}}'
# → {"msg_id": 1}

Queue names & quotas

Queue names must be lowercase [a-z_][a-z0-9_]*, up to 47 characters. Lowercase is required because PGMQ folds queue names to lowercase internally; a mixed-case name would otherwise read/send fine but break the archive/replay path.

There is no enforced per-project queue-count or message quota — queues are ordinary tables in your project database and consume project-DB storage like any other data. The only built-in bound is the standard per-scope rate limiter on the data-plane API. Cap queue growth in your own workers (e.g. dead-letter on read_ct, and prune the archive), and watch project-DB size as usual.

Visibility timeout (vt)

read(vt, qty) returns up to qty messages and hides them from other readers for vt seconds. Process each message, then ack (delete) it on success or archive it on failure. If you neither ack nor archive within vt, the message becomes visible again and is redelivered — this is the at-least-once retry mechanism. read_ct counts deliveries; dead-letter (archive) a message after N attempts so one bad message can’t block the queue.

Archive / DLQ model — be honest about the semantics

PGMQ has no separate dead-letter queue. archive(msg_id) moves a message from the active queue table (pgmq.q_<name>) to a per-queue archive table (pgmq.a_<name>). That archive is the DLQ-equivalent: listArchived() reads it and replay(msg_id) moves a message back onto the active queue (read the archived body, re-send it, delete the archive row — done in one transaction so a crash mid-replay never loses the message). There is no automatic redelivery from the archive; replay is explicit.

SDK example

Every AnvilBase SDK exposes a queues module (client.queues). It needs the service_role key, so use it from trusted backend code only.

import { createClient } from "@anvilbase/client";


// SERVER-SIDE: the service_role key drives the queue.
const client = createClient(ANVILBASE_URL, SERVICE_ROLE_KEY, { projectId });


await client.queues.create("jobs");


const { data: sent } = await client.queues.queue("jobs").send({ task: "email" });


// Worker loop
const { data: msgs } = await client.queues.queue("jobs").read(30, { qty: 10 });
for (const m of msgs ?? []) {
  try {
    await handle(m.message);
    await client.queues.queue("jobs").ack(m.msg_id);
  } catch {
    if (m.read_ct >= 5) await client.queues.queue("jobs").archive(m.msg_id); // → DLQ
  }
}


// Inspect + replay a dead-lettered message
const { data: dead } = await client.queues.queue("jobs").listArchived();
await client.queues.queue("jobs").replay(dead![0].msg_id);


const { data: metrics } = await client.queues.queue("jobs").metrics();

The Python, Go, Rust, Swift, Kotlin, and Elixir SDKs expose the same surface (create/list/drop + send/sendBatch/read/pop/ack/archive/ listArchived/replay/metrics).

How it works (provisioning note)

The data-plane API calls public.pgmq_* SECURITY DEFINER wrapper functions installed in each project DB. The RLS roles can’t reach the pgmq schema directly, so these wrappers (EXECUTE granted to service_role) are what let the API — and the console Queues inspector — drive PGMQ.

There are two families of wrapper, owned differently for a security reason:

  • Message operations (send/read/pop/ack/archive/metrics/ list/send_batch) are owned by the project DB owner (anvilbase). They only touch rows in the per-queue tables, so a regular owner is enough. They’re installed at project-provision time and lazily re-ensured for older project databases.
  • Queue lifecycle (create/drop) is owned by a superuser. Creating or dropping a queue runs ALTER EXTENSION pgmq ADD/DROP TABLE under the hood, which Postgres only allows for a superuser or the extension’s owner. On AnvilBase’s Postgres image the pgmq extension is owned by a role that has been demoted from superuser, so a regular-owner wrapper would fail with must be owner of extension pgmq. AnvilBase therefore installs the public.pgmq_create / public.pgmq_drop_queue wrappers as a superuser in template1 at image initialization; every project database (a clone of template1) inherits them automatically, so per-project queue create/drop works for the non-superuser control-plane role without weakening isolation (the wrappers pin their search_path, take only a validated queue name, and operate only inside the caller’s own project database).

Self-hosting on an older data volume (retrofit)

The lifecycle wrappers ship in template1 from a fresh initialization. If you upgraded a self-hosted deployment whose Postgres volume predates this change (Phase 10), new project databases created before the upgrade won’t have them, and queue create/drop will return a clear “retrofit required” error (message ops still work). To retrofit, run the snippet documented at the top of the template1 block in docker/postgres/owned/init-anvilbase.owned.sh as the supabase_admin superuser — once against template1 (so future project DBs inherit it) and once against each existing platform_* database. No queues are created until you start using them, so there’s no rush; the control plane reports the exact retrofit path if you hit the gap.

Enable PGMQ

Nothing to enable — PGMQ is already installed in every project database (each project DB is cloned from a template that has the extension and the public.pgmq_* wrappers baked in). Don’t run create extension pgmq: the extension is owned by a demoted, non-superuser role, so create extension (and bare pgmq.* calls) fail for the roles a project can reach with must be owner of extension pgmq / permission denied for schema pgmq. Drive queues through the data-plane API/SDK or the public.pgmq_* wrappers shown below.

Create a queue

Queue create/drop goes through the superuser-installed lifecycle wrapper (not pgmq.create() directly — see How it works above; a bare pgmq.create() fails with must be owner of extension pgmq). Run it server-side under a service_role context (the SQL Editor, the exec_sql RPC, or psql as the anvilbase role):

select public.pgmq_create('email_jobs');
-- drop:  select public.pgmq_drop_queue('email_jobs');

Most apps should just use the data-plane API instead — POST .../queues with { "name": "email_jobs" } (see Endpoints) — which calls this same wrapper for you.

Send messages

A message is JSONB. The key benefit of the SQL path is enqueuing inside the same transaction as your data write. Use the public.pgmq_send wrapper (bare pgmq.send fails with permission denied for schema pgmq — the RLS roles and service_role have no direct access to the pgmq schema):

-- enqueue as part of a larger transaction (server-side, service_role context)
begin;
  insert into orders (id, total) values ($1, $2);
  select public.pgmq_send('email_jobs', jsonb_build_object(
    'to', 'alice@example.com',
    'template', 'welcome',
    'user_id', '3f2b…'
  ));
commit;

Queue writes are a server-side operation. The public.pgmq_* wrappers are granted to service_role only, so you enqueue from trusted server code — the SDK / data-plane API (recommended), the exec_sql RPC, or psql — never from a client holding an anon/authenticated key. To let end-user actions enqueue, call a server route or an Edge Function that holds the service_role key and does the public.pgmq_send (or the data-plane send) there.

Read and process messages

A consumer reads a batch with a visibility timeout (vt, seconds): messages become invisible to other readers for that window so you can process them, then you delete (done) or archive (keep for audit):

-- read up to 10 messages, hide them for 30s (server-side, service_role)
select * from public.pgmq_read('email_jobs', 30, 10);
-- → msg_id, read_ct, enqueued_at, vt, message


-- after success
select public.pgmq_delete('email_jobs', :msg_id);
-- or keep a record
select public.pgmq_archive('email_jobs', :msg_id);

public.pgmq_pop('email_jobs') reads-and-deletes a single message in one call (at-most-once). read + delete gives at-least-once with retries (a message reappears after vt if you don’t delete it). read_ct tells you how many times a message has been delivered — use it to dead-letter poison messages after N attempts.

A simple worker loop

// server-side worker using the service_role key
while (true) {
  const { data: msgs } = await admin.rpc('exec_sql', {
    sql: `select * from public.pgmq_read('email_jobs', 30, 10)`,
  })
  if (!msgs?.length) { await sleep(1000); continue }
  for (const m of msgs) {
    try {
      await sendEmail(m.message)
      await admin.rpc('exec_sql', { sql: `select public.pgmq_delete('email_jobs', ${m.msg_id})` })
    } catch (e) {
      if (m.read_ct >= 5) {
        await admin.rpc('exec_sql', { sql: `select public.pgmq_archive('email_jobs', ${m.msg_id})` })
      }
      // else: leave it; it reappears after vt for retry
    }
  }
}

Prefer the SDK worker loop shown above over exec_sql — the SDK calls the same wrappers through the data-plane API without hand-writing SQL.

Inspecting queues in the console

The console’s Queues inspector lets you list queues and view, archive, and delete messages without writing SQL — handy for debugging stuck jobs and clearing poison messages. It reads through the same PGMQ functions (pgmq_list_queues, pgmq_read, pgmq_archive, pgmq_delete).

Scheduling consumers

Pair PGMQ with pg_cron to drain a queue on a schedule entirely inside Postgres, or run an external worker like the loop above. Both work; cron keeps everything in-database, an external worker scales independently. A cron-driven drain calls the same public.pgmq_read / public.pgmq_delete wrappers — schedule the job through the Cron data-plane API / SDK (service_role), which runs the job body inside your project DB with the right privileges.

Platform use

AnvilBase itself uses a PGMQ queue (auth_events) internally to durably ship auth events into the audit log — a good illustration of the pattern: write the event in the same transaction as the auth action, then a consumer persists it reliably.

Tips

  • Keep messages small — store a reference (id) and re-read the row, rather than embedding large payloads.
  • Make consumers idempotent — at-least-once delivery means a message can be processed more than once.
  • Dead-letter on read_ct so a single bad message can’t block a queue forever.

Next: Cache (Valkey).