fabi/PiCloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(queue): don't count never-executed transient releases against max_attempts

Browse Source 
`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) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-07-13 22:05:32 +02:00
parent bdcd3dc7f4
commit 31e6fc964d
5 changed files with 283 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
crates/manager-core/src/dispatcher.rs
View File

@@ -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<bool, String> {
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<bool, crate::group_queue_repo::GroupQueueRepoError> {
Ok(false)
}
async fn release(
&self,
_id: QueueMessageId,
_token: Uuid,
_delay: chrono::Duration,
) -> Result<bool, crate::group_queue_repo::GroupQueueRepoError> {
Ok(false)
}
#[allow(clippy::too_many_arguments)]
async fn dead_letter(
&self,
@@ -1966,6 +2002,7 @@ mod tests {
struct ClaimNackQueue {
claimed: ClaimedMessage,
nacked: Arc<AtomicBool>,
released: Arc<AtomicBool>,
}
#[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<bool, crate::queue_repo::QueueRepoError> {
self.released.store(true, Ordering::SeqCst);
Ok(true)
}
async fn reclaim_visibility_timeouts(
&self,
) -> Result<u64, crate::queue_repo::QueueRepoError> {
@@ -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<bool, crate::queue_repo::QueueRepoError> {
unimplemented!("not used by this test")
}
async fn release(
&self,
_message_id: QueueMessageId,
_claim_token: Uuid,
_retry_delay: chrono::Duration,
) -> Result<bool, crate::queue_repo::QueueRepoError> {
unimplemented!("not used by this test")
}
async fn reclaim_visibility_timeouts(
&self,
) -> Result<u64, crate::queue_repo::QueueRepoError> {

34
crates/manager-core/src/group_queue_repo.rs
View File

@@ -84,6 +84,16 @@ pub trait GroupQueueRepo: Send + Sync {
retry_delay: chrono::Duration,
) -> Result<bool, GroupQueueRepoError>;
/// 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<bool, GroupQueueRepoError>;
/// §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<bool, GroupQueueRepoError> {
// 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,

33
crates/manager-core/src/queue_repo.rs
View File

@@ -112,6 +112,18 @@ pub trait QueueRepo: Send + Sync {
retry_delay: chrono::Duration,
) -> Result<bool, QueueRepoError>;
/// 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<bool, QueueRepoError>;
/// 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<bool, QueueRepoError> {
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<u64, QueueRepoError> {
let res = sqlx::query(
"UPDATE queue_messages m \

8
crates/manager-core/src/queue_service.rs
View File

@@ -210,6 +210,14 @@ mod tests {
) -> Result<bool, crate::queue_repo::QueueRepoError> {
Ok(true)
}
async fn release(
&self,
_message_id: QueueMessageId,
_claim_token: uuid::Uuid,
_retry_delay: chrono::Duration,
) -> Result<bool, crate::queue_repo::QueueRepoError> {
Ok(true)
}
async fn reclaim_visibility_timeouts(
&self,
) -> Result<u64, crate::queue_repo::QueueRepoError> {