From e6c56ee5ee80eace3c9eddf0bf22a020fa8f48f3 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Tue, 20 Oct 2026 10:19:46 -0400 Subject: [PATCH] fix(relay): take NIP-43 snapshot reconcile off the startup path (#8467) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Before the relay opened its listener, it reconciled the NIP-43 membership snapshot for every active community. Each community costs two writer round trips, one after the other. On a production deployment with about 80k communities, this one step averaged 174s of a roughly 180s boot (max 324s). Pods in an availability zone farther from the writer took 3–4× longer than pods in the writer's zone. The step doesn't need to block startup. The snapshots live in the shared database, and the replicas that are already running serve the same snapshots. A new replica has nothing of its own to repair before it serves. The sweep before bind did the same work as the background maintenance sweep; only the metric labels differed. This PR: - Removes the NIP-43 sweep that ran before bind. - Moves the background loop into `side_effects::spawn_nip43_membership_reconciler`. Its first sweep now runs as soon as the task starts. Before, the loop skipped its first tick, so the first maintenance sweep waited a full `BUZZ_NIP43_RECONCILE_INTERVAL_SECS`. - Runs that loop on one replica only. The replica holding the `NIP43_RECONCILE_LOCK_KEY` session advisory lock sweeps; the others retry the lock every `NIP43_STANDBY_RETRY` (5s, capped at the interval). This reuses the usage-metrics leader lease (a detached Postgres session), renamed from `UsageMetricsLeader`/`try_lock_usage_metrics` to the generic `SessionLeader`/`try_lock_session_leader`. - Waits a full interval after each sweep (sleep after sweep), so a sweep longer than the interval never runs straight into the next one. - `reconcile_nip43_membership_snapshots` is now the only entry point and keeps its original signature (`-> Result`). It's no longer deprecated. ### What got simpler Deleted: - `Nip43ReconciliationPurpose` (the `Bootstrap`/`Maintenance` purposes) and `reconcile_nip43_membership_snapshots_with_purpose` - `Db::bootstrap_community_hosts` and `Db::nip43_membership_snapshot_needs_reconciliation_for_bootstrap` - The startup progress log (`NIP-43 startup reconciliation progress`, `nip43_progress_due`) and its test - The `nip43_reconcile` startup phase: `StartupStep::ALL` goes from 19 to 18 steps The leader lock reuses the existing usage-metrics lease rather than adding a second leader primitive. `StorageAccountingLeader` is still a near-duplicate of it; folding that in is left as identified debt. ### Trade-offs - **Repair before serving.** A newly started replica no longer guarantees it has repaired stale snapshots before serving. The running fleet doesn't guarantee that today either: the background sweep is what repairs drift, and the leader starts it as soon as it takes the lock. - **Rollout load.** Before, every replica swept before binding, and every replica also ran the background loop. Now there is one sweeper in steady state, so writer load from this job drops from one sweep per replica to one per deployment, and rollouts no longer stack boot sweeps. Standbys add one lock attempt every 5s each. - **Leader handoff.** When the leader's process exits (including SIGKILL), its connection closes, Postgres releases the session lock, and a standby retries the lock at least every `NIP43_STANDBY_RETRY` (5s) and sweeps as soon as it gets it. Each attempt also waits for a writer connection, so under database pressure takeover can take longer than 5s. Before this, a standby waited a full interval (3600s in production), so after a rollout replaced the leader nothing swept for up to an hour. If the leader is lost with its host or network instead, the lock is held until the database server's TCP keepalive drops the session, and nobody sweeps meanwhile. The usage-metrics leader already has the same exposure. 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 that overlap safe. - **Snapshot window.** A missing or stale kind:13534 snapshot is served until the leader's sweep reaches that community; before, a booting replica blocked until it was repaired. Admission reads canonical `relay_members`, so this window doesn't change who's admitted. It does affect clients that read the snapshot: desktop onboarding shows the membership-denied page to a member missing from it, the desktop/mobile member lists and join alerts lag, and desktop mesh discovery keeps its existing allowlist when the snapshot is missing. In production the inline publications that keep it current have not failed since 2026-10-07, and the sweep repaired no snapshots in the last 30 days. - **Dashboards and alerts.** The `buzz_startup_phase_seconds{phase="nip43_reconcile"}` series stops reporting. Anything charting or alerting on it should move to the sweep's existing counters (`buzz_nip43_membership_reconciliations_total`, `buzz_nip43_membership_reconciliation_failures_total`). ### Related issue None found. ### Merge order #8468 is stacked on this PR. Both remove a startup phase, and with the old fixed-length `StartupStep::ALL` array, merging both would have built a 17-entry `[Self; 18]` that doesn't compile. Merge this PR first; #8468 makes `ALL` a slice. ### Testing - `background_reconciler_sweeps_immediately` (private scratch database): spawns the production reconciler with a 24h interval and 24h standby retry, and requires the first sweep to publish a missing snapshot. Fails if the skipped first tick is restored. - `background_reconciler_standby_takes_over_on_short_retry` (private scratch database): with a 24h interval and 100ms standby retry, nothing publishes while another session holds the leader lock; once it's released the standby must publish within 10s. Fails if the standby sleeps the full interval again (checked by reverting that line). - `test_usage_metrics_lock_has_single_owner_and_releases_on_drop` covers the renamed lease. - `maintenance_sweep_skips_archived_and_deleted_communities` now calls the single entry point. - Runtime E2E at `574b28b3`: two real relays sharing one database, 3600s sweep interval, production 5s standby retry. While the leader was alive, the standby made only lock attempts (no sweep or snapshot-comparison reads). SIGKILL of the actual lock holder, 3/3 trials: the survivor took the lock in 4.81–4.86s and repaired the snapshot within 10ms after that. Both replicas served authenticated reads while the initial publication was stalled. - Blox, at `574b28b3`: `just fmt-check clippy security-review-check file-size-check` passes. Every `test-unit` step and `just test` (118 suites) passes except the `buzz-acp` `git_bootstrap` native-git tests, which fail the same way at base `b6aa5539` on Blox (environment, not this change). --------- Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/lib.rs | 2 +- crates/buzz-db/src/store/relay_members.rs | 22 +- crates/buzz-db/src/store/usage.rs | 39 +- crates/buzz-db/tests/observability_source.rs | 2 +- .../buzz-relay/src/handlers/side_effects.rs | 362 +++++++++++------- crates/buzz-relay/src/main.rs | 64 +--- crates/buzz-relay/src/startup_steps.rs | 15 +- deploy/charts/buzz/README.md | 7 +- 8 files changed, 269 insertions(+), 244 deletions(-) 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