agent-file-tools 0.47.3

//! Dispatch-path metrics and health-report helpers for the subc transport loop.

use std::time::{SystemTime, UNIX_EPOCH};

use super::{
    json, Arc, AtomicBool, AtomicU64, AtomicUsize, Duration, Executor, HashMap, HealthReport,
    HealthStatus, Instant, Ordering, PendingBind, RootHealthSnapshot, RouteChannel, Value,
    DISPATCH_PATH_BIND_WARN_AFTER, WRITER_QUEUE_CAPACITY,
};
use crate::context::{App, AppContext, RootHealthState};
use crate::executor::BindBlockerSnapshot;

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(super) struct ReapBlockerCensus {
    pub(super) deleted_retained: usize,
    pub(super) bound_routes: usize,
    pub(super) unbound_quiesced: usize,
    pub(super) bash_waits: usize,
    pub(super) maintenance_pending: usize,
    pub(super) maintenance_queued: usize,
    pub(super) pending_binds: usize,
    pub(super) actor_busy: usize,
    pub(super) actor_state_busy: usize,
    pub(super) artifact_eviction_blocked: usize,
    pub(super) artifact_eviction_failed: usize,
}

impl ReapBlockerCensus {
    pub(super) fn blocker_histogram(&self) -> String {
        format!(
            "bound_routes={},unbound_quiesced={},bash_waits={},maintenance_pending={},maintenance_queued={},pending_binds={},actor_busy={},actor_state_busy={},artifact_eviction_blocked={},artifact_eviction_failed={}",
            self.bound_routes,
            self.unbound_quiesced,
            self.bash_waits,
            self.maintenance_pending,
            self.maintenance_queued,
            self.pending_binds,
            self.actor_busy,
            self.actor_state_busy,
            self.artifact_eviction_blocked,
            self.artifact_eviction_failed,
        )
    }
}

struct ReapMetrics {
    last_sweep_ms: AtomicU64,
    deleted_retained: AtomicUsize,
    bound_routes: AtomicUsize,
    unbound_quiesced: AtomicUsize,
    bash_waits: AtomicUsize,
    maintenance_pending: AtomicUsize,
    maintenance_queued: AtomicUsize,
    pending_binds: AtomicUsize,
    actor_busy: AtomicUsize,
    actor_state_busy: AtomicUsize,
    artifact_eviction_blocked: AtomicUsize,
    artifact_eviction_failed: AtomicUsize,
}

impl ReapMetrics {
    fn new() -> Self {
        Self {
            last_sweep_ms: AtomicU64::new(0),
            deleted_retained: AtomicUsize::new(0),
            bound_routes: AtomicUsize::new(0),
            unbound_quiesced: AtomicUsize::new(0),
            bash_waits: AtomicUsize::new(0),
            maintenance_pending: AtomicUsize::new(0),
            maintenance_queued: AtomicUsize::new(0),
            pending_binds: AtomicUsize::new(0),
            actor_busy: AtomicUsize::new(0),
            actor_state_busy: AtomicUsize::new(0),
            artifact_eviction_blocked: AtomicUsize::new(0),
            artifact_eviction_failed: AtomicUsize::new(0),
        }
    }

    fn record(&self, now_ms: u64, census: ReapBlockerCensus) {
        self.last_sweep_ms.store(now_ms, Ordering::Relaxed);
        self.deleted_retained
            .store(census.deleted_retained, Ordering::Relaxed);
        self.bound_routes
            .store(census.bound_routes, Ordering::Relaxed);
        self.unbound_quiesced
            .store(census.unbound_quiesced, Ordering::Relaxed);
        self.bash_waits.store(census.bash_waits, Ordering::Relaxed);
        self.maintenance_pending
            .store(census.maintenance_pending, Ordering::Relaxed);
        self.maintenance_queued
            .store(census.maintenance_queued, Ordering::Relaxed);
        self.pending_binds
            .store(census.pending_binds, Ordering::Relaxed);
        self.actor_busy.store(census.actor_busy, Ordering::Relaxed);
        self.actor_state_busy
            .store(census.actor_state_busy, Ordering::Relaxed);
        self.artifact_eviction_blocked
            .store(census.artifact_eviction_blocked, Ordering::Relaxed);
        self.artifact_eviction_failed
            .store(census.artifact_eviction_failed, Ordering::Relaxed);
    }

