Add durable live member authorization checks

Signed-off-by: seiler <seiler@squareup.com>
Co-authored-by: Codex <noreply@openai.com>
(cherry picked from commit 92f13229ed3038d63f2a2d9c889bfd4c212e38e2)
Signed-off-by: Brad Seiler <seiler@squareup.com>

ARCHITECTURE.md

@@ -259,6 +259,83 @@ A `CancellationToken` coordinates shutdown across all three loops.
 
 Slow clients: `ConnectionState::send()` uses `try_send` — if the send buffer is full, a grace counter increments. After `SLOW_CLIENT_GRACE_LIMIT` (3) consecutive full-buffer events, the connection is cancelled. A successful send resets the counter.
 
+### Live Community Authorization Revalidation
+
+Authenticated root and huddle/audio sockets retain their server-resolved
+community, principal, admission-captured owner, admission route, whether that
+owner was a relay member at admission, and whether closed-relay membership was
+granted through that owner. A relay-local worker checks
+these records against batched writer-authoritative snapshots every 5 seconds. It
+keeps the immediate local disconnect and Redis fan-out paths; this scan is the
+durable backstop when a pod misses fan-out.
+
+The maximum authorization-staleness budget is 30 seconds. A database failure
+may preserve the last successful decision for 10 seconds, measured from the
+start of that session's last successful authoritative check; repeated failures
+never extend the grace. The worker handles up to 500 distinct
+`(community, principal, captured owner)`
+targets per query, runs at most 4 queries at a time, gives each query 1 second,
+and starts one shared 5-second lookup deadline before it snapshots and groups
+the live-session registry. No database batch starts after that deadline;
+unresolved targets are treated as lookup failures. Closing decisions are then applied in one
+bounded sweep over the connection snapshot. At the default 10,000-connection
+limit, the in-memory regression harness processed a full 10,000-session scan in
+70 ms in the latest measured workstation run. A disposable PostgreSQL run
+fetched all 10,000 authorization targets in 129 ms through the same 500-row,
+four-query batching shape. Database lookups are bounded by the shared 5-second
+deadline. The conservative staleness bound is 10 seconds of database grace, one
+5-second cadence, two 5-second scan windows, and one 1-second query deadline:
+26 seconds, below the 30-second budget. The relay's shared
+`BUZZ_MAX_CONNECTIONS` semaphore caps root and audio sockets together at
+10,000 by default; that is at most 20 batches and five query waves. The worker
+creates no per-socket timers or per-connection revalidation queries. A scan
+that exceeds its deadline treats unreturned targets as lookup failures and
+closes their sessions when their existing grace has expired. If a result batch
+is incomplete or malformed, every target in
+that batch is treated as a lookup failure; no partial rows refresh
+authorization.
+
+Membership is required only when `require_relay_membership` is enabled. An
+open-relay nonmember remains allowed unless directly or through its current
+recorded owner it has an active community ban. An owner-derived member remains
+allowed only while the relay, owner-delegation setting, and current owner
+membership still permit that path. Removal of the admission-captured owner
+revokes its session even if the agent's stored owner changed; captured-owner
+membership never grants access to a directly admitted principal. The first
+stored owner link also invalidates sessions admitted without one, matching the
+existing reconnect path when NIP-OA materializes ownership; the durable check
+detects that transition if its disconnect fan-out is missed. A directly
+admitted member with a stable owner link still relies on its own membership,
+and a change between already-recorded owners alone does not create a new
+membership requirement for that principal. An active ban on the captured owner
+also revokes that session. On open relays, an owner who was already a nonmember
+at admission does not become a membership requirement. Moderation timeouts are
+not read by this worker because they restrict writes, not an established
+socket. All membership, ownership, and ban predicates include the bound
+community. Admission reads the captured owner's membership from the writer
+when the delegation check did not already establish it; that bounded point
+lookup records the baseline needed to distinguish a later membership removal
+from pre-existing open-relay nonmembership.
+
+Each batch snapshot is the revalidation ordering point. A removal or ban
+committed before that snapshot closes the session; a re-add or unban committed
+before it keeps the session authorized. A later concurrent change is observed
+on the next scan, and a close based on an earlier snapshot can be recovered by
+reconnecting against current state. Admission binds the authenticated identity
+before its final writer read and only publishes an admitted live record after
+that check, so admission racing a removal is either cancelled by the immediate
+path or rejected by the final read / next durable scan.
+
+If the writer remains unavailable past the grace, the relay fails closed and
+cancels the socket. NIP-FI root and audio sockets receive their existing
+route-specific `authorization_denied` or `authorization_unavailable` terminal
+response; other sockets receive the normal access-revoked or unavailable close
+reason. This trades connection availability during a prolonged database
+outage for a bounded authorization window. Reconnecting after the database
+recovers performs the ordinary admission checks. The worker shares graceful
+shutdown cancellation and has no migration or rollout flag; all relay pods
+must run this code for the bound to hold deployment-wide.
+
 ### Step 5: Cleanup
 
 On disconnect (any cause):

