fix: defer blocked deliveries, claim owner token

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 {

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 ────────────────────────────────────────────

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 {

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<String>, 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 {

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> {
+    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<Begin, DomainError> {
-        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(claim_stamp(now - self.lease))
-            .bind(stamp(stale_before))
             .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<String> = sqlx::query_scalar(
+            let row: Option<(String, Option<String>)> = sqlx::query_as(
-                "SELECT status FROM processed_events WHERE event_id = ? AND handler = ?",
+                "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() {
+            match row {
-                Some("done") => return Ok(Begin::AlreadyDone),
+                Some((status, _)) if status == "done" => return Ok(Begin::AlreadyDone),
-                Some(_) => return Ok(Begin::InProgressElsewhere),
+                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);
+        token(ledger.begin(&id, "a").await.unwrap());
-        assert_eq!(
+        let retry_after = blocked(ledger.begin(&id, "a").await.unwrap());
-            ledger.begin(&id, "a").await.unwrap(),
+        token(ledger.begin(&id, "b").await.unwrap());
-            Begin::InProgressElsewhere
+        token(ledger.begin(&EventId::new(), "a").await.unwrap());
-        );
+
-        assert_eq!(ledger.begin(&id, "b").await.unwrap(), Begin::Run);
+        // The blocked delivery is told to come back when the lease runs out.
-        assert_eq!(
+        let lease = std::time::Duration::from_secs(HANDLER_LEASE_SECONDS as u64);
-            ledger.begin(&EventId::new(), "a").await.unwrap(),
+        assert!(retry_after <= lease && retry_after > lease - std::time::Duration::from_secs(5));
-            Begin::Run
-        );
     }
 
     #[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();
+        let t = token(ledger.begin(&id, "a").await.unwrap());
-        ledger.complete(&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!(
+        blocked(ledger.begin(&id, "h").await.unwrap());
-            ledger.begin(&id, "h").await.unwrap(),
+        let processed_at: String = sqlx::query_scalar("SELECT processed_at FROM processed_events")
-            Begin::InProgressElsewhere
+            .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;
             }
         }

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<Vec<Arc<dyn EventHandler>>>,
     ledger: Arc<dyn ProcessedEventLedger>,
@@ -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<Duration> = 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) => {
+                Ok(Begin::InProgressElsewhere { retry_after }) => {
-                    tracing::debug!(
+                    tracing::debug!(handler = name, "running in another delivery, deferring");
-                        handler = name,
+                    blocked_for = Some(blocked_for.map_or(retry_after, |b| b.max(retry_after)));
-                        "running in another delivery, retrying later"
-                    );
-                    any_transient = true;
                     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
+                if let (Some(id), Some(token)) = (&id, &claim)
-                    && let Err(e) = ledger.complete(id, name).await
+                    && 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
+                if let (Some(id), Some(token)) = (&id, &claim)
-                    && let Err(le) = ledger.release(id, name).await
+                    && 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}");
         }

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);
 }
 

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<Mutex<u32>>,
+    defers: Arc<Mutex<Vec<std::time::Duration>>>,
 }
 
-struct CountNack(Arc<Mutex<u32>>);
+struct CountNack(Arc<Mutex<u32>>, Arc<Mutex<Vec<std::time::Duration>>>);
 
 #[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<EventEnvelope, DomainError>> {
         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 =
+            let envelope = EventEnvelope::new(
-                EventEnvelope::new(id, event.clone(), Box::new(CountNack(Arc::clone(&nacks))));
+                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<Mutex<Vec<std::time::Duration>>> = 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<Mutex<Vec<std::time::Duration>>> = 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())

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<AtomicU32>);
+
+#[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<Mutex<Vec<&'static str>>>);
+
+#[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<Option<EventEnvelope>>);
+
+impl EventConsumer for Once {
+    fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>> {
+        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");
+}

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
     }

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`.
+    /// `complete` or `release` with the token.
-    Run,
+    Run(ClaimToken),
     /// The handler already finished this event; skip it.
     AlreadyDone,
-    /// Another delivery holds a live claim; do not run, retry the delivery later.
+    /// Another delivery holds a live claim; do not run. Retry the delivery
-    InProgressElsewhere,
+    /// 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<Begin, DomainError>;
 
-    /// Marks the claim done: the handler will not run for this event again.
+    /// 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) -> Result<(), DomainError>;
+    async fn complete(
+        &self,
+        id: &EventId,
+        handler: &str,
+        token: &ClaimToken,
+    ) -> Result<(), DomainError>;
 
-    /// Drops an `in_progress` claim so a retry can run immediately.
+    /// Drops an `in_progress` claim, if `token` still owns it, so a retry can run immediately.
-    async fn release(&self, id: &EventId, handler: &str) -> Result<(), DomainError>;
+    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.

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 ─────────────────────────────────────────────────────────

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<Begin, DomainError> {
-        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(())