    fn snapshot(&self) -> Value {
        json!({
            "deleted_retained": self.deleted_retained.load(Ordering::Relaxed),
            "blockers": {
                "bound_routes": self.bound_routes.load(Ordering::Relaxed),
                "unbound_quiesced": self.unbound_quiesced.load(Ordering::Relaxed),
                "bash_waits": self.bash_waits.load(Ordering::Relaxed),
                "maintenance_pending": self.maintenance_pending.load(Ordering::Relaxed),
                "maintenance_queued": self.maintenance_queued.load(Ordering::Relaxed),
                "pending_binds": self.pending_binds.load(Ordering::Relaxed),
                "actor_busy": self.actor_busy.load(Ordering::Relaxed),
                "actor_state_busy": self.actor_state_busy.load(Ordering::Relaxed),
                "artifact_eviction_blocked": self.artifact_eviction_blocked.load(Ordering::Relaxed),
                "artifact_eviction_failed": self.artifact_eviction_failed.load(Ordering::Relaxed),
            },
            "last_sweep_ms": self.last_sweep_ms.load(Ordering::Relaxed),
        })
    }
}

pub(super) struct DispatchPathMetrics {
    pub(super) origin: Instant,
    pub(super) frame_loop_last_tick_ms: AtomicU64,
    pub(super) writer_queued: AtomicUsize,
    pub(super) writer_active: AtomicBool,
    pub(super) writer_saturation_count: AtomicU64,
    pub(super) control_completion_queued: AtomicUsize,
    pub(super) maintenance_queued: AtomicUsize,
    pub(super) bash_deferred_queued: AtomicUsize,
    pub(super) bash_poll_touch_queued: AtomicUsize,
    pub(super) reliable_push_budget_deferrals: AtomicU64,
    pub(super) maintenance_budget_deferrals: AtomicU64,
    pub(super) response_tasks_live: AtomicUsize,
    reap: ReapMetrics,
}

impl DispatchPathMetrics {
    pub(super) fn new() -> Self {
        Self {
            origin: Instant::now(),
            frame_loop_last_tick_ms: AtomicU64::new(0),
            writer_queued: AtomicUsize::new(0),
            writer_active: AtomicBool::new(false),
            writer_saturation_count: AtomicU64::new(0),
            control_completion_queued: AtomicUsize::new(0),
            maintenance_queued: AtomicUsize::new(0),
            bash_deferred_queued: AtomicUsize::new(0),
            bash_poll_touch_queued: AtomicUsize::new(0),
            reliable_push_budget_deferrals: AtomicU64::new(0),
            maintenance_budget_deferrals: AtomicU64::new(0),
            response_tasks_live: AtomicUsize::new(0),
            reap: ReapMetrics::new(),
        }
    }

    fn now_ms(&self) -> u64 {
        duration_millis_u64(self.origin.elapsed())
    }

    pub(super) fn mark_frame_loop_tick(&self) {
        self.frame_loop_last_tick_ms
            .store(self.now_ms(), Ordering::Relaxed);
    }

    pub(super) fn record_reap(&self, census: ReapBlockerCensus) {
        let last_sweep_ms = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(duration_millis_u64)
            .unwrap_or(0);
        self.reap.record(last_sweep_ms, census);
    }

    fn reap_snapshot(&self) -> Value {
        self.reap.snapshot()
    }

    fn snapshot(
        &self,
        pending_binds: &HashMap<RouteChannel, PendingBind>,
        executor: &Executor,
    ) -> Value {
        let now = Instant::now();
        let oldest_pending_age_ms = pending_binds
            .values()
            .map(|bind| duration_millis_u64(now.saturating_duration_since(bind.started_at)))
            .max();
        let last_tick_ms = self.frame_loop_last_tick_ms.load(Ordering::Relaxed);
        json!({
            "frame_loop": {
                "last_tick_age_ms": self.now_ms().saturating_sub(last_tick_ms),
            },
            "pending_binds": {
                "count": pending_binds.len(),
                "oldest_age_ms": oldest_pending_age_ms,
            },
            "completion_channels": {
                "control": self.control_completion_queued.load(Ordering::Relaxed),
                "maintenance": self.maintenance_queued.load(Ordering::Relaxed),
                "bash_deferred": self.bash_deferred_queued.load(Ordering::Relaxed),
                "bash_poll_touch": self.bash_poll_touch_queued.load(Ordering::Relaxed),
            },
            "budget_deferrals": {
                "reliable_push": self.reliable_push_budget_deferrals.load(Ordering::Relaxed),
                "maintenance": self.maintenance_budget_deferrals.load(Ordering::Relaxed),
            },
            "writer": {
                "queued": self.writer_queued.load(Ordering::Relaxed),
                "active": self.writer_active.load(Ordering::Relaxed),
                "capacity": WRITER_QUEUE_CAPACITY,
                "saturation_count": self.writer_saturation_count.load(Ordering::Relaxed),
            },
            "response_tasks": {
                "live": self.response_tasks_live.load(Ordering::Relaxed),
            },
            "mutating_lanes": mutating_lanes_metrics(executor),
        })
    }
}

