fix(relay): await Redis task cancellation

Signed-off-by: Ravneet Arora <rarora@squareup.com>

.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-relay/src/redis_subscription_runtime.rs

@@ -111,11 +111,13 @@ impl RedisSubscriptionRuntime {
         }
     }
 
-    /// Cancel and await every owned task within one shared deadline.
+    /// 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;
-        for task in &mut self.tasks {
+        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()
@@ -123,8 +125,14 @@ impl RedisSubscriptionRuntime {
                 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);
         }
@@ -258,6 +266,15 @@ async fn run_control_consumer(
 mod tests {
     use super::*;
     use buzz_pubsub::health::SubscriptionPathState;
+    use std::sync::atomic::{AtomicBool, Ordering};
+
+    struct DropFlag(Arc<AtomicBool>);
+
+    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() {
@@ -335,4 +352,73 @@ mod tests {
             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
+        );
+    }
 }

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