fix: relay shutdown, claim lease, dead-letter undecodable, dedup window

crates/adapters/nats/src/consumer.rs

@@ -106,6 +106,12 @@ 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.
+const DEDUP_WINDOW: std::time::Duration = std::time::Duration::from_secs(60 * 60);
+
 // ── JetStream consumer ────────────────────────────────────────────────────────
 
 pub struct NatsJetStreamConsumer {
@@ -121,6 +127,7 @@ impl NatsJetStreamConsumer {
                 name: cfg.stream_name.clone(),
                 subjects: vec![consumer_subject_filter(&cfg.subject_prefix)],
                 max_messages: 100_000,
+                duplicate_window: DEDUP_WINDOW,
                 ..Default::default()
             })
             .await?;

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

@@ -2,6 +2,9 @@
 -- (event_id, handler) pairs so redeliveries are skipped.
 ALTER TABLE event_queue ADD COLUMN event_id TEXT;
 
+-- When the current claim started; a claim older than the lease is reclaimable.
+ALTER TABLE event_queue ADD COLUMN claimed_at TEXT;
+
 -- Legacy rows get a random v4-shaped uuid. The subquery references the row id
 -- so it is re-evaluated per row rather than cached.
 UPDATE event_queue SET event_id = (

crates/adapters/sqlite/src/event_queue/mod.rs

@@ -16,6 +16,11 @@ use tokio::sync::{Mutex, mpsc};
 use crate::shared::enqueue_events;
 pub(crate) use payload::DbEventPayload;
 
+/// How long a claim holds a row. A worker that dies between claim and
+/// ack/nack strands its rows in `processing`; once the lease lapses the claim
+/// query takes them again. Must exceed the slowest handler.
+pub(crate) const CLAIM_LEASE_SECONDS: i64 = 600;
+
 pub struct DbEventQueueConfig {
     pub poll_interval_ms: u64,
     pub batch_size: i64,
@@ -100,7 +105,11 @@ impl EventConsumer for SqliteEventQueue {
                     }
                     Ok(rows) => {
                         for row in rows {
+                            let row_id = row.id;
                             let envelope = decode_row(&pool, row, config.max_attempts);
+                            if let Err(e) = &envelope {
+                                dead_letter_undecodable(&pool, row_id, e).await;
+                            }
                             if tx.send(envelope).await.is_err() {
                                 tracing::info!("sqlite event queue consumer closed");
                                 return;
@@ -137,27 +146,50 @@ async fn claim_batch(
     pool: &SqlitePool,
     config: &DbEventQueueConfig,
 ) -> Result<Vec<QueueRow>, DomainError> {
-    let mut rows = sqlx::query_as::<_, QueueRow>(
+    let sql = format!(
-        "UPDATE event_queue SET status = 'processing'
+        "UPDATE event_queue
+         SET status = 'processing', claimed_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
          WHERE id IN (
              SELECT id FROM event_queue
-             WHERE status = 'pending'
+             WHERE (status = 'pending'
-               AND next_attempt_at <= strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
+                    AND next_attempt_at <= strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
+                OR (status = 'processing'
+                    AND (claimed_at IS NULL
+                         OR claimed_at < strftime('%Y-%m-%dT%H:%M:%SZ', 'now', '-{CLAIM_LEASE_SECONDS} seconds')))
              ORDER BY next_attempt_at ASC
              LIMIT ?
          )
-         RETURNING id, payload, event_id, attempts, next_attempt_at",
+         RETURNING id, payload, event_id, attempts, next_attempt_at"
-    )
+    );
-    .bind(config.batch_size)
+    let mut rows = sqlx::query_as::<_, QueueRow>(&sql)
-    .fetch_all(pool)
+        .bind(config.batch_size)
-    .await
+        .fetch_all(pool)
-    .map_err(|e| DomainError::InfrastructureError(format!("claim batch: {e}")))?;
+        .await
+        .map_err(|e| DomainError::InfrastructureError(format!("claim batch: {e}")))?;
 
     // RETURNING order is unspecified; deliver oldest first.
     rows.sort_by(|a, b| (&a.next_attempt_at, a.id).cmp(&(&b.next_attempt_at, b.id)));
     Ok(rows)
 }
 
+/// A row that cannot be decoded can never succeed, and has no ack handle to
+/// settle it with; park it so it is neither retried forever nor left claimed.
+async fn dead_letter_undecodable(pool: &SqlitePool, row_id: i64, error: &DomainError) {
+    tracing::error!(
+        row_id,
+        "event queue row is undecodable, dead-lettering: {error}"
+    );
+    let result =
+        sqlx::query("UPDATE event_queue SET status = 'dead_lettered', last_error = ? WHERE id = ?")
+            .bind(format!("undecodable: {error}"))
+            .bind(row_id)
+            .execute(pool)
+            .await;
+    if let Err(e) = result {
+        tracing::error!(row_id, "could not dead-letter undecodable row: {e}");
+    }
+}
+
 fn decode_row(
     pool: &SqlitePool,
     row: QueueRow,

crates/adapters/sqlite/src/event_queue/tests.rs

@@ -190,3 +190,89 @@ mod identity {
         assert!(envelope.id().is_none());
     }
 }
+
+mod lease {
+    use domain::ports::EventConsumer;
+    use futures::StreamExt;
+
+    use super::{claim_batch, config};
+    use crate::event_queue::SqliteEventQueue;
+
+    async fn pool() -> sqlx::SqlitePool {
+        let pool = sqlx::sqlite::SqlitePoolOptions::new()
+            .max_connections(1)
+            .connect("sqlite::memory:")
+            .await
+            .unwrap();
+        crate::migrate(&pool).await.unwrap();
+        pool
+    }
+
+    async fn claim_then_age(pool: &sqlx::SqlitePool, claimed_at: Option<&str>) {
+        super::insert(pool, 1).await;
+        assert_eq!(claim_batch(pool, &config(10)).await.unwrap().len(), 1);
+        sqlx::query("UPDATE event_queue SET claimed_at = ?")
+            .bind(claimed_at)
+            .execute(pool)
+            .await
+            .unwrap();
+    }
+
+    fn ago(seconds: i64) -> String {
+        (chrono::Utc::now() - chrono::Duration::seconds(seconds))
+            .format("%Y-%m-%dT%H:%M:%SZ")
+            .to_string()
+    }
+
+    #[tokio::test]
+    async fn an_expired_claim_is_reclaimed() {
+        let pool = pool().await;
+        claim_then_age(&pool, Some(&ago(super::super::CLAIM_LEASE_SECONDS + 60))).await;
+
+        let again = claim_batch(&pool, &config(10)).await.unwrap();
+
+        assert_eq!(again.len(), 1, "stranded row is claimed again");
+        let claimed: String = sqlx::query_scalar("SELECT claimed_at FROM event_queue")
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+        assert!(claimed > ago(60), "lease restarted: {claimed}");
+    }
+
+    #[tokio::test]
+    async fn a_legacy_processing_row_without_claimed_at_counts_as_expired() {
+        let pool = pool().await;
+        claim_then_age(&pool, None).await;
+
+        assert_eq!(claim_batch(&pool, &config(10)).await.unwrap().len(), 1);
+    }
+
+    #[tokio::test]
+    async fn a_fresh_claim_is_not_reclaimed() {
+        let pool = pool().await;
+        claim_then_age(&pool, Some(&ago(30))).await;
+
+        assert!(claim_batch(&pool, &config(10)).await.unwrap().is_empty());
+    }
+
+    #[tokio::test]
+    async fn an_undecodable_row_is_dead_lettered_not_left_processing() {
+        let pool = pool().await;
+        sqlx::query("INSERT INTO event_queue (event_type, payload) VALUES ('t', 'not json')")
+            .execute(&pool)
+            .await
+            .unwrap();
+        let q = SqliteEventQueue::create(pool.clone(), config(10));
+
+        let first = q.consume().next().await.unwrap();
+        assert!(first.is_err());
+
+        let (status, error): (String, Option<String>) =
+            sqlx::query_as("SELECT status, last_error FROM event_queue")
+                .fetch_one(&pool)
+                .await
+                .unwrap();
+        assert_eq!(status, "dead_lettered");
+        assert!(error.unwrap().contains("undecodable"));
+    }
+}

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

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

crates/bootstrap/src/event_bus.rs

@@ -1,5 +1,6 @@
 use std::sync::Arc;
 
+use crate::outbox_relay::OutboxRelay;
 #[cfg(feature = "nats")]
 use anyhow::Context;
 use domain::ports::{EventConsumer, EventPublisher, EventRelaySink};
@@ -42,7 +43,7 @@ pub async fn create_publisher(
 struct Wiring {
     publisher: Arc<dyn EventPublisher>,
     consumer: Arc<dyn EventConsumer>,
-    relay: Option<(Arc<dyn EventConsumer>, Arc<dyn EventRelaySink>)>,
+    relay: Option<OutboxRelay>,
 }
 
 /// The worker always writes to the outbox. With an external bus
@@ -61,7 +62,7 @@ fn compose(
         Some((consumer, sink)) => Wiring {
             publisher,
             consumer,
-            relay: Some((outbox, sink)),
+            relay: Some(OutboxRelay::new(outbox, sink)),
         },
     }
 }
@@ -69,7 +70,11 @@ fn compose(
 pub async fn create_channel(
     db_pool: &sqlx::SqlitePool,
     config: &config::AppConfig,
-) -> anyhow::Result<(Arc<dyn EventPublisher>, Arc<dyn EventConsumer>)> {
+) -> anyhow::Result<(
+    Arc<dyn EventPublisher>,
+    Arc<dyn EventConsumer>,
+    Option<OutboxRelay>,
+)> {
     let outbox = sqlite::SqliteEventQueue::create_channel(db_pool.clone(), &config.events).await?;
     let wiring = match EventBusBackend::from_config(&config.events)? {
         EventBusBackend::Db => {
@@ -84,10 +89,7 @@ pub async fn create_channel(
             compose(outbox, Some(nats::create_bus(cfg).await?))
         }
     };
-    if let Some((outbox, sink)) = wiring.relay {
+    Ok((wiring.publisher, wiring.consumer, wiring.relay))
-        crate::outbox_relay::spawn(outbox, sink);
-    }
-    Ok((wiring.publisher, wiring.consumer))
 }
 
 #[cfg(test)]
@@ -135,15 +137,13 @@ mod tests {
     #[test]
     fn bus_mode_relays_the_outbox_and_consumes_the_bus() {
         let (publisher, queue) = outbox();
-        let queue_ptr = Arc::as_ptr(&queue);
         let bus_consumer: Arc<dyn EventConsumer> = Arc::new(Nop);
         let bus_ptr = Arc::as_ptr(&bus_consumer);
         let sink: Arc<dyn EventRelaySink> = Arc::new(Nop);
 
         let wiring = compose((publisher, queue), Some((bus_consumer, sink)));
 
-        let (source, _) = wiring.relay.expect("relay spawned in bus mode");
+        assert!(wiring.relay.is_some(), "relay built in bus mode");
-        assert!(std::ptr::addr_eq(Arc::as_ptr(&source), queue_ptr));
         assert!(std::ptr::addr_eq(Arc::as_ptr(&wiring.consumer), bus_ptr));
     }
 }

crates/bootstrap/src/outbox_relay.rs

@@ -1,11 +1,16 @@
 //! Moves outbox rows onto the message bus. The worker runs it when the bus is
 //! not the outbox itself (`EVENT_BUS_BACKEND=nats`).
 //!
-//! Rows are claimed by the SQLite queue's own atomic claim, so a row is only
+//! Rows are claimed by the SQLite queue's own atomic claim, so a row is
-//! ever relayed by one claimant and failures take the queue's existing
+//! relayed by one claimant at a time, and failures take the queue's backoff
-//! backoff and dead-letter path. A row is marked done only after the sink
+//! and dead-letter path. A row is marked done only after the sink reports the
-//! reports the bus accepted it; a crash in between re-relays the row, and the
+//! bus accepted it.
-//! bus de-duplicates on the event id.
+//!
+//! Delivery is at-least-once. A crash between publish and mark-done leaves the
+//! row `processing`; when its claim lease lapses it is claimed and relayed
+//! again. The bus drops that repeat while it is inside the broker's dedup
+//! window (same `Nats-Msg-Id`); beyond the window, or in core mode, the
+//! consumer ledger makes the handlers skip it.
 
 use std::sync::Arc;
 
@@ -14,21 +19,43 @@ use domain::{
     ports::{EventConsumer, EventRelaySink},
 };
 use futures::StreamExt;
-use tokio::task::JoinHandle;
+use tokio::sync::watch;
 
-pub fn spawn(source: Arc<dyn EventConsumer>, sink: Arc<dyn EventRelaySink>) -> JoinHandle<()> {
+pub struct OutboxRelay {
-    tokio::spawn(async move {
+    source: Arc<dyn EventConsumer>,
-        let mut rows = source.consume();
+    sink: Arc<dyn EventRelaySink>,
-        while let Some(next) = rows.next().await {
+}
-            match next {
+
-                Ok(envelope) => relay_envelope(envelope, sink.as_ref()).await,
+impl OutboxRelay {
-                // A row that cannot even be decoded stays claimed; there is
+    pub fn new(source: Arc<dyn EventConsumer>, sink: Arc<dyn EventRelaySink>) -> Self {
-                // no handle to settle it with.
+        Self { source, sink }
-                Err(e) => tracing::error!("outbox relay: undecodable row: {e}"),
+    }
+
+    /// Runs until `shutdown` fires or the source ends. A shutdown never
+    /// interrupts a row mid-relay: it is acked or nacked before this returns.
+    /// Rows the claim already took but the relay had not started stay
+    /// `processing` and come back when their lease lapses.
+    pub async fn run(self, mut shutdown: watch::Receiver<bool>) {
+        let mut rows = self.source.consume();
+        loop {
+            tokio::select! {
+                biased;
+                _ = shutdown.changed() => {
+                    tracing::info!("outbox relay: shutdown signal received, stopping");
+                    return;
+                }
+                next = rows.next() => match next {
+                    Some(Ok(envelope)) => relay_envelope(envelope, self.sink.as_ref()).await,
+                    // The queue already dead-lettered it and logged the row id.
+                    Some(Err(e)) => tracing::error!("outbox relay: skipped undecodable row: {e}"),
+                    None => {
+                        tracing::error!("outbox relay: source closed");
+                        return;
+                    }
+                },
             }
         }
-        tracing::warn!("outbox relay: source closed, relay stopped");
+    }
-    })
 }
 
 async fn relay_envelope(envelope: EventEnvelope, sink: &dyn EventRelaySink) {

crates/bootstrap/src/worker.rs

@@ -8,9 +8,65 @@ use domain::value_objects::InstanceIdentity;
 
 use crate::database::{self, DatabaseOutput};
 use crate::event_bus;
+use crate::outbox_relay::OutboxRelay;
+use tokio::sync::watch;
 
 pub struct WorkerApp {
-    pub service: WorkerService,
+    service: WorkerService,
+    /// Present when the bus is external: drains the outbox into it.
+    relay: Option<OutboxRelay>,
+}
+
+impl WorkerApp {
+    /// Runs the handlers and, when there is one, the outbox relay, until
+    /// `shutdown` fires. Both stop together and finish their in-flight work.
+    ///
+    /// If the relay stops or panics on its own, the worker stops too and
+    /// returns an error so the process exits non-zero and the supervisor
+    /// restarts it: a worker that keeps handling events but no longer
+    /// delivers the outbox would fail silently.
+    pub async fn run(self, mut shutdown: watch::Receiver<bool>) -> anyhow::Result<()> {
+        let (stop_tx, stop_rx) = watch::channel(false);
+        let mut relay = self.relay.map(|r| tokio::spawn(r.run(stop_rx.clone())));
+        let service = self.service.run(stop_rx);
+        tokio::pin!(service);
+
+        let mut stopping = false;
+        let mut relay_failure = None;
+        loop {
+            tokio::select! {
+                _ = &mut service => break,
+                _ = shutdown.changed(), if !stopping => {
+                    stopping = true;
+                    let _ = stop_tx.send(true);
+                }
+                result = async { relay.as_mut().expect("guarded").await }, if relay.is_some() => {
+                    relay = None;
+                    if !stopping {
+                        relay_failure = Some(match result {
+                            Ok(()) => "outbox relay stopped unexpectedly".to_owned(),
+                            Err(e) => format!("outbox relay died: {e}"),
+                        });
+                        stopping = true;
+                        let _ = stop_tx.send(true);
+                    }
+                }
+            }
+        }
+        let _ = stop_tx.send(true);
+        if let Some(handle) = relay
+            && let Err(e) = handle.await
+        {
+            relay_failure.get_or_insert(format!("outbox relay died: {e}"));
+        }
+        match relay_failure {
+            Some(message) => {
+                tracing::error!("{message}");
+                Err(anyhow::anyhow!(message))
+            }
+            None => Ok(()),
+        }
+    }
 }
 
 /// Wires the background worker: event handlers, periodic jobs (already
@@ -58,7 +114,7 @@ async fn assemble(
         .clone()
         .map(|c| c as Arc<dyn PersonEnrichmentClient>);
 
-    let (event_publisher_arc, consumer_arc) =
+    let (event_publisher_arc, consumer_arc, relay) =
         event_bus::create_channel(&db.db_pool, config).await?;
 
     let worker_services = application::WorkerServices {
@@ -246,5 +302,6 @@ async fn assemble(
 
     Ok(WorkerApp {
         service: WorkerService::new(consumer_arc, handlers),
+        relay,
     })
 }

crates/bootstrap/tests/outbox_relay_test.rs

@@ -5,6 +5,7 @@ mod support;
 
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
+use tokio::sync::Notify;
 
 use async_trait::async_trait;
 use domain::{
@@ -19,6 +20,9 @@ use uuid::Uuid;
 #[derive(Default)]
 struct FakeSink {
     fail: bool,
+    /// When set, `relay` signals `started` then takes this long.
+    slow: Option<Duration>,
+    started: Notify,
     seen: Mutex<Vec<Option<EventId>>>,
 }
 
@@ -26,6 +30,10 @@ struct FakeSink {
 impl EventRelaySink for FakeSink {
     async fn relay(&self, id: Option<&EventId>, _event: &DomainEvent) -> Result<(), DomainError> {
         self.seen.lock().unwrap().push(id.cloned());
+        if let Some(delay) = self.slow {
+            self.started.notify_one();
+            tokio::time::sleep(delay).await;
+        }
         if self.fail {
             Err(DomainError::InfrastructureError("bus down".into()))
         } else {
@@ -65,6 +73,18 @@ async fn wait_for(pool: &sqlx::SqlitePool, status: &str) {
     panic!("row never reached {status}: {:?}", row(pool).await);
 }
 
+fn start(
+    queue: SqliteEventQueue,
+    sink: Arc<FakeSink>,
+) -> (
+    tokio::sync::watch::Sender<bool>,
+    tokio::task::JoinHandle<()>,
+) {
+    let (stop, rx) = tokio::sync::watch::channel(false);
+    let relay = bootstrap::outbox_relay::OutboxRelay::new(Arc::new(queue), sink);
+    (stop, tokio::spawn(relay.run(rx)))
+}
+
 fn event() -> DomainEvent {
     DomainEvent::UserUpdated {
         user_id: UserId::from_uuid(Uuid::new_v4()),
@@ -77,9 +97,10 @@ async fn relay_marks_rows_done_only_after_publish_succeeds() {
     publisher.publish(&event()).await.unwrap();
     let sink = Arc::new(FakeSink::default());
 
-    let relay = bootstrap::outbox_relay::spawn(Arc::new(queue), sink.clone());
+    let (stop, relay) = start(queue, sink.clone());
     wait_for(&pool, "done").await;
     relay.abort();
+    drop(stop);
 
     let (_, attempts, event_id) = row(&pool).await;
     assert_eq!(attempts, 0);
@@ -101,7 +122,7 @@ async fn a_failed_publish_leaves_the_row_pending_with_an_attempt_counted() {
         ..Default::default()
     });
 
-    let relay = bootstrap::outbox_relay::spawn(Arc::new(queue), sink.clone());
+    let (stop, relay) = start(queue, sink.clone());
     for _ in 0..200 {
         if row(&pool).await.1 == 1 {
             break;
@@ -109,7 +130,35 @@ async fn a_failed_publish_leaves_the_row_pending_with_an_attempt_counted() {
         tokio::time::sleep(Duration::from_millis(10)).await;
     }
     relay.abort();
+    drop(stop);
 
     let (status, attempts, _) = row(&pool).await;
     assert_eq!((status.as_str(), attempts), ("pending", 1));
 }
+
+#[tokio::test]
+async fn shutdown_lets_the_in_flight_row_finish_then_stops_the_relay() {
+    let (pool, publisher, queue) = outbox().await;
+    publisher.publish(&event()).await.unwrap();
+    publisher.publish(&event()).await.unwrap();
+    let sink = Arc::new(FakeSink {
+        slow: Some(Duration::from_millis(200)),
+        ..Default::default()
+    });
+    let (stop, relay) = start(queue, sink.clone());
+
+    sink.started.notified().await;
+    stop.send(true).unwrap();
+    tokio::time::timeout(Duration::from_secs(5), relay)
+        .await
+        .expect("relay stops on shutdown")
+        .unwrap();
+
+    let statuses: Vec<String> = sqlx::query_scalar("SELECT status FROM event_queue ORDER BY id")
+        .fetch_all(&pool)
+        .await
+        .unwrap();
+    assert_eq!(statuses[0], "done", "in-flight row was completed");
+    assert_ne!(statuses[0], "processing");
+    assert_eq!(sink.seen.lock().unwrap().len(), 1, "no new row after stop");
+}

crates/worker/src/main.rs

@@ -15,7 +15,5 @@ async fn main() -> anyhow::Result<()> {
     });
 
     tracing::info!("worker started");
-    app.service.run(shutdown_rx).await;
+    app.run(shutdown_rx).await
-
-    Ok(())
 }