crates/buzz-db/src/lib.rs

@@ -66,7 +66,7 @@ pub(crate) use runtime::{
 };
 pub use store::{
     admin_moderation, allowlist, api_token, archived_identities, artifact, channel,
-    channel_members, community, deletion, dm, entity_deletion, event, feed, git_repo, moderation,
+    channel_members, community, deletion, dm, entity_deletion, event, feed, git_repo, live_authorization, moderation,
     nip_fi_binding, operator_listener, partition, personal_read, product_feedback, push, reaction,
     read_state, relay_admin_actions, relay_invite, relay_members, relay_operators, reminder,
     replaceable, storage_accounting, thread, thread_window, usage, user, user_deletion,
@@ -83,6 +83,7 @@ pub use community::{
 pub use error::{DbError, Result};
 pub use event::{ChannelHeadPrecondition, ChannelHeadWriteStatus};
 pub use event::{EventQuery, DEFAULT_MAX_PAGE_LIMIT};
+pub use live_authorization::{LiveAuthorizationState, LiveAuthorizationTarget};
 pub use nip_fi_binding::{NipFiBindingRecord, NipFiUnbind, RowCapExceeded};
 pub use reaction::ReactionEventInsertOutcome;
 pub use reminder::DueReminder;

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

@@ -0,0 +1,406 @@
+//! Bounded, tenant-scoped authorization snapshots for already-open sessions.
+
+use buzz_core::CommunityId;
+use buzz_datastore_tracing::datastore_span;
+use sqlx::Row as _;
+use uuid::Uuid;
+
+use crate::{observability, Db, DbError, Result};
+
+/// One admitted principal and captured owner whose live authorization should be revalidated.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct LiveAuthorizationTarget {
+    /// The community resolved for the connection.
+    pub community_id: CommunityId,
+    /// The NIP-42-authenticated principal.
+    pub pubkey: [u8; 32],
+    /// The owner captured when this socket was admitted, if any.
+    pub session_owner_pubkey: Option<[u8; 32]>,
+}
+
+/// Authoritative access state for one live principal/owner pair in one community.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct LiveAuthorizationState {
+    /// The community used to scope every part of this lookup.
+    pub community_id: CommunityId,
+    /// The NIP-42-authenticated principal.
+    pub pubkey: [u8; 32],
+    /// The owner captured when this socket was admitted, if any.
+    pub session_owner_pubkey: Option<[u8; 32]>,
+    /// Whether that captured owner currently belongs to the community roster.
+    pub session_owner_is_relay_member: bool,
+    /// The principal's currently recorded agent owner, if any.
+    pub agent_owner_pubkey: Option<[u8; 32]>,
+    /// Whether the principal is currently in the community relay roster.
+    pub relay_member: bool,
+    /// Whether its current owner is currently in the community relay roster.
+    pub owner_is_relay_member: bool,
+    /// Whether the principal, captured owner, or current owner has an active community ban.
+    pub banned: bool,
+}
+
+/// Return one writer-authoritative access snapshot for each requested target.
+///
+/// Membership, current agent ownership, and active bans are read by one SQL
+/// statement so a relay can revalidate many live sessions without a query per
+/// connection. Every input is an explicit `(community_id, pubkey, captured
+/// owner)` tuple; the lookup never expands an owner or ban across tenant
+/// boundaries. Timeouts are intentionally omitted because they restrict writes
+/// rather than live access.
+async fn live_authorization_state_batch(
+    pool: &sqlx::PgPool,
+    targets: &[LiveAuthorizationTarget],
+) -> Result<Vec<LiveAuthorizationState>> {
+    if targets.is_empty() {
+        return Ok(Vec::new());
+    }
+
+    let community_ids = targets
+        .iter()
+        .map(|target| *target.community_id.as_uuid())
+        .collect::<Vec<_>>();
+    let pubkeys = targets
+        .iter()
+        .map(|target| target.pubkey.to_vec())
+        .collect::<Vec<_>>();
+    let session_owner_pubkeys = targets
+        .iter()
+        .map(|target| target.session_owner_pubkey.map(|owner| owner.to_vec()))
+        .collect::<Vec<_>>();
+    let mut connection =
+        observability::acquire_writer(pool, observability::WriterOperation::Authorization).await?;
+    let rows = sqlx::query(
+        r#"
+        WITH requested AS (
+            SELECT DISTINCT community_id, pubkey, session_owner_pubkey
+            FROM unnest($1::uuid[], $2::bytea[], $3::bytea[])
+                 AS target(community_id, pubkey, session_owner_pubkey)
+        )
+        SELECT requested.community_id,
+               requested.pubkey,
+               requested.session_owner_pubkey,
+               EXISTS (
+                   SELECT 1
+                   FROM relay_members
+                   WHERE relay_members.community_id = requested.community_id
+                     AND relay_members.pubkey = encode(requested.session_owner_pubkey, 'hex')
+               ) AS session_owner_is_relay_member,
+               users.agent_owner_pubkey,
+               EXISTS (
+                   SELECT 1
+                   FROM relay_members
+                   WHERE relay_members.community_id = requested.community_id
+                     AND relay_members.pubkey = encode(requested.pubkey, 'hex')
+               ) AS relay_member,
+               EXISTS (
+                   SELECT 1
+                   FROM relay_members
+                   WHERE relay_members.community_id = requested.community_id
+                     AND relay_members.pubkey = encode(users.agent_owner_pubkey, 'hex')
+               ) AS owner_is_relay_member,
+               EXISTS (
+                   SELECT 1
+                   FROM community_bans
+                   WHERE community_bans.community_id = requested.community_id
+                     AND community_bans.pubkey IN
+                         (requested.pubkey, requested.session_owner_pubkey, users.agent_owner_pubkey)
+                     AND community_bans.banned
+                     AND (community_bans.ban_expires_at IS NULL
+                          OR community_bans.ban_expires_at > now())
+               ) AS banned
+        FROM requested
+        LEFT JOIN users
+          ON users.community_id = requested.community_id
+         AND users.pubkey = requested.pubkey
+        "#,
+    )
+    .bind(community_ids)
+    .bind(pubkeys)
+    .bind(session_owner_pubkeys)
+    .fetch_all(&mut *connection)
+    .await?;
+
+    rows.into_iter()
+        .map(|row| {
+            let community_id: Uuid = row.try_get("community_id")?;
+            let pubkey = decode_pubkey(row.try_get("pubkey")?, "requested pubkey")?;
+            let session_owner_pubkey = row
+                .try_get::<Option<Vec<u8>>, _>("session_owner_pubkey")?
+                .map(|owner| decode_pubkey(owner, "session owner pubkey"))
+                .transpose()?;
+            let agent_owner_pubkey = row
+                .try_get::<Option<Vec<u8>>, _>("agent_owner_pubkey")?
+                .map(|owner| decode_pubkey(owner, "agent owner pubkey"))
+                .transpose()?;
+
+            Ok(LiveAuthorizationState {
+                community_id: CommunityId::from_uuid(community_id),
+                pubkey,
+                session_owner_pubkey,
+                session_owner_is_relay_member: row.try_get("session_owner_is_relay_member")?,
+                agent_owner_pubkey,
+                relay_member: row.try_get("relay_member")?,
+                owner_is_relay_member: row.try_get("owner_is_relay_member")?,
+                banned: row.try_get("banned")?,
+            })
+        })
+        .collect()
+}
+
+fn decode_pubkey(bytes: Vec<u8>, field: &'static str) -> Result<[u8; 32]> {
+    bytes
+        .try_into()
+        .map_err(|_| DbError::InvalidData(format!("{field} must contain exactly 32 bytes")))
+}
+
+impl Db {
+    /// Read current membership, agent ownership, and ban state for a bounded
+    /// batch of live connections from the authoritative writer.
+    ///
+    /// The caller is responsible for bounding the batch size and total work.
+    /// Each result is scoped to the exact community/principal/captured-owner tuple supplied.
+    #[datastore_span(name = "live_authorization_state_batch", system = "postgresql")]
+    pub async fn live_authorization_state_batch(
+        &self,
+        targets: &[LiveAuthorizationTarget],
+    ) -> Result<Vec<LiveAuthorizationState>> {
+        live_authorization_state_batch(&self.pool, targets).await
+    }
+}
+
+#[cfg(test)]
+mod postgres_tests {
+    use super::*;
+    use chrono::{Duration, Utc};
+    use nostr::Keys;
+    use sqlx::PgPool;
+
+    async fn setup_db() -> Db {
+        let pool = PgPool::connect(&crate::test_support::database_url())
+            .await
+            .expect("connect to disposable test database");
+        Db::from_pool(pool)
+    }
+
+    async fn make_community(db: &Db) -> CommunityId {
+        let id = Uuid::new_v4();
+        let host = format!("live-auth-test-{}.example", id.simple());
+        sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+            .bind(id)
+            .bind(host)
+            .execute(&db.pool)
+            .await
+            .expect("insert test community");
+        CommunityId::from_uuid(id)
+    }
+
+    fn pubkey() -> [u8; 32] {
+        Keys::generate().public_key().to_bytes()
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn batch_is_tenant_scoped_reads_owner_bans_ignores_timeouts_and_scales_to_10k() {
+        let db = setup_db().await;
+        let community_a = make_community(&db).await;
+        let community_b = make_community(&db).await;
+        let owner = pubkey();
+        let current_owner = pubkey();
+        let agent = pubkey();
+        let late_link_agent = pubkey();
+        let direct = pubkey();
+        let timed_out = pubkey();
+        let actor = pubkey();
+
+        for key in [
+            owner,
+            current_owner,
+            agent,
+            late_link_agent,
+            direct,
+            timed_out,
+            actor,
+        ] {
+            db.ensure_user_for_authorization(community_a, &key)
+                .await
+                .expect("create community A user");
+        }
+        db.ensure_user_for_authorization(community_b, &direct)
+            .await
+            .expect("create community B user");
+        db.ensure_user_for_authorization(community_b, &late_link_agent)
+            .await
+            .expect("create community B agent");
+        db.ensure_user_for_authorization(community_b, &current_owner)
+            .await
+            .expect("create community B owner");
+        db.set_agent_owner_for_authorization(community_a, &agent, &current_owner)
+            .await
+            .expect("record agent owner");
+        db.add_relay_member(community_a, &hex::encode(owner), "member", None)
+            .await
+            .expect("add owner to community A");
+        db.add_relay_member(community_a, &hex::encode(current_owner), "member", None)
+            .await
+            .expect("add current owner to community A");
+        db.add_relay_member(community_b, &hex::encode(direct), "member", None)
+            .await
+            .expect("add principal to community B");
+        db.add_relay_member(community_b, &hex::encode(current_owner), "member", None)
+            .await
+            .expect("add current owner to community B");
+        // Capture the session as ownerless, then commit its first owner link.
+        // This is the supported None -> Some ownership transition.
+        db.set_agent_owner_for_authorization(community_a, &late_link_agent, &current_owner)
+            .await
+            .expect("record late owner link in community A");
+        db.set_agent_owner_for_authorization(community_b, &late_link_agent, &current_owner)
+            .await
+            .expect("record same owner link in community B");
+        db.ban_community_member(community_a, &owner, &actor, None, None)
+            .await
+            .expect("ban agent owner in community A");
+        db.ban_community_member(community_a, &current_owner, &actor, None, None)
+            .await
+            .expect("ban current agent owner in community A");
+        db.timeout_community_member(
+            community_a,
+            &timed_out,
+            &actor,
+            Utc::now() + Duration::minutes(5),
+            None,
+        )
+        .await
+        .expect("time out principal in community A");
+
+        let states = db
+            .live_authorization_state_batch(&[
+                LiveAuthorizationTarget {
+                    community_id: community_a,
+                    pubkey: agent,
+                    session_owner_pubkey: Some(owner),
+                },
+                LiveAuthorizationTarget {
+                    community_id: community_a,
+                    pubkey: late_link_agent,
+                    session_owner_pubkey: None,
+                },
+                LiveAuthorizationTarget {
+                    community_id: community_b,
+                    pubkey: late_link_agent,
+                    session_owner_pubkey: None,
+                },
+                LiveAuthorizationTarget {
+                    community_id: community_b,
+                    pubkey: agent,
+                    session_owner_pubkey: Some(owner),
+                },
+                LiveAuthorizationTarget {
+                    community_id: community_a,
+                    pubkey: direct,
+                    session_owner_pubkey: None,
+                },
+                LiveAuthorizationTarget {
+                    community_id: community_b,
+                    pubkey: direct,
+                    session_owner_pubkey: None,
+                },
+                LiveAuthorizationTarget {
+                    community_id: community_a,
+                    pubkey: timed_out,
+                    session_owner_pubkey: None,
+                },
+            ])
+            .await
+            .expect("batch live authorization state");
+        let state = |community_id, pubkey| {
+            states
+                .iter()
+                .find(|state| state.community_id == community_id && state.pubkey == pubkey)
+                .expect("state for requested tenant/principal")
+        };
+
+        let agent_state = state(community_a, agent);
+        assert_eq!(agent_state.agent_owner_pubkey, Some(current_owner));
+        assert_eq!(agent_state.session_owner_pubkey, Some(owner));
+        assert!(agent_state.session_owner_is_relay_member);
+        assert!(!agent_state.relay_member);
+        assert!(
+            agent_state.owner_is_relay_member,
+            "active bans are reported independently of roster membership"
+        );
+        assert!(
+            agent_state.banned,
+            "captured owner ban applies after owner change"
+        );
+        let late_link_state = state(community_a, late_link_agent);
+        assert_eq!(
+            late_link_state.agent_owner_pubkey,
+            Some(current_owner),
+            "the lookup reads a first owner link added after admission"
+        );
+        assert!(
+            late_link_state.banned,
+            "a ban on a newly recorded current owner is returned for an ownerless session"
+        );
+        assert!(
+            !state(community_b, late_link_agent).banned,
+            "a ban on the current owner does not cross communities"
+        );
+        assert!(
+            !state(community_b, agent).banned,
+            "a ban on the same captured owner does not cross communities"
+        );
+        assert!(
+            !state(community_b, agent).session_owner_is_relay_member,
+            "captured-owner membership is community scoped"
+        );
+
+        assert!(!state(community_a, direct).relay_member);
+        assert!(state(community_b, direct).relay_member);
+        let timeout_state = state(community_a, timed_out);
+        assert!(!timeout_state.banned, "timeouts do not revoke live access");
+
+        // Exercise the relay's actual 500-principal batches and four-query
+        // concurrency at the default 10,000-socket limit. This is a disposable
+        // local fixture; no user or moderation rows are needed for absent keys.
+        const SUPPORTED_PRINCIPALS: usize = 10_000;
+        const BATCH_SIZE: usize = 500;
+        const QUERY_CONCURRENCY: usize = 4;
+        let targets = (0..SUPPORTED_PRINCIPALS)
+            .map(|index| {
+                let mut pubkey = [0; 32];
+                pubkey[..8].copy_from_slice(&(index as u64).to_be_bytes());
+                LiveAuthorizationTarget {
+                    community_id: community_a,
+                    pubkey,
+                    session_owner_pubkey: None,
+                }
+            })
+            .collect::<Vec<_>>();
+        let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(QUERY_CONCURRENCY));
+        let mut queries = tokio::task::JoinSet::new();
+        let started_at = std::time::Instant::now();
+        for batch in targets.chunks(BATCH_SIZE).map(|batch| batch.to_vec()) {
+            let db = db.clone();
+            let semaphore = std::sync::Arc::clone(&semaphore);
+            queries.spawn(async move {
+                let _permit = semaphore.acquire_owned().await.expect("query permit");
+                db.live_authorization_state_batch(&batch)
+                    .await
+                    .map(|states| states.len())
+            });
+        }
+        let mut rows = 0;
+        while let Some(result) = queries.join_next().await {
+            rows += result.expect("batch task").expect("batch query");
+        }
+        let elapsed = started_at.elapsed();
+        assert_eq!(rows, SUPPORTED_PRINCIPALS);
+        eprintln!(
+            "10k-principal PostgreSQL scan: {rows} rows, batches of {BATCH_SIZE}, \
+             concurrency {QUERY_CONCURRENCY}, {} ms",
+            elapsed.as_millis(),
+        );
+    }
+}

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

