fix(relay): apply channel setting changes atomically and lock every membership writer (#8516)

🤖
## Summary
This PR fixes two ways the relay could report a channel change that did
not fully happen.

**1. Channel settings could be saved only in part, while the app was
told that they all worked.**
When an owner or admin changed a channel's settings (name, description,
topic, public or private, message lifetime, archived or not, who can
join), the relay saved the request first and then applied each setting
with its own database write. If one write failed, the relay only logged
the error. It skipped the remaining settings and still reported success.
Sending the request again did nothing, because the relay saw it as
already saved. Example: a switch from public to private could fail while
the app showed success, and the channel stayed public.

Now saving the request, applying every setting and saving the "settings
changed" messages in the channel happen in one database transaction.
Either all of it is saved or none of it is:
- If anything fails, nothing is saved and the client gets an error. It
can send the same request again.
- A repeat of a request that already succeeded is reported as a
duplicate and changes nothing. This is also true when the repeat arrives
while the first copy is still being saved.
- A request to restore a channel that is not archived, or to change an
archived channel without restoring it, is refused and nothing is saved.
Before, the relay saved the request and a "settings changed" message,
and then the restore failed.
- Live updates go out only after the save commits. These are: removing
access for people who lose it, and pushing the new messages and channel
info to clients.

**2. A published member list could include someone who had just been
removed.**
For some channels the relay publishes a list of the channel's members.
While it builds that list, it holds a per-channel membership lock. Kicks
by a relay admin and the automatic kick after a deploy did not take that
lock. A kick could commit after the list was read, so the published list
still showed the removed member. Channel deletion also did not wait for
the lock. Kicks, channel deletion and the automatic add when someone
joins a Huddle (voice call) now all take the lock first. The published
list then includes the change, or the change waits until the list is
saved.

## Details
- **One lock order for all channel settings writes.** Before, each
writer took its own locks. Review kept finding writers that took them in
a different order or forgot one. `lock_channel_for_settings_write`
(`crates/buzz-db/src/store/channel.rs`) is now the only place that takes
them. The order is:
1. the membership lock, only when the public/private setting or the "who
can join" setting (`join_role`) changes. Without it, a member add could
read the old rule while the change commits.
2. the message-lifetime (TTL) lock: exclusive for a lifetime change,
shared otherwise.
3. the channel row (`FOR NO KEY UPDATE`). This does not block the key
lock that a member insert takes. The row lock also orders two copies of
the same request, so the second copy sees the first as already stored.
- The function returns a `ChannelSettingsWriter`. The writer borrows the
transaction that took the locks, so the compiler stops it from being
used with another transaction or after commit. The settings writes
(`update_channel_on`, `set_topic_on`, `set_purpose_on`,
`set_archived_on`) require the writer. They refuse a change its locks do
not cover.
- `Db::apply_channel_metadata_event` stores the channel-edit event
(kind:9002) and its mentions, applies the settings and stores the
"settings changed" messages (kind:40099) in one transaction. The older
single-purpose writes (`update_channel`, `set_topic`, `set_purpose`,
`archive_channel`, `unarchive_channel`) use the same helpers and behave
as before.
- `ingest.rs` and `validate_admin_event`: channel edits no longer check
archived/active state before the write. The transaction checks it after
its duplicate check, under the row lock. A refusal there goes back to
the client as `invalid:`. Channel edits have their own save path and
skip the general post-save step. `handle_side_effects` refuses them, so
no caller can go around the transaction.
- `side_effects.rs`: `emit_system_message` is split into signing and
sending. The messages are signed before the transaction and sent after
it. All "settings changed" messages from one edit share one timestamp
and come in a fixed order (topic, purpose, visibility, lifetime,
archive).
- A change to "who can join" (`join_role`) is applied in the same
transaction and announced with its own `join_role_changed` message.
- **Membership lock outside settings** (`relay_admin_actions.rs`,
`channel_members.rs`, `audio/handler.rs`): both kick paths
(`execute_kick_with_marker`, `deploy_kick_member`) take the membership
lock before they remove the member. The marker kick still undoes the
removal if its lease check fails after waiting. The Huddle join takes
its locks through a new `ChannelMembershipLocks` token, and the auto-add
is a method on that token, so it cannot run without the lock.
`soft_delete_channel` takes the full settings lock order.
- **Limit of the membership lock:** it covers member changes,
public/private and "who can join" changes, and channel deletion. It does
not cover archiving: archive writes and the cleanup of expired temporary
channels do not take it. The cleanup keeps its single UPDATE. A comment
in `channel.rs` explains why that is safe and why it must not be split
into a read and a write.

### Related issue
Follow-up to the review of #8143 (which added the "who can join"
setting). That review found that channel edits could be applied in part.
No separate issue.

### Testing
New PostgreSQL tests in
`crates/buzz-relay/src/handlers/edit_metadata_postgres_tests.rs`:
- `failed_edit_stores_nothing`: a test trigger makes one write fail
partway through the transaction. Nothing is stored or changed, on the
first send or a resend. Without the trigger, the same signed request
applies in full.
- `resent_edit_is_a_duplicate`,
`resent_archive_and_unarchive_are_duplicates`: a resend after success is
a duplicate and adds nothing.
- `resend_validated_after_the_first_commit_is_a_duplicate`: a resend
validated after its first copy committed is a duplicate, with no second
change and no second message.
- `concurrent_copies_of_a_restore_apply_once`: two copies of one
restore, held at the channel lock until both are waiting, apply once.
The other copy is a duplicate.
- `unarchive_of_an_active_channel_is_refused`,
`join_role_edit_is_applied_and_announced`.

Lock tests in buzz-db: each one holds a member list in progress and
checks that the writer waits for it. They cover a "who can join" change
(`channel_members.rs`), the deploy kick and the marker kick
(`relay_admin_actions.rs`), and channel deletion (`channel.rs`). The
kick and delete tests fail when the new locks are removed.

These tests and the existing channel-edit tests in
`group_state_postgres_tests` pass locally against a migrated PostgreSQL
database.

---------

Signed-off-by: Logan Johnson <loganj@squareup.com>
Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>

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

@@ -768,6 +768,24 @@ pub struct ChannelUpdate {
     pub join_role: Option<String>,
 }
 
+impl ChannelUpdate {
+    /// True when no field is set.
+    pub fn is_empty(&self) -> bool {
+        self.name.is_none()
+            && self.description.is_none()
+            && self.visibility.is_none()
+            && self.ttl_seconds.is_none()
+            && self.labels.is_none()
+            && self.join_role.is_none()
+    }
+
+    /// True when the update changes a setting that decides admission:
+    /// visibility or `join_role`.
+    pub fn changes_admission(&self) -> bool {
+        self.visibility.is_some() || self.join_role.is_some()
+    }
+}
+
 /// Updates channel metadata dynamically.
 ///
 /// At least one field must be provided; returns `InvalidData` otherwise.