pub(super) struct ResponseTaskGuard {
    metrics: Arc<DispatchPathMetrics>,
}

impl ResponseTaskGuard {
    pub(super) fn new(metrics: &Arc<DispatchPathMetrics>) -> Self {
        metrics.response_tasks_live.fetch_add(1, Ordering::Relaxed);
        Self {
            metrics: Arc::clone(metrics),
        }
    }
}

impl Drop for ResponseTaskGuard {
    fn drop(&mut self) {
        self.metrics
            .response_tasks_live
            .fetch_sub(1, Ordering::Relaxed);
    }
}

fn duration_millis_u64(duration: Duration) -> u64 {
    duration.as_millis().min(u128::from(u64::MAX)) as u64
}

pub(super) fn warn_slow_pending_binds(
    pending_binds: &mut HashMap<RouteChannel, PendingBind>,
    executor: &Executor,
) {
    let now = Instant::now();
    for (route, pending) in pending_binds.iter_mut() {
        if pending.warned_half_deadline {
            continue;
        }
        let age = now.saturating_duration_since(pending.started_at);
        if age < DISPATCH_PATH_BIND_WARN_AFTER {
            continue;
        }
        pending.warned_half_deadline = true;
        let snapshot = executor
            .try_bind_blocker_snapshot(&pending.bind_root_id, &pending.configure_request_id)
            .unwrap_or_else(|| BindBlockerSnapshot {
                configure_state: "scheduler_busy",
                configure_phase_timings: None,
                blockers: vec!["scheduler_busy".to_string()],
            });
        crate::slog_warn!(
            "{}",
            pending_bind_breadcrumb(
                *route,
                &pending.bind_root_id,
                age,
                &pending.configure_request_id,
                &snapshot,
            )
        );
    }
}

fn pending_bind_breadcrumb(
    route: RouteChannel,
    root_id: &crate::path_identity::ProjectRootId,
    age: Duration,
    configure_request_id: &str,
    snapshot: &BindBlockerSnapshot,
) -> String {
    let blockers = if snapshot.blockers.is_empty() {
        "none".to_string()
    } else {
        snapshot.blockers.join(", ")
    };
    let phase_timings = snapshot
        .configure_phase_timings
        .as_deref()
        .unwrap_or("unavailable");
    format!(
        "subc attach: pending RouteBind route {route} for root {} crossed {}ms (configure_request_id={}, configure_state={}, configure_phase_timings=[{}], blockers=[{}])",
        root_id.as_path().display(),
        duration_millis_u64(age),
        configure_request_id,
        snapshot.configure_state,
        phase_timings,
        blockers,
    )
}

/// Per-root attributed-bytes rollup for health metrics: top roots by size
/// (same cap as the status payload's memory section) plus process RSS and
/// attributed totals. `None` scheduler entries (contended) yield a busy
/// marker instead of waiting.
fn memory_rollup_metrics(
    actor_entries: Option<Vec<(cortexkit_paths::ProjectRootId, Arc<AppContext>)>>,
) -> Value {
    let Some(entries) = actor_entries else {
        return json!({ "status": "busy" });
    };
    let mut roots = std::collections::BTreeMap::new();
    for (root_id, ctx) in entries {
        roots.insert(
            root_id.as_path().display().to_string(),
            ctx.memory_root_snapshot(),
        );
    }
    let snapshot = crate::memory::MemorySnapshot::new("ready", roots);
    let per_root: Value = snapshot
        .roots
        .iter()
        .map(|(root, detail)| {
            (
                root.clone(),
                json!({
                    "attributed_bytes": detail.attributed_bytes,
                    "status": detail.status,
                }),
            )
        })
        .collect::<serde_json::Map<String, Value>>()
        .into();
    json!({
        "status": "ready",
        "roots": per_root,
        "roots_total": snapshot.roots_total,
        "roots_omitted": snapshot.roots_omitted,
        "roots_omitted_bytes": snapshot.roots_omitted_bytes,
        "rss_bytes": snapshot.process.rss_bytes,
        "total_attributed_bytes": snapshot.process.total_attributed_bytes,
        "sqlite_bytes": snapshot.process.sqlite.memory_used_bytes,
    })
}

