/metrics: the dispatcher's bb_* counters, one HELP/TYPE family each (stacked on #952)

CHANGELOG.md

@@ -721,6 +721,11 @@ each dialect translates, field by field, is listed in the generated
 
 ### Added
 
+- **`/metrics` carries two plugin-health counters.** `bb_deadline_without_wake_total` counts plugin
+  operations settled by their deadline without ever being woken, and `bb_plugin_quarantined_total`
+  counts plugin instances the crossing watchdog quarantined. Each is one counter family with its
+  `# HELP` and `# TYPE` lines, summed over every plugin, and is 0 on a healthy node. Nothing that
+  1.5.5 exposed changed. See [the observability guide](docs/observability.md).
 - **A plugin's inbound listener binds what its settings block states, capped at 1024 connections.**
   A need a plugin declares inbound names a settings block `{listen, tls: {cert, key, client_ca?},
   max_conns?}` (1.5.5's `listen` / `tls` shape; `cert` and `key` are secret references). The listener

crates/busbar-kernel/src/export/scrape.rs

@@ -101,12 +101,13 @@ pub(crate) fn read(scope: u32) -> Snapshot {
     answer
 }
 
-/// The WHOLE snapshot: with the recorder installed and a live `app`, its scrape-time gauges are
-/// refreshed and every export sink's `status` folded first; NOT READY before the recorder is.
+/// The WHOLE snapshot: with the recorder installed and a live `app`, its scrape-time gauges and the
+/// host's own counters are refreshed and every export sink's `status` folded first; NOT READY before the recorder is.
 fn whole(app: Option<&App>) -> Snapshot {
     let installed = crate::snapshot::recorder_installed();
     if let Some(app) = app.filter(|_| installed) {
         crate::snapshot::refresh_scrape_gauges(app);
+        crate::snapshot::refresh_host_counters();
         super::plugin::status();
     }
     crate::snapshot::snapshot().map_or(Snapshot::NotReady, Snapshot::Families)

crates/busbar-kernel/src/preflight.rs

@@ -133,8 +133,15 @@ pub struct RootInstall {
     /// the process's one dispatcher (ARCHITECT ruling 2026-09-29, the opener seam). `None` = no
     /// export module resolves.
     pub export_axis: Option<&'static dyn busbar_contract::export_calls::ExportAxis>,
+    /// The host's own counters every `/metrics` scrape exposes, one described family each: the
+    /// root's dispatcher's `bb_*` counters, read when the scrape is (`None` = none).
+    pub host_counters: Option<HostCounters>,
 }
 
+/// THE HOST'S OWN COUNTERS: each counter's family name, its HELP text and its cumulative value,
+/// read at scrape time ([`RootInstall::host_counters`]).
+pub type HostCounters = fn() -> Vec<(&'static str, &'static str, u64)>;
+
 /// THE ROOT'S HOOK AXIS over one plugin registry (each configuration's registry gets its own).
 ///
 /// # Errors
@@ -157,6 +164,7 @@ const STAND_IN: RootInstall = RootInstall {
     store_axis: Some(crate::test_support::store_axis_stand_in),
     secret_axis: Some(&crate::test_support::SecretsStandIn),
     export_axis: Some(&crate::test_support::export_axis::STAND_IN),
+    host_counters: None,
 };
 
 /// The hook doors a test build links in place of the root's (the stand-in hook axis,

crates/busbar-kernel/src/snapshot/mod.rs

@@ -357,6 +357,23 @@ pub fn refresh_scrape_gauges(app: &App) {
     }
 }
 
+/// THE HOST'S OWN COUNTERS onto the recorder at scrape time: what the composition root installed
+/// ([`crate::preflight::RootInstall::host_counters`]), none when it installed nothing.
+pub fn refresh_host_counters() {
+    if let Some(read) = crate::preflight::root_rows().host_counters {
+        emit_host_counters(&read());
+    }
+}
+
+/// Each of `counters` as one described counter family, set to its cumulative value: the host
+/// holds the count, so a scrape states it ([`metrics::Counter::absolute`]) and never adds to it.
+pub(crate) fn emit_host_counters(counters: &[(&'static str, &'static str, u64)]) {
+    for &(name, help, value) in counters {
+        metrics::describe_counter!(name, help);
+        metrics::counter!(name).absolute(value);
+    }
+}
+
 /// `busbar_lane_state` for ONE breaker cell, read from that cell alone — never folded with any
 /// other pool's cell on the same lane:
 /// * `1` — the cell is HalfOpen: a recovery probe is in flight on THIS cell right now.

crates/busbar/src/root/dispatch.rs

@@ -42,6 +42,17 @@ pub fn build(workers: usize, services: Arc<dyn HostServices>) -> Dispatcher {
     Dispatcher::with_services(config(workers), services)
 }
 
+/// THE DISPATCHER'S `bb_*` COUNTERS for the kernel's `/metrics` scrape
+/// (`busbar_kernel::preflight::RootInstall::host_counters`): every family its stats hold
+/// (`DispatchStats::families`), none before the dispatcher is built — a scrape never builds one.
+#[must_use]
+pub fn host_counters() -> Vec<(&'static str, &'static str, u64)> {
+    DISPATCHER
+        .get()
+        .map(|d| d.stats().families().to_vec())
+        .unwrap_or_default()
+}
+
 /// The process's dispatcher (built at boot; built with one worker and no installed services on
 /// first use where no boot ran).
 #[must_use]

crates/busbar/src/root/linked.rs

@@ -355,6 +355,7 @@ pub fn register_stores(linked: &Linked) {
             store_axis: Some(store_axis),
             secret_axis: Some(secrets),
             export_axis: Some(&crate::root::exports::EXPORTS),
+            host_counters: Some(crate::root::dispatch::host_counters),
         }),
         Err(refusal) => {
             eprintln!("busbar: {refusal}");

crates/busbar/tests/metrics_scrape_boot_window.rs

@@ -233,6 +233,73 @@ fn metrics_scrape_is_never_a_200_with_an_empty_body() {
     let _ = std::fs::remove_dir_all(&dir);
 }
 
+/// RED (coordinator ruling 2026-10-14, THE DESIGN A.4.3 / A.8): the process dispatcher's `bb_*`
+/// counters are on the real binary's `/metrics`, one family each with its `# HELP` and
+/// `# TYPE … counter` lines and a sample — `bb_deadline_without_wake_total` and
+/// `bb_plugin_quarantined_total`. Before the composition root handed the scrape its dispatcher's
+/// counters (`RootInstall::host_counters`) neither family was exported at all.
+#[test]
+fn the_dispatchers_bb_counters_are_scraped_with_help_and_type() {
+    let dir = fixture_dir();
+    let data_port = free_port();
+    let admin_port = free_port();
+    write_configs(&dir, data_port, admin_port);
+    let log_path = dir.join("out.log");
+    let log = std::fs::File::create(&log_path).unwrap();
+    let log_err = log.try_clone().unwrap();
+    let mut child = Command::new(common::boot::exe())
+        .env("BUSBAR_CONFIG", dir.join("config.yaml"))
+        .env("BUSBAR_PROVIDERS", dir.join("providers.yaml"))
+        .env("MOCK_KEY", "x")
+        .env("RUST_LOG", "warn")
+        .stdout(log)
+        .stderr(log_err)
+        .spawn()
+        .expect("spawn busbar");
+    let scrape = wait_for_full_scrape(data_port, Duration::from_secs(30));
+    let _ = child.kill();
+    let _ = child.wait();
+    let scrape = scrape.unwrap_or_else(|| {
+        panic!(
+            "busbar never answered a full /metrics exposition; log:\n{}",
+            read_to_string(&log_path)
+        )
+    });
+    for (name, help) in [
+        (
+            "bb_deadline_without_wake_total",
+            "Plugin operations that waited and were settled by their deadline without ever being \
+             woken; 0 while every plugin wakes what it parks",
+        ),
+        (
+            "bb_plugin_quarantined_total",
+            "Plugin instances the crossing watchdog quarantined for overrunning their call budget, \
+             summed over every plugin",
+        ),
+    ] {
+        let lines: Vec<&str> = scrape.body.lines().collect();
+        assert!(
+            lines.contains(&format!("# HELP {name} {help}").as_str()),
+            "{name}: no HELP line; exposition:\n{}",
+            scrape.body
+        );
+        assert!(
+            lines.contains(&format!("# TYPE {name} counter").as_str()),
+            "{name}: no TYPE counter line; exposition:\n{}",
+            scrape.body
+        );
+        assert!(
+            lines.iter().any(|l| l
+                .strip_prefix(name)
+                .and_then(|v| v.strip_prefix(' '))
+                .is_some_and(|v| v.parse::<u64>().is_ok())),
+            "{name}: no sample; exposition:\n{}",
+            scrape.body
+        );
+    }
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
 /// Extract the set of `# HELP <name> ...` metric names from a Prometheus exposition — the closed
 /// set `describe()` registers, independent of which series happen to have been emitted yet.
 fn help_names(body: &str) -> BTreeSet<String> {

crates/plugin-loader/src/dispatch/mod.rs

@@ -92,7 +92,10 @@ pub use plugin::{
 };
 pub use services::{HostServices, Later, Ran, Reading, Stored};
 pub use validate::Violation;
-pub use worker::{Adopter, Budgets, DispatchConfig, DispatchStats, Dispatcher, Done, Lent, Reply};
+pub use worker::{
+    Adopter, Budgets, CounterFamily, DispatchConfig, DispatchStats, Dispatcher, Done, Lent, Reply,
+    DEADLINE_WITHOUT_WAKE_TOTAL, DEFAULT_QUARANTINE_AFTER, QUARANTINED_TOTAL,
+};
 
 /// A kind, as the dispatcher sees it: its code, its table and its timeout outcome. Implemented once
 /// per kind ([`kinds`]); `Ops` is that kind's `abi/<kind>/Ops`.

crates/plugin-loader/src/dispatch/plugin.rs

@@ -370,6 +370,14 @@ pub(crate) struct Instance {
     /// (an `open` in flight that pended) is not open.
     opened: AtomicBool,
     pub(crate) faulted: AtomicBool,
+    /// The watchdog's breaches of this instance's watch budget so far; the
+    /// [`DispatchConfig::quarantine_after`](super::DispatchConfig::quarantine_after)th quarantines
+    /// it ([`Instance::faulted`]).
+    pub(crate) breaches: AtomicU32,
+    /// The ticket-less calls the watchdog has already counted as breaches: every call id below
+    /// this one (ids are minted in order, and a call stays hung on the list until it returns, so
+    /// each is counted once, not once per scan).
+    pub(crate) ticketless_counted: AtomicU64,
     /// `close` of an OPEN instance answered READY, or any `close` answered FAULT: every later op
     /// answers FAULT without a crossing. (Closing a half-open box frees the box; the instance never
     /// opened, so it is not closed.)
@@ -1459,6 +1467,8 @@ impl<K: Kind> Plugin<K> {
                 ptr: AtomicPtr::new(std::ptr::null_mut()),
                 opened: AtomicBool::new(false),
                 faulted: AtomicBool::new(false),
+                breaches: AtomicU32::new(0),
+                ticketless_counted: AtomicU64::new(0),
                 closed: AtomicBool::new(false),
                 gate: AtomicU32::new(0),
                 crossings: AtomicU64::new(0),

crates/plugin-loader/src/dispatch/tests/auth_outbound_tests.rs

@@ -246,6 +246,7 @@ fn opened() -> OutboundInstance {
             ..Budgets::default()
         },
         watchdog_period: Duration::from_millis(20),
+        ..DispatchConfig::default()
     }));
     let row = LinkedRow::of(plugin::door).expect("the test auth door states its Statement");
     let p = load_linked::<Auth>(

crates/plugin-loader/src/dispatch/tests/auth_ticks_tests.rs

@@ -132,6 +132,7 @@ fn dispatcher() -> Arc<Dispatcher> {
         workers: 2,
         budgets: Budgets::default(),
         watchdog_period: Duration::from_millis(20),
+        ..DispatchConfig::default()
     }))
 }
 

