fix(relay): multi-pod subscription coherence (one access-gated fan-out path + cross-pod cache invalidation + REQ/COUNT DB guard) (#1261)

Signed-off-by: Tyler Longwell <tlongwell@block.xyz>
Signed-off-by: npub1mprnacetjua2xx3p5eddmhxyk6wv929ymm5py8kd2xfxurxahspqqlgyta <d8473ee32b973aa31a21a65adddcc4b69cc2a8a4dee8121ecd51926e0cddbc02@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: Tyler Longwell <tlongwell@block.xyz>
Co-authored-by: npub1mprnacetjua2xx3p5eddmhxyk6wv929ymm5py8kd2xfxurxahspqqlgyta <d8473ee32b973aa31a21a65adddcc4b69cc2a8a4dee8121ecd51926e0cddbc02@sprout-oss.stage.blox.sqprod.co>

crates/buzz-pubsub/src/cache_invalidation.rs

@@ -0,0 +1,160 @@
+//! Cross-pod cache-key invalidation over Redis pub/sub.
+//!
+//! Each relay pod keeps in-memory (moka) membership / accessible-channels /
+//! visibility caches. A membership or visibility change is applied to the local
+//! caches only on the pod that processed the write; other pods would otherwise
+//! rely on the 10s TTL to expire stale entries. This module carries the same
+//! key drops to every pod immediately.
+//!
+//! The message is a pure cache-key drop — never an "evict these subscriptions"
+//! payload. The per-event access gate (`filter_fanout_by_access`) is the
+//! universal delivery-enforcement point, so dropping the stale key is
+//! sufficient: the next read re-fetches authoritative state from the DB.
+
+use futures_util::StreamExt;
+use serde::{Deserialize, Serialize};
+use tokio::sync::broadcast;
+use uuid::Uuid;
+
+/// Redis pub/sub channel for cache-invalidation messages. Distinct from the
+/// `buzz:channel:*` event topic so the two streams never interfere.
+pub const CACHE_INVALIDATION_CHANNEL: &str = "buzz:cache-invalidate";
+
+/// A cache-key drop to apply on every pod. Each variant mirrors exactly one of
+/// the relay's local `invalidate_*` operations.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(tag = "op")]
+pub enum CacheInvalidation {
+    /// Drop the `(channel_id, pubkey)` membership entry and the user's
+    /// accessible-channels entry. Mirrors `invalidate_membership`.
+    Membership {
+        /// Channel whose membership changed.
+        channel_id: Uuid,
+        /// Affected member's pubkey bytes.
+        pubkey: Vec<u8>,
+    },
+    /// Drop every user's accessible-channels entry. Mirrors
+    /// `invalidate_all_accessible_channels` (e.g. a new open channel).
+    AccessibleAll,
+    /// Drop the cached visibility for a single channel. Mirrors
+    /// `invalidate_channel_visibility` (e.g. an open→private flip).
+    Visibility {
+        /// Channel whose visibility changed.
+        channel_id: Uuid,
+    },
+    /// Drop all membership / accessible / visibility caches. Mirrors
+    /// `invalidate_channel_deleted`.
+    ChannelDeleted,
+}
+
+/// Initial reconnect backoff (1 second).
+const BACKOFF_INITIAL_SECS: u64 = 1;
+/// Maximum reconnect backoff (30 seconds).
+const BACKOFF_MAX_SECS: u64 = 30;
+
+/// Subscribes to `buzz:cache-invalidate` and forwards drops to the broadcast.
+///
+/// Mirrors `subscriber::run_subscriber`: a reconnect loop with exponential
+/// backoff (1s → 2s → 4s → … → 30s max). Never returns — runs for the lifetime
+/// of the relay.
+pub async fn run_cache_invalidation_subscriber(
+    redis_url: String,
+    broadcast_tx: broadcast::Sender<CacheInvalidation>,
+) {
+    let mut backoff_secs = BACKOFF_INITIAL_SECS;
+
+    loop {
+        match connect_and_subscribe(&redis_url, &broadcast_tx).await {
+            Ok(()) => {
+                backoff_secs = BACKOFF_INITIAL_SECS;
+                tracing::warn!(
+                    "Redis cache-invalidation stream ended (clean disconnect) — reconnecting in {backoff_secs}s"
+                );
+            }
+            Err(e) => {
+                tracing::error!(
+                    "Redis cache-invalidation error: {e} — reconnecting in {backoff_secs}s"
+                );
+            }
+        }
+
+        tokio::time::sleep(tokio::time::Duration::from_secs(backoff_secs)).await;
+        backoff_secs = (backoff_secs * 2).min(BACKOFF_MAX_SECS);
+
+        tracing::info!("Attempting to reconnect to Redis cache-invalidation...");
+    }
+}
+
+async fn connect_and_subscribe(
+    redis_url: &str,
+    broadcast_tx: &broadcast::Sender<CacheInvalidation>,
+) -> Result<(), redis::RedisError> {
+    let client = redis::Client::open(redis_url)?;
+    let mut conn = client.get_async_pubsub().await?;
+
+    conn.subscribe(CACHE_INVALIDATION_CHANNEL).await?;
+
+    tracing::info!(
+        "Redis cache-invalidation subscriber connected — listening on {CACHE_INVALIDATION_CHANNEL}"
+    );
+
+    let mut stream = conn.on_message();
+    while let Some(msg) = stream.next().await {
+        let payload: String = match msg.get_payload() {
+            Ok(p) => p,
+            Err(e) => {
+                tracing::warn!("Failed to get cache-invalidation payload: {e}");
+                continue;
+            }
+        };
+
+        let invalidation: CacheInvalidation = match serde_json::from_str(&payload) {
+            Ok(v) => v,
+            Err(e) => {
+                tracing::warn!("Failed to deserialize cache-invalidation message: {e}");
+                continue;
+            }
+        };
+
+        if broadcast_tx.send(invalidation).is_err() {
+            tracing::trace!("No cache-invalidation receivers — message dropped");
+        }
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn membership_roundtrips_through_json() {
+        let msg = CacheInvalidation::Membership {
+            channel_id: Uuid::from_u128(0x1234),
+            pubkey: vec![1, 2, 3, 4],
+        };
+        let json = serde_json::to_string(&msg).unwrap();
+        assert_eq!(
+            serde_json::from_str::<CacheInvalidation>(&json).unwrap(),
+            msg
+        );
+    }
+
+    #[test]
+    fn unit_variants_roundtrip_through_json() {
+        for msg in [
+            CacheInvalidation::AccessibleAll,
+            CacheInvalidation::ChannelDeleted,
+            CacheInvalidation::Visibility {
+                channel_id: Uuid::from_u128(0xabcd),
+            },
+        ] {
+            let json = serde_json::to_string(&msg).unwrap();
+            assert_eq!(
+                serde_json::from_str::<CacheInvalidation>(&json).unwrap(),
+                msg
+            );
+        }
+    }
+}

crates/buzz-pubsub/src/lib.rs

@@ -21,6 +21,8 @@
 //! Pool connections handle all other commands.
 //! Lagged receivers get `RecvError::Lagged`.
 
+/// Cross-pod cache-key invalidation over Redis pub/sub.
+pub mod cache_invalidation;
 /// Error types for pub/sub operations.
 pub mod error;
 /// Online/offline presence tracking in Redis.
@@ -41,6 +43,8 @@ use nostr::PublicKey;
 use tokio::sync::broadcast;
 use uuid::Uuid;
 
+use crate::cache_invalidation::{CacheInvalidation, CACHE_INVALIDATION_CHANNEL};
+
 /// A Nostr event received on a specific channel, broadcast to local subscribers.
 #[derive(Debug, Clone)]
 pub struct ChannelEvent {
@@ -72,17 +76,20 @@ pub struct PubSubManager {
     /// Redis URL used by the reconnect loop to re-establish pub/sub connections.
     redis_url: String,
     broadcast_tx: broadcast::Sender<ChannelEvent>,
+    cache_invalidation_tx: broadcast::Sender<CacheInvalidation>,
 }
 
 impl PubSubManager {
     /// Creates a new `PubSubManager` connected to the given Redis URL.
     pub async fn new(redis_url: &str, pool: deadpool_redis::Pool) -> Result<Self, PubSubError> {
         let (broadcast_tx, _) = broadcast::channel(4096);
+        let (cache_invalidation_tx, _) = broadcast::channel(4096);
 
         Ok(Self {
             pool,
             redis_url: redis_url.to_string(),
             broadcast_tx,
+            cache_invalidation_tx,
         })
     }
 
@@ -94,11 +101,44 @@ impl PubSubManager {
         subscriber::run_subscriber(self.redis_url.clone(), self.broadcast_tx.clone()).await;
     }
 
+    /// Starts the cache-invalidation subscriber loop with automatic
+    /// reconnection. Runs forever — spawn this in a background task.
+    pub async fn run_cache_invalidation_subscriber(self: Arc<Self>) {
+        cache_invalidation::run_cache_invalidation_subscriber(
+            self.redis_url.clone(),
+            self.cache_invalidation_tx.clone(),
+        )
+        .await;
+    }
+
     /// Returns a new broadcast receiver for locally-published channel events.
     pub fn subscribe_local(&self) -> broadcast::Receiver<ChannelEvent> {
         self.broadcast_tx.subscribe()
     }
 
+    /// Returns a new broadcast receiver for cross-pod cache-invalidation drops.
+    pub fn subscribe_cache_invalidations(&self) -> broadcast::Receiver<CacheInvalidation> {
+        self.cache_invalidation_tx.subscribe()
+    }
+
+    /// Publish a cache-key drop to all pods. Fire-and-forget at the call site:
+    /// the local cache is already dropped synchronously; this carries the same
+    /// drop cross-pod. A dropped publish is backstopped by the REQ denial-path
+    /// DB confirmation, so callers may spawn this without awaiting delivery.
+    pub async fn publish_cache_invalidation(
+        &self,
+        invalidation: &CacheInvalidation,
+    ) -> Result<i64, PubSubError> {
+        let mut conn = self.pool.get().await?;
+        let payload = serde_json::to_string(invalidation)?;
+        let subscriber_count: i64 = redis::cmd("PUBLISH")
+            .arg(CACHE_INVALIDATION_CHANNEL)
+            .arg(&payload)
+            .query_async(&mut conn)
+            .await?;
+        Ok(subscriber_count)
+    }
+
     /// Publish an event to the Redis channel. Returns subscriber count.
     ///
     /// Routing note (NIP-ER author-private reminders): events are keyed by
@@ -202,6 +242,36 @@ mod tests {
         assert_eq!(received.event.id, event_id);
     }
 
+    #[tokio::test]
+    #[ignore = "requires Redis"]
+    async fn test_cache_invalidation_roundtrip() {
+        let manager = make_manager().await;
+        let mut rx = manager.subscribe_cache_invalidations();
+
+        let manager_clone = manager.clone();
+        tokio::spawn(async move { manager_clone.run_cache_invalidation_subscriber().await });
+        tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
+
+        let channel_id = Uuid::new_v4();
+        let pubkey = Keys::generate().public_key().to_bytes().to_vec();
+        let sent = CacheInvalidation::Membership {
+            channel_id,
+            pubkey: pubkey.clone(),
+        };
+
+        manager
+            .publish_cache_invalidation(&sent)
+            .await
+            .expect("publish failed");
+
+        let received = tokio::time::timeout(tokio::time::Duration::from_secs(2), rx.recv())
+            .await
+            .expect("timeout")
+            .expect("channel closed");
+
+        assert_eq!(received, sent);
+    }
+
     #[tokio::test]
     #[ignore = "requires Redis"]
     async fn test_presence_set_and_get() {

crates/buzz-relay/src/api/git/transport.rs

@@ -1113,13 +1113,11 @@ async fn finalize_push(state: &Arc<AppState>, ctx: PushContext) -> Response {
         match build_ref_state_event(&inputs, &state.relay_keypair) {
             Ok(event) => match state.db.insert_event(&event, None).await {
                 Ok((stored, true)) => {
-                    let matches = state.sub_registry.fan_out(&stored);
-                    for (conn_id, sub_id) in matches {
-                        let _ = state.conn_manager.send_to(
-                            conn_id,
-                            crate::protocol::RelayMessage::event(&sub_id, &stored.event),
-                        );
-                    }
+                    // Routed through the guarded send path for uniformity; the
+                    // access gate no-ops for this globally-scoped
+                    // (channel_id = None) ref-state event.
+                    crate::handlers::event::fan_out_event_to_local_subscribers(state, &stored)
+                        .await;
                     info!(
                         owner = %ctx.owner,
                         repo = %ctx.repo_id,

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

@@ -788,27 +788,11 @@ async fn emit_participant_event(
     //    double-delivery when the event echoes back through the subscriber loop.
     state.mark_local_event(&event.id);
 
-    // 3. Local fan-out to WS subscribers on this node (same pattern as
+    // 3. Local fan-out to WS subscribers on this node, through the guarded send
+    //    path so a stale subscription on a removed/non-member connection cannot
+    //    receive this channel's audio lifecycle event (same gate as
     //    dispatch_persistent_event in the ingest handler).
-    let matches = state.sub_registry.fan_out(&stored);
-    if !matches.is_empty() {
-        let event_json = serde_json::to_string(&event)
-            .expect("nostr::Event serialization is infallible for well-formed events");
-        let mut drop_count = 0u32;
-        for (target_conn_id, sub_id) in &matches {
-            let msg = format!(r#"["EVENT","{}",{}]"#, sub_id, event_json);
-            if !state.conn_manager.send_to(*target_conn_id, msg) {
-                drop_count += 1;
-            }
-        }
-        if drop_count > 0 {
-            warn!(
-                event_id = %event_id_hex,
-                drop_count,
-                "audio lifecycle fan-out: {drop_count} connection(s) dropped"
-            );
-        }
-    }
+    crate::handlers::event::fan_out_event_to_local_subscribers(state, &stored).await;
 
     // 4. Cross-node broadcast via Redis pub/sub.
     if let Err(e) = state.pubsub.publish_event(parent_channel_id, &event).await {

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

@@ -31,10 +31,12 @@ pub async fn handle_count(
     state: Arc<AppState>,
 ) {
     // Require auth
-    let pubkey_bytes = {
+    let (pubkey_bytes, token_channel_ids) = {
         let auth = conn.auth_state.read().await;
         match &*auth {
-            AuthState::Authenticated(ctx) => ctx.pubkey.to_bytes().to_vec(),
+            AuthState::Authenticated(ctx) => {
+                (ctx.pubkey.to_bytes().to_vec(), ctx.channel_ids.clone())
+            }
             _ => {
                 conn.send(RelayMessage::closed(
                     &sub_id,
@@ -71,7 +73,8 @@ pub async fn handle_count(
     }
 
     // Get channels this user can access — same enforcement as WS REQ handler.
-    let accessible_channels = match state.get_accessible_channel_ids_cached(&pubkey_bytes).await {
+    let mut accessible_channels = match state.get_accessible_channel_ids_cached(&pubkey_bytes).await
+    {
         Ok(ids) => ids,
         Err(e) => {
             warn!(sub_id = %sub_id, "Failed to get accessible channels: {e}");
@@ -79,6 +82,14 @@ pub async fn handle_count(
             return;
         }
     };
+    // Narrow to the token's channel scope, mirroring the WS REQ handler. Without
+    // this, a scoped token would COUNT events in channels outside its scope via
+    // the no-channel-filter SQL pushdown below (which counts every accessible
+    // channel). The per-filter targeted-channel repair is bounded by the same
+    // scope through `resolve_request_local_access`'s `token_allows` argument.
+    if let Some(allowed) = token_channel_ids.as_deref() {
+        accessible_channels.retain(|channel_id| allowed.contains(channel_id));
+    }
 
     // For each filter, count matching events with channel access enforcement.
     let mut total: u64 = 0;
@@ -89,8 +100,31 @@ pub async fn handle_count(
         let needs_author_only_filtering = super::req::filter_can_match_author_only_kinds(filter);
 
         if let Some(ch_id) = extract_channel_from_filter(filter) {
-            // Filter targets a specific channel — verify access.
-            if !accessible_channels.contains(&ch_id) {
+            // Filter targets a specific channel — verify access. Mirrors the WS
+            // REQ handler: a cache-negative may be a stale miss on a non-writer
+            // pod, so confirm uncached and repair the Vec request-locally via
+            // `super::req::resolve_request_local_access` (so a just-added channel
+            // is counted, and any later filter on the same channel sees it too).
+            let db_is_member = if accessible_channels.contains(&ch_id) {
+                None
+            } else {
+                match state.db.is_member(ch_id, &pubkey_bytes).await {
+                    Ok(member) => Some(member),
+                    Err(e) => {
+                        warn!(sub_id = %sub_id, "Channel membership confirmation failed: {e}");
+                        conn.send(RelayMessage::closed(&sub_id, "error: database error"));
+                        return;
+                    }
+                }
+            };
+            if !super::req::resolve_request_local_access(
+                &mut accessible_channels,
+                ch_id,
+                token_channel_ids
+                    .as_deref()
+                    .is_none_or(|allowed| allowed.contains(&ch_id)),
+                db_is_member,
+            ) {
                 continue; // Skip filters targeting inaccessible channels.
             }
             // Channel is accessible — count with pushability check.

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

@@ -57,7 +57,7 @@ fn bounded_kind_label(kind: u32) -> String {
 /// survives on another node after an open->private flip, its events are not
 /// delivered here.
 pub async fn filter_fanout_by_access(
-    state: &Arc<AppState>,
+    state: &AppState,
     stored_event: &StoredEvent,
     matches: Vec<(crate::subscription::ConnId, crate::subscription::SubId)>,
 ) -> Vec<(crate::subscription::ConnId, crate::subscription::SubId)> {
@@ -112,6 +112,54 @@ pub async fn filter_fanout_by_access(
     allowed
 }
 
+/// Deliver one event to this relay's local subscribers through the access gate.
+///
+/// This is the single guarded send path for relay-local EVENT delivery. It runs
+/// `fan_out()` to find matching subscriptions, then `filter_fanout_by_access()`
+/// to drop recipients without access (private-channel non-members, author-only
+/// kinds delivered to non-authors), then writes the EVENT frames. The invariant
+/// it enforces: a registered subscription is never sufficient for delivery —
+/// delivery always revalidates access on the sending pod, so a stale
+/// subscription surviving a membership/visibility change (e.g. after an
+/// open→private flip or a cross-pod cache lag) cannot leak events.
+///
+/// All relay-local live fan-out routes through here. The two exceptions are
+/// `dispatch_persistent_event` (persistent ingest) and `fan_out_pubsub_event`
+/// (Redis cross-node), which call `filter_fanout_by_access` inline: the former
+/// layers an additional per-recipient DM-visibility-owner gate on top, the
+/// latter skips local echoes — both are equivalent to this helper plus their
+/// own extra step.
+pub(crate) async fn fan_out_event_to_local_subscribers(state: &AppState, stored: &StoredEvent) {
+    let matches = state.sub_registry.fan_out(stored);
+    let matches = filter_fanout_by_access(state, stored, matches).await;
+    metrics::histogram!("buzz_fanout_recipients").record(matches.len() as f64);
+    if matches.is_empty() {
+        return;
+    }
+
+    let event_json = match serde_json::to_string(&stored.event) {
+        Ok(json) => json,
+        Err(e) => {
+            error!(event_id = %stored.event.id.to_hex(), "Failed to serialize event for fan-out: {e}");
+            return;
+        }
+    };
+    let mut drop_count = 0u32;
+    for (conn_id, sub_id) in &matches {
+        let msg = format!(r#"["EVENT","{}",{}]"#, sub_id, event_json);
+        if !state.conn_manager.send_to(*conn_id, msg) {
+            drop_count += 1;
+        }
+    }
+    if drop_count > 0 {
+        tracing::warn!(
+            event_id = %stored.event.id.to_hex(),
+            drop_count,
+            "fan-out: {drop_count} connection(s) cancelled due to full/closed buffers"
+        );
+    }
+}
+
 /// Fan out one event received from Redis pub/sub to this relay's local subscribers.
 pub async fn fan_out_pubsub_event(state: &Arc<AppState>, channel_event: buzz_pubsub::ChannelEvent) {
     // Nil UUID is the sentinel for channel-less global events (see
@@ -594,27 +642,12 @@ async fn handle_ephemeral_event(
             warn!(conn_id = %conn_id, event_id = %event_id_hex, "Ephemeral publish failed: {e}");
         }
 
-        // Direct fan-out to local WS subscribers.
+        // Direct fan-out to local WS subscribers, through the guarded send path
+        // so a stale subscription on a removed/non-member connection cannot
+        // receive this private-channel ephemeral event.
         // Pass the channel_id so fan_out() uses the channel-kind index.
         let stored_event = StoredEvent::new(event.clone(), Some(ch_id));
-        let matches = state.sub_registry.fan_out(&stored_event);
-        metrics::histogram!("buzz_fanout_recipients").record(matches.len() as f64);
-        let event_json = serde_json::to_string(&event)
-            .expect("nostr::Event serialization is infallible for well-formed events");
-        let mut drop_count = 0u32;
-        for (target_conn_id, sub_id) in &matches {
-            let msg = format!(r#"["EVENT","{}",{}]"#, sub_id, event_json);
-            if !state.conn_manager.send_to(*target_conn_id, msg) {
-                drop_count += 1;
-            }
-        }
-        if drop_count > 0 {
-            tracing::warn!(
-                event_id = %event_id_hex,
-                drop_count,
-                "fan-out: {drop_count} connection(s) cancelled due to full/closed buffers"
-            );
-        }
+        fan_out_event_to_local_subscribers(&state, &stored_event).await;
     } else {
         // Channel-less ephemeral events (e.g., NIP-AB pairing kind:24134).
         //
@@ -631,27 +664,12 @@ async fn handle_ephemeral_event(
             warn!(conn_id = %conn_id, event_id = %event_id_hex, "Ephemeral global publish failed: {e}");
         }
 
-        // Direct fan-out to local WS subscribers.
-        // Pass channel_id=None so fan_out() uses the global subscriber index.
+        // Direct fan-out to local WS subscribers through the guarded send path.
+        // Pass channel_id=None so fan_out() uses the global subscriber index;
+        // filter_fanout_by_access no-ops for channel-less events except the
+        // author-only-kind gate.
         let stored_event = StoredEvent::new(event.clone(), None);
-        let matches = state.sub_registry.fan_out(&stored_event);
-        metrics::histogram!("buzz_fanout_recipients").record(matches.len() as f64);
-        let event_json = serde_json::to_string(&event)
-            .expect("nostr::Event serialization is infallible for well-formed events");
-        let mut drop_count = 0u32;
-        for (target_conn_id, sub_id) in &matches {
-            let msg = format!(r#"["EVENT","{}",{}]"#, sub_id, event_json);
-            if !state.conn_manager.send_to(*target_conn_id, msg) {
-                drop_count += 1;
-            }
-        }
-        if drop_count > 0 {
-            tracing::warn!(
-                event_id = %event_id_hex,
-                drop_count,
-                "fan-out: {drop_count} connection(s) cancelled due to full/closed buffers"
-            );
-        }
+        fan_out_event_to_local_subscribers(&state, &stored_event).await;
     }
 
     conn.send(RelayMessage::ok(event_id_hex, true, ""));
@@ -806,19 +824,6 @@ async fn handle_agent_observer_event(
         }
     }
 
-    let event_json = match serde_json::to_string(&event) {
-        Ok(json) => json,
-        Err(e) => {
-            error!(event_id = %event_id_hex, "Failed to serialize agent observer event: {e}");
-            conn.send(RelayMessage::ok(
-                event_id_hex,
-                false,
-                "error: internal server error",
-            ));
-            return;
-        }
-    };
-
     state.mark_local_event(&event.id);
     if let Err(e) = state.pubsub.publish_event(uuid::Uuid::nil(), &event).await {
         state.local_event_ids.invalidate(&event.id.to_bytes());
@@ -826,31 +831,14 @@ async fn handle_agent_observer_event(
     }
 
     let stored_event = StoredEvent::new(event.clone(), None);
-    let matches = state.sub_registry.fan_out(&stored_event);
-    metrics::histogram!("buzz_fanout_recipients").record(matches.len() as f64);
     debug!(
         event_id = %event_id_hex,
         agent = %route.agent.to_hex(),
         owner = %route.owner.to_hex(),
         direction = ?route.direction,
-        match_count = matches.len(),
         "Agent observer fan-out"
     );
-
-    let mut drop_count = 0u32;
-    for (target_conn_id, sub_id) in &matches {
-        let msg = format!(r#"["EVENT","{}",{}]"#, sub_id, event_json);
-        if !state.conn_manager.send_to(*target_conn_id, msg) {
-            drop_count += 1;
-        }
-    }
-    if drop_count > 0 {
-        tracing::warn!(
-            event_id = %event_id_hex,
-            drop_count,
-            "agent observer fan-out: {drop_count} connection(s) cancelled due to full/closed buffers"
-        );
-    }
+    fan_out_event_to_local_subscribers(&state, &stored_event).await;
 
     conn.send(RelayMessage::ok(event_id_hex, true, ""));
 }

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

@@ -316,14 +316,9 @@ async fn publish_channelless_ephemeral(state: &Arc<AppState>, event: &nostr::Eve
         tracing::warn!(event_id = %event.id, "mesh call-me-now global publish failed: {e}");
     }
     let stored = StoredEvent::new(event.clone(), None);
-    let matches = state.sub_registry.fan_out(&stored);
-    metrics::histogram!("buzz_fanout_recipients").record(matches.len() as f64);
-    if let Ok(event_json) = serde_json::to_string(event) {
-        for (target_conn_id, sub_id) in &matches {
-            let msg = format!(r#"["EVENT","{sub_id}",{event_json}]"#);
-            let _ = state.conn_manager.send_to(*target_conn_id, msg);
-        }
-    }
+    // Routed through the guarded send path for uniformity; the access gate
+    // no-ops for this globally-scoped (channel_id = None) call-me-now event.
+    crate::handlers::event::fan_out_event_to_local_subscribers(state, &stored).await;
 }
 
 /// Handle a verified KIND_MESH_STATUS_REPORT (24620) from an authenticated relay

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

@@ -91,6 +91,47 @@ pub async fn handle_req(
 
     let channel_id = extract_channel_id_from_filters(&filters);
 
+    // ── Channel access + stale-cache repair (BEFORE search & registration) ───
+    // Confirm channel access up front so the repaired `accessible_channels`
+    // vector reaches every downstream consumer: the NIP-50 search branch
+    // below, subscription registration, historical delivery, and COUNT. A
+    // cache-negative may be a stale miss on a non-writer pod (member just added
+    // on the pod that processed the write, before the 10s TTL expires or the
+    // cross-pod invalidation lands), so on a miss we confirm uncached against
+    // the DB; a verified positive repairs the vector request-locally (see
+    // `resolve_request_local_access`). Running this ahead of the search branch
+    // is what fixes the search false-miss: a `#h=<just-added>` search would
+    // otherwise be scoped against the stale vector and return empty.
+    if let Some(ch_id) = channel_id {
+        let token_allows = token_channel_ids
+            .as_deref()
+            .is_none_or(|allowed| allowed.contains(&ch_id));
+        let db_is_member = if !token_allows || accessible_channels.contains(&ch_id) {
+            None
+        } else {
+            match state.db.is_member(ch_id, &pubkey_bytes).await {
+                Ok(member) => Some(member),
+                Err(e) => {
+                    warn!(conn_id = %conn_id, "Channel membership confirmation failed: {e}");
+                    conn.send(RelayMessage::closed(&sub_id, "error: database error"));
+                    return;
+                }
+            }
+        };
+        if !resolve_request_local_access(
+            &mut accessible_channels,
+            ch_id,
+            token_allows,
+            db_is_member,
+        ) {
+            conn.send(RelayMessage::closed(
+                &sub_id,
+                "restricted: not a channel member",
+            ));
+            return;
+        }
+    }
+
     // ── #p / engram gating for globally-stored sensitive kinds ───────────────
     // Applied BEFORE the NIP-50 search branch so that an authenticated member
     // cannot use `{"search":"...","kinds":[30174]}` (or similar for p-gated
@@ -153,17 +194,6 @@ pub async fn handle_req(
         return;
     }
 
-    // Check channel access BEFORE registering the subscription.
-    if let Some(ch_id) = channel_id {
-        if !accessible_channels.contains(&ch_id) {
-            conn.send(RelayMessage::closed(
-                &sub_id,
-                "restricted: not a channel member",
-            ));
-            return;
-        }
-    }
-
     {
         let mut subs = conn.subscriptions.lock().await;
         subs.insert(sub_id.clone(), filters.clone());
@@ -274,6 +304,52 @@ pub async fn handle_req(
 /// Maximum Typesense pages to fetch per filter (prevents unbounded loops).
 const MAX_SEARCH_PAGES: u32 = 10;
 
+/// Resolve request-local channel access, repairing a stale cache-negative.
+///
+/// `accessible_channels` is the per-request membership vector — built once from
+/// the 10s cache (and already narrowed by any scoped-auth `token_channel_ids`
+/// via `retain`) and reused for subscription registration, historical delivery,
+/// search scope, and COUNT. On a multi-pod relay it can be stale on a non-writer
+/// pod (a member just added on another pod, before the TTL expires or the
+/// cross-pod invalidation lands), so the cache-negative branch confirms against
+/// the DB uncached and passes the result here.
+///
+/// `token_allows` is the scoped-auth upper bound: `false` when a scoped token is
+/// present and does NOT cover `ch_id`. The DB-positive repair must never push a
+/// channel back in past that bound, or a token scoped to channel A could reach
+/// channel B merely because the user is a DB member of B.
+///
+/// Truth table:
+/// - token denies `ch_id`               → denied, no DB needed, no repair
+/// - cached contains `ch_id`            → allowed, no repair, no DB needed
+/// - cache-miss + DB says member        → allowed, `ch_id` pushed once (repair)
+/// - cache-miss + DB says not a member  → denied, vector unchanged
+///
+/// The push is what makes the confirmation request-local-authoritative: every
+/// downstream consumer reads the same repaired vector, so a stale negative
+/// cannot stay sticky for the rest of the request. `db_is_member` is `None` when
+/// the cache hit or the token bound denied (DB was never consulted).
+pub(crate) fn resolve_request_local_access(
+    accessible_channels: &mut Vec<uuid::Uuid>,
+    ch_id: uuid::Uuid,
+    token_allows: bool,
+    db_is_member: Option<bool>,
+) -> bool {
+    if !token_allows {
+        return false;
+    }
+    if accessible_channels.contains(&ch_id) {
+        return true;
+    }
+    match db_is_member {
+        Some(true) => {
+            accessible_channels.push(ch_id);
+            true
+        }
+        _ => false,
+    }
+}
+
 pub(crate) fn build_search_channel_scope_filter(
     accessible_channels: &[uuid::Uuid],
     include_global: bool,
@@ -889,6 +965,75 @@ mod tests {
     use super::*;
     use nostr::{Alphabet, Filter, SingleLetterTag};
 
+    #[test]
+    fn request_local_access_cache_positive_no_db_no_repair() {
+        let ch = uuid::Uuid::new_v4();
+        let mut accessible = vec![ch];
+        // Cache hit: DB was never consulted (None), allowed, vector unchanged.
+        assert!(resolve_request_local_access(
+            &mut accessible,
+            ch,
+            true,
+            None
+        ));
+        assert_eq!(accessible, vec![ch], "no repair, no duplicate on cache hit");
+    }
+
+    #[test]
+    fn request_local_access_cache_negative_db_member_repairs() {
+        let ch = uuid::Uuid::new_v4();
+        let mut accessible: Vec<uuid::Uuid> = vec![];
+        // Stale cache-miss but DB confirms membership: allowed AND repaired.
+        assert!(resolve_request_local_access(
+            &mut accessible,
+            ch,
+            true,
+            Some(true)
+        ));
+        assert!(
+            accessible.contains(&ch),
+            "verified positive must push ch_id so the rest of the request sees it"
+        );
+    }
+
+    #[test]
+    fn request_local_access_cache_negative_db_nonmember_denied() {
+        let ch = uuid::Uuid::new_v4();
+        let mut accessible: Vec<uuid::Uuid> = vec![];
+        // Cache-miss and DB confirms non-membership: denied, vector unchanged.
+        assert!(!resolve_request_local_access(
+            &mut accessible,
+            ch,
+            true,
+            Some(false)
+        ));
+        assert!(
+            accessible.is_empty(),
+            "denied access must not mutate the request-local vector"
+        );
+    }
+
+    #[test]
+    fn request_local_access_token_denies_never_repairs() {
+        let ch = uuid::Uuid::new_v4();
+        let mut accessible: Vec<uuid::Uuid> = vec![];
+        // Scoped token does NOT cover ch_id: denied even though the DB confirms
+        // membership. The token scope is an upper bound on the repair — a DB
+        // positive must never push a channel back in past a narrower token, or
+        // a token scoped to channel A could reach channel B merely because the
+        // user is a DB member of B.
+        assert!(!resolve_request_local_access(
+            &mut accessible,
+            ch,
+            false,
+            Some(true)
+        ));
+        assert!(
+            accessible.is_empty(),
+            "token-denied access must not be repaired into the vector"
+        );
+    }
+
     fn filter_with_channel(channel_id: uuid::Uuid) -> Filter {
         Filter::new().custom_tag(
             SingleLetterTag::lowercase(Alphabet::H),

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

@@ -75,8 +75,9 @@ async fn evict_conn_channel_subscriptions(
 
 /// Revoke live channel subscriptions held by connections whose authenticated
 /// pubkey is not a current member. Used when an open channel flips to private:
-/// non-members could have subscribed while it was open, and fan-out does not
-/// re-check membership per event, so their subscriptions must be closed.
+/// non-members could have subscribed while it was open. Fan-out now re-checks
+/// membership per event as the delivery-time safety net; this eviction closes
+/// subscriptions promptly so clients stop treating the channel as live.
 async fn evict_non_member_channel_subscriptions(
     state: &Arc<AppState>,
     channel_id: Uuid,
@@ -604,21 +605,10 @@ pub async fn emit_membership_notification(
         return Ok(());
     }
 
-    // Fan-out only — skip search indexing and workflow evaluation.
-    let matches = state.sub_registry.fan_out(&stored);
-    if !matches.is_empty() {
-        let event_json = match serde_json::to_string(&stored.event) {
-            Ok(json) => json,
-            Err(e) => {
-                warn!("failed to serialize membership notification for fan-out: {e}");
-                return Ok(());
-            }
-        };
-        for (target_conn_id, sub_id) in &matches {
-            let msg = format!(r#"["EVENT","{}",{}]"#, sub_id, event_json);
-            state.conn_manager.send_to(*target_conn_id, msg);
-        }
-    }
+    // Fan-out only — skip search indexing and workflow evaluation. Routed
+    // through the guarded send path for uniformity; the access gate no-ops for
+    // these globally-scoped (channel_id = None) events.
+    crate::handlers::event::fan_out_event_to_local_subscribers(state, &stored).await;
 
     info!(
         channel = %channel_id,
@@ -2150,13 +2140,9 @@ async fn emit_initial_ref_state(
         .await
         .map_err(|e| anyhow::anyhow!("insert kind:30618: {e}"))?;
     if was_inserted {
-        let matches = state.sub_registry.fan_out(&stored);
-        for (conn_id, sub_id) in matches {
-            let _ = state.conn_manager.send_to(
-                conn_id,
-                crate::protocol::RelayMessage::event(&sub_id, &stored.event),
-            );
-        }
+        // Routed through the guarded send path for uniformity; the access gate
+        // no-ops for this globally-scoped (channel_id = None) ref-state event.
+        crate::handlers::event::fan_out_event_to_local_subscribers(state, &stored).await;
     }
     Ok(())
 }
@@ -2239,20 +2225,9 @@ async fn publish_nip43_delta(
         return Ok(());
     }
 
-    let matches = state.sub_registry.fan_out(&stored);
-    if !matches.is_empty() {
-        let event_json = match serde_json::to_string(&stored.event) {
-            Ok(json) => json,
-            Err(e) => {
-                warn!("failed to serialize kind:{kind} for fan-out: {e}");
-                return Ok(());
-            }
-        };
-        for (target_conn_id, sub_id) in &matches {
-            let msg = format!(r#"["EVENT","{}",{}]"#, sub_id, event_json);
-            state.conn_manager.send_to(*target_conn_id, msg);
-        }
-    }
+    // Routed through the guarded send path for uniformity; the access gate
+    // no-ops for this globally-scoped (channel_id = None) NIP-43 event.
+    crate::handlers::event::fan_out_event_to_local_subscribers(state, &stored).await;
 
     info!(
         target = %target_pubkey_hex,

crates/buzz-relay/src/main.rs

@@ -187,6 +187,12 @@ async fn main() -> anyhow::Result<()> {
     let pubsub_for_sub = Arc::clone(&pubsub);
     tokio::spawn(async move { pubsub_for_sub.run_subscriber().await });
 
+    // Spawn Redis pub/sub subscriber for cross-pod cache-key invalidation.
+    // Membership / visibility changes on other pods are received here and the
+    // matching local moka caches are dropped (via the consumer loop below).
+    let pubsub_for_cache = Arc::clone(&pubsub);
+    tokio::spawn(async move { pubsub_for_cache.run_cache_invalidation_subscriber().await });
+
     let auth = AuthService::new(config.auth.clone());
 
     let search_config = SearchConfig {
@@ -513,6 +519,32 @@ async fn main() -> anyhow::Result<()> {
         });
     }
 
+    // Cross-pod cache-invalidation consumer: receive cache-key drops from Redis
+    // pub/sub (published by other relay instances when membership/visibility
+    // changes) and apply the matching local moka drop. Uses the `*_local` drop
+    // variants so a received drop is never re-published.
+    {
+        let state_for_cache = Arc::clone(&state);
+        let mut rx = state_for_cache.pubsub.subscribe_cache_invalidations();
+        tokio::spawn(async move {
+            loop {
+                match rx.recv().await {
+                    Ok(invalidation) => {
+                        state_for_cache.apply_cache_invalidation(invalidation);
+                    }
+                    Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
+                        metrics::counter!("buzz_cache_invalidation_lag_total").increment(n);
+                        tracing::warn!("Cache-invalidation consumer lagged by {n} messages");
+                    }
+                    Err(tokio::sync::broadcast::error::RecvError::Closed) => {
+                        tracing::error!("Cache-invalidation broadcast channel closed");
+                        break;
+                    }
+                }
+            }
+        });
+    }
+
     let router = build_router(Arc::clone(&state));
     let health_router = build_health_router(Arc::clone(&state));
 

crates/buzz-relay/src/state.rs

@@ -17,6 +17,7 @@ use buzz_auth::AuthService;
 use buzz_core::event::StoredEvent;
 use buzz_db::Db;
 use buzz_media::MediaStorage;
+use buzz_pubsub::cache_invalidation::CacheInvalidation;
 use buzz_pubsub::PubSubManager;
 use buzz_search::SearchService;
 use buzz_workflow::WorkflowEngine;
@@ -448,7 +449,22 @@ impl AppState {
     }
 
     /// Invalidate caches after a membership change (add/remove member).
+    ///
+    /// Drops the local moka entries AND fire-and-forget publishes the same drop
+    /// to every other pod over Redis (see [`apply_cache_invalidation`]). The
+    /// publish is spawned, not awaited: the local drop is already done, and a
+    /// dropped publish is backstopped by the REQ denial-path DB confirmation.
     pub fn invalidate_membership(&self, channel_id: Uuid, pubkey: &[u8]) {
+        self.invalidate_membership_local(channel_id, pubkey);
+        self.spawn_cache_invalidation(CacheInvalidation::Membership {
+            channel_id,
+            pubkey: pubkey.to_vec(),
+        });
+    }
+
+    /// Local-only membership drop. The cross-pod consumer calls this directly so
+    /// applying a received drop never re-publishes it.
+    pub(crate) fn invalidate_membership_local(&self, channel_id: Uuid, pubkey: &[u8]) {
         self.membership_cache
             .invalidate(&(channel_id, pubkey.to_vec()));
         self.accessible_channels_cache.invalidate(&pubkey.to_vec());
@@ -456,11 +472,23 @@ impl AppState {
 
     /// Invalidate all users' accessible-channels cache (e.g. new open channel created).
     pub fn invalidate_all_accessible_channels(&self) {
+        self.invalidate_all_accessible_channels_local();
+        self.spawn_cache_invalidation(CacheInvalidation::AccessibleAll);
+    }
+
+    /// Local-only accessible-channels drop. See [`invalidate_membership_local`].
+    pub(crate) fn invalidate_all_accessible_channels_local(&self) {
         self.accessible_channels_cache.invalidate_all();
     }
 
     /// Invalidate the cached visibility for a single channel (e.g. after a flip).
     pub fn invalidate_channel_visibility(&self, channel_id: Uuid) {
+        self.invalidate_channel_visibility_local(channel_id);
+        self.spawn_cache_invalidation(CacheInvalidation::Visibility { channel_id });
+    }
+
+    /// Local-only visibility drop. See [`invalidate_membership_local`].
+    pub(crate) fn invalidate_channel_visibility_local(&self, channel_id: Uuid) {
         self.channel_visibility_cache.invalidate(&channel_id);
     }
 
@@ -471,11 +499,48 @@ impl AppState {
     /// keys, and stale `is_member=true` entries for a deleted channel would bypass
     /// the DB's `deleted_at IS NULL` guard.
     pub fn invalidate_channel_deleted(&self) {
+        self.invalidate_channel_deleted_local();
+        self.spawn_cache_invalidation(CacheInvalidation::ChannelDeleted);
+    }
+
+    /// Local-only channel-deleted drop. See [`invalidate_membership_local`].
+    pub(crate) fn invalidate_channel_deleted_local(&self) {
         self.membership_cache.invalidate_all();
         self.accessible_channels_cache.invalidate_all();
         self.channel_visibility_cache.invalidate_all();
     }
 
+    /// Fire-and-forget publish of a cache-key drop to all other pods. Failures
+    /// are logged and swallowed — the REQ denial-path DB confirmation is the
+    /// backstop, so a missed publish degrades to a <=10s TTL wait, never a leak.
+    fn spawn_cache_invalidation(&self, invalidation: CacheInvalidation) {
+        let pubsub = Arc::clone(&self.pubsub);
+        tokio::spawn(async move {
+            if let Err(e) = pubsub.publish_cache_invalidation(&invalidation).await {
+                tracing::warn!("Failed to publish cache invalidation {invalidation:?}: {e}");
+            }
+        });
+    }
+
+    /// Apply a cache-key drop received from another pod. Calls the local-only
+    /// drop variants so a received drop is never re-published (no fan-out loop).
+    pub fn apply_cache_invalidation(&self, invalidation: CacheInvalidation) {
+        match invalidation {
+            CacheInvalidation::Membership { channel_id, pubkey } => {
+                self.invalidate_membership_local(channel_id, &pubkey);
+            }
+            CacheInvalidation::AccessibleAll => {
+                self.invalidate_all_accessible_channels_local();
+            }
+            CacheInvalidation::Visibility { channel_id } => {
+                self.invalidate_channel_visibility_local(channel_id);
+            }
+            CacheInvalidation::ChannelDeleted => {
+                self.invalidate_channel_deleted_local();
+            }
+        }
+    }
+
     /// Get accessible channel IDs with a 10-second cache. Falls back to DB on miss.
     pub async fn get_accessible_channel_ids_cached(
         &self,