fn mutating_lanes_metrics(executor: &Executor) -> Value {
    match executor.try_mutating_lane_snapshots() {
        Some(snapshots) => Value::Array(
            snapshots
                .into_iter()
                .map(|snapshot| {
                    json!({
                        "root": snapshot.root_id.as_path().to_string_lossy(),
                        "request_id": snapshot.request_id,
                        "job": snapshot.command,
                        "started_age_ms": snapshot.started_age_ms,
                    })
                })
                .collect(),
        ),
        None => json!({ "scheduler_busy": true }),
    }
}

fn dispatch_liveness_metrics(executor: &Executor) -> Value {
    match executor.try_dispatch_liveness_snapshot() {
        Some(snapshot) => json!({
            "interactive": {
                "queued": snapshot.interactive.queued,
                "oldest_age_ms": snapshot.interactive.oldest_age_ms,
            },
            "maintenance": {
                "queued": snapshot.maintenance.queued,
                "oldest_age_ms": snapshot.maintenance.oldest_age_ms,
            },
            "running": {
                "interactive": snapshot.running.interactive,
                "maintenance": snapshot.running.maintenance,
            },
            "interactive_reserve": snapshot.interactive_reserve,
            "maintenance_cap": snapshot.maintenance_cap,
        }),
        None => json!({ "scheduler_busy": true }),
    }
}

pub(super) fn build_health_report(
    executor: &Executor,
    pending_binds: &HashMap<RouteChannel, PendingBind>,
    dispatch_path_metrics: &DispatchPathMetrics,
    shared_app: &App,
) -> HealthReport {
    // Health replies must stay cheap under any load (subc-health rule: a
    // probe that queues behind busy state lies about liveness and gets the
    // module health-killed). Every read below is try-lock-only: a contended
    // scheduler lock produces a degraded report with an empty root list instead
    // of waiting.
    let actor_entries = executor.try_actor_entries();
    let scheduler_busy = actor_entries.is_none();
    let actor_entries_for_memory = actor_entries.clone();
    let mut roots: Vec<RootHealthSnapshot> = actor_entries
        .unwrap_or_default()
        .into_iter()
        .map(|(root_id, ctx)| ctx.try_health_snapshot(root_id.as_path()))
        .collect();
    roots.sort_by(|left, right| left.project_root.cmp(&right.project_root));

    // Compact memory rollup for operator drill-down (`ck health aft`): per-root
    // attributed bytes with the same top-N cap as the status payload, plus
    // process totals. Memory estimates are themselves try-lock-only (busy
    // subsystems report as such), so this stays within the health-path lock
    // doctrine.
    let memory = memory_rollup_metrics(actor_entries_for_memory);

    // Health-verdict rule: DEGRADED means dispatch is impaired (an actor we
    // could not even snapshot without contention), never "a background index
    // is still warming". A serving root with search/callgraph mid-build is
    // healthy — component build states are informational detail, otherwise a
    // module with any active mason worktree reads permanently degraded and
    // the daemon's on-failing policies treat routine warmup as wreckage.
    let busy_roots = roots
        .iter()
        .filter(|root| matches!(root.state, RootHealthState::Busy))
        .count();
    let warming_roots = roots
        .iter()
        .filter(|root| !matches!(root.state, RootHealthState::Busy) && !root.is_fully_ready())
        .count();
    let detail = if scheduler_busy {
        Some("executor scheduler state could not be snapshotted without contention".to_string())
    } else if busy_roots > 0 {
        Some(format!(
            "{busy_roots} root actor(s) could not be snapshotted without contention"
        ))
    } else if warming_roots > 0 {
        Some(format!(
            "{warming_roots} root(s) warming background indexes (serving normally)"
        ))
    } else {
        None
    };

    HealthReport {
        // Failing to acquire the scheduler snapshot is itself dispatch contention;
        // reporting Ok with an empty root list would hide the condition being probed.
        status: if scheduler_busy || busy_roots > 0 {
            HealthStatus::Degraded
        } else {
            HealthStatus::Ok
        },
        detail,
        metrics: Some(json!({
            "actor_count": roots.iter().map(|root| root.actor_count).sum::<usize>(),
            "root_count": roots.len(),
            // Lifecycle audit counters (fleet leak tracking): process-wide
            // watcher runtimes, registered actor roots, and open routes.
            "runtime": {
                "live_watchers": shared_app.watcher_count(),
                "live_actor_roots": shared_app.actor_root_count(),
                "open_routes": shared_app.open_route_count(),
            },
            "memory": memory,
            "roots": roots,
            "reap": dispatch_path_metrics.reap_snapshot(),
            "dispatch_liveness": dispatch_liveness_metrics(executor),
            "dispatch_path": dispatch_path_metrics.snapshot(pending_binds, executor),
        })),
    }
}

