Fence fleet-wide enumerators against non-active tenant lifecycle states

Three fleet-wide background scans filtered candidates only by
communities.archived_at, ignoring the deletion lifecycle
(deletion_state / deleted_at) introduced with the community write
fence:

- list_all_enabled_workflows (cron scheduler input)
- query_due_reminders (NIP-ER reminder sweep)
- list_large_channel_rosters_needing_reconciliation (roster repair)

Each downstream write is already blocked by the universal write-fence
trigger, but an unfenced scan re-surfaces a quiescing tenant's rows on
every tick, turning the trigger rejection into a recurring background
error and wasted work. Gate all three scans on
archived_at IS NULL AND deletion_state = 'active' AND deleted_at IS NULL,
matching the community_write_allowed predicate used by the ephemeral
channel reaper and push-match queue.

Adds quiescing-tenant/active-bystander regression tests for each scan,
mirroring the existing push-matcher fence tests.

Signed-off-by: Logan Johnson <loganj@squareup.com>
Co-authored-by: Wakko <wakko@buzz.block.builderlab.xyz>

crates/buzz-db/src/channel.rs

@@ -1095,6 +1095,12 @@ pub struct LargeChannelRoster {
 ///
 /// This is an internal cross-community maintenance read. Callers must preserve
 /// the returned community id when reading or rewriting discovery state.
+///
+/// Lifecycle fencing: the scan feeds fleet-wide snapshot rewrites, so it must
+/// skip tenants that are archived or in any non-`active` deletion state. The
+/// rewrite itself is independently blocked by the universal write-fence
+/// trigger on `events`; this filter keeps a fenced tenant from surfacing as a
+/// candidate (and failing) on every reconciliation pass.
 pub async fn list_large_channel_rosters_needing_reconciliation(
     pool: &PgPool,
     minimum_members: i64,
@@ -1115,7 +1121,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.deletion_state = 'active'
+         AND community.deleted_at IS NULL
         JOIN LATERAL (
             SELECT roster.tags
             FROM events roster
@@ -2574,6 +2584,140 @@ mod tests {
         assert!(converged.is_empty());
     }
 
+    /// Lifecycle fence on the roster-reconciliation scan: a quiescing tenant's
+    /// stale snapshot must not surface as a candidate, while an active
+    /// bystander's identical stale snapshot still does. The rewrite is
+    /// independently blocked by the write-fence trigger on `events`; this
+    /// filter keeps the fenced tenant from failing reconciliation every pass.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn roster_reconciliation_scan_skips_quiescing_tenant_and_keeps_active_bystander() {
+        let pool = setup_pool().await;
+        let relay_pubkey = random_pubkey();
+        let creator = random_pubkey();
+
+        // Build one community with a channel whose live 39002 snapshot was
+        // canonical (creator-owner + 2 members), then gains a 3rd member — a
+        // reconciliation candidate at minimum_members = 2. The snapshot is
+        // inserted while it exactly matches canonical membership (4-field p
+        // tags, roles included, ordered by pubkey bytes) so the roster INSERT
+        // guard admits it; membership then grows, making the live snapshot
+        // stale without touching `events`.
+        let creator_hex = creator
+            .iter()
+            .map(|b| format!("{b:02x}"))
+            .collect::<String>();
+        let mut fixtures = Vec::new();
+        for _ in 0..2 {
+            let community_id = make_test_community(&pool).await;
+            let channel = create_test_channel(
+                &pool,
+                community_id,
+                "fence-roster",
+                ChannelType::Stream,
+                ChannelVisibility::Open,
+                None,
+                &creator,
+                None,
+            )
+            .await
+            .expect("create channel");
+            sqlx::query(
+                r#"
+                INSERT INTO channel_members (community_id, channel_id, pubkey, role, joined_at)
+                SELECT $1, $2, decode(lpad(to_hex(n), 64, '0'), 'hex'), 'member',
+                       NOW() + (n || ' seconds')::interval
+                FROM generate_series(1, 2) n
+                "#,
+            )
+            .bind(community_id)
+            .bind(channel.id)
+            .execute(&pool)
+            .await
+            .expect("insert members");
+            // Canonical membership = creator (owner, from create_test_channel)
+            // plus the two synthetic members. The guard orders p tags by raw
+            // pubkey bytes, which matches lexicographic order of the hex.
+            let mut roster: Vec<(String, &str)> = vec![
+                (creator_hex.clone(), "owner"),
+                (format!("{:064x}", 1), "member"),
+                (format!("{:064x}", 2), "member"),
+            ];
+            roster.sort_by(|a, b| a.0.cmp(&b.0));
+            let canonical_tags: Vec<serde_json::Value> =
+                std::iter::once(serde_json::json!(["d", channel.id.to_string()]))
+                    .chain(
+                        roster
+                            .iter()
+                            .map(|(pk, role)| serde_json::json!(["p", pk, "", role])),
+                    )
+                    .collect();
+            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(serde_json::Value::Array(canonical_tags))
+            .bind(vec![0u8; 64])
+            .bind(channel.id)
+            .bind(channel.id.to_string())
+            .execute(&pool)
+            .await
+            .expect("insert canonical snapshot");
+            // Grow membership past the snapshot: the live 39002 head is now
+            // stale relative to canonical membership.
+            sqlx::query(
+                r#"
+                INSERT INTO channel_members (community_id, channel_id, pubkey, role, joined_at)
+                VALUES ($1, $2, decode(lpad(to_hex(3), 64, '0'), 'hex'), 'member', NOW())
+                "#,
+            )
+            .bind(community_id)
+            .bind(channel.id)
+            .execute(&pool)
+            .await
+            .expect("insert late-joining member");
+            fixtures.push((CommunityId::from_uuid(community_id), channel.id));
+        }
+        let (active, active_channel) = fixtures[0];
+        let (target, _) = fixtures[1];
+
+        let mut lifecycle = 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(target.to_string())
+        .execute(&mut *lifecycle)
+        .await
+        .expect("authorize target lifecycle");
+        sqlx::query("UPDATE communities SET deletion_state = 'quiescing' WHERE id = $1")
+            .bind(target.as_uuid())
+            .execute(&mut *lifecycle)
+            .await
+            .expect("quiesce target");
+        lifecycle.commit().await.expect("commit target lifecycle");
+
+        let candidates = list_large_channel_rosters_needing_reconciliation(&pool, 2, &relay_pubkey)
+            .await
+            .expect("scan candidates");
+        assert!(
+            candidates
+                .iter()
+                .any(|c| c.community_id == active && c.channel_id == active_channel),
+            "active bystander's stale snapshot must still surface"
+        );
+        assert!(
+            !candidates.iter().any(|c| c.community_id == target),
+            "quiescing tenant must not surface as a reconciliation candidate"
+        );
+    }
+
     /// A random non-admin, non-owner user cannot remove someone else's bot.
     #[tokio::test]
     #[ignore = "requires Postgres"]

crates/buzz-db/src/event.rs

@@ -1460,6 +1460,8 @@ pub async fn query_due_reminders(
           AND e.deleted_at IS NULL
           AND e.delivered_at IS NULL
           AND c.archived_at IS NULL
+          AND c.deletion_state = 'active'
+          AND c.deleted_at IS NULL
         ORDER BY e.community_id, e.pubkey, e.d_tag, e.created_at DESC, e.id ASC
         LIMIT $3
         "#,
@@ -2539,6 +2541,67 @@ mod tests {
         }));
     }
 
