From 4e81ffe145993937df07683401e36b6a70e5ccb7 Mon Sep 17 00:00:00 2001 From: Matt Toohey Date: Thu, 29 Oct 2026 16:38:34 +1100 Subject: [PATCH] fix(db): refuse new members of an archived channel under the channel row lock An auto-join could add a new community member to a default channel after it was archived (jedwards27, second review on #8594). The auto-join lists the effective default channels and `join_channel_as_self` checks the archive state before the write, but `put_members` never read `archived_at`, and nothing ordered an archive against the membership write. An archive landing between the list and the insert still created the row and emitted the kind:40099, 39000/39002 and 44100. `put_members` and `put_members_with_event` now take the membership lock, the shared TTL lock and then the channel row `FOR NO KEY UPDATE`, the order a huddle join already uses, and refuse a target who is not an active member with `AccessDenied("channel is archived")` while `archived_at` is set. An auto-join that loses the race is logged and counted as a failed channel, with no row and no side effects. Active members keep their rows; only admission is refused. The review suggested treating archive and unarchive as admission changes so they take the membership lock. This fix locks the row instead, because the row is what every archive writes: `archive_channel`, a kind:9002 `archived` tag and the TTL reaper's single UPDATE, which takes no advisory lock at all and so would still race under the membership-lock version. It also leaves every archive writer unchanged. The refusal is not scoped to self-joins: the only production callers are kind:9000 (already refused on an archived channel at ingest), kind:9021 and the auto-join. No restore or import path re-adds members through `put_members`, and no test relies on adding to an archived channel. The cost is that a membership write now holds the channel row for its transaction. Writers of that row (settings writes, huddle joins, TTL refreshes, the reaper) wait for it, and reads do not. NOSTR.md "Default channels" now says a channel archived or made private while the join runs is one of the logged, unretried failures. Tests: - `put_member_waits_for_a_concurrent_archive` (buzz-db Postgres): an uncommitted reaper-style archive makes a join wait. Once the archive commits the join is refused with no row, and re-adding a removed member is refused too. In the other order, `archive_channel` waits for an admission that holds the locks, and the admitted member stays. - `auto_join_skips_a_default_channel_archived_while_it_joins` (buzz-relay Postgres): a flagged channel is archived in a held transaction. The auto-join lists it, passes the join body's check, then waits in `put_member`. After the archive commits there is no membership row, no kind:39002 entry, no kind:44100 and no `member_joined` notice. - Mutations: dropping the row lock from `lock_channel_for_member_put` fails both tests, and so does dropping the archived refusal from `put_members_on`. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Matt Toohey --- NOSTR.md | 4 +- crates/buzz-db/src/store/channel.rs | 8 +- crates/buzz-db/src/store/channel_members.rs | 174 +++++++++++++++++- crates/buzz-relay/src/api/invites.rs | 125 +++++++++++++ .../buzz-relay/src/handlers/side_effects.rs | 8 +- 5 files changed, 305 insertions(+), 14 deletions(-) diff --git a/NOSTR.md b/NOSTR.md index 856b6f414a..0973bba412 100644 --- a/NOSTR.md +++ b/NOSTR.md @@ -226,7 +226,9 @@ admin flags. It does this once, at that moment, and never again on login or reconnect, so a later kind:9022 leave stands. A channel that fails to join is logged and counted (`buzz_auto_join_failures_total`), not retried: the kind:39000 tag is the -record a client reconciles against. `buzz-admin add-member` writes the +record a client reconciles against. That includes a channel archived or made +private while the join runs: the membership write decides both again under +the channel's locks, so the member is not added and nothing is announced. `buzz-admin add-member` writes the roster row only and does not join default channels. Auto-join fires only at those roster-insert sites, so a community that does not require relay membership advertises the kind:39000 tag but auto-joins nobody; a client that diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 3487ce14c4..69db951b2a 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -988,9 +988,11 @@ impl ChannelSettingsWriter<'_> { /// 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. +/// the TTL lock and updates the row. A huddle join and a `put_members` +/// membership write take the membership lock, then the TTL lock, then the +/// row; the row lock is what orders them against an archive, which does not +/// take the membership lock. 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>( diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 796031fb69..ca95fb88fa 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -651,6 +651,11 @@ pub async fn add_member( /// neither a membership write nor a setting change can slip between the /// decision and the write. /// +/// An archived channel admits nobody new: a target who is not an active +/// member is refused with `AccessDenied` while `archived_at` is set. The +/// archive state is read under the channel row lock, which every archive +/// write needs too (see [`lock_channel_for_member_put`]). +/// /// There is no creator exception: channel creation installs the creator as /// owner in the same transaction, so a creator without an active row was /// removed and is admitted like anyone else. @@ -718,9 +723,10 @@ pub async fn put_members( ) .await?; - // First statement: serialize the whole decision / owner-count / upsert - // sequence against concurrent membership writes on this channel. - acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await?; + // First statements: serialize the whole decision / owner-count / upsert + // sequence against concurrent membership writes and archives of this + // channel. + lock_channel_for_member_put(&mut tx, channel_id).await?; let results = put_members_on(&mut tx, channel_id, targets, invited_by).await?; tx.commit().await?; Ok(results) @@ -747,9 +753,8 @@ pub enum MemberPutEventWrite { /// transaction. If any write fails, nothing is stored, so the client can /// resend the same event and have it applied. /// -/// The membership lock comes first, as in [`put_members`] and the channel -/// settings writers, so a second copy of the event waits for the first and -/// then finds it stored. +/// The locks of [`put_members`] come first, so a second copy of the event +/// waits for the first and then finds it stored. pub(crate) async fn put_members_with_event( pool: &PgPool, push_enqueue: Option, @@ -767,7 +772,7 @@ pub(crate) async fn put_members_with_event( ) .await?; tx.set_push_enqueue(push_enqueue); - acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await?; + lock_channel_for_member_put(&mut tx, channel_id).await?; // The same writes as `Db::insert_event_with_thread_metadata`. let (stored, inserted) = crate::event::insert_event_with_thread_metadata_tx( @@ -795,6 +800,23 @@ pub(crate) async fn put_members_with_event( Ok(MemberPutEventWrite::Applied { stored, results }) } +/// Take the locks of a [`put_members`] write on `tx`, in the channel lock +/// order (see `channel::lock_channel_for_settings_write`): the membership +/// lock, the shared TTL lock, then the channel row `FOR NO KEY UPDATE`. +/// +/// The row lock is what orders the write against an archive. Every archive +/// and unarchive updates that row (`archive_channel`, a kind:9002 `archived` +/// tag, the TTL reaper's single UPDATE), and none of them takes the +/// membership lock. So an archive either commits before [`put_members_on`] +/// reads `archived_at`, or waits until this write commits. +async fn lock_channel_for_member_put(tx: &mut AdmittedTx, channel_id: Uuid) -> Result<()> { + let community_id = tx.community(); + acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await?; + lock_channel_ttl_shared_in_transaction(tx, channel_id).await?; + lock_channel_archived_at_in_transaction(tx, channel_id).await?; + Ok(()) +} + fn check_distinct_targets(targets: &[MemberPut<'_>]) -> Result<()> { let mut seen = std::collections::HashSet::with_capacity(targets.len()); match targets.iter().find(|t| !seen.insert(t.pubkey)) { @@ -807,7 +829,8 @@ fn check_distinct_targets(targets: &[MemberPut<'_>]) -> Result<()> { } /// The decisions and upsert of [`put_members`], on a transaction that holds -/// `channel_id`'s membership lock. The caller owns the commit. +/// `channel_id`'s [`lock_channel_for_member_put`] locks. The caller owns the +/// commit. async fn put_members_on( tx: &mut AdmittedTx, channel_id: Uuid, @@ -816,6 +839,7 @@ async fn put_members_on( ) -> Result>> { let community_id = tx.community(); let channel = get_channel_tx(tx.conn(), community_id, channel_id).await?; + let archived = channel.archived_at.is_some(); let private = channel.visibility == "private"; let join_role = buzz_core::channel::ChannelJoinRole::parse_fail_closed(&channel.join_role); @@ -853,6 +877,12 @@ async fn put_members_on( } let actor = invited_by.unwrap_or(target.pubkey); let current_role = roles.get(target.pubkey).cloned(); + if archived && current_role.is_none() { + decided.push(Err(DbError::AccessDenied( + "channel is archived".to_string(), + ))); + continue; + } let actor_role: Option = roles .get(actor) .map(|r| { @@ -3852,6 +3882,134 @@ mod postgres_tests { .expect("join_role change"); } + /// An archive and an admission are ordered by the channel row lock, and an + /// archived channel admits nobody new. A join that waits on an uncommitted + /// archive is refused once it commits, with no row, and re-adding a + /// removed member is refused too. The other order: an archive waits for + /// an admission that holds the locks. + /// The archive here is the TTL reaper's single UPDATE, which takes no + /// advisory lock. Mutations: drop the row lock from + /// `lock_channel_for_member_put`, or the archived refusal from + /// `put_members_on` → RED. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn put_member_waits_for_a_concurrent_archive() { + 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 joiner = random_pubkey(); + let removed = random_pubkey(); + for pk in [&owner, &joiner, &removed] { + ensure_user(&pool, community, pk) + .await + .expect("ensure user"); + } + let channel = create_test_channel( + &pool, + community_id, + "put-member-archive-race", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner, + None, + ) + .await + .expect("create channel"); + put_member(&pool, community, channel.id, &removed, None, None) + .await + .expect("join before the archive"); + remove_member(&pool, community, channel.id, &removed, &removed) + .await + .expect("leave before the archive"); + + let mut archive = pool.begin().await.expect("begin archive"); + sqlx::query( + "UPDATE channels SET archived_at = NOW() \ + WHERE community_id = $1 AND id = $2 AND archived_at IS NULL", + ) + .bind(community_id) + .bind(channel.id) + .execute(&mut *archive) + .await + .expect("archive"); + + let join_pool = crate::test_support::named_pool("put-member-archive-race").await; + let join = tokio::spawn({ + let channel_id = channel.id; + let joiner = joiner.clone(); + async move { put_member(&join_pool, community, channel_id, &joiner, None, None).await } + }); + crate::test_support::wait_for_lock_wait(&pool, "put-member-archive-race").await; + assert!(!join.is_finished(), "the join must wait for the archive"); + archive.commit().await.expect("commit archive"); + + let err = join + .await + .expect("join task") + .expect_err("the join is decided on the archived channel"); + assert!( + matches!(&err, DbError::AccessDenied(reason) if reason == "channel is archived"), + "{err:?}" + ); + let readd = put_member(&pool, community, channel.id, &removed, None, Some(&owner)) + .await + .expect_err("an archived channel does not reactivate a removed member"); + assert!(matches!(readd, DbError::AccessDenied(_)), "{readd:?}"); + let mut roles = Vec::new(); + for pk in [&owner, &joiner, &removed] { + roles.push( + get_member_role(&pool, community, channel.id, pk) + .await + .expect("read role"), + ); + } + assert_eq!(roles, [Some("owner".to_string()), None, None]); + + // The other order: an archive waits for an admission that holds the + // locks, and the member it admitted stays a member. + crate::channel::unarchive_channel(&pool, community, channel.id) + .await + .expect("unarchive"); + let mut admission = crate::begin_community_write_transaction( + &pool, + community, + crate::observability::WriterOperation::EventWrite, + ) + .await + .expect("begin admission"); + lock_channel_for_member_put(&mut admission, channel.id) + .await + .expect("lock for the admission"); + let targets = [MemberPut { + pubkey: &joiner, + requested: None, + }]; + let admitted = put_members_on(&mut admission, channel.id, &targets, None) + .await + .expect("admission"); + assert!(admitted[0].is_ok(), "{admitted:?}"); + let archive_pool = crate::test_support::named_pool("archive-vs-admission").await; + let archive = tokio::spawn({ + let channel_id = channel.id; + async move { crate::channel::archive_channel(&archive_pool, community, channel_id).await } + }); + crate::test_support::wait_for_lock_wait(&pool, "archive-vs-admission").await; + assert!( + !archive.is_finished(), + "an archive must wait for an admission in progress" + ); + admission.commit().await.expect("commit admission"); + archive.await.expect("archive task").expect("archive"); + assert_eq!( + get_member_role(&pool, community, channel.id, &joiner) + .await + .expect("read role"), + Some("member".to_string()) + ); + } + /// A `join_role` change on an ephemeral channel and a roster publication /// take the channel's locks in one order. The roster holds the membership /// lock and refreshes the channel's TTL (TTL lock, then row) at commit. A diff --git a/crates/buzz-relay/src/api/invites.rs b/crates/buzz-relay/src/api/invites.rs index dc3e48b3b6..7d8cc7648d 100644 --- a/crates/buzz-relay/src/api/invites.rs +++ b/crates/buzz-relay/src/api/invites.rs @@ -1460,6 +1460,131 @@ mod postgres_tests { ); } + /// A default channel archived after the auto-join listed it is skipped: + /// no membership row, no kind:39002 roster entry, no kind:44100 and no + /// `member_joined` notice. The test holds an uncommitted archive of the + /// listed channel, so the auto-join passes the list and the shared join + /// body's open/unarchived check on the committed state and then waits in + /// `put_member`; the archive commits and the join resumes on an archived + /// channel. Mutation: drop the channel row lock or the archived refusal + /// from `put_members` → RED. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn auto_join_skips_a_default_channel_archived_while_it_joins() { + use crate::handlers::group_state_postgres_tests::{ + create_channel, edit_channel, group_state_tags, + }; + use crate::handlers::side_effects::auto_join_default_channels; + + let host = format!( + "invites-auto-join-archive-{}.example", + Uuid::new_v4().simple() + ); + let owner = Keys::generate(); + let state = invite_test_state(&host) + .await + .expect("requires reachable Postgres and relay test state"); + let community = state + .db + .lookup_community_by_host(&host) + .await + .expect("lookup") + .expect("community exists") + .id; + let tenant = buzz_core::TenantContext::resolved(community, &host); + let joiner = Keys::generate(); + let joiner_hex = joiner.public_key().to_hex(); + for (keys, role) in [(&owner, "owner"), (&joiner, "member")] { + state + .db + .add_relay_member( + community, + &keys.public_key().to_hex(), + role, + None, + &Keys::generate(), + ) + .await + .map(|write| write.outcome) + .expect("seed relay member"); + } + let starter = create_channel(&state, &tenant, &owner, &[]).await; + edit_channel(&state, &tenant, &owner, starter, &[&["auto_join", "true"]]) + .await + .expect("relay owner flags channel"); + + let pool = state.db.pool().clone(); + let mut archive = pool.begin().await.expect("begin archive"); + let archive_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *archive) + .await + .expect("archive backend pid"); + sqlx::query("UPDATE channels SET archived_at = NOW() WHERE community_id = $1 AND id = $2") + .bind(community.as_uuid()) + .bind(starter) + .execute(&mut *archive) + .await + .expect("archive the default channel"); + + let auto_join = tokio::spawn({ + let state = Arc::clone(&state); + let tenant = tenant.clone(); + let joiner_hex = joiner_hex.clone(); + async move { auto_join_default_channels(&tenant, &state, &joiner_hex).await } + }); + // Wait until the join is blocked behind the archive, so it has + // already listed the channel and passed the join body's check. + let deadline = std::time::Instant::now() + Duration::from_secs(10); + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_stat_activity \ + WHERE $1 = ANY(pg_blocking_pids(pid)))", + ) + .bind(archive_pid) + .fetch_one(&pool) + .await + .expect("read lock waiters"); + if waiting { + break; + } + assert!( + !auto_join.is_finished(), + "the auto-join finished without waiting for the archive" + ); + assert!( + std::time::Instant::now() < deadline, + "the auto-join never waited for the archive" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } + archive.commit().await.expect("commit archive"); + tokio::time::timeout(Duration::from_secs(10), auto_join) + .await + .expect("the auto-join returns after the archive commits") + .expect("auto-join task must not panic"); + + assert_eq!( + channel_roles(&state, community, &[starter], &joiner).await, + [None], + "an archived default channel must not admit the new member" + ); + assert!( + !group_state_tags(&state, &tenant, starter, 39002) + .await + .iter() + .any(|tag| tag[0] == "p" && tag[1] == joiner_hex), + "kind:39002 must not list the new member" + ); + assert_eq!( + added_notification_count(&state, community, &joiner_hex).await, + 0 + ); + assert_eq!( + member_joined_notice_count(&state, community, starter, &joiner_hex).await, + 0 + ); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn owner_mints_and_new_pubkey_claims() { diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 3e078c8e01..b3ccabc7e4 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -2731,7 +2731,10 @@ pub(crate) async fn join_channel_as_self( // A self-join gets the channel's `join_role`, read by `put_member` under // the membership lock: member in a normal channel, guest in an announce - // channel, where a member must then make them a member. + // channel, where a member must then make them a member. `put_member` + // also decides visibility and archive state again under its locks, so a + // channel made private or archived since the checks above is refused + // here, with no side effects. state .db .put_member(tenant.community(), channel_id, pubkey, None, None) @@ -2779,7 +2782,8 @@ pub(crate) async fn join_channel_as_self( /// Best effort: the relay membership has already committed, so a channel /// that fails is logged and counted, never retried, and never fails the /// caller. The `auto_join` tag on kind:39000 is the durable record a client -/// reconciles against. +/// reconciles against. A channel archived or made private after the list +/// was read is one such failure: `put_member` refuses it under its locks. pub(crate) async fn auto_join_default_channels( tenant: &TenantContext, state: &Arc,