From fd4336e883c87570201c648260fdff1820eb82da Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Sat, 11 Jul 2026 14:41:04 +0200 Subject: [PATCH] =?UTF-8?q?feat(queue):=20dead-letter=20store=20for=20grou?= =?UTF-8?q?p=20shared=20queues=20(=C2=A711.6=20D3=20/=20Track=20A=20M2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An exhausted SHARED durable-queue message was `drop_exhausted()`-ed with a warning — silent data loss. Per-app queues persist to `dead_letters` (0010); add the symmetric group store so an exhausted shared-queue message is preserved and operator-visible. - Migration 0068: `group_dead_letters`, keyed by (group_id, collection), CASCADE on the group (an app delete leaves the data — it belongs to the group, not the consuming app). - `GroupQueueRepo::dead_letter` (replaces `drop_exhausted`): one tx that INSERTs the dead-letter + DELETEs the live message, filtered by claim_token so a lost lease can't dead-letter a re-claimed message (mirrors queue_repo::dead_letter). - Dispatcher `q_terminal` shared arm now dead-letters instead of dropping. It returns None (not the dl id) so the per-app `fan_out_dead_letter` is SKIPPED — firing the consuming app's per-app handlers on a shared message (competing consumers → nondeterministic app) would be wrong. Fan-out to a *shared* dead_letter trigger is deferred (needs a new trigger kind). - Read side: `GroupDeadLetterRepo::list_for_group` backs a new read-only operator endpoint GET /api/v1/admin/groups/{id}/dead-letters (GroupKvRead, mirrors the M4 group-blobs surface). `pic dead-letters ls --group` deferred (optional; the HTTP endpoint is the operator surface). Pinned by group_queue::dead_letter_moves_an_exhausted_message_to_the_group_store (store move + claim-token-mismatch guard + operator read). Schema golden reblessed. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../migrations/0068_group_dead_letters.sql | 43 ++++++ crates/manager-core/src/dispatcher.rs | 51 +++++-- crates/manager-core/src/group_blobs_api.rs | 74 ++++++++++ .../src/group_dead_letter_repo.rs | 136 ++++++++++++++++++ crates/manager-core/src/group_queue_repo.rs | 103 ++++++++++--- crates/manager-core/src/lib.rs | 1 + crates/manager-core/tests/expected_schema.txt | 28 ++++ crates/manager-core/tests/group_queue.rs | 122 ++++++++++++++++ crates/picloud/src/lib.rs | 6 + 9 files changed, 536 insertions(+), 28 deletions(-) create mode 100644 crates/manager-core/migrations/0068_group_dead_letters.sql create mode 100644 crates/manager-core/src/group_dead_letter_repo.rs diff --git a/crates/manager-core/migrations/0068_group_dead_letters.sql b/crates/manager-core/migrations/0068_group_dead_letters.sql new file mode 100644 index 0000000..05e5d0b --- /dev/null +++ b/crates/manager-core/migrations/0068_group_dead_letters.sql @@ -0,0 +1,43 @@ +-- §11.6 D3 (dead-letter store for group SHARED durable queues). +-- +-- Previously an exhausted shared-queue message was `drop_exhausted()`-ed with a +-- warning — silent data loss. Per-app queues persist to `dead_letters` (0010); +-- this is the symmetric group store, keyed by `group_id` (the shared-queue owner) +-- + `collection` (a group has many shared queues) instead of `app_id`. +-- +-- CASCADE on the group mirrors `group_queue_messages` (0065): deleting the group +-- drops its dead-letters; an app delete leaves them (the data belongs to the +-- group, not the consuming app). No `app_id` — a shared-queue message is not +-- owned by whichever descendant happened to consume it. +-- +-- Fan-out to a *shared* dead-letter TRIGGER is deferred (would need a new shared +-- dead-letter trigger kind); M2 stops the data loss + makes exhausted messages +-- operator-visible (GET /api/v1/admin/groups/{id}/dead-letters). + +CREATE TABLE group_dead_letters ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + group_id UUID NOT NULL REFERENCES groups(id) ON DELETE CASCADE, + -- The shared-queue collection name the exhausted message came from. + collection TEXT NOT NULL, + -- The group_queue_messages.id row that exhausted retries (already deleted). + original_event_id UUID NOT NULL, + source TEXT NOT NULL, + op TEXT NOT NULL, + -- The materialized consumer copy's trigger/script (nullable, mirrors 0010). + trigger_id UUID, + script_id UUID, + payload JSONB NOT NULL, + attempt_count INT NOT NULL, + first_attempt_at TIMESTAMPTZ NOT NULL, + last_attempt_at TIMESTAMPTZ NOT NULL, + last_error TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + resolved_at TIMESTAMPTZ, + resolution TEXT + CHECK (resolution IN + ('replayed', 'ignored', 'handled_by_script', 'handler_failed')) +); + +-- Operator listing: newest-first per group (GET .../groups/{id}/dead-letters). +CREATE INDEX idx_group_dead_letters_group + ON group_dead_letters (group_id, created_at DESC); diff --git a/crates/manager-core/src/dispatcher.rs b/crates/manager-core/src/dispatcher.rs index 558008e..c448675 100644 --- a/crates/manager-core/src/dispatcher.rs +++ b/crates/manager-core/src/dispatcher.rs @@ -408,19 +408,37 @@ impl Dispatcher { reason: &str, ) -> Option { match c.shared_group { - Some(_) => { - if let Err(e) = self + Some(group_id) => { + // §11.6 D3: persist the exhausted message to the group dead-letter + // store instead of dropping it. We return None (not the dl id) so + // the per-app `fan_out_dead_letter` below is SKIPPED — firing the + // consuming app's per-app dead_letter handlers on a shared-queue + // message (competing consumers → nondeterministic app) would be + // wrong. Fan-out to a *shared* dead_letter trigger is deferred; the + // row is operator-visible via the group dead-letters admin API. + match self .group_queue - .drop_exhausted(claimed.id, claimed.claim_token) + .dead_letter( + claimed.id, + claimed.claim_token, + group_id, + &claimed.queue_name, + trigger_id, + script_id, + claimed.attempt, + claimed.enqueued_at, + reason, + ) .await { - tracing::warn!(?e, "shared-queue drop failed"); + Ok(dl_id) => tracing::warn!( + reason, + queue = %claimed.queue_name, + dead_letter_id = %dl_id.into_inner(), + "shared-queue message dead-lettered" + ), + Err(e) => tracing::error!(?e, "shared-queue dead-letter write failed"), } - tracing::warn!( - reason, - queue = %claimed.queue_name, - "shared-queue message dropped (no group dead-letter store yet)" - ); None } None => match self @@ -1663,12 +1681,21 @@ mod tests { ) -> Result { Ok(false) } - async fn drop_exhausted( + #[allow(clippy::too_many_arguments)] + async fn dead_letter( &self, _id: QueueMessageId, _token: Uuid, - ) -> Result { - Ok(false) + _group_id: picloud_shared::GroupId, + _collection: &str, + _trigger_id: Option, + _script_id: Option, + _attempt: u32, + _first_attempt_at: chrono::DateTime, + _last_error: &str, + ) -> Result + { + unreachable!("shared queue not exercised") } async fn reclaim_visibility_timeouts( &self, diff --git a/crates/manager-core/src/group_blobs_api.rs b/crates/manager-core/src/group_blobs_api.rs index ff149e3..a1a7d70 100644 --- a/crates/manager-core/src/group_blobs_api.rs +++ b/crates/manager-core/src/group_blobs_api.rs @@ -24,16 +24,23 @@ use serde_json::json; use uuid::Uuid; use crate::authz::{require, AuthzDenied, AuthzRepo, Capability}; +use crate::group_dead_letter_repo::GroupDeadLetterRepo; use crate::group_docs_repo::GroupDocsRepo; use crate::group_files_repo::GroupFilesRepo; use crate::group_kv_repo::GroupKvRepo; use crate::group_repo::GroupRepository; +/// Default/max page size for the dead-letters listing (an operator view, not a +/// bulk export). +const DEAD_LETTERS_LIMIT_DEFAULT: i64 = 100; +const DEAD_LETTERS_LIMIT_MAX: i64 = 500; + #[derive(Clone)] pub struct GroupBlobsState { pub kv: Arc, pub docs: Arc, pub files: Arc, + pub dead_letters: Arc, pub groups: Arc, pub authz: Arc, } @@ -46,6 +53,7 @@ pub fn group_blobs_router(state: GroupBlobsState) -> Router { .route("/groups/{id}/docs/{collection}/{doc_id}", get(get_doc)) .route("/groups/{id}/files", get(list_files)) .route("/groups/{id}/files/{collection}/{file_id}", get(get_file)) + .route("/groups/{id}/dead-letters", get(list_dead_letters)) .with_state(state) } @@ -64,6 +72,44 @@ struct ListKeysResponse { next_cursor: Option, } +#[derive(Debug, Deserialize)] +pub struct DeadLettersQuery { + #[serde(default)] + pub unresolved: bool, + #[serde(default)] + pub limit: Option, +} + +/// One group dead-letter, as returned by the operator listing. +#[derive(Debug, Serialize)] +struct DeadLetterDto { + id: Uuid, + collection: String, + source: String, + op: String, + attempt_count: u32, + last_error: String, + created_at: String, + resolved_at: Option, + payload: serde_json::Value, +} + +impl From for DeadLetterDto { + fn from(r: crate::group_dead_letter_repo::GroupDeadLetterRow) -> Self { + Self { + id: r.id.into_inner(), + collection: r.collection, + source: r.source, + op: r.op, + attempt_count: r.attempt_count, + last_error: r.last_error, + created_at: r.created_at.to_rfc3339(), + resolved_at: r.resolved_at.map(|t| t.to_rfc3339()), + payload: r.payload, + } + } +} + async fn list_kv( State(s): State, Extension(principal): Extension, @@ -251,6 +297,34 @@ async fn get_file( .into_response()) } +/// §11.6 D3: list a group's shared-queue dead-letters (newest-first). Guarded by +/// `GroupKvRead` (the same read tier as the shared collections above; no new +/// capability). `?unresolved=true` filters to still-open rows; `?limit=` caps +/// the page (default 100, max 500). +async fn list_dead_letters( + State(s): State, + Extension(principal): Extension, + Path(ident): Path, + Query(q): Query, +) -> Result>, GroupBlobsError> { + let group_id = resolve_group(&*s.groups, &ident).await?; + require( + s.authz.as_ref(), + &principal, + Capability::GroupKvRead(group_id), + ) + .await?; + let limit = q.limit.map_or(DEAD_LETTERS_LIMIT_DEFAULT, |n| { + i64::from(n).clamp(1, DEAD_LETTERS_LIMIT_MAX) + }); + let rows = s + .dead_letters + .list_for_group(group_id, q.unresolved, limit) + .await + .map_err(|e| GroupBlobsError::Backend(e.to_string()))?; + Ok(Json(rows.into_iter().map(DeadLetterDto::from).collect())) +} + async fn resolve_group( groups: &dyn GroupRepository, ident: &str, diff --git a/crates/manager-core/src/group_dead_letter_repo.rs b/crates/manager-core/src/group_dead_letter_repo.rs new file mode 100644 index 0000000..ae6a3c3 --- /dev/null +++ b/crates/manager-core/src/group_dead_letter_repo.rs @@ -0,0 +1,136 @@ +//! `GroupDeadLetterRepo` — READ side over `group_dead_letters` (§11.6 D3). +//! +//! The group counterpart to the per-app [`crate::dead_letter_repo::DeadLetterRepo`] +//! read surface, keyed by `(group_id, collection)`. The WRITE side lives in +//! [`crate::group_queue_repo::GroupQueueRepo::dead_letter`] (co-located with the +//! queue-message delete for one-transaction atomicity, exactly as the per-app +//! `queue_repo::dead_letter` inlines its own INSERT). This repo backs the +//! read-only operator surface GET /api/v1/admin/groups/{id}/dead-letters, so an +//! operator can see shared-queue messages that exhausted their retries rather +//! than having them vanish. + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use picloud_shared::{DeadLetterId, GroupId, ScriptId, TriggerId}; +use sqlx::PgPool; +use uuid::Uuid; + +#[derive(Debug, thiserror::Error)] +pub enum GroupDeadLetterRepoError { + #[error("database error: {0}")] + Db(#[from] sqlx::Error), +} + +/// One group dead-letter row (mirror of `dead_letter_repo::DeadLetterRow`, +/// group-keyed). +#[derive(Debug, Clone)] +pub struct GroupDeadLetterRow { + pub id: DeadLetterId, + pub group_id: GroupId, + pub collection: String, + pub original_event_id: Uuid, + pub source: String, + pub op: String, + pub trigger_id: Option, + pub script_id: Option, + pub payload: serde_json::Value, + pub attempt_count: u32, + pub first_attempt_at: DateTime, + pub last_attempt_at: DateTime, + pub last_error: String, + pub created_at: DateTime, + pub resolved_at: Option>, + pub resolution: Option, +} + +#[async_trait] +pub trait GroupDeadLetterRepo: Send + Sync { + /// A group's dead-letters, newest-first, capped at `limit` (bounded by the + /// caller). `unresolved_only` filters to `resolved_at IS NULL`. + async fn list_for_group( + &self, + group_id: GroupId, + unresolved_only: bool, + limit: i64, + ) -> Result, GroupDeadLetterRepoError>; +} + +pub struct PostgresGroupDeadLetterRepo { + pool: PgPool, +} + +impl PostgresGroupDeadLetterRepo { + #[must_use] + pub fn new(pool: PgPool) -> Self { + Self { pool } + } +} + +const COLS: &str = "id, group_id, collection, original_event_id, source, op, \ + trigger_id, script_id, payload, attempt_count, first_attempt_at, \ + last_attempt_at, last_error, created_at, resolved_at, resolution"; + +#[async_trait] +impl GroupDeadLetterRepo for PostgresGroupDeadLetterRepo { + async fn list_for_group( + &self, + group_id: GroupId, + unresolved_only: bool, + limit: i64, + ) -> Result, GroupDeadLetterRepoError> { + let rows: Vec = sqlx::query_as(&format!( + "SELECT {COLS} FROM group_dead_letters \ + WHERE group_id = $1 AND ($2 = FALSE OR resolved_at IS NULL) \ + ORDER BY created_at DESC LIMIT $3" + )) + .bind(group_id.into_inner()) + .bind(unresolved_only) + .bind(limit) + .fetch_all(&self.pool) + .await?; + Ok(rows.into_iter().map(Into::into).collect()) + } +} + +#[derive(sqlx::FromRow)] +struct RawRow { + id: Uuid, + group_id: Uuid, + collection: String, + original_event_id: Uuid, + source: String, + op: String, + trigger_id: Option, + script_id: Option, + payload: serde_json::Value, + attempt_count: i32, + first_attempt_at: DateTime, + last_attempt_at: DateTime, + last_error: String, + created_at: DateTime, + resolved_at: Option>, + resolution: Option, +} + +impl From for GroupDeadLetterRow { + fn from(r: RawRow) -> Self { + Self { + id: DeadLetterId(r.id), + group_id: r.group_id.into(), + collection: r.collection, + original_event_id: r.original_event_id, + source: r.source, + op: r.op, + trigger_id: r.trigger_id.map(Into::into), + script_id: r.script_id.map(Into::into), + payload: r.payload, + attempt_count: u32::try_from(r.attempt_count).unwrap_or(0), + first_attempt_at: r.first_attempt_at, + last_attempt_at: r.last_attempt_at, + last_error: r.last_error, + created_at: r.created_at, + resolved_at: r.resolved_at, + resolution: r.resolution, + } + } +} diff --git a/crates/manager-core/src/group_queue_repo.rs b/crates/manager-core/src/group_queue_repo.rs index d1755ca..5ec14bf 100644 --- a/crates/manager-core/src/group_queue_repo.rs +++ b/crates/manager-core/src/group_queue_repo.rs @@ -9,9 +9,10 @@ //! all claiming this one store with `FOR UPDATE SKIP LOCKED`, so each message is //! delivered at-most-once across the subtree). //! -//! Deferred (documented): shared-queue dead-lettering — an exhausted message is -//! nacked with backoff by the dispatcher (never silently dropped), but there is -//! no group dead-letter store yet. +//! §11.6 D3 (dead-lettering): an exhausted shared-queue message moves to the +//! group dead-letter store (`group_dead_letters`, 0068) via [`GroupQueueRepo::dead_letter`] +//! instead of being dropped — operator-visible at GET .../groups/{id}/dead-letters. +//! Deferred: fan-out to a *shared* dead-letter TRIGGER (needs a new trigger kind). use async_trait::async_trait; use chrono::{DateTime, Utc}; @@ -83,13 +84,24 @@ pub trait GroupQueueRepo: Send + Sync { retry_delay: chrono::Duration, ) -> Result; - /// Drop a message that exhausted its attempts (no group dead-letter store - /// yet — see the module deferral note). Deletes iff `claim_token` matches. - async fn drop_exhausted( + /// §11.6 D3: a message exhausted its attempts — move it to the group dead- + /// letter store (`group_dead_letters`) and delete it from the live queue, in + /// one transaction (mirrors `queue_repo::dead_letter`). Filtered by + /// `claim_token` so a lost lease can't dead-letter a re-claimed message. + /// Returns the new dead-letter id. + #[allow(clippy::too_many_arguments)] + async fn dead_letter( &self, message_id: QueueMessageId, claim_token: Uuid, - ) -> Result; + group_id: GroupId, + collection: &str, + trigger_id: Option, + script_id: Option, + attempt: u32, + first_attempt_at: DateTime, + last_error: &str, + ) -> Result; /// Periodic safety net: clear claims older than a shared-queue consumer's /// `visibility_timeout_secs`. Returns the number of rows reclaimed. @@ -209,18 +221,69 @@ impl GroupQueueRepo for PostgresGroupQueueRepo { Ok(res.rows_affected() == 1) } - async fn drop_exhausted( + #[allow(clippy::too_many_arguments)] + async fn dead_letter( &self, message_id: QueueMessageId, claim_token: Uuid, - ) -> Result { - let res = - sqlx::query("DELETE FROM group_queue_messages WHERE id = $1 AND claim_token = $2") - .bind(message_id.into_inner()) - .bind(claim_token) - .execute(&self.pool) - .await?; - Ok(res.rows_affected() == 1) + group_id: GroupId, + collection: &str, + trigger_id: Option, + script_id: Option, + attempt: u32, + first_attempt_at: DateTime, + last_error: &str, + ) -> Result { + let mut tx = self.pool.begin().await?; + // Pull the row inside the tx for its payload; filtered by claim_token so + // a lost lease can't dead-letter a re-claimed message. + let row: DeadLetterSourceRow = sqlx::query_as( + "SELECT id, payload, enqueued_at FROM group_queue_messages \ + WHERE id = $1 AND claim_token = $2", + ) + .bind(message_id.into_inner()) + .bind(claim_token) + .fetch_one(&mut *tx) + .await?; + + let payload = serde_json::json!({ + "source": "queue", + "queue_name": collection, + "message": row.payload, + "enqueued_at": row.enqueued_at, + "attempt": attempt, + "message_id": row.id.to_string(), + }); + + let dl_id = picloud_shared::DeadLetterId::new(); + sqlx::query( + "INSERT INTO group_dead_letters ( \ + id, group_id, collection, original_event_id, source, op, \ + trigger_id, script_id, payload, attempt_count, \ + first_attempt_at, last_attempt_at, last_error \ + ) VALUES ($1, $2, $3, $4, 'queue', 'receive', $5, $6, $7, $8, $9, NOW(), $10)", + ) + .bind(dl_id.into_inner()) + .bind(group_id.into_inner()) + .bind(collection) + .bind(row.id) + .bind(trigger_id.map(picloud_shared::TriggerId::into_inner)) + .bind(script_id.map(picloud_shared::ScriptId::into_inner)) + .bind(&payload) + .bind(i32::try_from(attempt).unwrap_or(0)) + .bind(first_attempt_at) + .bind(last_error) + .execute(&mut *tx) + .await?; + + sqlx::query("DELETE FROM group_queue_messages WHERE id = $1 AND claim_token = $2") + .bind(message_id.into_inner()) + .bind(claim_token) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + Ok(dl_id) } async fn reclaim_visibility_timeouts(&self) -> Result { @@ -292,3 +355,11 @@ struct ClaimedRow { attempt: i32, max_attempts: i32, } + +/// Minimal projection of a to-be-dead-lettered message: its payload + timing. +#[derive(sqlx::FromRow)] +struct DeadLetterSourceRow { + id: Uuid, + payload: serde_json::Value, + enqueued_at: DateTime, +} diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index 005b379..a3a31f7 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -52,6 +52,7 @@ pub mod files_sweep; pub mod gc; pub mod group_blobs_api; pub mod group_collection_repo; +pub mod group_dead_letter_repo; pub mod group_docs_repo; pub mod group_docs_service; pub mod group_files_repo; diff --git a/crates/manager-core/tests/expected_schema.txt b/crates/manager-core/tests/expected_schema.txt index f9421d6..30664af 100644 --- a/crates/manager-core/tests/expected_schema.txt +++ b/crates/manager-core/tests/expected_schema.txt @@ -231,6 +231,24 @@ table: group_collections created_at: timestamp with time zone NOT NULL default=now() updated_at: timestamp with time zone NOT NULL default=now() +table: group_dead_letters + id: uuid NOT NULL default=gen_random_uuid() + group_id: uuid NOT NULL + collection: text NOT NULL + original_event_id: uuid NOT NULL + source: text NOT NULL + op: text NOT NULL + trigger_id: uuid NULL + script_id: uuid NULL + payload: jsonb NOT NULL + attempt_count: integer NOT NULL + first_attempt_at: timestamp with time zone NOT NULL + last_attempt_at: timestamp with time zone NOT NULL + last_error: text NOT NULL + created_at: timestamp with time zone NOT NULL default=now() + resolved_at: timestamp with time zone NULL + resolution: text NULL + table: group_docs group_id: uuid NOT NULL collection: text NOT NULL @@ -567,6 +585,10 @@ indexes on group_collections: group_collections_group_uidx: public.group_collections USING btree (group_id, lower(name), kind) WHERE (group_id IS NOT NULL) group_collections_pkey: public.group_collections USING btree (id) +indexes on group_dead_letters: + group_dead_letters_pkey: public.group_dead_letters USING btree (id) + idx_group_dead_letters_group: public.group_dead_letters USING btree (group_id, created_at DESC) + indexes on group_docs: group_docs_pkey: public.group_docs USING btree (group_id, collection, id) idx_group_docs_data_gin: public.group_docs USING gin (data jsonb_path_ops) @@ -814,6 +836,11 @@ constraints on group_collections: [FOREIGN KEY] group_collections_group_id_fkey: FOREIGN KEY (group_id) REFERENCES groups(id) ON DELETE CASCADE [PRIMARY KEY] group_collections_pkey: PRIMARY KEY (id) +constraints on group_dead_letters: + [CHECK] group_dead_letters_resolution_check: CHECK ((resolution = ANY (ARRAY['replayed'::text, 'ignored'::text, 'handled_by_script'::text, 'handler_failed'::text]))) + [FOREIGN KEY] group_dead_letters_group_id_fkey: FOREIGN KEY (group_id) REFERENCES groups(id) ON DELETE CASCADE + [PRIMARY KEY] group_dead_letters_pkey: PRIMARY KEY (id) + constraints on group_docs: [FOREIGN KEY] group_docs_group_id_fkey: FOREIGN KEY (group_id) REFERENCES groups(id) ON DELETE CASCADE [PRIMARY KEY] group_docs_pkey: PRIMARY KEY (group_id, collection, id) @@ -1005,3 +1032,4 @@ constraints on vars: 0065: group queues 0066: projects 0067: project environments + 0068: group dead letters diff --git a/crates/manager-core/tests/group_queue.rs b/crates/manager-core/tests/group_queue.rs index 63cf35e..231b65a 100644 --- a/crates/manager-core/tests/group_queue.rs +++ b/crates/manager-core/tests/group_queue.rs @@ -12,6 +12,9 @@ use std::collections::HashSet; +use picloud_manager_core::group_dead_letter_repo::{ + GroupDeadLetterRepo, PostgresGroupDeadLetterRepo, +}; use picloud_manager_core::group_queue_repo::{ GroupQueueRepo, NewGroupQueueMessage, PostgresGroupQueueRepo, }; @@ -134,3 +137,122 @@ async fn competing_consumers_claim_each_message_exactly_once() { .execute(&pool) .await; } + +#[tokio::test] +async fn dead_letter_moves_an_exhausted_message_to_the_group_store() { + // §11.6 D3: an exhausted shared-queue message is MOVED to the group dead- + // letter store (not dropped) — the row leaves the live queue and appears in + // `group_dead_letters`, operator-visible via GroupDeadLetterRepo. + let Some(pool) = pool_or_skip().await else { + return; + }; + let sfx = Uuid::new_v4().simple().to_string(); + let g: (Uuid,) = sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id") + .bind(format!("gdl-g-{sfx}")) + .fetch_one(&pool) + .await + .unwrap(); + let group = GroupId::from(g.0); + let queue = PostgresGroupQueueRepo::new(pool.clone()); + let dlq = PostgresGroupDeadLetterRepo::new(pool.clone()); + + let id = queue + .enqueue(NewGroupQueueMessage { + group_id: group, + collection: "jobs".into(), + payload: serde_json::json!({ "task": "boom" }), + deliver_after: None, + max_attempts: 1, + enqueued_by_principal: None, + }) + .await + .unwrap(); + let msg = queue.claim(group, "jobs").await.unwrap().expect("claimed"); + assert_eq!(msg.id, id); + + // Exhausted → dead-letter (mirrors the dispatcher's q_terminal shared arm). + let dl_id = queue + .dead_letter( + msg.id, + msg.claim_token, + group, + "jobs", + None, + None, + msg.attempt, + msg.enqueued_at, + "handler exploded", + ) + .await + .expect("dead_letter"); + + // The live queue no longer holds it. + assert_eq!( + queue.depth(group, "jobs").await.unwrap(), + 0, + "the dead-lettered message left the live queue" + ); + + // It is preserved + operator-visible in the group dead-letter store. + let rows = dlq.list_for_group(group, false, 100).await.unwrap(); + assert_eq!(rows.len(), 1, "exactly one dead-letter recorded"); + let row = &rows[0]; + assert_eq!(row.id.into_inner(), dl_id.into_inner()); + assert_eq!(row.collection, "jobs"); + assert_eq!(row.source, "queue"); + assert_eq!(row.last_error, "handler exploded"); + assert_eq!(row.payload["message"]["task"], "boom"); + assert!( + row.resolved_at.is_none(), + "a fresh dead-letter is unresolved" + ); + + // A claim_token mismatch cannot dead-letter a re-claimed message: enqueue + // another, claim it, then try to DL with a bogus token → the source-row + // fetch finds nothing (RowNotFound), leaving the live queue intact. + let id2 = queue + .enqueue(NewGroupQueueMessage { + group_id: group, + collection: "jobs".into(), + payload: serde_json::json!({ "task": "keep" }), + deliver_after: None, + max_attempts: 1, + enqueued_by_principal: None, + }) + .await + .unwrap(); + let m2 = queue + .claim(group, "jobs") + .await + .unwrap() + .expect("claimed 2"); + let bogus = Uuid::new_v4(); + assert!( + queue + .dead_letter( + m2.id, + bogus, + group, + "jobs", + None, + None, + 1, + m2.enqueued_at, + "x" + ) + .await + .is_err(), + "a claim-token mismatch must not dead-letter the message" + ); + assert_eq!( + queue.depth(group, "jobs").await.unwrap(), + 1, + "the mismatched message is still live" + ); + let _ = id2; + + let _ = sqlx::query("DELETE FROM groups WHERE id = $1") + .bind(g.0) + .execute(&pool) + .await; +} diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index 0f4cd53..dbf3893 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -273,6 +273,11 @@ pub async fn build_app( )); // §11.6 D3: the group shared-queue store + enqueue service. let group_queue_repo = Arc::new(PostgresGroupQueueRepo::new(pool.clone())); + let group_dead_letter_repo = Arc::new( + picloud_manager_core::group_dead_letter_repo::PostgresGroupDeadLetterRepo::new( + pool.clone(), + ), + ); let group_queue: Arc = Arc::new(GroupQueueServiceImpl::new( group_queue_repo.clone(), @@ -722,6 +727,7 @@ pub async fn build_app( kv: group_kv_repo.clone(), docs: group_docs_repo.clone(), files: group_files_repo.clone(), + dead_letters: group_dead_letter_repo.clone(), groups: groups_repo.clone(), authz: authz.clone(), },