+    /// Lifecycle fence on the reminder sweep: a quiescing tenant's due
+    /// reminder must not surface, while an active bystander's identical
+    /// reminder still does. The claim UPDATE is independently blocked by the
+    /// universal write-fence trigger on `events`; this filter keeps the fenced
+    /// tenant's row from surfacing (and erroring) on every sweep tick.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn due_reminder_sweep_skips_quiescing_tenant_and_keeps_active_bystander() {
+        let pool = setup_pool().await;
+        let active = CommunityId::from_uuid(make_test_community(&pool).await);
+        let target = CommunityId::from_uuid(make_test_community(&pool).await);
+
+        let not_before = Utc::now().timestamp() - 1;
+        let make_reminder = |scope: &str| {
+            EventBuilder::new(Kind::Custom(KIND_EVENT_REMINDER as u16), "r")
+                .tags([
+                    Tag::parse(["d", scope]).unwrap(),
+                    Tag::parse(["not_before", &not_before.to_string()]).unwrap(),
+                ])
+                .sign_with_keys(&Keys::generate())
+                .expect("sign reminder")
+        };
+        let active_reminder = make_reminder("fence-active");
+        let target_reminder = make_reminder("fence-target");
+        insert_event(&pool, active, &active_reminder, None)
+            .await
+            .expect("insert active reminder");
+        insert_event(&pool, target, &target_reminder, None)
+            .await
+            .expect("insert target reminder");
+
+        let mut lifecycle = 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(target.to_string())
+        .execute(&mut *lifecycle)
+        .await
+        .expect("authorize target lifecycle");
+        sqlx::query("UPDATE communities SET deletion_state = 'quiescing' WHERE id = $1")
+            .bind(target.as_uuid())
+            .execute(&mut *lifecycle)
+            .await
+            .expect("quiesce target");
+        lifecycle.commit().await.expect("commit target lifecycle");
+
+        let due = query_due_reminders(&pool, Utc::now().timestamp(), 100)
+            .await
+            .expect("query due reminders");
+        assert!(
+            due.iter()
+                .any(|row| row.community_id == active && row.id == active_reminder.id.as_bytes()),
+            "active bystander's due reminder must still surface"
+        );
+        assert!(
+            !due.iter().any(|row| row.community_id == target),
+            "quiescing tenant's reminder must not surface in the sweep"
+        );
+    }
+
     /// Two pods race to claim the same due reminder: exactly one wins. The
     /// scheduler publishes only on a winning claim (`Ok(true)`) and `continue`s
     /// on the loser (`Ok(false)`), so a single winning claim *is* the proof of