#[cfg(test)]
mod tests {
    use super::super::test_support::{test_ctx, test_root};
    use super::super::{Lane, Response};
    use super::*;
    use serde_json::json;

    #[test]
    fn pending_bind_breadcrumb_names_every_blocker_class() {
        let (_dir, root) = test_root("breadcrumb-blockers");
        let cases = [
            "queued_behind_configure(2)",
            "queued_behind_maintenance(job=subc-maintenance-drain-watcher lane=Mutating root=/tmp/a age_ms=1)",
            "waiting_on_readers",
            "idle_workers==0(job=subc-bind-other lane=Mutating root=/tmp/b age_ms=2)",
        ];

        for blocker in cases {
            let breadcrumb = pending_bind_breadcrumb(
                RouteChannel {
                    channel: 7,
                    epoch: 1,
                },
                &root,
                Duration::from_secs(6),
                "subc-bind-7",
                &BindBlockerSnapshot {
                    configure_state: "queued",
                    configure_phase_timings: Some("artifact_owner_claim=12ms".to_string()),
                    blockers: vec![blocker.to_string()],
                },
            );
            assert!(
                breadcrumb.contains(blocker),
                "breadcrumb omitted blocker class: {breadcrumb}"
            );
            assert!(
                breadcrumb.contains("configure_phase_timings=[artifact_owner_claim=12ms]"),
                "breadcrumb omitted configure phase timings: {breadcrumb}"
            );
        }
    }

    #[test]
    fn health_report_includes_nonblocking_dispatch_liveness_for_queued_interactive() {
        let executor = Executor::with_config(crate::executor::ExecutorConfig {
            pool_size: 2,
            read_cap: 1,
            actor_cap: 1,
            heavy_permits: 2,
            drr_quantum: 1,
        });
        let (_dir_a, root_a) = test_root("health-liveness-a");
        let (_dir_b, root_b) = test_root("health-liveness-b");
        let (_dir_c, root_c) = test_root("health-liveness-c");
        executor.register_actor(root_a.clone(), test_ctx());
        executor.register_actor(root_b.clone(), test_ctx());
        executor.register_actor(root_c.clone(), test_ctx());

        let (started_tx, started_rx) = crossbeam_channel::bounded(2);
        let (release_tx, release_rx) = crossbeam_channel::bounded(2);
        let mut blockers = Vec::new();
        for (index, root) in [root_a, root_b].into_iter().enumerate() {
            let started_tx = started_tx.clone();
            let release_rx = release_rx.clone();
            blockers.push(executor.submit(
                root,
                Lane::PureRead,
                format!("health-blocker-{index}"),
                Box::new(move |_| {
                    started_tx.send(index).expect("signal blocker start");
                    release_rx
                        .recv_timeout(Duration::from_secs(2))
                        .expect("release blocker");
                    Response::success(format!("blocker-{index}"), json!({ "ok": true }))
                }),
            ));
        }
        for _ in 0..2 {
            started_rx
                .recv_timeout(Duration::from_secs(1))
                .expect("blocker starts");
        }

        let queued = executor.submit(
            root_c,
            Lane::PureRead,
            "queued-interactive".to_string(),
            Box::new(|_| Response::success("queued-interactive", json!({ "ok": true }))),
        );
        std::thread::sleep(Duration::from_millis(75));

        let metrics = DispatchPathMetrics::new();
        let pending_binds = HashMap::new();
        let report = build_health_report(
            &executor,
            &pending_binds,
            &metrics,
            &crate::context::App::default_shared(),
        );
        let dispatch = report
            .metrics
            .as_ref()
            .and_then(|metrics| metrics.get("dispatch_liveness"))
            .expect("dispatch_liveness metric");
        assert_eq!(dispatch.get("scheduler_busy"), None);
        assert_eq!(dispatch["interactive"]["queued"].as_u64(), Some(1));
        assert!(dispatch["interactive"]["oldest_age_ms"].as_u64().is_some());

        for _ in 0..2 {
            release_tx.send(()).expect("release blocker");
        }
        for blocker in blockers {
            blocker
                .recv_timeout(Duration::from_secs(1))
                .expect("blocker completion response");
        }
        queued
            .recv_timeout(Duration::from_secs(1))
            .expect("queued completion response");
    }

