From a88c18f029bf6d36d28be3023ba593f6f2014844 Mon Sep 17 00:00:00 2001 From: npub17jjz49l9jjmhhk7cac63j8yt9z555n9cw8vk7v5jz4vzw4ppld5qgj57cc Date: Fri, 26 Jun 2026 14:28:53 -0400 Subject: [PATCH] fix(relay): reorder reminder scheduler to claim FIRST, publish SECOND MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pre-build §5c gate for the multi-tenant relay rewrite: the NIP-ER reminder producer at `crates/buzz-relay/src/main.rs:454-487` inherited the old relay's publish-then-claim shape, with the comment "duplicate fan-out on the next tick is harmless (subscribers dedup by event ID)" That premise is wrong — subscribers do NOT dedup reminder fan-outs by event id, so a publish that wins fan-out but loses the claim leaks a duplicate delivery to every connected client. Reorder to: claim_due_reminder → if Ok(Some(stamp)) publish_event → if Ok(()) we're done if Err(_) rollback_reminder_claim(..., stamp) This mirrors the `claim → side-effect` ordering Max + Sami landed for the workflow cron producer (`df997d7cc` + `910d82228`) and keeps the DB claim row as the single cross-pod side-effect boundary. DB API change (single prod caller, `main.rs` only) claim_due_reminder returns `Result>` instead of `Result`. The `Some(stamp)` carries the `delivered_at` value the winner wrote, which the rollback uses as its `AND delivered_at = $stamp` guard. Without that guard, a stale rollback could clobber another pod's later claim — a silent duplicate-publish window. rollback_reminder_claim(id, created_at, stamp) -> Result is the inverse: stamp-guarded UPDATE that sets delivered_at = NULL. Returns true on revert, false when the row's delivered_at no longer equals stamp (caller leaves it alone; bounded duplicate fan-out is the worst case, never silent data loss). Tests (`#[ignore = "requires Postgres"]`, follows F1/F4 conventions) two_concurrent_claims_yield_exactly_one_winner Two `claim_due_reminder` calls under `tokio::join!` on the same reminder. Asserts exactly one Some(stamp), one None, and a third claim still loses. The §5c "two-pod concurrency" acceptance test Eva called for. Adversarial: drop the `AND delivered_at IS NULL` guard from the claim UPDATE and both calls win — verified. rollback_with_matching_stamp_reverts_then_reclaim_wins Happy path: claim → rollback with matching stamp returns true → second claim wins (no longer held). Adversarial: replace `SET delivered_at = NULL` with `SET delivered_at = delivered_at` (no-op rollback) and the second claim fails to win — verified. rollback_with_stale_stamp_does_not_clobber_live_claim ABA guard: claim with stamp_old → manually advance delivered_at to stamp_new (simulates a later pod's claim) → rollback with stamp_old returns false WITHOUT reverting; re-read confirms delivered_at is still stamp_new. Adversarial: drop the `AND delivered_at = $3` guard and the stale rollback clobbers the live claim — verified. Comment fix The misleading "subscribers dedup by event ID" comment is replaced with the actual invariant: claim is the side-effect boundary, rollback is best-effort, on rollback failure bounded duplicate fan-out is the worst case. Validation cargo fmt --check -p buzz-db -p buzz-relay ✅ cargo clippy -p buzz-db --tests --no-deps ✅ cargo clippy -p buzz-relay --tests --no-deps ✅ cargo test -p buzz-db --lib -- --include-ignored --test-threads=1 ✅ 87 pass cargo test -p buzz-relay --lib ✅ 369 pass cargo check -p buzz-relay ✅ Refs §5c reminder scheduler claim-first reorder gate. Co-authored-by: Tyler Longwell Signed-off-by: Tyler Longwell --- crates/buzz-db/src/event.rs | 255 +++++++++++++++++++++++++++++++++- crates/buzz-db/src/lib.rs | 16 ++- crates/buzz-relay/src/main.rs | 84 +++++++---- 3 files changed, 323 insertions(+), 32 deletions(-) diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs index 9cef6c53c3..ba057b422f 100644 --- a/crates/buzz-db/src/event.rs +++ b/crates/buzz-db/src/event.rs @@ -1074,15 +1074,20 @@ pub async fn query_due_reminders( Ok(results) } -/// Atomically claim a due reminder for delivery. Returns `Some(id)` if this -/// caller won the claim (set `delivered_at`), or `None` if another pod already -/// claimed it. Mirrors the reaper's `archived_at IS NULL` guard for cross-pod -/// idempotency. +/// Atomically claim a due reminder for delivery. Returns `Some(stamp)` if +/// this caller won the claim (the `delivered_at` value it wrote), or `None` +/// if another pod already claimed it. Mirrors the reaper's `archived_at IS +/// NULL` guard for cross-pod idempotency. +/// +/// The returned stamp is the value the caller must pass to +/// [`rollback_reminder_claim`] if a subsequent side effect (e.g. Redis +/// publish) fails — the rollback guards on `delivered_at = stamp` so a +/// stale rollback can never clobber a fresh winner's claim. pub async fn claim_due_reminder( pool: &PgPool, event_id: &[u8], event_created_at: DateTime, -) -> Result { +) -> Result> { let now_epoch = Utc::now().timestamp(); let result = sqlx::query( r#" @@ -1097,6 +1102,42 @@ pub async fn claim_due_reminder( .execute(pool) .await?; + if result.rows_affected() > 0 { + Ok(Some(now_epoch)) + } else { + Ok(None) + } +} + +/// Best-effort rollback of a prior winning [`claim_due_reminder`] call. +/// +/// Used when the caller won the claim but its follow-on side effect (e.g. +/// Redis publish) failed and it wants the reminder to be re-tried on a +/// subsequent scheduler tick. The `stamp` argument MUST be the value +/// returned by the original `claim_due_reminder` call — the SQL guard +/// `delivered_at = $stamp` is what prevents an ABA hazard where another pod +/// has since rewon the claim. Returns `Ok(true)` if the row was reverted, +/// `Ok(false)` if not (someone else holds a live claim now; leave it alone +/// — bounded duplicate fan-out is the worst case). +pub async fn rollback_reminder_claim( + pool: &PgPool, + event_id: &[u8], + event_created_at: DateTime, + stamp: i64, +) -> Result { + let result = sqlx::query( + r#" + UPDATE events + SET delivered_at = NULL + WHERE created_at = $1 AND id = $2 AND delivered_at = $3 + "#, + ) + .bind(event_created_at) + .bind(event_id) + .bind(stamp) + .execute(pool) + .await?; + Ok(result.rows_affected() > 0) } @@ -1241,3 +1282,207 @@ mod tests { assert_eq!(extract_not_before(&event), None); } } + +/// Live-PG dedup tests for the NIP-ER reminder claim/rollback API. +/// +/// Reuses the `#[ignore = "requires Postgres"]` + serial-on-fresh-schema +/// convention Sami's F1/F4 and Quinn's engine-layer audit landed (see +/// `crates/buzz-workflow/src/lib.rs` and `crates/buzz-db/src/workflow.rs`). +/// These pin the §5c reminder-scheduler invariants that the `claim FIRST → +/// publish SECOND → rollback on publish failure` shape in +/// `buzz-relay/src/main.rs` depends on. +#[cfg(test)] +mod claim_dedup_tests { + use super::*; + use buzz_core::kind::KIND_EVENT_REMINDER; + use sqlx::PgPool; + + const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; + + async fn setup_pool() -> PgPool { + let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| TEST_DB_URL.to_owned()); + + PgPool::connect(&database_url) + .await + .expect("connect to test DB") + } + + /// Insert a due reminder row directly: minimal columns to satisfy NOT + /// NULL + the predicates `query_due_reminders` filters on + /// (`kind = KIND_EVENT_REMINDER`, `not_before <= now`, + /// `deleted_at IS NULL`, `delivered_at IS NULL`). Returns (id_bytes, + /// created_at) — the composite PK the claim API targets. + async fn insert_due_reminder(pool: &PgPool) -> (Vec, DateTime) { + // 32-byte id: two uuids concatenated for uniqueness. + let id_bytes: Vec = Uuid::new_v4() + .as_bytes() + .iter() + .chain(Uuid::new_v4().as_bytes().iter()) + .copied() + .collect(); + let pubkey: Vec = vec![0xb1; 32]; + let created_at = Utc::now(); + let not_before = created_at.timestamp() - 1; // already due. + + sqlx::query( + r#" + INSERT INTO events + (id, pubkey, created_at, kind, tags, content, sig, not_before) + VALUES ($1, $2, $3, $4, '[]'::jsonb, '', $5, $6) + "#, + ) + .bind(&id_bytes) + .bind(&pubkey) + .bind(created_at) + .bind(KIND_EVENT_REMINDER as i32) + .bind(vec![0u8; 64]) + .bind(not_before) + .execute(pool) + .await + .expect("insert due reminder"); + + (id_bytes, created_at) + } + + /// (§5c acceptance) Two concurrent `claim_due_reminder` calls for the + /// same reminder must serialize: exactly one returns `Some(stamp)`, the + /// other `None`. This is the cross-pod invariant the reorder relies on + /// — without it the "claim first" ordering would still allow duplicate + /// publishes. + /// + /// The DB-layer mechanism is `UPDATE ... WHERE delivered_at IS NULL`: + /// Postgres serializes the row-level write under the partition, + /// guaranteeing at most one UPDATE flips `delivered_at` from NULL to a + /// stamp. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn two_concurrent_claims_yield_exactly_one_winner() { + let pool = setup_pool().await; + let (id, created_at) = insert_due_reminder(&pool).await; + + let pool_a = pool.clone(); + let id_a = id.clone(); + let pool_b = pool.clone(); + let id_b = id.clone(); + + let (a, b) = tokio::join!( + tokio::spawn(async move { claim_due_reminder(&pool_a, &id_a, created_at).await }), + tokio::spawn(async move { claim_due_reminder(&pool_b, &id_b, created_at).await }), + ); + let a = a.unwrap().expect("claim A succeeded"); + let b = b.unwrap().expect("claim B succeeded"); + + // Exactly one Some(stamp), exactly one None. + let winners = [a.is_some(), b.is_some()]; + assert_eq!( + winners.iter().filter(|w| **w).count(), + 1, + "exactly one of the two concurrent claims must win: a={a:?}, b={b:?}" + ); + + // A third claim must also lose — the claim row is the dedupe boundary. + let third = claim_due_reminder(&pool, &id, created_at) + .await + .expect("third claim runs"); + assert_eq!( + third, None, + "third claim must lose; reminder already claimed" + ); + } + + /// (§5c acceptance) The publish-failure rollback round-trip: a winner + /// can revert its own claim with the stamp it was returned, and a + /// subsequent `claim_due_reminder` then succeeds. This is the path the + /// scheduler takes when Redis publish fails after a successful claim. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn rollback_with_matching_stamp_reverts_then_reclaim_wins() { + let pool = setup_pool().await; + let (id, created_at) = insert_due_reminder(&pool).await; + + let stamp = claim_due_reminder(&pool, &id, created_at) + .await + .expect("first claim runs") + .expect("first claim wins"); + + let reverted = rollback_reminder_claim(&pool, &id, created_at, stamp) + .await + .expect("rollback runs"); + assert!(reverted, "rollback with matching stamp must revert the row"); + + // After rollback `delivered_at` is NULL again — a second claim wins. + // (We don't compare the two stamps: `Utc::now().timestamp()` has + // 1-second resolution, so back-to-back claims often share a value. + // What matters is that the second claim returns Some, not None.) + let _stamp_2 = claim_due_reminder(&pool, &id, created_at) + .await + .expect("second claim runs") + .expect("second claim wins after rollback"); + } + + /// (§5c acceptance — the ABA guard, NOT a happy path) The rollback's + /// `delivered_at = $stamp` guard must reject a stale stamp. Without + /// this guard, a slow pod could win-rollback-rewon-rollback racing + /// another pod's later claim and clobber it, producing a silent + /// duplicate publish window. This test simulates the racy shape by + /// hand: the first claim sets `delivered_at = stamp1`, the operator + /// (or a second pod via DB write) replaces it with `stamp2`, and the + /// first pod's late rollback with `stamp1` must return false WITHOUT + /// reverting the row. + /// + /// Adversarial: drop the `AND delivered_at = $3` guard from + /// `rollback_reminder_claim` and this test fails — the stale rollback + /// silently clobbers the live claim. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn rollback_with_stale_stamp_does_not_clobber_live_claim() { + let pool = setup_pool().await; + let (id, created_at) = insert_due_reminder(&pool).await; + + let stamp_old = claim_due_reminder(&pool, &id, created_at) + .await + .expect("first claim runs") + .expect("first claim wins"); + + // Simulate the row's `delivered_at` advancing (operator manual + // clear-then-reclaim, or any post-claim mutation). A stamp 1 + // second later is sufficient — the guard compares exact equality. + let stamp_new = stamp_old + 1; + sqlx::query( + r#" + UPDATE events SET delivered_at = $1 + WHERE created_at = $2 AND id = $3 + "#, + ) + .bind(stamp_new) + .bind(created_at) + .bind(&id) + .execute(&pool) + .await + .expect("advance delivered_at"); + + let reverted = rollback_reminder_claim(&pool, &id, created_at, stamp_old) + .await + .expect("rollback runs"); + assert!( + !reverted, + "rollback with stale stamp must NOT revert — would clobber the live claim" + ); + + // Confirm the row's `delivered_at` is still `stamp_new`, not NULL. + let current: Option = + sqlx::query_scalar("SELECT delivered_at FROM events WHERE created_at = $1 AND id = $2") + .bind(created_at) + .bind(&id) + .fetch_one(&pool) + .await + .expect("re-read delivered_at"); + assert_eq!( + current, + Some(stamp_new), + "delivered_at must remain stamp_new — the stale rollback was correctly a no-op" + ); + } +} diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9dd4d3e107..cb7e2b0a42 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -545,14 +545,28 @@ impl Db { } /// Atomically claim a due reminder for delivery (cross-pod dedup). + /// Returns `Some(stamp)` if this caller won the claim — the stamp must + /// be passed to [`Self::rollback_reminder_claim`] if the follow-on side + /// effect fails. pub async fn claim_due_reminder( &self, event_id: &[u8], event_created_at: chrono::DateTime, - ) -> Result { + ) -> Result> { event::claim_due_reminder(&self.pool, event_id, event_created_at).await } + /// Best-effort rollback of a prior winning [`Self::claim_due_reminder`] + /// — the `stamp` MUST be the value returned by that call. + pub async fn rollback_reminder_claim( + &self, + event_id: &[u8], + event_created_at: chrono::DateTime, + stamp: i64, + ) -> Result { + event::rollback_reminder_claim(&self.pool, event_id, event_created_at, stamp).await + } + /// Ensure a user record exists (upsert). pub async fn ensure_user(&self, pubkey: &[u8]) -> Result<()> { user::ensure_user(&self.pool, pubkey).await diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index b63c19758e..96d766dee1 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -410,11 +410,13 @@ async fn main() -> anyhow::Result<()> { }); } - // NIP-ER reminder scheduler — polls for due reminders and publishes them - // to Redis pub/sub for cross-pod fan-out. Each pod's existing - // subscribe_local consumer picks them up and applies the author-only gate. - // Mirrors the channel reaper pattern. Cross-pod dedup via `delivered_at` - // column: only the pod that wins the atomic claim publishes. + // NIP-ER reminder scheduler — polls for due reminders, claims them + // atomically, then publishes the winner to Redis pub/sub for cross-pod + // fan-out. Each pod's existing subscribe_local consumer picks them up + // and applies the author-only gate. Mirrors the channel reaper pattern. + // Cross-pod dedup via `delivered_at` column: claim FIRST → publish + // SECOND → rollback on publish failure, so at most one pod publishes + // per due reminder. { let scheduler_state = Arc::clone(&state); let scheduler_interval_secs: u64 = std::env::var("SPROUT_REMINDER_SCHEDULER_INTERVAL_SECS") @@ -454,35 +456,65 @@ async fn main() -> anyhow::Result<()> { info!(count = due.len(), "Reminder scheduler: due reminders found"); for reminder in due { - // Publish first, then claim. If publish fails the reminder - // stays unclaimed and will be retried next tick. If claim - // fails after a successful publish, duplicate fan-out on the - // next tick is harmless (subscribers dedup by event ID). + // Claim FIRST, publish SECOND, roll the claim back if + // publish fails. The claim row is the cross-pod side-effect + // boundary: at most one pod publishes per due reminder, and + // if that pod's publish fails we revert its `delivered_at` + // so a future tick can retry. Subscribers do NOT dedup by + // event id — publishing under a lost claim would leak a + // duplicate fan-out to every client. Mirrors the + // `claim → side-effect` ordering Max + Sami landed for the + // cron producer (`df997d7cc` + `910d82228`). + let stamp = match scheduler_state + .db + .claim_due_reminder(&reminder.id, reminder.created_at) + .await + { + Ok(Some(stamp)) => stamp, + Ok(None) => continue, // Another pod claimed it. + Err(e) => { + warn!( + event_id = hex::encode(&reminder.id), + "Reminder scheduler: claim failed, skipping: {e}" + ); + continue; + } + }; + if let Err(e) = scheduler_state .pubsub .publish_event(uuid::Uuid::nil(), &reminder_to_event(&reminder)) .await { + // Publish failed after we won the claim. Roll back so + // a subsequent tick can retry. The rollback is + // stamp-guarded, so if another pod has somehow + // already rewon the claim (e.g. by an operator + // manually clearing `delivered_at`), we leave their + // claim alone — bounded duplicate fan-out is the + // worst case, but never silent data loss. error!( event_id = hex::encode(&reminder.id), - "Reminder scheduler: Redis publish failed, skipping claim: {e}" + "Reminder scheduler: Redis publish failed, rolling back claim: {e}" ); - continue; - } - - // Atomic cross-pod claim — only the winner marks it delivered. - match scheduler_state - .db - .claim_due_reminder(&reminder.id, reminder.created_at) - .await - { - Ok(true) => {} - Ok(false) => {} // Another pod claimed it; duplicate publish is harmless. - Err(e) => { - warn!( - event_id = hex::encode(&reminder.id), - "Reminder scheduler: claim failed after publish (duplicate delivery possible): {e}" - ); + match scheduler_state + .db + .rollback_reminder_claim(&reminder.id, reminder.created_at, stamp) + .await + { + Ok(true) => {} // reverted; retry on next tick. + Ok(false) => { + warn!( + event_id = hex::encode(&reminder.id), + "Reminder scheduler: rollback no-op (claim already changed); leaving as-is" + ); + } + Err(e) => { + warn!( + event_id = hex::encode(&reminder.id), + "Reminder scheduler: rollback failed (claim stuck until manual recovery): {e}" + ); + } } } }