crates/buzz-db/src/workflow.rs

@@ -545,6 +545,13 @@ pub async fn list_enabled_channel_workflows(
 /// Used by the cron scheduler. Filters by trigger type in SQL to avoid loading
 /// event-triggered workflows that the cron loop would immediately discard.
 /// Results are bounded to [`LIST_MAX_LIMIT`] rows.
+///
+/// Lifecycle fencing: this is a fleet-wide scan, so it must skip tenants that
+/// are archived or in any non-`active` deletion state. The downstream claim
+/// INSERT is independently blocked by the universal write-fence trigger on
+/// `scheduled_workflow_fires`, but without this filter a quiescing tenant's
+/// workflows would surface every tick and turn the trigger rejection into a
+/// recurring scheduler error.
 pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result<Vec<WorkflowRecord>> {
     let rows = sqlx::query(
         r#"
@@ -556,6 +563,8 @@ pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result<Vec<WorkflowRec
           AND w.enabled = TRUE
           AND w.definition->'trigger'->>'on' = 'schedule'
           AND c.archived_at IS NULL
+          AND c.deletion_state = 'active'
+          AND c.deleted_at IS NULL
         ORDER BY w.created_at ASC
         LIMIT $1
         "#,
@@ -2133,6 +2142,60 @@ mod tests {
         (workflow_id, community)
     }
 
+    /// Move a community into `quiescing` using the executor-authorized
+    /// lifecycle path, mirroring the push-matcher fence fixtures.
+    async fn quiesce_test_community(pool: &PgPool, community: CommunityId) {
+        let mut lifecycle = 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(community.to_string())
+        .execute(&mut *lifecycle)
+        .await
+        .expect("authorize target lifecycle");
+        sqlx::query("UPDATE communities SET deletion_state = 'quiescing' WHERE id = $1")
+            .bind(community.as_uuid())
+            .execute(&mut *lifecycle)
+            .await
+            .expect("quiesce target");
+        lifecycle.commit().await.expect("commit target lifecycle");
+    }
+
+    /// Lifecycle fence on the cron scheduler's fleet-wide scan: a quiescing
+    /// tenant's enabled schedule workflow must not surface, while an active
+    /// bystander's identical workflow still does. Without the
+    /// `deletion_state = 'active'` filter the quiescing tenant's workflow
+    /// reaches the claim INSERT every tick, where the write-fence trigger
+    /// turns it into a recurring scheduler error.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn scheduler_scan_skips_quiescing_tenant_and_keeps_active_bystander() {
+        let pool = setup_pool().await;
+
+        let active = make_community(&pool).await;
+        let target = make_community(&pool).await;
+        let (active_workflow, _) = make_workflow_in(&pool, active).await;
+        let (target_workflow, _) = make_workflow_in(&pool, target).await;
+        quiesce_test_community(&pool, target).await;
+
+        let workflows = list_all_enabled_workflows(&pool)
+            .await
+            .expect("list enabled workflows");
+        assert!(
+            workflows
+                .iter()
+                .any(|w| w.community_id == active && w.id == active_workflow),
+            "active bystander's schedule workflow must still be scanned"
+        );
+        assert!(
+            !workflows
+                .iter()
+                .any(|w| w.community_id == target && w.id == target_workflow),
+            "quiescing tenant's workflow must not surface in the fleet-wide scan"
+        );
+    }
+
     /// Confinement: a duplicate workflow UUID existing in both community A and
     /// community B must claim independently. Claiming `(A, id, t)` must NOT
     /// consume `(B, id, t)` — B's identical instant stays claimable, and the