From 53892847f24d8bbed084d8e63c13fe90467d41fb Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Mon, 26 Oct 2026 00:24:56 +0100 Subject: [PATCH] fix: per-handler claim lease in ledger, nats max_deliver, pin handler names --- crates/adapters/nats/src/config.rs | 8 +- crates/adapters/nats/src/consumer.rs | 6 +- .../adapters/nats/tests/nats_config_test.rs | 44 +++- .../migrations/0101_outbox_identity.sql | 7 +- .../src/repositories/processed_events/mod.rs | 243 ++++++++++++++---- .../tests/fixtures/migration_ledger.txt | 2 +- crates/application/src/worker.rs | 51 ++-- .../jobs/processed_events_cleanup_test.rs | 8 +- .../application/tests/worker/worker_test.rs | 80 ++++++ crates/bootstrap/src/event_bus.rs | 2 +- crates/bootstrap/tests/bootstrap_test.rs | 25 +- crates/domain/src/ports/processed_events.rs | 40 ++- crates/domain/src/testing/fakes/store.rs | 26 +- .../testing/fakes/store_processed_events.rs | 65 ++++- 14 files changed, 489 insertions(+), 118 deletions(-) diff --git a/crates/adapters/nats/src/config.rs b/crates/adapters/nats/src/config.rs index 34003cac..e4f31404 100644 --- a/crates/adapters/nats/src/config.rs +++ b/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 { + pub fn from_config( + config: &::config::NatsConfig, + events: &::config::EventsConfig, + ) -> anyhow::Result { 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), }) } } diff --git a/crates/adapters/nats/src/consumer.rs b/crates/adapters/nats/src/consumer.rs index 59508212..fcea56fb 100644 --- a/crates/adapters/nats/src/consumer.rs +++ b/crates/adapters/nats/src/consumer.rs @@ -108,8 +108,9 @@ fn consume_from_rx(rx: EnvelopeRx) -> BoxStream<'static, Result, 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(§ion(Some("nats://test:4222"), "core")).unwrap(); + let cfg = NatsConfig::from_config( + §ion(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(§ion(Some("nats://test:4222"), "kafka")).is_err()); + assert!( + NatsConfig::from_config( + §ion(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(§ion(Some("nats://test:4222"), "jetstream"), &events).unwrap(); + assert_eq!(cfg.max_deliver, 9); } diff --git a/crates/adapters/sqlite/migrations/0101_outbox_identity.sql b/crates/adapters/sqlite/migrations/0101_outbox_identity.sql index f045ae91..a7aa1f41 100644 --- a/crates/adapters/sqlite/migrations/0101_outbox_identity.sql +++ b/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) ); diff --git a/crates/adapters/sqlite/src/repositories/processed_events/mod.rs b/crates/adapters/sqlite/src/repositories/processed_events/mod.rs index 6dd277b7..453707e9 100644 --- a/crates/adapters/sqlite/src/repositories/processed_events/mod.rs +++ b/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 { - let found: Option = - 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 { + 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 = 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 { 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 + ); } } diff --git a/crates/adapters/sqlite/tests/fixtures/migration_ledger.txt b/crates/adapters/sqlite/tests/fixtures/migration_ledger.txt index 33b259e2..704ea667 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 C0EE8C5B9FA6A2BE75674A10A008B895FFAF8056139FE0C0F9936B27BA06AFC71B1DC5BD8318948F6E9FCF0689823B71 +101 19A2542D4343F3921A8BF0A10E195DA1915E3EA97A2BCF2D3BF9079504DB2053E4BD2175D94C95BCA363BC2FDF14AA88 diff --git a/crates/application/src/worker.rs b/crates/application/src/worker.rs index 7f970d30..3a768e9b 100644 --- a/crates/application/src/worker.rs +++ b/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}" + ); + } } } } diff --git a/crates/application/tests/jobs/processed_events_cleanup_test.rs b/crates/application/tests/jobs/processed_events_cleanup_test.rs index dcefa601..8ce87ede 100644 --- a/crates/application/tests/jobs/processed_events_cleanup_test.rs +++ b/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] diff --git a/crates/application/tests/worker/worker_test.rs b/crates/application/tests/worker/worker_test.rs index c3a049f0..2d19f26a 100644 --- a/crates/application/tests/worker/worker_test.rs +++ b/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); +} diff --git a/crates/bootstrap/src/event_bus.rs b/crates/bootstrap/src/event_bus.rs index 34fd7f00..ba7ded83 100644 --- a/crates/bootstrap/src/event_bus.rs +++ b/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?)) diff --git a/crates/bootstrap/tests/bootstrap_test.rs b/crates/bootstrap/tests/bootstrap_test.rs index 7263272b..93792411 100644 --- a/crates/bootstrap/tests/bootstrap_test.rs +++ b/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); } diff --git a/crates/domain/src/ports/processed_events.rs b/crates/domain/src/ports/processed_events.rs index 76b35c89..d71a2553 100644 --- a/crates/domain/src/ports/processed_events.rs +++ b/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; + /// 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; - /// 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; } diff --git a/crates/domain/src/testing/fakes/store.rs b/crates/domain/src/testing/fakes/store.rs index 808c1fb9..94f3ca69 100644 --- a/crates/domain/src/testing/fakes/store.rs +++ b/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>, pub(super) follows: RwLock>, pub(super) blocked: RwLock>, - pub(super) processed_events: RwLock>, + pub(super) processed_events: RwLock>, } 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 { diff --git a/crates/domain/src/testing/fakes/store_processed_events.rs b/crates/domain/src/testing/fakes/store_processed_events.rs index f87b14a8..f1f96eb7 100644 --- a/crates/domain/src/testing/fakes/store_processed_events.rs +++ b/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 { - Ok(self - .processed_events - .read() - .unwrap() - .contains_key(&(id.value(), handler.to_owned()))) + async fn begin(&self, id: &EventId, handler: &str) -> Result { + 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 { + 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 { 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) } }