diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index c78a99eab5b..470337f8733 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -777,7 +777,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 80); + assert_eq!(migrations.len(), 81); for (index, migration) in migrations.iter().enumerate() { assert_eq!(migration.version, (index + 1) as i64); } @@ -785,6 +785,19 @@ mod postgres_tests { .sql .as_str() .contains("ADD COLUMN auto_join BOOLEAN NOT NULL DEFAULT false")); + assert_eq!(migrations[80].version, 81); + let retire_roster_fence = strip_sql_comments(migrations[80].sql.as_str()); + for statement in [ + "SET LOCAL lock_timeout = '1s';", + "SET LOCAL statement_timeout = '5s';", + "DROP TRIGGER trg_events_guard_channel_roster_snapshot ON events;", + "DROP FUNCTION guard_channel_roster_snapshot();", + ] { + assert!( + retire_roster_fence.contains(statement), + "0081 must run {statement}" + ); + } assert!(migrations[70] .sql .as_str() @@ -1336,31 +1349,8 @@ mod postgres_tests { 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. - // The desired state keeps the function and its trigger in separate - // files, so compare each statement rather than one contiguous span. - fn extract_span<'a>(sql: &'a str, start: &str, end: &str) -> &'a str { - let begin = sql.find(start).expect("span start"); - let relative_end = sql[begin..].find(end).expect("span end"); - &sql[begin..begin + relative_end + end.len()] - } - for (start, end) in [ - ( - "CREATE OR REPLACE FUNCTION guard_channel_roster_snapshot()", - "$$ LANGUAGE plpgsql;", - ), - ( - "CREATE TRIGGER trg_events_guard_channel_roster_snapshot", - " FOR EACH ROW EXECUTE FUNCTION guard_channel_roster_snapshot();", - ), - ] { - assert_eq!( - extract_span(roster_fence, start, end), - extract_span(&desired_schema, start, end) - ); - } + // Migration 0081 retires this fence; the desired state never installs it. + assert!(!desired_schema.contains("guard_channel_roster_snapshot")); // The single-row heartbeat table is updated continuously. Prevent // autovacuum from truncating its heap so standby queries are not diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 2c139228a8e..4ed7d14dfce 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -49,53 +49,6 @@ pub struct MemberRecord { /// `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. -/// -/// Only tests use this. The relay no longer requires the trigger at startup or -/// after migrating: [`validate_channel_roster_snapshot`] enforces the same rule -/// in every roster writer, so the trigger can be dropped. -#[cfg(test)] -pub(crate) 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(()) -} - /// 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. @@ -118,7 +71,8 @@ pub(crate) async fn acquire_channel_membership_lock( Ok(()) } -/// Check the same p-tag shape and canonical active membership as migration 0032. +/// Check that a kind:39002 roster has canonical p tags and names exactly the +/// active members, each with its role. /// The caller holds this channel's membership advisory lock on the admitted /// transaction before calling. Every writer that changes the membership of an /// existing channel takes that key before reading roles or changing rows @@ -126,14 +80,11 @@ pub(crate) async fn acquire_channel_membership_lock( /// including both kick paths /// (`execute_kick_with_marker` and `deploy_kick_member` in /// relay_admin_actions.rs), so no removal can commit between this check and -/// the roster commit. -/// Legacy replacement takes the replacement key, locks -/// old-head rows in its UPDATE, then takes the membership key at trigger INSERT. -/// The app path takes replacement key -> membership key -> old-head row locks, -/// removing the row-lock/membership-key cycle rather than preserving it. +/// the roster commit. The replacement writer takes replacement key -> +/// membership key -> old-head row locks. /// -/// On failure, DO raises the trigger's exact PostgreSQL error during mixed-version -/// rollout. Phase 2 can replace this with a typed DbError once legacy pods retire. +/// On failure, DO raises the PostgreSQL error that migration 0032's retired +/// trigger raised, so callers see the same SQLSTATE and message. pub(crate) async fn validate_channel_roster_snapshot( tx: &mut AdmittedTx, channel_id: Uuid, @@ -187,9 +138,9 @@ pub(crate) async fn validate_channel_roster_snapshot( )) } -/// Apply the app-side roster fence to a general event writer. The trigger -/// reacquires this transaction's advisory key until the fleet has upgraded; -/// PostgreSQL advisory locks are reentrant for their current owner. +/// Apply the roster fence to a general event writer. A caller that already +/// holds the membership key takes it again; PostgreSQL advisory locks are +/// reentrant for their current owner. pub(crate) async fn guard_channel_roster_event( tx: &mut AdmittedTx, kind: i32, @@ -559,10 +510,9 @@ pub async fn lock_member_snapshot( crate::observability::WriterOperation::EventWrite, ) .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. + // Match the canonical replacement writer's lock order: the replacement + // key, then the membership key (`guard_channel_roster_event` takes it + // again before the insert; advisory locks are reentrant). let replacement_lock = crate::replaceable::event_replacement_lock_key( community_id, 39002, @@ -2983,7 +2933,7 @@ mod postgres_tests { .await .expect("insert large roster"); - // Migration 0032's roster guard requires canonical four-field p tags + // The roster check requires canonical four-field p tags // whose roles exactly match channel_members, including the creator's // owner row created by create_test_channel. let creator_hex = hex::encode(&creator); @@ -3027,9 +2977,8 @@ mod postgres_tests { .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. + // with UPDATE to model a stale snapshot that predates the roster + // fence. Writers cannot insert new stale snapshots. sqlx::query( r#" INSERT INTO events @@ -4640,39 +4589,6 @@ mod postgres_tests { (community, channel.id, owner_a, owner_b) } - /// Commits `DISABLE TRIGGER` on the test database. Safe only under the - /// per-test database wrapper; do not run these against a shared database, - /// or later trigger-dependent tests there will see no trigger. - async fn disable_roster_trigger(pool: &PgPool) { - let partitions: Vec = sqlx::query_scalar( - "SELECT quote_ident(n.nspname) || '.' || quote_ident(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", - ) - .fetch_all(pool) - .await - .expect("list event partitions"); - for partition in partitions { - sqlx::query(sqlx::AssertSqlSafe(format!( - "ALTER TABLE {partition} DISABLE TRIGGER trg_events_guard_channel_roster_snapshot" - ))) - .execute(pool) - .await - .expect("disable only the roster trigger on partition"); - } - sqlx::query( - "ALTER TABLE ONLY events DISABLE TRIGGER trg_events_guard_channel_roster_snapshot", - ) - .execute(pool) - .await - .expect("disable only the roster trigger on parent"); - assert!( - verify_channel_roster_fence_catalog(pool).await.is_err(), - "app-only tests must run without the catalog-verified roster fence" - ); - } - fn signed_roster( keys: &Keys, channel_id: Uuid, @@ -4694,7 +4610,6 @@ mod postgres_tests { #[ignore = "requires Postgres"] async fn app_roster_guard_locked_writer_rejects_stale_and_accepts_current_without_trigger() { let pool = setup_pool().await; - disable_roster_trigger(&pool).await; let community_uuid = make_test_community(&pool).await; let community = CommunityId::from_uuid(community_uuid); let owner = random_pubkey(); @@ -4752,14 +4667,11 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] - async fn app_roster_guard_matches_trigger_for_roster_input_table() { + async fn app_roster_guard_verdicts_for_roster_input_table() { use nostr::{EventBuilder, Kind, Tag}; use serde_json::json; let pool = setup_pool().await; - verify_channel_roster_fence_catalog(&pool) - .await - .expect("trigger side must start with the real fence enabled"); let owner = random_pubkey(); let owner_hex = hex::encode(&owner); let extra_hex = hex::encode(random_pubkey()); @@ -4866,10 +4778,7 @@ mod postgres_tests { error => panic!("expected a database verdict, got {error:?}"), }; let keys = Keys::generate(); - let mut measured = Vec::new(); - // Measure all trigger verdicts first with raw inserts, bypassing every - // app guard. Each savepoint and transaction is rolled back so the app - // later sees exactly the same canonical membership and no event head. + let db = Db::from_pool(pool.clone()); for (name, tags, role, expected_error) in cases { let community_uuid = make_test_community(&pool).await; let community = CommunityId::from_uuid(community_uuid); @@ -4907,52 +4816,17 @@ mod postgres_tests { .tags(tags.iter().map(|tag| Tag::parse(tag).expect("case tag"))) .sign_with_keys(&keys) .expect("sign case"); - let mut tx = pool.begin().await.expect("trigger transaction"); - sqlx::query("SAVEPOINT roster_case") - .execute(&mut *tx) - .await - .expect("case savepoint"); - let trigger = 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, 39002, $5, '', $6, NOW(), $7, $8)", - ) - .bind(community_uuid) - .bind(event.id.as_bytes().as_slice()) - .bind(event.pubkey.to_bytes().as_slice()) - .bind(DateTime::from_timestamp(event.created_at.as_secs() as i64, 0).expect("timestamp")) - .bind(serde_json::to_value(&event.tags).expect("tags")) - .bind(event.sig.serialize().as_slice()) - .bind(channel.id) - .bind(channel.id.to_string()) - .execute(&mut *tx) - .await - .map(|result| result.rows_affected() == 1) - .map_err(database_error); - sqlx::query("ROLLBACK TO SAVEPOINT roster_case") - .execute(&mut *tx) - .await - .expect("undo trigger case"); - tx.rollback().await.expect("undo trigger transaction"); - let expected = - expected_error.map_or(Ok(true), |message| Err(("23514".into(), message.into()))); - assert_eq!(trigger, expected, "trigger contract: {name}"); - measured.push((name, community, channel.id, event, trigger)); - } - - // Only the app side runs with the trigger disabled; this helper also - // asserts that the retained catalog check now rejects the database. - disable_roster_trigger(&pool).await; - let db = Db::from_pool(pool.clone()); - for (name, community, channel, event, trigger) in measured { let app = db - .replace_addressable_event(community, &event, Some(channel)) + .replace_addressable_event(community, &event, Some(channel.id)) .await .map(|(_, inserted)| inserted) .map_err(|error| match error { DbError::Sqlx(error) => database_error(error), error => panic!("expected app database verdict for {name}, got {error:?}"), }); - assert_eq!(app, trigger, "app/trigger parity: {name}"); + let expected = + expected_error.map_or(Ok(true), |message| Err(("23514".into(), message.into()))); + assert_eq!(app, expected, "app roster check: {name}"); } } @@ -4960,7 +4834,6 @@ mod postgres_tests { #[ignore = "requires Postgres"] async fn app_roster_guard_covers_generic_and_parameterized_writers_without_trigger() { let pool = setup_pool().await; - disable_roster_trigger(&pool).await; let community_uuid = make_test_community(&pool).await; let community = CommunityId::from_uuid(community_uuid); let owner = random_pubkey(); @@ -5002,7 +4875,6 @@ mod postgres_tests { #[ignore = "requires Postgres"] async fn app_roster_guard_rechecks_after_concurrent_membership_commit_without_trigger() { let pool = setup_pool().await; - disable_roster_trigger(&pool).await; let community_uuid = make_test_community(&pool).await; let community = CommunityId::from_uuid(community_uuid); let owner = random_pubkey(); @@ -5593,165 +5465,72 @@ mod postgres_tests { .expect("create channel"); } - /// The relay must run without the 0032 roster trigger: the app-side check - /// in every roster writer is the fence, and a later migration drops the - /// trigger. Migrating a database that has no trigger must succeed. + /// Migration 0081 retires the 0032 roster trigger. Before it, the trigger + /// and its function are installed; after it, both are gone and the + /// database accepts a raw kind:39002 insert that disagrees with + /// membership, because only the app-side check guards rosters now. /// - /// Mutation: restore the catalog check after `MIGRATOR.run` in - /// `run_migrations_locked` → migrating fails → RED. + /// Mutation: remove the DROP statements from 0081 → the trigger refuses + /// the insert → RED. #[tokio::test] #[ignore = "requires Postgres"] - async fn migrations_run_without_the_roster_trigger() { - let (pool, _scratch) = - ScratchDb::migrated(&crate::test_support::database_url(), "roster_fence_absent").await; - sqlx::raw_sql( - "DROP TRIGGER trg_events_guard_channel_roster_snapshot ON events; \ - DROP FUNCTION guard_channel_roster_snapshot();", - ) - .execute(&pool) - .await - .expect("drop the roster trigger"); - assert!(verify_channel_roster_fence_catalog(&pool).await.is_err()); - - migration::run_migrations(&pool) + async fn migration_0081_retires_the_roster_trigger() { + use nostr::{EventBuilder, Keys, Kind, Tag}; + + async fn roster_trigger_installed(pool: &PgPool) -> bool { + sqlx::query_scalar( + "SELECT to_regprocedure('public.guard_channel_roster_snapshot()') IS NOT NULL \ + OR EXISTS (SELECT 1 FROM pg_trigger \ + WHERE tgname = 'trg_events_guard_channel_roster_snapshot')", + ) + .fetch_one(pool) .await - .expect("migrating must not require the roster trigger"); - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn desired_schema_rejects_stale_legacy_roster_role() { - use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + .expect("look up the roster trigger") + } - let scratch = - ScratchDb::create(&crate::test_support::database_url(), "schema_roster_role").await; - let pool = PgPoolOptions::new() - .max_connections(1) - .connect(scratch.url()) + let (pool, _scratch) = ScratchDb::migrated_through( + &crate::test_support::database_url(), + "roster_fence_retired", + 80, + ) + .await; + assert!( + roster_trigger_installed(&pool).await, + "before 0081 the trigger is installed" + ); + migration::run_migrations(&pool) .await - .expect("connect desired-schema scratch db"); - sqlx::raw_sql(sqlx::AssertSqlSafe( - crate::test_support::desired_state_schema_sql(), - )) - .execute(&pool) - .await - .expect("apply desired-state schema"); + .expect("apply migration 0081"); + assert!( + !roster_trigger_installed(&pool).await, + "0081 must drop the roster trigger and its function" + ); - 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(); + let stranger = Keys::generate().public_key().to_hex(); + let event = EventBuilder::new(Kind::Custom(39002), "") + .tags(vec![ + Tag::parse(["d", channel.to_string().as_str()]).expect("d tag"), + Tag::parse(["p", stranger.as_str(), "", "owner"]).expect("p tag"), + ]) + .sign_with_keys(&Keys::generate()) + .expect("sign roster"); sqlx::query( - "INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) \ - VALUES ($1, $2, $3, 'admin', $4)", + "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_uuid) + .bind(event.id.as_bytes().as_slice()) + .bind(event.pubkey.to_bytes().as_slice()) + .bind(serde_json::to_value(&event.tags).expect("tags")) + .bind(event.sig.serialize().as_slice()) .bind(channel) - .bind(member.as_slice()) - .bind(owner.as_slice()) + .bind(channel.to_string()) .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); - // Reproduce the old binary on one connection: replacement key -> - // old-head UPDATE -> raw INSERT. Do not use the app API, whose guard - // rejects before the retained trigger can exercise this legacy fence. - let error = async { - let mut tx = pool.begin().await.expect("begin legacy replacement"); - let relay_pubkey = relay_keys.public_key().to_bytes(); - sqlx::query("SELECT pg_advisory_xact_lock($1)") - .bind(crate::store::replaceable::event_replacement_lock_key( - community, - 39002, - &relay_pubkey, - Some(channel.as_bytes().as_slice()), - )) - .execute(&mut *tx) - .await - .expect("legacy replacement key"); - let deleted = sqlx::query( - "UPDATE events SET deleted_at = NOW() \ - WHERE community_id=$1 AND kind=39002 AND pubkey=$2 \ - AND channel_id=$3 AND deleted_at IS NULL", - ) - .bind(community_uuid) - .bind(relay_pubkey.as_slice()) - .bind(channel) - .execute(&mut *tx) - .await - .expect("legacy soft-delete of current head"); - assert_eq!(deleted.rows_affected(), 1, "legacy path must delete the live head"); - let result = 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, 39002, $5, $6, $7, NOW(), $8, $9)", - ) - .bind(community_uuid) - .bind(stale.id.as_bytes().as_slice()) - .bind(relay_pubkey.as_slice()) - .bind(chrono::DateTime::from_timestamp(stale.created_at.as_secs() as i64, 0).expect("timestamp")) - .bind(serde_json::to_value(&stale.tags).expect("tags")) - .bind(&stale.content) - .bind(stale.sig.serialize().as_slice()) - .bind(channel) - .bind(channel.to_string()) - .execute(&mut *tx) - .await; - tx.rollback().await.expect("restore live head after legacy attempt"); - result - } - .await - .expect_err("desired-state fence must reject stale role"); - assert!( - matches!(error, sqlx::Error::Database(ref db_error) - if db_error.code().as_deref() == Some("23514") - && db_error.message() == "kind 39002 roster does not match canonical channel membership"), - "expected trigger roster fence check violation, got {error:?}" - ); - - 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()); - let stale_rows: i64 = - sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1 AND id=$2") - .bind(community_uuid) - .bind(stale.id.as_bytes().as_slice()) - .fetch_one(&pool) - .await - .expect("count rejected stale roster"); - assert_eq!(stale_rows, 0, "legacy stale insert must roll back"); + .expect("without the trigger, a raw roster insert is not checked"); } } diff --git a/crates/buzz-db/src/store/replaceable.rs b/crates/buzz-db/src/store/replaceable.rs index 7c1789c3c49..743e6c529dc 100644 --- a/crates/buzz-db/src/store/replaceable.rs +++ b/crates/buzz-db/src/store/replaceable.rs @@ -784,173 +784,6 @@ mod postgres_tests { assert_eq!(new_rows, 0, "new roster must roll back with its index"); } - #[tokio::test] - #[ignore = "requires Postgres"] - async fn stale_legacy_roster_cannot_replace_new_locked_snapshot() { - use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; - - let (setup_pool, scratch) = - ScratchDb::migrated(&admin_url().await, "mixed_roster_writer").await; - 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(&fresh_b) - .await - .expect("new writer publishes B") - .1 - ); - snapshot - .release() - .await - .expect("commit B and release locks"); - - // Reproduce the old binary on one connection: replacement key -> - // old-head UPDATE -> raw INSERT. Do not use the app API, whose guard - // rejects before the retained trigger can exercise this legacy fence. - // A size-one pool and bounded attempt also catch self-deadlock. - let error = tokio::time::timeout(Duration::from_secs(3), async { - let mut tx = pool.begin().await.expect("begin legacy replacement"); - let relay_pubkey = relay_keys.public_key().to_bytes(); - sqlx::query("SELECT pg_advisory_xact_lock($1)") - .bind(crate::store::replaceable::event_replacement_lock_key( - community, - 39002, - &relay_pubkey, - Some(channel.as_bytes().as_slice()), - )) - .execute(&mut *tx) - .await - .expect("legacy replacement key"); - let deleted = sqlx::query( - "UPDATE events SET deleted_at = NOW() \ - WHERE community_id=$1 AND kind=39002 AND pubkey=$2 \ - AND channel_id=$3 AND deleted_at IS NULL", - ) - .bind(community_uuid) - .bind(relay_pubkey.as_slice()) - .bind(channel) - .execute(&mut *tx) - .await - .expect("legacy soft-delete of current head"); - assert_eq!(deleted.rows_affected(), 1, "legacy path must delete the live head"); - let result = 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, 39002, $5, $6, $7, NOW(), $8, $9)", - ) - .bind(community_uuid) - .bind(stale_a.id.as_bytes().as_slice()) - .bind(relay_pubkey.as_slice()) - .bind(chrono::DateTime::from_timestamp(stale_a.created_at.as_secs() as i64, 0).expect("timestamp")) - .bind(serde_json::to_value(&stale_a.tags).expect("tags")) - .bind(&stale_a.content) - .bind(stale_a.sig.serialize().as_slice()) - .bind(channel) - .bind(channel.to_string()) - .execute(&mut *tx) - .await; - tx.rollback().await.expect("restore live head after legacy attempt"); - result - }) - .await - .expect("legacy replacement must not deadlock") - .expect_err("stale captured roster A must be rejected"); - assert!( - matches!(error, sqlx::Error::Database(ref db_error) - if db_error.code().as_deref() == Some("23514") - && db_error.message() == "kind 39002 roster does not match canonical channel membership"), - "expected trigger 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"); - } - #[tokio::test] #[ignore = "requires Postgres"] async fn nip_rs_replacement_hard_deletes_payload_and_watermark_rejects_replay() { diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 8f2977bb04e..a10b74d04e6 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1405,7 +1405,7 @@ async fn store_group_members_event( // relay's moderation-DM key participates in the {relay, recipient} DM used // for moderation notices) MUST retain the relay's own `p` tag. nostr's // default `build_with_ctx` strips any `p` tag matching the signer, which - // would drop the relay from the snapshot and fail migration 0032's roster + // would drop the relay from the snapshot and fail the roster // fence against the canonical two-member DM. `allow_self_tagging` keeps the // snapshot faithful to `channel_members`. let event = EventBuilder::new(Kind::Custom(KIND_NIP29_GROUP_MEMBERS as u16), "") diff --git a/migrations/0081_retire_channel_roster_trigger.sql b/migrations/0081_retire_channel_roster_trigger.sql new file mode 100644 index 00000000000..4c077739f97 --- /dev/null +++ b/migrations/0081_retire_channel_roster_trigger.sql @@ -0,0 +1,20 @@ +-- Every kind:39002 roster writer checks the roster against canonical channel +-- membership in the application (validate_channel_roster_snapshot, from +-- #8316), and since #8611 no relay requires this trigger at startup. Retire +-- the trigger that repeats that check: it compares roster labels with the +-- stored role exactly, so it would refuse rosters that show the computed +-- `member`/`bot` label. Every relay and rollback target must run #8611 or +-- later before this migration runs; older relays refuse to start without the +-- trigger. Published migrations remain checksum-frozen. +-- +-- DROP TRIGGER takes ACCESS EXCLUSIVE on each partition of events in turn, +-- and keeps each lock while it waits for the next. While the locks are queued +-- or held, event reads and writes both block. +-- lock_timeout bounds each lock wait and statement_timeout bounds each whole +-- statement, as in 0049. Either timeout rolls the migration back with no +-- change, and it can be retried. +SET LOCAL lock_timeout = '1s'; +SET LOCAL statement_timeout = '5s'; + +DROP TRIGGER trg_events_guard_channel_roster_snapshot ON events; +DROP FUNCTION guard_channel_roster_snapshot(); diff --git a/schema/functions/guard_channel_roster_snapshot.sql b/schema/functions/guard_channel_roster_snapshot.sql deleted file mode 100644 index bcd5c9e534b..00000000000 --- a/schema/functions/guard_channel_roster_snapshot.sql +++ /dev/null @@ -1,72 +0,0 @@ --- 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; diff --git a/schema/schema.sql b/schema/schema.sql index 710345e5275..8022f4f13f3 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -64,7 +64,6 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; -- functions \i functions/channels_community_id_immutable.sql -\i functions/guard_channel_roster_snapshot.sql \i functions/events_created_at_floor_guard.sql \i functions/prevent_community_deletion_request_retargeting.sql \i functions/prevent_community_deletion_approval_removal.sql diff --git a/schema/triggers/events.sql b/schema/triggers/events.sql index 64d9bbdd5d3..45fc796d4b7 100644 --- a/schema/triggers/events.sql +++ b/schema/triggers/events.sql @@ -1,7 +1,3 @@ -CREATE TRIGGER trg_events_guard_channel_roster_snapshot - BEFORE INSERT ON events - FOR EACH ROW EXECUTE FUNCTION guard_channel_roster_snapshot(); - -- INSERT OR UPDATE OF: an UPDATE can move a previously exempt row into the -- guarded set (channel_id NULL -> NOT NULL) or move a channel row's -- created_at below the fence, so both mutation paths re-run the guard on the diff --git a/scripts/reconcile-schema-after-pgschema.sql b/scripts/reconcile-schema-after-pgschema.sql index 0c84398aa3d..c61b002f87f 100644 --- a/scripts/reconcile-schema-after-pgschema.sql +++ b/scripts/reconcile-schema-after-pgschema.sql @@ -18,7 +18,6 @@ BEGIN -- triggers while attaching and rejects same-named child triggers. 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; @@ -30,7 +29,6 @@ BEGIN ) THEN 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; @@ -42,7 +40,6 @@ BEGIN ) THEN 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; @@ -54,7 +51,6 @@ BEGIN ) THEN 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; @@ -66,7 +62,6 @@ BEGIN ) THEN 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; @@ -78,7 +73,6 @@ BEGIN ) THEN 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; @@ -90,7 +84,6 @@ BEGIN ) THEN 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; @@ -102,7 +95,6 @@ BEGIN ) THEN 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;