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 <Olympus.roots@outlook.com>

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"]

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<Vec<CommunityHost>> {
-    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<Vec<CommunityHost>> {
     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<CommunityHost> {
+    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<Vec<CommunityHost>> {
         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<Vec<CommunityHost>> {
+        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<Vec<CommunityHost>> {
-        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<Uuid> = 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<Uuid> = 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"]

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");
+}

crates/buzz-relay/src/handlers/side_effects.rs

@@ -3129,7 +3129,7 @@ pub async fn reconcile_nip43_membership_snapshots_with_purpose(
 ) -> anyhow::Result<usize> {
     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");
+        }
+    }
 }

crates/buzz-relay/src/push_runtime.rs

@@ -350,7 +350,7 @@ pub async fn run_delivery_worker(state: Arc<AppState>) {
     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);