@@ -38,6 +38,8 @@ mod event_follow_up_postgres_tests;
 pub mod feed;
 /// Git repository name registry (NIP-34 kind:30617).
 pub mod git_repo;
+/// Batched authoritative access-state reads for live connections.
+pub mod live_authorization;
 /// Community moderation: reports, bans/timeouts, audit actions.
 pub mod moderation;
 /// Deployment-global operator-listener mention registrations and delivery queues.

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

@@ -997,7 +997,7 @@ async fn run_active_audio_connection(
     }
 
     // Bind, then take the final ban/membership decision the root socket
-    // applies (see `final_admission_denial`), before any huddle lease. A ban
+    // applies (see `final_admission_check`), before any huddle lease. A ban
     // or removal whose disconnect ran before the bind is seen by these fresh
     // reads; one that runs after it cancels this socket (`check_cancel!`).
     // Same order as root AUTH: a proven owner before the pubkey; otherwise
@@ -1007,27 +1007,40 @@ async fn run_active_audio_connection(
         control.bind_owner(owner.to_bytes());
     }
     control.bind_pubkey(pubkey.to_bytes());
-    let owner =
+    let admitted_owner =
         crate::handlers::auth::admitted_owner(&state, tenant.community(), pubkey, nip_oa_owner)
             .await;
