From c859c9aab23067eae9ebd449107df9f3503db27f Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Tue, 14 Jul 2026 20:38:29 +0200 Subject: [PATCH] =?UTF-8?q?fix(outbox):=20reclaim=20stale=20claims=20?= =?UTF-8?q?=E2=80=94=20a=20crash=20mid-dispatch=20stranded=20events=20fore?= =?UTF-8?q?ver?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- CLAUDE.md | 1 + crates/manager-core/src/dispatcher.rs | 29 ++++ crates/manager-core/src/invoke_service.rs | 3 + crates/manager-core/src/outbox_repo.rs | 27 ++++ crates/manager-core/src/trigger_config.rs | 23 +++ crates/manager-core/tests/outbox_reclaim.rs | 153 ++++++++++++++++++++ 6 files changed, 236 insertions(+) create mode 100644 crates/manager-core/tests/outbox_reclaim.rs diff --git a/CLAUDE.md b/CLAUDE.md index c46befd..c1d7459 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -134,6 +134,7 @@ Environment variables consumed by the `picloud` binary: | `PICLOUD_DB_MAX_CONNECTIONS` | `32` | Postgres pool size. Matched to `PICLOUD_MAX_CONCURRENT_EXECUTIONS` so the data plane can't starve background workers. | | `PICLOUD_SESSION_TTL_HOURS` | `24` | Sliding-window admin session lifetime (idle timeout). | | `PICLOUD_SESSION_ABSOLUTE_TTL_HOURS` | `720` (30 days) | Absolute hard cap on an admin session's lifetime (audit 2026-07-11 C1). The sliding `touch` bump is clamped at `login_time + this`, so even a continuously-used or stolen-but-warm admin token self-expires. Mirrors the data-plane app-user session cap. | +| `PICLOUD_OUTBOX_CLAIM_TIMEOUT_SEC` | `600` (10 min) | How long a dispatcher may hold an OUTBOX claim before the reclaim task returns the row to the queue. **Without this a crash or restart mid-dispatch stranded every in-flight row permanently** — the dispatcher claims a row and only clears the claim on success (delete) or failure (reschedule), and `claim_due` selects `claimed_at IS NULL`, so a process that died in between left rows nothing would ever pick up again. Every other claim-based store (queue, group queue, workflow steps) already had a reclaimer; the outbox was the gap, and since it is the universal trigger path, the loss covered kv/docs/files/cron/pubsub/email/`invoke_async`/dead-letter alike. The default is deliberately generous: a script may run 300s, so a claim older than twice that is abandoned rather than slow. A reclaim does **not** bump `attempt_count` (the handler never ran — same reasoning as the transient queue `release`). Runs on the existing `PICLOUD_QUEUE_RECLAIM_INTERVAL_MS` ticker. | | `PICLOUD_REALTIME_BROADCAST_CAPACITY` | `64` | Per-channel SSE broadcast buffer depth (a slow consumer sees oldest events dropped). | | `PICLOUD_REALTIME_MAX_CHANNELS` | `100000` | Max live SSE channels per map (per-app and per-group). A subscribe that would open a NEW channel past this is refused with 503 (audit 2026-07-11 B6), so a client naming unbounded distinct topics can't grow the broadcaster maps to OOM. | | `PICLOUD_SANDBOX_MAX_*` | conservative defaults | Per-knob admin ceilings on Rhai sandbox overrides. See `manager-core::sandbox::SandboxCeiling`. | diff --git a/crates/manager-core/src/dispatcher.rs b/crates/manager-core/src/dispatcher.rs index 077bc4f..8f37912 100644 --- a/crates/manager-core/src/dispatcher.rs +++ b/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 { + 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 { + Ok(0) + } async fn insert( &self, _row: NewOutboxRow, diff --git a/crates/manager-core/src/invoke_service.rs b/crates/manager-core/src/invoke_service.rs index 6ab0ad2..a908d1b 100644 --- a/crates/manager-core/src/invoke_service.rs +++ b/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 { + Ok(0) + } async fn insert(&self, row: NewOutboxRow) -> Result { let id = uuid::Uuid::new_v4(); *self.last.lock().await = Some(row); diff --git a/crates/manager-core/src/outbox_repo.rs b/crates/manager-core/src/outbox_repo.rs index 7da669b..8ebd81e 100644 --- a/crates/manager-core/src/outbox_repo.rs +++ b/crates/manager-core/src/outbox_repo.rs @@ -120,6 +120,21 @@ pub trait OutboxRepo: Send + Sync { limit: i64, ) -> Result, 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; + /// 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 { + 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, diff --git a/crates/manager-core/src/trigger_config.rs b/crates/manager-core/src/trigger_config.rs index 8859afb..dcae043 100644 --- a/crates/manager-core/src/trigger_config.rs +++ b/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 } } diff --git a/crates/manager-core/tests/outbox_reclaim.rs b/crates/manager-core/tests/outbox_reclaim.rs new file mode 100644 index 0000000..5f1c9a2 --- /dev/null +++ b/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 { + 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"); +}