diff --git a/Justfile b/Justfile index 67f5c391be..8add201370 100644 --- a/Justfile +++ b/Justfile @@ -366,8 +366,9 @@ test-unit: if command -v cargo-nextest &>/dev/null; then cargo nextest run -p buzz-core -p buzz-auth --lib cargo nextest run -p buzz-audit --lib - # S4 cross-pod NIP-FI disconnect payload tests (infra-free). - cargo nextest run -p buzz-pubsub --lib -E 'test(/^conn_control::tests::nip_fi_disconnect_/)' + # Cross-pod disconnect payload tests (NIP-FI and community archive + # fence); all infra-free. + cargo nextest run -p buzz-pubsub --lib -E 'test(/^conn_control::tests::/)' # buzz-auth NIP-FI verifier doctests. The sealed-authority # `compile_fail` doctests prove the default-feature public API alone # cannot forge the issuer→JWKS authority; nextest does not run @@ -406,8 +407,10 @@ test-unit: # suites are otherwise absent from the infra-free unit lane. cargo nextest run -p buzz-media --lib \ -E 'test(=bucket_index::tests::bucket_snapshot_json_round_trip_preserves_community_keys)' + # buzz-admin: storage-snapshot worker plus the community archive + # command parser, validation, and propagation-evidence tests. cargo nextest run -p buzz-admin \ - -E 'test(storage_snapshot)' + -E 'test(storage_snapshot) + test(/^communities::tests::/) + test(/^tests::communities_/)' # Multi-tenant conformance gate (buzz-conformance): the independent # replay checker + golden fixtures. No infra — pure in-process trace # replay — so it belongs in the unit job. Run all targets (lib + the @@ -486,7 +489,7 @@ test-unit: # are infra-free; config parses env into a struct and never connects. # `^config::` is anchored so it does not also select nip_fi_config. cargo nextest run -p buzz-relay --lib --bin buzz-relay \ - -E 'test(/^api::admin::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' + -E 'test(/^api::admin::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' # Note on audio::join::tests scope: the full suite is infra-free (no DB, # no Redis). The infra-free audio/FI regression witnesses — bootstrap # ordering barrier, CommitConfirmed arm, pending-close invisibility, diff --git a/crates/buzz-admin/src/communities.rs b/crates/buzz-admin/src/communities.rs new file mode 100644 index 0000000000..d3232990ca --- /dev/null +++ b/crates/buzz-admin/src/communities.rs @@ -0,0 +1,516 @@ +//! 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. Echoed in the + /// command's JSON output only; not stored. + #[arg(long)] + operator_id: String, + /// Human-readable reason for the action. Echoed in the command's JSON + /// output only; not stored. + #[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. Echoed in the + /// command's JSON output only; not stored. + #[arg(long)] + operator_id: String, + /// Human-readable reason for the action. Echoed in the command's JSON + /// output only; not stored. + #[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 = match db + .unarchive_community_owned_by(&host, &owner_pubkey) + .await? + { + buzz_db::UnarchiveCommunityResult::Unarchived(record) => record, + buzz_db::UnarchiveCommunityResult::DeletionPending => { + anyhow::bail!("community deletion is pending; unarchive is refused") + } + buzz_db::UnarchiveCommunityResult::NotFound => anyhow::bail!( + "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 a48d8297a9..8700f926df 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; mod storage_snapshot_startup; @@ -102,6 +103,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)] @@ -181,6 +187,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?; @@ -911,4 +918,153 @@ mod tests { Command::PartitionAudit { months_ahead: 6 } )); } + + 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/runtime/observability.rs b/crates/buzz-db/src/runtime/observability.rs index 3708c26dfa..99256f3c2f 100644 --- a/crates/buzz-db/src/runtime/observability.rs +++ b/crates/buzz-db/src/runtime/observability.rs @@ -247,17 +247,21 @@ pub(crate) enum TransactionOperation { AcceptPushLeaseEvent, BeginCommunityDeletionQuiescing, FenceCommunityDeletion, + CommunityArchiveFence, + InactiveCommunityFence, } impl TransactionOperation { #[cfg(test)] - pub(crate) const ALL: [Self; 6] = [ + pub(crate) const ALL: [Self; 8] = [ Self::ReplaceParameterizedEvent, Self::ReplaceAddressableEvent, Self::PublishNip43MembershipLocked, Self::AcceptPushLeaseEvent, Self::BeginCommunityDeletionQuiescing, Self::FenceCommunityDeletion, + Self::CommunityArchiveFence, + Self::InactiveCommunityFence, ]; pub(crate) const fn as_str(self) -> &'static str { @@ -268,6 +272,24 @@ impl TransactionOperation { Self::AcceptPushLeaseEvent => "accept_push_lease_event", Self::BeginCommunityDeletionQuiescing => "begin_community_deletion_quiescing", Self::FenceCommunityDeletion => "fence_community_deletion", + Self::CommunityArchiveFence => "community_archive_fence", + Self::InactiveCommunityFence => "inactive_community_fence", + } + } + + /// Whether this transaction's writer checkout also feeds the legacy + /// `buzz_db_pool_acquire_wait_seconds` / `buzz_db_pool_acquisitions_total` + /// families. Only seams that predate operation-labelled pool metrics do; + /// later seams must not change those families' observed population. + const fn emits_legacy_pool_metrics(self) -> bool { + match self { + Self::ReplaceParameterizedEvent + | Self::ReplaceAddressableEvent + | Self::PublishNip43MembershipLocked + | Self::AcceptPushLeaseEvent + | Self::BeginCommunityDeletionQuiescing + | Self::FenceCommunityDeletion => true, + Self::CommunityArchiveFence | Self::InactiveCommunityFence => false, } } @@ -277,9 +299,10 @@ impl TransactionOperation { | Self::ReplaceAddressableEvent | Self::PublishNip43MembershipLocked | Self::AcceptPushLeaseEvent => WriterOperation::EventWrite, - Self::BeginCommunityDeletionQuiescing | Self::FenceCommunityDeletion => { - WriterOperation::Maintenance - } + Self::BeginCommunityDeletionQuiescing + | Self::FenceCommunityDeletion + | Self::CommunityArchiveFence + | Self::InactiveCommunityFence => WriterOperation::Maintenance, } } } @@ -525,7 +548,12 @@ pub(crate) async fn begin_transaction( pool: &sqlx::PgPool, operation: TransactionOperation, ) -> sqlx::Result<(sqlx::Transaction<'static, sqlx::Postgres>, TransactionTimer)> { - let connection = acquire_writer_with_legacy_metrics(pool, operation.writer_operation()).await?; + let writer = operation.writer_operation(); + let connection = if operation.emits_legacy_pool_metrics() { + acquire_writer_with_legacy_metrics(pool, writer).await? + } else { + acquire_writer(pool, writer).await? + }; let transaction = sqlx::Transaction::begin(connection, None).await?; Ok((transaction, TransactionTimer::start(operation))) } @@ -597,9 +625,9 @@ impl Drop for TransactionTimer { mod tests { use super::{ acquire_reader_with_legacy_metrics, acquire_writer, acquire_writer_with_legacy_metrics, - observe_advisory_lock, record_pool_acquire, refresh_pool_waiters, LockType, Outcome, - PoolAcquireAttempt, PoolOperation, ReaderOperation, TransactionOperation, TransactionTimer, - WriterOperation, + begin_transaction, observe_advisory_lock, record_pool_acquire, refresh_pool_waiters, + LockType, Outcome, PoolAcquireAttempt, PoolOperation, ReaderOperation, + TransactionOperation, TransactionTimer, WriterOperation, }; use metrics_util::debugging::{DebugValue, DebuggingRecorder}; use std::collections::{BTreeMap, BTreeSet}; @@ -658,6 +686,8 @@ mod tests { "accept_push_lease_event", "begin_community_deletion_quiescing", "fence_community_deletion", + "community_archive_fence", + "inactive_community_fence", ] ); } @@ -1007,6 +1037,21 @@ mod tests { 0 ); + for operation in [ + TransactionOperation::CommunityArchiveFence, + TransactionOperation::InactiveCommunityFence, + ] { + let Err(error) = begin_transaction(&pool, operation).await else { + panic!("closed newly instrumented transaction seam errors"); + }; + assert!(matches!(error, sqlx::Error::PoolClosed)); + } + assert_eq!( + legacy_acquisition_count(&snapshotter.snapshot().into_vec()), + 0, + "community lifecycle fences must not expand the legacy pool families" + ); + let error = acquire_writer_with_legacy_metrics(&pool, WriterOperation::EventWrite) .await .expect_err("closed legacy seam errors"); @@ -1015,6 +1060,20 @@ mod tests { legacy_acquisition_count(&snapshotter.snapshot().into_vec()), 1 ); + + let Err(error) = + begin_transaction(&pool, TransactionOperation::FenceCommunityDeletion).await + else { + panic!("closed legacy transaction seam errors"); + }; + assert!(matches!(error, sqlx::Error::PoolClosed)); + // `snapshot()` drains counters, so each assertion sees only the + // acquisitions recorded since the previous snapshot. + assert_eq!( + legacy_acquisition_count(&snapshotter.snapshot().into_vec()), + 1, + "pre-existing deletion fences keep their legacy pool population" + ); } fn legacy_acquisition_count( @@ -1505,7 +1564,7 @@ mod tests { // writer acquisition has been attributed to `event_write`. let _ = writer_db.begin_event_write_transaction(test_scope).await; let _ = writer_db - .is_community_active_for_maintenance(test_scope) + .with_inactive_community_fence(test_scope, |_| ()) .await; let _ = writer_db.usage_community_count().await; let _ = writer_db.reap_expired_ephemeral_channels().await; diff --git a/crates/buzz-db/src/store/community.rs b/crates/buzz-db/src/store/community.rs index cf4597662f..353162f22e 100644 --- a/crates/buzz-db/src/store/community.rs +++ b/crates/buzz-db/src/store/community.rs @@ -149,32 +149,11 @@ impl Db { /// Returns whether a community id still exists in the active lifecycle state. #[datastore_span(name = "is_community_active", system = "postgresql")] pub async fn is_community_active(&self, community_id: CommunityId) -> Result { - self.is_community_active_with_operation( - community_id, + let mut connection = crate::observability::acquire_writer( + &self.pool, crate::observability::WriterOperation::Authorization, ) - .await - } - - /// Background lifecycle revalidation variant of [`Self::is_community_active`]. - #[datastore_span(name = "is_community_active_for_maintenance", system = "postgresql")] - pub async fn is_community_active_for_maintenance( - &self, - community_id: CommunityId, - ) -> Result { - self.is_community_active_with_operation( - community_id, - crate::observability::WriterOperation::Maintenance, - ) - .await - } - - async fn is_community_active_with_operation( - &self, - community_id: CommunityId, - operation: crate::observability::WriterOperation, - ) -> Result { - let mut connection = crate::observability::acquire_writer(&self.pool, operation).await?; + .await?; let active = sqlx::query_scalar::<_, bool>( "SELECT EXISTS(SELECT 1 FROM communities WHERE id = $1 AND archived_at IS NULL AND deleted_at IS NULL AND deletion_state = 'active')", ) @@ -184,6 +163,107 @@ 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. + /// + /// `FOR NO KEY UPDATE` conflicts with unarchive's `FOR UPDATE` and its + /// `archived_at` update, but not with the `FOR KEY SHARE` locks that child + /// inserts referencing `communities(id)` take, so the fence never waits + /// on ordinary community writes. + #[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, transaction_timer) = crate::observability::begin_transaction( + &self.pool, + crate::observability::TransactionOperation::CommunityArchiveFence, + ) + .await?; + transaction_timer + .observe(async move { + 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 NO KEY 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)) + }) + .await + } + + /// 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. + /// Revalidation calls it for each bound community whose unlocked + /// lifecycle read is not active, so it takes `FOR NO KEY UPDATE` for the + /// same reason as [`Self::with_community_archive_fence`]. + #[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, transaction_timer) = crate::observability::begin_transaction( + &self.pool, + crate::observability::TransactionOperation::InactiveCommunityFence, + ) + .await?; + transaction_timer + .observe(async move { + 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 NO KEY 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)) + }) + .await + } + /// Returns a community by host regardless of lifecycle state. Operator-plane only. #[datastore_span( name = "lookup_community_by_host_for_management", @@ -526,6 +606,14 @@ impl Db { } /// Idempotently archives a community when the asserted pubkey is its current owner. + /// + /// Locks the community row before reading ownership so a concurrent owner + /// transfer, whose `FOR UPDATE` conflicts with this `FOR NO KEY UPDATE`, + /// has one serial order with the archive. The weaker lock lets the archive + /// proceed past foreign-key child inserts' `FOR KEY SHARE`. A single + /// `UPDATE ... FROM relay_members` would keep its pre-lock ownership + /// snapshot after waiting and could archive the new owner's community on + /// the old owner's assertion. #[datastore_span(name = "archive_community_owned_by", system = "postgresql")] pub async fn archive_community_owned_by( &self, @@ -533,37 +621,52 @@ impl Db { owner_pubkey: &str, protected_deployment_host: &str, ) -> Result> { - let mut connection = crate::observability::acquire_writer( + let connection = crate::observability::acquire_writer( &self.pool, crate::observability::WriterOperation::Authorization, ) .await?; - 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) - AND c.deletion_state = 'active' - AND c.deleted_at IS NULL - RETURNING c.id, c.host, c.archived_at"#, + let mut tx = sqlx::Transaction::begin(connection, None).await?; + let target = sqlx::query( + "SELECT id FROM communities \ + WHERE lower(host) = lower($1) AND lower(host) <> lower($2) \ + AND deletion_state = 'active' AND deleted_at IS NULL FOR NO KEY UPDATE", ) .bind(normalized_host) - .bind(owner_pubkey) .bind(protected_deployment_host) - .fetch_optional(&mut *connection) + .fetch_optional(&mut *tx) .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() + let Some(target) = target else { + tx.rollback().await?; + return Ok(None); + }; + let community_id: Uuid = target.try_get("id")?; + let is_owner: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM relay_members \ + WHERE community_id = $1 AND lower(pubkey) = lower($2) AND role = 'owner')", + ) + .bind(community_id) + .bind(owner_pubkey) + .fetch_one(&mut *tx) + .await?; + if !is_owner { + tx.rollback().await?; + return Ok(None); + } + let row = sqlx::query( + "UPDATE communities SET archived_at = COALESCE(archived_at, now()) \ + WHERE id = $1 RETURNING id, host, archived_at", + ) + .bind(community_id) + .fetch_one(&mut *tx) + .await?; + let archived = ArchivedCommunityRecord { + id: CommunityId::from_uuid(row.try_get("id")?), + host: row.try_get("host")?, + archived_at: row.try_get("archived_at")?, + }; + tx.commit().await?; + Ok(Some(archived)) } /// Idempotently restores a community when the asserted pubkey is its current owner. @@ -773,6 +876,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", @@ -786,14 +891,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", ); @@ -831,6 +938,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", @@ -1012,6 +1120,332 @@ mod postgres_tests { assert_eq!(retry, UnarchiveCommunityResult::Unarchived(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") + ); + assert!(matches!( + db.unarchive_community_owned_by(&host, &owner) + .await + .expect("unarchive community"), + UnarchiveCommunityResult::Unarchived(_) + )); + 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") + ); + } + + async fn named_contender_db(application_name: &str) -> Db { + use std::str::FromStr as _; + let options = + sqlx::postgres::PgConnectOptions::from_str(&crate::test_support::database_url()) + .expect("parse test database URL") + .application_name(application_name); + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect_with(options) + .await + .expect("connect contender DB"); + Db::from_pool(pool) + } + + async fn wait_for_lock_wait(db: &Db, application_name: &str) { + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pg_stat_activity \ + WHERE datname = current_database() AND application_name = $1 \ + AND wait_event_type = 'Lock')", + ) + .bind(application_name) + .fetch_one(&db.pool) + .await + .expect("inspect contender lock wait"); + if waiting { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("contender reached a lock wait"); + } + + /// An archive that queues behind an ownership transfer must re-check + /// ownership after the transfer commits. Neither serial order lets the + /// previous owner archive the new owner's community. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[ignore = "requires Postgres"] + async fn archive_queued_behind_transfer_rechecks_the_current_owner() { + let db = setup_db().await; + let host = format!("archive-transfer-{}.example", Uuid::new_v4().simple()); + let previous_owner = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple()); + let new_owner = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple()); + let CreateCommunityWithOwnerResult::Created(created) = db + .create_community_with_owner(&host, &previous_owner) + .await + .expect("create community") + else { + panic!("expected new community"); + }; + + // Pause the transfer after it locks the community row: it next locks + // the owner rows, which this transaction holds. + let mut owner_rows_gate = db.pool.begin().await.expect("begin owner-row gate"); + sqlx::query( + "SELECT 1 FROM relay_members WHERE community_id = $1 AND role = 'owner' FOR UPDATE", + ) + .bind(created.id.as_uuid()) + .execute(&mut *owner_rows_gate) + .await + .expect("hold owner rows"); + + let transfer_name = format!("archive-transfer-tx-{}", Uuid::new_v4().simple()); + let transfer_db = named_contender_db(&transfer_name).await; + let transfer = tokio::spawn({ + let community = created.id; + let previous_owner = previous_owner.clone(); + let new_owner = new_owner.clone(); + async move { + transfer_db + .transfer_ownership(community, &new_owner, &previous_owner) + .await + } + }); + wait_for_lock_wait(&db, &transfer_name).await; + + let archive_name = format!("archive-transfer-archive-{}", Uuid::new_v4().simple()); + let archive_db = named_contender_db(&archive_name).await; + let archive = tokio::spawn({ + let host = host.clone(); + let previous_owner = previous_owner.clone(); + async move { + archive_db + .archive_community_owned_by(&host, &previous_owner, "protected.example") + .await + } + }); + wait_for_lock_wait(&db, &archive_name).await; + + owner_rows_gate + .rollback() + .await + .expect("release owner-row gate"); + let transfer = tokio::time::timeout(std::time::Duration::from_secs(5), transfer) + .await + .expect("transfer completes") + .expect("join transfer") + .expect("transfer"); + let archive = tokio::time::timeout(std::time::Duration::from_secs(5), archive) + .await + .expect("archive completes") + .expect("join archive") + .expect("archive"); + + assert!(matches!( + transfer, + crate::relay_members::TransferResult::Transferred { .. } + )); + assert_eq!( + archive, None, + "the previous owner's archive must be refused" + ); + assert!(db + .is_community_active(created.id) + .await + .expect("community state")); + assert!(db + .archive_community_owned_by(&host, &new_owner, "protected.example") + .await + .expect("new-owner archive") + .is_some()); + } + + #[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!(matches!( + db.unarchive_community_owned_by(&host, &owner) + .await + .expect("unarchive community"), + UnarchiveCommunityResult::Unarchived(_) + )); + assert!(db + .is_community_active(created.id) + .await + .expect("restored community state")); + } + + /// Foreign-key child inserts hold `FOR KEY SHARE` on the community row for + /// their whole transaction. Archive and the lifecycle fences must not queue + /// behind them, or a busy community stalls the operator archive and + /// revalidation and conn-control on every pod. `FOR UPDATE` would block + /// here; `FOR NO KEY UPDATE` must not. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn lifecycle_fences_do_not_wait_on_foreign_key_share_locks() { + let db = setup_db().await; + let host = format!("fence-key-share-{}.example", Uuid::new_v4().simple()); + let owner = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple()); + let CreateCommunityWithOwnerResult::Created(created) = db + .create_community_with_owner(&host, &owner) + .await + .expect("create community") + else { + panic!("expected new community"); + }; + let mut child_writer = db.pool.begin().await.expect("begin child writer"); + sqlx::query("SELECT 1 FROM communities WHERE id = $1 FOR KEY SHARE") + .bind(created.id.as_uuid()) + .execute(&mut *child_writer) + .await + .expect("hold the lock a child-row insert takes"); + + let wait = std::time::Duration::from_secs(5); + let archived = tokio::time::timeout( + wait, + db.archive_community_owned_by(&host, &owner, "protected.example"), + ) + .await + .expect("archive must not wait on FOR KEY SHARE") + .expect("archive community") + .expect("owned community"); + assert_eq!( + tokio::time::timeout( + wait, + db.with_inactive_community_fence(created.id, |_| "inactive") + ) + .await + .expect("inactive fence must not wait on FOR KEY SHARE") + .expect("inactive fence"), + Some("inactive") + ); + assert_eq!( + tokio::time::timeout( + wait, + db.with_community_archive_fence(created.id, archived.archived_at, || "archived") + ) + .await + .expect("archive fence must not wait on FOR KEY SHARE") + .expect("archive fence"), + Some("archived") + ); + child_writer.rollback().await.expect("release child writer"); + } + #[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 0bf88689ea..e610e18259 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; @@ -108,7 +109,18 @@ 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` + /// (permanent deletion, or archive from a relay predating the timestamp) + /// closes only while the row is inactive, and fails closed as a deletion + /// when the receiver cannot evaluate that fence. + 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 @@ -329,18 +341,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 78c4c3eeff..4a1fb717c1 100644 --- a/crates/buzz-relay/src/api/operator.rs +++ b/crates/buzz-relay/src/api/operator.rs @@ -374,7 +374,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"); @@ -2381,6 +2384,359 @@ mod postgres_tests { ); } + /// A relay that predates the archive transition timestamp publishes a bare + /// `DisconnectCommunity`. A current relay must still apply it under the + /// lifecycle fence: never against a restored community, and with the + /// archive close reason against an archived one. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn legacy_disconnect_command_is_fenced_by_community_lifecycle() { + use crate::state::{CommunityConnectionControl, CommunityDisconnectReason}; + use tokio_util::sync::CancellationToken; + + 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()); + assert_eq!( + provision_community(Arc::clone(&state), &operator, &host, &owner) + .await + .status(), + StatusCode::OK + ); + let community = state + .db + .lookup_community_by_host(&host) + .await + .expect("active admission lookup") + .expect("active community") + .id; + let owner_hex = owner.public_key().to_hex(); + + // Active (for example, unarchived before the command arrived): keep sockets. + let active_cancel = CancellationToken::new(); + let active_control = CommunityConnectionControl::new(active_cancel.clone()); + let active_reason = active_control.disconnect_reason(); + let active_guard = + state + .community_connections + .register(Uuid::new_v4(), community, active_control); + assert_eq!( + state + .apply_community_disconnect(community, None) + .await + .expect("fenced legacy disconnect"), + None + ); + assert!( + !active_cancel.is_cancelled(), + "a legacy disconnect must not close an active community" + ); + assert_eq!(*active_reason.borrow(), None); + + // Archived: close with the archive reason, not `community deleted`. + state + .db + .archive_community_owned_by(&host, &owner_hex, "protected.example") + .await + .expect("archive community") + .expect("owned community"); + assert_eq!( + state + .apply_community_disconnect(community, None) + .await + .expect("fenced legacy disconnect"), + Some(1) + ); + assert!(active_cancel.is_cancelled()); + assert_eq!( + *active_reason.borrow(), + Some(CommunityDisconnectReason::CommunityArchived) + ); + + // Fenced deletion: this is the state `buzz-deletion` publishes the bare + // command from, so it must close with the deletion reason even though + // the row is also archived. + drop(active_guard); + let deleted_cancel = CancellationToken::new(); + let deleted_control = CommunityConnectionControl::new(deleted_cancel.clone()); + let deleted_reason = deleted_control.disconnect_reason(); + let _deleted_guard = + state + .community_connections + .register(Uuid::new_v4(), community, deleted_control); + let mut tx = state + .db + .pool() + .begin() + .await + .expect("begin deletion fixture"); + sqlx::query( + "SELECT set_config('buzz.deletion_executor_community', $1, true), \ + set_config('buzz.deletion_fence_generation', '0', true)", + ) + .bind(community.to_string()) + .execute(&mut *tx) + .await + .expect("enter deletion executor scope"); + sqlx::query("UPDATE communities SET deletion_state = 'fenced' WHERE id = $1") + .bind(community.as_uuid()) + .execute(&mut *tx) + .await + .expect("fence community deletion"); + tx.commit().await.expect("commit deletion fixture"); + assert_eq!( + state + .apply_community_disconnect(community, None) + .await + .expect("fenced legacy disconnect"), + Some(1) + ); + assert!(deleted_cancel.is_cancelled()); + assert_eq!( + *deleted_reason.borrow(), + Some(CommunityDisconnectReason::CommunityDeleted) + ); + } + + /// Lifecycle revalidation runs for every live community on every pod each + /// tick. Active communities must be checked without the row lock, so a + /// conflicting lock on an active row must not stall the tick, while an + /// archived community is still closed under the fence. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn revalidation_skips_the_row_lock_for_active_communities() { + use crate::state::{CommunityConnectionControl, CommunityDisconnectReason}; + use tokio_util::sync::CancellationToken; + + 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()); + assert_eq!( + provision_community(Arc::clone(&state), &operator, &host, &owner) + .await + .status(), + StatusCode::OK + ); + let community = state + .db + .lookup_community_by_host(&host) + .await + .expect("active admission lookup") + .expect("active community") + .id; + let cancel = CancellationToken::new(); + let control = CommunityConnectionControl::new(cancel.clone()); + let reason = control.disconnect_reason(); + let guard = state + .community_connections + .register(Uuid::new_v4(), community, control); + + let mut row_lock = state.db.pool().begin().await.expect("begin row lock"); + sqlx::query("SELECT 1 FROM communities WHERE id = $1 FOR UPDATE") + .bind(community.as_uuid()) + .execute(&mut *row_lock) + .await + .expect("hold the community row lock"); + assert_eq!( + tokio::time::timeout( + std::time::Duration::from_secs(5), + state.revalidate_live_communities() + ) + .await + .expect("revalidation must not wait on an active community's row lock"), + 0 + ); + row_lock.rollback().await.expect("release row lock"); + assert!(!cancel.is_cancelled()); + + state + .db + .archive_community_owned_by(&host, &owner.public_key().to_hex(), "protected.example") + .await + .expect("archive community") + .expect("owned community"); + assert_eq!(state.revalidate_live_communities().await, 1); + assert!(cancel.is_cancelled()); + assert_eq!( + *reason.borrow(), + Some(CommunityDisconnectReason::CommunityArchived) + ); + + drop(guard); + + // A community fenced for deletion but never archived must not pass the + // unlocked active pre-check; it closes with the deletion reason. + let deleted_host = format!("community-{}.example", Uuid::new_v4().simple()); + assert_eq!( + provision_community(Arc::clone(&state), &operator, &deleted_host, &owner) + .await + .status(), + StatusCode::OK + ); + let deleted_community = state + .db + .lookup_community_by_host(&deleted_host) + .await + .expect("active admission lookup") + .expect("active community") + .id; + let deleted_cancel = CancellationToken::new(); + let deleted_control = CommunityConnectionControl::new(deleted_cancel.clone()); + let deleted_reason = deleted_control.disconnect_reason(); + let _deleted_guard = state.community_connections.register( + Uuid::new_v4(), + deleted_community, + deleted_control, + ); + let mut tx = state + .db + .pool() + .begin() + .await + .expect("begin deletion fixture"); + sqlx::query( + "SELECT set_config('buzz.deletion_executor_community', $1, true), \ + set_config('buzz.deletion_fence_generation', '0', true)", + ) + .bind(deleted_community.to_string()) + .execute(&mut *tx) + .await + .expect("enter deletion executor scope"); + sqlx::query("UPDATE communities SET deletion_state = 'fenced' WHERE id = $1") + .bind(deleted_community.as_uuid()) + .execute(&mut *tx) + .await + .expect("fence community deletion"); + tx.commit().await.expect("commit deletion fixture"); + assert_eq!(state.revalidate_live_communities().await, 1); + assert!(deleted_cancel.is_cancelled()); + assert_eq!( + *deleted_reason.borrow(), + Some(CommunityDisconnectReason::CommunityDeleted) + ); + } + + /// The connection-control consumer serves ban disconnects and archive + /// disconnects from one receiver. An archive disconnect waits on the + /// community row lock, so a pubkey disconnect queued behind it must still + /// close its socket promptly, and the archive disconnect must still land + /// once the lock is released. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn conn_control_consumer_does_not_queue_pubkey_disconnects_behind_a_locked_archive() { + use crate::state::{CommunityConnectionControl, CommunityDisconnectReason}; + use buzz_pubsub::conn_control::{ConnControl, ScopedConnControl}; + use std::time::Duration; + use tokio_util::sync::CancellationToken; + + 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()); + assert_eq!( + provision_community(Arc::clone(&state), &operator, &host, &owner) + .await + .status(), + StatusCode::OK + ); + let community = state + .db + .lookup_community_by_host(&host) + .await + .expect("active admission lookup") + .expect("active community") + .id; + let community_control = CommunityConnectionControl::new(CancellationToken::new()); + let community_reason = community_control.disconnect_reason(); + let _community_guard = state.community_connections.register( + Uuid::new_v4(), + community, + community_control.clone(), + ); + let banned = [7u8; 32]; + let banned_control = CommunityConnectionControl::new(CancellationToken::new()); + banned_control.bind_pubkey(banned); + let banned_reason = banned_control.disconnect_reason(); + let _banned_guard = + state + .community_connections + .register(Uuid::new_v4(), community, banned_control.clone()); + + let archived = state + .db + .archive_community_owned_by(&host, &owner.public_key().to_hex(), "protected.example") + .await + .expect("archive community") + .expect("owned community"); + let mut row_lock = state.db.pool().begin().await.expect("begin row lock"); + sqlx::query("SELECT 1 FROM communities WHERE id = $1 FOR UPDATE") + .bind(community.as_uuid()) + .execute(&mut *row_lock) + .await + .expect("hold the community row lock"); + + let (tx, rx) = tokio::sync::broadcast::channel(8); + let consumer = tokio::spawn(Arc::clone(&state).run_conn_control_consumer(rx)); + tx.send(ScopedConnControl { + community_id: community, + command: ConnControl::DisconnectCommunity { + archived_at: Some(archived.archived_at), + }, + }) + .expect("send community disconnect"); + tx.send(ScopedConnControl { + community_id: community, + command: ConnControl::DisconnectPubkey { + pubkey: banned.to_vec(), + event_id: "ban".to_string(), + reason: "blocked: you are banned from this community".to_string(), + unowned_only: false, + }, + }) + .expect("send pubkey disconnect"); + + tokio::time::timeout( + Duration::from_secs(1), + banned_control.cancellation_token().cancelled(), + ) + .await + .expect("a pubkey disconnect must not wait behind a locked community disconnect"); + assert_eq!( + *banned_reason.borrow(), + Some(CommunityDisconnectReason::AccessRevoked) + ); + assert!( + !community_control.cancellation_token().is_cancelled(), + "the community disconnect must wait for the row lock" + ); + + row_lock.rollback().await.expect("release row lock"); + tokio::time::timeout( + Duration::from_secs(5), + community_control.cancellation_token().cancelled(), + ) + .await + .expect("the community disconnect must land once the row lock is released"); + assert_eq!( + *community_reason.borrow(), + Some(CommunityDisconnectReason::CommunityArchived) + ); + + drop(tx); + tokio::time::timeout(Duration::from_secs(5), consumer) + .await + .expect("the consumer must stop when the broadcast closes") + .expect("consumer task"); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn unarchive_restores_admission_and_is_idempotent_without_changing_ownership() { diff --git a/crates/buzz-relay/src/audio/handler.rs b/crates/buzz-relay/src/audio/handler.rs index c73db1686d..10b0bdd5b8 100644 --- a/crates/buzz-relay/src/audio/handler.rs +++ b/crates/buzz-relay/src/audio/handler.rs @@ -3763,7 +3763,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 f961ea4d67..b341f9b214 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -2103,6 +2103,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() @@ -2464,6 +2471,37 @@ 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, + mpsc::channel(1).1, + 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 ab855308b3..9021d6644f 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1465,43 +1465,8 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { // ban row is the durable backstop; even a dropped command still refuses the // banned member's next auth attempt at the auth seam. { - let state_for_conn_ctrl = Arc::clone(&state); - let mut rx = state_for_conn_ctrl.pubsub.subscribe_conn_control(); - tokio::spawn(async move { - 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, - reason, - unowned_only, - } => { - state_for_conn_ctrl.disconnect_pubkey_local( - scoped.community_id, - &pubkey, - &event_id, - &reason, - unowned_only, - ); - } - }, - Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { - metrics::counter!("buzz_conn_control_lag_total").increment(n); - tracing::warn!("Connection-control consumer lagged by {n} messages"); - } - Err(tokio::sync::broadcast::error::RecvError::Closed) => { - tracing::error!("Connection-control broadcast channel closed"); - break; - } - } - } - }); + let rx = state.pubsub.subscribe_conn_control(); + tokio::spawn(Arc::clone(&state).run_conn_control_consumer(rx)); } // Cross-pod NIP-FI disconnect consumer: receive deny entries from remote diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 954b220685..be6e9ed7c4 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -22,11 +22,12 @@ use buzz_core::CommunityId; use buzz_db::Db; use buzz_media::MediaStorage; use buzz_pubsub::cache_invalidation::CacheInvalidation; -use buzz_pubsub::conn_control::ConnControl; +use buzz_pubsub::conn_control::{ConnControl, ScopedConnControl}; 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,11 +39,11 @@ 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, /// NIP-FI: the connection's proven pubkey was added to the deny set. AuthorizationDenied, @@ -53,6 +54,10 @@ pub(crate) enum CommunityDisconnectReason { 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"), @@ -253,9 +258,8 @@ impl CommunityConnectionControl { self.cancel.cancel(); } - 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(); } @@ -359,13 +363,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; } } @@ -626,21 +642,20 @@ pub(crate) async fn run_registered_community_connection< cancel.cancel(); } -async fn revalidate_registered_communities( +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)), } } @@ -1960,29 +1975,178 @@ 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) } + /// Consumes cross-pod connection-control commands until the broadcast closes. + /// + /// Pubkey disconnects are in-memory and run inline. Community disconnects + /// wait on a writer connection and the community row lock, so each runs in + /// its own task and a ban disconnect never queues behind one. Every + /// community disconnect is fenced against the row, so their relative order + /// does not matter. + pub async fn run_conn_control_consumer( + self: Arc, + mut rx: tokio::sync::broadcast::Receiver, + ) { + loop { + match rx.recv().await { + Ok(scoped) => match scoped.command { + ConnControl::DisconnectCommunity { archived_at } => { + let state = Arc::clone(&self); + let community_id = scoped.community_id; + tokio::spawn(async move { + match state + .apply_community_disconnect(community_id, archived_at) + .await + { + Ok(Some(_)) => {} + Ok(None) => tracing::info!( + community = %community_id, + ?archived_at, + "ignored community disconnect for a community that is active or re-archived" + ), + Err(error) => tracing::warn!( + community = %community_id, + ?archived_at, + %error, + "could not verify archive disconnect; retaining sockets until lifecycle revalidation" + ), + } + }); + } + ConnControl::DisconnectPubkey { + pubkey, + event_id, + reason, + unowned_only, + } => { + self.disconnect_pubkey_local( + scoped.community_id, + &pubkey, + &event_id, + &reason, + unowned_only, + ); + } + }, + Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { + metrics::counter!("buzz_conn_control_lag_total").increment(n); + tracing::warn!("Connection-control consumer lagged by {n} messages"); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + tracing::error!("Connection-control broadcast channel closed"); + break; + } + } + } + } + + /// Applies a cross-pod `DisconnectCommunity` command to local sockets. + /// + /// Both forms act only under the community row lock, so neither can close + /// a community that an unarchive has already restored: + /// - `Some(archived_at)` closes only while that exact archive transition is + /// still current. + /// - `None` (community deletion, or archive published by a relay that + /// predates the transition timestamp) closes only while the row is + /// inactive, with the close reason taken from the row itself. + /// + /// The two forms fail in opposite directions when the fence cannot be + /// evaluated. `Some` is only ever a reversible archive, so it retains the + /// sockets for lifecycle revalidation and returns the error. `None` may be + /// a permanent deletion, so it fails closed with `community deleted`. + /// + /// Returns `None` when the fence found nothing to close. + pub async fn apply_community_disconnect( + &self, + community_id: CommunityId, + archived_at: Option>, + ) -> Result, buzz_db::DbError> { + match archived_at { + Some(archived_at) => { + self.db + .with_community_archive_fence(community_id, archived_at, || { + self.community_connections + .disconnect_archived_community(community_id) + }) + .await + } + None => match self.disconnect_if_inactive(community_id).await { + Ok(disconnected) => Ok(disconnected), + Err(error) => { + tracing::warn!( + community = %community_id, + %error, + "could not verify bare community disconnect; failing closed as a deletion" + ); + Ok(Some( + self.community_connections + .disconnect_deleted_community(community_id), + )) + } + }, + } + } + + async fn disconnect_if_inactive( + &self, + community_id: CommunityId, + ) -> Result, buzz_db::DbError> { + 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 + } + /// 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_for_maintenance(community_id) - }) - .await; + let (closed, failures) = revalidate_registered_communities( + &self.community_connections, + |community_id| async move { + // Nearly every live community is active. An unlocked read keeps + // the per-tick row lock off them; the fence re-checks under the + // lock before closing anything. + if self.db.is_community_active(community_id).await? { + return Ok(0); + } + self.disconnect_if_inactive(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"); } @@ -2516,54 +2680,63 @@ pub(crate) mod tests { #[test] fn community_disconnect_then_nip_fi_keeps_community_deleted_reason() { - // CommunityDeleted fires first, AuthorizationDenied arrives second. - // The slot must retain CommunityDeleted. - let cancel = CancellationToken::new(); - let control = CommunityConnectionControl::new(cancel.clone()); - let reason_rx = control.disconnect_reason(); + // A community close fires first, AuthorizationDenied arrives second. + // The slot must retain the community reason, archived or deleted. + for reason in [ + CommunityDisconnectReason::CommunityDeleted, + CommunityDisconnectReason::CommunityArchived, + ] { + let cancel = CancellationToken::new(); + let control = CommunityConnectionControl::new(cancel.clone()); + let reason_rx = control.disconnect_reason(); - // First writer: CommunityDeleted (via disconnect_community). - control.disconnect_community(); - // Second writer: AuthorizationDenied — must be ignored (via disconnect_nip_fi). - control.disconnect_nip_fi(); + // First writer: the community reason (via disconnect_community). + control.disconnect_community(reason); + // Second writer: AuthorizationDenied — must be ignored (via disconnect_nip_fi). + control.disconnect_nip_fi(); - assert_eq!( - *reason_rx.borrow(), - Some(CommunityDisconnectReason::CommunityDeleted), - "CommunityDeleted (first writer) must not be clobbered by AuthorizationDenied" - ); + assert_eq!( + *reason_rx.borrow(), + Some(reason), + "{reason:?} (first writer) must not be clobbered by AuthorizationDenied" + ); + } } #[test] fn disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame() { - // disconnect_community fires first → wins reason → no payload (community-deleted - // path is intentionally payload-less). + // disconnect_community fires first → wins reason → no payload (community + // closes are intentionally payload-less). // disconnect_nip_fi fires second → loses reason → must NOT enqueue a denial - // frame against the CommunityDeleted close. - let (terminal_tx, mut terminal_rx) = tokio::sync::mpsc::channel(1); + // frame against the community close. + for reason in [ + CommunityDisconnectReason::CommunityDeleted, + CommunityDisconnectReason::CommunityArchived, + ] { + let (terminal_tx, mut terminal_rx) = tokio::sync::mpsc::channel(1); - let cancel = CancellationToken::new(); - let control = CommunityConnectionControl::new(cancel.clone()); - control.set_terminal_frame_sender(terminal_tx); + let cancel = CancellationToken::new(); + let control = CommunityConnectionControl::new(cancel.clone()); + control.set_terminal_frame_sender(terminal_tx); - // First writer: disconnect_community. - control.disconnect_community(); - // Second writer: disconnect_nip_fi — loses reason slot. - control.disconnect_nip_fi(); + // First writer: disconnect_community. + control.disconnect_community(reason); + // Second writer: disconnect_nip_fi — loses reason slot. + control.disconnect_nip_fi(); - // Reason slot retains CommunityDeleted. - assert_eq!( - *control.disconnect_reason().borrow(), - Some(CommunityDisconnectReason::CommunityDeleted), - "CommunityDeleted must be retained when community wins reason" - ); + assert_eq!( + *control.disconnect_reason().borrow(), + Some(reason), + "{reason:?} must be retained when community wins reason" + ); - // No frame queued — losing deny must not send an authorization_denied payload - // against a community-deleted close. - assert!( - terminal_rx.try_recv().is_err(), - "losing disconnect_nip_fi must not enqueue a denial frame when community wins reason" - ); + // No frame queued — losing deny must not send an authorization_denied + // payload against a community close. + assert!( + terminal_rx.try_recv().is_err(), + "losing disconnect_nip_fi must not enqueue a denial frame against {reason:?}" + ); + } } #[test] @@ -3151,17 +3324,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); } @@ -3217,7 +3390,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()); @@ -3507,17 +3680,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()); @@ -3531,6 +3707,91 @@ 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()); + } + + /// A bare `DisconnectCommunity` may be a permanent deletion, so when the + /// lifecycle fence cannot be evaluated it must fail closed as deletion did + /// before the fence existed. The timestamped form is only ever a reversible + /// archive, so it retains sockets for lifecycle revalidation instead. + #[tokio::test] + async fn bare_community_disconnect_fails_closed_when_the_fence_is_unavailable() { + let state = test_state_with_database_url( + "postgres://unused:unused@127.0.0.1:1/unused", // sadscan:disable np.postgres.1 + ) + .await; + let community = CommunityId::from_uuid(Uuid::new_v4()); + let cancel = CancellationToken::new(); + let control = CommunityConnectionControl::new(cancel.clone()); + let reason = control.disconnect_reason(); + let _guard = state + .community_connections + .register(Uuid::new_v4(), community, control); + + state + .apply_community_disconnect(community, Some(Utc::now())) + .await + .expect_err("an unreachable fence must surface for the archive form"); + assert!( + !cancel.is_cancelled(), + "a timestamped archive disconnect must retain sockets when its fence fails" + ); + assert_eq!(*reason.borrow(), None); + + assert_eq!( + state + .apply_community_disconnect(community, None) + .await + .expect("a bare disconnect fails closed instead of erroring"), + Some(1) + ); + assert!( + cancel.is_cancelled(), + "a bare disconnect must close sockets when its fence fails" + ); + assert_eq!( + *reason.borrow(), + Some(CommunityDisconnectReason::CommunityDeleted) + ); + } + #[test] fn community_lifecycle_guard_deregisters_on_early_return() { let registry = CommunityConnectionRegistry::new(); @@ -3546,7 +3807,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()); } diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index d70eeee409..fededb6842 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -87,10 +87,10 @@ run_unit_tests() { run_test_step "buzz-auth unit tests" \ cargo test -p buzz-auth --lib -- --nocapture - # S4 cross-pod NIP-FI disconnect payload tests (infra-free). Mirrors - # `just test-unit`. - run_test_step "buzz-pubsub conn_control NIP-FI tests" \ - cargo test -p buzz-pubsub --lib conn_control::tests::nip_fi_disconnect_ -- --nocapture + # Cross-pod disconnect payload tests (NIP-FI and community archive fence; + # infra-free). Mirrors `just test-unit`. + run_test_step "buzz-pubsub conn_control tests" \ + cargo test -p buzz-pubsub --lib conn_control::tests:: -- --nocapture run_test_step "buzz-voice tests" \ cargo test -p buzz-voice --lib -- --nocapture @@ -132,6 +132,9 @@ run_unit_tests() { run_test_step "buzz-admin storage snapshot tests" \ cargo test -p buzz-admin storage_snapshot -- --nocapture + run_test_step "buzz-admin community archive command tests" \ + cargo test -p buzz-admin communities -- --nocapture + # Multi-tenant conformance gate: independent replay checker + golden # fixtures (buzz-conformance). Pure in-process trace replay, no infra. run_test_step "buzz-conformance tests" \ @@ -241,6 +244,21 @@ run_unit_tests() { run_test_step "buzz-relay admission regression test" \ cargo test -p buzz-relay --lib state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket -- --nocapture + run_test_step "buzz-relay archive revalidation fence test" \ + cargo test -p buzz-relay --lib state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback -- --exact --nocapture + + run_test_step "buzz-relay bare disconnect fail-closed test" \ + cargo test -p buzz-relay --lib state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable -- --exact --nocapture + + run_test_step "buzz-relay archived close-reason test" \ + cargo test -p buzz-relay --lib connection::tests::send_loop_sends_policy_close_when_community_is_archived -- --exact --nocapture + + run_test_step "buzz-relay community close first-writer tests" \ + cargo test -p buzz-relay --lib state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason -- --exact --nocapture + + run_test_step "buzz-relay community close suppresses losing denial frame test" \ + cargo test -p buzz-relay --lib state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame -- --exact --nocapture + run_test_step "buzz-relay router tests" \ cargo test -p buzz-relay --lib router::tests:: -- --nocapture