feat(relay): add durable community archival (#1834)

Signed-off-by: npub16ynjrn8tm9gxg4kpgfgj24eup8met48z58m9g0cvmpq9msjht0lq7r96r6 <d12721ccebd9506456c1425125573c09f795d4e2a1f6543f0cd8405dc2575bfe@sprout-oss.stage.blox.sqprod.co>
Signed-off-by: npub1ks3yhswfl2z70myka7wltdg39qpvqhspt2w3l2ajtaee8tmz5qkqrvl4et <b4224bc1c9fa85e7ec96ef9df5b5112802c05e015a9d1fabb25f7393af62a02c@sprout-oss.stage.blox.sqprod.co>
Signed-off-by: npub1dccv64krpcpse5cmkzfeh998cftungyatw3djt8jwdw6g43f7fyqzzmrf7 <6e30cd56c30e030cd31bb0939b94a7c257c9a09d5ba2d92cf2735da45629f248@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: npub16ynjrn8tm9gxg4kpgfgj24eup8met48z58m9g0cvmpq9msjht0lq7r96r6 <d12721ccebd9506456c1425125573c09f795d4e2a1f6543f0cd8405dc2575bfe@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: npub1ks3yhswfl2z70myka7wltdg39qpvqhspt2w3l2ajtaee8tmz5qkqrvl4et <b4224bc1c9fa85e7ec96ef9df5b5112802c05e015a9d1fabb25f7393af62a02c@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: npub1dccv64krpcpse5cmkzfeh998cftungyatw3djt8jwdw6g43f7fyqzzmrf7 <6e30cd56c30e030cd31bb0939b94a7c257c9a09d5ba2d92cf2735da45629f248@sprout-oss.stage.blox.sqprod.co>

crates/buzz-db/src/channel.rs

@@ -1345,6 +1345,7 @@ pub async fn reap_expired_ephemeral_channels(pool: &PgPool) -> Result<Vec<Reaped
            AND ch.ttl_deadline < NOW() \
            AND ch.archived_at IS NULL \
            AND ch.deleted_at IS NULL \
+           AND c.archived_at IS NULL \
          RETURNING ch.community_id, c.host, ch.id",
     )
     .fetch_all(pool)

crates/buzz-db/src/event.rs

@@ -1306,6 +1306,7 @@ pub async fn query_due_reminders(
           AND e.not_before <= $2
           AND e.deleted_at IS NULL
           AND e.delivered_at IS NULL
+          AND c.archived_at IS NULL
         ORDER BY e.community_id, e.pubkey, e.d_tag, e.created_at DESC, e.id ASC
         LIMIT $3
         "#,

crates/buzz-db/src/lib.rs

@@ -261,6 +261,19 @@ pub struct OwnedCommunityRecord {
     pub host: String,
     /// When the community row was created.
     pub created_at: DateTime<Utc>,
+    /// When the community was archived; absent while active.
+    pub archived_at: Option<DateTime<Utc>>,
+}
+
+/// Community row returned by an owner-authorized archive operation.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ArchivedCommunityRecord {
+    /// Stable server-resolved community id.
+    pub id: CommunityId,
+    /// Reserved canonical host.
+    pub host: String,
+    /// Durable first-archive timestamp.
+    pub archived_at: DateTime<Utc>,
 }
 
 /// Token summary returned by [`Db::list_active_tokens`].
@@ -407,6 +420,7 @@ impl Db {
             SELECT id, host
             FROM communities
             WHERE lower(host) = lower($1)
+              AND archived_at IS NULL
             "#,
         )
         .bind(normalized_host)
@@ -425,6 +439,35 @@ impl Db {
         .transpose()
     }
 
+    /// Returns whether a community id still exists in the active lifecycle state.
+    pub async fn is_community_active(&self, community_id: CommunityId) -> Result<bool> {
+        let active = sqlx::query_scalar::<_, bool>(
+            "SELECT EXISTS(SELECT 1 FROM communities WHERE id = $1 AND archived_at IS NULL)",
+        )
+        .bind(community_id.as_uuid())
+        .fetch_one(&self.pool)
+        .await?;
+        Ok(active)
+    }
+
+    /// Returns a community by host regardless of lifecycle state. Operator-plane only.
+    pub async fn lookup_community_by_host_for_management(
+        &self,
+        normalized_host: &str,
+    ) -> Result<Option<CommunityRecord>> {
+        let row = sqlx::query("SELECT id, host FROM communities WHERE lower(host) = lower($1)")
+            .bind(normalized_host)
+            .fetch_optional(&self.pool)
+            .await?;
+        row.map(|row| {
+            Ok(CommunityRecord {
+                id: CommunityId::from_uuid(row.try_get("id")?),
+                host: row.try_get("host")?,
+            })
+        })
+        .transpose()
+    }
+
     /// Lists communities where `owner_pubkey` currently holds the `owner` role.
     ///
     /// This is an operator-plane helper, not a tenant-scoped data-plane read:
