diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 14886aee2c..92f1773501 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -157,7 +157,6 @@ async fn event_write_transaction_preserves_legacy_acquisition_metrics() { #[test] fn nip43_reconciliation_compatibility_alias_is_preserved() { - #[allow(deprecated)] async fn call( db: &Db, community_id: CommunityId, diff --git a/crates/buzz-db/src/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs index 93b270efc9..e8939ecb3b 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -155,6 +155,17 @@ async fn nip43_membership_snapshot_drift( ) -> Result> { let mut connection = observability::acquire_writer(pool, observability::WriterOperation::Maintenance).await?; + nip43_membership_snapshot_drift_on(&mut connection, relay_pubkey, only).await +} + +/// [`nip43_membership_snapshot_drift`] on a caller-held connection, so the +/// snapshot publisher can re-check one community inside its locked +/// transaction. +async fn nip43_membership_snapshot_drift_on( + connection: &mut sqlx::PgConnection, + relay_pubkey: &nostr::PublicKey, + only: Option, +) -> Result> { let rows = sqlx::query_as::<_, (Uuid, String)>( r#" SELECT c.id, c.host @@ -209,7 +220,7 @@ async fn nip43_membership_snapshot_drift( .bind(relay_pubkey.to_bytes().to_vec()) .bind(buzz_core::kind::KIND_NIP43_MEMBERSHIP_LIST as i32) .bind(only.map(|community| *community.as_uuid())) - .fetch_all(&mut *connection) + .fetch_all(connection) .await?; Ok(rows .into_iter() @@ -1132,7 +1143,6 @@ impl Db { name = "nip43_membership_snapshot_needs_reconciliation", system = "postgresql" )] - #[deprecated(note = "use nip43_membership_snapshots_needing_reconciliation")] pub async fn nip43_membership_snapshot_needs_reconciliation( &self, community_id: CommunityId, @@ -1175,6 +1185,39 @@ impl Db { community_id: CommunityId, relay_keypair: &nostr::Keys, ) -> Result<(StoredEvent, bool, usize)> { + self.write_nip43_membership_snapshot(community_id, relay_keypair, false) + .await? + .ok_or_else(|| { + DbError::InvalidData("unconditional NIP-43 publication was skipped".to_string()) + }) + } + + /// Like [`Self::publish_nip43_membership_locked`], but after taking the + /// per-community lock it re-checks whether the live snapshot still + /// differs from the canonical rows, and writes nothing (`None`) when it + /// does not. + /// + /// The background reconciler lists drifted communities once per pass and + /// then publishes them one by one; several pods may run passes at the + /// same time, and live membership writes republish too. Checking under + /// the lock means each community is repaired once however many passes + /// overlap. + #[datastore_span(name = "republish_nip43_membership_if_drifted", system = "postgresql")] + pub async fn republish_nip43_membership_if_drifted( + &self, + community_id: CommunityId, + relay_keypair: &nostr::Keys, + ) -> Result> { + self.write_nip43_membership_snapshot(community_id, relay_keypair, true) + .await + } + + async fn write_nip43_membership_snapshot( + &self, + community_id: CommunityId, + relay_keypair: &nostr::Keys, + only_if_drifted: bool, + ) -> Result> { use nostr::{EventBuilder, Kind, Tag}; let kind_i32 = buzz_core::kind::KIND_NIP43_MEMBERSHIP_LIST as i32; @@ -1196,7 +1239,7 @@ impl Db { let transaction_timer = observability::TransactionTimer::start( observability::TransactionOperation::PublishNip43MembershipLocked, ); - let (event, received_at, was_inserted, member_count) = transaction_timer + let Some((event, received_at, was_inserted, member_count)) = transaction_timer .observe(async { // Acquire the per-community snapshot lock BEFORE reading members. @@ -1211,6 +1254,19 @@ impl Db { ) .await?; + if only_if_drifted + && nip43_membership_snapshot_drift_on( + &mut tx, + &relay_keypair.public_key(), + Some(community_id), + ) + .await? + .is_empty() + { + tx.rollback().await?; + return Ok(None); + } + // Read current members inside the locked transaction. let rows = sqlx::query( "SELECT pubkey, role FROM relay_members \ @@ -1289,9 +1345,12 @@ impl Db { } else { tx.rollback().await?; } - Ok::<_, DbError>((event, received_at, was_inserted, member_count)) + Ok::<_, DbError>(Some((event, received_at, was_inserted, member_count))) }) - .await?; + .await? + else { + return Ok(None); + }; if was_inserted { if let Err(e) = crate::insert_mentions(&self.pool, community_id, &event, None).await { @@ -1299,11 +1358,11 @@ impl Db { } } - Ok(( + Ok(Some(( StoredEvent::with_received_at(event, received_at, None, was_inserted), was_inserted, member_count, - )) + ))) } } @@ -2042,4 +2101,81 @@ mod postgres_tests { ); } } + + /// The reconciler's publish re-checks drift under the per-community lock: + /// a community whose snapshot became current after the pass listed it is + /// left alone, so overlapping passes on several pods repair it once. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn republish_if_drifted_writes_only_when_the_locked_snapshot_is_stale() { + let pool = setup_pool().await; + let db = Db::from_pool(pool.clone()); + let relay = nostr::Keys::generate(); + // Older than any publish below, so a write cannot collide with the + // fixture's event id and a skipped write is visible in the row count. + let earlier = chrono::Utc::now().timestamp() - 60; + + async fn snapshot_rows(pool: &PgPool, community: CommunityId) -> i64 { + sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id = $1 AND kind = $2") + .bind(community.as_uuid()) + .bind(buzz_core::kind::KIND_NIP43_MEMBERSHIP_LIST as i32) + .fetch_one(pool) + .await + .expect("count snapshots") + } + + let (current, owner) = owned_community(&pool).await; + insert_snapshot( + &pool, + current, + &relay.public_key(), + Snapshot::at(earlier, 1), + member_tags(&[(&owner, "owner")]), + ) + .await; + assert!( + db.republish_nip43_membership_if_drifted(current, &relay) + .await + .expect("republish current") + .is_none(), + "a current snapshot is not republished" + ); + assert_eq!(snapshot_rows(&pool, current).await, 1, "nothing written"); + + let (stale, owner) = owned_community(&pool).await; + insert_snapshot( + &pool, + stale, + &relay.public_key(), + Snapshot::at(earlier, 2), + member_tags(&[(&owner, "member")]), + ) + .await; + let (_, inserted, members) = db + .republish_nip43_membership_if_drifted(stale, &relay) + .await + .expect("republish stale") + .expect("a stale snapshot is republished"); + assert!(inserted); + assert_eq!(members, 1); + assert!( + nip43_membership_snapshot_drift(&pool, &relay.public_key(), Some(stale)) + .await + .expect("drift after repair") + .is_empty(), + "the republished snapshot matches the roster" + ); + assert!( + db.republish_nip43_membership_if_drifted(stale, &relay) + .await + .expect("second republish") + .is_none(), + "a second pass that listed the same community writes nothing" + ); + assert_eq!( + snapshot_rows(&pool, stale).await, + 2, + "fixture plus one repair" + ); + } } diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 8522b8f623..bd7bc4a9fb 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -3122,21 +3122,50 @@ pub async fn reconcile_nip43_membership_snapshots(state: &Arc) -> anyh /// deployment costs one query and produces no events. A failure in one /// community is logged and counted but does not prevent the remaining /// communities from being repaired; the next sweep retries it. +/// +/// Each community is re-checked under its publication lock, so a community +/// that another pass or a live membership write already repaired is left +/// alone and counts as neither repaired nor failed. A pass whose drift query +/// fails increments `buzz_nip43_membership_reconciliation_sweep_failures_total` +/// before returning the error. pub async fn reconcile_nip43_membership_snapshot_drift( state: &Arc, ) -> anyhow::Result { - let communities = state + let communities = match state .db .nip43_membership_snapshots_needing_reconciliation(&state.relay_keypair.public_key()) - .await?; + .await + { + Ok(communities) => communities, + Err(error) => { + metrics::counter!("buzz_nip43_membership_reconciliation_sweep_failures_total") + .increment(1); + return Err(error.into()); + } + }; + let summary = repair_nip43_snapshot_drift(state, communities).await; + metrics::counter!("buzz_nip43_membership_reconciliations_total") + .increment(summary.repaired as u64); + Ok(summary) +} + +/// Republish each listed community's snapshot if it is still drifted once +/// its publication lock is held. The list can be stale by then: another pod's +/// pass or a live membership write may already have repaired a community, and +/// that community counts as neither repaired nor failed. +async fn repair_nip43_snapshot_drift( + state: &Arc, + communities: Vec, +) -> ReconcileSummary { let mut summary = ReconcileSummary::default(); for community in communities { let community_id = buzz_core::CommunityId::from_uuid(community.id); let host = community.host; let tenant = TenantContext::resolved(community_id, host.clone()); - match publish_nip43_membership_list(&tenant, state).await { - Ok(()) => summary.repaired += 1, + match publish_nip43_membership_snapshot(&tenant, state, true).await { + Ok(true) => summary.repaired += 1, + Ok(false) => {} Err(error) => { summary.failed += 1; metrics::counter!("buzz_nip43_membership_reconciliation_failures_total") @@ -3146,9 +3175,44 @@ pub async fn reconcile_nip43_membership_snapshot_drift( } } - metrics::counter!("buzz_nip43_membership_reconciliations_total") - .increment(summary.repaired as u64); - Ok(summary) + summary +} + +/// Start the background NIP-43 snapshot reconciler. +/// +/// The first pass starts as soon as the task is scheduled; later passes follow +/// `interval`, and a slow pass delays the next tick rather than bursting to +/// catch up. This is a plain `fn` returning the task handle, so relay startup +/// cannot await a pass before opening its listener. The first pass always logs +/// its counts and duration; later passes log only when they repaired or +/// failed something. +pub fn spawn_nip43_snapshot_reconciler( + state: Arc, + interval: std::time::Duration, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let mut interval = tokio::time::interval(interval); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + let mut first_pass = true; + loop { + interval.tick().await; + let started = std::time::Instant::now(); + match reconcile_nip43_membership_snapshot_drift(&state).await { + Ok(summary) if first_pass || summary.repaired > 0 || summary.failed > 0 => info!( + count = summary.repaired, + failed = summary.failed, + elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX), + "NIP-43 membership snapshots reconciled" + ), + Ok(_) => {} + Err(error) => warn!( + %error, + "periodic NIP-43 membership snapshot reconciliation failed" + ), + } + first_pass = false; + } + }) } /// Publish a kind:13534 relay membership list event (NIP-43). @@ -3165,10 +3229,23 @@ pub async fn publish_nip43_membership_list( tenant: &TenantContext, state: &Arc, ) -> anyhow::Result<()> { + publish_nip43_membership_snapshot(tenant, state, false) + .await + .map(|_| ()) +} + +/// Publish the kind:13534 snapshot, or with `only_if_drifted` skip it when the +/// live snapshot already matches the canonical rows. Returns whether a new +/// snapshot was written. +async fn publish_nip43_membership_snapshot( + tenant: &TenantContext, + state: &Arc, + only_if_drifted: bool, +) -> anyhow::Result { let started_at = std::time::Instant::now(); metrics::counter!("buzz_nip43_membership_publications_total", "result" => "attempted") .increment(1); - let result = publish_nip43_membership_list_inner(tenant, state).await; + let result = publish_nip43_membership_list_inner(tenant, state, only_if_drifted).await; metrics::histogram!("buzz_nip43_membership_publication_seconds") .record(started_at.elapsed().as_secs_f64()); metrics::counter!( @@ -3182,7 +3259,8 @@ pub async fn publish_nip43_membership_list( async fn publish_nip43_membership_list_inner( tenant: &TenantContext, state: &Arc, -) -> anyhow::Result<()> { + only_if_drifted: bool, +) -> anyhow::Result { let relay_pubkey_hex = state.relay_keypair.public_key().to_hex(); let community = tenant.community(); @@ -3191,10 +3269,22 @@ async fn publish_nip43_membership_list_inner( // concurrent publication that reads newer state will block until our // replacement commits, then read the updated membership. This prevents a // stale snapshot from winning by arrival order. - let (stored, was_inserted, member_count) = state - .db - .publish_nip43_membership_locked(community, &state.relay_keypair) - .await?; + let published = if only_if_drifted { + state + .db + .republish_nip43_membership_if_drifted(community, &state.relay_keypair) + .await? + } else { + Some( + state + .db + .publish_nip43_membership_locked(community, &state.relay_keypair) + .await?, + ) + }; + let Some((stored, was_inserted, member_count)) = published else { + return Ok(false); + }; if was_inserted { dispatch_persistent_event( @@ -3209,7 +3299,7 @@ async fn publish_nip43_membership_list_inner( } info!(member_count, "NIP-43 membership list published"); - Ok(()) + Ok(was_inserted) } /// Shared helper: publish a NIP-43 membership delta event (kind 8000 or 8001). @@ -3983,19 +4073,10 @@ mod tests { } fn reconciliation_failures(recorder: &DebuggingRecorder) -> u64 { - recorder - .snapshotter() - .snapshot() - .into_vec() - .into_iter() - .filter(|(key, _, _, _)| { - key.key().name() == "buzz_nip43_membership_reconciliation_failures_total" - }) - .map(|(_, _, _, value)| match value { - DebugValue::Counter(value) => value, - other => panic!("reconciliation failures must be a counter: {other:?}"), - }) - .sum() + counter_total( + recorder, + "buzz_nip43_membership_reconciliation_failures_total", + ) } /// Regression for #7558 on the worker path: repeated NIP-43 @@ -4038,7 +4119,7 @@ mod tests { "no community may fail reconciliation on any sweep" ); assert_eq!( - membership_snapshots(&pool, active).await, + relay_signed_snapshots(&pool, active, &state).await, 1, "the active community is reconciled once and then left alone" ); @@ -4062,5 +4143,130 @@ mod tests { .expect("tombstone row is retained"); assert_eq!(tombstone_state, "tombstone"); } + + async fn relay_signed_snapshots( + pool: &sqlx::PgPool, + id: Uuid, + state: &Arc, + ) -> i64 { + sqlx::query_scalar( + "SELECT count(*) FROM events \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3", + ) + .bind(id) + .bind(KIND_NIP43_MEMBERSHIP_LIST as i32) + .bind(state.relay_keypair.public_key().to_bytes().to_vec()) + .fetch_one(pool) + .await + .expect("count relay-signed membership snapshots") + } + + fn counter_total(recorder: &DebuggingRecorder, name: &str) -> u64 { + recorder + .snapshotter() + .snapshot() + .into_vec() + .into_iter() + .filter(|(key, _, _, _)| key.key().name() == name) + .map(|(_, _, _, value)| match value { + DebugValue::Counter(value) => value, + other => panic!("{name} must be a counter: {other:?}"), + }) + .sum() + } + + /// The reconciler's first pass runs as soon as it is spawned, not one + /// interval later: with an hour-long interval a community missing its + /// snapshot is repaired long before the hour. Real time, not a paused clock, + /// because a paused clock auto-advances while the task waits on + /// Postgres and would hide a first-tick delay. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn reconciler_first_pass_runs_without_waiting_an_interval() { + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect to test DB"); + let state = crate::state::tests::test_state_with_database_pool(pool.clone()).await; + let community = insert_community(&pool).await; + + let reconciler = spawn_nip43_snapshot_reconciler( + Arc::clone(&state), + std::time::Duration::from_secs(3600), + ); + // The deadline is generous because the pass sweeps every community + // in the shared test database; it only has to be well short of the + // hour a delayed first tick would take. + let repaired = tokio::time::timeout(std::time::Duration::from_secs(300), async { + // Count only this state's snapshots: concurrent tests sweep the + // same database with their own relay keys. + while relay_signed_snapshots(&pool, community, &state).await == 0 { + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + }) + .await; + reconciler.abort(); + assert!( + repaired.is_ok(), + "the first reconcile pass must not wait for the interval" + ); + } + + /// A pass whose drift query fails surfaces as a counter, not only a + /// log line, so a backstop that fails every pass is visible. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn failed_sweep_query_increments_the_sweep_failure_counter() { + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect to test DB"); + let state = crate::state::tests::test_state_with_database_pool(pool.clone()).await; + pool.close().await; + + let recorder = DebuggingRecorder::new(); + let _guard = metrics::set_default_local_recorder(&recorder); + assert!(reconcile_nip43_membership_snapshot_drift(&state) + .await + .is_err()); + assert_eq!( + counter_total( + &recorder, + "buzz_nip43_membership_reconciliation_sweep_failures_total" + ), + 1 + ); + } + + /// A community the drift query listed, but that another pod's pass or + /// a live write repaired before this pass took its lock, is skipped: + /// no write, and it counts as neither repaired nor failed. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn community_repaired_after_listing_is_skipped_not_failed() { + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect to test DB"); + let state = crate::state::tests::test_state_with_database_pool(pool.clone()).await; + let id = insert_community(&pool).await; + let host = format!("nip43-sweep-{}.example", id.simple()); + let tenant = + TenantContext::resolved(buzz_core::CommunityId::from_uuid(id), host.clone()); + assert!(publish_nip43_membership_snapshot(&tenant, &state, false) + .await + .expect("publish current snapshot")); + + let recorder = DebuggingRecorder::new(); + let summary = { + let _guard = metrics::set_default_local_recorder(&recorder); + repair_nip43_snapshot_drift( + &state, + vec![buzz_db::usage::CommunityHost { id, host }], + ) + .await + }; + + assert_eq!((summary.repaired, summary.failed), (0, 0)); + assert_eq!(reconciliation_failures(&recorder), 0); + assert_eq!(relay_signed_snapshots(&pool, id, &state).await, 1); + } } } diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index b36dae7054..eeb2b75d06 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -855,9 +855,9 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { // in step with the canonical `relay_members` rows. Membership writes // republish on every change; this background sweep repairs snapshots that // are missing or whose publication failed after the membership commit. - // It runs off the startup path: the first pass starts now and each later - // pass follows the interval, so a large deployment opens its listener - // without waiting on it. A failed community is retried on the next pass. + // It runs off the startup path (see `spawn_nip43_snapshot_reconciler`), so + // a large deployment opens its listener without waiting on it. A failed + // community is retried on the next pass. if config.require_relay_membership { let reconcile_state = Arc::clone(&state); let interval_secs = std::env::var("BUZZ_NIP43_RECONCILE_INTERVAL_SECS") @@ -865,36 +865,10 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { .and_then(|value| value.parse::().ok()) .unwrap_or(60) .max(1); - tokio::spawn(async move { - let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs)); - interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - let mut first_pass = true; - loop { - interval.tick().await; - let started = std::time::Instant::now(); - match buzz_relay::handlers::side_effects::reconcile_nip43_membership_snapshot_drift( - &reconcile_state, - ) - .await - { - Ok(summary) if first_pass || summary.repaired > 0 || summary.failed > 0 => { - info!( - count = summary.repaired, - failed = summary.failed, - elapsed_ms = - u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX), - "NIP-43 membership snapshots reconciled" - ) - } - Ok(_) => {} - Err(error) => tracing::warn!( - %error, - "periodic NIP-43 membership snapshot reconciliation failed" - ), - } - first_pass = false; - } - }); + buzz_relay::handlers::side_effects::spawn_nip43_snapshot_reconciler( + reconcile_state, + std::time::Duration::from_secs(interval_secs), + ); } // Emit kind:39000/39002 discovery events for channels that exist in the DB