From 4054af41edc899e8fdd9a13b86abe3be2565bf33 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Sat, 6 Jun 2026 19:00:58 +0200 Subject: [PATCH] feat(v1.1.9): migrations 0034 + 0035 (queue_messages, queue_triggers) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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) --- .../migrations/0034_queue_messages.sql | 53 +++++++++++++++++++ .../migrations/0035_queue_triggers.sql | 41 ++++++++++++++ 2 files changed, 94 insertions(+) create mode 100644 crates/manager-core/migrations/0034_queue_messages.sql create mode 100644 crates/manager-core/migrations/0035_queue_triggers.sql diff --git a/crates/manager-core/migrations/0034_queue_messages.sql b/crates/manager-core/migrations/0034_queue_messages.sql new file mode 100644 index 0000000..52fbfdd --- /dev/null +++ b/crates/manager-core/migrations/0034_queue_messages.sql @@ -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; diff --git a/crates/manager-core/migrations/0035_queue_triggers.sql b/crates/manager-core/migrations/0035_queue_triggers.sql new file mode 100644 index 0000000..f2a0821 --- /dev/null +++ b/crates/manager-core/migrations/0035_queue_triggers.sql @@ -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);