feat(queue): dead-letter store for group shared queues (§11.6 D3 / Track A M2)

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) <noreply@anthropic.com>

crates/manager-core/src/dispatcher.rs

@@ -408,19 +408,37 @@ impl Dispatcher {
         reason: &str,
     ) -> Option<DeadLetterId> {
         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<bool, crate::group_queue_repo::GroupQueueRepoError> {
             Ok(false)
         }
-        async fn drop_exhausted(
+        #[allow(clippy::too_many_arguments)]
+        async fn dead_letter(
             &self,
             _id: QueueMessageId,
             _token: Uuid,
-        ) -> Result<bool, crate::group_queue_repo::GroupQueueRepoError> {
-            Ok(false)
+            _group_id: picloud_shared::GroupId,
+            _collection: &str,
+            _trigger_id: Option<picloud_shared::TriggerId>,
+            _script_id: Option<ScriptId>,
+            _attempt: u32,
+            _first_attempt_at: chrono::DateTime<chrono::Utc>,
+            _last_error: &str,
+        ) -> Result<picloud_shared::DeadLetterId, crate::group_queue_repo::GroupQueueRepoError>
+        {
+            unreachable!("shared queue not exercised")
         }
         async fn reclaim_visibility_timeouts(
             &self,

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<dyn GroupKvRepo>,
     pub docs: Arc<dyn GroupDocsRepo>,
     pub files: Arc<dyn GroupFilesRepo>,
+    pub dead_letters: Arc<dyn GroupDeadLetterRepo>,
     pub groups: Arc<dyn GroupRepository>,
     pub authz: Arc<dyn AuthzRepo>,
 }
@@ -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<String>,
 }
 
+#[derive(Debug, Deserialize)]
+pub struct DeadLettersQuery {
+    #[serde(default)]
+    pub unresolved: bool,
+    #[serde(default)]
+    pub limit: Option<u32>,
+}
+
+/// 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<String>,
+    payload: serde_json::Value,
+}
+
+impl From<crate::group_dead_letter_repo::GroupDeadLetterRow> 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<GroupBlobsState>,
     Extension(principal): Extension<Principal>,
@@ -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<GroupBlobsState>,
+    Extension(principal): Extension<Principal>,
+    Path(ident): Path<String>,
+    Query(q): Query<DeadLettersQuery>,
+) -> Result<Json<Vec<DeadLetterDto>>, 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,

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<TriggerId>,
+    pub script_id: Option<ScriptId>,
+    pub payload: serde_json::Value,
+    pub attempt_count: u32,
+    pub first_attempt_at: DateTime<Utc>,
+    pub last_attempt_at: DateTime<Utc>,
+    pub last_error: String,
+    pub created_at: DateTime<Utc>,
+    pub resolved_at: Option<DateTime<Utc>>,
+    pub resolution: Option<String>,
+}
+
+#[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<Vec<GroupDeadLetterRow>, 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<Vec<GroupDeadLetterRow>, GroupDeadLetterRepoError> {
+        let rows: Vec<RawRow> = 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<Uuid>,
+    script_id: Option<Uuid>,
+    payload: serde_json::Value,
+    attempt_count: i32,
+    first_attempt_at: DateTime<Utc>,
+    last_attempt_at: DateTime<Utc>,
+    last_error: String,
+    created_at: DateTime<Utc>,
+    resolved_at: Option<DateTime<Utc>>,
+    resolution: Option<String>,
+}
+
+impl From<RawRow> 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,
+        }
+    }
+}

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<bool, GroupQueueRepoError>;
 
-    /// 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<bool, GroupQueueRepoError>;
+        group_id: GroupId,
+        collection: &str,
+        trigger_id: Option<picloud_shared::TriggerId>,
+        script_id: Option<picloud_shared::ScriptId>,
+        attempt: u32,
+        first_attempt_at: DateTime<Utc>,
+        last_error: &str,
+    ) -> Result<picloud_shared::DeadLetterId, GroupQueueRepoError>;
 
     /// 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<bool, GroupQueueRepoError> {
-        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<picloud_shared::TriggerId>,
+        script_id: Option<picloud_shared::ScriptId>,
+        attempt: u32,
+        first_attempt_at: DateTime<Utc>,
+        last_error: &str,
+    ) -> Result<picloud_shared::DeadLetterId, GroupQueueRepoError> {
+        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<u64, GroupQueueRepoError> {
@@ -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<Utc>,
+}

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;