From 9ce9252cc3f731bdf18e4b72fdcbfde162a5cc0a Mon Sep 17 00:00:00 2001 From: Helen Tian Date: Wed, 4 Nov 2026 14:32:19 +1100 Subject: [PATCH 1/9] fix(relay): recover interrupted default-channel auto-joins Record pending joins atomically with roster admissions. Resume them on repeated claims/adds and authenticated connections, preserving leaves, original actor policy and admission generations. Keep client discovery pending until its required state and notification have been written. Co-authored-by: Flipper <1ef64189596befbc8b93f9021bff35ae5d02b155d9e05fc124bca7349927d597@buzz.block.builderlab.xyz> Signed-off-by: Helen Tian --- crates/buzz-db/src/runtime/migration.rs | 7 +- crates/buzz-db/src/store/channel_members.rs | 84 +++- crates/buzz-db/src/store/relay_invite.rs | 4 +- .../src/store/relay_invite/claim_owner.rs | 4 +- crates/buzz-db/src/store/relay_members.rs | 53 ++- crates/buzz-relay/src/api/invites.rs | 30 +- .../postgres_tests/recovery_postgres_tests.rs | 426 ++++++++++++++++++ crates/buzz-relay/src/handlers/auth.rs | 230 +++++++--- crates/buzz-relay/src/handlers/relay_admin.rs | 11 +- .../buzz-relay/src/handlers/side_effects.rs | 181 +++++--- migrations/0084_default_channel_recovery.sql | 5 + schema/tables/public/relay_members.sql | 1 + 12 files changed, 866 insertions(+), 170 deletions(-) create mode 100644 crates/buzz-relay/src/api/invites/postgres_tests/recovery_postgres_tests.rs create mode 100644 migrations/0084_default_channel_recovery.sql 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/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 5d42dd40003..0926fd72bbe 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -721,6 +721,17 @@ pub async fn put_members( Ok(results) } +/// Outcome of a default-channel membership write during roster recovery. +#[derive(Debug)] +pub enum DefaultMemberOutcome { + /// This transaction joined the member. + Joined(MemberRecord), + /// Already active; required discovery may still need recovery. + Active, + /// Left during this admission, or the admission token is no longer current. + Skipped, +} + /// Self-join `pubkey` to `channel_id` as [`put_member`] with no actor and no /// requested role (so the channel's `join_role` applies), but only if /// `pubkey` has never had a membership row in the channel, active or removed. @@ -737,25 +748,69 @@ pub async fn put_first_member( channel_id: Uuid, pubkey: &[u8], ) -> Result> { + Ok( + match put_first_member_with_pending(pool, community_id, channel_id, pubkey, None).await? { + DefaultMemberOutcome::Joined(member) => Some(member), + DefaultMemberOutcome::Active | DefaultMemberOutcome::Skipped => None, + }, + ) +} + +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, community_id, 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(DefaultMemberOutcome::Skipped); + }; + 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(if removed_at.is_none() { + DefaultMemberOutcome::Active + } else { + DefaultMemberOutcome::Skipped + }); + } } let target = MemberPut { pubkey, @@ -768,7 +823,7 @@ pub async fn put_first_member( match put { Ok(put) => { tx.commit().await?; - Ok(Some(put.member)) + Ok(DefaultMemberOutcome::Joined(put.member)) } Err(refused) => { tx.rollback().await?; @@ -2259,6 +2314,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 a95f3f69ea6..4b06df37194 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -158,8 +158,8 @@ 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, gen_random_uuid()) ON CONFLICT (community_id, pubkey) DO NOTHING", ) .bind(community.as_uuid()) .bind(pubkey) @@ -1189,6 +1189,55 @@ pub async fn rewrite_nip43_membership_list( } 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)>> { + let mut conn = observability::acquire_writer( + &self.pool, + observability::WriterOperation::Authorization, + ) + .await?; + Ok(sqlx::query_as( + "SELECT auto_join_pending, 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(&mut *conn) + .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 diff --git a/crates/buzz-relay/src/api/invites.rs b/crates/buzz-relay/src/api/invites.rs index dc1709d32d1..4de96e46a6c 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 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 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}"); + } } - // 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; 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..0ecb2cf847a --- /dev/null +++ b/crates/buzz-relay/src/api/invites/postgres_tests/recovery_postgres_tests.rs @@ -0,0 +1,426 @@ +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"); + } + + /// All roster admission paths commit recovery with membership. A failed list + /// leaves work behind; a repeated claim/add resumes it and a later leave stands. + #[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( + community, + &keys.public_key().to_hex(), + role, + Some(&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(); + assert!(matches!( + state + .db + .put_pending_default_member(community, a, &member.public_key().to_bytes(), old) + .await + .unwrap(), + buzz_db::channel_members::DefaultMemberOutcome::Joined(_) + )); + 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( + community, + &member_hex, + "member", + Some(&owner.public_key().to_hex()), + &state.relay_keypair, + ) + .await + .unwrap(); + let (new, _) = state + .db + .pending_default_channel_join(community, &member_hex) + .await + .unwrap() + .unwrap(); + assert_ne!(old, new); + let c = create_channel(&state, &tenant, &owner, &[]).await; + edit_channel(&state, &tenant, &owner, c, &[&["auto_join", "true"]]) + .await + .unwrap(); + assert!(matches!( + state + .db + .put_pending_default_member(community, c, &member.public_key().to_bytes(), old) + .await + .unwrap(), + buzz_db::channel_members::DefaultMemberOutcome::Skipped + )); + 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() + .0, + new + ); + assert!( + matches!( + state + .db + .put_pending_default_member(community, a, &member.public_key().to_bytes(), new,) + .await + .unwrap(), + buzz_db::channel_members::DefaultMemberOutcome::Joined(_) + ), + "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()); + } + + /// A crash or failed discovery write after membership commits must recover the + /// client-visible roster, not just retain the already-correct membership row. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn pending_join_recovers_failed_discovery_for_an_active_membership() { + use crate::handlers::group_state_postgres_tests::group_state_tags; + let host = format!("join-discovery-{}.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( + community, + &keys.public_key().to_hex(), + role, + 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(); + let (pending, _) = state + .db + .pending_default_channel_join(community, &member_hex) + .await + .unwrap() + .unwrap(); + let _ = state + .db + .put_pending_default_member( + community, + channel, + &member.public_key().to_bytes(), + pending, + ) + .await + .unwrap(); + assert!(!group_state_tags(&state, &tenant, channel, 39002) + .await + .iter() + .any(|tag| tag[0] == "p" && tag[1] == member_hex)); + + let mut blocker = state.db.pool().begin().await.unwrap(); + sqlx::query("LOCK TABLE events IN SHARE MODE") + .execute(&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_blocked_query(state.db.pool(), "%events%").await; + tokio::time::timeout(Duration::from_secs(10), retry) + .await + .unwrap() + .unwrap(); + blocker.rollback().await.unwrap(); + assert!( + state + .db + .pending_default_channel_join(community, &member_hex) + .await + .unwrap() + .is_some(), + "failed discovery must leave recovery pending" + ); + auto_join_default_channels(&tenant, &state, &member_hex, None).await; + assert!(group_state_tags(&state, &tenant, channel, 39002) + .await + .iter() + .any(|tag| tag[0] == "p" && tag[1] == member_hex)); + assert_eq!( + added_notification_count(&state, community, &member_hex).await, + 1 + ); + 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 76792fe6491..733c758ecef 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) => { @@ -2982,6 +2988,71 @@ 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( + community, + &keys.public_key().to_hex(), + "member", + Some("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()); + } + 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 @@ -2994,67 +3065,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( + community, + &person.public_key().to_hex(), + "member", + Some("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 bd5f7c56a41..6831c269052 100644 --- a/crates/buzz-relay/src/handlers/relay_admin.rs +++ b/crates/buzz-relay/src/handlers/relay_admin.rs @@ -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 0740206c2a6..74fc0930820 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -2987,13 +2987,40 @@ pub(crate) async fn join_channel_as_self( } /// The side effects of a committed self-join: the membership cache, the -/// `member_joined` kind:40099, kind:39000/39002 and kind:44100. Only the -/// system message is fatal, as for kind:9021. +/// `member_joined` kind:40099, kind:39000/39002 and kind:44100. async fn announce_self_join( tenant: &TenantContext, state: &Arc, channel_id: Uuid, pubkey: &[u8], +) -> anyhow::Result<()> { + emit_self_join_message(tenant, state, channel_id, pubkey).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"); + } + + if let Err(e) = emit_membership_notification( + tenant, + state, + channel_id, + pubkey, + pubkey, + buzz_core::kind::KIND_MEMBER_ADDED_NOTIFICATION, + ) + .await + { + warn!("membership notification for join failed: {e}"); + } + + Ok(()) +} + +async fn emit_self_join_message( + tenant: &TenantContext, + state: &Arc, + channel_id: Uuid, + pubkey: &[u8], ) -> anyhow::Result<()> { state.invalidate_membership(tenant, channel_id, pubkey); @@ -3011,11 +3038,18 @@ async fn announce_self_join( ) .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(()) +} - if let Err(e) = emit_membership_notification( +async fn refresh_self_join_discovery( + tenant: &TenantContext, + state: &Arc, + channel_id: Uuid, + pubkey: &[u8], +) -> anyhow::Result<()> { + state.invalidate_membership(tenant, channel_id, pubkey); + emit_group_discovery_events(tenant, state, channel_id).await?; + emit_membership_notification( tenant, state, channel_id, @@ -3023,87 +3057,104 @@ async fn announce_self_join( pubkey, buzz_core::kind::KIND_MEMBER_ADDED_NOTIFICATION, ) - .await - { - warn!("membership notification for join failed: {e}"); - } - + .await?; 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 channels.is_empty() { - 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); } - if let Some(actor) = added_by.filter(|actor| *actor != pubkey.as_slice()) { - match state +} + +async fn finish_default_channel_joins( + tenant: &TenantContext, + state: &Arc, + pubkey_hex: &str, + gate: Option<&Arc>, +) -> anyhow::Result<()> { + let Some((pending, added_by)) = state + .db + .pending_default_channel_join(tenant.community(), pubkey_hex) + .await? + else { + return Ok(()); + }; + let pubkey = hex::decode(pubkey_hex)?; + // Use the persisted original actor even when a retry comes from the member. + let actor = 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, pending) + .await?; + return Ok(()); } } } + let channels = state.db.list_auto_join_channels(tenant.community()).await?; for channel in channels { - match join_channel_as_self(tenant, state, channel.id, &pubkey).await { - Ok(SelfJoinOutcome::Joined) => { + #[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, + }; + match state + .db + .put_pending_default_member(tenant.community(), channel.id, &pubkey, pending) + .await + { + Ok(buzz_db::channel_members::DefaultMemberOutcome::Joined(_)) => { + // Membership is durable even if an announcement fails. A retry + // observes the row rather than resurrecting a later leave. + emit_self_join_message(tenant, state, channel.id, &pubkey).await?; + refresh_self_join_discovery(tenant, state, channel.id, &pubkey).await?; 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) => { - metrics::counter!("buzz_auto_join_failures_total").increment(1); - warn!(channel = %channel.id, member = %pubkey_hex, error = %e, "auto-join of default channel failed"); + Ok(buzz_db::channel_members::DefaultMemberOutcome::Active) => { + refresh_self_join_discovery(tenant, state, channel.id, &pubkey).await?; } + Ok(buzz_db::channel_members::DefaultMemberOutcome::Skipped) => {} + // A channel archived, privatized or deleted since listing is no + // longer eligible. Database errors must retain the pending work. + Err(buzz_db::DbError::AccessDenied(_) | buzz_db::DbError::ChannelNotFound(_)) => {} + Err(e) => return Err(e.into()), } } + state + .db + .finish_default_channel_join(tenant.community(), pubkey_hex, pending) + .await?; + 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) From 0984152f722984003a73402aa66add4bf5a70a15 Mon Sep 17 00:00:00 2001 From: Helen Tian Date: Wed, 4 Nov 2026 14:54:13 +1100 Subject: [PATCH 2/9] fix(relay): fence and deduplicate recovered join notices Keep healthy default-channel joins progressing when another channel fails. Use the persisted admission identity for retried notices, and serialize notice insertion with membership removal and admission generation changes. Cover failures, later builder clocks, leaves, kicks and roster removal. Co-authored-by: Flipper <1ef64189596befbc8b93f9021bff35ae5d02b155d9e05fc124bca7349927d597@buzz.block.builderlab.xyz> Signed-off-by: Helen Tian --- crates/buzz-db/src/store/channel_members.rs | 45 +- crates/buzz-db/src/store/relay_members.rs | 15 +- .../postgres_tests/recovery_postgres_tests.rs | 420 +++++++++++++++++- .../buzz-relay/src/handlers/side_effects.rs | 124 ++++-- 4 files changed, 561 insertions(+), 43 deletions(-) diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 0926fd72bbe..3a4337ade58 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -827,7 +827,11 @@ async fn put_first_member_with_pending( } Err(refused) => { tx.rollback().await?; - Err(refused) + if pending.is_some() && matches!(refused, DbError::AccessDenied(_)) { + Ok(DefaultMemberOutcome::Skipped) + } else { + Err(refused) + } } } } @@ -2314,6 +2318,45 @@ impl Db { .await } + /// Persist an admission's add notice only while that admission and channel + /// membership are current. The channel lock spans the check and durable + /// insertion, ordering this notice before any subsequent leave/removal. + #[datastore_span(name = "insert_pending_default_notification", system = "postgresql")] + pub async fn insert_pending_default_notification( + &self, + community: CommunityId, + channel: Uuid, + pubkey: &[u8], + pending: Uuid, + event: &nostr::Event, + ) -> Result> { + let mut tx = self.begin_event_write_transaction(community).await?; + let current: Option = sqlx::query_scalar( + "SELECT auto_join_pending FROM relay_members \ + WHERE community_id = $1 AND pubkey = $2 AND auto_join_pending = $3 FOR SHARE", + ) + .bind(community.as_uuid()) + .bind(hex::encode(pubkey)) + .bind(pending) + .fetch_optional(tx.conn()) + .await?; + if current.is_none() { + tx.rollback().await?; + return Ok(None); + } + acquire_channel_membership_lock(tx.conn(), community, channel).await?; + if !is_member_in_transaction(&mut tx, channel, pubkey).await? { + tx.rollback().await?; + return Ok(None); + } + let stored = crate::event::insert_event_in_transaction(&mut tx, event, None).await?; + if stored.1 { + crate::insert_mentions_in_transaction(&mut tx, event, None).await?; + } + tx.commit().await?; + Ok(Some(stored)) + } + /// 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")] diff --git a/crates/buzz-db/src/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs index 4b06df37194..e6eaba9d492 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -1188,6 +1188,17 @@ 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, + /// Stable timestamp for this admission's idempotent notifications. + pub admitted_at: DateTime, +} + impl Db { /// Read unfinished default-channel work and its original adding actor. #[datastore_span(name = "pending_default_channel_join", system = "postgresql")] @@ -1195,14 +1206,14 @@ impl Db { &self, community: CommunityId, pubkey: &str, - ) -> Result)>> { + ) -> Result> { let mut conn = observability::acquire_writer( &self.pool, observability::WriterOperation::Authorization, ) .await?; Ok(sqlx::query_as( - "SELECT auto_join_pending, added_by FROM relay_members \ + "SELECT auto_join_pending AS token, added_by, created_at AS admitted_at FROM relay_members \ WHERE community_id = $1 AND pubkey = $2 AND auto_join_pending IS NOT NULL", ) .bind(community.as_uuid()) 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 index 0ecb2cf847a..31a4281ef99 100644 --- 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 @@ -2,7 +2,7 @@ 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; + use crate::handlers::side_effects::{auto_join_default_channels, notice_barrier}; // Cancel the real listing query only after PostgreSQL reports that it is // blocked by our table lock. No timer or production-only failure hook. @@ -29,6 +29,407 @@ mod postgres_tests { .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( + community, + &member.public_key().to_hex(), + "member", + None, + &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"); + } + + async fn added_notifications( + state: &AppState, + community: buzz_core::CommunityId, + member_hex: &str, + ) -> Vec { + state + .db + .query_events(&buzz_db::EventQuery { + kinds: Some(vec![buzz_core::kind::KIND_MEMBER_ADDED_NOTIFICATION as i32]), + p_tag_hex: Some(member_hex.to_string()), + global_only: true, + ..buzz_db::EventQuery::for_community(community) + }) + .await + .unwrap() + .into_iter() + .map(|stored| stored.event) + .collect() + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn channel_local_failures_continue_later_defaults_and_retain_retry_work() { + for failure in ["membership", "announcement"] { + let (state, tenant, _, member, channels) = recovery_fixture(failure, 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(); + if failure == "membership" { + 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(); + } else { + sqlx::query("LOCK TABLE events IN SHARE MODE") + .execute(&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 announcements 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, + [ + if failure == "membership" { + None + } else { + Some("member".into()) + }, + Some("member".into()) + ], + "{failure}: 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 recovered_notices_keep_the_admission_identity_across_later_builder_clocks() { + let (state, tenant, _, member, channels) = recovery_fixture("notice-retry", 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 barrier = notice_barrier::install(&member.public_key().to_bytes()); + let first_clock = pending.admitted_at.timestamp() as u64 + 60; + barrier.pin_created_at(first_clock); + 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[1]) + .fetch_one(&mut *blocker) + .await + .unwrap(); + let retry = || { + let (state, tenant, member_hex) = (state.clone(), tenant.clone(), member_hex.clone()); + tokio::spawn(async move { + auto_join_default_channels(&tenant, &state, &member_hex, None).await + }) + }; + let first = retry(); + let (channel, release) = tokio::time::timeout(Duration::from_secs(10), barrier.next()) + .await + .unwrap() + .unwrap(); + assert_eq!(channel, channels[0]); + release.send(()).unwrap(); + cancel_query_blocked_by(state.db.pool(), blocker_pid).await; + blocker.rollback().await.unwrap(); + tokio::time::timeout(Duration::from_secs(10), first) + .await + .unwrap() + .unwrap(); + let before = added_notifications(&state, community, &member_hex).await; + assert_eq!(before.len(), 1); + assert_eq!( + state + .db + .pending_default_channel_join(community, &member_hex) + .await + .unwrap() + .unwrap() + .token, + pending.token + ); + + barrier.pin_created_at(first_clock + 3600); + let second = retry(); + for expected in &channels { + let (channel, release) = tokio::time::timeout(Duration::from_secs(10), barrier.next()) + .await + .unwrap() + .unwrap(); + assert_eq!(&channel, expected); + release.send(()).unwrap(); + } + tokio::time::timeout(Duration::from_secs(10), second) + .await + .unwrap() + .unwrap(); + let after = added_notifications(&state, community, &member_hex).await; + assert_eq!( + after.len(), + 2, + "retry must not persist another first-channel notice" + ); + assert!(after.iter().any(|notice| notice.id == before[0].id)); + for channel in &channels { + let channel = channel.to_string(); + assert_eq!( + after + .iter() + .filter(|notice| notice + .tags + .iter() + .any(|tag| tag.as_slice() == ["h", channel.as_str()])) + .count(), + 1, + "each default channel needs exactly one persisted notice", + ); + } + let token = pending.token.to_string(); + for notice in after { + assert_eq!( + notice.created_at.as_secs(), + pending.admitted_at.timestamp() as u64 + ); + assert!(notice + .tags + .iter() + .any(|tag| tag.as_slice() == ["admission", token.as_str()])); + } + assert!(state + .db + .pending_default_channel_join(community, &member_hex) + .await + .unwrap() + .is_none()); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn recovered_notice_does_not_outlive_a_committed_leave_or_removal() { + for removal in ["leave", "kick", "relay-removal"] { + let (state, tenant, owner, member, channels) = recovery_fixture(removal, 1).await; + let community = tenant.community(); + let member_hex = member.public_key().to_hex(); + let member_bytes = member.public_key().to_bytes(); + let pending = state + .db + .pending_default_channel_join(community, &member_hex) + .await + .unwrap() + .unwrap(); + assert!(matches!( + state + .db + .put_pending_default_member( + community, + channels[0], + &member_bytes, + pending.token + ) + .await + .unwrap(), + buzz_db::channel_members::DefaultMemberOutcome::Joined(_) + )); + let mut barrier = notice_barrier::install(&member_bytes); + 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 } + }); + let (channel, release) = tokio::time::timeout(Duration::from_secs(10), barrier.next()) + .await + .unwrap() + .unwrap(); + assert_eq!(channel, channels[0]); + if removal == "relay-removal" { + let _ = tokio::time::timeout( + Duration::from_secs(10), + state + .db + .remove_relay_member(community, &member_hex, &state.relay_keypair), + ) + .await + .expect("roster removal must commit while the notice is parked") + .unwrap(); + assert!(!state + .db + .is_relay_member(community, &member_hex) + .await + .unwrap()); + } else { + let actor = if removal == "leave" { + member_bytes + } else { + owner.public_key().to_bytes() + }; + tokio::time::timeout( + Duration::from_secs(10), + state + .db + .remove_member(community, channel, &member_bytes, &actor), + ) + .await + .expect("channel removal must commit while the notice is parked") + .unwrap(); + assert!(!state + .db + .is_member(community, channel, &member_bytes) + .await + .unwrap()); + } + release.send(()).unwrap(); + tokio::time::timeout(Duration::from_secs(10), retry) + .await + .unwrap() + .unwrap(); + assert_eq!( + added_notification_count(&state, community, &member_hex).await, + 0, + "{removal}: the parked add notice became stale before its insert" + ); + assert!(state + .db + .pending_default_channel_join(community, &member_hex) + .await + .unwrap() + .is_none()); + } + } + /// All roster admission paths commit recovery with membership. A failed list /// leaves work behind; a repeated claim/add resumes it and a later leave stands. #[tokio::test] @@ -208,12 +609,13 @@ mod postgres_tests { .await .unwrap(); } - let (old, _) = state + let old = state .db .pending_default_channel_join(community, &member_hex) .await .unwrap() - .unwrap(); + .unwrap() + .token; assert!(matches!( state .db @@ -254,12 +656,13 @@ mod postgres_tests { ) .await .unwrap(); - let (new, _) = state + let new = state .db .pending_default_channel_join(community, &member_hex) .await .unwrap() - .unwrap(); + .unwrap() + .token; assert_ne!(old, new); let c = create_channel(&state, &tenant, &owner, &[]).await; edit_channel(&state, &tenant, &owner, c, &[&["auto_join", "true"]]) @@ -285,7 +688,7 @@ mod postgres_tests { .await .unwrap() .unwrap() - .0, + .token, new ); assert!( @@ -362,12 +765,13 @@ mod postgres_tests { edit_channel(&state, &tenant, &owner, channel, &[&["auto_join", "true"]]) .await .unwrap(); - let (pending, _) = state + let pending = state .db .pending_default_channel_join(community, &member_hex) .await .unwrap() - .unwrap(); + .unwrap() + .token; let _ = state .db .put_pending_default_member( diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 74fc0930820..1ea3150197d 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1342,6 +1342,7 @@ pub(crate) async fn emit_membership_notification_tagged( actor_pubkey, notification_kind, extra_tags, + None, ) .await } @@ -1358,6 +1359,7 @@ async fn emit_typed_membership_notification( actor_pubkey: &[u8], notification_kind: u32, extra_tags: &[Tag], + admission: Option<&buzz_db::relay_members::PendingDefaultChannelJoin>, ) -> anyhow::Result<()> { let target_hex = hex::encode(target_pubkey); let actor_hex = hex::encode(actor_pubkey); @@ -1398,16 +1400,44 @@ async fn emit_typed_membership_notification( Some(at) => builder.custom_created_at(nostr::Timestamp::from(at)), None => builder, }; + let builder = if let Some(admission) = admission { + builder + .custom_created_at(nostr::Timestamp::from( + admission.admitted_at.timestamp() as u64 + )) + .tags([Tag::parse(["admission", &admission.token.to_string()])?]) + } else { + builder + }; let event = builder .sign_with_keys(&state.relay_keypair) .map_err(|e| anyhow::anyhow!("failed to sign membership notification: {e}"))?; - // Store with channel_id = None → globally scoped, reachable by global subscribers. - let (stored, was_inserted) = state - .db - .insert_event(tenant.community(), &event, None) - .await - .map_err(|e| anyhow::anyhow!("membership notification insert failed: {}", e.kind()))?; + // Recovery serializes the durable add with departures and the admission + // generation. Ordinary notices retain the existing insertion path. + let stored = match admission { + Some(admission) => { + state + .db + .insert_pending_default_notification( + tenant.community(), + channel_id, + target_pubkey, + admission.token, + &event, + ) + .await? + } + None => Some( + state + .db + .insert_event(tenant.community(), &event, None) + .await?, + ), + }; + let Some((stored, was_inserted)) = stored else { + return Ok(()); + }; if !was_inserted { return Ok(()); } @@ -2203,6 +2233,7 @@ pub(crate) async fn persist_put_user( &actor_bytes, KIND_MEMBER_ADDED_NOTIFICATION, &[], + None, ) .await { @@ -2903,6 +2934,7 @@ async fn handle_delete_group( &actor_bytes, KIND_MEMBER_REMOVED_NOTIFICATION, &[], + None, ) .await { @@ -3046,16 +3078,27 @@ async fn refresh_self_join_discovery( state: &Arc, channel_id: Uuid, pubkey: &[u8], + admission: &buzz_db::relay_members::PendingDefaultChannelJoin, ) -> anyhow::Result<()> { state.invalidate_membership(tenant, channel_id, pubkey); emit_group_discovery_events(tenant, state, channel_id).await?; - emit_membership_notification( + let Some(channel_type) = state + .db + .channel_type_of(tenant.community(), channel_id) + .await? + else { + return Ok(()); + }; + emit_typed_membership_notification( tenant, state, channel_id, + &channel_type, pubkey, pubkey, - buzz_core::kind::KIND_MEMBER_ADDED_NOTIFICATION, + KIND_MEMBER_ADDED_NOTIFICATION, + &[], + Some(admission), ) .await?; Ok(()) @@ -3082,7 +3125,7 @@ async fn finish_default_channel_joins( pubkey_hex: &str, gate: Option<&Arc>, ) -> anyhow::Result<()> { - let Some((pending, added_by)) = state + let Some(admission) = state .db .pending_default_channel_join(tenant.community(), pubkey_hex) .await? @@ -3091,7 +3134,8 @@ async fn finish_default_channel_joins( }; let pubkey = hex::decode(pubkey_hex)?; // Use the persisted original actor even when a retry comes from the member. - let actor = added_by + let actor = admission + .added_by .as_deref() .filter(|actor| *actor != "invite") .map(hex::decode) @@ -3108,13 +3152,14 @@ async fn finish_default_channel_joins( info!(member = %pubkey_hex, error = %denied, "default-channel add refused by member policy"); state .db - .finish_default_channel_join(tenant.community(), pubkey_hex, pending) + .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 { #[cfg(test)] if gate.is_some() { @@ -3127,36 +3172,51 @@ async fn finish_default_channel_joins( }, None => None, }; - match state - .db - .put_pending_default_member(tenant.community(), channel.id, &pubkey, pending) - .await + if let Err(error) = + finish_default_channel_join(tenant, state, channel.id, &pubkey, &admission).await { - Ok(buzz_db::channel_members::DefaultMemberOutcome::Joined(_)) => { - // Membership is durable even if an announcement fails. A retry - // observes the row rather than resurrecting a later leave. - emit_self_join_message(tenant, state, channel.id, &pubkey).await?; - refresh_self_join_discovery(tenant, state, channel.id, &pubkey).await?; - metrics::counter!("buzz_auto_join_channels_total").increment(1); - info!(channel = %channel.id, member = %pubkey_hex, "auto-joined default channel"); - } - Ok(buzz_db::channel_members::DefaultMemberOutcome::Active) => { - refresh_self_join_discovery(tenant, state, channel.id, &pubkey).await?; - } - Ok(buzz_db::channel_members::DefaultMemberOutcome::Skipped) => {} - // A channel archived, privatized or deleted since listing is no - // longer eligible. Database errors must retain the pending work. - Err(buzz_db::DbError::AccessDenied(_) | buzz_db::DbError::ChannelNotFound(_)) => {} - Err(e) => return Err(e.into()), + 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, pending) + .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(buzz_db::channel_members::DefaultMemberOutcome::Joined(_)) => { + emit_self_join_message(tenant, state, channel, pubkey).await?; + refresh_self_join_discovery(tenant, state, channel, pubkey, admission).await?; + metrics::counter!("buzz_auto_join_channels_total").increment(1); + } + Ok(buzz_db::channel_members::DefaultMemberOutcome::Active) => { + refresh_self_join_discovery(tenant, state, channel, pubkey, admission).await?; + } + Ok(buzz_db::channel_members::DefaultMemberOutcome::Skipped) => {} + Err(buzz_db::DbError::ChannelNotFound(_)) => {} + Err(error) => return Err(error.into()), + } + Ok(()) +} + /// Join a person admitted to a community that keeps no membership roster /// (`require_relay_membership = false`) to every effective `auto_join` /// channel they have never been a member of, as kind:9021 self-joins. From 0afb6a74e2fa1443931b7dced75c5927865a058e Mon Sep 17 00:00:00 2001 From: Helen Tian Date: Wed, 4 Nov 2026 15:05:24 +1100 Subject: [PATCH 3/9] fix(relay): deliver recovered join notices to recent subscriptions Date first recovery notices at emission time so startup subscription cutoffs include them. Reuse the first persisted notice as the per-channel admission identity, checking its tags under the membership lock to keep later retries idempotent without backdating newly recovered notices. Co-authored-by: Flipper <1ef64189596befbc8b93f9021bff35ae5d02b155d9e05fc124bca7349927d597@buzz.block.builderlab.xyz> Signed-off-by: Helen Tian --- crates/buzz-db/src/store/channel_members.rs | 23 +++++++++ crates/buzz-db/src/store/relay_members.rs | 4 +- .../postgres_tests/recovery_postgres_tests.rs | 51 ++++++++++++++----- .../buzz-relay/src/handlers/side_effects.rs | 6 +-- 4 files changed, 64 insertions(+), 20 deletions(-) diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 3a4337ade58..05af50499f6 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -2349,6 +2349,29 @@ impl Db { tx.rollback().await?; return Ok(None); } + // The first event is the durable per-channel delivery identity. A + // retry uses its admission tag instead of minting a new timestamp/ID. + // Check under the same membership lock as the insert, so concurrent + // workers cannot both create the first notice. Include deleted rows: + // recovery must not resurrect a notice deliberately removed later. + let recorded: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM events \ + WHERE community_id = $1 AND pubkey = $2 AND kind = $3 AND tags @> $4)", + ) + .bind(community.as_uuid()) + .bind(event.pubkey.to_bytes().as_slice()) + .bind(buzz_core::kind::KIND_MEMBER_ADDED_NOTIFICATION as i32) + .bind(serde_json::json!([ + ["admission", pending.to_string()], + ["h", channel.to_string()], + ["p", hex::encode(pubkey)], + ])) + .fetch_one(tx.conn()) + .await?; + if recorded { + tx.rollback().await?; + return Ok(None); + } let stored = crate::event::insert_event_in_transaction(&mut tx, event, None).await?; if stored.1 { crate::insert_mentions_in_transaction(&mut tx, event, None).await?; diff --git a/crates/buzz-db/src/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs index e6eaba9d492..20bfc047989 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -1195,8 +1195,6 @@ pub struct PendingDefaultChannelJoin { pub token: Uuid, /// Original actor; `invite` denotes a self-claim. pub added_by: Option, - /// Stable timestamp for this admission's idempotent notifications. - pub admitted_at: DateTime, } impl Db { @@ -1213,7 +1211,7 @@ impl Db { ) .await?; Ok(sqlx::query_as( - "SELECT auto_join_pending AS token, added_by, created_at AS admitted_at FROM relay_members \ + "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()) 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 index 31a4281ef99..4689f1bf7df 100644 --- 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 @@ -231,6 +231,18 @@ mod postgres_tests { let (state, tenant, _, member, channels) = recovery_fixture("notice-retry", 2).await; let community = tenant.community(); let member_hex = member.public_key().to_hex(); + let first_clock = nostr::Timestamp::now().as_secs(); + sqlx::query( + "UPDATE relay_members SET created_at = $3 WHERE community_id = $1 AND pubkey = $2", + ) + .bind(community.as_uuid()) + .bind(&member_hex) + .bind( + chrono::DateTime::::from_timestamp(first_clock as i64 - 7200, 0).unwrap(), + ) + .execute(state.db.pool()) + .await + .unwrap(); let pending = state .db .pending_default_channel_join(community, &member_hex) @@ -238,7 +250,6 @@ mod postgres_tests { .unwrap() .unwrap(); let mut barrier = notice_barrier::install(&member.public_key().to_bytes()); - let first_clock = pending.admitted_at.timestamp() as u64 + 60; barrier.pin_created_at(first_clock); let mut blocker = state.db.pool().begin().await.unwrap(); let blocker_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") @@ -272,6 +283,19 @@ mod postgres_tests { .unwrap(); let before = added_notifications(&state, community, &member_hex).await; assert_eq!(before.len(), 1); + let membership_filter: nostr::Filter = serde_json::from_value(serde_json::json!({ + "kinds": [buzz_core::kind::KIND_MEMBER_ADDED_NOTIFICATION, buzz_core::kind::KIND_MEMBER_REMOVED_NOTIFICATION], + "#p": [member_hex], + "since": first_clock, + })).unwrap(); + assert!( + buzz_core::filter::filters_match( + &[membership_filter], + &buzz_core::StoredEvent::new(before[0].clone(), None), + ), + "a recovered notice must reach the membership subscription started after admission", + ); + assert_eq!(before[0].created_at.as_secs(), first_clock); assert_eq!( state .db @@ -304,26 +328,29 @@ mod postgres_tests { "retry must not persist another first-channel notice" ); assert!(after.iter().any(|notice| notice.id == before[0].id)); - for channel in &channels { + for (index, channel) in channels.iter().enumerate() { let channel = channel.to_string(); - assert_eq!( - after - .iter() - .filter(|notice| notice + let notices = after + .iter() + .filter(|notice| { + notice .tags .iter() - .any(|tag| tag.as_slice() == ["h", channel.as_str()])) - .count(), + .any(|tag| tag.as_slice() == ["h", channel.as_str()]) + }) + .collect::>(); + assert_eq!( + notices.len(), 1, "each default channel needs exactly one persisted notice", ); + assert_eq!( + notices[0].created_at.as_secs(), + first_clock + index as u64 * 3600 + ); } let token = pending.token.to_string(); for notice in after { - assert_eq!( - notice.created_at.as_secs(), - pending.admitted_at.timestamp() as u64 - ); assert!(notice .tags .iter() diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 1ea3150197d..50b47d5c684 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1401,11 +1401,7 @@ async fn emit_typed_membership_notification( None => builder, }; let builder = if let Some(admission) = admission { - builder - .custom_created_at(nostr::Timestamp::from( - admission.admitted_at.timestamp() as u64 - )) - .tags([Tag::parse(["admission", &admission.token.to_string()])?]) + builder.tags([Tag::parse(["admission", &admission.token.to_string()])?]) } else { builder }; From f19e31a836f3a1aec6761a25bdfccd147adf63f7 Mon Sep 17 00:00:00 2001 From: Helen Tian Date: Thu, 5 Nov 2026 10:19:57 +1100 Subject: [PATCH 4/9] fix(relay): limit default-channel recovery to membership Keep pending admission work and safe membership retries. Use the existing announcement path only for newly joined channels; remove discovery replay and admission-specific notice persistence. Co-authored-by: Flipper <1ef64189596befbc8b93f9021bff35ae5d02b155d9e05fc124bca7349927d597@buzz.block.builderlab.xyz> Signed-off-by: Helen Tian --- crates/buzz-db/src/store/channel_members.rs | 96 +--- .../postgres_tests/recovery_postgres_tests.rs | 451 ++---------------- .../buzz-relay/src/handlers/side_effects.rs | 120 +---- 3 files changed, 70 insertions(+), 597 deletions(-) diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 05af50499f6..acbc1198aa5 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -721,17 +721,6 @@ pub async fn put_members( Ok(results) } -/// Outcome of a default-channel membership write during roster recovery. -#[derive(Debug)] -pub enum DefaultMemberOutcome { - /// This transaction joined the member. - Joined(MemberRecord), - /// Already active; required discovery may still need recovery. - Active, - /// Left during this admission, or the admission token is no longer current. - Skipped, -} - /// Self-join `pubkey` to `channel_id` as [`put_member`] with no actor and no /// requested role (so the channel's `join_role` applies), but only if /// `pubkey` has never had a membership row in the channel, active or removed. @@ -748,12 +737,7 @@ pub async fn put_first_member( channel_id: Uuid, pubkey: &[u8], ) -> Result> { - Ok( - match put_first_member_with_pending(pool, community_id, channel_id, pubkey, None).await? { - DefaultMemberOutcome::Joined(member) => Some(member), - DefaultMemberOutcome::Active | DefaultMemberOutcome::Skipped => None, - }, - ) + put_first_member_with_pending(pool, community_id, channel_id, pubkey, None).await } async fn put_first_member_with_pending( @@ -762,7 +746,7 @@ async fn put_first_member_with_pending( channel_id: Uuid, pubkey: &[u8], pending: Option, -) -> Result { +) -> Result> { let mut tx = crate::begin_community_write_transaction( pool, community_id, @@ -783,7 +767,7 @@ async fn put_first_member_with_pending( .await?; let Some(created_at) = current else { tx.rollback().await?; - return Ok(DefaultMemberOutcome::Skipped); + return Ok(None); }; Some(created_at) } else { @@ -805,11 +789,7 @@ async fn put_first_member_with_pending( // 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(if removed_at.is_none() { - DefaultMemberOutcome::Active - } else { - DefaultMemberOutcome::Skipped - }); + return Ok(None); } } let target = MemberPut { @@ -823,12 +803,12 @@ async fn put_first_member_with_pending( match put { Ok(put) => { tx.commit().await?; - Ok(DefaultMemberOutcome::Joined(put.member)) + Ok(Some(put.member)) } Err(refused) => { tx.rollback().await?; if pending.is_some() && matches!(refused, DbError::AccessDenied(_)) { - Ok(DefaultMemberOutcome::Skipped) + Ok(None) } else { Err(refused) } @@ -2318,68 +2298,6 @@ impl Db { .await } - /// Persist an admission's add notice only while that admission and channel - /// membership are current. The channel lock spans the check and durable - /// insertion, ordering this notice before any subsequent leave/removal. - #[datastore_span(name = "insert_pending_default_notification", system = "postgresql")] - pub async fn insert_pending_default_notification( - &self, - community: CommunityId, - channel: Uuid, - pubkey: &[u8], - pending: Uuid, - event: &nostr::Event, - ) -> Result> { - let mut tx = self.begin_event_write_transaction(community).await?; - let current: Option = sqlx::query_scalar( - "SELECT auto_join_pending FROM relay_members \ - WHERE community_id = $1 AND pubkey = $2 AND auto_join_pending = $3 FOR SHARE", - ) - .bind(community.as_uuid()) - .bind(hex::encode(pubkey)) - .bind(pending) - .fetch_optional(tx.conn()) - .await?; - if current.is_none() { - tx.rollback().await?; - return Ok(None); - } - acquire_channel_membership_lock(tx.conn(), community, channel).await?; - if !is_member_in_transaction(&mut tx, channel, pubkey).await? { - tx.rollback().await?; - return Ok(None); - } - // The first event is the durable per-channel delivery identity. A - // retry uses its admission tag instead of minting a new timestamp/ID. - // Check under the same membership lock as the insert, so concurrent - // workers cannot both create the first notice. Include deleted rows: - // recovery must not resurrect a notice deliberately removed later. - let recorded: bool = sqlx::query_scalar( - "SELECT EXISTS (SELECT 1 FROM events \ - WHERE community_id = $1 AND pubkey = $2 AND kind = $3 AND tags @> $4)", - ) - .bind(community.as_uuid()) - .bind(event.pubkey.to_bytes().as_slice()) - .bind(buzz_core::kind::KIND_MEMBER_ADDED_NOTIFICATION as i32) - .bind(serde_json::json!([ - ["admission", pending.to_string()], - ["h", channel.to_string()], - ["p", hex::encode(pubkey)], - ])) - .fetch_one(tx.conn()) - .await?; - if recorded { - tx.rollback().await?; - return Ok(None); - } - let stored = crate::event::insert_event_in_transaction(&mut tx, event, None).await?; - if stored.1 { - crate::insert_mentions_in_transaction(&mut tx, event, None).await?; - } - tx.commit().await?; - Ok(Some(stored)) - } - /// 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")] @@ -2389,7 +2307,7 @@ impl Db { channel: Uuid, pubkey: &[u8], pending: Uuid, - ) -> Result { + ) -> Result> { put_first_member_with_pending(&self.pool, community, channel, pubkey, Some(pending)).await } 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 index 4689f1bf7df..5fd001fedf9 100644 --- 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 @@ -2,7 +2,7 @@ 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, notice_barrier}; + 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. @@ -120,137 +120,18 @@ mod postgres_tests { .expect("recovery must reach the held PostgreSQL lock"); } - async fn added_notifications( - state: &AppState, - community: buzz_core::CommunityId, - member_hex: &str, - ) -> Vec { - state - .db - .query_events(&buzz_db::EventQuery { - kinds: Some(vec![buzz_core::kind::KIND_MEMBER_ADDED_NOTIFICATION as i32]), - p_tag_hex: Some(member_hex.to_string()), - global_only: true, - ..buzz_db::EventQuery::for_community(community) - }) - .await - .unwrap() - .into_iter() - .map(|stored| stored.event) - .collect() - } - #[tokio::test] #[ignore = "requires Postgres"] async fn channel_local_failures_continue_later_defaults_and_retain_retry_work() { - for failure in ["membership", "announcement"] { - let (state, tenant, _, member, channels) = recovery_fixture(failure, 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(); - if failure == "membership" { - 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(); - } else { - sqlx::query("LOCK TABLE events IN SHARE MODE") - .execute(&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 announcements 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, - [ - if failure == "membership" { - None - } else { - Some("member".into()) - }, - Some("member".into()) - ], - "{failure}: 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 recovered_notices_keep_the_admission_identity_across_later_builder_clocks() { - let (state, tenant, _, member, channels) = recovery_fixture("notice-retry", 2).await; + let (state, tenant, _, member, channels) = recovery_fixture("membership", 2).await; let community = tenant.community(); let member_hex = member.public_key().to_hex(); - let first_clock = nostr::Timestamp::now().as_secs(); - sqlx::query( - "UPDATE relay_members SET created_at = $3 WHERE community_id = $1 AND pubkey = $2", - ) - .bind(community.as_uuid()) - .bind(&member_hex) - .bind( - chrono::DateTime::::from_timestamp(first_clock as i64 - 7200, 0).unwrap(), - ) - .execute(state.db.pool()) - .await - .unwrap(); let pending = state .db .pending_default_channel_join(community, &member_hex) .await .unwrap() .unwrap(); - let mut barrier = notice_barrier::install(&member.public_key().to_bytes()); - barrier.pin_created_at(first_clock); let mut blocker = state.db.pool().begin().await.unwrap(); let blocker_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") .fetch_one(&mut *blocker) @@ -258,44 +139,26 @@ mod postgres_tests { .unwrap(); sqlx::query("SELECT id FROM channels WHERE community_id = $1 AND id = $2 FOR UPDATE") .bind(community.as_uuid()) - .bind(channels[1]) + .bind(channels[0]) .fetch_one(&mut *blocker) .await .unwrap(); - let retry = || { + let retry = tokio::spawn({ let (state, tenant, member_hex) = (state.clone(), tenant.clone(), member_hex.clone()); - tokio::spawn(async move { - auto_join_default_channels(&tenant, &state, &member_hex, None).await - }) - }; - let first = retry(); - let (channel, release) = tokio::time::timeout(Duration::from_secs(10), barrier.next()) - .await - .unwrap() - .unwrap(); - assert_eq!(channel, channels[0]); - release.send(()).unwrap(); + 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), first) + tokio::time::timeout(Duration::from_secs(10), retry) .await .unwrap() .unwrap(); - let before = added_notifications(&state, community, &member_hex).await; - assert_eq!(before.len(), 1); - let membership_filter: nostr::Filter = serde_json::from_value(serde_json::json!({ - "kinds": [buzz_core::kind::KIND_MEMBER_ADDED_NOTIFICATION, buzz_core::kind::KIND_MEMBER_REMOVED_NOTIFICATION], - "#p": [member_hex], - "since": first_clock, - })).unwrap(); - assert!( - buzz_core::filter::filters_match( - &[membership_filter], - &buzz_core::StoredEvent::new(before[0].clone(), None), - ), - "a recovered notice must reach the membership subscription started after admission", + assert_eq!( + channel_roles(&state, community, &channels, &member).await, + [None, Some("member".into())], + "a local failure must not strand the later channel", ); - assert_eq!(before[0].created_at.as_secs(), first_clock); assert_eq!( state .db @@ -306,159 +169,23 @@ mod postgres_tests { .token, pending.token ); - - barrier.pin_created_at(first_clock + 3600); - let second = retry(); - for expected in &channels { - let (channel, release) = tokio::time::timeout(Duration::from_secs(10), barrier.next()) - .await - .unwrap() - .unwrap(); - assert_eq!(&channel, expected); - release.send(()).unwrap(); - } - tokio::time::timeout(Duration::from_secs(10), second) - .await - .unwrap() - .unwrap(); - let after = added_notifications(&state, community, &member_hex).await; + auto_join_default_channels(&tenant, &state, &member_hex, None).await; assert_eq!( - after.len(), - 2, - "retry must not persist another first-channel notice" + channel_roles(&state, community, &channels, &member).await, + [Some("member".into()), Some("member".into())] ); - assert!(after.iter().any(|notice| notice.id == before[0].id)); - for (index, channel) in channels.iter().enumerate() { - let channel = channel.to_string(); - let notices = after - .iter() - .filter(|notice| { - notice - .tags - .iter() - .any(|tag| tag.as_slice() == ["h", channel.as_str()]) - }) - .collect::>(); - assert_eq!( - notices.len(), - 1, - "each default channel needs exactly one persisted notice", - ); - assert_eq!( - notices[0].created_at.as_secs(), - first_clock + index as u64 * 3600 - ); - } - let token = pending.token.to_string(); - for notice in after { - assert!(notice - .tags - .iter() - .any(|tag| tag.as_slice() == ["admission", token.as_str()])); - } 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 recovered_notice_does_not_outlive_a_committed_leave_or_removal() { - for removal in ["leave", "kick", "relay-removal"] { - let (state, tenant, owner, member, channels) = recovery_fixture(removal, 1).await; - let community = tenant.community(); - let member_hex = member.public_key().to_hex(); - let member_bytes = member.public_key().to_bytes(); - let pending = state - .db - .pending_default_channel_join(community, &member_hex) - .await - .unwrap() - .unwrap(); - assert!(matches!( - state - .db - .put_pending_default_member( - community, - channels[0], - &member_bytes, - pending.token - ) - .await - .unwrap(), - buzz_db::channel_members::DefaultMemberOutcome::Joined(_) - )); - let mut barrier = notice_barrier::install(&member_bytes); - 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 } - }); - let (channel, release) = tokio::time::timeout(Duration::from_secs(10), barrier.next()) - .await - .unwrap() - .unwrap(); - assert_eq!(channel, channels[0]); - if removal == "relay-removal" { - let _ = tokio::time::timeout( - Duration::from_secs(10), - state - .db - .remove_relay_member(community, &member_hex, &state.relay_keypair), - ) - .await - .expect("roster removal must commit while the notice is parked") - .unwrap(); - assert!(!state - .db - .is_relay_member(community, &member_hex) - .await - .unwrap()); - } else { - let actor = if removal == "leave" { - member_bytes - } else { - owner.public_key().to_bytes() - }; - tokio::time::timeout( - Duration::from_secs(10), - state - .db - .remove_member(community, channel, &member_bytes, &actor), - ) - .await - .expect("channel removal must commit while the notice is parked") - .unwrap(); - assert!(!state - .db - .is_member(community, channel, &member_bytes) - .await - .unwrap()); - } - release.send(()).unwrap(); - tokio::time::timeout(Duration::from_secs(10), retry) - .await - .unwrap() - .unwrap(); - assert_eq!( - added_notification_count(&state, community, &member_hex).await, - 0, - "{removal}: the parked add notice became stale before its insert" - ); - assert!(state - .db - .pending_default_channel_join(community, &member_hex) - .await - .unwrap() - .is_none()); - } - } - - /// All roster admission paths commit recovery with membership. A failed list - /// leaves work behind; a repeated claim/add resumes it and a later leave stands. #[tokio::test] #[ignore = "requires Postgres"] async fn temporary_listing_failure_is_recovered_for_each_admission_path() { @@ -643,14 +370,12 @@ mod postgres_tests { .unwrap() .unwrap() .token; - assert!(matches!( - state - .db - .put_pending_default_member(community, a, &member.public_key().to_bytes(), old) - .await - .unwrap(), - buzz_db::channel_members::DefaultMemberOutcome::Joined(_) - )); + assert!(state + .db + .put_pending_default_member(community, a, &member.public_key().to_bytes(), old) + .await + .unwrap() + .is_some()); state .db .remove_member( @@ -695,14 +420,12 @@ mod postgres_tests { edit_channel(&state, &tenant, &owner, c, &[&["auto_join", "true"]]) .await .unwrap(); - assert!(matches!( - state - .db - .put_pending_default_member(community, c, &member.public_key().to_bytes(), old) - .await - .unwrap(), - buzz_db::channel_members::DefaultMemberOutcome::Skipped - )); + 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) @@ -719,14 +442,12 @@ mod postgres_tests { new ); assert!( - matches!( - state - .db - .put_pending_default_member(community, a, &member.public_key().to_bytes(), new,) - .await - .unwrap(), - buzz_db::channel_members::DefaultMemberOutcome::Joined(_) - ), + 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 @@ -756,102 +477,4 @@ mod postgres_tests { .unwrap() .is_none()); } - - /// A crash or failed discovery write after membership commits must recover the - /// client-visible roster, not just retain the already-correct membership row. - #[tokio::test] - #[ignore = "requires Postgres"] - async fn pending_join_recovers_failed_discovery_for_an_active_membership() { - use crate::handlers::group_state_postgres_tests::group_state_tags; - let host = format!("join-discovery-{}.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( - community, - &keys.public_key().to_hex(), - role, - 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(); - let pending = state - .db - .pending_default_channel_join(community, &member_hex) - .await - .unwrap() - .unwrap() - .token; - let _ = state - .db - .put_pending_default_member( - community, - channel, - &member.public_key().to_bytes(), - pending, - ) - .await - .unwrap(); - assert!(!group_state_tags(&state, &tenant, channel, 39002) - .await - .iter() - .any(|tag| tag[0] == "p" && tag[1] == member_hex)); - - let mut blocker = state.db.pool().begin().await.unwrap(); - sqlx::query("LOCK TABLE events IN SHARE MODE") - .execute(&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_blocked_query(state.db.pool(), "%events%").await; - tokio::time::timeout(Duration::from_secs(10), retry) - .await - .unwrap() - .unwrap(); - blocker.rollback().await.unwrap(); - assert!( - state - .db - .pending_default_channel_join(community, &member_hex) - .await - .unwrap() - .is_some(), - "failed discovery must leave recovery pending" - ); - auto_join_default_channels(&tenant, &state, &member_hex, None).await; - assert!(group_state_tags(&state, &tenant, channel, 39002) - .await - .iter() - .any(|tag| tag[0] == "p" && tag[1] == member_hex)); - assert_eq!( - added_notification_count(&state, community, &member_hex).await, - 1 - ); - assert!(state - .db - .pending_default_channel_join(community, &member_hex) - .await - .unwrap() - .is_none()); - } } diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 50b47d5c684..3de074602ef 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1342,7 +1342,6 @@ pub(crate) async fn emit_membership_notification_tagged( actor_pubkey, notification_kind, extra_tags, - None, ) .await } @@ -1359,7 +1358,6 @@ async fn emit_typed_membership_notification( actor_pubkey: &[u8], notification_kind: u32, extra_tags: &[Tag], - admission: Option<&buzz_db::relay_members::PendingDefaultChannelJoin>, ) -> anyhow::Result<()> { let target_hex = hex::encode(target_pubkey); let actor_hex = hex::encode(actor_pubkey); @@ -1400,40 +1398,16 @@ async fn emit_typed_membership_notification( Some(at) => builder.custom_created_at(nostr::Timestamp::from(at)), None => builder, }; - let builder = if let Some(admission) = admission { - builder.tags([Tag::parse(["admission", &admission.token.to_string()])?]) - } else { - builder - }; let event = builder .sign_with_keys(&state.relay_keypair) .map_err(|e| anyhow::anyhow!("failed to sign membership notification: {e}"))?; - // Recovery serializes the durable add with departures and the admission - // generation. Ordinary notices retain the existing insertion path. - let stored = match admission { - Some(admission) => { - state - .db - .insert_pending_default_notification( - tenant.community(), - channel_id, - target_pubkey, - admission.token, - &event, - ) - .await? - } - None => Some( - state - .db - .insert_event(tenant.community(), &event, None) - .await?, - ), - }; - let Some((stored, was_inserted)) = stored else { - return Ok(()); - }; + // Store with channel_id = None → globally scoped, reachable by global subscribers. + let (stored, was_inserted) = state + .db + .insert_event(tenant.community(), &event, None) + .await + .map_err(|e| anyhow::anyhow!("membership notification insert failed: {}", e.kind()))?; if !was_inserted { return Ok(()); } @@ -2229,7 +2203,6 @@ pub(crate) async fn persist_put_user( &actor_bytes, KIND_MEMBER_ADDED_NOTIFICATION, &[], - None, ) .await { @@ -2930,7 +2903,6 @@ async fn handle_delete_group( &actor_bytes, KIND_MEMBER_REMOVED_NOTIFICATION, &[], - None, ) .await { @@ -3015,40 +2987,13 @@ pub(crate) async fn join_channel_as_self( } /// The side effects of a committed self-join: the membership cache, the -/// `member_joined` kind:40099, kind:39000/39002 and kind:44100. +/// `member_joined` kind:40099, kind:39000/39002 and kind:44100. Only the +/// system message is fatal, as for kind:9021. async fn announce_self_join( tenant: &TenantContext, state: &Arc, channel_id: Uuid, pubkey: &[u8], -) -> anyhow::Result<()> { - emit_self_join_message(tenant, state, channel_id, pubkey).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"); - } - - if let Err(e) = emit_membership_notification( - tenant, - state, - channel_id, - pubkey, - pubkey, - buzz_core::kind::KIND_MEMBER_ADDED_NOTIFICATION, - ) - .await - { - warn!("membership notification for join failed: {e}"); - } - - Ok(()) -} - -async fn emit_self_join_message( - tenant: &TenantContext, - state: &Arc, - channel_id: Uuid, - pubkey: &[u8], ) -> anyhow::Result<()> { state.invalidate_membership(tenant, channel_id, pubkey); @@ -3066,37 +3011,23 @@ async fn emit_self_join_message( ) .await?; - Ok(()) -} + 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"); + } -async fn refresh_self_join_discovery( - tenant: &TenantContext, - state: &Arc, - channel_id: Uuid, - pubkey: &[u8], - admission: &buzz_db::relay_members::PendingDefaultChannelJoin, -) -> anyhow::Result<()> { - state.invalidate_membership(tenant, channel_id, pubkey); - emit_group_discovery_events(tenant, state, channel_id).await?; - let Some(channel_type) = state - .db - .channel_type_of(tenant.community(), channel_id) - .await? - else { - return Ok(()); - }; - emit_typed_membership_notification( + if let Err(e) = emit_membership_notification( tenant, state, channel_id, - &channel_type, pubkey, pubkey, - KIND_MEMBER_ADDED_NOTIFICATION, - &[], - Some(admission), + buzz_core::kind::KIND_MEMBER_ADDED_NOTIFICATION, ) - .await?; + .await + { + warn!("membership notification for join failed: {e}"); + } + Ok(()) } @@ -3198,15 +3129,16 @@ async fn finish_default_channel_join( .put_pending_default_member(tenant.community(), channel, pubkey, admission.token) .await { - Ok(buzz_db::channel_members::DefaultMemberOutcome::Joined(_)) => { - emit_self_join_message(tenant, state, channel, pubkey).await?; - refresh_self_join_discovery(tenant, state, channel, pubkey, admission).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); + } metrics::counter!("buzz_auto_join_channels_total").increment(1); } - Ok(buzz_db::channel_members::DefaultMemberOutcome::Active) => { - refresh_self_join_discovery(tenant, state, channel, pubkey, admission).await?; - } - Ok(buzz_db::channel_members::DefaultMemberOutcome::Skipped) => {} + Ok(None) => {} Err(buzz_db::DbError::ChannelNotFound(_)) => {} Err(error) => return Err(error.into()), } From 83e94f5a40da38bd3994d746c4d734d5907be8a1 Mon Sep 17 00:00:00 2001 From: Helen Tian Date: Thu, 5 Nov 2026 10:32:25 +1100 Subject: [PATCH 5/9] fix(relay): preserve roster-only operator additions Opt into durable default-channel work only from the existing kind:9030 path. Keep add_relay_member roster-only for buzz-admin and verify that authentication does not enroll its additions. Co-authored-by: Flipper <1ef64189596befbc8b93f9021bff35ae5d02b155d9e05fc124bca7349927d597@buzz.block.builderlab.xyz> Signed-off-by: Helen Tian --- crates/buzz-db/src/store/relay_members.rs | 39 ++++++++++++++++++- .../postgres_tests/recovery_postgres_tests.rs | 12 +++--- crates/buzz-relay/src/handlers/auth.rs | 35 +++++++++++++++-- crates/buzz-relay/src/handlers/relay_admin.rs | 4 +- 4 files changed, 77 insertions(+), 13 deletions(-) diff --git a/crates/buzz-db/src/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs index 20bfc047989..f2d5f9540bd 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, @@ -159,12 +172,13 @@ pub async fn add_relay_member( .await?; let result = sqlx::query( "INSERT INTO relay_members (community_id, pubkey, role, added_by, auto_join_pending) \ - VALUES ($1, $2, $3, $4, gen_random_uuid()) ON CONFLICT (community_id, pubkey) DO NOTHING", + 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; @@ -1333,6 +1347,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/postgres_tests/recovery_postgres_tests.rs b/crates/buzz-relay/src/api/invites/postgres_tests/recovery_postgres_tests.rs index 5fd001fedf9..8481240ea71 100644 --- 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 @@ -83,11 +83,11 @@ mod postgres_tests { ); let _ = state .db - .add_relay_member( + .add_relay_member_with_default_channels( community, &member.public_key().to_hex(), "member", - None, + &member.public_key().to_hex(), &state.relay_keypair, ) .await @@ -346,11 +346,11 @@ mod postgres_tests { for (keys, role) in [(&owner, "owner"), (&member, "member")] { let _ = state .db - .add_relay_member( + .add_relay_member_with_default_channels( community, &keys.public_key().to_hex(), role, - Some(&owner.public_key().to_hex()), + &owner.public_key().to_hex(), &state.relay_keypair, ) .await @@ -399,11 +399,11 @@ mod postgres_tests { .unwrap(); let _ = state .db - .add_relay_member( + .add_relay_member_with_default_channels( community, &member_hex, "member", - Some(&owner.public_key().to_hex()), + &owner.public_key().to_hex(), &state.relay_keypair, ) .await diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index 733c758ecef..cc6c09ac9a5 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/crates/buzz-relay/src/handlers/auth.rs @@ -3006,11 +3006,11 @@ mod tests { for keys in [&person, &agent] { let _ = state .db - .add_relay_member( + .add_relay_member_with_default_channels( community, &keys.public_key().to_hex(), "member", - Some("invite"), + "invite", &state.relay_keypair, ) .await @@ -3040,6 +3040,33 @@ mod tests { .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!( @@ -3104,11 +3131,11 @@ mod tests { if roster { let _ = state .db - .add_relay_member( + .add_relay_member_with_default_channels( community, &person.public_key().to_hex(), "member", - Some("invite"), + "invite", &state.relay_keypair, ) .await diff --git a/crates/buzz-relay/src/handlers/relay_admin.rs b/crates/buzz-relay/src/handlers/relay_admin.rs index 6831c269052..b514340a323 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 From ef5669c21d9656ee25770b43e6991dbb28326f6c Mon Sep 17 00:00:00 2001 From: Helen Tian Date: Thu, 5 Nov 2026 11:18:13 +1100 Subject: [PATCH 6/9] test(relay): seed pending admission in archive race Use the explicit default-channel admission entry for the joiner while keeping the owner roster-only. Preserve the archive lock barrier and all membership and notice assertions. Co-authored-by: Flipper <1ef64189596befbc8b93f9021bff35ae5d02b155d9e05fc124bca7349927d597@buzz.block.builderlab.xyz> Signed-off-by: Helen Tian --- crates/buzz-relay/src/api/invites.rs | 47 ++++++++++++++++------------ 1 file changed, 27 insertions(+), 20 deletions(-) diff --git a/crates/buzz-relay/src/api/invites.rs b/crates/buzz-relay/src/api/invites.rs index 4de96e46a6c..ef4cc044543 100644 --- a/crates/buzz-relay/src/api/invites.rs +++ b/crates/buzz-relay/src/api/invites.rs @@ -1566,10 +1566,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"] @@ -1597,20 +1596,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"); - } + state + .db + .add_relay_member( + community, + &owner.public_key().to_hex(), + "owner", + None, + &Keys::generate(), + ) + .await + .expect("seed relay owner"); + 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 @@ -1636,8 +1643,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) From 476cd335f191cd0f279b95381af73aff6c578ba3 Mon Sep 17 00:00:00 2001 From: Helen Tian Date: Thu, 5 Nov 2026 11:43:25 +1100 Subject: [PATCH 7/9] test(relay): explicitly discard fixture membership writes Keep snapshot fan-out intentionally unused in the isolated archive-race fixture, matching existing seed helpers and satisfying warnings-as-errors builds. Co-authored-by: Flipper <1ef64189596befbc8b93f9021bff35ae5d02b155d9e05fc124bca7349927d597@buzz.block.builderlab.xyz> Signed-off-by: Helen Tian --- crates/buzz-relay/src/api/invites.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/buzz-relay/src/api/invites.rs b/crates/buzz-relay/src/api/invites.rs index ef4cc044543..b66f26bcfd0 100644 --- a/crates/buzz-relay/src/api/invites.rs +++ b/crates/buzz-relay/src/api/invites.rs @@ -1596,7 +1596,7 @@ mod postgres_tests { let tenant = buzz_core::TenantContext::resolved(community, &host); let joiner = Keys::generate(); let joiner_hex = joiner.public_key().to_hex(); - state + let _ = state .db .add_relay_member( community, @@ -1607,7 +1607,7 @@ mod postgres_tests { ) .await .expect("seed relay owner"); - state + let _ = state .db .add_relay_member_with_default_channels( community, From b920d0092eb80c2fcbc8b3cb58a84a53a30b1cdf Mon Sep 17 00:00:00 2001 From: Helen Tian Date: Fri, 6 Nov 2026 10:02:20 +1100 Subject: [PATCH 8/9] fix(relay): read pending default-channel work from the replica on AUTH The recovery check runs on every AUTH admission, including roster communities that spawned nothing before. Reading it through the writer added a primary-pool acquire per connection. Route the AUTH read through the bounded replica path, as is_relay_member is. A stale answer only defers the work or carries an old token, and every write re-checks the token on the writer. Invite claims and admin adds keep the writer read, because they just committed the token. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Helen Tian --- crates/buzz-db/src/runtime/tests.rs | 59 +++++++++++++ crates/buzz-db/src/store/relay_members.rs | 82 ++++++++++++++++--- .../buzz-relay/src/handlers/side_effects.rs | 23 ++++-- 3 files changed, 146 insertions(+), 18 deletions(-) diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 7ef4cc6f957..6a14f5e3dca 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/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs index f2d5f9540bd..dfd52472a6b 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -1211,6 +1211,31 @@ pub struct PendingDefaultChannelJoin { 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")] @@ -1219,19 +1244,50 @@ impl Db { community: CommunityId, pubkey: &str, ) -> Result> { - let mut conn = observability::acquire_writer( - &self.pool, - observability::WriterOperation::Authorization, - ) - .await?; - 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(&mut *conn) - .await?) + 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. diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 3de074602ef..32ae26b40f3 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -3052,11 +3052,24 @@ async fn finish_default_channel_joins( pubkey_hex: &str, gate: Option<&Arc>, ) -> anyhow::Result<()> { - let Some(admission) = state - .db - .pending_default_channel_join(tenant.community(), pubkey_hex) - .await? - else { + // 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? + } + }; + let Some(admission) = pending else { return Ok(()); }; let pubkey = hex::decode(pubkey_hex)?; From 81273906fac72ad475ee566bda0755dd1c4b02fc Mon Sep 17 00:00:00 2001 From: Helen Tian Date: Fri, 6 Nov 2026 13:39:31 +1100 Subject: [PATCH 9/9] test(db): check listed-name migration independently of later versions Assert that migration 0083 has a successful ledger entry after startup migrations complete. Later migrations, including default-channel recovery 0084, must not invalidate the listed-name index tests. Co-authored-by: Codex Signed-off-by: Helen Tian --- .../tests/listed_channel_name_postgres_tests.rs | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) 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"); }