From efb2ca9377f9e69086a3f1c0fbfb7bc8a63b9e7c Mon Sep 17 00:00:00 2001 From: npub16ynjrn8tm9gxg4kpgfgj24eup8met48z58m9g0cvmpq9msjht0lq7r96r6 Date: Mon, 13 Jul 2026 15:35:30 -0700 Subject: [PATCH 1/6] Add durable community archival Co-authored-by: npub16ynjrn8tm9gxg4kpgfgj24eup8met48z58m9g0cvmpq9msjht0lq7r96r6 Signed-off-by: npub16ynjrn8tm9gxg4kpgfgj24eup8met48z58m9g0cvmpq9msjht0lq7r96r6 --- crates/buzz-db/src/channel.rs | 1 + crates/buzz-db/src/event.rs | 1 + crates/buzz-db/src/lib.rs | 68 +++++++++++++++++++++++++- crates/buzz-db/src/migration.rs | 2 +- crates/buzz-db/src/workflow.rs | 16 +++--- crates/buzz-pubsub/src/conn_control.rs | 9 ++++ crates/buzz-relay/src/api/operator.rs | 55 ++++++++++++++++++++- crates/buzz-relay/src/main.rs | 5 ++ crates/buzz-relay/src/router.rs | 4 ++ crates/buzz-relay/src/state.rs | 57 +++++++++++++++++++++ migrations/0016_community_archival.sql | 3 ++ schema/schema.sql | 1 + 12 files changed, 212 insertions(+), 10 deletions(-) create mode 100644 migrations/0016_community_archival.sql diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index f1e796a0b2..c522d27bc2 100644 --- a/crates/buzz-db/src/channel.rs +++ b/crates/buzz-db/src/channel.rs @@ -1345,6 +1345,7 @@ pub async fn reap_expired_ephemeral_channels(pool: &PgPool) -> Result, + /// When the community was archived; absent while active. + pub archived_at: Option>, +} + +/// Community row returned by an owner-authorized archive operation. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ArchivedCommunityRecord { + /// Stable server-resolved community id. + pub id: CommunityId, + /// Reserved canonical host. + pub host: String, + /// Durable first-archive timestamp. + pub archived_at: DateTime, } /// Token summary returned by [`Db::list_active_tokens`]. @@ -407,6 +420,7 @@ impl Db { SELECT id, host FROM communities WHERE lower(host) = lower($1) + AND archived_at IS NULL "#, ) .bind(normalized_host) @@ -425,6 +439,24 @@ impl Db { .transpose() } + /// Returns a community by host regardless of lifecycle state. Operator-plane only. + pub async fn lookup_community_by_host_for_management( + &self, + normalized_host: &str, + ) -> Result> { + let row = sqlx::query("SELECT id, host FROM communities WHERE lower(host) = lower($1)") + .bind(normalized_host) + .fetch_optional(&self.pool) + .await?; + row.map(|row| { + Ok(CommunityRecord { + id: CommunityId::from_uuid(row.try_get("id")?), + host: row.try_get("host")?, + }) + }) + .transpose() + } + /// Lists communities where `owner_pubkey` currently holds the `owner` role. /// /// This is an operator-plane helper, not a tenant-scoped data-plane read: @@ -436,7 +468,7 @@ impl Db { let owner_pubkey = owner_pubkey.to_ascii_lowercase(); let rows = sqlx::query( r#" - SELECT c.id, c.host, c.created_at + SELECT c.id, c.host, c.created_at, c.archived_at FROM communities c JOIN relay_members rm ON rm.community_id = c.id WHERE rm.pubkey = $1 @@ -453,10 +485,12 @@ impl Db { let id: Uuid = row.try_get("id")?; let host: String = row.try_get("host")?; let created_at: DateTime = row.try_get("created_at")?; + let archived_at: Option> = row.try_get("archived_at")?; Ok(OwnedCommunityRecord { id: CommunityId::from_uuid(id), host, created_at, + archived_at, }) }) .collect() @@ -478,6 +512,7 @@ impl Db { SELECT host FROM communities WHERE id = $1 + AND archived_at IS NULL "#, ) .bind(community_id.as_uuid()) @@ -632,6 +667,7 @@ impl Db { WHERE lower(c.host) = lower($1) AND lower(rm.pubkey) = lower($2) AND rm.role = 'owner' + AND c.archived_at IS NULL "#, ) .bind(normalized_host) @@ -654,6 +690,36 @@ impl Db { )) } + /// Idempotently archives a community when the asserted pubkey is its current owner. + pub async fn archive_community_owned_by( + &self, + normalized_host: &str, + owner_pubkey: &str, + ) -> Result> { + let row = sqlx::query( + r#"UPDATE communities c + SET archived_at = COALESCE(c.archived_at, now()) + FROM relay_members rm + WHERE lower(c.host) = lower($1) + AND rm.community_id = c.id + AND lower(rm.pubkey) = lower($2) + AND rm.role = 'owner' + RETURNING c.id, c.host, c.archived_at"#, + ) + .bind(normalized_host) + .bind(owner_pubkey) + .fetch_optional(&self.pool) + .await?; + row.map(|row| { + Ok(ArchivedCommunityRecord { + id: CommunityId::from_uuid(row.try_get("id")?), + host: row.try_get("host")?, + archived_at: row.try_get("archived_at")?, + }) + }) + .transpose() + } + /// Returns the community that owns a channel, if the channel exists. /// /// Internal relay producers use this to derive tenant context from the row diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 0147572515..3d4cb32835 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -548,7 +548,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 15); + assert_eq!(migrations.len(), 16); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index 2ff90280ae..9c02f162c9 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -457,13 +457,15 @@ pub async fn list_enabled_channel_workflows( pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result> { let rows = sqlx::query( r#" - SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, - status::text AS status, enabled, created_at, updated_at - FROM workflows - WHERE status = 'active' - AND enabled = TRUE - AND definition->'trigger'->>'on' = 'schedule' - ORDER BY created_at ASC + SELECT w.id, w.community_id, w.name, w.owner_pubkey, w.channel_id, w.definition, w.definition_hash, + w.status::text AS status, w.enabled, w.created_at, w.updated_at + FROM workflows w + JOIN communities c ON c.id = w.community_id + WHERE w.status = 'active' + AND w.enabled = TRUE + AND w.definition->'trigger'->>'on' = 'schedule' + AND c.archived_at IS NULL + ORDER BY w.created_at ASC LIMIT $1 "#, ) diff --git a/crates/buzz-pubsub/src/conn_control.rs b/crates/buzz-pubsub/src/conn_control.rs index a562980a79..099027fc8a 100644 --- a/crates/buzz-pubsub/src/conn_control.rs +++ b/crates/buzz-pubsub/src/conn_control.rs @@ -54,6 +54,8 @@ pub fn parse_conn_control_channel(channel: &str) -> Option { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(tag = "op")] pub enum ConnControl { + /// Disconnect every live socket bound to the carrying community. + DisconnectCommunity, /// Disconnect every live connection authenticated as `pubkey` in the /// carrying community — live ban enforcement. `pubkey` is 32 raw bytes. /// `event_id` and `reason` reproduce the same NIP-01 `OK` frame the origin @@ -197,6 +199,13 @@ mod tests { assert_eq!(parse_conn_control_channel(&extended), None); } + #[test] + fn disconnect_community_command_serde_round_trips() { + let cmd = ConnControl::DisconnectCommunity; + let json = serde_json::to_string(&cmd).unwrap(); + assert_eq!(serde_json::from_str::(&json).unwrap(), cmd); + } + #[test] fn disconnect_command_serde_round_trips() { let cmd = ConnControl::DisconnectPubkey { diff --git a/crates/buzz-relay/src/api/operator.rs b/crates/buzz-relay/src/api/operator.rs index 55be86cba5..91693dda70 100644 --- a/crates/buzz-relay/src/api/operator.rs +++ b/crates/buzz-relay/src/api/operator.rs @@ -192,6 +192,58 @@ pub async fn provision_community( } } +/// Owner assertion supplied by the trusted operator client. +#[derive(Debug, Deserialize)] +pub struct ArchiveCommunityRequest { + host: String, + owner_pubkey: String, +} + +/// Idempotently archive a community owned by the asserted end-user identity. +pub async fn archive_community( + State(state): State>, + headers: HeaderMap, + body: axum::body::Bytes, +) -> Result, (StatusCode, Json)> { + const PATH: &str = "/operator/communities/archive"; + authorize_operator_request(&state, &headers, "POST", PATH, None, Some(&body)).await?; + let request: ArchiveCommunityRequest = serde_json::from_slice(&body).map_err(|e| { + api_error( + StatusCode::BAD_REQUEST, + &format!("invalid archive-community JSON: {e}"), + ) + })?; + let normalized_host = normalize_candidate_host(&request.host) + .map_err(|msg| api_error(StatusCode::BAD_REQUEST, &msg))?; + if normalized_host == buzz_core::tenant::relay_url_authority(&state.config.relay_url) { + return Err(api_error( + StatusCode::CONFLICT, + "the deployment community cannot be archived", + )); + } + let owner = validate_pubkey_hex(&request.owner_pubkey).ok_or_else(|| { + api_error( + StatusCode::BAD_REQUEST, + "invalid owner_pubkey: expected 64-char hex pubkey", + ) + })?; + let record = state + .db + .archive_community_owned_by(&normalized_host, &owner) + .await + .map_err(|e| internal_error(&format!("archive community: {e}")))? + .ok_or_else(|| api_error(StatusCode::NOT_FOUND, "community not found"))?; + let tenant = TenantContext::resolved(record.id, &record.host); + let closed = state.disconnect_community_clusterwide(&tenant); + tracing::info!(community = %record.id, host = %record.host, local_connections_closed = closed, "community archived"); + Ok(Json(serde_json::json!({ + "community_id": record.id.to_string(), + "host": record.host, + "archived_at": record.archived_at, + "status": "archived", + }))) +} + /// List communities where a pubkey currently holds the `owner` role. pub async fn list_owned_communities( State(state): State>, @@ -228,6 +280,7 @@ pub async fn list_owned_communities( "community_id": row.id.to_string(), "host": row.host, "created_at": row.created_at, + "archived_at": row.archived_at, })).collect::>(), }))) } @@ -378,7 +431,7 @@ pub async fn community_availability( .map_err(|msg| api_error(StatusCode::BAD_REQUEST, &msg))?; let existing = state .db - .lookup_community_by_host(&normalized_host) + .lookup_community_by_host_for_management(&normalized_host) .await .map_err(|e| internal_error(&format!("check community availability: {e}")))?; diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index df68ec2700..978125815b 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -794,6 +794,11 @@ async fn main() -> anyhow::Result<()> { loop { match rx.recv().await { Ok(scoped) => match scoped.command { + buzz_pubsub::conn_control::ConnControl::DisconnectCommunity => { + state_for_conn_ctrl + .conn_manager + .disconnect_community(scoped.community_id); + } buzz_pubsub::conn_control::ConnControl::DisconnectPubkey { pubkey, event_id, diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index 902a0ac1b4..a7836b678d 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -64,6 +64,10 @@ pub fn build_router(state: Arc) -> Router { "/operator/communities", get(api::operator::list_owned_communities).post(api::operator::provision_community), ) + .route( + "/operator/communities/archive", + post(api::operator::archive_community), + ) .route( "/operator/communities/availability", get(api::operator::community_availability), diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index b772745f81..e60bae226b 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -195,6 +195,18 @@ impl ConnectionManager { closed } + /// Disconnect every live socket bound to one community. + pub fn disconnect_community(&self, community: CommunityId) -> usize { + let mut closed = 0; + for entry in self.connections.iter() { + if entry.community_id == community { + entry.cancel.cancel(); + closed += 1; + } + } + closed + } + /// Return the server-resolved community that the connection's host bound to. pub fn community_for_conn(&self, conn_id: Uuid) -> Option { self.connections @@ -811,6 +823,22 @@ impl AppState { closed } + /// Disconnect a community locally and publish the command to every relay pod. + pub fn disconnect_community_clusterwide(&self, tenant: &TenantContext) -> usize { + let closed = self.conn_manager.disconnect_community(tenant.community()); + let pubsub = Arc::clone(&self.pubsub); + let tenant = tenant.clone(); + tokio::spawn(async move { + if let Err(e) = pubsub + .publish_conn_control(&tenant, &ConnControl::DisconnectCommunity) + .await + { + tracing::warn!(community = %tenant.community(), error = %e, "failed to publish community disconnect"); + } + }); + closed + } + /// Get accessible channel IDs with a 10-second cache. Falls back to DB on miss. pub async fn get_accessible_channel_ids_cached( &self, @@ -1296,6 +1324,35 @@ mod tests { ); } + #[tokio::test] + async fn disconnect_community_closes_only_matching_tenant() { + let mgr = ConnectionManager::new(); + let community_a = CommunityId::from_uuid(Uuid::from_u128(0xa)); + let community_b = CommunityId::from_uuid(Uuid::from_u128(0xb)); + let register = |community| { + let conn_id = Uuid::new_v4(); + let (tx, _rx) = mpsc::channel(8); + let (ctrl_tx, _ctrl_rx) = mpsc::channel(8); + let cancel = CancellationToken::new(); + mgr.register( + conn_id, + tx, + ctrl_tx, + cancel.clone(), + community, + Arc::new(AtomicU8::new(0)), + Arc::new(Mutex::new(HashMap::new())), + 3, + ); + cancel + }; + let cancel_a = register(community_a); + let cancel_b = register(community_b); + assert_eq!(mgr.disconnect_community(community_a), 1); + assert!(cancel_a.is_cancelled()); + assert!(!cancel_b.is_cancelled()); + } + #[tokio::test] async fn disconnect_pubkey_closes_matching_conns_with_reason() { let (mgr, id, _rx, mut ctrl_rx, cancel, _bp) = setup_conn(8); diff --git a/migrations/0016_community_archival.sql b/migrations/0016_community_archival.sql new file mode 100644 index 0000000000..823c38bf7a --- /dev/null +++ b/migrations/0016_community_archival.sql @@ -0,0 +1,3 @@ +-- Durable community archival state. Archived hosts remain reserved by the existing +-- full unique index and continue to count toward owner quotas. +ALTER TABLE communities ADD COLUMN archived_at TIMESTAMPTZ; diff --git a/schema/schema.sql b/schema/schema.sql index 0ba06258c8..ccba36c2fc 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -55,6 +55,7 @@ CREATE TABLE communities ( host VARCHAR(255) NOT NULL, signing_key BYTEA, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + archived_at TIMESTAMPTZ, CONSTRAINT chk_communities_id_not_nil CHECK (id <> '00000000-0000-0000-0000-000000000000'::uuid) ); From e43e04d6b19df74794c7a2fca4282a342117e72d Mon Sep 17 00:00:00 2001 From: npub16ynjrn8tm9gxg4kpgfgj24eup8met48z58m9g0cvmpq9msjht0lq7r96r6 Date: Mon, 13 Jul 2026 15:37:18 -0700 Subject: [PATCH 2/6] Harden archive rollout behavior Co-authored-by: npub16ynjrn8tm9gxg4kpgfgj24eup8met48z58m9g0cvmpq9msjht0lq7r96r6 Signed-off-by: npub16ynjrn8tm9gxg4kpgfgj24eup8met48z58m9g0cvmpq9msjht0lq7r96r6 --- crates/buzz-db/src/lib.rs | 3 +++ crates/buzz-pubsub/src/conn_control.rs | 10 ++++++++++ crates/buzz-relay/src/api/operator.rs | 5 +++-- 3 files changed, 16 insertions(+), 2 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 411c37365c..17506468a2 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -695,6 +695,7 @@ impl Db { &self, normalized_host: &str, owner_pubkey: &str, + protected_deployment_host: &str, ) -> Result> { let row = sqlx::query( r#"UPDATE communities c @@ -704,10 +705,12 @@ impl Db { AND rm.community_id = c.id AND lower(rm.pubkey) = lower($2) AND rm.role = 'owner' + AND lower(c.host) <> lower($3) RETURNING c.id, c.host, c.archived_at"#, ) .bind(normalized_host) .bind(owner_pubkey) + .bind(protected_deployment_host) .fetch_optional(&self.pool) .await?; row.map(|row| { diff --git a/crates/buzz-pubsub/src/conn_control.rs b/crates/buzz-pubsub/src/conn_control.rs index 099027fc8a..bc177cff13 100644 --- a/crates/buzz-pubsub/src/conn_control.rs +++ b/crates/buzz-pubsub/src/conn_control.rs @@ -206,6 +206,16 @@ mod tests { assert_eq!(serde_json::from_str::(&json).unwrap(), cmd); } + #[test] + fn unknown_command_is_rejected_without_affecting_later_messages() { + assert!(serde_json::from_str::(r#"{"op":"FutureCommand"}"#).is_err()); + let known = serde_json::to_string(&ConnControl::DisconnectCommunity).unwrap(); + assert_eq!( + serde_json::from_str::(&known).unwrap(), + ConnControl::DisconnectCommunity + ); + } + #[test] fn disconnect_command_serde_round_trips() { let cmd = ConnControl::DisconnectPubkey { diff --git a/crates/buzz-relay/src/api/operator.rs b/crates/buzz-relay/src/api/operator.rs index 91693dda70..d3413d5aa8 100644 --- a/crates/buzz-relay/src/api/operator.rs +++ b/crates/buzz-relay/src/api/operator.rs @@ -215,7 +215,8 @@ pub async fn archive_community( })?; let normalized_host = normalize_candidate_host(&request.host) .map_err(|msg| api_error(StatusCode::BAD_REQUEST, &msg))?; - if normalized_host == buzz_core::tenant::relay_url_authority(&state.config.relay_url) { + let deployment_host = buzz_core::tenant::relay_url_authority(&state.config.relay_url); + if normalized_host == deployment_host { return Err(api_error( StatusCode::CONFLICT, "the deployment community cannot be archived", @@ -229,7 +230,7 @@ pub async fn archive_community( })?; let record = state .db - .archive_community_owned_by(&normalized_host, &owner) + .archive_community_owned_by(&normalized_host, &owner, &deployment_host) .await .map_err(|e| internal_error(&format!("archive community: {e}")))? .ok_or_else(|| api_error(StatusCode::NOT_FOUND, "community not found"))?; From a6565e07245db51bd48b455f5a203fc1ec523886 Mon Sep 17 00:00:00 2001 From: npub1ks3yhswfl2z70myka7wltdg39qpvqhspt2w3l2ajtaee8tmz5qkqrvl4et Date: Mon, 13 Jul 2026 16:05:16 -0700 Subject: [PATCH 3/6] Fix stale Postgres-ignored test baselines The archival migration bumped the migration count to 12, but the pre-0007 retry test still asserted the last applied version was 11. list_communities_owned_by_returns_only_owner_rows used fixed shared pubkeys, so it counted communities leaked by sibling ignored tests on a shared database; use per-run unique pubkeys instead. Both failures reproduced on a fresh fully-migrated database and are test-baseline issues, not archival logic defects. 79/79 buzz-db and 11/11 buzz-relay ignored tests pass after this change. Co-authored-by: npub1ks3yhswfl2z70myka7wltdg39qpvqhspt2w3l2ajtaee8tmz5qkqrvl4et Signed-off-by: npub1ks3yhswfl2z70myka7wltdg39qpvqhspt2w3l2ajtaee8tmz5qkqrvl4et --- crates/buzz-db/src/lib.rs | 9 +++++++-- crates/buzz-db/src/migration.rs | 2 +- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 17506468a2..697ef5bc4c 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -4197,8 +4197,13 @@ mod tests { let community_a = CommunityId::from_uuid(make_community(&db.pool).await); let community_b = CommunityId::from_uuid(make_community(&db.pool).await); let community_c = CommunityId::from_uuid(make_community(&db.pool).await); - let owner = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; - let other = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; + // Unique per run: `list_communities_owned_by` is keyed only by pubkey, + // so a shared fixed pubkey picks up communities leaked by sibling + // ignored tests running against the same database. + let owner = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple()); + let owner = owner.as_str(); + let other = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple()); + let other = other.as_str(); db.bootstrap_owner(community_a, owner) .await diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 3d4cb32835..e4cb57322c 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -1005,7 +1005,7 @@ mod tests { run_migrations(&pool) .await .expect("retry succeeds after operator repair"); - assert_eq!(applied_versions(&pool).await.last().copied(), Some(15)); + assert_eq!(applied_versions(&pool).await.last().copied(), Some(16)); } #[tokio::test] From b383b1ff459c3dca1e7d027f0f54d9bd682dab07 Mon Sep 17 00:00:00 2001 From: npub16ynjrn8tm9gxg4kpgfgj24eup8met48z58m9g0cvmpq9msjht0lq7r96r6 Date: Mon, 13 Jul 2026 16:27:47 -0700 Subject: [PATCH 4/6] Close community archival socket lifecycle races Co-authored-by: npub16ynjrn8tm9gxg4kpgfgj24eup8met48z58m9g0cvmpq9msjht0lq7r96r6 Signed-off-by: npub16ynjrn8tm9gxg4kpgfgj24eup8met48z58m9g0cvmpq9msjht0lq7r96r6 --- crates/buzz-db/src/lib.rs | 11 + crates/buzz-relay/src/api/operator.rs | 117 ++++++++- crates/buzz-relay/src/audio/handler.rs | 57 +++-- crates/buzz-relay/src/connection.rs | 29 ++- crates/buzz-relay/src/main.rs | 28 ++- crates/buzz-relay/src/state.rs | 318 ++++++++++++++++++++++--- 6 files changed, 502 insertions(+), 58 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 697ef5bc4c..f02d3dd5c4 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -439,6 +439,17 @@ impl Db { .transpose() } + /// Returns whether a community id still exists in the active lifecycle state. + pub async fn is_community_active(&self, community_id: CommunityId) -> Result { + let active = sqlx::query_scalar::<_, bool>( + "SELECT EXISTS(SELECT 1 FROM communities WHERE id = $1 AND archived_at IS NULL)", + ) + .bind(community_id.as_uuid()) + .fetch_one(&self.pool) + .await?; + Ok(active) + } + /// Returns a community by host regardless of lifecycle state. Operator-plane only. pub async fn lookup_community_by_host_for_management( &self, diff --git a/crates/buzz-relay/src/api/operator.rs b/crates/buzz-relay/src/api/operator.rs index d3413d5aa8..3b1596b90a 100644 --- a/crates/buzz-relay/src/api/operator.rs +++ b/crates/buzz-relay/src/api/operator.rs @@ -235,7 +235,23 @@ pub async fn archive_community( .map_err(|e| internal_error(&format!("archive community: {e}")))? .ok_or_else(|| api_error(StatusCode::NOT_FOUND, "community not found"))?; let tenant = TenantContext::resolved(record.id, &record.host); - let closed = state.disconnect_community_clusterwide(&tenant); + let closed = match state.disconnect_community_clusterwide(&tenant).await { + Ok(closed) => closed, + Err(error) => { + tracing::warn!(community = %record.id, host = %record.host, %error, "community archived but disconnect propagation is pending"); + return Err(( + StatusCode::SERVICE_UNAVAILABLE, + Json(serde_json::json!({ + "community_id": record.id.to_string(), + "host": record.host, + "archived_at": record.archived_at, + "status": "archived", + "propagation": "pending", + "error": "connection propagation pending — retry this request", + })), + )); + } + }; tracing::info!(community = %record.id, host = %record.host, local_connections_closed = closed, "community archived"); Ok(Json(serde_json::json!({ "community_id": record.id.to_string(), @@ -782,6 +798,105 @@ mod tests { ); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn archive_publish_failure_is_retryable_and_preserves_timestamp() { + let operator = Keys::generate(); + let owner = Keys::generate(); + let Some(state) = operator_test_state(std::slice::from_ref(&operator)).await else { + return; + }; + let host = format!("community-{}.example", Uuid::new_v4().simple()); + let owner_hex = owner.public_key().to_hex(); + let create_body = serde_json::json!({ + "host": host, + "initial_owner_pubkey": owner_hex, + "create_only": true, + }) + .to_string(); + let create_url = format!("http://{INGRESS_HOST}/operator/communities"); + let create_auth = + nip98_auth_header(&operator, &create_url, "POST", Some(create_body.as_bytes())); + let create_response = build_router(Arc::clone(&state)) + .oneshot( + Request::builder() + .method("POST") + .uri("/operator/communities") + .header(header::HOST, INGRESS_HOST) + .header(header::AUTHORIZATION, create_auth) + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(create_body)) + .expect("create request"), + ) + .await + .expect("create response"); + assert_eq!(create_response.status(), StatusCode::OK); + + let archive_body = serde_json::json!({ + "host": host, + "owner_pubkey": owner.public_key().to_hex(), + }) + .to_string(); + let archive_url = format!("http://{INGRESS_HOST}/operator/communities/archive"); + let archive_once = |state: Arc| { + let auth = nip98_auth_header( + &operator, + &archive_url, + "POST", + Some(archive_body.as_bytes()), + ); + let body = archive_body.clone(); + async move { + build_router(state) + .oneshot( + Request::builder() + .method("POST") + .uri("/operator/communities/archive") + .header(header::HOST, INGRESS_HOST) + .header(header::AUTHORIZATION, auth) + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(body)) + .expect("archive request"), + ) + .await + .expect("archive response") + } + }; + + let first = archive_once(Arc::clone(&state)).await; + assert_eq!(first.status(), StatusCode::SERVICE_UNAVAILABLE); + let first_json = read_json(first).await; + assert_eq!(first_json["status"], "archived"); + assert_eq!(first_json["propagation"], "pending"); + let first_archived_at = first_json["archived_at"].clone(); + assert!(!first_archived_at.is_null()); + assert!(state + .db + .lookup_community_by_host(&host) + .await + .expect("active lookup") + .is_none()); + + let second = archive_once(Arc::clone(&state)).await; + assert_eq!(second.status(), StatusCode::SERVICE_UNAVAILABLE); + let second_json = read_json(second).await; + assert_eq!(second_json["archived_at"], first_archived_at); + + let owned = state + .db + .list_communities_owned_by(&owner.public_key().to_hex()) + .await + .expect("owned communities"); + let row = owned + .iter() + .find(|row| row.host == host) + .expect("archived row"); + assert_eq!( + serde_json::to_value(row.archived_at).expect("timestamp JSON"), + first_archived_at + ); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn happy_path_create_returns_created_and_bootstraps_owner() { diff --git a/crates/buzz-relay/src/audio/handler.rs b/crates/buzz-relay/src/audio/handler.rs index 9e930ccc40..e1d422ac5c 100644 --- a/crates/buzz-relay/src/audio/handler.rs +++ b/crates/buzz-relay/src/audio/handler.rs @@ -38,7 +38,7 @@ use buzz_core::StoredEvent; use buzz_pubsub::EventTopic; use crate::audio::room::PeerCtrl; -use crate::state::AppState; +use crate::state::{run_registered_community_connection, AppState}; /// Maximum binary frame size: 4 KB is generous for a single Opus packet. const MAX_AUDIO_FRAME_BYTES: usize = 4096; @@ -112,6 +112,29 @@ async fn handle_audio_connection( state: Arc, tenant: TenantContext, channel_id: Uuid, +) { + let cancel = CancellationToken::new(); + let community_id = tenant.community(); + let registry = Arc::clone(&state.community_connections); + let check_state = Arc::clone(&state); + let run_state = Arc::clone(&state); + run_registered_community_connection( + ®istry, + Uuid::new_v4(), + community_id, + cancel.clone(), + move || async move { check_state.db.is_community_active(community_id).await }, + move || handle_active_audio_connection(socket, run_state, tenant, channel_id, cancel), + ) + .await; +} + +async fn handle_active_audio_connection( + socket: WebSocket, + state: Arc, + tenant: TenantContext, + channel_id: Uuid, + cancel: CancellationToken, ) { let (mut ws_send, mut ws_recv) = socket.split(); @@ -126,23 +149,26 @@ async fn handle_audio_connection( return; } - let auth_result = tokio::time::timeout(AUTH_TIMEOUT, async { - while let Some(Ok(msg)) = ws_recv.next().await { - if let WsMessage::Text(text) = msg { - if text.len() > MAX_TEXT_FRAME_BYTES { - warn!(channel_id = %channel_id, "auth text frame too large — dropping"); - continue; - } - if let Ok(auth) = serde_json::from_str::(&text) { - if auth.msg_type == "auth" { - return Some(auth); + let auth_result = tokio::select! { + biased; + _ = cancel.cancelled() => return, + result = tokio::time::timeout(AUTH_TIMEOUT, async { + while let Some(Ok(msg)) = ws_recv.next().await { + if let WsMessage::Text(text) = msg { + if text.len() > MAX_TEXT_FRAME_BYTES { + warn!(channel_id = %channel_id, "auth text frame too large — dropping"); + continue; + } + if let Ok(auth) = serde_json::from_str::(&text) { + if auth.msg_type == "auth" { + return Some(auth); + } } } } - } - None - }) - .await; + None + }) => result, + }; let auth_msg = match auth_result { Ok(Some(a)) => a, @@ -395,7 +421,6 @@ async fn handle_audio_connection( ) .await; - let cancel = CancellationToken::new(); let missed_pongs = Arc::new(AtomicU8::new(0)); // Dual-channel pattern (matches connection.rs): data channel for audio, diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 294c3b1b4e..5c53d53ef4 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -20,7 +20,7 @@ use nostr::Filter; use crate::handlers; use crate::protocol::{ClientMessage, RelayMessage}; -use crate::state::AppState; +use crate::state::{run_registered_community_connection, AppState}; use buzz_pubsub::EventTopic; /// Maximum time a new socket may hold a connection slot without completing NIP-42 auth. @@ -120,6 +120,31 @@ pub async fn handle_connection( state: Arc, addr: SocketAddr, tenant: TenantContext, +) { + let conn_id = Uuid::new_v4(); + let cancel = CancellationToken::new(); + let community_id = tenant.community(); + let registry = Arc::clone(&state.community_connections); + let check_state = Arc::clone(&state); + let run_state = Arc::clone(&state); + run_registered_community_connection( + ®istry, + conn_id, + community_id, + cancel.clone(), + move || async move { check_state.db.is_community_active(community_id).await }, + move || handle_active_connection(socket, run_state, addr, tenant, conn_id, cancel), + ) + .await; +} + +async fn handle_active_connection( + socket: WebSocket, + state: Arc, + addr: SocketAddr, + tenant: TenantContext, + conn_id: Uuid, + cancel: CancellationToken, ) { let permit = match state.conn_semaphore.clone().try_acquire_owned() { Ok(p) => p, @@ -129,9 +154,7 @@ pub async fn handle_connection( } }; - let conn_id = Uuid::new_v4(); let challenge = generate_challenge(); - let cancel = CancellationToken::new(); let (tx, rx) = mpsc::channel::(state.config.send_buffer_size); // Control channel for Pong/Close — small capacity, guaranteed delivery diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 978125815b..c7814033e5 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -781,6 +781,32 @@ async fn main() -> anyhow::Result<()> { }); } + // Durable lifecycle backstop: Redis pub/sub cannot deliver to a pod that was + // offline. Periodically revalidate only communities with local live sockets + // so missed archive commands still converge without a global DB scan. + { + let lifecycle_state = Arc::clone(&state); + let interval_secs = std::env::var("BUZZ_COMMUNITY_REVALIDATE_INTERVAL_SECS") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(30) + .clamp(1, 300); + tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + interval.tick().await; + let closed = lifecycle_state.revalidate_live_communities().await; + if closed > 0 { + tracing::info!( + closed, + "closed sockets for inactive communities during lifecycle revalidation" + ); + } + } + }); + } + // Cross-pod connection-control consumer: receive disconnect commands from // Redis pub/sub (published by the pod that recorded a ban) and close any // matching local sockets. A member's live connections may land on any pod, @@ -796,7 +822,7 @@ async fn main() -> anyhow::Result<()> { Ok(scoped) => match scoped.command { buzz_pubsub::conn_control::ConnControl::DisconnectCommunity => { state_for_conn_ctrl - .conn_manager + .community_connections .disconnect_community(scoped.community_id); } buzz_pubsub::conn_control::ConnControl::DisconnectPubkey { diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index e60bae226b..4cc8222041 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -1,6 +1,7 @@ //! Shared application state — Arc-wrapped, shared across all connections. use std::collections::{HashMap, HashSet}; +use std::future::Future; use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::sync::Arc; use std::time::Instant; @@ -55,7 +56,128 @@ struct ConnEntry { grace_limit: u8, } -/// Tracks active WebSocket connections and provides message routing by connection ID. +/// Community-scoped lifecycle registry shared by every long-lived socket type. +/// +/// A handler registers before durable active-state revalidation. Archival after +/// registration cancels the token; archival before registration is observed by +/// the revalidation. The returned guard removes the entry on every exit path. +pub struct CommunityConnectionRegistry { + connections: Arc>, +} + +impl Default for CommunityConnectionRegistry { + fn default() -> Self { + Self::new() + } +} + +impl CommunityConnectionRegistry { + /// Creates an empty lifecycle registry. + pub fn new() -> Self { + Self { + connections: Arc::new(DashMap::new()), + } + } + + /// Registers one socket and returns a guard that deregisters it on drop. + pub fn register( + &self, + connection_id: Uuid, + community_id: CommunityId, + cancel: CancellationToken, + ) -> CommunityConnectionGuard { + self.connections + .insert(connection_id, (community_id, cancel)); + CommunityConnectionGuard { + connection_id, + connections: Arc::clone(&self.connections), + } + } + + /// Cancels every socket type currently bound to `community_id`. + pub fn disconnect_community(&self, community_id: CommunityId) -> usize { + let mut closed = 0; + for entry in self.connections.iter() { + if entry.value().0 == community_id { + entry.value().1.cancel(); + closed += 1; + } + } + closed + } + + /// Returns the distinct communities with live sockets on this pod. + pub fn bound_communities(&self) -> HashSet { + self.connections + .iter() + .map(|entry| entry.value().0) + .collect() + } +} + +/// Removes a socket lifecycle registration on every handler exit path. +pub struct CommunityConnectionGuard { + connection_id: Uuid, + connections: Arc>, +} + +impl Drop for CommunityConnectionGuard { + fn drop(&mut self) { + self.connections.remove(&self.connection_id); + } +} + +/// Registers a socket, durably revalidates its community, then runs it. +/// +/// The ordering is the archival admission invariant: archive-before-query is +/// observed by the query, while archive-after-registration sees the token. +pub async fn run_registered_community_connection( + registry: &CommunityConnectionRegistry, + connection_id: Uuid, + community_id: CommunityId, + cancel: CancellationToken, + check_active: Check, + run: Run, +) where + Check: FnOnce() -> CheckFuture, + CheckFuture: Future>, + Run: FnOnce() -> RunFuture, + RunFuture: Future, +{ + let _guard = registry.register(connection_id, community_id, cancel.clone()); + if !matches!(check_active().await, Ok(true)) { + cancel.cancel(); + return; + } + if cancel.is_cancelled() { + return; + } + run().await; + cancel.cancel(); +} + +async fn revalidate_registered_communities( + registry: &CommunityConnectionRegistry, + mut check_active: Check, +) -> (usize, Vec<(CommunityId, buzz_db::DbError)>) +where + Check: FnMut(CommunityId) -> CheckFuture, + CheckFuture: Future>, +{ + let communities = registry.bound_communities(); + let mut closed = 0; + let mut failures = Vec::new(); + for community_id in communities { + match check_active(community_id).await { + Ok(false) => closed += registry.disconnect_community(community_id), + Ok(true) => {} + Err(error) => failures.push((community_id, error)), + } + } + (closed, failures) +} + +/// Tracks active Nostr WebSocket connections and provides message routing by connection ID. pub struct ConnectionManager { connections: DashMap, } @@ -339,6 +461,8 @@ pub struct AppState { pub sub_registry: Arc, /// Registry of active WebSocket connections. pub conn_manager: Arc, + /// Lifecycle cancellation for every long-lived socket, including huddle audio. + pub community_connections: Arc, /// Semaphore limiting total concurrent connections. pub conn_semaphore: Arc, /// Semaphore limiting concurrent message handler tasks. @@ -519,6 +643,7 @@ impl AppState { search: search_arc, sub_registry: Arc::new(SubscriptionRegistry::new()), conn_manager: Arc::new(ConnectionManager::new()), + community_connections: Arc::new(CommunityConnectionRegistry::new()), conn_semaphore: Arc::new(Semaphore::new(max_connections)), handler_semaphore: Arc::new(Semaphore::new(max_concurrent_handlers)), git_semaphore: Arc::new(Semaphore::new(git_max_concurrent_ops)), @@ -814,6 +939,10 @@ impl AppState { event_id: event_id.to_string(), reason: reason.to_string(), }; + // This pre-existing ban path may remain fire-and-forget because the + // durable ban row rejects the member again at auth. Community archival + // is different: its API awaits publication and live sockets also have a + // periodic durable-state revalidation backstop below. tokio::spawn(async move { if let Err(e) = pubsub.publish_conn_control(&tenant, &command).await { tracing::warn!("Failed to publish conn-control disconnect: {e}"); @@ -824,18 +953,36 @@ impl AppState { } /// Disconnect a community locally and publish the command to every relay pod. - pub fn disconnect_community_clusterwide(&self, tenant: &TenantContext) -> usize { - let closed = self.conn_manager.disconnect_community(tenant.community()); - let pubsub = Arc::clone(&self.pubsub); - let tenant = tenant.clone(); - tokio::spawn(async move { - if let Err(e) = pubsub - .publish_conn_control(&tenant, &ConnControl::DisconnectCommunity) - .await - { - tracing::warn!(community = %tenant.community(), error = %e, "failed to publish community disconnect"); - } - }); + /// + /// Publication is awaited so the archive API can distinguish durable state + /// from propagation completion and offer a retryable response on failure. + pub async fn disconnect_community_clusterwide( + &self, + tenant: &TenantContext, + ) -> Result { + let closed = self + .community_connections + .disconnect_community(tenant.community()); + self.pubsub + .publish_conn_control(tenant, &ConnControl::DisconnectCommunity) + .await?; + Ok(closed) + } + + /// Revalidate all communities with live sockets and cancel inactive ones. + /// + /// This is the durable backstop for Redis pub/sub's lossy offline-subscriber + /// semantics: a pod that missed a successful publish eventually observes the + /// archived row directly. + pub async fn revalidate_live_communities(&self) -> usize { + let (closed, failures) = + revalidate_registered_communities(&self.community_connections, |community_id| { + self.db.is_community_active(community_id) + }) + .await; + for (community_id, error) in failures { + tracing::warn!(%community_id, %error, "community lifecycle revalidation failed; retaining its sockets until next tick"); + } closed } @@ -1324,33 +1471,130 @@ mod tests { ); } - #[tokio::test] - async fn disconnect_community_closes_only_matching_tenant() { - let mgr = ConnectionManager::new(); + #[test] + fn community_lifecycle_disconnect_covers_socket_types_and_preserves_tenant_fence() { + let registry = CommunityConnectionRegistry::new(); let community_a = CommunityId::from_uuid(Uuid::from_u128(0xa)); let community_b = CommunityId::from_uuid(Uuid::from_u128(0xb)); - let register = |community| { - let conn_id = Uuid::new_v4(); - let (tx, _rx) = mpsc::channel(8); - let (ctrl_tx, _ctrl_rx) = mpsc::channel(8); - let cancel = CancellationToken::new(); - mgr.register( - conn_id, - tx, - ctrl_tx, - cancel.clone(), - community, - Arc::new(AtomicU8::new(0)), - Arc::new(Mutex::new(HashMap::new())), - 3, - ); - cancel - }; - let cancel_a = register(community_a); - let cancel_b = register(community_b); - assert_eq!(mgr.disconnect_community(community_a), 1); + let ordinary_a = CancellationToken::new(); + let audio_a = CancellationToken::new(); + let ordinary_b = CancellationToken::new(); + let _ordinary_a_guard = registry.register(Uuid::new_v4(), community_a, ordinary_a.clone()); + let _audio_a_guard = registry.register(Uuid::new_v4(), community_a, audio_a.clone()); + let _ordinary_b_guard = registry.register(Uuid::new_v4(), community_b, ordinary_b.clone()); + + assert_eq!(registry.disconnect_community(community_a), 2); + assert!(ordinary_a.is_cancelled()); + assert!(audio_a.is_cancelled()); + assert!(!ordinary_b.is_cancelled()); + } + + #[tokio::test] + async fn register_then_revalidate_closes_both_archive_race_orderings() { + let registry = CommunityConnectionRegistry::new(); + let community = CommunityId::from_uuid(Uuid::from_u128(0xa)); + + // Archive wins before durable revalidation: the check observes inactive + // and the socket body never starts. + let cancel_before = CancellationToken::new(); + let started_before = Arc::new(AtomicBool::new(false)); + let started_before_run = Arc::clone(&started_before); + run_registered_community_connection( + ®istry, + Uuid::new_v4(), + community, + cancel_before.clone(), + || async { Ok(false) }, + move || async move { started_before_run.store(true, Ordering::SeqCst) }, + ) + .await; + assert!(cancel_before.is_cancelled()); + assert!(!started_before.load(Ordering::SeqCst)); + + // Archive wins after registration but while revalidation is paused: its + // sweep sees the token, and even an active query result cannot start the + // socket body afterward. + let cancel_during = CancellationToken::new(); + let registered = Arc::new(tokio::sync::Notify::new()); + let resume = Arc::new(tokio::sync::Notify::new()); + let registered_check = Arc::clone(®istered); + let resume_check = Arc::clone(&resume); + let started_during = Arc::new(AtomicBool::new(false)); + let started_during_run = Arc::clone(&started_during); + let future = run_registered_community_connection( + ®istry, + Uuid::new_v4(), + community, + cancel_during.clone(), + move || async move { + registered_check.notify_one(); + resume_check.notified().await; + Ok(true) + }, + move || async move { started_during_run.store(true, Ordering::SeqCst) }, + ); + tokio::pin!(future); + tokio::select! { + _ = registered.notified() => {} + _ = &mut future => panic!("revalidation should be paused"), + } + assert_eq!(registry.disconnect_community(community), 1); + resume.notify_one(); + future.await; + assert!(cancel_during.is_cancelled()); + assert!(!started_during.load(Ordering::SeqCst)); + } + + #[tokio::test] + async fn revalidation_continues_after_one_community_lookup_failure() { + let registry = CommunityConnectionRegistry::new(); + let archived_a = CommunityId::from_uuid(Uuid::from_u128(0xa)); + let failed = CommunityId::from_uuid(Uuid::from_u128(0xb)); + let archived_c = CommunityId::from_uuid(Uuid::from_u128(0xc)); + let cancel_a = CancellationToken::new(); + let cancel_failed = CancellationToken::new(); + let cancel_c = CancellationToken::new(); + let _guard_a = registry.register(Uuid::new_v4(), archived_a, cancel_a.clone()); + let _guard_failed = registry.register(Uuid::new_v4(), failed, cancel_failed.clone()); + let _guard_c = registry.register(Uuid::new_v4(), archived_c, cancel_c.clone()); + + let (closed, failures) = + revalidate_registered_communities(®istry, |community| async move { + if community == failed { + Err(buzz_db::DbError::InvalidData( + "injected lookup failure".into(), + )) + } else { + Ok(false) + } + }) + .await; + + assert_eq!(closed, 2); assert!(cancel_a.is_cancelled()); - assert!(!cancel_b.is_cancelled()); + assert!(!cancel_failed.is_cancelled()); + assert!(cancel_c.is_cancelled()); + assert_eq!(failures.len(), 1); + assert_eq!(failures[0].0, failed); + assert_eq!( + registry.bound_communities(), + HashSet::from([archived_a, failed, archived_c]) + ); + } + + #[test] + fn community_lifecycle_guard_deregisters_on_early_return() { + let registry = CommunityConnectionRegistry::new(); + let community = CommunityId::from_uuid(Uuid::from_u128(0xa)); + let cancel = CancellationToken::new(); + let guard = registry.register(Uuid::new_v4(), community, cancel.clone()); + assert_eq!(registry.bound_communities(), HashSet::from([community])); + + drop(guard); + + assert!(registry.bound_communities().is_empty()); + assert_eq!(registry.disconnect_community(community), 0); + assert!(!cancel.is_cancelled()); } #[tokio::test] From 3b2a3a1bd92d4bd6936629ee7b62bf66633d85c2 Mon Sep 17 00:00:00 2001 From: npub16ynjrn8tm9gxg4kpgfgj24eup8met48z58m9g0cvmpq9msjht0lq7r96r6 Date: Mon, 13 Jul 2026 16:31:51 -0700 Subject: [PATCH 5/6] Harden archival propagation lifecycle Co-authored-by: npub16ynjrn8tm9gxg4kpgfgj24eup8met48z58m9g0cvmpq9msjht0lq7r96r6 Signed-off-by: npub16ynjrn8tm9gxg4kpgfgj24eup8met48z58m9g0cvmpq9msjht0lq7r96r6 --- crates/buzz-relay/src/api/operator.rs | 13 ++++ crates/buzz-relay/src/main.rs | 90 ++++++++++++++++++++++----- crates/buzz-relay/src/state.rs | 10 ++- 3 files changed, 97 insertions(+), 16 deletions(-) diff --git a/crates/buzz-relay/src/api/operator.rs b/crates/buzz-relay/src/api/operator.rs index 3b1596b90a..d2ef6ade0c 100644 --- a/crates/buzz-relay/src/api/operator.rs +++ b/crates/buzz-relay/src/api/operator.rs @@ -877,10 +877,23 @@ mod tests { .expect("active lookup") .is_none()); + assert_eq!( + state + .community_disconnect_publish_attempts + .load(std::sync::atomic::Ordering::Relaxed), + 1 + ); let second = archive_once(Arc::clone(&state)).await; assert_eq!(second.status(), StatusCode::SERVICE_UNAVAILABLE); let second_json = read_json(second).await; assert_eq!(second_json["archived_at"], first_archived_at); + assert_eq!( + state + .community_disconnect_publish_attempts + .load(std::sync::atomic::Ordering::Relaxed), + 2, + "idempotent archive retry must republish the disconnect" + ); let owned = state .db diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index c7814033e5..c2dbaa3e04 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -19,6 +19,7 @@ use buzz_relay::router::{build_health_router, build_router}; use buzz_relay::state::AppState; use buzz_relay::telemetry; use buzz_workflow::WorkflowEngine; +use tokio_util::sync::CancellationToken; fn buzz_auto_migrate_enabled(value: Option<&str>) -> bool { value.map(str::trim).is_some_and(|value| { @@ -791,20 +792,12 @@ async fn main() -> anyhow::Result<()> { .and_then(|value| value.parse::().ok()) .unwrap_or(30) .clamp(1, 300); - tokio::spawn(async move { - let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs)); - interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); - loop { - interval.tick().await; - let closed = lifecycle_state.revalidate_live_communities().await; - if closed > 0 { - tracing::info!( - closed, - "closed sockets for inactive communities during lifecycle revalidation" - ); - } - } - }); + let cancel = lifecycle_state.community_revalidator_cancel.clone(); + tokio::spawn(run_community_revalidator( + lifecycle_state, + std::time::Duration::from_secs(interval_secs), + cancel, + )); } // Cross-pod connection-control consumer: receive disconnect commands from @@ -927,6 +920,7 @@ async fn main() -> anyhow::Result<()> { } serve(router, health_router, Arc::clone(&state)).await?; + state.community_revalidator_cancel.cancel(); // Signal the audit worker to stop accepting, flush buffered entries, and // exit. Uses a CancellationToken so it works regardless of how many @@ -945,6 +939,42 @@ async fn main() -> anyhow::Result<()> { Ok(()) } +async fn run_community_revalidator( + state: Arc, + period: std::time::Duration, + cancel: CancellationToken, +) { + run_periodic_until_cancelled(period, cancel, || async { + let closed = state.revalidate_live_communities().await; + if closed > 0 { + tracing::info!( + closed, + "closed sockets for inactive communities during lifecycle revalidation" + ); + } + }) + .await; +} + +async fn run_periodic_until_cancelled( + period: std::time::Duration, + cancel: CancellationToken, + mut tick: Tick, +) where + Tick: FnMut() -> TickFuture, + TickFuture: std::future::Future, +{ + let mut interval = tokio::time::interval(period); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + biased; + _ = cancel.cancelled() => break, + _ = interval.tick() => tick().await, + } + } +} + /// Bind all listeners and run with graceful shutdown. /// /// ```text @@ -1502,7 +1532,37 @@ async fn run_usage_metrics_tick( #[cfg(test)] mod tests { - use super::buzz_auto_migrate_enabled; + use std::sync::Arc; + use std::time::Duration; + + use tokio_util::sync::CancellationToken; + + use super::{buzz_auto_migrate_enabled, run_periodic_until_cancelled}; + + #[tokio::test(start_paused = true)] + async fn periodic_loop_exits_immediately_on_cancellation() { + let cancel = CancellationToken::new(); + let tick_count = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let count_for_tick = Arc::clone(&tick_count); + let task_cancel = cancel.clone(); + let task = tokio::spawn(async move { + run_periodic_until_cancelled(Duration::from_secs(300), task_cancel, move || { + let count = Arc::clone(&count_for_tick); + async move { + count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } + }) + .await; + }); + + tokio::task::yield_now().await; + cancel.cancel(); + tokio::time::timeout(Duration::from_millis(1), task) + .await + .expect("loop must not wait for the next interval") + .expect("loop task"); + assert!(tick_count.load(std::sync::atomic::Ordering::Relaxed) <= 1); + } #[test] fn buzz_auto_migrate_is_opt_in() { diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 4cc8222041..b0fcd6a4e2 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet}; use std::future::Future; -use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicU8, Ordering}; use std::sync::Arc; use std::time::Instant; @@ -463,6 +463,10 @@ pub struct AppState { pub conn_manager: Arc, /// Lifecycle cancellation for every long-lived socket, including huddle audio. pub community_connections: Arc, + /// Stops only the periodic lifecycle revalidator during graceful shutdown. + pub community_revalidator_cancel: CancellationToken, + /// Test/telemetry counter for archive disconnect publication attempts. + pub community_disconnect_publish_attempts: Arc, /// Semaphore limiting total concurrent connections. pub conn_semaphore: Arc, /// Semaphore limiting concurrent message handler tasks. @@ -644,6 +648,8 @@ impl AppState { sub_registry: Arc::new(SubscriptionRegistry::new()), conn_manager: Arc::new(ConnectionManager::new()), community_connections: Arc::new(CommunityConnectionRegistry::new()), + community_revalidator_cancel: CancellationToken::new(), + community_disconnect_publish_attempts: Arc::new(AtomicU64::new(0)), conn_semaphore: Arc::new(Semaphore::new(max_connections)), handler_semaphore: Arc::new(Semaphore::new(max_concurrent_handlers)), git_semaphore: Arc::new(Semaphore::new(git_max_concurrent_ops)), @@ -963,6 +969,8 @@ impl AppState { let closed = self .community_connections .disconnect_community(tenant.community()); + self.community_disconnect_publish_attempts + .fetch_add(1, Ordering::Relaxed); self.pubsub .publish_conn_control(tenant, &ConnControl::DisconnectCommunity) .await?; From aa1f4f0fafbe25a57534d41cecafd6fb15d1dd57 Mon Sep 17 00:00:00 2001 From: npub1dccv64krpcpse5cmkzfeh998cftungyatw3djt8jwdw6g43f7fyqzzmrf7 <6e30cd56c30e030cd31bb0939b94a7c257c9a09d5ba2d92cf2735da45629f248@sprout-oss.stage.blox.sqprod.co> Date: Mon, 13 Jul 2026 19:36:01 -0700 Subject: [PATCH 6/6] refactor: remove obsolete community disconnect method Co-authored-by: npub1dccv64krpcpse5cmkzfeh998cftungyatw3djt8jwdw6g43f7fyqzzmrf7 <6e30cd56c30e030cd31bb0939b94a7c257c9a09d5ba2d92cf2735da45629f248@sprout-oss.stage.blox.sqprod.co> Signed-off-by: npub1dccv64krpcpse5cmkzfeh998cftungyatw3djt8jwdw6g43f7fyqzzmrf7 <6e30cd56c30e030cd31bb0939b94a7c257c9a09d5ba2d92cf2735da45629f248@sprout-oss.stage.blox.sqprod.co> --- crates/buzz-relay/src/state.rs | 12 ------------ 1 file changed, 12 deletions(-) diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index b0fcd6a4e2..87650aaaf0 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -317,18 +317,6 @@ impl ConnectionManager { closed } - /// Disconnect every live socket bound to one community. - pub fn disconnect_community(&self, community: CommunityId) -> usize { - let mut closed = 0; - for entry in self.connections.iter() { - if entry.community_id == community { - entry.cancel.cancel(); - closed += 1; - } - } - closed - } - /// Return the server-resolved community that the connection's host bound to. pub fn community_for_conn(&self, conn_id: Uuid) -> Option { self.connections