@@ -436,7 +479,7 @@ impl Db {
         let owner_pubkey = owner_pubkey.to_ascii_lowercase();
         let rows = sqlx::query(
             r#"
-            SELECT c.id, c.host, c.created_at
+            SELECT c.id, c.host, c.created_at, c.archived_at
             FROM communities c
             JOIN relay_members rm ON rm.community_id = c.id
             WHERE rm.pubkey = $1
@@ -453,10 +496,12 @@ impl Db {
                 let id: Uuid = row.try_get("id")?;
                 let host: String = row.try_get("host")?;
                 let created_at: DateTime<Utc> = row.try_get("created_at")?;
+                let archived_at: Option<DateTime<Utc>> = row.try_get("archived_at")?;
                 Ok(OwnedCommunityRecord {
                     id: CommunityId::from_uuid(id),
                     host,
                     created_at,
+                    archived_at,
                 })
             })
             .collect()
@@ -478,6 +523,7 @@ impl Db {
             SELECT host
             FROM communities
             WHERE id = $1
+              AND archived_at IS NULL
             "#,
         )
         .bind(community_id.as_uuid())
@@ -632,6 +678,7 @@ impl Db {
                 WHERE lower(c.host) = lower($1)
                   AND lower(rm.pubkey) = lower($2)
                   AND rm.role = 'owner'
+                  AND c.archived_at IS NULL
                 "#,
             )
             .bind(normalized_host)
@@ -654,6 +701,39 @@ impl Db {
         ))
     }
 
+    /// Idempotently archives a community when the asserted pubkey is its current owner.
+    pub async fn archive_community_owned_by(
+        &self,
+        normalized_host: &str,
+        owner_pubkey: &str,
+        protected_deployment_host: &str,
+    ) -> Result<Option<ArchivedCommunityRecord>> {
+        let row = sqlx::query(
+            r#"UPDATE communities c
+               SET archived_at = COALESCE(c.archived_at, now())
+               FROM relay_members rm
+               WHERE lower(c.host) = lower($1)
+                 AND rm.community_id = c.id
+                 AND lower(rm.pubkey) = lower($2)
+                 AND rm.role = 'owner'
+                 AND lower(c.host) <> lower($3)
+               RETURNING c.id, c.host, c.archived_at"#,
+        )
+        .bind(normalized_host)
+        .bind(owner_pubkey)
+        .bind(protected_deployment_host)
+        .fetch_optional(&self.pool)
+        .await?;
+        row.map(|row| {
+            Ok(ArchivedCommunityRecord {
+                id: CommunityId::from_uuid(row.try_get("id")?),
+                host: row.try_get("host")?,
+                archived_at: row.try_get("archived_at")?,
+            })
+        })
+        .transpose()
+    }
+
     /// Returns the community that owns a channel, if the channel exists.
     ///
     /// Internal relay producers use this to derive tenant context from the row
@@ -4128,8 +4208,13 @@ mod tests {
         let community_a = CommunityId::from_uuid(make_community(&db.pool).await);
         let community_b = CommunityId::from_uuid(make_community(&db.pool).await);
         let community_c = CommunityId::from_uuid(make_community(&db.pool).await);
-        let owner = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
-        let other = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
+        // Unique per run: `list_communities_owned_by` is keyed only by pubkey,
+        // so a shared fixed pubkey picks up communities leaked by sibling
+        // ignored tests running against the same database.
+        let owner = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple());
+        let owner = owner.as_str();
+        let other = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple());
+        let other = other.as_str();
 
         db.bootstrap_owner(community_a, owner)
             .await

crates/buzz-db/src/migration.rs

@@ -548,7 +548,7 @@ mod tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 15);
+        assert_eq!(migrations.len(), 16);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -1005,7 +1005,7 @@ mod tests {
         run_migrations(&pool)
             .await
             .expect("retry succeeds after operator repair");
-        assert_eq!(applied_versions(&pool).await.last().copied(), Some(15));
+        assert_eq!(applied_versions(&pool).await.last().copied(), Some(16));
     }
 
     #[tokio::test]

crates/buzz-db/src/workflow.rs

