diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs index b50b92613..07be8811b 100644 --- a/crates/buzz-db/src/event.rs +++ b/crates/buzz-db/src/event.rs @@ -1500,4 +1500,111 @@ mod tests { row.id == event_b.id.as_bytes() && row.community_id == community_b && row.host == host_b })); } + + /// Two pods race to claim the same due reminder: exactly one wins. The + /// scheduler publishes only on a winning claim (`Ok(true)`) and `continue`s + /// on the loser (`Ok(false)`), so a single winning claim *is* the proof of + /// exactly one publish side effect across N pods. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn claim_due_reminder_is_won_by_exactly_one_of_two_racing_pods() { + let pool = setup_pool().await; + let community = CommunityId::from_uuid(make_test_community(&pool).await); + let not_before = Utc::now().timestamp() - 1; + let keys = Keys::generate(); + let event = EventBuilder::new(Kind::Custom(KIND_EVENT_REMINDER as u16), "due") + .tags([ + Tag::parse(["d", "due-reminder-claim-race"]).unwrap(), + Tag::parse(["not_before", ¬_before.to_string()]).unwrap(), + ]) + .sign_with_keys(&keys) + .expect("sign reminder"); + insert_event(&pool, community, &event, None) + .await + .expect("insert reminder"); + + let id = event.id.as_bytes().to_vec(); + let created_at = event.created_at.as_secs() as i64; + let created_at = chrono::DateTime::from_timestamp(created_at, 0).expect("created_at"); + + // Two pods, two distinct per-attempt stamps, same reminder. + let stamp_p1: i64 = 0x1111_1111_1111_1111; + let stamp_p2: i64 = 0x2222_2222_2222_2222; + let won_p1 = claim_due_reminder_with_stamp(&pool, &id, created_at, stamp_p1) + .await + .expect("p1 claim"); + let won_p2 = claim_due_reminder_with_stamp(&pool, &id, created_at, stamp_p2) + .await + .expect("p2 claim"); + + assert!( + won_p1 ^ won_p2, + "exactly one pod must win the claim (p1={won_p1}, p2={won_p2}) — \ + the loser never reaches the publish side effect" + ); + } + + /// A failed publish releases the claim so the reminder is redeliverable, + /// and the compare-and-clear stamp guard prevents one pod from rolling back + /// another pod's claim. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn release_due_reminder_rolls_back_only_the_matching_stamp() { + let pool = setup_pool().await; + let community = CommunityId::from_uuid(make_test_community(&pool).await); + let not_before = Utc::now().timestamp() - 1; + let keys = Keys::generate(); + let event = EventBuilder::new(Kind::Custom(KIND_EVENT_REMINDER as u16), "due") + .tags([ + Tag::parse(["d", "due-reminder-release"]).unwrap(), + Tag::parse(["not_before", ¬_before.to_string()]).unwrap(), + ]) + .sign_with_keys(&keys) + .expect("sign reminder"); + insert_event(&pool, community, &event, None) + .await + .expect("insert reminder"); + + let id = event.id.as_bytes().to_vec(); + let created_at = event.created_at.as_secs() as i64; + let created_at = chrono::DateTime::from_timestamp(created_at, 0).expect("created_at"); + let stamp: i64 = 0x3333_3333_3333_3333; + + assert!( + claim_due_reminder_with_stamp(&pool, &id, created_at, stamp) + .await + .expect("claim"), + "first claim wins" + ); + + // A release with the *wrong* stamp must be a no-op (does not clear + // another pod's claim). + assert!( + !release_due_reminder(&pool, &id, created_at, stamp ^ 0xFFFF) + .await + .expect("wrong-stamp release"), + "release with a non-matching stamp must not clear the claim" + ); + assert!( + !claim_due_reminder_with_stamp(&pool, &id, created_at, stamp) + .await + .expect("re-claim after no-op release"), + "reminder must still be claimed after a no-op release" + ); + + // The matching-stamp release rolls the claim back; the reminder is + // redeliverable and a subsequent claim wins again. + assert!( + release_due_reminder(&pool, &id, created_at, stamp) + .await + .expect("matching-stamp release"), + "release with the claiming stamp must clear the claim" + ); + assert!( + claim_due_reminder_with_stamp(&pool, &id, created_at, stamp) + .await + .expect("re-claim after release"), + "released reminder must be reclaimable for retry" + ); + } } diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index dd6e92e5d..499d8245d 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -515,14 +515,45 @@ 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 before side effect (§5c: claim-before-publish). A + // unique per-attempt stamp lets a failed publish roll back + // exactly this pod's claim via compare-and-clear, without a + // racing pod's later claim being clobbered. `delivered_at` + // is only ever read as a NULL/non-NULL sentinel (the + // due-reminder query guard and the partial index), never as + // a wall-clock value, so an opaque stamp is safe to store. let reminder_tenant = buzz_core::tenant::TenantContext::resolved( reminder.community_id, reminder.host.clone(), ); + let delivery_stamp = chrono::Utc::now() + .timestamp_nanos_opt() + .unwrap_or_else(|| chrono::Utc::now().timestamp()) + ^ rand::random::(); + + match scheduler_state + .db + .claim_due_reminder_with_stamp( + &reminder.id, + reminder.created_at, + delivery_stamp, + ) + .await + { + Ok(true) => {} // We won the claim — proceed to publish. + Ok(false) => continue, // Another pod claimed it; no side effect here. + Err(e) => { + warn!( + event_id = hex::encode(&reminder.id), + "Reminder scheduler: claim failed, skipping publish: {e}" + ); + continue; + } + } + + // Publish the single side effect. On failure, release our + // claim so the next tick (this pod or another) can retry — + // the stamp guard ensures we only clear our own claim. if let Err(e) = scheduler_state .pubsub .publish_event( @@ -534,23 +565,21 @@ async fn main() -> anyhow::Result<()> { { error!( event_id = hex::encode(&reminder.id), - "Reminder scheduler: Redis publish failed, skipping claim: {e}" + "Reminder scheduler: Redis publish failed after claim, releasing: {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) => { + if let Err(release_err) = scheduler_state + .db + .release_due_reminder( + &reminder.id, + reminder.created_at, + delivery_stamp, + ) + .await + { warn!( event_id = hex::encode(&reminder.id), - "Reminder scheduler: claim failed after publish (duplicate delivery possible): {e}" + "Reminder scheduler: release after failed publish errored \ + (reminder stays claimed, will not retry): {release_err}" ); } }