From fe9f28453115dc6a06d639ed865d50fb8f98fa5c Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Mon, 26 Oct 2026 01:52:28 +0100 Subject: [PATCH] sqlite ack/nack/defer only if claim still held; use shared backoff --- crates/adapters/sqlite/src/event_queue/mod.rs | 97 ++++++++++++------- .../adapters/sqlite/src/event_queue/tests.rs | 81 ++++++++++++++++ 2 files changed, 141 insertions(+), 37 deletions(-) diff --git a/crates/adapters/sqlite/src/event_queue/mod.rs b/crates/adapters/sqlite/src/event_queue/mod.rs index 302ccbee..5253f1af 100644 --- a/crates/adapters/sqlite/src/event_queue/mod.rs +++ b/crates/adapters/sqlite/src/event_queue/mod.rs @@ -137,6 +137,7 @@ struct QueueRow { event_id: Option, attempts: i32, next_attempt_at: String, + claimed_at: Option, } /// Claims a batch in one atomic statement. A read-then-write transaction can @@ -159,7 +160,7 @@ async fn claim_batch( ORDER BY next_attempt_at ASC LIMIT ? ) - RETURNING id, payload, event_id, attempts, next_attempt_at" + RETURNING id, payload, event_id, attempts, next_attempt_at, claimed_at" ); let mut rows = sqlx::query_as::<_, QueueRow>(&sql) .bind(config.batch_size) @@ -212,6 +213,7 @@ fn decode_row( row_id: row.id, attempts: row.attempts, max_attempts, + claimed_at: row.claimed_at, }), )) } @@ -221,43 +223,74 @@ struct DbAckHandle { row_id: i64, attempts: i32, max_attempts: i32, + /// The claim this handle belongs to. Settling is conditional on it, so a + /// worker whose lease lapsed and was reclaimed cannot flip the new claim's row. + claimed_at: Option, +} + +impl DbAckHandle { + /// Runs a settle statement whose last two binds are the row id and claim + /// stamp. Zero rows changed means the claim is no longer ours. + async fn settle<'q>( + &self, + what: &str, + query: sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'q>>, + ) -> Result<(), DomainError> { + let done = query + .bind(self.row_id) + .bind(self.claimed_at.clone()) + .execute(&self.pool) + .await + .map_err(|e| DomainError::InfrastructureError(format!("{what}: {e}")))?; + if done.rows_affected() == 0 { + tracing::warn!( + row_id = self.row_id, + "{what}: claim no longer held (lease lapsed and row reclaimed); ignoring" + ); + } + Ok(()) + } } #[async_trait] impl AckHandle for DbAckHandle { async fn ack(&self) -> Result<(), DomainError> { - sqlx::query("UPDATE event_queue SET status = 'done' WHERE id = ?") - .bind(self.row_id) - .execute(&self.pool) - .await - .map_err(|e| DomainError::InfrastructureError(format!("ack: {e}")))?; - Ok(()) + self.settle( + "ack", + sqlx::query( + "UPDATE event_queue SET status = 'done' \ + WHERE id = ? AND status = 'processing' AND claimed_at IS ?", + ), + ) + .await } async fn nack(&self) -> Result<(), DomainError> { let new_attempts = self.attempts + 1; if new_attempts >= self.max_attempts { - sqlx::query( - "UPDATE event_queue SET status = 'dead_lettered', attempts = ?, last_error = 'max attempts reached' WHERE id = ?" - ) - .bind(new_attempts) - .bind(self.row_id) - .execute(&self.pool) - .await - .map_err(|e| DomainError::InfrastructureError(format!("nack dead-letter: {e}")))?; - } else { - let backoff = backoff_seconds(new_attempts); - let sql = format!( - "UPDATE event_queue SET status = 'pending', attempts = ?, next_attempt_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now', '+{backoff} seconds'), last_error = 'nack' WHERE id = ?" + tracing::error!( + row_id = self.row_id, + attempts = new_attempts, + "event exhausted max attempts, dead-lettering" ); - sqlx::query(&sql) - .bind(new_attempts) - .bind(self.row_id) - .execute(&self.pool) + self.settle( + "nack dead-letter", + sqlx::query( + "UPDATE event_queue SET status = 'dead_lettered', attempts = ?, last_error = 'max attempts reached' \ + WHERE id = ? AND status = 'processing' AND claimed_at IS ?", + ) + .bind(new_attempts), + ) + .await + } else { + let backoff = domain::events::retry_backoff(new_attempts.max(0) as u32).as_secs(); + let sql = format!( + "UPDATE event_queue SET status = 'pending', attempts = ?, next_attempt_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now', '+{backoff} seconds'), last_error = 'nack' \ + WHERE id = ? AND status = 'processing' AND claimed_at IS ?" + ); + self.settle("nack retry", sqlx::query(&sql).bind(new_attempts)) .await - .map_err(|e| DomainError::InfrastructureError(format!("nack retry: {e}")))?; } - Ok(()) } async fn defer(&self, delay: std::time::Duration) -> Result<(), DomainError> { @@ -266,21 +299,11 @@ impl AckHandle for DbAckHandle { "UPDATE event_queue SET status = 'pending', claimed_at = NULL, last_error = 'deferred', next_attempt_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now', '+{secs} seconds') - WHERE id = ?" + WHERE id = ? AND status = 'processing' AND claimed_at IS ?" ); - sqlx::query(&sql) - .bind(self.row_id) - .execute(&self.pool) - .await - .map_err(|e| DomainError::InfrastructureError(format!("defer: {e}")))?; - Ok(()) + self.settle("defer", sqlx::query(&sql)).await } } -fn backoff_seconds(attempts: i32) -> i64 { - let base: i64 = 5 * (1i64 << attempts.min(6)); - base.min(300) -} - #[cfg(test)] mod tests; diff --git a/crates/adapters/sqlite/src/event_queue/tests.rs b/crates/adapters/sqlite/src/event_queue/tests.rs index 7978df3c..94f86cd9 100644 --- a/crates/adapters/sqlite/src/event_queue/tests.rs +++ b/crates/adapters/sqlite/src/event_queue/tests.rs @@ -303,3 +303,84 @@ mod lease { assert!(error.unwrap().contains("undecodable")); } } + +mod stale_handle { + use domain::{ + events::{DomainEvent, EventEnvelope}, + ports::{EventConsumer, EventPublisher}, + value_objects::UserId, + }; + use futures::StreamExt; + + use super::{claim_batch, config}; + use crate::event_queue::{CLAIM_LEASE_SECONDS, SqliteEventQueue}; + + /// Claims a row, lets its lease lapse, and has another worker reclaim it. + /// Returns the first (now stale) worker's envelope. + async fn reclaimed() -> (sqlx::SqlitePool, EventEnvelope) { + let pool = sqlx::sqlite::SqlitePoolOptions::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .unwrap(); + crate::migrate(&pool).await.unwrap(); + let q = SqliteEventQueue::create(pool.clone(), config(10)); + q.publish(&DomainEvent::UserUpdated { + user_id: UserId::generate(), + }) + .await + .unwrap(); + let stale = q.consume().next().await.unwrap().unwrap(); + let old = (chrono::Utc::now() - chrono::Duration::seconds(CLAIM_LEASE_SECONDS + 60)) + .format("%Y-%m-%dT%H:%M:%SZ") + .to_string(); + sqlx::query("UPDATE event_queue SET claimed_at = ?") + .bind(old) + .execute(&pool) + .await + .unwrap(); + assert_eq!(claim_batch(&pool, &config(10)).await.unwrap().len(), 1); + // Real reclaims happen a lease later, so their timestamp differs from + // the stale handle's; the claim clock has only second resolution. + let second_claim = (chrono::Utc::now() - chrono::Duration::seconds(5)) + .format("%Y-%m-%dT%H:%M:%SZ") + .to_string(); + sqlx::query("UPDATE event_queue SET claimed_at = ?") + .bind(second_claim) + .execute(&pool) + .await + .unwrap(); + (pool, stale) + } + + async fn state(pool: &sqlx::SqlitePool) -> (String, i32) { + sqlx::query_as("SELECT status, attempts FROM event_queue") + .fetch_one(pool) + .await + .unwrap() + } + + #[tokio::test] + async fn a_stale_ack_cannot_complete_a_reclaimed_row() { + let (pool, stale) = reclaimed().await; + stale.ack().await.unwrap(); + assert_eq!(state(&pool).await, ("processing".into(), 0)); + } + + #[tokio::test] + async fn a_stale_nack_cannot_spend_an_attempt_on_a_reclaimed_row() { + let (pool, stale) = reclaimed().await; + stale.nack().await.unwrap(); + assert_eq!(state(&pool).await, ("processing".into(), 0)); + } + + #[tokio::test] + async fn a_stale_defer_cannot_release_a_reclaimed_row() { + let (pool, stale) = reclaimed().await; + stale + .defer(std::time::Duration::from_secs(60)) + .await + .unwrap(); + assert_eq!(state(&pool).await, ("processing".into(), 0)); + } +}