mirror of
https://github.com/block/buzz.git
synced 2026-08-18 06:50:31 +02:00
fix(pubsub): replace PSUBSCRIBE patterns with exact per-community subscriptions
Two subscribers rode wildcard patterns: `buzz:*:cache-invalidate` and `buzz:*:conn-control`. ElastiCache Serverless restricts `PSUBSCRIBE`, so both fail loudly there. Replace each with exact `SUBSCRIBE`/`UNSUBSCRIBE` on the per-community channels this pod actually needs. Standalone Redis and cluster-mode behave identically — there is no separate serverless mode, and no `SSUBSCRIBE`. Reconciliation is level-triggered. A `desired` closure is re-read inside the subscriber task immediately before each command, never snapshotted, so a queued unsubscribe cannot be applied against interest that has since been renewed. Establishment is recorded only after Redis acks the `SUBSCRIBE`: `sink.subscribe().await` returning `Ok` *is* the acknowledgement, which is what lets a cross-crate gate read acknowledged state without adding a oneshot ack path to fire-and-forget retain. Convergence is driven by a 1s tick, with wakes as a latency shortcut only — a lost or coalesced wake costs latency, never convergence, and the tick is the restore path that fires with no external event. On disconnect the whole established set is cleared, and the next connect rebuilds from `desired` rather than from a remembered active set. The two families have different interest owners. Connection control follows live sockets: a pod needs a community's disconnect commands only while it holds a socket a remote ban could close. Cache invalidation follows *cache residency*, not connection lifetime, because a cached authorization decision outlives the socket that populated it; scoping it to connections would leave a pod serving authz for a community it no longer listens to. Residency lapses on a timer, so no call path owes an unsubscribe. It cannot be driven off a moka eviction listener: moka's notifications are lazy and do not fire at TTL on an idle cache, so an event-driven release would never run on the pod that most needs it. `CacheResidency::admit` deliberately both records residency and returns the gate. Split into a separate gate and a separate record, only an already-admitted community would ever be recorded — so on a fresh pod nothing enters the desired set, nothing is subscribed, nothing is established, and all three authorization caches read through to Postgres for the life of the process with no error and no log. Folding them makes that ordering unrepresentable. The gate covers cache *inserts only*, never reads. Gating reads would discard a live, still-invalidatable cache on every reconnect and turn a pub/sub blip into a DB stampede. A refused insert means the pod reads through — degraded, not stale — and bumps `buzz_authz_cache_insert_skipped_total`. `AUTHZ_CACHE_TTL` is now one constant shared by the three moka caches and by residency, because residency must outlast every entry it protects. Both subscribers moved from an early spawn to alongside their consumer loops: each needs a desired-set closure read off `AppState`, which did not exist at the old spawn point. Tests: five Redis-backed reconciliation tests, asserting on `PUBSUB NUMSUB` rather than `PUBLISH`'s return. `PUBLISH` counts pattern matches, so any unrelated `PSUBSCRIBE buzz:*` client on a shared server inflates it — and exact-subscriber count is precisely the property this change is about, since it cannot be satisfied by a leftover wildcard. 100 exact subscriptions on one connection route every payload to its own community (no cross-wiring under the wrong tenant label); a withdrawn subset unsubscribes while the rest keep routing; a newly desired community converges on the tick with no wake at all; and a severed connection rebuilds the entire desired set. The sever kills only its own client by id — `CLIENT KILL TYPE pubsub` would take down every other pub/sub client on the server, including concurrent tests. Relay-side tests assert the bootstrap closes: the refused first `admit` must still make the community desired, which is the property whose absence is the deadlock above. Co-authored-by: npub17jjz49l9jjmhhk7cac63j8yt9z555n9cw8vk7v5jz4vzw4ppld5qgj57cc <f4a42a97e594b77bdbd8ee35191c8b28a94a4cb871d96f32921558275421fb68@buzz.block.builderlab.xyz> Signed-off-by: npub17jjz49l9jjmhhk7cac63j8yt9z555n9cw8vk7v5jz4vzw4ppld5qgj57cc <f4a42a97e594b77bdbd8ee35191c8b28a94a4cb871d96f32921558275421fb68@buzz.block.builderlab.xyz>
This commit is contained in:
npub17jjz49l9jjmhhk7cac63j8yt9z555n9cw8vk7v5jz4vzw4ppld5qgj57cc
parent
ba87121a18
commit
ab490cff5a
@@ -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);
|
||||
|
||||
tracing::info!("Attempting to reconnect to Redis cache-invalidation...");
|
||||
}
|
||||
run_community_subscriber(
|
||||
redis_url,
|
||||
CacheInvalidationFamily { broadcast_tx },
|
||||
topics,
|
||||
desired,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
async fn connect_and_subscribe(
|
||||
redis_url: &str,
|
||||
broadcast_tx: &broadcast::Sender<ScopedCacheInvalidation>,
|
||||
) -> Result<(), redis::RedisError> {
|
||||
let client = redis::Client::open(redis_url)?;
|
||||
let mut conn = client.get_async_pubsub().await?;
|
||||
struct CacheInvalidationFamily {
|
||||
broadcast_tx: broadcast::Sender<ScopedCacheInvalidation>,
|
||||
}
|
||||
|
||||
conn.psubscribe(CACHE_INVALIDATION_PATTERN).await?;
|
||||
impl CommunityChannelFamily for CacheInvalidationFamily {
|
||||
fn channel(&self, community: CommunityId) -> String {
|
||||
cache_invalidation_channel_for(community)
|
||||
}
|
||||
|
||||
tracing::info!(
|
||||
"Redis cache-invalidation subscriber connected — listening on {CACHE_INVALIDATION_PATTERN}"
|
||||
);
|
||||
fn parse_channel(&self, channel: &str) -> Option<CommunityId> {
|
||||
parse_cache_invalidation_channel(channel)
|
||||
}
|
||||
|
||||
let mut stream = conn.on_message();
|
||||
while let Some(msg) = stream.next().await {
|
||||
let channel = msg.get_channel_name();
|
||||
let Some(community_id) = parse_cache_invalidation_channel(channel) else {
|
||||
tracing::warn!("Received cache-invalidation message on unexpected channel: {channel}");
|
||||
continue;
|
||||
};
|
||||
|
||||
let payload: String = match msg.get_payload() {
|
||||
Ok(p) => p,
|
||||
Err(e) => {
|
||||
tracing::warn!("Failed to get cache-invalidation payload: {e}");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let invalidation: CacheInvalidation = match serde_json::from_str(&payload) {
|
||||
fn deliver(&self, community_id: CommunityId, payload: &str) {
|
||||
let invalidation: CacheInvalidation = match serde_json::from_str(payload) {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
tracing::warn!("Failed to deserialize cache-invalidation message: {e}");
|
||||
continue;
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
@@ -167,12 +155,10 @@ async fn connect_and_subscribe(
|
||||
invalidation,
|
||||
};
|
||||
|
||||
if broadcast_tx.send(scoped).is_err() {
|
||||
if self.broadcast_tx.send(scoped).is_err() {
|
||||
tracing::trace!("No cache-invalidation receivers — message dropped");
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -0,0 +1,693 @@
|
||||
//! Level-triggered subscription state for community-scoped Redis pub/sub
|
||||
//! channel families (cache invalidation, connection control).
|
||||
//!
|
||||
//! ElastiCache Serverless rejects `PSUBSCRIBE`, so these families can no longer
|
||||
//! ride a single `buzz:*:...` wildcard. Each pod instead subscribes to the exact
|
||||
//! per-community channels it currently needs. Two sets carry that:
|
||||
//!
|
||||
//! * **desired** — owned by whoever holds the interest (live sockets for
|
||||
//! connection control; local cache residency for cache invalidation),
|
||||
//! supplied as a closure and recomputed from scratch on every reconcile. This
|
||||
//! module never caches it, so a queued command can never be applied against a
|
||||
//! stale view.
|
||||
//! * **established** — the communities Redis has acknowledged a `SUBSCRIBE` for
|
||||
//! on the *current* connection. A community enters only after the ack returns,
|
||||
//! and the whole set is cleared when the connection ends.
|
||||
//!
|
||||
//! Reconciliation is level-triggered: the subscriber task diffs desired against
|
||||
//! established on connect, on a wake, and on a fixed tick. A coalesced or lost
|
||||
//! wake therefore costs latency, never convergence — and the tick is the
|
||||
//! automatic restore path that any cleared state needs. A healthy subscription
|
||||
//! must never sit marked unestablished waiting on an unrelated external event.
|
||||
|
||||
use std::collections::HashSet;
|
||||
use std::sync::{Arc, RwLock};
|
||||
use std::time::Duration;
|
||||
|
||||
use buzz_core::CommunityId;
|
||||
use futures_util::StreamExt;
|
||||
use tokio::sync::Notify;
|
||||
|
||||
/// Initial reconnect backoff (1 second).
|
||||
const BACKOFF_INITIAL_SECS: u64 = 1;
|
||||
/// Maximum reconnect backoff (30 seconds).
|
||||
const BACKOFF_MAX_SECS: u64 = 30;
|
||||
/// Upper bound on how long desired and established can diverge without a wake.
|
||||
/// This is the unconditional convergence path, so it must never be disabled.
|
||||
const RECONCILE_INTERVAL: Duration = Duration::from_secs(1);
|
||||
|
||||
/// Communities whose channel this pod wants subscribed, recomputed on demand.
|
||||
pub type DesiredCommunities = Arc<dyn Fn() -> HashSet<CommunityId> + 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<HashSet<CommunityId>>,
|
||||
/// Latency shortcut: producers signal new interest instead of waiting for
|
||||
/// the next tick. Coalescing and loss are both harmless.
|
||||
wake: Notify,
|
||||
}
|
||||
|
||||
impl CommunityTopics {
|
||||
/// Creates empty subscription state for a named channel family.
|
||||
pub fn new(name: &'static str) -> Self {
|
||||
Self {
|
||||
name,
|
||||
established: RwLock::new(HashSet::new()),
|
||||
wake: Notify::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Log label for this channel family.
|
||||
pub fn name(&self) -> &'static str {
|
||||
self.name
|
||||
}
|
||||
|
||||
/// Whether Redis has acknowledged this community's subscription on the
|
||||
/// current connection. This is the gate for anything that must not become
|
||||
/// readable before its invalidation topic is established.
|
||||
pub fn is_established(&self, community: CommunityId) -> bool {
|
||||
self.established
|
||||
.read()
|
||||
.expect("community topics lock poisoned")
|
||||
.contains(&community)
|
||||
}
|
||||
|
||||
/// Number of established communities, for the connect log line and for
|
||||
/// tests asserting how much of a desired set has been acked.
|
||||
pub fn established_count(&self) -> usize {
|
||||
self.established
|
||||
.read()
|
||||
.expect("community topics lock poisoned")
|
||||
.len()
|
||||
}
|
||||
|
||||
/// Ask the subscriber task to reconcile now rather than at the next tick.
|
||||
pub fn wake(&self) {
|
||||
self.wake.notify_one();
|
||||
}
|
||||
|
||||
/// Waits for a [`Self::wake`]. The wait side of the same signal, public so
|
||||
/// interest holders in other crates can assert they really nudge the
|
||||
/// reconciler; the subscriber loop below is its production caller.
|
||||
///
|
||||
/// One permit is stored, so a wake raised before the wait still completes.
|
||||
pub async fn wake_notified(&self) {
|
||||
self.wake.notified().await;
|
||||
}
|
||||
|
||||
/// Marks `community` established, standing in for a Redis ack.
|
||||
///
|
||||
/// Test seam only: in production the subscriber records establishment, and
|
||||
/// only after `SUBSCRIBE` returns. It is public because the gate's consumers
|
||||
/// live in another crate and have to be able to exercise both sides of it.
|
||||
#[doc(hidden)]
|
||||
pub fn insert_established_for_test(&self, community: CommunityId) {
|
||||
self.insert_established(community);
|
||||
}
|
||||
|
||||
fn snapshot(&self) -> HashSet<CommunityId> {
|
||||
self.established
|
||||
.read()
|
||||
.expect("community topics lock poisoned")
|
||||
.clone()
|
||||
}
|
||||
|
||||
fn insert_established(&self, community: CommunityId) {
|
||||
self.established
|
||||
.write()
|
||||
.expect("community topics lock poisoned")
|
||||
.insert(community);
|
||||
}
|
||||
|
||||
fn remove_established(&self, community: CommunityId) {
|
||||
self.established
|
||||
.write()
|
||||
.expect("community topics lock poisoned")
|
||||
.remove(&community);
|
||||
}
|
||||
|
||||
fn clear_established(&self) {
|
||||
self.established
|
||||
.write()
|
||||
.expect("community topics lock poisoned")
|
||||
.clear();
|
||||
}
|
||||
}
|
||||
|
||||
/// The per-family parts of a community-scoped subscriber: channel naming,
|
||||
/// channel parsing, and payload decode plus local delivery.
|
||||
pub(crate) trait CommunityChannelFamily: Send + 'static {
|
||||
/// Exact Redis channel for one community. Never a pattern.
|
||||
fn channel(&self, community: CommunityId) -> String;
|
||||
/// Recover the community from a received channel name.
|
||||
fn parse_channel(&self, channel: &str) -> Option<CommunityId>;
|
||||
/// Decode one payload and hand it to local consumers.
|
||||
fn deliver(&self, community: CommunityId, payload: &str);
|
||||
}
|
||||
|
||||
/// Runs a community-scoped subscriber with automatic reconnection.
|
||||
///
|
||||
/// Never returns — spawn it in a background task. Reconnects with exponential
|
||||
/// backoff (1s → 2s → 4s → … → 30s max), rebuilding the exact subscription set
|
||||
/// from `desired` on every connect.
|
||||
pub(crate) async fn run_community_subscriber<F: CommunityChannelFamily>(
|
||||
redis_url: String,
|
||||
family: F,
|
||||
topics: Arc<CommunityTopics>,
|
||||
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<F: CommunityChannelFamily>(
|
||||
redis_url: &str,
|
||||
family: &F,
|
||||
topics: &CommunityTopics,
|
||||
desired: &DesiredCommunities,
|
||||
) -> Result<(), redis::RedisError> {
|
||||
let name = topics.name();
|
||||
let client = redis::Client::open(redis_url)?;
|
||||
let conn = client.get_async_pubsub().await?;
|
||||
let (mut sink, mut stream) = conn.split();
|
||||
|
||||
// A fresh connection has nothing subscribed, whatever the previous one had.
|
||||
topics.clear_established();
|
||||
reconcile(&mut sink, family, topics, desired).await?;
|
||||
|
||||
tracing::info!(
|
||||
established = topics.established_count(),
|
||||
"Redis {name} subscriber connected with exact per-community subscriptions"
|
||||
);
|
||||
|
||||
let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
|
||||
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = topics.wake.notified() => {
|
||||
reconcile(&mut sink, family, topics, desired).await?;
|
||||
}
|
||||
_ = ticker.tick() => {
|
||||
reconcile(&mut sink, family, topics, desired).await?;
|
||||
}
|
||||
msg = stream.next() => {
|
||||
let Some(msg) = msg else {
|
||||
// Stream returned None — Redis connection closed.
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
let channel = msg.get_channel_name();
|
||||
let Some(community_id) = family.parse_channel(channel) else {
|
||||
tracing::warn!("Received {name} message on unexpected channel: {channel}");
|
||||
continue;
|
||||
};
|
||||
|
||||
let payload: String = match msg.get_payload() {
|
||||
Ok(p) => p,
|
||||
Err(e) => {
|
||||
tracing::warn!("Failed to get {name} payload: {e}");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
family.deliver(community_id, &payload);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Bring the live connection's subscriptions in line with current desire.
|
||||
///
|
||||
/// `desired` is read here, inside the subscriber task, immediately before each
|
||||
/// command — so an unsubscribe can never be applied against a stale view of
|
||||
/// interest that has since been renewed.
|
||||
async fn reconcile<F: CommunityChannelFamily>(
|
||||
sink: &mut redis::aio::PubSubSink,
|
||||
family: &F,
|
||||
topics: &CommunityTopics,
|
||||
desired: &DesiredCommunities,
|
||||
) -> Result<(), redis::RedisError> {
|
||||
let want = desired();
|
||||
let have = topics.snapshot();
|
||||
|
||||
for community in want.difference(&have) {
|
||||
// `subscribe` awaits Redis's reply, so this returning `Ok` *is* the
|
||||
// acknowledgement. Only then does the community count as established.
|
||||
sink.subscribe(&family.channel(*community)).await?;
|
||||
topics.insert_established(*community);
|
||||
}
|
||||
|
||||
for community in have.difference(&want) {
|
||||
// Withdraw establishment before the unsubscribe, never after: a reader
|
||||
// that consults `is_established` between the two would otherwise cache
|
||||
// an entry this connection is about to stop protecting.
|
||||
topics.remove_established(*community);
|
||||
sink.unsubscribe(&family.channel(*community)).await?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use uuid::Uuid;
|
||||
|
||||
fn community(id: u128) -> CommunityId {
|
||||
CommunityId::from_uuid(Uuid::from_u128(id))
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nothing_is_established_before_an_ack() {
|
||||
let topics = CommunityTopics::new("test");
|
||||
assert!(!topics.is_established(community(1)));
|
||||
assert_eq!(topics.established_count(), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn established_is_cleared_wholesale_on_disconnect() {
|
||||
let topics = CommunityTopics::new("test");
|
||||
topics.insert_established(community(1));
|
||||
topics.insert_established(community(2));
|
||||
assert_eq!(topics.established_count(), 2);
|
||||
|
||||
topics.clear_established();
|
||||
|
||||
assert!(!topics.is_established(community(1)));
|
||||
assert!(!topics.is_established(community(2)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn removing_one_community_leaves_the_others_established() {
|
||||
let topics = CommunityTopics::new("test");
|
||||
topics.insert_established(community(1));
|
||||
topics.insert_established(community(2));
|
||||
|
||||
topics.remove_established(community(1));
|
||||
|
||||
assert!(!topics.is_established(community(1)));
|
||||
assert!(topics.is_established(community(2)));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_wake_raised_before_the_wait_is_not_lost() {
|
||||
let topics = CommunityTopics::new("test");
|
||||
topics.wake();
|
||||
|
||||
// Notify stores one permit, so a wake that arrives before the subscriber
|
||||
// reaches its select still reconciles immediately. Convergence does not
|
||||
// depend on this — the tick covers a lost wake — but latency does.
|
||||
tokio::time::timeout(Duration::from_millis(50), topics.wake.notified())
|
||||
.await
|
||||
.expect("a wake raised before the wait must still complete");
|
||||
}
|
||||
}
|
||||
|
||||
/// Redis-backed reconciliation tests.
|
||||
///
|
||||
/// These are `#[ignore]`d because they need a live server, and CI selects them
|
||||
/// explicitly (`--run-ignored all` over `package(buzz-pubsub)`). They route
|
||||
/// through `test_util::test_redis_url()`, so `REDIS_URL` points them at either
|
||||
/// standalone Redis or a cluster-mode server; when it is set but unreachable
|
||||
/// they must **fail**, never self-skip — a suite that quietly passes when the
|
||||
/// server is missing is exactly the fake-green class this lane is guarding
|
||||
/// against.
|
||||
#[cfg(test)]
|
||||
mod redis_tests {
|
||||
use super::*;
|
||||
use crate::cache_invalidation::{
|
||||
cache_invalidation_channel_for, run_cache_invalidation_subscriber, CacheInvalidation,
|
||||
ScopedCacheInvalidation,
|
||||
};
|
||||
use crate::test_util::{await_established, await_unestablished, test_redis_url};
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Mutex as StdMutex;
|
||||
use tokio::sync::broadcast;
|
||||
use uuid::Uuid;
|
||||
|
||||
/// A mutable desired set a test can steer, plus the closure over it.
|
||||
fn steerable() -> (Arc<StdMutex<HashSet<CommunityId>>>, 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<StdMutex<HashSet<CommunityId>>>, 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<CommunityId> {
|
||||
(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<i64> {
|
||||
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<CommunityTopics>,
|
||||
broadcast::Receiver<ScopedCacheInvalidation>,
|
||||
) {
|
||||
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<CommunityId, CacheInvalidation> = 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<i64> = 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;
|
||||
}
|
||||
}
|
||||
@@ -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<ScopedConnControl>,
|
||||
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 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<ScopedConnControl>,
|
||||
) -> Result<(), redis::RedisError> {
|
||||
let client = redis::Client::open(redis_url)?;
|
||||
let mut conn = client.get_async_pubsub().await?;
|
||||
struct ConnControlFamily {
|
||||
broadcast_tx: broadcast::Sender<ScopedConnControl>,
|
||||
}
|
||||
|
||||
conn.psubscribe(CONN_CONTROL_PATTERN).await?;
|
||||
impl CommunityChannelFamily for ConnControlFamily {
|
||||
fn channel(&self, community: CommunityId) -> String {
|
||||
conn_control_channel_for(community)
|
||||
}
|
||||
|
||||
tracing::info!("Redis conn-control subscriber connected — listening on {CONN_CONTROL_PATTERN}");
|
||||
fn parse_channel(&self, channel: &str) -> Option<CommunityId> {
|
||||
parse_conn_control_channel(channel)
|
||||
}
|
||||
|
||||
let mut stream = conn.on_message();
|
||||
while let Some(msg) = stream.next().await {
|
||||
let channel = msg.get_channel_name();
|
||||
let Some(community_id) = parse_conn_control_channel(channel) else {
|
||||
tracing::warn!("Received conn-control message on unexpected channel: {channel}");
|
||||
continue;
|
||||
};
|
||||
|
||||
let payload: String = match msg.get_payload() {
|
||||
Ok(p) => p,
|
||||
Err(e) => {
|
||||
tracing::warn!("Failed to get conn-control payload: {e}");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let command: ConnControl = match serde_json::from_str(&payload) {
|
||||
fn deliver(&self, community_id: CommunityId, payload: &str) {
|
||||
let command: ConnControl = match serde_json::from_str(payload) {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
tracing::warn!("Failed to deserialize conn-control message: {e}");
|
||||
continue;
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
@@ -153,12 +140,10 @@ async fn connect_and_subscribe(
|
||||
command,
|
||||
};
|
||||
|
||||
if broadcast_tx.send(scoped).is_err() {
|
||||
if self.broadcast_tx.send(scoped).is_err() {
|
||||
tracing::trace!("No conn-control receivers — message dropped");
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -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<ChannelEvent>,
|
||||
cache_invalidation_tx: broadcast::Sender<ScopedCacheInvalidation>,
|
||||
conn_control_tx: broadcast::Sender<ScopedConnControl>,
|
||||
/// Communities whose cache-invalidation channel Redis has acknowledged.
|
||||
cache_invalidation_topics: Arc<CommunityTopics>,
|
||||
/// Communities whose connection-control channel Redis has acknowledged.
|
||||
conn_control_topics: Arc<CommunityTopics>,
|
||||
}
|
||||
|
||||
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<Self>) {
|
||||
///
|
||||
/// `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<Self>, 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<Self>) {
|
||||
///
|
||||
/// `desired` returns the communities holding live sockets on this pod.
|
||||
pub async fn run_conn_control_subscriber(self: Arc<Self>, 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<CommunityTopics> {
|
||||
&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<CommunityTopics> {
|
||||
&self.conn_control_topics
|
||||
}
|
||||
|
||||
/// Returns a new broadcast receiver for locally-published channel events.
|
||||
pub fn subscribe_local(&self) -> broadcast::Receiver<ChannelEvent> {
|
||||
self.broadcast_tx.subscribe()
|
||||
@@ -386,12 +425,66 @@ pub(crate) mod test_util {
|
||||
cfg.create_pool(Some(deadpool_redis::Runtime::Tokio1))
|
||||
.expect("Failed to create Redis pool")
|
||||
}
|
||||
|
||||
/// A fixed desired-community set, for tests that do not exercise churn.
|
||||
pub fn fixed_desired(
|
||||
communities: impl IntoIterator<Item = buzz_core::CommunityId>,
|
||||
) -> crate::community_topics::DesiredCommunities {
|
||||
let set: std::collections::HashSet<_> = communities.into_iter().collect();
|
||||
std::sync::Arc::new(move || set.clone())
|
||||
}
|
||||
|
||||
/// Waits until `topics` reports every community in `expected` established,
|
||||
/// then returns. Panics on timeout.
|
||||
///
|
||||
/// Every Redis-backed test here waits on the acknowledgement rather than
|
||||
/// sleeping a guessed interval: a fixed sleep either flakes under load or
|
||||
/// passes vacuously when the subscribe silently never happened.
|
||||
pub async fn await_established(
|
||||
topics: &crate::community_topics::CommunityTopics,
|
||||
expected: &[buzz_core::CommunityId],
|
||||
what: &str,
|
||||
) {
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
|
||||
loop {
|
||||
if expected.iter().all(|c| topics.is_established(*c)) {
|
||||
return;
|
||||
}
|
||||
assert!(
|
||||
std::time::Instant::now() < deadline,
|
||||
"timed out waiting for {what} to be established \
|
||||
({} of {} acked)",
|
||||
topics.established_count(),
|
||||
expected.len()
|
||||
);
|
||||
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Waits until `topics` reports `community` NOT established. Panics on timeout.
|
||||
pub async fn await_unestablished(
|
||||
topics: &crate::community_topics::CommunityTopics,
|
||||
community: buzz_core::CommunityId,
|
||||
what: &str,
|
||||
) {
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
|
||||
loop {
|
||||
if !topics.is_established(community) {
|
||||
return;
|
||||
}
|
||||
assert!(
|
||||
std::time::Instant::now() < deadline,
|
||||
"timed out waiting for {what} to be unestablished"
|
||||
);
|
||||
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::test_util::{make_test_pool, test_redis_url};
|
||||
use crate::test_util::{await_established, fixed_desired, make_test_pool, test_redis_url};
|
||||
use buzz_core::{CommunityId, TenantContext};
|
||||
use nostr::{EventBuilder, Keys, Kind};
|
||||
use uuid::Uuid;
|
||||
@@ -452,10 +545,24 @@ mod tests {
|
||||
async fn test_cache_invalidation_roundtrip() {
|
||||
let manager = make_manager().await;
|
||||
let mut rx = manager.subscribe_cache_invalidations();
|
||||
let ctx = ctx(0xaaaa, "a.example");
|
||||
|
||||
let manager_clone = manager.clone();
|
||||
tokio::spawn(async move { manager_clone.run_cache_invalidation_subscriber().await });
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
|
||||
let desired = fixed_desired([ctx.community()]);
|
||||
tokio::spawn(async move {
|
||||
manager_clone
|
||||
.run_cache_invalidation_subscriber(desired)
|
||||
.await
|
||||
});
|
||||
// Wait for the exact SUBSCRIBE to be acked rather than sleeping: with
|
||||
// per-community channels, publishing before the ack is a lost message,
|
||||
// not a late one.
|
||||
await_established(
|
||||
manager.cache_invalidation_topics(),
|
||||
&[ctx.community()],
|
||||
"cache-invalidation roundtrip",
|
||||
)
|
||||
.await;
|
||||
|
||||
let channel_id = Uuid::new_v4();
|
||||
let pubkey = Keys::generate().public_key().to_bytes().to_vec();
|
||||
@@ -464,8 +571,6 @@ mod tests {
|
||||
pubkey: pubkey.clone(),
|
||||
};
|
||||
|
||||
let ctx = ctx(0xaaaa, "a.example");
|
||||
|
||||
manager
|
||||
.publish_cache_invalidation(&ctx, &sent)
|
||||
.await
|
||||
|
||||
@@ -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 {
|
||||
|
||||
+245
-21
@@ -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<DashMap<Uuid, (CommunityId, CancellationToken)>>,
|
||||
}
|
||||
|
||||
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<CommunityTopics>,
|
||||
}
|
||||
|
||||
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<CommunityTopics>) -> Self {
|
||||
Self {
|
||||
connections: Arc::new(DashMap::new()),
|
||||
conn_control_topics,
|
||||
}
|
||||
}
|
||||
|
||||
/// Registers one socket and returns a guard that deregisters it on drop.
|
||||
///
|
||||
/// The wake only shortens the subscribe latency; the subscriber's periodic
|
||||
/// reconcile is what guarantees the community is eventually subscribed, so
|
||||
/// a lost or coalesced wake cannot strand this connection unprotected.
|
||||
pub fn register(
|
||||
&self,
|
||||
connection_id: Uuid,
|
||||
@@ -89,6 +103,7 @@ impl CommunityConnectionRegistry {
|
||||
) -> CommunityConnectionGuard {
|
||||
self.connections
|
||||
.insert(connection_id, (community_id, cancel));
|
||||
self.conn_control_topics.wake();
|
||||
CommunityConnectionGuard {
|
||||
connection_id,
|
||||
connections: Arc::clone(&self.connections),
|
||||
@@ -128,6 +143,82 @@ impl Drop for CommunityConnectionGuard {
|
||||
}
|
||||
}
|
||||
|
||||
/// Communities whose authorization caches this pod may still serve, and the
|
||||
/// desired set for the cache-invalidation subscriber.
|
||||
///
|
||||
/// Cached authorization outlives the socket that populated it, so residency —
|
||||
/// not connection lifetime — is what decides whether this pod must keep hearing
|
||||
/// `buzz:{community}:cache-invalidate`. A community becomes resident when this
|
||||
/// pod first resolves authorization under it and stays resident until `ttl`
|
||||
/// after the most recent resolution, which covers the whole window in which a
|
||||
/// cached entry could still be read.
|
||||
///
|
||||
/// Residency lapses on a timer rather than on an event, so no unsubscribe is
|
||||
/// owed by any call path: a community that stops being read simply ages out.
|
||||
pub struct CacheResidency {
|
||||
/// Community → deadline after which no cached entry can still be served.
|
||||
resident: Arc<DashMap<CommunityId, Instant>>,
|
||||
/// Cache-invalidation subscription state. Woken on a newly resident
|
||||
/// community; read by the establishment gate before a cache insert.
|
||||
topics: Arc<CommunityTopics>,
|
||||
/// 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<CommunityTopics>, ttl: Duration) -> Self {
|
||||
Self {
|
||||
resident: Arc::new(DashMap::new()),
|
||||
topics,
|
||||
ttl,
|
||||
}
|
||||
}
|
||||
|
||||
/// Records that this pod is resolving authorization for `community` and
|
||||
/// reports whether the result may be cached.
|
||||
///
|
||||
/// Recording and gating are one call on purpose. Residency *is* the desired
|
||||
/// set, so a community that is only gated and never recorded would never be
|
||||
/// subscribed, never become established, and never pass the gate — the
|
||||
/// authorization caches would stay off for the life of the pod. Folding the
|
||||
/// two makes that ordering unrepresentable.
|
||||
///
|
||||
/// The first call for a community therefore returns `false` (nothing is
|
||||
/// acked yet) while still asking for the subscription; once the reconciler
|
||||
/// establishes it, subsequent calls return `true`. A `false` means the pod
|
||||
/// reads through to the DB — degraded, not stale.
|
||||
///
|
||||
/// Gating **inserts only** is deliberate: gating reads would discard a live,
|
||||
/// still-invalidatable cache on every reconnect and turn a pub/sub blip into
|
||||
/// a DB stampede.
|
||||
#[must_use]
|
||||
pub fn admit(&self, community: CommunityId) -> bool {
|
||||
let deadline = Instant::now() + self.ttl;
|
||||
if self.resident.insert(community, deadline).is_none() {
|
||||
// Newly resident: shorten the wait for the subscribe. Convergence
|
||||
// does not depend on this — the reconcile tick is the guarantee.
|
||||
self.topics.wake();
|
||||
}
|
||||
let admitted = self.topics.is_established(community);
|
||||
if !admitted {
|
||||
metrics::counter!("buzz_authz_cache_insert_skipped_total").increment(1);
|
||||
}
|
||||
admitted
|
||||
}
|
||||
|
||||
/// Communities this pod may still serve a cached entry for. Expired entries
|
||||
/// are dropped here, so this is also the tracker's only reclamation path —
|
||||
/// and it is driven by the subscriber's own periodic reconcile, never by an
|
||||
/// external event.
|
||||
pub fn resident_communities(&self) -> HashSet<CommunityId> {
|
||||
let now = Instant::now();
|
||||
self.resident.retain(|_, deadline| *deadline > now);
|
||||
self.resident.iter().map(|entry| *entry.key()).collect()
|
||||
}
|
||||
}
|
||||
|
||||
/// Registers a socket, durably revalidates its community, then runs it.
|
||||
///
|
||||
/// The ordering is the archival admission invariant: archive-before-query is
|
||||
@@ -506,6 +597,10 @@ pub struct AppState {
|
||||
pub conn_manager: Arc<ConnectionManager>,
|
||||
/// Lifecycle cancellation for every long-lived socket, including huddle audio.
|
||||
pub community_connections: Arc<CommunityConnectionRegistry>,
|
||||
/// 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<CacheResidency>,
|
||||
/// Stops only the periodic lifecycle revalidator during graceful shutdown.
|
||||
pub community_revalidator_cancel: CancellationToken,
|
||||
/// Test/telemetry counter for archive disconnect publication attempts.
|
||||
@@ -712,6 +807,8 @@ impl AppState {
|
||||
Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone()));
|
||||
let admission_rate_limiter = Arc::new(RedisRateLimiter::new(redis_pool.clone()));
|
||||
let audit_enabled = audit_arc.is_some();
|
||||
let conn_control_topics = Arc::clone(pubsub.conn_control_topics());
|
||||
let cache_invalidation_topics = Arc::clone(pubsub.cache_invalidation_topics());
|
||||
let state = Self {
|
||||
config: Arc::new(config),
|
||||
db,
|
||||
@@ -722,7 +819,11 @@ impl AppState {
|
||||
search: search_arc,
|
||||
sub_registry: Arc::new(SubscriptionRegistry::new()),
|
||||
conn_manager: Arc::new(ConnectionManager::new()),
|
||||
community_connections: Arc::new(CommunityConnectionRegistry::new()),
|
||||
community_connections: Arc::new(CommunityConnectionRegistry::new(conn_control_topics)),
|
||||
cache_residency: Arc::new(CacheResidency::new(
|
||||
cache_invalidation_topics,
|
||||
AUTHZ_CACHE_TTL,
|
||||
)),
|
||||
community_revalidator_cancel: CancellationToken::new(),
|
||||
community_disconnect_publish_attempts: Arc::new(AtomicU64::new(0)),
|
||||
conn_semaphore: Arc::new(Semaphore::new(max_connections)),
|
||||
@@ -741,21 +842,21 @@ impl AppState {
|
||||
membership_cache: Arc::new(
|
||||
moka::sync::Cache::builder()
|
||||
.max_capacity(10_000)
|
||||
.time_to_live(std::time::Duration::from_secs(10))
|
||||
.time_to_live(AUTHZ_CACHE_TTL)
|
||||
.support_invalidation_closures()
|
||||
.build(),
|
||||
),
|
||||
accessible_channels_cache: Arc::new(
|
||||
moka::sync::Cache::builder()
|
||||
.max_capacity(10_000)
|
||||
.time_to_live(std::time::Duration::from_secs(10))
|
||||
.time_to_live(AUTHZ_CACHE_TTL)
|
||||
.support_invalidation_closures()
|
||||
.build(),
|
||||
),
|
||||
channel_visibility_cache: Arc::new(
|
||||
moka::sync::Cache::builder()
|
||||
.max_capacity(10_000)
|
||||
.time_to_live(std::time::Duration::from_secs(10))
|
||||
.time_to_live(AUTHZ_CACHE_TTL)
|
||||
.support_invalidation_closures()
|
||||
.build(),
|
||||
),
|
||||
@@ -825,6 +926,11 @@ impl AppState {
|
||||
}
|
||||
|
||||
/// Check channel membership with a 10-second cache. Falls back to DB on miss.
|
||||
///
|
||||
/// The result is cached only while this pod's cache-invalidation channel
|
||||
/// for `community_id` is established — see [`CacheResidency::may_cache`].
|
||||
/// Ungated, a pod could cache an authorization decision it would never hear
|
||||
/// the invalidation for, and serve it for the whole TTL.
|
||||
pub async fn is_member_cached(
|
||||
&self,
|
||||
community_id: CommunityId,
|
||||
@@ -838,7 +944,9 @@ impl AppState {
|
||||
}
|
||||
metrics::counter!("buzz_membership_cache_misses_total").increment(1);
|
||||
let result = self.db.is_member(community_id, channel_id, pubkey).await?;
|
||||
self.membership_cache.insert(key, result);
|
||||
if self.cache_residency.admit(community_id) {
|
||||
self.membership_cache.insert(key, result);
|
||||
}
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
@@ -1102,7 +1210,9 @@ impl AppState {
|
||||
.db
|
||||
.get_accessible_channel_ids(community_id, pubkey)
|
||||
.await?;
|
||||
self.accessible_channels_cache.insert(key, result.clone());
|
||||
if self.cache_residency.admit(community_id) {
|
||||
self.accessible_channels_cache.insert(key, result.clone());
|
||||
}
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
@@ -1144,8 +1254,16 @@ impl AppState {
|
||||
}
|
||||
};
|
||||
if visibility == "private" {
|
||||
self.channel_visibility_cache
|
||||
.insert((community_id, channel_id), visibility.clone());
|
||||
// Gated like the other two caches: the residency set must cover
|
||||
// every cache `apply_cache_invalidation` can drop, or a community
|
||||
// could hold an entry no remote invalidation reaches. Here that
|
||||
// costs liveness rather than confidentiality — a stale `private`
|
||||
// is over-restrictive, not a leak — but a private->open flip going
|
||||
// unheard for the full TTL is still a real delivery bug.
|
||||
if self.cache_residency.admit(community_id) {
|
||||
self.channel_visibility_cache
|
||||
.insert((community_id, channel_id), visibility.clone());
|
||||
}
|
||||
}
|
||||
Ok(visibility)
|
||||
}
|
||||
@@ -1255,6 +1373,12 @@ mod tests {
|
||||
(mgr, conn_id, rx, ctrl_rx, cancel, bp)
|
||||
}
|
||||
|
||||
/// A lifecycle registry whose conn-control wakes go nowhere. These tests
|
||||
/// assert registry bookkeeping, not subscription reconciliation.
|
||||
fn test_registry() -> CommunityConnectionRegistry {
|
||||
CommunityConnectionRegistry::new(Arc::new(CommunityTopics::new("test")))
|
||||
}
|
||||
|
||||
async fn test_state() -> Arc<AppState> {
|
||||
let mut config = crate::config::Config::from_env().expect("default config loads");
|
||||
config.require_relay_membership = false;
|
||||
@@ -1573,7 +1697,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn community_lifecycle_disconnect_covers_socket_types_and_preserves_tenant_fence() {
|
||||
let registry = CommunityConnectionRegistry::new();
|
||||
let registry = test_registry();
|
||||
let community_a = CommunityId::from_uuid(Uuid::from_u128(0xa));
|
||||
let community_b = CommunityId::from_uuid(Uuid::from_u128(0xb));
|
||||
let ordinary_a = CancellationToken::new();
|
||||
@@ -1591,7 +1715,7 @@ mod tests {
|
||||
|
||||
#[tokio::test]
|
||||
async fn register_then_revalidate_closes_both_archive_race_orderings() {
|
||||
let registry = CommunityConnectionRegistry::new();
|
||||
let registry = test_registry();
|
||||
let community = CommunityId::from_uuid(Uuid::from_u128(0xa));
|
||||
|
||||
// Archive wins before durable revalidation: the check observes inactive
|
||||
@@ -1647,7 +1771,7 @@ mod tests {
|
||||
|
||||
#[tokio::test]
|
||||
async fn revalidation_continues_after_one_community_lookup_failure() {
|
||||
let registry = CommunityConnectionRegistry::new();
|
||||
let registry = test_registry();
|
||||
let archived_a = CommunityId::from_uuid(Uuid::from_u128(0xa));
|
||||
let failed = CommunityId::from_uuid(Uuid::from_u128(0xb));
|
||||
let archived_c = CommunityId::from_uuid(Uuid::from_u128(0xc));
|
||||
@@ -1684,7 +1808,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn community_lifecycle_guard_deregisters_on_early_return() {
|
||||
let registry = CommunityConnectionRegistry::new();
|
||||
let registry = test_registry();
|
||||
let community = CommunityId::from_uuid(Uuid::from_u128(0xa));
|
||||
let cancel = CancellationToken::new();
|
||||
let guard = registry.register(Uuid::new_v4(), community, cancel.clone());
|
||||
@@ -1930,4 +2054,104 @@ mod tests {
|
||||
other => panic!("expected a restart close frame, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// Residency drives the cache-invalidation desired set, so it must cover a
|
||||
/// community for at least as long as an entry cached under it can be read.
|
||||
#[test]
|
||||
fn residency_outlives_the_entry_it_protects() {
|
||||
// The two are the same constant by construction; asserting the relation
|
||||
// is what fails the build if a future change gives the caches a longer
|
||||
// TTL than residency, which would silently reopen the stale-authz gap.
|
||||
let topics = Arc::new(CommunityTopics::new("test"));
|
||||
let residency = CacheResidency::new(Arc::clone(&topics), AUTHZ_CACHE_TTL);
|
||||
let community = CommunityId::from_uuid(Uuid::from_u128(0xa));
|
||||
|
||||
let _ = residency.admit(community);
|
||||
assert_eq!(
|
||||
residency.resident_communities(),
|
||||
HashSet::from([community]),
|
||||
"a just-resolved community is resident"
|
||||
);
|
||||
|
||||
// A residency shorter than the cache TTL is the bug; prove the tracker
|
||||
// reports lapse purely on its own clock by using a zero-length window.
|
||||
let expired = CacheResidency::new(topics, Duration::ZERO);
|
||||
let _ = expired.admit(community);
|
||||
assert!(
|
||||
expired.resident_communities().is_empty(),
|
||||
"residency must lapse on its own timer, with no external event"
|
||||
);
|
||||
}
|
||||
|
||||
/// The bootstrap must close: recording residency and gating the insert are
|
||||
/// one call, so a community this pod reads becomes desired *before* it is
|
||||
/// admitted, and is admitted as soon as the reconciler acks.
|
||||
///
|
||||
/// Split into a `may_cache` gate plus a separate `touch`, only an admitted
|
||||
/// community would ever be recorded — nothing would enter the desired set,
|
||||
/// nothing would be subscribed, and the authorization caches would stay off
|
||||
/// for the life of the pod. This test is the deadlock detector, so it
|
||||
/// asserts the *first* call is both refused and residency-forming.
|
||||
#[test]
|
||||
fn admit_makes_a_community_desired_before_it_is_established() {
|
||||
let topics = Arc::new(CommunityTopics::new("test"));
|
||||
let residency = CacheResidency::new(Arc::clone(&topics), AUTHZ_CACHE_TTL);
|
||||
let community = CommunityId::from_uuid(Uuid::from_u128(0xa));
|
||||
|
||||
assert!(
|
||||
!residency.admit(community),
|
||||
"nothing is cacheable before Redis acks the invalidation channel"
|
||||
);
|
||||
assert_eq!(
|
||||
residency.resident_communities(),
|
||||
HashSet::from([community]),
|
||||
"the refused call must still make the community desired, or the \
|
||||
subscriber never subscribes and the gate never opens"
|
||||
);
|
||||
|
||||
// Stand in for the reconciler acking the SUBSCRIBE it was just asked for.
|
||||
topics.insert_established_for_test(community);
|
||||
assert!(
|
||||
residency.admit(community),
|
||||
"an acked community must be admitted"
|
||||
);
|
||||
}
|
||||
|
||||
/// A newly resident community wakes the cache-invalidation subscriber, so
|
||||
/// the subscribe does not wait for the reconcile tick.
|
||||
#[tokio::test]
|
||||
async fn a_newly_resident_community_signals_the_cache_subscriber() {
|
||||
let topics = Arc::new(CommunityTopics::new("test"));
|
||||
let residency = CacheResidency::new(Arc::clone(&topics), AUTHZ_CACHE_TTL);
|
||||
let community = CommunityId::from_uuid(Uuid::from_u128(0xa));
|
||||
|
||||
let _ = residency.admit(community);
|
||||
|
||||
tokio::time::timeout(Duration::from_millis(500), topics.wake_notified())
|
||||
.await
|
||||
.expect("first residency must nudge the cache-invalidation reconciler");
|
||||
}
|
||||
|
||||
/// A registration wakes the conn-control subscriber. The wake is latency
|
||||
/// only — the reconcile tick is the guarantee — but a registry that never
|
||||
/// signals would make every new community's ban enforcement wait a tick.
|
||||
#[tokio::test]
|
||||
async fn registering_a_socket_signals_the_conn_control_subscriber() {
|
||||
let topics = Arc::new(CommunityTopics::new("test"));
|
||||
let registry = CommunityConnectionRegistry::new(Arc::clone(&topics));
|
||||
let community = CommunityId::from_uuid(Uuid::from_u128(0xa));
|
||||
|
||||
let guard = registry.register(Uuid::new_v4(), community, CancellationToken::new());
|
||||
assert_eq!(registry.bound_communities(), HashSet::from([community]));
|
||||
|
||||
// The wake stores one permit, so waiting after the fact still observes
|
||||
// the registration's signal. A bounded timeout rather than a bare await:
|
||||
// a registry that never signals must fail this, not hang the suite.
|
||||
tokio::time::timeout(Duration::from_millis(500), topics.wake_notified())
|
||||
.await
|
||||
.expect("registration must nudge the conn-control reconciler");
|
||||
|
||||
drop(guard);
|
||||
assert!(registry.bound_communities().is_empty());
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user