From 9199d38ecc3c8fb2a69021cc2bb5765039159023 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 25 Aug 2026 11:51:47 -0400 Subject: [PATCH] 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 Co-authored-by: Wakko --- crates/buzz-db/src/channel.rs | 146 ++++++++++++++++++++++++++++++++- crates/buzz-db/src/event.rs | 63 ++++++++++++++ crates/buzz-db/src/workflow.rs | 63 ++++++++++++++ 3 files changed, 271 insertions(+), 1 deletion(-) diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index 5311335551..c484d67def 100644 --- a/crates/buzz-db/src/channel.rs +++ b/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::(); + 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 = + 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"] diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs index dcfc97c844..d0e236d048 100644 --- a/crates/buzz-db/src/event.rs +++ b/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", ¬_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 diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index d4bebfb8fe..1c90de53b2 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/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> { let rows = sqlx::query( r#" @@ -556,6 +563,8 @@ pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result'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