crates/plugin-loader/src/dispatch/watchdog.rs

@@ -2,10 +2,15 @@
 // Copyright (C) 2026 Busbar Inc and contributors
 
 //! THE CROSSING WATCHDOG, in every build. A crossing never blocks by contract; one that has not
-//! RETURNED within its class budget (`Budgets`) is a wedged plugin, and the watchdog:
+//! RETURNED within its class budget (`Budgets`) is a BREACH, a wedged plugin, and the watchdog:
 //!
-//! 1. marks the INSTANCE faulted — it is never called again; every later op on it, on any worker,
-//!    answers FAULT without a crossing (a pending op resumed or cancelled on another worker too);
+//! 1. counts the breach against the INSTANCE, and on its Nth (`DispatchConfig::quarantine_after`,
+//!    THE DESIGN A.8) QUARANTINES it: it is never called again; every later op on it, on any
+//!    worker, answers FAULT without a crossing (a pending op resumed or cancelled on another worker
+//!    too), its connector streams go, a diagnostic names it, and `bb_plugin_quarantined_total`
+//!    ([`super::DispatchStats::quarantined`]) counts it. Below N the instance keeps serving; the
+//!    breach's own op and worker are settled by 2 and 3 all the same. A hook comes back out of
+//!    quarantine through its door's timed trial (R2, `hook_door`);
 //! 2. marks the WORKER dead and settles EVERY ticket it held as FAULT — the hung op, every pending
 //!    and queued op of any instance on that worker, and its driver tickets (which stop driving);
 //!    their `max_inflight` units, drain counts and lifecycle exclusions are given back, their
