fix(relay): reorder reminder scheduler to claim FIRST, publish SECOND

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<Option<i64>>` instead of
  `Result<bool>`. 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<bool> 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 <tlongwell@block.xyz>
Signed-off-by: Tyler Longwell <tlongwell@block.xyz>

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<Utc>,
-) -> Result<bool> {
+) -> Result<Option<i64>> {
     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<Utc>,
+    stamp: i64,
+) -> Result<bool> {
+    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<u8>, DateTime<Utc>) {
+        // 32-byte id: two uuids concatenated for uniqueness.
+        let id_bytes: Vec<u8> = Uuid::new_v4()
+            .as_bytes()
+            .iter()
+            .chain(Uuid::new_v4().as_bytes().iter())
+            .copied()
+            .collect();
+        let pubkey: Vec<u8> = 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<i64> =
+            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"
+        );
+    }
+}

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<chrono::Utc>,
-    ) -> Result<bool> {
+    ) -> Result<Option<i64>> {
         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<chrono::Utc>,
+        stamp: i64,
+    ) -> Result<bool> {
+        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

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}"
+                                );
+                            }
                         }
                     }
                 }