diff --git a/crates/adapters/nats/src/consumer.rs b/crates/adapters/nats/src/consumer.rs index f428ab80..59508212 100644 --- a/crates/adapters/nats/src/consumer.rs +++ b/crates/adapters/nats/src/consumer.rs @@ -106,6 +106,12 @@ fn consume_from_rx(rx: EnvelopeRx) -> BoxStream<'static, Result { 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, DomainError> { - let mut rows = sqlx::query_as::<_, QueueRow>( - "UPDATE event_queue SET status = 'processing' + let sql = format!( + "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' - AND next_attempt_at <= strftime('%Y-%m-%dT%H:%M:%SZ', 'now') + WHERE (status = 'pending' + 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", - ) - .bind(config.batch_size) - .fetch_all(pool) - .await - .map_err(|e| DomainError::InfrastructureError(format!("claim batch: {e}")))?; + RETURNING id, payload, event_id, attempts, next_attempt_at" + ); + let mut rows = sqlx::query_as::<_, QueueRow>(&sql) + .bind(config.batch_size) + .fetch_all(pool) + .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, diff --git a/crates/adapters/sqlite/src/event_queue/tests.rs b/crates/adapters/sqlite/src/event_queue/tests.rs index 79b66e9f..8025b5f6 100644 --- a/crates/adapters/sqlite/src/event_queue/tests.rs +++ b/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) = + 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")); + } +} diff --git a/crates/adapters/sqlite/tests/fixtures/migration_ledger.txt b/crates/adapters/sqlite/tests/fixtures/migration_ledger.txt index 93d4a5c3..33b259e2 100644 --- a/crates/adapters/sqlite/tests/fixtures/migration_ledger.txt +++ b/crates/adapters/sqlite/tests/fixtures/migration_ledger.txt @@ -29,4 +29,4 @@ 29 56F4F7B41A09F07211D7543204F0EAC93B7BFB15AFD9BC90DF2654DAE5BE96B1E358B50E2190BF25739C5F0B1FD4B806 30 DAAA0AF35B4ACB271A9D8068B82F02A72EEC62F3AC07ED504C7A7887976C2C8BACB35FFF477B63A931DF15F79018390D 100 310ADC9ABF0CC2F8E9EDAB33168D009D64228FF5BD8D2A14185513699EBB782606A1EEA8FEE6EC6AF65D72396153F13F -101 D4712BF715F8BD86502ED8162488998B1BAFF41C241072173BC2210A3676C05B5B373DD3ED33A16B473A2EC983211302 +101 C0EE8C5B9FA6A2BE75674A10A008B895FFAF8056139FE0C0F9936B27BA06AFC71B1DC5BD8318948F6E9FCF0689823B71 diff --git a/crates/bootstrap/src/event_bus.rs b/crates/bootstrap/src/event_bus.rs index d83486a1..34fd7f00 100644 --- a/crates/bootstrap/src/event_bus.rs +++ b/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, consumer: Arc, - relay: Option<(Arc, Arc)>, + relay: Option, } /// 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, Arc)> { +) -> anyhow::Result<( + Arc, + Arc, + Option, +)> { 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 { - crate::outbox_relay::spawn(outbox, sink); - } - Ok((wiring.publisher, wiring.consumer)) + Ok((wiring.publisher, wiring.consumer, wiring.relay)) } #[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 = Arc::new(Nop); let bus_ptr = Arc::as_ptr(&bus_consumer); let sink: Arc = Arc::new(Nop); let wiring = compose((publisher, queue), Some((bus_consumer, sink))); - let (source, _) = wiring.relay.expect("relay spawned in bus mode"); - assert!(std::ptr::addr_eq(Arc::as_ptr(&source), queue_ptr)); + assert!(wiring.relay.is_some(), "relay built in bus mode"); assert!(std::ptr::addr_eq(Arc::as_ptr(&wiring.consumer), bus_ptr)); } } diff --git a/crates/bootstrap/src/outbox_relay.rs b/crates/bootstrap/src/outbox_relay.rs index 5264adf0..86d25152 100644 --- a/crates/bootstrap/src/outbox_relay.rs +++ b/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 -//! ever relayed by one claimant and failures take the queue's existing -//! backoff and dead-letter path. A row is marked done only after the sink -//! reports the bus accepted it; a crash in between re-relays the row, and the -//! bus de-duplicates on the event id. +//! Rows are claimed by the SQLite queue's own atomic claim, so a row is +//! relayed by one claimant at a time, and failures take the queue's backoff +//! and dead-letter path. A row is marked done only after the sink reports the +//! bus accepted it. +//! +//! 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, sink: Arc) -> JoinHandle<()> { - tokio::spawn(async move { - let mut rows = source.consume(); - while let Some(next) = rows.next().await { - match next { - Ok(envelope) => relay_envelope(envelope, sink.as_ref()).await, - // A row that cannot even be decoded stays claimed; there is - // no handle to settle it with. - Err(e) => tracing::error!("outbox relay: undecodable row: {e}"), +pub struct OutboxRelay { + source: Arc, + sink: Arc, +} + +impl OutboxRelay { + pub fn new(source: Arc, sink: Arc) -> Self { + Self { source, sink } + } + + /// 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) { + 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) { diff --git a/crates/bootstrap/src/worker.rs b/crates/bootstrap/src/worker.rs index 1c49dce9..7530520e 100644 --- a/crates/bootstrap/src/worker.rs +++ b/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, +} + +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) -> 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); - 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, }) } diff --git a/crates/bootstrap/tests/outbox_relay_test.rs b/crates/bootstrap/tests/outbox_relay_test.rs index 3cad162e..e424af6a 100644 --- a/crates/bootstrap/tests/outbox_relay_test.rs +++ b/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, + started: Notify, seen: Mutex>>, } @@ -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, +) -> ( + tokio::sync::watch::Sender, + 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 = 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"); +} diff --git a/crates/worker/src/main.rs b/crates/worker/src/main.rs index 592476b4..6be9ed37 100644 --- a/crates/worker/src/main.rs +++ b/crates/worker/src/main.rs @@ -15,7 +15,5 @@ async fn main() -> anyhow::Result<()> { }); tracing::info!("worker started"); - app.service.run(shutdown_rx).await; - - Ok(()) + app.run(shutdown_rx).await }