@@ -23,9 +28,39 @@ use std::time::{Duration, Instant};
 
 use busbar_contract::abi::mechanism::call::DeadlineClass;
 
+use super::plugin::Instance;
 use super::ticket::recycled_generation;
 use super::worker::{spawn_worker, Pool, Worker};
 
+/// Count one breach against `inst`; quarantine it on the pool's Nth. `true` when the instance is
+/// quarantined now (by this breach or an earlier one).
+fn breach(pool: &Pool, inst: &Instance) -> bool {
+    let n = inst
+        .breaches
+        .fetch_add(1, Ordering::AcqRel)
+        .saturating_add(1);
+    let of = pool.env.quarantine_after;
+    if n < of {
+        tracing::warn!(
+            plugin = inst.name(),
+            breach = n,
+            quarantine_after = of,
+            "plugin crossing outlived its watch budget; its worker is replaced and the op is FAULT"
+        );
+        return inst.faulted.load(Ordering::Acquire);
+    }
+    if !inst.faulted.swap(true, Ordering::AcqRel) {
+        pool.env.stats.quarantined.fetch_add(1, Ordering::Relaxed);
+        tracing::warn!(
+            plugin = inst.name(),
+            breaches = n,
+            metric = super::worker::QUARANTINED_TOTAL,
+            "plugin quarantined: every new call answers FAULT without a crossing"
+        );
+    }
+    true
+}
+
 /// Start the watchdog of `pool`; it ends when the pool does.
 pub(crate) fn spawn(pool: Weak<Pool>, period: Duration) {
     std::thread::Builder::new()
@@ -71,12 +106,28 @@ fn ticketless(pool: &Pool) {
         .collect();
     drop(adopted);
     for inst in live {
-        let hung = inst.lock_calls().iter().any(|&(_, started, s)| {
-            started.elapsed() > pool.env.budgets.of(s, DeadlineClass::Call)
-        });
-        if hung {
-            inst.faulted.store(true, Ordering::Release);
-            // A faulted instance never crosses again: its connector streams go now.
+        // Each hung call is ONE breach, however many scans see it hung: only calls at or above
+        // the instance's counted mark are new.
+        let from = inst.ticketless_counted.load(Ordering::Acquire);
+        let fresh: Vec<u64> = inst
+            .lock_calls()
+            .iter()
+            .filter(|&&(id, started, s)| {
+                id >= from && started.elapsed() > pool.env.budgets.of(s, DeadlineClass::Call)
+            })
+            .map(|&(id, _, _)| id)
+            .collect();
+        let Some(&last) = fresh.iter().max() else {
+            continue;
+        };
+        inst.ticketless_counted
+            .fetch_max(last + 1, Ordering::AcqRel);
+        let mut quarantined = false;
+        for _ in &fresh {
+            quarantined = breach(pool, &inst);
+        }
+        if quarantined {
+            // A quarantined instance never crosses again: its connector streams go now.
             if let Some((id, table)) = inst.wake.conn.get() {
                 super::conn_services::purge(*id, table);
             }
@@ -87,7 +138,7 @@ fn ticketless(pool: &Pool) {
 /// Replace worker `i` if it is STILL inside the crossing that began at `started`: re-checked under
 /// the state lock and the crossing lock, so a crossing that returned meanwhile is never faulted.
 fn replace(pool: &Pool, i: usize, old: &Arc<Worker>, started: Instant) {
-    let (gone, gens, faulted) = {
+    let (gone, gens, breached, quarantined) = {
         let mut st = old.lock();
         if st.dead {
             return;
@@ -96,20 +147,22 @@ fn replace(pool: &Pool, i: usize, old: &Arc<Worker>, started: Instant) {
         let Some(record) = crossing.as_ref().filter(|c| c.started == started) else {
             return;
         };
-        record.instance.faulted.store(true, Ordering::Release);
-        let faulted = Arc::clone(&record.instance);
+        let quarantined = breach(pool, &record.instance);
+        let breached = Arc::clone(&record.instance);
         drop(crossing);
         st.dead = true;
         let gens: Vec<u32> = st
             .entries
             .iter()
             .map(|e| recycled_generation(e.generation))
             .collect();
-        (Worker::fault_all(&mut st), gens, faulted)
+        (Worker::fault_all(&mut st), gens, breached, quarantined)
     };
-    // A faulted instance never crosses again: its connector streams go now, outside the locks.
-    if let Some((id, table)) = faulted.wake.conn.get() {
-        super::conn_services::purge(*id, table);
+    // A quarantined instance never crosses again: its connector streams go now, outside the locks.
+    if quarantined {
+        if let Some((id, table)) = breached.wake.conn.get() {
+            super::conn_services::purge(*id, table);
+        }
     }
     pool.env.services.forget_worker(old.index);
     let conns: Vec<_> = gone.iter().flat_map(|e| e.conns.iter().copied()).collect();