From 0236a488fadd61a8082c853d4a52cc85f9be52b8 Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Mon, 26 Oct 2026 00:33:29 +0100 Subject: [PATCH] fix: defer blocked deliveries, claim owner token --- crates/adapters/event-publisher/src/lib.rs | 3 + crates/adapters/nats/src/consumer.rs | 13 ++ crates/adapters/sqlite/src/event_queue/mod.rs | 16 ++ .../adapters/sqlite/src/event_queue/tests.rs | 27 +++ .../src/repositories/processed_events/mod.rs | 176 +++++++++++++----- crates/application/src/worker.rs | 37 ++-- .../jobs/processed_events_cleanup_test.rs | 5 +- .../application/tests/worker/worker_test.rs | 109 ++++++++++- .../bootstrap/tests/ledger_recovery_test.rs | 105 +++++++++++ crates/domain/src/events.rs | 9 + crates/domain/src/ports/processed_events.rs | 45 ++++- crates/domain/src/testing/fakes/noops.rs | 3 + .../testing/fakes/store_processed_events.rs | 60 ++++-- 13 files changed, 528 insertions(+), 80 deletions(-) create mode 100644 crates/bootstrap/tests/ledger_recovery_test.rs diff --git a/crates/adapters/event-publisher/src/lib.rs b/crates/adapters/event-publisher/src/lib.rs index 8c68fc40..e3c6289b 100644 --- a/crates/adapters/event-publisher/src/lib.rs +++ b/crates/adapters/event-publisher/src/lib.rs @@ -47,6 +47,9 @@ impl AckHandle for NoopAck { async fn nack(&self) -> Result<(), DomainError> { Ok(()) } + async fn defer(&self, _: std::time::Duration) -> Result<(), DomainError> { + Ok(()) + } } pub struct ChannelEventConsumer { diff --git a/crates/adapters/nats/src/consumer.rs b/crates/adapters/nats/src/consumer.rs index fcea56fb..d26b2156 100644 --- a/crates/adapters/nats/src/consumer.rs +++ b/crates/adapters/nats/src/consumer.rs @@ -48,6 +48,16 @@ impl AckHandle for NatsJetStreamAckHandle { .await .map_err(|e| DomainError::InfrastructureError(e.to_string())) } + + /// JetStream counts every redelivery toward `max_deliver`, deferrals + /// included. The delay is at least the remaining claim lease, so the + /// deliveries allowed by `max_deliver` still span more than one lease. + async fn defer(&self, delay: std::time::Duration) -> Result<(), DomainError> { + self.message + .ack_with(AckKind::Nak(Some(delay))) + .await + .map_err(|e| DomainError::InfrastructureError(e.to_string())) + } } // ── Core NATS ack handle (no-op) ───────────────────────────────────────────── @@ -62,6 +72,9 @@ impl AckHandle for NoopAck { async fn nack(&self) -> Result<(), DomainError> { Ok(()) } + async fn defer(&self, _: std::time::Duration) -> Result<(), DomainError> { + Ok(()) + } } // ── Envelope construction helpers ──────────────────────────────────────────── diff --git a/crates/adapters/sqlite/src/event_queue/mod.rs b/crates/adapters/sqlite/src/event_queue/mod.rs index f672216e..302ccbee 100644 --- a/crates/adapters/sqlite/src/event_queue/mod.rs +++ b/crates/adapters/sqlite/src/event_queue/mod.rs @@ -259,6 +259,22 @@ impl AckHandle for DbAckHandle { } Ok(()) } + + async fn defer(&self, delay: std::time::Duration) -> Result<(), DomainError> { + let secs = delay.as_secs_f64().ceil().max(1.0) as i64; + let sql = format!( + "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 = ?" + ); + sqlx::query(&sql) + .bind(self.row_id) + .execute(&self.pool) + .await + .map_err(|e| DomainError::InfrastructureError(format!("defer: {e}")))?; + Ok(()) + } } fn backoff_seconds(attempts: i32) -> i64 { diff --git a/crates/adapters/sqlite/src/event_queue/tests.rs b/crates/adapters/sqlite/src/event_queue/tests.rs index 8025b5f6..7978df3c 100644 --- a/crates/adapters/sqlite/src/event_queue/tests.rs +++ b/crates/adapters/sqlite/src/event_queue/tests.rs @@ -189,6 +189,33 @@ mod identity { assert!(envelope.id().is_none()); } + + #[tokio::test] + async fn defer_reschedules_without_counting_an_attempt() { + let (q, pool) = queue().await; + q.publish(&event()).await.unwrap(); + let envelope = q.consume().next().await.unwrap().unwrap(); + + envelope + .defer(std::time::Duration::from_secs(600)) + .await + .unwrap(); + + let (status, attempts, claimed, due_in): (String, i32, Option, i64) = + sqlx::query_as( + "SELECT status, attempts, claimed_at, + CAST(strftime('%s', next_attempt_at) AS INTEGER) + - CAST(strftime('%s', 'now') AS INTEGER) + FROM event_queue", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(status, "pending"); + assert_eq!(attempts, 0, "a deferral is not a failed attempt"); + assert!(claimed.is_none(), "claim is released"); + assert!((595..=600).contains(&due_in), "due in {due_in}s"); + } } mod lease { diff --git a/crates/adapters/sqlite/src/repositories/processed_events/mod.rs b/crates/adapters/sqlite/src/repositories/processed_events/mod.rs index 453707e9..756b46b9 100644 --- a/crates/adapters/sqlite/src/repositories/processed_events/mod.rs +++ b/crates/adapters/sqlite/src/repositories/processed_events/mod.rs @@ -2,20 +2,34 @@ use async_trait::async_trait; use domain::{ errors::DomainError, events::EventId, - ports::{Begin, HANDLER_LEASE_SECONDS, ProcessedEventLedger}, + ports::{Begin, ClaimToken, HANDLER_LEASE_SECONDS, ProcessedEventLedger}, }; use sqlx::SqlitePool; /// Matches the column default in migration 0101, so string comparison orders rows by time. const TIMESTAMP_FORMAT: &str = "%Y-%m-%dT%H:%M:%SZ"; +const CLAIM_FORMAT: &str = "%Y-%m-%dT%H:%M:%S%.3fZ"; pub struct SqliteProcessedEventLedger { pool: SqlitePool, + lease: chrono::Duration, } impl SqliteProcessedEventLedger { pub fn new(pool: SqlitePool) -> Self { - Self { pool } + Self::with_lease( + pool, + std::time::Duration::from_secs(HANDLER_LEASE_SECONDS as u64), + ) + } + + /// A shorter lease than production's, so lease takeover can be tested + /// without waiting half an hour. + pub fn with_lease(pool: SqlitePool, lease: std::time::Duration) -> Self { + Self { + pool, + lease: chrono::Duration::from_std(lease).expect("lease fits a chrono duration"), + } } } @@ -23,74 +37,110 @@ fn stamp(at: chrono::NaiveDateTime) -> String { at.format(TIMESTAMP_FORMAT).to_string() } +/// Millisecond precision, so two claims taken in one second have distinct tokens. +fn claim_stamp(at: chrono::NaiveDateTime) -> String { + at.format(CLAIM_FORMAT).to_string() +} + +fn parse_claim_stamp(value: &str) -> Option { + chrono::NaiveDateTime::parse_from_str(value, CLAIM_FORMAT) + .or_else(|_| chrono::NaiveDateTime::parse_from_str(value, TIMESTAMP_FORMAT)) + .ok() +} + #[async_trait] impl ProcessedEventLedger for SqliteProcessedEventLedger { async fn begin(&self, id: &EventId, handler: &str) -> Result { - let now = chrono::Utc::now().naive_utc(); - let stale_before = now - chrono::Duration::seconds(HANDLER_LEASE_SECONDS); // Retried once: the row can vanish (released) between the claim attempt // and the state lookup. for _ in 0..2 { + let now = chrono::Utc::now().naive_utc(); + let started = claim_stamp(now); // One statement, so two workers cannot both win: insert if absent, // or take over an in_progress claim past its lease. let claimed = sqlx::query( "INSERT INTO processed_events (event_id, handler, status, started_at, processed_at) VALUES (?, ?, 'in_progress', ?, ?) - ON CONFLICT (event_id, handler) DO UPDATE SET started_at = excluded.started_at + ON CONFLICT (event_id, handler) DO UPDATE + SET started_at = excluded.started_at, processed_at = excluded.processed_at WHERE processed_events.status = 'in_progress' AND processed_events.started_at < ?", ) .bind(id.value().to_string()) .bind(handler) + .bind(&started) .bind(stamp(now)) - .bind(stamp(now)) - .bind(stamp(stale_before)) + .bind(claim_stamp(now - self.lease)) .execute(&self.pool) .await .map_err(crate::shared::map_sqlx_error)? .rows_affected(); if claimed == 1 { - return Ok(Begin::Run); + return Ok(Begin::Run(ClaimToken::new(started))); } - let status: Option = sqlx::query_scalar( - "SELECT status FROM processed_events WHERE event_id = ? AND handler = ?", + let row: Option<(String, Option)> = sqlx::query_as( + "SELECT status, started_at FROM processed_events WHERE event_id = ? AND handler = ?", ) .bind(id.value().to_string()) .bind(handler) .fetch_optional(&self.pool) .await .map_err(crate::shared::map_sqlx_error)?; - match status.as_deref() { - Some("done") => return Ok(Begin::AlreadyDone), - Some(_) => return Ok(Begin::InProgressElsewhere), + match row { + Some((status, _)) if status == "done" => return Ok(Begin::AlreadyDone), + Some((_, started_at)) => { + let lease_end = started_at + .as_deref() + .and_then(parse_claim_stamp) + .map_or(now + self.lease, |s| s + self.lease); + let retry_after = (lease_end - now) + .to_std() + .unwrap_or_default() + .max(std::time::Duration::from_secs(1)); + return Ok(Begin::InProgressElsewhere { retry_after }); + } None => continue, } } - Ok(Begin::InProgressElsewhere) + Ok(Begin::InProgressElsewhere { + retry_after: std::time::Duration::from_secs(1), + }) } - async fn complete(&self, id: &EventId, handler: &str) -> Result<(), DomainError> { + async fn complete( + &self, + id: &EventId, + handler: &str, + token: &ClaimToken, + ) -> Result<(), DomainError> { sqlx::query( "UPDATE processed_events SET status = 'done', processed_at = ? - WHERE event_id = ? AND handler = ?", + WHERE event_id = ? AND handler = ? AND status = 'in_progress' AND started_at = ?", ) .bind(stamp(chrono::Utc::now().naive_utc())) .bind(id.value().to_string()) .bind(handler) + .bind(token.as_str()) .execute(&self.pool) .await .map_err(crate::shared::map_sqlx_error)?; Ok(()) } - async fn release(&self, id: &EventId, handler: &str) -> Result<(), DomainError> { + async fn release( + &self, + id: &EventId, + handler: &str, + token: &ClaimToken, + ) -> Result<(), DomainError> { sqlx::query( "DELETE FROM processed_events - WHERE event_id = ? AND handler = ? AND status = 'in_progress'", + WHERE event_id = ? AND handler = ? AND status = 'in_progress' AND started_at = ?", ) .bind(id.value().to_string()) .bind(handler) + .bind(token.as_str()) .execute(&self.pool) .await .map_err(crate::shared::map_sqlx_error)?; @@ -111,13 +161,18 @@ impl ProcessedEventLedger for SqliteProcessedEventLedger { mod tests { use super::*; - async fn ledger() -> (SqliteProcessedEventLedger, SqlitePool) { + async fn memory_pool() -> SqlitePool { let pool = sqlx::sqlite::SqlitePoolOptions::new() .max_connections(1) .connect("sqlite::memory:") .await .unwrap(); crate::migrate(&pool).await.unwrap(); + pool + } + + async fn ledger() -> (SqliteProcessedEventLedger, SqlitePool) { + let pool = memory_pool().await; (SqliteProcessedEventLedger::new(pool.clone()), pool) } @@ -135,30 +190,42 @@ mod tests { .unwrap(); } + fn token(begin: Begin) -> ClaimToken { + match begin { + Begin::Run(token) => token, + other => panic!("expected Run, got {other:?}"), + } + } + + fn blocked(begin: Begin) -> std::time::Duration { + match begin { + Begin::InProgressElsewhere { retry_after } => retry_after, + other => panic!("expected InProgressElsewhere, got {other:?}"), + } + } + #[tokio::test] async fn the_first_begin_runs_and_a_second_sees_it_in_progress() { let (ledger, _) = ledger().await; let id = EventId::new(); - assert_eq!(ledger.begin(&id, "a").await.unwrap(), Begin::Run); - assert_eq!( - ledger.begin(&id, "a").await.unwrap(), - Begin::InProgressElsewhere - ); - assert_eq!(ledger.begin(&id, "b").await.unwrap(), Begin::Run); - assert_eq!( - ledger.begin(&EventId::new(), "a").await.unwrap(), - Begin::Run - ); + token(ledger.begin(&id, "a").await.unwrap()); + let retry_after = blocked(ledger.begin(&id, "a").await.unwrap()); + token(ledger.begin(&id, "b").await.unwrap()); + token(ledger.begin(&EventId::new(), "a").await.unwrap()); + + // The blocked delivery is told to come back when the lease runs out. + let lease = std::time::Duration::from_secs(HANDLER_LEASE_SECONDS as u64); + assert!(retry_after <= lease && retry_after > lease - std::time::Duration::from_secs(5)); } #[tokio::test] async fn a_completed_claim_is_already_done() { let (ledger, _) = ledger().await; let id = EventId::new(); - ledger.begin(&id, "a").await.unwrap(); + let t = token(ledger.begin(&id, "a").await.unwrap()); - ledger.complete(&id, "a").await.unwrap(); + ledger.complete(&id, "a", &t).await.unwrap(); assert_eq!(ledger.begin(&id, "a").await.unwrap(), Begin::AlreadyDone); } @@ -167,21 +234,21 @@ mod tests { async fn a_released_claim_can_run_again() { let (ledger, _) = ledger().await; let id = EventId::new(); - ledger.begin(&id, "a").await.unwrap(); + let t = token(ledger.begin(&id, "a").await.unwrap()); - ledger.release(&id, "a").await.unwrap(); + ledger.release(&id, "a", &t).await.unwrap(); - assert_eq!(ledger.begin(&id, "a").await.unwrap(), Begin::Run); + token(ledger.begin(&id, "a").await.unwrap()); } #[tokio::test] async fn release_never_removes_a_done_claim() { let (ledger, _) = ledger().await; let id = EventId::new(); - ledger.begin(&id, "a").await.unwrap(); - ledger.complete(&id, "a").await.unwrap(); + let t = token(ledger.begin(&id, "a").await.unwrap()); + ledger.complete(&id, "a", &t).await.unwrap(); - ledger.release(&id, "a").await.unwrap(); + ledger.release(&id, "a", &t).await.unwrap(); assert_eq!(ledger.begin(&id, "a").await.unwrap(), Begin::AlreadyDone); } @@ -192,11 +259,16 @@ mod tests { let id = EventId::new(); insert(&pool, &id, "in_progress", "2020-01-01T00:00:00Z").await; - assert_eq!(ledger.begin(&id, "h").await.unwrap(), Begin::Run); + token(ledger.begin(&id, "h").await.unwrap()); // The takeover refreshed the lease, so a third delivery is blocked. - assert_eq!( - ledger.begin(&id, "h").await.unwrap(), - Begin::InProgressElsewhere + blocked(ledger.begin(&id, "h").await.unwrap()); + let processed_at: String = sqlx::query_scalar("SELECT processed_at FROM processed_events") + .fetch_one(&pool) + .await + .unwrap(); + assert!( + processed_at.as_str() > "2020-01-01T00:00:00Z", + "takeover refreshes processed_at" ); } @@ -209,6 +281,26 @@ mod tests { assert_eq!(ledger.begin(&id, "h").await.unwrap(), Begin::AlreadyDone); } + #[tokio::test] + async fn a_lapsed_owner_cannot_release_a_claim_taken_over_by_another() { + let pool = memory_pool().await; + let ledger = + SqliteProcessedEventLedger::with_lease(pool.clone(), std::time::Duration::from_secs(1)); + let id = EventId::new(); + let lapsed = token(ledger.begin(&id, "a").await.unwrap()); + tokio::time::sleep(std::time::Duration::from_millis(2100)).await; + let owner = token(ledger.begin(&id, "a").await.unwrap()); + assert_ne!(lapsed, owner); + + ledger.release(&id, "a", &lapsed).await.unwrap(); + ledger.complete(&id, "a", &lapsed).await.unwrap(); + + // Still the new owner's live claim: neither call touched it. + blocked(ledger.begin(&id, "a").await.unwrap()); + ledger.complete(&id, "a", &owner).await.unwrap(); + assert_eq!(ledger.begin(&id, "a").await.unwrap(), Begin::AlreadyDone); + } + #[tokio::test] async fn concurrent_begins_have_exactly_one_winner() { let dir = tempfile::tempdir().unwrap(); @@ -234,7 +326,7 @@ mod tests { .collect(); let mut runs = 0; for t in tasks { - if t.await.unwrap() == Begin::Run { + if matches!(t.await.unwrap(), Begin::Run(_)) { runs += 1; } } diff --git a/crates/application/src/worker.rs b/crates/application/src/worker.rs index 3a768e9b..f01d64b9 100644 --- a/crates/application/src/worker.rs +++ b/crates/application/src/worker.rs @@ -1,4 +1,5 @@ use std::sync::Arc; +use std::time::Duration; use domain::{ errors::DomainError, @@ -88,6 +89,10 @@ impl WorkerService { } } +/// Margin added to a claim's remaining lease before a blocked delivery is +/// retried, so the retry lands after the lease has lapsed rather than on it. +const DEFER_MARGIN: Duration = Duration::from_secs(5); + async fn dispatch( handlers: Arc>>, ledger: Arc, @@ -95,21 +100,21 @@ async fn dispatch( ) { let id = envelope.id().copied(); let mut any_transient = false; + // Longest wait among handlers blocked behind another delivery's live claim. + let mut blocked_for: Option = None; for handler in handlers.iter() { let name = handler.name(); + let mut claim = None; if let Some(id) = &id { match ledger.begin(id, name).await { - Ok(Begin::Run) => {} + Ok(Begin::Run(token)) => claim = Some(token), Ok(Begin::AlreadyDone) => { tracing::debug!(handler = name, "already processed, skipping"); continue; } - Ok(Begin::InProgressElsewhere) => { - tracing::debug!( - handler = name, - "running in another delivery, retrying later" - ); - any_transient = true; + Ok(Begin::InProgressElsewhere { retry_after }) => { + tracing::debug!(handler = name, "running in another delivery, deferring"); + blocked_for = Some(blocked_for.map_or(retry_after, |b| b.max(retry_after))); continue; } Err(e) => { @@ -123,8 +128,8 @@ async fn dispatch( let result = handler.handle(&envelope.event).await; match &result { Ok(()) => { - if let Some(id) = &id - && let Err(e) = ledger.complete(id, name).await + if let (Some(id), Some(token)) = (&id, &claim) + && let Err(e) = ledger.complete(id, name, token).await { // The work is done; failing the delivery would only repeat it. // The claim lapses after its lease and the handler may run once more. @@ -135,8 +140,8 @@ async fn dispatch( // Permanent failures are released too, not recorded: if a sibling // handler's transient failure redelivers this event, a handler that // failed permanently runs (and fails) again. - if let Some(id) = &id - && let Err(le) = ledger.release(id, name).await + if let (Some(id), Some(token)) = (&id, &claim) + && let Err(le) = ledger.release(id, name, token).await { tracing::warn!(handler = name, "releasing ledger claim failed: {le}"); } @@ -152,7 +157,15 @@ async fn dispatch( } } } - if any_transient { + // A blocked handler cannot be retried before its claim is done or lapsed, + // so wait that long without spending a retry attempt; any transient failure + // in the same delivery is retried then too. A plain nack would burn the + // retry budget (and dead-letter) long before a crashed claim's lease ends. + if let Some(wait) = blocked_for { + if let Err(e) = envelope.defer(wait + DEFER_MARGIN).await { + tracing::error!("defer failed: {e}"); + } + } else if any_transient { if let Err(e) = envelope.nack().await { tracing::error!("nack failed: {e}"); } diff --git a/crates/application/tests/jobs/processed_events_cleanup_test.rs b/crates/application/tests/jobs/processed_events_cleanup_test.rs index 8ce87ede..cc685db3 100644 --- a/crates/application/tests/jobs/processed_events_cleanup_test.rs +++ b/crates/application/tests/jobs/processed_events_cleanup_test.rs @@ -26,7 +26,10 @@ async fn removes_only_rows_older_than_the_retention_window() { .unwrap(); use domain::ports::{Begin, ProcessedEventLedger}; - assert_eq!(store.begin(&stale, "h").await.unwrap(), Begin::Run); + assert!(matches!( + store.begin(&stale, "h").await.unwrap(), + Begin::Run(_) + )); assert_eq!(store.begin(&fresh, "h").await.unwrap(), Begin::AlreadyDone); } diff --git a/crates/application/tests/worker/worker_test.rs b/crates/application/tests/worker/worker_test.rs index 2d19f26a..34a24e64 100644 --- a/crates/application/tests/worker/worker_test.rs +++ b/crates/application/tests/worker/worker_test.rs @@ -28,6 +28,9 @@ impl AckHandle for TrackingAck { *self.nacked.lock().unwrap() = true; Ok(()) } + async fn defer(&self, _: std::time::Duration) -> Result<(), DomainError> { + Ok(()) + } } struct VecConsumer { @@ -380,9 +383,10 @@ struct RedeliveringConsumer { event: DomainEvent, times: usize, nacks: Arc>, + defers: Arc>>, } -struct CountNack(Arc>); +struct CountNack(Arc>, Arc>>); #[async_trait] impl AckHandle for CountNack { @@ -393,16 +397,24 @@ impl AckHandle for CountNack { *self.0.lock().unwrap() += 1; Ok(()) } + async fn defer(&self, delay: std::time::Duration) -> Result<(), DomainError> { + self.1.lock().unwrap().push(delay); + Ok(()) + } } impl EventConsumer for RedeliveringConsumer { fn consume(&self) -> BoxStream<'_, Result> { let nacks = Arc::clone(&self.nacks); + let defers = Arc::clone(&self.defers); let id = self.id; let event = self.event.clone(); Box::pin(stream::iter(0..self.times).then(move |_| { - let envelope = - EventEnvelope::new(id, event.clone(), Box::new(CountNack(Arc::clone(&nacks)))); + let envelope = EventEnvelope::new( + id, + event.clone(), + Box::new(CountNack(Arc::clone(&nacks), Arc::clone(&defers))), + ); async move { // Let the previous delivery's handlers finish first: a real // broker redelivers only after the nack. @@ -423,6 +435,7 @@ async fn a_redelivered_event_skips_handlers_that_already_succeeded() { event: movie_discovered(), times: 2, nacks: Arc::clone(&nacks), + defers: Default::default(), }; WorkerService::new( @@ -455,6 +468,7 @@ async fn events_without_id_run_every_time() { event: movie_discovered(), times: 2, nacks: Arc::new(Mutex::new(0)), + defers: Default::default(), }; WorkerService::new(Arc::new(consumer), vec![handler], InMemoryStore::new()) @@ -485,13 +499,18 @@ async fn a_handler_already_running_elsewhere_is_not_run_concurrently() { let store = InMemoryStore::new(); let id = domain::events::EventId::new(); // Another delivery holds a live claim on this (event, handler). - assert_eq!(store.begin(&id, "slow").await.unwrap(), Begin::Run); + assert!(matches!( + store.begin(&id, "slow").await.unwrap(), + Begin::Run(_) + )); let nacks = Arc::new(Mutex::new(0)); + let defers: Arc>> = Default::default(); let consumer = RedeliveringConsumer { id: Some(id), event: movie_discovered(), times: 1, nacks: Arc::clone(&nacks), + defers: Arc::clone(&defers), }; WorkerService::new(Arc::new(consumer), vec![handler], store) @@ -504,7 +523,85 @@ async fn a_handler_already_running_elsewhere_is_not_run_concurrently() { 0, "must not run beside the claim holder" ); - assert_eq!(*nacks.lock().unwrap(), 1, "delivery is retried later"); + assert_eq!( + *nacks.lock().unwrap(), + 0, + "a blocked delivery is not a failure" + ); + assert_eq!(defers.lock().unwrap().len(), 1, "delivery is retried later"); +} + +#[tokio::test] +async fn a_delivery_blocked_by_a_live_claim_is_deferred_without_consuming_an_attempt() { + use domain::ports::{Begin, HANDLER_LEASE_SECONDS, ProcessedEventLedger}; + let (handler, calls) = CountingHandler::new("slow", 0); + let store = InMemoryStore::new(); + let id = domain::events::EventId::new(); + assert!(matches!( + store.begin(&id, "slow").await.unwrap(), + Begin::Run(_) + )); + let nacks = Arc::new(Mutex::new(0)); + let defers: Arc>> = Default::default(); + let consumer = RedeliveringConsumer { + id: Some(id), + event: movie_discovered(), + times: 1, + nacks: Arc::clone(&nacks), + defers: Arc::clone(&defers), + }; + + WorkerService::new(Arc::new(consumer), vec![handler], store) + .unwrap() + .run(tokio::sync::watch::channel(false).1) + .await; + + assert_eq!(*calls.lock().unwrap(), 0); + assert_eq!(*nacks.lock().unwrap(), 0, "a nack would consume an attempt"); + let defers = defers.lock().unwrap(); + assert_eq!(defers.len(), 1); + // Retried only once the claim's lease has run out (plus a margin). + let lease = std::time::Duration::from_secs(HANDLER_LEASE_SECONDS as u64); + assert!( + defers[0] > lease - std::time::Duration::from_secs(5), + "{:?}", + defers[0] + ); + assert!( + defers[0] <= lease + std::time::Duration::from_secs(6), + "{:?}", + defers[0] + ); +} + +#[tokio::test] +async fn a_lapsed_owner_cannot_release_a_claim_taken_over_by_another() { + use domain::ports::{Begin, HANDLER_LEASE_SECONDS, ProcessedEventLedger}; + let store = InMemoryStore::new(); + let id = domain::events::EventId::new(); + store.seed_processed_event( + &id, + "h", + false, + chrono::Utc::now().naive_utc() - chrono::Duration::seconds(HANDLER_LEASE_SECONDS + 1), + ); + let Begin::Run(lapsed) = store.begin(&id, "h").await.unwrap() else { + panic!("stale claim should be taken over"); + }; + // Simulate: the original owner's token is from before the takeover. + let original = domain::ports::ClaimToken::new("2020-01-01T00:00:00.000Z".into()); + assert_ne!(original, lapsed); + + store.release(&id, "h", &original).await.unwrap(); + store.complete(&id, "h", &original).await.unwrap(); + + assert!( + matches!( + store.begin(&id, "h").await.unwrap(), + Begin::InProgressElsewhere { .. } + ), + "the new owner's claim must survive the old owner's release and complete" + ); } #[tokio::test] @@ -525,6 +622,7 @@ async fn a_stale_in_progress_claim_is_taken_over() { event: movie_discovered(), times: 1, nacks: Arc::clone(&nacks), + defers: Default::default(), }; WorkerService::new(Arc::new(consumer), vec![handler], store) @@ -546,6 +644,7 @@ async fn a_failed_handler_releases_its_claim_so_redelivery_retries() { event: movie_discovered(), times: 3, nacks: Arc::clone(&nacks), + defers: Default::default(), }; WorkerService::new(Arc::new(consumer), vec![handler], store.clone()) diff --git a/crates/bootstrap/tests/ledger_recovery_test.rs b/crates/bootstrap/tests/ledger_recovery_test.rs new file mode 100644 index 00000000..72e02ac8 --- /dev/null +++ b/crates/bootstrap/tests/ledger_recovery_test.rs @@ -0,0 +1,105 @@ +//! Crash recovery through the real worker dispatch and the real sqlite ledger. + +mod support; + +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use application::worker::WorkerService; +use async_trait::async_trait; +use domain::{ + errors::DomainError, + events::{AckHandle, DomainEvent, EventEnvelope, EventId}, + ports::{Begin, EventConsumer, EventHandler, ProcessedEventLedger}, +}; +use futures::stream::BoxStream; +use sqlite::SqliteProcessedEventLedger; + +struct Counting(Arc); + +#[async_trait] +impl EventHandler for Counting { + fn name(&self) -> &'static str { + "crashy" + } + async fn handle(&self, _: &DomainEvent) -> Result<(), DomainError> { + self.0.fetch_add(1, Ordering::SeqCst); + Ok(()) + } +} + +struct Outcome(Arc>>); + +#[async_trait] +impl AckHandle for Outcome { + async fn ack(&self) -> Result<(), DomainError> { + self.0.lock().unwrap().push("ack"); + Ok(()) + } + async fn nack(&self) -> Result<(), DomainError> { + self.0.lock().unwrap().push("nack"); + Ok(()) + } + async fn defer(&self, _: Duration) -> Result<(), DomainError> { + self.0.lock().unwrap().push("defer"); + Ok(()) + } +} + +struct Once(Mutex>); + +impl EventConsumer for Once { + fn consume(&self) -> BoxStream<'_, Result> { + let item = self.0.lock().unwrap().take().map(Ok); + Box::pin(futures::stream::iter(item)) + } +} + +/// A claim abandoned mid-handler blocks redeliveries (deferred, not failed) +/// until its lease lapses; then the next delivery takes it over and runs the +/// handler once, and later deliveries skip it. +#[tokio::test] +async fn an_abandoned_claim_defers_deliveries_until_the_lease_lapses_then_runs_once() { + let pool = support::memory_pool().await; + sqlite::migrate(&pool).await.unwrap(); + let ledger = Arc::new(SqliteProcessedEventLedger::with_lease( + pool, + Duration::from_secs(1), + )); + let id = EventId::new(); + // The crashed worker: claimed, never completed or released. + assert!(matches!( + ledger.begin(&id, "crashy").await.unwrap(), + Begin::Run(_) + )); + + let runs = Arc::new(AtomicU32::new(0)); + let outcomes = Arc::new(Mutex::new(Vec::new())); + let deliver = || { + let envelope = EventEnvelope::new( + Some(id), + DomainEvent::SearchReindexRequested, + Box::new(Outcome(outcomes.clone())), + ); + let service = WorkerService::new( + Arc::new(Once(Mutex::new(Some(envelope)))), + vec![Arc::new(Counting(runs.clone()))], + ledger.clone(), + ) + .unwrap(); + async move { service.run(tokio::sync::watch::channel(false).1).await } + }; + + deliver().await; + assert_eq!(*outcomes.lock().unwrap(), ["defer"]); + assert_eq!(runs.load(Ordering::SeqCst), 0); + + tokio::time::sleep(Duration::from_millis(2100)).await; + deliver().await; + assert_eq!(*outcomes.lock().unwrap(), ["defer", "ack"]); + assert_eq!(runs.load(Ordering::SeqCst), 1); + + deliver().await; + assert_eq!(runs.load(Ordering::SeqCst), 1, "done claims never re-run"); +} diff --git a/crates/domain/src/events.rs b/crates/domain/src/events.rs index 113d5fe8..3543cf9e 100644 --- a/crates/domain/src/events.rs +++ b/crates/domain/src/events.rs @@ -148,6 +148,11 @@ pub enum DomainEvent { pub trait AckHandle: Send + Sync { async fn ack(&self) -> Result<(), DomainError>; async fn nack(&self) -> Result<(), DomainError>; + /// Redeliver after `delay` without counting a failed attempt: for a + /// delivery that was not tried (it is blocked behind a live claim), so it + /// must not burn the retry budget that `nack` spends. A broker may still + /// count the redelivery itself: JetStream does, toward `max_deliver`. + async fn defer(&self, delay: std::time::Duration) -> Result<(), DomainError>; } /// Stable identity of one queued event; the key for de-duplicating deliveries. @@ -194,6 +199,10 @@ impl EventEnvelope { self.ack.ack().await } + pub async fn defer(self, delay: std::time::Duration) -> Result<(), DomainError> { + self.ack.defer(delay).await + } + pub async fn nack(self) -> Result<(), DomainError> { self.ack.nack().await } diff --git a/crates/domain/src/ports/processed_events.rs b/crates/domain/src/ports/processed_events.rs index d71a2553..ea542000 100644 --- a/crates/domain/src/ports/processed_events.rs +++ b/crates/domain/src/ports/processed_events.rs @@ -7,15 +7,32 @@ use crate::{errors::DomainError, events::EventId}; /// the next `begin` takes it over. Must exceed any handler's runtime. pub const HANDLER_LEASE_SECONDS: i64 = 1800; -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +/// Proof of ownership of one claim, returned by `begin`. `complete` and +/// `release` act only while the claim still carries this token, so an owner +/// whose lease lapsed cannot disturb whoever took the claim over. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClaimToken(String); + +impl ClaimToken { + pub fn new(value: String) -> Self { + Self(value) + } + + pub fn as_str(&self) -> &str { + &self.0 + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] pub enum Begin { /// The caller now holds the claim and must run the handler, then - /// `complete` or `release`. - Run, + /// `complete` or `release` with the token. + Run(ClaimToken), /// The handler already finished this event; skip it. AlreadyDone, - /// Another delivery holds a live claim; do not run, retry the delivery later. - InProgressElsewhere, + /// Another delivery holds a live claim; do not run. Retry the delivery + /// after `retry_after`, when the claim is done or its lease has lapsed. + InProgressElsewhere { retry_after: std::time::Duration }, } /// Per-(event, handler) claim ledger that stops redelivered events from @@ -32,11 +49,21 @@ pub trait ProcessedEventLedger: Send + Sync { /// reports the existing state. async fn begin(&self, id: &EventId, handler: &str) -> Result; - /// Marks the claim done: the handler will not run for this event again. - async fn complete(&self, id: &EventId, handler: &str) -> Result<(), DomainError>; + /// Marks the claim done, if `token` still owns it: the handler will not run for this event again. + async fn complete( + &self, + id: &EventId, + handler: &str, + token: &ClaimToken, + ) -> Result<(), DomainError>; - /// Drops an `in_progress` claim so a retry can run immediately. - async fn release(&self, id: &EventId, handler: &str) -> Result<(), DomainError>; + /// Drops an `in_progress` claim, if `token` still owns it, so a retry can run immediately. + async fn release( + &self, + id: &EventId, + handler: &str, + token: &ClaimToken, + ) -> Result<(), DomainError>; /// Removes rows (done, or abandoned `in_progress`) first recorded before /// `cutoff`. Returns the number removed. diff --git a/crates/domain/src/testing/fakes/noops.rs b/crates/domain/src/testing/fakes/noops.rs index 19beaf94..4a633de4 100644 --- a/crates/domain/src/testing/fakes/noops.rs +++ b/crates/domain/src/testing/fakes/noops.rs @@ -159,6 +159,9 @@ impl AckHandle for NoopAckHandle { async fn nack(&self) -> Result<(), DomainError> { Ok(()) } + async fn defer(&self, _: std::time::Duration) -> Result<(), DomainError> { + Ok(()) + } } // ── NoopPersonCommand ───────────────────────────────────────────────────────── diff --git a/crates/domain/src/testing/fakes/store_processed_events.rs b/crates/domain/src/testing/fakes/store_processed_events.rs index f1f96eb7..f26c71d7 100644 --- a/crates/domain/src/testing/fakes/store_processed_events.rs +++ b/crates/domain/src/testing/fakes/store_processed_events.rs @@ -1,21 +1,36 @@ use async_trait::async_trait; -use chrono::{Duration, NaiveDateTime, Utc}; +use chrono::{Duration, NaiveDateTime, Timelike, Utc}; use crate::{ errors::DomainError, events::EventId, - ports::{Begin, HANDLER_LEASE_SECONDS, ProcessedEventLedger}, + ports::{Begin, ClaimToken, HANDLER_LEASE_SECONDS, ProcessedEventLedger}, }; use super::InMemoryStore; use super::store::LedgerRow; +/// Claim stamps are millisecond-precision UTC, as in the sqlite adapter. +const STAMP: &str = "%Y-%m-%dT%H:%M:%S%.3fZ"; + +fn now_millis() -> NaiveDateTime { + let now = Utc::now().naive_utc(); + now.with_nanosecond(now.nanosecond() / 1_000_000 * 1_000_000) + .expect("truncated nanoseconds are valid") +} + +fn token(at: NaiveDateTime) -> ClaimToken { + ClaimToken::new(at.format(STAMP).to_string()) +} + /// Mirrors the sqlite adapter: `begin` inserts, or takes over only a stale -/// `in_progress` claim; `(event_id, handler)` is the key. +/// `in_progress` claim; `(event_id, handler)` is the key; `complete` and +/// `release` act only for the token that owns the claim. #[async_trait] impl ProcessedEventLedger for InMemoryStore { async fn begin(&self, id: &EventId, handler: &str) -> Result { - let now = Utc::now().naive_utc(); + let now = now_millis(); + let lease = Duration::seconds(HANDLER_LEASE_SECONDS); let mut rows = self.processed_events.write().unwrap(); let key = (id.value(), handler.to_owned()); match rows.get_mut(&key) { @@ -28,23 +43,38 @@ impl ProcessedEventLedger for InMemoryStore { processed_at: now, }, ); - Ok(Begin::Run) + Ok(Begin::Run(token(now))) } Some(row) if row.done => Ok(Begin::AlreadyDone), - Some(row) if row.started_at < now - Duration::seconds(HANDLER_LEASE_SECONDS) => { + Some(row) if row.started_at < now - lease => { row.started_at = now; - Ok(Begin::Run) + row.processed_at = now; + Ok(Begin::Run(token(now))) + } + Some(row) => { + let remaining = (row.started_at + lease - now) + .to_std() + .unwrap_or_default() + .max(std::time::Duration::from_secs(1)); + Ok(Begin::InProgressElsewhere { + retry_after: remaining, + }) } - Some(_) => Ok(Begin::InProgressElsewhere), } } - async fn complete(&self, id: &EventId, handler: &str) -> Result<(), DomainError> { + async fn complete( + &self, + id: &EventId, + handler: &str, + owner: &ClaimToken, + ) -> Result<(), DomainError> { if let Some(row) = self .processed_events .write() .unwrap() .get_mut(&(id.value(), handler.to_owned())) + && token(row.started_at) == *owner { row.done = true; row.processed_at = Utc::now().naive_utc(); @@ -52,10 +82,18 @@ impl ProcessedEventLedger for InMemoryStore { Ok(()) } - async fn release(&self, id: &EventId, handler: &str) -> Result<(), DomainError> { + async fn release( + &self, + id: &EventId, + handler: &str, + owner: &ClaimToken, + ) -> Result<(), DomainError> { let mut rows = self.processed_events.write().unwrap(); let key = (id.value(), handler.to_owned()); - if rows.get(&key).is_some_and(|r| !r.done) { + if rows + .get(&key) + .is_some_and(|r| !r.done && token(r.started_at) == *owner) + { rows.remove(&key); } Ok(())