fix(relay): close legacy sockets after community FI upgrades (#8991)

## Summary

When a community starts requiring NIP-FI, the periodic community
revalidator now closes its existing legacy root and audio WebSockets
with Close 1008. It reads current policy from the writer by immutable
community ID, with a 500 ms budget per read. Failed reads retry on later
sweeps; late registrations are caught by the next pass.

Enforced FI admissions survive. Optional-shadow admissions remain
legacy. Off and repair mode retain their existing behavior. No client
request or Redis notification is needed.

The default sweep interval is 30 seconds, not a deadline during database
failure, overload or paused processes. Already-running requests and
descendant work may finish. Receipts prove policy acceptance and remain
`pending`. The `pending_reasons` value changes from
`registry_baseline_unverified` to `connection_closure_unverified`;
clients must tolerate unknown reasons. This change does not certify
fleet-wide completion.

### Related issue

[BUZZ-512](https://linear.app/squareup/issue/BUZZ-512), under the
revised eventual-closure scope. Builds on merged #8985 and #8988. The
replacement starts from main and adds no activation schema or migration.

### Rollout and limits

There is no new flag. Count already-required communities before
deployment: each relay closes their legacy sockets on its first
successful sweep. Closures within a community can cause simultaneous
reconnects. Keep new upgrades disabled until every serving relay runs
this version.

Observe `buzz_community_auth_policy_checks_total{outcome}` and
`buzz_community_auth_policy_disconnects_total` per relay, and verify
real idle root/audio Close 1008 probes plus FI reconnects. Counters
record checks and cancellation requests, not fleet completion. A binary
rollback requires no migration and restores the old pending reason, but
loses eventual closure; it does not undo durable policy. See the
[operator
contract](1c74532516/docs/community-auth-upgrades.md).

### Testing

The real-socket journey runs a local relay against isolated PostgreSQL
and Redis. It covers idle root/audio closure after a durable change
without pubsub, actual read timeouts and recovery, late registration on
both routes, FI reconnects, unrelated sockets, and Off/repair/Enforce
controls.

With local PostgreSQL and Redis configured per the repository's test
guide, run:

```sh
CARGO_TARGET_DIR=target/fi-closure-review scripts/postgres-test-run.sh -p buzz-relay --lib -E 'test(sweep_closes_idle_root_audio_and_late_registrations_after_read_recovery)'
```

Expected: one passing journey, covering both late-registration routes
and all socket controls.

The journey passed at `1c74532516f697c2f06498646326f8c479574b14` in a
branch-specific build target (5.56 seconds). Independent source review
has no remaining P1/P2 findings. Earlier shared-target results are
unreliable and are not acceptance evidence. All mandatory push hooks
passed in the same isolated target, including Rust unit tests, both
Tauri Clippy modes, Tauri tests and file-size checks. Hosted CI passed
at this head. The full local `just ci` chain was not completed. Human
runtime acceptance remains open. No merge or deployment is included.

Generated with Codex

---------

Signed-off-by: Brad Seiler <seiler@squareup.com>
Co-authored-by: Honor <e0bf479eeebe47dde1e35705212a4df70616a7bc834d907ddfb18cf0405f8b1b@seiler-test.communities.public.builderlabstaging.xyz>

crates/buzz-db/src/store/community_auth.rs

@@ -112,7 +112,7 @@ const RECEIPT: &str = "SELECT u.*, true AS requires_nip_fi,
         AND EXISTS (SELECT 1 FROM community_nip_fi_issuers i
             WHERE i.community_id = c.id AND i.issuer = u.expected_issuer)
     THEN 'pending' ELSE 'unavailable' END AS state,
-    ARRAY['registry_baseline_unverified']::text[] AS pending_reasons
+    ARRAY['connection_closure_unverified']::text[] AS pending_reasons
     FROM community_auth_upgrades u
     LEFT JOIN communities c ON c.id = u.target_community_id
     WHERE u.operation_id = $1";

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

@@ -3243,7 +3243,7 @@ mod postgres_tests {
         assert_eq!(record["state"], "pending");
         assert_eq!(
             record["pending_reasons"],
-            serde_json::json!(["registry_baseline_unverified"])
+            serde_json::json!(["connection_closure_unverified"])
         );
         // A fresh signed retry recovers the same tuple despite the lost response.
         let retry =

crates/buzz-relay/src/main.rs

@@ -1962,7 +1962,7 @@ async fn run_community_revalidator(
         if closed > 0 {
             tracing::info!(
                 closed,
-                "closed sockets for inactive communities during lifecycle revalidation"
+                "closed sockets for inactive communities or required authentication policy"
             );
         }
     })

crates/buzz-relay/src/nip_fi_policy.rs

@@ -11,7 +11,7 @@ use buzz_auth::{DenialClass, NipFiMode};
 use buzz_db::community_auth::CommunityAuthPolicy;
 use futures_util::future::{BoxFuture, FutureExt};
 
-const POLICY_READ_BUDGET: std::time::Duration = std::time::Duration::from_millis(500);
+pub(crate) const POLICY_READ_BUDGET: std::time::Duration = std::time::Duration::from_millis(500);
 
 /// Host policy source, shared by admission and discovery.
 pub(crate) trait PolicySource: Send + Sync {

crates/buzz-relay/src/nip_fi_test_hooks.rs

@@ -154,6 +154,7 @@ macro_rules! make_hook {
     };
 }
 
+make_hook!(community_registration_hook, before_community_registration);
 make_hook!(auth_commit_hook, before_auth_commit);
 make_hook!(event_ingest_hook, before_event_ingest);
 make_hook!(req_registration_hook, before_req_registration);

crates/buzz-relay/src/router.rs

@@ -2792,6 +2792,9 @@ mod tests {
         );
     }
 
+    #[path = "policy_sweep_postgres_tests.rs"]
+    mod policy_sweep;
+
     // ── F6: document fallback (postgres-only) ───────────────────────────────────
     //
     // A no-`Accept` plain GET / to a successfully mapped host must bypass the
@@ -3193,7 +3196,7 @@ mod tests {
             }
         }
 