@@ -776,20 +794,198 @@ pub async fn update_channel(
     pool: &PgPool,
     community_id: CommunityId,
     channel_id: Uuid,
-    mut updates: ChannelUpdate,
+    updates: ChannelUpdate,
 ) -> Result<ChannelRecord> {
-    if updates.name.is_none()
-        && updates.description.is_none()
-        && updates.visibility.is_none()
-        && updates.ttl_seconds.is_none()
-        && updates.labels.is_none()
-        && updates.join_role.is_none()
-    {
+    if updates.is_empty() {
         return Err(DbError::InvalidData(
             "at least one field must be provided for update".to_string(),
         ));
     }
 
+    let mut tx = crate::begin_community_write_transaction(
+        pool,
+        community_id,
+        crate::observability::WriterOperation::EventWrite,
+    )
+    .await?;
+    let mut locked = lock_channel_for_settings_write(&mut tx, channel_id, &updates).await?;
+    update_channel_on(&mut locked, updates).await?;
+    tx.commit().await?;
+
+    get_channel_with_operation(
+        pool,
+        community_id,
+        channel_id,
+        crate::observability::WriterOperation::EventWrite,
+    )
+    .await
+}
+
+#[derive(Clone, Copy)]
+enum TtlLock {
+    Shared,
+    Exclusive,
+}
+
+/// Takes the per-channel TTL advisory lock (see [`channel_ttl_lock_key`]) on `tx`.
+async fn lock_channel_ttl(
+    tx: &mut crate::AdmittedTx,
+    channel_id: Uuid,
+    mode: TtlLock,
+) -> Result<()> {
+    let sql = match mode {
+        TtlLock::Shared => "SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))",
+        TtlLock::Exclusive => "SELECT pg_advisory_xact_lock(hashtextextended($1, 0))",
+    };
+    sqlx::query(sql)
+        .bind(channel_ttl_lock_key(tx.community(), channel_id))
+        .execute(tx.conn())
+        .await?;
+    Ok(())
+}
+
+/// The only way to write a channel's settings. Only
+/// [`lock_channel_for_settings_write`] makes one, after it takes the locks
+/// on the transaction it borrows, and every settings write is a method on
+/// it. The borrow binds the locks to that transaction: a writer cannot use
+/// them with another transaction or after its commit, skip a lock, or take
+/// the locks in a different order.
+pub(crate) struct ChannelSettingsWriter<'tx> {
+    tx: &'tx mut crate::AdmittedTx,
+    channel_id: Uuid,
+    admission: bool,
+    ttl: TtlLock,
+    previous_visibility: String,
+    archived: bool,
+}
+
+impl ChannelSettingsWriter<'_> {
+    /// The channel's visibility when the lock was taken (`"open"` or `"private"`).
+    pub(crate) fn previous_visibility(&self) -> &str {
+        &self.previous_visibility
+    }
+
+    /// Whether the channel was archived when the lock was taken. The row
+    /// lock keeps this current until the transaction ends.
+    pub(crate) fn archived(&self) -> bool {
+        self.archived
+    }
+
+    /// The locked transaction, for writes that are not channel settings
+    /// (for example, storing the event that requested the change).
+    pub(crate) fn tx(&mut self) -> &mut crate::AdmittedTx {
+        self.tx
+    }
+
+    /// Refuse `updates` if this lock does not cover them. This catches a
+    /// caller that took the lock for one set of changes and then writes
+    /// another set.
+    fn check_covers(&self, updates: &ChannelUpdate) -> Result<()> {
+        let covered = (self.admission || !updates.changes_admission())
+            && (matches!(self.ttl, TtlLock::Exclusive) || updates.ttl_seconds.is_none());
+        if covered {
+            Ok(())
+        } else {
+            Err(DbError::InvalidData(
+                "channel settings write is not covered by its lock".into(),
+            ))
+        }
+    }
+}
+
+/// Takes, on `tx`, every lock a write to `channel_id`'s settings needs, in
+/// the one order that all channel writers use:
+///
+/// 1. the channel's membership lock, when `updates` changes visibility or
+///    `join_role`. Those two settings decide admission, and `put_member` and
+///    a huddle join decide admission under this lock. So an admission either
+///    sees the old setting and commits first, or sees the new one.
+/// 2. the channel's TTL lock: EXCLUSIVE for a TTL change, SHARED otherwise.
+///    A TTL change can switch the event-commit fast path
+///    (`event_follow_up::refresh_channel_ttls`, which reads `ttl_seconds`
+///    under the SHARED lock). With the EXCLUSIVE lock, a concurrent event
+///    either sees the new TTL or commits before this change.
+/// 3. the channel row, `FOR NO KEY UPDATE`.
+///
+/// Roster publication holds the membership lock and then, at commit, takes
+/// the TTL lock and updates the row. A huddle join takes the membership
+/// lock, then the TTL lock, then the row. A writer that took these locks in
+/// any other order could wait for one of them while it waits for this one.
+///
+/// Returns `ChannelNotFound` when the channel is missing or deleted.
+pub(crate) async fn lock_channel_for_settings_write<'tx>(
+    tx: &'tx mut crate::AdmittedTx,
+    channel_id: Uuid,
+    updates: &ChannelUpdate,
+) -> Result<ChannelSettingsWriter<'tx>> {
+    let ttl = if updates.ttl_seconds.is_some() {
+        TtlLock::Exclusive
+    } else {
+        TtlLock::Shared
+    };
+    lock_channel_row(tx, channel_id, updates.changes_admission(), ttl).await
+}
+
+/// Takes the locks that deleting `channel_id` needs, in the same order as
+/// [`lock_channel_for_settings_write`]. A delete ends admission, so it takes
+/// the membership lock: a membership write or a huddle join either commits
+/// first or, under the lock, finds the channel deleted. A delete does not
+/// change the TTL, so the TTL lock is SHARED.
+async fn lock_channel_for_delete(
+    tx: &mut crate::AdmittedTx,
+    channel_id: Uuid,
+) -> Result<ChannelSettingsWriter<'_>> {
+    lock_channel_row(tx, channel_id, true, TtlLock::Shared).await
+}
+
+async fn lock_channel_row(
+    tx: &mut crate::AdmittedTx,
+    channel_id: Uuid,
+    admission: bool,
+    ttl: TtlLock,
+) -> Result<ChannelSettingsWriter<'_>> {
+    let community_id = tx.community();
+    if admission {
+        crate::channel_members::acquire_channel_membership_lock(
+            tx.conn(),
+            community_id,
+            channel_id,
+        )
+        .await?;
+    }
+    lock_channel_ttl(tx, channel_id, ttl).await?;
+    // `FOR NO KEY UPDATE`, like a huddle join: it orders this write against
+    // every other settings write and a huddle join, but not against the
+    // `KEY SHARE` lock that a membership INSERT takes for its foreign key.
+    let (previous_visibility, archived): (String, bool) = sqlx::query_as(
+        "SELECT visibility::text, archived_at IS NOT NULL FROM channels \
+         WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL FOR NO KEY UPDATE",
+    )
+    .bind(community_id.as_uuid())
+    .bind(channel_id)
+    .fetch_optional(tx.conn())
+    .await?
+    .ok_or(DbError::ChannelNotFound(channel_id))?;
+    Ok(ChannelSettingsWriter {
+        tx,
+        channel_id,
+        admission,
+        ttl,
+        previous_visibility,
+        archived,
+    })
+}
+
+/// Writes the set fields of `updates` to the locked channel row on `tx`.
+/// The caller owns the commit.
+async fn update_channel_on(
+    w: &mut ChannelSettingsWriter<'_>,
+    mut updates: ChannelUpdate,
+) -> Result<()> {
+    w.check_covers(&updates)?;
+    let channel_id = w.channel_id;
+    let tx = &mut *w.tx;
+    let community_id = tx.community();
     if let Some(name) = updates.name.as_mut() {
         *name = buzz_core::channel::canonical_channel_name(name).to_owned();
         if name.is_empty() {
@@ -861,57 +1057,11 @@ pub async fn update_channel(
     q = q.bind(community_id.as_uuid());
     q = q.bind(channel_id);
 
-    // Every writer that takes more than one of a channel's locks takes them
-    // in this order: membership lock, then TTL lock, then the channel row.
-    // Roster publication holds the membership lock and then refreshes the
-    // TTL (TTL lock, row update) at commit; a huddle join takes the
-    // membership lock before it locks the row. Taking the row first here
-    // would let either of them wait on this transaction while it waits on
-    // them.
-    let mut tx = crate::begin_community_write_transaction(
-        pool,
-        community_id,
-        crate::observability::WriterOperation::EventWrite,
-    )
-    .await?;
-    // Visibility and `join_role` decide admission, which `put_member` and a
-    // huddle join decide under the channel's membership lock. Take that lock
-    // too, so an admission either sees the old setting and commits first, or
-    // sees the new one.
-    if updates.visibility.is_some() || updates.join_role.is_some() {
-        crate::channel_members::acquire_channel_membership_lock(
-            tx.conn(),
-            community_id,
-            channel_id,
-        )
-        .await?;
-    }
-    // T1a repair: a TTL change can flip this channel's event-commit fast
-    // path (`event_follow_up::refresh_channel_ttls` reads ttl_seconds under
-    // a SHARED per-channel advisory lock). Take the same key EXCLUSIVE
-    // before the UPDATE so a concurrent event either sees the committed TTL
-    // or strictly precedes this transition — whose own deadline reset is
-    // then the latest word. Non-TTL updates don't touch the fast path and
-    // skip the lock.
-    if updates.ttl_seconds.is_some() {
-        sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
-            .bind(channel_ttl_lock_key(community_id, channel_id))
-            .execute(tx.conn())
-            .await?;
-    }
     let result = q.execute(tx.conn()).await?;
     if result.rows_affected() == 0 {
         return Err(DbError::ChannelNotFound(channel_id));
     }
-    tx.commit().await?;
-
-    get_channel_with_operation(
-        pool,
-        community_id,
-        channel_id,
-        crate::observability::WriterOperation::EventWrite,
-    )
-    .await
+    Ok(())
 }
 
 /// Sets the topic for a channel, recording who set it and when.
@@ -928,20 +1078,29 @@ pub async fn set_topic(
         crate::observability::WriterOperation::EventWrite,
     )
     .await?;
+    let mut locked =
+        lock_channel_for_settings_write(&mut tx, channel_id, &ChannelUpdate::default()).await?;
+    set_topic_on(&mut locked, topic, set_by).await?;
+    tx.commit().await?;
+    Ok(())
+}
+
+async fn set_topic_on(w: &mut ChannelSettingsWriter<'_>, topic: &str, set_by: &[u8]) -> Result<()> {
+    let channel_id = w.channel_id;
+    let tx = &mut *w.tx;
     let result = sqlx::query(
         "UPDATE channels SET topic = $1, topic_set_by = $2, topic_set_at = NOW() \
          WHERE community_id = $3 AND id = $4 AND deleted_at IS NULL",
     )
     .bind(topic)
     .bind(set_by)
-    .bind(community_id.as_uuid())
+    .bind(tx.community().as_uuid())
     .bind(channel_id)
     .execute(tx.conn())
     .await?;
     if result.rows_affected() == 0 {
         return Err(DbError::ChannelNotFound(channel_id));
     }
-    tx.commit().await?;
     Ok(())
 }
 
@@ -959,20 +1118,33 @@ pub async fn set_purpose(
         crate::observability::WriterOperation::EventWrite,
     )
     .await?;
+    let mut locked =
+        lock_channel_for_settings_write(&mut tx, channel_id, &ChannelUpdate::default()).await?;
+    set_purpose_on(&mut locked, purpose, set_by).await?;
+    tx.commit().await?;
+    Ok(())
+}
+
+async fn set_purpose_on(
+    w: &mut ChannelSettingsWriter<'_>,
+    purpose: &str,
+    set_by: &[u8],
+) -> Result<()> {
+    let channel_id = w.channel_id;
+    let tx = &mut *w.tx;
     let result = sqlx::query(
         "UPDATE channels SET purpose = $1, purpose_set_by = $2, purpose_set_at = NOW() \
          WHERE community_id = $3 AND id = $4 AND deleted_at IS NULL",
     )
     .bind(purpose)
     .bind(set_by)
-    .bind(community_id.as_uuid())
+    .bind(tx.community().as_uuid())
     .bind(channel_id)
     .execute(tx.conn())
     .await?;
     if result.rows_affected() == 0 {
         return Err(DbError::ChannelNotFound(channel_id));
     }
-    tx.commit().await?;
     Ok(())
 }
 
@@ -991,36 +1163,9 @@ pub async fn archive_channel(
         crate::observability::WriterOperation::EventWrite,
     )
     .await?;
-    // First check: does the channel exist and what is its state?
-    let row = sqlx::query(
-        "SELECT archived_at FROM channels WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
-    )
-        .bind(community_id.as_uuid())
-        .bind(channel_id)
-        .fetch_optional(tx.conn())
-        .await?;
-
-    match row {
-        None => return Err(DbError::ChannelNotFound(channel_id)),
-        Some(r) => {
-            let archived_at: Option<DateTime<Utc>> = r.try_get("archived_at")?;
-            if archived_at.is_some() {
-                return Err(DbError::AccessDenied(
-                    "channel is already archived".to_string(),
-                ));
-            }
-        }
-    }
-
-    sqlx::query(
-        "UPDATE channels SET archived_at = NOW() \
-         WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL AND archived_at IS NULL",
-    )
-    .bind(community_id.as_uuid())
-    .bind(channel_id)
-    .execute(tx.conn())
-    .await?;
-
+    let mut locked =
+        lock_channel_for_settings_write(&mut tx, channel_id, &ChannelUpdate::default()).await?;
+    set_archived_on(&mut locked, true).await?;
     tx.commit().await?;
     Ok(())
 }
@@ -1040,42 +1185,174 @@ pub async fn unarchive_channel(
         crate::observability::WriterOperation::EventWrite,
     )
     .await?;
