diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index a27698b3d2..831d07159c 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -516,6 +516,15 @@ async fn get_channel_with_operation( operation: crate::observability::WriterOperation, ) -> Result { let mut connection = crate::observability::acquire_writer(pool, operation).await?; + select_channel(&mut connection, community_id, channel_id).await +} + +/// Read a live channel on `conn`. +async fn select_channel( + conn: &mut sqlx::PgConnection, + community_id: CommunityId, + channel_id: Uuid, +) -> Result { let row = sqlx::query( r#" SELECT id, name, channel_type::text AS channel_type, visibility::text AS visibility, @@ -530,7 +539,7 @@ async fn get_channel_with_operation( ) .bind(community_id.as_uuid()) .bind(channel_id) - .fetch_optional(&mut *connection) + .fetch_optional(conn) .await? .ok_or(DbError::ChannelNotFound(channel_id))?; @@ -784,11 +793,23 @@ pub(crate) fn row_to_channel_record(row: sqlx::postgres::PgRow) -> Result bool { + channel.channel_type == ChannelType::Dm.as_str() +} + +/// The roles a channel's relay-signed NIP-29 kind:39001 admins event lists. +/// The locked snapshot that feeds it reads only these members. +pub const GROUP_ADMIN_ROLES: [MemberRole; 2] = [MemberRole::Owner, MemberRole::Admin]; + /// The tags of a channel's relay-signed NIP-29 kind:39000 metadata event. /// /// This is the single builder for kind:39000: the relay's discovery emitter /// and the `buzz-admin` backfill both call it, so the published flags cannot -/// drift apart. `members` is used only for DM participant `p` tags. +/// drift apart. `members` is read only when [`group_metadata_uses_members`] +/// holds, for DM participant `p` tags. /// /// The NIP-29 access flags follow the channel's settings: /// - `private` (members-only read), or Buzz's explicit `public`; @@ -838,7 +859,7 @@ pub fn group_metadata_tags(channel: &ChannelRecord, members: &[MemberRecord]) -> if channel.channel_type == ChannelType::System.as_str() && !open { tags.push(tag(&["hidden"])); } - if channel.channel_type == ChannelType::Dm.as_str() { + if group_metadata_uses_members(channel) { tags.push(tag(&["hidden"])); // Include participant pubkeys in kind:39000 for DMs so clients can // resolve display names without a separate kind:39002 fetch. @@ -888,6 +909,139 @@ pub fn group_metadata_tags(channel: &ChannelRecord, members: &[MemberRecord]) -> tags } +/// A channel and its active members, read behind the replacement lock of the +/// channel's relay-signed kind:39000 or kind:39001 event, with the +/// `created_at` of that event's live copy. +/// +/// The event built from this snapshot is stored by [`Self::replace`] before +/// the lock is released, so racing publishers store in the order they read, +/// each one second past the last: the live event never falls back to state +/// read before a newer publisher's. kind:39002 has the same fence in +/// [`LockedMemberSnapshot`]. +pub struct LockedGroupStateSnapshot { + /// The channel, read behind the lock. + pub channel: ChannelRecord, + /// The active members the kind's tags use, read behind the lock: every + /// member for a DM's kind:39000 (its `p` tags), owners and admins for + /// kind:39001, and none for any other kind:39000, whose tags do not + /// mention members. Reading only those keeps the lock short on large + /// channels. + pub members: Vec, + /// `created_at` of the coordinate's live event, if it has one. + pub latest_created_at: Option, + kind: i32, + relay_pubkey: Vec, + tx: crate::AdmittedTx, + /// Times lock, reads and replacement as one `replace_addressable_event` + /// transaction, which is where these writes were timed before. + timer: crate::observability::TransactionTimer, +} + +impl LockedGroupStateSnapshot { + /// Store `event`, built from this snapshot, as the coordinate's live + /// event, and release the lock. Returns `(event, false)`, storing + /// nothing, when an equal or newer event is already live. + pub async fn replace(self, event: &nostr::Event) -> Result<(StoredEvent, bool)> { + let Self { + channel, + kind, + relay_pubkey, + mut tx, + timer, + .. + } = self; + let channel_id = channel.id; + if buzz_core::kind::event_kind_i32(event) != kind + || event.pubkey.to_bytes().as_slice() != relay_pubkey.as_slice() + || crate::event::extract_d_tag(event) != Some(channel_id.to_string()) + { + return Err(DbError::InvalidData( + "group-state replacement does not match its locked coordinate".into(), + )); + } + timer + .observe(async { + let (stored, inserted) = crate::replaceable::replace_addressable_event_under_lock( + &mut tx, + event, + Some(channel_id), + ) + .await?; + if inserted { + tx.commit().await?; + } else { + tx.rollback().await?; + } + Ok((stored, inserted)) + }) + .await + } +} + +/// Take the replacement lock of `channel_id`'s relay-signed `kind` event +/// (39000 or 39001), then read the channel, the members that kind's tags +/// use and the live event's `created_at` behind it. See +/// [`LockedGroupStateSnapshot`]. Any other `kind` is refused. +pub async fn lock_group_state_snapshot( + pool: &PgPool, + community_id: CommunityId, + channel_id: Uuid, + kind: i32, + relay_pubkey: &[u8], +) -> Result { + if kind != 39000 && kind != 39001 { + return Err(DbError::InvalidData(format!( + "group-state snapshots cover kind:39000 and kind:39001, not kind:{kind}" + ))); + } + // The same writer population and transaction timer as + // `Db::replace_addressable_event`, which stored these events before. + let mut tx = crate::begin_community_event_write_transaction_with_legacy_metrics( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; + let timer = crate::observability::TransactionTimer::start( + crate::observability::TransactionOperation::ReplaceAddressableEvent, + ); + crate::replaceable::take_replacement_lock(&mut tx, kind, relay_pubkey, Some(channel_id)) + .await?; + let channel = select_channel(tx.conn(), community_id, channel_id).await?; + let members = if kind == 39001 { + crate::channel_members::select_active_members_in_roles( + tx.conn(), + community_id, + channel_id, + &GROUP_ADMIN_ROLES, + ) + .await? + } else if group_metadata_uses_members(&channel) { + crate::channel_members::select_active_members(tx.conn(), community_id, channel_id).await? + } else { + Vec::new() + }; + let latest_created_at: Option> = sqlx::query_scalar( + "SELECT created_at 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(relay_pubkey) + .bind(channel_id) + .fetch_optional(tx.conn()) + .await?; + Ok(LockedGroupStateSnapshot { + channel, + members, + latest_created_at: latest_created_at.map(|created_at| created_at.timestamp() as u64), + kind, + relay_pubkey: relay_pubkey.to_vec(), + tx, + timer, + }) +} + /// Partial update for channel metadata. Every field is `None` to leave the /// column unchanged. #[derive(Default)] @@ -1923,6 +2077,19 @@ impl Db { .await } + /// Read a channel's kind:39000 or kind:39001 inputs behind that event's + /// replacement lock. See [`LockedGroupStateSnapshot`]. + #[datastore_span(name = "lock_group_state_snapshot", system = "postgresql")] + pub async fn lock_group_state_snapshot( + &self, + community_id: CommunityId, + channel_id: Uuid, + kind: i32, + relay_pubkey: &[u8], + ) -> Result { + lock_group_state_snapshot(&self.pool, community_id, channel_id, kind, relay_pubkey).await + } + /// Returns the canvas content for a channel, if any. #[datastore_span(name = "get_canvas", system = "postgresql")] pub async fn get_canvas( @@ -2975,4 +3142,265 @@ mod postgres_tests { "a second delete finds no live channel" ); } + + /// A group-state snapshot reads behind its coordinate's replacement lock. + /// One taken while another is held waits for it, then sees a channel + /// change committed while it waited and the held snapshot's stored + /// `created_at`. Mutation: read the channel or the live timestamp before + /// taking the lock → RED. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn group_state_snapshot_reads_behind_the_replacement_lock() { + let pool = setup_pool().await; + let community_id = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_id); + let channel = create_test_channel( + &pool, + community_id, + "group-state-snapshot", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &random_pubkey(), + None, + ) + .await + .expect("create channel"); + let relay = Keys::generate(); + let relay_pubkey = relay.public_key().to_bytes(); + + let held = lock_group_state_snapshot(&pool, community, channel.id, 39000, &relay_pubkey) + .await + .expect("hold snapshot"); + assert_eq!(held.latest_created_at, None); + let waiter_pool = crate::test_support::named_pool("group-state-snapshot").await; + let channel_id = channel.id; + let waiter = tokio::spawn(async move { + lock_group_state_snapshot(&waiter_pool, community, channel_id, 39000, &relay_pubkey) + .await + }); + crate::test_support::wait_for_lock_wait(&pool, "group-state-snapshot").await; + sqlx::query("UPDATE channels SET topic = 'changed while waiting' WHERE id = $1") + .bind(channel.id) + .execute(&pool) + .await + .expect("change the channel"); + + let held_at = 1_800_000_000; + let event = nostr::EventBuilder::new(nostr::Kind::Custom(39000), "") + .tags( + group_metadata_tags(&held.channel, &held.members) + .into_iter() + .map(|tag| nostr::Tag::parse(tag).expect("tag")), + ) + .custom_created_at(nostr::Timestamp::from(held_at)) + .sign_with_keys(&relay) + .expect("sign 39000"); + assert!(held.replace(&event).await.expect("store held snapshot").1); + + let next = waiter.await.expect("waiter task").expect("next snapshot"); + assert_eq!(next.channel.topic.as_deref(), Some("changed while waiting")); + assert_eq!(next.latest_created_at, Some(held_at)); + } + + /// A snapshot reads only the members its kind's tags use, and the tags + /// built from it are the ones the full roster would give: none of the + /// roster for a stream's kind:39000, owners and admins in roster order + /// for kind:39001, the whole roster for a DM's kind:39000. Any other + /// kind is refused. Mutation: read the roster for a stream's kind:39000, + /// or drop the role filter for kind:39001 → RED. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn group_state_snapshot_reads_the_members_its_kind_publishes() { + let pool = setup_pool().await; + let community_id = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_id); + let relay_pubkey = Keys::generate().public_key().to_bytes(); + let roster = async |channel_id: Uuid| { + let mut conn = pool.acquire().await.expect("acquire"); + crate::channel_members::select_active_members(&mut conn, community, channel_id) + .await + .expect("roster") + }; + let who = |members: &[MemberRecord]| { + members + .iter() + .map(|m| (m.pubkey.clone(), m.role.clone())) + .collect::>() + }; + // `create_test_channel` seats the creator as owner; the rest join + // directly, past the join-role policy `add_member` applies. + let seat = async |channel_id: Uuid, role: &str| { + sqlx::query( + "INSERT INTO channel_members (community_id, channel_id, pubkey, role) \ + VALUES ($1, $2, $3, $4::member_role)", + ) + .bind(community_id) + .bind(channel_id) + .bind(random_pubkey()) + .bind(role) + .execute(&pool) + .await + .expect("seat member"); + }; + + let owner = random_pubkey(); + let stream = create_test_channel( + &pool, + community_id, + "group-state-members", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner, + None, + ) + .await + .expect("create stream"); + for role in ["member", "admin", "guest"] { + seat(stream.id, role).await; + } + let all = roster(stream.id).await; + assert_eq!(all.len(), 4); + + let metadata = lock_group_state_snapshot(&pool, community, stream.id, 39000, &relay_pubkey) + .await + .expect("stream 39000 snapshot"); + assert!(metadata.members.is_empty(), "{:?}", who(&metadata.members)); + assert_eq!( + group_metadata_tags(&metadata.channel, &metadata.members), + group_metadata_tags(&metadata.channel, &all) + ); + drop(metadata); + + let admins = lock_group_state_snapshot(&pool, community, stream.id, 39001, &relay_pubkey) + .await + .expect("stream 39001 snapshot"); + let expected: Vec<_> = all + .iter() + .filter(|m| m.role == "owner" || m.role == "admin") + .cloned() + .collect(); + assert_eq!(expected.len(), 2); + assert_eq!(who(&admins.members), who(&expected)); + drop(admins); + + let dm = create_test_channel( + &pool, + community_id, + "group-state-members-dm", + ChannelType::Dm, + ChannelVisibility::Private, + None, + &owner, + None, + ) + .await + .expect("create dm"); + seat(dm.id, "member").await; + let dm_all = roster(dm.id).await; + assert_eq!(dm_all.len(), 2); + let dm_metadata = lock_group_state_snapshot(&pool, community, dm.id, 39000, &relay_pubkey) + .await + .expect("dm 39000 snapshot"); + assert_eq!(who(&dm_metadata.members), who(&dm_all)); + let tags = group_metadata_tags(&dm_metadata.channel, &dm_metadata.members); + assert_eq!(tags, group_metadata_tags(&dm_metadata.channel, &dm_all)); + assert_eq!(tags.iter().filter(|t| t[0] == "p").count(), 2); + drop(dm_metadata); + + let refused = lock_group_state_snapshot(&pool, community, dm.id, 39002, &relay_pubkey) + .await + .err() + .expect("kind:39002 is not a group-state snapshot"); + assert!(matches!(refused, DbError::InvalidData(_)), "{refused:?}"); + } + + /// A group-state write takes its replacement lock once: the snapshot + /// takes it and the replacement runs under it, so the lock metrics see + /// one acquisition per write, as they do for + /// `Db::replace_addressable_event`. Mutation: take the lock again in + /// `replace_addressable_event_under_lock` → RED. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn group_state_replacement_takes_its_lock_once() { + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + + let pool = setup_pool().await; + let community_id = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_id); + let channel = create_test_channel( + &pool, + community_id, + "group-state-lock-once", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &random_pubkey(), + None, + ) + .await + .expect("create channel"); + let relay = Keys::generate(); + let relay_pubkey = relay.public_key().to_bytes(); + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let _guard = metrics::set_default_local_recorder(&recorder); + // A debugging snapshot drains what it reads, so each call returns the + // acquisitions since the previous one. + let replacement_acquisitions_since_last = || { + snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(key, _, _, _)| { + key.key().name() == "buzz_db_advisory_lock_acquisitions_total" + && key.key().labels().any(|label| { + label.key() == "lock_type" + && label.value() + == crate::observability::LockType::Replacement.as_str() + }) + }) + .map(|(_, _, _, value)| match value { + DebugValue::Counter(count) => count, + other => panic!("acquisitions are a counter, got {other:?}"), + }) + .sum::() + }; + + let snapshot = + lock_group_state_snapshot(&pool, community, channel.id, 39000, &relay_pubkey) + .await + .expect("snapshot"); + let event = nostr::EventBuilder::new(nostr::Kind::Custom(39000), "") + .tags( + group_metadata_tags(&snapshot.channel, &snapshot.members) + .into_iter() + .map(|tag| nostr::Tag::parse(tag).expect("tag")), + ) + .sign_with_keys(&relay) + .expect("sign 39000"); + assert!(snapshot.replace(&event).await.expect("store").1); + assert_eq!( + replacement_acquisitions_since_last(), + 1, + "one lock per group-state write" + ); + + let profile = nostr::EventBuilder::new(nostr::Kind::Custom(0), "{}") + .sign_with_keys(&relay) + .expect("sign kind:0"); + assert!( + Db::from_pool(pool.clone()) + .replace_addressable_event(community, &profile, None) + .await + .expect("store kind:0") + .1 + ); + assert_eq!( + replacement_acquisitions_since_last(), + 1, + "one lock per addressable replace" + ); + } } diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 936e2b9078..5d42dd4000 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -579,23 +579,7 @@ pub async fn lock_member_snapshot( ) .await?; acquire_channel_membership_lock(tx.conn(), 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(tx.conn()) - .await?; - let members = rows - .into_iter() - .map(row_to_member_record) - .collect::>>()?; + let members = select_active_members(tx.conn(), community_id, channel_id).await?; let identity = sqlx::query_as::<_, (String, Vec, Vec)>( "SELECT channel_type::text, labels, created_by FROM channels WHERE community_id = $1 AND id = $2", ) @@ -1438,18 +1422,50 @@ async fn get_members_with_operation( operation: crate::observability::WriterOperation, ) -> Result> { let mut connection = crate::observability::acquire_writer(pool, operation).await?; + select_active_members(&mut connection, community_id, channel_id).await +} + +/// Read a live channel's active members on `conn`, oldest first. +pub(crate) async fn select_active_members( + conn: &mut PgConnection, + community_id: CommunityId, + channel_id: Uuid, +) -> Result> { + select_active_members_filtered(conn, community_id, channel_id, None).await +} + +/// The active members of `channel_id` whose role is one of `roles`, in the +/// order [`select_active_members`] returns them. +pub(crate) async fn select_active_members_in_roles( + conn: &mut PgConnection, + community_id: CommunityId, + channel_id: Uuid, + roles: &[MemberRole], +) -> Result> { + let roles: Vec<&str> = roles.iter().map(MemberRole::as_str).collect(); + select_active_members_filtered(conn, community_id, channel_id, Some(&roles)).await +} + +async fn select_active_members_filtered( + conn: &mut PgConnection, + community_id: CommunityId, + channel_id: Uuid, + roles: Option<&[&str]>, +) -> Result> { 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 + AND ($3::text[] IS NULL OR cm.role::text = ANY($3)) ORDER BY cm.joined_at ASC "#, ) .bind(community_id.as_uuid()) .bind(channel_id) - .fetch_all(&mut *connection) + .bind(roles) + .fetch_all(conn) .await?; rows.into_iter().map(row_to_member_record).collect() } diff --git a/crates/buzz-db/src/store/replaceable.rs b/crates/buzz-db/src/store/replaceable.rs index ffc56214bd..ef0ba2d4ff 100644 --- a/crates/buzz-db/src/store/replaceable.rs +++ b/crates/buzz-db/src/store/replaceable.rs @@ -389,9 +389,159 @@ async fn replace_parameterized_event_in_transaction_impl( )) } +/// Take the replacement lock of the `(kind, pubkey, channel_id)` coordinate +/// on `tx`, serializing every writer of that coordinate until the transaction +/// ends. Key collisions only cause extra serialization; they cannot change +/// behavior. Taken once per write: it is reentrant, but each take records an +/// acquisition and a wait sample. +pub(crate) async fn take_replacement_lock( + tx: &mut AdmittedTx, + kind: i32, + pubkey: &[u8], + channel_id: Option, +) -> Result<()> { + let lock_key = event_replacement_lock_key( + tx.community(), + kind, + pubkey, + channel_id.as_ref().map(|id| id.as_bytes().as_slice()), + ); + observability::observe_advisory_lock( + LockType::Replacement, + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(lock_key) + .execute(tx.conn()), + ) + .await?; + Ok(()) +} + +/// Replace a replaceable event on a caller-owned transaction that already +/// holds the coordinate's replacement lock ([`take_replacement_lock`]): the +/// body of [`Db::replace_addressable_event`], shared with the relay's locked +/// group-state snapshots. +/// +/// Returns `(event, false)` for a stale write or a duplicate ID. The caller +/// commits only when the event was inserted and rolls back otherwise: a +/// duplicate leaves the previous event soft-deleted on the transaction. +pub(crate) async fn replace_addressable_event_under_lock( + tx: &mut AdmittedTx, + event: &nostr::Event, + channel_id: Option, +) -> Result<(StoredEvent, bool)> { + let community_id = tx.community(); + let kind_i32 = buzz_core::kind::event_kind_i32(event); + let pubkey_bytes = 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))?; + + // Check for the newest existing event. ORDER BY + LIMIT 1 is defensive against + // historical data where prior bugs may have left multiple live rows. + 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 IS NOT DISTINCT FROM $4 \ + AND deleted_at IS NULL \ + ORDER BY created_at DESC, id ASC LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(kind_i32) + .bind(pubkey_bytes.as_slice()) + .bind(channel_id) + .fetch_optional(tx.conn()) + .await?; + + // Stale-write protection: reject if incoming is not newer. + // NIP-16: created_at is second-resolution. On same-second tie, lowest + // event id (lexicographic) wins — deterministic across relays. + let incoming_id = event.id.as_bytes().as_slice(); + if let Some((existing_ts, existing_id)) = existing { + let dominated = created_at < existing_ts + || (created_at == existing_ts && incoming_id >= existing_id.as_slice()); + if dominated { + let received_at = chrono::Utc::now(); + return Ok(( + StoredEvent::with_received_at(event.clone(), received_at, channel_id, false), + false, + )); + } + } + + let tags_json = serde_json::to_value(&event.tags)?; + crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json) + .await?; + + // Soft-delete the old event (if any). IS NOT DISTINCT FROM for NULL safety. + sqlx::query(superseded_with_mentions_sql!( + "UPDATE events SET deleted_at = NOW() \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \ + AND channel_id IS NOT DISTINCT FROM $4 \ + AND deleted_at IS NULL" + )) + .bind(community_id.as_uuid()) + .bind(kind_i32) + .bind(pubkey_bytes.as_slice()) + .bind(channel_id) + .execute(tx.conn()) + .await?; + + // Insert the new event inside the same transaction. + let sig_bytes = event.sig.serialize(); + let received_at = chrono::Utc::now(); + let d_tag = crate::event::extract_d_tag(event); + + let insert_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, $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_bytes.as_slice()) + .bind(created_at) + .bind(kind_i32) + .bind(&tags_json) + .bind(&event.content) + .bind(sig_bytes.as_slice()) + .bind(received_at) + .bind(channel_id) + .bind(d_tag.as_deref()) + .execute(tx.conn()) + .await?; + + if insert_result.rows_affected() == 0 { + // ON CONFLICT fired — the event ID already exists. The caller's + // rollback restores the previous replaceable event. + return Ok(( + StoredEvent::with_received_at(event.clone(), received_at, channel_id, false), + false, + )); + } + + crate::store::event_follow_up::after_admitted_insert( + tx, + event.id.as_bytes().as_slice(), + kind_i32, + channel_id, + ) + .await?; + + // The replaceable event and its denormalized mention index are one + // authoritative discovery write. An indexing error must roll back the + // new event and restore the previously-live event. + crate::insert_mentions_in_transaction(tx, event, channel_id).await?; + + Ok(( + StoredEvent::with_received_at(event.clone(), received_at, channel_id, true), + true, + )) +} + impl Db { /// Atomically replace a replaceable event: NIP-16 kinds (0, 3, 41, 10000–19999) - /// and NIP-29 discovery state (39000–39002, called from side_effects.rs). + /// and NIP-29 discovery state (39000–39002) written outside the relay's + /// locked group-state snapshots, e.g. by `buzz-admin`. /// /// Keeps only the event with the highest `created_at` per (kind, pubkey, channel_id). /// Same-second ties are broken by lowest event `id` (NIP-16 deterministic ordering). @@ -404,20 +554,6 @@ impl Db { event: &nostr::Event, channel_id: Option, ) -> Result<(StoredEvent, bool)> { - let kind_i32 = buzz_core::kind::event_kind_i32(event); - let pubkey_bytes = 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))?; - - // Collisions only cause extra serialization; they cannot change behavior. - let lock_key = event_replacement_lock_key( - community_id, - kind_i32, - pubkey_bytes.as_slice(), - channel_id.as_ref().map(|id| id.as_bytes().as_slice()), - ); - let mut tx = crate::begin_community_event_write_transaction_with_legacy_metrics( &self.pool, community_id, @@ -431,136 +567,21 @@ impl Db { transaction_timer .observe(async { - // Serialize all writers for the same (kind, pubkey, channel_id) tuple. - // Advisory lock is transaction-scoped — released on commit/rollback. - observability::observe_advisory_lock( - observability::LockType::Replacement, - sqlx::query("SELECT pg_advisory_xact_lock($1)") - .bind(lock_key) - .execute(tx.conn()), - ) - .await?; - - // Check for the newest existing event. ORDER BY + LIMIT 1 is defensive against - // historical data where prior bugs may have left multiple live rows. - 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 IS NOT DISTINCT FROM $4 \ - AND deleted_at IS NULL \ - ORDER BY created_at DESC, id ASC LIMIT 1", - ) - .bind(community_id.as_uuid()) - .bind(kind_i32) - .bind(pubkey_bytes.as_slice()) - .bind(channel_id) - .fetch_optional(tx.conn()) - .await?; - - // Stale-write protection: reject if incoming is not newer. - // NIP-16: created_at is second-resolution. On same-second tie, lowest - // event id (lexicographic) wins — deterministic across relays. - let incoming_id = event.id.as_bytes().as_slice(); - if let Some((existing_ts, existing_id)) = existing { - let dominated = created_at < existing_ts - || (created_at == existing_ts - && incoming_id >= existing_id.as_slice()); - if dominated { - tx.rollback().await?; - let received_at = chrono::Utc::now(); - return Ok(( - StoredEvent::with_received_at( - event.clone(), - received_at, - channel_id, - false, - ), - false, - )); - } - } - - 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(superseded_with_mentions_sql!( - "UPDATE events SET deleted_at = NOW() \ - WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \ - AND channel_id IS NOT DISTINCT FROM $4 \ - AND deleted_at IS NULL" - )) - .bind(community_id.as_uuid()) - .bind(kind_i32) - .bind(pubkey_bytes.as_slice()) - .bind(channel_id) - .execute(tx.conn()) - .await?; - - // Insert the new event inside the same transaction. - let sig_bytes = event.sig.serialize(); - let received_at = chrono::Utc::now(); - let d_tag = crate::event::extract_d_tag(event); - - let insert_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, $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_bytes.as_slice()) - .bind(created_at) - .bind(kind_i32) - .bind(&tags_json) - .bind(&event.content) - .bind(sig_bytes.as_slice()) - .bind(received_at) - .bind(channel_id) - .bind(d_tag.as_deref()) - .execute(tx.conn()) - .await?; - - let was_inserted = insert_result.rows_affected() > 0; - if !was_inserted { - // ON CONFLICT fired — the event ID already exists. Rollback the - // soft-delete so we don't lose the previous replaceable event. - tx.rollback().await?; - return Ok(( - StoredEvent::with_received_at( - event.clone(), - received_at, - channel_id, - false, - ), - false, - )); - } - - crate::store::event_follow_up::after_admitted_insert( + take_replacement_lock( &mut tx, - event.id.as_bytes().as_slice(), - kind_i32, + buzz_core::kind::event_kind_i32(event), + &event.pubkey.to_bytes(), channel_id, ) .await?; - - // The replaceable event and its denormalized mention index are one - // authoritative discovery write. An indexing error must roll back the - // new event and restore the previously-live event. - crate::insert_mentions_in_transaction(&mut tx, event, channel_id) - .await?; - - tx.commit().await?; - - Ok(( - StoredEvent::with_received_at(event.clone(), received_at, channel_id, true), - true, - )) + let (stored, was_inserted) = + replace_addressable_event_under_lock(&mut tx, event, channel_id).await?; + if was_inserted { + tx.commit().await?; + } else { + tx.rollback().await?; + } + Ok((stored, was_inserted)) }) .await } diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 14d9c9d514..2c5271eb03 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -264,7 +264,9 @@ fn p0_pool_acquisitions_use_typed_operation_pairs_without_other() { .expect("channel lookup helper must precede canvas reads") .0; assert!(get_channel.contains("acquire_writer(pool, operation)")); - assert!(get_channel.contains("fetch_optional(&mut *connection)")); + assert!(get_channel.contains("select_channel(&mut connection, community_id, channel_id)")); + assert!(get_channel.contains("conn: &mut sqlx::PgConnection")); + assert!(get_channel.contains("fetch_optional(conn)")); assert!(!get_channel.contains("fetch_optional(pool)")); assert!(channel.contains("pub async fn get_channel_for_event_write(")); let list_channels = channel diff --git a/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs b/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs index 5fbfd6799b..010240aeb4 100644 --- a/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs @@ -1046,6 +1046,205 @@ async fn removed_creator_join_racing_a_private_switch_is_refused() { ); } +/// Where a join's kind:39000 publisher is held while a kind:9002 edit races it. +#[derive(Clone, Copy, Debug)] +enum HoldPublisher { + /// Before it takes the replacement lock: the edit publishes in full + /// while the join is held, so anything the join read before the lock is + /// stale by the time it stores. + BeforeLock, + /// After it has read behind the lock: the edit's publisher waits for the + /// join's, then must still win with the edit's state. + AfterRead, +} + +/// A join's kind:39000 publisher cannot replace a racing edit's kind:39000 +/// with state read before the edit. In the e2e +/// `test_nip29_announce_channel_wire`, an admission auto-joined a default +/// channel while a relay admin's kind:9002 cleared `auto_join`; the join read +/// `auto_join` first, stored after the edit at a later second, and the stale +/// tag stayed live. +/// +/// Two legs. `BeforeLock` holds the join immediately before it takes the +/// lock, lets the edit publish in full, then releases it: a channel or +/// member read done before the lock (`get_channel_for_event_write`, +/// `get_members_for_event_write`) stores stale tags last → RED. `AfterRead` +/// holds the join after its read behind the lock until the edit is waiting +/// for that lock, so the edit must still win with its own state. +/// +/// Both legs also check the timestamps: every kind:39000 publication for +/// the coordinate, replaced rows included, leaves exactly one row, and +/// their `created_at` strictly increase in storage order. The live event is +/// first pinned an hour ahead, so each publication's timestamp is the live +/// one plus a second and the clock cannot separate the two publishers. +/// Mutation: take the live `created_at` before the lock → RED every time: +/// the publisher that locks second computes the same second as the first, +/// and either loses the id tie and leaves no row, or wins it and repeats +/// the second. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_publisher_that_read_before_an_edit_cannot_replace_it() { + for hold in [HoldPublisher::BeforeLock, HoldPublisher::AfterRead] { + join_publisher_held_against_an_auto_join_edit(hold).await; + } +} + +/// `created_at` of every relay-signed kind:39000 row for `channel_id`, +/// replaced rows included, in storage order. +async fn stored_metadata_created_at( + state: &AppState, + tenant: &TenantContext, + channel_id: Uuid, +) -> Vec> { + sqlx::query_scalar( + "SELECT created_at FROM events WHERE community_id = $1 AND kind = 39000 \ + AND pubkey = $2 AND channel_id = $3 ORDER BY received_at ASC", + ) + .bind(tenant.community().as_uuid()) + .bind(state.relay_keypair.public_key().to_bytes().as_slice()) + .bind(channel_id) + .fetch_all(state.db.pool()) + .await + .expect("read every kind:39000 row") +} + +async fn join_publisher_held_against_an_auto_join_edit(hold: HoldPublisher) { + let state = state().await; + let tenant = community(&state, "group-state-stale-publisher").await; + let owner = Keys::generate(); + let channel_id = create_channel(&state, &tenant, &owner, &[]).await; + let admin = Keys::generate(); + state + .db + .add_relay_member( + tenant.community(), + &admin.public_key().to_hex(), + "admin", + None, + &Keys::generate(), + ) + .await + .map(|write| write.outcome) + .expect("seed relay admin"); + edit_channel( + &state, + &tenant, + &admin, + channel_id, + &[&["auto_join", "true"]], + ) + .await + .expect("set auto_join"); + // Pin the live event an hour ahead: from here on every publication is + // stored at the live `created_at` plus one, whatever the clock says. + let pinned = EventBuilder::new(Kind::Custom(39000), "") + .tags( + group_state_tags(&state, &tenant, channel_id, 39000) + .await + .into_iter() + .map(|tag| Tag::parse(tag).expect("tag")), + ) + .custom_created_at(nostr::Timestamp::now() + 3600) + .sign_with_keys(&state.relay_keypair) + .expect("sign pinned kind:39000"); + assert!( + state + .db + .replace_addressable_event(tenant.community(), &pinned, Some(channel_id)) + .await + .expect("store pinned kind:39000") + .1 + ); + let rows_before = stored_metadata_created_at(&state, &tenant, channel_id) + .await + .len(); + + let (arrived, release) = match hold { + HoldPublisher::BeforeLock => { + crate::nip_fi_test_hooks::group_state_lock_hook::arm(tenant.community()) + } + HoldPublisher::AfterRead => { + crate::nip_fi_test_hooks::group_state_read_hook::arm(tenant.community()) + } + }; + let joiner = Keys::generate(); + let join = EventBuilder::new(Kind::Custom(9021), "") + .tags([Tag::parse(["h", &channel_id.to_string()]).unwrap()]) + .sign_with_keys(&joiner) + .unwrap(); + let join_task = tokio::spawn({ + let state = Arc::clone(&state); + let tenant = tenant.clone(); + let auth = http_auth(&joiner); + async move { ingest_event(&state, &tenant, join, auth).await } + }); + tokio::time::timeout(std::time::Duration::from_secs(10), arrived) + .await + .unwrap_or_else(|_| panic!("{hold:?}: the join publishes kind:39000")) + .unwrap_or_else(|_| panic!("{hold:?}: the hook stays armed until the join reaches it")); + + let edit_task = tokio::spawn({ + let state = Arc::clone(&state); + let tenant = tenant.clone(); + async move { + edit_channel( + &state, + &tenant, + &admin, + channel_id, + &[&["auto_join", "false"]], + ) + .await + } + }); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); + while !edit_task.is_finished() { + if matches!(hold, HoldPublisher::AfterRead) { + 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(state.db.pool()) + .await + .expect("read lock waiters"); + if waiting { + break; + } + } + assert!( + std::time::Instant::now() < deadline, + "{hold:?}: the edit neither finished nor waited for the held publisher" + ); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + release.notify_one(); + + let joined = join_task + .await + .expect("join task must not panic") + .unwrap_or_else(|error| panic!("{hold:?}: join: {error:?}")); + assert!(joined.accepted, "{hold:?}: join: {}", joined.message); + edit_task + .await + .expect("edit task must not panic") + .unwrap_or_else(|error| panic!("{hold:?}: clear auto_join: {error}")); + let tags = group_state_tags(&state, &tenant, channel_id, 39000).await; + assert!( + !tags.contains(&vec!["auto_join".to_string()]), + "{hold:?}: the join's stale kind:39000 replaced the edit's: {tags:?}" + ); + let created_at = stored_metadata_created_at(&state, &tenant, channel_id).await; + assert_eq!( + created_at.len(), + rows_before + 2, + "{hold:?}: the join and the edit each leave one kind:39000 row: {created_at:?}" + ); + assert!( + created_at.windows(2).all(|pair| pair[0] < pair[1]), + "{hold:?}: kind:39000 timestamps must strictly increase in storage order: {created_at:?}" + ); +} + /// A create or rename that would list a second open channel under a taken /// name is refused with its reason as a client error, and stores nothing. /// Mutation: let `ChannelNameTaken` fall through to the internal-error arm diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 38b6dbc38a..0740206c2a 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1513,41 +1513,51 @@ async fn actor_is_relay_operator( .map_err(|e| anyhow::anyhow!("relay staff lookup failed: {}", e.message)) } -/// Sign, store (replacing previous), and fan-out a single addressable discovery event. +/// Sign, store (replacing previous), and fan-out one of a channel's +/// relay-signed kind:39000/39001 events, with the tags `build_tags` makes +/// from the channel and its members. +/// +/// The inputs, the timestamp and the replacement all sit behind the event's +/// replacement lock (see `LockedGroupStateSnapshot`). Read outside it, an +/// auto-join that read `auto_join` before a racing kind:9002 cleared it +/// stored its snapshot after the edit's, at the same or a later second, and +/// the stale tag stayed live. async fn emit_addressable_discovery_event( tenant: &TenantContext, state: &Arc, channel_id: Uuid, kind: u32, - tags: Vec, relay_pubkey_hex: &str, + build_tags: impl FnOnce( + &buzz_db::channel::ChannelRecord, + &[MemberRecord], + ) -> anyhow::Result>, ) -> anyhow::Result<()> { - // Ensure the new event's created_at is strictly greater than any existing event - // of the same (kind, pubkey, channel_id). Without this, rapid successive updates - // (e.g. set topic then set purpose in the same second) can produce events with - // identical created_at, causing the second to be rejected by stale-write protection - // (NIP-16 tiebreaker: lower event ID wins, which is random). + #[cfg(test)] + crate::nip_fi_test_hooks::before_group_state_lock(tenant.community()).await; + let snapshot = state + .db + .lock_group_state_snapshot( + tenant.community(), + channel_id, + kind as i32, + &state.relay_keypair.public_key().to_bytes(), + ) + .await?; + #[cfg(test)] + crate::nip_fi_test_hooks::after_group_state_read(tenant.community()).await; + let tags = build_tags(&snapshot.channel, &snapshot.members)?; + // One second past the live event, so the replacement never loses a + // same-second tiebreak (NIP-16: lower event ID wins, which is random). let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_secs(); - let min_ts = { - let existing = state - .db - .query_events_for_event_write(&buzz_db::event::EventQuery { - kinds: Some(vec![kind as i32]), - channel_id: Some(channel_id), - limit: Some(1), - ..buzz_db::event::EventQuery::for_community(tenant.community()) - }) - .await - .unwrap_or_default(); - existing - .first() - .map(|e| e.event.created_at.as_secs() + 1) - .unwrap_or(now) - }; - let ts = now.max(min_ts); + let ts = snapshot + .latest_created_at + .map(|timestamp| timestamp + 1) + .unwrap_or(now) + .max(now); let event = EventBuilder::new(Kind::Custom(kind as u16), "") .tags(tags) @@ -1555,9 +1565,8 @@ async fn emit_addressable_discovery_event( .sign_with_keys(&state.relay_keypair) .map_err(|e| anyhow::anyhow!("failed to sign kind:{kind}: {e}"))?; - let (stored, was_inserted) = state - .db - .replace_addressable_event(tenant.community(), &event, Some(channel_id)) + let (stored, was_inserted) = snapshot + .replace(&event) .await .map_err(|e| anyhow::anyhow!("channel state write failed: {}", e.kind()))?; if was_inserted { @@ -1667,58 +1676,45 @@ pub async fn emit_group_discovery_events( state: &Arc, channel_id: Uuid, ) -> anyhow::Result<()> { - let channel = state - .db - .get_channel_for_event_write(tenant.community(), channel_id) - .await?; - let members = state - .db - .get_members_for_event_write(tenant.community(), channel_id) - .await?; - let relay_pubkey_hex = hex::encode(state.relay_keypair.public_key().to_bytes()); let group_id = channel_id.to_string(); - { - let tags = buzz_db::channel::group_metadata_tags(&channel, &members) - .into_iter() - .map(Tag::parse) - .collect::, _>>()?; - emit_addressable_discovery_event( - tenant, - state, - channel_id, - KIND_NIP29_GROUP_METADATA, - tags, - &relay_pubkey_hex, - ) - .await?; - } + emit_addressable_discovery_event( + tenant, + state, + channel_id, + KIND_NIP29_GROUP_METADATA, + &relay_pubkey_hex, + |channel, members| { + Ok(buzz_db::channel::group_metadata_tags(channel, members) + .into_iter() + .map(Tag::parse) + .collect::, _>>()?) + }, + ) + .await?; - { - let mut tags: Vec = vec![Tag::parse(["d", &group_id])?]; - tags.extend(group_state_identity_tags(&channel.identity())?); - for m in members - .iter() - .filter(|m| m.role == "owner" || m.role == "admin") - { - let pubkey_hex = hex::encode(&m.pubkey); - tags.push(Tag::parse(["p", &pubkey_hex, &m.role])?); - } - emit_addressable_discovery_event( - tenant, - state, - channel_id, - KIND_NIP29_GROUP_ADMINS, - tags, - &relay_pubkey_hex, - ) - .await?; - } + emit_addressable_discovery_event( + tenant, + state, + channel_id, + KIND_NIP29_GROUP_ADMINS, + &relay_pubkey_hex, + |channel, members| { + let mut tags: Vec = vec![Tag::parse(["d", &group_id])?]; + tags.extend(group_state_identity_tags(&channel.identity())?); + // The snapshot already scoped `members` to `GROUP_ADMIN_ROLES`. + for m in members { + let pubkey_hex = hex::encode(&m.pubkey); + tags.push(Tag::parse(["p", &pubkey_hex, &m.role])?); + } + Ok(tags) + }, + ) + .await?; - // Re-capture membership behind the writer lock immediately before the - // authoritative 39002 replacement. Metadata/admin snapshots retain their - // existing behavior; only membership publication needs this freshness fence. + // The roster takes the same fence behind the channel's membership lock, + // which its replacement also validates against. let relay_pubkey = state.relay_keypair.public_key().to_bytes(); let mut member_snapshot = state .db diff --git a/crates/buzz-relay/src/nip_fi_test_hooks.rs b/crates/buzz-relay/src/nip_fi_test_hooks.rs index 6ed3624ce5..9fa89d30fa 100644 --- a/crates/buzz-relay/src/nip_fi_test_hooks.rs +++ b/crates/buzz-relay/src/nip_fi_test_hooks.rs @@ -211,6 +211,15 @@ make_hook!(auto_join_channel_hook, before_auto_join_channel); // `after_auto_join`: fires when an admission auto-join task has finished, so // a test can assert what it did (or did not) write. make_hook!(auto_join_done_hook, after_auto_join); +// `before_group_state_lock`: fires immediately before a kind:39000/39001 +// publisher takes its replacement lock. Anything it read before this point +// was read outside the lock; a publisher held here while the channel is +// edited and republished must still store the edit's state. +make_hook!(group_state_lock_hook, before_group_state_lock); +// `after_group_state_read`: fires when a kind:39000/39001 publisher has read +// the channel and its members behind the lock, before it signs and stores. +// A publisher held here makes a racing edit wait for it. +make_hook!(group_state_read_hook, after_group_state_read); // `before_http_ban_recheck`: fires in `buzz_v1` and Git settings after the // shared membership check passes, immediately before the fresh ban reread. // A ban or closed pool here makes only the reread refuse.