From 7f6ffd5ee5ae1b66d660245acf465b88e17639bc Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Mon, 5 Oct 2026 13:25:21 -0400 Subject: [PATCH] fix(db): enumerate only active communities for maintenance (#8061) Fixes #7558. Supersedes #7630. The SQL predicate here comes from #7630 by @olympusbuildz, who is credited as co-author on the commit. This PR keeps that fix unchanged and adds the lifecycle fixtures and worker-path coverage the fix needed. ## Problem `community_hosts_with_operation` ran `SELECT id, host FROM communities` with no filter. After a whole-community deletion, the `communities` row stays behind as a `tombstone` until retention expires, which is by design. But the NIP-43 maintenance sweep kept handing that community to `publish_nip43_membership_list`. The community write fence rejected the write, so the relay logged `NIP-43 membership reconciliation failed` about every 60s, indefinitely, for every deleted community. ## Change The helper is split by caller purpose: - `active_community_hosts` (maintenance) and `bootstrap_community_hosts` filter to `archived_at IS NULL AND deleted_at IS NULL AND deletion_state = 'active'`. That's the same active-community predicate already used in `community.rs` and `deletion.rs`. - `usage_community_hosts` still enumerates every row, exactly as on main. The startup large-roster reconciliation query (`list_large_channel_rosters_needing_reconciliation`) joined `communities` without any lifecycle filter, so it gets the same active predicate on its `JOIN communities`. Tombstone rows aren't modified. ### Caller impact | Caller | Enumeration | Effect | |---|---|---| | NIP-43 maintenance sweep (`side_effects.rs`) | active | Stops trying fenced writes and logging warnings forever. This is the bug fix. | | NIP-43 bootstrap reconciliation | active | Same as maintenance, at startup. | | Push-wake worker (`push_runtime.rs`) | active | Stops claiming wakes for these communities. Archived communities are unreachable through host resolution; quiescing, fenced and tombstone writes are DB-fenced. Wakes already queued for an archived community aren't delivered late after unarchive, because claim and send both re-check `expires_at`. | | Startup large-roster reconciliation (`side_effects.rs`, kind 39002) | active | Stops rewriting roster snapshots for archived communities and attempting fenced writes for deleting or tombstoned ones. | | Usage-metrics host map (`main.rs`) | all rows | Unchanged. Per-community labels, fleet totals and storage attribution keep covering the same rows as the unfiltered usage count queries. | ## Tests #7630's test tombstoned a row with a plain `UPDATE`. On current main, the `enforce_community_tombstone` trigger from migration 0029 rejects that unless the transaction carries the deletion-executor GUCs. The fixtures here set those GUCs the way the executor does. - `buzz-db` `store::usage::postgres_tests::test_community_hosts_excludes_archived_and_deleted_communities` sets up active, archived, quiescing, fenced and tombstoned fixtures. Across two sweeps it checks that `active_community_hosts` (maintenance) and `bootstrap_community_hosts` both return only the active community, and that `usage_community_hosts` still returns all five. It also checks that the tombstone row still exists with `deleted_at` set. - `buzz-relay` `handlers::side_effects::tests::postgres_tests::maintenance_sweep_skips_archived_and_deleted_communities` is the worker-path regression. It runs `reconcile_nip43_membership_snapshots_with_purpose(Maintenance)` three times against the same five fixtures. It checks that `buzz_nip43_membership_reconciliation_failures_total` stays at 0, the active community gets exactly one membership snapshot, the four non-active communities get no maintenance write, and the tombstone row is still there. - `buzz-db` `store::channel_members::postgres_tests::large_roster_reconciliation_skips_archived_and_deleted_communities` seeds a channel whose roster has diverged from its live relay-authored kind-39002 snapshot in each of the five fixture communities. Only the active community's channel may be returned. All three tests are `#[ignore = "requires Postgres"]` inside `postgres_tests::`, so the PostgreSQL CI lane picks them up through its existing structural filter. ## What got simpler The background writers share one definition of "community eligible for maintenance": the same active predicate the rest of the store uses. Usage metrics keep their existing semantics, so this PR changes no metric. One enumeration helper is added (`active_community_hosts`), and the old `community_hosts_with_operation` indirection is gone. The executor-GUC lifecycle fixture lives once in `test_support` and is shared by both DB tests instead of being duplicated. --------- Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Olympusbuildz --- crates/buzz-db/src/store/channel_members.rs | 103 ++++++++++++- crates/buzz-db/src/store/usage.rs | 139 ++++++++++++++++-- crates/buzz-db/src/test_support.rs | 26 ++++ .../buzz-relay/src/handlers/side_effects.rs | 134 ++++++++++++++++- crates/buzz-relay/src/push_runtime.rs | 2 +- 5 files changed, 389 insertions(+), 15 deletions(-) diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 7312bf3c01..0dbea01dd4 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -1098,6 +1098,10 @@ pub struct LargeChannelRoster { /// Returns active channels whose canonical roster exceeds `minimum_members`. /// +/// Only active communities are scanned: archived communities must not have +/// their discovery state rewritten, and deleting or tombstoned communities +/// would only produce fenced-write failures. +/// /// This is an internal cross-community maintenance read. Callers must preserve /// the returned community id when reading or rewriting discovery state. pub async fn list_large_channel_rosters_needing_reconciliation( @@ -1125,7 +1129,11 @@ pub async fn list_large_channel_rosters_needing_reconciliation( ) SELECT lr.community_id, community.host, lr.channel_id, lr.member_count FROM large_rosters lr - JOIN communities community ON community.id = lr.community_id + JOIN communities community + ON community.id = lr.community_id + AND community.archived_at IS NULL + AND community.deleted_at IS NULL + AND community.deletion_state = 'active' JOIN LATERAL ( SELECT roster.tags FROM events roster @@ -2334,6 +2342,99 @@ mod postgres_tests { assert!(converged.is_empty()); } + /// Seed a channel whose roster (owner + one member) has diverged from its + /// live relay-authored kind-39002 snapshot (owner only). + async fn seed_divergent_roster(pool: &PgPool, community_id: Uuid, relay_pubkey: &[u8]) -> Uuid { + let creator = random_pubkey(); + let channel = create_test_channel( + pool, + community_id, + "divergent-roster", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &creator, + None, + ) + .await + .expect("create test channel"); + let tags = serde_json::json!([ + ["d", channel.id.to_string()], + ["p", hex::encode(&creator), "", "owner"] + ]); + sqlx::query( + r#" + INSERT INTO events + (community_id, id, pubkey, created_at, kind, tags, content, sig, channel_id, d_tag) + VALUES ($1, $2, $3, NOW(), 39002, $4, '', $5, $6, $7) + "#, + ) + .bind(community_id) + .bind(random_pubkey()) + .bind(relay_pubkey) + .bind(tags) + .bind(vec![0u8; 64]) + .bind(channel.id) + .bind(channel.id.to_string()) + .execute(pool) + .await + .expect("insert roster snapshot"); + sqlx::query( + "INSERT INTO channel_members (community_id, channel_id, pubkey, role, joined_at) \ + VALUES ($1, $2, $3, 'member', NOW())", + ) + .bind(community_id) + .bind(channel.id) + .bind(random_pubkey()) + .execute(pool) + .await + .expect("diverge roster"); + channel.id + } + + /// Startup roster reconciliation must not rewrite discovery state for + /// archived communities, nor attempt fenced writes for deleting or + /// tombstoned ones. Same active-community predicate as #7558. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn large_roster_reconciliation_skips_archived_and_deleted_communities() { + let pool = setup_pool().await; + let relay_pubkey = random_pubkey(); + let mut fixtures = Vec::new(); + for label in ["active", "archived", "quiescing", "fenced", "tombstone"] { + let community_id = make_test_community(&pool).await; + let channel_id = seed_divergent_roster(&pool, community_id, &relay_pubkey).await; + fixtures.push((label, community_id, channel_id)); + } + for (label, community_id, _) in &fixtures { + match *label { + "active" => {} + "archived" => { + sqlx::query("UPDATE communities SET archived_at = now() WHERE id = $1") + .bind(community_id) + .execute(&pool) + .await + .expect("archive fixture"); + } + state => crate::test_support::set_deletion_state(&pool, *community_id, state).await, + } + } + + let candidates = list_large_channel_rosters_needing_reconciliation(&pool, 1, &relay_pubkey) + .await + .expect("list candidates"); + let found: Vec<(CommunityId, Uuid)> = candidates + .iter() + .map(|c| (c.community_id, c.channel_id)) + .collect(); + let (_, active_community, active_channel) = fixtures[0]; + assert_eq!( + found, + vec![(CommunityId::from_uuid(active_community), active_channel)], + "only the active community's divergent roster may be reconciled" + ); + } + /// A random non-admin, non-owner user cannot remove someone else's bot. #[tokio::test] #[ignore = "requires Postgres"] diff --git a/crates/buzz-db/src/store/usage.rs b/crates/buzz-db/src/store/usage.rs index 9d557c8b18..8b8664f484 100644 --- a/crates/buzz-db/src/store/usage.rs +++ b/crates/buzz-db/src/store/usage.rs @@ -386,23 +386,46 @@ pub struct CommunityHost { pub host: String, } -/// Fetch all community id → host mappings in one query. +/// Fetch every community id → host mapping in one query, including archived, +/// deleting, and tombstoned communities. +/// +/// Usage metrics label per-community series and attribute storage with this +/// map, so it must cover the same rows as the unfiltered usage count queries. pub async fn community_hosts(pool: &PgPool) -> Result> { - community_hosts_with_operation(pool, observability::WriterOperation::Maintenance).await + let mut connection = + observability::acquire_writer(pool, observability::WriterOperation::Maintenance).await?; + let rows = sqlx::query_as::<_, (Uuid, String)>("SELECT id, host FROM communities") + .fetch_all(&mut *connection) + .await?; + Ok(into_community_hosts(rows)) } -async fn community_hosts_with_operation( +/// Fetch the id → host mapping for every active community in one query. +/// +/// Archived communities are unreachable through host resolution, and +/// communities anywhere in the deletion lifecycle (`quiescing`, `fenced`, +/// `tombstone`) have their writes DB-fenced, so background workers that write +/// per community must skip them. Tombstone rows stay in place for retention. +async fn active_community_hosts( pool: &PgPool, operation: observability::WriterOperation, ) -> Result> { let mut connection = observability::acquire_writer(pool, operation).await?; - let rows = sqlx::query_as::<_, (Uuid, String)>("SELECT id, host FROM communities") - .fetch_all(&mut *connection) - .await?; - Ok(rows - .into_iter() + let rows = sqlx::query_as::<_, (Uuid, String)>( + "SELECT id, host FROM communities \ + WHERE archived_at IS NULL \ + AND deleted_at IS NULL \ + AND deletion_state = 'active'", + ) + .fetch_all(&mut *connection) + .await?; + Ok(into_community_hosts(rows)) +} + +fn into_community_hosts(rows: Vec<(Uuid, String)>) -> Vec { + rows.into_iter() .map(|(id, host)| CommunityHost { id, host }) - .collect()) + .collect() } impl Db { @@ -497,16 +520,22 @@ impl Db { active_channel_counts(&self.pool, interval_sql).await } - /// Return all community id → host mappings. + /// Return all community id → host mappings, for usage-metrics labels. #[datastore_span(name = "usage_community_hosts", system = "postgresql")] pub async fn usage_community_hosts(&self) -> Result> { community_hosts(&self.pool).await } - /// Return community host mappings during startup bootstrap work. + /// Return active community host mappings for background maintenance writers. + #[datastore_span(name = "active_community_hosts", system = "postgresql")] + 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> { - community_hosts_with_operation(&self.pool, observability::WriterOperation::Bootstrap).await + active_community_hosts(&self.pool, observability::WriterOperation::Bootstrap).await } } @@ -805,6 +834,92 @@ mod postgres_tests { assert_eq!(found.unwrap().host, host); } + /// Regression for #7558: maintenance and bootstrap enumeration return only + /// active communities. Archived, logically deleted (quiescing/fenced), and + /// tombstoned rows are skipped on every sweep, and the tombstone row itself + /// is left intact for retention. The usage-metrics map stays unfiltered so + /// its per-community labels and storage attribution keep covering every + /// row the unfiltered usage count queries return. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn test_community_hosts_excludes_archived_and_deleted_communities() { + let pool = get_pool().await; + let (active, _, active_host) = make_community(&pool).await; + let (archived, _, _) = make_community(&pool).await; + let (quiescing, _, _) = make_community(&pool).await; + let (fenced, _, _) = make_community(&pool).await; + let (tombstone, _, _) = make_community(&pool).await; + + sqlx::query("UPDATE communities SET archived_at = now() WHERE id = $1") + .bind(archived) + .execute(&pool) + .await + .expect("archive fixture"); + crate::test_support::set_deletion_state(&pool, quiescing, "quiescing").await; + crate::test_support::set_deletion_state(&pool, fenced, "fenced").await; + crate::test_support::set_deletion_state(&pool, tombstone, "tombstone").await; + + 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), + ] { + 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!( + hosts + .iter() + .find(|h| h.id == active) + .map(|h| h.host.as_str()), + Some(active_host.as_str()), + "{caller} sweep {sweep}: active community must be returned" + ); + for (label, id) in [ + ("archived", archived), + ("quiescing", quiescing), + ("fenced", fenced), + ("tombstone", tombstone), + ] { + assert!( + !ids.contains(&id), + "{caller} sweep {sweep}: {label} community must be excluded" + ); + } + } + } + + let metrics_ids: std::collections::HashSet = db + .usage_community_hosts() + .await + .expect("usage-metrics hosts") + .iter() + .map(|h| h.id) + .collect(); + for (label, id) in [ + ("active", active), + ("archived", archived), + ("quiescing", quiescing), + ("fenced", fenced), + ("tombstone", tombstone), + ] { + assert!( + metrics_ids.contains(&id), + "usage-metrics map must still include the {label} community" + ); + } + + let (state, deleted): (String, bool) = sqlx::query_as( + "SELECT deletion_state, deleted_at IS NOT NULL FROM communities WHERE id = $1", + ) + .bind(tombstone) + .fetch_one(&pool) + .await + .expect("tombstone row is retained"); + assert_eq!(state, "tombstone"); + assert!(deleted, "tombstone keeps its deleted_at"); + } + /// community_count reflects newly inserted communities. #[tokio::test] #[ignore = "requires Postgres"] diff --git a/crates/buzz-db/src/test_support.rs b/crates/buzz-db/src/test_support.rs index 7699313d63..1cf722998d 100644 --- a/crates/buzz-db/src/test_support.rs +++ b/crates/buzz-db/src/test_support.rs @@ -7,3 +7,29 @@ pub(crate) fn database_url() -> String { .or_else(|_| std::env::var("DATABASE_URL")) .unwrap_or_else(|_| DEFAULT_DATABASE_URL.to_owned()) } + +/// Move a community fixture into a deletion lifecycle state the way the +/// executor does: the tombstone trigger only accepts lifecycle changes from a +/// transaction carrying the executor GUCs for that community. +pub(crate) async fn set_deletion_state(pool: &sqlx::PgPool, id: uuid::Uuid, state: &str) { + let mut tx = pool.begin().await.expect("begin lifecycle fixture"); + sqlx::query( + "SELECT set_config('buzz.deletion_executor_community', $1, true), \ + set_config('buzz.deletion_fence_generation', '0', true)", + ) + .bind(id.to_string()) + .execute(&mut *tx) + .await + .expect("authorize lifecycle fixture"); + sqlx::query( + "UPDATE communities SET deletion_state = $2, \ + deleted_at = CASE WHEN $2 = 'tombstone' THEN now() END \ + WHERE id = $1", + ) + .bind(id) + .bind(state) + .execute(&mut *tx) + .await + .expect("set lifecycle state"); + tx.commit().await.expect("commit lifecycle fixture"); +} diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index a2e260c9fc..37d6a22627 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -3129,7 +3129,7 @@ pub async fn reconcile_nip43_membership_snapshots_with_purpose( ) -> anyhow::Result { let communities = match purpose { Nip43ReconciliationPurpose::Bootstrap => state.db.bootstrap_community_hosts().await?, - Nip43ReconciliationPurpose::Maintenance => state.db.usage_community_hosts().await?, + Nip43ReconciliationPurpose::Maintenance => state.db.active_community_hosts().await?, }; let mut reconciled = 0usize; @@ -3960,4 +3960,136 @@ mod tests { assert!(actor_is_channel_owner_or_admin(&members, &actor)); } + + mod postgres_tests { + use super::*; + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + + async fn insert_community(pool: &sqlx::PgPool) -> Uuid { + let id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(format!("nip43-sweep-{}.example", id.simple())) + .execute(pool) + .await + .expect("insert community fixture"); + id + } + + async fn set_deletion_state(pool: &sqlx::PgPool, id: Uuid, state: &str) { + let mut tx = pool.begin().await.expect("begin lifecycle fixture"); + sqlx::query( + "SELECT set_config('buzz.deletion_executor_community', $1, true), \ + set_config('buzz.deletion_fence_generation', '0', true)", + ) + .bind(id.to_string()) + .execute(&mut *tx) + .await + .expect("authorize lifecycle fixture"); + sqlx::query( + "UPDATE communities SET deletion_state = $2, \ + deleted_at = CASE WHEN $2 = 'tombstone' THEN now() END \ + WHERE id = $1", + ) + .bind(id) + .bind(state) + .execute(&mut *tx) + .await + .expect("set lifecycle state"); + tx.commit().await.expect("commit lifecycle fixture"); + } + + async fn membership_snapshots(pool: &sqlx::PgPool, id: Uuid) -> i64 { + sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id = $1 AND kind = $2") + .bind(id) + .bind(KIND_NIP43_MEMBERSHIP_LIST as i32) + .fetch_one(pool) + .await + .expect("count membership snapshots") + } + + 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() + } + + /// 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 + /// publisher, so the fence never rejects a write and nothing warns. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn maintenance_sweep_skips_archived_and_deleted_communities() { + 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 active = insert_community(&pool).await; + let archived = insert_community(&pool).await; + let quiescing = insert_community(&pool).await; + let fenced = insert_community(&pool).await; + let tombstone = insert_community(&pool).await; + sqlx::query("UPDATE communities SET archived_at = now() WHERE id = $1") + .bind(archived) + .execute(&pool) + .await + .expect("archive fixture"); + set_deletion_state(&pool, quiescing, "quiescing").await; + set_deletion_state(&pool, fenced, "fenced").await; + set_deletion_state(&pool, tombstone, "tombstone").await; + + 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}")); + } + + assert_eq!( + reconciliation_failures(&recorder), + 0, + "no community may fail reconciliation on any sweep" + ); + assert_eq!( + membership_snapshots(&pool, active).await, + 1, + "the active community is reconciled once and then left alone" + ); + for (label, id) in [ + ("archived", archived), + ("quiescing", quiescing), + ("fenced", fenced), + ("tombstone", tombstone), + ] { + assert_eq!( + membership_snapshots(&pool, id).await, + 0, + "{label} community must not receive a maintenance write" + ); + } + let tombstone_state: String = + sqlx::query_scalar("SELECT deletion_state FROM communities WHERE id = $1") + .bind(tombstone) + .fetch_one(&pool) + .await + .expect("tombstone row is retained"); + assert_eq!(tombstone_state, "tombstone"); + } + } } diff --git a/crates/buzz-relay/src/push_runtime.rs b/crates/buzz-relay/src/push_runtime.rs index d47dcc9d61..e25e7c0d17 100644 --- a/crates/buzz-relay/src/push_runtime.rs +++ b/crates/buzz-relay/src/push_runtime.rs @@ -350,7 +350,7 @@ pub async fn run_delivery_worker(state: Arc) { let mut idle_delay = Duration::from_millis(500); loop { let mut found = false; - match state.db.usage_community_hosts().await { + match state.db.active_community_hosts().await { Ok(communities) => { for community in communities { let community = buzz_core::CommunityId::from_uuid(community.id);