-    if let Ok(Some(owner)) = owner {
-        control.bind_owner(owner);
-    }
-    let denial = match owner {
-        Err(denial) => Some(denial),
-        Ok(_) => {
-            crate::handlers::auth::final_admission_denial(
-                &state,
-                tenant.community(),
-                pubkey,
-                auth_tag_json.as_deref(),
-                Some(signed_auth_created_at),
-                token_owner,
+    let admitted_owner = match admitted_owner {
+        Ok(owner) => owner,
+        Err(denial) => {
+            shadow_attempt.refused();
+            exit_authorization_refusal(
+                &mut ws_send,
+                &control,
+                &mut terminal_ctrl_rx,
+                nip_fi_assertion.is_some(),
+                denial.class,
+                serde_json::json!({"type": "error", "message": denial.reason}),
             )
-            .await
+            .await;
+            return;
         }
     };
-    if let Some(denial) = denial {
+    if let Some(owner) = admitted_owner {
+        control.bind_owner(owner);
+    }
+    let final_check_started_at = tokio::time::Instant::now();
+    let final_check = crate::handlers::auth::final_admission_check(
+        &state,
+        tenant.community(),
+        pubkey,
+        auth_tag_json.as_deref(),
+        Some(signed_auth_created_at),
+        token_owner,
+        admitted_owner,
+    )
+    .await;
+    if let Some(denial) = final_check.denial {
         shadow_attempt.refused();
         let (class, message) = (denial.class, denial.reason);
         warn!(channel_id = %channel_id, pubkey = %pubkey_hex, reason = message, "audio: denied at final admission check");
@@ -1043,6 +1056,14 @@ async fn run_active_audio_connection(
         return;
     }
     check_cancel!(cancel, terminal_ctrl_rx, disconnect_reason, ws_send);
+    control.mark_live_authorized(
+        pubkey.to_bytes(),
+        admitted_owner,
+        final_check.owner_was_relay_member_at_admission,
+        final_check.membership_via_owner,
+        crate::nip_fi_session::NipFiWsRoute::Audio,
+        final_check_started_at,
+    );
 
     // Huddle cross-pod routing (mesh) OR single-pod guardrail.
     //
@@ -4188,7 +4209,7 @@ mod tests {
         // Fix 5: use Config::for_test() which holds NIP_FI_ENV_LOCK internally. [FI-TRACE-ENV-RACE]
         let mut config = crate::config::Config::for_test();
         config.require_relay_membership = require_relay_membership;
-        config.database_url = "postgres://buzz:buzz_dev@127.0.0.1:1/buzz".to_string(); // sadscan:disable np.postgres.1 -- unreachable test-only URL
+        config.database_url = "postgres://127.0.0.1:1/buzz".to_string();
         config.redis_url = "redis://127.0.0.1:1".to_string();
         let mut pool_options = sqlx::postgres::PgPoolOptions::new();
         if let Some(timeout) = acquire_timeout {

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

@@ -56,6 +56,38 @@ fn classify_relay_membership(
     }
 }
 
+fn final_admission_from_membership(
+    membership: PolicyCheck<Option<nostr::PublicKey>>,
+) -> FinalAdmissionCheck {
+    match membership {
+        PolicyCheck::Allowed(owner) => FinalAdmissionCheck {
+            denial: None,
+            membership_via_owner: owner.is_some(),
+            owner_was_relay_member_at_admission: false,
+        },
+        PolicyCheck::Denied => FinalAdmissionCheck {
+            denial: Some(AdmissionDenial {
+                metric: "not_relay_member",
+                reason: "restricted: not a relay member",
+                outcome: AuthOutcome::NotRelayMember,
+                class: buzz_auth::DenialClass::AuthorizationDenied,
+            }),
+            membership_via_owner: false,
+            owner_was_relay_member_at_admission: false,
+        },
+        PolicyCheck::DependencyError => FinalAdmissionCheck {
+            denial: Some(AdmissionDenial {
+                metric: "relay_membership_check_error",
+                reason: "error: internal error checking relay membership",
+                outcome: AuthOutcome::RelayMembershipCheckError,
+                class: buzz_auth::DenialClass::AuthorizationUnavailable,
+            }),
+            membership_via_owner: false,
+            owner_was_relay_member_at_admission: false,
+        },
+    }
+}
+
 /// Community-ban verdict for an authenticating principal: shared by every
 /// socket auth seam (root and audio) so they cannot drift.
 ///
@@ -189,6 +221,18 @@ pub(crate) fn ban_admission_denial(ban: BanOutcome) -> Option<AdmissionDenial> {
     })
 }
 
+/// Final admission result, including whether relay membership depended on the
+/// agent's owner. Live revalidation must apply the same policy path that
+/// admitted the socket.
+pub(crate) struct FinalAdmissionCheck {
+    pub(crate) denial: Option<AdmissionDenial>,
+    pub(crate) membership_via_owner: bool,
+    /// Whether the socket's captured owner was a relay member at admission.
+    /// This detects later owner removal without granting direct sessions the
+    /// owner's membership.
+    pub(crate) owner_was_relay_member_at_admission: bool,
+}
+
 /// Final ban and relay-membership verdict for an authenticating socket.
 ///
 /// Callers bind the socket to `pubkey` first (the registry a ban's or
@@ -196,15 +240,16 @@ pub(crate) fn ban_admission_denial(ban: BanOutcome) -> Option<AdmissionDenial> {
 /// runs after the bind and cancels the socket, or it ran before, so its
 /// committed ban or removal is visible to these fresh reads. Checking before
 /// binding leaves a gap where both are missed. Callers must also refuse a
-/// socket whose cancellation token fired. Both reads fail closed.
-pub(crate) async fn final_admission_denial(
+/// socket whose cancellation token fired. Every writer read fails closed.
+pub(crate) async fn final_admission_check(
     state: &AppState,
     community: buzz_core::CommunityId,
     pubkey: nostr::PublicKey,
     auth_tag_json: Option<&str>,
     signed_auth_created_at: Option<u64>,
     token_owner: Option<nostr::PublicKey>,
-) -> Option<AdmissionDenial> {
+    admitted_owner: Option<[u8; 32]>,
+) -> FinalAdmissionCheck {
     let ban = community_ban_outcome(
         state,
         community,
@@ -216,9 +261,13 @@ pub(crate) async fn final_admission_denial(
     )
     .await;
     if let Some(denial) = ban_admission_denial(ban) {
-        return Some(denial);
+        return FinalAdmissionCheck {
+            denial: Some(denial),
+            membership_via_owner: false,
+            owner_was_relay_member_at_admission: false,
+        };
     }
-    match crate::api::relay_members::check_relay_membership_authoritative(
+    let membership = crate::api::relay_members::check_relay_membership_authoritative(
         state,
         community,
         pubkey.as_bytes(),
@@ -226,25 +275,48 @@ pub(crate) async fn final_admission_denial(
         signed_auth_created_at,
         token_owner,
     )
-    .await
-    {
-        Ok(crate::api::relay_members::MembershipDecision::Denied) => Some(AdmissionDenial {
-            metric: "not_relay_member",
-            reason: "restricted: not a relay member",
-            outcome: AuthOutcome::NotRelayMember,
-            class: buzz_auth::DenialClass::AuthorizationDenied,
-        }),
-        Ok(_) => None,
-        Err(e) => {
-            warn!(pubkey = %pubkey.to_hex(), error = %e, "relay membership recheck failed, denying (fail-closed)");
-            Some(AdmissionDenial {
-                metric: "relay_membership_check_error",
-                reason: "error: internal error checking relay membership",
-                outcome: AuthOutcome::RelayMembershipCheckError,
-                class: buzz_auth::DenialClass::AuthorizationUnavailable,
-            })
-        }
+    .await;
+    if let Err(error) = &membership {
+        warn!(pubkey = %pubkey.to_hex(), error, "relay membership recheck failed, denying (fail-closed)");
     }
+    let mut final_check = final_admission_from_membership(classify_relay_membership(membership));
+    if final_check.denial.is_some() {
+        return final_check;
+    }
+
+    final_check.owner_was_relay_member_at_admission =
+        match (admitted_owner, final_check.membership_via_owner) {
+            (None, false) => false,
+            (Some(_), true) => true,
+            (Some(owner), false) => match state
+                .db
+                .is_relay_member_writer(community, &hex::encode(owner))
+                .await
+            {
+                Ok(is_member) => is_member,
+                Err(_) => {
+                    warn!("captured-owner membership snapshot failed at admission; denying");
+                    final_check.denial = Some(AdmissionDenial {
+                        metric: "owner_membership_check_error",
+                        reason: "error: internal error checking relay membership",
+                        outcome: AuthOutcome::RelayMembershipCheckError,
+                        class: buzz_auth::DenialClass::AuthorizationUnavailable,
+                    });
+                    return final_check;
+                }
+            },
+            (None, true) => {
+                warn!("owner-derived membership admission has no captured owner; denying");
+                final_check.denial = Some(AdmissionDenial {
+                    metric: "owner_membership_check_error",
+                    reason: "error: internal error checking relay membership",
+                    outcome: AuthOutcome::RelayMembershipCheckError,
+                    class: buzz_auth::DenialClass::AuthorizationUnavailable,
+                });
+                return final_check;
+            }
+        };
+    final_check
 }
 
 /// Owner to record on an admitted socket: the proven NIP-OA owner, else the
@@ -726,7 +798,7 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, state:
             }
 
             // Bind, then take the final ban/membership decision (see
-            // `final_admission_denial`), then refuse if a disconnect already
+            // `final_admission_check`), then refuse if a disconnect already
             // cancelled this socket. No await separates the last check from
             // `authenticate`.
             // A proven owner is recorded before the pubkey, so this socket's
@@ -750,24 +822,29 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, state:
                     .as_ref()
                     .map(crate::nip_fi_core::NipFiSession::from_assertion),
             );
-            match admitted_owner(&state, conn.tenant.community(), pubkey, nip_oa_owner).await {
-                Ok(Some(owner)) => state.conn_manager.set_admitted_owner(conn_id, owner),
-                Ok(None) => {}
-                Err(denial) => {
-                    deny_admission(&conn, &event_id_hex, denial);
-                    return;
-                }
+            let admitted_owner =
+                match admitted_owner(&state, conn.tenant.community(), pubkey, nip_oa_owner).await {
+                    Ok(owner) => owner,
+                    Err(denial) => {
+                        deny_admission(&conn, &event_id_hex, denial);
+                        return;
+                    }
+                };
+            if let Some(owner) = admitted_owner {
+                state.conn_manager.set_admitted_owner(conn_id, owner);
             }
-            let denial = match final_admission_denial(
+            let final_check_started_at = tokio::time::Instant::now();
+            let final_check = final_admission_check(
                 &state,
                 conn.tenant.community(),
                 pubkey,
                 auth_tag_json.as_deref(),
                 Some(signed_auth_created_at),
                 token_owner,
+                admitted_owner,
             )
-            .await
-            {
+            .await;
+            let denial = match final_check.denial {
                 None if conn.cancel.is_cancelled() => Some(AdmissionDenial {
                     metric: "revoked_during_auth",
                     reason: "blocked: access revoked",
@@ -823,6 +900,14 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, state:
             if conn.cancel.is_cancelled() {
                 return;
             }
+            conn.community_control.mark_live_authorized(
+                pubkey.to_bytes(),
+                admitted_owner,
+                final_check.owner_was_relay_member_at_admission,
+                final_check.membership_via_owner,
+                crate::nip_fi_session::NipFiWsRoute::Root,
+                final_check_started_at,
+            );
             state.conn_manager.mark_admitted(conn_id);
             conn.send(RelayMessage::ok(&event_id_hex, true, ""));
             // _auth_permit drops here — expiry's write guard may proceed.
@@ -850,7 +935,7 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, state:
 mod tests {
     use super::{
         ban_denial, classify_allowlist, classify_relay_membership, extract_auth_tag_json,
-        handle_auth, nip42_denial_class, BanOutcome, PolicyCheck,
+        final_admission_from_membership, handle_auth, nip42_denial_class, BanOutcome, PolicyCheck,
     };
     use crate::api::relay_members::MembershipDecision;
     use crate::connection::{tests::test_conn_with_auth, AuthState};
@@ -966,7 +1051,7 @@ mod tests {
         // Fix 5: use Config::for_test() which holds NIP_FI_ENV_LOCK internally. [FI-TRACE-ENV-RACE]
         let mut config = crate::config::Config::for_test();
         config.require_relay_membership = false;
-        config.database_url = "postgres://buzz:buzz_dev@127.0.0.1:1/buzz".to_string(); // sadscan:disable np.postgres.1
+        config.database_url = "postgres://127.0.0.1:1/buzz".to_string();
         config.redis_url = "redis://127.0.0.1:1".to_string();
         // 100ms acquire timeout: a request that falls through to the stub
         // pool still waits, but for 100ms instead of sqlx's 30s default.
@@ -1966,7 +2051,7 @@ mod tests {
         /// after the bind. A user in another community, paused at the same
         /// point, is still admitted.
         ///
-        /// Mutation: remove the `final_admission_denial` call from
+        /// Mutation: remove the `final_admission_check` call from
         /// `handle_auth` → the banned socket is admitted → RED.
         #[tokio::test]
         #[ignore = "requires Postgres — runs in postgres-ci nextest lane"]
@@ -2063,7 +2148,7 @@ mod tests {
         /// member: the final check reads membership from the writer. A user
         /// in another community, paused at the same point, is still admitted.
         ///
-        /// Mutation: make `final_admission_denial` call the replica-routed
+        /// Mutation: make `final_admission_check` call the replica-routed
         /// `check_relay_membership` → the removed socket is admitted → RED.
         #[tokio::test]
         #[ignore = "requires Postgres — runs in postgres-ci nextest lane"]
@@ -3849,6 +3934,23 @@ mod tests {
             classify_relay_membership(Err("database unavailable".to_owned())),
             PolicyCheck::DependencyError
         );
+
+        let open = final_admission_from_membership(PolicyCheck::Allowed(None));
+        assert!(open.denial.is_none());
+        assert!(!open.membership_via_owner);
+        let via_owner = final_admission_from_membership(PolicyCheck::Allowed(Some(owner)));
+        assert!(via_owner.denial.is_none());
+        assert!(via_owner.membership_via_owner);
+        let denied = final_admission_from_membership(PolicyCheck::Denied);
+        assert_eq!(
+            denied.denial.as_ref().map(|denial| denial.class),
+            Some(buzz_auth::DenialClass::AuthorizationDenied)
+        );
+        let unavailable = final_admission_from_membership(PolicyCheck::DependencyError);
+        assert_eq!(
+            unavailable.denial.as_ref().map(|denial| denial.class),
+            Some(buzz_auth::DenialClass::AuthorizationUnavailable)
+        );
     }
 
     /// The handler owns retry classification, so drive its real terminal-state
@@ -3940,7 +4042,7 @@ mod tests {
         let snapshotter = recorder.snapshotter();
         let _recorder_guard = metrics::set_default_local_recorder(&recorder);
         let state = crate::state::tests::test_state_with_unreachable_database_url(
-            "postgres://buzz:buzz_dev@127.0.0.1:1/buzz",
+            "postgres://127.0.0.1:1/buzz",
         )
         .await;
 

crates/buzz-relay/src/main.rs

@@ -1539,6 +1539,16 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
         ));
     }
 
+    // Durable member-access backstop: unlike lossy Redis disconnect fan-out,
+    // this fixed cadence rechecks only admitted local sockets against the
+    // authoritative writer. Its bounded scan and DB-error grace establish the
+    // live authorization staleness budget independently of archive maintenance.
+    {
+        let access_state = Arc::clone(&state);
+        let cancel = access_state.community_revalidator_cancel.clone();
+        tokio::spawn(run_live_authorization_revalidator(access_state, cancel));
+    }
+
     // Per-pod dependency diagnostics runtime: one seam starts the dependency
     // sampler and its independent completion-epoch republisher together.
     {
@@ -2151,6 +2161,17 @@ async fn run_community_revalidator(
     .await;
 }
 
+async fn run_live_authorization_revalidator(state: Arc<AppState>, cancel: CancellationToken) {
+    run_periodic_until_cancelled(
+        buzz_relay::state::LIVE_AUTHORIZATION_REVALIDATION_INTERVAL,
+        cancel,
+        || async {
+            state.revalidate_live_authorizations().await;
+        },
+    )
+    .await;
+}
+
 async fn run_periodic_until_cancelled<Tick, TickFuture>(
     period: std::time::Duration,
     cancel: CancellationToken,