From 386ccc21e2095b563f4d59d3e732bb6a7058a4f0 Mon Sep 17 00:00:00 2001 From: Wes Date: Tue, 18 Aug 2026 12:50:12 -0600 Subject: [PATCH 1/8] Repair stale large channel roster snapshots Detect relay-authored NIP-29 member snapshots truncated by the former 1,000-member query cap and republish only kind 39002 during relay startup. Keep the maintenance scan tenant-aware, signer-scoped, unpaginated, and convergent after a successful repair. Co-authored-by: Carl Signed-off-by: Wes --- crates/buzz-db/src/channel.rs | 245 ++++++++++++++++++ crates/buzz-db/src/lib.rs | 18 ++ .../buzz-relay/src/handlers/side_effects.rs | 92 ++++++- crates/buzz-relay/src/main.rs | 13 + 4 files changed, 356 insertions(+), 12 deletions(-) diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index 8035ab58adb..203c80fe8e7 100644 --- a/crates/buzz-db/src/channel.rs +++ b/crates/buzz-db/src/channel.rs @@ -781,6 +781,81 @@ pub async fn get_accessible_channel_ids( .collect() } +/// A large channel whose canonical active-member count may need its legacy +/// discovery snapshot repaired. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LargeChannelRoster { + /// Community that owns the channel. + pub community_id: CommunityId, + /// Canonical host for the owning community. + pub host: String, + /// Channel whose roster snapshot differs from canonical membership. + pub channel_id: Uuid, + /// Canonical active-member count. + pub member_count: i64, +} + +/// Returns active channels whose canonical roster exceeds `minimum_members`. +/// +/// 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( + pool: &PgPool, + minimum_members: i64, + relay_pubkey: &[u8], +) -> Result> { + let rows = sqlx::query( + r#" + WITH large_rosters AS ( + SELECT cm.community_id, cm.channel_id, COUNT(*) AS member_count + FROM channel_members cm + JOIN channels ch + ON ch.community_id = cm.community_id + AND ch.id = cm.channel_id + AND ch.deleted_at IS NULL + WHERE cm.removed_at IS NULL + GROUP BY cm.community_id, cm.channel_id + HAVING COUNT(*) > $1 + ) + 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 LATERAL ( + SELECT roster.tags + FROM events roster + WHERE roster.community_id = lr.community_id + AND roster.channel_id = lr.channel_id + AND roster.kind = 39002 + AND roster.pubkey = $2 + AND roster.deleted_at IS NULL + ORDER BY roster.created_at DESC, roster.id ASC + LIMIT 1 + ) live_roster ON true + WHERE lr.member_count <> ( + SELECT COUNT(*) + FROM jsonb_array_elements(live_roster.tags) tag + WHERE tag->>0 = 'p' + ) + ORDER BY lr.community_id, lr.channel_id + "#, + ) + .bind(minimum_members) + .bind(relay_pubkey) + .fetch_all(pool) + .await?; + + rows.into_iter() + .map(|row| { + Ok(LargeChannelRoster { + community_id: CommunityId::from_uuid(row.try_get("community_id")?), + host: row.try_get("host")?, + channel_id: row.try_get("channel_id")?, + member_count: row.try_get("member_count")?, + }) + }) + .collect() +} + /// Lists channels in a community, optionally filtered by visibility string. pub async fn list_channels( pool: &PgPool, @@ -2016,6 +2091,176 @@ mod tests { ); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn large_roster_reconciliation_candidates_respect_snapshot_count_and_signer() { + let pool = setup_pool().await; + let community_id = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_id); + let creator = random_pubkey(); + let relay_pubkey = random_pubkey(); + let other_relay_pubkey = random_pubkey(); + let channel = create_test_channel( + &pool, + community_id, + "stale-large-roster", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &creator, + None, + ) + .await + .expect("create test channel"); + + let extra_members = 1_500; + 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, $3) n + "#, + ) + .bind(community_id) + .bind(channel.id) + .bind(extra_members) + .execute(&pool) + .await + .expect("insert large roster"); + + let stale_tags: Vec = + std::iter::once(serde_json::json!(["d", channel.id.to_string()])) + .chain((0..1_000).map(|n| serde_json::json!(["p", format!("{n:064x}")]))) + .collect(); + let complete_tags: Vec = + std::iter::once(serde_json::json!(["d", channel.id.to_string()])) + .chain((0..1_501).map(|n| serde_json::json!(["p", format!("{n:064x}")]))) + .collect(); + + // A historical duplicate live row can exist from earlier replacement + // bugs. Candidate selection must apply NIP-16 head ordering instead of + // producing one candidate per row or accepting this older complete row. + 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() - INTERVAL '1 minute', 39002, $4, '', $5, $6, $7), + ($1, $8, $3, NOW(), 39002, $9, '', $5, $6, $7) + "#, + ) + .bind(community_id) + .bind(random_pubkey()) + .bind(&relay_pubkey) + .bind(serde_json::Value::Array(complete_tags.clone())) + .bind(vec![0u8; 64]) + .bind(channel.id) + .bind(channel.id.to_string()) + .bind(random_pubkey()) + .bind(serde_json::Value::Array(stale_tags)) + .execute(&pool) + .await + .expect("insert historical complete and live stale snapshots"); + + // The same channel UUID in another tenant is deliberately valid. A + // complete snapshot there must not mask this tenant's stale head. + let other_community_id = make_test_community(&pool).await; + sqlx::query( + r#" + INSERT INTO channels + (id, community_id, name, channel_type, visibility, created_by) + VALUES ($1, $2, 'same-id-complete-roster', 'stream', 'open', $3) + "#, + ) + .bind(channel.id) + .bind(other_community_id) + .bind(&creator) + .execute(&pool) + .await + .expect("insert same channel id in other tenant"); + 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(0, 1500) n + "#, + ) + .bind(other_community_id) + .bind(channel.id) + .execute(&pool) + .await + .expect("insert complete other-tenant roster"); + 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(other_community_id) + .bind(random_pubkey()) + .bind(&relay_pubkey) + .bind(serde_json::Value::Array(complete_tags.clone())) + .bind(vec![0u8; 64]) + .bind(channel.id) + .bind(channel.id.to_string()) + .execute(&pool) + .await + .expect("insert complete other-tenant snapshot"); + + // Put the stale channel behind the 1,000 newest channels that the old + // list_channels-based sweep could see. This set-based scan has no such + // pagination ceiling. + sqlx::query( + r#" + INSERT INTO channels + (id, community_id, name, channel_type, visibility, created_by, created_at) + SELECT gen_random_uuid(), $1, 'newer-decoy-' || n, 'stream', 'open', $2, + NOW() + (n || ' seconds')::interval + FROM generate_series(1, 1000) n + "#, + ) + .bind(community_id) + .bind(&creator) + .execute(&pool) + .await + .expect("insert channels beyond old list ceiling"); + + let candidates = + list_large_channel_rosters_needing_reconciliation(&pool, 1_000, &relay_pubkey) + .await + .expect("find stale snapshot"); + assert_eq!(candidates.len(), 1); + assert_eq!(candidates[0].community_id, community); + assert_eq!(candidates[0].channel_id, channel.id); + assert_eq!(candidates[0].member_count, 1_501); + + let other_signer_candidates = + list_large_channel_rosters_needing_reconciliation(&pool, 1_000, &other_relay_pubkey) + .await + .expect("other signer is isolated from relay-authored snapshot"); + assert!(other_signer_candidates.is_empty()); + + sqlx::query( + "UPDATE events SET tags = $1, created_at = NOW() + INTERVAL '1 minute' WHERE community_id = $2 AND channel_id = $3 AND kind = 39002 AND pubkey = $4 AND created_at = (SELECT MAX(created_at) FROM events WHERE community_id = $2 AND channel_id = $3 AND kind = 39002 AND pubkey = $4 AND deleted_at IS NULL)", + ) + .bind(serde_json::Value::Array(complete_tags)) + .bind(community_id) + .bind(channel.id) + .bind(&relay_pubkey) + .execute(&pool) + .await + .expect("complete snapshot"); + + let converged = + list_large_channel_rosters_needing_reconciliation(&pool, 1_000, &relay_pubkey) + .await + .expect("check converged snapshot"); + assert!(converged.is_empty()); + } + /// 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/lib.rs b/crates/buzz-db/src/lib.rs index 330525d310d..9c1ac1d06ec 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -2476,6 +2476,24 @@ impl Db { channel::get_accessible_channel_ids(&self.pool, community_id, pubkey).await } + /// Returns large active-channel rosters whose relay-authored snapshots differ. + #[datastore_span( + name = "list_large_channel_rosters_needing_reconciliation", + system = "postgresql" + )] + pub async fn list_large_channel_rosters_needing_reconciliation( + &self, + minimum_members: i64, + relay_pubkey: &[u8], + ) -> Result> { + channel::list_large_channel_rosters_needing_reconciliation( + &self.pool, + minimum_members, + relay_pubkey, + ) + .await + } + /// Lists channels, optionally filtered by visibility. #[datastore_span(name = "list_channels", system = "postgresql")] pub async fn list_channels( diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 282ea776577..ca0c4bcc3a6 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1049,6 +1049,26 @@ fn group_members_tags(group_id: &str, members: &[MemberRecord]) -> anyhow::Resul Ok(tags) } +async fn emit_group_members_event( + tenant: &TenantContext, + state: &Arc, + channel_id: Uuid, + members: &[MemberRecord], + relay_pubkey_hex: &str, +) -> anyhow::Result<()> { + let group_id = channel_id.to_string(); + let tags = group_members_tags(&group_id, members)?; + emit_addressable_discovery_event( + tenant, + state, + channel_id, + KIND_NIP29_GROUP_MEMBERS, + tags, + relay_pubkey_hex, + ) + .await +} + /// Emit NIP-29 group discovery events (39000, 39001, 39002) signed by the relay keypair. /// Called after group creation, metadata changes, or membership changes. /// Events are stored channel-scoped (`channel_id = Some(...)`) so that existing @@ -1151,18 +1171,7 @@ pub async fn emit_group_discovery_events( .await?; } - { - let tags = group_members_tags(&group_id, &members)?; - emit_addressable_discovery_event( - tenant, - state, - channel_id, - KIND_NIP29_GROUP_MEMBERS, - tags, - &relay_pubkey_hex, - ) - .await?; - } + emit_group_members_event(tenant, state, channel_id, &members, &relay_pubkey_hex).await?; Ok(()) } @@ -3052,6 +3061,65 @@ pub async fn publish_nip43_member_removed( publish_nip43_delta(tenant, state, 8001, target_pubkey_hex, "member-removed").await } +/// Repair legacy kind:39002 snapshots truncated by the former 1,000-member +/// database cap. +/// +/// The scan is deliberately limited to canonical rosters above that boundary, +/// so normal-sized channels and already-correct large snapshots incur no +/// rewrites. Community identity travels with every candidate; a shared relay +/// never resolves a channel against a neighboring tenant. +pub async fn reconcile_large_channel_member_snapshots( + state: &Arc, +) -> anyhow::Result { + const LEGACY_ROSTER_LIMIT: i64 = 1_000; + + let relay_pubkey = state.relay_keypair.public_key(); + let candidates = state + .db + .list_large_channel_rosters_needing_reconciliation( + LEGACY_ROSTER_LIMIT, + &relay_pubkey.to_bytes(), + ) + .await?; + let relay_pubkey_hex = relay_pubkey.to_hex(); + let mut reconciled = 0usize; + + for candidate in candidates { + let result = async { + let channel_id = candidate.channel_id; + // Re-read the canonical roster immediately before publication. A + // membership mutation may have landed after the candidate scan. + let members = state + .db + .get_members(candidate.community_id, channel_id) + .await?; + let tenant = TenantContext::resolved(candidate.community_id, candidate.host.clone()); + emit_group_members_event(&tenant, state, channel_id, &members, &relay_pubkey_hex) + .await?; + Ok::(true) + } + .await; + + match result { + Ok(true) => reconciled += 1, + Ok(false) => {} + Err(error) => { + metrics::counter!("buzz_channel_roster_reconciliation_failures_total").increment(1); + warn!( + community_id = %candidate.community_id, + host = %candidate.host, + channel_id = %candidate.channel_id, + %error, + "large channel roster reconciliation failed" + ); + } + } + } + + metrics::counter!("buzz_channel_roster_reconciliations_total").increment(reconciled as u64); + Ok(reconciled) +} + /// Reconcile channels that exist in the DB but don't have kind:39000 events. /// /// This handles the case where channels were created via direct SQL inserts diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 3584e1849d1..e383946d262 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -534,6 +534,19 @@ async fn main() -> anyhow::Result<()> { ); } + // Repair legacy NIP-29 channel rosters that were persisted while the + // canonical member query still truncated at 1,000 rows. This runs before + // listeners open, is tenant-preserving across shared deployments, and + // rewrites only mismatched large-channel snapshots. + match buzz_relay::handlers::side_effects::reconcile_large_channel_member_snapshots(&state).await + { + Ok(count) if count > 0 => info!(count, "large channel member snapshots repaired"), + Ok(_) => {} + Err(error) => { + tracing::warn!(%error, "large channel member snapshot startup reconciliation failed") + } + } + // 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 From 0a1f685f9d7e583a5f2e1beed47d797e24e335ff Mon Sep 17 00:00:00 2001 From: Wes Date: Tue, 18 Aug 2026 16:59:34 -0600 Subject: [PATCH 2/8] Serialize roster publication with membership writes Hold the per-channel membership advisory lock from canonical roster capture through discovery-event replacement. This prevents startup reconciliation from superseding a roster published for a concurrently committed mutation. Apply the same fence to normal discovery publication and cover the lock contract with a deterministic Postgres regression. Co-authored-by: Carl Signed-off-by: Wes --- crates/buzz-db/src/channel.rs | 119 ++++++++++++++++++ crates/buzz-db/src/lib.rs | 10 ++ .../buzz-relay/src/handlers/side_effects.rs | 34 +++-- 3 files changed, 154 insertions(+), 9 deletions(-) diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index 203c80fe8e7..cb2985e19ed 100644 --- a/crates/buzz-db/src/channel.rs +++ b/crates/buzz-db/src/channel.rs @@ -366,6 +366,59 @@ async fn acquire_channel_membership_lock( Ok(()) } +/// An active member roster captured while holding the channel's membership +/// serialization lock. +/// +/// Keep this value alive until every membership-derived discovery event has +/// been stored. Dropping it rolls back the read-only transaction and releases +/// the lock; [`LockedMemberSnapshot::release`] does so explicitly. +pub struct LockedMemberSnapshot { + /// Canonical active members captured behind the lock. + pub members: Vec, + tx: Transaction<'static, Postgres>, +} + +impl LockedMemberSnapshot { + /// Release the membership lock after publishing the captured snapshot. + pub async fn release(self) -> Result<()> { + self.tx.commit().await?; + Ok(()) + } +} + +/// Capture all active members while holding the same per-channel lock used by +/// membership writers. +/// +/// The returned guard must remain alive through publication. This prevents a +/// rolling relay from publishing an older roster after a concurrent add or +/// remove has committed and published newer membership state. +pub async fn lock_member_snapshot( + pool: &PgPool, + community_id: CommunityId, + channel_id: Uuid, +) -> Result { + let mut tx = pool.begin().await?; + acquire_channel_membership_lock(&mut tx, community_id, channel_id).await?; + let rows = sqlx::query( + r#" + SELECT cm.channel_id, cm.pubkey, cm.role::text AS role, cm.joined_at, cm.invited_by, cm.removed_at + FROM channel_members cm + JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL + WHERE cm.community_id = $1 AND cm.channel_id = $2 AND cm.removed_at IS NULL + ORDER BY cm.joined_at ASC + "#, + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_all(&mut *tx) + .await?; + let members = rows + .into_iter() + .map(row_to_member_record) + .collect::>>()?; + Ok(LockedMemberSnapshot { members, tx }) +} + /// Add a member to a channel. /// /// Role enforcement: @@ -2744,6 +2797,72 @@ mod tests { (community, channel.id, owner_a, owner_b) } + /// A captured roster holds the same lock as membership writers until the + /// publisher explicitly releases it. This is the freshness fence used by + /// rolling-deploy reconciliation. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn locked_member_snapshot_blocks_post_capture_membership_mutation() { + let pool = setup_pool().await; + let community_id = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_id); + let owner = random_pubkey(); + let newcomer = random_pubkey(); + let channel = create_test_channel( + &pool, + community_id, + "snapshot-freshness-fence", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner, + None, + ) + .await + .expect("create channel"); + + let snapshot = lock_member_snapshot(&pool, community, channel.id) + .await + .expect("capture locked roster"); + assert_eq!(snapshot.members.len(), 1); + + let mut contender = pool.begin().await.expect("begin membership writer"); + let acquired: bool = + sqlx::query_scalar("SELECT pg_try_advisory_xact_lock(hashtextextended($1, 0))") + .bind(format!( + "{CHANNEL_MEMBERSHIP_LOCK_NAMESPACE}{}:{}", + community.as_uuid(), + channel.id + )) + .fetch_one(&mut *contender) + .await + .expect("try membership writer lock"); + assert!( + !acquired, + "membership mutation must wait until the captured roster is published" + ); + contender.rollback().await.expect("rollback contender"); + + snapshot.release().await.expect("release snapshot fence"); + add_member( + &pool, + community, + channel.id, + &newcomer, + MemberRole::Member, + None, + ) + .await + .expect("membership mutation after publication"); + assert_eq!( + get_members(&pool, community, channel.id) + .await + .expect("fresh roster") + .len(), + 2 + ); + } + /// The lock must be shared with `remove_member`: a demotion racing an owner /// removal goes through a separate count/update path, so both must serialize /// on the same key or they can jointly empty the owner set. diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9c1ac1d06ec..68d93273277 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -2390,6 +2390,16 @@ impl Db { channel::set_canvas(&self.pool, community_id, channel_id, canvas).await } + /// Capture the active roster while holding the membership-writer lock. + #[datastore_span(name = "lock_member_snapshot", system = "postgresql")] + pub async fn lock_member_snapshot( + &self, + community_id: CommunityId, + channel_id: Uuid, + ) -> Result { + channel::lock_member_snapshot(&self.pool, community_id, channel_id).await + } + /// Adds a member to a channel. #[datastore_span(name = "add_member", system = "postgresql")] pub async fn add_member( diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index ca0c4bcc3a6..d6eab18d1d0 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1083,7 +1083,14 @@ pub async fn emit_group_discovery_events( channel_id: Uuid, ) -> anyhow::Result<()> { let channel = state.db.get_channel(tenant.community(), channel_id).await?; - let members = state.db.get_members(tenant.community(), channel_id).await?; + // Keep the membership-writer lock through all membership-derived event + // replacements. This orders normal publication and startup repair by the + // canonical roster they captured, rather than by wall-clock timestamps. + let member_snapshot = state + .db + .lock_member_snapshot(tenant.community(), channel_id) + .await?; + let members = &member_snapshot.members; let relay_pubkey_hex = hex::encode(state.relay_keypair.public_key().to_bytes()); let group_id = channel_id.to_string(); @@ -1109,7 +1116,7 @@ pub async fn emit_group_discovery_events( tags.push(Tag::parse(["hidden"])?); // Include participant pubkeys in kind:39000 for DMs so clients can // resolve display names without a separate kind:39002 fetch. - for m in &members { + for m in members { let pubkey_hex = hex::encode(&m.pubkey); tags.push(Tag::parse(["p", &pubkey_hex])?); } @@ -1171,7 +1178,8 @@ pub async fn emit_group_discovery_events( .await?; } - emit_group_members_event(tenant, state, channel_id, &members, &relay_pubkey_hex).await?; + emit_group_members_event(tenant, state, channel_id, members, &relay_pubkey_hex).await?; + member_snapshot.release().await?; Ok(()) } @@ -3087,15 +3095,23 @@ pub async fn reconcile_large_channel_member_snapshots( for candidate in candidates { let result = async { let channel_id = candidate.channel_id; - // Re-read the canonical roster immediately before publication. A - // membership mutation may have landed after the candidate scan. - let members = state + // Hold the membership-writer lock from roster capture through + // replacement. Otherwise a rolling deployment can publish stale + // roster A after another relay commits and publishes roster B. + let member_snapshot = state .db - .get_members(candidate.community_id, channel_id) + .lock_member_snapshot(candidate.community_id, channel_id) .await?; let tenant = TenantContext::resolved(candidate.community_id, candidate.host.clone()); - emit_group_members_event(&tenant, state, channel_id, &members, &relay_pubkey_hex) - .await?; + emit_group_members_event( + &tenant, + state, + channel_id, + &member_snapshot.members, + &relay_pubkey_hex, + ) + .await?; + member_snapshot.release().await?; Ok::(true) } .await; From e1bd17696be2c282fa8f66129394959fc8ab98bd Mon Sep 17 00:00:00 2001 From: Wes Date: Tue, 18 Aug 2026 17:32:32 -0600 Subject: [PATCH 3/8] Publish roster snapshots on the locked connection Keep roster capture, timestamp selection, and kind 39002 replacement inside the same membership-locked transaction and writer connection. This preserves the rolling-deploy freshness fence without deadlocking supported one-connection pools. Cover replacement and lock exclusion using a pool capped at one. Co-authored-by: Carl Signed-off-by: Wes --- crates/buzz-db/src/channel.rs | 117 ++++++++++++++++-- .../buzz-relay/src/handlers/side_effects.rs | 92 +++++++++----- 2 files changed, 169 insertions(+), 40 deletions(-) diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index cb2985e19ed..ff7d3b83d35 100644 --- a/crates/buzz-db/src/channel.rs +++ b/crates/buzz-db/src/channel.rs @@ -367,11 +367,7 @@ async fn acquire_channel_membership_lock( } /// An active member roster captured while holding the channel's membership -/// serialization lock. -/// -/// Keep this value alive until every membership-derived discovery event has -/// been stored. Dropping it rolls back the read-only transaction and releases -/// the lock; [`LockedMemberSnapshot::release`] does so explicitly. +/// serialization lock on one writer connection. pub struct LockedMemberSnapshot { /// Canonical active members captured behind the lock. pub members: Vec, @@ -379,7 +375,94 @@ pub struct LockedMemberSnapshot { } impl LockedMemberSnapshot { - /// Release the membership lock after publishing the captured snapshot. + /// Return the newest relay-authored member snapshot timestamp using this + /// guard's existing connection. + pub async fn latest_member_event_timestamp( + &mut self, + community_id: CommunityId, + channel_id: Uuid, + relay_pubkey: &[u8], + ) -> Result> { + let value: Option> = sqlx::query_scalar( + "SELECT created_at FROM events WHERE community_id = $1 AND kind = 39002 AND pubkey = $2 AND channel_id = $3 AND deleted_at IS NULL ORDER BY created_at DESC, id ASC LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(relay_pubkey) + .bind(channel_id) + .fetch_optional(&mut *self.tx) + .await?; + Ok(value.map(|timestamp| timestamp.timestamp() as u64)) + } + + /// Replace the relay-authored member snapshot on this guard's existing + /// connection. The membership lock therefore spans capture and replacement + /// without a nested pool checkout. + pub async fn replace_member_event( + &mut self, + community_id: CommunityId, + channel_id: Uuid, + event: &nostr::Event, + ) -> Result<(buzz_core::StoredEvent, bool)> { + let kind = buzz_core::kind::event_kind_i32(event); + let pubkey = event.pubkey.to_bytes(); + let created_at_secs = event.created_at.as_secs() as i64; + let created_at = chrono::DateTime::from_timestamp(created_at_secs, 0) + .ok_or(DbError::InvalidTimestamp(created_at_secs))?; + let existing: Option<(chrono::DateTime, Vec)> = sqlx::query_as( + "SELECT created_at, id FROM events WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND channel_id = $4 AND deleted_at IS NULL ORDER BY created_at DESC, id ASC LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(kind) + .bind(pubkey.as_slice()) + .bind(channel_id) + .fetch_optional(&mut *self.tx) + .await?; + let incoming_id = event.id.as_bytes().as_slice(); + if let Some((existing_ts, existing_id)) = existing { + if created_at < existing_ts + || (created_at == existing_ts && incoming_id >= existing_id.as_slice()) + { + return Ok(( + buzz_core::StoredEvent::with_received_at( + event.clone(), + Utc::now(), + Some(channel_id), + false, + ), + false, + )); + } + } + sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND channel_id = $4 AND deleted_at IS NULL") + .bind(community_id.as_uuid()).bind(kind).bind(pubkey.as_slice()).bind(channel_id) + .execute(&mut *self.tx).await?; + let received_at = Utc::now(); + let tags = serde_json::to_value(&event.tags)?; + let sig = event.sig.serialize(); + let inserted = sqlx::query("INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) ON CONFLICT DO NOTHING") + .bind(community_id.as_uuid()).bind(event.id.as_bytes().as_slice()) + .bind(pubkey.as_slice()).bind(created_at).bind(kind).bind(tags) + .bind(&event.content).bind(sig.as_slice()).bind(received_at).bind(channel_id) + .bind(crate::event::extract_d_tag(event)).execute(&mut *self.tx).await?; + if inserted.rows_affected() == 0 { + return Err(DbError::InvalidData( + "member snapshot event id already exists".into(), + )); + } + crate::insert_mentions_in_transaction(&mut self.tx, community_id, event, Some(channel_id)) + .await?; + Ok(( + buzz_core::StoredEvent::with_received_at( + event.clone(), + received_at, + Some(channel_id), + true, + ), + true, + )) + } + + /// Commit the replacement and release the membership lock. pub async fn release(self) -> Result<()> { self.tx.commit().await?; Ok(()) @@ -1663,6 +1746,7 @@ mod tests { use super::*; use crate::user::{ensure_user, set_agent_owner}; use nostr::Keys; + use sqlx::postgres::PgPoolOptions; const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 -- local test-only credentials @@ -2821,10 +2905,29 @@ mod tests { .await .expect("create channel"); - let snapshot = lock_member_snapshot(&pool, community, channel.id) + let snapshot_pool = PgPoolOptions::new() + .max_connections(1) + .acquire_timeout(std::time::Duration::from_secs(1)) + .connect(TEST_DB_URL) + .await + .expect("connect one-connection pool"); + let mut snapshot = lock_member_snapshot(&snapshot_pool, community, channel.id) .await .expect("capture locked roster"); assert_eq!(snapshot.members.len(), 1); + let relay_keys = Keys::generate(); + let event = nostr::EventBuilder::new(nostr::Kind::Custom(39002), "") + .tags(vec![ + nostr::Tag::parse(["d", &channel.id.to_string()]).expect("d tag"), + nostr::Tag::parse(["p", &hex::encode(&owner)]).expect("p tag"), + ]) + .sign_with_keys(&relay_keys) + .expect("sign roster"); + let (_, inserted) = snapshot + .replace_member_event(community, channel.id, &event) + .await + .expect("replace roster on held connection"); + assert!(inserted); let mut contender = pool.begin().await.expect("begin membership writer"); let acquired: bool = diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 77a9e56d1dc..c71981ffbfd 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1049,24 +1049,53 @@ fn group_members_tags(group_id: &str, members: &[MemberRecord]) -> anyhow::Resul Ok(tags) } -async fn emit_group_members_event( +async fn store_group_members_event( tenant: &TenantContext, state: &Arc, channel_id: Uuid, - members: &[MemberRecord], - relay_pubkey_hex: &str, -) -> anyhow::Result<()> { + member_snapshot: &mut buzz_db::channel::LockedMemberSnapshot, +) -> anyhow::Result> { let group_id = channel_id.to_string(); - let tags = group_members_tags(&group_id, members)?; - emit_addressable_discovery_event( - tenant, - state, - channel_id, - KIND_NIP29_GROUP_MEMBERS, - tags, - relay_pubkey_hex, - ) - .await + let tags = group_members_tags(&group_id, &member_snapshot.members)?; + let relay_pubkey = state.relay_keypair.public_key().to_bytes(); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + let ts = member_snapshot + .latest_member_event_timestamp(tenant.community(), channel_id, &relay_pubkey) + .await? + .map(|timestamp| timestamp + 1) + .unwrap_or(now) + .max(now); + let event = EventBuilder::new(Kind::Custom(KIND_NIP29_GROUP_MEMBERS as u16), "") + .tags(tags) + .custom_created_at(nostr::Timestamp::from(ts)) + .sign_with_keys(&state.relay_keypair) + .map_err(|error| anyhow::anyhow!("failed to sign member snapshot: {error}"))?; + let (stored, inserted) = member_snapshot + .replace_member_event(tenant.community(), channel_id, &event) + .await?; + Ok(inserted.then_some(stored)) +} + +async fn dispatch_group_members_event( + tenant: &TenantContext, + state: &Arc, + stored: Option, + relay_pubkey_hex: &str, +) { + if let Some(stored) = stored { + dispatch_persistent_event( + tenant, + state, + &stored, + KIND_NIP29_GROUP_MEMBERS, + relay_pubkey_hex, + None, + ) + .await; + } } /// Emit NIP-29 group discovery events (39000, 39001, 39002) signed by the relay keypair. @@ -1083,14 +1112,7 @@ pub async fn emit_group_discovery_events( channel_id: Uuid, ) -> anyhow::Result<()> { let channel = state.db.get_channel(tenant.community(), channel_id).await?; - // Keep the membership-writer lock through all membership-derived event - // replacements. This orders normal publication and startup repair by the - // canonical roster they captured, rather than by wall-clock timestamps. - let member_snapshot = state - .db - .lock_member_snapshot(tenant.community(), channel_id) - .await?; - let members = &member_snapshot.members; + let members = state.db.get_members(tenant.community(), channel_id).await?; let relay_pubkey_hex = hex::encode(state.relay_keypair.public_key().to_bytes()); let group_id = channel_id.to_string(); @@ -1116,7 +1138,7 @@ pub async fn emit_group_discovery_events( tags.push(Tag::parse(["hidden"])?); // Include participant pubkeys in kind:39000 for DMs so clients can // resolve display names without a separate kind:39002 fetch. - for m in members { + for m in &members { let pubkey_hex = hex::encode(&m.pubkey); tags.push(Tag::parse(["p", &pubkey_hex])?); } @@ -1178,8 +1200,17 @@ pub async fn emit_group_discovery_events( .await?; } - emit_group_members_event(tenant, state, channel_id, members, &relay_pubkey_hex).await?; + // Re-capture membership behind the writer lock immediately before the + // authoritative 39002 replacement. Metadata/admin snapshots retain their + // existing behavior; only membership publication needs this freshness fence. + let mut member_snapshot = state + .db + .lock_member_snapshot(tenant.community(), channel_id) + .await?; + let stored_members = + store_group_members_event(tenant, state, channel_id, &mut member_snapshot).await?; member_snapshot.release().await?; + dispatch_group_members_event(tenant, state, stored_members, &relay_pubkey_hex).await; Ok(()) } @@ -3098,20 +3129,15 @@ pub async fn reconcile_large_channel_member_snapshots( // Hold the membership-writer lock from roster capture through // replacement. Otherwise a rolling deployment can publish stale // roster A after another relay commits and publishes roster B. - let member_snapshot = state + let mut member_snapshot = state .db .lock_member_snapshot(candidate.community_id, channel_id) .await?; let tenant = TenantContext::resolved(candidate.community_id, candidate.host.clone()); - emit_group_members_event( - &tenant, - state, - channel_id, - &member_snapshot.members, - &relay_pubkey_hex, - ) - .await?; + let stored_members = + store_group_members_event(&tenant, state, channel_id, &mut member_snapshot).await?; member_snapshot.release().await?; + dispatch_group_members_event(&tenant, state, stored_members, &relay_pubkey_hex).await; Ok::(true) } .await; From 0638a9d571aca2ebedfb3e9bff918807594e6541 Mon Sep 17 00:00:00 2001 From: Wes Date: Tue, 18 Aug 2026 18:01:55 -0600 Subject: [PATCH 4/8] Fence stale rosters during rolling relay deploys Serialize the locked roster writer with the canonical replacement key, then validate every kind 39002 insert against canonical active membership in Postgres. This keeps old and new relay binaries safe during a rolling deploy: a legacy stale capture cannot replace a newer roster, and its soft-delete rolls back with the rejected insert. Cover the actual mixed-version schedule on a one-connection pool and keep the desired-state schema in sync with the additive migration. Co-authored-by: Carl Signed-off-by: Wes --- crates/buzz-db/src/channel.rs | 74 ++++++++-- crates/buzz-db/src/lib.rs | 132 +++++++++++++++++- crates/buzz-db/src/migration.rs | 14 +- .../buzz-relay/src/handlers/side_effects.rs | 5 +- .../0032_channel_roster_snapshot_fence.sql | 66 +++++++++ schema/schema.sql | 69 +++++++++ 6 files changed, 343 insertions(+), 17 deletions(-) create mode 100644 migrations/0032_channel_roster_snapshot_fence.sql diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index ff7d3b83d35..2f5e7cdbd7f 100644 --- a/crates/buzz-db/src/channel.rs +++ b/crates/buzz-db/src/channel.rs @@ -371,6 +371,9 @@ async fn acquire_channel_membership_lock( pub struct LockedMemberSnapshot { /// Canonical active members captured behind the lock. pub members: Vec, + community_id: CommunityId, + channel_id: Uuid, + relay_pubkey: Vec, tx: Transaction<'static, Postgres>, } @@ -403,7 +406,20 @@ impl LockedMemberSnapshot { channel_id: Uuid, event: &nostr::Event, ) -> Result<(buzz_core::StoredEvent, bool)> { + if community_id != self.community_id + || channel_id != self.channel_id + || event.pubkey.to_bytes().as_slice() != self.relay_pubkey.as_slice() + { + return Err(DbError::InvalidData( + "member snapshot replacement does not match its locked coordinate".into(), + )); + } let kind = buzz_core::kind::event_kind_i32(event); + if kind != 39002 { + return Err(DbError::InvalidData( + "member snapshot replacement requires kind 39002".into(), + )); + } let pubkey = event.pubkey.to_bytes(); let created_at_secs = event.created_at.as_secs() as i64; let created_at = chrono::DateTime::from_timestamp(created_at_secs, 0) @@ -479,8 +495,23 @@ pub async fn lock_member_snapshot( pool: &PgPool, community_id: CommunityId, channel_id: Uuid, + relay_pubkey: &[u8], ) -> Result { let mut tx = pool.begin().await?; + // Match the canonical replacement writer's lock order. Old binaries take + // this key before INSERT; migration 0032 then takes the membership key in + // the INSERT trigger. Taking both in that order avoids mixed-version + // duplicate heads without introducing a lock-order inversion. + let replacement_lock = crate::event_replacement_lock_key( + community_id, + 39002, + relay_pubkey, + Some(channel_id.as_bytes()), + ); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(replacement_lock) + .execute(&mut *tx) + .await?; acquire_channel_membership_lock(&mut tx, community_id, channel_id).await?; let rows = sqlx::query( r#" @@ -499,7 +530,13 @@ pub async fn lock_member_snapshot( .into_iter() .map(row_to_member_record) .collect::>>()?; - Ok(LockedMemberSnapshot { members, tx }) + Ok(LockedMemberSnapshot { + members, + community_id, + channel_id, + relay_pubkey: relay_pubkey.to_vec(), + tx, + }) } /// Add a member to a channel. @@ -2275,16 +2312,17 @@ mod tests { .chain((0..1_501).map(|n| serde_json::json!(["p", format!("{n:064x}")]))) .collect(); - // A historical duplicate live row can exist from earlier replacement - // bugs. Candidate selection must apply NIP-16 head ordering instead of - // producing one candidate per row or accepting this older complete row. + // Insert canonical-looking history first, then corrupt the newest row + // with UPDATE to model a stale snapshot that predates migration 0032's + // INSERT fence. New stale snapshots cannot be inserted once that fence + // is deployed. 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() - INTERVAL '1 minute', 39002, $4, '', $5, $6, $7), - ($1, $8, $3, NOW(), 39002, $9, '', $5, $6, $7) + ($1, $8, $3, NOW(), 39002, $4, '', $5, $6, $7) "#, ) .bind(community_id) @@ -2295,10 +2333,21 @@ mod tests { .bind(channel.id) .bind(channel.id.to_string()) .bind(random_pubkey()) + .execute(&pool) + .await + .expect("insert historical duplicate snapshots"); + sqlx::query( + "UPDATE events SET tags = $1 WHERE community_id = $2 AND channel_id = $3 \ + AND kind = 39002 AND pubkey = $4 AND created_at = (SELECT MAX(created_at) \ + FROM events WHERE community_id = $2 AND channel_id = $3 AND kind = 39002 AND pubkey = $4)", + ) .bind(serde_json::Value::Array(stale_tags)) + .bind(community_id) + .bind(channel.id) + .bind(&relay_pubkey) .execute(&pool) .await - .expect("insert historical complete and live stale snapshots"); + .expect("simulate pre-fence stale live snapshot"); // The same channel UUID in another tenant is deliberately valid. A // complete snapshot there must not mask this tenant's stale head. @@ -2911,11 +2960,16 @@ mod tests { .connect(TEST_DB_URL) .await .expect("connect one-connection pool"); - let mut snapshot = lock_member_snapshot(&snapshot_pool, community, channel.id) - .await - .expect("capture locked roster"); - assert_eq!(snapshot.members.len(), 1); let relay_keys = Keys::generate(); + let mut snapshot = lock_member_snapshot( + &snapshot_pool, + community, + channel.id, + &relay_keys.public_key().to_bytes(), + ) + .await + .expect("capture locked roster"); + assert_eq!(snapshot.members.len(), 1); let event = nostr::EventBuilder::new(nostr::Kind::Custom(39002), "") .tags(vec![ nostr::Tag::parse(["d", &channel.id.to_string()]).expect("d tag"), diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 68d93273277..3610f1f20bb 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -68,7 +68,7 @@ use uuid::Uuid; use buzz_core::{CommunityId, StoredEvent}; -fn event_replacement_lock_key( +pub(crate) fn event_replacement_lock_key( community_id: CommunityId, kind: i32, pubkey: &[u8], @@ -2396,8 +2396,9 @@ impl Db { &self, community_id: CommunityId, channel_id: Uuid, + relay_pubkey: &[u8], ) -> Result { - channel::lock_member_snapshot(&self.pool, community_id, channel_id).await + channel::lock_member_snapshot(&self.pool, community_id, channel_id, relay_pubkey).await } /// Adds a member to a channel. @@ -5521,9 +5522,10 @@ mod tests { let community_uuid = Uuid::new_v4(); let channel = Uuid::new_v4(); let keys = Keys::generate(); - seed_community_channel(&pool, community_uuid, channel, &keys).await; + let owner_keys = Keys::generate(); + seed_community_channel(&pool, community_uuid, channel, &owner_keys).await; let community = CommunityId::from_uuid(community_uuid); - let member = Keys::generate().public_key().to_hex(); + let member = owner_keys.public_key().to_hex(); let tags = || { vec![ Tag::parse(["d", channel.to_string().as_str()]).expect("d tag"), @@ -5589,6 +5591,128 @@ mod tests { drop_scratch_db(&admin, pool, &scratch_name).await; } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn stale_legacy_roster_cannot_replace_new_locked_snapshot() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let admin = PgPool::connect(&admin_url().await) + .await + .expect("connect admin"); + let (setup_pool, scratch_name) = create_scratch_db(&admin, "mixed_roster_writer").await; + let base_url = admin_url().await; + let slash = base_url.rfind('/').expect("database URL has path segment"); + let scratch_url = format!("{}/{}", &base_url[..slash], scratch_name); + let pool = PgPoolOptions::new() + .max_connections(1) + .acquire_timeout(Duration::from_secs(1)) + .connect(&scratch_url) + .await + .expect("connect one-connection scratch pool"); + setup_pool.close().await; + let db = Db::from_pool(pool.clone()); + let community_uuid = Uuid::new_v4(); + let community = CommunityId::from_uuid(community_uuid); + let channel = Uuid::new_v4(); + let relay_keys = Keys::generate(); + let owner_keys = Keys::generate(); + let owner = owner_keys.public_key().to_bytes(); + seed_community_channel(&pool, community_uuid, channel, &owner_keys).await; + + // This is the old pod's unlocked capture A. It remains in process memory + // while canonical membership advances and the new pod publishes B. + let base = Timestamp::now().as_secs(); + let roster = |members: &[&[u8]], timestamp| { + let tags = + std::iter::once(Tag::parse(["d", channel.to_string().as_str()]).expect("d tag")) + .chain(members.iter().map(|member| { + Tag::parse(["p", hex::encode(member).as_str(), "", "member"]) + .expect("p tag") + })) + .collect::>(); + EventBuilder::new(Kind::Custom(39002), "") + .tags(tags) + .custom_created_at(Timestamp::from(timestamp)) + .sign_with_keys(&relay_keys) + .expect("sign roster") + }; + let stale_a = roster(&[owner.as_slice()], base + 2); + + let newcomer = Keys::generate().public_key().to_bytes(); + sqlx::query( + "INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) \ + VALUES ($1, $2, $3, 'member', $4)", + ) + .bind(community_uuid) + .bind(channel) + .bind(newcomer.as_slice()) + .bind(owner.as_slice()) + .execute(&pool) + .await + .expect("commit newer canonical membership"); + + let relay_pubkey = relay_keys.public_key().to_bytes(); + let mut snapshot = db + .lock_member_snapshot(community, channel, &relay_pubkey) + .await + .expect("new writer captures locked roster B"); + let fresh_b = roster(&[owner.as_slice(), newcomer.as_slice()], base + 1); + assert!( + snapshot + .replace_member_event(community, channel, &fresh_b) + .await + .expect("new writer publishes B") + .1 + ); + snapshot + .release() + .await + .expect("commit B and release locks"); + + // The legacy canonical path takes the replacement key, soft-deletes B, + // then attempts its newer-timestamp stale A. Migration 0032 rejects the + // INSERT; transaction rollback must restore B. A one-connection pool + // proves the lock order does not turn this compatibility path into a + // self-deadlock. + let error = tokio::time::timeout( + Duration::from_secs(3), + db.replace_addressable_event(community, &stale_a, Some(channel)), + ) + .await + .expect("legacy replacement must not deadlock") + .expect_err("stale captured roster A must be rejected"); + assert!( + matches!( + error, + DbError::Sqlx(sqlx::Error::Database(ref db_error)) + if db_error.code().as_deref() == Some("23514") + ), + "expected roster fence check violation, got {error:?}" + ); + + let live_ids: Vec> = sqlx::query_scalar( + "SELECT id FROM events WHERE community_id=$1 AND channel_id=$2 \ + AND kind=39002 AND pubkey=$3 AND deleted_at IS NULL", + ) + .bind(community_uuid) + .bind(channel) + .bind(relay_pubkey.as_slice()) + .fetch_all(&pool) + .await + .expect("load live roster heads"); + assert_eq!(live_ids, vec![fresh_b.id.as_bytes().to_vec()]); + let stale_rows: i64 = + sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1 AND id=$2") + .bind(community_uuid) + .bind(stale_a.id.as_bytes().as_slice()) + .fetch_one(&pool) + .await + .expect("count rejected stale roster"); + assert_eq!(stale_rows, 0, "stale roster insert must roll back"); + + drop_scratch_db(&admin, pool, &scratch_name).await; + } + #[tokio::test] #[ignore = "requires Postgres"] async fn nip_rs_replacement_hard_deletes_payload_and_watermark_rejects_replay() { diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index be87faa1ac2..0a6709c86a4 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -625,7 +625,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 31); + assert_eq!(migrations.len(), 32); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -1036,6 +1036,18 @@ mod tests { assert_eq!(migrations[29].version, 30); let deletion_recovery = migrations[29].sql.as_str(); assert!(deletion_recovery.contains("SET LOCAL lock_timeout = '5s'")); + + // Mixed-version channel-roster fence: old canonical replacement writers + // acquire their replacement key before INSERT; this trigger then takes + // the membership key and validates the exact active p-tag set. + assert_eq!(migrations[31].version, 32); + let roster_fence = migrations[31].sql.as_str(); + assert!(roster_fence.contains("CREATE TRIGGER trg_events_guard_channel_roster_snapshot")); + assert!(roster_fence.contains("NEW.kind <> 39002")); + assert!(roster_fence.contains("'buzz_channel_membership:'")); + assert!(roster_fence.contains("cm.removed_at IS NULL")); + assert!(roster_fence.contains("snapshot_members IS DISTINCT FROM canonical_members")); + assert!(roster_fence.contains("ERRCODE = '23514'")); } #[test] diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index c71981ffbfd..89595fbee17 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1203,9 +1203,10 @@ pub async fn emit_group_discovery_events( // Re-capture membership behind the writer lock immediately before the // authoritative 39002 replacement. Metadata/admin snapshots retain their // existing behavior; only membership publication needs this freshness fence. + let relay_pubkey = state.relay_keypair.public_key().to_bytes(); let mut member_snapshot = state .db - .lock_member_snapshot(tenant.community(), channel_id) + .lock_member_snapshot(tenant.community(), channel_id, &relay_pubkey) .await?; let stored_members = store_group_members_event(tenant, state, channel_id, &mut member_snapshot).await?; @@ -3131,7 +3132,7 @@ pub async fn reconcile_large_channel_member_snapshots( // roster A after another relay commits and publishes roster B. let mut member_snapshot = state .db - .lock_member_snapshot(candidate.community_id, channel_id) + .lock_member_snapshot(candidate.community_id, channel_id, &relay_pubkey.to_bytes()) .await?; let tenant = TenantContext::resolved(candidate.community_id, candidate.host.clone()); let stored_members = diff --git a/migrations/0032_channel_roster_snapshot_fence.sql b/migrations/0032_channel_roster_snapshot_fence.sql new file mode 100644 index 00000000000..1d99c05f330 --- /dev/null +++ b/migrations/0032_channel_roster_snapshot_fence.sql @@ -0,0 +1,66 @@ +-- Prevent mixed-version relay pods from publishing a stale NIP-29 member +-- snapshot after a newer canonical roster has been committed. +-- +-- Old binaries already serialize kind 39002 replacement on the replacement +-- advisory key. This trigger adds the channel-membership key at INSERT time, +-- after that canonical key, and validates every p tag against the current +-- active membership set. New binaries take both keys in the same order before +-- capture and replacement. Thus old and new writers remain compatible during +-- a rolling deploy. +CREATE OR REPLACE FUNCTION guard_channel_roster_snapshot() +RETURNS TRIGGER AS $$ +DECLARE + canonical_members BYTEA[]; + snapshot_members BYTEA[]; +BEGIN + IF NEW.kind <> 39002 OR NEW.channel_id IS NULL THEN + RETURN NEW; + END IF; + + PERFORM pg_advisory_xact_lock(hashtextextended( + 'buzz_channel_membership:' || NEW.community_id::text || ':' || NEW.channel_id::text, + 0 + )); + + SELECT COALESCE(array_agg(cm.pubkey ORDER BY cm.pubkey), ARRAY[]::BYTEA[]) + INTO canonical_members + FROM channel_members cm + WHERE cm.community_id = NEW.community_id + AND cm.channel_id = NEW.channel_id + AND cm.removed_at IS NULL; + + -- A roster is canonical only when every p tag is a 32-byte hex pubkey, + -- no member is duplicated, and the normalized tag set exactly matches the + -- active membership set. Ignoring malformed tags would let an otherwise + -- complete roster smuggle invalid discovery data through the fence. + IF EXISTS ( + SELECT 1 + FROM jsonb_array_elements(NEW.tags) tag + WHERE tag->>0 = 'p' + AND (jsonb_array_length(tag) < 2 OR COALESCE(tag->>1, '') !~ '^[0-9a-fA-F]{64}$') + ) THEN + RAISE EXCEPTION 'kind 39002 roster contains an invalid p tag' + USING ERRCODE = '23514'; + END IF; + + SELECT COALESCE(array_agg(member ORDER BY member), ARRAY[]::BYTEA[]) + INTO snapshot_members + FROM ( + SELECT decode(tag->>1, 'hex') AS member + FROM jsonb_array_elements(NEW.tags) tag + WHERE tag->>0 = 'p' + ) snapshot; + + IF snapshot_members IS DISTINCT FROM canonical_members THEN + RAISE EXCEPTION 'kind 39002 roster does not match canonical channel membership' + USING ERRCODE = '23514'; + END IF; + + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events; +CREATE TRIGGER trg_events_guard_channel_roster_snapshot + BEFORE INSERT ON events + FOR EACH ROW EXECUTE FUNCTION guard_channel_roster_snapshot(); diff --git a/schema/schema.sql b/schema/schema.sql index 9ef7bc0a4b8..a6bc57e55c1 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -990,6 +990,75 @@ AFTER INSERT ON events DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION refresh_channel_ttl_after_event_insert(); +-- Channel roster snapshot fence (keep in sync with migrations/0032). +-- Prevent mixed-version relay pods from publishing a stale NIP-29 member +-- snapshot after a newer canonical roster has been committed. +-- +-- Old binaries already serialize kind 39002 replacement on the replacement +-- advisory key. This trigger adds the channel-membership key at INSERT time, +-- after that canonical key, and validates every p tag against the current +-- active membership set. New binaries take both keys in the same order before +-- capture and replacement. Thus old and new writers remain compatible during +-- a rolling deploy. +CREATE OR REPLACE FUNCTION guard_channel_roster_snapshot() +RETURNS TRIGGER AS $$ +DECLARE + canonical_members BYTEA[]; + snapshot_members BYTEA[]; +BEGIN + IF NEW.kind <> 39002 OR NEW.channel_id IS NULL THEN + RETURN NEW; + END IF; + + PERFORM pg_advisory_xact_lock(hashtextextended( + 'buzz_channel_membership:' || NEW.community_id::text || ':' || NEW.channel_id::text, + 0 + )); + + SELECT COALESCE(array_agg(cm.pubkey ORDER BY cm.pubkey), ARRAY[]::BYTEA[]) + INTO canonical_members + FROM channel_members cm + WHERE cm.community_id = NEW.community_id + AND cm.channel_id = NEW.channel_id + AND cm.removed_at IS NULL; + + -- A roster is canonical only when every p tag is a 32-byte hex pubkey, + -- no member is duplicated, and the normalized tag set exactly matches the + -- active membership set. Ignoring malformed tags would let an otherwise + -- complete roster smuggle invalid discovery data through the fence. + IF EXISTS ( + SELECT 1 + FROM jsonb_array_elements(NEW.tags) tag + WHERE tag->>0 = 'p' + AND (jsonb_array_length(tag) < 2 OR COALESCE(tag->>1, '') !~ '^[0-9a-fA-F]{64}$') + ) THEN + RAISE EXCEPTION 'kind 39002 roster contains an invalid p tag' + USING ERRCODE = '23514'; + END IF; + + SELECT COALESCE(array_agg(member ORDER BY member), ARRAY[]::BYTEA[]) + INTO snapshot_members + FROM ( + SELECT decode(tag->>1, 'hex') AS member + FROM jsonb_array_elements(NEW.tags) tag + WHERE tag->>0 = 'p' + ) snapshot; + + IF snapshot_members IS DISTINCT FROM canonical_members THEN + RAISE EXCEPTION 'kind 39002 roster does not match canonical channel membership' + USING ERRCODE = '23514'; + END IF; + + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events; +CREATE TRIGGER trg_events_guard_channel_roster_snapshot + BEFORE INSERT ON events + FOR EACH ROW EXECUTE FUNCTION guard_channel_roster_snapshot(); + + -- Replica-fence floor guard (keep in sync with migrations/0021). A deferred -- constraint trigger re-checks, inside COMMIT processing, that channel-bearing -- event rows are no older than `buzz.created_at_floor` seconds before commit From 8f109bca79669f4f7dc65e8499016919d2be81b3 Mon Sep 17 00:00:00 2001 From: Wes Date: Tue, 18 Aug 2026 18:20:30 -0600 Subject: [PATCH 5/8] Fence stale roster roles during rolling deploys Compare each kind 39002 member's authoritative role as well as its pubkey against canonical active membership. Reject malformed role tags and cover the legacy stale-capture schedule with a role-only mutation. Co-authored-by: Carl Signed-off-by: Wes --- crates/buzz-db/src/lib.rs | 34 +++++++++---- crates/buzz-db/src/migration.rs | 5 +- .../0032_channel_roster_snapshot_fence.sql | 48 +++++++++++-------- 3 files changed, 58 insertions(+), 29 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 3610f1f20bb..9339009afa3 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -5529,7 +5529,7 @@ mod tests { let tags = || { vec![ Tag::parse(["d", channel.to_string().as_str()]).expect("d tag"), - Tag::parse(["p", member.as_str(), "", "member"]).expect("p tag"), + Tag::parse(["p", member.as_str(), "", "owner"]).expect("p tag"), ] }; let base = Timestamp::now().as_secs(); @@ -5620,14 +5620,13 @@ mod tests { seed_community_channel(&pool, community_uuid, channel, &owner_keys).await; // This is the old pod's unlocked capture A. It remains in process memory - // while canonical membership advances and the new pod publishes B. + // while a role-only canonical mutation advances and the new pod publishes B. let base = Timestamp::now().as_secs(); - let roster = |members: &[&[u8]], timestamp| { + let roster = |members: &[(&[u8], &str)], timestamp| { let tags = std::iter::once(Tag::parse(["d", channel.to_string().as_str()]).expect("d tag")) - .chain(members.iter().map(|member| { - Tag::parse(["p", hex::encode(member).as_str(), "", "member"]) - .expect("p tag") + .chain(members.iter().map(|(member, role)| { + Tag::parse(["p", hex::encode(member).as_str(), "", *role]).expect("p tag") })) .collect::>(); EventBuilder::new(Kind::Custom(39002), "") @@ -5636,7 +5635,6 @@ mod tests { .sign_with_keys(&relay_keys) .expect("sign roster") }; - let stale_a = roster(&[owner.as_slice()], base + 2); let newcomer = Keys::generate().public_key().to_bytes(); sqlx::query( @@ -5649,14 +5647,32 @@ mod tests { .bind(owner.as_slice()) .execute(&pool) .await - .expect("commit newer canonical membership"); + .expect("seed member before legacy capture"); + let stale_a = roster( + &[(owner.as_slice(), "owner"), (newcomer.as_slice(), "member")], + base + 2, + ); + + sqlx::query( + "UPDATE channel_members SET role = 'admin' \ + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3", + ) + .bind(community_uuid) + .bind(channel) + .bind(newcomer.as_slice()) + .execute(&pool) + .await + .expect("commit newer canonical role"); let relay_pubkey = relay_keys.public_key().to_bytes(); let mut snapshot = db .lock_member_snapshot(community, channel, &relay_pubkey) .await .expect("new writer captures locked roster B"); - let fresh_b = roster(&[owner.as_slice(), newcomer.as_slice()], base + 1); + let fresh_b = roster( + &[(owner.as_slice(), "owner"), (newcomer.as_slice(), "admin")], + base + 1, + ); assert!( snapshot .replace_member_event(community, channel, &fresh_b) diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 0a6709c86a4..2c9824d4a5b 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -1039,13 +1039,16 @@ mod tests { // Mixed-version channel-roster fence: old canonical replacement writers // acquire their replacement key before INSERT; this trigger then takes - // the membership key and validates the exact active p-tag set. + // the membership key and validates the exact active pubkey/role p-tag set. assert_eq!(migrations[31].version, 32); let roster_fence = migrations[31].sql.as_str(); assert!(roster_fence.contains("CREATE TRIGGER trg_events_guard_channel_roster_snapshot")); assert!(roster_fence.contains("NEW.kind <> 39002")); assert!(roster_fence.contains("'buzz_channel_membership:'")); assert!(roster_fence.contains("cm.removed_at IS NULL")); + assert!(roster_fence.contains("cm.role::text")); + assert!(roster_fence.contains("jsonb_array_length(roster_tag.tag_json) <> 4")); + assert!(roster_fence.contains("roster_tag.tag_json->>3")); assert!(roster_fence.contains("snapshot_members IS DISTINCT FROM canonical_members")); assert!(roster_fence.contains("ERRCODE = '23514'")); } diff --git a/migrations/0032_channel_roster_snapshot_fence.sql b/migrations/0032_channel_roster_snapshot_fence.sql index 1d99c05f330..cdc7bc4b93e 100644 --- a/migrations/0032_channel_roster_snapshot_fence.sql +++ b/migrations/0032_channel_roster_snapshot_fence.sql @@ -4,14 +4,14 @@ -- Old binaries already serialize kind 39002 replacement on the replacement -- advisory key. This trigger adds the channel-membership key at INSERT time, -- after that canonical key, and validates every p tag against the current --- active membership set. New binaries take both keys in the same order before --- capture and replacement. Thus old and new writers remain compatible during --- a rolling deploy. +-- active membership set and roles. New binaries take both keys in the same +-- order before capture and replacement. Thus old and new writers remain +-- compatible during a rolling deploy. CREATE OR REPLACE FUNCTION guard_channel_roster_snapshot() RETURNS TRIGGER AS $$ DECLARE - canonical_members BYTEA[]; - snapshot_members BYTEA[]; + canonical_members TEXT[]; + snapshot_members TEXT[]; BEGIN IF NEW.kind <> 39002 OR NEW.channel_id IS NULL THEN RETURN NEW; @@ -22,34 +22,44 @@ BEGIN 0 )); - SELECT COALESCE(array_agg(cm.pubkey ORDER BY cm.pubkey), ARRAY[]::BYTEA[]) + SELECT COALESCE( + array_agg(encode(cm.pubkey, 'hex') || ':' || cm.role::text ORDER BY cm.pubkey), + ARRAY[]::TEXT[] + ) INTO canonical_members FROM channel_members cm WHERE cm.community_id = NEW.community_id AND cm.channel_id = NEW.channel_id AND cm.removed_at IS NULL; - -- A roster is canonical only when every p tag is a 32-byte hex pubkey, - -- no member is duplicated, and the normalized tag set exactly matches the - -- active membership set. Ignoring malformed tags would let an otherwise - -- complete roster smuggle invalid discovery data through the fence. + -- A roster is canonical only when every p tag uses the emitted four-field + -- shape, contains a 32-byte hex pubkey and valid authoritative role, has no + -- duplicate members, and exactly matches the active membership rows. IF EXISTS ( SELECT 1 - FROM jsonb_array_elements(NEW.tags) tag - WHERE tag->>0 = 'p' - AND (jsonb_array_length(tag) < 2 OR COALESCE(tag->>1, '') !~ '^[0-9a-fA-F]{64}$') + FROM jsonb_array_elements(NEW.tags) AS roster_tag(tag_json) + WHERE roster_tag.tag_json->>0 = 'p' + AND ( + jsonb_array_length(roster_tag.tag_json) <> 4 + OR COALESCE(roster_tag.tag_json->>1, '') !~ '^[0-9a-fA-F]{64}$' + OR roster_tag.tag_json->>2 <> '' + OR COALESCE(roster_tag.tag_json->>3, '') NOT IN ('owner', 'admin', 'bot', 'member', 'guest') + ) ) THEN RAISE EXCEPTION 'kind 39002 roster contains an invalid p tag' USING ERRCODE = '23514'; END IF; - SELECT COALESCE(array_agg(member ORDER BY member), ARRAY[]::BYTEA[]) + SELECT COALESCE( + array_agg( + lower((roster_tag.tag_json->>1)) || ':' || (roster_tag.tag_json->>3) + ORDER BY decode((roster_tag.tag_json->>1), 'hex') + ), + ARRAY[]::TEXT[] + ) INTO snapshot_members - FROM ( - SELECT decode(tag->>1, 'hex') AS member - FROM jsonb_array_elements(NEW.tags) tag - WHERE tag->>0 = 'p' - ) snapshot; + FROM jsonb_array_elements(NEW.tags) AS roster_tag(tag_json) + WHERE roster_tag.tag_json->>0 = 'p'; IF snapshot_members IS DISTINCT FROM canonical_members THEN RAISE EXCEPTION 'kind 39002 roster does not match canonical channel membership' From 897f01bebfe80210db5da9d9d114fa48db21bcdb Mon Sep 17 00:00:00 2001 From: Wes Date: Tue, 18 Aug 2026 18:30:39 -0600 Subject: [PATCH 6/8] Keep desired-state roster fence role-aware Install the same pubkey-and-role fence in fresh schema bootstraps as migration 0032. Assert exact executable parity and exercise stale-role rejection against a database created directly from schema.sql. Co-authored-by: Carl Signed-off-by: Wes --- crates/buzz-db/src/lib.rs | 94 +++++++++++++++++++++++++++++++++ crates/buzz-db/src/migration.rs | 15 ++++++ schema/schema.sql | 48 ++++++++++------- 3 files changed, 138 insertions(+), 19 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9339009afa3..4d6f84ecf95 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -5729,6 +5729,100 @@ mod tests { drop_scratch_db(&admin, pool, &scratch_name).await; } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn desired_schema_rejects_stale_legacy_roster_role() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let admin = PgPool::connect(&admin_url().await) + .await + .expect("connect admin"); + let scratch_name = format!("schema_roster_role_{}", Uuid::new_v4().simple()); + sqlx::query(sqlx::AssertSqlSafe(format!( + "CREATE DATABASE {scratch_name}" + ))) + .execute(&admin) + .await + .expect("create desired-schema scratch db"); + let base_url = admin_url().await; + let slash = base_url.rfind('/').expect("database URL has path segment"); + let scratch_url = format!("{}/{}", &base_url[..slash], scratch_name); + let pool = PgPoolOptions::new() + .max_connections(1) + .connect(&scratch_url) + .await + .expect("connect desired-schema scratch db"); + sqlx::raw_sql(include_str!("../../../schema/schema.sql")) + .execute(&pool) + .await + .expect("apply desired-state schema"); + + let db = Db::from_pool(pool.clone()); + let community_uuid = Uuid::new_v4(); + let community = CommunityId::from_uuid(community_uuid); + let channel = Uuid::new_v4(); + let relay_keys = Keys::generate(); + let owner_keys = Keys::generate(); + let owner = owner_keys.public_key().to_bytes(); + seed_community_channel(&pool, community_uuid, channel, &owner_keys).await; + let member = Keys::generate().public_key().to_bytes(); + sqlx::query( + "INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) \ + VALUES ($1, $2, $3, 'admin', $4)", + ) + .bind(community_uuid) + .bind(channel) + .bind(member.as_slice()) + .bind(owner.as_slice()) + .execute(&pool) + .await + .expect("seed canonical admin"); + + let roster = |role: &str, timestamp| { + EventBuilder::new(Kind::Custom(39002), "") + .tags(vec![ + Tag::parse(["d", channel.to_string().as_str()]).expect("d tag"), + Tag::parse(["p", hex::encode(owner).as_str(), "", "owner"]) + .expect("owner p tag"), + Tag::parse(["p", hex::encode(member).as_str(), "", role]) + .expect("member p tag"), + ]) + .custom_created_at(Timestamp::from(timestamp)) + .sign_with_keys(&relay_keys) + .expect("sign roster") + }; + let base = Timestamp::now().as_secs(); + let fresh = roster("admin", base); + assert!( + db.replace_addressable_event(community, &fresh, Some(channel)) + .await + .expect("publish canonical role") + .1 + ); + let stale = roster("member", base + 1); + let error = db + .replace_addressable_event(community, &stale, Some(channel)) + .await + .expect_err("desired-state fence must reject stale role"); + assert!(matches!( + error, + DbError::Sqlx(sqlx::Error::Database(ref db_error)) + if db_error.code().as_deref() == Some("23514") + )); + let live_id: Vec = sqlx::query_scalar( + "SELECT id FROM events WHERE community_id=$1 AND channel_id=$2 \ + AND kind=39002 AND deleted_at IS NULL", + ) + .bind(community_uuid) + .bind(channel) + .fetch_one(&pool) + .await + .expect("load desired-state live roster"); + assert_eq!(live_id, fresh.id.as_bytes().to_vec()); + + drop_scratch_db(&admin, pool, &scratch_name).await; + } + #[tokio::test] #[ignore = "requires Postgres"] async fn nip_rs_replacement_hard_deletes_payload_and_watermark_rejects_replay() { diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 2c9824d4a5b..4c1e664fc85 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -1051,6 +1051,21 @@ mod tests { assert!(roster_fence.contains("roster_tag.tag_json->>3")); assert!(roster_fence.contains("snapshot_members IS DISTINCT FROM canonical_members")); assert!(roster_fence.contains("ERRCODE = '23514'")); + + // Fresh desired-state bootstrap must install the identical executable + // fence as migration 0032. CI and isolated relay startup use schema.sql + // without running migrations, so drift reopens rolling-deploy races. + fn extract_roster_fence(sql: &str) -> &str { + let fence_start = "CREATE OR REPLACE FUNCTION guard_channel_roster_snapshot()"; + let fence_end = " FOR EACH ROW EXECUTE FUNCTION guard_channel_roster_snapshot();"; + let start = sql.find(fence_start).expect("roster fence function"); + let relative_end = sql[start..].find(fence_end).expect("roster fence trigger"); + &sql[start..start + relative_end + fence_end.len()] + } + assert_eq!( + extract_roster_fence(roster_fence), + extract_roster_fence(desired_schema) + ); } #[test] diff --git a/schema/schema.sql b/schema/schema.sql index a6bc57e55c1..6e14e6be1bf 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -997,14 +997,14 @@ FOR EACH ROW EXECUTE FUNCTION refresh_channel_ttl_after_event_insert(); -- Old binaries already serialize kind 39002 replacement on the replacement -- advisory key. This trigger adds the channel-membership key at INSERT time, -- after that canonical key, and validates every p tag against the current --- active membership set. New binaries take both keys in the same order before --- capture and replacement. Thus old and new writers remain compatible during --- a rolling deploy. +-- active membership set and roles. New binaries take both keys in the same +-- order before capture and replacement. Thus old and new writers remain +-- compatible during a rolling deploy. CREATE OR REPLACE FUNCTION guard_channel_roster_snapshot() RETURNS TRIGGER AS $$ DECLARE - canonical_members BYTEA[]; - snapshot_members BYTEA[]; + canonical_members TEXT[]; + snapshot_members TEXT[]; BEGIN IF NEW.kind <> 39002 OR NEW.channel_id IS NULL THEN RETURN NEW; @@ -1015,34 +1015,44 @@ BEGIN 0 )); - SELECT COALESCE(array_agg(cm.pubkey ORDER BY cm.pubkey), ARRAY[]::BYTEA[]) + SELECT COALESCE( + array_agg(encode(cm.pubkey, 'hex') || ':' || cm.role::text ORDER BY cm.pubkey), + ARRAY[]::TEXT[] + ) INTO canonical_members FROM channel_members cm WHERE cm.community_id = NEW.community_id AND cm.channel_id = NEW.channel_id AND cm.removed_at IS NULL; - -- A roster is canonical only when every p tag is a 32-byte hex pubkey, - -- no member is duplicated, and the normalized tag set exactly matches the - -- active membership set. Ignoring malformed tags would let an otherwise - -- complete roster smuggle invalid discovery data through the fence. + -- A roster is canonical only when every p tag uses the emitted four-field + -- shape, contains a 32-byte hex pubkey and valid authoritative role, has no + -- duplicate members, and exactly matches the active membership rows. IF EXISTS ( SELECT 1 - FROM jsonb_array_elements(NEW.tags) tag - WHERE tag->>0 = 'p' - AND (jsonb_array_length(tag) < 2 OR COALESCE(tag->>1, '') !~ '^[0-9a-fA-F]{64}$') + FROM jsonb_array_elements(NEW.tags) AS roster_tag(tag_json) + WHERE roster_tag.tag_json->>0 = 'p' + AND ( + jsonb_array_length(roster_tag.tag_json) <> 4 + OR COALESCE(roster_tag.tag_json->>1, '') !~ '^[0-9a-fA-F]{64}$' + OR roster_tag.tag_json->>2 <> '' + OR COALESCE(roster_tag.tag_json->>3, '') NOT IN ('owner', 'admin', 'bot', 'member', 'guest') + ) ) THEN RAISE EXCEPTION 'kind 39002 roster contains an invalid p tag' USING ERRCODE = '23514'; END IF; - SELECT COALESCE(array_agg(member ORDER BY member), ARRAY[]::BYTEA[]) + SELECT COALESCE( + array_agg( + lower((roster_tag.tag_json->>1)) || ':' || (roster_tag.tag_json->>3) + ORDER BY decode((roster_tag.tag_json->>1), 'hex') + ), + ARRAY[]::TEXT[] + ) INTO snapshot_members - FROM ( - SELECT decode(tag->>1, 'hex') AS member - FROM jsonb_array_elements(NEW.tags) tag - WHERE tag->>0 = 'p' - ) snapshot; + FROM jsonb_array_elements(NEW.tags) AS roster_tag(tag_json) + WHERE roster_tag.tag_json->>0 = 'p'; IF snapshot_members IS DISTINCT FROM canonical_members THEN RAISE EXCEPTION 'kind 39002 roster does not match canonical channel membership' From f707f732cc59578819e590462ada38cd8ce69954 Mon Sep 17 00:00:00 2001 From: Wes Date: Tue, 18 Aug 2026 19:16:53 -0600 Subject: [PATCH 7/8] Fix pgschema partition trigger attachment pgschema copies parent triggers onto standalone event partitions. Remove the new roster snapshot trigger copy before attaching each partition so PostgreSQL can recreate the inherited trigger without a name collision. Co-authored-by: Carl Signed-off-by: Wes --- scripts/attach-schema-partitions.sql | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/scripts/attach-schema-partitions.sql b/scripts/attach-schema-partitions.sql index 5837676f842..a67bb706b1b 100644 --- a/scripts/attach-schema-partitions.sql +++ b/scripts/attach-schema-partitions.sql @@ -16,12 +16,12 @@ BEGIN ) THEN -- pgschema may copy parent triggers onto standalone children. Drop -- those copies before ATTACH; PostgreSQL recreates inherited parent - -- triggers while attaching and rejects same-named child triggers - -- (both the push-match trigger and the replica-fence floor guard). + -- triggers while attaching and rejects same-named child triggers. DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p_past; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p_past; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p_past; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p_past; + DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p_past; ALTER TABLE events ATTACH PARTITION events_p_past FOR VALUES FROM (MINVALUE) TO ('2026-01-01'); END IF; @@ -35,6 +35,7 @@ BEGIN DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_01; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_01; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_01; + DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_01; ALTER TABLE events ATTACH PARTITION events_p2026_01 FOR VALUES FROM ('2026-01-01') TO ('2026-02-01'); END IF; @@ -48,6 +49,7 @@ BEGIN DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_02; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_02; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_02; + DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_02; ALTER TABLE events ATTACH PARTITION events_p2026_02 FOR VALUES FROM ('2026-02-01') TO ('2026-03-01'); END IF; @@ -61,6 +63,7 @@ BEGIN DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_03; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_03; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_03; + DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_03; ALTER TABLE events ATTACH PARTITION events_p2026_03 FOR VALUES FROM ('2026-03-01') TO ('2026-04-01'); END IF; @@ -74,6 +77,7 @@ BEGIN DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_04; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_04; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_04; + DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_04; ALTER TABLE events ATTACH PARTITION events_p2026_04 FOR VALUES FROM ('2026-04-01') TO ('2026-05-01'); END IF; @@ -87,6 +91,7 @@ BEGIN DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_05; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_05; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_05; + DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_05; ALTER TABLE events ATTACH PARTITION events_p2026_05 FOR VALUES FROM ('2026-05-01') TO ('2026-06-01'); END IF; @@ -100,6 +105,7 @@ BEGIN DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_06; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_06; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_06; + DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_06; ALTER TABLE events ATTACH PARTITION events_p2026_06 FOR VALUES FROM ('2026-06-01') TO ('2026-07-01'); END IF; @@ -113,6 +119,7 @@ BEGIN DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p_future; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p_future; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p_future; + DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p_future; ALTER TABLE events ATTACH PARTITION events_p_future FOR VALUES FROM ('2026-07-01') TO (MAXVALUE); END IF; From be8ea0084f4d4c78c7c2550baad4399e4df8ce73 Mon Sep 17 00:00:00 2001 From: Wes Date: Wed, 19 Aug 2026 19:43:23 -0600 Subject: [PATCH 8/8] Fail closed without the channel roster fence Verify migration 0032 across the partitioned events catalog and through a rolled-back behavior probe before any relay listener or roster publisher starts. Cover pre-0032, inert-function, and partition-trigger failures, and document the schema-before-code rollout boundary. Co-authored-by: Carl <32a2e2c9d428ee08902cab75d956da2c1d235a22d4766b0dd4138bf6e2e5db1d@buzz.block.builderlab.xyz> Signed-off-by: Wes --- crates/buzz-db/src/channel.rs | 123 ++++++++++++++++++++++++++++ crates/buzz-db/src/lib.rs | 140 ++++++++++++++++++++++++++++++-- crates/buzz-db/src/migration.rs | 43 ++++++++-- crates/buzz-relay/src/main.rs | 18 +++- deploy/charts/buzz/README.md | 2 + 5 files changed, 309 insertions(+), 17 deletions(-) diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index 2f5e7cdbd7f..109a9367d7a 100644 --- a/crates/buzz-db/src/channel.rs +++ b/crates/buzz-db/src/channel.rs @@ -347,6 +347,129 @@ pub async fn set_canvas( /// `buzz_channel_ttl:`. const CHANNEL_MEMBERSHIP_LOCK_NAMESPACE: &str = "buzz_channel_membership:"; +/// Verify that migration 0032's roster fence is active on the partitioned +/// `events` parent and every attached partition. +/// +/// New roster publishers depend on this database-side guard to serialize with +/// legacy publishers during a rolling deployment. If the migration has not +/// been applied, publishing with the new lock protocol would falsely appear +/// safe while an old pod could still overwrite it with stale membership. +pub async fn verify_channel_roster_fence_catalog<'e>( + executor: impl sqlx::PgExecutor<'e>, +) -> Result<()> { + // tgtype bits: 1 = ROW, 2 = BEFORE, 4 = INSERT, 16 = UPDATE, 64 = INSTEAD. + // Required: ROW + BEFORE + INSERT set; UPDATE + INSTEAD clear. + let missing: Vec = sqlx::query_scalar( + r#" + SELECT n.nspname || '.' || c.relname + FROM ( + SELECT 'public.events'::regclass AS oid + UNION ALL + SELECT inhrelid FROM pg_inherits WHERE inhparent = 'public.events'::regclass + ) rels + JOIN pg_class c ON c.oid = rels.oid + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE NOT EXISTS ( + SELECT 1 FROM pg_trigger t + WHERE t.tgrelid = rels.oid + AND t.tgname = 'trg_events_guard_channel_roster_snapshot' + AND t.tgfoid = to_regprocedure('public.guard_channel_roster_snapshot()') + AND t.tgenabled IN ('O', 'A') + AND t.tgtype & 1 = 1 -- row-level + AND t.tgtype & 2 = 2 -- BEFORE + AND t.tgtype & 4 = 4 -- fires on INSERT + AND t.tgtype & 16 = 0 -- not UPDATE + AND t.tgtype & 64 = 0 -- not INSTEAD OF + ) + "#, + ) + .fetch_all(executor) + .await?; + if !missing.is_empty() { + return Err(DbError::InvalidData(format!( + "channel roster fence trigger missing, disabled, or mis-shaped on: {}", + missing.join(", ") + ))); + } + Ok(()) +} + +/// Prove migration 0032's roster fence semantics through the live writer pool. +/// +/// The catalog check cannot detect a no-op or otherwise corrupted trigger +/// function. This rolled-back probe verifies that a canonical empty roster is +/// accepted while a stale roster member is rejected with `check_violation`. +pub async fn verify_channel_roster_fence_behavior(pool: &sqlx::PgPool) -> Result<()> { + let mut tx = pool.begin().await?; + let community_id = Uuid::new_v4(); + let channel_id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_id) + .bind(format!( + "roster-fence-verify-{}.invalid", + community_id.simple() + )) + .execute(&mut *tx) + .await?; + + let insert = |id: Vec, tags: serde_json::Value| { + sqlx::query( + "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag) \ + VALUES ($1, $2, $3, NOW(), 39002, $4, '', $5, NOW(), $6, $7)", + ) + .bind(community_id) + .bind(id) + .bind(vec![0u8; 32]) + .bind(tags) + .bind(vec![0u8; 64]) + .bind(channel_id) + .bind(channel_id.to_string()) + }; + + insert( + vec![0u8; 32], + serde_json::json!([["d", channel_id.to_string()]]), + ) + .execute(&mut *tx) + .await + .map_err(|error| { + DbError::InvalidData(format!( + "channel roster fence rejected a canonical probe roster: {error}" + )) + })?; + + sqlx::query("SAVEPOINT roster_fence_probe") + .execute(&mut *tx) + .await?; + let stale = insert( + vec![1u8; 32], + serde_json::json!([ + ["d", channel_id.to_string()], + ["p", hex::encode([2u8; 32]), "", "member"] + ]), + ) + .execute(&mut *tx) + .await; + match stale { + Err(sqlx::Error::Database(error)) if error.code().as_deref() == Some("23514") => {} + Ok(_) => { + return Err(DbError::InvalidData( + "channel roster fence is inert: a stale probe roster was accepted".into(), + )); + } + Err(error) => { + return Err(DbError::InvalidData(format!( + "channel roster fence probe failed unexpectedly: {error}" + ))); + } + } + sqlx::query("ROLLBACK TO SAVEPOINT roster_fence_probe") + .execute(&mut *tx) + .await?; + tx.rollback().await?; + Ok(()) +} + /// Take the per-channel membership lock. MUST be the first statement in the /// transaction that then reads roles/owner counts and writes membership, so the /// whole check-then-write sequence is atomic against a concurrent one. diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 4d6f84ecf95..3ff230f9503 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -2390,6 +2390,13 @@ impl Db { channel::set_canvas(&self.pool, community_id, channel_id, canvas).await } + /// Verify the mixed-version channel-roster database fence end to end. + #[datastore_span(name = "verify_channel_roster_fence", system = "postgresql")] + pub async fn verify_channel_roster_fence(&self) -> Result<()> { + channel::verify_channel_roster_fence_catalog(&self.pool).await?; + channel::verify_channel_roster_fence_behavior(&self.pool).await + } + /// Capture the active roster while holding the membership-writer lock. #[datastore_span(name = "lock_member_snapshot", system = "postgresql")] pub async fn lock_member_snapshot( @@ -5509,6 +5516,113 @@ mod tests { id } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn unmigrated_roster_fence_blocks_startup_until_0032_is_applied() { + let admin = PgPool::connect(&admin_url().await) + .await + .expect("connect admin"); + let (pool, scratch_name) = + create_scratch_db_through(&admin, "roster_fence_unmigrated", Some(31)).await; + let db = Db::from_pool(pool.clone()); + + let error = db + .verify_channel_roster_fence() + .await + .expect_err("pre-0032 schema must block roster publishers"); + assert!( + error.to_string().contains("channel roster fence trigger"), + "startup gate must report the missing schema fence: {error}" + ); + let rows_before: i64 = sqlx::query_scalar("SELECT count(*) FROM events WHERE kind = 39002") + .fetch_one(&pool) + .await + .expect("count pre-migration rosters"); + assert_eq!( + rows_before, 0, + "failed startup gate must not publish a roster" + ); + + migration::run_migrations(&pool) + .await + .expect("apply migration 0032"); + db.verify_channel_roster_fence() + .await + .expect("0032 must open the startup gate"); + + drop_scratch_db(&admin, pool, &scratch_name).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn channel_roster_fence_behavior_verification_detects_inert_function() { + let admin = PgPool::connect(&admin_url().await) + .await + .expect("connect admin"); + let (pool, scratch_name) = create_scratch_db(&admin, "roster_fence_inert").await; + let db = Db::from_pool(pool.clone()); + + sqlx::raw_sql( + "CREATE OR REPLACE FUNCTION guard_channel_roster_snapshot() \ + RETURNS TRIGGER AS $$ BEGIN RETURN NEW; END; $$ LANGUAGE plpgsql;", + ) + .execute(&pool) + .await + .expect("replace roster fence with inert body"); + let error = db + .verify_channel_roster_fence() + .await + .expect_err("inert roster fence must fail closed"); + assert!( + error + .to_string() + .contains("stale probe roster was accepted"), + "behavior probe must identify inert semantics: {error}" + ); + + drop_scratch_db(&admin, pool, &scratch_name).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn channel_roster_fence_catalog_verification_fails_closed() { + let admin = PgPool::connect(&admin_url().await) + .await + .expect("connect admin"); + let (pool, scratch_name) = create_scratch_db(&admin, "roster_fence_catalog").await; + let db = Db::from_pool(pool.clone()); + + db.verify_channel_roster_fence() + .await + .expect("migrated roster fence must verify"); + + let child: String = sqlx::query_scalar( + "SELECT n.nspname || '.' || c.relname \ + FROM pg_inherits i JOIN pg_class c ON c.oid = i.inhrelid \ + JOIN pg_namespace n ON n.oid = c.relnamespace \ + WHERE i.inhparent = 'public.events'::regclass ORDER BY i.inhrelid LIMIT 1", + ) + .fetch_one(&pool) + .await + .expect("load event partition"); + sqlx::query(sqlx::AssertSqlSafe(format!( + "ALTER TABLE {child} DISABLE TRIGGER trg_events_guard_channel_roster_snapshot" + ))) + .execute(&pool) + .await + .expect("disable partition roster trigger"); + let error = db + .verify_channel_roster_fence() + .await + .expect_err("disabled partition roster fence must fail closed"); + assert!( + error.to_string().contains(&child), + "verification must identify the unfenced partition: {error}" + ); + + drop_scratch_db(&admin, pool, &scratch_name).await; + } + #[tokio::test] #[ignore = "requires Postgres"] async fn addressable_replacement_rolls_back_when_mention_indexing_fails() { @@ -7179,9 +7293,12 @@ mod tests { std::env::var("TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.into()) } - /// Create a fresh scratch database on the same server and run migrations. - /// Returns (pool, db_name); callers should `drop_scratch_db` when done. - async fn create_scratch_db(admin: &PgPool, prefix: &str) -> (PgPool, String) { + /// Create a fresh scratch database on the same server and optionally run migrations. + async fn create_scratch_db_through( + admin: &PgPool, + prefix: &str, + target: Option, + ) -> (PgPool, String) { let name = format!("{}_{}", prefix, Uuid::new_v4().simple()); sqlx::query(sqlx::AssertSqlSafe(format!("CREATE DATABASE {name}"))) .execute(admin) @@ -7196,12 +7313,23 @@ mod tests { let pool = PgPool::connect(&scratch_url) .await .expect("connect scratch db"); - migration::run_migrations(&pool) - .await - .expect("migrate scratch db"); + match target { + Some(target) => migration::run_migrations_through(&pool, target) + .await + .expect("migrate scratch db through target"), + None => migration::run_migrations(&pool) + .await + .expect("migrate scratch db"), + } (pool, name) } + /// Create a fresh scratch database on the same server and run all migrations. + /// Returns (pool, db_name); callers should `drop_scratch_db` when done. + async fn create_scratch_db(admin: &PgPool, prefix: &str) -> (PgPool, String) { + create_scratch_db_through(admin, prefix, None).await + } + async fn drop_scratch_db(admin: &PgPool, pool: PgPool, name: &str) { pool.close().await; let _ = sqlx::query(sqlx::AssertSqlSafe(format!( diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 4c1e664fc85..94c7aea2faf 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -32,6 +32,20 @@ pub async fn run_migrations(pool: &PgPool) -> Result<()> { .await } +#[cfg(test)] +pub(crate) async fn run_migrations_through(pool: &PgPool, target: i64) -> Result<()> { + with_exclusive_schema_destruction_lock(pool, |mut conn| async move { + let outcome = async { + reject_legacy_nip_rs_cardinality_ambiguity(&mut conn).await?; + MIGRATOR.run_to(target, &mut conn).await?; + Ok(()) + } + .await; + (conn, outcome) + }) + .await +} + async fn run_migrations_locked(conn: &mut PgConnection) -> Result<()> { reject_legacy_nip_rs_cardinality_ambiguity(conn).await?; MIGRATOR.run(&mut *conn).await?; @@ -43,6 +57,7 @@ async fn run_migrations_locked(conn: &mut PgConnection) -> Result<()> { // guard, so migration fails closed if any is missing. (The fence probe // re-runs this same check at startup on non-migrating relays.) crate::replica_fence::verify_floor_guard_catalog(&mut *conn).await?; + crate::channel::verify_channel_roster_fence_catalog(&mut *conn).await?; Ok(()) } @@ -1254,6 +1269,7 @@ mod tests { // Build the needles so this test's own source never matches them. let migrate_macro = ["sqlx", "::migrate!"].concat(); let migrator_run = ["MIGRATOR", ".run("].concat(); + let migrator_run_to = ["MIGRATOR", ".run_to("].concat(); let manifest_dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")); let this_file = manifest_dir.join("src/migration.rs"); @@ -1280,23 +1296,24 @@ mod tests { rust_sources(crates_dir, &mut files); for path in &files { let source = std::fs::read_to_string(path).expect("read rust source"); - let (macro_hits, run_hits) = ( + let (macro_hits, run_hits, run_to_hits) = ( count(&source, &migrate_macro), count(&source, &migrator_run), + count(&source, &migrator_run_to), ); if *path == this_file { assert_eq!( - (macro_hits, run_hits), - (1, 1), - "migration.rs must embed the migrator once and run it exactly once, \ - inside the locked wrapper" + (macro_hits, run_hits, run_to_hits), + (1, 1, 1), + "migration.rs must embed the migrator once, run it once in production, \ + and expose exactly one test-only bounded run" ); } else if *path == push_gateway_exception { continue; } else { assert_eq!( - (macro_hits, run_hits), - (0, 0), + (macro_hits, run_hits, run_to_hits), + (0, 0, 0), "{} embeds or runs a SQLx migrator outside the schema/destruction \ lock contract; route migration execution through \ buzz_db migration::run_migrations", @@ -1319,13 +1336,23 @@ mod tests { .find("async fn with_exclusive_schema_destruction_lock") .expect("exclusive lock wrapper"); let run_site = source.find(&migrator_run).expect("migrator run site"); + let run_to_site = source + .find(&migrator_run_to) + .expect("bounded test migrator run site"); assert!( source[entry..locked].contains("with_exclusive_schema_destruction_lock("), "run_migrations must delegate through the exclusive schema/destruction lock" ); assert!( run_site > locked && run_site < wrapper, - "the migrator run site must live inside run_migrations_locked" + "the production migrator run site must live inside run_migrations_locked" + ); + assert!( + run_to_site > entry + && run_to_site < locked + && source[entry..run_to_site].contains("#[cfg(test)]") + && source[entry..run_to_site].contains("with_exclusive_schema_destruction_lock("), + "the bounded migrator run must remain test-only and use the exclusive lock wrapper" ); assert!( source[wrapper..].contains("pg_advisory_lock($1)") diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index e383946d262..566b684f830 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -534,10 +534,22 @@ async fn main() -> anyhow::Result<()> { ); } + match state.db.verify_channel_roster_fence().await { + Ok(()) => { + info!("Channel roster fence verified"); + } + Err(error) => { + error!(%error, "Channel roster fence validation failed"); + return Err(anyhow::anyhow!( + "Channel roster fence is unsafe; apply or repair migration 0032 before starting this relay: {error}" + )); + } + } + // Repair legacy NIP-29 channel rosters that were persisted while the - // canonical member query still truncated at 1,000 rows. This runs before - // listeners open, is tenant-preserving across shared deployments, and - // rewrites only mismatched large-channel snapshots. + // canonical member query still truncated at 1,000 rows. Validation above + // makes migration 0032 a code/schema compatibility gate before the new + // replacement protocol or listener can serve traffic. match buzz_relay::handlers::side_effects::reconcile_large_channel_member_snapshots(&state).await { Ok(count) if count > 0 => info!(count, "large channel member snapshots repaired"), diff --git a/deploy/charts/buzz/README.md b/deploy/charts/buzz/README.md index 86989676604..30cee4f4063 100644 --- a/deploy/charts/buzz/README.md +++ b/deploy/charts/buzz/README.md @@ -205,6 +205,8 @@ default so long-lived WebSocket connections have time to drain. Schema migrations are embedded in the relay binary via `sqlx::migrate!` and run at startup, gated by `BUZZ_AUTO_MIGRATE` (default `true`). Multiple replicas race-safely behind a Postgres advisory lock. `helm upgrade` is the entire upgrade procedure. +Migration 0032 is a hard compatibility boundary for relay versions that publish repaired channel rosters. The relay verifies the roster-fence trigger catalog and behavior before opening listeners and refuses to start if 0032 is missing or inert. Apply migrations before rolling the relay; for large installations, prefer a controlled `buzz-admin migrate` job with PostgreSQL lock monitoring before the code rollout. + If you prefer decoupling migrations from serving, set `migrate.autoMigrate=false`. **In that mode the chart does not run migrations for you** — you own running `buzz-admin migrate` (separate Pod / one-shot Job) against the database before every `helm install` / `helm upgrade`. Readiness probes only verify DB connectivity, not schema freshness, so a pod will appear healthy against an unmigrated schema and fail under load. A pre-upgrade Helm Job for this is on the chart roadmap; the values knob `migrate.preUpgradeJob.enabled` is reserved. ## Backups