diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index 8035ab58adb..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. @@ -366,6 +489,179 @@ async fn acquire_channel_membership_lock( Ok(()) } +/// An active member roster captured while holding the channel's membership +/// serialization lock on one writer connection. +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>, +} + +impl LockedMemberSnapshot { + /// 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)> { + 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) + .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(()) + } +} + +/// 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, + 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#" + 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, + community_id, + channel_id, + relay_pubkey: relay_pubkey.to_vec(), + tx, + }) +} + /// Add a member to a channel. /// /// Role enforcement: @@ -781,6 +1077,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, @@ -1535,6 +1906,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 @@ -2016,6 +2388,188 @@ 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(); + + // 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, $4, '', $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()) + .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("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. + 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"] @@ -2499,6 +3053,96 @@ 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_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 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"), + 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 = + 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 330525d310d..3ff230f9503 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], @@ -2390,6 +2390,24 @@ 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( + &self, + community_id: CommunityId, + channel_id: Uuid, + relay_pubkey: &[u8], + ) -> Result { + channel::lock_member_snapshot(&self.pool, community_id, channel_id, relay_pubkey).await + } + /// Adds a member to a channel. #[datastore_span(name = "add_member", system = "postgresql")] pub async fn add_member( @@ -2476,6 +2494,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( @@ -5480,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() { @@ -5493,13 +5636,14 @@ 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"), - 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(); @@ -5561,6 +5705,238 @@ 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 a role-only canonical mutation advances and the new pod publishes B. + let base = Timestamp::now().as_secs(); + 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, role)| { + Tag::parse(["p", hex::encode(member).as_str(), "", *role]).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 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("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(), "owner"), (newcomer.as_slice(), "admin")], + 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 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() { @@ -6917,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) @@ -6934,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 be87faa1ac2..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(()) } @@ -625,7 +640,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 +1051,36 @@ 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 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'")); + + // 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] @@ -1224,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"); @@ -1250,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", @@ -1289,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/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index f2b58937ab5..89595fbee17 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1049,6 +1049,55 @@ fn group_members_tags(group_id: &str, members: &[MemberRecord]) -> anyhow::Resul Ok(tags) } +async fn store_group_members_event( + tenant: &TenantContext, + state: &Arc, + channel_id: Uuid, + member_snapshot: &mut buzz_db::channel::LockedMemberSnapshot, +) -> anyhow::Result> { + let group_id = channel_id.to_string(); + 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. /// Called after group creation, metadata changes, or membership changes. /// Events are stored channel-scoped (`channel_id = Some(...)`) so that existing @@ -1151,18 +1200,18 @@ 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, - ) + // 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, &relay_pubkey) .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(()) } @@ -3052,6 +3101,68 @@ 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; + // 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 mut member_snapshot = state + .db + .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 = + 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; + + 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..566b684f830 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -534,6 +534,31 @@ 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. 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"), + 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 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 diff --git a/migrations/0032_channel_roster_snapshot_fence.sql b/migrations/0032_channel_roster_snapshot_fence.sql new file mode 100644 index 00000000000..cdc7bc4b93e --- /dev/null +++ b/migrations/0032_channel_roster_snapshot_fence.sql @@ -0,0 +1,76 @@ +-- 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 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 TEXT[]; + snapshot_members TEXT[]; +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(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 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) 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( + 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 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' + 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..6e14e6be1bf 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -990,6 +990,85 @@ 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 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 TEXT[]; + snapshot_members TEXT[]; +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(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 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) 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( + 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 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' + 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 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;