fix(relay): read kind:39000/39001 inputs behind the replacement lock (#9035)

## Problem

`test_nip29_announce_channel_wire` flaked in CI (9/30 local solo runs
failed). `emit_group_discovery_events` read the channel, its members and
the latest live `created_at` *before* taking the coordinate's
replacement lock, and only locked for the write. Two publishers for one
channel could therefore store in the reverse order of their reads, or
land on the same second where NIP-16's lower-id tie-break picks a random
winner. Either way a kind:39000/39001 built from stale state could stay
live.

In the test, the relay owner reconnects to send an `auto_join=false`
edit, gets auto-joined into the channel, and the join's kind:39000 —
built from the old `auto_join=true` state — replaces the edit's.

## Fix

- **buzz-db:** `lock_group_state_snapshot` takes the replacement lock
first, then reads the channel, the members the kind actually publishes
(full roster for a DM's kind:39000, owners/admins for kind:39001, none
otherwise) and the latest timestamp; `LockedGroupStateSnapshot::replace`
stores on the same transaction. This is the fence kind:39002 already had
via `LockedMemberSnapshot`. The lock is taken once per write
(`take_replacement_lock`), and the write keeps its
`ReplaceAddressableEvent` transaction timer and writer population so
metrics stay continuous.
- **buzz-relay:** `emit_addressable_discovery_event` builds tags and
`created_at` (live + 1s) from that snapshot.

## Tests

- buzz-db: `group_state_snapshot_reads_behind_the_replacement_lock`,
`group_state_snapshot_reads_the_members_its_kind_publishes`,
`group_state_replacement_takes_its_lock_once`.
- buzz-relay: `a_publisher_that_read_before_an_edit_cannot_replace_it`,
with `BeforeLock` and `AfterRead` legs driven by new
`nip_fi_test_hooks`.
- Every new assertion was confirmed to fail when its guard is reverted.
- `test_nip29_announce_channel_wire` solo against a relay built from
this tree: 30/30 passed (9/30 failed before).

Merged with current `main`; the only conflict kept main's bounded
`channel state write failed: {kind}` error mapping on the new snapshot
write.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Matt Toohey <contact@matttoohey.com>
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

crates/buzz-db/src/store/channel.rs

@@ -516,6 +516,15 @@ async fn get_channel_with_operation(
     operation: crate::observability::WriterOperation,
 ) -> Result<ChannelRecord> {
     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<ChannelRecord> {
     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<Channe
     })
 }
 
+/// Whether [`group_metadata_tags`] reads the member roster for `channel`:
+/// only a DM's kind:39000 lists its participants. The locked snapshot that
+/// feeds it reads the roster under the same predicate.
+pub fn group_metadata_uses_members(channel: &ChannelRecord) -> 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<MemberRecord>,
+    /// `created_at` of the coordinate's live event, if it has one.
+    pub latest_created_at: Option<u64>,
+    kind: i32,
+    relay_pubkey: Vec<u8>,
+    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<LockedGroupStateSnapshot> {
+    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<DateTime<Utc>> = 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<LockedGroupStateSnapshot> {
+        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::<Vec<_>>()
+        };
+        // `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::<u64>()
+        };
+
+        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"
+        );
+    }
 }

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::<Result<Vec<_>>>()?;
+    let members = select_active_members(tx.conn(), community_id, channel_id).await?;
     let identity = sqlx::query_as::<_, (String, Vec<String>, Vec<u8>)>(
         "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<Vec<MemberRecord>> {
     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<Vec<MemberRecord>> {
+    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<Vec<MemberRecord>> {
+    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<Vec<MemberRecord>> {
     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()
 }

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<Uuid>,
+) -> 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<Uuid>,
+) -> 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<chrono::Utc>, Vec<u8>)> = 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<Uuid>,
     ) -> 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<chrono::Utc>, Vec<u8>)> =
-                    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
     }

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

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<chrono::DateTime<chrono::Utc>> {
+    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

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<AppState>,
     channel_id: Uuid,
     kind: u32,
-    tags: Vec<Tag>,
     relay_pubkey_hex: &str,
+    build_tags: impl FnOnce(
+        &buzz_db::channel::ChannelRecord,
+        &[MemberRecord],
+    ) -> anyhow::Result<Vec<Tag>>,
 ) -> 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<AppState>,
     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::<Result<Vec<_>, _>>()?;
-        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::<Result<Vec<_>, _>>()?)
+        },
+    )
+    .await?;
 
-    {
-        let mut tags: Vec<Tag> = 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<Tag> = 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

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.