diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index f178ab5814..4855d32ccc 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -197,6 +197,94 @@ async fn acquire_channel_membership_lock( Ok(()) } +/// Check the same p-tag shape and canonical active membership as migration 0032. +/// The caller holds this channel's membership advisory lock on the admitted +/// transaction before calling. The membership writers in this module take that +/// key before reading roles or changing rows. Not every writer does yet: +/// `execute_kick_with_marker` and `deploy_kick_member` (relay_admin_actions.rs) +/// set `removed_at` without it, so a kick can commit after this check passes +/// and before the roster commits. The retained trigger has the same gap. Both +/// kick paths must take the key, with a concurrent kick-vs-roster-publication +/// test, before anything relies on complete serialization (phase 2 included). +/// 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. +/// +/// 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. +pub(crate) async fn validate_channel_roster_snapshot( + tx: &mut AdmittedTx, + channel_id: Uuid, + tags: &serde_json::Value, +) -> Result<()> { + let verdict: i32 = sqlx::query_scalar( + r#" + SELECT CASE WHEN EXISTS ( + SELECT 1 FROM jsonb_array_elements($3::jsonb) 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 1 + WHEN + (SELECT COALESCE(array_agg(encode(cm.pubkey, 'hex') || ':' || cm.role::text + ORDER BY cm.pubkey), ARRAY[]::text[]) + FROM channel_members cm + WHERE cm.community_id = $1 AND cm.channel_id = $2 + AND cm.removed_at IS NULL) + IS DISTINCT FROM + (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[]) + FROM jsonb_array_elements($3::jsonb) AS roster_tag(tag_json) + WHERE roster_tag.tag_json->>0 = 'p') + THEN 2 ELSE 0 END + "#, + ) + .bind(tx.community().as_uuid()) + .bind(channel_id) + .bind(tags) + .fetch_one(tx.conn()) + .await?; + match verdict { + 0 => return Ok(()), + 1 => sqlx::query( + "DO $$ BEGIN RAISE EXCEPTION 'kind 39002 roster contains an invalid p tag' USING ERRCODE = '23514'; END $$", + ), + _ => sqlx::query( + "DO $$ BEGIN RAISE EXCEPTION 'kind 39002 roster does not match canonical channel membership' USING ERRCODE = '23514'; END $$", + ), + } + .execute(tx.conn()) + .await?; + Err(DbError::InvalidData( + "roster check did not reject invalid snapshot".into(), + )) +} + +/// 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. +pub(crate) async fn guard_channel_roster_event( + tx: &mut AdmittedTx, + kind: i32, + channel_id: Option, + tags: &serde_json::Value, +) -> Result<()> { + if kind == 39002 { + if let Some(channel_id) = channel_id { + let community_id = tx.community(); + acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await?; + validate_channel_roster_snapshot(tx, channel_id, tags).await?; + } + } + Ok(()) +} + // ── Transaction-level membership helpers (for commit_participant_join) ──────── /// Acquire the per-channel membership advisory lock on a caller-owned transaction. @@ -398,11 +486,12 @@ impl LockedMemberSnapshot { )); } } + let tags = serde_json::to_value(&event.tags)?; + validate_channel_roster_snapshot(&mut self.tx, channel_id, &tags).await?; 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(self.tx.conn()).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()) @@ -3081,6 +3170,447 @@ 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, + members: &[(&[u8], &str, &str)], + ) -> nostr::Event { + use nostr::{EventBuilder, Kind, Tag}; + + let mut tags = vec![Tag::parse(["d", &channel_id.to_string()]).expect("d tag")]; + for (pubkey, relay, role) in members { + tags.push(Tag::parse(["p", &hex::encode(pubkey), relay, role]).expect("p tag")); + } + EventBuilder::new(Kind::Custom(39002), "") + .tags(tags) + .sign_with_keys(keys) + .expect("sign roster") + } + + #[tokio::test] + #[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(); + let channel = create_test_channel( + &pool, + community_uuid, + "app-roster-locked", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner, + None, + ) + .await + .expect("create channel"); + let keys = Keys::generate(); + let stale = signed_roster(&keys, channel.id, &[]); + let mut snapshot = + lock_member_snapshot(&pool, community, channel.id, &keys.public_key().to_bytes()) + .await + .expect("lock snapshot"); + let error = snapshot + .replace_member_event(&stale) + .await + .expect_err("app must reject stale roster without trigger"); + assert!(matches!(error, DbError::Sqlx(sqlx::Error::Database(ref db)) + if db.code().as_deref() == Some("23514") + && db.message() == "kind 39002 roster does not match canonical channel membership")); + drop(snapshot); + + let current = signed_roster(&keys, channel.id, &[(&owner, "", "owner")]); + let mut snapshot = + lock_member_snapshot(&pool, community, channel.id, &keys.public_key().to_bytes()) + .await + .expect("lock current snapshot"); + assert!( + snapshot + .replace_member_event(¤t) + .await + .expect("current roster") + .1 + ); + snapshot.release().await.expect("commit current roster"); + 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.id) + .fetch_one(&pool) + .await + .expect("load current roster"); + assert_eq!(live_id, current.id.as_bytes().to_vec()); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn app_roster_guard_matches_trigger_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()); + let invalid = "kind 39002 roster contains an invalid p tag"; + let mismatch = "kind 39002 roster does not match canonical channel membership"; + let mut cases = vec![ + ( + "three-field p", + json!([["p", owner_hex, ""]]), + Some("owner"), + Some(invalid), + ), + ( + "five-field p", + json!([["p", owner_hex, "", "owner", "extra"]]), + Some("owner"), + Some(invalid), + ), + ( + "63-char pubkey", + json!([["p", &owner_hex[..63], "", "owner"]]), + Some("owner"), + Some(invalid), + ), + ( + "non-hex pubkey", + json!([["p", "z".repeat(64), "", "owner"]]), + Some("owner"), + Some(invalid), + ), + ( + "nonempty relay", + json!([["p", owner_hex, "wss://relay.invalid", "owner"]]), + Some("owner"), + Some(invalid), + ), + ( + "unknown role", + json!([["p", owner_hex, "", "moderator"]]), + Some("owner"), + Some(invalid), + ), + ( + "duplicate member", + json!([["p", owner_hex, "", "owner"], ["p", owner_hex, "", "owner"]]), + Some("owner"), + Some(mismatch), + ), + ( + "role mismatch", + json!([["p", owner_hex, "", "admin"]]), + Some("owner"), + Some(mismatch), + ), + ("missing member", json!([]), Some("owner"), Some(mismatch)), + ( + "extra member", + json!([ + ["p", owner_hex, "", "owner"], + ["p", extra_hex, "", "member"] + ]), + Some("owner"), + Some(mismatch), + ), + ( + "uppercase real member", + json!([["p", owner_hex.to_uppercase(), "", "owner"]]), + Some("owner"), + None, + ), + ( + "non-p tags ignored", + json!([ + ["p", owner_hex, "", "owner"], + ["name", "ignored"], + ["x"], + ["P", "not a p tag"] + ]), + Some("owner"), + None, + ), + ("empty roster and membership", json!([]), None, None), + ( + "nonempty roster and empty membership", + json!([["p", owner_hex, "", "owner"]]), + None, + Some(mismatch), + ), + ]; + for role in ["owner", "admin", "bot", "member", "guest"] { + cases.push((role, json!([["p", owner_hex, "", role]]), Some(role), None)); + } + assert_ne!( + owner_hex, + owner_hex.to_uppercase(), + "uppercase case must change the input" + ); + + let database_error = |error: sqlx::Error| match error { + sqlx::Error::Database(error) => ( + error.code().expect("SQLSTATE").into_owned(), + error.message().to_owned(), + ), + 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. + for (name, tags, role, expected_error) in cases { + let community_uuid = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_uuid); + let channel = create_test_channel( + &pool, + community_uuid, + name, + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner, + None, + ) + .await + .expect("create case channel"); + if let Some(role) = role { + sqlx::query("UPDATE channel_members SET role=$3::member_role WHERE community_id=$1 AND channel_id=$2") + .bind(community_uuid) + .bind(channel.id) + .bind(role) + .execute(&pool) + .await + .expect("set canonical case role"); + } else { + sqlx::query("UPDATE channel_members SET removed_at=NOW() WHERE community_id=$1 AND channel_id=$2") + .bind(community_uuid) + .bind(channel.id) + .execute(&pool) + .await + .expect("empty active membership"); + } + let mut tags: Vec> = serde_json::from_value(tags).expect("case tag arrays"); + tags.insert(0, vec!["d".into(), channel.id.to_string()]); + let event = EventBuilder::new(Kind::Custom(39002), "") + .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)) + .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}"); + } + } + + #[tokio::test] + #[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(); + let channel = create_test_channel( + &pool, + community_uuid, + "app-roster-generic", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner, + None, + ) + .await + .expect("create channel"); + let stale = signed_roster(&Keys::generate(), channel.id, &[]); + let db = Db::from_pool(pool.clone()); + + for result in [ + db.insert_event(community, &stale, Some(channel.id)).await, + db.insert_event_with_thread_metadata(community, &stale, Some(channel.id), None) + .await, + db.replace_parameterized_event( + community, + &stale, + &channel.id.to_string(), + Some(channel.id), + ) + .await, + ] { + let error = result.expect_err("every event writer must reject stale roster"); + assert!(matches!(error, DbError::Sqlx(sqlx::Error::Database(ref db)) + if db.code().as_deref() == Some("23514") + && db.message() == "kind 39002 roster does not match canonical channel membership")); + } + } + + #[tokio::test] + #[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(); + let newcomer = random_pubkey(); + let channel = create_test_channel( + &pool, + community_uuid, + "app-roster-race", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner, + None, + ) + .await + .expect("create channel"); + let old = signed_roster(&Keys::generate(), channel.id, &[(&owner, "", "owner")]); + + let mut holder = pool.begin().await.expect("hold membership transaction"); + acquire_channel_membership_lock(&mut holder, community, channel.id) + .await + .expect("hold membership key"); + 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.id) + .bind(&newcomer) + .bind(&owner) + .execute(&mut *holder) + .await + .expect("stage newer canonical roster"); + + let writer_pool = pool.clone(); + let old_id = old.id.as_bytes().to_vec(); + let writer = tokio::spawn(async move { + Db::from_pool(writer_pool) + .replace_addressable_event(community, &old, Some(channel.id)) + .await + }); + // A waiting advisory lock proves the old publication has reached the + // app fence before the newer canonical membership commits. + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_locks WHERE locktype='advisory' \ + AND NOT granted AND database=(SELECT oid FROM pg_database WHERE datname=current_database()))", + ) + .fetch_one(&pool) + .await + .expect("observe waiting roster writer"); + if waiting { break; } + tokio::task::yield_now().await; + } + }) + .await + .expect("old roster writer must wait for membership key"); + holder + .commit() + .await + .expect("commit newer canonical roster"); + + let error = writer + .await + .expect("join old writer") + .expect_err("old in-flight roster must be rejected"); + assert!(matches!(error, DbError::Sqlx(sqlx::Error::Database(ref db)) + if db.code().as_deref() == Some("23514") + && db.message() == "kind 39002 roster does not match canonical channel membership")); + let count: i64 = sqlx::query_scalar("SELECT count(*) FROM events WHERE id=$1") + .bind(old_id) + .fetch_one(&pool) + .await + .expect("count rejected old roster"); + assert_eq!(count, 0); + } + /// 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. @@ -3804,15 +4334,61 @@ mod postgres_tests { .1 ); let stale = roster("member", base + 1); - let error = db - .replace_addressable_event(community, &stale, Some(channel)) + // 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_err("desired-state fence must reject stale role"); - assert!(matches!( - error, - DbError::Sqlx(sqlx::Error::Database(ref db_error)) + .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", @@ -3823,6 +4399,14 @@ mod postgres_tests { .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"); drop_scratch_db(&admin, pool, &scratch_name).await; } diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 9474623da9..209f09e573 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -482,6 +482,8 @@ async fn insert_event_on( let received_at = Utc::now(); let d_tag = extract_d_tag(event); let not_before = extract_not_before(event); + crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json) + .await?; let result = sqlx::query( r#" INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) @@ -1784,6 +1786,9 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( let d_tag = extract_d_tag(event); let not_before = extract_not_before(event); + crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json) + .await?; + let result = sqlx::query( r#" INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) diff --git a/crates/buzz-db/src/store/replaceable.rs b/crates/buzz-db/src/store/replaceable.rs index 5cb76d7262..1ccee16332 100644 --- a/crates/buzz-db/src/store/replaceable.rs +++ b/crates/buzz-db/src/store/replaceable.rs @@ -259,6 +259,9 @@ async fn replace_parameterized_event_in_transaction_impl( .execute(tx.conn()) .await?; let written: Result = async { + let tags_json = serde_json::to_value(&event.tags)?; + crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json) + .await?; if existing.is_some() { let previous_nip_rs_hard_delete: Option = if is_nip_rs { sqlx::query_scalar( @@ -309,7 +312,6 @@ async fn replace_parameterized_event_in_transaction_impl( } let sig_bytes = event.sig.serialize(); - let tags_json = serde_json::to_value(&event.tags)?; let insert_result = sqlx::query( "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) \ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) \ @@ -485,6 +487,12 @@ impl Db { } } + let tags_json = serde_json::to_value(&event.tags)?; + crate::channel_members::guard_channel_roster_event( + &mut tx, kind_i32, channel_id, &tags_json, + ) + .await?; + // Soft-delete the old event (if any). IS NOT DISTINCT FROM for NULL safety. sqlx::query( "UPDATE events SET deleted_at = NOW() \ @@ -501,7 +509,6 @@ impl Db { // Insert the new event inside the same transaction. let sig_bytes = event.sig.serialize(); - let tags_json = serde_json::to_value(&event.tags)?; let received_at = chrono::Utc::now(); let d_tag = crate::event::extract_d_tag(event); @@ -921,25 +928,61 @@ mod postgres_tests { .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)), - ) + // 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, - DbError::Sqlx(sqlx::Error::Database(ref db_error)) - if db_error.code().as_deref() == Some("23514") - ), - "expected roster fence check violation, got {error:?}" + 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( diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 9b05eeb94a..60917166dd 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -1877,8 +1877,7 @@ mod tests { ); } -#[test] -fn every_production_event_insert_runs_the_follow_up_hook() { +fn production_event_writer_source_files() -> Vec { use std::path::{Path, PathBuf}; fn collect_rs_files(dir: &Path, out: &mut Vec) { @@ -1903,6 +1902,12 @@ fn every_production_event_insert_runs_the_follow_up_hook() { } } + files +} + +#[test] +fn every_production_event_insert_runs_the_follow_up_hook() { + let files = production_event_writer_source_files(); let mut hooked_writers = 0usize; let mut violations = Vec::new(); for path in files { @@ -1935,3 +1940,123 @@ fn every_production_event_insert_runs_the_follow_up_hook() { "expected at least seven hooked event writers, found {hooked_writers}" ); } + +// Fixed channel-less kinds cannot be channel rosters. The startup probes must +// exercise the retained DB backstops directly and roll back all their writes. +// Exemptions match by function header, so a kind 39002 insert added inside one +// of these functions would pass the policy; keep them single-purpose. +const EVENT_INSERT_ROSTER_GUARD_EXCEPTIONS: [&str; 4] = [ + // NIP-43 membership is kind 13534 with channel_id NULL. + "pub async fn publish_nip43_membership_locked(", + // Push leases are kind 30350 with channel_id NULL. + "pub async fn accept_lease_event(", + // Rolled-back replica-floor startup probe. + "pub async fn verify_floor_guard_behavior(", + // Rolled-back roster-fence startup probe. + "pub async fn verify_channel_roster_fence_behavior(", +]; + +fn event_insert_roster_guard_violations(production_source: &str) -> Vec { + function_slices(production_source) + .into_iter() + .filter(|function_source| { + let header = function_header(function_source); + let code = mask_comments_and_literals(function_source); + inserts_events_row(function_source) + && !EVENT_INSERT_ROSTER_GUARD_EXCEPTIONS + .iter() + .any(|exception| header.starts_with(exception)) + && !code.contains("guard_channel_roster_event(") + && !code.contains("validate_channel_roster_snapshot(") + }) + .map(|function_source| function_header(function_source).to_owned()) + .collect() +} + +#[test] +fn event_insert_roster_guard_policy_rejects_writers_without_the_guard() { + let source = r#" +pub(crate) async fn unguarded_writer(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)"); +} +pub(crate) async fn guarded_writer(tx: &mut AdmittedTx) { + guard_channel_roster_event(tx, kind, channel, tags).await?; + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)"); +} +pub(crate) async fn locked_writer(tx: &mut AdmittedTx) { + validate_channel_roster_snapshot(tx, channel, tags).await?; + sqlx::query("insert into public.events (community_id, id) VALUES ($1, $2)"); +} +pub(crate) async fn mention_writer(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO event_mentions (community_id, event_id) VALUES ($1, $2)"); +} +pub(crate) async fn commented_guard_writer(tx: &mut AdmittedTx) { + // guard_channel_roster_event(tx, kind, channel, tags).await?; + /* validate_channel_roster_snapshot(tx, channel, tags).await?; */ + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)"); +} +pub(crate) async fn quoted_guard_writer(tx: &mut AdmittedTx) { + tracing::debug!("guard_channel_roster_event("); + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)"); +} +pub async fn publish_nip43_membership_locked(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)"); +} +pub async fn accept_lease_event(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)"); +} +pub async fn verify_floor_guard_behavior(pool: &PgPool) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)"); +} +pub async fn verify_channel_roster_fence_behavior(pool: &PgPool) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)"); +} +#[cfg(test)] +mod tests { + fn fixture() { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)"); + } +} +"#; + assert_eq!( + event_insert_roster_guard_violations(&strip_cfg_test_items(source)), + [ + "pub(crate) async fn unguarded_writer(tx: &mut AdmittedTx) {", + "pub(crate) async fn commented_guard_writer(tx: &mut AdmittedTx) {", + "pub(crate) async fn quoted_guard_writer(tx: &mut AdmittedTx) {", + ], + "only real roster-guard calls in production writers count" + ); +} + +/// Presence check: every production `INSERT INTO events` function must contain +/// a real (uncommented, unquoted) roster-guard call. It does not prove the call +/// is awaited, reachable, or given the event's real kind and channel; the PG +/// `app_roster_guard_*` tests cover behavior. +#[test] +fn every_production_event_insert_runs_the_roster_guard() { + let mut checked_writers = 0; + let mut violations = Vec::new(); + for path in production_event_writer_source_files() { + let source = std::fs::read_to_string(&path).expect("read source file"); + let production = strip_cfg_test_items(&source); + checked_writers += function_slices(&production) + .into_iter() + .filter(|function| inserts_events_row(function)) + .count(); + violations.extend( + event_insert_roster_guard_violations(&production) + .into_iter() + .map(|header| format!("{}: {header}", path.display())), + ); + } + assert!( + violations.is_empty(), + "production `INSERT INTO events` writers must run the app-side roster guard: {violations:?}" + ); + // Five guarded writers plus the four explicitly exempt writers/probes. + assert!( + checked_writers >= 9, + "expected at least nine event writers, found {checked_writers}" + ); +}