diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 660a55fef..c6e705f88 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -2872,39 +2872,53 @@ async fn emit_initial_ref_state( Ok(()) } +/// Reconcile one community's event-backed NIP-43 membership view. +/// +/// `relay_members` is canonical. The snapshot is rebuilt only when it is +/// absent or its member/role set differs from the canonical rows, so this is +/// cheap (two queries) when nothing changed. Returns whether a repair +/// publication happened. +pub async fn reconcile_nip43_membership_snapshot( + tenant: &TenantContext, + state: &Arc, +) -> anyhow::Result { + if !state + .db + .nip43_membership_snapshot_needs_reconciliation( + tenant.community(), + &state.relay_keypair.public_key(), + ) + .await? + { + return Ok(false); + } + + publish_nip43_membership_list(tenant, state).await?; + Ok(true) +} + /// Reconcile every community's event-backed NIP-43 membership view. /// -/// `relay_members` is canonical. A snapshot is rebuilt only when it is absent -/// or its member/role set differs from the canonical rows. This makes the sweep -/// safe to run at startup and periodically without producing an event stream -/// when nothing changed. A failure in one community is logged and counted but -/// does not prevent the remaining communities from being repaired. +/// A failure in one community is logged and counted but does not prevent the +/// remaining communities from being repaired. +/// +/// This sweep is O(communities) — sequential, two queries each, plus a +/// publication per repair. On a large deployment it takes minutes, so it MUST +/// NOT run on the pre-bind startup path (a startup probe SIGKILLs the pod +/// mid-sweep and the restart begins again at community #1, forever). It runs +/// post-bind, jittered, and leader-gated — see the sweep task in `main.rs`. pub async fn reconcile_nip43_membership_snapshots(state: &Arc) -> anyhow::Result { + let started_at = std::time::Instant::now(); let communities = state.db.usage_community_hosts().await?; + let total = communities.len(); + let mut scanned = 0usize; let mut reconciled = 0usize; for community in communities { let community_id = buzz_core::CommunityId::from_uuid(community.id); let host = community.host; - let result = async { - if !state - .db - .nip43_membership_snapshot_needs_reconciliation( - community_id, - &state.relay_keypair.public_key(), - ) - .await? - { - return Ok::(false); - } - - let tenant = TenantContext::resolved(community_id, host.clone()); - publish_nip43_membership_list(&tenant, state).await?; - Ok::(true) - } - .await; - - match result { + let tenant = TenantContext::resolved(community_id, host.clone()); + match reconcile_nip43_membership_snapshot(&tenant, state).await { Ok(true) => reconciled += 1, Ok(false) => {} Err(error) => { @@ -2913,9 +2927,21 @@ pub async fn reconcile_nip43_membership_snapshots(state: &Arc) -> anyh warn!(%community_id, %host, %error, "NIP-43 membership reconciliation failed"); } } + scanned += 1; + if scanned.is_multiple_of(1000) { + info!( + scanned, + total, + reconciled, + elapsed_ms = started_at.elapsed().as_millis() as u64, + "NIP-43 membership sweep progress" + ); + } } metrics::counter!("buzz_nip43_membership_reconciliations_total").increment(reconciled as u64); + metrics::histogram!("buzz_nip43_membership_sweep_seconds") + .record(started_at.elapsed().as_secs_f64()); Ok(reconciled) } diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 799cf9cf6..543641e76 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -83,6 +83,13 @@ impl EmissionScope { const USAGE_METRICS_LOCK_KEY: i64 = 0x4255_5A5A_4D45_5452; +/// Session advisory lock serializing the fleet-wide NIP-43 membership sweep. +/// +/// The sweep is O(communities) — minutes at fleet scale — so only one replica +/// may run it at a time; the rest skip their tick. Same detached-session +/// leadership mechanism as the usage-metrics poller, different key. +const NIP43_SWEEP_LOCK_KEY: i64 = 0x4255_5A5A_4E50_3433; + #[tokio::main] async fn main() -> anyhow::Result<()> { // Install the ring CryptoProvider for rustls. Required before any rustls @@ -527,16 +534,35 @@ async fn main() -> anyhow::Result<()> { ); } - // NIP-43: reconcile the event-backed roster for every provisioned - // community before opening the listener. `relay_members` is canonical; - // this repairs pre-snapshot communities and any publication that failed - // after a membership transaction committed. + // NIP-43: reconcile the event-backed roster for the deployment's own + // community before opening the listener. The fleet-wide sweep across every + // provisioned community deliberately does NOT run here: it is + // O(communities) — minutes at fleet scale — and a startup probe that + // SIGKILLs the pod mid-sweep restarts it from community #1 forever + // (permanent crashloop). The fleet sweep runs post-bind, jittered, and + // leader-gated in the periodic task spawned below. if config.require_relay_membership { - match buzz_relay::handlers::side_effects::reconcile_nip43_membership_snapshots(&state).await - { - Ok(count) => info!(count, "NIP-43 membership snapshots reconciled on startup"), - Err(error) => { - tracing::warn!(%error, "NIP-43 membership snapshot startup reconciliation failed") + // `deployment_community` is always Some here: startup fails fast above + // when membership is enforced and the community cannot be ensured. + if let Some(community) = deployment_community { + let host = buzz_relay::tenant::relay_url_authority(&config.relay_url); + let tenant = buzz_core::tenant::TenantContext::resolved(community, host); + let started_at = std::time::Instant::now(); + info!(community = %community, "NIP-43 startup phase: reconciling deployment community snapshot"); + match buzz_relay::handlers::side_effects::reconcile_nip43_membership_snapshot( + &tenant, &state, + ) + .await + { + Ok(repaired) => info!( + community = %community, + repaired, + elapsed_ms = started_at.elapsed().as_millis() as u64, + "NIP-43 startup phase: deployment community snapshot reconciled" + ), + Err(error) => { + tracing::warn!(%error, "NIP-43 deployment community startup reconciliation failed") + } } } @@ -547,24 +573,62 @@ async fn main() -> anyhow::Result<()> { .unwrap_or(60) .max(1); tokio::spawn(async move { + // Jitter the first tick by a random fraction of the interval so a + // rolling deploy of N pods doesn't contend for the sweep lock (and + // hammer `communities`) simultaneously at boot. True per-process + // randomness — PID-derived seeds are unsafe when every pod is PID 1. + let jitter_secs = rand::random::() % interval_secs; + tokio::time::sleep(std::time::Duration::from_secs(jitter_secs)).await; + let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs)); - interval.tick().await; + // A fleet-scale sweep takes longer than the interval; skip ticks + // rather than scheduling a catch-up burst behind it. + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { interval.tick().await; + // Per-tick leadership: hold the sweep lock only while sweeping, + // so a replica that dies mid-sweep releases it with its session + // and any peer picks up on its next tick. + let leader = match reconcile_state + .db + .try_lock_usage_metrics(NIP43_SWEEP_LOCK_KEY) + .await + { + Ok(Some(guard)) => guard, + Ok(None) => continue, + Err(error) => { + tracing::warn!(%error, "NIP-43 sweep leadership check failed"); + continue; + } + }; + let started_at = std::time::Instant::now(); + info!("NIP-43 membership sweep starting (leader)"); match buzz_relay::handlers::side_effects::reconcile_nip43_membership_snapshots( &reconcile_state, ) .await { - Ok(count) if count > 0 => { - info!(count, "NIP-43 membership snapshots repaired") + Ok(count) => { + let elapsed_ms = started_at.elapsed().as_millis() as u64; + if count > 0 { + info!( + count, + elapsed_ms, "NIP-43 membership sweep complete: snapshots repaired" + ) + } else { + info!( + elapsed_ms, + "NIP-43 membership sweep complete: nothing to repair" + ) + } } - Ok(_) => {} Err(error) => tracing::warn!( %error, - "periodic NIP-43 membership snapshot reconciliation failed" + elapsed_ms = started_at.elapsed().as_millis() as u64, + "NIP-43 membership sweep failed" ), } + drop(leader); } }); }