feat(v1.1.9): migrations 0034 + 0035 (queue_messages, queue_triggers)

- 0034_queue_messages.sql: per-app durable named queues. (app_id, queue_name)
  is the identity tuple; no queue registry table. Partial indexes on
  (claim_token IS NULL) for the dispatch hot path and (claim_token IS NOT NULL)
  for the visibility-timeout reclaim scan. The queue table IS the outbox
  for queue semantics — no double-buffering.
- 0035_queue_triggers.sql: widens triggers.kind CHECK to admit 'queue';
  widens outbox.source_kind CHECK to admit 'invoke' (for invoke_async).
  Adds queue_trigger_details(trigger_id PK, queue_name, visibility_timeout_secs,
  last_fired_at). Retry policy lives on the parent triggers row — same
  pattern as every other kind. One-consumer-per-queue is API-layer enforced
  via pg_advisory_xact_lock (partial unique index can't span parent.app_id).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-06-06 19:00:58 +02:00
parent b9e002a707
commit 4054af41ed
2 changed files with 94 additions and 0 deletions

View File

@@ -0,0 +1,53 @@
-- v1.1.9: durable per-app named queues (queue::*).
--
-- Producer: queue::enqueue(name, msg) — INSERTs into this table.
-- Consumer: a registered queue:receive trigger fires when a message is
-- available; the dispatcher claims with FOR UPDATE SKIP LOCKED + a
-- visibility-timeout window.
--
-- "The queue table IS the outbox" — there is no double-buffering. The
-- dispatcher's queue arm claims directly from queue_messages; the
-- visibility-timeout reclaim task resets stale claims so crashed
-- consumers don't lose work.
--
-- Identity tuple: (app_id, queue_name). Queue names are implicit — no
-- queue registry table; a queue exists once the first message is
-- enqueued under that name.
CREATE TABLE queue_messages (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
app_id UUID NOT NULL REFERENCES apps(id) ON DELETE CASCADE,
queue_name TEXT NOT NULL,
payload JSONB NOT NULL,
enqueued_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-- deliver_after: NULL = immediate; else dispatcher won't claim until NOW() >= deliver_after.
deliver_after TIMESTAMPTZ NULL,
-- claim_token: NULL = unclaimed; UUID = currently leased by a dispatcher.
claim_token UUID NULL,
claimed_at TIMESTAMPTZ NULL,
attempt INT NOT NULL DEFAULT 0,
max_attempts INT NOT NULL DEFAULT 3,
-- Forensic: who enqueued. NEVER used for authz — the trigger fires
-- as the principal that REGISTERED the consumer (design notes §4).
enqueued_by_principal UUID NULL
);
-- Dispatch hot path: scan for unclaimed messages in (app_id, queue_name)
-- order by enqueued_at. The (deliver_after IS NULL OR deliver_after <= NOW())
-- condition cannot live in the partial WHERE (NOW() is non-immutable);
-- Postgres applies it as a filter on the partial result set, which is
-- already small.
CREATE INDEX idx_queue_messages_dispatch
ON queue_messages (app_id, queue_name, enqueued_at)
WHERE claim_token IS NULL;
-- depth() / depth_pending() helpers and the dashboard's queue list view.
CREATE INDEX idx_queue_messages_app_queue
ON queue_messages (app_id, queue_name);
-- Reclaim-task scan: find currently-leased messages whose claim is older
-- than the per-queue visibility_timeout_secs. Bounded by the number of
-- in-flight messages, which is small.
CREATE INDEX idx_queue_messages_claimed
ON queue_messages (claimed_at)
WHERE claim_token IS NOT NULL;

View File

@@ -0,0 +1,41 @@
-- v1.1.9: queue:receive trigger kind + OutboxSourceKind::Invoke.
--
-- queue:receive is the new trigger kind that fires a script per claimed
-- message. Layout E parent + per-kind detail (mirrors pubsub).
--
-- 'invoke' is added to outbox.source_kind for invoke_async() — a
-- fire-and-forget function-composition call writes an outbox row that
-- the dispatcher fires through the standard executor path.
--
-- Queue itself does NOT need an outbox.source_kind variant (the queue
-- table IS the outbox for queue semantics — see 0034).
ALTER TABLE triggers DROP CONSTRAINT triggers_kind_check;
ALTER TABLE triggers ADD CONSTRAINT triggers_kind_check
CHECK (kind IN ('kv', 'dead_letter', 'docs', 'cron',
'files', 'pubsub', 'email', 'queue'));
ALTER TABLE outbox DROP CONSTRAINT outbox_source_kind_check;
ALTER TABLE outbox ADD CONSTRAINT outbox_source_kind_check
CHECK (source_kind IN ('http', 'kv', 'dead_letter', 'docs',
'cron', 'files', 'pubsub', 'email', 'invoke'));
-- Per-queue-trigger config. Retry policy lives on the parent triggers
-- row (retry_max_attempts, retry_backoff, retry_base_ms) — same
-- pattern as every other trigger kind. visibility_timeout_secs is
-- queue-specific.
--
-- "Exactly one consumer per (app_id, queue_name)" is enforced at the
-- API layer via a pg_advisory_xact_lock on hashtext(app_id || queue_name)
-- + a SELECT-then-INSERT in one transaction (a partial unique index
-- can't reference the parent's app_id column from the detail table).
CREATE TABLE queue_trigger_details (
trigger_id UUID PRIMARY KEY REFERENCES triggers(id) ON DELETE CASCADE,
queue_name TEXT NOT NULL,
visibility_timeout_secs INT NOT NULL DEFAULT 30,
last_fired_at TIMESTAMPTZ NULL
);
-- Help the dispatcher's "find all active queue consumers" query.
CREATE INDEX idx_queue_trigger_details_queue_name
ON queue_trigger_details (queue_name);