diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index e13b06c81d..f495983a28 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -85,6 +85,6 @@ pub use event::{EventQuery, DEFAULT_MAX_PAGE_LIMIT}; pub use nip_fi_binding::{NipFiBindingRecord, NipFiUnbind, RowCapExceeded}; pub use reaction::ReactionEventInsertOutcome; pub use reminder::DueReminder; -pub use usage::UsageMetricsLeader; +pub use usage::SessionLeader; use buzz_core::CommunityId; diff --git a/crates/buzz-db/src/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs index 6b21c7ce01..1e8ca64fa9 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -1169,9 +1169,7 @@ impl Db { name = "nip43_membership_snapshot_needs_reconciliation", system = "postgresql" )] - #[deprecated( - note = "use nip43_membership_snapshot_needs_reconciliation_for_bootstrap or nip43_membership_snapshot_needs_reconciliation_for_maintenance" - )] + #[deprecated(note = "use nip43_membership_snapshot_needs_reconciliation_for_maintenance")] pub async fn nip43_membership_snapshot_needs_reconciliation( &self, community_id: CommunityId, @@ -1185,24 +1183,6 @@ impl Db { .await } - /// Startup-attributed variant of the NIP-43 snapshot comparison. - #[datastore_span( - name = "nip43_membership_snapshot_needs_reconciliation_for_bootstrap", - system = "postgresql" - )] - pub async fn nip43_membership_snapshot_needs_reconciliation_for_bootstrap( - &self, - community_id: CommunityId, - relay_pubkey: &nostr::PublicKey, - ) -> Result { - self.nip43_membership_snapshot_needs_reconciliation_with_operation( - community_id, - relay_pubkey, - observability::WriterOperation::Bootstrap, - ) - .await - } - /// Periodic maintenance variant of the NIP-43 snapshot comparison. #[datastore_span( name = "nip43_membership_snapshot_needs_reconciliation_for_maintenance", diff --git a/crates/buzz-db/src/store/usage.rs b/crates/buzz-db/src/store/usage.rs index ea626c4ad6..09c0140262 100644 --- a/crates/buzz-db/src/store/usage.rs +++ b/crates/buzz-db/src/store/usage.rs @@ -195,16 +195,16 @@ where .await?) } -/// Owns the detached Postgres session holding the relay usage-metrics advisory lock. +/// Owns the detached Postgres session holding a deployment-wide leader advisory lock. /// /// The connection deliberately does not return to the main pool: session advisory -/// locks must remain bound to this exact physical connection, and the poller -/// pings it before each leader-only collection tick. -pub struct UsageMetricsLeader { +/// locks must remain bound to this exact physical connection, and each leader-only +/// loop (usage metrics, NIP-43 snapshot reconciliation) pings it before working. +pub struct SessionLeader { connection: PgConnection, } -impl UsageMetricsLeader { +impl SessionLeader { /// Returns whether the lock-owning session is still reachable. /// /// Bounded to 5 seconds — a blackholed connection (no RST) would otherwise @@ -604,17 +604,15 @@ fn into_community_hosts(rows: Vec<(Uuid, String)>) -> Vec { } impl Db { - /// Try to acquire the detached session advisory lock for relay usage metrics. + /// Try to acquire the detached session advisory lock `lock_key`, making this + /// replica the leader for the job that key names. /// /// The returned guard owns the exact connection that acquired the lock. It is /// detached from the shared pool so a stable leader neither returns a locked /// session to other callers nor permanently consumes a pool slot. Dropping the /// guard closes the connection and releases the session-scoped lock. - #[datastore_span(name = "try_lock_usage_metrics", system = "postgresql")] - pub async fn try_lock_usage_metrics( - &self, - lock_key: i64, - ) -> Result> { + #[datastore_span(name = "try_lock_session_leader", system = "postgresql")] + pub async fn try_lock_session_leader(&self, lock_key: i64) -> Result> { let mut connection = observability::acquire_writer_with_legacy_metrics( &self.pool, observability::WriterOperation::Maintenance, @@ -625,7 +623,7 @@ impl Db { .fetch_one(&mut *connection) .await?; if acquired { - Ok(Some(UsageMetricsLeader { + Ok(Some(SessionLeader { connection: connection.detach(), })) } else { @@ -770,12 +768,6 @@ impl Db { pub async fn active_community_hosts(&self) -> Result> { active_community_hosts(&self.pool, observability::WriterOperation::Maintenance).await } - - /// Return active community host mappings during startup bootstrap work. - #[datastore_span(name = "bootstrap_community_hosts", system = "postgresql")] - pub async fn bootstrap_community_hosts(&self) -> Result> { - active_community_hosts(&self.pool, observability::WriterOperation::Bootstrap).await - } } #[cfg(test)] @@ -913,14 +905,14 @@ mod postgres_tests { let key = 0x4255_5A5A_4D45_5452; let mut leader = first - .try_lock_usage_metrics(key) + .try_lock_session_leader(key) .await .expect("first lock attempt") .expect("first database handle becomes leader"); assert!(leader.is_live().await, "lock owner remains reachable"); assert!( second - .try_lock_usage_metrics(key) + .try_lock_session_leader(key) .await .expect("second lock attempt") .is_none(), @@ -933,7 +925,7 @@ mod postgres_tests { let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5); let successor = loop { let attempt = second - .try_lock_usage_metrics(key) + .try_lock_session_leader(key) .await .expect("lock attempt after leader drop"); if attempt.is_some() || tokio::time::Instant::now() >= deadline { @@ -1183,10 +1175,7 @@ mod postgres_tests { let db = Db::from_pool(pool.clone()); for sweep in 0..2 { - for (caller, hosts) in [ - ("maintenance", db.active_community_hosts().await), - ("bootstrap", db.bootstrap_community_hosts().await), - ] { + for (caller, hosts) in [("maintenance", db.active_community_hosts().await)] { let hosts = hosts.unwrap_or_else(|e| panic!("{caller} sweep {sweep}: {e}")); let ids: std::collections::HashSet = hosts.iter().map(|h| h.id).collect(); assert_eq!( diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 2b505e01ec..87a35e2d92 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -472,7 +472,7 @@ fn p0_pool_acquisitions_use_typed_operation_pairs_without_other() { .next() .expect("usage production source"); let usage_leader_lock = usage_production - .split_once("pub async fn try_lock_usage_metrics(") + .split_once("pub async fn try_lock_session_leader(") .expect("usage store must expose the legacy leader-lock acquisition") .1 .split_once("pub async fn usage_community_count(") diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index f5b55410d3..426d268e44 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -3350,21 +3350,6 @@ async fn emit_initial_ref_state( Ok(()) } -/// 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. -#[derive(Clone, Copy)] -pub enum Nip43ReconciliationPurpose { - /// Before listener admission opens. - Bootstrap, - /// Periodic background repair after startup. - Maintenance, -} - /// Per-item counts from one snapshot reconciliation sweep. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub struct ReconcileSummary { @@ -3374,65 +3359,29 @@ pub struct ReconcileSummary { pub failed: usize, } -/// Preserve the original maintenance reconciliation API for downstream callers. -#[deprecated(note = "use reconcile_nip43_membership_snapshots_with_purpose")] +/// Reconcile every active 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 +/// periodic background sweep safe to repeat 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. Returns the +/// number of snapshots republished. pub async fn reconcile_nip43_membership_snapshots(state: &Arc) -> anyhow::Result { - reconcile_nip43_membership_snapshots_with_purpose( - state, - Nip43ReconciliationPurpose::Maintenance, - ) - .await - .map(|summary| summary.repaired) -} - -/// Reconcile NIP-43 snapshots with explicit startup or maintenance attribution. -pub async fn reconcile_nip43_membership_snapshots_with_purpose( - state: &Arc, - purpose: Nip43ReconciliationPurpose, -) -> anyhow::Result { - let communities = match purpose { - Nip43ReconciliationPurpose::Bootstrap => state.db.bootstrap_community_hosts().await?, - Nip43ReconciliationPurpose::Maintenance => state.db.active_community_hosts().await?, - }; + let communities = state.db.active_community_hosts().await?; let mut reconciled = 0usize; - let mut failed = 0usize; - let total = communities.len(); - let started = std::time::Instant::now(); - for (index, community) in communities.into_iter().enumerate() { - if nip43_progress_due(purpose, index, total) { - info!( - done = index, - total, - republished = reconciled, - failed, - elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX), - "NIP-43 startup reconciliation progress" - ); - } + for community in communities { let community_id = buzz_core::CommunityId::from_uuid(community.id); let host = community.host; let result = async { - let needs_reconciliation = match purpose { - Nip43ReconciliationPurpose::Bootstrap => { - state - .db - .nip43_membership_snapshot_needs_reconciliation_for_bootstrap( - community_id, - &state.relay_keypair.public_key(), - ) - .await? - } - Nip43ReconciliationPurpose::Maintenance => { - state - .db - .nip43_membership_snapshot_needs_reconciliation_for_maintenance( - community_id, - &state.relay_keypair.public_key(), - ) - .await? - } - }; + let needs_reconciliation = state + .db + .nip43_membership_snapshot_needs_reconciliation_for_maintenance( + community_id, + &state.relay_keypair.public_key(), + ) + .await?; if !needs_reconciliation { return Ok::(false); } @@ -3447,7 +3396,6 @@ pub async fn reconcile_nip43_membership_snapshots_with_purpose( Ok(true) => reconciled += 1, Ok(false) => {} Err(error) => { - failed += 1; metrics::counter!("buzz_nip43_membership_reconciliation_failures_total") .increment(1); warn!(%community_id, %host, %error, "NIP-43 membership reconciliation failed"); @@ -3456,24 +3404,77 @@ pub async fn reconcile_nip43_membership_snapshots_with_purpose( } metrics::counter!("buzz_nip43_membership_reconciliations_total").increment(reconciled as u64); - Ok(ReconcileSummary { - repaired: reconciled, - failed, - }) + Ok(reconciled) } -/// Communities between startup NIP-43 reconciliation progress logs. -const NIP43_BOOTSTRAP_PROGRESS_INTERVAL: usize = 5_000; +/// Deployment-wide leader lock for the NIP-43 snapshot reconciler. +pub const NIP43_RECONCILE_LOCK_KEY: i64 = 0x4255_5A5A_4E34_3352; -/// Whether a sweep logs progress before processing community `done` of -/// `total`. Only the startup sweep logs (the 60 s maintenance sweep stays -/// quiet): every [`NIP43_BOOTSTRAP_PROGRESS_INTERVAL`] communities, never at -/// the start (nothing done yet) and never past the end. -fn nip43_progress_due(purpose: Nip43ReconciliationPurpose, done: usize, total: usize) -> bool { - matches!(purpose, Nip43ReconciliationPurpose::Bootstrap) - && done > 0 - && done < total - && done.is_multiple_of(NIP43_BOOTSTRAP_PROGRESS_INTERVAL) +/// How often a standby retries the NIP-43 reconciler leader lock. +pub const NIP43_STANDBY_RETRY: std::time::Duration = std::time::Duration::from_secs(5); + +/// Spawn the background NIP-43 snapshot reconciler. +/// +/// In steady state only the replica holding the `lock_key` session advisory +/// lock sweeps. The leader starts each sweep `every` after the previous one +/// finished; a standby retries the lock every `standby_retry` (capped at +/// `every`), so it takes over within `standby_retry` of the leader's session +/// ending and sweeps at once. Postgres ends that session when the leader's +/// process exits; a leader lost with its host or network holds the lock until +/// the server's TCP keepalive drops the session, and nobody sweeps meanwhile. +/// The lease is checked only between sweeps, so a leader that loses its +/// session mid-sweep finishes that sweep while the new leader starts one; the +/// per-community publication lock keeps the overlap safe. The first lock +/// attempt runs as soon as the task does. Startup never waits on it: the +/// snapshots live in the shared database, so a new replica has nothing of its +/// own to repair before it serves. Trade-off: a missing or stale kind:13534 +/// snapshot is served until the leader's sweep reaches that community. +pub fn spawn_nip43_membership_reconciler( + state: Arc, + lock_key: i64, + every: std::time::Duration, + standby_retry: std::time::Duration, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let mut leader: Option = None; + loop { + if let Some(guard) = leader.as_mut() { + if !guard.is_live().await { + warn!("NIP-43 reconciler leader lock connection lost; demoting"); + leader = None; + } + } + if leader.is_none() { + match state.db.try_lock_session_leader(lock_key).await { + Ok(Some(guard)) => { + info!("Acquired NIP-43 reconciler leader lock"); + leader = Some(guard); + } + Ok(None) => {} + Err(error) => { + warn!(%error, "NIP-43 reconciler leader lock attempt failed") + } + } + } + if leader.is_some() { + match reconcile_nip43_membership_snapshots(&state).await { + Ok(repaired) if repaired > 0 => { + info!(count = repaired, "NIP-43 membership snapshots repaired") + } + Ok(_) => {} + Err(error) => { + warn!(%error, "periodic NIP-43 membership snapshot reconciliation failed") + } + } + } + let pause = if leader.is_some() { + every + } else { + standby_retry.min(every) + }; + tokio::time::sleep(pause).await; + } + }) } /// Publish a kind:13534 relay membership list event (NIP-43). @@ -4148,37 +4149,6 @@ mod tests { assert!(initial_members(&headless).is_err()); } - #[test] - fn nip43_progress_logs_every_interval_inside_the_startup_sweep_only() { - use Nip43ReconciliationPurpose::{Bootstrap, Maintenance}; - const STEP: usize = NIP43_BOOTSTRAP_PROGRESS_INTERVAL; - let total = 79_942; - let due: Vec = (0..total) - .filter(|&done| nip43_progress_due(Bootstrap, done, total)) - .collect(); - assert_eq!(due.first(), Some(&STEP)); - assert_eq!(due.len(), total / STEP); - assert!(due.iter().all(|done| done.is_multiple_of(STEP))); - - // (purpose, done, total, expected) - for (purpose, done, total, expected) in [ - (Bootstrap, STEP, total, true), - (Maintenance, STEP, total, false), - (Bootstrap, 0, total, false), - (Bootstrap, STEP - 1, total, false), - (Bootstrap, STEP, STEP, false), - (Bootstrap, STEP, STEP + 1, true), - ] { - assert_eq!( - nip43_progress_due(purpose, done, total), - expected, - "done={done} total={total} bootstrap={}", - matches!(purpose, Bootstrap) - ); - } - assert!(!(0..total).any(|done| nip43_progress_due(Maintenance, done, total))); - } - #[test] fn workflow_deletion_retry_matches_authorized_dispatch() { let keys = nostr::Keys::generate(); @@ -4226,16 +4196,6 @@ mod tests { } } - #[test] - fn nip43_reconciliation_compatibility_alias_is_preserved() { - #[allow(deprecated)] - async fn call(state: &Arc) -> anyhow::Result { - reconcile_nip43_membership_snapshots(state).await - } - - let _ = call; - } - #[test] fn group_members_snapshot_keeps_members_past_one_thousand() { let channel_id = Uuid::new_v4(); @@ -4416,6 +4376,149 @@ mod tests { .sum() } + /// A freshly migrated private database, so no other test's sweep can + /// publish the snapshot a reconciler test is waiting for. + struct ScratchDb { + admin: sqlx::PgPool, + name: String, + pool: sqlx::PgPool, + } + + impl ScratchDb { + async fn create(prefix: &str) -> Self { + let admin_url = crate::test_support::database_url(); + let admin = sqlx::PgPool::connect(&admin_url) + .await + .expect("connect admin to create scratch db"); + let name = format!("{prefix}_{}", Uuid::new_v4().simple()); + sqlx::query(sqlx::AssertSqlSafe(format!("CREATE DATABASE {name}"))) + .execute(&admin) + .await + .expect("create scratch db"); + let slash = admin_url + .rfind('/') + .expect("URL must have a path component"); + let pool = sqlx::PgPool::connect(&format!("{}/{name}", &admin_url[..slash])) + .await + .expect("connect scratch db"); + buzz_db::migration::run_migrations(&pool) + .await + .expect("migrate scratch db"); + Self { admin, name, pool } + } + + async fn drop_db(self) { + self.pool.close().await; + let _ = sqlx::query(sqlx::AssertSqlSafe(format!( + "DROP DATABASE IF EXISTS {} WITH (FORCE)", + self.name + ))) + .execute(&self.admin) + .await; + } + } + + /// A per-test leader key, so concurrent tests and any relay sharing + /// the test database never contend for the production lock. + fn scratch_lock_key() -> i64 { + Uuid::new_v4().as_u64_pair().0 as i64 + } + + /// Polls until `community` has a membership snapshot, up to `budget`. + async fn snapshot_within( + pool: &sqlx::PgPool, + community: Uuid, + budget: std::time::Duration, + ) -> bool { + tokio::time::timeout(budget, async { + while membership_snapshots(pool, community).await == 0 { + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + }) + .await + .is_ok() + } + + const DAY: std::time::Duration = std::time::Duration::from_secs(24 * 60 * 60); + + /// The background reconciler's first sweep runs right away rather + /// than one interval after startup; with the pre-bind sweep gone this + /// is the only thing that repairs a missing snapshot on boot. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn background_reconciler_sweeps_immediately() { + let db = ScratchDb::create("nip43_first_sweep").await; + let state = crate::state::tests::test_state_with_database_pool(db.pool.clone()).await; + let community = insert_community(&db.pool).await; + assert_eq!(membership_snapshots(&db.pool, community).await, 0); + + // Interval and standby retry both far longer than the test: only + // the immediate first sweep can publish the snapshot. + let reconciler = + spawn_nip43_membership_reconciler(Arc::clone(&state), scratch_lock_key(), DAY, DAY); + let published = + snapshot_within(&db.pool, community, std::time::Duration::from_secs(60)).await; + reconciler.abort(); + let snapshots = membership_snapshots(&db.pool, community).await; + drop(state); + db.drop_db().await; + + assert!( + published, + "the first background sweep must publish the missing snapshot without waiting an interval" + ); + assert_eq!(snapshots, 1); + } + + /// Only the leader-lock holder sweeps. A replica that can't take the + /// lock never publishes, and it takes over on its short standby retry + /// once the holder lets go, not after a full sweep interval. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn background_reconciler_standby_takes_over_on_short_retry() { + let db = ScratchDb::create("nip43_leader").await; + let state = crate::state::tests::test_state_with_database_pool(db.pool.clone()).await; + let community = insert_community(&db.pool).await; + + // Another "replica" holds the leader lock. + let key = scratch_lock_key(); + let mut holder = db.pool.acquire().await.expect("acquire lock holder"); + sqlx::query("SELECT pg_advisory_lock($1)") + .bind(key) + .execute(&mut *holder) + .await + .expect("take leader lock"); + + // A sweep interval far longer than the test, so only the standby + // retry can explain a takeover. + let retry = std::time::Duration::from_millis(100); + let reconciler = spawn_nip43_membership_reconciler(Arc::clone(&state), key, DAY, retry); + // Many standby retries' worth of time: none may sweep. + tokio::time::sleep(retry * 20).await; + let standby_snapshots = membership_snapshots(&db.pool, community).await; + + sqlx::query("SELECT pg_advisory_unlock($1)") + .bind(key) + .execute(&mut *holder) + .await + .expect("release leader lock"); + let published = + snapshot_within(&db.pool, community, std::time::Duration::from_secs(10)).await; + reconciler.abort(); + drop(holder); + drop(state); + db.drop_db().await; + + assert_eq!( + standby_snapshots, 0, + "a replica without the leader lock must not sweep" + ); + assert!( + published, + "the standby must take the lock on its short retry and sweep, not wait a full interval" + ); + } + /// Regression for #7558 on the worker path: repeated NIP-43 /// maintenance sweeps reconcile the active community once and never /// hand archived, quiescing, fenced, or tombstoned communities to the @@ -4445,12 +4548,9 @@ mod tests { let recorder = DebuggingRecorder::new(); let _guard = metrics::set_default_local_recorder(&recorder); for sweep in 0..3 { - reconcile_nip43_membership_snapshots_with_purpose( - &state, - Nip43ReconciliationPurpose::Maintenance, - ) - .await - .unwrap_or_else(|e| panic!("maintenance sweep {sweep}: {e}")); + reconcile_nip43_membership_snapshots(&state) + .await + .unwrap_or_else(|e| panic!("maintenance sweep {sweep}: {e}")); } assert_eq!( diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index a154f70757..40a3d059bc 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1095,63 +1095,21 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { } step.finish(); - // 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: repair event-backed rosters (`relay_members` is canonical) in + // the background, on the one replica holding the reconciler leader lock. + // It never blocks the listener; see `spawn_nip43_membership_reconciler`. if config.require_relay_membership { - let mut step = StepTimer::start(StartupStep::Nip43Reconcile); - match buzz_relay::handlers::side_effects::reconcile_nip43_membership_snapshots_with_purpose( - &state, - buzz_relay::handlers::side_effects::Nip43ReconciliationPurpose::Bootstrap, - ) - .await - { - Ok(summary) => { - if summary.failed > 0 { - step.degrade(); - } - info!( - count = summary.repaired, - failed = summary.failed, - "NIP-43 membership snapshots reconciled on startup" - ); - } - Err(error) => { - step.degrade(); - tracing::warn!(%error, "NIP-43 membership snapshot startup reconciliation failed") - } - } - step.finish(); - - let reconcile_state = Arc::clone(&state); let interval_secs = std::env::var("BUZZ_NIP43_RECONCILE_INTERVAL_SECS") .ok() .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.tick().await; - loop { - interval.tick().await; - match buzz_relay::handlers::side_effects::reconcile_nip43_membership_snapshots_with_purpose( - &reconcile_state, - buzz_relay::handlers::side_effects::Nip43ReconciliationPurpose::Maintenance, - ) - .await - { - Ok(summary) if summary.repaired > 0 => { - info!(count = summary.repaired, "NIP-43 membership snapshots repaired") - } - Ok(_) => {} - Err(error) => tracing::warn!( - %error, - "periodic NIP-43 membership snapshot reconciliation failed" - ), - } - } - }); + buzz_relay::handlers::side_effects::spawn_nip43_membership_reconciler( + Arc::clone(&state), + buzz_relay::handlers::side_effects::NIP43_RECONCILE_LOCK_KEY, + std::time::Duration::from_secs(interval_secs), + buzz_relay::handlers::side_effects::NIP43_STANDBY_RETRY, + ); } // Emit kind:39000/39002 discovery events for channels that exist in the DB @@ -2458,7 +2416,7 @@ fn dropped_in_memory_keys( async fn run_usage_metrics_tick( state: &AppState, emission_scope: &EmissionScope, - leader: &mut Option, + leader: &mut Option, emitted_in_memory: &mut HashSet, fleet_schedule: &mut FleetUsageSchedule, ) -> anyhow::Result<()> { @@ -2500,7 +2458,7 @@ async fn run_usage_metrics_tick( if leader.is_none() && !demoted { *leader = state .db - .try_lock_usage_metrics(USAGE_METRICS_LOCK_KEY) + .try_lock_session_leader(USAGE_METRICS_LOCK_KEY) .await?; if leader.is_some() { info!("Acquired usage metrics leader lock"); diff --git a/crates/buzz-relay/src/startup_steps.rs b/crates/buzz-relay/src/startup_steps.rs index 189731fd6b..0bc8308fd4 100644 --- a/crates/buzz-relay/src/startup_steps.rs +++ b/crates/buzz-relay/src/startup_steps.rs @@ -71,8 +71,6 @@ pub enum StartupStep { ChannelRosterFence, /// Repair large NIP-29 channel roster snapshots. LargeRosterReconcile, - /// Reconcile every community's NIP-43 membership snapshot. - Nip43Reconcile, /// Bind the private health listener. HealthBind, /// Bind the public relay listener. @@ -81,7 +79,7 @@ pub enum StartupStep { impl StartupStep { /// The complete `phase` label vocabulary. - pub const ALL: [Self; 19] = [ + pub const ALL: [Self; 18] = [ Self::DbConnect, Self::DbMigrate, Self::PartitionEnsure, @@ -98,7 +96,6 @@ impl StartupStep { Self::GitConformanceProbe, Self::ChannelRosterFence, Self::LargeRosterReconcile, - Self::Nip43Reconcile, Self::HealthBind, Self::ListenerBind, ]; @@ -122,7 +119,6 @@ impl StartupStep { Self::GitConformanceProbe => "git_conformance_probe", Self::ChannelRosterFence => "channel_roster_fence", Self::LargeRosterReconcile => "large_roster_reconcile", - Self::Nip43Reconcile => "nip43_reconcile", Self::HealthBind => "health_bind", Self::ListenerBind => "listener_bind", } @@ -318,9 +314,12 @@ mod tests { #[test] fn running_step_is_current_and_has_no_duration() { let mut held = None; - let gauges = gauges(|| held = Some(StepTimer::start(StartupStep::Nip43Reconcile))); - assert_eq!(get(&gauges, CURRENT_METRIC, "nip43_reconcile"), Some(1.0)); - assert_eq!(get(&gauges, SECONDS_METRIC, "nip43_reconcile"), None); + let gauges = gauges(|| held = Some(StepTimer::start(StartupStep::LargeRosterReconcile))); + assert_eq!( + get(&gauges, CURRENT_METRIC, "large_roster_reconcile"), + Some(1.0) + ); + assert_eq!(get(&gauges, SECONDS_METRIC, "large_roster_reconcile"), None); std::mem::forget(held); } diff --git a/deploy/charts/buzz/README.md b/deploy/charts/buzz/README.md index f2ef677c9e..073bbd0fe0 100644 --- a/deploy/charts/buzz/README.md +++ b/deploy/charts/buzz/README.md @@ -173,8 +173,8 @@ two gauges labelled `phase`: - `buzz_startup_phase_seconds` is the finished step's duration. `degraded` means the step hit a non-fatal error and startup continued — for -example, some communities failed NIP-43 reconciliation or a NIP-FI issuer did -not warm. `failed` means startup aborted inside the step. +example, some large channel rosters failed reconciliation or a NIP-FI issuer +did not warm. `failed` means startup aborted inside the step. These are gauges, not counters, because each value is written once per boot and a monotonic-counter scrape drops a series' first sample. Each is written @@ -183,8 +183,7 @@ field of the `Prometheus metrics exporter started` log (2700 s with default intervals). A step that runs longer than that timeout loses its `_current` series, so a pod stuck that long shows no current step. The `phase` values are the closed `StartupStep` vocabulary in `crates/buzz-relay/src/startup_steps.rs`; -they never reuse an early lifecycle phase name. The NIP-43 startup reconcile -also logs `NIP-43 startup reconciliation progress` every 5,000 communities. +they never reuse an early lifecycle phase name. ### Readiness contract