fix(outbox): reclaim stale claims — a crash mid-dispatch stranded events forever

Chasing the e2e flakiness turned up a production durability bug, not a test
bug.

The dispatcher claims an outbox row, executes it, then either deletes it
(success) or reschedules it (failure) — both of which clear the claim. If
the PROCESS DIES in between, neither runs. And `claim_due` only ever selects
`claimed_at IS NULL`. Nothing else in the codebase clears `outbox.claimed_at`
— grep it: there are exactly three writers, and those are two of them.

So a crash or restart mid-dispatch stranded every in-flight row PERMANENTLY.
Its trigger never fired and no retry could notice. The outbox is the
universal trigger path, so the loss covered kv / docs / files / cron /
pubsub / email / invoke_async / dead-letter alike. This is the same
durability class as the audit's #6 (lost trigger event), which was just
fixed at the WRITE end — this is the same hole at the READ end.

Every other claim-based store already had the safety net: `queue_messages`
and `group_queue_messages` have `reclaim_visibility_timeouts`,
`workflow_steps` has its own reclaim. The outbox was the one that didn't.

`OutboxRepo::reclaim_stale_claims(timeout)` returns rows whose claim is older
than `PICLOUD_OUTBOX_CLAIM_TIMEOUT_SEC` (default 600s), run from the
dispatcher's existing reclaim ticker. The default is deliberately generous —
a script may run for up to 300s (the `scripts.timeout_seconds` CHECK), so a
claim held past twice that is abandoned rather than slow; reclaiming a row a
LIVE dispatcher is still working on would double-execute it (survivable —
dispatch is at-least-once — but not worth courting).

A reclaim does NOT bump `attempt_count`: the handler never ran, so it must
not consume the row's retry budget, or repeated restarts would dead-letter an
event that executed zero times. (Same reasoning as the transient queue
`release` fixed earlier in this branch.)

This is also the root cause of the flaky e2e suites: each test drops its
dispatcher, and `claim_due` is not scoped per app or per dispatcher, so a
test's dispatcher could claim another test's row and strand it on teardown.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/manager-core/src/dispatcher.rs

