ccd3644aa4
The producer side of shared durable queues:
- migration 0065 group_queue_messages (mirrors 0034, keyed by (group_id,
collection); CASCADE on group delete).
- GroupQueueRepo/PostgresGroupQueueRepo: enqueue + the competing-consumer
claim (FOR UPDATE SKIP LOCKED) + ack/nack/drop_exhausted/reclaim/depth.
- GroupQueueService trait (shared) + GroupQueueServiceImpl: resolve owning
group (kind='queue') from cx.app_id's chain, require editor+
(GroupQueueEnqueue, fails closed on anon), size-cap, enqueue.
- SDK: `queue::shared_collection("name")` -> GroupQueueHandle with
`.enqueue(msg[, opts])` / `.depth()` / `.depth_pending()`; wired through
Services + the picloud binary.
- authz: Capability::GroupQueueEnqueue(GroupId), editor+ / script:write.
Deterministic test proves competing consumers claim each message exactly
once. Consumption wiring (materialized consumers + dispatcher branch) lands
in D3.2.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
47 lines
2.2 KiB
SQL
47 lines
2.2 KiB
SQL
-- §11.6 D3: group-shared durable queues.
|
|
--
|
|
-- The queue counterpart to the group_kv/docs/files shared stores (0053-0055).
|
|
-- A group declares a `queue` shared collection (kind admitted in 0064); any
|
|
-- subtree app enqueues into ONE group-owned store via
|
|
-- `queue::shared_collection(name).enqueue(...)`. Consumption is by COMPETING
|
|
-- CONSUMERS: a group `[[triggers.queue]] shared = true` template materializes a
|
|
-- consumer copy per descendant app (like the M5 stateful templates), and all
|
|
-- copies claim from this shared store with FOR UPDATE SKIP LOCKED — each message
|
|
-- delivered at-most-once across the whole subtree, scaling horizontally.
|
|
--
|
|
-- Identity tuple: (group_id, collection). Keyed by the OWNING GROUP, not an app
|
|
-- (the shared rows belong to the group). CASCADE on group delete (data dies with
|
|
-- its group; an app delete leaves it), matching the other group stores. Mirrors
|
|
-- queue_messages (0034) column-for-column, swapping (app_id, queue_name) for
|
|
-- (group_id, collection).
|
|
|
|
CREATE TABLE group_queue_messages (
|
|
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
|
group_id UUID NOT NULL REFERENCES groups(id) ON DELETE CASCADE,
|
|
collection TEXT NOT NULL,
|
|
payload JSONB NOT NULL,
|
|
enqueued_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
deliver_after TIMESTAMPTZ NULL,
|
|
claim_token UUID NULL,
|
|
claimed_at TIMESTAMPTZ NULL,
|
|
attempt INT NOT NULL DEFAULT 0,
|
|
max_attempts INT NOT NULL DEFAULT 3,
|
|
enqueued_by_principal UUID NULL
|
|
);
|
|
|
|
-- Dispatch hot path: scan for unclaimed messages in (group_id, collection)
|
|
-- order by enqueued_at (the NOW() deliver_after filter is applied on the small
|
|
-- partial result, as in 0034).
|
|
CREATE INDEX idx_group_queue_messages_dispatch
|
|
ON group_queue_messages (group_id, collection, enqueued_at)
|
|
WHERE claim_token IS NULL;
|
|
|
|
-- depth() / depth_pending() helpers.
|
|
CREATE INDEX idx_group_queue_messages_group_collection
|
|
ON group_queue_messages (group_id, collection);
|
|
|
|
-- Reclaim-task scan: currently-leased messages past their visibility timeout.
|
|
CREATE INDEX idx_group_queue_messages_claimed
|
|
ON group_queue_messages (claimed_at)
|
|
WHERE claim_token IS NOT NULL;
|