crates/adapters/nats/src/config.rs

@@ -11,10 +11,15 @@ pub struct NatsConfig {
     pub subject_prefix: String,
     pub stream_name: String,
     pub consumer_name: String,
+    /// JetStream redelivery bound, same as the DB queue's attempt limit.
+    pub max_deliver: i64,
 }
 
 impl NatsConfig {
-    pub fn from_config(config: &::config::NatsConfig) -> anyhow::Result<Self> {
+    pub fn from_config(
+        config: &::config::NatsConfig,
+        events: &::config::EventsConfig,
+    ) -> anyhow::Result<Self> {
         let url = config
             .url
             .clone()
@@ -32,6 +37,7 @@ impl NatsConfig {
             subject_prefix: config.subject_prefix.clone(),
             stream_name: config.stream_name.clone(),
             consumer_name: config.consumer_name.clone(),
+            max_deliver: i64::from(events.queue_max_attempts),
         })
     }
 }

crates/adapters/nats/src/consumer.rs

@@ -108,8 +108,9 @@ fn consume_from_rx(rx: EnvelopeRx) -> BoxStream<'static, Result<EventEnvelope, D
 
 /// How long JetStream remembers `Nats-Msg-Id`s. A row re-relayed within this
 /// window (crash between publish and mark-done, or a lease reclaim) is dropped
-/// by the broker; beyond it the consumer ledger is the backstop. Set
-/// explicitly so it does not silently follow the server default.
+/// by the broker; a duplicate arriving later is caught by the processed-event
+/// ledger, which keeps its rows for 30 days. Set explicitly so it does not
+/// silently follow the server default.
 const DEDUP_WINDOW: std::time::Duration = std::time::Duration::from_secs(60 * 60);
 
 // ── JetStream consumer ────────────────────────────────────────────────────────
@@ -140,6 +141,7 @@ impl NatsJetStreamConsumer {
                     durable_name: Some(cfg.consumer_name.clone()),
                     filter_subject: subject_filter,
                     ack_wait: std::time::Duration::from_secs(600),
+                    max_deliver: cfg.max_deliver,
                     ..Default::default()
                 },
             )

crates/adapters/nats/tests/nats_config_test.rs

