fix(relay): disconnect removed members cluster-wide

crates/buzz-relay/src/handlers/auth.rs

@@ -275,10 +275,17 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, state:
             }
 
             info!(conn_id = %conn_id, pubkey = %pubkey.to_hex(), "NIP-42 auth successful");
+            let membership_pubkey = auth_ctx
+                .agent_owner_pubkey
+                .unwrap_or(pubkey)
+                .to_bytes()
+                .to_vec();
             *conn.auth_state.write().await = AuthState::Authenticated(auth_ctx);
-            state
-                .conn_manager
-                .set_authenticated_pubkey(conn_id, pubkey.to_bytes().to_vec());
+            state.conn_manager.set_authenticated_membership(
+                conn_id,
+                pubkey.to_bytes().to_vec(),
+                membership_pubkey,
+            );
             conn.send(RelayMessage::ok(&event_id_hex, true, ""));
         }
         Err(e) => {

crates/buzz-relay/src/handlers/event.rs

@@ -608,13 +608,14 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
     )
     .increment(1);
 
-    let (conn_id, pubkey_bytes, auth_pubkey, scopes, channel_ids) = {
+    let (conn_id, pubkey_bytes, auth_pubkey, membership_pubkey, scopes, channel_ids) = {
         let auth = conn.auth_state.read().await;
         match &*auth {
             AuthState::Authenticated(ctx) => (
                 conn.conn_id,
                 ctx.pubkey.to_bytes().to_vec(),
                 ctx.pubkey,
+                ctx.agent_owner_pubkey.unwrap_or(ctx.pubkey),
                 ctx.scopes.clone(),
                 ctx.channel_ids.clone(),
             ),
@@ -630,6 +631,40 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
         }
     };
 
+    // Redis disconnect delivery is best-effort. On closed relays, every new
+    // EVENT therefore rechecks the durable member principal before processing
+    // either the ephemeral or persistent path. NIP-OA agents are authorized by
+    // their verified owner, not by an absent direct relay_members row.
+    if state.config.require_relay_membership {
+        match state
+            .db
+            .is_relay_member(conn.tenant.community(), &membership_pubkey.to_hex())
+            .await
+        {
+            Ok(true) => {}
+            Ok(false) => {
+                reject("membership");
+                state.conn_manager.disconnect_membership_principal(
+                    conn.tenant.community(),
+                    membership_pubkey.as_bytes(),
+                    &event_id_hex,
+                    "blocked: relay membership removed",
+                );
+                return;
+            }
+            Err(error) => {
+                warn!(%error, "relay membership recheck failed");
+                reject("membership");
+                conn.send(RelayMessage::ok(
+                    &event_id_hex,
+                    false,
+                    "restricted: relay membership could not be verified",
+                ));
+                return;
+            }
+        }
+    }
+
     // Must run before both ephemeral and persistent branches. Persistent
     // events get a second check inside ingest_event() (step 3), but
     // ephemeral events bypass the pipeline entirely.

crates/buzz-relay/src/handlers/relay_admin.rs

@@ -29,6 +29,38 @@ use crate::handlers::side_effects::{
 };
 use crate::state::AppState;
 
+const RELAY_MEMBERSHIP_REMOVED_REASON: &str = "blocked: relay membership removed";
+
+#[derive(Debug, Eq, PartialEq)]
+enum RemovalDisposition {
+    Removed,
+    AlreadyAbsent,
+}
+
+fn removal_disposition(
+    result: RemoveResult,
+    target_is_virtual_agent: bool,
+    require_relay_membership: bool,
+) -> Result<RemovalDisposition, String> {
+    match result {
+        RemoveResult::Removed => Ok(RemovalDisposition::Removed),
+        // A repeated command must be safe and must re-fan-out the disconnect
+        // in case the first Redis publish was lost. Durable absence is still
+        // the authorization backstop. A NIP-OA virtual agent is deliberately
+        // absent from relay_members, so it is not a valid idempotent retry.
+        RemoveResult::NotFound if require_relay_membership && !target_is_virtual_agent => {
+            Ok(RemovalDisposition::AlreadyAbsent)
+        }
+        RemoveResult::NotFound => {
+            Err("member not found: absence does not prove a removable member".to_string())
+        }
+        RemoveResult::IsOwner => Err("cannot remove the relay owner".to_string()),
+        RemoveResult::RoleMismatch => {
+            Err("actor not authorized: admins can only remove members".to_string())
+        }
+    }
+}
+
 /// Extract the hex pubkey from the first `p` tag, returning it as a `String`.
 fn extract_p_tag_hex(event: &Event) -> Option<String> {
     for tag in event.tags.iter() {
@@ -252,30 +284,50 @@ pub async fn handle_relay_admin_event(
                     .map_err(|e| format!("database error: {e}"))?
             };
 
-            match remove_result {
-                RemoveResult::Removed => {}
-                RemoveResult::IsOwner => {
-                    return Err("cannot remove the relay owner".to_string());
-                }
-                RemoveResult::NotFound => {
-                    return Err(format!("member not found: {target_hex}"));
-                }
-                RemoveResult::RoleMismatch => {
-                    return Err("actor not authorized: admins can only remove members".to_string());
-                }
-            }
+            let target_pubkey = hex::decode(&target_hex)
+                .map_err(|e| format!("invalid target pubkey encoding: {e}"))?;
+            let target_is_virtual_agent = if remove_result == RemoveResult::NotFound {
+                state
+                    .db
+                    .get_agent_channel_policy(tenant.community(), &target_pubkey)
+                    .await
+                    .map_err(|e| format!("database error: {e}"))?
+                    .is_some_and(|(_, owner)| owner.is_some())
+            } else {
+                false
+            };
+            let disposition = removal_disposition(
+                remove_result,
+                target_is_virtual_agent,
+                state.config.require_relay_membership,
+            )?;
+
+            // The DB mutation/absence check above is the durable authorization
+            // backstop. Close matching sockets on this pod and fan the same
+            // community-scoped command to every other pod. Repeating an
+            // already-completed removal intentionally replays this fan-out.
+            let closed_locally = state.disconnect_pubkey_clusterwide(
+                tenant,
+                &target_pubkey,
+                &event.id.to_hex(),
+                RELAY_MEMBERSHIP_REMOVED_REASON,
+            );
 
             info!(
                 sender = %sender_hex,
                 target = %target_hex,
-                "relay member removed"
+                ?disposition,
+                closed_locally,
+                "relay member removal enforced"
             );
 
-            if let Err(e) = publish_nip43_member_removed(tenant, state, &target_hex).await {
-                warn!(error = %e, "failed to publish NIP-43 member removed event");
-            }
-            if let Err(e) = publish_nip43_membership_list(tenant, state).await {
-                warn!(error = %e, "failed to publish NIP-43 membership list");
+            if disposition == RemovalDisposition::Removed {
+                if let Err(e) = publish_nip43_member_removed(tenant, state, &target_hex).await {
+                    warn!(error = %e, "failed to publish NIP-43 member removed event");
+                }
+                if let Err(e) = publish_nip43_membership_list(tenant, state).await {
+                    warn!(error = %e, "failed to publish NIP-43 membership list");
+                }
             }
         }
 
@@ -431,6 +483,46 @@ mod tests {
         assert_eq!(extract_tag_value(&event, "p"), None);
     }
 
+    #[test]
+    fn removal_disposition_accepts_idempotent_retry() {
+        assert_eq!(
+            removal_disposition(RemoveResult::Removed, false, true),
+            Ok(RemovalDisposition::Removed)
+        );
+        assert_eq!(
+            removal_disposition(RemoveResult::NotFound, false, true),
+            Ok(RemovalDisposition::AlreadyAbsent)
+        );
+    }
+
+    #[test]
+    fn removal_disposition_rejects_virtual_agent_as_retry() {
+        assert_eq!(
+            removal_disposition(RemoveResult::NotFound, true, true),
+            Err("member not found: absence does not prove a removable member".to_string())
+        );
+    }
+
+    #[test]
+    fn removal_disposition_rejects_absent_target_on_open_relay() {
+        assert_eq!(
+            removal_disposition(RemoveResult::NotFound, false, false),
+            Err("member not found: absence does not prove a removable member".to_string())
+        );
+    }
+
+    #[test]
+    fn removal_disposition_preserves_protected_role_errors() {
+        assert_eq!(
+            removal_disposition(RemoveResult::IsOwner, false, true),
+            Err("cannot remove the relay owner".to_string())
+        );
+        assert_eq!(
+            removal_disposition(RemoveResult::RoleMismatch, false, true),
+            Err("actor not authorized: admins can only remove members".to_string())
+        );
+    }
+
     #[test]
     fn workspace_icon_empty_ok() {
         assert!(validate_workspace_icon("").is_ok());

crates/buzz-relay/src/main.rs

@@ -892,6 +892,21 @@ async fn main() -> anyhow::Result<()> {
         ));
     }
 
+    // Durable relay-membership backstop: a pod can miss a Redis disconnect
+    // while offline or disconnected. Recheck only principals that currently
+    // authorize local sockets and converge within 15 seconds.
+    if state.config.require_relay_membership {
+        let membership_state = Arc::clone(&state);
+        let interval_value = std::env::var("BUZZ_RELAY_MEMBERSHIP_REVALIDATE_INTERVAL_SECS").ok();
+        let interval_secs = membership_revalidate_interval_secs(interval_value.as_deref());
+        let cancel = membership_state.community_revalidator_cancel.clone();
+        tokio::spawn(run_membership_revalidator(
+            membership_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,
@@ -1077,6 +1092,30 @@ async fn run_community_revalidator(
     .await;
 }
 
+async fn run_membership_revalidator(
+    state: Arc<AppState>,
+    period: std::time::Duration,
+    cancel: CancellationToken,
+) {
+    run_periodic_until_cancelled(period, cancel, || async {
+        let closed = state.revalidate_live_memberships().await;
+        if closed > 0 {
+            tracing::info!(
+                closed,
+                "closed sockets for removed relay members during revalidation"
+            );
+        }
+    })
+    .await;
+}
+
+fn membership_revalidate_interval_secs(value: Option<&str>) -> u64 {
+    value
+        .and_then(|value| value.parse::<u64>().ok())
+        .unwrap_or(10)
+        .clamp(1, 15)
+}
+
 async fn run_periodic_until_cancelled<Tick, TickFuture>(
     period: std::time::Duration,
     cancel: CancellationToken,
@@ -1805,8 +1844,8 @@ mod tests {
 
     use super::{
         buzz_auto_migrate_enabled, dropped_in_memory_keys, idle_timeout_secs,
-        refresh_legacy_active_gauge_recency, run_periodic_until_cancelled, EmissionScope,
-        InMemoryMetricKey,
+        membership_revalidate_interval_secs, refresh_legacy_active_gauge_recency,
+        run_periodic_until_cancelled, EmissionScope, InMemoryMetricKey,
     };
     use metrics::GaugeFn;
     use metrics_util::{
@@ -1854,6 +1893,14 @@ mod tests {
         assert!(buzz_auto_migrate_enabled(Some("on")));
     }
 
+    #[test]
+    fn membership_revalidation_interval_never_exceeds_fifteen_seconds() {
+        assert_eq!(membership_revalidate_interval_secs(None), 10);
+        assert_eq!(membership_revalidate_interval_secs(Some("0")), 1);
+        assert_eq!(membership_revalidate_interval_secs(Some("15")), 15);
+        assert_eq!(membership_revalidate_interval_secs(Some("300")), 15);
+    }
+
     #[test]
     fn test_emission_scope_off_disallows_every_community() {
         assert!(EmissionScope::All.allows(&Uuid::new_v4()));