    #[test]
    fn health_metrics_memory_rollup_reports_per_root_and_process_totals() {
        let executor = Executor::with_config(crate::executor::ExecutorConfig {
            pool_size: 1,
            read_cap: 1,
            actor_cap: 1,
            heavy_permits: 1,
            drr_quantum: 1,
        });
        let dispatch_path_metrics = Arc::new(DispatchPathMetrics::new());
        let report = build_health_report(
            &executor,
            &HashMap::new(),
            &dispatch_path_metrics,
            &crate::context::App::default_shared(),
        );
        let metrics = report.metrics.expect("health metrics present");
        let memory = metrics.get("memory").expect("memory rollup present");
        // No actors registered: ready rollup with zero roots and process totals.
        assert_eq!(memory.get("status").and_then(Value::as_str), Some("ready"));
        assert_eq!(memory.get("roots_total").and_then(Value::as_u64), Some(0));
        assert!(memory.get("total_attributed_bytes").is_some());
        assert!(memory.get("rss_bytes").is_some());
        // Lifecycle audit counters ride the same probe so operator drill-downs
        // and fleet leak checks read one surface.
        let runtime = metrics.get("runtime").expect("runtime counters present");
        for key in ["live_watchers", "live_actor_roots", "open_routes"] {
            assert!(
                runtime.get(key).and_then(Value::as_u64).is_some(),
                "runtime.{key} must be a number"
            );
        }
    }

    #[test]
    fn health_snapshot_fast_fails_while_mutating_job_holds_component_lock() {
        let executor = Executor::with_config(crate::executor::ExecutorConfig {
            pool_size: 1,
            read_cap: 1,
            actor_cap: 1,
            heavy_permits: 1,
            drr_quantum: 1,
        });
        let (_dir, root) = test_root("health-mutating-lock");
        let ctx = test_ctx();
        executor.register_actor(root.clone(), Arc::clone(&ctx));
        let (started_tx, started_rx) = crossbeam_channel::bounded(1);
        let (release_tx, release_rx) = crossbeam_channel::bounded(1);
        let blocker = executor.submit(
            root,
            Lane::Mutating,
            "health-lock-blocker".to_string(),
            Box::new(move |ctx| {
                let _index = ctx
                    .search_index()
                    .write()
                    .unwrap_or_else(std::sync::PoisonError::into_inner);
                started_tx.send(()).expect("signal held health lock");
                release_rx.recv().expect("release held health lock");
                Response::success("health-lock-blocker", json!({ "ok": true }))
            }),
        );
        started_rx
            .recv_timeout(Duration::from_secs(1))
            .expect("mutating lock holder starts");

        let report = build_health_report(
            &executor,
            &HashMap::new(),
            &DispatchPathMetrics::new(),
            &crate::context::App::default_shared(),
        );

        // Settle the holder before asserting so a verdict failure cannot drop the release
        // sender and make the worker panic while the test unwinds.
        release_tx.send(()).expect("release held health lock");
        blocker
            .recv_timeout(Duration::from_secs(3))
            .expect("mutating lock holder completes");

        assert_eq!(report.status, HealthStatus::Degraded);
    }
}