From bde8040cc34e767a4719ee8796ae854b7bffae9b Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Mon, 26 Oct 2026 01:56:38 +0100 Subject: [PATCH] watchlist add: event only if inserted; adr 0015 jetstream/relay accuracy; drop vacuous assert --- .../sqlite/src/repositories/watchlist/mod.rs | 11 +++++++--- .../tests/repositories/outbox_writes_test.rs | 18 +++++++++++++++ .../tests/diary/review_logger_test.rs | 1 - .../application/tests/watchlist/add_test.rs | 5 +++++ crates/bootstrap/tests/outbox_relay_test.rs | 2 +- .../src/testing/fakes/store_watchlist.rs | 22 +++++++++++++------ docs/adr/0015-transactional-outbox.md | 13 +++++++++-- 7 files changed, 58 insertions(+), 14 deletions(-) diff --git a/crates/adapters/sqlite/src/repositories/watchlist/mod.rs b/crates/adapters/sqlite/src/repositories/watchlist/mod.rs index a6046ce3..8d747023 100644 --- a/crates/adapters/sqlite/src/repositories/watchlist/mod.rs +++ b/crates/adapters/sqlite/src/repositories/watchlist/mod.rs @@ -39,7 +39,7 @@ impl WatchlistRepository for SqliteWatchlistRepository { .begin() .await .map_err(crate::shared::map_sqlx_error)?; - sqlx::query( + let inserted = sqlx::query( "INSERT OR IGNORE INTO watchlist_entries (id, user_id, movie_id, added_at) \ VALUES (?, ?, ?, ?)", ) @@ -49,8 +49,13 @@ impl WatchlistRepository for SqliteWatchlistRepository { .bind(&added_at) .execute(&mut *tx) .await - .map_err(crate::shared::map_sqlx_error)?; - enqueue_events(&mut tx, events).await?; + .map_err(crate::shared::map_sqlx_error)? + .rows_affected() + > 0; + // A repeat add changed nothing, so it announces nothing. + if inserted { + enqueue_events(&mut tx, events).await?; + } tx.commit().await.map_err(crate::shared::map_sqlx_error) } diff --git a/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs b/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs index 0a64450a..14f46307 100644 --- a/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs +++ b/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs @@ -200,6 +200,24 @@ async fn adding_a_watchlist_entry_writes_the_entry_and_its_event_in_one_commit() assert_eq!(count(&pool, "event_queue").await, 1); } +#[tokio::test] +async fn re_adding_a_watchlist_entry_enqueues_no_second_event() { + let (pool, user_id, movie) = watchlist_fixture().await; + let repo = SqliteWatchlistRepository::new(pool.clone()); + let first = WatchlistEntry::new(user_id.clone(), movie.id().clone()); + repo.add(&first, &[watchlist_added(&first, &movie)]) + .await + .unwrap(); + let again = WatchlistEntry::new(user_id, movie.id().clone()); + + repo.add(&again, &[watchlist_added(&again, &movie)]) + .await + .unwrap(); + + assert_eq!(count(&pool, "watchlist_entries").await, 1); + assert_eq!(count(&pool, "event_queue").await, 1, "no phantom event"); +} + #[tokio::test] async fn a_failing_enqueue_rolls_back_the_watchlist_add() { let (pool, user_id, movie) = watchlist_fixture().await; diff --git a/crates/application/tests/diary/review_logger_test.rs b/crates/application/tests/diary/review_logger_test.rs index b4e4ebdf..af524b7d 100644 --- a/crates/application/tests/diary/review_logger_test.rs +++ b/crates/application/tests/diary/review_logger_test.rs @@ -270,7 +270,6 @@ async fn a_refused_save_propagates_and_records_nothing() { .await; assert!(result.is_err()); - assert_eq!(movies.review_count(), 0); assert!(movies.recorded_events().is_empty()); } diff --git a/crates/application/tests/watchlist/add_test.rs b/crates/application/tests/watchlist/add_test.rs index 2d1fb01d..d297cdec 100644 --- a/crates/application/tests/watchlist/add_test.rs +++ b/crates/application/tests/watchlist/add_test.rs @@ -54,6 +54,11 @@ async fn adding_a_movie_already_on_the_watchlist_is_idempotent() { add::execute(&deps, make_cmd()).await.unwrap(); add::execute(&deps, make_cmd()).await.unwrap(); + assert_eq!( + movies.recorded_events().len(), + 1, + "the repeat inserted nothing, so it announces nothing" + ); assert_eq!( watchlist.watchlist_count(), 1, diff --git a/crates/bootstrap/tests/outbox_relay_test.rs b/crates/bootstrap/tests/outbox_relay_test.rs index 857a6ab5..4c453261 100644 --- a/crates/bootstrap/tests/outbox_relay_test.rs +++ b/crates/bootstrap/tests/outbox_relay_test.rs @@ -38,7 +38,7 @@ impl EventRelaySink for FakeSink { } let outage = self .fail_first - .fetch_update( + .try_update( std::sync::atomic::Ordering::SeqCst, std::sync::atomic::Ordering::SeqCst, |n| n.checked_sub(1), diff --git a/crates/domain/src/testing/fakes/store_watchlist.rs b/crates/domain/src/testing/fakes/store_watchlist.rs index c19d6820..ab523cad 100644 --- a/crates/domain/src/testing/fakes/store_watchlist.rs +++ b/crates/domain/src/testing/fakes/store_watchlist.rs @@ -16,15 +16,23 @@ use super::InMemoryStore; #[async_trait] impl WatchlistRepository for InMemoryStore { - /// `INSERT OR IGNORE` on UNIQUE(user_id, movie_id): adding twice is a no-op. + /// `INSERT OR IGNORE` on UNIQUE(user_id, movie_id): adding twice is a no-op + /// and, like the adapter, records events only for the row it inserted. async fn add(&self, entry: &WatchlistEntry, events: &[DomainEvent]) -> Result<(), DomainError> { let key = (entry.user_id.value(), entry.movie_id.value()); - self.watchlist - .write() - .unwrap() - .entry(key) - .or_insert_with(|| entry.clone()); - self.record_events(events); + let inserted = { + let mut entries = self.watchlist.write().unwrap(); + match entries.entry(key) { + std::collections::hash_map::Entry::Vacant(slot) => { + slot.insert(entry.clone()); + true + } + std::collections::hash_map::Entry::Occupied(_) => false, + } + }; + if inserted { + self.record_events(events); + } Ok(()) } diff --git a/docs/adr/0015-transactional-outbox.md b/docs/adr/0015-transactional-outbox.md index 07f6a535..3846834b 100644 --- a/docs/adr/0015-transactional-outbox.md +++ b/docs/adr/0015-transactional-outbox.md @@ -25,8 +25,17 @@ Use cases never open transactions; the adapter does. **The server never talks to NATS.** With `EVENT_BUS_BACKEND=nats` the server still only inserts outbox rows. The worker runs a relay that claims pending rows and publishes them to JetStream with `Nats-Msg-Id` set to the event id, then marks the row done. With `db` the worker consumes the rows -directly. `max_deliver` on the JetStream consumer equals `EVENT_QUEUE_MAX_ATTEMPTS`, so both -backends dead-letter after the same number of attempts. +directly. `max_deliver` on the JetStream consumer equals `EVENT_QUEUE_MAX_ATTEMPTS`, and a JetStream +nak uses the same backoff as the SQLite queue (`domain::events::retry_backoff`: 5 s doubling per +attempt, capped at 300 s), so both backends give up after the same number of deliveries on the same +schedule. They differ in what giving up means. SQLite marks the row `dead_lettered`. JetStream has no +DLQ: after `max_deliver` the message is no longer delivered but stays in the stream (until stream +limits evict it). The consumer logs `event exhausted max deliveries` at error level with the event id +on the final nak. To find such messages, watch the JetStream max-deliveries advisory +(`$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES..`) or inspect the stream +(`nats stream view `) for sequences the consumer has not acked. The relay never spends +attempts on transport failures: when the bus is unreachable it defers the row with growing backoff and +pauses, so an outage cannot dead-letter rows. **Three dedup layers.** 1. *Broker.* JetStream drops a repeated `Nats-Msg-Id` inside `duplicate_window`, set explicitly to 1