-    // First check: does the channel exist and what is its state?
+    let mut locked =
+        lock_channel_for_settings_write(&mut tx, channel_id, &ChannelUpdate::default()).await?;
+    set_archived_on(&mut locked, false).await?;
+    tx.commit().await?;
+    Ok(())
+}
+
+/// Archives (`archive = true`) or restores the channel on `tx`. Refuses with
+/// `AccessDenied` when the channel is already in that state. Restoring an
+/// ephemeral channel restarts its TTL deadline.
+async fn set_archived_on(w: &mut ChannelSettingsWriter<'_>, archive: bool) -> Result<()> {
+    let channel_id = w.channel_id;
+    let tx = &mut *w.tx;
+    let community_id = tx.community();
     let row = sqlx::query(
         "SELECT archived_at FROM channels WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
     )
-        .bind(community_id.as_uuid())
-        .bind(channel_id)
-        .fetch_optional(tx.conn())
-        .await?;
+    .bind(community_id.as_uuid())
+    .bind(channel_id)
+    .fetch_optional(tx.conn())
+    .await?;
 
     match row {
         None => return Err(DbError::ChannelNotFound(channel_id)),
         Some(r) => {
             let archived_at: Option<DateTime<Utc>> = r.try_get("archived_at")?;
-            if archived_at.is_none() {
+            if archive && archived_at.is_some() {
+                return Err(DbError::AccessDenied(
+                    "channel is already archived".to_string(),
+                ));
+            }
+            if !archive && archived_at.is_none() {
                 return Err(DbError::AccessDenied("channel is not archived".to_string()));
             }
         }
     }
 
-    sqlx::query(
+    let sql = if archive {
+        "UPDATE channels SET archived_at = NOW() \
+         WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL AND archived_at IS NULL"
+    } else {
         "UPDATE channels SET archived_at = NULL, \
              ttl_deadline = CASE \
                  WHEN ttl_seconds IS NOT NULL THEN NOW() + (ttl_seconds || ' seconds')::interval \
                  ELSE ttl_deadline \
              END \
-         WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL AND archived_at IS NOT NULL",
-    )
-    .bind(community_id.as_uuid())
-    .bind(channel_id)
-    .execute(tx.conn())
-    .await?;
-
-    tx.commit().await?;
+         WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL AND archived_at IS NOT NULL"
+    };
+    sqlx::query(sql)
+        .bind(community_id.as_uuid())
+        .bind(channel_id)
+        .execute(tx.conn())
+        .await?;
     Ok(())
 }
 
+/// The channel changes named by one kind:9002 event. `None` leaves a value
+/// unchanged.
+#[derive(Default)]
+pub struct ChannelMetadataEdit {
+    /// Name, description, visibility, TTL and labels.
+    pub columns: ChannelUpdate,
+    /// New topic, recorded as set by the event's author.
+    pub topic: Option<String>,
+    /// New purpose, recorded as set by the event's author.
+    pub purpose: Option<String>,
+    /// `Some(true)` archives the channel; `Some(false)` restores it.
+    pub archived: Option<bool>,
+}
+
+/// The result of [`Db::apply_channel_metadata_event`].
+pub struct ChannelMetadataWrite {
+    /// The stored kind:9002 event.
+    pub stored: StoredEvent,
+    /// `false` when the event was already stored. Nothing was changed then:
+    /// its changes committed with it the first time.
+    pub inserted: bool,
+    /// The channel's visibility before this event (`"open"` or `"private"`).
+    pub previous_visibility: String,
+}
+
+pub(crate) async fn apply_channel_metadata_event(
+    pool: &PgPool,
+    push_enqueue: Option<crate::runtime::push_enqueue::PushEnqueue>,
+    community_id: CommunityId,
+    channel_id: Uuid,
+    event: &nostr::Event,
+    edit: ChannelMetadataEdit,
+    system_messages: &[nostr::Event],
+) -> Result<ChannelMetadataWrite> {
+    let mut tx = crate::begin_community_event_write_transaction(
+        pool,
+        community_id,
+        crate::observability::WriterOperation::EventWrite,
+    )
+    .await?;
+    tx.set_push_enqueue(push_enqueue);
+
+    // The row lock also orders two copies of the same event: the second
+    // waits, then sees the first stored.
+    let mut locked = lock_channel_for_settings_write(&mut tx, channel_id, &edit.columns).await?;
+    let previous_visibility = locked.previous_visibility().to_owned();
+
+    // The same writes as `Db::insert_event_with_thread_metadata`.
+    let (stored, inserted) = crate::event::insert_event_with_thread_metadata_tx(
+        locked.tx(),
+        event,
+        Some(channel_id),
+        None,
+    )
+    .await?;
+    if !inserted {
+        tx.rollback().await?;
+        return Ok(ChannelMetadataWrite {
+            stored,
+            inserted,
+            previous_visibility,
+        });
+    }
+    // Decide the archive state here, after the duplicate check and under
+    // the row lock, not during validation. A resend that races the first
+    // commit then sees that commit as a duplicate (above), never as a
+    // channel already in the requested state.
+    match (locked.archived(), edit.archived) {
+        (true, Some(false)) | (false, None | Some(true)) => {}
+        (true, _) => return Err(DbError::AccessDenied("channel is archived".to_string())),
+        (false, Some(false)) => {
+            return Err(DbError::AccessDenied("channel is not archived".to_string()));
+        }
+    }
+    crate::insert_mentions_in_transaction(locked.tx(), event, Some(channel_id)).await?;
+
+    let ChannelMetadataEdit {
+        columns,
+        topic,
+        purpose,
+        archived,
+    } = edit;
+    let actor = event.pubkey.to_bytes();
+    if !columns.is_empty() {
+        update_channel_on(&mut locked, columns).await?;
+    }
+    if let Some(topic) = topic {
+        set_topic_on(&mut locked, &topic, &actor).await?;
+    }
+    if let Some(purpose) = purpose {
+        set_purpose_on(&mut locked, &purpose, &actor).await?;
+    }
+    if let Some(archive) = archived {
+        set_archived_on(&mut locked, archive).await?;
+    }
+    // The same writes as `Db::insert_event`.
+    for message in system_messages {
+        let (_, message_inserted) =
+            crate::event::insert_event_in_transaction(locked.tx(), message, Some(channel_id))
+                .await?;
+        if message_inserted {
+            crate::insert_mentions_in_transaction(locked.tx(), message, Some(channel_id)).await?;
+        }
+    }
+    tx.commit().await?;
+    Ok(ChannelMetadataWrite {
+        stored,
+        inserted,
+        previous_visibility,
+    })
+}
+
 /// Soft-delete a channel by setting `deleted_at = NOW()`.
 ///
 /// Returns `Ok(true)` if the channel was deleted, `Ok(false)` if already
@@ -1091,16 +1368,30 @@ pub async fn soft_delete_channel(
         crate::observability::WriterOperation::EventWrite,
     )
     .await?;
-    let result = sqlx::query(
-        "UPDATE channels SET deleted_at = NOW() WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
-    )
-            .bind(community_id.as_uuid())
-            .bind(channel_id)
-            .execute(tx.conn())
-            .await?;
-
+    let mut writer = match lock_channel_for_delete(&mut tx, channel_id).await {
+        Ok(writer) => writer,
+        Err(DbError::ChannelNotFound(_)) => return Ok(false),
+        Err(error) => return Err(error),
+    };
+    soft_delete_on(&mut writer).await?;
     tx.commit().await?;
-    Ok(result.rows_affected() > 0)
+    Ok(true)
+}
+
+async fn soft_delete_on(w: &mut ChannelSettingsWriter<'_>) -> Result<()> {
+    if !w.admission {
+        return Err(DbError::InvalidData(
+            "channel delete is not covered by its lock".into(),
+        ));
+    }
+    let channel_id = w.channel_id;
+    let community_id = w.tx.community();
+    sqlx::query("UPDATE channels SET deleted_at = NOW() WHERE community_id = $1 AND id = $2")
+        .bind(community_id.as_uuid())
+        .bind(channel_id)
+        .execute(w.tx.conn())
+        .await?;
+    Ok(())
 }
 
 /// Archive ephemeral channels whose TTL deadline has passed.
@@ -1133,6 +1424,16 @@ pub async fn reap_expired_ephemeral_channels(pool: &PgPool) -> Result<Vec<Reaped
         crate::observability::WriterOperation::Maintenance,
         "ephemeral_channel_reap",
         |id, mut tx| async move {
+            // This archive takes no advisory lock, unlike
+            // `lock_channel_for_settings_write`. It does not need one: it is
+            // a single UPDATE whose WHERE clause holds every condition for
+            // the archive. The UPDATE waits for the row lock of any writer
+            // that holds the row (a TTL refresh, a settings write, a huddle
+            // join). Under READ COMMITTED, PostgreSQL then checks the WHERE
+            // clause again against the version that writer committed. So a
+            // refresh that moved the deadline wins, a channel archived or
+            // deleted meanwhile is skipped, and a join that held the row
+            // commits first. Do not split this into a read and a write.
             let rows = sqlx::query(
                 "UPDATE channels AS ch SET archived_at = NOW() \
                  FROM communities AS c \
@@ -1248,6 +1549,32 @@ impl Db {
         .await
     }
 
+    /// Stores the kind:9002 `event` for `channel_id`, applies `edit`, and
+    /// stores the relay-signed `system_messages` that announce it, all in one
+    /// transaction. If any write fails, nothing is stored, so a client can
+    /// resend the same event. A resend of a stored event changes nothing and
+    /// returns `inserted: false`.
+    #[datastore_span(name = "apply_channel_metadata_event", system = "postgresql")]
+    pub async fn apply_channel_metadata_event(
+        &self,
+        community_id: CommunityId,
+        channel_id: Uuid,
+        event: &nostr::Event,
+        edit: ChannelMetadataEdit,
+        system_messages: &[nostr::Event],
+    ) -> Result<ChannelMetadataWrite> {
+        apply_channel_metadata_event(
+            &self.pool,
+            self.push_enqueue.clone(),
+            community_id,
+            channel_id,
+            event,
+            edit,
+            system_messages,
+        )
+        .await
+    }
+
     /// Creates a channel with a relay-chosen UUID and an initial label set.
     #[allow(clippy::too_many_arguments)]
     #[datastore_span(name = "create_labeled_channel", system = "postgresql")]
@@ -1420,7 +1747,7 @@ impl Db {
 
 /// The per-channel TTL advisory lock key. Event commits take it SHARED
 /// (`event_follow_up::refresh_channel_ttls`); TTL transitions in
-/// [`update_channel`] take it EXCLUSIVE. Both sides build it here, so the keys
+/// [`update_channel`] and [`apply_channel_metadata_event`] take it EXCLUSIVE. Both sides build it here, so the keys
 /// cannot drift.
 pub(crate) fn channel_ttl_lock_key(community: CommunityId, channel: Uuid) -> String {
     format!("buzz_channel_ttl:{}:{channel}", community.as_uuid())
@@ -2077,4 +2404,52 @@ mod postgres_tests {
             .expect("members");
         assert_eq!(members.len(), 1, "only the first creator is a member");
     }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn soft_delete_waits_for_membership_lock() {
+        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,
+            "delete-vs-roster",
+            ChannelType::Stream,
+            ChannelVisibility::Open,
+            None,
+            &owner,
+            None,
+        )
+        .await
+        .expect("create channel");
+
+        // A roster publication holds the membership lock. A delete ends
+        // admission, so it must wait, like any membership write.
+        let snapshot =
+            crate::channel_members::lock_member_snapshot(&pool, community, channel.id, &[9u8; 32])
+                .await
+                .expect("capture locked roster");
+        let delete_pool = crate::test_support::named_pool("delete-vs-roster").await;
+        let channel_id = channel.id;
+        let delete =
+            tokio::spawn(
+                async move { soft_delete_channel(&delete_pool, community, channel_id).await },
+            );
+        crate::test_support::wait_for_lock_wait(&pool, "delete-vs-roster").await;
+        assert!(
+            !delete.is_finished(),
+            "delete must wait for the roster fence"
+        );
+        snapshot.release().await.expect("release roster fence");
+        assert!(delete.await.expect("delete task").expect("delete"));
+
+        assert!(
+            !soft_delete_channel(&pool, community, channel.id)
+                .await
+                .expect("second delete"),
+            "a second delete finds no live channel"
+        );
+    }
 }

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

@@ -199,13 +199,13 @@ pub(crate) async fn acquire_channel_membership_lock(
 
 /// Check the same p-tag shape and canonical active membership as migration 0032.
 /// The caller holds this channel's membership advisory lock on the admitted
-/// transaction before calling. The membership writers in this module take that
-/// key before reading roles or changing rows. Not every writer does yet:
-/// `execute_kick_with_marker` and `deploy_kick_member` (relay_admin_actions.rs)
-/// set `removed_at` without it, so a kick can commit after this check passes
-/// and before the roster commits. The retained trigger has the same gap. Both
-/// kick paths must take the key, with a concurrent kick-vs-roster-publication
-/// test, before anything relies on complete serialization (phase 2 included).
+/// transaction before calling. Every writer that changes the membership of an
+/// existing channel takes that key before reading roles or changing rows
+/// (channel creation inserts rows that no other transaction can see yet),
+/// including both kick paths
+/// (`execute_kick_with_marker` and `deploy_kick_member` in
+/// relay_admin_actions.rs), so no removal can commit between this check and
+/// the roster commit.
 /// Legacy replacement takes the replacement key, locks
 /// old-head rows in its UPDATE, then takes the membership key at trigger INSERT.
 /// The app path takes replacement key -> membership key -> old-head row locks,
@@ -287,17 +287,89 @@ pub(crate) async fn guard_channel_roster_event(
 
 // ── Transaction-level membership helpers (for commit_participant_join) ────────
 
-/// Acquire the per-channel membership advisory lock on a caller-owned transaction.
+/// Proof that a caller-owned transaction holds the membership locks of a set
+/// of channels. Only [`lock_channel_memberships_in_transaction`] makes one,
+/// after it takes the locks on the transaction it borrows. The borrow binds
+/// the locks to that transaction: the proof cannot be used with another
+/// transaction or after its commit. Membership writes that compose with other
+/// work on the caller's transaction (a huddle join) are methods on it, so
+/// they cannot run without the lock.
+pub struct ChannelMembershipLocks<'tx> {
+    tx: &'tx mut AdmittedTx,
+    channels: Vec<Uuid>,
+}
+
+impl ChannelMembershipLocks<'_> {
+    /// The locked transaction, for the other reads and writes of the caller's
+    /// operation.
+    pub fn tx(&mut self) -> &mut AdmittedTx {
+        self.tx
+    }
+
+    fn check_covers(&self, channel_id: Uuid) -> Result<()> {
+        if self.channels.contains(&channel_id) {
+            Ok(())
+        } else {
+            Err(DbError::InvalidData(
+                "channel membership write is not covered by its lock".into(),
+            ))
+        }
+    }
+
+    /// Auto-add a member (for ephemeral-channel admission).
+    ///
+    /// Inserts or reactivates the membership row at `Member` role with the
+    /// given `invited_by` (channel creator for huddle auto-add). Refuses a
+    /// channel whose membership lock this proof does not hold.
+    ///
+    /// Used by `commit_participant_join` to atomically add membership and the
+    /// `48101` event in a single transaction under a session effect permit.
+    pub async fn insert_auto_membership(
+        &mut self,
+        channel_id: Uuid,
+        pubkey: &[u8],
+        invited_by: &[u8],
+    ) -> Result<()> {
+        self.check_covers(channel_id)?;
+        let community_id = self.tx.community();
+        sqlx::query(
+            r#"
+            INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by)
+            VALUES ($1, $2, $3, 'member'::member_role, $4)
+            ON CONFLICT (community_id, channel_id, pubkey) DO UPDATE SET
+                removed_at = NULL,
+                removed_by = NULL,
+                role = EXCLUDED.role
+            "#,
+        )
+        .bind(community_id.as_uuid())
+        .bind(channel_id)
+        .bind(pubkey)
+        .bind(invited_by)
+        .execute(self.tx.conn())
+        .await?;
+        Ok(())
+    }
+}
+
+/// Take the membership lock of each channel in `channels`, in that order, on
+/// a caller-owned transaction, and return the proof that holds them.
 ///
