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,