diff --git a/.github/workflows/_ci-relay.yml b/.github/workflows/_ci-relay.yml index 97d32f912c7..4a21a6dff76 100644 --- a/.github/workflows/_ci-relay.yml +++ b/.github/workflows/_ci-relay.yml @@ -88,6 +88,7 @@ jobs: cargo nextest archive \ --cargo-profile ci \ -p buzz-db \ + -p buzz-pubsub \ -p buzz-relay \ -p buzz-test-client \ --lib \ @@ -205,6 +206,13 @@ jobs: run: | scripts/postgres-test-run.sh \ --archive-file target/ci/postgres-tests.tar.zst + - name: Redis subscription integration tests + run: | + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E 'package(buzz-pubsub)' \ + --run-ignored ignored-only \ + --test-threads 1 desktop-e2e-integration-shard: name: Desktop E2E Integration (${{ matrix.shard }}/2) diff --git a/Cargo.lock b/Cargo.lock index e57698003fc..bb8102327c6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1214,12 +1214,14 @@ dependencies = [ "chrono", "deadpool-redis", "futures-util", + "metrics", "nostr 0.44.7", "redis", "serde", "serde_json", "thiserror 2.0.18", "tokio", + "tokio-util", "tracing", "uuid", ] diff --git a/Justfile b/Justfile index 00c4fecc2f9..6d3626c0327 100644 --- a/Justfile +++ b/Justfile @@ -385,6 +385,11 @@ test-unit: # #[ignore]d, so --lib runs only the infra-free set. Without this gate a # stray file in migrations/ or a broken lint ships green. cargo nextest run -p buzz-db --lib + # Redis subscription state, supervision inputs, and metric bookkeeping + # are infrastructure-free except for the explicitly ignored Redis + # round trips. Run the complete non-ignored package here; the required + # PostgreSQL/Redis lane selects the ignored cases separately. + cargo nextest run -p buzz-pubsub # Multi-tenant conformance gate (buzz-conformance): the independent # replay checker + golden fixtures. No infra — pure in-process trace # replay — so it belongs in the unit job. Run all targets (lib + the @@ -449,9 +454,15 @@ test-unit: # `handlers::`: the wider set is mostly Postgres-backed, and five of its # non-postgres_tests cases only "pass" without a database by waiting out # the ~30s sqlx acquire timeout, so they do not belong in the infra-free - # unit job either. + # unit job either. The final two clauses cover the process-owned Redis + # runtime and its raw metric contract without widening into unrelated + # relay tests that require infrastructure. cargo nextest run -p buzz-relay --lib \ - -E '(test(/^api::admin::/) - test(=api::admin::tests::disabled_mode_allows_unauthenticated_requests_on_the_admin_host) - test(=api::admin::tests::nip98_mode_unrostered_signer_does_not_consume_a_replay_slot)) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/)' + -E '(test(/^api::admin::/) - test(=api::admin::tests::disabled_mode_allows_unauthenticated_requests_on_the_admin_host) - test(=api::admin::tests::nip98_mode_unrostered_signer_does_not_consume_a_replay_slot)) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^metrics::contract_tests::/) + test(/^redis_subscription_runtime::tests::/)' + # The production startup call site and child-process boot contract live + # outside the relay library target, so enumerate both targets explicitly. + cargo nextest run -p buzz-relay --bin buzz-relay + cargo nextest run -p buzz-relay --test boot_lifecycle # ACP author-gate and queue tests protect the trust boundary between # relay events and agent prompts. They are infra-free; ignored lifecycle # tests remain excluded and run in their dedicated integration lanes. diff --git a/crates/buzz-pubsub/Cargo.toml b/crates/buzz-pubsub/Cargo.toml index 2ee2d6b47dd..2ff425b9d76 100644 --- a/crates/buzz-pubsub/Cargo.toml +++ b/crates/buzz-pubsub/Cargo.toml @@ -13,6 +13,7 @@ buzz-auth = { workspace = true } redis = { workspace = true } deadpool-redis = { workspace = true } tokio = { workspace = true } +tokio-util = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } uuid = { workspace = true } @@ -21,6 +22,7 @@ tracing = { workspace = true } thiserror = { workspace = true } nostr = { workspace = true } futures-util = { workspace = true } +metrics = { workspace = true } [dev-dependencies] -tokio = { workspace = true } +tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/buzz-pubsub/src/cache_invalidation.rs b/crates/buzz-pubsub/src/cache_invalidation.rs index f48158c9c57..b104fee25fd 100644 --- a/crates/buzz-pubsub/src/cache_invalidation.rs +++ b/crates/buzz-pubsub/src/cache_invalidation.rs @@ -11,12 +11,16 @@ //! universal delivery-enforcement point, so dropping the stale key is //! sufficient: the next read re-fetches authoritative state from the DB. +use std::sync::Arc; + use buzz_core::{CommunityId, TenantContext}; use futures_util::StreamExt; use serde::{Deserialize, Serialize}; use tokio::sync::broadcast; +use tokio_util::sync::CancellationToken; use uuid::Uuid; +use crate::health::{SubscriptionHealth, SubscriptionPath, SubscriptionTransitionReason}; use crate::topic::BUZZ_PREFIX; /// Tenant-local Redis pub/sub channel suffix for cache-invalidation messages. @@ -101,24 +105,56 @@ pub async fn run_cache_invalidation_subscriber( redis_url: String, broadcast_tx: broadcast::Sender, ) { - let mut backoff_secs = BACKOFF_INITIAL_SECS; + let health = Arc::new(SubscriptionHealth::new()); + health.consumer_attached(SubscriptionPath::Cache); + run_cache_invalidation_subscriber_until_cancelled( + redis_url, + broadcast_tx, + CancellationToken::new(), + health, + ) + .await; +} +/// Run cache invalidation until explicit process cancellation while reporting +/// the shared end-to-end subscription-path health contract. +pub async fn run_cache_invalidation_subscriber_until_cancelled( + redis_url: String, + broadcast_tx: broadcast::Sender, + cancel: CancellationToken, + health: Arc, +) { + let mut backoff_secs = BACKOFF_INITIAL_SECS; loop { - match connect_and_subscribe(&redis_url, &broadcast_tx).await { + health.connecting(SubscriptionPath::Cache); + let attempt = connect_and_subscribe(&redis_url, &broadcast_tx, &health); + let result = tokio::select! { + () = cancel.cancelled() => return, + result = attempt => result, + }; + match result { Ok(()) => { backoff_secs = BACKOFF_INITIAL_SECS; + health.reconnecting( + SubscriptionPath::Cache, + SubscriptionTransitionReason::StreamClosed, + ); tracing::warn!( "Redis cache-invalidation stream ended (clean disconnect) — reconnecting in {backoff_secs}s" ); } - Err(e) => { + Err((reason, e)) => { + health.reconnecting(SubscriptionPath::Cache, reason); tracing::error!( "Redis cache-invalidation error: {e} — reconnecting in {backoff_secs}s" ); } } - tokio::time::sleep(tokio::time::Duration::from_secs(backoff_secs)).await; + tokio::select! { + () = cancel.cancelled() => return, + () = tokio::time::sleep(tokio::time::Duration::from_secs(backoff_secs)) => {} + } backoff_secs = (backoff_secs * 2).min(BACKOFF_MAX_SECS); tracing::info!("Attempting to reconnect to Redis cache-invalidation..."); @@ -128,11 +164,19 @@ pub async fn run_cache_invalidation_subscriber( 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?; + health: &SubscriptionHealth, +) -> Result<(), (SubscriptionTransitionReason, redis::RedisError)> { + let client = redis::Client::open(redis_url) + .map_err(|error| (SubscriptionTransitionReason::Dial, error))?; + let mut conn = client + .get_async_pubsub() + .await + .map_err(|error| (SubscriptionTransitionReason::Dial, error))?; - conn.psubscribe(CACHE_INVALIDATION_PATTERN).await?; + conn.psubscribe(CACHE_INVALIDATION_PATTERN) + .await + .map_err(|error| (SubscriptionTransitionReason::Subscribe, error))?; + health.network_ready(SubscriptionPath::Cache); tracing::info!( "Redis cache-invalidation subscriber connected — listening on {CACHE_INVALIDATION_PATTERN}" diff --git a/crates/buzz-pubsub/src/conn_control.rs b/crates/buzz-pubsub/src/conn_control.rs index bc177cff139..4e18f612b52 100644 --- a/crates/buzz-pubsub/src/conn_control.rs +++ b/crates/buzz-pubsub/src/conn_control.rs @@ -14,12 +14,16 @@ //! 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 tokio_util::sync::CancellationToken; use uuid::Uuid; +use crate::health::{SubscriptionHealth, SubscriptionPath, SubscriptionTransitionReason}; use crate::topic::BUZZ_PREFIX; /// Tenant-local Redis pub/sub channel suffix for connection-control messages. @@ -91,22 +95,54 @@ pub async fn run_conn_control_subscriber( redis_url: String, broadcast_tx: broadcast::Sender, ) { - let mut backoff_secs = BACKOFF_INITIAL_SECS; + let health = Arc::new(SubscriptionHealth::new()); + health.consumer_attached(SubscriptionPath::ConnectionControl); + run_conn_control_subscriber_until_cancelled( + redis_url, + broadcast_tx, + CancellationToken::new(), + health, + ) + .await; +} +/// Run connection control until explicit process cancellation while reporting +/// the shared end-to-end subscription-path health contract. +pub async fn run_conn_control_subscriber_until_cancelled( + redis_url: String, + broadcast_tx: broadcast::Sender, + cancel: CancellationToken, + health: Arc, +) { + let mut backoff_secs = BACKOFF_INITIAL_SECS; loop { - match connect_and_subscribe(&redis_url, &broadcast_tx).await { + health.connecting(SubscriptionPath::ConnectionControl); + let attempt = connect_and_subscribe(&redis_url, &broadcast_tx, &health); + let result = tokio::select! { + () = cancel.cancelled() => return, + result = attempt => result, + }; + match result { Ok(()) => { backoff_secs = BACKOFF_INITIAL_SECS; + health.reconnecting( + SubscriptionPath::ConnectionControl, + SubscriptionTransitionReason::StreamClosed, + ); tracing::warn!( "Redis conn-control stream ended (clean disconnect) — reconnecting in {backoff_secs}s" ); } - Err(e) => { + Err((reason, e)) => { + health.reconnecting(SubscriptionPath::ConnectionControl, reason); tracing::error!("Redis conn-control error: {e} — reconnecting in {backoff_secs}s"); } } - tokio::time::sleep(tokio::time::Duration::from_secs(backoff_secs)).await; + tokio::select! { + () = cancel.cancelled() => return, + () = tokio::time::sleep(tokio::time::Duration::from_secs(backoff_secs)) => {} + } backoff_secs = (backoff_secs * 2).min(BACKOFF_MAX_SECS); tracing::info!("Attempting to reconnect to Redis conn-control..."); @@ -116,11 +152,19 @@ pub async fn run_conn_control_subscriber( 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?; + health: &SubscriptionHealth, +) -> Result<(), (SubscriptionTransitionReason, redis::RedisError)> { + let client = redis::Client::open(redis_url) + .map_err(|error| (SubscriptionTransitionReason::Dial, error))?; + let mut conn = client + .get_async_pubsub() + .await + .map_err(|error| (SubscriptionTransitionReason::Dial, error))?; - conn.psubscribe(CONN_CONTROL_PATTERN).await?; + conn.psubscribe(CONN_CONTROL_PATTERN) + .await + .map_err(|error| (SubscriptionTransitionReason::Subscribe, error))?; + health.network_ready(SubscriptionPath::ConnectionControl); tracing::info!("Redis conn-control subscriber connected — listening on {CONN_CONTROL_PATTERN}"); diff --git a/crates/buzz-pubsub/src/health.rs b/crates/buzz-pubsub/src/health.rs new file mode 100644 index 00000000000..12219ca765a --- /dev/null +++ b/crates/buzz-pubsub/src/health.rs @@ -0,0 +1,672 @@ +//! Fixed-cardinality health for the relay's cross-pod Redis subscription paths. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Mutex; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use tokio::sync::Notify; +use tokio::time::Instant; +use uuid::Uuid; + +/// Number of required cross-pod Redis subscription paths in every relay pod. +pub const REQUIRED_PATH_COUNT: usize = 3; + +/// Every required cross-pod Redis subscription path. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub enum SubscriptionPath { + /// Nostr event delivery from Redis to local WebSocket subscribers. + Event, + /// Membership and visibility cache invalidation from other relay pods. + Cache, + /// Live connection-control commands from other relay pods. + ConnectionControl, +} + +impl SubscriptionPath { + /// Complete fixed path vocabulary. + pub const ALL: [Self; REQUIRED_PATH_COUNT] = + [Self::Event, Self::Cache, Self::ConnectionControl]; + + /// Stable metric and lifecycle value. + pub const fn as_str(self) -> &'static str { + match self { + Self::Event => "event", + Self::Cache => "cache", + Self::ConnectionControl => "connection_control", + } + } + + const fn index(self) -> usize { + match self { + Self::Event => 0, + Self::Cache => 1, + Self::ConnectionControl => 2, + } + } +} + +/// Current end-to-end state of one Redis subscription path. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum SubscriptionPathState { + /// The downstream consumer is attached but the first Redis handshake is incomplete. + Connecting, + /// The downstream consumer is alive and the Redis subscription handshake completed. + Ready, + /// The Redis connection or subscription failed and its retry backoff is active. + Reconnecting, + /// An owned network or consumer task exited unexpectedly. + Failed, + /// Explicit relay shutdown completed for the path. + Stopped, +} + +impl SubscriptionPathState { + /// Complete fixed state vocabulary. + pub const ALL: [Self; 5] = [ + Self::Connecting, + Self::Ready, + Self::Reconnecting, + Self::Failed, + Self::Stopped, + ]; + + /// Stable metric and lifecycle value. + pub const fn as_str(self) -> &'static str { + match self { + Self::Connecting => "connecting", + Self::Ready => "ready", + Self::Reconnecting => "reconnecting", + Self::Failed => "failed", + Self::Stopped => "stopped", + } + } +} + +/// Bounded transition reason; never contains Redis endpoints, topics, or raw errors. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum SubscriptionTransitionReason { + /// The Redis connection could not be established. + Dial, + /// A `SUBSCRIBE`, `PSUBSCRIBE`, or dynamic subscription command failed. + Subscribe, + /// An established Redis message stream ended. + StreamClosed, + /// The relay-side broadcast consumer closed unexpectedly. + ConsumerClosed, + /// An owned task panicked. + Panic, + /// Explicit relay shutdown cancelled the task. + Cancelled, + /// The runtime owner disappeared without completing shutdown. + OwnerDropped, + /// Bounded shutdown did not finish before its deadline. + ShutdownTimeout, +} + +impl SubscriptionTransitionReason { + /// Stable metric and lifecycle value. + pub const fn as_str(self) -> &'static str { + match self { + Self::Dial => "dial", + Self::Subscribe => "subscribe", + Self::StreamClosed => "stream_closed", + Self::ConsumerClosed => "consumer_closed", + Self::Panic => "panic", + Self::Cancelled => "cancelled", + Self::OwnerDropped => "owner_dropped", + Self::ShutdownTimeout => "shutdown_timeout", + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum SubscriptionTransition { + Connected, + ConnectFailed, + Disconnected, + Reconnected, + Terminal, +} + +impl SubscriptionTransition { + const fn as_str(self) -> &'static str { + match self { + Self::Connected => "connected", + Self::ConnectFailed => "connect_failed", + Self::Disconnected => "disconnected", + Self::Reconnected => "reconnected", + Self::Terminal => "terminal", + } + } +} + +/// Read-only state for one subscription path. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct SubscriptionPathSnapshot { + /// Current end-to-end path state. + pub state: SubscriptionPathState, + /// Whether the relay-side consumer was attached before Redis could deliver. + pub consumer_attached: bool, + /// Unix timestamp of the latest completed end-to-end ready transition. + pub last_ready_timestamp_seconds: Option, + /// Time spent reaching the latest completed end-to-end ready transition. + pub last_readiness_duration: Option, + /// Time spent continuously outside `ready`; zero while ready. + pub not_ready_duration: Duration, + /// Time spent in the current state. + pub state_duration: Duration, + /// Consecutive failed connection/subscription attempts since the latest ready state. + pub consecutive_failures: u64, +} + +#[derive(Clone, Copy, Debug)] +struct PathStatus { + state: SubscriptionPathState, + consumer_attached: bool, + network_ready: bool, + ever_ready: bool, + terminal: bool, + last_ready_timestamp_seconds: Option, + last_readiness_duration: Option, + not_ready_since: Instant, + state_since: Instant, + consecutive_failures: u64, +} + +impl PathStatus { + fn initial(now: Instant) -> Self { + Self { + state: SubscriptionPathState::Connecting, + consumer_attached: false, + network_ready: false, + ever_ready: false, + terminal: false, + last_ready_timestamp_seconds: None, + last_readiness_duration: None, + not_ready_since: now, + state_since: now, + consecutive_failures: 0, + } + } + + fn snapshot(self, now: Instant) -> SubscriptionPathSnapshot { + SubscriptionPathSnapshot { + state: self.state, + consumer_attached: self.consumer_attached, + last_ready_timestamp_seconds: self.last_ready_timestamp_seconds, + last_readiness_duration: self.last_readiness_duration, + not_ready_duration: if self.state == SubscriptionPathState::Ready { + Duration::ZERO + } else { + now.saturating_duration_since(self.not_ready_since) + }, + state_duration: now.saturating_duration_since(self.state_since), + consecutive_failures: self.consecutive_failures, + } + } + + fn set_state(&mut self, state: SubscriptionPathState, now: Instant) { + if self.state != state { + self.state = state; + self.state_since = now; + } + } + + fn mark_ready(&mut self, now: Instant) { + self.last_readiness_duration = Some(now.saturating_duration_since(self.not_ready_since)); + self.set_state(SubscriptionPathState::Ready, now); + self.ever_ready = true; + self.last_ready_timestamp_seconds = Some(unix_timestamp_seconds()); + self.consecutive_failures = 0; + } + + fn mark_not_ready(&mut self, state: SubscriptionPathState, now: Instant) { + if self.state == SubscriptionPathState::Ready { + self.not_ready_since = now; + } + self.set_state(state, now); + } +} + +/// Shared health snapshot and metric publisher for all Redis subscription paths. +pub struct SubscriptionHealth { + group_id: Uuid, + sequence: AtomicU64, + statuses: Mutex<[PathStatus; REQUIRED_PATH_COUNT]>, + publish_lock: Mutex<()>, + changed: Notify, +} + +impl Default for SubscriptionHealth { + fn default() -> Self { + Self::new() + } +} + +impl SubscriptionHealth { + /// Create the per-process fixed path snapshot. + pub fn new() -> Self { + let now = Instant::now(); + Self { + group_id: Uuid::new_v4(), + sequence: AtomicU64::new(1), + statuses: Mutex::new([PathStatus::initial(now); REQUIRED_PATH_COUNT]), + publish_lock: Mutex::new(()), + changed: Notify::new(), + } + } + + /// Record that the downstream relay consumer is attached and able to buffer messages. + pub fn consumer_attached(&self, path: SubscriptionPath) { + let _publish_guard = self.lock_publish(); + let mut transition = None; + let snapshot = { + let mut statuses = self.lock_statuses(); + let status = &mut statuses[path.index()]; + if status.terminal || status.consumer_attached { + return; + } + status.consumer_attached = true; + if status.network_ready { + transition = Some(if status.ever_ready { + SubscriptionTransition::Reconnected + } else { + SubscriptionTransition::Connected + }); + status.mark_ready(Instant::now()); + } + status.snapshot(Instant::now()) + }; + self.publish( + path, + snapshot, + transition, + transition.map(|_| SubscriptionTransitionReason::Subscribe), + ); + } + + /// Record that a Redis connection/subscription attempt has started. + pub fn connecting(&self, path: SubscriptionPath) { + if self.update_nonterminal(path, |status| { + status.network_ready = false; + status.mark_not_ready(SubscriptionPathState::Connecting, Instant::now()); + (None, None) + }) { + metrics::counter!( + "buzz_redis_subscription_attempts_total", + "path" => path.as_str(), + ) + .increment(1); + } + } + + /// Record a completed Redis subscription handshake. + pub fn network_ready(&self, path: SubscriptionPath) { + self.update_nonterminal(path, |status| { + status.network_ready = true; + if !status.consumer_attached { + return (None, None); + } + let transition = if status.ever_ready { + SubscriptionTransition::Reconnected + } else { + SubscriptionTransition::Connected + }; + status.mark_ready(Instant::now()); + ( + Some(transition), + Some(SubscriptionTransitionReason::Subscribe), + ) + }); + } + + /// Record a failed connection/subscription or an established stream closure. + pub fn reconnecting(&self, path: SubscriptionPath, reason: SubscriptionTransitionReason) { + self.update_nonterminal(path, |status| { + status.network_ready = false; + status.mark_not_ready(SubscriptionPathState::Reconnecting, Instant::now()); + status.consecutive_failures = status.consecutive_failures.saturating_add(1); + let transition = if status.ever_ready { + SubscriptionTransition::Disconnected + } else { + SubscriptionTransition::ConnectFailed + }; + (Some(transition), Some(reason)) + }); + } + + /// Record an unexpected terminal task outcome. Failure is latched until process exit. + pub fn failed(&self, path: SubscriptionPath, reason: SubscriptionTransitionReason) { + self.update(path, |status| { + status.network_ready = false; + status.mark_not_ready(SubscriptionPathState::Failed, Instant::now()); + status.terminal = true; + (Some(SubscriptionTransition::Terminal), Some(reason)) + }); + } + + /// Record successful explicit shutdown unless the path already failed. + pub fn stopped(&self, path: SubscriptionPath) { + self.update_nonterminal(path, |status| { + status.network_ready = false; + status.consumer_attached = false; + status.mark_not_ready(SubscriptionPathState::Stopped, Instant::now()); + status.terminal = true; + ( + Some(SubscriptionTransition::Terminal), + Some(SubscriptionTransitionReason::Cancelled), + ) + }); + } + + /// Return the current state for one path. + pub fn snapshot(&self, path: SubscriptionPath) -> SubscriptionPathSnapshot { + self.lock_statuses()[path.index()].snapshot(Instant::now()) + } + + /// Return all path snapshots in [`SubscriptionPath::ALL`] order. + pub fn snapshots(&self) -> [SubscriptionPathSnapshot; REQUIRED_PATH_COUNT] { + let statuses = self.lock_statuses(); + let now = Instant::now(); + [ + statuses[0].snapshot(now), + statuses[1].snapshot(now), + statuses[2].snapshot(now), + ] + } + + /// Number of required paths whose end-to-end state is currently ready. + pub fn ready_path_count(&self) -> usize { + self.lock_statuses() + .iter() + .filter(|status| status.state == SubscriptionPathState::Ready) + .count() + } + + /// Whether every required path is currently ready. + pub fn all_ready(&self) -> bool { + self.ready_path_count() == REQUIRED_PATH_COUNT + } + + /// Refresh current gauges so stable paths survive the recorder's idle timeout. + pub fn refresh_metrics(&self) { + let _publish_guard = self.lock_publish(); + let snapshots = self.snapshots(); + for (index, path) in SubscriptionPath::ALL.iter().copied().enumerate() { + publish_path_gauges(path, snapshots[index]); + } + publish_aggregate_gauges(&snapshots); + } + + /// Wait without polling or sleeps until one path reaches `expected`. + pub async fn wait_for_state( + &self, + path: SubscriptionPath, + expected: SubscriptionPathState, + timeout: Duration, + ) -> Result { + tokio::time::timeout(timeout, async { + loop { + let changed = self.changed.notified(); + let snapshot = self.snapshot(path); + if snapshot.state == expected { + return snapshot; + } + changed.await; + } + }) + .await + } + + fn update_nonterminal( + &self, + path: SubscriptionPath, + mutate: impl FnOnce( + &mut PathStatus, + ) -> ( + Option, + Option, + ), + ) -> bool { + self.update(path, mutate) + } + + fn update( + &self, + path: SubscriptionPath, + mutate: impl FnOnce( + &mut PathStatus, + ) -> ( + Option, + Option, + ), + ) -> bool { + let _publish_guard = self.lock_publish(); + let (snapshot, transition, reason) = { + let mut statuses = self.lock_statuses(); + let status = &mut statuses[path.index()]; + if status.terminal { + return false; + } + let (transition, reason) = mutate(status); + (status.snapshot(Instant::now()), transition, reason) + }; + self.publish(path, snapshot, transition, reason); + true + } + + fn publish( + &self, + path: SubscriptionPath, + snapshot: SubscriptionPathSnapshot, + transition: Option, + reason: Option, + ) { + publish_path_gauges(path, snapshot); + let snapshots = self.snapshots(); + publish_aggregate_gauges(&snapshots); + if let (Some(transition), Some(reason)) = (transition, reason) { + metrics::counter!( + "buzz_redis_subscription_transitions_total", + "path" => path.as_str(), + "transition" => transition.as_str(), + "reason" => reason.as_str(), + ) + .increment(1); + } + let sequence = self.sequence.fetch_add(1, Ordering::Relaxed); + tracing::info!( + event_name = "buzz_redis_subscription_lifecycle", + schema_version = 1_u64, + subscription_group_id = %self.group_id, + sequence, + path = path.as_str(), + state = snapshot.state.as_str(), + consumer_attached = snapshot.consumer_attached, + transition = transition.map(SubscriptionTransition::as_str).unwrap_or("none"), + reason = reason.map(SubscriptionTransitionReason::as_str).unwrap_or("none"), + "Redis subscription path lifecycle" + ); + self.changed.notify_waiters(); + } + + fn lock_statuses(&self) -> std::sync::MutexGuard<'_, [PathStatus; REQUIRED_PATH_COUNT]> { + self.statuses + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + fn lock_publish(&self) -> std::sync::MutexGuard<'_, ()> { + self.publish_lock + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } +} + +fn publish_path_gauges(path: SubscriptionPath, snapshot: SubscriptionPathSnapshot) { + for state in SubscriptionPathState::ALL { + metrics::gauge!( + "buzz_redis_subscription_path_state", + "path" => path.as_str(), + "state" => state.as_str(), + ) + .set(if snapshot.state == state { 1.0 } else { 0.0 }); + } + if let Some(timestamp) = snapshot.last_ready_timestamp_seconds { + metrics::gauge!( + "buzz_redis_subscription_last_ready_timestamp_seconds", + "path" => path.as_str(), + ) + .set(timestamp as f64); + } + if let Some(duration) = snapshot.last_readiness_duration { + metrics::gauge!( + "buzz_redis_subscription_last_readiness_duration_seconds", + "path" => path.as_str(), + ) + .set(duration.as_secs_f64()); + } + metrics::gauge!( + "buzz_redis_subscription_not_ready_duration_seconds", + "path" => path.as_str(), + ) + .set(snapshot.not_ready_duration.as_secs_f64()); + metrics::gauge!( + "buzz_redis_subscription_state_duration_seconds", + "path" => path.as_str(), + ) + .set(snapshot.state_duration.as_secs_f64()); + metrics::gauge!( + "buzz_redis_subscription_consecutive_failures", + "path" => path.as_str(), + ) + .set(snapshot.consecutive_failures as f64); +} + +fn publish_aggregate_gauges(snapshots: &[SubscriptionPathSnapshot; REQUIRED_PATH_COUNT]) { + let ready = snapshots + .iter() + .filter(|snapshot| snapshot.state == SubscriptionPathState::Ready) + .count(); + metrics::gauge!("buzz_redis_subscription_paths_ready").set(ready as f64); + metrics::gauge!("buzz_redis_subscription_all_ready").set(if ready == REQUIRED_PATH_COUNT { + 1.0 + } else { + 0.0 + }); +} + +fn unix_timestamp_seconds() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + + #[test] + fn ready_requires_both_consumer_and_network() { + let health = SubscriptionHealth::new(); + health.network_ready(SubscriptionPath::Event); + assert_eq!( + health.snapshot(SubscriptionPath::Event).state, + SubscriptionPathState::Connecting + ); + + health.consumer_attached(SubscriptionPath::Event); + let snapshot = health.snapshot(SubscriptionPath::Event); + assert_eq!(snapshot.state, SubscriptionPathState::Ready); + assert!(snapshot.consumer_attached); + assert!(snapshot.last_ready_timestamp_seconds.is_some()); + } + + #[test] + fn all_ready_requires_all_three_paths() { + let health = SubscriptionHealth::new(); + for path in SubscriptionPath::ALL { + health.consumer_attached(path); + health.network_ready(path); + } + assert_eq!(health.ready_path_count(), REQUIRED_PATH_COUNT); + assert!(health.all_ready()); + } + + #[test] + fn failure_is_latched_against_later_network_updates() { + let health = SubscriptionHealth::new(); + health.consumer_attached(SubscriptionPath::Cache); + health.failed( + SubscriptionPath::Cache, + SubscriptionTransitionReason::ConsumerClosed, + ); + health.network_ready(SubscriptionPath::Cache); + assert_eq!( + health.snapshot(SubscriptionPath::Cache).state, + SubscriptionPathState::Failed + ); + } + + #[tokio::test(start_paused = true)] + async fn readiness_duration_includes_retries_and_resets_after_recovery() { + let health = SubscriptionHealth::new(); + health.consumer_attached(SubscriptionPath::Event); + health.connecting(SubscriptionPath::Event); + tokio::time::advance(Duration::from_secs(2)).await; + health.reconnecting(SubscriptionPath::Event, SubscriptionTransitionReason::Dial); + tokio::time::advance(Duration::from_secs(3)).await; + health.connecting(SubscriptionPath::Event); + tokio::time::advance(Duration::from_secs(4)).await; + health.network_ready(SubscriptionPath::Event); + + let initial = health.snapshot(SubscriptionPath::Event); + assert_eq!( + initial.last_readiness_duration, + Some(Duration::from_secs(9)) + ); + assert_eq!(initial.not_ready_duration, Duration::ZERO); + assert_eq!(initial.consecutive_failures, 0); + + health.reconnecting( + SubscriptionPath::Event, + SubscriptionTransitionReason::StreamClosed, + ); + tokio::time::advance(Duration::from_secs(5)).await; + let recovering = health.snapshot(SubscriptionPath::Event); + assert_eq!(recovering.not_ready_duration, Duration::from_secs(5)); + assert_eq!(recovering.state_duration, Duration::from_secs(5)); + assert_eq!(recovering.consecutive_failures, 1); + + health.connecting(SubscriptionPath::Event); + tokio::time::advance(Duration::from_secs(2)).await; + health.network_ready(SubscriptionPath::Event); + let recovered = health.snapshot(SubscriptionPath::Event); + assert_eq!( + recovered.last_readiness_duration, + Some(Duration::from_secs(7)) + ); + assert_eq!(recovered.consecutive_failures, 0); + } + + #[tokio::test] + async fn wait_for_state_observes_transition_without_sleep() { + let health = Arc::new(SubscriptionHealth::new()); + let waiter_health = Arc::clone(&health); + let waiter = tokio::spawn(async move { + waiter_health + .wait_for_state( + SubscriptionPath::ConnectionControl, + SubscriptionPathState::Ready, + Duration::from_secs(1), + ) + .await + }); + health.consumer_attached(SubscriptionPath::ConnectionControl); + health.network_ready(SubscriptionPath::ConnectionControl); + let snapshot = waiter.await.expect("waiter task").expect("ready state"); + assert_eq!(snapshot.state, SubscriptionPathState::Ready); + } +} diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index 4f1690beefb..90bcc40cda5 100644 --- a/crates/buzz-pubsub/src/lib.rs +++ b/crates/buzz-pubsub/src/lib.rs @@ -27,6 +27,8 @@ pub mod cache_invalidation; pub mod conn_control; /// Error types for pub/sub operations. pub mod error; +/// Fixed-cardinality health for cross-pod Redis subscription paths. +pub mod health; /// Redis-backed NIP-98 replay seen-set. pub mod nip98_replay; pub use nip98_replay::RedisNip98ReplayGuard; @@ -50,6 +52,7 @@ use std::time::Duration; use buzz_core::TenantContext; use nostr::PublicKey; use tokio::sync::{broadcast, mpsc, Mutex}; +use tokio_util::sync::CancellationToken; use crate::cache_invalidation::{ cache_invalidation_channel, CacheInvalidation, ScopedCacheInvalidation, @@ -110,6 +113,7 @@ pub struct PubSubManager { broadcast_tx: broadcast::Sender, cache_invalidation_tx: broadcast::Sender, conn_control_tx: broadcast::Sender, + subscription_health: Arc, } impl PubSubManager { @@ -138,6 +142,7 @@ impl PubSubManager { broadcast_tx, cache_invalidation_tx, conn_control_tx, + subscription_health: Arc::new(health::SubscriptionHealth::new()), }) } @@ -146,6 +151,12 @@ impl PubSubManager { /// Runs forever — spawn this in a background task. The loop reconnects /// with exponential backoff on Redis disconnect (1s → 2s → 4s → … → 30s). pub async fn run_subscriber(self: Arc) { + self.run_subscriber_until_cancelled(CancellationToken::new()) + .await; + } + + /// Run the event subscriber until explicit process cancellation. + pub async fn run_subscriber_until_cancelled(self: Arc, cancel: CancellationToken) { let Some(subscription_rx) = self.subscription_rx.lock().await.take() else { tracing::error!("Redis pub/sub subscriber already started"); return; @@ -156,6 +167,8 @@ impl PubSubManager { self.broadcast_tx.clone(), self.desired_topics.clone(), subscription_rx, + cancel, + Arc::clone(&self.subscription_health), ) .await; } @@ -163,9 +176,20 @@ 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) { - cache_invalidation::run_cache_invalidation_subscriber( + self.run_cache_invalidation_subscriber_until_cancelled(CancellationToken::new()) + .await; + } + + /// Run the cache-invalidation subscriber until explicit process cancellation. + pub async fn run_cache_invalidation_subscriber_until_cancelled( + self: Arc, + cancel: CancellationToken, + ) { + cache_invalidation::run_cache_invalidation_subscriber_until_cancelled( self.redis_url.clone(), self.cache_invalidation_tx.clone(), + cancel, + Arc::clone(&self.subscription_health), ) .await; } @@ -173,13 +197,29 @@ impl PubSubManager { /// 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) { - conn_control::run_conn_control_subscriber( + self.run_conn_control_subscriber_until_cancelled(CancellationToken::new()) + .await; + } + + /// Run the connection-control subscriber until explicit process cancellation. + pub async fn run_conn_control_subscriber_until_cancelled( + self: Arc, + cancel: CancellationToken, + ) { + conn_control::run_conn_control_subscriber_until_cancelled( self.redis_url.clone(), self.conn_control_tx.clone(), + cancel, + Arc::clone(&self.subscription_health), ) .await; } + /// Return the shared end-to-end Redis subscription health snapshot. + pub fn subscription_health(&self) -> Arc { + Arc::clone(&self.subscription_health) + } + /// Returns a new broadcast receiver for locally-published channel events. pub fn subscribe_local(&self) -> broadcast::Receiver { self.broadcast_tx.subscribe() @@ -378,6 +418,7 @@ pub(crate) mod test_util { #[cfg(test)] mod tests { use super::*; + use crate::health::{SubscriptionPath, SubscriptionPathState}; use crate::test_util::make_test_pool; use buzz_core::{CommunityId, TenantContext}; use nostr::{EventBuilder, Keys, Kind}; @@ -401,13 +442,33 @@ mod tests { async fn test_publish_and_subscribe_roundtrip() { let manager = make_manager().await; let mut rx = manager.subscribe_local(); + manager + .subscription_health() + .consumer_attached(SubscriptionPath::Event); + let ctx = ctx(0xaaaa, "a.example"); + let channel_id = Uuid::new_v4(); + manager + .retain_topic(&ctx, EventTopic::Channel(channel_id)) + .await; + let cancel = CancellationToken::new(); let manager_clone = manager.clone(); - tokio::spawn(async move { manager_clone.run_subscriber().await }); - tokio::time::sleep(tokio::time::Duration::from_millis(200)).await; + let task_cancel = cancel.clone(); + let task = tokio::spawn(async move { + manager_clone + .run_subscriber_until_cancelled(task_cancel) + .await + }); + manager + .subscription_health() + .wait_for_state( + SubscriptionPath::Event, + SubscriptionPathState::Ready, + Duration::from_secs(2), + ) + .await + .expect("event Redis subscription handshake"); - let ctx = ctx(0xaaaa, "a.example"); - let channel_id = Uuid::new_v4(); let keys = Keys::generate(); let event = EventBuilder::new(Kind::TextNote, "hello pubsub") .tags([]) @@ -415,10 +476,6 @@ mod tests { .expect("signing failed"); let event_id = event.id; - manager - .retain_topic(&ctx, EventTopic::Channel(channel_id)) - .await; - manager .publish_event(&ctx, EventTopic::Channel(channel_id), &event) .await @@ -432,6 +489,8 @@ mod tests { assert_eq!(received.community_id, ctx.community()); assert_eq!(received.topic, EventTopic::Channel(channel_id)); assert_eq!(received.event.id, event_id); + cancel.cancel(); + task.await.expect("event subscriber shutdown"); } #[tokio::test] @@ -439,10 +498,27 @@ mod tests { async fn test_cache_invalidation_roundtrip() { let manager = make_manager().await; let mut rx = manager.subscribe_cache_invalidations(); + manager + .subscription_health() + .consumer_attached(SubscriptionPath::Cache); + let cancel = CancellationToken::new(); 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 task_cancel = cancel.clone(); + let task = tokio::spawn(async move { + manager_clone + .run_cache_invalidation_subscriber_until_cancelled(task_cancel) + .await + }); + manager + .subscription_health() + .wait_for_state( + SubscriptionPath::Cache, + SubscriptionPathState::Ready, + Duration::from_secs(2), + ) + .await + .expect("cache Redis subscription handshake"); let channel_id = Uuid::new_v4(); let pubkey = Keys::generate().public_key().to_bytes().to_vec(); @@ -470,6 +546,58 @@ mod tests { invalidation: sent, } ); + cancel.cancel(); + task.await.expect("cache subscriber shutdown"); + } + + #[tokio::test] + #[ignore = "requires Redis"] + async fn test_connection_control_roundtrip() { + let manager = make_manager().await; + let mut rx = manager.subscribe_conn_control(); + manager + .subscription_health() + .consumer_attached(SubscriptionPath::ConnectionControl); + + let cancel = CancellationToken::new(); + let manager_clone = manager.clone(); + let task_cancel = cancel.clone(); + let task = tokio::spawn(async move { + manager_clone + .run_conn_control_subscriber_until_cancelled(task_cancel) + .await + }); + manager + .subscription_health() + .wait_for_state( + SubscriptionPath::ConnectionControl, + SubscriptionPathState::Ready, + Duration::from_secs(2), + ) + .await + .expect("connection-control Redis subscription handshake"); + + let ctx = ctx(0xaaaa, "a.example"); + let sent = ConnControl::DisconnectCommunity; + manager + .publish_conn_control(&ctx, &sent) + .await + .expect("publish failed"); + + let received = tokio::time::timeout(Duration::from_secs(2), rx.recv()) + .await + .expect("timeout") + .expect("channel closed"); + assert_eq!( + received, + ScopedConnControl { + community_id: ctx.community(), + command: sent, + } + ); + + cancel.cancel(); + task.await.expect("connection-control subscriber shutdown"); } #[tokio::test] diff --git a/crates/buzz-pubsub/src/subscriber.rs b/crates/buzz-pubsub/src/subscriber.rs index 88826ed99be..27b1dea2d5a 100644 --- a/crates/buzz-pubsub/src/subscriber.rs +++ b/crates/buzz-pubsub/src/subscriber.rs @@ -7,7 +7,9 @@ use std::time::Duration; use futures_util::StreamExt; use nostr::JsonUtil; use tokio::sync::{broadcast, mpsc, Mutex}; +use tokio_util::sync::CancellationToken; +use crate::health::{SubscriptionHealth, SubscriptionPath, SubscriptionTransitionReason}; use crate::topic::EventTopicKey; use crate::ChannelEvent; @@ -39,31 +41,45 @@ pub(crate) async fn run_subscriber( broadcast_tx: broadcast::Sender, desired_topics: DesiredTopics, mut subscription_rx: mpsc::Receiver, + cancel: CancellationToken, + health: Arc, ) { let mut backoff_secs = BACKOFF_INITIAL_SECS; - loop { - match connect_and_subscribe( + health.connecting(SubscriptionPath::Event); + let attempt = connect_and_subscribe( &redis_url, &broadcast_tx, desired_topics.clone(), &mut subscription_rx, - ) - .await - { + &health, + ); + let result = tokio::select! { + () = cancel.cancelled() => return, + result = attempt => result, + }; + match result { Ok(()) => { // Stream ended cleanly (Redis returned None). The connection was // established and ran successfully, so reset backoff to the initial // value — a brief Redis restart should reconnect quickly. backoff_secs = BACKOFF_INITIAL_SECS; + health.reconnecting( + SubscriptionPath::Event, + SubscriptionTransitionReason::StreamClosed, + ); tracing::warn!("Redis pub/sub stream ended (clean disconnect) — reconnecting in {backoff_secs}s"); } - Err(e) => { + Err((reason, e)) => { + health.reconnecting(SubscriptionPath::Event, reason); tracing::error!("Redis pub/sub error: {e} — reconnecting in {backoff_secs}s"); } } - tokio::time::sleep(Duration::from_secs(backoff_secs)).await; + tokio::select! { + () = cancel.cancelled() => return, + () = tokio::time::sleep(Duration::from_secs(backoff_secs)) => {} + } backoff_secs = (backoff_secs * 2).min(BACKOFF_MAX_SECS); tracing::info!("Attempting to reconnect to Redis pub/sub..."); @@ -77,9 +93,14 @@ async fn connect_and_subscribe( broadcast_tx: &broadcast::Sender, desired_topics: DesiredTopics, subscription_rx: &mut mpsc::Receiver, -) -> Result<(), redis::RedisError> { - let client = redis::Client::open(redis_url)?; - let conn = client.get_async_pubsub().await?; + health: &SubscriptionHealth, +) -> Result<(), (SubscriptionTransitionReason, redis::RedisError)> { + let client = redis::Client::open(redis_url) + .map_err(|error| (SubscriptionTransitionReason::Dial, error))?; + let conn = client + .get_async_pubsub() + .await + .map_err(|error| (SubscriptionTransitionReason::Dial, error))?; let (mut sink, mut stream) = conn.split(); let mut active_topics = HashSet::new(); @@ -93,10 +114,14 @@ async fn connect_and_subscribe( for topic in initial_topics { let channel = topic.redis_channel(); - sink.subscribe(&channel).await?; + sink.subscribe(&channel) + .await + .map_err(|error| (SubscriptionTransitionReason::Subscribe, error))?; active_topics.insert(channel); } + health.network_ready(SubscriptionPath::Event); + tracing::info!( topic_count = active_topics.len(), "Redis pub/sub subscriber connected with dynamic scoped subscriptions" @@ -109,14 +134,18 @@ async fn connect_and_subscribe( SubscriptionCommand::Subscribe(topic) => { let channel = topic.redis_channel(); if active_topics.insert(channel.clone()) { - sink.subscribe(&channel).await?; + sink.subscribe(&channel) + .await + .map_err(|error| (SubscriptionTransitionReason::Subscribe, error))?; } } SubscriptionCommand::UnsubscribeIfIdle(topic) => { if desired_refcount(&desired_topics, topic).await == 0 { let channel = topic.redis_channel(); if active_topics.remove(&channel) { - sink.unsubscribe(&channel).await?; + sink.unsubscribe(&channel) + .await + .map_err(|error| (SubscriptionTransitionReason::Subscribe, error))?; } } } diff --git a/crates/buzz-relay/src/lib.rs b/crates/buzz-relay/src/lib.rs index 18ea187fc7d..f225006c8e8 100644 --- a/crates/buzz-relay/src/lib.rs +++ b/crates/buzz-relay/src/lib.rs @@ -38,6 +38,8 @@ pub mod protocol; /// Durable NIP-PL matcher and delivery worker. pub mod push_runtime; mod readiness; +/// Owned runtime for the relay's three Redis cross-pod subscription paths. +pub mod redis_subscription_runtime; /// Axum router construction. pub mod router; /// Shared application state. diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 206f0329c0e..cf29433dcc4 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -460,25 +460,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { .await .map_err(|e| anyhow::anyhow!("PubSub init failed: {e}"))?, ); - info!("Redis pub/sub connected"); - - // Spawn Redis pub/sub subscriber for multi-node fan-out. - // Events published by other relay instances are received here and - // fanned out to local WebSocket subscribers. - 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 }); + info!("Redis pub/sub manager initialized"); let auth = AuthService::new(config.auth.clone()); @@ -525,6 +507,9 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { media_storage, ); let state = Arc::new(app_state); + let redis_subscriptions = + buzz_relay::redis_subscription_runtime::RedisSubscriptionRuntime::start(Arc::clone(&state)); + info!("Redis subscription runtime started"); // Inter-relay mesh (BUZZ_MESH seam). `boot_mesh` returns None when the // kill switch is off — nothing is bound, published, or spawned, so the @@ -964,65 +949,6 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { }); } - // Multi-node fan-out consumer: receive events from Redis pub/sub - // (published by other relay instances) and fan out to local WS subscribers. - { - let state_for_sub = Arc::clone(&state); - let mut rx = state_for_sub.pubsub.subscribe_local(); - tokio::spawn(async move { - loop { - match rx.recv().await { - Ok(channel_event) => { - buzz_relay::handlers::event::fan_out_pubsub_event( - &state_for_sub, - channel_event, - ) - .await; - } - Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { - metrics::counter!("buzz_multinode_fanout_lag_total").increment(n); - tracing::warn!("Multi-node fan-out lagged by {n} messages"); - } - Err(tokio::sync::broadcast::error::RecvError::Closed) => { - tracing::error!("Multi-node fan-out broadcast channel closed"); - break; - } - } - } - }); - } - - // Cross-pod cache-invalidation consumer: receive cache-key drops from Redis - // pub/sub (published by other relay instances when membership/visibility - // changes) and apply the matching local moka drop. Uses the `*_local` drop - // variants so a received drop is never re-published. - { - let state_for_cache = Arc::clone(&state); - let mut rx = state_for_cache.pubsub.subscribe_cache_invalidations(); - tokio::spawn(async move { - loop { - match rx.recv().await { - Ok(scoped) => { - // The Redis topic carries the originating community, - // and the local moka keys carry that same label. Apply - // only the matching tenant-local drop; a mutation in A - // must not flush B's derived state. - state_for_cache - .apply_cache_invalidation(scoped.community_id, scoped.invalidation); - } - Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { - metrics::counter!("buzz_cache_invalidation_lag_total").increment(n); - tracing::warn!("Cache-invalidation consumer lagged by {n} messages"); - } - Err(tokio::sync::broadcast::error::RecvError::Closed) => { - tracing::error!("Cache-invalidation broadcast channel closed"); - break; - } - } - } - }); - } - // Durable lifecycle backstop: Redis pub/sub cannot deliver to a pod that was // offline. Periodically revalidate only communities with local live sockets // so missed archive commands still converge without a global DB scan. @@ -1041,50 +967,6 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { )); } - // Cross-pod connection-control consumer: receive disconnect commands from - // Redis pub/sub (published by the pod that recorded a ban) and close any - // matching local sockets. A member's live connections may land on any pod, - // so this is how a ban reaches sockets the banning pod does not hold. The DB - // ban row is the durable backstop; even a dropped command still refuses the - // banned member's next auth attempt at the auth seam. - { - let state_for_conn_ctrl = Arc::clone(&state); - let mut rx = state_for_conn_ctrl.pubsub.subscribe_conn_control(); - tokio::spawn(async move { - loop { - match rx.recv().await { - Ok(scoped) => match scoped.command { - buzz_pubsub::conn_control::ConnControl::DisconnectCommunity => { - state_for_conn_ctrl - .community_connections - .disconnect_community(scoped.community_id); - } - buzz_pubsub::conn_control::ConnControl::DisconnectPubkey { - pubkey, - event_id, - reason, - } => { - state_for_conn_ctrl.conn_manager.disconnect_pubkey( - scoped.community_id, - &pubkey, - &event_id, - &reason, - ); - } - }, - Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { - metrics::counter!("buzz_conn_control_lag_total").increment(n); - tracing::warn!("Connection-control consumer lagged by {n} messages"); - } - Err(tokio::sync::broadcast::error::RecvError::Closed) => { - tracing::error!("Connection-control broadcast channel closed"); - break; - } - } - } - }); - } - let router = build_router(Arc::clone(&state)); let health_router = build_health_router(Arc::clone(&state)); @@ -1141,6 +1023,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { metrics::gauge!("buzz_redis_pool_size").set(rs.size as f64); metrics::gauge!("buzz_redis_pool_max").set(rs.max_size as f64); metrics::gauge!("buzz_redis_pool_waiting").set(rs.waiting as f64); + pool_state.pubsub.subscription_health().refresh_metrics(); let deletion_store = pool_state.db.deletion_store(); match deletion_store.reap_expired_serving_write_leases(1000).await { @@ -1214,7 +1097,11 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { }); } - serve(router, health_router, Arc::clone(&state)).await?; + let serve_result = serve(router, health_router, Arc::clone(&state)).await; + redis_subscriptions + .shutdown(std::time::Duration::from_secs(5)) + .await; + serve_result?; state.community_revalidator_cancel.cancel(); // Signal the audit worker to stop accepting, flush buffered entries, and diff --git a/crates/buzz-relay/src/metrics.rs b/crates/buzz-relay/src/metrics.rs index f71894116c3..4801c4abc19 100644 --- a/crates/buzz-relay/src/metrics.rs +++ b/crates/buzz-relay/src/metrics.rs @@ -206,6 +206,7 @@ pub fn try_install(port: u16, gauge_idle_timeout_secs: u64) -> Result<(), Metric .map_err(|_error| MetricsInstallError::RecorderConflict)?; describe_readiness_metrics(); describe_db_pool_metrics(); + describe_redis_subscription_metrics(); tokio::spawn(exporter); Ok(()) } @@ -257,6 +258,54 @@ pub(crate) fn describe_db_pool_metrics() { ); } +/// Register the fixed-cardinality Redis subscription-path metric contract. +pub(crate) fn describe_redis_subscription_metrics() { + metrics::describe_gauge!( + "buzz_redis_subscription_path_state", + "One-hot end-to-end state by required Redis subscription path and bounded state" + ); + metrics::describe_gauge!( + "buzz_redis_subscription_last_ready_timestamp_seconds", + metrics::Unit::Seconds, + "Unix timestamp of the latest completed consumer-plus-Redis subscription handshake" + ); + metrics::describe_gauge!( + "buzz_redis_subscription_last_readiness_duration_seconds", + metrics::Unit::Seconds, + "Seconds spent continuously not ready before the latest completed end-to-end subscription handshake" + ); + metrics::describe_gauge!( + "buzz_redis_subscription_not_ready_duration_seconds", + metrics::Unit::Seconds, + "Seconds spent continuously outside ready, or zero while the subscription path is ready" + ); + metrics::describe_gauge!( + "buzz_redis_subscription_state_duration_seconds", + metrics::Unit::Seconds, + "Seconds spent in the current bounded subscription path state" + ); + metrics::describe_gauge!( + "buzz_redis_subscription_consecutive_failures", + "Consecutive connection or subscription failures since the path was last ready" + ); + metrics::describe_counter!( + "buzz_redis_subscription_attempts_total", + "Redis connection and subscription attempts by required path" + ); + metrics::describe_counter!( + "buzz_redis_subscription_transitions_total", + "Redis subscription path transitions by path, bounded transition, and bounded reason" + ); + metrics::describe_gauge!( + "buzz_redis_subscription_paths_ready", + "Number of required Redis subscription paths currently ready in this relay process" + ); + metrics::describe_gauge!( + "buzz_redis_subscription_all_ready", + "Whether all required Redis subscription paths are currently ready in this relay process" + ); +} + #[cfg(test)] pub(crate) fn readiness_test_recorder() -> ( metrics_exporter_prometheus::PrometheusRecorder, @@ -330,6 +379,8 @@ pub async fn track_metrics(req: Request, next: Next) -> Response { mod contract_tests { use std::collections::BTreeSet; + use buzz_pubsub::health::{SubscriptionHealth, SubscriptionPath}; + const OUTCOMES: [&str; 4] = ["success", "timeout", "error", "cancelled"]; fn label_keys(line: &str) -> BTreeSet<&str> { @@ -344,6 +395,98 @@ mod contract_tests { .unwrap_or_default() } + #[test] + fn redis_subscription_scrape_has_fixed_labels_and_one_hot_state() { + let (recorder, handle) = super::readiness_test_recorder(); + metrics::with_local_recorder(&recorder, || { + super::describe_redis_subscription_metrics(); + let health = SubscriptionHealth::new(); + for path in SubscriptionPath::ALL { + health.consumer_attached(path); + health.connecting(path); + health.network_ready(path); + } + health.refresh_metrics(); + }); + + let scrape = handle.render(); + assert!(scrape.contains("# TYPE buzz_redis_subscription_path_state gauge")); + assert!( + scrape.contains("# TYPE buzz_redis_subscription_last_ready_timestamp_seconds gauge") + ); + assert!( + scrape.contains("# TYPE buzz_redis_subscription_last_readiness_duration_seconds gauge") + ); + assert!(scrape.contains("# TYPE buzz_redis_subscription_not_ready_duration_seconds gauge")); + assert!(scrape.contains("# TYPE buzz_redis_subscription_state_duration_seconds gauge")); + assert!(scrape.contains("# TYPE buzz_redis_subscription_consecutive_failures gauge")); + assert!(scrape.contains("# TYPE buzz_redis_subscription_attempts_total counter")); + assert!(scrape.contains("# TYPE buzz_redis_subscription_transitions_total counter")); + assert!(scrape.contains("# TYPE buzz_redis_subscription_paths_ready gauge")); + assert!(scrape.contains("# TYPE buzz_redis_subscription_all_ready gauge")); + + let state_lines = scrape + .lines() + .filter(|line| line.starts_with("buzz_redis_subscription_path_state{")) + .collect::>(); + assert_eq!(state_lines.len(), 15, "unexpected state series:\n{scrape}"); + assert_eq!( + state_lines + .iter() + .filter(|line| line.ends_with(" 1")) + .count(), + 3, + "exactly one state per path must be active:\n{scrape}" + ); + for line in state_lines { + assert_eq!(label_keys(line), BTreeSet::from(["path", "state"])); + } + + for line in scrape.lines().filter(|line| { + line.starts_with("buzz_redis_subscription_last_ready_timestamp_seconds{") + }) { + assert_eq!(label_keys(line), BTreeSet::from(["path"])); + } + for prefix in [ + "buzz_redis_subscription_last_readiness_duration_seconds{", + "buzz_redis_subscription_not_ready_duration_seconds{", + "buzz_redis_subscription_state_duration_seconds{", + "buzz_redis_subscription_consecutive_failures{", + "buzz_redis_subscription_attempts_total{", + ] { + let lines = scrape + .lines() + .filter(|line| line.starts_with(prefix)) + .collect::>(); + assert_eq!(lines.len(), 3, "unexpected series for {prefix}:\n{scrape}"); + for line in lines { + assert_eq!(label_keys(line), BTreeSet::from(["path"])); + } + } + for line in scrape + .lines() + .filter(|line| line.starts_with("buzz_redis_subscription_transitions_total{")) + { + assert_eq!( + label_keys(line), + BTreeSet::from(["path", "reason", "transition"]) + ); + } + assert!(scrape.contains("buzz_redis_subscription_paths_ready 3")); + assert!(scrape.contains("buzz_redis_subscription_all_ready 1")); + for forbidden in [ + "redis_url", + "channel", + "community", + "tenant", + "credential", + "payload", + "error", + ] { + assert!(!scrape.contains(&format!("{forbidden}=\""))); + } + } + #[test] fn production_builder_exports_frozen_db_pool_contract_and_187_series_budget() { let (recorder, handle) = super::readiness_test_recorder(); diff --git a/crates/buzz-relay/src/redis_subscription_runtime.rs b/crates/buzz-relay/src/redis_subscription_runtime.rs new file mode 100644 index 00000000000..1157a5e08eb --- /dev/null +++ b/crates/buzz-relay/src/redis_subscription_runtime.rs @@ -0,0 +1,424 @@ +//! Process-owned Redis subscription sockets and their relay-side consumers. + +use std::future::Future; +use std::panic::AssertUnwindSafe; +use std::sync::Arc; +use std::time::Duration; + +use buzz_pubsub::health::{SubscriptionHealth, SubscriptionPath, SubscriptionTransitionReason}; +use futures_util::FutureExt; +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; + +use crate::AppState; + +const OWNED_TASK_COUNT: usize = 6; + +#[derive(Clone, Copy)] +enum TaskKind { + Network, + Consumer, +} + +struct OwnedTask { + path: SubscriptionPath, + handle: JoinHandle<()>, +} + +/// Owns both halves of every required cross-pod subscription path. +/// +/// Each path is ready only when its relay-side consumer is attached and its +/// dedicated Redis connection has completed `SUBSCRIBE` or `PSUBSCRIBE`. +#[must_use = "the Redis subscription runtime must be retained and shut down"] +pub struct RedisSubscriptionRuntime { + cancel: CancellationToken, + health: Arc, + tasks: Vec, + shutdown_complete: bool, +} + +impl RedisSubscriptionRuntime { + /// Attach all consumers before starting any Redis socket, then retain all + /// six task handles for supervision and bounded shutdown. + pub fn start(state: Arc) -> Self { + let cancel = CancellationToken::new(); + let health = state.pubsub.subscription_health(); + let mut tasks = Vec::with_capacity(OWNED_TASK_COUNT); + + let event_rx = state.pubsub.subscribe_local(); + health.consumer_attached(SubscriptionPath::Event); + tasks.push(spawn_owned( + SubscriptionPath::Event, + TaskKind::Consumer, + Arc::clone(&health), + cancel.clone(), + run_event_consumer(Arc::clone(&state), event_rx, cancel.clone()), + )); + + let cache_rx = state.pubsub.subscribe_cache_invalidations(); + health.consumer_attached(SubscriptionPath::Cache); + tasks.push(spawn_owned( + SubscriptionPath::Cache, + TaskKind::Consumer, + Arc::clone(&health), + cancel.clone(), + run_cache_consumer(Arc::clone(&state), cache_rx, cancel.clone()), + )); + + let control_rx = state.pubsub.subscribe_conn_control(); + health.consumer_attached(SubscriptionPath::ConnectionControl); + tasks.push(spawn_owned( + SubscriptionPath::ConnectionControl, + TaskKind::Consumer, + Arc::clone(&health), + cancel.clone(), + run_control_consumer(Arc::clone(&state), control_rx, cancel.clone()), + )); + + let pubsub = Arc::clone(&state.pubsub); + tasks.push(spawn_owned( + SubscriptionPath::Event, + TaskKind::Network, + Arc::clone(&health), + cancel.clone(), + pubsub.run_subscriber_until_cancelled(cancel.clone()), + )); + + let pubsub = Arc::clone(&state.pubsub); + tasks.push(spawn_owned( + SubscriptionPath::Cache, + TaskKind::Network, + Arc::clone(&health), + cancel.clone(), + pubsub.run_cache_invalidation_subscriber_until_cancelled(cancel.clone()), + )); + + let pubsub = Arc::clone(&state.pubsub); + tasks.push(spawn_owned( + SubscriptionPath::ConnectionControl, + TaskKind::Network, + Arc::clone(&health), + cancel.clone(), + pubsub.run_conn_control_subscriber_until_cancelled(cancel.clone()), + )); + + debug_assert_eq!(tasks.len(), OWNED_TASK_COUNT); + Self { + cancel, + health, + tasks, + shutdown_complete: false, + } + } + + /// Cancel and await every owned task within one shared graceful deadline. + /// Tasks that miss it are aborted and joined before shutdown reports completion. + pub async fn shutdown(mut self, timeout: Duration) { + self.cancel.cancel(); + let deadline = tokio::time::Instant::now() + timeout; + let mut timed_out = Vec::new(); + for mut task in self.tasks.drain(..) { + if tokio::time::timeout_at(deadline, &mut task.handle) + .await + .is_err() + { + task.handle.abort(); + self.health + .failed(task.path, SubscriptionTransitionReason::ShutdownTimeout); + timed_out.push(task); + } + } + + for task in timed_out { + let _ = task.handle.await; + } + + for path in SubscriptionPath::ALL { + self.health.stopped(path); + } + self.shutdown_complete = true; + } +} + +impl Drop for RedisSubscriptionRuntime { + fn drop(&mut self) { + if self.shutdown_complete { + return; + } + self.cancel.cancel(); + for task in &self.tasks { + task.handle.abort(); + } + for path in SubscriptionPath::ALL { + self.health + .failed(path, SubscriptionTransitionReason::OwnerDropped); + } + } +} + +fn spawn_owned( + path: SubscriptionPath, + kind: TaskKind, + health: Arc, + cancel: CancellationToken, + future: impl Future + Send + 'static, +) -> OwnedTask { + let handle = tokio::spawn(async move { + let result = AssertUnwindSafe(future).catch_unwind().await; + if cancel.is_cancelled() { + return; + } + let reason = match result { + Err(_) => SubscriptionTransitionReason::Panic, + Ok(()) => match kind { + TaskKind::Network => SubscriptionTransitionReason::StreamClosed, + TaskKind::Consumer => SubscriptionTransitionReason::ConsumerClosed, + }, + }; + health.failed(path, reason); + }); + OwnedTask { path, handle } +} + +async fn run_event_consumer( + state: Arc, + mut rx: tokio::sync::broadcast::Receiver, + cancel: CancellationToken, +) { + loop { + tokio::select! { + () = cancel.cancelled() => return, + result = rx.recv() => match result { + Ok(channel_event) => { + crate::handlers::event::fan_out_pubsub_event(&state, channel_event).await; + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { + metrics::counter!("buzz_multinode_fanout_lag_total").increment(n); + tracing::warn!("Multi-node fan-out lagged by {n} messages"); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => return, + } + } + } +} + +async fn run_cache_consumer( + state: Arc, + mut rx: tokio::sync::broadcast::Receiver< + buzz_pubsub::cache_invalidation::ScopedCacheInvalidation, + >, + cancel: CancellationToken, +) { + loop { + tokio::select! { + () = cancel.cancelled() => return, + result = rx.recv() => match result { + Ok(scoped) => { + state.apply_cache_invalidation(scoped.community_id, scoped.invalidation); + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { + metrics::counter!("buzz_cache_invalidation_lag_total").increment(n); + tracing::warn!("Cache-invalidation consumer lagged by {n} messages"); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => return, + } + } + } +} + +async fn run_control_consumer( + state: Arc, + mut rx: tokio::sync::broadcast::Receiver, + cancel: CancellationToken, +) { + loop { + tokio::select! { + () = cancel.cancelled() => return, + result = rx.recv() => match result { + Ok(scoped) => match scoped.command { + buzz_pubsub::conn_control::ConnControl::DisconnectCommunity => { + state.community_connections.disconnect_community(scoped.community_id); + } + buzz_pubsub::conn_control::ConnControl::DisconnectPubkey { + pubkey, + event_id, + reason, + } => { + state.conn_manager.disconnect_pubkey( + scoped.community_id, + &pubkey, + &event_id, + &reason, + ); + } + }, + Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { + metrics::counter!("buzz_conn_control_lag_total").increment(n); + tracing::warn!("Connection-control consumer lagged by {n} messages"); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => return, + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use buzz_pubsub::health::SubscriptionPathState; + use std::sync::atomic::{AtomicBool, Ordering}; + + struct DropFlag(Arc); + + impl Drop for DropFlag { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + + #[tokio::test] + async fn production_runtime_owns_both_halves_and_attaches_consumers_first() { + let state = crate::state::tests::test_state().await; + let health = state.pubsub.subscription_health(); + let runtime = RedisSubscriptionRuntime::start(state); + + assert_eq!(runtime.tasks.len(), OWNED_TASK_COUNT); + for path in SubscriptionPath::ALL { + assert!(health.snapshot(path).consumer_attached); + } + + for path in SubscriptionPath::ALL { + health + .wait_for_state( + path, + SubscriptionPathState::Reconnecting, + Duration::from_secs(2), + ) + .await + .expect("unreachable Redis must be observable as reconnecting"); + } + + runtime.shutdown(Duration::from_secs(1)).await; + for path in SubscriptionPath::ALL { + assert_eq!(health.snapshot(path).state, SubscriptionPathState::Stopped); + } + } + + #[tokio::test] + async fn dropping_owner_latches_every_path_failed() { + let state = crate::state::tests::test_state().await; + let health = state.pubsub.subscription_health(); + let runtime = RedisSubscriptionRuntime::start(state); + + drop(runtime); + + for path in SubscriptionPath::ALL { + assert_eq!(health.snapshot(path).state, SubscriptionPathState::Failed); + } + } + + #[tokio::test] + async fn owned_task_panic_latches_path_failed() { + let health = Arc::new(SubscriptionHealth::new()); + let task = spawn_owned( + SubscriptionPath::Event, + TaskKind::Consumer, + Arc::clone(&health), + CancellationToken::new(), + async { panic!("test panic") }, + ); + + task.handle.await.expect("panic is supervised inside task"); + assert_eq!( + health.snapshot(SubscriptionPath::Event).state, + SubscriptionPathState::Failed + ); + } + + #[tokio::test] + async fn unexpected_consumer_return_latches_path_failed() { + let health = Arc::new(SubscriptionHealth::new()); + let task = spawn_owned( + SubscriptionPath::Cache, + TaskKind::Consumer, + Arc::clone(&health), + CancellationToken::new(), + async {}, + ); + + task.handle.await.expect("task join"); + assert_eq!( + health.snapshot(SubscriptionPath::Cache).state, + SubscriptionPathState::Failed + ); + } + + #[tokio::test] + async fn shutdown_joins_every_task_after_aborting_timed_out_work() { + let cancel = CancellationToken::new(); + let health = Arc::new(SubscriptionHealth::new()); + let event_dropped = Arc::new(AtomicBool::new(false)); + let cache_dropped = Arc::new(AtomicBool::new(false)); + + let (event_started_tx, event_started_rx) = tokio::sync::oneshot::channel(); + let event_flag = Arc::clone(&event_dropped); + let event_task = spawn_owned( + SubscriptionPath::Event, + TaskKind::Network, + Arc::clone(&health), + cancel.clone(), + async move { + let _drop_flag = DropFlag(event_flag); + event_started_tx + .send(()) + .expect("signal event task started"); + std::future::pending::<()>().await; + }, + ); + + let (cache_started_tx, cache_started_rx) = tokio::sync::oneshot::channel(); + let cache_flag = Arc::clone(&cache_dropped); + let cache_task = spawn_owned( + SubscriptionPath::Cache, + TaskKind::Consumer, + Arc::clone(&health), + cancel.clone(), + async move { + let _drop_flag = DropFlag(cache_flag); + cache_started_tx + .send(()) + .expect("signal cache task started"); + std::future::pending::<()>().await; + }, + ); + + event_started_rx.await.expect("event task started"); + cache_started_rx.await.expect("cache task started"); + + RedisSubscriptionRuntime { + cancel, + health: Arc::clone(&health), + tasks: vec![event_task, cache_task], + shutdown_complete: false, + } + .shutdown(Duration::ZERO) + .await; + + assert!( + event_dropped.load(Ordering::SeqCst), + "event resource must be released before shutdown returns" + ); + assert!( + cache_dropped.load(Ordering::SeqCst), + "cache resource must be released before shutdown returns" + ); + assert_eq!( + health.snapshot(SubscriptionPath::Event).state, + SubscriptionPathState::Failed + ); + assert_eq!( + health.snapshot(SubscriptionPath::Cache).state, + SubscriptionPathState::Failed + ); + } +} diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 6a14fe0cfee..097f81b7367 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -104,6 +104,12 @@ run_unit_tests() { run_test_step "buzz-db unit tests" \ cargo test -p buzz-db --lib -- --nocapture + # Mirror the complete infrastructure-free buzz-pubsub suite selected by the + # nextest path in `just test-unit`. Ignored real-Redis tests run in CI's + # required PostgreSQL/Redis lane. + run_test_step "buzz-pubsub unit tests" \ + cargo test -p buzz-pubsub -- --nocapture + # Multi-tenant conformance gate: independent replay checker + golden # fixtures (buzz-conformance). Pure in-process trace replay, no infra. run_test_step "buzz-conformance tests" \ @@ -142,6 +148,18 @@ run_unit_tests() { run_test_step "buzz-relay side-effects helper tests" \ cargo test -p buzz-relay --lib handlers::side_effects::tests:: -- --nocapture + + run_test_step "buzz-relay Redis metric contract tests" \ + cargo test -p buzz-relay --lib metrics::contract_tests:: -- --nocapture + + run_test_step "buzz-relay Redis runtime tests" \ + cargo test -p buzz-relay --lib redis_subscription_runtime::tests:: -- --nocapture + + run_test_step "buzz-relay binary tests" \ + cargo test -p buzz-relay --bin buzz-relay -- --nocapture + + run_test_step "buzz-relay boot lifecycle tests" \ + cargo test -p buzz-relay --test boot_lifecycle -- --nocapture } # ---- DB / integration tests (infra required) --------------------------------