From 12604c67f54a8b3196af88a79b646f2fd0011de6 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Tue, 6 Oct 2026 23:10:10 -0400 Subject: [PATCH] fix(relay): close review gaps in the NIP-43 drift reconciler - Re-check drift for each community under its publication lock and skip the write when the snapshot is already current, so passes that overlap across pods (or with live membership writes) repair a community once instead of once per pass. - Extract spawn_nip43_snapshot_reconciler, a sync fn, so startup cannot await a pass, and pin that the first pass does not wait an interval. - Count a pass whose drift query fails in buzz_nip43_membership_reconciliation_sweep_failures_total. - Pin that a community repaired between listing and locking is skipped, not counted as failed. - Un-deprecate the single-community drift check; it is now a wrapper over the same query, and the sweep is not a drop-in replacement for it. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/runtime/tests.rs | 1 - crates/buzz-db/src/store/relay_members.rs | 150 +++++++++- .../buzz-relay/src/handlers/side_effects.rs | 262 ++++++++++++++++-- crates/buzz-relay/src/main.rs | 40 +-- 4 files changed, 384 insertions(+), 69 deletions(-) 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