From d0e0307aeb7901cd1de11dac6f94c79783499396 Mon Sep 17 00:00:00 2001 From: Bryan Pfremmer <1383402+pfremm@users.noreply.github.com> Date: Sun, 1 Nov 2026 20:43:47 +0000 Subject: [PATCH] fix(relay): fence kind:39000/39001 snapshots behind the replacement lock emit_group_discovery_events read the channel and roster outside any lock and only took the replacement lock around the write, so an emission holding an older snapshot could store it after a newer one. Since #8594 this happens routinely: a relay admin reconnecting to clear auto_join spawns an admission auto-join that read the channel while auto_join was still true, and its kind:39000 lands after the false edit's, advertising auto_join durably. That is the Relay E2E failure in test_nip29_announce_channel_wire on main (CI run 41341005385, e2e_relay.rs:3847). Read the channel, roster and newest head inside the same transaction that holds the (kind, relay pubkey, channel) replacement lock, as kind:39002 already does, and replace before releasing it. Signed-off-by: Bryan Pfremmer <1383402+pfremm@users.noreply.github.com> --- crates/buzz-db/src/store/channel.rs | 6 +- crates/buzz-db/src/store/channel_members.rs | 280 +++++++++++++++-- crates/buzz-db/src/store/replaceable.rs | 281 +++++++++--------- .../buzz-relay/src/handlers/side_effects.rs | 134 ++++----- 4 files changed, 474 insertions(+), 227 deletions(-) diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 36ae4a1dc6b..56a6d68e959 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -27,9 +27,9 @@ pub use crate::channel_members::{ add_member, get_accessible_channel_ids, get_accessible_channels, get_bot_members, get_member_count, get_member_counts_bulk, get_member_role, get_members, get_members_bulk, get_users_bulk, is_member, list_large_channel_rosters_needing_reconciliation, - lock_member_snapshot, membership_pairs, remove_member, AccessibleChannel, BotChannelEntry, - BotMemberRecord, ChannelIdentity, LargeChannelRoster, LockedMemberSnapshot, MemberRecord, - UserRecord, + lock_group_state_snapshot, lock_member_snapshot, membership_pairs, remove_member, + AccessibleChannel, BotChannelEntry, BotMemberRecord, ChannelIdentity, LargeChannelRoster, + LockedGroupStateSnapshot, LockedMemberSnapshot, MemberRecord, UserRecord, }; /// A channel row as returned from the database. diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 2c139228a8e..0411bab88c9 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -413,6 +413,135 @@ pub struct ChannelIdentity { pub created_by: Vec, } +/// A channel row and active roster read while holding the replacement lock +/// for one relay-authored group-state coordinate (kind 39000 or 39001). +/// +/// Holding the coordinate lock across read, sign, and replace means a +/// publisher that read older channel state cannot store its event after a +/// publisher that read newer state: every state writer commits before it +/// publishes, so the later publisher waits here and reads the committed row. +pub struct LockedGroupStateSnapshot { + /// Channel row captured behind the lock. + pub channel: ChannelRecord, + /// Active members captured behind the lock. + pub members: Vec, + /// `created_at` (Unix seconds) of the live event at the locked coordinate. + pub latest_created_at: Option, + kind: i32, + relay_pubkey: Vec, + tx: AdmittedTx, +} + +impl LockedGroupStateSnapshot { + /// Replace the locked coordinate's event, then commit and release the + /// lock. Returns `(event, false)`, rolling back, when the event is stale + /// or a duplicate; callers should skip fan-out in that case. + pub async fn replace_and_commit( + self, + event: &nostr::Event, + ) -> Result<(buzz_core::StoredEvent, bool)> { + let channel_id = self.channel.id; + if event.pubkey.to_bytes().as_slice() != self.relay_pubkey.as_slice() + || buzz_core::kind::event_kind_i32(event) != self.kind + || 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(), + )); + } + let mut tx = self.tx; + let (stored, inserted) = crate::replaceable::replace_addressable_event_in_transaction( + &mut tx, + event, + Some(channel_id), + ) + .await?; + if inserted { + tx.commit().await?; + } else { + tx.rollback().await?; + } + Ok((stored, inserted)) + } +} + +/// Lock a relay-authored kind:39000 or kind:39001 coordinate and read the +/// channel and active roster on the same transaction. +/// +/// Lock order: the coordinate's replacement key only, the same first key that +/// [`crate::Db::replace_addressable_event`] and [`lock_member_snapshot`] take. +/// The membership key is not needed: the reads take no locks, and channel and +/// membership writers commit before they publish, so they never hold a row or +/// membership lock while waiting on this key. +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( + "group-state snapshot lock requires kind 39000 or 39001".into(), + )); + } + let mut tx = crate::begin_community_event_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; + let replacement_lock = crate::replaceable::event_replacement_lock_key( + community_id, + kind, + relay_pubkey, + Some(channel_id.as_bytes()), + ); + crate::replaceable::acquire_event_replacement_lock(&mut tx, replacement_lock).await?; + // Each statement below takes its snapshot after the lock is granted. + let channel = get_channel_tx(tx.conn(), community_id, channel_id).await?; + let members = get_active_members_tx(tx.conn(), community_id, channel_id).await?; + let latest: 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.map(|timestamp| timestamp.timestamp() as u64), + kind, + relay_pubkey: relay_pubkey.to_vec(), + tx, + }) +} + +/// Read a channel's active members, oldest first, on `conn`. +async fn get_active_members_tx( + conn: &mut PgConnection, + community_id: CommunityId, + channel_id: Uuid, +) -> 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 + ORDER BY cm.joined_at ASC + "#, + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_all(&mut *conn) + .await?; + rows.into_iter().map(row_to_member_record).collect() +} + /// An active member roster captured while holding the channel's membership /// serialization lock on one writer connection. pub struct LockedMemberSnapshot { @@ -577,23 +706,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 = get_active_members_tx(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", ) @@ -2176,6 +2289,23 @@ pub async fn get_member_role( } impl Db { + /// Lock a relay-authored kind:39000/39001 coordinate and capture the + /// channel and roster it will publish. See [`lock_group_state_snapshot`]. + #[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 { + let mut snapshot = + lock_group_state_snapshot(&self.pool, community_id, channel_id, kind, relay_pubkey) + .await?; + snapshot.tx.set_push_enqueue(self.push_enqueue.clone()); + Ok(snapshot) + } + /// Capture the active roster while holding the membership-writer lock. #[datastore_span(name = "lock_member_snapshot", system = "postgresql")] pub async fn lock_member_snapshot( @@ -4690,6 +4820,122 @@ mod postgres_tests { .expect("sign roster") } + /// A kind:39000 publisher that read the channel before an `auto_join` + /// flip committed cannot leave its stale snapshot as the live head: the + /// flip's own publisher waits on the coordinate lock, then reads the + /// committed row and the stale head, and replaces it with a newer event. + /// + /// Mutations: drop the replacement lock from `lock_group_state_snapshot` + /// → the second publisher does not wait (RED); read the live head before + /// taking the lock → the second publisher misses the stale head's + /// timestamp (RED). + #[tokio::test] + #[ignore = "requires Postgres"] + async fn group_state_snapshot_fences_stale_metadata_publication() { + use crate::channel::{update_channel, ChannelUpdate}; + let pool = setup_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, + "group-state-fence", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner, + None, + ) + .await + .expect("create channel"); + let set_auto_join = |auto_join| { + update_channel( + &pool, + community, + channel.id, + ChannelUpdate { + auto_join: Some(auto_join), + ..Default::default() + }, + ) + }; + set_auto_join(true).await.expect("flag channel"); + let keys = Keys::generate(); + let relay = keys.public_key().to_bytes(); + let sign = |snapshot: &LockedGroupStateSnapshot| { + let created_at = snapshot.latest_created_at.map_or(1_000, |ts| ts + 1); + let tags = crate::channel::group_metadata_tags(&snapshot.channel, &snapshot.members) + .into_iter() + .map(|tag| nostr::Tag::parse(tag).expect("metadata tag")); + nostr::EventBuilder::new(nostr::Kind::Custom(39000), "") + .tags(tags) + .custom_created_at(nostr::Timestamp::from(created_at)) + .sign_with_keys(&keys) + .expect("sign metadata") + }; + let has_auto_join = |event: &nostr::Event| { + event + .tags + .iter() + .any(|tag| tag.as_slice()[0] == "auto_join") + }; + + let stale = lock_group_state_snapshot(&pool, community, channel.id, 39000, &relay) + .await + .expect("lock stale snapshot"); + assert!(stale.channel.auto_join); + set_auto_join(false).await.expect("clear auto_join"); + + let fresh = tokio::spawn({ + let pool = pool.clone(); + async move { lock_group_state_snapshot(&pool, community, channel.id, 39000, &relay).await } + }); + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + assert!( + !fresh.is_finished(), + "the second publisher waits for the first" + ); + + let stale_event = sign(&stale); + assert!(has_auto_join(&stale_event)); + assert!( + stale + .replace_and_commit(&stale_event) + .await + .expect("replace stale") + .1 + ); + + let fresh = fresh.await.expect("join").expect("lock fresh snapshot"); + assert!(!fresh.channel.auto_join, "reads the committed flip"); + assert_eq!( + fresh.latest_created_at, + Some(stale_event.created_at.as_secs()), + "reads the head committed while it waited" + ); + let fresh_event = sign(&fresh); + assert!(!has_auto_join(&fresh_event)); + assert!( + fresh + .replace_and_commit(&fresh_event) + .await + .expect("replace fresh") + .1 + ); + + let live_id: Vec = sqlx::query_scalar( + "SELECT id FROM events WHERE community_id=$1 AND channel_id=$2 \ + AND kind=39000 AND deleted_at IS NULL", + ) + .bind(community_uuid) + .bind(channel.id) + .fetch_one(&pool) + .await + .expect("load live metadata"); + assert_eq!(live_id, fresh_event.id.as_bytes().to_vec()); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn app_roster_guard_locked_writer_rejects_stale_and_accepts_current_without_trigger() { diff --git a/crates/buzz-db/src/store/replaceable.rs b/crates/buzz-db/src/store/replaceable.rs index 7c1789c3c49..d871bdf37cd 100644 --- a/crates/buzz-db/src/store/replaceable.rs +++ b/crates/buzz-db/src/store/replaceable.rs @@ -96,6 +96,142 @@ pub(crate) fn event_replacement_lock_key( hash as i64 } +/// Serialize all writers for one `(kind, pubkey, coordinate)` tuple on `tx`. +/// +/// The advisory lock is transaction-scoped and released on commit or rollback. +pub(crate) async fn acquire_event_replacement_lock( + tx: &mut AdmittedTx, + lock_key: i64, +) -> Result<()> { + observability::observe_advisory_lock( + LockType::Replacement, + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(lock_key) + .execute(tx.conn()), + ) + .await?; + Ok(()) +} + +/// Replace an addressable event on a caller-owned transaction that already +/// holds the coordinate's [`acquire_event_replacement_lock`]. +/// +/// Returns `(event, false)` for stale writes and duplicate IDs. The caller +/// must commit only when the event was inserted and roll back otherwise: a +/// duplicate ID is detected after the previous head was soft-deleted. +pub(crate) async fn replace_addressable_event_in_transaction( + 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 { + return Ok(( + StoredEvent::with_received_at(event.clone(), chrono::Utc::now(), 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( + "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, + )) +} + /// Replace a parameterized event in a caller-owned transaction. /// /// This function acquires a transaction-scoped advisory lock but never commits @@ -410,17 +546,11 @@ 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(), + buzz_core::kind::event_kind_i32(event), + event.pubkey.to_bytes().as_slice(), channel_id.as_ref().map(|id| id.as_bytes().as_slice()), ); @@ -437,136 +567,15 @@ 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( - "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. + acquire_event_replacement_lock(&mut tx, lock_key).await?; + let (stored, was_inserted) = + replace_addressable_event_in_transaction(&mut tx, event, channel_id).await?; + if was_inserted { + tx.commit().await?; + } else { 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( - &mut 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(&mut tx, event, channel_id) - .await?; - - tx.commit().await?; - - Ok(( - StoredEvent::with_received_at(event.clone(), received_at, channel_id, true), - true, - )) + Ok((stored, was_inserted)) }) .await } diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 8f2977bb04e..edea56a5a19 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1297,15 +1297,31 @@ async fn emit_typed_membership_notification( Ok(()) } -/// Sign, store (replacing previous), and fan-out a single addressable discovery event. +/// Sign, store (replacing previous), and fan-out one relay-authored kind:39000 +/// or kind:39001 event. +/// +/// `build_tags` receives the channel and roster read while holding the +/// coordinate's replacement lock, which is held until the replacement commits. +/// An emission that started before a concurrent channel or membership change +/// committed therefore cannot store its stale snapshot after the newer one. async fn emit_addressable_discovery_event( tenant: &TenantContext, state: &Arc, channel_id: Uuid, kind: u32, - tags: Vec, + build_tags: impl FnOnce( + &buzz_db::channel::ChannelRecord, + &[MemberRecord], + ) -> anyhow::Result>, relay_pubkey_hex: &str, ) -> anyhow::Result<()> { + let relay_pubkey = state.relay_keypair.public_key().to_bytes(); + let snapshot = state + .db + .lock_group_state_snapshot(tenant.community(), channel_id, kind as i32, &relay_pubkey) + .await?; + let tags = build_tags(&snapshot.channel, &snapshot.members)?; + // 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 @@ -1315,23 +1331,11 @@ async fn emit_addressable_discovery_event( .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) @@ -1339,10 +1343,7 @@ 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)) - .await?; + let (stored, was_inserted) = snapshot.replace_and_commit(&event).await?; if was_inserted { let kind_u32 = event_kind_u32(&stored.event); dispatch_persistent_event(tenant, state, &stored, kind_u32, relay_pubkey_hex, None).await; @@ -1450,58 +1451,49 @@ 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?; - } + // Each kind re-reads the channel and roster behind its own coordinate lock, + // so a concurrent emission holding an older snapshot cannot overwrite a + // newer one (e.g. auto-join publishing `auto_join` after it was cleared). + emit_addressable_discovery_event( + tenant, + state, + channel_id, + KIND_NIP29_GROUP_METADATA, + |channel, members| { + Ok(buzz_db::channel::group_metadata_tags(channel, members) + .into_iter() + .map(Tag::parse) + .collect::, _>>()?) + }, + &relay_pubkey_hex, + ) + .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, + |channel, members| { + let mut tags: Vec = vec![Tag::parse(["d", &channel.id.to_string()])?]; + 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])?); + } + Ok(tags) + }, + &relay_pubkey_hex, + ) + .await?; - // Re-capture membership behind the writer lock immediately before the - // authoritative 39002 replacement. Metadata/admin snapshots retain their - // existing behavior; only membership publication needs this freshness fence. + // Re-capture membership behind the replacement and membership-writer locks + // immediately before the authoritative 39002 replacement. let relay_pubkey = state.relay_keypair.public_key().to_bytes(); let mut member_snapshot = state .db