-/// Equivalent to the internal `acquire_channel_membership_lock`, but exposed
-/// for callers that need to compose multiple operations in one transaction
-/// (e.g., `commit_participant_join` in `audio/handler.rs`).
-pub async fn acquire_channel_membership_lock_in_transaction(
-    tx: &mut AdmittedTx,
-    channel_id: Uuid,
-) -> Result<()> {
+/// These locks come first in the channel lock order (membership, then TTL,
+/// then the channel row; see `channel::lock_channel_for_settings_write`).
+/// Take them before any other channel lock on `tx`.
+pub async fn lock_channel_memberships_in_transaction<'tx>(
+    tx: &'tx mut AdmittedTx,
+    channels: &[Uuid],
+) -> Result<ChannelMembershipLocks<'tx>> {
     let community_id = tx.community();
-    acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await
+    for channel_id in channels {
+        acquire_channel_membership_lock(tx.conn(), community_id, *channel_id).await?;
+    }
+    Ok(ChannelMembershipLocks {
+        tx,
+        channels: channels.to_vec(),
+    })
 }
 
 /// Lock the live channel row on a caller-owned transaction and return its
@@ -405,41 +477,6 @@ pub async fn active_role_in_transaction(
     get_active_role_tx(tx.conn(), community_id, channel_id, pubkey).await
 }
 
-/// Auto-add a member on a caller-owned transaction (for ephemeral-channel admission).
-///
-/// Inserts or reactivates the membership row at `Member` role with the given
-/// `invited_by` (channel creator for huddle auto-add). Does NOT acquire the
-/// advisory lock — callers must have already called
-/// `acquire_channel_membership_lock_in_transaction` before calling this.
-///
-/// Used by `commit_participant_join` to atomically add membership and the
-/// `48101` event in a single transaction under a session effect permit.
-pub async fn insert_auto_membership_in_transaction(
-    tx: &mut AdmittedTx,
-    channel_id: Uuid,
-    pubkey: &[u8],
-    invited_by: &[u8],
-) -> Result<()> {
-    let community_id = tx.community();
-    sqlx::query(
-        r#"
-        INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by)
-        VALUES ($1, $2, $3, 'member'::member_role, $4)
-        ON CONFLICT (community_id, channel_id, pubkey) DO UPDATE SET
-            removed_at = NULL,
-            removed_by = NULL,
-            role = EXCLUDED.role
-        "#,
-    )
-    .bind(community_id.as_uuid())
-    .bind(channel_id)
-    .bind(pubkey)
-    .bind(invited_by)
-    .execute(tx.conn())
-    .await?;
-    Ok(())
-}
-
 // ── End transaction-level helpers ─────────────────────────────────────────────
 
 /// The immutable facts that every relay-signed group-state event
@@ -3386,6 +3423,25 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn join_role_change_waits_for_a_roster_publication_on_an_ephemeral_channel() {
+        join_role_change_waits_for_roster_publication(SettingsWriter::UpdateChannel).await;
+    }
+
+    /// The kind:9002 path stores the event and applies its changes in its own
+    /// transaction. It must take the same locks in the same order as
+    /// `update_channel`.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn kind_9002_join_role_change_waits_for_a_roster_publication_on_an_ephemeral_channel() {
+        join_role_change_waits_for_roster_publication(SettingsWriter::Kind9002).await;
+    }
+
+    #[derive(Clone, Copy)]
+    enum SettingsWriter {
+        UpdateChannel,
+        Kind9002,
+    }
+
+    async fn join_role_change_waits_for_roster_publication(writer: SettingsWriter) {
         let pool = setup_pool().await;
         let community_id = make_test_community(&pool).await;
         let community = CommunityId::from_uuid(community_id);
@@ -3424,19 +3480,45 @@ mod postgres_tests {
             let pool = pool.clone();
             let channel_id = channel.id;
             async move {
-                crate::channel::update_channel(
-                    &pool,
-                    community,
-                    channel_id,
-                    crate::channel::ChannelUpdate {
-                        join_role: Some("guest".into()),
-                        ..Default::default()
-                    },
-                )
-                .await
+                let updates = crate::channel::ChannelUpdate {
+                    join_role: Some("guest".into()),
+                    ..Default::default()
+                };
+                match writer {
+                    SettingsWriter::UpdateChannel => {
+                        crate::channel::update_channel(&pool, community, channel_id, updates).await
+                    }
+                    SettingsWriter::Kind9002 => {
+                        let edit_event = nostr::EventBuilder::new(nostr::Kind::Custom(9002), "")
+                            .tags(vec![
+                                nostr::Tag::parse(["h", &channel_id.to_string()]).expect("h tag"),
+                                nostr::Tag::parse(["join_role", "guest"]).expect("join_role tag"),
+                            ])
+                            .sign_with_keys(&Keys::generate())
+                            .expect("sign kind:9002");
+                        let write = crate::channel::apply_channel_metadata_event(
+                            &pool,
+                            None,
+                            community,
+                            channel_id,
+                            &edit_event,
+                            crate::channel::ChannelMetadataEdit {
+                                columns: updates,
+                                ..Default::default()
+                            },
+                            &[],
+                        )
+                        .await?;
+                        assert!(write.inserted, "the kind:9002 event must be stored");
+                        crate::channel::get_channel(&pool, community, channel_id).await
+                    }
+                }
             }
         });
