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,35 @@ 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, + 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 +479,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 +496,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 +523,7 @@ impl Db { SELECT host FROM communities WHERE id = $1 + AND archived_at IS NULL "#, ) .bind(community_id.as_uuid()) @@ -632,6 +678,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 +701,39 @@ 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, + protected_deployment_host: &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' + 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| { + 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 @@ -4128,8 +4208,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 0147572515..e4cb57322c 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] @@ -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] 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..bc177cff13 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,23 @@ 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 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 55be86cba5..d2ef6ade0c 100644 --- a/crates/buzz-relay/src/api/operator.rs +++ b/crates/buzz-relay/src/api/operator.rs @@ -192,6 +192,75 @@ 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))?; + 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", + )); + } + 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, &deployment_host) + .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 = 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(), + "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 +297,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 +448,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}")))?; @@ -728,6 +798,118 @@ 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()); + + 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 + .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 df68ec2700..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| { @@ -781,6 +782,24 @@ 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); + 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 // 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, @@ -794,6 +813,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 + .community_connections + .disconnect_community(scoped.community_id); + } buzz_pubsub::conn_control::ConnControl::DisconnectPubkey { pubkey, event_id, @@ -896,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 @@ -914,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 @@ -1471,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/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..87650aaaf0 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -1,7 +1,8 @@ //! Shared application state — Arc-wrapped, shared across all connections. use std::collections::{HashMap, HashSet}; -use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; +use std::future::Future; +use std::sync::atomic::{AtomicBool, AtomicU64, 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, } @@ -327,6 +449,12 @@ 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, + /// 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. @@ -507,6 +635,9 @@ impl AppState { search: search_arc, 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)), @@ -802,6 +933,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}"); @@ -811,6 +946,42 @@ impl AppState { closed } + /// Disconnect a community locally and publish the command to every relay pod. + /// + /// 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.community_disconnect_publish_attempts + .fetch_add(1, Ordering::Relaxed); + 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 + } + /// 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 +1467,132 @@ mod tests { ); } + #[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 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_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] 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) );