feat(relay): observe Redis subscription paths

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

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.

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"] }

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<ScopedCacheInvalidation>,
 ) {
-    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<ScopedCacheInvalidation>,
+    cancel: CancellationToken,
+    health: Arc<SubscriptionHealth>,
+) {
+    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<ScopedCacheInvalidation>,
-) -> 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}"

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<ScopedConnControl>,
 ) {
-    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<ScopedConnControl>,
+    cancel: CancellationToken,
+    health: Arc<SubscriptionHealth>,
+) {
+    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<ScopedConnControl>,
-) -> 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}");