-        tokio::time::sleep(std::time::Duration::from_millis(300)).await;
+        // The kind:9002 writer also stores an event before it would reach
+        // the row, so give an unblocked writer time to finish. A blocked
+        // writer waits until the roster commits, so this cannot pass early.
+        tokio::time::sleep(std::time::Duration::from_secs(2)).await;
         assert!(
             !update.is_finished(),
             "the join_role change must wait for the roster publication"

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

@@ -703,6 +703,12 @@ pub async fn execute_kick_with_marker(
         .bind(action_id)
         .fetch_optional(tx.conn())
         .await?;
+    // Like every membership writer, remove under the channel's membership
+    // lock, so a roster publication either sees the member removed or
+    // commits before the removal. No channel lock is held yet, so the lock
+    // order holds.
+    crate::channel_members::acquire_channel_membership_lock(tx.conn(), community_id, channel_id)
+        .await?;
 
     let kick = sqlx::query(
         r#"
@@ -1689,6 +1695,11 @@ pub async fn deploy_kick_member(
         crate::observability::WriterOperation::Authorization,
     )
     .await?;
+    // Like every membership writer, remove under the channel's membership
+    // lock, so a roster publication either sees the member removed or
+    // commits before the removal.
+    crate::channel_members::acquire_channel_membership_lock(tx.conn(), community_id, channel_id)
+        .await?;
     // Use a direct UPDATE to avoid the tenant ownership check in channel::remove_member.
     // This is the deployment-authority primitive: no actor role check.
     let result = sqlx::query(
@@ -3840,14 +3851,27 @@ mod postgres_tests {
         };
 
         // First kick: member is present → Removed + step_marker committed.
-        let r1 = execute_kick_with_marker(
+        // It runs while a roster publication holds the member snapshot, and
+        // must wait for it.
+        let (kicked, kick_actor) = (target.clone(), actork.clone());
+        let r1 = assert_kick_waits_for_roster(
             &pool,
-            action_id,
-            lease_token1,
-            cid,
+            community_id,
             channel_id,
             &target,
-            &actork,
+            "marker-kick-vs-roster",
+            move |kick_pool| async move {
+                execute_kick_with_marker(
+                    &kick_pool,
+                    action_id,
+                    lease_token1,
+                    cid,
+                    channel_id,
+                    &kicked,
+                    &kick_actor,
+                )
+                .await
+            },
         )
         .await
         .expect("first kick");
@@ -4054,6 +4078,106 @@ mod postgres_tests {
         );
     }
 
+    /// Runs `kick` on its own session while a roster publication holds the
+    /// channel's member snapshot. The kick must wait for the snapshot's
+    /// membership lock, so the published roster cannot list a member whose
+    /// removal committed before it.
+    async fn assert_kick_waits_for_roster<F, T>(
+        pool: &PgPool,
+        community_id: Uuid,
+        channel_id: Uuid,
+        target: &[u8],
+        application_name: &str,
+        kick: impl FnOnce(PgPool) -> F,
+    ) -> T
+    where
+        F: std::future::Future<Output = T> + Send + 'static,
+        T: Send + 'static,
+    {
+        let snapshot = crate::channel_members::lock_member_snapshot(
+            pool,
+            CommunityId::from_uuid(community_id),
+            channel_id,
+            &[9u8; 32],
+        )
+        .await
+        .expect("capture locked roster");
+        assert!(
+            snapshot.members.iter().any(|m| m.pubkey == target),
+            "the captured roster lists the member before the kick"
+        );
+        let kick_pool = crate::test_support::named_pool(application_name).await;
+        let kick = tokio::spawn(kick(kick_pool));
+        crate::test_support::wait_for_lock_wait(pool, application_name).await;
+        assert!(!kick.is_finished(), "kick must wait for the roster fence");
+        let active: bool = sqlx::query_scalar(
+            "SELECT removed_at IS NULL FROM channel_members \
+             WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3",
+        )
+        .bind(community_id)
+        .bind(channel_id)
+        .bind(target)
+        .fetch_one(pool)
+        .await
+        .expect("read membership");
+        assert!(active, "the member stays active while the roster is held");
+        snapshot.release().await.expect("release roster fence");
+        kick.await.expect("kick task")
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn deploy_kick_member_waits_for_roster_publication() {
+        let pool = setup_pool().await;
+        let community_id = make_community(&pool).await;
+        let actor = actor();
+        let target = vec![4u8; 32];
+        let channel_id = Uuid::new_v4();
+        sqlx::query(
+            r#"
+            INSERT INTO channels (id, community_id, name, channel_type, visibility, created_by)
+            VALUES ($1, $2, 'kick-vs-roster', 'stream', 'open', $3)
+            "#,
+        )
+        .bind(channel_id)
+        .bind(community_id)
+        .bind(&actor)
+        .execute(&pool)
+        .await
+        .expect("create channel");
+        sqlx::query(
+            "INSERT INTO channel_members (community_id, channel_id, pubkey, role) VALUES ($1, $2, $3, 'member')",
+        )
+        .bind(community_id)
+        .bind(channel_id)
+        .bind(&target)
+        .execute(&pool)
+        .await
+        .expect("add member");
+
+        let kicked = target.clone();
+        let result = assert_kick_waits_for_roster(
+            &pool,
+            community_id,
+            channel_id,
+            &target,
+            "deploy-kick-vs-roster",
+            move |kick_pool| async move {
+                deploy_kick_member(
+                    &kick_pool,
+                    CommunityId::from_uuid(community_id),
+                    channel_id,
+                    &kicked,
+                    &actor,
+                )
+                .await
+            },
+        )
+        .await
+        .expect("kick after roster publication");
+        assert_eq!(result, KickResult::Removed);
+    }
+
     // ── Reopen: terminal → open CAS + durable audit row ───────────────────────
 
     async fn set_report_status(pool: &PgPool, community_id: Uuid, report_id: Uuid, status: &str) {

crates/buzz-relay/src/audio/handler.rs

@@ -3282,17 +3282,21 @@ async fn commit_participant_join(
     #[cfg(test)]
     crate::nip_fi_test_hooks::before_membership_lock(tenant.community()).await;
 
-    buzz_db::channel_members::acquire_channel_membership_lock_in_transaction(&mut tx, channel_id)
-        .await?;
-    if parent_channel_id != channel_id {
-        buzz_db::channel_members::acquire_channel_membership_lock_in_transaction(
-            &mut tx,
-            parent_channel_id,
-        )
-        .await?;
-    }
-    buzz_db::channel_members::lock_channel_ttl_shared_in_transaction(&mut tx, parent_channel_id)
-        .await?;
+    let locked_channels = if parent_channel_id == channel_id {
+        vec![channel_id]
+    } else {
+        vec![channel_id, parent_channel_id]
+    };
+    let mut membership = buzz_db::channel_members::lock_channel_memberships_in_transaction(
+        &mut tx,
+        &locked_channels,
+    )
+    .await?;
+    buzz_db::channel_members::lock_channel_ttl_shared_in_transaction(
+        membership.tx(),
+        parent_channel_id,
+    )
+    .await?;
 
     // Archive re-check (ALL paths): re-read archived_at inside the
     // transaction before any write, taking a row-level write lock
@@ -3306,9 +3310,11 @@ async fn commit_participant_join(
     // is held only for the duration of the join transaction (typically
     // sub-millisecond). See `lock_channel_archived_at_in_transaction` for why
     // the lock is `FOR NO KEY UPDATE` rather than `FOR UPDATE`.
-    let channel_archived =
-        buzz_db::channel_members::lock_channel_archived_at_in_transaction(&mut tx, channel_id)
-            .await?;
+    let channel_archived = buzz_db::channel_members::lock_channel_archived_at_in_transaction(
+        membership.tx(),
+        channel_id,
+    )
+    .await?;
 
     // Test hook: fires after the FOR UPDATE lock is acquired but before the
     // archived check / any write. A test can attempt a concurrent archive here
@@ -3338,9 +3344,12 @@ async fn commit_participant_join(
         // IMPORTANT 4b: Re-read parent membership under the lock. A parent
         // membership revocation in the same window would make the auto-add
         // unjustified; reject rather than grant access from stale authority.
-        let parent_still_member =
-            buzz_db::channel_members::is_member_in_transaction(&mut tx, *parent_id, pubkey_bytes)
-                .await?;
+        let parent_still_member = buzz_db::channel_members::is_member_in_transaction(
+            membership.tx(),
+            *parent_id,
+            pubkey_bytes,
+        )
+        .await?;
 
         if !parent_still_member {
             retire_shadow();
@@ -3355,7 +3364,7 @@ async fn commit_participant_join(
         // but before this transaction acquires the lock; committing a join into
         // an unlinked channel violates the "creator authority" invariant.
         let link_still_exists = buzz_db::event::huddle_started_link_exists_in_transaction(
-            &mut tx,
+            membership.tx(),
             *parent_id,
             channel_id,
             channel_created_by.as_slice(),
@@ -3370,18 +3379,17 @@ async fn commit_participant_join(
 
         // Re-read child membership — a concurrent legitimate add may have
         // already provided access; do not overwrite role/provenance.
-        let still_absent =
-            !buzz_db::channel_members::is_member_in_transaction(&mut tx, channel_id, pubkey_bytes)
-                .await?;
+        let still_absent = !buzz_db::channel_members::is_member_in_transaction(
+            membership.tx(),
+            channel_id,
+            pubkey_bytes,
+        )
+        .await?;
 
         if still_absent {
-            buzz_db::channel_members::insert_auto_membership_in_transaction(
-                &mut tx,
-                channel_id,
-                pubkey_bytes,
-                channel_created_by.as_slice(),
-            )
-            .await?;
+            membership
+                .insert_auto_membership(channel_id, pubkey_bytes, channel_created_by.as_slice())
+                .await?;
         }
         // If not still_absent: concurrent add observed — membership preserved.
     }
@@ -3392,22 +3400,22 @@ async fn commit_participant_join(
     //     an open channel whose `join_role` is `member`. A non-member of a
     //     private scope has lost the membership the pre-join check saw (the
     //     auto-add above already made a private huddle's joiner a member).
-    let scopes = if parent_channel_id == channel_id {
-        vec![channel_id]
-    } else {
-        vec![channel_id, parent_channel_id]
-    };
-    for scope in scopes {
-        let role =
-            buzz_db::channel_members::active_role_in_transaction(&mut tx, scope, pubkey_bytes)
-                .await?;
+    for scope in locked_channels.iter().copied() {
+        let role = buzz_db::channel_members::active_role_in_transaction(
+            membership.tx(),
+            scope,
+            pubkey_bytes,
+        )
+        .await?;
         let refusal = match role.as_deref() {
             Some(role) => (!crate::handlers::ingest::role_may_post(Some(role)))
                 .then_some(JoinCommitError::WriteRoleLost),
             None => {
-                let settings =
-                    buzz_db::channel_members::channel_write_settings_in_transaction(&mut tx, scope)
-                        .await?;
+                let settings = buzz_db::channel_members::channel_write_settings_in_transaction(
+                    membership.tx(),
+                    scope,
+                )
+                .await?;
                 match settings {
                     Some((visibility, _)) if visibility != "open" => {
                         Some(JoinCommitError::MembershipLost)
@@ -9524,7 +9532,7 @@ mod tests {
         // Exercises the concurrent-external-add path in the AutoAddRequired branch
         // of `commit_participant_join`. An external transaction inserts the
         // membership row while our transaction is paused at `before_membership_lock`
-        // — just before `acquire_channel_membership_lock_in_transaction`. When our
+        // — just before `lock_channel_memberships_in_transaction`. When our
         // transaction resumes:
         //   1. It acquires the channel membership lock.
         //   2. Re-reads membership — the external insert is committed and visible.

crates/buzz-relay/src/handlers/edit_metadata_postgres_tests.rs

@@ -0,0 +1,507 @@
+//! Postgres-backed tests for kind:9002: the event and the channel changes it
+//! names commit together, or not at all.
+
+use buzz_db::event::EventQuery;
+use nostr::{EventBuilder, Keys, Kind, Tag};
+use uuid::Uuid;
+
+use crate::handlers::group_state_postgres_tests::{
+    community, create_channel, edit_channel, http_auth, state,
+};
+use crate::handlers::ingest::ingest_event;
+
+/// Count stored events of `kind` in `channel_id`.
+async fn count_kind(
+    state: &crate::state::AppState,
+    tenant: &buzz_core::TenantContext,
+    channel_id: Uuid,
+    kind: i32,
+) -> usize {
+    state
+        .db
+        .query_events(&EventQuery {
+            kinds: Some(vec![kind]),
+            channel_id: Some(channel_id),
+            ..EventQuery::for_community(tenant.community())
+        })
+        .await
+        .expect("query events")
+        .len()
+}
+
+/// A change that fails inside the write stores nothing and changes nothing,
+/// and the same event succeeds once the failure is gone. A test trigger makes
+/// the topic UPDATE fail after the event row and the `about` change were
+/// written in the same transaction. Before, the relay stored the event,
+/// applied `about`, skipped `topic`, and reported success; a resend then
+/// returned `duplicate:`. Mutation: store the event in its own transaction
+/// → RED.
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn failed_edit_stores_nothing() {
+    let state = state().await;
+    let tenant = community(&state, "edit-atomic-fail").await;
+    let owner = Keys::generate();
+    let channel_id = create_channel(&state, &tenant, &owner, &[]).await;
+    let before = state
+        .db
+        .get_channel(tenant.community(), channel_id)
+        .await
+        .expect("channel");
+    let notices_before = count_kind(&state, &tenant, channel_id, 40099).await;
+
+    // Each test runs in its own database, so this trigger affects only it.
+    for statement in [
+        "CREATE FUNCTION edit_atomic_fail() RETURNS trigger LANGUAGE plpgsql AS \
+         $$ BEGIN RAISE EXCEPTION 'injected topic failure'; END $$",
+        "CREATE TRIGGER edit_atomic_fail BEFORE UPDATE ON channels FOR EACH ROW \
+         WHEN (NEW.topic = 'fail-topic') EXECUTE FUNCTION edit_atomic_fail()",
+    ] {
+        sqlx::query(statement)
+            .execute(state.db.pool())
+            .await
+            .expect("install failure trigger");
+    }
+
+    let event = EventBuilder::new(Kind::Custom(9002), "")
+        .tags([
+            Tag::parse(["h", &channel_id.to_string()]).unwrap(),
+            Tag::parse(["about", "changed"]).unwrap(),
+            Tag::parse(["topic", "fail-topic"]).unwrap(),
+        ])
+        .sign_with_keys(&owner)
+        .expect("sign 9002");
+
+    for attempt in ["first send", "resend"] {
+        match ingest_event(&state, &tenant, event.clone(), http_auth(&owner)).await {
+            Ok(result) => assert!(
+                !result.accepted,
+                "{attempt}: a failed edit was reported as accepted: {:?}",
+                result.message
+            ),
+            Err(error) => assert!(
+                format!("{error:?}").contains("channel update failed"),
+                "{attempt}: {error:?}"
+            ),
+        }
+        let after = state
+            .db
+            .get_channel(tenant.community(), channel_id)
+            .await
+            .expect("channel");
+        assert_eq!(
+            after.description, before.description,
+            "{attempt}: about is unchanged"
+        );
+        assert_eq!(after.topic, before.topic, "{attempt}: topic is unchanged");
+        assert_eq!(
+            count_kind(&state, &tenant, channel_id, 9002).await,
+            0,
+            "{attempt}"
+        );
+        assert_eq!(
+            count_kind(&state, &tenant, channel_id, 40099).await,
+            notices_before,
+            "{attempt}: no system message announces a change that did not happen"
+        );
+    }
+
+    // With the failure gone, the same signed event applies in full.
+    sqlx::query("DROP TRIGGER edit_atomic_fail ON channels")
+        .execute(state.db.pool())
+        .await
+        .expect("drop failure trigger");
+    let retry = ingest_event(&state, &tenant, event, http_auth(&owner))
+        .await
+        .expect("retry");
+    assert!(retry.accepted, "{:?}", retry.message);
+    assert!(retry.message.is_empty(), "{:?}", retry.message);
+    let after = state
+        .db
+        .get_channel(tenant.community(), channel_id)
+        .await
+        .expect("channel");
+    assert_eq!(after.description.as_deref(), Some("changed"));
+    assert_eq!(after.topic.as_deref(), Some("fail-topic"));
+    assert_eq!(count_kind(&state, &tenant, channel_id, 9002).await, 1);
+    assert_eq!(
+        count_kind(&state, &tenant, channel_id, 40099).await,
+        notices_before + 1
+    );
+}
+
+/// A resend of a stored edit is a duplicate and changes nothing: its changes
+/// and system messages committed with it the first time.
+/// Mutation: apply the edit again on a duplicate → RED (second notice).
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn resent_edit_is_a_duplicate() {
+    let state = state().await;
+    let tenant = community(&state, "edit-atomic-resend").await;
+    let owner = Keys::generate();
+    let channel_id = create_channel(&state, &tenant, &owner, &[]).await;
+    let notices_before = count_kind(&state, &tenant, channel_id, 40099).await;
+
+    let event = EventBuilder::new(Kind::Custom(9002), "")
+        .tags([
+            Tag::parse(["h", &channel_id.to_string()]).unwrap(),
+            Tag::parse(["about", "about-1"]).unwrap(),
+            Tag::parse(["topic", "topic-1"]).unwrap(),
+            Tag::parse(["purpose", "purpose-1"]).unwrap(),
+            Tag::parse(["ttl", "3600"]).unwrap(),
+            Tag::parse(["t", "label-1"]).unwrap(),
+        ])
+        .sign_with_keys(&owner)
+        .expect("sign 9002");
+
+    let first = ingest_event(&state, &tenant, event.clone(), http_auth(&owner))
+        .await
+        .expect("first send");
+    assert!(first.accepted, "{:?}", first.message);
+    assert!(first.message.is_empty(), "{:?}", first.message);
+
+    let channel = state
+        .db
+        .get_channel(tenant.community(), channel_id)
+        .await
+        .expect("channel");
+    assert_eq!(channel.description.as_deref(), Some("about-1"));
+    assert_eq!(channel.topic.as_deref(), Some("topic-1"));
+    assert_eq!(
+        channel.topic_set_by.as_deref(),
+        Some(owner.public_key().to_bytes().as_slice())
+    );
+    assert_eq!(channel.purpose.as_deref(), Some("purpose-1"));
+    assert_eq!(channel.ttl_seconds, Some(3600));
+    assert!(channel.ttl_deadline.is_some());
+    assert_eq!(channel.labels, vec!["label-1"]);
+    // topic, purpose and ttl each announce their change.
+    assert_eq!(
+        count_kind(&state, &tenant, channel_id, 40099).await,
+        notices_before + 3
+    );
+
+    let resend = ingest_event(&state, &tenant, event, http_auth(&owner))
+        .await
+        .expect("resend");
+    assert!(resend.accepted, "{:?}", resend.message);
+    assert_eq!(resend.message, "duplicate:");
+    assert_eq!(count_kind(&state, &tenant, channel_id, 9002).await, 1);
+    assert_eq!(
+        count_kind(&state, &tenant, channel_id, 40099).await,
+        notices_before + 3
+    );
+}
+
+/// Restoring a channel that is not archived is refused before storage.
+/// Before, the event was stored and reported accepted while the restore
+/// failed. Mutation: drop the pre-storage check → RED.
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn unarchive_of_an_active_channel_is_refused() {
+    let state = state().await;
+    let tenant = community(&state, "edit-atomic-unarchive").await;
+    let owner = Keys::generate();
+    let channel_id = create_channel(&state, &tenant, &owner, &[]).await;
+    let notices_before = count_kind(&state, &tenant, channel_id, 40099).await;
+
+    let error = edit_channel(
+        &state,
+        &tenant,
+        &owner,
+        channel_id,
+        &[&["archived", "false"]],
+    )
+    .await
+    .expect_err("unarchive of an active channel");
+    assert!(error.contains("channel is not archived"), "{error}");
+    assert_eq!(count_kind(&state, &tenant, channel_id, 9002).await, 0);
+
+    // Archive then restore still works, and each step is announced.
+    edit_channel(
+        &state,
+        &tenant,
+        &owner,
+        channel_id,
+        &[&["archived", "true"]],
+    )
+    .await
+    .expect("archive");
+    edit_channel(
+        &state,
+        &tenant,
+        &owner,
+        channel_id,
+        &[&["archived", "false"]],
+    )
+    .await
+    .expect("unarchive");
+    let channel = state
+        .db
+        .get_channel(tenant.community(), channel_id)
+        .await
+        .expect("channel");
+    assert!(channel.archived_at.is_none());
+    assert_eq!(count_kind(&state, &tenant, channel_id, 9002).await, 2);
+    assert_eq!(
+        count_kind(&state, &tenant, channel_id, 40099).await,
+        notices_before + 2
+    );
+}
+
+/// Resending a committed archive or unarchive is a duplicate, not a refusal.
+/// After the first send commits, the channel is already in the requested
+/// state, so an archive-state check before the duplicate check would refuse
+/// the identical resend (a lost response would look like a failure).
+/// Mutation: check the archive state before the duplicate check in the
+/// atomic write → RED ("channel is not archived").
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn resent_archive_and_unarchive_are_duplicates() {
+    let state = state().await;
+    let tenant = community(&state, "edit-atomic-archive-resend").await;
+    let owner = Keys::generate();
+    let channel_id = create_channel(&state, &tenant, &owner, &[]).await;
+    let notices_before = count_kind(&state, &tenant, channel_id, 40099).await;
+
+    for (value, archived) in [("true", true), ("false", false)] {
+        let event = EventBuilder::new(Kind::Custom(9002), "")
+            .tags([
+                Tag::parse(["h", &channel_id.to_string()]).unwrap(),
+                Tag::parse(["archived", value]).unwrap(),
+            ])
+            .sign_with_keys(&owner)
+            .expect("sign 9002");
+        let first = ingest_event(&state, &tenant, event.clone(), http_auth(&owner))
+            .await
+            .expect("first send");
+        assert!(first.accepted, "archived={value}: {:?}", first.message);
+        assert!(
+            first.message.is_empty(),
+            "archived={value}: {:?}",
+            first.message
+        );
+
+        let resend = ingest_event(&state, &tenant, event, http_auth(&owner))
+            .await
+            .expect("resend");
+        assert!(resend.accepted, "archived={value}: {:?}", resend.message);
+        assert_eq!(resend.message, "duplicate:", "archived={value}");
+
+        let channel = state
+            .db
+            .get_channel(tenant.community(), channel_id)
+            .await
+            .expect("channel");
+        assert_eq!(channel.archived_at.is_some(), archived, "archived={value}");
+    }
+    assert_eq!(count_kind(&state, &tenant, channel_id, 9002).await, 2);
+    assert_eq!(
+        count_kind(&state, &tenant, channel_id, 40099).await,
+        notices_before + 2
+    );
+}
+
+/// A kind:9002 `join_role` change goes through the same transaction as the
+/// other settings: it is stored, and it is announced by one
+/// `join_role_changed` system message. Mutation: drop the `join_role` arm
+/// from the edit → RED (the event changes nothing).
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn join_role_edit_is_applied_and_announced() {
+    let state = state().await;
+    let tenant = community(&state, "edit-atomic-join-role").await;
+    let owner = Keys::generate();
+    let channel_id = create_channel(&state, &tenant, &owner, &[]).await;
+    let notices_before = count_kind(&state, &tenant, channel_id, 40099).await;
+
+    let event = EventBuilder::new(Kind::Custom(9002), "")
+        .tags([
+            Tag::parse(["h", &channel_id.to_string()]).unwrap(),
+            Tag::parse(["join_role", "guest"]).unwrap(),
+        ])
+        .sign_with_keys(&owner)
+        .expect("sign 9002");
+    let result = ingest_event(&state, &tenant, event, http_auth(&owner))
+        .await
+        .expect("ingest");
+    assert!(result.accepted, "{:?}", result.message);
+
+    let after = state
+        .db
+        .get_channel(tenant.community(), channel_id)
+        .await
+        .expect("channel");
+    assert_eq!(after.join_role, "guest");
+    let notices = state
+        .db
+        .query_events(&EventQuery {
+            kinds: Some(vec![40099]),
+            channel_id: Some(channel_id),
+            ..EventQuery::for_community(tenant.community())
+        })
+        .await
+        .expect("query events");
+    assert_eq!(notices.len(), notices_before + 1);
+    let announced = notices
+        .iter()
+        .filter_map(|stored| serde_json::from_str::<serde_json::Value>(&stored.event.content).ok())
+        .any(|content| content["type"] == "join_role_changed" && content["join_role"] == "guest");
+    assert!(announced, "one join_role_changed system message");
+}
+
+/// Sign a kind:9002 that sets `archived` to `value` on `channel_id`.
+fn archive_event(owner: &Keys, channel_id: Uuid, value: &str) -> nostr::Event {
+    EventBuilder::new(Kind::Custom(9002), "")
+        .tags([
+            Tag::parse(["h", &channel_id.to_string()]).unwrap(),
+            Tag::parse(["archived", value]).unwrap(),
+        ])
+        .sign_with_keys(owner)
+        .expect("sign 9002")
+}
+
+/// A resend checked after its first copy committed is a duplicate.
+/// Validation runs on the new archive state and must not refuse the resend;
+/// the write must then report it as a duplicate, with no change and no
+/// notice. This is what a resend sees when it is validated after a racing
+/// first copy commits. Before, validation read the archive state and refused
+/// the resend ("channel is not archived"). Mutation: refuse a restore of an
+/// active channel during validation → RED.
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn resend_validated_after_the_first_commit_is_a_duplicate() {
+    let state = state().await;
+    let tenant = community(&state, "edit-atomic-resend-window").await;
+    let owner = Keys::generate();
+    let channel_id = create_channel(&state, &tenant, &owner, &[]).await;
+
+    for value in ["true", "false"] {
+        let event = archive_event(&owner, channel_id, value);
+        let first = ingest_event(&state, &tenant, event.clone(), http_auth(&owner))
+            .await
+            .expect("first send");
+        assert!(
+            first.accepted && first.message.is_empty(),
+            "{:?}",
+            first.message
+        );
+        let notices = count_kind(&state, &tenant, channel_id, 40099).await;
+
+        crate::handlers::side_effects::validate_admin_event(&tenant, 9002, &event, &state)
+            .await
+            .unwrap_or_else(|e| panic!("archived={value}: resend refused in validation: {e}"));
+        let (_, inserted) =
+            crate::handlers::side_effects::persist_edit_metadata(&tenant, &event, &state)
+                .await
+                .expect("resend write");
+        assert!(
+            !inserted,
+            "archived={value}: the resend changed the channel"
+        );
+        assert_eq!(
+            count_kind(&state, &tenant, channel_id, 40099).await,
+            notices,
+            "archived={value}: the resend added a notice"
+        );
+    }
+    assert_eq!(count_kind(&state, &tenant, channel_id, 9002).await, 2);
+}
+
+/// Two copies of one restore, sent at the same time, change the channel
+/// once. Both are held at the channel's TTL lock until both wait there, so
+/// both pass validation before either commits. One applies the restore; the
+/// other is reported as a duplicate, with no second notice. The case
+/// where a copy validates after the other commits is covered by
+/// `resend_validated_after_the_first_commit_is_a_duplicate`.
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn concurrent_copies_of_a_restore_apply_once() {
+    let pool = sqlx::PgPool::connect(&crate::test_support::database_url())
+        .await
+        .expect("connect test pool");
+    let state = crate::state::tests::test_state_with_database_pool(pool.clone()).await;
+    let tenant = community(&state, "edit-atomic-restore-race").await;
+    let owner = Keys::generate();
+    let channel_id = create_channel(&state, &tenant, &owner, &[]).await;
+    edit_channel(
+        &state,
+        &tenant,
+        &owner,
+        channel_id,
+        &[&["archived", "true"]],
+    )
+    .await
+    .expect("archive");
+    let notices = count_kind(&state, &tenant, channel_id, 40099).await;
+
+    let lock_key = format!(
+        "buzz_channel_ttl:{}:{channel_id}",
+        tenant.community().as_uuid()
+    );
+    let mut holder = pool.begin().await.expect("begin lock holder");
+    sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
+        .bind(&lock_key)
+        .execute(&mut *holder)
+        .await
+        .expect("hold the TTL lock");
+
+    let event = archive_event(&owner, channel_id, "false");
+    let tasks: Vec<_> = (0..2)
+        .map(|_| {
+            let state = std::sync::Arc::clone(&state);
+            let tenant = tenant.clone();
+            let event = event.clone();
+            let auth = http_auth(&owner);
+            tokio::spawn(async move { ingest_event(&state, &tenant, event, auth).await })
+        })
+        .collect();
+
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
+    loop {
+        let waiting: i64 = sqlx::query_scalar(
+            "SELECT count(*) FROM pg_locks WHERE locktype = 'advisory' AND NOT granted \
+             AND ((classid::bigint << 32) | objid::bigint) = hashtextextended($1, 0)",
+        )
+        .bind(&lock_key)
+        .fetch_one(&pool)
+        .await
+        .expect("read lock waiters");
+        if waiting == 2 {
+            break;
+        }
+        assert!(
+            tasks.iter().all(|task| !task.is_finished()),
+            "a copy finished without waiting for the lock"
+        );
+        assert!(
+            std::time::Instant::now() < deadline,
+            "both copies never waited for the TTL lock"
+        );
+        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+    }
+    holder.commit().await.expect("release the TTL lock");
+
+    let mut messages = Vec::new();
+    for task in tasks {
+        let result = tokio::time::timeout(std::time::Duration::from_secs(10), task)
+            .await
+            .expect("a copy returns after the lock is released")
+            .expect("task must not panic")
+            .expect("both copies are accepted");
+        assert!(result.accepted, "{:?}", result.message);
+        messages.push(result.message);
+    }
+    messages.sort();
+    assert_eq!(messages, ["", "duplicate:"]);
+    let channel = state
+        .db
+        .get_channel(tenant.community(), channel_id)
+        .await
+        .expect("channel");
+    assert!(channel.archived_at.is_none());
+    assert_eq!(
+        count_kind(&state, &tenant, channel_id, 40099).await,
+        notices + 1
+    );
+}

crates/buzz-relay/src/handlers/ingest.rs

@@ -3031,17 +3031,15 @@ async fn ingest_event_inner(
     }
 
     if channel_id.is_some() {
-        // Allow kind:9002 with archived=false (unarchive operation)
-        let is_unarchive = kind_u32 == KIND_NIP29_EDIT_METADATA
-            && event.tags.iter().any(|t| {
-                let parts = t.as_slice();
-                parts.len() >= 2 && parts[0] == "archived" && parts[1] == "false"
-            });
+        // kind:9002 is not checked here: its atomic write decides the
+        // archive state under the channel row lock, after its duplicate
+        // check, so a resend that races the first commit is a duplicate.
+        let is_edit_metadata = kind_u32 == KIND_NIP29_EDIT_METADATA;
 
         // A member may still leave an archived channel by themselves.
         let is_self_departure = crate::handlers::side_effects::is_self_departure(&event);
 
-        if !is_unarchive && !is_self_departure {
+        if !is_edit_metadata && !is_self_departure {
             if let Some(channel) = &channel_row {
                 if channel.archived_at.is_some() {
                     return Err(match target_denial {
@@ -3523,6 +3521,19 @@ async fn ingest_event_inner(
         crate::handlers::side_effects::persist_workflow_deletion(tenant, &event, state)
             .await
             .map_err(|e| IngestError::Internal(format!("error: workflow deletion failed: {e}")))?
+    } else if kind_u32 == KIND_NIP29_EDIT_METADATA {
+        // One commit stores the event and applies its channel changes. A
+        // failure stores nothing, so the client is told and can resend it.
+        // A refusal decided under the lock (for example, restoring a channel
+        // that is not archived) is the client's error; anything else is ours.
+        crate::handlers::side_effects::persist_edit_metadata(tenant, &event, state)
+            .await
+            .map_err(|e| match e.downcast_ref::<buzz_db::DbError>() {
+                Some(buzz_db::DbError::AccessDenied(reason)) => {
+                    IngestError::Rejected(format!("invalid: {reason}"))
+                }
+                _ => IngestError::Internal(format!("error: channel update failed: {e}")),
+            })?
     } else if buzz_core::kind::is_replaceable(kind_u32) {
         // NIP-16 replaceable event — atomic replace with stale-write protection.
         // channel_id is None for global kinds (0, 1, 3) due to step 5b above.
@@ -3614,7 +3625,10 @@ async fn ingest_event_inner(
         });
     }
 
-    if !workflow_deletion && crate::handlers::side_effects::is_side_effect_kind(kind_u32) {
+    // Workflow deletion and kind:9002 applied their changes in the insert's
+    // own transaction above.
+    let persisted_atomically = workflow_deletion || kind_u32 == KIND_NIP29_EDIT_METADATA;
+    if !persisted_atomically && crate::handlers::side_effects::is_side_effect_kind(kind_u32) {
         if let Err(e) =
             crate::handlers::side_effects::handle_side_effects(tenant, kind_u32, &event, state)
                 .await

crates/buzz-relay/src/handlers/mod.rs

@@ -14,6 +14,8 @@ pub mod community_provisioning;
 pub mod count;
 #[cfg(test)]
 mod create_with_members_postgres_tests;
+#[cfg(test)]
+mod edit_metadata_postgres_tests;
 /// EVENT handler — WS dispatcher → ingest pipeline → fan-out.
 pub mod event;
 #[cfg(test)]

crates/buzz-relay/src/handlers/side_effects.rs

@@ -337,7 +337,8 @@ pub async fn handle_side_effects(
         5 => handle_standard_deletion_event(tenant, event, state).await,
         9000 => handle_put_user(tenant, event, state).await,
         9001 => handle_remove_user(tenant, event, state).await,
-        9002 => handle_edit_metadata(tenant, event, state).await,
+        // kind:9002 belongs to the atomic persistence path in ingest.
+        9002 => Err(anyhow::anyhow!("kind:9002 requires atomic persistence")),
         9005 => handle_delete_event_side_effect(tenant, event, state).await,
         9007 => handle_create_group(tenant, event, state).await,
         9008 => handle_delete_group(tenant, event, state).await,
@@ -578,20 +579,17 @@ pub async fn validate_admin_event(
 
     let actor_bytes = event.pubkey.to_bytes().to_vec();
 
-    // Reject mutations on archived channels — except kind:9002 with archived=false
-    // (unarchive), which must be allowed through so the channel can be restored,
-    // and a member leaving by themselves (still subject to the last-owner rule).
+    // Reject mutations on archived channels, except a member leaving by
+    // themselves (still subject to the last-owner rule). kind:9002 is not
+    // checked here: its atomic write decides the archive state under the
+    // channel row lock, after its duplicate check, so a resend that races
+    // the first commit is a duplicate and not a refusal.
     let channel = state
         .db
         .get_channel_for_event_write(tenant.community(), channel_id)
         .await
         .map_err(|_| anyhow::anyhow!("channel not found"))?;
-    let is_unarchive_request = kind == 9002
-        && event.tags.iter().any(|t| {
-            let parts = t.as_slice();
-            parts.len() >= 2 && parts[0] == "archived" && parts[1] == "false"
-        });
-    if channel.archived_at.is_some() && !is_unarchive_request && !is_self_departure(event) {
+    if kind != 9002 && channel.archived_at.is_some() && !is_self_departure(event) {
         return Err(anyhow::anyhow!("channel is archived"));
     }
 
@@ -974,14 +972,7 @@ pub async fn emit_system_message(
     content: serde_json::Value,
     idempotency_ts: chrono::DateTime<chrono::Utc>,
 ) -> anyhow::Result<()> {
-    let channel_tag = Tag::parse(["h", &channel_id.to_string()])?;
-
-    let ts = nostr::Timestamp::from(idempotency_ts.timestamp() as u64);
-    let event = EventBuilder::new(Kind::Custom(40099), content.to_string())
-        .tags([channel_tag])
-        .custom_created_at(ts)
-        .sign_with_keys(&state.relay_keypair)
-        .map_err(|e| anyhow::anyhow!("failed to sign system message: {e}"))?;
+    let event = sign_system_message(state, channel_id, content, idempotency_ts)?;
 
     // Durable insert is the completion boundary — propagate failure.
     state
@@ -990,16 +981,41 @@ pub async fn emit_system_message(
         .await
         .map_err(|e| anyhow::anyhow!("system message insert failed: {e}"))?;
 
-    // Fan out to subscribers: best-effort, clients can retrieve the persisted event.
+    fan_out_system_message(tenant, state, channel_id, &event).await;
+    Ok(())
+}
+
+/// Sign a kind:40099 system message for `channel_id` with the relay keypair.
+fn sign_system_message(
+    state: &Arc<AppState>,
+    channel_id: Uuid,
+    content: serde_json::Value,
+    created_at: chrono::DateTime<chrono::Utc>,
+) -> anyhow::Result<Event> {
+    let channel_tag = Tag::parse(["h", &channel_id.to_string()])?;
+    let ts = nostr::Timestamp::from(created_at.timestamp() as u64);
+    EventBuilder::new(Kind::Custom(40099), content.to_string())
+        .tags([channel_tag])
+        .custom_created_at(ts)
+        .sign_with_keys(&state.relay_keypair)
+        .map_err(|e| anyhow::anyhow!("failed to sign system message: {e}"))
+}
+
+/// Fan out a stored system message. Best-effort: clients can retrieve the
+/// persisted event.
+async fn fan_out_system_message(
+    tenant: &TenantContext,
+    state: &Arc<AppState>,
+    channel_id: Uuid,
+    event: &Event,
+) {
     if let Err(e) = state
         .pubsub
-        .publish_event(tenant, EventTopic::Channel(channel_id), &event)
+        .publish_event(tenant, EventTopic::Channel(channel_id), event)
         .await
     {
         warn!("System message fan-out failed: {e}");
     }
-
-    Ok(())
 }
 
 /// Sign and fan out a fresh relay-signed `kind:39005` thread-summary overlay
@@ -1691,281 +1707,205 @@ async fn handle_remove_user(
     Ok(())
 }
 
-async fn handle_edit_metadata(
+/// Persist an authorized kind:9002 event and the channel changes it names.
+///
+/// One transaction stores the event, applies every change and stores the
+/// system messages that announce them. If it fails, nothing is stored and the
+/// client is told so, so it can resend the same event. Live effects (cache
+/// invalidation, subscription eviction, fan-out, discovery events) run only
+/// after the commit.
+pub(crate) async fn persist_edit_metadata(
     tenant: &TenantContext,
     event: &Event,
     state: &Arc<AppState>,
-) -> anyhow::Result<()> {
+) -> anyhow::Result<(buzz_core::StoredEvent, bool)> {
     let channel_id =
         extract_h_tag_channel(event).ok_or_else(|| anyhow::anyhow!("missing h tag"))?;
     let actor_bytes = event.pubkey.to_bytes().to_vec();
     let actor_hex = hex::encode(&actor_bytes);
 
+    // kind:9002 is a partial update: a missing tag means "no change". If a
+    // tag repeats, its last value wins.
+    let mut edit = buzz_db::channel::ChannelMetadataEdit::default();
     for tag in event.tags.iter() {
-        let key = tag.kind().to_string();
-        if let Some(val) = tag.content() {
-            match key.as_str() {
-                "name" => {
-                    state
-                        .db
-                        .update_channel(
-                            tenant.community(),
-                            channel_id,
-                            buzz_db::channel::ChannelUpdate {
-                                name: Some(val.to_string()),
-                                ..Default::default()
-                            },
-                        )
-                        .await?;
-                }
-                "about" => {
-                    state
-                        .db
-                        .update_channel(
-                            tenant.community(),
-                            channel_id,
-                            buzz_db::channel::ChannelUpdate {
-                                description: Some(val.to_string()),
-                                ..Default::default()
-                            },
-                        )
-                        .await?;
-                }
-                "topic" => {
-                    state
-                        .db
-                        .set_topic(tenant.community(), channel_id, val, &actor_bytes)
-                        .await?;
-                    emit_system_message(
-                        tenant,
-                        state,
-                        channel_id,
-                        serde_json::json!({
-                            "type": "topic_changed", "actor": actor_hex, "topic": val
-                        }),
-                        chrono::Utc::now(),
-                    )
-                    .await?;
-                }
-                "purpose" => {
-                    state
-                        .db
-                        .set_purpose(tenant.community(), channel_id, val, &actor_bytes)
-                        .await?;
-                    emit_system_message(
-                        tenant,
-                        state,
-                        channel_id,
-                        serde_json::json!({
-                            "type": "purpose_changed", "actor": actor_hex, "purpose": val
-                        }),
-                        chrono::Utc::now(),
-                    )
-                    .await?;
-                }
-                "visibility" => {
-                    let was_open = state
-                        .db
-                        .get_channel_for_event_write(tenant.community(), channel_id)
-                        .await
-                        .map(|c| c.visibility == "open")
-                        .unwrap_or(false);
-                    state
-                        .db
-                        .update_channel(
-                            tenant.community(),
-                            channel_id,
-                            buzz_db::channel::ChannelUpdate {
-                                visibility: Some(val.to_string()),
-                                ..Default::default()
-                            },
-                        )
-                        .await?;
-                    // A visibility flip changes who can see the channel, so the
-                    // accessible-channels and visibility caches must be cleared
-                    // before any later event for this channel fans out.
-                    state.invalidate_all_accessible_channels(tenant);
-                    state.invalidate_channel_visibility(tenant, channel_id);
-                    // On open -> private, eagerly close non-members' live subs
-                    // for an immediate CLOSED on this node. The fan-out access
-                    // filter is the cluster-wide correctness backstop.
-                    if was_open && val == "private" {
-                        evict_non_member_channel_subscriptions(tenant, state, channel_id).await?;
-                    }
-                    emit_system_message(
-                        tenant,
-                        state,
-                        channel_id,
-                        serde_json::json!({
-                            "type": "visibility_changed", "actor": actor_hex, "visibility": val
-                        }),
-                        chrono::Utc::now(),
-                    )
-                    .await?;
-                }
-                "join_role" => {
-                    // Validated during authorization; parse again so a bad
-                    // value can never reach the column.
-                    let join_role: buzz_db::channel::ChannelJoinRole =
-                        val.parse().map_err(|e: String| anyhow::anyhow!(e))?;
-                    state
-                        .db
-                        .update_channel(
-                            tenant.community(),
-                            channel_id,
-                            buzz_db::channel::ChannelUpdate {
-                                join_role: Some(join_role.as_str().to_string()),
-                                ..Default::default()
-                            },
-                        )
-                        .await?;
-                    emit_system_message(
-                        tenant,
-                        state,
-                        channel_id,
-                        serde_json::json!({
-                            "type": "join_role_changed", "actor": actor_hex, "join_role": join_role.as_str()
-                        }),
-                        chrono::Utc::now(),
-                    )
-                    .await?;
-                }
-                "ttl" => {
-                    // Empty string clears the TTL (permanent); otherwise it is a
-                    // positive integer of seconds, validated during authorization.
-                    // Fail closed: a parse failure must reject, never silently
-                    // clear the TTL to permanent.
-                    let ttl_change: Option<i32> = if val.is_empty() {
-                        None
-                    } else {
-                        Some(val.parse::<i32>().map_err(|_| {
-                            anyhow::anyhow!("invalid ttl value: {val} (must be a positive integer)")
-                        })?)
-                    };
-                    state
-                        .db
-                        .update_channel(
-                            tenant.community(),
-                            channel_id,
-                            buzz_db::channel::ChannelUpdate {
-                                ttl_seconds: Some(ttl_change),
-                                ..Default::default()
-                            },
-                        )
-                        .await?;
-                    emit_system_message(
-                        tenant,
-                        state,
-                        channel_id,
-                        serde_json::json!({
-                            "type": "ttl_changed", "actor": actor_hex, "ttl_seconds": ttl_change
-                        }),
-                        chrono::Utc::now(),
-                    )
-                    .await?;
-                }
-                "archived" => {
-                    match val {
-                        "true" => {
-                            state
-                                .db
-                                .archive_channel(tenant.community(), channel_id)
-                                .await?;
-                            emit_system_message(
-                                tenant,
-                                state,
-                                channel_id,
-                                serde_json::json!({
-                                    "type": "channel_archived", "actor": actor_hex
-                                }),
-                                chrono::Utc::now(),
-                            )
-                            .await?;
-                        }
-                        "false" => {
-                            state
-                                .db
-                                .unarchive_channel(tenant.community(), channel_id)
-                                .await?;
-                            emit_system_message(
-                                tenant,
-                                state,
-                                channel_id,
-                                serde_json::json!({
-                                    "type": "channel_unarchived", "actor": actor_hex
-                                }),
-                                chrono::Utc::now(),
-                            )
-                            .await?;
+        let Some(val) = tag.content() else { continue };
+        match tag.kind().to_string().as_str() {
+            "name" => edit.columns.name = Some(val.to_string()),
+            "about" => edit.columns.description = Some(val.to_string()),
+            "topic" => edit.topic = Some(val.to_string()),
+            "purpose" => edit.purpose = Some(val.to_string()),
+            "visibility" => edit.columns.visibility = Some(val.to_string()),
+            "join_role" => {
+                // Validated during authorization; parse again so a bad
+                // value can never reach the column.
+                let join_role: buzz_db::channel::ChannelJoinRole =
+                    val.parse().map_err(|e: String| anyhow::anyhow!(e))?;
+                edit.columns.join_role = Some(join_role.as_str().to_string());
+            }
+            "ttl" => {
+                // Empty string clears the TTL (permanent); otherwise it is a
+                // positive integer of seconds, validated during authorization.
+                // Fail closed: a parse failure must reject, never silently
+                // clear the TTL to permanent.
+                edit.columns.ttl_seconds = Some(if val.is_empty() {
+                    None
+                } else {
+                    Some(val.parse::<i32>().map_err(|_| {
+                        anyhow::anyhow!("invalid ttl value: {val} (must be a positive integer)")
+                    })?)
+                });
+            }
+            "archived" => match val {
+                "true" => edit.archived = Some(true),
+                "false" => edit.archived = Some(false),
+                _ => {} // refused during authorization
+            },
+            _ => {}
+        }
+    }
+    // All `t` tags together form the new label set. `["t", ""]` alone clears it.
+    if let Some(labels) = channel_labels_from_tags(event) {
+        edit.columns.labels = Some(labels.map_err(|e| anyhow::anyhow!("{e}"))?);
+    }
 
-                            // Resubscribe connected agents after restore: archiving evicts their
-                            // live subscriptions (CLOSED "channel access revoked") and unarchive
-                            // otherwise emits no signal that makes a connected agent resubscribe.
-                            // We reuse the member_added notification (44100) purely as a resubscribe
-                            // trigger — no membership actually changed here — because it flows on the
-                            // agent's always-live global membership subscription, the same path
-                            // remove/re-add uses to recover. Humans self-heal via the re-emitted
-                            // kind:39000 discovery, so this is intentionally agent-scoped.
-                            //
-                            // Known limitation: emit_membership_notification builds a created_at=now
-                            // event with no nonce, and insert_event skips fan-out on a duplicate id.
-                            // Four sub-second toggles (archive->unarchive->archive->unarchive) on the
-                            // same channel by the same actor could collide ids and skip a fan-out.
-                            // Not reachable in practice — unarchive has a single human-driven caller;
-                            // the reaper only auto-archives — so we don't engineer around it.
-                            for member in state
-                                .db
-                                .get_members_for_event_write(tenant.community(), channel_id)
-                                .await?
-                            {
-                                if let Err(e) = emit_membership_notification(
-                                    tenant,
-                                    state,
-                                    channel_id,
-                                    &member.pubkey,
-                                    &actor_bytes,
-                                    KIND_MEMBER_ADDED_NOTIFICATION,
-                                )
-                                .await
-                                {
-                                    warn!(
-                                        channel = %channel_id,
-                                        error = %e,
-                                        "post-unarchive resubscribe notification failed"
-                                    );
-                                }
-                            }
-                        }
-                        _ => {} // ignore invalid values
-                    }
-                }
-                _ => {}
+    let now = chrono::Utc::now();
+    let mut notices = Vec::new();
+    if let Some(topic) = &edit.topic {
+        notices.push(serde_json::json!({
+            "type": "topic_changed", "actor": actor_hex, "topic": topic
+        }));
+    }
+    if let Some(purpose) = &edit.purpose {
+        notices.push(serde_json::json!({
+            "type": "purpose_changed", "actor": actor_hex, "purpose": purpose
+        }));
+    }
+    if let Some(visibility) = &edit.columns.visibility {
+        notices.push(serde_json::json!({
+            "type": "visibility_changed", "actor": actor_hex, "visibility": visibility
+        }));
+    }
+    if let Some(join_role) = &edit.columns.join_role {
+        notices.push(serde_json::json!({
+            "type": "join_role_changed", "actor": actor_hex, "join_role": join_role
+        }));
+    }
+    if let Some(ttl_seconds) = edit.columns.ttl_seconds {
+        notices.push(serde_json::json!({
+            "type": "ttl_changed", "actor": actor_hex, "ttl_seconds": ttl_seconds
+        }));
+    }
+    match edit.archived {
+        Some(true) => notices.push(serde_json::json!({
+            "type": "channel_archived", "actor": actor_hex
+        })),
+        Some(false) => notices.push(serde_json::json!({
+            "type": "channel_unarchived", "actor": actor_hex
+        })),
+        None => {}
+    }
+    let system_messages = notices
+        .into_iter()
+        .map(|content| sign_system_message(state, channel_id, content, now))
+        .collect::<anyhow::Result<Vec<_>>>()?;
+
+    let new_visibility = edit.columns.visibility.clone();
+    let unarchived = edit.archived == Some(false);
+    let write = state
+        .db
+        .apply_channel_metadata_event(
+            tenant.community(),
+            channel_id,
+            event,
+            edit,
+            &system_messages,
+        )
+        .await?;
+    if !write.inserted {
+        return Ok((write.stored, false));
+    }
+
+    // The changes are committed. What follows only updates live state, so a
+    // failure is logged; the next read sees the committed channel.
+    if let Some(visibility) = new_visibility.as_deref() {
+        // A visibility flip changes who can see the channel, so the
+        // accessible-channels and visibility caches must be cleared
+        // before any later event for this channel fans out.
+        state.invalidate_all_accessible_channels(tenant);
+        state.invalidate_channel_visibility(tenant, channel_id);
+        // On open -> private, eagerly close non-members' live subs
+        // for an immediate CLOSED on this node. The fan-out access
+        // filter is the cluster-wide correctness backstop.
+        if write.previous_visibility == "open" && visibility == "private" {
+            if let Err(e) = evict_non_member_channel_subscriptions(tenant, state, channel_id).await
+            {
+                warn!(channel = %channel_id, error = %e, "non-member subscription eviction failed");
             }
         }
     }
-
-    // All `t` tags together form the new label set, so they are applied
-    // once after the loop. `["t", ""]` alone clears the set.
-    if let Some(labels) = channel_labels_from_tags(event) {
-        let labels = labels.map_err(|e| anyhow::anyhow!("{e}"))?;
-        state
-            .db
-            .update_channel(
-                tenant.community(),
-                channel_id,
-                buzz_db::channel::ChannelUpdate {
-                    labels: Some(labels),
-                    ..Default::default()
-                },
-            )
-            .await?;
+    for message in &system_messages {
+        fan_out_system_message(tenant, state, channel_id, message).await;
+    }
+    if unarchived {
+        resubscribe_members_after_unarchive(tenant, state, channel_id, &actor_bytes).await;
     }
-
     if let Err(e) = emit_group_discovery_events(tenant, state, channel_id).await {
         warn!(channel = %channel_id, error = %e, "NIP-29 group discovery emission failed");
     }
 
-    Ok(())
+    Ok((write.stored, true))
+}
+
+/// Resubscribe connected agents after restore: archiving evicts their
+/// live subscriptions (CLOSED "channel access revoked") and unarchive
+/// otherwise emits no signal that makes a connected agent resubscribe.
+/// We reuse the member_added notification (44100) purely as a resubscribe
+/// trigger — no membership actually changed here — because it flows on the
+/// agent's always-live global membership subscription, the same path
+/// remove/re-add uses to recover. Humans self-heal via the re-emitted
+/// kind:39000 discovery, so this is intentionally agent-scoped.
+///
+/// Known limitation: emit_membership_notification builds a created_at=now
+/// event with no nonce, and insert_event skips fan-out on a duplicate id.
+/// Four sub-second toggles (archive->unarchive->archive->unarchive) on the
+/// same channel by the same actor could collide ids and skip a fan-out.
+/// Not reachable in practice — unarchive has a single human-driven caller;
+/// the reaper only auto-archives — so we don't engineer around it.
+async fn resubscribe_members_after_unarchive(
+    tenant: &TenantContext,
+    state: &Arc<AppState>,
+    channel_id: Uuid,
+    actor_bytes: &[u8],
+) {
+    let members = match state
+        .db
+        .get_members_for_event_write(tenant.community(), channel_id)
+        .await
+    {
+        Ok(members) => members,
+        Err(e) => {
+            warn!(channel = %channel_id, error = %e, "post-unarchive member read failed");
+            return;
+        }
+    };
+    for member in members {
+        if let Err(e) = emit_membership_notification(
+            tenant,
+            state,
+            channel_id,
+            &member.pubkey,
+            actor_bytes,
+            KIND_MEMBER_ADDED_NOTIFICATION,
+        )
+        .await
+        {
+            warn!(
+                channel = %channel_id,
+                error = %e,
+                "post-unarchive resubscribe notification failed"
+            );
+        }
+    }
 }
 
 /// The channel label set named by an event's `["t", <label>]` tags, or