From 6c05f7953894c78bff42efdcd8bbe990c09e0a48 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Mon, 31 Aug 2026 10:56:20 -0400 Subject: [PATCH 1/7] feat(admin): add community archive commands Signed-off-by: Luke Tornquist --- crates/buzz-admin/src/communities.rs | 505 ++++++++++++++++++ crates/buzz-admin/src/main.rs | 161 ++++++ ...2026-08-31-buzz-admin-community-archive.md | 100 ++++ 3 files changed, 766 insertions(+) create mode 100644 crates/buzz-admin/src/communities.rs create mode 100644 docs/plans/2026-08-31-buzz-admin-community-archive.md diff --git a/crates/buzz-admin/src/communities.rs b/crates/buzz-admin/src/communities.rs new file mode 100644 index 00000000000..426bd7a8495 --- /dev/null +++ b/crates/buzz-admin/src/communities.rs @@ -0,0 +1,505 @@ +//! 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 (propagation, exit_code) = match publish_disconnect(&tenant).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) -> 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, &ConnControl::DisconnectCommunity) + .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/docs/plans/2026-08-31-buzz-admin-community-archive.md b/docs/plans/2026-08-31-buzz-admin-community-archive.md new file mode 100644 index 00000000000..46db09c8b7f --- /dev/null +++ b/docs/plans/2026-08-31-buzz-admin-community-archive.md @@ -0,0 +1,100 @@ +# Buzz Admin Community Archive Commands Implementation Plan + +> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** Add reversible, single-operator `buzz-admin communities archive` and `buzz-admin communities unarchive` commands with exact-host/current-owner safety guards and live connection propagation. + +**Architecture:** Add a thin `communities` command module to `buzz-admin`. It will reuse the existing owner-aware, idempotent `buzz-db` lifecycle methods, construct the same deployment-host guard used by the relay operator endpoint, and publish the existing community disconnect command to Redis after archival. Operator identity and reason are required command inputs and included in machine-readable output; no second-operator approval record is created. + +**Tech Stack:** Rust, clap, buzz-db, buzz-core tenant normalization, buzz-pubsub/Redis, serde_json, existing Buzz unit and database integration tests. + +--- + +### Task 1: Specify the CLI surface with failing tests + +**Files:** +- Modify: `crates/buzz-admin/src/main.rs` +- Test: `crates/buzz-admin/src/main.rs` + +**Step 1: Write failing parser tests** + +Add tests that parse: + +```text +buzz-admin communities archive --host example.communities.buzz.xyz --owner-pubkey <64-hex> --operator-id codex --reason requested-deletion +buzz-admin communities unarchive --host example.communities.buzz.xyz --owner-pubkey <64-hex> --operator-id codex --reason rollback +``` + +Assert both subcommands capture the exact host, owner pubkey, operator identity, and reason. Assert omitting `--owner-pubkey`, `--operator-id`, or `--reason` fails parsing. + +**Step 2: Run the test and verify RED** + +Run: `. ./bin/activate-hermit && cargo test -p buzz-admin communities_command` + +Expected: compilation/parser failure because the `communities` command surface does not exist. + +### Task 2: Add lifecycle validation and execution + +**Files:** +- Create: `crates/buzz-admin/src/communities.rs` +- Modify: `crates/buzz-admin/src/main.rs` + +**Step 1: Implement the minimum command surface** + +Define a clap `CommunitiesCommand` enum with `Archive` and `Unarchive` variants. Both require `--host`, `--owner-pubkey`, `--operator-id`, and `--reason`. + +**Step 2: Add validation tests before helpers** + +Test normalization of uppercase/default-port/trailing-dot authorities and rejection of schemes, paths, whitespace, malformed authorities, and invalid owner pubkeys. Run those tests and verify they fail before adding helpers. + +**Step 3: Implement validation helpers** + +Normalize only a bare authority using the shared tenant normalization rules and parse the owner as a Nostr public key, returning canonical lowercase hex. + +**Step 4: Implement archive** + +Connect to the existing `DATABASE_URL`. Resolve the protected deployment host from `RELAY_URL`. Call `Db::archive_community_owned_by`; fail closed when no exact host/current-owner row is updated. After commit, connect to `REDIS_URL` and publish `ConnControl::DisconnectCommunity` under the returned community tenant. Emit JSON containing `community_id`, `host`, `archived_at`, `status`, `operator_id`, `reason`, and propagation subscriber count. If Redis publication fails, report that archival committed but propagation is pending and return nonzero so a safe idempotent retry is visible. + +**Step 5: Implement unarchive** + +Call `Db::unarchive_community_owned_by` with the same exact host/current-owner guard. Emit JSON containing `community_id`, `host`, `archived_at: null`, `status: active`, `operator_id`, and `reason`. No disconnect is published. + +**Step 6: Run focused tests and verify GREEN** + +Run: `. ./bin/activate-hermit && cargo test -p buzz-admin` + +Expected: all `buzz-admin` tests pass. + +### Task 3: Verify behavior and publish the PR + +**Files:** +- Modify only files required by Tasks 1–2 plus this plan. + +**Step 1: Run formatting and lint/build gates** + +Run: + +```bash +. ./bin/activate-hermit +cargo fmt --all -- --check +cargo clippy -p buzz-admin --all-targets -- -D warnings +cargo test -p buzz-admin +cargo build -p buzz-admin +``` + +Expected: every command exits zero with no warnings or failed tests. + +**Step 2: Verify help output** + +Run: + +```bash +target/debug/buzz-admin communities archive --help +target/debug/buzz-admin communities unarchive --help +``` + +Expected: required exact-host, current-owner, operator identity, and reason arguments are documented; no approval argument exists. + +**Step 3: Commit and publish** + +Commit the plan, tests, and implementation on `codex/community-archive-admin`, push it from the local machine after reviewing the transferred patch, and open a draft PR describing reversibility, safety guards, propagation semantics, and test evidence. From f1db65c80d5dc3e16773eae4cf68b64ebc16d533 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Mon, 31 Aug 2026 11:35:40 -0400 Subject: [PATCH 2/7] docs: remove implementation plan Signed-off-by: Luke Tornquist --- ...2026-08-31-buzz-admin-community-archive.md | 100 ------------------ 1 file changed, 100 deletions(-) delete mode 100644 docs/plans/2026-08-31-buzz-admin-community-archive.md diff --git a/docs/plans/2026-08-31-buzz-admin-community-archive.md b/docs/plans/2026-08-31-buzz-admin-community-archive.md deleted file mode 100644 index 46db09c8b7f..00000000000 --- a/docs/plans/2026-08-31-buzz-admin-community-archive.md +++ /dev/null @@ -1,100 +0,0 @@ -# Buzz Admin Community Archive Commands Implementation Plan - -> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. - -**Goal:** Add reversible, single-operator `buzz-admin communities archive` and `buzz-admin communities unarchive` commands with exact-host/current-owner safety guards and live connection propagation. - -**Architecture:** Add a thin `communities` command module to `buzz-admin`. It will reuse the existing owner-aware, idempotent `buzz-db` lifecycle methods, construct the same deployment-host guard used by the relay operator endpoint, and publish the existing community disconnect command to Redis after archival. Operator identity and reason are required command inputs and included in machine-readable output; no second-operator approval record is created. - -**Tech Stack:** Rust, clap, buzz-db, buzz-core tenant normalization, buzz-pubsub/Redis, serde_json, existing Buzz unit and database integration tests. - ---- - -### Task 1: Specify the CLI surface with failing tests - -**Files:** -- Modify: `crates/buzz-admin/src/main.rs` -- Test: `crates/buzz-admin/src/main.rs` - -**Step 1: Write failing parser tests** - -Add tests that parse: - -```text -buzz-admin communities archive --host example.communities.buzz.xyz --owner-pubkey <64-hex> --operator-id codex --reason requested-deletion -buzz-admin communities unarchive --host example.communities.buzz.xyz --owner-pubkey <64-hex> --operator-id codex --reason rollback -``` - -Assert both subcommands capture the exact host, owner pubkey, operator identity, and reason. Assert omitting `--owner-pubkey`, `--operator-id`, or `--reason` fails parsing. - -**Step 2: Run the test and verify RED** - -Run: `. ./bin/activate-hermit && cargo test -p buzz-admin communities_command` - -Expected: compilation/parser failure because the `communities` command surface does not exist. - -### Task 2: Add lifecycle validation and execution - -**Files:** -- Create: `crates/buzz-admin/src/communities.rs` -- Modify: `crates/buzz-admin/src/main.rs` - -**Step 1: Implement the minimum command surface** - -Define a clap `CommunitiesCommand` enum with `Archive` and `Unarchive` variants. Both require `--host`, `--owner-pubkey`, `--operator-id`, and `--reason`. - -**Step 2: Add validation tests before helpers** - -Test normalization of uppercase/default-port/trailing-dot authorities and rejection of schemes, paths, whitespace, malformed authorities, and invalid owner pubkeys. Run those tests and verify they fail before adding helpers. - -**Step 3: Implement validation helpers** - -Normalize only a bare authority using the shared tenant normalization rules and parse the owner as a Nostr public key, returning canonical lowercase hex. - -**Step 4: Implement archive** - -Connect to the existing `DATABASE_URL`. Resolve the protected deployment host from `RELAY_URL`. Call `Db::archive_community_owned_by`; fail closed when no exact host/current-owner row is updated. After commit, connect to `REDIS_URL` and publish `ConnControl::DisconnectCommunity` under the returned community tenant. Emit JSON containing `community_id`, `host`, `archived_at`, `status`, `operator_id`, `reason`, and propagation subscriber count. If Redis publication fails, report that archival committed but propagation is pending and return nonzero so a safe idempotent retry is visible. - -**Step 5: Implement unarchive** - -Call `Db::unarchive_community_owned_by` with the same exact host/current-owner guard. Emit JSON containing `community_id`, `host`, `archived_at: null`, `status: active`, `operator_id`, and `reason`. No disconnect is published. - -**Step 6: Run focused tests and verify GREEN** - -Run: `. ./bin/activate-hermit && cargo test -p buzz-admin` - -Expected: all `buzz-admin` tests pass. - -### Task 3: Verify behavior and publish the PR - -**Files:** -- Modify only files required by Tasks 1–2 plus this plan. - -**Step 1: Run formatting and lint/build gates** - -Run: - -```bash -. ./bin/activate-hermit -cargo fmt --all -- --check -cargo clippy -p buzz-admin --all-targets -- -D warnings -cargo test -p buzz-admin -cargo build -p buzz-admin -``` - -Expected: every command exits zero with no warnings or failed tests. - -**Step 2: Verify help output** - -Run: - -```bash -target/debug/buzz-admin communities archive --help -target/debug/buzz-admin communities unarchive --help -``` - -Expected: required exact-host, current-owner, operator identity, and reason arguments are documented; no approval argument exists. - -**Step 3: Commit and publish** - -Commit the plan, tests, and implementation on `codex/community-archive-admin`, push it from the local machine after reviewing the transferred patch, and open a draft PR describing reversibility, safety guards, propagation semantics, and test evidence. From 915f73d90be6b4010ca5419762b6c8c43f69beb9 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Mon, 31 Aug 2026 12:44:55 -0400 Subject: [PATCH 3/7] fix(relay): fence archive disconnects Signed-off-by: Luke Tornquist --- crates/buzz-admin/src/communities.rs | 9 ++- crates/buzz-db/src/store/community.rs | 107 ++++++++++++++++++++++++- crates/buzz-pubsub/src/conn_control.rs | 33 +++++++- crates/buzz-relay/src/api/operator.rs | 5 +- crates/buzz-relay/src/main.rs | 39 ++++++++- crates/buzz-relay/src/state.rs | 20 ++++- 6 files changed, 193 insertions(+), 20 deletions(-) diff --git a/crates/buzz-admin/src/communities.rs b/crates/buzz-admin/src/communities.rs index 426bd7a8495..175347a64e7 100644 --- a/crates/buzz-admin/src/communities.rs +++ b/crates/buzz-admin/src/communities.rs @@ -95,7 +95,10 @@ async fn archive( })?; let tenant = TenantContext::resolved(record.id, &record.host); - let (propagation, exit_code) = match publish_disconnect(&tenant).await { + 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!( @@ -141,7 +144,7 @@ async fn unarchive( Ok(0) } -async fn publish_disconnect(tenant: &TenantContext) -> Result { +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)) @@ -150,7 +153,7 @@ async fn publish_disconnect(tenant: &TenantContext) -> Result { .await .context("PubSub init failed")?; pubsub - .publish_conn_control(tenant, &ConnControl::DisconnectCommunity) + .publish_conn_control(tenant, command) .await .context("publishing DisconnectCommunity failed") } diff --git a/crates/buzz-db/src/store/community.rs b/crates/buzz-db/src/store/community.rs index 5e8462345bb..11b72e2467b 100644 --- a/crates/buzz-db/src/store/community.rs +++ b/crates/buzz-db/src/store/community.rs @@ -129,6 +129,44 @@ 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)) + } + /// Returns a community by host regardless of lifecycle state. Operator-plane only. #[datastore_span( name = "lookup_community_by_host_for_management", @@ -606,6 +644,7 @@ mod tests { let operations = [ "lookup_community_by_host", "is_community_active", + "with_community_archive_fence", "lookup_community_by_host_for_management", "list_communities_owned_by", "lookup_community_host", @@ -619,14 +658,16 @@ mod 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", ); @@ -663,6 +704,7 @@ mod 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", @@ -841,6 +883,63 @@ mod 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] #[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..24a4cdf1320 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,33 @@ 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 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 b59fd840c6d..caaf323386f 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/main.rs b/crates/buzz-relay/src/main.rs index d81602e2019..8984e8c30ca 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -990,10 +990,41 @@ async fn main() -> anyhow::Result<()> { loop { match rx.recv().await { Ok(scoped) => match scoped.command { - buzz_pubsub::conn_control::ConnControl::DisconnectCommunity => { - state_for_conn_ctrl - .community_connections - .disconnect_community(scoped.community_id); + buzz_pubsub::conn_control::ConnControl::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_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_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 efdb2846148..7923419baa9 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; @@ -1204,14 +1205,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_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) } From c9190f52447b65673862a7cc63efcdde5a2a32c5 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Mon, 31 Aug 2026 13:23:17 -0400 Subject: [PATCH 4/7] fix(relay): fence lifecycle revalidation Signed-off-by: Luke Tornquist --- crates/buzz-db/src/store/community.rs | 114 ++++++++++++++++++++++++++ crates/buzz-relay/src/state.rs | 84 +++++++++++++++---- 2 files changed, 181 insertions(+), 17 deletions(-) diff --git a/crates/buzz-db/src/store/community.rs b/crates/buzz-db/src/store/community.rs index 11b72e2467b..0ee5d63169d 100644 --- a/crates/buzz-db/src/store/community.rs +++ b/crates/buzz-db/src/store/community.rs @@ -167,6 +167,41 @@ impl Db { Ok(Some(result)) } + /// Runs `apply` while holding the community row lock only when the + /// community is not active. + /// + /// 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() -> T, + ) -> Result> { + let mut tx = self.pool.begin().await?; + let active = sqlx::query_scalar::<_, bool>( + r#"SELECT archived_at IS NULL + AND deleted_at IS NULL + AND deletion_state = 'active' + FROM communities + WHERE id = $1 + FOR UPDATE"#, + ) + .bind(community_id.as_uuid()) + .fetch_optional(&mut *tx) + .await?; + + if active == Some(true) { + tx.rollback().await?; + return Ok(None); + } + + let result = apply(); + 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", @@ -645,6 +680,7 @@ mod tests { "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", @@ -940,6 +976,84 @@ mod tests { ); } + #[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"); + }; + 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 fence = tokio::spawn(async move { + fenced_db + .with_inactive_community_fence(community_id, move || { + entered_tx.send(()).expect("report entered fence"); + release_rx.recv().expect("release fenced disconnect"); + "disconnected" + }) + .await + }); + 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!( + fence + .await + .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-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 7923419baa9..1ce4889078b 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -213,21 +213,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)), } } @@ -1234,11 +1233,19 @@ 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(community_id) - }) - .await; + let (closed, failures) = revalidate_registered_communities( + &self.community_connections, + |community_id| async move { + self.db + .with_inactive_community_fence(community_id, || { + self.community_connections + .disconnect_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"); } @@ -1996,17 +2003,20 @@ 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_community(community)) } - }) - .await; + } + }) + .await; assert_eq!(closed, 2); assert!(cancel_a.is_cancelled()); @@ -2020,6 +2030,46 @@ 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_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(); From 6dfa901ead8f3a47a55c4e2db83dc54af5a59daf Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Mon, 31 Aug 2026 16:27:40 -0400 Subject: [PATCH 5/7] fix(relay): distinguish archived disconnects Signed-off-by: Luke Tornquist --- crates/buzz-db/src/store/community.rs | 32 +++++++------ crates/buzz-pubsub/src/conn_control.rs | 22 +++++++++ crates/buzz-relay/src/audio/handler.rs | 2 +- crates/buzz-relay/src/connection.rs | 37 +++++++++++++++ crates/buzz-relay/src/main.rs | 4 +- crates/buzz-relay/src/state.rs | 62 +++++++++++++++++--------- 6 files changed, 122 insertions(+), 37 deletions(-) diff --git a/crates/buzz-db/src/store/community.rs b/crates/buzz-db/src/store/community.rs index 0ee5d63169d..0cd8e6497a8 100644 --- a/crates/buzz-db/src/store/community.rs +++ b/crates/buzz-db/src/store/community.rs @@ -168,7 +168,8 @@ impl Db { } /// Runs `apply` while holding the community row lock only when the - /// community is not active. + /// 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 @@ -177,13 +178,12 @@ impl Db { pub async fn with_inactive_community_fence( &self, community_id: CommunityId, - apply: impl FnOnce() -> T, + apply: impl FnOnce(Option>) -> T, ) -> Result> { let mut tx = self.pool.begin().await?; - let active = sqlx::query_scalar::<_, bool>( - r#"SELECT archived_at IS NULL - AND deleted_at IS NULL - AND deletion_state = 'active' + 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"#, @@ -192,12 +192,16 @@ impl Db { .fetch_optional(&mut *tx) .await?; - if active == Some(true) { - tx.rollback().await?; - return Ok(None); - } + 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(); + let result = apply(archived_at); tx.commit().await?; Ok(Some(result)) } @@ -992,7 +996,8 @@ mod tests { let CreateCommunityWithOwnerResult::Created(created) = created else { panic!("expected new community"); }; - db.archive_community_owned_by(&host, &owner, "protected.example") + let archived = db + .archive_community_owned_by(&host, &owner, "protected.example") .await .expect("archive community") .expect("owned community"); @@ -1003,7 +1008,8 @@ mod tests { let (release_tx, release_rx) = std::sync::mpsc::sync_channel(0); let fence = tokio::spawn(async move { fenced_db - .with_inactive_community_fence(community_id, move || { + .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" diff --git a/crates/buzz-pubsub/src/conn_control.rs b/crates/buzz-pubsub/src/conn_control.rs index 24a4cdf1320..81722124dc9 100644 --- a/crates/buzz-pubsub/src/conn_control.rs +++ b/crates/buzz-pubsub/src/conn_control.rs @@ -230,6 +230,28 @@ mod tests { 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()); diff --git a/crates/buzz-relay/src/audio/handler.rs b/crates/buzz-relay/src/audio/handler.rs index 06d3a32b43d..ca3aba6f21c 100644 --- a/crates/buzz-relay/src/audio/handler.rs +++ b/crates/buzz-relay/src/audio/handler.rs @@ -1663,7 +1663,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 5fcfe70b91c..a57e0594338 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -824,6 +824,13 @@ 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() @@ -1038,6 +1045,36 @@ 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 8984e8c30ca..6d3a5711d7c 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1002,7 +1002,7 @@ async fn main() -> anyhow::Result<()> { || { state_for_conn_ctrl .community_connections - .disconnect_community(scoped.community_id) + .disconnect_archived_community(scoped.community_id) }, ) .await @@ -1023,7 +1023,7 @@ async fn main() -> anyhow::Result<()> { } else { state_for_conn_ctrl .community_connections - .disconnect_community(scoped.community_id); + .disconnect_deleted_community(scoped.community_id); } } buzz_pubsub::conn_control::ConnControl::DisconnectPubkey { diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 1ce4889078b..070a7ab95f4 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -39,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"), @@ -79,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(); } } @@ -149,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; } } @@ -1210,7 +1225,7 @@ impl AppState { .db .with_community_archive_fence(tenant.community(), archived_at, || { self.community_connections - .disconnect_community(tenant.community()) + .disconnect_archived_community(tenant.community()) }) .await? .unwrap_or(0); @@ -1237,9 +1252,14 @@ impl AppState { &self.community_connections, |community_id| async move { self.db - .with_inactive_community_fence(community_id, || { - self.community_connections - .disconnect_community(community_id) + .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)) @@ -1907,17 +1927,17 @@ 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); } @@ -1971,7 +1991,7 @@ 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()); @@ -2012,7 +2032,7 @@ mod tests { "injected lookup failure".into(), )) } else { - Ok(registry.disconnect_community(community)) + Ok(registry.disconnect_archived_community(community)) } } }) @@ -2050,7 +2070,7 @@ mod tests { async move { entered.notify_one(); resume.notified().await; - Ok(registry.disconnect_community(community_id)) + Ok(registry.disconnect_archived_community(community_id)) } }); tokio::pin!(future); @@ -2085,7 +2105,7 @@ 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()); } From 8af98306986c1612860f8aa7ca11ae1479ec2428 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Mon, 31 Aug 2026 16:28:05 -0400 Subject: [PATCH 6/7] ci: run community archive fence tests Signed-off-by: Luke Tornquist --- .github/workflows/ci.yml | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 44966c28de6..15edf9d81e6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -710,6 +710,18 @@ jobs: env: DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz TEST_DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz + - name: Community archive fence PostgreSQL tests + # Exact archive-transition matching and row-lock serialization require + # real Postgres and are ignored by the infrastructure-free unit job. + run: | + filter='(package(buzz-db) and test(=store::community::tests::archive_disconnect_fence_tracks_the_exact_archive_transition)) or (package(buzz-db) and test(=store::community::tests::inactive_community_fence_holds_the_row_lock_through_disconnect))' + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E "${filter}" \ + --run-ignored ignored-only + env: + DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz + TEST_DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz - name: Writer session timeout guardrails run: | cargo nextest run \ From 4b604647ed59c7f3bc09b3ef30130c9bef14b72f Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Mon, 31 Aug 2026 17:15:15 -0400 Subject: [PATCH 7/7] test(db): avoid blocking archive fence runtime Signed-off-by: Luke Tornquist --- crates/buzz-db/src/store/community.rs | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/crates/buzz-db/src/store/community.rs b/crates/buzz-db/src/store/community.rs index 0cd8e6497a8..2e1ede7e66b 100644 --- a/crates/buzz-db/src/store/community.rs +++ b/crates/buzz-db/src/store/community.rs @@ -1006,15 +1006,17 @@ mod tests { 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 fence = tokio::spawn(async move { - fenced_db - .with_inactive_community_fence(community_id, move |archived_at| { + 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" - }) - .await + }, + )) }); entered_rx.await.expect("fence acquired row lock"); @@ -1043,8 +1045,9 @@ mod tests { release_tx.send(()).expect("release fence"); assert_eq!( - fence + 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")