diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index ef2259dde37..ba864271adb 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -898,7 +898,11 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 83); + assert_eq!(migrations.len(), 84); + assert!(migrations[83] + .sql + .as_str() + .contains("ADD COLUMN auto_join_pending UUID")); for (index, migration) in migrations.iter().enumerate() { assert_eq!(migration.version, (index + 1) as i64); } @@ -3026,6 +3030,7 @@ mod postgres_tests { .expect("apply migrations for schema parity"); for table in [ + "relay_members", "relay_admin_actions", "relay_admin_outbox", "relay_operator_audit", diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 204efaa2ef1..002d1eecd24 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -2090,6 +2090,65 @@ async fn is_relay_member_is_bounded_routed_and_fails_closed() { ); } +/// The per-AUTH pending default-channel read is bounded-routed like +/// membership; the roster writers' read stays on the writer. A different +/// token on each pool proves which pool answered. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn pending_default_channel_join_routed_reads_the_bounded_replica() { + let (writer, _writer_db) = ScratchDb::migrated(&admin_url().await, "pend_w").await; + let (replica, _replica_db) = ScratchDb::migrated(&admin_url().await, "pend_r").await; + + let community = Uuid::new_v4(); + let pubkey = "cc".repeat(32); + let (writer_token, replica_token) = (Uuid::new_v4(), Uuid::new_v4()); + for (pool, token) in [(&writer, writer_token), (&replica, replica_token)] { + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community) + .bind(format!("pending-routing-{}.example", community.simple())) + .execute(pool) + .await + .expect("insert community"); + sqlx::query( + "INSERT INTO relay_members (community_id, pubkey, role, auto_join_pending) \ + VALUES ($1, $2, 'member', $3)", + ) + .bind(community) + .bind(&pubkey) + .bind(token) + .execute(pool) + .await + .expect("seed pending work"); + } + let cid = CommunityId::from_uuid(community); + let token = |pending: Option| { + pending.expect("pending work").token + }; + + let mut db = Db::from_pools(writer.clone(), replica.clone()); + db.fence().force_open_for_tests(chrono::Utc::now()); + + // Budget unset ⇒ writer. + let pending = db.pending_default_channel_join_routed(cid, &pubkey).await; + assert_eq!(token(pending.expect("gate off")), writer_token); + + // Budget set + fresh entry ⇒ replica; the writer-only read ignores it. + db.set_replica_read_max_age_for_tests(Some(std::time::Duration::from_secs(5))); + let pending = db.pending_default_channel_join_routed(cid, &pubkey).await; + assert_eq!(token(pending.expect("gate on")), replica_token); + let pending = db.pending_default_channel_join(cid, &pubkey).await; + assert_eq!(token(pending.expect("writer read")), writer_token); + + // Entry older than the budget ⇒ fail closed to the writer. + db.fence().close(); + db.fence().force_open_for_tests_at( + chrono::Utc::now(), + std::time::Instant::now() - std::time::Duration::from_secs(10), + ); + let pending = db.pending_default_channel_join_routed(cid, &pubkey).await; + assert_eq!(token(pending.expect("entry too old")), writer_token); +} + /// Community separation across every routed seam, verified on /// REPLICA-SERVED reads. /// diff --git a/crates/buzz-db/src/runtime/tests/listed_channel_name_postgres_tests.rs b/crates/buzz-db/src/runtime/tests/listed_channel_name_postgres_tests.rs index 8ebd4a49680..a07a3858ef6 100644 --- a/crates/buzz-db/src/runtime/tests/listed_channel_name_postgres_tests.rs +++ b/crates/buzz-db/src/runtime/tests/listed_channel_name_postgres_tests.rs @@ -505,6 +505,16 @@ async fn ledger_version(pool: &PgPool) -> i64 { .expect("read migration ledger") } +async fn listed_name_migration_recorded(pool: &PgPool) -> bool { + sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM _sqlx_migrations WHERE version = $1 AND success)", + ) + .bind(BEFORE_INDEX + 1) + .fetch_one(pool) + .await + .expect("read listed-name migration ledger entry") +} + async fn index_oid(pool: &PgPool) -> Option { sqlx::query_scalar("SELECT to_regclass('public.idx_channels_listed_name')::oid::bigint") .fetch_one(pool) @@ -556,7 +566,7 @@ async fn migration_schema_listed_name_index_needs_the_renames_then_builds_inline migration::run_migrations(&pool) .await .expect("migrate a populated table after the renames"); - assert_eq!(ledger_version(&pool).await, BEFORE_INDEX + 1); + assert!(listed_name_migration_recorded(&pool).await); let mut expected = vec![ (kept, "square-ios".to_owned()), renamed(newer, "Square-iOS"), @@ -633,6 +643,6 @@ async fn migration_schema_listed_name_index_validates_a_prebuild() { .await .expect("accept the prebuild without waiting for channel writers"); writer.rollback().await.unwrap(); - assert_eq!(ledger_version(&pool).await, BEFORE_INDEX + 1); + assert!(listed_name_migration_recorded(&pool).await); assert_eq!(index_oid(&pool).await, oid, "the prebuild is not replaced"); } diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index cf9c05cb2c5..4d82367c984 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -737,6 +737,16 @@ pub async fn put_first_member( community_id: CommunityId, channel_id: Uuid, pubkey: &[u8], +) -> Result> { + put_first_member_with_pending(pool, community_id, channel_id, pubkey, None).await +} + +async fn put_first_member_with_pending( + pool: &PgPool, + community_id: CommunityId, + channel_id: Uuid, + pubkey: &[u8], + pending: Option, ) -> Result> { let mut tx = crate::begin_community_write_transaction( pool, @@ -744,19 +754,44 @@ pub async fn put_first_member( crate::observability::WriterOperation::EventWrite, ) .await?; + let admitted_at = if let Some(pending) = pending { + // Fence removal/re-admission and completion against an older worker. + // Keep this lock through the channel write, on the same connection. + let current = sqlx::query_scalar::<_, DateTime>( + "SELECT created_at FROM relay_members \ + WHERE community_id = $1 AND pubkey = $2 AND auto_join_pending = $3 FOR SHARE", + ) + .bind(community_id.as_uuid()) + .bind(hex::encode(pubkey)) + .bind(pending) + .fetch_optional(tx.conn()) + .await?; + let Some(created_at) = current else { + tx.rollback().await?; + return Ok(None); + }; + Some(created_at) + } else { + None + }; lock_channel_for_member_put(&mut tx, channel_id).await?; - let seen: bool = sqlx::query_scalar( - "SELECT EXISTS (SELECT 1 FROM channel_members \ - WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3)", + let history: Option<(Option>,)> = sqlx::query_as( + "SELECT removed_at FROM channel_members \ + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3", ) .bind(community_id.as_uuid()) .bind(channel_id) .bind(pubkey) - .fetch_one(tx.conn()) + .fetch_optional(tx.conn()) .await?; - if seen { - tx.rollback().await?; - return Ok(None); + if let Some((removed_at,)) = history { + // Ordinary first joins preserve all history. Recovery preserves a + // leave from this admission, but fresh roster admissions retain the + // original behavior of restoring earlier removed memberships. + if admitted_at.is_none() || removed_at.is_none() || removed_at >= admitted_at { + tx.rollback().await?; + return Ok(None); + } } let target = MemberPut { pubkey, @@ -773,7 +808,11 @@ pub async fn put_first_member( } Err(refused) => { tx.rollback().await?; - Err(refused) + if pending.is_some() && matches!(refused, DbError::AccessDenied(_)) { + Ok(None) + } else { + Err(refused) + } } } } @@ -2260,6 +2299,19 @@ impl Db { .await } + /// Complete one pending default-channel join, preserving leaves since admission. + /// A stale token or a removed relay member writes nothing. + #[datastore_span(name = "put_pending_default_member", system = "postgresql")] + pub async fn put_pending_default_member( + &self, + community: CommunityId, + channel: Uuid, + pubkey: &[u8], + pending: Uuid, + ) -> Result> { + put_first_member_with_pending(&self.pool, community, channel, pubkey, Some(pending)).await + } + /// Self-joins `pubkey` with the channel's `join_role` only if they have /// never been a member of the channel (see [`put_first_member`]). #[datastore_span(name = "put_first_member", system = "postgresql")] diff --git a/crates/buzz-db/src/store/relay_invite.rs b/crates/buzz-db/src/store/relay_invite.rs index 36576ae9ef7..c6db801d279 100644 --- a/crates/buzz-db/src/store/relay_invite.rs +++ b/crates/buzz-db/src/store/relay_invite.rs @@ -611,8 +611,8 @@ pub async fn claim_relay_invite_with_owner( } let inserted = sqlx::query( - "INSERT INTO relay_members (community_id, pubkey, role, added_by) \ - VALUES ($1, $2, 'member', 'invite') \ + "INSERT INTO relay_members (community_id, pubkey, role, added_by, auto_join_pending) \ + VALUES ($1, $2, 'member', 'invite', gen_random_uuid()) \ ON CONFLICT (community_id, pubkey) DO NOTHING", ) .bind(community.as_uuid()) diff --git a/crates/buzz-db/src/store/relay_invite/claim_owner.rs b/crates/buzz-db/src/store/relay_invite/claim_owner.rs index 7e275b1263d..a3106fc6bad 100644 --- a/crates/buzz-db/src/store/relay_invite/claim_owner.rs +++ b/crates/buzz-db/src/store/relay_invite/claim_owner.rs @@ -116,8 +116,8 @@ pub async fn claim_relay_membership_with_owner( return Err(DbError::InviteRestricted); } let inserted = sqlx::query( - "INSERT INTO relay_members (community_id, pubkey, role, added_by) \ - VALUES ($1, $2, $3, 'invite') \ + "INSERT INTO relay_members (community_id, pubkey, role, added_by, auto_join_pending) \ + VALUES ($1, $2, $3, 'invite', gen_random_uuid()) \ ON CONFLICT (community_id, pubkey) DO NOTHING", ) .bind(community.as_uuid()) diff --git a/crates/buzz-db/src/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs index 812cfc4987e..ac7f50491d4 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -150,6 +150,19 @@ pub async fn add_relay_member( role: &str, added_by: Option<&str>, relay_keypair: &nostr::Keys, +) -> Result> { + add_relay_member_with_pending(pool, community, pubkey, role, added_by, None, relay_keypair) + .await +} + +async fn add_relay_member_with_pending( + pool: &PgPool, + community: CommunityId, + pubkey: &str, + role: &str, + added_by: Option<&str>, + pending: Option, + relay_keypair: &nostr::Keys, ) -> Result> { let mut tx = crate::begin_community_write_transaction( pool, @@ -158,13 +171,14 @@ pub async fn add_relay_member( ) .await?; let result = sqlx::query( - "INSERT INTO relay_members (community_id, pubkey, role, added_by) \ - VALUES ($1, $2, $3, $4) ON CONFLICT (community_id, pubkey) DO NOTHING", + "INSERT INTO relay_members (community_id, pubkey, role, added_by, auto_join_pending) \ + VALUES ($1, $2, $3, $4, $5) ON CONFLICT (community_id, pubkey) DO NOTHING", ) .bind(community.as_uuid()) .bind(pubkey) .bind(role) .bind(added_by) + .bind(pending) .execute(tx.conn()) .await?; let added = result.rows_affected() > 0; @@ -1189,7 +1203,121 @@ pub async fn rewrite_nip43_membership_list( commit_membership_write(tx, (), true, relay_keypair).await } +/// Durable default-channel work for one roster admission. +#[derive(Debug, sqlx::FromRow)] +pub struct PendingDefaultChannelJoin { + /// Generation checked before writing or completing this work. + pub token: Uuid, + /// Original actor; `invite` denotes a self-claim. + pub added_by: Option, +} + +async fn pending_default_channel_join( + pool: &PgPool, + community: CommunityId, + pubkey: &str, +) -> Result> { + let mut conn = + observability::acquire_writer(pool, observability::WriterOperation::Authorization).await?; + pending_default_channel_join_on(&mut conn, community, pubkey).await +} + +async fn pending_default_channel_join_on( + conn: &mut sqlx::PgConnection, + community: CommunityId, + pubkey: &str, +) -> Result> { + Ok(sqlx::query_as( + "SELECT auto_join_pending AS token, added_by FROM relay_members \ + WHERE community_id = $1 AND pubkey = $2 AND auto_join_pending IS NOT NULL", + ) + .bind(community.as_uuid()) + .bind(pubkey) + .fetch_optional(conn) + .await?) +} + impl Db { + /// Read unfinished default-channel work and its original adding actor. + #[datastore_span(name = "pending_default_channel_join", system = "postgresql")] + pub async fn pending_default_channel_join( + &self, + community: CommunityId, + pubkey: &str, + ) -> Result> { + pending_default_channel_join(&self.pool, community, pubkey).await + } + + /// [`Db::pending_default_channel_join`] on the bounded replica arm, for + /// the retry that runs on every AUTH. A stale answer is safe: it can only + /// defer the work to a later admission or carry an old token, and every + /// write re-checks the token on the writer. Falls back to the writer as + /// [`Db::is_relay_member`] does. + #[datastore_span(name = "pending_default_channel_join_routed", system = "postgresql")] + pub async fn pending_default_channel_join_routed( + &self, + community: CommunityId, + pubkey: &str, + ) -> Result> { + let path = "pending_default_channel_join"; + match self + .route_read( + path, + RoutePredicate::Bounded, + crate::observability::ReaderOperation::Authorization, + ) + .await? + { + RouteDecision::Replica(mut tx, _entry, reason) => { + match pending_default_channel_join_on(&mut tx, community, pubkey).await { + Ok(pending) => { + Self::record_route(path, "replica", reason); + Ok(pending) + } + Err(e) if e.is_statement_cancelled() => { + Self::record_route(path, "replica", "statement_cancelled"); + Err(e) + } + Err(e) => { + tracing::warn!(path, "replica read failed; re-running on writer: {e}"); + Self::record_route(path, "writer", "replica_error"); + pending_default_channel_join(&self.pool, community, pubkey).await + } + } + } + RouteDecision::Writer => { + pending_default_channel_join(&self.pool, community, pubkey).await + } + } + } + + /// Finish only the admission generation that was actually processed. + #[datastore_span(name = "finish_default_channel_join", system = "postgresql")] + pub async fn finish_default_channel_join( + &self, + community: CommunityId, + pubkey: &str, + pending: Uuid, + ) -> Result<()> { + let mut tx = crate::begin_community_write_transaction( + &self.pool, + community, + observability::WriterOperation::Authorization, + ) + .await?; + sqlx::query( + "UPDATE relay_members SET auto_join_pending = NULL \ + WHERE community_id = $1 AND pubkey = $2 AND auto_join_pending = $3", + ) + .bind(community.as_uuid()) + .bind(pubkey) + .bind(pending) + .execute(tx.conn()) + .await?; + tx.commit().await?; + Ok(()) + } + /// Returns `true` if `pubkey` (64-char hex) is a member of `community`. /// /// Replica-routed on the bounded arm — the one PERMISSION read routed by @@ -1276,6 +1404,29 @@ impl Db { add_relay_member(&self.pool, community, pubkey, role, added_by, relay_keypair).await } + /// Adds a roster member and records default-channel work atomically. + /// Unlike the roster-only `add_relay_member`, this opts into auto-join. + #[datastore_span(name = "add_relay_member_with_default_channels", system = "postgresql")] + pub async fn add_relay_member_with_default_channels( + &self, + community: CommunityId, + pubkey: &str, + role: &str, + added_by: &str, + relay_keypair: &nostr::Keys, + ) -> Result> { + add_relay_member_with_pending( + &self.pool, + community, + pubkey, + role, + Some(added_by), + Some(Uuid::new_v4()), + relay_keypair, + ) + .await + } + /// Claims relay membership via an invite and atomically persists the /// accepted policy version when a policy is configured. #[datastore_span(name = "claim_relay_membership", system = "postgresql")] diff --git a/crates/buzz-relay/src/api/invites.rs b/crates/buzz-relay/src/api/invites.rs index 94650c7dbd1..cf783c09cfc 100644 --- a/crates/buzz-relay/src/api/invites.rs +++ b/crates/buzz-relay/src/api/invites.rs @@ -532,13 +532,19 @@ async fn claim_invite_authed( if write.outcome.owner_link_created { state.disconnect_unowned_agent_clusterwide(&tenant, pubkey.as_bytes()); } - // The member-added delta and the auto-join only on Joined, never on - // other outcomes; the membership-list fan-out on every outcome. + // Publish the member-added delta only on Joined; successful repeated + // claims may still have durable default-channel work to finish. let joined = matches!( write.outcome.outcome, buzz_db::relay_invite::ClaimOutcome::Joined { .. } ); - finish_committed_claim(&tenant, state, &claimer_hex, write.snapshot, joined).await; + if matches!( + write.outcome.outcome, + buzz_db::relay_invite::ClaimOutcome::Joined { .. } + | buzz_db::relay_invite::ClaimOutcome::AlreadyMember { .. } + ) { + finish_committed_claim(&tenant, state, &claimer_hex, write.snapshot, joined).await; + } return match write.outcome.outcome { buzz_db::relay_invite::ClaimOutcome::Joined { .. } => Ok(Json(serde_json::json!({ "status": "joined", @@ -633,14 +639,13 @@ async fn claim_invite_authed( } /// The follow-up of a claim whose roster write committed: the NIP-43 -/// membership-list fan-out and, when the claim added the member (`joined`), -/// the member-added delta and the default-channel auto-join. +/// membership-list fan-out, the member-added delta for new members, and +/// any unfinished default-channel joins (including on a repeated claim). /// /// It runs on its own task, which the handler awaits, so the response still /// waits for the joins. A client that disconnects first makes the server /// drop the handler's future: that detaches the task rather than cancelling -/// it partway, which nothing would retry, because a re-claim is -/// `already_member`. +/// it partway. The durable token also survives a relay restart. async fn finish_committed_claim( tenant: &buzz_core::TenantContext, state: &Arc, @@ -653,13 +658,11 @@ async fn finish_committed_claim( (tenant.clone(), Arc::clone(state), claimer_hex.to_owned()); async move { fan_out_nip43_membership_list(&tenant, &state, snapshot).await; - if !joined { - return; + if joined { + if let Err(e) = publish_nip43_member_added(&tenant, &state, &claimer_hex).await { + tracing::warn!("failed to publish NIP-43 member-added delta after claim: {e}"); + } } - if let Err(e) = publish_nip43_member_added(&tenant, &state, &claimer_hex).await { - tracing::warn!("failed to publish NIP-43 member-added delta after claim: {e}"); - } - // A claimer joins for themselves: no third-party add policy. auto_join_default_channels(&tenant, &state, &claimer_hex, None).await; } }); @@ -697,6 +700,7 @@ fn claim_key_rate_limited( #[cfg(test)] mod postgres_tests { mod owner_admission_tests; + mod recovery_postgres_tests; mod restriction_tests; use std::sync::Arc; use std::time::Duration; @@ -1560,10 +1564,9 @@ 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 + /// listed channel, so the auto-join reads the committed channel state and + /// then waits for the membership write's channel row lock. 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"] @@ -1591,20 +1594,28 @@ mod postgres_tests { 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 _ = state + .db + .add_relay_member( + community, + &owner.public_key().to_hex(), + "owner", + None, + &Keys::generate(), + ) + .await + .expect("seed relay owner"); + let _ = state + .db + .add_relay_member_with_default_channels( + community, + &joiner_hex, + "member", + &joiner_hex, + &Keys::generate(), + ) + .await + .expect("seed pending default-channel admission"); let starter = create_channel(&state, &tenant, &owner, &[]).await; edit_channel(&state, &tenant, &owner, starter, &[&["auto_join", "true"]]) .await @@ -1630,8 +1641,8 @@ mod postgres_tests { async move { auto_join_default_channels(&tenant, &state, &joiner_hex, None).await } }); // Wait until the join is blocked on the channel row lock in - // `put_member`, behind the archive, so it has already listed the - // channel and passed the join body's check. + // the membership write, behind the archive, so it has already listed + // the channel. wait_for_put_member_behind(&pool, archive_pid, &auto_join).await; archive.commit().await.expect("commit archive"); tokio::time::timeout(Duration::from_secs(10), auto_join) diff --git a/crates/buzz-relay/src/api/invites/postgres_tests/recovery_postgres_tests.rs b/crates/buzz-relay/src/api/invites/postgres_tests/recovery_postgres_tests.rs new file mode 100644 index 00000000000..8481240ea71 --- /dev/null +++ b/crates/buzz-relay/src/api/invites/postgres_tests/recovery_postgres_tests.rs @@ -0,0 +1,480 @@ +mod postgres_tests { + use super::super::*; + use crate::handlers::group_state_postgres_tests::{create_channel, edit_channel, http_auth}; + use crate::handlers::ingest::ingest_event; + use crate::handlers::side_effects::auto_join_default_channels; + + // Cancel the real listing query only after PostgreSQL reports that it is + // blocked by our table lock. No timer or production-only failure hook. + async fn cancel_blocked_query(pool: &sqlx::PgPool, pattern: &str) { + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let cancelled: Vec = sqlx::query_scalar( + "SELECT pg_cancel_backend(pid) FROM pg_stat_activity \ + WHERE datname = current_database() AND wait_event_type = 'Lock' \ + AND query LIKE $1", + ) + .bind(pattern) + .fetch_all(pool) + .await + .expect("cancel blocked default-channel listing"); + if !cancelled.is_empty() { + assert!(cancelled.into_iter().all(|cancelled| cancelled)); + return; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("the admission must reach the actual default-channel listing"); + } + + async fn recovery_fixture( + label: &str, + channel_count: usize, + ) -> ( + Arc, + buzz_core::TenantContext, + Keys, + Keys, + Vec, + ) { + let host = format!("join-{label}-{}.example", Uuid::new_v4().simple()); + let state = invite_test_state(&host).await.expect("isolated Postgres"); + let community = state + .db + .lookup_community_by_host(&host) + .await + .unwrap() + .unwrap() + .id; + let tenant = buzz_core::TenantContext::resolved(community, &host); + let (owner, member) = (Keys::generate(), Keys::generate()); + let _ = state + .db + .add_relay_member( + community, + &owner.public_key().to_hex(), + "owner", + None, + &state.relay_keypair, + ) + .await + .unwrap(); + let mut channels = Vec::new(); + for _ in 0..channel_count { + let channel = create_channel(&state, &tenant, &owner, &[]).await; + edit_channel(&state, &tenant, &owner, channel, &[&["auto_join", "true"]]) + .await + .unwrap(); + channels.push(channel); + } + assert_eq!( + state + .db + .list_auto_join_channels(community) + .await + .unwrap() + .iter() + .map(|channel| channel.id) + .collect::>(), + channels, + "the failing channel must precede the healthy one", + ); + let _ = state + .db + .add_relay_member_with_default_channels( + community, + &member.public_key().to_hex(), + "member", + &member.public_key().to_hex(), + &state.relay_keypair, + ) + .await + .unwrap(); + (state, tenant, owner, member, channels) + } + + // Target only the connection blocked by this test's transaction, even + // when another recovery test is running against the same database. + async fn cancel_query_blocked_by(pool: &sqlx::PgPool, blocker: i32) { + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let cancelled: Vec = sqlx::query_scalar( + "SELECT pg_cancel_backend(pid) FROM pg_stat_activity \ + WHERE datname = current_database() AND wait_event_type = 'Lock' \ + AND $1 = ANY(pg_blocking_pids(pid))", + ) + .bind(blocker) + .fetch_all(pool) + .await + .unwrap(); + if !cancelled.is_empty() { + assert_eq!(cancelled, [true], "exactly our recovery query was blocked"); + return; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("recovery must reach the held PostgreSQL lock"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn channel_local_failures_continue_later_defaults_and_retain_retry_work() { + let (state, tenant, _, member, channels) = recovery_fixture("membership", 2).await; + let community = tenant.community(); + let member_hex = member.public_key().to_hex(); + let pending = state + .db + .pending_default_channel_join(community, &member_hex) + .await + .unwrap() + .unwrap(); + let mut blocker = state.db.pool().begin().await.unwrap(); + let blocker_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *blocker) + .await + .unwrap(); + sqlx::query("SELECT id FROM channels WHERE community_id = $1 AND id = $2 FOR UPDATE") + .bind(community.as_uuid()) + .bind(channels[0]) + .fetch_one(&mut *blocker) + .await + .unwrap(); + let retry = tokio::spawn({ + let (state, tenant, member_hex) = (state.clone(), tenant.clone(), member_hex.clone()); + async move { auto_join_default_channels(&tenant, &state, &member_hex, None).await } + }); + cancel_query_blocked_by(state.db.pool(), blocker_pid).await; + // Only the first channel fails. Later memberships can write. + blocker.rollback().await.unwrap(); + tokio::time::timeout(Duration::from_secs(10), retry) + .await + .unwrap() + .unwrap(); + assert_eq!( + channel_roles(&state, community, &channels, &member).await, + [None, Some("member".into())], + "a local failure must not strand the later channel", + ); + assert_eq!( + state + .db + .pending_default_channel_join(community, &member_hex) + .await + .unwrap() + .unwrap() + .token, + pending.token + ); + auto_join_default_channels(&tenant, &state, &member_hex, None).await; + assert_eq!( + channel_roles(&state, community, &channels, &member).await, + [Some("member".into()), Some("member".into())] + ); + assert!(state + .db + .pending_default_channel_join(community, &member_hex) + .await + .unwrap() + .is_none()); + assert_eq!( + added_notification_count(&state, community, &member_hex).await, + 2 + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn temporary_listing_failure_is_recovered_for_each_admission_path() { + let host = format!("join-recovery-{}.example", Uuid::new_v4().simple()); + let state = invite_test_state(&host).await.expect("test state"); + let community = state + .db + .lookup_community_by_host(&host) + .await + .unwrap() + .unwrap() + .id; + let tenant = buzz_core::TenantContext::resolved(community, &host); + let owner = Keys::generate(); + let _ = state + .db + .add_relay_member( + community, + &owner.public_key().to_hex(), + "owner", + None, + &state.relay_keypair, + ) + .await + .unwrap(); + let channel = create_channel(&state, &tenant, &owner, &[]).await; + edit_channel(&state, &tenant, &owner, channel, &[&["auto_join", "true"]]) + .await + .unwrap(); + + for path in ["v1", "v2", "admin"] { + let member = Keys::generate(); + let member_hex = member.public_key().to_hex(); + let body = match path { + "v1" => serde_json::json!({"code": forge_v1_invite_code(&state, community, unix_now() + 3600)}).to_string(), + "v2" => serde_json::json!({"code": mint_code(state.clone(), &host, &owner, + serde_json::json!({"max_uses": 1})).await}).to_string(), + _ => String::new(), + }; + let add = EventBuilder::new(Kind::Custom(9030), "") + .tags([Tag::public_key(member.public_key())]) + .sign_with_keys(&owner) + .unwrap(); + let admit = || { + let (state, host, member, body, tenant, add, owner) = ( + state.clone(), + host.clone(), + member.clone(), + body.clone(), + tenant.clone(), + add.clone(), + owner.clone(), + ); + async move { + if path == "admin" { + let mut auth = http_auth(&owner); + if let crate::handlers::ingest::IngestAuth::Http { scopes, .. } = &mut auth + { + scopes.push(buzz_auth::Scope::AdminUsers); + } + let result = ingest_event(&state, &tenant, add, auth).await.unwrap(); + assert!(result.accepted, "{:?}", result.message); + } else { + let response = + post_json(state, &host, "/api/invites/claim", &member, body).await; + assert_eq!(response.status(), StatusCode::OK); + } + } + }; + let mut blocker = state.db.pool().begin().await.unwrap(); + sqlx::query("LOCK TABLE channels IN ACCESS EXCLUSIVE MODE") + .execute(&mut *blocker) + .await + .unwrap(); + let request = tokio::spawn(admit()); + cancel_blocked_query( + state.db.pool(), + "%FROM channels%WHERE community_id = $1 AND auto_join%", + ) + .await; + tokio::time::timeout(Duration::from_secs(10), request) + .await + .unwrap() + .unwrap(); + blocker.rollback().await.unwrap(); + assert!(state + .db + .is_relay_member(community, &member_hex) + .await + .unwrap()); + let pending = state + .db + .pending_default_channel_join(community, &member_hex) + .await + .unwrap(); + assert!( + pending.is_some(), + "{path}: committed admission lost its retry record" + ); + assert_eq!( + channel_roles(&state, community, &[channel], &member).await, + [None] + ); + + // A retry is already_member (and a single-use v2 invite is exhausted + // for new claimants), so recovery cannot depend on a fresh insertion. + admit().await; + assert_eq!( + channel_roles(&state, community, &[channel], &member).await, + [Some("member".to_string())], + "{path}: retry must finish the join" + ); + assert!(state + .db + .pending_default_channel_join(community, &member_hex) + .await + .unwrap() + .is_none()); + assert_eq!( + added_notification_count(&state, community, &member_hex).await, + 1 + ); + let leave = EventBuilder::new(Kind::Custom(9022), "") + .tags([Tag::parse(["h", &channel.to_string()]).unwrap()]) + .sign_with_keys(&member) + .unwrap(); + assert!( + ingest_event(&state, &tenant, leave, http_auth(&member)) + .await + .unwrap() + .accepted + ); + admit().await; + assert_eq!( + channel_roles(&state, community, &[channel], &member).await, + [None] + ); + } + } + + /// Recovery preserves a leave made during partial completion, rejects stale + /// generations after removal/re-addition, and still enforces the original actor. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn pending_join_preserves_leaves_and_fences_old_admissions() { + let host = format!("join-fence-{}.example", Uuid::new_v4().simple()); + let state = invite_test_state(&host).await.unwrap(); + let community = state + .db + .lookup_community_by_host(&host) + .await + .unwrap() + .unwrap() + .id; + let tenant = buzz_core::TenantContext::resolved(community, &host); + let (owner, member) = (Keys::generate(), Keys::generate()); + let member_hex = member.public_key().to_hex(); + for (keys, role) in [(&owner, "owner"), (&member, "member")] { + let _ = state + .db + .add_relay_member_with_default_channels( + community, + &keys.public_key().to_hex(), + role, + &owner.public_key().to_hex(), + &state.relay_keypair, + ) + .await + .unwrap(); + } + let a = create_channel(&state, &tenant, &owner, &[]).await; + let b = create_channel(&state, &tenant, &owner, &[]).await; + for channel in [a, b] { + edit_channel(&state, &tenant, &owner, channel, &[&["auto_join", "true"]]) + .await + .unwrap(); + } + let old = state + .db + .pending_default_channel_join(community, &member_hex) + .await + .unwrap() + .unwrap() + .token; + assert!(state + .db + .put_pending_default_member(community, a, &member.public_key().to_bytes(), old) + .await + .unwrap() + .is_some()); + state + .db + .remove_member( + community, + a, + &member.public_key().to_bytes(), + &member.public_key().to_bytes(), + ) + .await + .unwrap(); + auto_join_default_channels(&tenant, &state, &member_hex, None).await; + assert_eq!( + channel_roles(&state, community, &[a, b], &member).await, + [None, Some("member".to_string())] + ); + + let _ = state + .db + .remove_relay_member(community, &member_hex, &state.relay_keypair) + .await + .unwrap(); + let _ = state + .db + .add_relay_member_with_default_channels( + community, + &member_hex, + "member", + &owner.public_key().to_hex(), + &state.relay_keypair, + ) + .await + .unwrap(); + let new = state + .db + .pending_default_channel_join(community, &member_hex) + .await + .unwrap() + .unwrap() + .token; + assert_ne!(old, new); + let c = create_channel(&state, &tenant, &owner, &[]).await; + edit_channel(&state, &tenant, &owner, c, &[&["auto_join", "true"]]) + .await + .unwrap(); + assert!(state + .db + .put_pending_default_member(community, c, &member.public_key().to_bytes(), old) + .await + .unwrap() + .is_none()); + state + .db + .finish_default_channel_join(community, &member_hex, old) + .await + .unwrap(); + assert_eq!( + state + .db + .pending_default_channel_join(community, &member_hex) + .await + .unwrap() + .unwrap() + .token, + new + ); + assert!( + state + .db + .put_pending_default_member(community, a, &member.public_key().to_bytes(), new,) + .await + .unwrap() + .is_some(), + "a fresh roster admission restores a leave from the previous admission" + ); + // If the failed admin-add is retried by the member, it must still ask the + // original admin's permission rather than treating the retry as self-join. + state + .db + .ensure_user(community, &member.public_key().to_bytes()) + .await + .unwrap(); + sqlx::query( + "UPDATE users SET channel_add_policy = 'nobody' WHERE community_id = $1 AND pubkey = $2", + ) + .bind(community.as_uuid()) + .bind(member.public_key().to_bytes().as_slice()) + .execute(state.db.pool()) + .await + .unwrap(); + auto_join_default_channels(&tenant, &state, &member_hex, None).await; + assert_eq!( + channel_roles(&state, community, &[c], &member).await, + [None] + ); + assert!(state + .db + .pending_default_channel_join(community, &member_hex) + .await + .unwrap() + .is_none()); + } +} diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index 1eb45dfb5d4..ee24aca78a8 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/crates/buzz-relay/src/handlers/auth.rs @@ -998,26 +998,32 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: // the ones they have never been in, under this socket's effect // permits. Agents are not. let auto_join = !state.config.require_relay_membership && !admitted_agent; - if auto_join { + { let state = Arc::clone(&state); let tenant = conn.tenant.clone(); let gate = Arc::clone(&conn.nip_fi_gate); tokio::spawn(async move { - crate::handlers::side_effects::auto_join_never_joined_channels( + crate::handlers::side_effects::auto_join_default_channels( &tenant, &state, - &pubkey.to_bytes(), - &gate, + &pubkey.to_hex(), + Some(&gate), ) .await; + if auto_join { + crate::handlers::side_effects::auto_join_never_joined_channels( + &tenant, + &state, + &pubkey.to_bytes(), + &gate, + ) + .await; + } else { + #[cfg(test)] + crate::nip_fi_test_hooks::after_auto_join(tenant.community()).await; + } }); } - // A test waits on every admission's auto-join outcome, so a skip - // reports here as the spawned task does when it finishes. - #[cfg(test)] - if !auto_join { - crate::nip_fi_test_hooks::after_auto_join(conn.tenant.community()).await; - } // _auth_permit drops here — expiry's write guard may proceed. } Err(e) => { @@ -2978,6 +2984,98 @@ mod tests { assert_eq!(role(&person).await, None, "a leave stands"); } + /// Durable roster work is retried by real AUTH admission, including + /// agents. Completed work must not enroll members in defaults added later. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn root_auth_recovers_pending_roster_default_joins() { + let mut state = auth_test_state_real_db_expect().await; + std::sync::Arc::make_mut( + &mut std::sync::Arc::get_mut(&mut state) + .expect("unique test state") + .config, + ) + .require_relay_membership = true; + let community = seeded_community(&state).await; + let (owner, person, agent) = (Keys::generate(), Keys::generate(), Keys::generate()); + let channel = default_channels(&state, community, &owner, &["recovered"]).await[0]; + for keys in [&person, &agent] { + let _ = state + .db + .add_relay_member_with_default_channels( + community, + &keys.public_key().to_hex(), + "member", + "invite", + &state.relay_keypair, + ) + .await + .expect("admit with pending work"); + } + let credential = buzz_sdk::nip_oa::compute_auth_tag(&owner, &agent.public_key(), "") + .expect("credential"); + let tag: Vec = serde_json::from_str(&credential).expect("tag"); + for (keys, tag, challenge) in [ + (&person, None, "recover-person"), + (&agent, Some(tag), "recover-agent"), + ] { + admit_and_settle(&state, community, keys, challenge, tag).await; + assert_eq!( + state + .db + .get_member_role(community, channel, &keys.public_key().to_bytes()) + .await + .expect("membership") + .as_deref(), + Some("member") + ); + assert!(state + .db + .pending_default_channel_join(community, &keys.public_key().to_hex()) + .await + .expect("pending") + .is_none()); + } + // The operator CLI uses roster-only admission. AUTH must not + // turn that into default-channel enrollment. + let roster_only = Keys::generate(); + let _ = state + .db + .add_relay_member( + community, + &roster_only.public_key().to_hex(), + "member", + None, + &state.relay_keypair, + ) + .await + .expect("roster-only add"); + assert!(state + .db + .pending_default_channel_join(community, &roster_only.public_key().to_hex()) + .await + .expect("pending") + .is_none()); + admit_and_settle(&state, community, &roster_only, "roster-only", None).await; + assert!(state + .db + .get_member_role(community, channel, &roster_only.public_key().to_bytes()) + .await + .expect("membership") + .is_none()); + let later = default_channels(&state, community, &owner, &["later"]).await[0]; + admit_and_settle(&state, community, &person, "recover-completed", None).await; + assert!( + state + .db + .get_member_role(community, later, &person.public_key().to_bytes()) + .await + .expect("membership") + .is_none(), + "completed work must not become ongoing auto-enrollment" + ); + } + /// The admission auto-join holds the socket's effect permit for each /// channel, so a session that ends before a channel's write joins /// nothing more: cancelled before the first channel, nothing; between @@ -2990,67 +3088,90 @@ mod tests { #[ignore = "requires Postgres — runs in postgres-ci nextest lane"] async fn admission_auto_join_stops_when_the_session_ends() { use crate::nip_fi_test_hooks::{auto_join_channel_hook, auto_join_done_hook}; - let state = auth_test_state_real_db_expect().await; - let community = seeded_community(&state).await; - let owner = Keys::generate(); - let channels = - default_channels(&state, community, &owner, &["general", "announcements"]).await; - let roles = |keys: &Keys| { - let state = std::sync::Arc::clone(&state); - let pubkey = keys.public_key().to_bytes(); - let channels = channels.clone(); - async move { - let mut roles = Vec::new(); - for channel in channels { - roles.push( - state - .db - .get_member_role(community, channel, &pubkey) - .await - .expect("member role"), - ); - } - roles - } - }; - let member = Some("member".to_string()); - - // `stop_before` channels pass, then the socket is cancelled. - for (stop_before, expected) in [(0, [None, None]), (1, [member.clone(), None])] { - let person = Keys::generate(); - let (done, done_release) = auto_join_done_hook::arm(community); - let (mut reached, mut release) = auto_join_channel_hook::arm(community); - let (conn, _ctrl) = registered_pending_conn(&state, community, "auto-join-stop"); - handle_auth( - signed_auth(&person, "auto-join-stop", None), - std::sync::Arc::clone(&conn), - std::sync::Arc::clone(&state), + for roster in [false, true] { + let mut state = auth_test_state_real_db_expect().await; + std::sync::Arc::make_mut( + &mut std::sync::Arc::get_mut(&mut state) + .expect("unique test state") + .config, ) - .await; - assert!(matches!( - conn.auth_state_snapshot(), - AuthState::Authenticated(_) - )); - for _ in 0..stop_before { - reached.await.expect("task reaches a channel"); - // Re-arm before the task can run again: this task does - // not yield between the release and the arm. + .require_relay_membership = roster; + let community = seeded_community(&state).await; + let owner = Keys::generate(); + let channels = + default_channels(&state, community, &owner, &["general", "announcements"]) + .await; + let roles = |keys: &Keys| { + let state = std::sync::Arc::clone(&state); + let pubkey = keys.public_key().to_bytes(); + let channels = channels.clone(); + async move { + let mut roles = Vec::new(); + for channel in channels { + roles.push( + state + .db + .get_member_role(community, channel, &pubkey) + .await + .expect("member role"), + ); + } + roles + } + }; + let member = Some("member".to_string()); + + // `stop_before` channels pass, then the socket is cancelled. + for (stop_before, expected) in [(0, [None, None]), (1, [member.clone(), None])] { + let person = Keys::generate(); + if roster { + let _ = state + .db + .add_relay_member_with_default_channels( + community, + &person.public_key().to_hex(), + "member", + "invite", + &state.relay_keypair, + ) + .await + .expect("pending roster work"); + } + let (done, done_release) = auto_join_done_hook::arm(community); + let (mut reached, mut release) = auto_join_channel_hook::arm(community); + let (conn, _ctrl) = + registered_pending_conn(&state, community, "auto-join-stop"); + handle_auth( + signed_auth(&person, "auto-join-stop", None), + std::sync::Arc::clone(&conn), + std::sync::Arc::clone(&state), + ) + .await; + assert!(matches!( + conn.auth_state_snapshot(), + AuthState::Authenticated(_) + )); + for _ in 0..stop_before { + reached.await.expect("task reaches a channel"); + // Re-arm before the task can run again: this task does + // not yield between the release and the arm. + release.notify_one(); + (reached, release) = auto_join_channel_hook::arm(community); + } + reached.await.expect("task reaches the stop channel"); + conn.cancel.cancel(); release.notify_one(); - (reached, release) = auto_join_channel_hook::arm(community); + done.await.expect("task finishes"); + done_release.notify_one(); + assert_eq!(roles(&person).await, expected, "stop before {stop_before}"); + + admit_and_settle(&state, community, &person, "auto-join-resume", None).await; + assert_eq!( + roles(&person).await, + [member.clone(), member.clone()], + "the next admission joins what was skipped" + ); } - reached.await.expect("task reaches the stop channel"); - conn.cancel.cancel(); - release.notify_one(); - done.await.expect("task finishes"); - done_release.notify_one(); - assert_eq!(roles(&person).await, expected, "stop before {stop_before}"); - - admit_and_settle(&state, community, &person, "auto-join-resume", None).await; - assert_eq!( - roles(&person).await, - [member.clone(), member.clone()], - "the next admission joins what was skipped" - ); } } diff --git a/crates/buzz-relay/src/handlers/relay_admin.rs b/crates/buzz-relay/src/handlers/relay_admin.rs index 1f51b16f89e..28c1d11ba8b 100644 --- a/crates/buzz-relay/src/handlers/relay_admin.rs +++ b/crates/buzz-relay/src/handlers/relay_admin.rs @@ -341,11 +341,11 @@ async fn execute_relay_admin_command( // to change an existing member's role. let write = state .db - .add_relay_member( + .add_relay_member_with_default_channels( tenant.community(), &target_hex, &role, - Some(&sender_hex), + &sender_hex, &state.relay_keypair, ) .await @@ -366,16 +366,9 @@ async fn execute_relay_admin_command( if let Err(e) = publish_nip43_member_added(tenant, state, &target_hex).await { warn!(error = %e, "failed to publish NIP-43 member added event"); } - // The sender adds someone else, so the target's - // `channel_add_policy` decides whether they are auto-joined. - auto_join_default_channels( - tenant, - state, - &target_hex, - Some(&event.pubkey.to_bytes()), - ) - .await; } + // A no-op re-add also resumes unfinished work from the original actor. + auto_join_default_channels(tenant, state, &target_hex, None).await; fan_out_nip43_membership_list(tenant, state, write.snapshot).await; } diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 23aaaf5e974..0bcc267a3cb 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -3031,79 +3031,131 @@ async fn announce_self_join( Ok(()) } -/// Join a new community member to every effective `auto_join` channel, as -/// kind:9021 self-joins. Runs inline at the roster write sites (invite -/// claim, kind:9030) and nowhere else, so a later kind:9022 leave stands. A -/// community without a roster joins on admission instead -/// ([`auto_join_never_joined_channels`]). -/// -/// 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. A channel archived or made private after the list -/// was read is one such failure: `put_member` refuses it under its locks. -/// -/// `added_by` is the actor of a third-party add (kind:9030), `None` when the -/// member joined by themselves (an invite claim). A third-party add first -/// asks the member's `channel_add_policy`, once, as a kind:9000 from the same -/// actor would; a refusal skips every default channel, logged and counted. +/// Process the durable default-channel work of this roster admission. +/// Failed reads or writes retain the token for a later claim or admission. +/// Leaves made during this admission are preserved. pub(crate) async fn auto_join_default_channels( tenant: &TenantContext, state: &Arc, pubkey_hex: &str, - added_by: Option<&[u8]>, + gate: Option<&Arc>, ) { - let Ok(pubkey) = hex::decode(pubkey_hex) else { - warn!(member = %pubkey_hex, "auto-join skipped: member pubkey is not hex"); - return; - }; - let channels = match state.db.list_auto_join_channels(tenant.community()).await { - Ok(channels) => channels, - Err(e) => { - warn!(member = %pubkey_hex, error = %e, "auto-join skipped: listing default channels failed"); - metrics::counter!("buzz_auto_join_failures_total").increment(1); - return; + if let Err(e) = finish_default_channel_joins(tenant, state, pubkey_hex, gate).await { + warn!(member = %pubkey_hex, error = %e, "default-channel joins remain pending"); + metrics::counter!("buzz_auto_join_failures_total").increment(1); + } +} + +async fn finish_default_channel_joins( + tenant: &TenantContext, + state: &Arc, + pubkey_hex: &str, + gate: Option<&Arc>, +) -> anyhow::Result<()> { + // A socket's AUTH retry runs on every admission, so it reads the replica + // and keeps the writer pool free. A roster write (no gate) reads the + // writer, where it has just committed the token. + let pending = match gate { + Some(_) => { + state + .db + .pending_default_channel_join_routed(tenant.community(), pubkey_hex) + .await? + } + None => { + state + .db + .pending_default_channel_join(tenant.community(), pubkey_hex) + .await? } }; - if channels.is_empty() { - return; - } - if let Some(actor) = added_by.filter(|actor| *actor != pubkey.as_slice()) { - match state + let Some(admission) = pending else { + return Ok(()); + }; + let pubkey = hex::decode(pubkey_hex)?; + // Use the persisted original actor even when a retry comes from the member. + let actor = admission + .added_by + .as_deref() + .filter(|actor| *actor != "invite") + .map(hex::decode) + .transpose()?; + if let Some(actor) = actor.as_deref().filter(|actor| *actor != pubkey.as_slice()) { + if let Some((policy, owner)) = state .db .get_agent_channel_policy(tenant.community(), &pubkey) - .await + .await? { - Ok(None) => {} - Ok(Some((policy, owner))) => { - if let Err(denied) = - channel_authz::decide_channel_add_policy(&policy, owner.as_deref(), actor) - { - warn!(member = %pubkey_hex, error = %denied, "auto-join skipped: the member's channel_add_policy refuses the actor"); - metrics::counter!("buzz_auto_join_failures_total").increment(1); - return; - } - } - Err(e) => { - warn!(member = %pubkey_hex, error = %e, "auto-join skipped: reading the member's channel_add_policy failed"); - metrics::counter!("buzz_auto_join_failures_total").increment(1); - return; + if let Err(denied) = + channel_authz::decide_channel_add_policy(&policy, owner.as_deref(), actor) + { + info!(member = %pubkey_hex, error = %denied, "default-channel add refused by member policy"); + state + .db + .finish_default_channel_join(tenant.community(), pubkey_hex, admission.token) + .await?; + return Ok(()); } } } + let channels = state.db.list_auto_join_channels(tenant.community()).await?; + let mut failure = None; for channel in channels { - match join_channel_as_self(tenant, state, channel.id, &pubkey).await { - Ok(SelfJoinOutcome::Joined) => { - metrics::counter!("buzz_auto_join_channels_total").increment(1); - info!(channel = %channel.id, member = %pubkey_hex, "auto-joined default channel"); - } - Ok(SelfJoinOutcome::AlreadyMember) => {} - Err(e) => { + #[cfg(test)] + if gate.is_some() { + crate::nip_fi_test_hooks::before_auto_join_channel(tenant.community()).await; + } + let _permit = match gate { + Some(gate) => match gate.acquire_effect().await { + Ok(permit) => Some(permit), + Err(_) => return Ok(()), // Pending work belongs to the next admission. + }, + None => None, + }; + if let Err(error) = + finish_default_channel_join(tenant, state, channel.id, &pubkey, &admission).await + { + warn!(channel = %channel.id, member = %pubkey_hex, error = %error, + "default-channel join remains pending; continuing other channels"); + failure = Some(error); + } + } + if let Some(error) = failure { + return Err(error); + } + state + .db + .finish_default_channel_join(tenant.community(), pubkey_hex, admission.token) + .await?; + Ok(()) +} + +async fn finish_default_channel_join( + tenant: &TenantContext, + state: &Arc, + channel: Uuid, + pubkey: &[u8], + admission: &buzz_db::relay_members::PendingDefaultChannelJoin, +) -> anyhow::Result<()> { + match state + .db + .put_pending_default_member(tenant.community(), channel, pubkey, admission.token) + .await + { + Ok(Some(_)) => { + // Membership is durable now. Announcements keep their existing + // failure handling; retries must not replay them for active members. + if let Err(error) = announce_self_join(tenant, state, channel, pubkey).await { + warn!(channel = %channel, error = %error, "default-channel announcement failed"); metrics::counter!("buzz_auto_join_failures_total").increment(1); - warn!(channel = %channel.id, member = %pubkey_hex, error = %e, "auto-join of default channel failed"); } + metrics::counter!("buzz_auto_join_channels_total").increment(1); } + Ok(None) => {} + Err(buzz_db::DbError::ChannelNotFound(_)) => {} + Err(error) => return Err(error.into()), } + Ok(()) } /// Join a person admitted to a community that keeps no membership roster diff --git a/migrations/0084_default_channel_recovery.sql b/migrations/0084_default_channel_recovery.sql new file mode 100644 index 00000000000..7180b85cd7c --- /dev/null +++ b/migrations/0084_default_channel_recovery.sql @@ -0,0 +1,5 @@ +-- Expand only: old relays ignore this nullable recovery token. +-- New roster writers set it atomically; existing members are not backfilled. +SET LOCAL lock_timeout = '5s'; + +ALTER TABLE relay_members ADD COLUMN auto_join_pending UUID; diff --git a/schema/tables/public/relay_members.sql b/schema/tables/public/relay_members.sql index 8084a4db714..b739e0aa294 100644 --- a/schema/tables/public/relay_members.sql +++ b/schema/tables/public/relay_members.sql @@ -7,6 +7,7 @@ CREATE TABLE relay_members ( pubkey TEXT NOT NULL, role TEXT NOT NULL CHECK (role IN ('owner', 'admin', 'member')), added_by TEXT, + auto_join_pending UUID, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), PRIMARY KEY (community_id, pubkey)