From 31e6fc964d2ac043a9949c7014b8efa0a7e0e68f Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Mon, 13 Jul 2026 22:05:32 +0200 Subject: [PATCH] fix(queue): don't count never-executed transient releases against max_attempts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `claim` pre-increments `attempt` before the handler runs, so a real execution failure legitimately counts toward `max_attempts`. But the two TRANSIENT release paths — gate saturation (server at capacity) and script-disabled-at- fire-time — re-queue the message WITHOUT executing it, and previously used the same `nack` that leaves the pre-increment in place. Under sustained overload a message could therefore be dead-lettered after N gate-rejections having run zero times (each rejection burning a retry). Add a non-counting `release` to both `QueueRepo` and `GroupQueueRepo` (re-queue + `attempt = GREATEST(attempt - 1, 0)`, mirroring `nack` otherwise) and route the two transient paths through a new `Dispatcher::q_release`. A real handler failure still uses `nack` and counts. No schema change. Pinned by a new `queue_release` DB test (release undoes the claim increment; nack keeps it) and the updated `disabled_queue_consumer_...` dispatcher test (now asserts a release, not a nack). Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/manager-core/src/dispatcher.rs | 74 ++++++++++- crates/manager-core/src/group_queue_repo.rs | 34 +++++ crates/manager-core/src/queue_repo.rs | 33 +++++ crates/manager-core/src/queue_service.rs | 8 ++ crates/manager-core/tests/queue_release.rs | 140 ++++++++++++++++++++ 5 files changed, 283 insertions(+), 6 deletions(-) create mode 100644 crates/manager-core/tests/queue_release.rs diff --git a/crates/manager-core/src/dispatcher.rs b/crates/manager-core/src/dispatcher.rs index 1a2caa1..077bc4f 100644 --- a/crates/manager-core/src/dispatcher.rs +++ b/crates/manager-core/src/dispatcher.rs @@ -394,6 +394,31 @@ impl Dispatcher { } } + /// TRANSIENT release — the handler never ran (gate saturated, or the script + /// was disabled at fire time). Re-queue WITHOUT counting the claim's + /// pre-increment against `max_attempts`, so overload/disable windows can't + /// dead-letter a message that executed zero times. + async fn q_release( + &self, + c: &ActiveQueueConsumer, + id: QueueMessageId, + token: uuid::Uuid, + delay: chrono::Duration, + ) -> Result { + match c.shared_group { + Some(_) => self + .group_queue + .release(id, token, delay) + .await + .map_err(|e| e.to_string()), + None => self + .queue + .release(id, token, delay) + .await + .map_err(|e| e.to_string()), + } + } + /// Terminal disposition of a message that can't be processed (script /// missing / cross-app / exhausted). Per-app → dead-letter into `dead_letters` /// (+ the caller fans out `dead_letter` triggers). Group shared queue → @@ -485,8 +510,11 @@ impl Dispatcher { // immediate (which lets the next tick re-claim). Mirrors the // outbox arm. let Ok(permit) = self.gate.try_acquire() else { + // Transient: never executed → release without burning the retry + // budget (else sustained overload could dead-letter a message that + // ran zero times). let _ = self - .q_nack( + .q_release( consumer, claimed.id, claimed.claim_token, @@ -577,7 +605,7 @@ impl Dispatcher { "queue consumer script disabled at fire time; releasing claim" ); if let Err(e) = self - .q_nack( + .q_release( consumer, claimed.id, claimed.claim_token, @@ -585,7 +613,7 @@ impl Dispatcher { ) .await { - tracing::warn!(?e, "queue nack on disabled consumer failed"); + tracing::warn!(?e, "queue release on disabled consumer failed"); } drop(permit); return Ok(()); @@ -1683,6 +1711,14 @@ mod tests { ) -> Result { Ok(false) } + async fn release( + &self, + _id: QueueMessageId, + _token: Uuid, + _delay: chrono::Duration, + ) -> Result { + Ok(false) + } #[allow(clippy::too_many_arguments)] async fn dead_letter( &self, @@ -1966,6 +2002,7 @@ mod tests { struct ClaimNackQueue { claimed: ClaimedMessage, nacked: Arc, + released: Arc, } #[async_trait] @@ -2000,6 +2037,15 @@ mod tests { self.nacked.store(true, Ordering::SeqCst); Ok(true) } + async fn release( + &self, + _message_id: QueueMessageId, + _claim_token: Uuid, + _retry_delay: chrono::Duration, + ) -> Result { + self.released.store(true, Ordering::SeqCst); + Ok(true) + } async fn reclaim_visibility_timeouts( &self, ) -> Result { @@ -2417,6 +2463,7 @@ mod tests { }; let nacked = Arc::new(AtomicBool::new(false)); + let released = Arc::new(AtomicBool::new(false)); let executed = Arc::new(AtomicBool::new(false)); let dispatcher = Dispatcher { @@ -2437,6 +2484,7 @@ mod tests { queue: Arc::new(ClaimNackQueue { claimed, nacked: nacked.clone(), + released: released.clone(), }), group_queue: Arc::new(NoopGroupQueue), config: TriggerConfig::from_env(), @@ -2447,10 +2495,16 @@ mod tests { // 1. The disabled path returns Ok(()). assert!(result.is_ok(), "dispatch_one_queue returned {result:?}"); - // 2. The claim was released via nack. + // 2. The claim was RELEASED (transient — the handler never ran), not + // nacked: a disabled-at-fire release must not burn the retry + // budget (audit fix #4). assert!( - nacked.load(Ordering::SeqCst), - "expected nack to release the claim for a disabled consumer" + released.load(Ordering::SeqCst), + "expected a transient release for a disabled consumer" + ); + assert!( + !nacked.load(Ordering::SeqCst), + "a disabled-at-fire release must NOT count as a nack (retry budget)" ); // 3. The executor was never reached. If the `if !script.enabled` // gate is deleted, the flow falls through to resolve → @@ -2533,6 +2587,14 @@ mod tests { ) -> Result { unimplemented!("not used by this test") } + async fn release( + &self, + _message_id: QueueMessageId, + _claim_token: Uuid, + _retry_delay: chrono::Duration, + ) -> Result { + unimplemented!("not used by this test") + } async fn reclaim_visibility_timeouts( &self, ) -> Result { diff --git a/crates/manager-core/src/group_queue_repo.rs b/crates/manager-core/src/group_queue_repo.rs index 5ec14bf..5e5e52c 100644 --- a/crates/manager-core/src/group_queue_repo.rs +++ b/crates/manager-core/src/group_queue_repo.rs @@ -84,6 +84,16 @@ pub trait GroupQueueRepo: Send + Sync { retry_delay: chrono::Duration, ) -> Result; + /// TRANSIENT release (handler never ran): re-queue AND undo `claim`'s + /// pre-increment of `attempt`, so a non-execution doesn't burn the retry + /// budget. Mirrors `queue_repo::release`. Floors `attempt` at 0. + async fn release( + &self, + message_id: QueueMessageId, + claim_token: Uuid, + retry_delay: chrono::Duration, + ) -> Result; + /// §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 @@ -221,6 +231,30 @@ impl GroupQueueRepo for PostgresGroupQueueRepo { Ok(res.rows_affected() == 1) } + async fn release( + &self, + message_id: QueueMessageId, + claim_token: Uuid, + retry_delay: chrono::Duration, + ) -> Result { + // TRANSIENT release (handler never ran): re-queue AND undo `claim`'s + // pre-increment of `attempt` so a non-execution doesn't burn the retry + // budget. Mirrors `queue_repo::release`. Floors at 0. + let next = Utc::now() + retry_delay; + let res = sqlx::query( + "UPDATE group_queue_messages \ + SET claim_token = NULL, claimed_at = NULL, deliver_after = $3, \ + attempt = GREATEST(attempt - 1, 0) \ + WHERE id = $1 AND claim_token = $2", + ) + .bind(message_id.into_inner()) + .bind(claim_token) + .bind(next) + .execute(&self.pool) + .await?; + Ok(res.rows_affected() == 1) + } + #[allow(clippy::too_many_arguments)] async fn dead_letter( &self, diff --git a/crates/manager-core/src/queue_repo.rs b/crates/manager-core/src/queue_repo.rs index 588bceb..78765c4 100644 --- a/crates/manager-core/src/queue_repo.rs +++ b/crates/manager-core/src/queue_repo.rs @@ -112,6 +112,18 @@ pub trait QueueRepo: Send + Sync { retry_delay: chrono::Duration, ) -> Result; + /// TRANSIENT release: the handler never ran (gate saturated, or the script + /// was disabled at fire time), so re-queue AND undo `claim`'s pre-increment + /// of `attempt` — a non-execution must not burn the retry budget (else a + /// message could be dead-lettered under sustained overload having executed + /// zero times). Floors `attempt` at 0. IFF the claim_token matches. + async fn release( + &self, + message_id: QueueMessageId, + claim_token: Uuid, + retry_delay: chrono::Duration, + ) -> Result; + /// Periodic safety net: clear claims whose `claimed_at` exceeds the /// per-queue `visibility_timeout_secs` (joined from /// `queue_trigger_details`). Returns the number of rows reclaimed. @@ -251,6 +263,27 @@ impl QueueRepo for PostgresQueueRepo { Ok(res.rows_affected() == 1) } + async fn release( + &self, + message_id: QueueMessageId, + claim_token: Uuid, + retry_delay: chrono::Duration, + ) -> Result { + let next = Utc::now() + retry_delay; + let res = sqlx::query( + "UPDATE queue_messages \ + SET claim_token = NULL, claimed_at = NULL, deliver_after = $3, \ + attempt = GREATEST(attempt - 1, 0) \ + WHERE id = $1 AND claim_token = $2", + ) + .bind(message_id.into_inner()) + .bind(claim_token) + .bind(next) + .execute(&self.pool) + .await?; + Ok(res.rows_affected() == 1) + } + async fn reclaim_visibility_timeouts(&self) -> Result { let res = sqlx::query( "UPDATE queue_messages m \ diff --git a/crates/manager-core/src/queue_service.rs b/crates/manager-core/src/queue_service.rs index 0f3eed1..ee3b205 100644 --- a/crates/manager-core/src/queue_service.rs +++ b/crates/manager-core/src/queue_service.rs @@ -210,6 +210,14 @@ mod tests { ) -> Result { Ok(true) } + async fn release( + &self, + _message_id: QueueMessageId, + _claim_token: uuid::Uuid, + _retry_delay: chrono::Duration, + ) -> Result { + Ok(true) + } async fn reclaim_visibility_timeouts( &self, ) -> Result { diff --git a/crates/manager-core/tests/queue_release.rs b/crates/manager-core/tests/queue_release.rs new file mode 100644 index 0000000..e1a6ee4 --- /dev/null +++ b/crates/manager-core/tests/queue_release.rs @@ -0,0 +1,140 @@ +//! Audit fix #4: a TRANSIENT queue release (handler never ran — gate saturated +//! or script disabled at fire time) must NOT count against `max_attempts`. +//! +//! `claim` pre-increments `attempt` before the handler runs; a real failure +//! then legitimately counts. But the gate-saturation / disabled-at-fire paths +//! re-queue WITHOUT executing, so they must undo that pre-increment via +//! `release` — otherwise sustained overload could dead-letter a message that +//! executed zero times. This pins `release` (undoes the increment) against +//! `nack` (keeps it). Skips cleanly when `DATABASE_URL` is unset. + +use picloud_manager_core::queue_repo::{PostgresQueueRepo, QueueRepo}; +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!("queue_release: DATABASE_URL unset — skipping"); + return None; + }; + let pool = PgPoolOptions::new() + .max_connections(2) + .connect(&url) + .await + .expect("connect"); + sqlx::migrate!("./migrations") + .run(&pool) + .await + .expect("migrate"); + Some(pool) +} + +async fn mk_app(pool: &PgPool, slug: &str) -> Uuid { + // apps.group_id is NOT NULL — create a root group to hang the app on. + let g: (Uuid,) = sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id") + .bind(format!("{slug}-grp")) + .fetch_one(pool) + .await + .expect("insert group"); + let r: (Uuid,) = + sqlx::query_as("INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id") + .bind(slug) + .bind(g.0) + .fetch_one(pool) + .await + .expect("insert app"); + r.0 +} + +async fn attempt_of(pool: &PgPool, id: Uuid) -> i32 { + let r: (i32,) = sqlx::query_as("SELECT attempt FROM queue_messages WHERE id = $1") + .bind(id) + .fetch_one(pool) + .await + .expect("read attempt"); + r.0 +} + +fn unique(p: &str) -> String { + // No Date/rand in scripts, but tests can use a UUID for uniqueness. + format!("{p}-{}", Uuid::new_v4().simple()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn release_undoes_the_claim_increment_but_nack_keeps_it() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let repo = PostgresQueueRepo::new(pool.clone()); + let app = mk_app(&pool, &unique("qr-app")).await; + let app_id = AppId::from(app); + let queue = "jobs"; + + // --- release: claim (attempt 0→1) then release → back to 0 ------------ + let id_rel: (Uuid,) = sqlx::query_as( + "INSERT INTO queue_messages (app_id, queue_name, payload, max_attempts) \ + VALUES ($1, $2, '{}'::jsonb, 3) RETURNING id", + ) + .bind(app) + .bind(queue) + .fetch_one(&pool) + .await + .expect("insert msg"); + let claimed = repo + .claim(app_id, queue) + .await + .expect("claim") + .expect("a message"); + assert_eq!(claimed.attempt, 1, "claim pre-increments attempt"); + assert!(repo + .release( + claimed.id, + claimed.claim_token, + chrono::Duration::milliseconds(0) + ) + .await + .expect("release")); + assert_eq!( + attempt_of(&pool, id_rel.0).await, + 0, + "a transient release must undo the claim's pre-increment" + ); + + // --- nack: claim (0→1) then nack → stays 1 (a real failure counts) ---- + let id_nack: (Uuid,) = sqlx::query_as( + "INSERT INTO queue_messages (app_id, queue_name, payload, max_attempts) \ + VALUES ($1, $2, '{}'::jsonb, 3) RETURNING id", + ) + .bind(app) + .bind("jobs2") + .fetch_one(&pool) + .await + .expect("insert msg2"); + let claimed2 = repo + .claim(app_id, "jobs2") + .await + .expect("claim2") + .expect("a message"); + assert!(repo + .nack( + claimed2.id, + claimed2.claim_token, + chrono::Duration::milliseconds(0) + ) + .await + .expect("nack")); + assert_eq!( + attempt_of(&pool, id_nack.0).await, + 1, + "a real-failure nack keeps the attempt (it counts toward max_attempts)" + ); + + // cleanup + sqlx::query("DELETE FROM apps WHERE id = $1") + .bind(app) + .execute(&pool) + .await + .expect("cleanup"); +}