diff --git a/crates/buzz-admin/src/communities.rs b/crates/buzz-admin/src/communities.rs new file mode 100644 index 00000000000..175347a64e7 --- /dev/null +++ b/crates/buzz-admin/src/communities.rs @@ -0,0 +1,508 @@ +//! Reversible whole-community lifecycle commands. + +use anyhow::{Context, Result}; +use buzz_core::tenant::{normalize_host, relay_url_authority, TenantContext}; +use buzz_db::{ArchivedCommunityRecord, UnarchivedCommunityRecord}; +use buzz_pubsub::conn_control::ConnControl; +use buzz_pubsub::PubSubManager; +use clap::Subcommand; +use serde_json::{json, Value}; +use url::{Host, Url}; + +const MAX_HOST_LEN: usize = 255; + +/// Reversible community lifecycle actions. +#[derive(Debug, Subcommand)] +pub enum CommunitiesCommand { + /// Archive a community and disconnect its live clients. + Archive { + /// Exact community hostname or authority. + #[arg(long)] + host: String, + /// Current owner public key as 64-character hex. + #[arg(long)] + owner_pubkey: String, + /// Identity of the operator performing the action. + #[arg(long)] + operator_id: String, + /// Human-readable reason for the action. + #[arg(long)] + reason: String, + }, + /// Restore an archived community. + Unarchive { + /// Exact community hostname or authority. + #[arg(long)] + host: String, + /// Current owner public key as 64-character hex. + #[arg(long)] + owner_pubkey: String, + /// Identity of the operator performing the action. + #[arg(long)] + operator_id: String, + /// Human-readable reason for the action. + #[arg(long)] + reason: String, + }, +} + +/// Run a reversible community lifecycle action. +pub async fn run(command: CommunitiesCommand) -> anyhow::Result { + match command { + CommunitiesCommand::Archive { + host, + owner_pubkey, + operator_id, + reason, + } => archive(host, owner_pubkey, operator_id, reason).await, + CommunitiesCommand::Unarchive { + host, + owner_pubkey, + operator_id, + reason, + } => unarchive(host, owner_pubkey, operator_id, reason).await, + } +} + +async fn archive( + host: String, + owner_pubkey: String, + operator_id: String, + reason: String, +) -> Result { + let host = normalize_host_authority(&host).map_err(anyhow::Error::msg)?; + let owner_pubkey = parse_owner_pubkey(&owner_pubkey).map_err(anyhow::Error::msg)?; + let operator_id = required_audit_field("operator_id", &operator_id) + .map_err(anyhow::Error::msg)? + .to_string(); + let reason = required_audit_field("reason", &reason) + .map_err(anyhow::Error::msg)? + .to_string(); + + let relay_url = std::env::var("RELAY_URL") + .context("RELAY_URL is required to protect the deployment community")?; + let deployment_host = deployment_host_from_relay_url(&relay_url)?; + ensure_not_deployment_host(&host, &deployment_host)?; + + let db = crate::connect_db().await?; + let record = db + .archive_community_owned_by(&host, &owner_pubkey, &deployment_host) + .await? + .ok_or_else(|| { + anyhow::anyhow!( + "no active, undeleted community matched both the hostname and current owner pubkey" + ) + })?; + let tenant = TenantContext::resolved(record.id, &record.host); + + let command = ConnControl::DisconnectCommunity { + archived_at: Some(record.archived_at), + }; + let (propagation, exit_code) = match publish_disconnect(&tenant, &command).await { + Ok(subscriber_count) => classify_archive_publication(subscriber_count), + Err(error) => ( + ArchivePropagation::Pending(format!( + "connection propagation pending — retry this command: {error:#}" + )), + 1, + ), + }; + print_json(&archive_evidence( + &record, + &operator_id, + &reason, + propagation, + ))?; + Ok(exit_code) +} + +async fn unarchive( + host: String, + owner_pubkey: String, + operator_id: String, + reason: String, +) -> Result { + let host = normalize_host_authority(&host).map_err(anyhow::Error::msg)?; + let owner_pubkey = parse_owner_pubkey(&owner_pubkey).map_err(anyhow::Error::msg)?; + let operator_id = required_audit_field("operator_id", &operator_id) + .map_err(anyhow::Error::msg)? + .to_string(); + let reason = required_audit_field("reason", &reason) + .map_err(anyhow::Error::msg)? + .to_string(); + + let db = crate::connect_db().await?; + let record = db + .unarchive_community_owned_by(&host, &owner_pubkey) + .await? + .ok_or_else(|| { + anyhow::anyhow!( + "no active-deletion-state community matched both the hostname and current owner pubkey" + ) + })?; + print_json(&unarchive_evidence(&record, &operator_id, &reason))?; + Ok(0) +} + +async fn publish_disconnect(tenant: &TenantContext, command: &ConnControl) -> Result { + let redis_url = std::env::var("REDIS_URL").context("REDIS_URL is required")?; + let redis_pool = deadpool_redis::Config::from_url(&redis_url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .context("Redis pool creation failed")?; + let pubsub = PubSubManager::new(&redis_url, redis_pool) + .await + .context("PubSub init failed")?; + pubsub + .publish_conn_control(tenant, command) + .await + .context("publishing DisconnectCommunity failed") +} + +enum ArchivePropagation { + Published(i64), + Pending(String), +} + +fn classify_archive_publication(subscriber_count: i64) -> (ArchivePropagation, i32) { + if subscriber_count > 0 { + (ArchivePropagation::Published(subscriber_count), 0) + } else { + ( + ArchivePropagation::Pending( + "connection propagation pending — Redis reported zero subscribers; retry this command" + .to_string(), + ), + 1, + ) + } +} + +fn archive_evidence( + record: &ArchivedCommunityRecord, + operator_id: &str, + reason: &str, + propagation: ArchivePropagation, +) -> Value { + match propagation { + ArchivePropagation::Published(subscriber_count) => json!({ + "action": "archive", + "community_id": record.id.to_string(), + "host": record.host, + "archived_at": record.archived_at, + "status": "archived", + "operator_id": operator_id, + "reason": reason, + "propagation": "published", + "propagation_subscribers": subscriber_count, + "retryable": false, + }), + ArchivePropagation::Pending(error) => json!({ + "action": "archive", + "community_id": record.id.to_string(), + "host": record.host, + "archived_at": record.archived_at, + "status": "archived", + "operator_id": operator_id, + "reason": reason, + "propagation": "pending", + "propagation_subscribers": null, + "retryable": true, + "error": error, + }), + } +} + +fn unarchive_evidence( + record: &UnarchivedCommunityRecord, + operator_id: &str, + reason: &str, +) -> Value { + json!({ + "action": "unarchive", + "community_id": record.id.to_string(), + "host": record.host, + "archived_at": null, + "status": "active", + "operator_id": operator_id, + "reason": reason, + }) +} + +fn print_json(value: &Value) -> Result<()> { + println!("{}", serde_json::to_string_pretty(value)?); + Ok(()) +} + +fn required_audit_field<'a>(name: &str, value: &'a str) -> Result<&'a str, String> { + let value = value.trim(); + if value.is_empty() { + return Err(format!("{name} must not be empty")); + } + Ok(value) +} + +fn deployment_host_from_relay_url(relay_url: &str) -> Result { + let host = relay_url_authority(relay_url); + if host.is_empty() { + anyhow::bail!("RELAY_URL does not contain a valid deployment authority"); + } + Ok(host) +} + +fn ensure_not_deployment_host(host: &str, deployment_host: &str) -> Result<()> { + if host == deployment_host { + anyhow::bail!("the deployment community cannot be archived"); + } + Ok(()) +} + +fn normalize_host_authority(host: &str) -> Result { + if host.is_empty() { + return Err("host is empty".to_string()); + } + if host.len() > MAX_HOST_LEN { + return Err(format!( + "host too long: {} bytes (max {MAX_HOST_LEN})", + host.len() + )); + } + if host.chars().any(|c| c.is_control() || c.is_whitespace()) { + return Err("host contains invalid characters".to_string()); + } + if host.contains('/') || host.contains('?') || host.contains('#') || host.contains('@') { + return Err( + "host must be a bare authority (no scheme, path, query, or userinfo)".to_string(), + ); + } + + let normalized = normalize_host(host); + let parsed = Url::parse(&format!("http://{normalized}/")) + .map_err(|_| "host is not a valid authority".to_string())?; + let parsed_host = parsed + .host() + .ok_or_else(|| "host is not a valid authority".to_string())?; + let serialized_host = match parsed_host { + Host::Domain(domain) => { + validate_domain_labels(domain)?; + domain.to_string() + } + Host::Ipv4(addr) => addr.to_string(), + Host::Ipv6(addr) => format!("[{addr}]"), + }; + let canonical_authority = match parsed.port() { + Some(port) => format!("{serialized_host}:{port}"), + None => serialized_host, + }; + if canonical_authority != normalized { + return Err(format!( + "host is not a canonical authority: expected {canonical_authority:?}" + )); + } + + Ok(normalized) +} + +fn validate_domain_labels(domain: &str) -> Result<(), String> { + if domain.len() > 253 { + return Err("domain name too long".to_string()); + } + for label in domain.split('.') { + if label.is_empty() { + return Err("domain contains an empty label".to_string()); + } + if label.len() > 63 { + return Err("domain label too long".to_string()); + } + let valid_label = label + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b == b'-') + && !label.starts_with('-') + && !label.ends_with('-'); + if !valid_label { + return Err("domain label contains invalid characters".to_string()); + } + } + Ok(()) +} + +fn parse_owner_pubkey(input: &str) -> Result { + let normalized = input.to_ascii_lowercase(); + if normalized.len() != 64 || !normalized.chars().all(|c| c.is_ascii_hexdigit()) { + return Err("owner pubkey must be 64-character hex".to_string()); + } + nostr::PublicKey::parse(&normalized) + .map(|pubkey| pubkey.to_hex()) + .map_err(|error| format!("invalid owner pubkey: {error}")) +} + +#[cfg(test)] +mod tests { + use nostr::Keys; + + use super::*; + + #[test] + fn normalizes_bare_host_authority() { + assert_eq!( + normalize_host_authority("EXAMPLE.COMMUNITIES.BUZZ.XYZ.:443").unwrap(), + "example.communities.buzz.xyz" + ); + assert_eq!( + normalize_host_authority("Relay.Example:8443").unwrap(), + "relay.example:8443" + ); + } + + #[test] + fn rejects_non_authority_and_malformed_hosts() { + for invalid in [ + "https://relay.example", + "relay.example/path", + " relay.example", + "relay example", + "relay..example", + "-relay.example", + "relay.example:abc", + "[::1", + ] { + assert!( + normalize_host_authority(invalid).is_err(), + "accepted invalid host {invalid:?}" + ); + } + } + + #[test] + fn validates_and_canonicalizes_owner_pubkey() { + let owner = Keys::generate().public_key().to_hex().to_ascii_uppercase(); + assert_eq!( + parse_owner_pubkey(&owner).unwrap(), + owner.to_ascii_lowercase() + ); + + let invalid = [ + String::new(), + "abc".to_string(), + "g".repeat(64), + "0".repeat(63), + ]; + for invalid in &invalid { + assert!( + parse_owner_pubkey(invalid).is_err(), + "accepted invalid owner pubkey {invalid:?}" + ); + } + } + + fn archived_record() -> buzz_db::ArchivedCommunityRecord { + buzz_db::ArchivedCommunityRecord { + id: buzz_core::CommunityId::from_uuid(uuid::Uuid::from_u128(1)), + host: "relay.example".to_string(), + archived_at: "2026-08-31T12:00:00Z".parse().unwrap(), + } + } + + #[test] + fn archive_evidence_records_operator_reason_and_published_propagation() { + let evidence = archive_evidence( + &archived_record(), + "operator@example", + "owner requested archive", + ArchivePropagation::Published(3), + ); + + assert_eq!( + evidence["community_id"], + uuid::Uuid::from_u128(1).to_string() + ); + assert_eq!(evidence["host"], "relay.example"); + assert_eq!(evidence["status"], "archived"); + assert_eq!(evidence["operator_id"], "operator@example"); + assert_eq!(evidence["reason"], "owner requested archive"); + assert_eq!(evidence["propagation"], "published"); + assert_eq!(evidence["propagation_subscribers"], 3); + assert_eq!(evidence["retryable"], false); + } + + #[test] + fn archive_evidence_marks_committed_propagation_failure_retryable() { + let evidence = archive_evidence( + &archived_record(), + "operator@example", + "owner requested archive", + ArchivePropagation::Pending("redis unavailable".to_string()), + ); + + assert_eq!(evidence["status"], "archived"); + assert_eq!(evidence["propagation"], "pending"); + assert_eq!(evidence["retryable"], true); + assert_eq!(evidence["error"], "redis unavailable"); + } + + #[test] + fn zero_redis_subscribers_is_propagation_pending() { + let (propagation, exit_code) = classify_archive_publication(0); + let evidence = archive_evidence( + &archived_record(), + "operator@example", + "owner requested archive", + propagation, + ); + + assert_eq!(exit_code, 1); + assert_eq!(evidence["propagation"], "pending"); + assert_eq!(evidence["retryable"], true); + + let (propagation, exit_code) = classify_archive_publication(2); + let evidence = archive_evidence( + &archived_record(), + "operator@example", + "owner requested archive", + propagation, + ); + assert_eq!(exit_code, 0); + assert_eq!(evidence["propagation"], "published"); + assert_eq!(evidence["propagation_subscribers"], 2); + } + + #[test] + fn unarchive_evidence_records_operator_reason_without_disconnect() { + let record = buzz_db::UnarchivedCommunityRecord { + id: buzz_core::CommunityId::from_uuid(uuid::Uuid::from_u128(2)), + host: "relay.example".to_string(), + }; + let evidence = unarchive_evidence(&record, "operator@example", "rollback"); + + assert_eq!( + evidence["community_id"], + uuid::Uuid::from_u128(2).to_string() + ); + assert_eq!(evidence["host"], "relay.example"); + assert_eq!(evidence["archived_at"], serde_json::Value::Null); + assert_eq!(evidence["status"], "active"); + assert_eq!(evidence["operator_id"], "operator@example"); + assert_eq!(evidence["reason"], "rollback"); + assert!(evidence.get("propagation").is_none()); + } + + #[test] + fn audit_fields_must_be_nonempty() { + assert_eq!( + required_audit_field("operator_id", "operator@example").unwrap(), + "operator@example" + ); + assert!(required_audit_field("operator_id", " ").is_err()); + assert!(required_audit_field("reason", "\t").is_err()); + } + + #[test] + fn deployment_host_is_derived_and_protected() { + assert_eq!( + deployment_host_from_relay_url("wss://RELAY.EXAMPLE:443/path").unwrap(), + "relay.example" + ); + assert!(deployment_host_from_relay_url("not a URL").is_err()); + assert!(ensure_not_deployment_host("relay.example", "relay.example").is_err()); + assert!(ensure_not_deployment_host("other.example", "relay.example").is_ok()); + } +} diff --git a/crates/buzz-admin/src/main.rs b/crates/buzz-admin/src/main.rs index 19a3b1d9d48..72b69c7d1a0 100644 --- a/crates/buzz-admin/src/main.rs +++ b/crates/buzz-admin/src/main.rs @@ -20,6 +20,7 @@ //! newest timestamp and collide on the bumped second. run.sh serialization is //! the guard against parallel adds (e.g. `xargs -P`). +mod communities; mod deletions; use std::sync::Arc; @@ -83,6 +84,11 @@ enum Command { #[command(subcommand)] command: ProductFeedbackCommand, }, + /// Reversible whole-community lifecycle controls. + Communities { + #[command(subcommand)] + command: communities::CommunitiesCommand, + }, /// Durable CLI-only whole-community deletion control plane. Deletions { #[command(subcommand)] @@ -160,6 +166,7 @@ async fn run(cli: Cli) -> Result { Command::ProductFeedback { command: ProductFeedbackCommand::List { limit }, } => cmd_list_product_feedback(limit).await, + Command::Communities { command } => communities::run(command).await, Command::Deletions { command } => deletions::run(command).await, Command::ReconcileChannels { channel, relay_key } => { reconcile_channels(channel, relay_key).await?; @@ -631,3 +638,157 @@ async fn reconcile_channels( ); Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + const OWNER: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + + #[test] + fn communities_command_parses_archive() { + let cli = Cli::try_parse_from([ + "buzz-admin", + "communities", + "archive", + "--host", + "example.communities.buzz.xyz", + "--owner-pubkey", + OWNER, + "--operator-id", + "codex", + "--reason", + "requested deletion", + ]) + .unwrap(); + + match cli.command { + Command::Communities { + command: + communities::CommunitiesCommand::Archive { + host, + owner_pubkey, + operator_id, + reason, + }, + } => { + assert_eq!(host, "example.communities.buzz.xyz"); + assert_eq!(owner_pubkey, OWNER); + assert_eq!(operator_id, "codex"); + assert_eq!(reason, "requested deletion"); + } + _ => panic!("expected communities archive command"), + } + } + + #[test] + fn communities_command_parses_unarchive() { + let cli = Cli::try_parse_from([ + "buzz-admin", + "communities", + "unarchive", + "--host", + "example.communities.buzz.xyz", + "--owner-pubkey", + OWNER, + "--operator-id", + "codex", + "--reason", + "rollback", + ]) + .unwrap(); + + match cli.command { + Command::Communities { + command: + communities::CommunitiesCommand::Unarchive { + host, + owner_pubkey, + operator_id, + reason, + }, + } => { + assert_eq!(host, "example.communities.buzz.xyz"); + assert_eq!(owner_pubkey, OWNER); + assert_eq!(operator_id, "codex"); + assert_eq!(reason, "rollback"); + } + _ => panic!("expected communities unarchive command"), + } + } + + #[test] + fn communities_commands_require_all_safety_and_audit_arguments() { + let missing_argument_cases = [ + vec![ + "buzz-admin", + "communities", + "archive", + "--owner-pubkey", + OWNER, + "--operator-id", + "codex", + "--reason", + "requested deletion", + ], + vec![ + "buzz-admin", + "communities", + "archive", + "--host", + "example.communities.buzz.xyz", + "--operator-id", + "codex", + "--reason", + "requested deletion", + ], + vec![ + "buzz-admin", + "communities", + "unarchive", + "--host", + "example.communities.buzz.xyz", + "--owner-pubkey", + OWNER, + "--reason", + "rollback", + ], + vec![ + "buzz-admin", + "communities", + "unarchive", + "--host", + "example.communities.buzz.xyz", + "--owner-pubkey", + OWNER, + "--operator-id", + "codex", + ], + ]; + + for args in missing_argument_cases { + assert!(Cli::try_parse_from(args).is_err()); + } + } + + #[test] + fn communities_commands_do_not_expose_deletion_approval_arguments() { + let command = Cli::try_parse_from([ + "buzz-admin", + "communities", + "archive", + "--host", + "example.communities.buzz.xyz", + "--owner-pubkey", + OWNER, + "--operator-id", + "codex", + "--reason", + "requested deletion", + "--approved-by", + "second-operator", + ]); + + assert!(command.is_err()); + } +} diff --git a/crates/buzz-db/src/store/community.rs b/crates/buzz-db/src/store/community.rs index bcd38f4e5ce..72920bd8cbb 100644 --- a/crates/buzz-db/src/store/community.rs +++ b/crates/buzz-db/src/store/community.rs @@ -160,6 +160,83 @@ impl Db { Ok(active) } + /// Runs `apply` while holding the community row lock only when the exact + /// archive transition is still current. + /// + /// Serializing the synchronous action with unarchive prevents the final + /// lifecycle order from becoming "restored, then disconnected". The exact + /// timestamp also rejects delayed commands from an earlier archive cycle. + #[datastore_span(name = "with_community_archive_fence", system = "postgresql")] + pub async fn with_community_archive_fence( + &self, + community_id: CommunityId, + archived_at: DateTime, + apply: impl FnOnce() -> T, + ) -> Result> { + let mut tx = self.pool.begin().await?; + let matches = sqlx::query_scalar::<_, bool>( + r#"SELECT COALESCE(archived_at = $2, FALSE) + FROM communities + WHERE id = $1 + AND deletion_state = 'active' + AND deleted_at IS NULL + FOR UPDATE"#, + ) + .bind(community_id.as_uuid()) + .bind(archived_at) + .fetch_optional(&mut *tx) + .await? + .unwrap_or(false); + + if !matches { + tx.rollback().await?; + return Ok(None); + } + + let result = apply(); + tx.commit().await?; + Ok(Some(result)) + } + + /// Runs `apply` while holding the community row lock only when the + /// community is not active. The callback receives the current archive + /// transition, or `None` when the row is deleted, deleting, or missing. + /// + /// The row lock serializes the synchronous action with unarchive so a + /// periodic lifecycle revalidation cannot disconnect a community after it + /// has been restored. Missing community ids are also treated as inactive. + #[datastore_span(name = "with_inactive_community_fence", system = "postgresql")] + pub async fn with_inactive_community_fence( + &self, + community_id: CommunityId, + apply: impl FnOnce(Option>) -> T, + ) -> Result> { + let mut tx = self.pool.begin().await?; + let lifecycle = sqlx::query_as::<_, (Option>, bool)>( + r#"SELECT archived_at, + deleted_at IS NOT NULL OR deletion_state <> 'active' + FROM communities + WHERE id = $1 + FOR UPDATE"#, + ) + .bind(community_id.as_uuid()) + .fetch_optional(&mut *tx) + .await?; + + let archived_at = match lifecycle { + Some((None, false)) => { + tx.rollback().await?; + return Ok(None); + } + Some((archived_at, false)) => archived_at, + Some((_, true)) | None => None, + }; + + let result = apply(archived_at); + tx.commit().await?; + Ok(Some(result)) + } + /// Returns a community by host regardless of lifecycle state. Operator-plane only. #[datastore_span( name = "lookup_community_by_host_for_management", @@ -708,6 +785,8 @@ mod postgres_tests { let operations = [ "lookup_community_by_host", "is_community_active", + "with_community_archive_fence", + "with_inactive_community_fence", "lookup_community_by_host_for_management", "list_communities_owned_by", "lookup_community_host", @@ -721,14 +800,16 @@ mod postgres_tests { "communities_of_channels", ]; for operation in operations { - let method = format!("pub async fn {operation}("); + let standard_method = format!("pub async fn {operation}("); + let generic_method = format!("pub async fn {operation}<"); + let method_count = community_source.matches(&standard_method).count() + + community_source.matches(&generic_method).count(); assert_eq!( - community_source.matches(&method).count(), - 1, + method_count, 1, "{operation} implementation must live exactly once in community.rs", ); assert!( - !lib_source.contains(&method), + !lib_source.contains(&standard_method) && !lib_source.contains(&generic_method), "{operation} implementation must not remain in lib.rs", ); @@ -765,6 +846,7 @@ mod postgres_tests { "lookup_community_by_host_matches_case_insensitive_host_index", "create_community_with_owner_is_atomic_and_create_only", "unarchive_community_owned_by_restores_admission_idempotently", + "archive_disconnect_fence_tracks_the_exact_archive_transition", "create_community_with_owner_enforces_per_owner_limit", "concurrent_same_owner_create_returns_the_winning_row_to_both_callers", "ensure_configured_community_reports_insert_winner", @@ -943,6 +1025,146 @@ mod postgres_tests { assert_eq!(retry, restored); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn archive_disconnect_fence_tracks_the_exact_archive_transition() { + let db = setup_db().await; + let host = format!("archive-fence-{}.example", Uuid::new_v4().simple()); + let owner = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple()); + let created = db + .create_community_with_owner(&host, &owner) + .await + .expect("create community"); + let CreateCommunityWithOwnerResult::Created(created) = created else { + panic!("expected new community"); + }; + let first_archive = db + .archive_community_owned_by(&host, &owner, "protected.example") + .await + .expect("archive community") + .expect("owned community"); + + assert_eq!( + db.with_community_archive_fence(created.id, first_archive.archived_at, || "applied") + .await + .expect("matching archive fence"), + Some("applied") + ); + db.unarchive_community_owned_by(&host, &owner) + .await + .expect("unarchive community") + .expect("owned community"); + assert_eq!( + db.with_community_archive_fence(created.id, first_archive.archived_at, || "stale") + .await + .expect("unarchived fence"), + None + ); + + let replacement_archived_at = first_archive.archived_at + chrono::Duration::seconds(1); + sqlx::query("UPDATE communities SET archived_at = $2 WHERE id = $1") + .bind(created.id.as_uuid()) + .bind(replacement_archived_at) + .execute(&db.pool) + .await + .expect("replace archive transition"); + assert_eq!( + db.with_community_archive_fence(created.id, first_archive.archived_at, || "stale") + .await + .expect("stale archive fence"), + None + ); + assert_eq!( + db.with_community_archive_fence(created.id, replacement_archived_at, || "replacement") + .await + .expect("replacement archive fence"), + Some("replacement") + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[ignore = "requires Postgres"] + async fn inactive_community_fence_holds_the_row_lock_through_disconnect() { + let db = setup_db().await; + let host = format!( + "inactive-community-fence-{}.example", + Uuid::new_v4().simple() + ); + let owner = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple()); + let created = db + .create_community_with_owner(&host, &owner) + .await + .expect("create community"); + let CreateCommunityWithOwnerResult::Created(created) = created else { + panic!("expected new community"); + }; + let archived = db + .archive_community_owned_by(&host, &owner, "protected.example") + .await + .expect("archive community") + .expect("owned community"); + + let fenced_db = db.clone(); + let community_id = created.id; + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = std::sync::mpsc::sync_channel(0); + let runtime = tokio::runtime::Handle::current(); + let fence = tokio::task::spawn_blocking(move || { + runtime.block_on(fenced_db.with_inactive_community_fence( + community_id, + move |archived_at| { + assert_eq!(archived_at, Some(archived.archived_at)); + entered_tx.send(()).expect("report entered fence"); + release_rx.recv().expect("release fenced disconnect"); + "disconnected" + }, + )) + }); + entered_rx.await.expect("fence acquired row lock"); + + let mut contender = db.pool.begin().await.expect("begin lock contender"); + sqlx::query("SET LOCAL lock_timeout = '100ms'") + .execute(&mut *contender) + .await + .expect("set contender lock timeout"); + let lock_error = sqlx::query("UPDATE communities SET archived_at = NULL WHERE id = $1") + .bind(created.id.as_uuid()) + .execute(&mut *contender) + .await + .expect_err("unarchive update must contend on the fenced row lock"); + assert_eq!( + lock_error + .as_database_error() + .and_then(|error| error.code().map(|code| code.into_owned())) + .as_deref(), + Some("55P03"), + "the contender must fail specifically because the row lock is held" + ); + contender + .rollback() + .await + .expect("roll back timed-out contender"); + + release_tx.send(()).expect("release fence"); + assert_eq!( + tokio::time::timeout(std::time::Duration::from_secs(5), fence) + .await + .expect("fence completes after release") + .expect("fence task") + .expect("inactive community fence"), + Some("disconnected") + ); + assert!(db + .unarchive_community_owned_by(&host, &owner) + .await + .expect("unarchive community") + .is_some()); + assert!(db + .is_community_active(created.id) + .await + .expect("restored community state")); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn create_community_with_owner_enforces_per_owner_limit() { diff --git a/crates/buzz-pubsub/src/conn_control.rs b/crates/buzz-pubsub/src/conn_control.rs index bc177cff139..81722124dc9 100644 --- a/crates/buzz-pubsub/src/conn_control.rs +++ b/crates/buzz-pubsub/src/conn_control.rs @@ -15,6 +15,7 @@ //! dropped, the next auth attempt is refused at the auth seam. use buzz_core::{CommunityId, TenantContext}; +use chrono::{DateTime, Utc}; use futures_util::StreamExt; use serde::{Deserialize, Serialize}; use tokio::sync::broadcast; @@ -55,7 +56,16 @@ pub fn parse_conn_control_channel(channel: &str) -> Option { #[serde(tag = "op")] pub enum ConnControl { /// Disconnect every live socket bound to the carrying community. - DisconnectCommunity, + /// + /// Archive callers include the durable archive-transition timestamp. Each + /// receiver checks it under a row lock before disconnecting, so a delayed + /// command cannot disconnect a subsequently restored community. `None` + /// preserves the unconditional command used by permanent deletion. + DisconnectCommunity { + /// Exact archive transition this command belongs to, when reversible. + #[serde(default, skip_serializing_if = "Option::is_none")] + archived_at: Option>, + }, /// 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 @@ -201,18 +211,55 @@ mod tests { #[test] fn disconnect_community_command_serde_round_trips() { - let cmd = ConnControl::DisconnectCommunity; + let cmd = ConnControl::DisconnectCommunity { archived_at: None }; + let json = serde_json::to_string(&cmd).unwrap(); + assert_eq!(json, r#"{"op":"DisconnectCommunity"}"#); + assert_eq!(serde_json::from_str::(&json).unwrap(), cmd); + } + + #[test] + fn archived_community_disconnect_carries_the_lifecycle_fence() { + let archived_at = chrono::DateTime::parse_from_rfc3339("2026-08-31T12:34:56.123456Z") + .unwrap() + .with_timezone(&chrono::Utc); + let cmd = ConnControl::DisconnectCommunity { + archived_at: Some(archived_at), + }; let json = serde_json::to_string(&cmd).unwrap(); + assert!(json.contains(r#""archived_at":"2026-08-31T12:34:56.123456Z""#)); assert_eq!(serde_json::from_str::(&json).unwrap(), cmd); } + #[test] + fn legacy_consumer_accepts_archived_community_disconnect_payload() { + #[derive(Debug, Deserialize, PartialEq, Eq)] + #[serde(tag = "op")] + enum LegacyConnControl { + DisconnectCommunity, + } + + let archived_at = chrono::DateTime::parse_from_rfc3339("2026-08-31T12:34:56.123456Z") + .unwrap() + .with_timezone(&chrono::Utc); + let json = serde_json::to_string(&ConnControl::DisconnectCommunity { + archived_at: Some(archived_at), + }) + .unwrap(); + + assert_eq!( + serde_json::from_str::(&json).unwrap(), + LegacyConnControl::DisconnectCommunity + ); + } + #[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(); + let known = + serde_json::to_string(&ConnControl::DisconnectCommunity { archived_at: None }).unwrap(); assert_eq!( serde_json::from_str::(&known).unwrap(), - ConnControl::DisconnectCommunity + ConnControl::DisconnectCommunity { archived_at: None } ); } diff --git a/crates/buzz-relay/src/api/operator.rs b/crates/buzz-relay/src/api/operator.rs index 2c49ca6a5c3..c3900b79236 100644 --- a/crates/buzz-relay/src/api/operator.rs +++ b/crates/buzz-relay/src/api/operator.rs @@ -239,7 +239,10 @@ 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 = match state.disconnect_community_clusterwide(&tenant).await { + let closed = match state + .disconnect_community_clusterwide(&tenant, record.archived_at) + .await + { Ok(closed) => closed, Err(error) => { tracing::warn!(community = %record.id, host = %record.host, %error, "community archived but disconnect propagation is pending"); diff --git a/crates/buzz-relay/src/audio/handler.rs b/crates/buzz-relay/src/audio/handler.rs index 6e6d467d092..01f8dc6f153 100644 --- a/crates/buzz-relay/src/audio/handler.rs +++ b/crates/buzz-relay/src/audio/handler.rs @@ -1679,7 +1679,7 @@ mod tests { let registry = crate::state::CommunityConnectionRegistry::new(); let community = buzz_core::CommunityId::from_uuid(Uuid::new_v4()); let _guard = registry.register(Uuid::new_v4(), community, control); - assert_eq!(registry.disconnect_community(community), 1); + assert_eq!(registry.disconnect_deleted_community(community), 1); let messages = Arc::new(Mutex::new(Vec::new())); let sink = MockSink { messages: Arc::clone(&messages), diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index e284e7fa6a2..c98a3b6481e 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -868,6 +868,13 @@ pub(crate) mod tests { rx } + fn archived_community_disconnect_reason() -> watch::Receiver> + { + let (tx, rx) = watch::channel(None); + tx.send_replace(Some(CommunityDisconnectReason::CommunityArchived)); + rx + } + fn text_payloads(messages: &[WsMessage]) -> Vec { messages .iter() @@ -1069,6 +1076,36 @@ pub(crate) mod tests { } } + #[tokio::test] + async fn send_loop_sends_policy_close_when_community_is_archived() { + let (_data_tx, data_rx) = mpsc::channel(1); + let (_ctrl_tx, ctrl_rx) = mpsc::channel(1); + let (_restart_tx, restart_rx) = mpsc::channel(1); + let cancel = CancellationToken::new(); + cancel.cancel(); + + let (sink, state) = MockSink::new(None); + send_loop_inner( + sink, + data_rx, + ctrl_rx, + restart_rx, + cancel, + archived_community_disconnect_reason(), + ) + .await; + + let state = state.lock().expect("mock sink poisoned"); + assert_eq!(state.messages.len(), 1); + match &state.messages[0] { + WsMessage::Close(Some(close)) => { + assert_eq!(close.code, axum::extract::ws::close_code::POLICY); + assert_eq!(close.reason.as_str(), "community archived"); + } + other => panic!("expected one 1008 archive close, got {other:?}"), + } + } + #[tokio::test] async fn send_loop_sends_bare_close_for_ordinary_cancellation() { let (_data_tx, data_rx) = mpsc::channel(1); diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 206f0329c0e..e37eab35d32 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1054,10 +1054,41 @@ async fn run_relay_main(boot: BootTracker) -> 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::DisconnectCommunity { + archived_at, + } => { + if let Some(archived_at) = archived_at { + match state_for_conn_ctrl + .db + .with_community_archive_fence( + scoped.community_id, + archived_at, + || { + state_for_conn_ctrl + .community_connections + .disconnect_archived_community(scoped.community_id) + }, + ) + .await + { + Ok(Some(_)) => {} + Ok(None) => tracing::info!( + community = %scoped.community_id, + %archived_at, + "ignored stale archived-community disconnect" + ), + Err(error) => tracing::warn!( + community = %scoped.community_id, + %archived_at, + %error, + "could not verify archived-community disconnect; retaining sockets" + ), + } + } else { + state_for_conn_ctrl + .community_connections + .disconnect_deleted_community(scoped.community_id); + } } buzz_pubsub::conn_control::ConnControl::DisconnectPubkey { pubkey, diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 95372d5bc3b..5eeb3654a44 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -27,6 +27,7 @@ use buzz_pubsub::rate_limiter::RedisRateLimiter; use buzz_pubsub::{PubSubManager, RedisNip98ReplayGuard}; use buzz_search::SearchService; use buzz_workflow::WorkflowEngine; +use chrono::{DateTime, Utc}; use deadpool_redis; use crate::audio::AudioRoomManager; @@ -38,17 +39,21 @@ pub(crate) type ScopedPubkeyKey = (CommunityId, [u8; 32]); /// Why a community-bound socket is being asked to stop. /// -/// Only deletion is externally attributed today. Ordinary lifecycle exits keep -/// using cancellation alone and therefore retain the existing bare-close -/// behavior. +/// Ordinary lifecycle exits keep using cancellation alone and therefore retain +/// the existing bare-close behavior. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum CommunityDisconnectReason { + CommunityArchived, CommunityDeleted, } impl CommunityDisconnectReason { pub(crate) fn close_message(self) -> WsMessage { match self { + Self::CommunityArchived => WsMessage::Close(Some(axum::extract::ws::CloseFrame { + code: axum::extract::ws::close_code::POLICY, + reason: WsUtf8Bytes::from_static("community archived"), + })), Self::CommunityDeleted => WsMessage::Close(Some(axum::extract::ws::CloseFrame { code: axum::extract::ws::close_code::POLICY, reason: WsUtf8Bytes::from_static("community deleted"), @@ -78,9 +83,8 @@ impl CommunityConnectionControl { self.reason_tx.subscribe() } - fn disconnect_community(&self) { - self.reason_tx - .send_replace(Some(CommunityDisconnectReason::CommunityDeleted)); + fn disconnect_community(&self, reason: CommunityDisconnectReason) { + self.reason_tx.send_replace(Some(reason)); self.cancel.cancel(); } } @@ -148,13 +152,25 @@ impl CommunityConnectionRegistry { } } - /// Disconnects every socket type currently bound to `community_id` and - /// attributes the close to community deletion. - pub fn disconnect_community(&self, community_id: CommunityId) -> usize { + /// Disconnects every socket type currently bound to an archived community. + pub fn disconnect_archived_community(&self, community_id: CommunityId) -> usize { + self.disconnect_community(community_id, CommunityDisconnectReason::CommunityArchived) + } + + /// Disconnects every socket type currently bound to a permanently deleted community. + pub fn disconnect_deleted_community(&self, community_id: CommunityId) -> usize { + self.disconnect_community(community_id, CommunityDisconnectReason::CommunityDeleted) + } + + fn disconnect_community( + &self, + community_id: CommunityId, + reason: CommunityDisconnectReason, + ) -> usize { let mut closed = 0; for entry in self.connections.iter() { if entry.value().0 == community_id { - entry.value().1.disconnect_community(); + entry.value().1.disconnect_community(reason); closed += 1; } } @@ -212,21 +228,20 @@ pub(crate) async fn run_registered_community_connection( +async fn revalidate_registered_communities( registry: &CommunityConnectionRegistry, - mut check_active: Check, + mut revalidate: Revalidate, ) -> (usize, Vec<(CommunityId, buzz_db::DbError)>) where - Check: FnMut(CommunityId) -> CheckFuture, - CheckFuture: Future>, + Revalidate: FnMut(CommunityId) -> RevalidateFuture, + RevalidateFuture: 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) => {} + match revalidate(community_id).await { + Ok(disconnected) => closed += disconnected, Err(error) => failures.push((community_id, error)), } } @@ -1231,14 +1246,25 @@ impl AppState { pub async fn disconnect_community_clusterwide( &self, tenant: &TenantContext, - ) -> Result { + archived_at: DateTime, + ) -> anyhow::Result { let closed = self - .community_connections - .disconnect_community(tenant.community()); + .db + .with_community_archive_fence(tenant.community(), archived_at, || { + self.community_connections + .disconnect_archived_community(tenant.community()) + }) + .await? + .unwrap_or(0); self.community_disconnect_publish_attempts .fetch_add(1, Ordering::Relaxed); self.pubsub - .publish_conn_control(tenant, &ConnControl::DisconnectCommunity) + .publish_conn_control( + tenant, + &ConnControl::DisconnectCommunity { + archived_at: Some(archived_at), + }, + ) .await?; Ok(closed) } @@ -1249,11 +1275,24 @@ impl AppState { /// 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_for_maintenance(community_id) - }) - .await; + let (closed, failures) = revalidate_registered_communities( + &self.community_connections, + |community_id| async move { + self.db + .with_inactive_community_fence(community_id, |archived_at| { + if archived_at.is_some() { + self.community_connections + .disconnect_archived_community(community_id) + } else { + self.community_connections + .disconnect_deleted_community(community_id) + } + }) + .await + .map(|disconnected| disconnected.unwrap_or(0)) + }, + ) + .await; for (community_id, error) in failures { tracing::warn!(%community_id, %error, "community lifecycle revalidation failed; retaining its sockets until next tick"); } @@ -1924,17 +1963,17 @@ pub(crate) mod tests { let _audio_a_guard = registry.register(Uuid::new_v4(), community_a, audio_a_control); let _ordinary_b_guard = registry.register(Uuid::new_v4(), community_b, ordinary_b_control); - assert_eq!(registry.disconnect_community(community_a), 2); + assert_eq!(registry.disconnect_archived_community(community_a), 2); assert!(ordinary_a.is_cancelled()); assert!(audio_a.is_cancelled()); assert!(!ordinary_b.is_cancelled()); assert_eq!( *ordinary_a_reason.borrow(), - Some(CommunityDisconnectReason::CommunityDeleted) + Some(CommunityDisconnectReason::CommunityArchived) ); assert_eq!( *audio_a_reason.borrow(), - Some(CommunityDisconnectReason::CommunityDeleted) + Some(CommunityDisconnectReason::CommunityArchived) ); assert_eq!(*ordinary_b_reason.borrow(), None); } @@ -1988,7 +2027,7 @@ pub(crate) mod tests { _ = registered.notified() => {} _ = &mut future => panic!("revalidation should be paused"), } - assert_eq!(registry.disconnect_community(community), 1); + assert_eq!(registry.disconnect_archived_community(community), 1); resume.notify_one(); future.await; assert!(cancel_during.is_cancelled()); @@ -2020,17 +2059,20 @@ pub(crate) mod tests { CommunityConnectionControl::new(cancel_c.clone()), ); - let (closed, failures) = - revalidate_registered_communities(®istry, |community| async move { + let registry_for_revalidation = ®istry; + let (closed, failures) = revalidate_registered_communities(®istry, |community| { + let registry = registry_for_revalidation; + async move { if community == failed { Err(buzz_db::DbError::InvalidData( "injected lookup failure".into(), )) } else { - Ok(false) + Ok(registry.disconnect_archived_community(community)) } - }) - .await; + } + }) + .await; assert_eq!(closed, 2); assert!(cancel_a.is_cancelled()); @@ -2044,6 +2086,46 @@ pub(crate) mod tests { ); } + #[tokio::test] + async fn periodic_revalidation_disconnects_inside_the_fenced_callback() { + 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, + CommunityConnectionControl::new(cancel.clone()), + ); + let entered = Arc::new(tokio::sync::Notify::new()); + let resume = Arc::new(tokio::sync::Notify::new()); + + let future = revalidate_registered_communities(®istry, |community_id| { + let entered = Arc::clone(&entered); + let resume = Arc::clone(&resume); + let registry = ®istry; + async move { + entered.notify_one(); + resume.notified().await; + Ok(registry.disconnect_archived_community(community_id)) + } + }); + tokio::pin!(future); + tokio::select! { + _ = entered.notified() => {} + _ = &mut future => panic!("revalidation should be paused inside the fence"), + } + assert!( + !cancel.is_cancelled(), + "the helper must not disconnect outside the fenced callback" + ); + + resume.notify_one(); + let (closed, failures) = future.await; + assert_eq!(closed, 1); + assert!(failures.is_empty()); + assert!(cancel.is_cancelled()); + } + #[test] fn community_lifecycle_guard_deregisters_on_early_return() { let registry = CommunityConnectionRegistry::new(); @@ -2059,7 +2141,7 @@ pub(crate) mod tests { drop(guard); assert!(registry.bound_communities().is_empty()); - assert_eq!(registry.disconnect_community(community), 0); + assert_eq!(registry.disconnect_archived_community(community), 0); assert!(!cancel.is_cancelled()); }