@@ -457,13 +457,15 @@ pub async fn list_enabled_channel_workflows(
 pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result<Vec<WorkflowRecord>> {
     let rows = sqlx::query(
         r#"
-        SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash,
-               status::text AS status, enabled, created_at, updated_at
-        FROM workflows
-        WHERE status = 'active'
-          AND enabled = TRUE
-          AND definition->'trigger'->>'on' = 'schedule'
-        ORDER BY created_at ASC
+        SELECT w.id, w.community_id, w.name, w.owner_pubkey, w.channel_id, w.definition, w.definition_hash,
+               w.status::text AS status, w.enabled, w.created_at, w.updated_at
+        FROM workflows w
+        JOIN communities c ON c.id = w.community_id
+        WHERE w.status = 'active'
+          AND w.enabled = TRUE
+          AND w.definition->'trigger'->>'on' = 'schedule'
+          AND c.archived_at IS NULL
+        ORDER BY w.created_at ASC
         LIMIT $1
         "#,
     )

crates/buzz-pubsub/src/conn_control.rs

@@ -54,6 +54,8 @@ pub fn parse_conn_control_channel(channel: &str) -> Option<CommunityId> {
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(tag = "op")]
 pub enum ConnControl {
+    /// Disconnect every live socket bound to the carrying community.
+    DisconnectCommunity,
     /// Disconnect every live connection authenticated as `pubkey` in the
     /// carrying community — live ban enforcement. `pubkey` is 32 raw bytes.
     /// `event_id` and `reason` reproduce the same NIP-01 `OK` frame the origin
@@ -197,6 +199,23 @@ mod tests {
         assert_eq!(parse_conn_control_channel(&extended), None);
     }
 
+    #[test]
+    fn disconnect_community_command_serde_round_trips() {
+        let cmd = ConnControl::DisconnectCommunity;
+        let json = serde_json::to_string(&cmd).unwrap();
+        assert_eq!(serde_json::from_str::<ConnControl>(&json).unwrap(), cmd);
+    }
+
+    #[test]
+    fn unknown_command_is_rejected_without_affecting_later_messages() {
+        assert!(serde_json::from_str::<ConnControl>(r#"{"op":"FutureCommand"}"#).is_err());
+        let known = serde_json::to_string(&ConnControl::DisconnectCommunity).unwrap();
+        assert_eq!(
+            serde_json::from_str::<ConnControl>(&known).unwrap(),
+            ConnControl::DisconnectCommunity
+        );
+    }
+
     #[test]
     fn disconnect_command_serde_round_trips() {
         let cmd = ConnControl::DisconnectPubkey {

crates/buzz-relay/src/api/operator.rs

@@ -192,6 +192,75 @@ pub async fn provision_community(
     }
 }
 
+/// Owner assertion supplied by the trusted operator client.
+#[derive(Debug, Deserialize)]
+pub struct ArchiveCommunityRequest {
+    host: String,
+    owner_pubkey: String,
+}
+
+/// Idempotently archive a community owned by the asserted end-user identity.
+pub async fn archive_community(
+    State(state): State<Arc<AppState>>,
+    headers: HeaderMap,
+    body: axum::body::Bytes,
+) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
+    const PATH: &str = "/operator/communities/archive";
+    authorize_operator_request(&state, &headers, "POST", PATH, None, Some(&body)).await?;
+    let request: ArchiveCommunityRequest = serde_json::from_slice(&body).map_err(|e| {
+        api_error(
+            StatusCode::BAD_REQUEST,
+            &format!("invalid archive-community JSON: {e}"),
+        )
+    })?;
+    let normalized_host = normalize_candidate_host(&request.host)
+        .map_err(|msg| api_error(StatusCode::BAD_REQUEST, &msg))?;
+    let deployment_host = buzz_core::tenant::relay_url_authority(&state.config.relay_url);
+    if normalized_host == deployment_host {
+        return Err(api_error(
+            StatusCode::CONFLICT,
+            "the deployment community cannot be archived",
+        ));
+    }
+    let owner = validate_pubkey_hex(&request.owner_pubkey).ok_or_else(|| {
+        api_error(
+            StatusCode::BAD_REQUEST,
+            "invalid owner_pubkey: expected 64-char hex pubkey",
+        )
+    })?;
+    let record = state
+        .db
+        .archive_community_owned_by(&normalized_host, &owner, &deployment_host)
+        .await
+        .map_err(|e| internal_error(&format!("archive community: {e}")))?
+        .ok_or_else(|| api_error(StatusCode::NOT_FOUND, "community not found"))?;
+    let tenant = TenantContext::resolved(record.id, &record.host);
+    let closed = match state.disconnect_community_clusterwide(&tenant).await {
+        Ok(closed) => closed,
+        Err(error) => {
+            tracing::warn!(community = %record.id, host = %record.host, %error, "community archived but disconnect propagation is pending");
+            return Err((
+                StatusCode::SERVICE_UNAVAILABLE,
+                Json(serde_json::json!({
+                    "community_id": record.id.to_string(),
+                    "host": record.host,
+                    "archived_at": record.archived_at,
+                    "status": "archived",
+                    "propagation": "pending",
+                    "error": "connection propagation pending — retry this request",
+                })),
+            ));
+        }
+    };
+    tracing::info!(community = %record.id, host = %record.host, local_connections_closed = closed, "community archived");
+    Ok(Json(serde_json::json!({
+        "community_id": record.id.to_string(),
+        "host": record.host,
+        "archived_at": record.archived_at,
+        "status": "archived",
+    })))
+}
+
 /// List communities where a pubkey currently holds the `owner` role.
 pub async fn list_owned_communities(
     State(state): State<Arc<AppState>>,
@@ -228,6 +297,7 @@ pub async fn list_owned_communities(
             "community_id": row.id.to_string(),
             "host": row.host,
             "created_at": row.created_at,
+            "archived_at": row.archived_at,
         })).collect::<Vec<_>>(),
     })))
 }
