diff --git a/crates/buzz-pubsub/src/cache_invalidation.rs b/crates/buzz-pubsub/src/cache_invalidation.rs index f48158c9c..9567b2066 100644 --- a/crates/buzz-pubsub/src/cache_invalidation.rs +++ b/crates/buzz-pubsub/src/cache_invalidation.rs @@ -10,28 +10,47 @@ //! 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. +//! +//! # Subscription scope +//! +//! This pod subscribes to the exact per-community channels of the communities +//! whose entries it may hold — **cache residency**, not connection lifetime. +//! Cached authorization outlives the socket that populated it, so scoping by +//! live connections would leave a pod holding stale authz for a community it is +//! no longer listening to. See [`crate::community_topics`] for the desired / +//! established contract and why the caller must not insert a cache entry for a +//! community that is not yet established. + +use std::sync::Arc; use buzz_core::{CommunityId, TenantContext}; -use futures_util::StreamExt; use serde::{Deserialize, Serialize}; use tokio::sync::broadcast; use uuid::Uuid; +use crate::community_topics::{ + run_community_subscriber, CommunityChannelFamily, CommunityTopics, DesiredCommunities, +}; use crate::topic::BUZZ_PREFIX; /// Tenant-local Redis pub/sub channel suffix for cache-invalidation messages. pub const CACHE_INVALIDATION_SUFFIX: &str = "cache-invalidate"; -/// Pattern used by the subscriber to receive cache invalidations for all -/// communities this pod may have cached locally. -pub const CACHE_INVALIDATION_PATTERN: &str = "buzz:*:cache-invalidate"; +/// Log label for the cache-invalidation subscriber. +pub(crate) const CACHE_INVALIDATION_NAME: &str = "cache-invalidation"; /// Redis pub/sub channel for cache-invalidation messages under `ctx`. pub fn cache_invalidation_channel(ctx: &TenantContext) -> String { - format!( - "{BUZZ_PREFIX}:{}:{CACHE_INVALIDATION_SUFFIX}", - ctx.community() - ) + cache_invalidation_channel_for(ctx.community()) +} + +/// Redis pub/sub channel for cache-invalidation messages in `community`. +/// +/// The subscriber needs this without a [`TenantContext`]: it subscribes to the +/// exact channels of the communities whose cache entries this pod holds, and +/// those are known only as ids. +pub fn cache_invalidation_channel_for(community: CommunityId) -> String { + format!("{BUZZ_PREFIX}:{community}:{CACHE_INVALIDATION_SUFFIX}") } /// Parse a cache-invalidation Redis channel into its scoped community id. @@ -87,78 +106,47 @@ pub struct ScopedCacheInvalidation { pub invalidation: CacheInvalidation, } -/// 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 scoped drops to the broadcast. +/// Subscribes to the exact `buzz:{community}:cache-invalidate` channels this pod +/// needs and forwards scoped 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. +/// `desired` is the set of communities whose cache entries this pod may hold — +/// re-read on every reconcile, never cached here. Never returns; runs for the +/// lifetime of the relay. See [`crate::community_topics`] for the level-triggered +/// reconciliation and establishment contract. pub async fn run_cache_invalidation_subscriber( redis_url: String, broadcast_tx: broadcast::Sender, + topics: Arc, + desired: DesiredCommunities, ) { - 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..."); - } + run_community_subscriber( + redis_url, + CacheInvalidationFamily { broadcast_tx }, + topics, + desired, + ) + .await; } -async fn connect_and_subscribe( - redis_url: &str, - broadcast_tx: &broadcast::Sender, -) -> Result<(), redis::RedisError> { - let client = redis::Client::open(redis_url)?; - let mut conn = client.get_async_pubsub().await?; +struct CacheInvalidationFamily { + broadcast_tx: broadcast::Sender, +} - conn.psubscribe(CACHE_INVALIDATION_PATTERN).await?; +impl CommunityChannelFamily for CacheInvalidationFamily { + fn channel(&self, community: CommunityId) -> String { + cache_invalidation_channel_for(community) + } - tracing::info!( - "Redis cache-invalidation subscriber connected — listening on {CACHE_INVALIDATION_PATTERN}" - ); + fn parse_channel(&self, channel: &str) -> Option { + parse_cache_invalidation_channel(channel) + } - let mut stream = conn.on_message(); - while let Some(msg) = stream.next().await { - let channel = msg.get_channel_name(); - let Some(community_id) = parse_cache_invalidation_channel(channel) else { - tracing::warn!("Received cache-invalidation message on unexpected channel: {channel}"); - continue; - }; - - 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) { + fn deliver(&self, community_id: CommunityId, payload: &str) { + let invalidation: CacheInvalidation = match serde_json::from_str(payload) { Ok(v) => v, Err(e) => { tracing::warn!("Failed to deserialize cache-invalidation message: {e}"); - continue; + return; } }; @@ -167,12 +155,10 @@ async fn connect_and_subscribe( invalidation, }; - if broadcast_tx.send(scoped).is_err() { + if self.broadcast_tx.send(scoped).is_err() { tracing::trace!("No cache-invalidation receivers — message dropped"); } } - - Ok(()) } #[cfg(test)] diff --git a/crates/buzz-pubsub/src/community_topics.rs b/crates/buzz-pubsub/src/community_topics.rs new file mode 100644 index 000000000..33c4b50f0 --- /dev/null +++ b/crates/buzz-pubsub/src/community_topics.rs @@ -0,0 +1,693 @@ +//! Level-triggered subscription state for community-scoped Redis pub/sub +//! channel families (cache invalidation, connection control). +//! +//! ElastiCache Serverless rejects `PSUBSCRIBE`, so these families can no longer +//! ride a single `buzz:*:...` wildcard. Each pod instead subscribes to the exact +//! per-community channels it currently needs. Two sets carry that: +//! +//! * **desired** — owned by whoever holds the interest (live sockets for +//! connection control; local cache residency for cache invalidation), +//! supplied as a closure and recomputed from scratch on every reconcile. This +//! module never caches it, so a queued command can never be applied against a +//! stale view. +//! * **established** — the communities Redis has acknowledged a `SUBSCRIBE` for +//! on the *current* connection. A community enters only after the ack returns, +//! and the whole set is cleared when the connection ends. +//! +//! Reconciliation is level-triggered: the subscriber task diffs desired against +//! established on connect, on a wake, and on a fixed tick. A coalesced or lost +//! wake therefore costs latency, never convergence — and the tick is the +//! automatic restore path that any cleared state needs. A healthy subscription +//! must never sit marked unestablished waiting on an unrelated external event. + +use std::collections::HashSet; +use std::sync::{Arc, RwLock}; +use std::time::Duration; + +use buzz_core::CommunityId; +use futures_util::StreamExt; +use tokio::sync::Notify; + +/// Initial reconnect backoff (1 second). +const BACKOFF_INITIAL_SECS: u64 = 1; +/// Maximum reconnect backoff (30 seconds). +const BACKOFF_MAX_SECS: u64 = 30; +/// Upper bound on how long desired and established can diverge without a wake. +/// This is the unconditional convergence path, so it must never be disabled. +const RECONCILE_INTERVAL: Duration = Duration::from_secs(1); + +/// Communities whose channel this pod wants subscribed, recomputed on demand. +pub type DesiredCommunities = Arc HashSet + Send + Sync>; + +/// Subscription state for one community-scoped channel family, shared between +/// the subscriber task and the interest holders that read it. +pub struct CommunityTopics { + /// Log label, e.g. `cache-invalidation`. + name: &'static str, + /// Communities Redis has acknowledged on the live connection. Written only + /// by the subscriber task, after an ack. + established: RwLock>, + /// Latency shortcut: producers signal new interest instead of waiting for + /// the next tick. Coalescing and loss are both harmless. + wake: Notify, +} + +impl CommunityTopics { + /// Creates empty subscription state for a named channel family. + pub fn new(name: &'static str) -> Self { + Self { + name, + established: RwLock::new(HashSet::new()), + wake: Notify::new(), + } + } + + /// Log label for this channel family. + pub fn name(&self) -> &'static str { + self.name + } + + /// Whether Redis has acknowledged this community's subscription on the + /// current connection. This is the gate for anything that must not become + /// readable before its invalidation topic is established. + pub fn is_established(&self, community: CommunityId) -> bool { + self.established + .read() + .expect("community topics lock poisoned") + .contains(&community) + } + + /// Number of established communities, for the connect log line and for + /// tests asserting how much of a desired set has been acked. + pub fn established_count(&self) -> usize { + self.established + .read() + .expect("community topics lock poisoned") + .len() + } + + /// Ask the subscriber task to reconcile now rather than at the next tick. + pub fn wake(&self) { + self.wake.notify_one(); + } + + /// Waits for a [`Self::wake`]. The wait side of the same signal, public so + /// interest holders in other crates can assert they really nudge the + /// reconciler; the subscriber loop below is its production caller. + /// + /// One permit is stored, so a wake raised before the wait still completes. + pub async fn wake_notified(&self) { + self.wake.notified().await; + } + + /// Marks `community` established, standing in for a Redis ack. + /// + /// Test seam only: in production the subscriber records establishment, and + /// only after `SUBSCRIBE` returns. It is public because the gate's consumers + /// live in another crate and have to be able to exercise both sides of it. + #[doc(hidden)] + pub fn insert_established_for_test(&self, community: CommunityId) { + self.insert_established(community); + } + + fn snapshot(&self) -> HashSet { + self.established + .read() + .expect("community topics lock poisoned") + .clone() + } + + fn insert_established(&self, community: CommunityId) { + self.established + .write() + .expect("community topics lock poisoned") + .insert(community); + } + + fn remove_established(&self, community: CommunityId) { + self.established + .write() + .expect("community topics lock poisoned") + .remove(&community); + } + + fn clear_established(&self) { + self.established + .write() + .expect("community topics lock poisoned") + .clear(); + } +} + +/// The per-family parts of a community-scoped subscriber: channel naming, +/// channel parsing, and payload decode plus local delivery. +pub(crate) trait CommunityChannelFamily: Send + 'static { + /// Exact Redis channel for one community. Never a pattern. + fn channel(&self, community: CommunityId) -> String; + /// Recover the community from a received channel name. + fn parse_channel(&self, channel: &str) -> Option; + /// Decode one payload and hand it to local consumers. + fn deliver(&self, community: CommunityId, payload: &str); +} + +/// Runs a community-scoped subscriber with automatic reconnection. +/// +/// Never returns — spawn it in a background task. Reconnects with exponential +/// backoff (1s → 2s → 4s → … → 30s max), rebuilding the exact subscription set +/// from `desired` on every connect. +pub(crate) async fn run_community_subscriber( + redis_url: String, + family: F, + topics: Arc, + desired: DesiredCommunities, +) { + let name = topics.name(); + let mut backoff_secs = BACKOFF_INITIAL_SECS; + + loop { + let outcome = connect_and_serve(&redis_url, &family, &topics, &desired).await; + // The connection is gone, so nothing is subscribed any more. Clearing + // here (not only on the next connect) is what stops dependent state + // from being treated as protected during the backoff. + topics.clear_established(); + + match outcome { + Ok(()) => { + backoff_secs = BACKOFF_INITIAL_SECS; + tracing::warn!( + "Redis {name} stream ended (clean disconnect) — reconnecting in {backoff_secs}s" + ); + } + Err(e) => { + tracing::error!("Redis {name} error: {e} — reconnecting in {backoff_secs}s"); + } + } + + tokio::time::sleep(Duration::from_secs(backoff_secs)).await; + backoff_secs = (backoff_secs * 2).min(BACKOFF_MAX_SECS); + + tracing::info!("Attempting to reconnect to Redis {name}..."); + } +} + +async fn connect_and_serve( + redis_url: &str, + family: &F, + topics: &CommunityTopics, + desired: &DesiredCommunities, +) -> Result<(), redis::RedisError> { + let name = topics.name(); + let client = redis::Client::open(redis_url)?; + let conn = client.get_async_pubsub().await?; + let (mut sink, mut stream) = conn.split(); + + // A fresh connection has nothing subscribed, whatever the previous one had. + topics.clear_established(); + reconcile(&mut sink, family, topics, desired).await?; + + tracing::info!( + established = topics.established_count(), + "Redis {name} subscriber connected with exact per-community subscriptions" + ); + + let mut ticker = tokio::time::interval(RECONCILE_INTERVAL); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + + loop { + tokio::select! { + _ = topics.wake.notified() => { + reconcile(&mut sink, family, topics, desired).await?; + } + _ = ticker.tick() => { + reconcile(&mut sink, family, topics, desired).await?; + } + msg = stream.next() => { + let Some(msg) = msg else { + // Stream returned None — Redis connection closed. + return Ok(()); + }; + + let channel = msg.get_channel_name(); + let Some(community_id) = family.parse_channel(channel) else { + tracing::warn!("Received {name} message on unexpected channel: {channel}"); + continue; + }; + + let payload: String = match msg.get_payload() { + Ok(p) => p, + Err(e) => { + tracing::warn!("Failed to get {name} payload: {e}"); + continue; + } + }; + + family.deliver(community_id, &payload); + } + } + } +} + +/// Bring the live connection's subscriptions in line with current desire. +/// +/// `desired` is read here, inside the subscriber task, immediately before each +/// command — so an unsubscribe can never be applied against a stale view of +/// interest that has since been renewed. +async fn reconcile( + sink: &mut redis::aio::PubSubSink, + family: &F, + topics: &CommunityTopics, + desired: &DesiredCommunities, +) -> Result<(), redis::RedisError> { + let want = desired(); + let have = topics.snapshot(); + + for community in want.difference(&have) { + // `subscribe` awaits Redis's reply, so this returning `Ok` *is* the + // acknowledgement. Only then does the community count as established. + sink.subscribe(&family.channel(*community)).await?; + topics.insert_established(*community); + } + + for community in have.difference(&want) { + // Withdraw establishment before the unsubscribe, never after: a reader + // that consults `is_established` between the two would otherwise cache + // an entry this connection is about to stop protecting. + topics.remove_established(*community); + sink.unsubscribe(&family.channel(*community)).await?; + } + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use uuid::Uuid; + + fn community(id: u128) -> CommunityId { + CommunityId::from_uuid(Uuid::from_u128(id)) + } + + #[test] + fn nothing_is_established_before_an_ack() { + let topics = CommunityTopics::new("test"); + assert!(!topics.is_established(community(1))); + assert_eq!(topics.established_count(), 0); + } + + #[test] + fn established_is_cleared_wholesale_on_disconnect() { + let topics = CommunityTopics::new("test"); + topics.insert_established(community(1)); + topics.insert_established(community(2)); + assert_eq!(topics.established_count(), 2); + + topics.clear_established(); + + assert!(!topics.is_established(community(1))); + assert!(!topics.is_established(community(2))); + } + + #[test] + fn removing_one_community_leaves_the_others_established() { + let topics = CommunityTopics::new("test"); + topics.insert_established(community(1)); + topics.insert_established(community(2)); + + topics.remove_established(community(1)); + + assert!(!topics.is_established(community(1))); + assert!(topics.is_established(community(2))); + } + + #[tokio::test] + async fn a_wake_raised_before_the_wait_is_not_lost() { + let topics = CommunityTopics::new("test"); + topics.wake(); + + // Notify stores one permit, so a wake that arrives before the subscriber + // reaches its select still reconciles immediately. Convergence does not + // depend on this — the tick covers a lost wake — but latency does. + tokio::time::timeout(Duration::from_millis(50), topics.wake.notified()) + .await + .expect("a wake raised before the wait must still complete"); + } +} + +/// Redis-backed reconciliation tests. +/// +/// These are `#[ignore]`d because they need a live server, and CI selects them +/// explicitly (`--run-ignored all` over `package(buzz-pubsub)`). They route +/// through `test_util::test_redis_url()`, so `REDIS_URL` points them at either +/// standalone Redis or a cluster-mode server; when it is set but unreachable +/// they must **fail**, never self-skip — a suite that quietly passes when the +/// server is missing is exactly the fake-green class this lane is guarding +/// against. +#[cfg(test)] +mod redis_tests { + use super::*; + use crate::cache_invalidation::{ + cache_invalidation_channel_for, run_cache_invalidation_subscriber, CacheInvalidation, + ScopedCacheInvalidation, + }; + use crate::test_util::{await_established, await_unestablished, test_redis_url}; + use std::collections::HashMap; + use std::sync::Mutex as StdMutex; + use tokio::sync::broadcast; + use uuid::Uuid; + + /// A mutable desired set a test can steer, plus the closure over it. + fn steerable() -> (Arc>>, DesiredCommunities) { + let cell = Arc::new(StdMutex::new(HashSet::new())); + let reader = Arc::clone(&cell); + ( + cell, + Arc::new(move || reader.lock().expect("desired lock poisoned").clone()), + ) + } + + fn set_desired(cell: &Arc>>, want: &[CommunityId]) { + *cell.lock().expect("desired lock poisoned") = want.iter().copied().collect(); + } + + /// Distinct communities per test run: these tests share one Redis, and a + /// fixed id would let a previous run's channel traffic bleed in. + fn fresh(n: usize) -> Vec { + (0..n) + .map(|_| CommunityId::from_uuid(Uuid::new_v4())) + .collect() + } + + /// A plain connection for issuing commands. Panics if the server named by + /// `REDIS_URL` is unreachable — these tests must fail loudly rather than + /// self-skip when their dependency is missing. + async fn control_conn() -> redis::aio::MultiplexedConnection { + redis::Client::open(test_redis_url()) + .expect("redis client") + .get_multiplexed_async_connection() + .await + .expect("REDIS_URL must be reachable for the ignored Redis suite") + } + + /// Publishes `invalidation` on `community`'s exact channel. + /// + /// The return of `PUBLISH` is deliberately ignored: it counts *pattern* + /// matches as well as exact ones, so any unrelated `PSUBSCRIBE buzz:*` + /// client on the same server inflates it. Subscription state is asserted + /// with [`exact_subscribers`] instead, and delivery by actually receiving. + async fn publish(community: CommunityId, invalidation: &CacheInvalidation) { + let _: i64 = redis::cmd("PUBLISH") + .arg(cache_invalidation_channel_for(community)) + .arg(serde_json::to_string(invalidation).expect("serialize")) + .query_async(&mut control_conn().await) + .await + .expect("PUBLISH"); + } + + /// Exact (non-pattern) subscriber count for `community`'s channel. + /// + /// `PUBSUB NUMSUB` ignores pattern subscribers, which is exactly the + /// distinction this lane turns on: it proves an exact `SUBSCRIBE` is in + /// place, and cannot be satisfied by a leftover wildcard. Combined with a + /// per-run random community id, no other client on a shared server can + /// contribute to this count. + async fn exact_subscribers(community: CommunityId) -> i64 { + let channel = cache_invalidation_channel_for(community); + let (_, count): (String, i64) = redis::cmd("PUBSUB") + .arg("NUMSUB") + .arg(&channel) + .query_async(&mut control_conn().await) + .await + .expect("PUBSUB NUMSUB"); + count + } + + /// Client ids that currently hold `n` exact subscriptions and no patterns. + /// + /// Used to single out one subscriber's own connection for a targeted sever. + /// The tests in this module run in parallel inside one binary, so several + /// clients have exact subscriptions at any moment; the caller disambiguates + /// by diffing this before and after its own subscriber connects, and by + /// choosing an `n` no other test in the crate uses. + async fn clients_with_exact_subs( + conn: &mut redis::aio::MultiplexedConnection, + n: usize, + ) -> HashSet { + let list: String = redis::cmd("CLIENT") + .arg("LIST") + .query_async(conn) + .await + .expect("CLIENT LIST"); + list.lines() + .filter(|line| line.contains(" psub=0 ") && line.contains(&format!(" sub={n} "))) + .filter_map(|line| { + line.split_whitespace() + .find_map(|field| field.strip_prefix("id=")) + .and_then(|id| id.parse().ok()) + }) + .collect() + } + + fn membership(seed: u128) -> CacheInvalidation { + CacheInvalidation::Membership { + channel_id: Uuid::from_u128(seed), + pubkey: vec![(seed & 0xff) as u8; 32], + } + } + + /// Spawns a cache-invalidation subscriber over `desired` and returns its + /// topics handle plus a receiver of delivered invalidations. + fn spawn_subscriber( + desired: DesiredCommunities, + ) -> ( + Arc, + broadcast::Receiver, + ) { + let topics = Arc::new(CommunityTopics::new( + crate::cache_invalidation::CACHE_INVALIDATION_NAME, + )); + let (tx, rx) = broadcast::channel(4096); + let spawn_topics = Arc::clone(&topics); + tokio::spawn(async move { + run_cache_invalidation_subscriber(test_redis_url(), tx, spawn_topics, desired).await + }); + (topics, rx) + } + + /// The core replacement claim: 100 exact `SUBSCRIBE`s on ONE RESP2 + /// connection route every `(channel, payload)` pair byte-exactly to the + /// community it was published on. + /// + /// This is what `PSUBSCRIBE buzz:*:cache-invalidate` used to do in one + /// command. A wildcard cannot be used on ElastiCache Serverless, so the + /// property that has to hold now is that N exact subscriptions on a single + /// connection are equivalent — including that no message is delivered under + /// the wrong community label, which is the failure that would silently + /// cross-wire tenants. + #[tokio::test] + #[ignore = "requires Redis"] + async fn hundred_exact_subscriptions_route_every_payload_to_its_own_community() { + let communities = fresh(100); + let (cell, desired) = steerable(); + set_desired(&cell, &communities); + let (topics, mut rx) = spawn_subscriber(desired); + await_established(&topics, &communities, "100 exact subscriptions").await; + assert_eq!(topics.established_count(), 100); + + // One distinguishable payload per community, all on one connection. + let mut expected = HashMap::new(); + for (i, community) in communities.iter().enumerate() { + let invalidation = membership(0x1000 + i as u128); + assert_eq!( + exact_subscribers(*community).await, + 1, + "exactly one exact SUBSCRIBE should hold {community}'s channel" + ); + publish(*community, &invalidation).await; + expected.insert(*community, invalidation); + } + + let mut seen: HashMap = HashMap::new(); + while seen.len() < communities.len() { + let scoped = tokio::time::timeout(Duration::from_secs(10), rx.recv()) + .await + .expect("timed out before all 100 payloads arrived") + .expect("broadcast closed"); + assert!( + seen.insert(scoped.community_id, scoped.invalidation) + .is_none(), + "community {} delivered twice", + scoped.community_id + ); + } + assert_eq!( + seen, expected, + "payload must arrive under its own community" + ); + } + + /// Reconciliation is level-triggered in BOTH directions: growing the + /// desired set subscribes, shrinking it unsubscribes, and the communities + /// that stayed desired keep working across the change. + /// + /// The `subscriber_count == 0` assertion is the real teeth. Establishment + /// bookkeeping saying "unsubscribed" proves nothing on its own; Redis + /// reporting no subscriber for the channel proves the `UNSUBSCRIBE` was + /// actually issued. + #[tokio::test] + #[ignore = "requires Redis"] + async fn unsubscribing_a_subset_leaves_the_rest_routing() { + let c = fresh(3); + let (cell, desired) = steerable(); + set_desired(&cell, &c); + let (topics, mut rx) = spawn_subscriber(desired); + await_established(&topics, &c, "initial three").await; + + // Drop the middle one. No explicit command: only the desired set moves. + set_desired(&cell, &[c[0], c[2]]); + topics.wake(); + await_unestablished(&topics, c[1], "the withdrawn community").await; + + assert_eq!( + exact_subscribers(c[1]).await, + 0, + "Redis must report no exact subscriber for the unsubscribed channel" + ); + assert!(topics.is_established(c[0]) && topics.is_established(c[2])); + + // The retained communities still route, and the withdrawn one's publish + // does not arrive — checked by identity, not by absence of traffic. + let kept = membership(0x2002); + assert_eq!(exact_subscribers(c[2]).await, 1); + publish(c[2], &kept).await; + let scoped = tokio::time::timeout(Duration::from_secs(5), rx.recv()) + .await + .expect("timed out on retained community") + .expect("broadcast closed"); + assert_eq!(scoped.community_id, c[2]); + assert_eq!(scoped.invalidation, kept); + + // And re-adding it converges again with no command, closing the loop. + set_desired(&cell, &c); + topics.wake(); + await_established(&topics, &c, "re-added community").await; + assert_eq!(exact_subscribers(c[1]).await, 1); + } + + /// Convergence must not depend on the wake: with wakes deliberately never + /// issued, the periodic reconcile alone has to subscribe a newly desired + /// community. + /// + /// This is the restore-path property. A cleared or missing subscription + /// that only recovers when some external event fires is the bug class I want + /// structurally excluded, so the tick is asserted in isolation. + #[tokio::test] + #[ignore = "requires Redis"] + async fn the_periodic_reconcile_converges_without_any_wake() { + let c = fresh(1); + let (cell, desired) = steerable(); + let (topics, _rx) = spawn_subscriber(desired); + // Subscriber is up with an empty desired set. + await_established(&topics, &[], "empty start").await; + + set_desired(&cell, &c); + // Deliberately no `topics.wake()`. + await_established(&topics, &c, "tick-only convergence").await; + assert_eq!(exact_subscribers(c[0]).await, 1); + } + + /// Sever the connection under the subscriber and it must rebuild the whole + /// desired set on reconnect — and, before that, report nothing established, + /// so the cache gate cannot treat entries as protected during the gap. + /// + /// `CLIENT KILL` is the sever: it is a real server-side disconnect, not a + /// simulated error return, so the reconnect path runs for the same reason it + /// would in production. + #[tokio::test] + #[ignore = "requires Redis"] + async fn a_severed_connection_rebuilds_the_entire_desired_set() { + // Five communities: a subscription count no other test in this crate + // uses, so this subscriber's connection is identifiable in CLIENT LIST + // even with the rest of the suite running in parallel. + const SEVER_TEST_COMMUNITIES: usize = 5; + let c = fresh(SEVER_TEST_COMMUNITIES); + let mut conn = control_conn().await; + let before = clients_with_exact_subs(&mut conn, SEVER_TEST_COMMUNITIES).await; + + let (cell, desired) = steerable(); + set_desired(&cell, &c); + let (topics, _rx) = spawn_subscriber(desired); + await_established(&topics, &c, "pre-sever").await; + + // Sever ONLY this subscriber, by id. `CLIENT KILL TYPE pubsub` would + // also kill every other pub/sub client on the server — a concurrent + // test here, or an unrelated process sharing the dev Redis. + let after = clients_with_exact_subs(&mut conn, SEVER_TEST_COMMUNITIES).await; + let new_clients: Vec = after.difference(&before).copied().collect(); + assert_eq!( + new_clients.len(), + 1, + "expected exactly one new {SEVER_TEST_COMMUNITIES}-subscription client to be ours, got {new_clients:?}" + ); + let killed: i64 = redis::cmd("CLIENT") + .arg("KILL") + .arg("ID") + .arg(new_clients[0]) + .query_async(&mut conn) + .await + .expect("CLIENT KILL ID"); + assert_eq!(killed, 1, "expected to sever exactly our own subscriber"); + + // The sever must be observable as lost establishment, not papered over: + // the gate has to read closed while the connection is gone. + await_unestablished(&topics, c[0], "establishment during the reconnect gap").await; + + // Rebuilt from `desired`, not from a remembered active set: all five are + // back, and each channel really has an exact subscriber again. + await_established(&topics, &c, "post-sever rebuild").await; + for community in &c { + assert_eq!( + exact_subscribers(*community).await, + 1, + "channel for {community} must be re-subscribed after the sever" + ); + } + } + + /// The establishment gate's precondition: while nothing is established, + /// `is_established` is false for a community the pod desires, and it flips + /// to true only once Redis has acked. + /// + /// This is the stale-authz window. The relay consults exactly this to decide + /// whether caching an authorization decision is safe, so "unestablished + /// while disconnected" has to be observable rather than merely intended. + #[tokio::test] + #[ignore = "requires Redis"] + async fn a_desired_community_is_unestablished_until_redis_acks() { + let c = fresh(1); + let (cell, desired) = steerable(); + set_desired(&cell, &c); + + let topics = Arc::new(CommunityTopics::new( + crate::cache_invalidation::CACHE_INVALIDATION_NAME, + )); + // Desired, but no subscriber running: nothing can have been acked, so + // the gate must read closed even though interest exists. + assert!( + !topics.is_established(c[0]), + "desire alone must not open the cache gate" + ); + + let (tx, _rx) = broadcast::channel(16); + let spawn_topics = Arc::clone(&topics); + tokio::spawn(async move { + run_cache_invalidation_subscriber(test_redis_url(), tx, spawn_topics, desired).await + }); + await_established(&topics, &c, "gate opens on ack").await; + + // And it closes again for a community that leaves the desired set. + set_desired(&cell, &[]); + topics.wake(); + await_unestablished(&topics, c[0], "gate closes on withdrawal").await; + } +} diff --git a/crates/buzz-pubsub/src/conn_control.rs b/crates/buzz-pubsub/src/conn_control.rs index bc177cff1..d42fc1b3f 100644 --- a/crates/buzz-pubsub/src/conn_control.rs +++ b/crates/buzz-pubsub/src/conn_control.rs @@ -14,24 +14,36 @@ //! The DB ban row remains the durable backstop: even if a disconnect message is //! dropped, the next auth attempt is refused at the auth seam. +use std::sync::Arc; + use buzz_core::{CommunityId, TenantContext}; -use futures_util::StreamExt; use serde::{Deserialize, Serialize}; use tokio::sync::broadcast; use uuid::Uuid; +use crate::community_topics::{ + run_community_subscriber, CommunityChannelFamily, CommunityTopics, DesiredCommunities, +}; use crate::topic::BUZZ_PREFIX; /// Tenant-local Redis pub/sub channel suffix for connection-control messages. pub const CONN_CONTROL_SUFFIX: &str = "conn-control"; -/// Pattern the subscriber uses to receive connection-control messages for every -/// community this pod may hold connections for. -pub const CONN_CONTROL_PATTERN: &str = "buzz:*:conn-control"; +/// Log label for the connection-control subscriber. +pub(crate) const CONN_CONTROL_NAME: &str = "conn-control"; /// Redis pub/sub channel for connection-control messages under `ctx`. pub fn conn_control_channel(ctx: &TenantContext) -> String { - format!("{BUZZ_PREFIX}:{}:{CONN_CONTROL_SUFFIX}", ctx.community()) + conn_control_channel_for(ctx.community()) +} + +/// Redis pub/sub channel for connection-control messages in `community`. +/// +/// The subscriber needs this without a [`TenantContext`]: it subscribes to the +/// exact channels of the communities holding live sockets on this pod, which are +/// known only as ids. +pub fn conn_control_channel_for(community: CommunityId) -> String { + format!("{BUZZ_PREFIX}:{community}:{CONN_CONTROL_SUFFIX}") } /// Parse a connection-control Redis channel into its scoped community id. @@ -79,72 +91,47 @@ pub struct ScopedConnControl { pub command: ConnControl, } -/// 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:*:conn-control` and forwards scoped commands to the -/// broadcast. Mirrors [`crate::cache_invalidation::run_cache_invalidation_subscriber`]: -/// a reconnect loop with exponential backoff. Never returns. +/// Subscribes to the exact `buzz:{community}:conn-control` channels for the +/// communities holding live sockets on this pod and forwards scoped commands to +/// the broadcast. +/// +/// `desired` is re-read on every reconcile, so a community that gains or loses +/// its last socket converges without any explicit command. Never returns. See +/// [`crate::community_topics`] for the reconciliation contract. pub async fn run_conn_control_subscriber( redis_url: String, broadcast_tx: broadcast::Sender, + topics: Arc, + desired: DesiredCommunities, ) { - 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 conn-control stream ended (clean disconnect) — reconnecting in {backoff_secs}s" - ); - } - Err(e) => { - tracing::error!("Redis conn-control 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 conn-control..."); - } + run_community_subscriber( + redis_url, + ConnControlFamily { broadcast_tx }, + topics, + desired, + ) + .await; } -async fn connect_and_subscribe( - redis_url: &str, - broadcast_tx: &broadcast::Sender, -) -> Result<(), redis::RedisError> { - let client = redis::Client::open(redis_url)?; - let mut conn = client.get_async_pubsub().await?; +struct ConnControlFamily { + broadcast_tx: broadcast::Sender, +} - conn.psubscribe(CONN_CONTROL_PATTERN).await?; +impl CommunityChannelFamily for ConnControlFamily { + fn channel(&self, community: CommunityId) -> String { + conn_control_channel_for(community) + } - tracing::info!("Redis conn-control subscriber connected — listening on {CONN_CONTROL_PATTERN}"); + fn parse_channel(&self, channel: &str) -> Option { + parse_conn_control_channel(channel) + } - let mut stream = conn.on_message(); - while let Some(msg) = stream.next().await { - let channel = msg.get_channel_name(); - let Some(community_id) = parse_conn_control_channel(channel) else { - tracing::warn!("Received conn-control message on unexpected channel: {channel}"); - continue; - }; - - let payload: String = match msg.get_payload() { - Ok(p) => p, - Err(e) => { - tracing::warn!("Failed to get conn-control payload: {e}"); - continue; - } - }; - - let command: ConnControl = match serde_json::from_str(&payload) { + fn deliver(&self, community_id: CommunityId, payload: &str) { + let command: ConnControl = match serde_json::from_str(payload) { Ok(v) => v, Err(e) => { tracing::warn!("Failed to deserialize conn-control message: {e}"); - continue; + return; } }; @@ -153,12 +140,10 @@ async fn connect_and_subscribe( command, }; - if broadcast_tx.send(scoped).is_err() { + if self.broadcast_tx.send(scoped).is_err() { tracing::trace!("No conn-control receivers — message dropped"); } } - - Ok(()) } #[cfg(test)] diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index 0cdd81945..ab2b7f660 100644 --- a/crates/buzz-pubsub/src/lib.rs +++ b/crates/buzz-pubsub/src/lib.rs @@ -23,6 +23,8 @@ /// Cross-pod cache-key invalidation over Redis pub/sub. pub mod cache_invalidation; +/// Level-triggered exact subscriptions for community-scoped channel families. +pub mod community_topics; /// Cross-pod connection-control commands over Redis pub/sub. pub mod conn_control; /// Error types for pub/sub operations. @@ -54,6 +56,7 @@ use tokio::sync::{broadcast, mpsc, Mutex}; use crate::cache_invalidation::{ cache_invalidation_channel, CacheInvalidation, ScopedCacheInvalidation, }; +use crate::community_topics::{CommunityTopics, DesiredCommunities}; use crate::conn_control::{conn_control_channel, ConnControl, ScopedConnControl}; pub use crate::topic::{channel_key, global_key, EventTopic, EventTopicKey}; @@ -110,6 +113,10 @@ pub struct PubSubManager { broadcast_tx: broadcast::Sender, cache_invalidation_tx: broadcast::Sender, conn_control_tx: broadcast::Sender, + /// Communities whose cache-invalidation channel Redis has acknowledged. + cache_invalidation_topics: Arc, + /// Communities whose connection-control channel Redis has acknowledged. + conn_control_topics: Arc, } impl PubSubManager { @@ -138,6 +145,10 @@ impl PubSubManager { broadcast_tx, cache_invalidation_tx, conn_control_tx, + cache_invalidation_topics: Arc::new(CommunityTopics::new( + cache_invalidation::CACHE_INVALIDATION_NAME, + )), + conn_control_topics: Arc::new(CommunityTopics::new(conn_control::CONN_CONTROL_NAME)), }) } @@ -162,24 +173,52 @@ impl PubSubManager { /// 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) { + /// + /// `desired` returns the communities whose cache entries this pod may still + /// serve. It is re-read on every reconcile, never cached, so residency that + /// appears or lapses converges without any explicit command. + pub async fn run_cache_invalidation_subscriber(self: Arc, desired: DesiredCommunities) { cache_invalidation::run_cache_invalidation_subscriber( self.redis_url.clone(), self.cache_invalidation_tx.clone(), + self.cache_invalidation_topics.clone(), + desired, ) .await; } /// Starts the connection-control subscriber loop with automatic /// reconnection. Runs forever — spawn this in a background task. - pub async fn run_conn_control_subscriber(self: Arc) { + /// + /// `desired` returns the communities holding live sockets on this pod. + pub async fn run_conn_control_subscriber(self: Arc, desired: DesiredCommunities) { conn_control::run_conn_control_subscriber( self.redis_url.clone(), self.conn_control_tx.clone(), + self.conn_control_topics.clone(), + desired, ) .await; } + /// Subscription state for the cache-invalidation family. + /// + /// The relay consults this before caching authorization for a community: + /// an entry inserted while the community's invalidation channel is not + /// established could not be dropped by a remote invalidation. + pub fn cache_invalidation_topics(&self) -> &Arc { + &self.cache_invalidation_topics + } + + /// Subscription state for the connection-control family. + /// + /// The relay's socket registry holds this to wake the reconciler when a new + /// community gains its first connection, so the subscribe does not wait for + /// the next tick. + pub fn conn_control_topics(&self) -> &Arc { + &self.conn_control_topics + } + /// Returns a new broadcast receiver for locally-published channel events. pub fn subscribe_local(&self) -> broadcast::Receiver { self.broadcast_tx.subscribe() @@ -386,12 +425,66 @@ pub(crate) mod test_util { cfg.create_pool(Some(deadpool_redis::Runtime::Tokio1)) .expect("Failed to create Redis pool") } + + /// A fixed desired-community set, for tests that do not exercise churn. + pub fn fixed_desired( + communities: impl IntoIterator, + ) -> crate::community_topics::DesiredCommunities { + let set: std::collections::HashSet<_> = communities.into_iter().collect(); + std::sync::Arc::new(move || set.clone()) + } + + /// Waits until `topics` reports every community in `expected` established, + /// then returns. Panics on timeout. + /// + /// Every Redis-backed test here waits on the acknowledgement rather than + /// sleeping a guessed interval: a fixed sleep either flakes under load or + /// passes vacuously when the subscribe silently never happened. + pub async fn await_established( + topics: &crate::community_topics::CommunityTopics, + expected: &[buzz_core::CommunityId], + what: &str, + ) { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + if expected.iter().all(|c| topics.is_established(*c)) { + return; + } + assert!( + std::time::Instant::now() < deadline, + "timed out waiting for {what} to be established \ + ({} of {} acked)", + topics.established_count(), + expected.len() + ); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + } + + /// Waits until `topics` reports `community` NOT established. Panics on timeout. + pub async fn await_unestablished( + topics: &crate::community_topics::CommunityTopics, + community: buzz_core::CommunityId, + what: &str, + ) { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + if !topics.is_established(community) { + return; + } + assert!( + std::time::Instant::now() < deadline, + "timed out waiting for {what} to be unestablished" + ); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + } } #[cfg(test)] mod tests { use super::*; - use crate::test_util::{make_test_pool, test_redis_url}; + use crate::test_util::{await_established, fixed_desired, make_test_pool, test_redis_url}; use buzz_core::{CommunityId, TenantContext}; use nostr::{EventBuilder, Keys, Kind}; use uuid::Uuid; @@ -452,10 +545,24 @@ mod tests { async fn test_cache_invalidation_roundtrip() { let manager = make_manager().await; let mut rx = manager.subscribe_cache_invalidations(); + let ctx = ctx(0xaaaa, "a.example"); 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 desired = fixed_desired([ctx.community()]); + tokio::spawn(async move { + manager_clone + .run_cache_invalidation_subscriber(desired) + .await + }); + // Wait for the exact SUBSCRIBE to be acked rather than sleeping: with + // per-community channels, publishing before the ack is a lost message, + // not a late one. + await_established( + manager.cache_invalidation_topics(), + &[ctx.community()], + "cache-invalidation roundtrip", + ) + .await; let channel_id = Uuid::new_v4(); let pubkey = Keys::generate().public_key().to_bytes().to_vec(); @@ -464,8 +571,6 @@ mod tests { pubkey: pubkey.clone(), }; - let ctx = ctx(0xaaaa, "a.example"); - manager .publish_cache_invalidation(&ctx, &sent) .await diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 799cf9cf6..ab9e56e10 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -380,17 +380,9 @@ 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 }); - - // Spawn Redis pub/sub subscriber for cross-pod connection-control commands. - // Bans recorded on other pods are received here and applied to any local - // sockets (via the consumer loop below), enforcing live disconnect fan-out. - let pubsub_for_conn_ctrl = Arc::clone(&pubsub); - tokio::spawn(async move { pubsub_for_conn_ctrl.run_conn_control_subscriber().await }); + // The two community-scoped subscribers (cache invalidation, connection + // control) are spawned later, alongside their consumer loops: each needs a + // desired-community closure read off `AppState`, which does not exist yet. let auth = AuthService::new(config.auth.clone()); @@ -875,6 +867,21 @@ async fn main() -> anyhow::Result<()> { // changes) and apply the matching local moka drop. Uses the `*_local` drop // variants so a received drop is never re-published. { + // Exact per-community subscriptions, driven by cache residency: this pod + // must keep hearing a community's invalidations for as long as it can + // still serve a cached authorization decision under it. Reading the + // residency set through a closure (rather than snapshotting it) is what + // makes the subscriber level-triggered — see `buzz_pubsub::community_topics`. + let residency = Arc::clone(&state.cache_residency); + let pubsub_for_cache = Arc::clone(&state.pubsub); + tokio::spawn(async move { + pubsub_for_cache + .run_cache_invalidation_subscriber(Arc::new(move || { + residency.resident_communities() + })) + .await + }); + let state_for_cache = Arc::clone(&state); let mut rx = state_for_cache.pubsub.subscribe_cache_invalidations(); tokio::spawn(async move { @@ -926,6 +933,17 @@ async fn main() -> anyhow::Result<()> { // ban row is the durable backstop; even a dropped command still refuses the // banned member's next auth attempt at the auth seam. { + // Exact per-community subscriptions, driven by connection ownership: + // this pod only needs a community's disconnect commands while it holds + // a socket a remote ban could have to close. + let registry = Arc::clone(&state.community_connections); + let pubsub_for_conn_ctrl = Arc::clone(&state.pubsub); + tokio::spawn(async move { + pubsub_for_conn_ctrl + .run_conn_control_subscriber(Arc::new(move || registry.bound_communities())) + .await + }); + let state_for_conn_ctrl = Arc::clone(&state); let mut rx = state_for_conn_ctrl.pubsub.subscribe_conn_control(); tokio::spawn(async move { diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 58a869a99..0baf1ec29 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -4,7 +4,7 @@ use std::collections::{HashMap, HashSet}; use std::future::Future; use std::sync::atomic::{AtomicBool, AtomicU64, AtomicU8, Ordering}; use std::sync::Arc; -use std::time::Instant; +use std::time::{Duration, Instant}; use axum::body::Bytes; use axum::extract::ws::{Message as WsMessage, Utf8Bytes as WsUtf8Bytes}; @@ -22,6 +22,7 @@ use buzz_core::CommunityId; use buzz_db::Db; use buzz_media::MediaStorage; use buzz_pubsub::cache_invalidation::CacheInvalidation; +use buzz_pubsub::community_topics::CommunityTopics; use buzz_pubsub::conn_control::ConnControl; use buzz_pubsub::rate_limiter::RedisRateLimiter; use buzz_pubsub::{PubSubManager, RedisNip98ReplayGuard}; @@ -57,30 +58,43 @@ struct ConnEntry { grace_limit: u8, } +/// Lifetime of every cached authorization decision (membership, accessible +/// channels, channel visibility). +/// +/// [`CacheResidency`] is driven off this same value: residency must outlast the +/// last entry that can still be served, so the two cannot be allowed to drift. +pub const AUTHZ_CACHE_TTL: Duration = Duration::from_secs(10); + /// Community-scoped lifecycle registry shared by every long-lived socket type. /// /// A handler registers before durable active-state revalidation. Archival after /// registration cancels the token; archival before registration is observed by /// the revalidation. The returned guard removes the entry on every exit path. +/// +/// This is also the desired set for the connection-control subscriber: a pod +/// only needs `buzz:{community}:conn-control` while it holds a socket that a +/// remote ban could have to close. See [`Self::bound_communities`]. pub struct CommunityConnectionRegistry { connections: Arc>, -} - -impl Default for CommunityConnectionRegistry { - fn default() -> Self { - Self::new() - } + /// Connection-control subscription state, nudged on every registration so + /// a new socket's community is subscribed without waiting for the tick. + conn_control_topics: Arc, } impl CommunityConnectionRegistry { - /// Creates an empty lifecycle registry. - pub fn new() -> Self { + /// Creates an empty lifecycle registry feeding `conn_control_topics`. + pub fn new(conn_control_topics: Arc) -> Self { Self { connections: Arc::new(DashMap::new()), + conn_control_topics, } } /// Registers one socket and returns a guard that deregisters it on drop. + /// + /// The wake only shortens the subscribe latency; the subscriber's periodic + /// reconcile is what guarantees the community is eventually subscribed, so + /// a lost or coalesced wake cannot strand this connection unprotected. pub fn register( &self, connection_id: Uuid, @@ -89,6 +103,7 @@ impl CommunityConnectionRegistry { ) -> CommunityConnectionGuard { self.connections .insert(connection_id, (community_id, cancel)); + self.conn_control_topics.wake(); CommunityConnectionGuard { connection_id, connections: Arc::clone(&self.connections), @@ -128,6 +143,82 @@ impl Drop for CommunityConnectionGuard { } } +/// Communities whose authorization caches this pod may still serve, and the +/// desired set for the cache-invalidation subscriber. +/// +/// Cached authorization outlives the socket that populated it, so residency — +/// not connection lifetime — is what decides whether this pod must keep hearing +/// `buzz:{community}:cache-invalidate`. A community becomes resident when this +/// pod first resolves authorization under it and stays resident until `ttl` +/// after the most recent resolution, which covers the whole window in which a +/// cached entry could still be read. +/// +/// Residency lapses on a timer rather than on an event, so no unsubscribe is +/// owed by any call path: a community that stops being read simply ages out. +pub struct CacheResidency { + /// Community → deadline after which no cached entry can still be served. + resident: Arc>, + /// Cache-invalidation subscription state. Woken on a newly resident + /// community; read by the establishment gate before a cache insert. + topics: Arc, + /// Cache entry lifetime. Must be >= the moka TTL of every gated cache, or + /// residency would expire while a readable entry survives. + ttl: Duration, +} + +impl CacheResidency { + /// Creates an empty residency tracker for caches living `ttl`. + pub fn new(topics: Arc, ttl: Duration) -> Self { + Self { + resident: Arc::new(DashMap::new()), + topics, + ttl, + } + } + + /// Records that this pod is resolving authorization for `community` and + /// reports whether the result may be cached. + /// + /// Recording and gating are one call on purpose. Residency *is* the desired + /// set, so a community that is only gated and never recorded would never be + /// subscribed, never become established, and never pass the gate — the + /// authorization caches would stay off for the life of the pod. Folding the + /// two makes that ordering unrepresentable. + /// + /// The first call for a community therefore returns `false` (nothing is + /// acked yet) while still asking for the subscription; once the reconciler + /// establishes it, subsequent calls return `true`. A `false` means the pod + /// reads through to the DB — degraded, not stale. + /// + /// Gating **inserts only** is deliberate: gating reads would discard a live, + /// still-invalidatable cache on every reconnect and turn a pub/sub blip into + /// a DB stampede. + #[must_use] + pub fn admit(&self, community: CommunityId) -> bool { + let deadline = Instant::now() + self.ttl; + if self.resident.insert(community, deadline).is_none() { + // Newly resident: shorten the wait for the subscribe. Convergence + // does not depend on this — the reconcile tick is the guarantee. + self.topics.wake(); + } + let admitted = self.topics.is_established(community); + if !admitted { + metrics::counter!("buzz_authz_cache_insert_skipped_total").increment(1); + } + admitted + } + + /// Communities this pod may still serve a cached entry for. Expired entries + /// are dropped here, so this is also the tracker's only reclamation path — + /// and it is driven by the subscriber's own periodic reconcile, never by an + /// external event. + pub fn resident_communities(&self) -> HashSet { + let now = Instant::now(); + self.resident.retain(|_, deadline| *deadline > now); + self.resident.iter().map(|entry| *entry.key()).collect() + } +} + /// Registers a socket, durably revalidates its community, then runs it. /// /// The ordering is the archival admission invariant: archive-before-query is @@ -506,6 +597,10 @@ pub struct AppState { pub conn_manager: Arc, /// Lifecycle cancellation for every long-lived socket, including huddle audio. pub community_connections: Arc, + /// Communities whose authorization caches this pod may still serve — the + /// desired set for the cache-invalidation subscriber, and the gate the + /// `*_cached` readers consult before caching an authorization decision. + pub cache_residency: Arc, /// Stops only the periodic lifecycle revalidator during graceful shutdown. pub community_revalidator_cancel: CancellationToken, /// Test/telemetry counter for archive disconnect publication attempts. @@ -712,6 +807,8 @@ impl AppState { Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone())); let admission_rate_limiter = Arc::new(RedisRateLimiter::new(redis_pool.clone())); let audit_enabled = audit_arc.is_some(); + let conn_control_topics = Arc::clone(pubsub.conn_control_topics()); + let cache_invalidation_topics = Arc::clone(pubsub.cache_invalidation_topics()); let state = Self { config: Arc::new(config), db, @@ -722,7 +819,11 @@ impl AppState { search: search_arc, sub_registry: Arc::new(SubscriptionRegistry::new()), conn_manager: Arc::new(ConnectionManager::new()), - community_connections: Arc::new(CommunityConnectionRegistry::new()), + community_connections: Arc::new(CommunityConnectionRegistry::new(conn_control_topics)), + cache_residency: Arc::new(CacheResidency::new( + cache_invalidation_topics, + AUTHZ_CACHE_TTL, + )), community_revalidator_cancel: CancellationToken::new(), community_disconnect_publish_attempts: Arc::new(AtomicU64::new(0)), conn_semaphore: Arc::new(Semaphore::new(max_connections)), @@ -741,21 +842,21 @@ impl AppState { membership_cache: Arc::new( moka::sync::Cache::builder() .max_capacity(10_000) - .time_to_live(std::time::Duration::from_secs(10)) + .time_to_live(AUTHZ_CACHE_TTL) .support_invalidation_closures() .build(), ), accessible_channels_cache: Arc::new( moka::sync::Cache::builder() .max_capacity(10_000) - .time_to_live(std::time::Duration::from_secs(10)) + .time_to_live(AUTHZ_CACHE_TTL) .support_invalidation_closures() .build(), ), channel_visibility_cache: Arc::new( moka::sync::Cache::builder() .max_capacity(10_000) - .time_to_live(std::time::Duration::from_secs(10)) + .time_to_live(AUTHZ_CACHE_TTL) .support_invalidation_closures() .build(), ), @@ -825,6 +926,11 @@ impl AppState { } /// Check channel membership with a 10-second cache. Falls back to DB on miss. + /// + /// The result is cached only while this pod's cache-invalidation channel + /// for `community_id` is established — see [`CacheResidency::may_cache`]. + /// Ungated, a pod could cache an authorization decision it would never hear + /// the invalidation for, and serve it for the whole TTL. pub async fn is_member_cached( &self, community_id: CommunityId, @@ -838,7 +944,9 @@ impl AppState { } metrics::counter!("buzz_membership_cache_misses_total").increment(1); let result = self.db.is_member(community_id, channel_id, pubkey).await?; - self.membership_cache.insert(key, result); + if self.cache_residency.admit(community_id) { + self.membership_cache.insert(key, result); + } Ok(result) } @@ -1102,7 +1210,9 @@ impl AppState { .db .get_accessible_channel_ids(community_id, pubkey) .await?; - self.accessible_channels_cache.insert(key, result.clone()); + if self.cache_residency.admit(community_id) { + self.accessible_channels_cache.insert(key, result.clone()); + } Ok(result) } @@ -1144,8 +1254,16 @@ impl AppState { } }; if visibility == "private" { - self.channel_visibility_cache - .insert((community_id, channel_id), visibility.clone()); + // Gated like the other two caches: the residency set must cover + // every cache `apply_cache_invalidation` can drop, or a community + // could hold an entry no remote invalidation reaches. Here that + // costs liveness rather than confidentiality — a stale `private` + // is over-restrictive, not a leak — but a private->open flip going + // unheard for the full TTL is still a real delivery bug. + if self.cache_residency.admit(community_id) { + self.channel_visibility_cache + .insert((community_id, channel_id), visibility.clone()); + } } Ok(visibility) } @@ -1255,6 +1373,12 @@ mod tests { (mgr, conn_id, rx, ctrl_rx, cancel, bp) } + /// A lifecycle registry whose conn-control wakes go nowhere. These tests + /// assert registry bookkeeping, not subscription reconciliation. + fn test_registry() -> CommunityConnectionRegistry { + CommunityConnectionRegistry::new(Arc::new(CommunityTopics::new("test"))) + } + async fn test_state() -> Arc { let mut config = crate::config::Config::from_env().expect("default config loads"); config.require_relay_membership = false; @@ -1573,7 +1697,7 @@ mod tests { #[test] fn community_lifecycle_disconnect_covers_socket_types_and_preserves_tenant_fence() { - let registry = CommunityConnectionRegistry::new(); + let registry = test_registry(); let community_a = CommunityId::from_uuid(Uuid::from_u128(0xa)); let community_b = CommunityId::from_uuid(Uuid::from_u128(0xb)); let ordinary_a = CancellationToken::new(); @@ -1591,7 +1715,7 @@ mod tests { #[tokio::test] async fn register_then_revalidate_closes_both_archive_race_orderings() { - let registry = CommunityConnectionRegistry::new(); + let registry = test_registry(); let community = CommunityId::from_uuid(Uuid::from_u128(0xa)); // Archive wins before durable revalidation: the check observes inactive @@ -1647,7 +1771,7 @@ mod tests { #[tokio::test] async fn revalidation_continues_after_one_community_lookup_failure() { - let registry = CommunityConnectionRegistry::new(); + let registry = test_registry(); let archived_a = CommunityId::from_uuid(Uuid::from_u128(0xa)); let failed = CommunityId::from_uuid(Uuid::from_u128(0xb)); let archived_c = CommunityId::from_uuid(Uuid::from_u128(0xc)); @@ -1684,7 +1808,7 @@ mod tests { #[test] fn community_lifecycle_guard_deregisters_on_early_return() { - let registry = CommunityConnectionRegistry::new(); + let registry = test_registry(); let community = CommunityId::from_uuid(Uuid::from_u128(0xa)); let cancel = CancellationToken::new(); let guard = registry.register(Uuid::new_v4(), community, cancel.clone()); @@ -1930,4 +2054,104 @@ mod tests { other => panic!("expected a restart close frame, got {other:?}"), } } + + /// Residency drives the cache-invalidation desired set, so it must cover a + /// community for at least as long as an entry cached under it can be read. + #[test] + fn residency_outlives_the_entry_it_protects() { + // The two are the same constant by construction; asserting the relation + // is what fails the build if a future change gives the caches a longer + // TTL than residency, which would silently reopen the stale-authz gap. + let topics = Arc::new(CommunityTopics::new("test")); + let residency = CacheResidency::new(Arc::clone(&topics), AUTHZ_CACHE_TTL); + let community = CommunityId::from_uuid(Uuid::from_u128(0xa)); + + let _ = residency.admit(community); + assert_eq!( + residency.resident_communities(), + HashSet::from([community]), + "a just-resolved community is resident" + ); + + // A residency shorter than the cache TTL is the bug; prove the tracker + // reports lapse purely on its own clock by using a zero-length window. + let expired = CacheResidency::new(topics, Duration::ZERO); + let _ = expired.admit(community); + assert!( + expired.resident_communities().is_empty(), + "residency must lapse on its own timer, with no external event" + ); + } + + /// The bootstrap must close: recording residency and gating the insert are + /// one call, so a community this pod reads becomes desired *before* it is + /// admitted, and is admitted as soon as the reconciler acks. + /// + /// Split into a `may_cache` gate plus a separate `touch`, only an admitted + /// community would ever be recorded — nothing would enter the desired set, + /// nothing would be subscribed, and the authorization caches would stay off + /// for the life of the pod. This test is the deadlock detector, so it + /// asserts the *first* call is both refused and residency-forming. + #[test] + fn admit_makes_a_community_desired_before_it_is_established() { + let topics = Arc::new(CommunityTopics::new("test")); + let residency = CacheResidency::new(Arc::clone(&topics), AUTHZ_CACHE_TTL); + let community = CommunityId::from_uuid(Uuid::from_u128(0xa)); + + assert!( + !residency.admit(community), + "nothing is cacheable before Redis acks the invalidation channel" + ); + assert_eq!( + residency.resident_communities(), + HashSet::from([community]), + "the refused call must still make the community desired, or the \ + subscriber never subscribes and the gate never opens" + ); + + // Stand in for the reconciler acking the SUBSCRIBE it was just asked for. + topics.insert_established_for_test(community); + assert!( + residency.admit(community), + "an acked community must be admitted" + ); + } + + /// A newly resident community wakes the cache-invalidation subscriber, so + /// the subscribe does not wait for the reconcile tick. + #[tokio::test] + async fn a_newly_resident_community_signals_the_cache_subscriber() { + let topics = Arc::new(CommunityTopics::new("test")); + let residency = CacheResidency::new(Arc::clone(&topics), AUTHZ_CACHE_TTL); + let community = CommunityId::from_uuid(Uuid::from_u128(0xa)); + + let _ = residency.admit(community); + + tokio::time::timeout(Duration::from_millis(500), topics.wake_notified()) + .await + .expect("first residency must nudge the cache-invalidation reconciler"); + } + + /// A registration wakes the conn-control subscriber. The wake is latency + /// only — the reconcile tick is the guarantee — but a registry that never + /// signals would make every new community's ban enforcement wait a tick. + #[tokio::test] + async fn registering_a_socket_signals_the_conn_control_subscriber() { + let topics = Arc::new(CommunityTopics::new("test")); + let registry = CommunityConnectionRegistry::new(Arc::clone(&topics)); + let community = CommunityId::from_uuid(Uuid::from_u128(0xa)); + + let guard = registry.register(Uuid::new_v4(), community, CancellationToken::new()); + assert_eq!(registry.bound_communities(), HashSet::from([community])); + + // The wake stores one permit, so waiting after the fact still observes + // the registration's signal. A bounded timeout rather than a bare await: + // a registry that never signals must fail this, not hang the suite. + tokio::time::timeout(Duration::from_millis(500), topics.wake_notified()) + .await + .expect("registration must nudge the conn-control reconciler"); + + drop(guard); + assert!(registry.bound_communities().is_empty()); + } }