@@ -231,6 +231,8 @@ impl Dispatcher {
         // contend with the per-100ms dispatcher tick.
         let reclaim_queue = self.queue.clone();
         let reclaim_group_queue = self.group_queue.clone();
+        let reclaim_outbox = self.outbox.clone();
+        let outbox_claim_timeout = self.config.outbox_claim_timeout_secs;
         let reclaim_interval =
             Duration::from_millis(u64::from(self.config.queue_reclaim_interval_ms));
         tokio::spawn(async move {
@@ -238,6 +240,21 @@ impl Dispatcher {
             ticker.tick().await;
             loop {
                 ticker.tick().await;
+                // A dispatcher that died mid-dispatch left its claimed rows
+                // stranded — `claim_due` only takes unclaimed rows, so without
+                // this they would never fire again.
+                match reclaim_outbox
+                    .reclaim_stale_claims(outbox_claim_timeout)
+                    .await
+                {
+                    Ok(0) => {}
+                    Ok(n) => tracing::warn!(
+                        reclaimed = n,
+                        timeout_secs = outbox_claim_timeout,
+                        "reclaimed stale outbox claims — a dispatcher died mid-dispatch"
+                    ),
+                    Err(e) => tracing::warn!(?e, "outbox reclaim task errored"),
+                }
                 match reclaim_queue.reclaim_visibility_timeouts().await {
                     Ok(0) => {}
                     Ok(n) => tracing::info!(reclaimed = n, "queue visibility-timeout reclaim"),
@@ -2344,6 +2361,12 @@ mod tests {
 
         #[async_trait]
         impl OutboxRepo for UnusedOutbox {
+            async fn reclaim_stale_claims(
+                &self,
+                _timeout_secs: u32,
+            ) -> Result<u64, crate::outbox_repo::OutboxRepoError> {
+                Ok(0)
+            }
             async fn insert(
                 &self,
                 _row: NewOutboxRow,
@@ -2649,6 +2672,12 @@ mod tests {
 
         #[async_trait]
         impl OutboxRepo for RecordingOutbox {
+            async fn reclaim_stale_claims(
+                &self,
+                _timeout_secs: u32,
+            ) -> Result<u64, crate::outbox_repo::OutboxRepoError> {
+                Ok(0)
+            }
             async fn insert(
                 &self,
                 _row: NewOutboxRow,

crates/manager-core/src/invoke_service.rs

@@ -319,6 +319,9 @@ mod tests {
     }
     #[async_trait]
     impl OutboxRepo for CapturingOutbox {
+        async fn reclaim_stale_claims(&self, _timeout_secs: u32) -> Result<u64, OutboxRepoError> {
+            Ok(0)
+        }
         async fn insert(&self, row: NewOutboxRow) -> Result<uuid::Uuid, OutboxRepoError> {
             let id = uuid::Uuid::new_v4();
             *self.last.lock().await = Some(row);

crates/manager-core/src/outbox_repo.rs

@@ -120,6 +120,21 @@ pub trait OutboxRepo: Send + Sync {
         limit: i64,
     ) -> Result<Vec<OutboxRow>, OutboxRepoError>;
 
+    /// Return rows whose claim is older than `timeout_secs` to the queue.
+    ///
+    /// A dispatcher claims a row, executes it, then either deletes it (success)
+    /// or reschedules it (failure) — both of which clear the claim. If the
+    /// PROCESS DIES in between, neither runs, and since `claim_due` only selects
+    /// `claimed_at IS NULL`, that row is stranded forever: its trigger never
+    /// fires and no retry can ever notice. Every other claim-based store (queue,
+    /// group queue, workflow steps) has had this safety net; the outbox did not,
+    /// so a crash or restart mid-dispatch permanently lost every in-flight event.
+    ///
+    /// Returns how many rows were reclaimed. `attempt_count` is deliberately NOT
+    /// bumped: the handler never ran, so a reclaim must not consume the row's
+    /// retry budget (same reasoning as the transient queue `release`).
+    async fn reclaim_stale_claims(&self, timeout_secs: u32) -> Result<u64, OutboxRepoError>;
+
     /// Remove a row after a terminal outcome (success or dead-letter).
     async fn delete(&self, id: Uuid) -> Result<(), OutboxRepoError>;
 
@@ -215,6 +230,18 @@ impl OutboxRepo for PostgresOutboxRepo {
         Ok(())
     }
 
+    async fn reclaim_stale_claims(&self, timeout_secs: u32) -> Result<u64, OutboxRepoError> {
+        let res = sqlx::query(
+            "UPDATE outbox SET claimed_at = NULL, claimed_by = NULL \
+             WHERE claimed_at IS NOT NULL \
+               AND claimed_at < NOW() - ($1 || ' seconds')::INTERVAL",
+        )
+        .bind(i64::from(timeout_secs))
+        .execute(&self.pool)
+        .await?;
+        Ok(res.rows_affected())
+    }
+
     async fn reschedule(
         &self,
         id: Uuid,

crates/manager-core/src/trigger_config.rs

@@ -69,6 +69,24 @@ pub struct TriggerConfig {
     /// the message.
     pub queue_reclaim_interval_ms: u32,
 
+    /// How long an OUTBOX claim may be held before the reclaim task assumes the
+    /// dispatcher that took it is gone and returns the row to the queue
+    /// (`PICLOUD_OUTBOX_CLAIM_TIMEOUT_SEC`, default 600).
+    ///
+    /// Without this, a process that dies between claiming a row and finishing it
+    /// strands that row FOREVER — `claim_due` only selects `claimed_at IS NULL`,
+    /// and nothing else ever clears the claim. Every other claim-based store
+    /// (queue, group queue, workflow steps) already had a reclaimer; the outbox
+    /// did not, so a crash or restart mid-dispatch permanently lost every
+    /// in-flight trigger event.
+    ///
+    /// The default is deliberately generous: a script may run for up to 300s
+    /// (`scripts.timeout_seconds` CHECK), so a claim older than twice that is
+    /// abandoned rather than slow. Reclaiming a row a LIVE dispatcher is still
+    /// working on would double-execute it — dispatch is at-least-once, so that is
+    /// survivable, but not something to court by tuning this down.
+    pub outbox_claim_timeout_secs: u32,
+
     /// Default per-queue visibility-timeout in seconds (v1.1.9). Used
     /// at queue-trigger creation when the request omits an explicit
     /// value. Default 30. The per-trigger column overrides this.
@@ -88,6 +106,7 @@ impl TriggerConfig {
             abandoned_retention_days: 7,
             cron_tick_interval_ms: 30_000,
             queue_reclaim_interval_ms: 30_000,
+            outbox_claim_timeout_secs: 600,
             queue_default_visibility_timeout_secs: 30,
         }
     }
@@ -123,6 +142,10 @@ impl TriggerConfig {
             &mut c.queue_default_visibility_timeout_secs,
             "PICLOUD_QUEUE_DEFAULT_VISIBILITY_TIMEOUT_SECS",
         );
+        load_u32(
+            &mut c.outbox_claim_timeout_secs,
+            "PICLOUD_OUTBOX_CLAIM_TIMEOUT_SEC",
+        );
         c
     }
 }

crates/manager-core/tests/outbox_reclaim.rs

@@ -0,0 +1,153 @@
+//! A dispatcher that dies mid-dispatch must not strand its claimed outbox rows.
+//!
+//! The dispatcher claims a row, executes it, then deletes it (success) or
+//! reschedules it (failure) — both of which clear the claim. If the PROCESS DIES
+//! in between, neither runs. And `claim_due` only ever selects
+//! `claimed_at IS NULL`, so before `reclaim_stale_claims` that row was stranded
+//! FOREVER: its trigger never fired, and no retry could notice. The outbox is
+//! the universal trigger path (kv/docs/files/cron/pubsub/email/invoke_async/
+//! dead-letter), so a crash or restart mid-dispatch silently lost all of it.
+//!
+//! Every other claim-based store — `queue_messages`, `group_queue_messages`,
+//! `workflow_steps` — already had this reclaimer. The outbox was the one gap.
+//!
+//! Skips cleanly when `DATABASE_URL` is unset.
+
+use picloud_manager_core::outbox_repo::{
+    NewOutboxRow, OutboxRepo, OutboxSourceKind, PostgresOutboxRepo,
+};
+use picloud_shared::AppId;
+use sqlx::postgres::PgPoolOptions;
+use sqlx::PgPool;
+use uuid::Uuid;
+
+async fn pool_or_skip() -> Option<PgPool> {
+    let Ok(url) = std::env::var("DATABASE_URL") else {
+        eprintln!("outbox_reclaim: DATABASE_URL unset — skipping");
+        return None;
+    };
+    let pool = PgPoolOptions::new()
+        .max_connections(3)
+        .connect(&url)
+        .await
+        .expect("connect");
+    sqlx::migrate!("./migrations")
+        .run(&pool)
+        .await
+        .expect("migrate");
+    Some(pool)
+}
+
+async fn mk_app(pool: &PgPool) -> Uuid {
+    let uniq = Uuid::new_v4().simple().to_string();
+    let (group,): (Uuid,) =
+        sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id")
+            .bind(format!("or-grp-{uniq}"))
+            .fetch_one(pool)
+            .await
+            .expect("group");
+    let (app,): (Uuid,) =
+        sqlx::query_as("INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id")
+            .bind(format!("or-app-{uniq}"))
+            .bind(group)
+            .fetch_one(pool)
+            .await
+            .expect("app");
+    app
+}
+
+/// Exactly `claim_due`'s predicate: a row is dispatchable iff its claim is clear.
+async fn claimable(pool: &PgPool, id: Uuid) -> bool {
+    let (n,): (i64,) =
+        sqlx::query_as("SELECT COUNT(*) FROM outbox WHERE id = $1 AND claimed_at IS NULL")
+            .bind(id)
+            .fetch_one(pool)
+            .await
+            .expect("claimable");
+    n == 1
+}
+
+async fn attempt_count(pool: &PgPool, id: Uuid) -> i32 {
+    let (n,): (i32,) = sqlx::query_as("SELECT attempt_count FROM outbox WHERE id = $1")
+        .bind(id)
+        .fetch_one(pool)
+        .await
+        .expect("attempt_count");
+    n
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn a_stranded_claim_is_reclaimed_without_burning_the_retry_budget() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let repo = PostgresOutboxRepo::new(pool.clone());
+    let app = mk_app(&pool).await;
+
+    let id = repo
+        .insert(NewOutboxRow {
+            app_id: AppId::from(app),
+            source_kind: OutboxSourceKind::Kv,
+            trigger_id: None,
+            script_id: None,
+            reply_to: None,
+            payload: serde_json::json!({}),
+            origin_principal: None,
+            trigger_depth: 0,
+            root_execution_id: None,
+        })
+        .await
+        .expect("insert");
+
+    // A dispatcher claims it. (Asserted against `claim_due`'s own predicate
+    // rather than by calling it: the dev DB is shared, and a real `claim_due`
+    // would claim other tests' rows out from under them.)
+    sqlx::query("UPDATE outbox SET claimed_at = NOW(), claimed_by = 'dispatcher-a' WHERE id = $1")
+        .bind(id)
+        .execute(&pool)
+        .await
+        .expect("claim");
+    assert!(
+        !claimable(&pool, id).await,
+        "a claimed row is invisible to every other dispatcher — that is the trap"
+    );
+
+    // …and then the process dies. Nothing clears the claim, so before the
+    // reclaimer existed this row would sit here forever.
+    sqlx::query("UPDATE outbox SET claimed_at = NOW() - INTERVAL '20 minutes' WHERE id = $1")
+        .bind(id)
+        .execute(&pool)
+        .await
+        .expect("backdate");
+
+    let n = repo.reclaim_stale_claims(600).await.expect("reclaim");
+    assert!(n >= 1, "the stale claim must be reclaimed");
+    assert!(
+        claimable(&pool, id).await,
+        "the reclaimed row is dispatchable again — the trigger fires after all"
+    );
+    assert_eq!(
+        attempt_count(&pool, id).await,
+        0,
+        "the handler never ran, so a reclaim must NOT consume the retry budget — \
+         otherwise repeated restarts would dead-letter an event that executed zero times"
+    );
+
+    // A FRESH claim must not be reclaimed out from under a LIVE dispatcher.
+    sqlx::query("UPDATE outbox SET claimed_at = NOW(), claimed_by = 'dispatcher-b' WHERE id = $1")
+        .bind(id)
+        .execute(&pool)
+        .await
+        .expect("re-claim");
+    assert_eq!(
+        repo.reclaim_stale_claims(600).await.expect("reclaim again"),
+        0,
+        "a claim within the timeout belongs to a live dispatcher and must be left alone"
+    );
+
+    sqlx::query("DELETE FROM apps WHERE id = $1")
+        .bind(app)
+        .execute(&pool)
+        .await
+        .expect("cleanup");
+}