@@ -378,7 +448,7 @@ pub async fn community_availability(
         .map_err(|msg| api_error(StatusCode::BAD_REQUEST, &msg))?;
     let existing = state
         .db
-        .lookup_community_by_host(&normalized_host)
+        .lookup_community_by_host_for_management(&normalized_host)
         .await
         .map_err(|e| internal_error(&format!("check community availability: {e}")))?;
 
@@ -728,6 +798,118 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn archive_publish_failure_is_retryable_and_preserves_timestamp() {
+        let operator = Keys::generate();
+        let owner = Keys::generate();
+        let Some(state) = operator_test_state(std::slice::from_ref(&operator)).await else {
+            return;
+        };
+        let host = format!("community-{}.example", Uuid::new_v4().simple());
+        let owner_hex = owner.public_key().to_hex();
+        let create_body = serde_json::json!({
+            "host": host,
+            "initial_owner_pubkey": owner_hex,
+            "create_only": true,
+        })
+        .to_string();
+        let create_url = format!("http://{INGRESS_HOST}/operator/communities");
+        let create_auth =
+            nip98_auth_header(&operator, &create_url, "POST", Some(create_body.as_bytes()));
+        let create_response = build_router(Arc::clone(&state))
+            .oneshot(
+                Request::builder()
+                    .method("POST")
+                    .uri("/operator/communities")
+                    .header(header::HOST, INGRESS_HOST)
+                    .header(header::AUTHORIZATION, create_auth)
+                    .header(header::CONTENT_TYPE, "application/json")
+                    .body(Body::from(create_body))
+                    .expect("create request"),
+            )
+            .await
+            .expect("create response");
+        assert_eq!(create_response.status(), StatusCode::OK);
+
+        let archive_body = serde_json::json!({
+            "host": host,
+            "owner_pubkey": owner.public_key().to_hex(),
+        })
+        .to_string();
+        let archive_url = format!("http://{INGRESS_HOST}/operator/communities/archive");
+        let archive_once = |state: Arc<AppState>| {
+            let auth = nip98_auth_header(
+                &operator,
+                &archive_url,
+                "POST",
+                Some(archive_body.as_bytes()),
+            );
+            let body = archive_body.clone();
+            async move {
+                build_router(state)
+                    .oneshot(
+                        Request::builder()
+                            .method("POST")
+                            .uri("/operator/communities/archive")
+                            .header(header::HOST, INGRESS_HOST)
+                            .header(header::AUTHORIZATION, auth)
+                            .header(header::CONTENT_TYPE, "application/json")
+                            .body(Body::from(body))
+                            .expect("archive request"),
+                    )
+                    .await
+                    .expect("archive response")
+            }
+        };
+
+        let first = archive_once(Arc::clone(&state)).await;
+        assert_eq!(first.status(), StatusCode::SERVICE_UNAVAILABLE);
+        let first_json = read_json(first).await;
+        assert_eq!(first_json["status"], "archived");
+        assert_eq!(first_json["propagation"], "pending");
+        let first_archived_at = first_json["archived_at"].clone();
+        assert!(!first_archived_at.is_null());
+        assert!(state
+            .db
+            .lookup_community_by_host(&host)
+            .await
+            .expect("active lookup")
+            .is_none());
+
+        assert_eq!(
+            state
+                .community_disconnect_publish_attempts
+                .load(std::sync::atomic::Ordering::Relaxed),
+            1
+        );
+        let second = archive_once(Arc::clone(&state)).await;
+        assert_eq!(second.status(), StatusCode::SERVICE_UNAVAILABLE);
+        let second_json = read_json(second).await;
+        assert_eq!(second_json["archived_at"], first_archived_at);
+        assert_eq!(
+            state
+                .community_disconnect_publish_attempts
+                .load(std::sync::atomic::Ordering::Relaxed),
+            2,
+            "idempotent archive retry must republish the disconnect"
+        );
+
+        let owned = state
+            .db
+            .list_communities_owned_by(&owner.public_key().to_hex())
+            .await
+            .expect("owned communities");
+        let row = owned
+            .iter()
+            .find(|row| row.host == host)
+            .expect("archived row");
+        assert_eq!(
+            serde_json::to_value(row.archived_at).expect("timestamp JSON"),
+            first_archived_at
+        );
+    }
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn happy_path_create_returns_created_and_bootstraps_owner() {

crates/buzz-relay/src/audio/handler.rs

@@ -38,7 +38,7 @@ use buzz_core::StoredEvent;
 use buzz_pubsub::EventTopic;
 
 use crate::audio::room::PeerCtrl;
-use crate::state::AppState;
+use crate::state::{run_registered_community_connection, AppState};
 
 /// Maximum binary frame size: 4 KB is generous for a single Opus packet.
 const MAX_AUDIO_FRAME_BYTES: usize = 4096;
@@ -112,6 +112,29 @@ async fn handle_audio_connection(
     state: Arc<AppState>,
     tenant: TenantContext,
     channel_id: Uuid,
+) {
+    let cancel = CancellationToken::new();
+    let community_id = tenant.community();
+    let registry = Arc::clone(&state.community_connections);
+    let check_state = Arc::clone(&state);
+    let run_state = Arc::clone(&state);
+    run_registered_community_connection(
+        &registry,
+        Uuid::new_v4(),
+        community_id,
+        cancel.clone(),
+        move || async move { check_state.db.is_community_active(community_id).await },
+        move || handle_active_audio_connection(socket, run_state, tenant, channel_id, cancel),
+    )
+    .await;
+}
+
+async fn handle_active_audio_connection(
+    socket: WebSocket,
+    state: Arc<AppState>,
+    tenant: TenantContext,
+    channel_id: Uuid,
+    cancel: CancellationToken,
 ) {
     let (mut ws_send, mut ws_recv) = socket.split();
 
@@ -126,23 +149,26 @@ async fn handle_audio_connection(
         return;
     }
 
-    let auth_result = tokio::time::timeout(AUTH_TIMEOUT, async {
-        while let Some(Ok(msg)) = ws_recv.next().await {
-            if let WsMessage::Text(text) = msg {
-                if text.len() > MAX_TEXT_FRAME_BYTES {
-                    warn!(channel_id = %channel_id, "auth text frame too large — dropping");
-                    continue;
-                }
-                if let Ok(auth) = serde_json::from_str::<AuthMsg>(&text) {
-                    if auth.msg_type == "auth" {
-                        return Some(auth);
+    let auth_result = tokio::select! {
+        biased;
+        _ = cancel.cancelled() => return,
+        result = tokio::time::timeout(AUTH_TIMEOUT, async {
+            while let Some(Ok(msg)) = ws_recv.next().await {
+                if let WsMessage::Text(text) = msg {
+                    if text.len() > MAX_TEXT_FRAME_BYTES {
+                        warn!(channel_id = %channel_id, "auth text frame too large — dropping");
+                        continue;
+                    }
+                    if let Ok(auth) = serde_json::from_str::<AuthMsg>(&text) {
+                        if auth.msg_type == "auth" {
+                            return Some(auth);
+                        }
                     }
                 }
             }
-        }
-        None
-    })
-    .await;
+            None
+        }) => result,
+    };
 
     let auth_msg = match auth_result {
         Ok(Some(a)) => a,
@@ -395,7 +421,6 @@ async fn handle_audio_connection(
     )
     .await;
 
-    let cancel = CancellationToken::new();
     let missed_pongs = Arc::new(AtomicU8::new(0));
 
     // Dual-channel pattern (matches connection.rs): data channel for audio,

crates/buzz-relay/src/connection.rs

@@ -20,7 +20,7 @@ use nostr::Filter;
 
 use crate::handlers;
 use crate::protocol::{ClientMessage, RelayMessage};
-use crate::state::AppState;
+use crate::state::{run_registered_community_connection, AppState};
 use buzz_pubsub::EventTopic;
 
 /// Maximum time a new socket may hold a connection slot without completing NIP-42 auth.
@@ -120,6 +120,31 @@ pub async fn handle_connection(
     state: Arc<AppState>,
     addr: SocketAddr,
     tenant: TenantContext,
+) {
+    let conn_id = Uuid::new_v4();
+    let cancel = CancellationToken::new();
+    let community_id = tenant.community();
+    let registry = Arc::clone(&state.community_connections);
+    let check_state = Arc::clone(&state);
+    let run_state = Arc::clone(&state);
+    run_registered_community_connection(
+        &registry,
+        conn_id,
+        community_id,
+        cancel.clone(),
+        move || async move { check_state.db.is_community_active(community_id).await },
+        move || handle_active_connection(socket, run_state, addr, tenant, conn_id, cancel),
+    )
+    .await;
+}
+
+async fn handle_active_connection(
+    socket: WebSocket,
+    state: Arc<AppState>,
+    addr: SocketAddr,
+    tenant: TenantContext,
+    conn_id: Uuid,
+    cancel: CancellationToken,
 ) {
     let permit = match state.conn_semaphore.clone().try_acquire_owned() {
         Ok(p) => p,
@@ -129,9 +154,7 @@ pub async fn handle_connection(
         }
     };
 
-    let conn_id = Uuid::new_v4();
     let challenge = generate_challenge();
-    let cancel = CancellationToken::new();
 
     let (tx, rx) = mpsc::channel::<WsMessage>(state.config.send_buffer_size);
     // Control channel for Pong/Close — small capacity, guaranteed delivery

crates/buzz-relay/src/main.rs

@@ -19,6 +19,7 @@ use buzz_relay::router::{build_health_router, build_router};
 use buzz_relay::state::AppState;
 use buzz_relay::telemetry;
 use buzz_workflow::WorkflowEngine;
+use tokio_util::sync::CancellationToken;
 
 fn buzz_auto_migrate_enabled(value: Option<&str>) -> bool {
     value.map(str::trim).is_some_and(|value| {
@@ -781,6 +782,24 @@ async fn main() -> anyhow::Result<()> {
         });
     }
 
+    // Durable lifecycle backstop: Redis pub/sub cannot deliver to a pod that was
+    // offline. Periodically revalidate only communities with local live sockets
+    // so missed archive commands still converge without a global DB scan.
+    {
+        let lifecycle_state = Arc::clone(&state);
+        let interval_secs = std::env::var("BUZZ_COMMUNITY_REVALIDATE_INTERVAL_SECS")
+            .ok()
+            .and_then(|value| value.parse::<u64>().ok())
+            .unwrap_or(30)
+            .clamp(1, 300);
+        let cancel = lifecycle_state.community_revalidator_cancel.clone();
+        tokio::spawn(run_community_revalidator(
+            lifecycle_state,
+            std::time::Duration::from_secs(interval_secs),
+            cancel,
+        ));
+    }
+
     // Cross-pod connection-control consumer: receive disconnect commands from
     // Redis pub/sub (published by the pod that recorded a ban) and close any
     // matching local sockets. A member's live connections may land on any pod,
@@ -794,6 +813,11 @@ async fn main() -> anyhow::Result<()> {
             loop {
                 match rx.recv().await {
                     Ok(scoped) => match scoped.command {
+                        buzz_pubsub::conn_control::ConnControl::DisconnectCommunity => {
+                            state_for_conn_ctrl
+                                .community_connections
+                                .disconnect_community(scoped.community_id);
+                        }
                         buzz_pubsub::conn_control::ConnControl::DisconnectPubkey {
                             pubkey,
                             event_id,
@@ -896,6 +920,7 @@ async fn main() -> anyhow::Result<()> {
     }
 
     serve(router, health_router, Arc::clone(&state)).await?;
+    state.community_revalidator_cancel.cancel();
 
     // Signal the audit worker to stop accepting, flush buffered entries, and
     // exit. Uses a CancellationToken so it works regardless of how many
@@ -914,6 +939,42 @@ async fn main() -> anyhow::Result<()> {
     Ok(())
 }
 
+async fn run_community_revalidator(
+    state: Arc<AppState>,
+    period: std::time::Duration,
+    cancel: CancellationToken,
+) {
+    run_periodic_until_cancelled(period, cancel, || async {
+        let closed = state.revalidate_live_communities().await;
+        if closed > 0 {
+            tracing::info!(
+                closed,
+                "closed sockets for inactive communities during lifecycle revalidation"
+            );
+        }
+    })
+    .await;
+}
+
+async fn run_periodic_until_cancelled<Tick, TickFuture>(
+    period: std::time::Duration,
+    cancel: CancellationToken,
+    mut tick: Tick,
+) where
+    Tick: FnMut() -> TickFuture,
+    TickFuture: std::future::Future<Output = ()>,
+{
+    let mut interval = tokio::time::interval(period);
+    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    loop {
+        tokio::select! {
+            biased;
+            _ = cancel.cancelled() => break,
+            _ = interval.tick() => tick().await,
+        }
+    }
+}
+
 /// Bind all listeners and run with graceful shutdown.
 ///
 /// ```text
@@ -1471,7 +1532,37 @@ async fn run_usage_metrics_tick(
 
 #[cfg(test)]
 mod tests {
-    use super::buzz_auto_migrate_enabled;
+    use std::sync::Arc;
+    use std::time::Duration;
+
+    use tokio_util::sync::CancellationToken;
+
+    use super::{buzz_auto_migrate_enabled, run_periodic_until_cancelled};
+
+    #[tokio::test(start_paused = true)]
+    async fn periodic_loop_exits_immediately_on_cancellation() {
+        let cancel = CancellationToken::new();
+        let tick_count = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let count_for_tick = Arc::clone(&tick_count);
+        let task_cancel = cancel.clone();
+        let task = tokio::spawn(async move {
+            run_periodic_until_cancelled(Duration::from_secs(300), task_cancel, move || {
+                let count = Arc::clone(&count_for_tick);
+                async move {
+                    count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                }
+            })
+            .await;
+        });
+
+        tokio::task::yield_now().await;
+        cancel.cancel();
+        tokio::time::timeout(Duration::from_millis(1), task)
+            .await
+            .expect("loop must not wait for the next interval")
+            .expect("loop task");
+        assert!(tick_count.load(std::sync::atomic::Ordering::Relaxed) <= 1);
+    }
 
     #[test]
     fn buzz_auto_migrate_is_opt_in() {

crates/buzz-relay/src/router.rs

@@ -64,6 +64,10 @@ pub fn build_router(state: Arc<AppState>) -> Router {
             "/operator/communities",
             get(api::operator::list_owned_communities).post(api::operator::provision_community),
         )
+        .route(
+            "/operator/communities/archive",
+            post(api::operator::archive_community),
+        )
         .route(
             "/operator/communities/availability",
             get(api::operator::community_availability),

crates/buzz-relay/src/state.rs

@@ -1,7 +1,8 @@
 //! Shared application state — Arc-wrapped, shared across all connections.
 
 use std::collections::{HashMap, HashSet};
-use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
+use std::future::Future;
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicU8, Ordering};
 use std::sync::Arc;
 use std::time::Instant;
 
@@ -55,7 +56,128 @@ struct ConnEntry {
     grace_limit: u8,
 }
 
-/// Tracks active WebSocket connections and provides message routing by connection ID.
+/// Community-scoped lifecycle registry shared by every long-lived socket type.
+///
+/// A handler registers before durable active-state revalidation. Archival after
+/// registration cancels the token; archival before registration is observed by
+/// the revalidation. The returned guard removes the entry on every exit path.
+pub struct CommunityConnectionRegistry {
+    connections: Arc<DashMap<Uuid, (CommunityId, CancellationToken)>>,
+}
+
+impl Default for CommunityConnectionRegistry {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl CommunityConnectionRegistry {
+    /// Creates an empty lifecycle registry.
+    pub fn new() -> Self {
+        Self {
+            connections: Arc::new(DashMap::new()),
+        }
+    }
+
+    /// Registers one socket and returns a guard that deregisters it on drop.
+    pub fn register(
+        &self,
+        connection_id: Uuid,
+        community_id: CommunityId,
+        cancel: CancellationToken,
+    ) -> CommunityConnectionGuard {
+        self.connections
+            .insert(connection_id, (community_id, cancel));
+        CommunityConnectionGuard {
+            connection_id,
+            connections: Arc::clone(&self.connections),
+        }
+    }
+
+    /// Cancels every socket type currently bound to `community_id`.
+    pub fn disconnect_community(&self, community_id: CommunityId) -> usize {
+        let mut closed = 0;
+        for entry in self.connections.iter() {
+            if entry.value().0 == community_id {
+                entry.value().1.cancel();
+                closed += 1;
+            }
+        }
+        closed
+    }
+
+    /// Returns the distinct communities with live sockets on this pod.
+    pub fn bound_communities(&self) -> HashSet<CommunityId> {
+        self.connections
+            .iter()
+            .map(|entry| entry.value().0)
+            .collect()
+    }
+}
+
+/// Removes a socket lifecycle registration on every handler exit path.
+pub struct CommunityConnectionGuard {
+    connection_id: Uuid,
+    connections: Arc<DashMap<Uuid, (CommunityId, CancellationToken)>>,
+}
+
+impl Drop for CommunityConnectionGuard {
+    fn drop(&mut self) {
+        self.connections.remove(&self.connection_id);
+    }
+}
+
+/// Registers a socket, durably revalidates its community, then runs it.
+///
+/// The ordering is the archival admission invariant: archive-before-query is
+/// observed by the query, while archive-after-registration sees the token.
+pub async fn run_registered_community_connection<Check, CheckFuture, Run, RunFuture>(
+    registry: &CommunityConnectionRegistry,
+    connection_id: Uuid,
+    community_id: CommunityId,
+    cancel: CancellationToken,
+    check_active: Check,
+    run: Run,
+) where
+    Check: FnOnce() -> CheckFuture,
+    CheckFuture: Future<Output = Result<bool, buzz_db::DbError>>,
+    Run: FnOnce() -> RunFuture,
+    RunFuture: Future<Output = ()>,
+{
+    let _guard = registry.register(connection_id, community_id, cancel.clone());
+    if !matches!(check_active().await, Ok(true)) {
+        cancel.cancel();
+        return;
+    }
+    if cancel.is_cancelled() {
+        return;
+    }
+    run().await;
+    cancel.cancel();
+}
+
+async fn revalidate_registered_communities<Check, CheckFuture>(
+    registry: &CommunityConnectionRegistry,
+    mut check_active: Check,
+) -> (usize, Vec<(CommunityId, buzz_db::DbError)>)
+where
+    Check: FnMut(CommunityId) -> CheckFuture,
+    CheckFuture: Future<Output = Result<bool, buzz_db::DbError>>,
+{
+    let communities = registry.bound_communities();
+    let mut closed = 0;
+    let mut failures = Vec::new();
+    for community_id in communities {
+        match check_active(community_id).await {
+            Ok(false) => closed += registry.disconnect_community(community_id),
+            Ok(true) => {}
+            Err(error) => failures.push((community_id, error)),
+        }
+    }
+    (closed, failures)
+}
+
+/// Tracks active Nostr WebSocket connections and provides message routing by connection ID.
 pub struct ConnectionManager {
     connections: DashMap<Uuid, ConnEntry>,
 }
@@ -327,6 +449,12 @@ pub struct AppState {
     pub sub_registry: Arc<SubscriptionRegistry>,
     /// Registry of active WebSocket connections.
     pub conn_manager: Arc<ConnectionManager>,
+    /// Lifecycle cancellation for every long-lived socket, including huddle audio.
+    pub community_connections: Arc<CommunityConnectionRegistry>,
+    /// Stops only the periodic lifecycle revalidator during graceful shutdown.
+    pub community_revalidator_cancel: CancellationToken,
+    /// Test/telemetry counter for archive disconnect publication attempts.
+    pub community_disconnect_publish_attempts: Arc<AtomicU64>,
     /// Semaphore limiting total concurrent connections.
     pub conn_semaphore: Arc<Semaphore>,
     /// Semaphore limiting concurrent message handler tasks.
@@ -507,6 +635,9 @@ impl AppState {
             search: search_arc,
             sub_registry: Arc::new(SubscriptionRegistry::new()),
             conn_manager: Arc::new(ConnectionManager::new()),
+            community_connections: Arc::new(CommunityConnectionRegistry::new()),
+            community_revalidator_cancel: CancellationToken::new(),
+            community_disconnect_publish_attempts: Arc::new(AtomicU64::new(0)),
             conn_semaphore: Arc::new(Semaphore::new(max_connections)),
             handler_semaphore: Arc::new(Semaphore::new(max_concurrent_handlers)),
             git_semaphore: Arc::new(Semaphore::new(git_max_concurrent_ops)),
@@ -802,6 +933,10 @@ impl AppState {
             event_id: event_id.to_string(),
             reason: reason.to_string(),
         };
+        // This pre-existing ban path may remain fire-and-forget because the
+        // durable ban row rejects the member again at auth. Community archival
+        // is different: its API awaits publication and live sockets also have a
+        // periodic durable-state revalidation backstop below.
         tokio::spawn(async move {
             if let Err(e) = pubsub.publish_conn_control(&tenant, &command).await {
                 tracing::warn!("Failed to publish conn-control disconnect: {e}");
@@ -811,6 +946,42 @@ impl AppState {
         closed
     }
 
+    /// Disconnect a community locally and publish the command to every relay pod.
+    ///
+    /// Publication is awaited so the archive API can distinguish durable state
+    /// from propagation completion and offer a retryable response on failure.
+    pub async fn disconnect_community_clusterwide(
+        &self,
+        tenant: &TenantContext,
+    ) -> Result<usize, buzz_pubsub::PubSubError> {
+        let closed = self
+            .community_connections
+            .disconnect_community(tenant.community());
+        self.community_disconnect_publish_attempts
+            .fetch_add(1, Ordering::Relaxed);
+        self.pubsub
+            .publish_conn_control(tenant, &ConnControl::DisconnectCommunity)
+            .await?;
+        Ok(closed)
+    }
+
+    /// Revalidate all communities with live sockets and cancel inactive ones.
+    ///
+    /// This is the durable backstop for Redis pub/sub's lossy offline-subscriber
+    /// semantics: a pod that missed a successful publish eventually observes the
+    /// archived row directly.
+    pub async fn revalidate_live_communities(&self) -> usize {
+        let (closed, failures) =
+            revalidate_registered_communities(&self.community_connections, |community_id| {
+                self.db.is_community_active(community_id)
+            })
+            .await;
+        for (community_id, error) in failures {
+            tracing::warn!(%community_id, %error, "community lifecycle revalidation failed; retaining its sockets until next tick");
+        }
+        closed
+    }
+
     /// Get accessible channel IDs with a 10-second cache. Falls back to DB on miss.
     pub async fn get_accessible_channel_ids_cached(
         &self,
@@ -1296,6 +1467,132 @@ mod tests {
         );
     }
 
+    #[test]
+    fn community_lifecycle_disconnect_covers_socket_types_and_preserves_tenant_fence() {
+        let registry = CommunityConnectionRegistry::new();
+        let community_a = CommunityId::from_uuid(Uuid::from_u128(0xa));
+        let community_b = CommunityId::from_uuid(Uuid::from_u128(0xb));
+        let ordinary_a = CancellationToken::new();
+        let audio_a = CancellationToken::new();
+        let ordinary_b = CancellationToken::new();
+        let _ordinary_a_guard = registry.register(Uuid::new_v4(), community_a, ordinary_a.clone());
+        let _audio_a_guard = registry.register(Uuid::new_v4(), community_a, audio_a.clone());
+        let _ordinary_b_guard = registry.register(Uuid::new_v4(), community_b, ordinary_b.clone());
+
+        assert_eq!(registry.disconnect_community(community_a), 2);
+        assert!(ordinary_a.is_cancelled());
+        assert!(audio_a.is_cancelled());
+        assert!(!ordinary_b.is_cancelled());
+    }
+
+    #[tokio::test]
+    async fn register_then_revalidate_closes_both_archive_race_orderings() {
+        let registry = CommunityConnectionRegistry::new();
+        let community = CommunityId::from_uuid(Uuid::from_u128(0xa));
+
+        // Archive wins before durable revalidation: the check observes inactive
+        // and the socket body never starts.
+        let cancel_before = CancellationToken::new();
+        let started_before = Arc::new(AtomicBool::new(false));
+        let started_before_run = Arc::clone(&started_before);
+        run_registered_community_connection(
+            &registry,
+            Uuid::new_v4(),
+            community,
+            cancel_before.clone(),
+            || async { Ok(false) },
+            move || async move { started_before_run.store(true, Ordering::SeqCst) },
+        )
+        .await;
+        assert!(cancel_before.is_cancelled());
+        assert!(!started_before.load(Ordering::SeqCst));
+
+        // Archive wins after registration but while revalidation is paused: its
+        // sweep sees the token, and even an active query result cannot start the
+        // socket body afterward.
+        let cancel_during = CancellationToken::new();
+        let registered = Arc::new(tokio::sync::Notify::new());
+        let resume = Arc::new(tokio::sync::Notify::new());
+        let registered_check = Arc::clone(&registered);
+        let resume_check = Arc::clone(&resume);
+        let started_during = Arc::new(AtomicBool::new(false));
+        let started_during_run = Arc::clone(&started_during);
+        let future = run_registered_community_connection(
+            &registry,
+            Uuid::new_v4(),
+            community,
+            cancel_during.clone(),
+            move || async move {
+                registered_check.notify_one();
+                resume_check.notified().await;
+                Ok(true)
+            },
+            move || async move { started_during_run.store(true, Ordering::SeqCst) },
+        );
+        tokio::pin!(future);
+        tokio::select! {
+            _ = registered.notified() => {}
+            _ = &mut future => panic!("revalidation should be paused"),
+        }
+        assert_eq!(registry.disconnect_community(community), 1);
+        resume.notify_one();
+        future.await;
+        assert!(cancel_during.is_cancelled());
+        assert!(!started_during.load(Ordering::SeqCst));
+    }
+
+    #[tokio::test]
+    async fn revalidation_continues_after_one_community_lookup_failure() {
+        let registry = CommunityConnectionRegistry::new();
+        let archived_a = CommunityId::from_uuid(Uuid::from_u128(0xa));
+        let failed = CommunityId::from_uuid(Uuid::from_u128(0xb));
+        let archived_c = CommunityId::from_uuid(Uuid::from_u128(0xc));
+        let cancel_a = CancellationToken::new();
+        let cancel_failed = CancellationToken::new();
+        let cancel_c = CancellationToken::new();
+        let _guard_a = registry.register(Uuid::new_v4(), archived_a, cancel_a.clone());
+        let _guard_failed = registry.register(Uuid::new_v4(), failed, cancel_failed.clone());
+        let _guard_c = registry.register(Uuid::new_v4(), archived_c, cancel_c.clone());
+
+        let (closed, failures) =
+            revalidate_registered_communities(&registry, |community| async move {
+                if community == failed {
+                    Err(buzz_db::DbError::InvalidData(
+                        "injected lookup failure".into(),
+                    ))
+                } else {
+                    Ok(false)
+                }
+            })
+            .await;
+
+        assert_eq!(closed, 2);
+        assert!(cancel_a.is_cancelled());
+        assert!(!cancel_failed.is_cancelled());
+        assert!(cancel_c.is_cancelled());
+        assert_eq!(failures.len(), 1);
+        assert_eq!(failures[0].0, failed);
+        assert_eq!(
+            registry.bound_communities(),
+            HashSet::from([archived_a, failed, archived_c])
+        );
+    }
+
+    #[test]
+    fn community_lifecycle_guard_deregisters_on_early_return() {
+        let registry = CommunityConnectionRegistry::new();
+        let community = CommunityId::from_uuid(Uuid::from_u128(0xa));
+        let cancel = CancellationToken::new();
+        let guard = registry.register(Uuid::new_v4(), community, cancel.clone());
+        assert_eq!(registry.bound_communities(), HashSet::from([community]));
+
+        drop(guard);
+
+        assert!(registry.bound_communities().is_empty());
+        assert_eq!(registry.disconnect_community(community), 0);
+        assert!(!cancel.is_cancelled());
+    }
+
     #[tokio::test]
     async fn disconnect_pubkey_closes_matching_conns_with_reason() {
         let (mgr, id, _rx, mut ctrl_rx, cancel, _bp) = setup_conn(8);

migrations/0016_community_archival.sql

@@ -0,0 +1,3 @@
+-- Durable community archival state. Archived hosts remain reserved by the existing
+-- full unique index and continue to count toward owner quotas.
+ALTER TABLE communities ADD COLUMN archived_at TIMESTAMPTZ;

schema/schema.sql

@@ -55,6 +55,7 @@ CREATE TABLE communities (
     host            VARCHAR(255) NOT NULL,
     signing_key     BYTEA,
     created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    archived_at     TIMESTAMPTZ,
     CONSTRAINT chk_communities_id_not_nil CHECK (id <> '00000000-0000-0000-0000-000000000000'::uuid)
 );