diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d392170f6a..23c6ad8926 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -353,6 +353,7 @@ jobs: cargo nextest archive \ --cargo-profile ci \ -p buzz-db \ + -p buzz-pubsub \ -p buzz-relay \ -p buzz-test-client \ --lib \ @@ -684,6 +685,30 @@ jobs: done cat /tmp/buzz-relay.log exit 1 + - name: Redis compatibility tests (standalone) + # buzz-pubsub's Redis-backed tests are #[ignore]d, so nothing ran them + # in CI before. They assert the cluster-safe command shapes required by + # ElastiCache Serverless (single-key GETs for bulk presence, exact + # SUBSCRIBE) *and* that those shapes still behave on standard Redis — + # which is the compatibility half of the guarantee, and the half a + # serverless-only rig cannot check. Points at the standalone Redis this + # job already runs; a cluster-mode service container is deliberately not + # added here. + # + # The second selector is the composed cache-residency drill, and it is + # named because it CANNOT live in buzz-pubsub: it needs the real + # `CacheResidency` (buzz-relay) driving the real subscriber + # (buzz-pubsub), and the dependency edge only runs relay → pubsub. A + # `package(buzz-pubsub)` selector does not reach buzz-relay, so without + # this line the drill would pass locally and gate nothing — the exact + # fake-green class the rest of this step exists to remove. + run: | + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E 'package(buzz-pubsub) or (package(buzz-relay) and test(/state::redis_tests::/))' \ + --run-ignored all + env: + REDIS_URL: redis://localhost:6379 - name: Invite security tests run: | cargo nextest run \ @@ -692,6 +717,15 @@ jobs: --run-ignored ignored-only env: DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz + - name: Tunnel fence Redis integration tests + run: | + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E 'package(buzz-relay) and test(/tunnel::directory::tests/)' \ + --run-ignored ignored-only + env: + REDIS_URL: redis://localhost:6379 + - name: Workspace profile (kind:9033) gate tests # Call-site integration for the 9033 authorization gate: open relay # rosterless/steward transitions and the closed-relay admin/owner rule, diff --git a/crates/buzz-pubsub/src/cache_invalidation.rs b/crates/buzz-pubsub/src/cache_invalidation.rs index f48158c9c5..9567b20664 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); + run_community_subscriber( + redis_url, + CacheInvalidationFamily { broadcast_tx }, + topics, + desired, + ) + .await; +} - tracing::info!("Attempting to reconnect to Redis cache-invalidation..."); - } +struct CacheInvalidationFamily { + broadcast_tx: broadcast::Sender, } -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?; - - conn.psubscribe(CACHE_INVALIDATION_PATTERN).await?; - - tracing::info!( - "Redis cache-invalidation subscriber connected — listening on {CACHE_INVALIDATION_PATTERN}" - ); - - 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; - }; +impl CommunityChannelFamily for CacheInvalidationFamily { + fn channel(&self, community: CommunityId) -> String { + cache_invalidation_channel_for(community) + } - let payload: String = match msg.get_payload() { - Ok(p) => p, - Err(e) => { - tracing::warn!("Failed to get cache-invalidation payload: {e}"); - continue; - } - }; + fn parse_channel(&self, channel: &str) -> Option { + parse_cache_invalidation_channel(channel) + } - 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 0000000000..17670a03c8 --- /dev/null +++ b/crates/buzz-pubsub/src/community_topics.rs @@ -0,0 +1,1046 @@ +//! 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. Crucially, the *withdrawal* direction re-reads it +//! under the same lock that removes establishment (`withdraw_undesired`), so +//! interest renewed after the diff was computed cannot have its subscription +//! pulled out from under it. +//! * **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 an *observation*, not a gate: it is true only until the lock is + /// released. Anything that must not become readable without a live + /// invalidation topic has to run inside [`Self::with_established`]. + pub fn is_established(&self, community: CommunityId) -> bool { + self.established + .read() + .expect("community topics lock poisoned") + .contains(&community) + } + + /// Runs `f` with this community's establishment state, holding it against + /// withdrawal for the whole call. + /// + /// This is the gate for state whose readability depends on the + /// subscription — in the relay, an authorization cache insert. Withdrawal + /// takes the matching write lock (`withdraw_undesired`), so the two cannot + /// interleave: either `f` completes and the withdrawal decision that + /// follows re-reads desire and sees whatever interest `f` recorded, or the + /// withdrawal already happened and `f` is told `false`. Reading + /// establishment and then acting on it in a second step reopens that gap, + /// however small the gap looks. + /// + /// `f` must not call back into this `CommunityTopics`: the read lock is + /// held, so a nested write would deadlock and a nested read can too if a + /// writer is already queued. It may touch state the `desired` closure + /// reads — that closure only ever runs under the write lock, which excludes + /// this one, so there is no lock cycle. + pub fn with_established(&self, community: CommunityId, f: impl FnOnce(bool) -> R) -> R { + let established = self + .established + .read() + .expect("community topics lock poisoned"); + f(established.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); + } + + /// Re-reads `desired` and withdraws establishment for everything it no + /// longer names, **atomically**, returning the communities whose channel the + /// caller must now `UNSUBSCRIBE`. + /// + /// This is the ordering invariant the establishment gate rests on, and the + /// reason the decision cannot be made from a snapshot taken earlier in the + /// reconcile. Interest renewal (`CacheResidency::admit_and_insert` in the + /// relay) + /// records residency and *then* reads establishment under the read lock; + /// this method re-reads residency under the matching write lock. Mutual + /// exclusion makes the two orderings the only possibilities: + /// + /// * renewal's establishment read precedes this critical section — then this + /// `desired()` call happens after the renewal recorded interest, sees the + /// community, and does not withdraw it. The admitted entry stays covered. + /// * renewal's establishment read follows this critical section — then the + /// withdrawal already happened, the read returns false, and nothing is + /// admitted in the first place. + /// + /// So an entry admitted against establishment is never left readable behind + /// a withdrawn subscription. A `desired()` re-read *outside* the lock would + /// not give this: renewal could still land between the check and the removal. + /// + /// `desired` must not acquire this lock, or this would deadlock. Both + /// production closures read only their own registry. + pub(crate) fn withdraw_undesired(&self, desired: &DesiredCommunities) -> Vec { + let mut established = self + .established + .write() + .expect("community topics lock poisoned"); + Self::withdraw_locked(&mut established, desired) + } + + /// The withdrawal decision itself, given the write lock. Shared so the + /// non-blocking test twin below cannot drift away from the real path. + fn withdraw_locked( + established: &mut HashSet, + desired: &DesiredCommunities, + ) -> Vec { + let want = desired(); + let withdrawn: Vec = established.difference(&want).copied().collect(); + for community in &withdrawn { + established.remove(community); + } + withdrawn + } + + /// Test seam for `withdraw_undesired`, whose invariant is shared with + /// the cache-residency gate in another crate. + #[doc(hidden)] + pub fn withdraw_undesired_for_test(&self, desired: &DesiredCommunities) -> Vec { + self.withdraw_undesired(desired) + } + + /// Non-blocking `withdraw_undesired`: `None` when the establishment lock is + /// held elsewhere. + /// + /// Test seam for the exclusion itself. A test can call this from inside a + /// [`Self::with_established`] closure and assert `None` — structural proof + /// that no withdrawal can interleave with a gated insert, with no threads + /// and no scheduler dependence. Asserting that through a parked thread + /// makes the *mutation bite* probabilistic even though the passing case is + /// deterministic. + #[doc(hidden)] + pub fn try_withdraw_undesired_for_test( + &self, + desired: &DesiredCommunities, + ) -> Option> { + let mut established = match self.established.try_write() { + Ok(guard) => guard, + Err(std::sync::TryLockError::WouldBlock) => return None, + Err(e) => panic!("community topics lock poisoned: {e}"), + }; + Some(Self::withdraw_locked(&mut established, desired)) + } + + 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 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. +/// +/// The withdrawal ordering below is enforced for **both** families, not just +/// cache invalidation, because it lives in the shared reconciler. Cache +/// invalidation is what makes it a correctness requirement — a lost invalidation +/// leaves a readable authorization entry with nothing behind it. Connection +/// control's exposure is milder: a `disconnect` command lost in an unsubscribe +/// gap leaves a socket open that a ban has already closed elsewhere, and the DB +/// ban row still refuses that pubkey's next authorization, so the miss costs one +/// stale socket rather than a wrong access decision. The invariant is applied +/// uniformly anyway: one reconciler is less to get wrong than two, and nothing +/// here is per-family. +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. +/// +/// The two directions are deliberately asymmetric, because only one of them can +/// break an invariant: +/// +/// * **Subscribing** from the pass's initial `desired()` read is safe. Holding a +/// subscription nobody wants any more costs one idle channel until the next +/// pass withdraws it; it can never make an entry readable without coverage. +/// * **Unsubscribing** must not be decided from that same read. Between the read +/// and the command, interest can be renewed and an insert admitted against the +/// still-present establishment bit — and a pub/sub message lost in the +/// resulting gap is lost permanently, because pub/sub has no replay. The tick +/// restores *state*; nothing restores a dropped invalidation. So withdrawal +/// re-reads desire under the establishment write lock, via +/// [`CommunityTopics::withdraw_undesired`], and only the communities that +/// round-trip out of that critical section are unsubscribed. +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); + } + + // Establishment is withdrawn inside the lock, so a reader that admits an + // entry either observed the bit before the withdrawal (and its renewed + // interest was seen here, keeping the subscription) or observes it after + // (and admits nothing). + for community in topics.withdraw_undesired(desired) { + 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)); + + let keep = community(2); + let desired: DesiredCommunities = Arc::new(move || HashSet::from([keep])); + assert_eq!(topics.withdraw_undesired(&desired), vec![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"); + } + + /// Withdrawal must decide from desire read *inside* its own critical + /// section, not from a snapshot taken earlier in the reconcile pass. + /// + /// This is the race Wren found in review. The subscribe loop awaits Redis + /// round-trips, so an arbitrary amount of time passes between the pass's + /// `desired()` read and the withdrawal — long enough for interest to be + /// renewed and a cache entry admitted against the still-present + /// establishment bit. Here the stale view says "withdraw c", live desire says + /// "keep c", and the assertion is that live desire wins. + #[test] + fn withdrawal_reads_live_desire_not_the_passs_stale_view() { + let topics = CommunityTopics::new("test"); + let c = community(1); + topics.insert_established(c); + + // The pass's initial read saw an empty desired set (c not wanted). By + // the time withdrawal runs, interest has been renewed. + let desired: DesiredCommunities = Arc::new(move || HashSet::from([c])); + + assert!( + topics.withdraw_undesired(&desired).is_empty(), + "a community desired at withdrawal time must not be unsubscribed" + ); + assert!( + topics.is_established(c), + "renewed interest must keep its establishment bit" + ); + } + + /// The other direction of the same call: still-undesired communities are + /// withdrawn, and establishment is dropped before the caller unsubscribes. + #[test] + fn withdrawal_still_removes_communities_nobody_wants() { + let topics = CommunityTopics::new("test"); + let (kept, dropped) = (community(1), community(2)); + topics.insert_established(kept); + topics.insert_established(dropped); + + let desired: DesiredCommunities = Arc::new(move || HashSet::from([kept])); + + assert_eq!(topics.withdraw_undesired(&desired), vec![dropped]); + assert!( + !topics.is_established(dropped), + "establishment must be gone before the UNSUBSCRIBE is issued" + ); + assert!(topics.is_established(kept)); + } + + /// The invariant itself, forced deterministically: a renewal that observes + /// establishment can never have its subscription withdrawn by a + /// concurrently-running reconcile. + /// + /// The `desired` closure parks *inside* the withdrawal critical section and + /// releases the renewal thread from there, so the renewal's + /// `is_established` read is guaranteed to happen while withdrawal holds the + /// write lock. That is the interleaving a sleep-based test only reaches by + /// luck. The reader must then observe `false` (withdrawal won the lock and + /// this admission is correctly refused) — never `true` while the community + /// also ends up unsubscribed, which is the stale-authz hole. + #[test] + fn a_renewal_racing_withdrawal_never_sees_establishment_it_loses() { + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::mpsc; + + let topics = Arc::new(CommunityTopics::new("test")); + let c = community(1); + topics.insert_established(c); + + // Signals the renewal thread that withdrawal now holds the write lock. + let (in_section_tx, in_section_rx) = mpsc::channel::<()>(); + // Set by the renewal thread once its establishment read has returned. + let renewal_read = Arc::new(AtomicBool::new(false)); + let read_result = Arc::new(AtomicBool::new(false)); + + let renewal_topics = Arc::clone(&topics); + let renewal_read_flag = Arc::clone(&renewal_read); + let renewal_result = Arc::clone(&read_result); + let renewal = std::thread::spawn(move || { + in_section_rx + .recv() + .expect("withdrawal must signal from inside its critical section"); + // Blocks until withdrawal releases the write lock. + let admitted = renewal_topics.is_established(c); + renewal_result.store(admitted, Ordering::SeqCst); + renewal_read_flag.store(true, Ordering::SeqCst); + }); + + // Desire says "nobody wants c" — but hands the renewal thread its cue + // first, from inside the lock. + let desired: DesiredCommunities = Arc::new(move || { + in_section_tx.send(()).expect("renewal thread alive"); + // Give the renewal thread every chance to reach its read while the + // write lock is still held; RwLock write exclusion is what makes + // the outcome deterministic regardless of how far it gets. + std::thread::sleep(Duration::from_millis(50)); + HashSet::new() + }); + + assert_eq!(topics.withdraw_undesired(&desired), vec![c]); + renewal.join().expect("renewal thread panicked"); + + assert!(renewal_read.load(Ordering::SeqCst)); + assert!( + !read_result.load(Ordering::SeqCst), + "a renewal whose interest was not seen by withdrawal must be refused, \ + never admitted against an establishment bit that is being removed" + ); + assert!(!topics.is_established(c)); + } + + /// `with_established` must exclude withdrawal for the whole closure, not + /// just for the establishment read. + /// + /// This is the lock half of the lifetime invariant, and it is asserted + /// *structurally*: from inside the closure, `try_write()` must report + /// `WouldBlock`. No threads, no sleeps, no scheduler. Under a mutation that + /// releases the lock before calling `f`, `try_write()` succeeds and this + /// fails on every run — which is what a deterministic mutation kill means. + /// An earlier draft signalled a withdrawal thread and slept 50ms to "give + /// it every chance"; that is deterministic on correct code but its + /// *mutation bite* depends on the scheduler, so it could pass vacuously. + /// Wren caught it pre-push. + #[test] + fn with_established_holds_the_gate_for_the_whole_closure() { + let topics = CommunityTopics::new("test"); + let c = community(1); + topics.insert_established(c); + + let established = topics.with_established(c, |established| { + assert!( + matches!( + topics.established.try_write(), + Err(std::sync::TryLockError::WouldBlock) + ), + "withdrawal must be excluded for the whole closure; a gate that \ + releases the lock before the caller acts is the round-2 bug" + ); + established + }); + + assert!(established); + // And the exclusion ends with the closure, or the reconciler would + // never be able to withdraw anything. + assert!(topics.established.try_write().is_ok()); + } +} + +/// 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; + } + + /// End-to-end against live Redis: interest renewed *during* a reconcile pass + /// must not lose its exact subscription. + /// + /// The unit test above proves the lock discipline in isolation; this proves + /// the property survives the real `reconcile` path, where awaited Redis + /// round-trips sit between the pass's desire read and the withdrawal. + /// + /// The closure is a phase machine keyed on read count, which is what makes + /// the interleaving deterministic: once the race is armed, the pass's *first* + /// read reports the community undesired (so a stale-snapshot reconciler would + /// queue an `UNSUBSCRIBE`), and every read after it reports the community + /// desired again (the renewal). Correct code re-reads inside the withdrawal + /// critical section, sees the renewal, and keeps the subscription. Verified + /// to bite: reinstating the `have.difference(&want)` withdrawal fails this + /// test on the `PUBSUB NUMSUB` assertion. + #[tokio::test] + #[ignore = "requires Redis"] + async fn interest_renewed_mid_reconcile_keeps_its_exact_subscription() { + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + + let c = fresh(1)[0]; + let armed = Arc::new(AtomicBool::new(false)); + let reads = Arc::new(AtomicUsize::new(0)); + + let closure_armed = Arc::clone(&armed); + let closure_reads = Arc::clone(&reads); + let racing: DesiredCommunities = Arc::new(move || { + if !closure_armed.load(Ordering::SeqCst) { + return HashSet::from([c]); + } + // First read after arming: interest looks gone. Every later read + // (crucially, the one inside the withdrawal lock) sees it renewed. + if closure_reads.fetch_add(1, Ordering::SeqCst) == 0 { + HashSet::new() + } else { + HashSet::from([c]) + } + }); + + let (topics, _rx) = spawn_subscriber(racing); + await_established(&topics, &[c], "pre-race establishment").await; + assert_eq!(exact_subscribers(c).await, 1); + + armed.store(true, Ordering::SeqCst); + topics.wake(); + + // The discriminator, and why this bites where a coarser wait did not: + // correct code reads desire TWICE in one pass (the pass read, then the + // withdrawal read inside the lock), microseconds apart. A stale-snapshot + // reconciler reads once, unsubscribes, and only reads again at the next + // tick — a whole `RECONCILE_INTERVAL` later, by which point it has + // re-subscribed and a naive end-state assertion passes vacuously. + // + // So: poll far faster than the tick, require the second read inside a + // window well below it, and assert establishment never drops along the + // way. The transient loss is the bug; the repaired end state is not proof. + let window = RECONCILE_INTERVAL / 4; + let deadline = std::time::Instant::now() + window; + while reads.load(Ordering::SeqCst) < 2 { + assert!( + topics.is_established(c), + "establishment was withdrawn for renewed interest — an \ + invalidation published now would be lost, and the later tick \ + that re-subscribes cannot bring it back" + ); + assert!( + std::time::Instant::now() < deadline, + "withdrawal did not re-read desire within its own pass ({window:?}); \ + a second read this late means it decided from the stale snapshot" + ); + tokio::time::sleep(Duration::from_millis(1)).await; + } + + assert!( + topics.is_established(c), + "renewed interest must not lose establishment" + ); + assert_eq!( + exact_subscribers(c).await, + 1, + "renewed interest must still hold an exact SUBSCRIBE in Redis" + ); + } +} diff --git a/crates/buzz-pubsub/src/conn_control.rs b/crates/buzz-pubsub/src/conn_control.rs index bc177cff13..d42fc1b3f3 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?; - - conn.psubscribe(CONN_CONTROL_PATTERN).await?; - - tracing::info!("Redis conn-control subscriber connected — listening on {CONN_CONTROL_PATTERN}"); +struct ConnControlFamily { + broadcast_tx: broadcast::Sender, +} - 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; - }; +impl CommunityChannelFamily for ConnControlFamily { + fn channel(&self, community: CommunityId) -> String { + conn_control_channel_for(community) + } - let payload: String = match msg.get_payload() { - Ok(p) => p, - Err(e) => { - tracing::warn!("Failed to get conn-control payload: {e}"); - continue; - } - }; + fn parse_channel(&self, channel: &str) -> Option { + parse_conn_control_channel(channel) + } - 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/error.rs b/crates/buzz-pubsub/src/error.rs index 96bc1016c9..6461f9b4b6 100644 --- a/crates/buzz-pubsub/src/error.rs +++ b/crates/buzz-pubsub/src/error.rs @@ -26,6 +26,17 @@ pub enum PubSubError { /// A Redis channel key could not be parsed as a valid channel ID. #[error("Invalid channel key: {0}")] InvalidChannelKey(String), + + /// A bulk presence read returned a different number of values than keys + /// requested. Reassembly is positional, so continuing would attribute one + /// pubkey's status to another. + #[error("Presence bulk read returned {got} values for {expected} keys")] + PresenceReplyMismatch { + /// Number of presence keys requested. + expected: usize, + /// Number of values Redis returned. + got: usize, + }, } impl From for PubSubError { diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index 4f1690beef..ab2b7f660c 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() @@ -368,17 +407,84 @@ impl PubSubManager { #[cfg(test)] pub(crate) mod test_util { + /// The Redis endpoint under test. Honours `REDIS_URL` so the same + /// `--ignored` suite can be pointed at standalone Redis (the CI default) or + /// at a single-shard cluster-mode server, which is what enforces + /// `CROSSSLOT` locally. + /// + /// Every test that opens a connection must route through this — a + /// hardcoded URL alongside an env-var-driven one makes the pool and the + /// pub/sub client talk to *different servers*, which fails as a timeout and + /// reads like a cluster incompatibility. + pub fn test_redis_url() -> String { + std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string()) + } + pub fn make_test_pool() -> deadpool_redis::Pool { - let cfg = deadpool_redis::Config::from_url("redis://127.0.0.1:6379"); + let cfg = deadpool_redis::Config::from_url(test_redis_url()); 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; + 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; @@ -386,7 +492,7 @@ mod tests { async fn make_manager() -> Arc { let pool = make_test_pool(); Arc::new( - PubSubManager::new("redis://127.0.0.1:6379", pool) + PubSubManager::new(&test_redis_url(), pool) .await .expect("Failed to create PubSubManager"), ) @@ -439,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(); @@ -451,8 +571,6 @@ mod tests { pubkey: pubkey.clone(), }; - let ctx = ctx(0xaaaa, "a.example"); - manager .publish_cache_invalidation(&ctx, &sent) .await @@ -513,7 +631,7 @@ mod tests { let pool = make_test_pool(); let manager = Arc::new( PubSubManager::with_config( - PubSubConfig::new("redis://127.0.0.1:6379") + PubSubConfig::new(test_redis_url()) .with_unsubscribe_debounce(Duration::from_millis(25)), pool, ) @@ -593,8 +711,7 @@ mod tests { async fn retain_release_refcounts_and_debounces_last_release() { let pool = make_test_pool(); let manager = PubSubManager::with_config( - PubSubConfig::new("redis://127.0.0.1:6379") - .with_unsubscribe_debounce(Duration::from_millis(1)), + PubSubConfig::new(test_redis_url()).with_unsubscribe_debounce(Duration::from_millis(1)), pool, ) .await diff --git a/crates/buzz-pubsub/src/presence.rs b/crates/buzz-pubsub/src/presence.rs index e0c9dfd6c9..a5a357d488 100644 --- a/crates/buzz-pubsub/src/presence.rs +++ b/crates/buzz-pubsub/src/presence.rs @@ -71,6 +71,18 @@ pub async fn get_presence( } /// Returns `pubkey_hex → status` for all currently-set keys. +/// +/// Issues one single-key `GET` per pubkey in a **non-atomic** pipeline: presence +/// keys are per-pubkey, so on a cluster (ElastiCache Serverless included) they +/// span slots and any multi-key command over them fails `CROSSSLOT`. +/// +/// **Never call `.atomic()` on this pipeline.** `Pipeline::new` leaves +/// `transaction_mode: false`; `.atomic()` sets it, which makes `query_async` +/// wrap the batch in `MULTI`/`EXEC` and dispatch it as one unit — restoring +/// exactly the cross-slot failure this replaces. A non-atomic pipeline is +/// independent commands sharing one round trip, which is all this read needs: +/// each key is fetched on its own and the entries are TTL'd anyway, so there is +/// nothing for atomicity to protect. pub async fn get_presence_bulk( pool: &Pool, ctx: &TenantContext, @@ -80,11 +92,22 @@ pub async fn get_presence_bulk( return Ok(HashMap::new()); } let mut conn = pool.get().await?; - let keys: Vec = pubkeys - .iter() - .map(|pubkey| presence_key(ctx, pubkey)) - .collect(); - let values: Vec> = redis::cmd("MGET").arg(&keys).query_async(&mut conn).await?; + let mut pipe = redis::pipe(); + for pubkey in pubkeys { + pipe.cmd("GET").arg(presence_key(ctx, pubkey)); + } + let values: Vec> = pipe.query_async(&mut conn).await?; + + // Replies are positional — Redis returns values, not key names — so a reply + // of the wrong length would silently misattribute one pubkey's status to + // another under `zip`. Refuse instead. + if values.len() != pubkeys.len() { + return Err(PubSubError::PresenceReplyMismatch { + expected: pubkeys.len(), + got: values.len(), + }); + } + let result = pubkeys .iter() .zip(values.iter()) @@ -139,6 +162,49 @@ mod tests { ); } + /// CRC16-XMODEM over the cluster hash-tag-free key, mod 16384 — the Redis + /// Cluster slot function. Test-local on purpose: production never computes + /// slots (we use a plain, non-cluster client), this exists only to assert a + /// key-design property. + fn key_slot(key: &str) -> u16 { + let mut crc: u16 = 0; + for byte in key.as_bytes() { + crc ^= (*byte as u16) << 8; + for _ in 0..8 { + crc = if crc & 0x8000 != 0 { + (crc << 1) ^ 0x1021 + } else { + crc << 1 + }; + } + } + crc % 16384 + } + + #[test] + fn key_slot_matches_redis_cluster_reference_vector() { + // Redis Cluster spec's CRC16 check value: CRC16-XMODEM("123456789") = + // 0x31C3, which is below 16384 so the slot equals it outright. + assert_eq!(key_slot("123456789"), 0x31C3); + } + + /// Presence keys carry no hash tag, so pubkeys in one community spread across + /// slots. That is deliberate: tagging them by community would pin an entire + /// community's presence to a single slot, against AWS's uniform-distribution + /// guidance and into the per-slot ECPU ceiling. It is also why the bulk read + /// must be single-key `GET`s rather than any multi-key command. + #[test] + fn presence_keys_in_one_community_span_multiple_slots() { + let ctx = ctx(0xaaaa, "a.example"); + let slots: std::collections::HashSet = (0..16) + .map(|_| key_slot(&presence_key(&ctx, &make_pubkey()))) + .collect(); + assert!( + slots.len() > 1, + "presence keys must not co-slot; got {slots:?}" + ); + } + #[tokio::test] #[ignore = "requires Redis"] async fn test_presence_set_and_get() { @@ -189,6 +255,91 @@ mod tests { clear_presence(&pool, &ctx, &pk2).await.unwrap(); } + /// Asserts the **reassembled map**, not the reply vector or its length. + /// + /// The bulk read zips request positions against reply positions, so a + /// misalignment is count-correct and attributes one member's status to + /// another — silent server-side, and user-visible as other people's online + /// status. Every assertion here is `pubkey → status` for a *distinctly + /// valued* status, so any permutation of the reply fails. + #[tokio::test] + #[ignore = "requires Redis"] + async fn presence_bulk_reassembles_by_pubkey_across_slots() { + let pool = make_test_pool(); + let ctx = ctx(0xb301, "b3.example"); + + // Distinct statuses so a shifted reply can't coincidentally match, and + // enough keys that they span multiple slots (see the slot unit test). + let present: Vec<(PublicKey, String)> = (0..12) + .map(|i| (make_pubkey(), format!("status-{i}"))) + .collect(); + let missing: Vec = (0..4).map(|_| make_pubkey()).collect(); + + for (pk, status) in &present { + set_presence(&pool, &ctx, pk, status).await.unwrap(); + } + + let expected: HashMap = present + .iter() + .map(|(pk, status)| (pk.to_hex(), status.clone())) + .collect(); + + // Interleave set and unset keys, so a `None` in the middle of the reply + // must not shift the keys after it. + let mut requested: Vec = Vec::new(); + for (i, (pk, _)) in present.iter().enumerate() { + requested.push(*pk); + if let Some(absent) = missing.get(i / 3) { + requested.push(*absent); + } + } + + let got = get_presence_bulk(&pool, &ctx, &requested).await.unwrap(); + assert_eq!(got, expected, "reassembled map must be keyed by pubkey"); + + // Same set, reversed order: the map is order-independent by definition, + // so this catches positional misalignment without needing to know which + // order is "right". + let mut reversed = requested.clone(); + reversed.reverse(); + let got_reversed = get_presence_bulk(&pool, &ctx, &reversed).await.unwrap(); + assert_eq!( + got_reversed, expected, + "map must not depend on request order" + ); + + // Duplicate keys: the reply is one value per request position, and the + // map must collapse them rather than drift. + let duplicated: Vec = + requested.iter().chain(requested.iter()).copied().collect(); + let got_duplicated = get_presence_bulk(&pool, &ctx, &duplicated).await.unwrap(); + assert_eq!( + got_duplicated, expected, + "duplicate requests must collapse to the same map" + ); + + // All-missing: no entries, no error. + assert!(get_presence_bulk(&pool, &ctx, &missing) + .await + .unwrap() + .is_empty()); + + for (pk, _) in &present { + clear_presence(&pool, &ctx, pk).await.unwrap(); + } + } + + #[tokio::test] + #[ignore = "requires Redis"] + async fn presence_bulk_empty_request_returns_empty_map() { + let pool = make_test_pool(); + let ctx = ctx(0xb302, "b3.example"); + assert!(get_presence_bulk(&pool, &ctx, &[]) + .await + .unwrap() + .is_empty()); + } + #[tokio::test] #[ignore = "requires Redis"] async fn test_presence_ttl() { diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 85a0ca2efe..c4821a1302 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -152,6 +152,13 @@ pub struct Config { /// no-regression rollout. pub mesh: buzz_relay_mesh::MeshConfig, + /// Tunnel fence Redis key format used by the mesh session directory. + /// + /// Defaults to `legacy`; operators must explicitly select `tagged` only + /// after completing the drain phase documented in + /// `docs/tunnel-fence-key-migration.md`. + pub tunnel_fence_key_format: crate::tunnel::directory::SessionKeyFormat, + /// Testbed-only reliable-stream echo consumer (`BUZZ_MESH_DEMO_ECHO`). /// When `on`, the owner side of an inbound reliable mesh stream echoes /// every validated `Data` frame back to the sender — a transport/ @@ -279,6 +286,18 @@ pub struct Config { pub serve_git_web_gui: bool, } +fn parse_tunnel_fence_key_format( + value: Option<&str>, +) -> Result { + match value { + None | Some("legacy") => Ok(crate::tunnel::directory::SessionKeyFormat::Legacy), + Some("tagged") => Ok(crate::tunnel::directory::SessionKeyFormat::Tagged), + Some(value) => Err(ConfigError::InvalidValue(format!( + "BUZZ_TUNNEL_FENCE_KEY_FORMAT must be 'legacy' or 'tagged', got {value:?}" + ))), + } +} + fn parse_bind_addr(raw: &str) -> Result { raw.parse::() .map_err(|e| ConfigError::InvalidBindAddr(e.to_string())) @@ -560,6 +579,16 @@ impl Config { registry_refresh: std::time::Duration::from_secs(15), }; + let tunnel_fence_key_format = match std::env::var("BUZZ_TUNNEL_FENCE_KEY_FORMAT") { + Ok(value) => parse_tunnel_fence_key_format(Some(&value))?, + Err(std::env::VarError::NotPresent) => parse_tunnel_fence_key_format(None)?, + Err(std::env::VarError::NotUnicode(_)) => { + return Err(ConfigError::InvalidValue( + "BUZZ_TUNNEL_FENCE_KEY_FORMAT must be valid Unicode".to_string(), + )); + } + }; + // Demo echo opt-in: same strict pattern as BUZZ_MESH — explicit // `on`/`true`/`1` only, anything else (absent, `off`, typos) is off. let mesh_demo_echo = std::env::var("BUZZ_MESH_DEMO_ECHO") @@ -956,6 +985,7 @@ impl Config { require_relay_membership, huddle_audio_available, mesh, + tunnel_fence_key_format, mesh_demo_echo, relay_owner_pubkey, relay_operator_api_origin, @@ -997,6 +1027,25 @@ mod tests { // value set by `invalid_bind_addr_returns_error`, causing a flaky failure. static ENV_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(()); + #[test] + fn tunnel_fence_key_format_is_default_off_and_strict() { + use crate::tunnel::directory::SessionKeyFormat; + + assert_eq!( + parse_tunnel_fence_key_format(None).unwrap(), + SessionKeyFormat::Legacy + ); + assert_eq!( + parse_tunnel_fence_key_format(Some("legacy")).unwrap(), + SessionKeyFormat::Legacy + ); + assert_eq!( + parse_tunnel_fence_key_format(Some("tagged")).unwrap(), + SessionKeyFormat::Tagged + ); + assert!(parse_tunnel_fence_key_format(Some("on")).is_err()); + } + #[test] fn defaults_are_valid() { let _guard = ENV_MUTEX.lock().unwrap(); diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index a9cdffcdec..4d348df4c6 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -1692,7 +1692,7 @@ mod tests { register_presence_sub(&receiver, "receiver-presence"); // Under the community-scoped bus, Redis delivery is demand-driven: - // a relay only PSUBSCRIBEs `buzz:{community}:global` after it retains + // a relay only SUBSCRIBEs `buzz:{community}:global` after it retains // interest in that topic. Both relays share one explicit tenant and // retain Global before publishing — origin too, so the echo- // suppression assertion still exercises `mark_local_event` against a @@ -1710,7 +1710,7 @@ mod tests { .retain_topic(&tenant, EventTopic::Global) .await; - // Match buzz-pubsub's own Redis round-trip test: give PSUBSCRIBE a + // Match buzz-pubsub's own Redis round-trip test: give SUBSCRIBE a // bounded moment to attach before publishing the single test event. tokio::time::sleep(std::time::Duration::from_millis(200)).await; diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 799cf9cf60..ab9e56e109 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/mesh_boot.rs b/crates/buzz-relay/src/mesh_boot.rs index 20e550aa08..fc16e317f7 100644 --- a/crates/buzz-relay/src/mesh_boot.rs +++ b/crates/buzz-relay/src/mesh_boot.rs @@ -508,7 +508,11 @@ pub async fn boot_mesh( transport.set_inbound(Box::new(dispatcher.clone())); Ok(Some(MeshHandle { - directory: SessionDirectory::new(redis_pool), + directory: SessionDirectory::with_key_format( + redis_pool, + std::time::Duration::from_secs(30), + config.tunnel_fence_key_format, + ), transport, membership: membership_arc, local_runtime_id: runtime_id, diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 58a869a995..bbe17d1860 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,147 @@ 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, if + /// the invalidation topic is established, runs `insert` to place the entry + /// in its cache. + /// + /// Recording, gating and inserting are one call on purpose, for two + /// separate reasons this lane found the hard way. + /// + /// **Recording with gating.** 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. + /// + /// **Gating with inserting.** The residency window must cover the whole life + /// of every entry it protects. A gate that only returned a `bool` left the + /// caller to insert afterwards, so the moka entry's TTL started *after* the + /// residency deadline it was measured against — under-covering the entry by + /// the call gap, and by an unbounded amount if the thread is preempted in + /// between: residency could lapse, the reconciler withdraw and + /// `UNSUBSCRIBE`, and the resumed caller still land an entry readable for a + /// full TTL with nothing behind it. Equal TTL constants do not nest when the + /// clocks start in the wrong order. + /// + /// 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 insert and 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. + /// + /// **The three steps below are ordered, and each ordering is load-bearing.** + /// + /// 1. Residency is recorded *before* establishment is read, and the + /// reconciler withdraws establishment while re-reading residency under + /// the matching write lock (`CommunityTopics::withdraw_undesired`, + /// private to `buzz-pubsub`). + /// Whichever side wins the lock, either this renewal is visible to the + /// withdrawal decision (so the subscription survives) or the withdrawal + /// already happened (so nothing is admitted). + /// 2. `insert` runs *inside* [`CommunityTopics::with_established`], so no + /// withdrawal can interleave between the gate and the entry landing. + /// 3. The residency deadline is re-stamped *after* `insert` returns and + /// still inside that critical section, so it is based at or after the + /// entry's own clock start and therefore outlives it. + /// + /// `insert` must not call back into [`CommunityTopics`] (the read lock is + /// held); the three production callers do a single moka insert, which does + /// not. + pub fn admit_and_insert(&self, community: CommunityId, insert: impl FnOnce()) -> bool { + // Step 1: residency before the establishment read; see the ordering note. + if self + .resident + .insert(community, Instant::now() + self.ttl) + .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.with_established(community, |established| { + if established { + // Step 2: the entry lands while withdrawal is excluded. + insert(); + // Step 3: residency restarts at or after the entry's clock, so + // the entry can never outlive the subscription protecting it. + self.resident.insert(community, Instant::now() + self.ttl); + } + established + }); + 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() + } + + /// The deadline currently protecting `community`, if it is resident. + /// + /// Test seam: the nesting relation between a residency window and the cache + /// entry it protects is only checkable against the actual deadline. A test + /// comparing TTL *constants* passes while the clocks start in the wrong + /// order, which is exactly the bug this seam exists to catch. + #[doc(hidden)] + pub fn deadline_for_test(&self, community: CommunityId) -> Option { + self.resident.get(&community).map(|entry| *entry.value()) + } + + /// Forces every residency to lapse now. + /// + /// Test seam: lets a test reach the post-expiry state deterministically + /// instead of sleeping out a real TTL. + #[doc(hidden)] + pub fn expire_all_for_test(&self) { + self.resident.clear(); + } +} + /// Registers a socket, durably revalidates its community, then runs it. /// /// The ordering is the archival admission invariant: archive-before-query is @@ -506,6 +662,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 +872,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 +884,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 +907,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 +991,12 @@ 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::admit_and_insert`]. + /// 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 +1010,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); + self.cache_residency.admit_and_insert(community_id, || { + self.membership_cache.insert(key, result); + }); Ok(result) } @@ -1102,7 +1276,9 @@ impl AppState { .db .get_accessible_channel_ids(community_id, pubkey) .await?; - self.accessible_channels_cache.insert(key, result.clone()); + self.cache_residency.admit_and_insert(community_id, || { + self.accessible_channels_cache.insert(key, result.clone()); + }); Ok(result) } @@ -1144,8 +1320,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. + self.cache_residency.admit_and_insert(community_id, || { + self.channel_visibility_cache + .insert((community_id, channel_id), visibility.clone()); + }); } Ok(visibility) } @@ -1255,6 +1439,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 +1763,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 +1781,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 +1837,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 +1874,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 +2120,553 @@ mod tests { other => panic!("expected a restart close frame, got {other:?}"), } } + + /// Residency must cover an entry for at least as long as that entry can be + /// read — which is a statement about *nested lifetimes*, not about equal + /// constants. + /// + /// This test previously asserted `AUTHZ_CACHE_TTL == AUTHZ_CACHE_TTL` and + /// was green while the bug was live: `admit` stamped the residency deadline + /// and returned, and the caller started moka's identical TTL afterwards, so + /// the windows had the same length and the wrong offset. Equal durations do + /// not nest when their clocks start in the wrong order. It now asserts the + /// relation that actually matters: the residency deadline is taken *after* + /// the insert it protects. + #[test] + fn residency_outlives_the_entry_it_protects() { + 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)); + topics.insert_established_for_test(community); + + // Stand in for the cache entry: record when its own TTL clock starts. + let inserted_at = Arc::new(std::sync::Mutex::new(None)); + let recorder = Arc::clone(&inserted_at); + assert!(residency.admit_and_insert(community, move || { + *recorder.lock().unwrap() = Some(Instant::now()); + })); + let inserted_at = inserted_at.lock().unwrap().expect("insert ran"); + + let deadline = residency.deadline_for_test(community).expect("resident"); + assert!( + deadline >= inserted_at + AUTHZ_CACHE_TTL, + "the residency window must start at or after the entry's own clock, \ + or the subscription can be withdrawn while the entry is still readable" + ); + + // 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); + expired.admit_and_insert(community, || {}); + assert!( + expired.resident_communities().is_empty(), + "residency must lapse on its own timer, with no external event" + ); + } + + /// The lifetime boundary of the same invariant: an entry cannot land after + /// its subscription has been withdrawn, because the insert runs while + /// withdrawal is structurally excluded. + /// + /// This is Wren's round-2 blocker, asserted without threads. From inside + /// the insert closure — the exact point where the old code had already + /// returned and left the caller unprotected — residency is lapsed, which is + /// the state that used to permit an unprotected entry, and a withdrawal is + /// then attempted. `try_withdraw_undesired_for_test` reports `WouldBlock`, + /// proving the reconciler cannot unsubscribe while this entry is landing. + /// After the call returns the same withdrawal both *acquires* the lock (so + /// the exclusion is scoped to the insert, not a lock this test wedged) and + /// withdraws nothing (so step 3's re-stamp put the entry back under cover). + /// + /// An earlier draft did this with a parked thread and a 50ms sleep. On + /// correct code that is deterministic, but its *mutation bite* was + /// scheduler-dependent — a late-scheduled withdrawer makes the unlocked + /// mutation pass. Wren caught that pre-push; a probabilistic detector is + /// not a deterministic test. This shape fails on every run under the + /// unlocked mutation. + #[test] + fn no_entry_can_land_after_its_subscription_is_withdrawn() { + let topics = Arc::new(CommunityTopics::new("test")); + let residency = Arc::new(CacheResidency::new(Arc::clone(&topics), AUTHZ_CACHE_TTL)); + let community = CommunityId::from_uuid(Uuid::from_u128(0xa)); + topics.insert_established_for_test(community); + + // Exactly the closure main.rs hands the cache-invalidation subscriber. + let for_closure = Arc::clone(&residency); + let desired: buzz_pubsub::community_topics::DesiredCommunities = + Arc::new(move || for_closure.resident_communities()); + + let lapse = Arc::clone(&residency); + let topics_for_insert = Arc::clone(&topics); + let desired_for_insert = Arc::clone(&desired); + let admitted = residency.admit_and_insert(community, move || { + // The delayed-caller state: residency has lapsed, so a reconciler + // reaching the withdrawal decision now *would* unsubscribe. + lapse.expire_all_for_test(); + assert!( + topics_for_insert + .try_withdraw_undesired_for_test(&desired_for_insert) + .is_none(), + "the reconciler must not be able to withdraw while an admitted \ + entry is still landing — that gap is the stale-authz hole" + ); + }); + + assert!(admitted, "an established community is admitted"); + let after = topics + .try_withdraw_undesired_for_test(&desired) + .expect("the exclusion must end with the insert, or the reconciler stalls"); + assert!( + after.is_empty(), + "the entry landed inside a residency window re-stamped around it, so \ + the very next withdrawal must leave its subscription alone" + ); + } + + /// 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 separate gate and record, 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)); + + let inserted = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&inserted); + assert!( + !residency.admit_and_insert(community, move || flag.store(true, Ordering::SeqCst)), + "nothing is cacheable before Redis acks the invalidation channel" + ); + assert!( + !inserted.load(Ordering::SeqCst), + "a refused admission must not run the insert" + ); + 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); + let flag = Arc::clone(&inserted); + assert!( + residency.admit_and_insert(community, move || flag.store(true, Ordering::SeqCst)), + "an acked community must be admitted" + ); + assert!( + inserted.load(Ordering::SeqCst), + "an admitted community must actually get its entry inserted" + ); + } + + /// The cross-crate half of the ordering invariant: a community + /// `admit_and_insert()` just made resident must survive a withdrawal + /// decision taken concurrently by the reconciler. + /// + /// `admit_and_insert` records residency *before* reading establishment, and the + /// reconciler re-reads residency under the write lock, so a renewal that got + /// its residency in cannot be unsubscribed. This test pairs the two real + /// call sites — the relay's residency closure and the pubsub withdrawal — + /// because each is correct alone and only the pairing carries the property. + #[test] + fn admitted_residency_is_visible_to_a_concurrent_withdrawal() { + let topics = Arc::new(CommunityTopics::new("test")); + let residency = Arc::new(CacheResidency::new(Arc::clone(&topics), AUTHZ_CACHE_TTL)); + let community = CommunityId::from_uuid(Uuid::from_u128(0xa)); + + // Exactly the closure main.rs hands the cache-invalidation subscriber. + let for_closure = Arc::clone(&residency); + let desired: buzz_pubsub::community_topics::DesiredCommunities = + Arc::new(move || for_closure.resident_communities()); + + // Established and resident: the steady state. + topics.insert_established_for_test(community); + assert!(residency.admit_and_insert(community, || {})); + + // Reconciler asks whether to withdraw. Residency is live, so it must not. + assert!( + topics.withdraw_undesired_for_test(&desired).is_empty(), + "a resident community must never be unsubscribed" + ); + assert!( + residency.admit_and_insert(community, || {}), + "and it stays admissible" + ); + } + + /// 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)); + + residency.admit_and_insert(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()); + } +} + +/// Redis-backed composed drill for the cache-residency gate. +/// +/// `#[ignore]`d because it needs a live server. **It is selected by name in +/// `.github/workflows/ci.yml`**: the step that runs the reconciler's own Redis +/// suite selects `package(buzz-pubsub)`, which does not reach this crate, so a +/// drill added here without its own selector would pass locally and gate +/// nothing. Whoever moves or renames this module must move the selector too. +/// +/// The unit tests above assert each half of the invariant in isolation, against +/// `insert_established_for_test`. This module asserts the composition against a +/// real Redis connection that really goes away — which is the only place the +/// suppression can be observed for the reason production would observe it. +#[cfg(test)] +mod redis_tests { + use super::*; + use buzz_pubsub::community_topics::DesiredCommunities; + + /// Communities this drill makes resident. + /// + /// Doubles as the fingerprint that singles out this subscriber's own Redis + /// connection for a targeted `CLIENT KILL`: seven exact subscriptions on one + /// pub/sub connection is a count no other test in the workspace holds. + const DRILL_COMMUNITIES: usize = 7; + + fn test_redis_url() -> String { + std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string()) + } + + /// A plain connection for issuing `CLIENT` commands. Panics when `REDIS_URL` + /// is unreachable: a Redis test that self-skips its own dependency is + /// exactly the fake-green this lane exists to remove. + 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") + } + + /// Client ids holding exactly `n` exact subscriptions and no patterns. + /// + /// Test-private twin of `clients_with_exact_subs` in + /// `buzz-pubsub/src/community_topics.rs`; duplicated rather than exported + /// because a `pub` cross-crate seam whose only consumer is one test is + /// production API invented for coverage. `psub=0` is also the standing proof + /// that this subscriber holds no wildcard — the command ElastiCache + /// Serverless rejects. + 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() + } + + /// Polls `cond` until it holds or `within` elapses. Every wait here is on an + /// observable state change rather than a guessed sleep: a fixed sleep either + /// flakes under load or passes vacuously when the thing it waited for never + /// happened. + /// + /// The bound is a parameter because one of the waits below is a + /// *discriminator*, not just a convergence check — see the sever step. + async fn wait_for(cond: impl Fn() -> bool, within: Duration, what: &str) { + let deadline = Instant::now() + within; + loop { + if cond() { + return; + } + assert!( + Instant::now() < deadline, + "timed out after {within:?} waiting for {what}" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + + /// The composed invariant this whole lane exists to protect: while a + /// community's invalidation topic is not established, this pod must not + /// cache an authorization decision under it — and it must resume caching on + /// its own once the topic comes back. + /// + /// Each half is unit-tested above against `insert_established_for_test`. + /// Neither half proves the property, because the dangerous state is not + /// "the bit is false", it is "the bit is false *because the connection is + /// gone*, and a caller is resolving authorization right now". So this drill + /// composes them against live Redis: + /// + /// 1. **Bootstrap** — the first admission is refused yet still records + /// residency, so the community becomes desired and gets subscribed. A + /// gate that recorded only on success would leave the caches off for the + /// life of the pod, and would pass every isolated test. + /// 2. **Open gate** — once acked, the entry lands, and a real published + /// invalidation really arrives. Establishment bookkeeping is not evidence + /// that a channel routes. + /// 3. **Sever** — `CLIENT KILL ID` on this subscriber's own connection. A + /// real server-side disconnect, so the reconnect path runs for the reason + /// it runs in production, not because a test injected an error. + /// 4. **Suppression** — during the gap the admission is refused and the + /// insert does not run. This is the silent-stale-authz hole: an entry + /// admitted here would stay readable for a full TTL with nothing able to + /// drop it, and no error anywhere. Degraded (read through to the DB) is + /// the correct outcome; stale is not. + /// 5. **Resume** — recovery is driven only by residency and the reconciler's + /// own tick. Nothing in this test republishes, re-registers, or pokes the + /// subscriber, because any cleared state that needs an external event to + /// come back is a pod that never caches again. + #[tokio::test] + #[ignore = "requires Redis"] + async fn a_severed_invalidation_topic_suppresses_caching_until_it_is_rebuilt() { + // Fresh ids per run: this drill shares one Redis with the rest of the + // suite, and a fixed community would let another run's traffic bleed in. + let communities: Vec = (0..DRILL_COMMUNITIES) + .map(|_| CommunityId::from_uuid(Uuid::new_v4())) + .collect(); + let gated = communities[0]; + + let mut conn = control_conn().await; + let before = clients_with_exact_subs(&mut conn, DRILL_COMMUNITIES).await; + + // Exactly the objects main.rs wires together for the real relay: the + // production `PubSubManager`, its own cache-invalidation topics, and a + // `CacheResidency` over them at the production TTL. + let pool = deadpool_redis::Config::from_url(test_redis_url()) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("redis pool"); + let pubsub = Arc::new( + PubSubManager::new(&test_redis_url(), pool) + .await + .expect("pubsub manager"), + ); + let topics = Arc::clone(pubsub.cache_invalidation_topics()); + let residency = Arc::new(CacheResidency::new(Arc::clone(&topics), AUTHZ_CACHE_TTL)); + let mut invalidations = pubsub.subscribe_cache_invalidations(); + + // Step 1: bootstrap. Refused, but residency-forming. + let inserts = Arc::new(AtomicU64::new(0)); + for community in &communities { + let counter = Arc::clone(&inserts); + assert!( + !residency.admit_and_insert(*community, move || { + counter.fetch_add(1, Ordering::SeqCst); + }), + "nothing is cacheable before Redis acks the invalidation channel" + ); + } + assert_eq!( + inserts.load(Ordering::SeqCst), + 0, + "a refused admission must not run its insert" + ); + + // The production closure from main.rs: residency read live, never + // snapshotted, so the subscriber is level-triggered off it. + let residency_for_closure = Arc::clone(&residency); + let desired: DesiredCommunities = + Arc::new(move || residency_for_closure.resident_communities()); + let pubsub_for_cache = Arc::clone(&pubsub); + tokio::spawn(async move { + pubsub_for_cache + .run_cache_invalidation_subscriber(desired) + .await + }); + + let all_established = { + let topics = Arc::clone(&topics); + let communities = communities.clone(); + move || communities.iter().all(|c| topics.is_established(*c)) + }; + wait_for( + &all_established, + Duration::from_secs(15), + "the drill's exact subscriptions to be acked", + ) + .await; + + // Step 2: the gate is open, so the entry lands. + let counter = Arc::clone(&inserts); + assert!( + residency.admit_and_insert(gated, move || { + counter.fetch_add(1, Ordering::SeqCst); + }), + "an acked community must be admitted" + ); + assert_eq!( + inserts.load(Ordering::SeqCst), + 1, + "an admitted community must actually get its entry inserted" + ); + + // ...and the channel it was admitted against really routes. + let ctx = TenantContext::resolved(gated, "drill.example"); + let sent = CacheInvalidation::Membership { + channel_id: Uuid::new_v4(), + pubkey: vec![0x7a; 32], + }; + pubsub + .publish_cache_invalidation(&ctx, &sent) + .await + .expect("publish before the sever"); + let received = tokio::time::timeout(Duration::from_secs(5), invalidations.recv()) + .await + .expect("timed out waiting for the pre-sever invalidation") + .expect("broadcast closed"); + assert_eq!(received.community_id, gated); + assert_eq!(received.invalidation, sent); + + // Step 3: sever ONLY this subscriber, by id. `CLIENT KILL TYPE pubsub` + // would also kill every other pub/sub client on the server — a + // concurrent test, or an unrelated process sharing the dev Redis. + let after = clients_with_exact_subs(&mut conn, DRILL_COMMUNITIES).await; + let ours: Vec = after.difference(&before).copied().collect(); + assert_eq!( + ours.len(), + 1, + "expected exactly one new {DRILL_COMMUNITIES}-subscription client to be ours, got {ours:?}" + ); + let killed: i64 = redis::cmd("CLIENT") + .arg("KILL") + .arg("ID") + .arg(ours[0]) + .query_async(&mut conn) + .await + .expect("CLIENT KILL ID"); + assert_eq!(killed, 1, "expected to sever exactly our own subscriber"); + + // Step 4: the suppression. The sever must be observable as lost + // establishment first — a gap that reads "established" is the hole. + // + // The bound here is the discriminator, and it is deliberately far below + // the reconnect backoff (`BACKOFF_INITIAL_SECS` = 1s). `connect_and_serve` + // *also* clears establishment on its next connect, so a generous timeout + // would be satisfied by that later clear and would pass even if the + // reconnect-gap clear were deleted — the assertion would hold for the + // wrong reason. Requiring the gate to close within 500ms of the sever + // can only be met by the clear that runs when the connection drops, which + // is the one that protects the gap. + let gate_closed = { + let topics = Arc::clone(&topics); + move || !topics.is_established(gated) + }; + wait_for( + &gate_closed, + Duration::from_millis(500), + "the sever itself to close the cache gate, before any reconnect could", + ) + .await; + + let before_gap = inserts.load(Ordering::SeqCst); + let counter = Arc::clone(&inserts); + assert!( + !residency.admit_and_insert(gated, move || { + counter.fetch_add(1, Ordering::SeqCst); + }), + "a community whose invalidation topic is severed must not be cacheable: \ + an entry admitted here stays readable for a full TTL with nothing able \ + to drop it, and nothing anywhere reports an error" + ); + assert_eq!( + inserts.load(Ordering::SeqCst), + before_gap, + "the suppressed admission must not have run its insert" + ); + + // Step 5: resume, driven only by residency and the reconciler's tick. + // + // Only `gated` is asserted back, and that is the correct claim rather + // than a weaker one: the suppressed admission above re-stamped *its* + // residency, so `gated` is still desired. The other six were last + // touched at bootstrap, so they age out after `AUTHZ_CACHE_TTL` and the + // level-triggered reconciler is right not to re-subscribe them. Waiting + // on all seven here would assert a property the design deliberately does + // not have, and would pass or fail on whether reconnect beat a 10s TTL. + let gate_open = { + let topics = Arc::clone(&topics); + move || topics.is_established(gated) + }; + wait_for( + &gate_open, + Duration::from_secs(15), + "the reconnected subscriber to rebuild the still-resident subscription", + ) + .await; + + let counter = Arc::clone(&inserts); + assert!( + residency.admit_and_insert(gated, move || { + counter.fetch_add(1, Ordering::SeqCst); + }), + "caching must resume once the invalidation topic is re-established" + ); + assert_eq!( + inserts.load(Ordering::SeqCst), + before_gap + 1, + "the resumed admission must run its insert" + ); + + // And the rebuilt subscription carries traffic. Re-establishment + // bookkeeping is not evidence that Redis will deliver again. + let resumed = CacheInvalidation::Membership { + channel_id: Uuid::new_v4(), + pubkey: vec![0x8b; 32], + }; + pubsub + .publish_cache_invalidation(&ctx, &resumed) + .await + .expect("publish after the rebuild"); + let received = tokio::time::timeout(Duration::from_secs(5), invalidations.recv()) + .await + .expect("timed out waiting for the post-rebuild invalidation") + .expect("broadcast closed"); + assert_eq!(received.community_id, gated); + assert_eq!( + received.invalidation, resumed, + "the rebuilt subscription must deliver, not just report itself acked" + ); + } } diff --git a/crates/buzz-relay/src/tunnel/directory.rs b/crates/buzz-relay/src/tunnel/directory.rs index 7cd6e22a2d..4352418b58 100644 --- a/crates/buzz-relay/src/tunnel/directory.rs +++ b/crates/buzz-relay/src/tunnel/directory.rs @@ -14,18 +14,42 @@ use uuid::Uuid; const DEFAULT_LEASE_TTL: Duration = Duration::from_secs(30); +/// Redis key format used for tunnel fences during the staged cluster migration. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum SessionKeyFormat { + /// Original untagged keys. This remains the default during the drain deploy. + #[default] + Legacy, + /// Cluster-safe keys co-slotted by a first-position session hash tag. + Tagged, +} + const ACQUIRE_SCRIPT: &str = r#" local lease_key = KEYS[1] local generation_key = KEYS[2] local owner = ARGV[1] local profile = ARGV[2] local ttl_ms = tonumber(ARGV[3]) +local legacy_lease = ARGV[4] +local legacy_generation = ARGV[5] local current = redis.call('GET', lease_key) if current then return {'exists', current, redis.call('GET', generation_key) or ''} end +-- A live legacy lease is allowed to drain before this key format takes over. +-- Deployments must drain legacy writers before enabling the tagged format; the +-- separate legacy reads cannot be made atomic across Redis Cluster slots. +if not redis.call('GET', generation_key) and legacy_lease ~= '' then + return {'exists', legacy_lease, legacy_generation} +end + +-- Preserve the old non-expiring fence watermark on first use. SETNX accepts the +-- integer as an exact string, avoiding Lua-number precision loss. +if legacy_generation ~= '' then + redis.call('SETNX', generation_key, legacy_generation) +end local generation = redis.call('INCR', generation_key) local value = owner .. '|' .. tostring(generation) .. '|' .. profile redis.call('SET', lease_key, value, 'PX', ttl_ms) @@ -76,10 +100,15 @@ return {'lost', current, redis.call('GET', generation_key) or current_generation const VALIDATE_SCRIPT: &str = r#" local lease_key = KEYS[1] local generation_key = KEYS[2] +local legacy_lease = ARGV[1] +local legacy_generation = ARGV[2] local current = redis.call('GET', lease_key) or '' -local known_generation = redis.call('GET', generation_key) or '' -return {current, known_generation} +local known_generation = redis.call('GET', generation_key) +if known_generation then + return {current, known_generation} +end +return {legacy_lease, legacy_generation} "#; /// Redis-backed owner directory for mesh tunnel sessions. @@ -87,6 +116,7 @@ return {current, known_generation} pub struct SessionDirectory { pool: deadpool_redis::Pool, lease_ttl: Duration, + key_format: SessionKeyFormat, } /// Active session ownership lease read from Redis. @@ -184,12 +214,25 @@ pub enum DirectoryError { impl SessionDirectory { /// Create a directory backed by `pool` with the default lease TTL. pub fn new(pool: deadpool_redis::Pool) -> Self { - Self::with_lease_ttl(pool, DEFAULT_LEASE_TTL) + Self::with_key_format(pool, DEFAULT_LEASE_TTL, SessionKeyFormat::Legacy) } /// Create a directory backed by `pool` with an explicit lease TTL. pub fn with_lease_ttl(pool: deadpool_redis::Pool, lease_ttl: Duration) -> Self { - Self { pool, lease_ttl } + Self::with_key_format(pool, lease_ttl, SessionKeyFormat::Legacy) + } + + /// Create a directory using an explicitly selected migration key format. + pub fn with_key_format( + pool: deadpool_redis::Pool, + lease_ttl: Duration, + key_format: SessionKeyFormat, + ) -> Self { + Self { + pool, + lease_ttl, + key_format, + } } /// Attempt to create/take over the session lease. @@ -207,13 +250,27 @@ impl SessionDirectory { let keys = SessionKeys::new(community_id, session_id); let ttl_ms = ttl_ms(self.lease_ttl)?; let mut conn = self.pool.get().await?; + let (lease_key, generation_key, legacy) = match self.key_format { + SessionKeyFormat::Legacy => ( + &keys.legacy_lease, + &keys.legacy_generation, + LegacyValues::default(), + ), + SessionKeyFormat::Tagged => ( + &keys.lease, + &keys.generation, + read_legacy_keys(&mut conn, &keys).await?, + ), + }; let (status, value, _known_generation): (String, String, String) = Script::new(ACQUIRE_SCRIPT) - .key(&keys.lease) - .key(&keys.generation) + .key(lease_key) + .key(generation_key) .arg(owner_runtime_id.to_hex()) .arg(profile.as_wire_str()) .arg(ttl_ms) + .arg(&legacy.lease) + .arg(&legacy.generation) .invoke_async(&mut *conn) .await?; let lease = parse_lease(community_id, session_id, &value)?; @@ -248,8 +305,8 @@ impl SessionDirectory { let ttl_ms = ttl_ms(self.lease_ttl)?; let mut conn = self.pool.get().await?; let (status, value, known_generation): (String, String, String) = Script::new(RENEW_SCRIPT) - .key(&keys.lease) - .key(&keys.generation) + .key(keys.lease(self.key_format)) + .key(keys.generation(self.key_format)) .arg(lease.owner_runtime_id.to_hex()) .arg(lease.generation) .arg(ttl_ms) @@ -279,8 +336,8 @@ impl SessionDirectory { let mut conn = self.pool.get().await?; let (status, value, known_generation): (String, String, String) = Script::new(RELEASE_SCRIPT) - .key(&keys.lease) - .key(&keys.generation) + .key(keys.lease(self.key_format)) + .key(keys.generation(self.key_format)) .arg(lease.owner_runtime_id.to_hex()) .arg(lease.generation) .invoke_async(&mut *conn) @@ -310,14 +367,8 @@ impl SessionDirectory { ) -> Result, DirectoryError> { let keys = SessionKeys::new(community_id, session_id); let mut conn = self.pool.get().await?; - let value: Option = redis::cmd("GET") - .arg(&keys.lease) - .query_async(&mut *conn) - .await?; - value - .as_deref() - .map(|v| parse_lease(community_id, session_id, v)) - .transpose() + let (value, _known_generation) = read_keys(&mut conn, &keys, self.key_format).await?; + parse_optional_lease(community_id, session_id, &value) } /// Read the non-expiring generation counter for a session, if it exists. @@ -328,14 +379,8 @@ impl SessionDirectory { ) -> Result, DirectoryError> { let keys = SessionKeys::new(community_id, session_id); let mut conn = self.pool.get().await?; - let value: Option = redis::cmd("GET") - .arg(&keys.generation) - .query_async(&mut *conn) - .await?; - match value.as_deref() { - Some(value) => parse_optional_generation(community_id, session_id, value), - None => Ok(None), - } + let (_lease, generation) = read_keys(&mut conn, &keys, self.key_format).await?; + parse_optional_generation(community_id, session_id, &generation) } /// Validate a session-bearing mesh frame fence against Redis. @@ -356,11 +401,9 @@ impl SessionDirectory { .get() .await .map_err(|e| MeshError::Transport(format!("redis pool: {e}")))?; - let (lease_value, known_generation): (String, String) = Script::new(VALIDATE_SCRIPT) - .key(&keys.lease) - .key(&keys.generation) - .invoke_async(&mut *conn) - .await?; + let (lease_value, known_generation) = read_keys(&mut conn, &keys, self.key_format) + .await + .map_err(|e| MeshError::Transport(e.to_string()))?; let known_from_counter = parse_optional_generation(community_id, fenced.session_id, &known_generation) .map_err(|e| MeshError::Transport(e.to_string()))? @@ -453,18 +496,120 @@ impl SessionLease { struct SessionKeys { lease: String, generation: String, + legacy_lease: String, + legacy_generation: String, } impl SessionKeys { + fn lease(&self, format: SessionKeyFormat) -> &str { + match format { + SessionKeyFormat::Legacy => &self.legacy_lease, + SessionKeyFormat::Tagged => &self.lease, + } + } + + fn generation(&self, format: SessionKeyFormat) -> &str { + match format { + SessionKeyFormat::Legacy => &self.legacy_generation, + SessionKeyFormat::Tagged => &self.generation, + } + } + fn new(community_id: CommunityId, session_id: Uuid) -> Self { - let base = format!("buzz:{}:tunnel:{}", community_id, session_id); + // The session hash tag is deliberately the first `{...}` segment. Redis + // Cluster hashes both script keys to this session's slot, while standard + // Redis treats the braces as ordinary key bytes. + let base = format!("buzz:{{{session_id}}}:{community_id}:tunnel"); + let legacy_base = format!("buzz:{community_id}:tunnel:{session_id}"); Self { lease: format!("{base}:lease"), generation: format!("{base}:generation"), + legacy_lease: format!("{legacy_base}:lease"), + legacy_generation: format!("{legacy_base}:generation"), } } } +#[derive(Default)] +struct LegacyValues { + lease: String, + generation: String, +} + +async fn read_legacy_keys( + conn: &mut deadpool_redis::Connection, + keys: &SessionKeys, +) -> Result { + // This pipeline is intentionally non-atomic: MULTI/EXEC would put the two + // legacy keys in one cross-slot transaction on Redis Cluster. These reads + // bridge the drain-aware key migration; tagged state always wins once its + // non-expiring generation key has been initialized. + let (lease, generation): (Option, Option) = redis::pipe() + .cmd("GET") + .arg(&keys.legacy_lease) + .cmd("GET") + .arg(&keys.legacy_generation) + .query_async(&mut **conn) + .await?; + Ok(LegacyValues { + lease: lease.unwrap_or_default(), + generation: generation.unwrap_or_default(), + }) +} + +async fn read_current_keys( + conn: &mut deadpool_redis::Connection, + keys: &SessionKeys, + legacy: &LegacyValues, +) -> Result<(String, String), redis::RedisError> { + read_current_keys_for(conn, keys, SessionKeyFormat::Tagged, legacy).await +} + +async fn read_keys( + conn: &mut deadpool_redis::Connection, + keys: &SessionKeys, + format: SessionKeyFormat, +) -> Result<(String, String), redis::RedisError> { + match format { + SessionKeyFormat::Legacy => { + let empty = LegacyValues::default(); + read_current_keys_for(conn, keys, format, &empty).await + } + SessionKeyFormat::Tagged => read_keys_with_legacy_fallback(conn, keys).await, + } +} + +async fn read_current_keys_for( + conn: &mut deadpool_redis::Connection, + keys: &SessionKeys, + format: SessionKeyFormat, + legacy: &LegacyValues, +) -> Result<(String, String), redis::RedisError> { + Script::new(VALIDATE_SCRIPT) + .key(keys.lease(format)) + .key(keys.generation(format)) + .arg(&legacy.lease) + .arg(&legacy.generation) + .invoke_async(&mut **conn) + .await +} + +async fn read_keys_with_legacy_fallback( + conn: &mut deadpool_redis::Connection, + keys: &SessionKeys, +) -> Result<(String, String), redis::RedisError> { + let empty = LegacyValues { + lease: String::new(), + generation: String::new(), + }; + let current = read_current_keys(conn, keys, &empty).await?; + if !current.1.is_empty() { + return Ok(current); + } + let legacy = read_legacy_keys(conn, keys).await?; + read_current_keys(conn, keys, &legacy).await +} + trait ProfileWireExt { fn as_wire_str(&self) -> &'static str; } @@ -597,25 +742,36 @@ mod tests { .expect("create redis pool") } - async fn redis_directory_if_available() -> Option { + async fn redis_directory() -> SessionDirectory { let pool = pool(); - let mut conn = pool.get().await.ok()?; + let mut conn = pool + .get() + .await + .expect("REDIS_URL must be reachable for ignored tunnel directory tests"); redis::cmd("PING") .query_async::(&mut *conn) .await - .ok()?; - Some(SessionDirectory::with_lease_ttl( + .expect("REDIS_URL must answer PING for ignored tunnel directory tests"); + drop(conn); + SessionDirectory::with_key_format( pool, Duration::from_millis(150), - )) + SessionKeyFormat::Tagged, + ) } async fn clear_keys(directory: &SessionDirectory, community_id: CommunityId, session_id: Uuid) { let keys = SessionKeys::new(community_id, session_id); let mut conn = directory.pool.get().await.expect("redis conn"); - let _: () = redis::cmd("DEL") + let _: () = redis::pipe() + .cmd("DEL") .arg(keys.lease) + .cmd("DEL") .arg(keys.generation) + .cmd("DEL") + .arg(keys.legacy_lease) + .cmd("DEL") + .arg(keys.legacy_generation) .query_async(&mut *conn) .await .expect("clear keys"); @@ -647,20 +803,185 @@ mod tests { let keys = SessionKeys::new(community(), session()); assert_eq!( keys.lease, - format!("buzz:{}:tunnel:{}:lease", community(), session()) + format!("buzz:{{{}}}:{}:tunnel:lease", session(), community()) ); assert_eq!( keys.generation, + format!("buzz:{{{}}}:{}:tunnel:generation", session(), community()) + ); + assert_eq!( + keys.legacy_lease, + format!("buzz:{}:tunnel:{}:lease", community(), session()) + ); + assert_eq!( + keys.legacy_generation, format!("buzz:{}:tunnel:{}:generation", community(), session()) ); + assert_eq!(keys.lease.find('{'), Some(5)); + assert_eq!(keys.generation.find('{'), Some(5)); + let tag = format!("{{{}}}", session()); + assert!(keys.lease.starts_with(&format!("buzz:{tag}:"))); + assert!(keys.generation.starts_with(&format!("buzz:{tag}:"))); assert_ne!(keys.lease, keys.generation); } #[tokio::test] - async fn acquire_conflict_renew_release_and_monotonic_takeover() { - let Some(directory) = redis_directory_if_available().await else { - return; + #[ignore = "requires REDIS_URL; run by backend-integration"] + async fn legacy_generation_and_live_lease_migrate_without_fence_regression() { + let directory = redis_directory().await; + let community_id = community(); + let session_id = Uuid::new_v4(); + clear_keys(&directory, community_id, session_id).await; + let keys = SessionKeys::new(community_id, session_id); + let legacy_lease = format!("{}|41|reliable-stream", runtime(1).to_hex()); + let mut conn = directory.pool.get().await.expect("redis conn"); + let _: () = redis::pipe() + .cmd("SET") + .arg(&keys.legacy_generation) + .arg(41_u64) + .cmd("SET") + .arg(&keys.legacy_lease) + .arg(&legacy_lease) + .arg("PX") + .arg(5_000_u64) + .query_async(&mut *conn) + .await + .expect("seed legacy keys"); + drop(conn); + + let existing = directory + .acquire( + community_id, + session_id, + runtime(2), + Profile::ReliableStream, + ) + .await + .expect("legacy lease remains authoritative"); + assert!(matches!(existing, AcquireResult::Exists(ref lease) if lease.generation == 41)); + let legacy = match existing { + AcquireResult::Exists(lease) => lease, + AcquireResult::Acquired(_) => unreachable!(), + }; + assert_eq!( + directory.lookup(community_id, session_id).await.unwrap(), + Some(legacy) + ); + let mut conn = directory.pool.get().await.expect("redis conn"); + let _: () = redis::cmd("DEL") + .arg(&keys.legacy_lease) + .query_async(&mut *conn) + .await + .expect("drain legacy lease"); + drop(conn); + + let migrated = match directory + .acquire( + community_id, + session_id, + runtime(2), + Profile::ReliableStream, + ) + .await + .expect("acquire tagged lease") + { + AcquireResult::Acquired(lease) => lease, + AcquireResult::Exists(_) => panic!("released legacy lease must drain"), }; + assert_eq!(migrated.generation, 42); + assert_eq!( + directory + .known_generation(community_id, session_id) + .await + .unwrap(), + Some(42) + ); + } + + #[tokio::test] + #[ignore = "requires REDIS_URL; run by backend-integration"] + async fn tagged_state_is_authoritative_when_both_formats_conflict() { + let directory = redis_directory().await; + let community_id = community(); + let session_id = Uuid::new_v4(); + clear_keys(&directory, community_id, session_id).await; + let keys = SessionKeys::new(community_id, session_id); + let tagged = SessionLease { + community_id, + session_id, + owner_runtime_id: runtime(2), + generation: 42, + profile: Profile::ReliableStream, + }; + let legacy_value = format!("{}|99|reliable-stream", runtime(1).to_hex()); + let tagged_value = format!("{}|42|reliable-stream", runtime(2).to_hex()); + let mut conn = directory.pool.get().await.expect("redis conn"); + let _: () = redis::pipe() + .cmd("SET") + .arg(&keys.legacy_generation) + .arg(99_u64) + .cmd("SET") + .arg(&keys.legacy_lease) + .arg(&legacy_value) + .arg("PX") + .arg(5_000_u64) + .cmd("SET") + .arg(&keys.generation) + .arg(42_u64) + .cmd("SET") + .arg(&keys.lease) + .arg(&tagged_value) + .arg("PX") + .arg(5_000_u64) + .query_async(&mut *conn) + .await + .expect("seed conflicting key formats"); + drop(conn); + + assert_eq!( + directory.lookup(community_id, session_id).await.unwrap(), + Some(tagged.clone()) + ); + assert_eq!( + directory + .known_generation(community_id, session_id) + .await + .unwrap(), + Some(42) + ); + assert!(directory + .validate_fenced_header(community_id, &tagged.fenced_header()) + .await + .is_ok()); + assert!(matches!( + directory + .validate_fenced_header( + community_id, + &FencedHeader { + session_id, + generation: 99, + owner_runtime_id: runtime(1), + }, + ) + .await, + Err(MeshError::FutureGeneration { + known_generation: 42, + .. + }) + )); + assert!(matches!( + directory + .acquire(community_id, session_id, runtime(3), Profile::ReliableStream) + .await + .unwrap(), + AcquireResult::Exists(ref lease) if *lease == tagged + )); + } + + #[tokio::test] + #[ignore = "requires REDIS_URL; run by backend-integration"] + async fn acquire_conflict_renew_release_and_monotonic_takeover() { + let directory = redis_directory().await; let community_id = community(); let session_id = Uuid::new_v4(); clear_keys(&directory, community_id, session_id).await; @@ -733,10 +1054,9 @@ mod tests { } #[tokio::test] + #[ignore = "requires REDIS_URL; run by backend-integration"] async fn takeover_after_ttl_expiry_increments_non_expiring_counter() { - let Some(directory) = redis_directory_if_available().await else { - return; - }; + let directory = redis_directory().await; let community_id = community(); let session_id = Uuid::new_v4(); clear_keys(&directory, community_id, session_id).await; @@ -785,10 +1105,9 @@ mod tests { } #[tokio::test] + #[ignore = "requires REDIS_URL; run by backend-integration"] async fn validate_returns_typed_fence_rejections() { - let Some(directory) = redis_directory_if_available().await else { - return; - }; + let directory = redis_directory().await; let community_id = community(); let session_id = Uuid::new_v4(); clear_keys(&directory, community_id, session_id).await; @@ -880,10 +1199,9 @@ mod tests { } #[tokio::test] + #[ignore = "requires REDIS_URL; run by backend-integration"] async fn validate_returns_no_active_lease_after_expiry_before_takeover() { - let Some(directory) = redis_directory_if_available().await else { - return; - }; + let directory = redis_directory().await; let community_id = community(); let session_id = Uuid::new_v4(); clear_keys(&directory, community_id, session_id).await; diff --git a/crates/buzz-relay/src/tunnel/reliable.rs b/crates/buzz-relay/src/tunnel/reliable.rs index 5153e104df..fae9572646 100644 --- a/crates/buzz-relay/src/tunnel/reliable.rs +++ b/crates/buzz-relay/src/tunnel/reliable.rs @@ -694,7 +694,8 @@ mod tests { } async fn clear_keys(directory: &SessionDirectory, community_id: CommunityId, session_id: Uuid) { - let base = format!("buzz:{}:tunnel:{}", community_id, session_id); + let base = format!("buzz:{{{session_id}}}:{community_id}:tunnel"); + let legacy_base = format!("buzz:{community_id}:tunnel:{session_id}"); let _ = directory .release(&SessionLease { community_id, @@ -705,9 +706,15 @@ mod tests { }) .await; let mut conn = pool().get().await.expect("redis conn"); - let _: () = redis::cmd("DEL") + let _: () = redis::pipe() + .cmd("DEL") .arg(format!("{base}:lease")) + .cmd("DEL") .arg(format!("{base}:generation")) + .cmd("DEL") + .arg(format!("{legacy_base}:lease")) + .cmd("DEL") + .arg(format!("{legacy_base}:generation")) .query_async(&mut *conn) .await .expect("clear keys"); diff --git a/docs/tunnel-fence-key-migration.md b/docs/tunnel-fence-key-migration.md new file mode 100644 index 0000000000..4ef517135e --- /dev/null +++ b/docs/tunnel-fence-key-migration.md @@ -0,0 +1,103 @@ +# Tunnel fence Redis key migration + +Tunnel lease and generation keys must share a Redis Cluster hash slot. The +cluster-safe tagged format is deliberately gated by +`BUZZ_TUNNEL_FENCE_KEY_FORMAT`; its default is `legacy` so an ordinary image +rollout cannot activate new-format writers while old writers still run. + +## Required two-phase deployment + +1. **Drain deploy:** deploy the new binary everywhere with + `BUZZ_TUNNEL_FENCE_KEY_FORMAT=legacy` (or unset). Keep the existing + standalone Redis endpoint. Wait until every old binary has terminated and + its maximum graceful-shutdown/lease window has elapsed. Verify no old pod + can acquire or renew a legacy lease. +2. **Enable deploy (non-rolling):** scale the relay deployment to zero and wait + until the maximum graceful-shutdown/lease window has elapsed. Confirm that no + relay process remains and no legacy lease can still be renewed. While writers + remain stopped, build a manifest from the source endpoint containing every + legacy `buzz:*:tunnel:*:generation` key and its exact string value. Transfer + those non-expiring generation keys, under the same legacy key names, to the + cluster-enforcing target endpoint. Do **not** copy lease keys: all expiring + leases must remain drained and must not be resurrected on the target. One + executable procedure is below; run it from a controlled host with + `redis-cli` access to both endpoints: + + ```bash + set -euo pipefail + : "${SOURCE_REDIS_URL:?set the drained source Redis URL}" + : "${TARGET_REDIS_URL:?set the target Redis URL}" + + work=$(mktemp -d) + trap 'rm -rf "$work"' EXIT + + manifest() { + endpoint=$1 + output=$2 + redis-cli -u "$endpoint" --scan \ + --pattern 'buzz:*:tunnel:*:generation' \ + | LC_ALL=C sort -u \ + | while IFS= read -r key; do + value=$(redis-cli -u "$endpoint" --raw GET "$key") + test -n "$value" + printf '%s\t%s\n' "$key" "$value" + done >"$output" + } + + # Never overwrite initialized tagged state without operator reconciliation. + test -z "$(redis-cli -u "$TARGET_REDIS_URL" --scan \ + --pattern 'buzz:{*}:*:tunnel:generation' | head -n 1)" + + manifest "$SOURCE_REDIS_URL" "$work/source-before" + while IFS=$'\t' read -r key value; do + redis-cli -u "$TARGET_REDIS_URL" SET "$key" "$value" >/dev/null + done <"$work/source-before" + + # Writers remain stopped while these final manifests are produced. + manifest "$SOURCE_REDIS_URL" "$work/source-after" + manifest "$TARGET_REDIS_URL" "$work/target-after" + cmp "$work/source-before" "$work/source-after" + cmp "$work/source-after" "$work/target-after" + + # Neither legacy nor tagged leases may exist on the target. + test -z "$(redis-cli -u "$TARGET_REDIS_URL" --scan \ + --pattern 'buzz:*:tunnel:*:lease' | head -n 1)" + test -z "$(redis-cli -u "$TARGET_REDIS_URL" --scan \ + --pattern 'buzz:{*}:*:tunnel:lease' | head -n 1)" + ``` + + Treat any non-zero exit as a failed transfer; do not continue to scale-up. + + Before starting any relay, rescan both endpoints and verify that the complete + source and target legacy-generation key sets are identical and that every + corresponding string value is byte-for-byte equal to the manifest. Also + verify that the source manifest did not change during transfer and that the + target has no legacy or tagged tunnel `*:lease` keys. If a generation key is + missing, extra, changed, unreadable, or cannot be written, **abort the + cutover**, keep all relays scaled to zero, and repair/repeat the transfer; + never start tagged writers from a partial or unverified copy. If the target + already contains tagged generation keys, stop and perform an + operator-reviewed reconciliation rather than overwriting or assuming that + the legacy manifest is authoritative. + + Only after that verification succeeds, set + `BUZZ_TUNNEL_FENCE_KEY_FORMAT=tagged` on the whole deployment, switch the + configured Redis endpoint to the verified target, and scale up. Do **not** + start the first tagged pod while any legacy-mode pod is alive. The first + tagged acquisition reads the copied legacy watermark from its currently + configured target pool, seeds its non-expiring tagged generation, and then + increments it. Tagged state is authoritative once initialized. + +Do not mix `legacy` and `tagged` writers. The two legacy keys occupy different +cluster slots, so Redis cannot atomically prove that an old writer did not race +the migration. + +## Rollback rule + +Before phase 2, roll back normally with the format left `legacy`. After any pod +has enabled `tagged`, **do not roll back to an old binary or set the format back +to `legacy`**. Roll back only to a tagged-capable binary with the format still +`tagged`; otherwise a legacy writer can reuse an already-issued generation and +break fencing. Restoring legacy mode requires a full mesh outage, drainage of +all tagged leases, and an operator-reviewed generation migration; it is not a +rolling rollback.