From 5681d0c034e94a61c766ecf8c438b26a1e973670 Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Mon, 26 Oct 2026 00:17:28 +0100 Subject: [PATCH] feat: processed-event ledger, idempotent dispatch --- .../adapters/activitypub/src/event_handler.rs | 4 + .../adapters/image-converter/src/handler.rs | 4 + crates/adapters/object-storage/src/lib.rs | 4 + crates/adapters/sqlite/src/lib.rs | 12 +- .../adapters/sqlite/src/repositories/mod.rs | 2 + .../src/repositories/processed_events/mod.rs | 120 ++++++++++++ crates/application/src/jobs/mod.rs | 2 + .../src/jobs/processed_events_cleanup.rs | 39 ++++ .../src/movies/discovery_indexer.rs | 4 + .../application/src/movies/event_handler.rs | 4 + .../src/movies/poster_sync_handler.rs | 4 + .../application/src/movies/reindex_search.rs | 4 + .../application/src/movies/search_cleanup.rs | 4 + .../application/src/person/event_handler.rs | 4 + crates/application/src/worker.rs | 79 ++++++-- .../application/src/wrapup/event_handler.rs | 4 + crates/application/tests/jobs.rs | 2 + .../jobs/processed_events_cleanup_test.rs | 35 ++++ .../application/tests/worker/worker_test.rs | 181 +++++++++++++++++- crates/bootstrap/src/database.rs | 2 + .../bootstrap/src/follow_backfill_handler.rs | 4 + crates/bootstrap/src/worker.rs | 11 +- crates/bootstrap/tests/bootstrap_test.rs | 31 +++ crates/domain/src/ports/events.rs | 3 + crates/domain/src/ports/mod.rs | 2 + crates/domain/src/ports/processed_events.rs | 16 ++ crates/domain/src/testing/fakes/mod.rs | 1 + crates/domain/src/testing/fakes/store.rs | 19 ++ .../testing/fakes/store_processed_events.rs | 33 ++++ 29 files changed, 610 insertions(+), 24 deletions(-) create mode 100644 crates/adapters/sqlite/src/repositories/processed_events/mod.rs create mode 100644 crates/application/src/jobs/processed_events_cleanup.rs create mode 100644 crates/application/tests/jobs.rs create mode 100644 crates/application/tests/jobs/processed_events_cleanup_test.rs create mode 100644 crates/domain/src/ports/processed_events.rs create mode 100644 crates/domain/src/testing/fakes/store_processed_events.rs diff --git a/crates/adapters/activitypub/src/event_handler.rs b/crates/adapters/activitypub/src/event_handler.rs index 09f6b829..7ee4e60f 100644 --- a/crates/adapters/activitypub/src/event_handler.rs +++ b/crates/adapters/activitypub/src/event_handler.rs @@ -55,6 +55,10 @@ impl ActivityPubEventHandler { #[async_trait] impl EventHandler for ActivityPubEventHandler { + fn name(&self) -> &'static str { + "activitypub" + } + async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> { match event { DomainEvent::ReviewLogged { diff --git a/crates/adapters/image-converter/src/handler.rs b/crates/adapters/image-converter/src/handler.rs index e0ae7b14..4664e687 100644 --- a/crates/adapters/image-converter/src/handler.rs +++ b/crates/adapters/image-converter/src/handler.rs @@ -31,6 +31,10 @@ impl ImageConversionHandler { #[async_trait] impl EventHandler for ImageConversionHandler { + fn name(&self) -> &'static str { + "image_conversion" + } + async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> { let key = match event { DomainEvent::ImageStored { key } => key.clone(), diff --git a/crates/adapters/object-storage/src/lib.rs b/crates/adapters/object-storage/src/lib.rs index a1144e57..b55fde82 100644 --- a/crates/adapters/object-storage/src/lib.rs +++ b/crates/adapters/object-storage/src/lib.rs @@ -112,6 +112,10 @@ impl ImageCleanupHandler { #[async_trait] impl EventHandler for ImageCleanupHandler { + fn name(&self) -> &'static str { + "image_cleanup" + } + async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> { let poster_path = match event { DomainEvent::MovieDeleted { poster_path, .. } => poster_path, diff --git a/crates/adapters/sqlite/src/lib.rs b/crates/adapters/sqlite/src/lib.rs index 34bb266b..7a5c486f 100644 --- a/crates/adapters/sqlite/src/lib.rs +++ b/crates/adapters/sqlite/src/lib.rs @@ -10,12 +10,12 @@ pub use repositories::{ SqliteApContentQuery, SqliteDiaryRepository, SqliteImageRefAdapter, SqliteImportProfileRepository, SqliteImportSessionRepository, SqliteInsightsQuery, SqliteMovieDeduplicator, SqliteMovieProfileRepository, SqliteMovieRepository, - SqlitePersonAdapter, SqliteProfileFieldsRepository, SqliteRefreshSessionAdapter, - SqliteRemoteGoalRepository, SqliteReviewRepository, SqliteSocialRepository, - SqliteStatsRepository, SqliteUserRepository, SqliteWatchEventRepository, - SqliteWatchlistRepository, SqliteWebhookTokenRepository, SqliteWrapUpRepository, - SqliteWrapUpStatsQuery, create_federated_profile_query, create_image_ref, - create_person_adapter, create_search_adapter, + SqlitePersonAdapter, SqliteProcessedEventLedger, SqliteProfileFieldsRepository, + SqliteRefreshSessionAdapter, SqliteRemoteGoalRepository, SqliteReviewRepository, + SqliteSocialRepository, SqliteStatsRepository, SqliteUserRepository, + SqliteWatchEventRepository, SqliteWatchlistRepository, SqliteWebhookTokenRepository, + SqliteWrapUpRepository, SqliteWrapUpStatsQuery, create_federated_profile_query, + create_image_ref, create_person_adapter, create_search_adapter, }; use repositories::{SqliteGoalRepository, SqliteUserSettingsRepository}; diff --git a/crates/adapters/sqlite/src/repositories/mod.rs b/crates/adapters/sqlite/src/repositories/mod.rs index 53803c82..ac467d51 100644 --- a/crates/adapters/sqlite/src/repositories/mod.rs +++ b/crates/adapters/sqlite/src/repositories/mod.rs @@ -8,6 +8,7 @@ mod insights; mod movie; mod movie_dedup; mod persons; +mod processed_events; mod profile; mod profile_fields; mod refresh_sessions; @@ -32,6 +33,7 @@ pub use insights::SqliteInsightsQuery; pub use movie::SqliteMovieRepository; pub use movie_dedup::SqliteMovieDeduplicator; pub use persons::{SqlitePersonAdapter, create_person_adapter}; +pub use processed_events::SqliteProcessedEventLedger; pub use profile::SqliteMovieProfileRepository; pub use profile_fields::SqliteProfileFieldsRepository; pub use refresh_sessions::SqliteRefreshSessionAdapter; diff --git a/crates/adapters/sqlite/src/repositories/processed_events/mod.rs b/crates/adapters/sqlite/src/repositories/processed_events/mod.rs new file mode 100644 index 00000000..6dd277b7 --- /dev/null +++ b/crates/adapters/sqlite/src/repositories/processed_events/mod.rs @@ -0,0 +1,120 @@ +use async_trait::async_trait; +use domain::{errors::DomainError, events::EventId, ports::ProcessedEventLedger}; +use sqlx::SqlitePool; + +/// Matches the column default in migration 0101, so string comparison orders rows by time. +const TIMESTAMP_FORMAT: &str = "%Y-%m-%dT%H:%M:%SZ"; + +pub struct SqliteProcessedEventLedger { + pool: SqlitePool, +} + +impl SqliteProcessedEventLedger { + pub fn new(pool: SqlitePool) -> Self { + Self { pool } + } +} + +#[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 (?, ?)") + .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()) + .execute(&self.pool) + .await + .map_err(crate::shared::map_sqlx_error)?; + Ok(result.rows_affected()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + async fn ledger() -> (SqliteProcessedEventLedger, SqlitePool) { + let pool = sqlx::sqlite::SqlitePoolOptions::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .unwrap(); + crate::migrate(&pool).await.unwrap(); + (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()); + } + + #[tokio::test] + async fn recording_twice_is_not_an_error() { + let (ledger, pool) = ledger().await; + let id = EventId::new(); + ledger.record_processed(&id, "a").await.unwrap(); + ledger.record_processed(&id, "a").await.unwrap(); + + let rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM processed_events") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(rows, 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 cutoff = chrono::NaiveDate::from_ymd_opt(2026, 3, 1) + .unwrap() + .and_hms_opt(0, 0, 0) + .unwrap(); + + 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()); + } +} diff --git a/crates/application/src/jobs/mod.rs b/crates/application/src/jobs/mod.rs index 93fab650..58115c4a 100644 --- a/crates/application/src/jobs/mod.rs +++ b/crates/application/src/jobs/mod.rs @@ -1,6 +1,7 @@ mod enrichment_staleness; mod import_cleanup; mod movie_dedup; +mod processed_events_cleanup; mod refresh_session_cleanup; mod watch_event_cleanup; mod wrapup; @@ -8,6 +9,7 @@ mod wrapup; pub use enrichment_staleness::EnrichmentStalenessJob; pub use import_cleanup::ImportSessionCleanupJob; pub use movie_dedup::MovieDeduplicationJob; +pub use processed_events_cleanup::{PROCESSED_EVENTS_RETENTION_DAYS, ProcessedEventsCleanupJob}; pub use refresh_session_cleanup::RefreshSessionCleanupJob; pub use watch_event_cleanup::WatchEventCleanupJob; pub use wrapup::{WrapUpAutoGenerateJob, WrapUpCleanupJob}; diff --git a/crates/application/src/jobs/processed_events_cleanup.rs b/crates/application/src/jobs/processed_events_cleanup.rs new file mode 100644 index 00000000..7e83302d --- /dev/null +++ b/crates/application/src/jobs/processed_events_cleanup.rs @@ -0,0 +1,39 @@ +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use domain::{ + errors::DomainError, + ports::{PeriodicJob, ProcessedEventLedger}, +}; + +/// Ledger rows older than this can no longer matter: the queue gives up on an +/// event long before, so it cannot be redelivered. +pub const PROCESSED_EVENTS_RETENTION_DAYS: i64 = 30; + +pub struct ProcessedEventsCleanupJob { + ledger: Arc, +} + +impl ProcessedEventsCleanupJob { + pub fn new(ledger: Arc) -> Self { + Self { ledger } + } +} + +#[async_trait] +impl PeriodicJob for ProcessedEventsCleanupJob { + fn interval(&self) -> Duration { + Duration::from_secs(24 * 3600) + } + + async fn run(&self) -> Result<(), DomainError> { + let cutoff = chrono::Utc::now().naive_utc() + - chrono::Duration::days(PROCESSED_EVENTS_RETENTION_DAYS); + let n = self.ledger.delete_older_than(cutoff).await?; + if n > 0 { + tracing::info!("processed events cleanup: removed {n} ledger rows"); + } + Ok(()) + } +} diff --git a/crates/application/src/movies/discovery_indexer.rs b/crates/application/src/movies/discovery_indexer.rs index 82819fd3..1ba14266 100644 --- a/crates/application/src/movies/discovery_indexer.rs +++ b/crates/application/src/movies/discovery_indexer.rs @@ -27,6 +27,10 @@ impl MovieDiscoveryIndexer { #[async_trait] impl EventHandler for MovieDiscoveryIndexer { + fn name(&self) -> &'static str { + "movie_discovery_indexer" + } + async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> { let movie_id = match event { DomainEvent::MovieDiscovered { movie_id, .. } => movie_id.clone(), diff --git a/crates/application/src/movies/event_handler.rs b/crates/application/src/movies/event_handler.rs index fdc70b3f..d904cd64 100644 --- a/crates/application/src/movies/event_handler.rs +++ b/crates/application/src/movies/event_handler.rs @@ -70,6 +70,10 @@ impl MovieEnrichmentHandler { #[async_trait] impl EventHandler for MovieEnrichmentHandler { + fn name(&self) -> &'static str { + "movie_enrichment" + } + async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> { let (movie_id, external_metadata_id) = match event { DomainEvent::MovieEnrichmentRequested { diff --git a/crates/application/src/movies/poster_sync_handler.rs b/crates/application/src/movies/poster_sync_handler.rs index c7583013..874f6855 100644 --- a/crates/application/src/movies/poster_sync_handler.rs +++ b/crates/application/src/movies/poster_sync_handler.rs @@ -106,6 +106,10 @@ impl PosterSyncHandler { #[async_trait] impl EventHandler for PosterSyncHandler { + fn name(&self) -> &'static str { + "poster_sync" + } + async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> { let (movie_id, external_metadata_id) = match event { DomainEvent::MovieDiscovered { diff --git a/crates/application/src/movies/reindex_search.rs b/crates/application/src/movies/reindex_search.rs index bf3bba5e..d833eb0f 100644 --- a/crates/application/src/movies/reindex_search.rs +++ b/crates/application/src/movies/reindex_search.rs @@ -132,6 +132,10 @@ impl SearchReindexHandler { #[async_trait] impl EventHandler for SearchReindexHandler { + fn name(&self) -> &'static str { + "search_reindex" + } + async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> { if !matches!(event, DomainEvent::SearchReindexRequested) { return Ok(()); diff --git a/crates/application/src/movies/search_cleanup.rs b/crates/application/src/movies/search_cleanup.rs index 280c42a4..88f3fddc 100644 --- a/crates/application/src/movies/search_cleanup.rs +++ b/crates/application/src/movies/search_cleanup.rs @@ -24,6 +24,10 @@ impl SearchCleanupHandler { #[async_trait] impl EventHandler for SearchCleanupHandler { + fn name(&self) -> &'static str { + "search_cleanup" + } + async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> { let movie_id = match event { DomainEvent::MovieDeleted { movie_id, .. } => movie_id.value().to_string(), diff --git a/crates/application/src/person/event_handler.rs b/crates/application/src/person/event_handler.rs index 9d99b93e..9506669b 100644 --- a/crates/application/src/person/event_handler.rs +++ b/crates/application/src/person/event_handler.rs @@ -31,6 +31,10 @@ impl PersonEnrichmentHandler { #[async_trait] impl EventHandler for PersonEnrichmentHandler { + fn name(&self) -> &'static str { + "person_enrichment" + } + async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> { let (person_id, external_person_id) = match event { DomainEvent::PersonEnrichmentRequested { diff --git a/crates/application/src/worker.rs b/crates/application/src/worker.rs index 64cd50e0..7f970d30 100644 --- a/crates/application/src/worker.rs +++ b/crates/application/src/worker.rs @@ -1,8 +1,9 @@ use std::sync::Arc; use domain::{ + errors::DomainError, events::EventEnvelope, - ports::{EventConsumer, EventHandler}, + ports::{EventConsumer, EventHandler, ProcessedEventLedger}, }; use futures::StreamExt; use tokio::sync::Semaphore; @@ -12,20 +13,42 @@ const DEFAULT_CONCURRENCY: usize = 4; pub struct WorkerService { consumer: Arc, handlers: Vec>, + ledger: Arc, semaphore: Arc, } impl WorkerService { - pub fn new(consumer: Arc, handlers: Vec>) -> Self { - Self { + /// Fails if two handlers share a name: the name keys the processed-event + /// ledger, so a collision would make one handler's success skip the other. + pub fn new( + consumer: Arc, + handlers: Vec>, + ledger: Arc, + ) -> Result { + let mut seen = std::collections::HashSet::new(); + for handler in &handlers { + if !seen.insert(handler.name()) { + return Err(DomainError::ValidationError(format!( + "duplicate event handler name `{}`", + handler.name() + ))); + } + } + Ok(Self { consumer, handlers, + ledger, semaphore: Arc::new(Semaphore::new(DEFAULT_CONCURRENCY)), - } + }) + } + + pub fn handler_names(&self) -> Vec<&'static str> { + self.handlers.iter().map(|h| h.name()).collect() } pub async fn run(self, mut shutdown: tokio::sync::watch::Receiver) { let handlers = Arc::new(self.handlers); + let ledger = self.ledger; let mut tasks = tokio::task::JoinSet::new(); let mut stream = self.consumer.consume(); @@ -43,8 +66,9 @@ impl WorkerService { let permit = self.semaphore.clone().acquire_owned().await; let Ok(permit) = permit else { break }; let h = Arc::clone(&handlers); + let ledger = Arc::clone(&ledger); tasks.spawn(async move { - dispatch(h, envelope).await; + dispatch(h, ledger, envelope).await; drop(permit); }); } @@ -64,15 +88,48 @@ impl WorkerService { } } -async fn dispatch(handlers: Arc>>, envelope: EventEnvelope) { +async fn dispatch( + handlers: Arc>>, + ledger: Arc, + envelope: EventEnvelope, +) { + let id = envelope.id().copied(); let mut any_transient = false; for handler in handlers.iter() { - if let Err(e) = handler.handle(&envelope.event).await { - if e.is_transient() { - tracing::warn!("transient handler error, will retry: {e}"); + let name = handler.name(); + if let Some(id) = &id { + match ledger.has_processed(id, name).await { + Ok(true) => { + tracing::debug!(handler = name, "already processed, skipping"); + continue; + } + Ok(false) => {} + Err(e) => { + // Running anyway could repeat a side effect; retry the delivery instead. + tracing::warn!(handler = name, "ledger lookup failed, will retry: {e}"); + any_transient = true; + continue; + } + } + } + match handler.handle(&envelope.event).await { + Ok(()) => { + if let Some(id) = &id + && let Err(e) = ledger.record_processed(id, name).await + { + // The work is done; failing the delivery would only repeat it. + tracing::warn!(handler = name, "recording processed event failed: {e}"); + } + } + Err(e) if e.is_transient() => { + tracing::warn!(handler = name, "transient handler error, will retry: {e}"); any_transient = true; - } else { - tracing::warn!("permanent handler error (not retrying): {e}"); + } + Err(e) => { + tracing::warn!( + handler = name, + "permanent handler error (not retrying): {e}" + ); } } } diff --git a/crates/application/src/wrapup/event_handler.rs b/crates/application/src/wrapup/event_handler.rs index 351b7693..1d5de35d 100644 --- a/crates/application/src/wrapup/event_handler.rs +++ b/crates/application/src/wrapup/event_handler.rs @@ -32,6 +32,10 @@ impl WrapUpEventHandler { #[async_trait] impl EventHandler for WrapUpEventHandler { + fn name(&self) -> &'static str { + "wrapup" + } + async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> { match event { DomainEvent::WrapUpRequested { diff --git a/crates/application/tests/jobs.rs b/crates/application/tests/jobs.rs new file mode 100644 index 00000000..e61924a4 --- /dev/null +++ b/crates/application/tests/jobs.rs @@ -0,0 +1,2 @@ +#[path = "jobs/processed_events_cleanup_test.rs"] +mod processed_events_cleanup_test; diff --git a/crates/application/tests/jobs/processed_events_cleanup_test.rs b/crates/application/tests/jobs/processed_events_cleanup_test.rs new file mode 100644 index 00000000..dcefa601 --- /dev/null +++ b/crates/application/tests/jobs/processed_events_cleanup_test.rs @@ -0,0 +1,35 @@ +use application::jobs::{PROCESSED_EVENTS_RETENTION_DAYS, ProcessedEventsCleanupJob}; +use domain::{events::EventId, ports::PeriodicJob, testing::InMemoryStore}; + +#[tokio::test] +async fn removes_only_rows_older_than_the_retention_window() { + let store = InMemoryStore::new(); + let now = chrono::Utc::now().naive_utc(); + let stale = EventId::new(); + let fresh = EventId::new(); + store.seed_processed_event( + &stale, + "h", + now - chrono::Duration::days(PROCESSED_EVENTS_RETENTION_DAYS + 1), + ); + store.seed_processed_event( + &fresh, + "h", + now - chrono::Duration::days(PROCESSED_EVENTS_RETENTION_DAYS - 1), + ); + + ProcessedEventsCleanupJob::new(store.clone()) + .run() + .await + .unwrap(); + + use domain::ports::ProcessedEventLedger; + assert!(!store.has_processed(&stale, "h").await.unwrap()); + assert!(store.has_processed(&fresh, "h").await.unwrap()); +} + +#[test] +fn runs_daily() { + let job = ProcessedEventsCleanupJob::new(InMemoryStore::new()); + assert_eq!(job.interval(), std::time::Duration::from_secs(86_400)); +} diff --git a/crates/application/tests/worker/worker_test.rs b/crates/application/tests/worker/worker_test.rs index 61f4fc43..c3a049f0 100644 --- a/crates/application/tests/worker/worker_test.rs +++ b/crates/application/tests/worker/worker_test.rs @@ -1,6 +1,6 @@ use application::worker::*; use async_trait::async_trait; -use domain::testing::NoopAckHandle; +use domain::testing::{InMemoryStore, NoopAckHandle}; use domain::value_objects::{ExternalMetadataId, MovieId}; use domain::{ errors::DomainError, @@ -10,7 +10,7 @@ use domain::{ events::EventEnvelope, ports::{EventConsumer, EventHandler}, }; -use futures::{stream, stream::BoxStream}; +use futures::{StreamExt as _, stream, stream::BoxStream}; use std::sync::{Arc, Mutex}; struct TrackingAck { @@ -52,6 +52,9 @@ struct RecordingHandler { #[async_trait] impl EventHandler for RecordingHandler { + fn name(&self) -> &'static str { + "recording" + } async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> { let label = match event { DomainEvent::MovieDiscovered { .. } => "movie_discovered", @@ -109,9 +112,14 @@ async fn dispatches_to_all_handlers() { calls: Arc::clone(&calls), }; - WorkerService::new(Arc::new(consumer), vec![Arc::new(handler)]) - .run(tokio::sync::watch::channel(false).1) - .await; + WorkerService::new( + Arc::new(consumer), + vec![Arc::new(handler)], + InMemoryStore::new(), + ) + .unwrap() + .run(tokio::sync::watch::channel(false).1) + .await; assert_eq!(*calls.lock().unwrap(), vec!["movie_discovered"]); } @@ -147,7 +155,8 @@ async fn acks_when_all_handlers_succeed() { nacked: Arc::clone(&nacked), }; - WorkerService::new(Arc::new(consumer), vec![]) + WorkerService::new(Arc::new(consumer), vec![], InMemoryStore::new()) + .unwrap() .run(tokio::sync::watch::channel(false).1) .await; @@ -181,6 +190,9 @@ async fn nacks_on_transient_handler_failure() { struct TransientHandler; #[async_trait] impl EventHandler for TransientHandler { + fn name(&self) -> &'static str { + "transient" + } async fn handle(&self, _: &DomainEvent) -> Result<(), DomainError> { Err(DomainError::InfrastructureError("network timeout".into())) } @@ -193,7 +205,9 @@ async fn nacks_on_transient_handler_failure() { nacked: Arc::clone(&nacked), }), vec![Arc::new(TransientHandler)], + InMemoryStore::new(), ) + .unwrap() .run(tokio::sync::watch::channel(false).1) .await; @@ -227,6 +241,9 @@ async fn acks_on_permanent_handler_failure() { struct PermanentHandler; #[async_trait] impl EventHandler for PermanentHandler { + fn name(&self) -> &'static str { + "permanent" + } async fn handle(&self, _: &DomainEvent) -> Result<(), DomainError> { Err(DomainError::NotFound("movie not on tmdb".into())) } @@ -239,7 +256,9 @@ async fn acks_on_permanent_handler_failure() { nacked: Arc::clone(&nacked), }), vec![Arc::new(PermanentHandler)], + InMemoryStore::new(), ) + .unwrap() .run(tokio::sync::watch::channel(false).1) .await; @@ -276,6 +295,9 @@ async fn nacks_if_any_handler_is_transient() { struct OkHandler; #[async_trait] impl EventHandler for OkHandler { + fn name(&self) -> &'static str { + "ok" + } async fn handle(&self, _: &DomainEvent) -> Result<(), DomainError> { Ok(()) } @@ -283,6 +305,9 @@ async fn nacks_if_any_handler_is_transient() { struct TransientHandler; #[async_trait] impl EventHandler for TransientHandler { + fn name(&self) -> &'static str { + "transient" + } async fn handle(&self, _: &DomainEvent) -> Result<(), DomainError> { Err(DomainError::InfrastructureError("db gone".into())) } @@ -295,7 +320,9 @@ async fn nacks_if_any_handler_is_transient() { nacked: Arc::clone(&nacked), }), vec![Arc::new(OkHandler), Arc::new(TransientHandler)], + InMemoryStore::new(), ) + .unwrap() .run(tokio::sync::watch::channel(false).1) .await; @@ -308,3 +335,145 @@ async fn nacks_if_any_handler_is_transient() { "should nack when any handler is transient" ); } + +// ── idempotent dispatch ─────────────────────────────────────────────────────── + +struct CountingHandler { + name: &'static str, + calls: Arc>, + fail_transient_times: Mutex, +} + +impl CountingHandler { + fn new(name: &'static str, fail_transient_times: u32) -> (Arc, Arc>) { + let calls = Arc::new(Mutex::new(0)); + ( + Arc::new(Self { + name, + calls: Arc::clone(&calls), + fail_transient_times: Mutex::new(fail_transient_times), + }), + calls, + ) + } +} + +#[async_trait] +impl EventHandler for CountingHandler { + fn name(&self) -> &'static str { + self.name + } + async fn handle(&self, _: &DomainEvent) -> Result<(), DomainError> { + *self.calls.lock().unwrap() += 1; + let mut left = self.fail_transient_times.lock().unwrap(); + if *left > 0 { + *left -= 1; + return Err(DomainError::InfrastructureError("flaky".into())); + } + Ok(()) + } +} + +/// Delivers the same event (same id) `times` times, recording ack/nack per delivery. +struct RedeliveringConsumer { + id: Option, + event: DomainEvent, + times: usize, + nacks: Arc>, +} + +struct CountNack(Arc>); + +#[async_trait] +impl AckHandle for CountNack { + async fn ack(&self) -> Result<(), DomainError> { + Ok(()) + } + async fn nack(&self) -> Result<(), DomainError> { + *self.0.lock().unwrap() += 1; + Ok(()) + } +} + +impl EventConsumer for RedeliveringConsumer { + fn consume(&self) -> BoxStream<'_, Result> { + let nacks = Arc::clone(&self.nacks); + let id = self.id; + let event = self.event.clone(); + Box::pin(stream::iter(0..self.times).then(move |_| { + let envelope = + EventEnvelope::new(id, event.clone(), Box::new(CountNack(Arc::clone(&nacks)))); + async move { + // Let the previous delivery's handlers finish first: a real + // broker redelivers only after the nack. + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + Ok(envelope) + } + })) + } +} + +#[tokio::test] +async fn a_redelivered_event_skips_handlers_that_already_succeeded() { + let (first, first_calls) = CountingHandler::new("first", 0); + let (second, second_calls) = CountingHandler::new("second", 1); + let nacks = Arc::new(Mutex::new(0)); + let consumer = RedeliveringConsumer { + id: Some(domain::events::EventId::new()), + event: movie_discovered(), + times: 2, + nacks: Arc::clone(&nacks), + }; + + WorkerService::new( + Arc::new(consumer), + vec![first, second], + InMemoryStore::new(), + ) + .unwrap() + .run(tokio::sync::watch::channel(false).1) + .await; + + assert_eq!(*nacks.lock().unwrap(), 1, "only the first delivery nacks"); + assert_eq!( + *first_calls.lock().unwrap(), + 1, + "first ran once, skipped on redelivery" + ); + assert_eq!( + *second_calls.lock().unwrap(), + 2, + "second retried and succeeded" + ); +} + +#[tokio::test] +async fn events_without_id_run_every_time() { + let (handler, calls) = CountingHandler::new("only", 0); + let consumer = RedeliveringConsumer { + id: None, + event: movie_discovered(), + times: 2, + nacks: Arc::new(Mutex::new(0)), + }; + + WorkerService::new(Arc::new(consumer), vec![handler], InMemoryStore::new()) + .unwrap() + .run(tokio::sync::watch::channel(false).1) + .await; + + assert_eq!(*calls.lock().unwrap(), 2); +} + +#[tokio::test] +async fn duplicate_handler_names_are_rejected() { + let (a, _) = CountingHandler::new("same", 0); + let (b, _) = CountingHandler::new("same", 0); + let result = WorkerService::new( + Arc::new(VecConsumer { events: vec![] }), + vec![a, b], + InMemoryStore::new(), + ); + let err = result.err().expect("duplicate names must be rejected"); + assert!(err.to_string().contains("same"), "{err}"); +} diff --git a/crates/bootstrap/src/database.rs b/crates/bootstrap/src/database.rs index 89ef2cce..a2cc4c00 100644 --- a/crates/bootstrap/src/database.rs +++ b/crates/bootstrap/src/database.rs @@ -38,6 +38,7 @@ pub struct DatabaseOutput { pub federation_settings: std::sync::Arc, pub remote_goal: Arc, pub refresh_session: Arc, + pub processed_events: Arc, pub deduplicator: Arc, pub image_ref_command: Arc, pub image_ref_query: Arc, @@ -114,6 +115,7 @@ fn assemble( federation_settings: w.federation_settings, remote_goal: w.remote_goal, refresh_session: Arc::new(sqlite::SqliteRefreshSessionAdapter::new(w.pool.clone())) as _, + processed_events: Arc::new(sqlite::SqliteProcessedEventLedger::new(w.pool.clone())) as _, deduplicator: w.deduplicator, image_ref_command, image_ref_query, diff --git a/crates/bootstrap/src/follow_backfill_handler.rs b/crates/bootstrap/src/follow_backfill_handler.rs index 454c12ba..37b18070 100644 --- a/crates/bootstrap/src/follow_backfill_handler.rs +++ b/crates/bootstrap/src/follow_backfill_handler.rs @@ -14,6 +14,10 @@ pub struct FollowBackfillHandler { #[async_trait] impl EventHandler for FollowBackfillHandler { + fn name(&self) -> &'static str { + "follow_backfill" + } + async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> { match event { DomainEvent::FollowAccepted { diff --git a/crates/bootstrap/src/worker.rs b/crates/bootstrap/src/worker.rs index 7530520e..c2ba1242 100644 --- a/crates/bootstrap/src/worker.rs +++ b/crates/bootstrap/src/worker.rs @@ -18,6 +18,11 @@ pub struct WorkerApp { } impl WorkerApp { + /// Names of the registered event handlers, in registration order. + pub fn handler_names(&self) -> Vec<&'static str> { + self.service.handler_names() + } + /// Runs the handlers and, when there is one, the outbox relay, until /// `shutdown` fires. Both stop together and finish their in-flight work. /// @@ -142,6 +147,7 @@ async fn assemble( let search_command = db.search_command; let wrapup_repo = db.wrapup_repo; let refresh_session = db.refresh_session; + let processed_events = db.processed_events; let event_publisher = event_publisher_arc; let metadata = metadata_client; @@ -217,6 +223,9 @@ async fn assemble( Arc::new(application::jobs::RefreshSessionCleanupJob::new( Arc::clone(&refresh_session), )), + Arc::new(application::jobs::ProcessedEventsCleanupJob::new( + Arc::clone(&processed_events), + )), ]; if let Some(job) = enrichment_job { periodic_jobs.push(job); @@ -301,7 +310,7 @@ async fn assemble( }; Ok(WorkerApp { - service: WorkerService::new(consumer_arc, handlers), + service: WorkerService::new(consumer_arc, handlers, processed_events)?, relay, }) } diff --git a/crates/bootstrap/tests/bootstrap_test.rs b/crates/bootstrap/tests/bootstrap_test.rs index 5f0fc821..7263272b 100644 --- a/crates/bootstrap/tests/bootstrap_test.rs +++ b/crates/bootstrap/tests/bootstrap_test.rs @@ -107,3 +107,34 @@ async fn server_publishes_to_the_outbox_even_when_bus_is_nats() { .unwrap(); assert_eq!(queued, 1); } + +/// Handler names key the processed-event ledger, so each must be unique. The +/// check runs over the real wiring, in both federation builds. +#[tokio::test] +async fn handler_names_are_unique() { + let pool = support::memory_pool().await; + let worker = bootstrap::build_worker_on(&support::test_config(), pool) + .await + .expect("worker builds"); + + let names = worker.handler_names(); + let unique: std::collections::HashSet<_> = names.iter().collect(); + assert_eq!( + unique.len(), + names.len(), + "duplicate handler name in {names:?}" + ); + for core in [ + "poster_sync", + "image_cleanup", + "search_cleanup", + "search_reindex", + ] { + assert!(names.contains(&core), "{core} missing from {names:?}"); + } + assert_eq!( + names.contains(&"activitypub"), + cfg!(feature = "federation"), + "{names:?}" + ); +} diff --git a/crates/domain/src/ports/events.rs b/crates/domain/src/ports/events.rs index 4d3f72e3..c3b18e81 100644 --- a/crates/domain/src/ports/events.rs +++ b/crates/domain/src/ports/events.rs @@ -23,5 +23,8 @@ pub trait EventConsumer: Send + Sync { #[async_trait] pub trait EventHandler: Send + Sync { + /// Stable, unique snake_case identity. It keys the processed-event ledger, + /// so renaming a handler makes it re-run events still inside the retention window. + fn name(&self) -> &'static str; async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError>; } diff --git a/crates/domain/src/ports/mod.rs b/crates/domain/src/ports/mod.rs index 07793697..f25cec7d 100644 --- a/crates/domain/src/ports/mod.rs +++ b/crates/domain/src/ports/mod.rs @@ -14,6 +14,7 @@ pub mod media_server; pub mod movie; pub mod noop; pub mod person; +pub mod processed_events; pub mod rss; pub mod search; pub mod social; @@ -35,6 +36,7 @@ pub use jobs::*; pub use media_server::*; pub use movie::*; pub use person::*; +pub use processed_events::*; pub use rss::*; pub use search::*; pub use social::*; diff --git a/crates/domain/src/ports/processed_events.rs b/crates/domain/src/ports/processed_events.rs new file mode 100644 index 00000000..76b35c89 --- /dev/null +++ b/crates/domain/src/ports/processed_events.rs @@ -0,0 +1,16 @@ +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. +#[async_trait] +pub trait ProcessedEventLedger: Send + Sync { + async fn has_processed(&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>; + + /// Returns the number of rows removed. + async fn delete_older_than(&self, cutoff: chrono::NaiveDateTime) -> Result; +} diff --git a/crates/domain/src/testing/fakes/mod.rs b/crates/domain/src/testing/fakes/mod.rs index 3b0750fe..95a5bb52 100644 --- a/crates/domain/src/testing/fakes/mod.rs +++ b/crates/domain/src/testing/fakes/mod.rs @@ -6,6 +6,7 @@ mod store; mod store_goal; mod store_import; mod store_movie; +mod store_processed_events; mod store_review; mod store_session; mod store_social; diff --git a/crates/domain/src/testing/fakes/store.rs b/crates/domain/src/testing/fakes/store.rs index 857210a0..808c1fb9 100644 --- a/crates/domain/src/testing/fakes/store.rs +++ b/crates/domain/src/testing/fakes/store.rs @@ -64,6 +64,7 @@ pub struct InMemoryStore { pub(super) import_profiles: RwLock>, pub(super) follows: RwLock>, pub(super) blocked: RwLock>, + pub(super) processed_events: RwLock>, } impl InMemoryStore { @@ -85,6 +86,7 @@ impl InMemoryStore { import_profiles: RwLock::default(), follows: RwLock::default(), blocked: RwLock::default(), + processed_events: RwLock::default(), }) } @@ -122,6 +124,23 @@ impl InMemoryStore { self.import_sessions.read().unwrap().len() } + pub fn processed_event_count(&self) -> usize { + self.processed_events.read().unwrap().len() + } + + /// Inserts a ledger row with an explicit timestamp (for retention tests). + pub fn seed_processed_event( + &self, + id: &crate::events::EventId, + handler: &str, + processed_at: chrono::NaiveDateTime, + ) { + self.processed_events + .write() + .unwrap() + .insert((id.value(), handler.to_owned()), processed_at); + } + pub fn refresh_session_count(&self) -> usize { self.refresh_sessions.read().unwrap().len() } diff --git a/crates/domain/src/testing/fakes/store_processed_events.rs b/crates/domain/src/testing/fakes/store_processed_events.rs new file mode 100644 index 00000000..f87b14a8 --- /dev/null +++ b/crates/domain/src/testing/fakes/store_processed_events.rs @@ -0,0 +1,33 @@ +use async_trait::async_trait; + +use crate::{errors::DomainError, events::EventId, ports::ProcessedEventLedger}; + +use super::InMemoryStore; + +#[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()))) + } + + /// `(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 + .write() + .unwrap() + .entry((id.value(), handler.to_owned())) + .or_insert_with(|| chrono::Utc::now().naive_utc()); + Ok(()) + } + + async fn delete_older_than(&self, cutoff: chrono::NaiveDateTime) -> Result { + let mut rows = self.processed_events.write().unwrap(); + let before = rows.len(); + rows.retain(|_, at| *at >= cutoff); + Ok((before - rows.len()) as u64) + } +}