@@ -10,15 +10,24 @@ fn section(url: Option<&str>, mode: &str) -> config::NatsConfig {
 
 #[test]
 fn errors_without_nats_url() {
-    assert!(NatsConfig::from_config(&config::NatsConfig::default()).is_err());
+    assert!(
+        NatsConfig::from_config(
+            &config::NatsConfig::default(),
+            &config::EventsConfig::default()
+        )
+        .is_err()
+    );
 }
 
 #[test]
 fn defaults_with_only_url() {
-    let cfg = NatsConfig::from_config(&config::NatsConfig {
-        url: Some("nats://localhost:4222".into()),
-        ..Default::default()
-    })
+    let cfg = NatsConfig::from_config(
+        &config::NatsConfig {
+            url: Some("nats://localhost:4222".into()),
+            ..Default::default()
+        },
+        &config::EventsConfig::default(),
+    )
     .unwrap();
     assert_eq!(cfg.url, "nats://localhost:4222");
     assert_eq!(cfg.mode, NatsMode::JetStream);
@@ -29,12 +38,33 @@ fn defaults_with_only_url() {
 
 #[test]
 fn core_mode_parsed() {
-    let cfg = NatsConfig::from_config(&section(Some("nats://test:4222"), "core")).unwrap();
+    let cfg = NatsConfig::from_config(
+        &section(Some("nats://test:4222"), "core"),
+        &config::EventsConfig::default(),
+    )
+    .unwrap();
     assert_eq!(cfg.url, "nats://test:4222");
     assert_eq!(cfg.mode, NatsMode::Core);
 }
 
 #[test]
 fn invalid_mode_errors() {
-    assert!(NatsConfig::from_config(&section(Some("nats://test:4222"), "kafka")).is_err());
+    assert!(
+        NatsConfig::from_config(
+            &section(Some("nats://test:4222"), "kafka"),
+            &config::EventsConfig::default()
+        )
+        .is_err()
+    );
+}
+
+#[test]
+fn max_deliver_follows_the_queue_attempt_limit() {
+    let events = config::EventsConfig {
+        queue_max_attempts: 9,
+        ..Default::default()
+    };
+    let cfg =
+        NatsConfig::from_config(&section(Some("nats://test:4222"), "jetstream"), &events).unwrap();
+    assert_eq!(cfg.max_deliver, 9);
 }

crates/adapters/sqlite/migrations/0101_outbox_identity.sql

@@ -1,5 +1,6 @@
--- Outbox identity: every queued event carries a stable id; handlers record
--- (event_id, handler) pairs so redeliveries are skipped.
+-- Outbox identity: every queued event carries a stable id; handlers claim
+-- (event_id, handler) pairs so redeliveries are skipped and concurrent
+-- deliveries never run the same handler twice.
 ALTER TABLE event_queue ADD COLUMN event_id TEXT;
 
 -- When the current claim started; a claim older than the lease is reclaimable.
@@ -21,6 +22,8 @@ CREATE UNIQUE INDEX idx_event_queue_event_id
 CREATE TABLE processed_events (
     event_id     TEXT NOT NULL,
     handler      TEXT NOT NULL,
+    status       TEXT NOT NULL CHECK (status IN ('in_progress', 'done')),
+    started_at   TEXT,
     processed_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
     PRIMARY KEY (event_id, handler)
 );

crates/adapters/sqlite/src/repositories/processed_events/mod.rs

@@ -1,5 +1,9 @@
 use async_trait::async_trait;
-use domain::{errors::DomainError, events::EventId, ports::ProcessedEventLedger};
+use domain::{
+    errors::DomainError,
+    events::EventId,
+    ports::{Begin, HANDLER_LEASE_SECONDS, ProcessedEventLedger},
+};
 use sqlx::SqlitePool;
 
 /// Matches the column default in migration 0101, so string comparison orders rows by time.
@@ -15,32 +19,87 @@ impl SqliteProcessedEventLedger {
     }
 }
 
+fn stamp(at: chrono::NaiveDateTime) -> String {
+    at.format(TIMESTAMP_FORMAT).to_string()
+}
+
 #[async_trait]
 impl ProcessedEventLedger for SqliteProcessedEventLedger {
-    async fn has_processed(&self, id: &EventId, handler: &str) -> Result<bool, DomainError> {
-        let found: Option<i64> =
-            sqlx::query_scalar("SELECT 1 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)?;
-        Ok(found.is_some())
-    }
-
-    async fn record_processed(&self, id: &EventId, handler: &str) -> Result<(), DomainError> {
-        sqlx::query("INSERT OR IGNORE INTO processed_events (event_id, handler) VALUES (?, ?)")
+    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 {
+            // 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
+                 WHERE processed_events.status = 'in_progress'
+                   AND processed_events.started_at < ?",
+            )
             .bind(id.value().to_string())
             .bind(handler)
+            .bind(stamp(now))
+            .bind(stamp(now))
+            .bind(stamp(stale_before))
             .execute(&self.pool)
             .await
+            .map_err(crate::shared::map_sqlx_error)?
+            .rows_affected();
+            if claimed == 1 {
+                return Ok(Begin::Run);
+            }
+
+            let status: Option<String> = sqlx::query_scalar(
+                "SELECT status 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),
+                None => continue,
+            }
+        }
+        Ok(Begin::InProgressElsewhere)
+    }
+
+    async fn complete(&self, id: &EventId, handler: &str) -> Result<(), DomainError> {
+        sqlx::query(
+            "UPDATE processed_events SET status = 'done', processed_at = ?
+             WHERE event_id = ? AND handler = ?",
+        )
+        .bind(stamp(chrono::Utc::now().naive_utc()))
+        .bind(id.value().to_string())
+        .bind(handler)
+        .execute(&self.pool)
+        .await
+        .map_err(crate::shared::map_sqlx_error)?;
+        Ok(())
+    }
+
+    async fn release(&self, id: &EventId, handler: &str) -> Result<(), DomainError> {
+        sqlx::query(
+            "DELETE FROM processed_events
+             WHERE event_id = ? AND handler = ? AND status = 'in_progress'",
+        )
+        .bind(id.value().to_string())
+        .bind(handler)
+        .execute(&self.pool)
+        .await
+        .map_err(crate::shared::map_sqlx_error)?;
         Ok(())
     }
 
     async fn delete_older_than(&self, cutoff: chrono::NaiveDateTime) -> Result<u64, DomainError> {
         let result = sqlx::query("DELETE FROM processed_events WHERE processed_at < ?")
-            .bind(cutoff.format(TIMESTAMP_FORMAT).to_string())
+            .bind(stamp(cutoff))
             .execute(&self.pool)
             .await
             .map_err(crate::shared::map_sqlx_error)?;
@@ -62,50 +121,134 @@ mod tests {
         (SqliteProcessedEventLedger::new(pool.clone()), pool)
     }
 
-    #[tokio::test]
-    async fn a_recorded_pair_is_processed_and_others_are_not() {
-        let (ledger, _) = ledger().await;
-        let id = EventId::new();
-        assert!(!ledger.has_processed(&id, "a").await.unwrap());
-
-        ledger.record_processed(&id, "a").await.unwrap();
-
-        assert!(ledger.has_processed(&id, "a").await.unwrap());
-        assert!(!ledger.has_processed(&id, "b").await.unwrap());
-        assert!(!ledger.has_processed(&EventId::new(), "a").await.unwrap());
+    async fn insert(pool: &SqlitePool, id: &EventId, status: &str, at: &str) {
+        sqlx::query(
+            "INSERT INTO processed_events (event_id, handler, status, started_at, processed_at)
+             VALUES (?, 'h', ?, ?, ?)",
+        )
+        .bind(id.value().to_string())
+        .bind(status)
+        .bind(at)
+        .bind(at)
+        .execute(pool)
+        .await
+        .unwrap();
     }
 
     #[tokio::test]
-    async fn recording_twice_is_not_an_error() {
+    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
+        );
+    }
+
+    #[tokio::test]
+    async fn a_completed_claim_is_already_done() {
+        let (ledger, _) = ledger().await;
+        let id = EventId::new();
+        ledger.begin(&id, "a").await.unwrap();
+
+        ledger.complete(&id, "a").await.unwrap();
+
+        assert_eq!(ledger.begin(&id, "a").await.unwrap(), Begin::AlreadyDone);
+    }
+
+    #[tokio::test]
+    async fn a_released_claim_can_run_again() {
+        let (ledger, _) = ledger().await;
+        let id = EventId::new();
+        ledger.begin(&id, "a").await.unwrap();
+
+        ledger.release(&id, "a").await.unwrap();
+
+        assert_eq!(ledger.begin(&id, "a").await.unwrap(), Begin::Run);
+    }
+
+    #[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();
+
+        ledger.release(&id, "a").await.unwrap();
+
+        assert_eq!(ledger.begin(&id, "a").await.unwrap(), Begin::AlreadyDone);
+    }
+
+    #[tokio::test]
+    async fn a_stale_in_progress_claim_is_taken_over() {
         let (ledger, pool) = ledger().await;
         let id = EventId::new();
-        ledger.record_processed(&id, "a").await.unwrap();
-        ledger.record_processed(&id, "a").await.unwrap();
+        insert(&pool, &id, "in_progress", "2020-01-01T00:00:00Z").await;
 
-        let rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM processed_events")
-            .fetch_one(&pool)
+        assert_eq!(ledger.begin(&id, "h").await.unwrap(), Begin::Run);
+        // The takeover refreshed the lease, so a third delivery is blocked.
+        assert_eq!(
+            ledger.begin(&id, "h").await.unwrap(),
+            Begin::InProgressElsewhere
+        );
+    }
+
+    #[tokio::test]
+    async fn a_done_claim_is_never_taken_over_however_old() {
+        let (ledger, pool) = ledger().await;
+        let id = EventId::new();
+        insert(&pool, &id, "done", "2020-01-01T00:00:00Z").await;
+
+        assert_eq!(ledger.begin(&id, "h").await.unwrap(), Begin::AlreadyDone);
+    }
+
+    #[tokio::test]
+    async fn concurrent_begins_have_exactly_one_winner() {
+        let dir = tempfile::tempdir().unwrap();
+        let opts = sqlx::sqlite::SqliteConnectOptions::new()
+            .filename(dir.path().join("l.db"))
+            .create_if_missing(true)
+            .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
+            .busy_timeout(std::time::Duration::from_secs(5));
+        let pool = sqlx::sqlite::SqlitePoolOptions::new()
+            .max_connections(4)
+            .connect_with(opts)
             .await
             .unwrap();
-        assert_eq!(rows, 1);
+        crate::migrate(&pool).await.unwrap();
+        let ledger = std::sync::Arc::new(SqliteProcessedEventLedger::new(pool));
+        let id = EventId::new();
+
+        let tasks: Vec<_> = (0..8)
+            .map(|_| {
+                let ledger = ledger.clone();
+                tokio::spawn(async move { ledger.begin(&id, "a").await.unwrap() })
+            })
+            .collect();
+        let mut runs = 0;
+        for t in tasks {
+            if t.await.unwrap() == Begin::Run {
+                runs += 1;
+            }
+        }
+
+        assert_eq!(runs, 1);
     }
 
     #[tokio::test]
     async fn delete_older_than_removes_only_rows_before_the_cutoff() {
         let (ledger, pool) = ledger().await;
-        let (old, recent) = (EventId::new(), EventId::new());
-        for (id, at) in [
-            (old, "2026-01-01T00:00:00Z"),
-            (recent, "2026-06-01T00:00:00Z"),
-        ] {
-            sqlx::query(
-                "INSERT INTO processed_events (event_id, handler, processed_at) VALUES (?, 'h', ?)",
-            )
-            .bind(id.value().to_string())
-            .bind(at)
-            .execute(&pool)
-            .await
-            .unwrap();
-        }
+        let (old_done, old_stuck, recent) = (EventId::new(), EventId::new(), EventId::new());
+        insert(&pool, &old_done, "done", "2026-01-01T00:00:00Z").await;
+        insert(&pool, &old_stuck, "in_progress", "2026-01-02T00:00:00Z").await;
+        insert(&pool, &recent, "done", "2026-06-01T00:00:00Z").await;
         let cutoff = chrono::NaiveDate::from_ymd_opt(2026, 3, 1)
             .unwrap()
             .and_hms_opt(0, 0, 0)
@@ -113,8 +256,10 @@ mod tests {
 
         let removed = ledger.delete_older_than(cutoff).await.unwrap();
 
-        assert_eq!(removed, 1);
-        assert!(!ledger.has_processed(&old, "h").await.unwrap());
-        assert!(ledger.has_processed(&recent, "h").await.unwrap());
+        assert_eq!(removed, 2);
+        assert_eq!(
+            ledger.begin(&recent, "h").await.unwrap(),
+            Begin::AlreadyDone
+        );
     }
 }

crates/adapters/sqlite/tests/fixtures/migration_ledger.txt

@@ -29,4 +29,4 @@
 29 56F4F7B41A09F07211D7543204F0EAC93B7BFB15AFD9BC90DF2654DAE5BE96B1E358B50E2190BF25739C5F0B1FD4B806
 30 DAAA0AF35B4ACB271A9D8068B82F02A72EEC62F3AC07ED504C7A7887976C2C8BACB35FFF477B63A931DF15F79018390D
 100 310ADC9ABF0CC2F8E9EDAB33168D009D64228FF5BD8D2A14185513699EBB782606A1EEA8FEE6EC6AF65D72396153F13F
-101 C0EE8C5B9FA6A2BE75674A10A008B895FFAF8056139FE0C0F9936B27BA06AFC71B1DC5BD8318948F6E9FCF0689823B71
+101 19A2542D4343F3921A8BF0A10E195DA1915E3EA97A2BCF2D3BF9079504DB2053E4BD2175D94C95BCA363BC2FDF14AA88

crates/application/src/worker.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use domain::{
     errors::DomainError,
     events::EventEnvelope,
-    ports::{EventConsumer, EventHandler, ProcessedEventLedger},
+    ports::{Begin, EventConsumer, EventHandler, ProcessedEventLedger},
 };
 use futures::StreamExt;
 use tokio::sync::Semaphore;
@@ -98,38 +98,57 @@ async fn dispatch(
     for handler in handlers.iter() {
         let name = handler.name();
         if let Some(id) = &id {
-            match ledger.has_processed(id, name).await {
-                Ok(true) => {
+            match ledger.begin(id, name).await {
+                Ok(Begin::Run) => {}
+                Ok(Begin::AlreadyDone) => {
                     tracing::debug!(handler = name, "already processed, skipping");
                     continue;
                 }
-                Ok(false) => {}
+                Ok(Begin::InProgressElsewhere) => {
+                    tracing::debug!(
+                        handler = name,
+                        "running in another delivery, retrying later"
+                    );
+                    any_transient = true;
+                    continue;
+                }
                 Err(e) => {
                     // Running anyway could repeat a side effect; retry the delivery instead.
-                    tracing::warn!(handler = name, "ledger lookup failed, will retry: {e}");
+                    tracing::warn!(handler = name, "ledger claim failed, will retry: {e}");
                     any_transient = true;
                     continue;
                 }
             }
         }
-        match handler.handle(&envelope.event).await {
+        let result = handler.handle(&envelope.event).await;
+        match &result {
             Ok(()) => {
                 if let Some(id) = &id
-                    && let Err(e) = ledger.record_processed(id, name).await
+                    && let Err(e) = ledger.complete(id, name).await
                 {
                     // The work is done; failing the delivery would only repeat it.
-                    tracing::warn!(handler = name, "recording processed event failed: {e}");
+                    // The claim lapses after its lease and the handler may run once more.
+                    tracing::warn!(handler = name, "completing ledger claim failed: {e}");
                 }
             }
-            Err(e) if e.is_transient() => {
-                tracing::warn!(handler = name, "transient handler error, will retry: {e}");
-                any_transient = true;
-            }
             Err(e) => {
-                tracing::warn!(
-                    handler = name,
-                    "permanent handler error (not retrying): {e}"
-                );
+                // 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
+                {
+                    tracing::warn!(handler = name, "releasing ledger claim failed: {le}");
+                }
+                if e.is_transient() {
+                    tracing::warn!(handler = name, "transient handler error, will retry: {e}");
+                    any_transient = true;
+                } else {
+                    tracing::warn!(
+                        handler = name,
+                        "permanent handler error (not retrying): {e}"
+                    );
+                }
             }
         }
     }

crates/application/tests/jobs/processed_events_cleanup_test.rs

@@ -10,11 +10,13 @@ async fn removes_only_rows_older_than_the_retention_window() {
     store.seed_processed_event(
         &stale,
         "h",
+        true,
         now - chrono::Duration::days(PROCESSED_EVENTS_RETENTION_DAYS + 1),
     );
     store.seed_processed_event(
         &fresh,
         "h",
+        true,
         now - chrono::Duration::days(PROCESSED_EVENTS_RETENTION_DAYS - 1),
     );
 
@@ -23,9 +25,9 @@ async fn removes_only_rows_older_than_the_retention_window() {
         .await
         .unwrap();
 
-    use domain::ports::ProcessedEventLedger;
-    assert!(!store.has_processed(&stale, "h").await.unwrap());
-    assert!(store.has_processed(&fresh, "h").await.unwrap());
+    use domain::ports::{Begin, ProcessedEventLedger};
+    assert_eq!(store.begin(&stale, "h").await.unwrap(), Begin::Run);
+    assert_eq!(store.begin(&fresh, "h").await.unwrap(), Begin::AlreadyDone);
 }
 
 #[test]

crates/application/tests/worker/worker_test.rs

@@ -477,3 +477,83 @@ async fn duplicate_handler_names_are_rejected() {
     let err = result.err().expect("duplicate names must be rejected");
     assert!(err.to_string().contains("same"), "{err}");
 }
+
+#[tokio::test]
+async fn a_handler_already_running_elsewhere_is_not_run_concurrently() {
+    use domain::ports::{Begin, ProcessedEventLedger};
+    let (handler, calls) = CountingHandler::new("slow", 0);
+    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);
+    let nacks = Arc::new(Mutex::new(0));
+    let consumer = RedeliveringConsumer {
+        id: Some(id),
+        event: movie_discovered(),
+        times: 1,
+        nacks: Arc::clone(&nacks),
+    };
+
+    WorkerService::new(Arc::new(consumer), vec![handler], store)
+        .unwrap()
+        .run(tokio::sync::watch::channel(false).1)
+        .await;
+
+    assert_eq!(
+        *calls.lock().unwrap(),
+        0,
+        "must not run beside the claim holder"
+    );
+    assert_eq!(*nacks.lock().unwrap(), 1, "delivery is retried later");
+}
+
+#[tokio::test]
+async fn a_stale_in_progress_claim_is_taken_over() {
+    use domain::ports::HANDLER_LEASE_SECONDS;
+    let (handler, calls) = CountingHandler::new("crashed", 0);
+    let store = InMemoryStore::new();
+    let id = domain::events::EventId::new();
+    store.seed_processed_event(
+        &id,
+        "crashed",
+        false,
+        chrono::Utc::now().naive_utc() - chrono::Duration::seconds(HANDLER_LEASE_SECONDS + 1),
+    );
+    let nacks = Arc::new(Mutex::new(0));
+    let consumer = RedeliveringConsumer {
+        id: Some(id),
+        event: movie_discovered(),
+        times: 1,
+        nacks: Arc::clone(&nacks),
+    };
+
+    WorkerService::new(Arc::new(consumer), vec![handler], store)
+        .unwrap()
+        .run(tokio::sync::watch::channel(false).1)
+        .await;
+
+    assert_eq!(*calls.lock().unwrap(), 1);
+    assert_eq!(*nacks.lock().unwrap(), 0);
+}
+
+#[tokio::test]
+async fn a_failed_handler_releases_its_claim_so_redelivery_retries() {
+    let (handler, calls) = CountingHandler::new("flaky", 1);
+    let store = InMemoryStore::new();
+    let nacks = Arc::new(Mutex::new(0));
+    let consumer = RedeliveringConsumer {
+        id: Some(domain::events::EventId::new()),
+        event: movie_discovered(),
+        times: 3,
+        nacks: Arc::clone(&nacks),
+    };
+
+    WorkerService::new(Arc::new(consumer), vec![handler], store.clone())
+        .unwrap()
+        .run(tokio::sync::watch::channel(false).1)
+        .await;
+
+    // Delivery 1 fails and releases; 2 succeeds; 3 is skipped as done.
+    assert_eq!(*calls.lock().unwrap(), 2);
+    assert_eq!(*nacks.lock().unwrap(), 1);
+}

crates/bootstrap/src/event_bus.rs

@@ -83,7 +83,7 @@ pub async fn create_channel(
         }
         #[cfg(feature = "nats")]
         EventBusBackend::Nats => {
-            let cfg = nats::NatsConfig::from_config(&config.nats)
+            let cfg = nats::NatsConfig::from_config(&config.nats, &config.events)
                 .context("EVENT_BUS_BACKEND=nats requires NATS_URL to be set")?;
             tracing::info!("event bus: NATS ({}), relaying the outbox", cfg.url);
             compose(outbox, Some(nats::create_bus(cfg).await?))

crates/bootstrap/tests/bootstrap_test.rs

@@ -124,17 +124,24 @@ async fn handler_names_are_unique() {
         names.len(),
         "duplicate handler name in {names:?}"
     );
-    for core in [
-        "poster_sync",
+    let mut sorted = names.clone();
+    sorted.sort_unstable();
+    // Names key the ledger: a rename re-runs events inside the retention
+    // window, so changing this list must be a deliberate edit.
+    let mut expected = vec![
         "image_cleanup",
+        "movie_discovery_indexer",
+        "movie_enrichment",
+        "person_enrichment",
+        "poster_sync",
         "search_cleanup",
         "search_reindex",
-    ] {
-        assert!(names.contains(&core), "{core} missing from {names:?}");
+        "wrapup",
+    ];
+    if cfg!(feature = "federation") {
+        expected.extend(["activitypub", "follow_backfill"]);
     }
-    assert_eq!(
-        names.contains(&"activitypub"),
-        cfg!(feature = "federation"),
-        "{names:?}"
-    );
+    expected.sort_unstable();
+    // Image conversion ("image_conversion") is disabled in the test config.
+    assert_eq!(sorted, expected);
 }

crates/domain/src/ports/processed_events.rs

@@ -2,15 +2,43 @@ use async_trait::async_trait;
 
 use crate::{errors::DomainError, events::EventId};
 
-/// Records which handler has already completed which event, so a redelivered
-/// event does not re-run handlers that succeeded.
+/// How long an `in_progress` claim blocks other deliveries of the same
+/// (event, handler). Past it the claim is presumed dead (crashed worker) and
+/// 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)]
+pub enum Begin {
+    /// The caller now holds the claim and must run the handler, then
+    /// `complete` or `release`.
+    Run,
+    /// The handler already finished this event; skip it.
+    AlreadyDone,
+    /// Another delivery holds a live claim; do not run, retry the delivery later.
+    InProgressElsewhere,
+}
+
+/// Per-(event, handler) claim ledger that stops redelivered events from
+/// re-running handlers.
+///
+/// Guarantee: two deliveries never run the same handler for the same event
+/// concurrently (`begin` is atomic across workers). After a crash mid-handler
+/// the claim lapses after `HANDLER_LEASE_SECONDS` and the handler runs again,
+/// so delivery is at-least-once and handlers must be idempotent.
 #[async_trait]
 pub trait ProcessedEventLedger: Send + Sync {
-    async fn has_processed(&self, id: &EventId, handler: &str) -> Result<bool, DomainError>;
+    /// Atomically claims (event, handler): inserts a claim if there is none, or
+    /// takes over an `in_progress` claim older than the lease; otherwise
+    /// reports the existing state.
+    async fn begin(&self, id: &EventId, handler: &str) -> Result<Begin, DomainError>;
 
-    /// Idempotent: recording an existing pair is not an error.
-    async fn record_processed(&self, id: &EventId, handler: &str) -> Result<(), DomainError>;
+    /// Marks the claim done: the handler will not run for this event again.
+    async fn complete(&self, id: &EventId, handler: &str) -> Result<(), DomainError>;
 
-    /// Returns the number of rows removed.
+    /// Drops an `in_progress` claim so a retry can run immediately.
+    async fn release(&self, id: &EventId, handler: &str) -> Result<(), DomainError>;
+
+    /// Removes rows (done, or abandoned `in_progress`) first recorded before
+    /// `cutoff`. Returns the number removed.
     async fn delete_older_than(&self, cutoff: chrono::NaiveDateTime) -> Result<u64, DomainError>;
 }

crates/domain/src/testing/fakes/store.rs

@@ -9,6 +9,13 @@ use crate::models::{
 };
 use crate::value_objects::SocialIdentity;
 
+/// One `processed_events` row.
+pub(super) struct LedgerRow {
+    pub(super) done: bool,
+    pub(super) started_at: chrono::NaiveDateTime,
+    pub(super) processed_at: chrono::NaiveDateTime,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(super) enum FollowState {
     Pending,
@@ -64,7 +71,7 @@ pub struct InMemoryStore {
     pub(super) import_profiles: RwLock<HashMap<Uuid, ImportProfile>>,
     pub(super) follows: RwLock<Vec<(Uuid, SocialIdentity, FollowState)>>,
     pub(super) blocked: RwLock<Vec<(Uuid, SocialIdentity)>>,
-    pub(super) processed_events: RwLock<HashMap<(Uuid, String), chrono::NaiveDateTime>>,
+    pub(super) processed_events: RwLock<HashMap<(Uuid, String), LedgerRow>>,
 }
 
 impl InMemoryStore {
@@ -128,17 +135,22 @@ impl InMemoryStore {
         self.processed_events.read().unwrap().len()
     }
 
-    /// Inserts a ledger row with an explicit timestamp (for retention tests).
+    /// Inserts a ledger row with explicit timestamps (retention and lease tests).
     pub fn seed_processed_event(
         &self,
         id: &crate::events::EventId,
         handler: &str,
-        processed_at: chrono::NaiveDateTime,
+        done: bool,
+        at: chrono::NaiveDateTime,
     ) {
-        self.processed_events
-            .write()
-            .unwrap()
-            .insert((id.value(), handler.to_owned()), processed_at);
+        self.processed_events.write().unwrap().insert(
+            (id.value(), handler.to_owned()),
+            LedgerRow {
+                done,
+                started_at: at,
+                processed_at: at,
+            },
+        );
     }
 
     pub fn refresh_session_count(&self) -> usize {

crates/domain/src/testing/fakes/store_processed_events.rs

@@ -1,33 +1,70 @@
 use async_trait::async_trait;
+use chrono::{Duration, NaiveDateTime, Utc};
 
-use crate::{errors::DomainError, events::EventId, ports::ProcessedEventLedger};
+use crate::{
+    errors::DomainError,
+    events::EventId,
+    ports::{Begin, HANDLER_LEASE_SECONDS, ProcessedEventLedger},
+};
 
 use super::InMemoryStore;
+use super::store::LedgerRow;
 
+/// Mirrors the sqlite adapter: `begin` inserts, or takes over only a stale
+/// `in_progress` claim; `(event_id, handler)` is the key.
 #[async_trait]
 impl ProcessedEventLedger for InMemoryStore {
-    async fn has_processed(&self, id: &EventId, handler: &str) -> Result<bool, DomainError> {
-        Ok(self
-            .processed_events
-            .read()
-            .unwrap()
-            .contains_key(&(id.value(), handler.to_owned())))
+    async fn begin(&self, id: &EventId, handler: &str) -> Result<Begin, DomainError> {
+        let now = Utc::now().naive_utc();
+        let mut rows = self.processed_events.write().unwrap();
+        let key = (id.value(), handler.to_owned());
+        match rows.get_mut(&key) {
+            None => {
+                rows.insert(
+                    key,
+                    LedgerRow {
+                        done: false,
+                        started_at: now,
+                        processed_at: now,
+                    },
+                );
+                Ok(Begin::Run)
+            }
+            Some(row) if row.done => Ok(Begin::AlreadyDone),
+            Some(row) if row.started_at < now - Duration::seconds(HANDLER_LEASE_SECONDS) => {
+                row.started_at = now;
+                Ok(Begin::Run)
+            }
+            Some(_) => Ok(Begin::InProgressElsewhere),
+        }
     }
 
-    /// `(event_id, handler)` is the primary key; a repeat keeps the first row.
-    async fn record_processed(&self, id: &EventId, handler: &str) -> Result<(), DomainError> {
-        self.processed_events
+    async fn complete(&self, id: &EventId, handler: &str) -> Result<(), DomainError> {
+        if let Some(row) = self
+            .processed_events
             .write()
             .unwrap()
-            .entry((id.value(), handler.to_owned()))
-            .or_insert_with(|| chrono::Utc::now().naive_utc());
+            .get_mut(&(id.value(), handler.to_owned()))
+        {
+            row.done = true;
+            row.processed_at = Utc::now().naive_utc();
+        }
         Ok(())
     }
 
-    async fn delete_older_than(&self, cutoff: chrono::NaiveDateTime) -> Result<u64, DomainError> {
+    async fn release(&self, id: &EventId, handler: &str) -> 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) {
+            rows.remove(&key);
+        }
+        Ok(())
+    }
+
+    async fn delete_older_than(&self, cutoff: NaiveDateTime) -> Result<u64, DomainError> {
         let mut rows = self.processed_events.write().unwrap();
         let before = rows.len();
-        rows.retain(|_, at| *at >= cutoff);
+        rows.retain(|_, r| r.processed_at >= cutoff);
         Ok((before - rows.len()) as u64)
     }
 }