fix(relay): publish group state from one locked snapshot

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

@@ -706,6 +706,32 @@ pub struct ReapedEphemeralChannel {
     pub channel_id: Uuid,
 }
 
+/// The channel row, deleted or not, on a caller-owned transaction, or `None`
+/// when it is missing.
+pub(crate) async fn channel_in_transaction(
+    tx: &mut crate::AdmittedTx,
+    channel_id: Uuid,
+) -> Result<Option<ChannelRecord>> {
+    let community_id = tx.community();
+    let row = sqlx::query(
+        r#"
+        SELECT id, name, channel_type::text AS channel_type, visibility::text AS visibility,
+               description, canvas,
+               created_by, created_at, updated_at, archived_at, deleted_at,
+               nip29_group_id, topic_required, max_members,
+               topic, topic_set_by, topic_set_at,
+               purpose, purpose_set_by, purpose_set_at,
+               ttl_seconds, ttl_deadline, labels, join_role, unlisted, auto_join
+        FROM channels WHERE community_id = $1 AND id = $2
+        "#,
+    )
+    .bind(community_id.as_uuid())
+    .bind(channel_id)
+    .fetch_optional(tx.conn())
+    .await?;
+    row.map(row_to_channel_record).transpose()
+}
+
 pub(crate) fn row_to_channel_record(row: sqlx::postgres::PgRow) -> Result<ChannelRecord> {
     let id: Uuid = row.try_get("id")?;
     let topic_required: bool = row.try_get("topic_required")?;

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

@@ -413,15 +413,26 @@ pub struct ChannelIdentity {
     pub created_by: Vec<u8>,
 }
 
-/// An active member roster captured while holding the channel's membership
-/// serialization lock on one writer connection.
+/// A channel's group state (row and active roster) captured while holding the
+/// replacement locks of its relay-signed group-state events (kinds
+/// 39000–39002) and its membership serialization lock, on one writer
+/// connection.
+///
+/// Every channel writer commits before it publishes group state, so an
+/// emission that captures and replaces under these locks either finishes
+/// before a later emission starts or reads that writer's committed change.
+/// An older capture therefore can never replace a newer one.
 pub struct LockedMemberSnapshot {
     /// Canonical active members captured behind the lock.
     pub members: Vec<MemberRecord>,
-    /// The channel's type, labels and creator, read on the same connection, so the
-    /// roster carries the same identity tags as the channel's other
-    /// group-state events. `None` when no channel row exists.
+    /// The channel's type, labels and creator, from the same row read as
+    /// `channel`, so the roster carries the same identity tags as the
+    /// channel's other group-state events. Present for a deleted channel;
+    /// `None` when no channel row exists.
     pub identity: Option<ChannelIdentity>,
+    /// The live channel row, read on the same connection. `None` when the
+    /// channel is missing or deleted.
+    pub channel: Option<ChannelRecord>,
     channel_id: Uuid,
     relay_pubkey: Vec<u8>,
     tx: AdmittedTx,
@@ -433,24 +444,28 @@ impl LockedMemberSnapshot {
         self.channel_id
     }
 
-    /// Return the newest relay-authored member snapshot timestamp for the
-    /// locked coordinate, using this guard's existing connection.
-    pub async fn latest_member_event_timestamp(&mut self) -> Result<Option<u64>> {
+    /// Return the newest relay-authored group-state event timestamp of `kind`
+    /// (39000–39002) for the locked channel, using this guard's existing
+    /// connection.
+    pub async fn latest_group_state_timestamp(&mut self, kind: i32) -> Result<Option<u64>> {
+        require_group_state_kind(kind)?;
         let value: Option<chrono::DateTime<Utc>> = sqlx::query_scalar(
-            "SELECT created_at FROM events WHERE community_id = $1 AND kind = 39002 AND pubkey = $2 AND channel_id = $3 AND deleted_at IS NULL ORDER BY created_at DESC, id ASC LIMIT 1",
+            "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(self.tx.community().as_uuid())
+        .bind(kind)
         .bind(self.relay_pubkey.as_slice())
         .bind(self.channel_id)
         .fetch_optional(self.tx.conn())
         .await?;
         Ok(value.map(|timestamp| timestamp.timestamp() as u64))
     }
 
-    /// Replace the relay-authored member snapshot on this guard's existing
-    /// connection. The membership lock therefore spans capture and replacement
-    /// without a nested pool checkout.
-    pub async fn replace_member_event(
+    /// Replace a relay-authored group-state event (kind 39000, 39001 or
+    /// 39002) of the locked channel on this guard's existing connection. The
+    /// locks therefore span capture and replacement without a nested pool
+    /// checkout.
+    pub async fn replace_group_state_event(
         &mut self,
         event: &nostr::Event,
     ) -> Result<(buzz_core::StoredEvent, bool)> {
@@ -460,15 +475,11 @@ impl LockedMemberSnapshot {
             || crate::event::extract_d_tag(event) != Some(channel_id.to_string())
         {
             return Err(DbError::InvalidData(
-                "member snapshot replacement does not match its locked coordinate".into(),
+                "group-state replacement does not match its locked coordinate".into(),
             ));
         }
         let kind = buzz_core::kind::event_kind_i32(event);
-        if kind != 39002 {
-            return Err(DbError::InvalidData(
-                "member snapshot replacement requires kind 39002".into(),
-            ));
-        }
+        require_group_state_kind(kind)?;
         let pubkey = 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)
@@ -499,7 +510,9 @@ impl LockedMemberSnapshot {
             }
         }
         let tags = serde_json::to_value(&event.tags)?;
-        validate_channel_roster_snapshot(&mut self.tx, channel_id, &tags).await?;
+        if kind == 39002 {
+            validate_channel_roster_snapshot(&mut self.tx, channel_id, &tags).await?;
+        }
         sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND channel_id = $4 AND deleted_at IS NULL")
             .bind(community_id.as_uuid()).bind(kind).bind(pubkey.as_slice()).bind(channel_id)
             .execute(self.tx.conn()).await?;
@@ -512,7 +525,7 @@ impl LockedMemberSnapshot {
             .bind(crate::event::extract_d_tag(event)).execute(self.tx.conn()).await?;
         if inserted.rows_affected() == 0 {
             return Err(DbError::InvalidData(
-                "member snapshot event id already exists".into(),
+                "group-state event id already exists".into(),
             ));
         }
         crate::store::event_follow_up::after_admitted_insert(
@@ -534,19 +547,34 @@ impl LockedMemberSnapshot {
         ))
     }
 
-    /// Commit the replacement and release the membership lock.
+    /// Commit the replacements and release the locks.
     pub async fn release(self) -> Result<()> {
         self.tx.commit().await?;
         Ok(())
     }
 }
 
-/// Capture all active members while holding the same per-channel lock used by
-/// membership writers.
+/// The relay-signed group-state kinds a [`LockedMemberSnapshot`] fences, in
+/// the order it takes their replacement locks.
+const GROUP_STATE_KINDS: [i32; 3] = [39000, 39001, 39002];
+
+fn require_group_state_kind(kind: i32) -> Result<()> {
+    if GROUP_STATE_KINDS.contains(&kind) {
+        Ok(())
+    } else {
+        Err(DbError::InvalidData(format!(
+            "group-state replacement requires kind 39000, 39001 or 39002, not {kind}"
+        )))
+    }
+}
+
+/// Capture the channel row and all active members while holding the
+/// replacement locks of the channel's group-state events and the same
+/// per-channel lock used by membership writers.
 ///
 /// The returned guard must remain alive through publication. This prevents a
-/// rolling relay from publishing an older roster after a concurrent add or
-/// remove has committed and published newer membership state.
+/// relay from publishing older group state after a concurrent metadata edit,
+/// add or remove has committed and published newer state.
 pub async fn lock_member_snapshot(
     pool: &PgPool,
     community_id: CommunityId,
@@ -559,23 +587,27 @@ pub async fn lock_member_snapshot(
         crate::observability::WriterOperation::EventWrite,
     )
     .await?;
-    // Match the canonical replacement writer's lock order. Old binaries take
-    // this key before INSERT; migration 0032 then takes the membership key in
-    // the INSERT trigger. Taking both in that order avoids mixed-version
-    // duplicate heads without introducing a lock-order inversion.
-    let replacement_lock = crate::replaceable::event_replacement_lock_key(
-        community_id,
-        39002,
-        relay_pubkey,
-        Some(channel_id.as_bytes()),
-    );
-    crate::observability::observe_advisory_lock(
-        crate::observability::LockType::Replacement,
-        sqlx::query("SELECT pg_advisory_xact_lock($1)")
-            .bind(replacement_lock)
-            .execute(tx.conn()),
-    )
-    .await?;
+    // Match the canonical replacement writer's lock order: each coordinate's
+    // replacement key, then (migration 0032's INSERT trigger, for 39002) the
+    // membership key. The keys are the ones `replace_addressable_event` takes,
+    // so a relay that still writes each kind on its own transaction
+    // serializes with this one. No writer holds one of these keys while it
+    // waits for another, so taking all three in kind order cannot invert.
+    for kind in GROUP_STATE_KINDS {
+        let replacement_lock = crate::replaceable::event_replacement_lock_key(
+            community_id,
+            kind,
+            relay_pubkey,
+            Some(channel_id.as_bytes()),
+        );
+        crate::observability::observe_advisory_lock(
+            crate::observability::LockType::Replacement,
+            sqlx::query("SELECT pg_advisory_xact_lock($1)")
+                .bind(replacement_lock)
+                .execute(tx.conn()),
+        )
+        .await?;
+    }
     acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await?;
     let rows = sqlx::query(
         r#"
@@ -594,21 +626,16 @@ pub async fn lock_member_snapshot(
         .into_iter()
         .map(row_to_member_record)
         .collect::<Result<Vec<_>>>()?;
-    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",
-    )
-    .bind(community_id.as_uuid())
-    .bind(channel_id)
-    .fetch_optional(tx.conn())
-    .await?
-    .map(|(channel_type, labels, created_by)| ChannelIdentity {
-        channel_type,
-        labels,
-        created_by,
-    });
+    // One read feeds both: the roster's identity tags (kept for a deleted
+    // channel) and the live row behind 39000/39001. Label edits do not take
+    // the membership lock, so two reads could straddle one.
+    let row = crate::channel::channel_in_transaction(&mut tx, channel_id).await?;
+    let identity = row.as_ref().map(ChannelRecord::identity);
+    let channel = row.filter(|channel| channel.deleted_at.is_none());
     Ok(LockedMemberSnapshot {
         members,
         identity,
+        channel,
         channel_id,
         relay_pubkey: relay_pubkey.to_vec(),
         tx,
@@ -4379,7 +4406,7 @@ mod postgres_tests {
             .sign_with_keys(&relay_keys)
             .expect("sign roster");
         let (_, inserted) = snapshot
-            .replace_member_event(&event)
+            .replace_group_state_event(&event)
             .await
             .expect("replace roster");
         assert!(inserted);
@@ -4717,7 +4744,7 @@ mod postgres_tests {
                 .await
                 .expect("lock snapshot");
         let error = snapshot
-            .replace_member_event(&stale)
+            .replace_group_state_event(&stale)
             .await
             .expect_err("app must reject stale roster without trigger");
         assert!(matches!(error, DbError::Sqlx(sqlx::Error::Database(ref db))
@@ -4732,7 +4759,7 @@ mod postgres_tests {
                 .expect("lock current snapshot");
         assert!(
             snapshot
-                .replace_member_event(&current)
+                .replace_group_state_event(&current)
                 .await
                 .expect("current roster")
                 .1
@@ -5081,9 +5108,10 @@ mod postgres_tests {
         assert_eq!(count, 0);
     }
 
-    /// A captured roster holds the same lock as membership writers until the
-    /// publisher explicitly releases it. This is the freshness fence used by
-    /// rolling-deploy reconciliation.
+    /// A captured snapshot holds the same lock as membership writers, and the
+    /// replacement keys of every group-state kind, until the publisher
+    /// explicitly releases it. This is the freshness fence used by group-state
+    /// publication and rolling-deploy reconciliation.
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn locked_member_snapshot_blocks_post_capture_membership_mutation() {
@@ -5121,6 +5149,16 @@ mod postgres_tests {
         .await
         .expect("capture locked roster");
         assert_eq!(snapshot.members.len(), 1);
+        assert_eq!(
+            snapshot.channel.as_ref().map(|captured| captured.id),
+            Some(channel.id),
+            "the live channel row is captured behind the same locks"
+        );
+        assert_eq!(
+            snapshot.identity,
+            snapshot.channel.as_ref().map(ChannelRecord::identity),
+            "the roster's identity comes from the captured row, not a second read"
+        );
         let event = nostr::EventBuilder::new(nostr::Kind::Custom(39002), "")
             .tags(vec![
                 nostr::Tag::parse(["d", &channel.id.to_string()]).expect("d tag"),
@@ -5137,13 +5175,13 @@ mod postgres_tests {
             .expect("sign roster for another channel");
         assert!(
             matches!(
-                snapshot.replace_member_event(&other_channel).await,
+                snapshot.replace_group_state_event(&other_channel).await,
                 Err(DbError::InvalidData(_))
             ),
             "a roster whose d tag names another channel must not replace the locked one"
         );
         let (_, inserted) = snapshot
-            .replace_member_event(&event)
+            .replace_group_state_event(&event)
             .await
             .expect("replace roster on held connection");
         assert!(inserted);
@@ -5163,6 +5201,24 @@ mod postgres_tests {
             !acquired,
             "membership mutation must wait until the captured roster is published"
         );
+        // A relay that still replaces 39000 and 39001 on their own
+        // transactions takes these keys, so it waits for this publication too.
+        for kind in [39000, 39001] {
+            let acquired: bool = sqlx::query_scalar("SELECT pg_try_advisory_xact_lock($1)")
+                .bind(crate::replaceable::event_replacement_lock_key(
+                    community,
+                    kind,
+                    &relay_keys.public_key().to_bytes(),
+                    Some(channel.id.as_bytes()),
+                ))
+                .fetch_one(&mut *contender)
+                .await
+                .expect("try group-state replacement lock");
+            assert!(
+                !acquired,
+                "a kind:{kind} replacement must wait until the captured state is published"
+            );
+        }
         contender.rollback().await.expect("rollback contender");
 
         snapshot.release().await.expect("release snapshot fence");
@@ -5185,6 +5241,45 @@ mod postgres_tests {
         );
     }
 
+    /// A deleted channel has no live row for 39000/39001, but its roster keeps
+    /// the identity tags, so reconciliation still publishes a valid 39002.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn locked_member_snapshot_of_deleted_channel_keeps_identity() {
+        let pool = setup_pool().await;
+        let community_id = make_test_community(&pool).await;
+        let community = CommunityId::from_uuid(community_id);
+        let owner = random_pubkey();
+        let channel = create_test_channel(
+            &pool,
+            community_id,
+            "snapshot-deleted-identity",
+            ChannelType::Stream,
+            ChannelVisibility::Open,
+            None,
+            &owner,
+            None,
+        )
+        .await
+        .expect("create channel");
+        assert!(
+            crate::channel::soft_delete_channel(&pool, community, channel.id)
+                .await
+                .expect("soft delete channel")
+        );
+
+        let snapshot = lock_member_snapshot(&pool, community, channel.id, &[7u8; 32])
+            .await
+            .expect("capture deleted channel");
+        assert!(
+            snapshot.channel.is_none(),
+            "a deleted channel has no live row"
+        );
+        assert_eq!(snapshot.identity, Some(channel.identity()));
+        assert!(snapshot.members.is_empty());
+        snapshot.release().await.expect("release snapshot");
+    }
+
     /// Captured roster publication must hold the same shared community-deletion
     /// lock as serving event writes so deletion fencing can serialize against it.
     #[tokio::test]