diff --git a/.github/workflows/_ci-relay.yml b/.github/workflows/_ci-relay.yml index 97d32f912c..4a21a6dff7 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/Justfile b/Justfile index 00c4fecc2f..6d3626c032 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-relay/src/redis_subscription_runtime.rs b/crates/buzz-relay/src/redis_subscription_runtime.rs index 8ef38983e6..1157a5e08e 100644 --- a/crates/buzz-relay/src/redis_subscription_runtime.rs +++ b/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); + + 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 + ); + } } diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 6a14fe0cfe..097f81b736 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) --------------------------------