-        async fn real_db_state() -> Option<Arc<AppState>> {
+        pub(super) async fn real_db_state() -> Option<Arc<AppState>> {
             let db_url = crate::test_support::database_url();
             let pool = sqlx::PgPool::connect(&db_url).await.ok()?;
 

crates/buzz-relay/src/router/tests/policy_sweep_postgres_tests.rs

@@ -0,0 +1,338 @@
+mod postgres_tests {
+    use super::super::postgres_tests::real_db_state;
+    use super::super::*;
+    use futures_util::{SinkExt, StreamExt};
+    use tokio_tungstenite::tungstenite::{client::IntoClientRequest, Message};
+    type Client = tokio_tungstenite::WebSocketStream<
+        tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
+    >;
+
+    async fn frame(client: &mut Client, stage: &str) -> Message {
+        tokio::time::timeout(std::time::Duration::from_secs(10), async {
+            loop {
+                let message = client.next().await.unwrap().unwrap();
+                match message {
+                    Message::Ping(_) => {
+                        client.flush().await.unwrap();
+                    }
+                    Message::Pong(_) => {}
+                    message => return message,
+                }
+            }
+        })
+        .await
+        .unwrap_or_else(|error| panic!("{stage}: {error}"))
+    }
+
+    async fn authenticated_socket(
+        addr: std::net::SocketAddr,
+        host: &str,
+        path: &str,
+        key: &nostr::Keys,
+        assertion: Option<&str>,
+        audio: bool,
+    ) -> Client {
+        let mut request = format!("ws://{addr}{path}").into_client_request().unwrap();
+        request.headers_mut().insert("host", host.parse().unwrap());
+        if let Some(token) = assertion {
+            request.headers_mut().insert(
+                "nostr-federated-identity",
+                format!("Bearer {token}").parse().unwrap(),
+            );
+        }
+        let (mut client, _) = tokio_tungstenite::connect_async(request).await.unwrap();
+        let Message::Text(challenge) = frame(&mut client, "challenge").await else {
+            panic!("challenge")
+        };
+        let challenge: serde_json::Value = serde_json::from_str(&challenge).unwrap();
+        let challenge = if audio {
+            challenge["challenge"].as_str().unwrap()
+        } else {
+            challenge[1].as_str().unwrap()
+        };
+        let event = nostr::EventBuilder::new(nostr::Kind::Authentication, "")
+            .tag(nostr::Tag::parse(["relay", &format!("ws://{host}")]).unwrap())
+            .tag(nostr::Tag::parse(["challenge", challenge]).unwrap())
+            .sign_with_keys(key)
+            .unwrap();
+        let auth = if audio {
+            serde_json::json!({"type":"auth","event":event,"parent_channel_id":null,"protocol_version":1})
+        } else {
+            serde_json::json!(["AUTH", event])
+        };
+        client
+            .send(Message::Text(auth.to_string().into()))
+            .await
+            .unwrap();
+        loop {
+            let Message::Text(text) = frame(&mut client, "authenticated socket").await else {
+                panic!("admission closed")
+            };
+            let value: serde_json::Value = serde_json::from_str(&text).unwrap();
+            if audio && value["type"] == "joined" {
+                break;
+            }
+            if !audio && value[0] == "OK" {
+                assert_eq!(value[2], true, "{value}");
+                break;
+            }
+            assert!(value["type"] != "error", "{value}");
+        }
+        client
+    }
+
+    async fn ping_round_trip(client: &mut Client) {
+        client
+            .send(Message::Ping(bytes::Bytes::from_static(b"live")))
+            .await
+            .unwrap();
+        tokio::time::timeout(std::time::Duration::from_secs(10), async {
+            loop {
+                match client.next().await.unwrap().unwrap() {
+                    Message::Pong(bytes) => {
+                        assert_eq!(bytes.as_ref(), b"live");
+                        break;
+                    }
+                    Message::Ping(_) => {
+                        client.flush().await.unwrap();
+                    }
+                    Message::Close(close) => panic!("unexpected close {close:?}"),
+                    _ => {}
+                }
+            }
+        })
+        .await
+        .unwrap();
+    }
+    #[tokio::test]
+    #[ignore = "requires Postgres and Redis"]
+    async fn sweep_closes_idle_root_audio_and_late_registrations_after_read_recovery() {
+        for late_audio in [false, true] {
+            sweep_journey(late_audio).await;
+        }
+    }
+
+    async fn sweep_journey(late_audio: bool) {
+        use crate::nip_fi_core::test_support::{
+            assertion_verifier, sign_assertion, TEST_COMMUNITY_URI, TEST_ISSUER,
+        };
+        use buzz_db::community_auth::{UpgradeOutcome, UpgradeRequest};
+        let base = real_db_state().await.unwrap();
+        let mut configured = (*base).clone();
+        let config = Arc::make_mut(&mut configured.config);
+        config.nip_fi.mode = buzz_auth::NipFiMode::Shadow;
+        config.nip_fi_upgrades_enabled = false;
+        config.nip_fi.max_connection_lifetime_secs = 3600;
+        config.nip_fi.communities = crate::nip_fi_config::NipFiCommunities::for_test_any_host(
+            TEST_COMMUNITY_URI,
+            &[TEST_ISSUER],
+        );
+        configured.nip_fi_verifier = Some(Arc::new(assertion_verifier(
+            buzz_auth::TokenClass::DedicatedNipFi,
+        )));
+        configured.nip_fi_deny_map = Some(Arc::new(buzz_auth::NipFiDenyMap::new(
+            16,
+            vec![buzz_auth::IssuerCapacity {
+                issuer: TEST_ISSUER.into(),
+                capacity: 16,
+            }],
+        )));
+        let redis = buzz_pubsub::redis_connection(&std::env::var("REDIS_URL").unwrap()).unwrap();
+        configured.admission_rate_limiter =
+            Arc::new(buzz_pubsub::rate_limiter::RedisRateLimiter::new(redis));
+        let state = Arc::new(configured);
+        let key = nostr::Keys::generate();
+        let owner = key.public_key().to_hex();
+        let host = format!("dark-target-{}.example", uuid::Uuid::new_v4());
+        let created = state
+            .db
+            .create_community_with_owner_and_nip_fi_issuers(
+                &host,
+                &owner,
+                &[TEST_ISSUER.into()],
+                &key,
+            )
+            .await
+            .unwrap();
+        let buzz_db::CreateCommunityWithOwnerResult::Created(row) = created.outcome else {
+            panic!("target")
+        };
+        let free_host = format!("dark-free-{}.example", uuid::Uuid::new_v4());
+        let _ = state
+            .db
+            .create_community_with_owner(&free_host, &owner, &key)
+            .await
+            .unwrap();
+        state
+            .db
+            .ensure_user(row.id, &key.public_key().to_bytes())
+            .await
+            .unwrap();
+        let channel = uuid::Uuid::new_v4();
+        state
+            .db
+            .create_channel_with_id(
+                row.id,
+                channel,
+                "dark-audio",
+                buzz_db::channel::ChannelType::Stream,
+                buzz_db::channel::ChannelVisibility::Open,
+                None,
+                &key.public_key().to_bytes(),
+                None,
+            )
+            .await
+            .unwrap();
+        sqlx::query("INSERT INTO channel_members(community_id,channel_id,pubkey,role) VALUES($1,$2,$3,'member') ON CONFLICT DO NOTHING")
+            .bind(row.id.as_uuid()).bind(channel).bind(key.public_key().to_bytes().to_vec()).execute(state.db.pool()).await.unwrap();
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let app = build_router(state.clone())
+            .into_make_service_with_connect_info::<std::net::SocketAddr>();
+        let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+        let path = format!("/huddle/{channel}/audio");
+        let mut root = authenticated_socket(addr, &host, "/", &key, None, false).await;
+        let mut audio = authenticated_socket(addr, &host, &path, &key, None, true).await;
+        let mut free = authenticated_socket(addr, &free_host, "/", &key, None, false).await;
+
+        // A callback already admitted under optional policy registers only after
+        // the sweep. Exercise both transport types in the same journey.
+        let (arrived, release) = crate::nip_fi_test_hooks::community_registration_hook::arm(row.id);
+        let mut late_request = format!(
+            "ws://{addr}{}",
+            if late_audio { path.as_str() } else { "/" }
+        )
+        .into_client_request()
+        .unwrap();
+        late_request
+            .headers_mut()
+            .insert("host", host.parse().unwrap());
+        let (mut late, _) = tokio_tungstenite::connect_async(late_request)
+            .await
+            .unwrap();
+        tokio::time::timeout(Duration::from_secs(10), arrived)
+            .await
+            .unwrap()
+            .unwrap();
+        let request = UpgradeRequest {
+            operation_id: uuid::Uuid::new_v4(),
+            community_id: *row.id.as_uuid(),
+            expected_host: host.clone(),
+            actor_pubkey: owner.clone(),
+            expected_owner_pubkey: owner.clone(),
+            expected_issuer: TEST_ISSUER.into(),
+            operator_pubkey: owner,
+        };
+        let UpgradeOutcome::Accepted(record) = state
+            .db
+            .enable_community_nip_fi(&request, Ok(()))
+            .await
+            .unwrap()
+        else {
+            panic!("enable")
+        };
+
+        // No invalidation or request on an existing client. A writer table lock
+        // blocks real policy reads until their 500ms budget expires.
+        let mut lock = state.db.pool().begin().await.unwrap();
+        sqlx::query("LOCK TABLE communities IN ACCESS EXCLUSIVE MODE")
+            .execute(&mut *lock)
+            .await
+            .unwrap();
+        assert_eq!(
+            tokio::time::timeout(Duration::from_secs(3), state.revalidate_live_communities())
+                .await
+                .unwrap(),
+            0
+        );
+        lock.rollback().await.unwrap();
+
+        let token = sign_assertion("nip-fi+jwt", &key.public_key().to_hex(), None);
+        let mut compliant = authenticated_socket(addr, &host, "/", &key, Some(&token), false).await;
+        // Keep compliant audio on a different channel so room cleanup is independent.
+        let next_channel = uuid::Uuid::new_v4();
+        state
+            .db
+            .create_channel_with_id(
+                row.id,
+                next_channel,
+                "fi-audio",
+                buzz_db::channel::ChannelType::Stream,
+                buzz_db::channel::ChannelVisibility::Open,
+                None,
+                &key.public_key().to_bytes(),
+                None,
+            )
+            .await
+            .unwrap();
+        sqlx::query("INSERT INTO channel_members(community_id,channel_id,pubkey,role) VALUES($1,$2,$3,'member') ON CONFLICT DO NOTHING")
+            .bind(row.id.as_uuid()).bind(next_channel).bind(key.public_key().to_bytes().to_vec()).execute(state.db.pool()).await.unwrap();
+        let mut compliant_audio = authenticated_socket(
+            addr,
+            &host,
+            &format!("/huddle/{next_channel}/audio"),
+            &key,
+            Some(&token),
+            true,
+        )
+        .await;
+        // Explicit Off and repair mode retain lifecycle-only sweep behavior.
+        for mode in [
+            buzz_auth::NipFiMode::Off,
+            buzz_auth::NipFiMode::DenyProtected,
+        ] {
+            let mut bypass = (*state).clone();
+            Arc::make_mut(&mut bypass.config).nip_fi.mode = mode;
+            assert_eq!(bypass.revalidate_live_communities().await, 0);
+        }
+        assert_eq!(state.revalidate_live_communities().await, 2);
+        for client in [&mut root, &mut audio] {
+            policy_close(client).await;
+        }
+        release.notify_one();
+        assert!(matches!(
+            frame(&mut late, "late root registers").await,
+            Message::Text(_)
+        ));
+        assert_eq!(state.revalidate_live_communities().await, 1);
+        policy_close(&mut late).await;
+
+        assert_eq!(state.revalidate_live_communities().await, 0);
+        for client in [&mut compliant, &mut compliant_audio, &mut free] {
+            ping_round_trip(client).await;
+        }
+        // Global Enforce also closes an optional community's legacy socket.
+        let mut enforce = (*state).clone();
+        Arc::make_mut(&mut enforce.config).nip_fi.mode = buzz_auth::NipFiMode::Enforce;
+        assert_eq!(enforce.revalidate_live_communities().await, 1);
+        policy_close(&mut free).await;
+        for client in [&mut compliant, &mut compliant_audio] {
+            ping_round_trip(client).await;
+        }
+        let status = state
+            .db
+            .community_auth_upgrade_status(record.operation_id, &request.operator_pubkey)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(status.state, "pending");
+        assert_eq!(
+            status.pending_reasons,
+            vec!["connection_closure_unverified"]
+        );
+        drop((compliant, compliant_audio, free));
+        server.abort();
+        let _ = server.await;
+    }
+
+    async fn policy_close(client: &mut Client) {
+        loop {
+            if let Message::Close(Some(close)) = frame(client, "policy close").await {
+                assert_eq!(u16::from(close.code), 1008);
+                assert_eq!(close.reason, "community authentication upgraded");
+                let _ = client.flush().await;
+                break;
+            }
+        }
+    }
+}

crates/buzz-relay/src/state.rs

@@ -45,6 +45,7 @@ pub(crate) type ScopedPubkeyKey = (CommunityId, [u8; 32]);
 pub(crate) enum CommunityDisconnectReason {
     CommunityArchived,
     CommunityDeleted,
+    AuthPolicyUpgraded,
     /// NIP-FI: the connection's proven pubkey was added to the deny set.
     AuthorizationDenied,
     /// The authenticated pubkey lost access to the community (e.g. a ban).
@@ -65,6 +66,10 @@ impl CommunityDisconnectReason {
                 code: axum::extract::ws::close_code::POLICY,
                 reason: WsUtf8Bytes::from_static("community deleted"),
             })),
+            Self::AuthPolicyUpgraded => WsMessage::Close(Some(axum::extract::ws::CloseFrame {
+                code: axum::extract::ws::close_code::POLICY,
+                reason: WsUtf8Bytes::from_static("community authentication upgraded"),
+            })),
             Self::AuthorizationDenied => WsMessage::Close(Some(axum::extract::ws::CloseFrame {
                 code: axum::extract::ws::close_code::POLICY,
                 reason: WsUtf8Bytes::from_static("authorization denied"),
@@ -507,6 +512,23 @@ impl CommunityConnectionRegistry {
         closed
     }
 
+    /// Enforced FI admission is attached before registration. Later NIP-42 AUTH
+    /// cannot turn a legacy or observed-shadow socket into an FI admission.
+    fn disconnect_auth_policy(&self, community_id: CommunityId) -> usize {
+        let mut closed = 0;
+        for entry in self.connections.iter() {
+            let (community, control) = entry.value();
+            if *community == community_id
+                && control.nip_fi_enforce().is_none()
+                && !control.cancel.is_cancelled()
+            {
+                control.disconnect_community(CommunityDisconnectReason::AuthPolicyUpgraded);
+                closed += 1;
+            }
+        }
+        closed
+    }
+
     /// Disconnects every registered socket admitted under NIP-FI `issuer` whose
     /// proven pubkey matches `pubkey`, across all communities.  A same-key
     /// socket admitted under a different issuer (or with no assertion) is not
@@ -713,6 +735,8 @@ pub(crate) async fn run_registered_community_connection<
     OnNotRun: FnOnce() -> OnNotRunFuture,
     OnNotRunFuture: Future<Output = ()>,
 {
+    #[cfg(test)]
+    crate::nip_fi_test_hooks::before_community_registration(community_id).await;
     let cancel = control.cancel.clone();
     let _guard = registry.register(connection_id, community_id, control.clone());
 
@@ -2558,21 +2582,66 @@ impl AppState {
             .await
     }
 
-    /// Revalidate all communities with live sockets and cancel inactive ones.
+    /// Revalidate lifecycle and effective FI policy for every locally bound community.
     ///
-    /// 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.
+    /// Reads the writer by immutable ID, independent of host-cache/pubsub delivery.
+    /// Failed reads retain sockets until a later sweep; late registrations are
+    /// included next time. This closes sockets, not already-running work.
     pub async fn revalidate_live_communities(&self) -> usize {
         let (closed, failures) = revalidate_registered_communities(
             &self.community_connections,
             |community_id| async move {
-                // Nearly every live community is active. An unlocked read keeps
-                // the per-tick row lock off them; the fence re-checks under the
-                // lock before closing anything.
-                if self.db.is_community_active(community_id).await? {
+                let mode = self.config.nip_fi.mode;
+                if mode.evaluates() {
+                    let policy = match tokio::time::timeout(
+                        crate::nip_fi_policy::POLICY_READ_BUDGET,
+                        self.db.community_auth_policy(community_id),
+                    )
+                    .await
+                    {
+                        Ok(Ok(policy)) => policy,
+                        Ok(Err(error)) => {
+                            metrics::counter!(
+                                "buzz_community_auth_policy_checks_total",
+                                "outcome" => "error",
+                            ).increment(1);
+                            return Err(error);
+                        }
+                        Err(_) => {
+                            metrics::counter!(
+                                "buzz_community_auth_policy_checks_total",
+                                "outcome" => "timeout",
+                            ).increment(1);
+                            tracing::warn!(%community_id, "community policy read timed out; retaining sockets until next tick");
+                            return Ok(0);
+                        }
+                    };
+                    let required = policy.as_ref().is_some_and(|policy| {
+                        crate::nip_fi_policy::effective(mode, policy.requires_nip_fi).enforces()
+                    });
+                    let outcome = if required {
+                        "required"
+                    } else if policy.is_some() {
+                        "optional"
+                    } else {
+                        "inactive"
+                    };
+                    metrics::counter!("buzz_community_auth_policy_checks_total", "outcome" => outcome).increment(1);
+                    if policy.is_some() {
+                        let closed = if required {
+                            self.community_connections.disconnect_auth_policy(community_id)
+                        } else {
+                            0
+                        };
+                        // Counts cancellation requests, not transport retirement or fleet completion.
+                        metrics::counter!("buzz_community_auth_policy_disconnects_total").increment(closed as u64);
+                        return Ok(closed);
+                    }
+                } else if self.db.is_community_active(community_id).await? {
+                    // Off and repair mode retain their existing lifecycle-only behavior.
                     return Ok(0);
                 }
+                // Recheck inactive rows under the existing archive/deletion fence.
                 self.disconnect_if_inactive(community_id)
                     .await
                     .map(|disconnected| disconnected.unwrap_or(0))
@@ -2580,7 +2649,7 @@ impl AppState {
         )
         .await;
         for (community_id, error) in failures {
-            tracing::warn!(%community_id, %error, "community lifecycle revalidation failed; retaining its sockets until next tick");
+            tracing::warn!(%community_id, %error, "community lifecycle or auth policy revalidation failed; retaining its sockets until next tick");
         }
         closed
     }

docs/community-auth-upgrades.md

@@ -1,10 +1,10 @@
 # Community NIP-FI upgrade operator contract
 
-The acceptance layer records durable policy. The per-request enforcement layer
-adds community checks and discovery; the activation layer updates existing sessions.
-Do not expose upgrade calls in a deployment until those releases and the serving
-fleet baseline have been verified. This change alone cannot establish enforcement.
-It never reports `active` and never substitutes a timer for activation evidence.
+The acceptance layer records durable policy. Admission enforces the effective
+requirement, and the periodic community revalidator eventually closes existing
+legacy root and audio WebSockets. Deploy and verify these behaviors on every
+serving relay before enabling upgrade calls. The receipt proves acceptance; it
+never reports `active` or certifies that all old work has finished.
 
 The operator HTTP plane is separate from tenant event ingest. These management
 operations extend the existing provisioning API; community clients cannot invoke
@@ -19,8 +19,8 @@ Responses use `Cache-Control: no-store`.
 acceptances return 503 `upgrades_disabled` with no policy mutation or
 receipt. Authorized GET and identical POST receipt recovery remain available;
 principal checks and payload conflicts are unchanged. Enabling this deployment
-switch is an explicit operator action after the enforcement and activation fleet
-baseline is verified; caller release discipline alone cannot enable acceptance.
+switch is an explicit operator action after admission and eventual socket closure
+are verified on every serving relay; caller release discipline alone cannot enable acceptance.
 
 ## Generation storage
 
@@ -29,8 +29,8 @@ nonnegative database CHECK. Supported writers retain the current generation
 when the community already requires FI, or increment it by one on acceptance.
 Starting from zero, those writes preserve the nonnegative invariant. Direct SQL
 can insert a negative community generation and is outside that writer contract.
-Positive-generation CHECKs on acceptance receipts remain in place; activation
-and receipt matching still require the exact accepted generation.
+Positive-generation CHECKs on acceptance receipts remain in place; receipt
+matching still requires the exact accepted generation.
 
 The community CHECK is omitted to avoid validating existing rows while adding
 the column. The current schema planner folds a column and its CHECK together,
@@ -96,12 +96,12 @@ Enable and readback return HTTP 200 for an accepted receipt:
   "policy_generation": 1,
   "requires_nip_fi": true,
   "state": "pending",
-  "pending_reasons": ["registry_baseline_unverified"]
+  "pending_reasons": ["connection_closure_unverified"]
 }
 ```
 
 `requires_nip_fi: true` records durable acceptance. `state` reports availability
-and activation separately. The acceptance layer returns `pending` while the target's policy,
+and connection closure separately. The acceptance layer returns `pending` while the target's policy,
 lifecycle and issuer binding still match, and `unavailable` if they do not or
 trusted issuer configuration is no longer applicable. A missing database response
 is HTTP 503, never a fabricated receipt. A receipt is not proof of current service
@@ -110,10 +110,11 @@ health. No response in this version reports `active`.
 `pending_reasons` applies only while `state` is `pending`; clients must ignore
 those obligations when state is `unavailable`, which always takes precedence.
 `pending_reasons` is extensible. Clients must preserve pending on unknown values.
-`registry_baseline_unverified` prevents completion until the activation layer verifies fleet
-membership. That layer may add `boot_application_pending`, `boot_drain_pending`
-and `boot_unreachable`, and must bind activation evidence to the exact operation
-and generation. Baseline invalidation must prevent `active`.
+`connection_closure_unverified` means that this version does not certify socket
+closure. It remains present after successful sweeps: neither a timer nor local
+socket counts prove fleet-wide completion. Clients must treat `pending` as
+accepted policy whose connection closure is unverified, rather than wait for an
+activation ACK that this version does not produce.
 
 The UUID binds the normalized community/host/actor/owner/issuer/operator tuple.
 Identical retries recover that tuple and generation before rechecking mutable
@@ -257,8 +258,8 @@ community row, preserving the deletion-shared → row → host-lock ordering.
 Policy resolutions that start after acceptance commits therefore cannot use an
 optional lease for that community, even on a pod that missed invalidation.
 Admissions that resolved policy before commit can still be in flight, exactly
-as with primary reads; their completion and cancellation belong to BUZZ-512's
-generation-registration and drain fence. Cache expiry is not activation evidence.
+as with primary reads; existing sockets are caught by a later community sweep, and already-running
+requests may finish. Cache expiry does not certify socket closure.
 The acceptance drain blocks new policy fills and other acceptances for the
 host. Its community row lock also blocks archive/unarchive, deletion admission
 and fences, ownership transfer/convergence and issuer-binding writers for that
@@ -318,9 +319,47 @@ provider address is added to discovery. Existing clients can keep reading their
 current fields; new clients must use the requirement field to distinguish optional
 capability from mandatory evidence.
 
-This is admission enforcement only. Existing requests and sockets may predate the
-new policy. Only BUZZ-512's application and drain evidence may establish completed
-activation; the operator receipt remains pending until that evidence is recorded.
+### Existing connections
+
+The existing community revalidator also reads durable authentication policy from
+the writer by immutable community ID for communities with local sockets. Each
+policy read has a 500 ms budget. A failed read retains that community's sockets,
+continues the sweep, and retries on a later pass. No Redis notification or client
+request is needed. An upgrade callback that registers after a scan is caught on
+the next pass.
+
+When effective mode requires FI, legacy and optional-shadow sockets receive
+Close 1008 (`community authentication upgraded`) through the existing transport
+cleanup paths. Only the enforced FI admission marker preserves a socket; later
+NIP-42 AUTH does not turn a legacy socket into an FI admission. Valid FI sessions
+retain their normal expiry and revocation rules. Other communities stay open.
+Off and repair mode retain their existing behavior; Off suspends FI enforcement.
+
+`BUZZ_COMMUNITY_REVALIDATE_INTERVAL_SECS` defaults to 30 seconds, clamped to 1–300 seconds.
+This interval is not a deadline through database failure, sweep duration,
+overload or a paused relay. Closure requires a successful sweep on each running
+compatible relay. Already-running HTTP requests and descendant work may finish.
+There is no boot inventory, drain ACK, or durable fleet-completion certification.
+The operator receipt remains `pending` even after successful local closure.
+
+For each serving relay, observe
+`buzz_community_auth_policy_checks_total{outcome="required"}` advancing after a
+required policy change, with no sustained `error` or `timeout` outcomes.
+Other bounded outcomes are `optional` and `inactive`; no community labels are
+emitted. `buzz_community_auth_policy_disconnects_total` counts legacy socket
+cancellation requests. Verify actual Close 1008 on idle root/audio probes and
+successful FI reconnects on that relay; the counters alone do not certify
+transport retirement or fleet completion. A relay without local sockets has no
+policy checks to report. Account for that when probing each serving relay.
+
+There is no new feature flag. The first successful sweep after deployment can
+close existing legacy sockets in already-required communities. Count required rows and assess concurrent
+reconnect load before rollout; this change does not spread those reconnects.
+Keep new upgrades disabled throughout mixed-version deployment. A binary rollback
+needs no migration and restores the old receipt reason; clients must tolerate
+unknown pending reasons. An older binary cannot provide this eventual-closure
+behavior, so retain a compatible binary or contain affected communities instead
+of assuming rollback undoes accepted policy.
 
 ### Staging and containment
 
@@ -330,7 +369,7 @@ return a generic document; it now returns 404. Policy failures return 503.
 Database-source startup loads its FI binding snapshot only in evaluating modes;
 Off does not load it. A failed load still fails startup in Shadow or Enforce. The local dependency gauge
 `buzz_nip_fi_required_admission_ready` must be 1 before accepting any upgrade;
-it covers verifier, deny map and finite lifetime, not fleet baseline verification.
+it covers verifier, deny map and finite lifetime, not verification of socket closure.
 Binding and JWKS readiness must also be verified through a signed request.
 
 Keep `BUZZ_NIP_FI_UPGRADES_ENABLED=false` throughout a mixed-version rollout
@@ -340,7 +379,7 @@ writer cannot safely coexist with cached optional leases. Before rolling an
 acceptance writer back or removing this protocol, disable new acceptance on
 every writer and verify that disablement; keep the lease-aware writers until
 in-flight acceptance has finished. Disabling new calls does not undo existing
-requirements. Existing required-community enforcement and BUZZ-512 activation
+requirements. Existing required-community admission, eventual socket closure
 and revocation prerequisites still apply. Never enable an older acceptance path
 while any caching pod serves optional communities.
 Older pods may log an unknown `CommunityHost` cache-invalidation message during
@@ -369,9 +408,9 @@ dependency-failure 503 with no-store explicitly.
 Then test a disposable required community with complete FI dependencies, in both
 Shadow and Enforce: unsigned/missing evidence denies, signed valid FI succeeds,
 root/audio upgrades agree, discovery agrees, and a pending deletion request does
-not weaken the requirement. Before production acceptance, complete BUZZ-512's
-session application/drain and the production baseline-verifier gate. No such
-production verification is established by this PR.
+not weaken the requirement. Before production acceptance, verify idle legacy root/audio closure after a
+durable policy change, FI reconnects and isolation on every serving relay. No
+production verification is established by local tests.
 
 Before any required row exists, an operator may stop the rollout and restore the
 previous binary after verifying the primary contains zero required rows. Once a
@@ -387,9 +426,9 @@ while new acceptances are disabled.
 
 Required sessions admitted under global Shadow are real enforced FI sessions.
 The existing global-Shadow admin disconnect path still only observes would-close;
-BUZZ-512 must close these enforced sessions independently of global rollout mode,
+A separate revocation change must close these enforced sessions independently of global rollout mode,
 while preserving the separate optional-shadow registry. Before enabling upgrade
-acceptance for any serving community in Shadow, deploy and verify BUZZ-512's local
+acceptance for any serving community in Shadow, deploy and verify that local
 and cross-pod disconnect behavior, including real Redis propagation. Keep upgrade
 acceptance disabled until that prerequisite is met; admission tests alone do not
-establish it. Production also requires the baseline-verifier gate above.
+establish it. This socket sweep does not change that independent revocation prerequisite.