feat(redis): ElastiCache Serverless compatibility with full standard-Redis parity

.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,

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<ScopedCacheInvalidation>,
+    topics: Arc<CommunityTopics>,
+    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<ScopedCacheInvalidation>,
 }
 
-async fn connect_and_subscribe(
-    redis_url: &str,
-    broadcast_tx: &broadcast::Sender<ScopedCacheInvalidation>,
-) -> 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<CommunityId> {
+        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)]