From 83079d57f124030f80aeb846c4d92dc7ee71613c Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Tue, 13 Oct 2026 22:03:38 -0700 Subject: [PATCH 1/2] dispatch watchdog: an instance is quarantined on its Nth breach, counted as bb_plugin_quarantined_total THE DESIGN A.8 (BUSBAR-1.6.0.md l.3689-3690): "on a breach of the threshold, N times, the instance is quarantined: new calls get FAULT (hooks go to on_error), plus a diagnostic and bb_plugin_quarantined_total". predev faulted the instance on the first breach with no count, no diagnostic and no counter (dispatch/watchdog.rs:4-13). - DispatchConfig::quarantine_after (N, clamped >= 1; DEFAULT_QUARANTINE_AFTER = 1 keeps the first-breach quarantine every caller has today). - every breach (a worker's hung crossing, or a hung ticket-less call, counted ONCE however many scans see it) is counted on the instance and warned with the plugin's name; below N the worker is replaced and the op is FAULT but the instance keeps serving; the Nth quarantines it: FAULT without a crossing, its connector streams purged, a diagnostic, and DispatchStats::quarantined (QUARANTINED_TOTAL = "bb_plugin_quarantined_total"). - the timed-probe exit (R2) is the hook door's existing trial (hook_door.rs, accepted difference Q133, owner-signed 2026-10-07); unchanged. RED: red_an_instance_is_quarantined_on_its_nth_breach_and_not_before (N = 2: one hung call is one breach across many scans, the instance serves after it, the second breach quarantines and counts); red_a_hung_ticketless_call_faults_its_instance now also asserts the counter. --- crates/plugin-loader/src/dispatch/mod.rs | 5 +- crates/plugin-loader/src/dispatch/plugin.rs | 10 +++ .../src/dispatch/tests/auth_outbound_tests.rs | 1 + .../src/dispatch/tests/auth_ticks_tests.rs | 1 + crates/plugin-loader/src/dispatch/watchdog.rs | 85 +++++++++++++++---- crates/plugin-loader/src/dispatch/worker.rs | 23 +++++ .../src/tests/auth_axis_tests.rs | 1 + .../src/tests/auth_conformance_tests.rs | 1 + .../src/tests/auth_door_tests.rs | 1 + .../tests/auth_verify_conformance_tests.rs | 1 + .../src/tests/conformance_suite_tests.rs | 1 + .../plugin-loader/src/tests/dispatch_tests.rs | 57 +++++++++++++ 12 files changed, 170 insertions(+), 17 deletions(-) diff --git a/crates/plugin-loader/src/dispatch/mod.rs b/crates/plugin-loader/src/dispatch/mod.rs index 2096b9af19..9207a03844 100644 --- a/crates/plugin-loader/src/dispatch/mod.rs +++ b/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, DispatchConfig, DispatchStats, Dispatcher, Done, Lent, Reply, + 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//Ops`. diff --git a/crates/plugin-loader/src/dispatch/plugin.rs b/crates/plugin-loader/src/dispatch/plugin.rs index b5070e1560..03632fdd62 100644 --- a/crates/plugin-loader/src/dispatch/plugin.rs +++ b/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 Plugin { 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), diff --git a/crates/plugin-loader/src/dispatch/tests/auth_outbound_tests.rs b/crates/plugin-loader/src/dispatch/tests/auth_outbound_tests.rs index 97e7cd7185..4df5c0b89e 100644 --- a/crates/plugin-loader/src/dispatch/tests/auth_outbound_tests.rs +++ b/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::( diff --git a/crates/plugin-loader/src/dispatch/tests/auth_ticks_tests.rs b/crates/plugin-loader/src/dispatch/tests/auth_ticks_tests.rs index ec66a1c6e4..aeb4a2f3a1 100644 --- a/crates/plugin-loader/src/dispatch/tests/auth_ticks_tests.rs +++ b/crates/plugin-loader/src/dispatch/tests/auth_ticks_tests.rs @@ -132,6 +132,7 @@ fn dispatcher() -> Arc { workers: 2, budgets: Budgets::default(), watchdog_period: Duration::from_millis(20), + ..DispatchConfig::default() })) } diff --git a/crates/plugin-loader/src/dispatch/watchdog.rs b/crates/plugin-loader/src/dispatch/watchdog.rs index a0acd4c18a..55cde4be26 100644 --- a/crates/plugin-loader/src/dispatch/watchdog.rs +++ b/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, 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 = 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, started: Instant) { - let (gone, gens, faulted) = { + let (gone, gens, breached, quarantined) = { let mut st = old.lock(); if st.dead { return; @@ -96,8 +147,8 @@ fn replace(pool: &Pool, i: usize, old: &Arc, 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 = st @@ -105,11 +156,13 @@ fn replace(pool: &Pool, i: usize, old: &Arc, started: Instant) { .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(); diff --git a/crates/plugin-loader/src/dispatch/worker.rs b/crates/plugin-loader/src/dispatch/worker.rs index 308a336178..770998949c 100644 --- a/crates/plugin-loader/src/dispatch/worker.rs +++ b/crates/plugin-loader/src/dispatch/worker.rs @@ -104,14 +104,29 @@ pub struct DispatchConfig { pub budgets: Budgets, /// How often the watchdog looks. pub watchdog_period: Duration, + /// THE QUARANTINE COUNT, N (THE DESIGN A.8): the breaches of the watch budget an instance + /// survives before it is quarantined. Every breach replaces its worker and settles the hung + /// op as FAULT; the Nth also quarantines the instance — every new call answers FAULT without a + /// crossing (a hook's caller goes to its `on_error`, and the hook door's timed trial brings a + /// fresh instance back, R2), with a diagnostic and `bb_plugin_quarantined_total`. Clamped to + /// at least 1; the default [`DEFAULT_QUARANTINE_AFTER`] quarantines on the first breach. + pub quarantine_after: u32, } +/// The default [`DispatchConfig::quarantine_after`]: the first breach quarantines, as before the +/// count existed. +pub const DEFAULT_QUARANTINE_AFTER: u32 = 1; + +/// The quarantine counter's exposition name (THE DESIGN A.8), labelled by plugin. +pub const QUARANTINED_TOTAL: &str = "bb_plugin_quarantined_total"; + impl Default for DispatchConfig { fn default() -> Self { Self { workers: 1, budgets: Budgets::default(), watchdog_period: Duration::from_millis(50), + quarantine_after: DEFAULT_QUARANTINE_AFTER, } } } @@ -123,6 +138,9 @@ pub struct DispatchStats { pub stale_wakes: u64, /// Workers the watchdog replaced. pub replacements: u64, + /// Instances the watchdog quarantined: `bb_plugin_quarantined_total` (THE DESIGN A.8), the + /// process's sum over every plugin. + pub quarantined: u64, /// WriteBehind ops that completed after their caller stopped waiting. pub write_behind_late: u64, /// Library unloads still running on their reapers, process-wide: a count that stays up names a @@ -141,6 +159,7 @@ pub struct DispatchStats { pub(crate) struct Stats { stale_wakes: AtomicU64, pub(crate) replacements: AtomicU64, + pub(crate) quarantined: AtomicU64, write_behind_late: AtomicU64, driver_kept_high: AtomicU64, deadline_without_wake: AtomicU64, @@ -149,6 +168,8 @@ pub(crate) struct Stats { /// What a worker thread needs besides its worker. pub(crate) struct Env { pub(crate) budgets: Budgets, + /// [`DispatchConfig::quarantine_after`], at least 1. + pub(crate) quarantine_after: u32, pub(crate) stats: Arc, /// The host services' stored results. pub(crate) services: Arc, @@ -1430,6 +1451,7 @@ impl Dispatcher { let n = config.workers.clamp(1, MAX_WORKERS - 1); let env = Arc::new(Env { budgets: config.budgets, + quarantine_after: config.quarantine_after.max(1), stats: Arc::default(), services: Arc::default(), provider, @@ -1477,6 +1499,7 @@ impl Dispatcher { DispatchStats { stale_wakes: s.stale_wakes.load(Ordering::Relaxed), replacements: s.replacements.load(Ordering::Relaxed), + quarantined: s.quarantined.load(Ordering::Relaxed), write_behind_late: s.write_behind_late.load(Ordering::Relaxed), live_reapers: super::load::live_reapers(), driver_kept_high: s.driver_kept_high.load(Ordering::Relaxed), diff --git a/crates/plugin-loader/src/tests/auth_axis_tests.rs b/crates/plugin-loader/src/tests/auth_axis_tests.rs index fb87fb3a9c..1442893599 100644 --- a/crates/plugin-loader/src/tests/auth_axis_tests.rs +++ b/crates/plugin-loader/src/tests/auth_axis_tests.rs @@ -15,6 +15,7 @@ fn empty() -> AuthRows { workers: 1, budgets: Budgets::default(), watchdog_period: Duration::from_millis(20), + ..DispatchConfig::default() })); AuthRows::new(Arc::new(PluginRegistry::empty()), d) } diff --git a/crates/plugin-loader/src/tests/auth_conformance_tests.rs b/crates/plugin-loader/src/tests/auth_conformance_tests.rs index f26d159406..efd21f9dea 100644 --- a/crates/plugin-loader/src/tests/auth_conformance_tests.rs +++ b/crates/plugin-loader/src/tests/auth_conformance_tests.rs @@ -90,6 +90,7 @@ fn dispatcher() -> Arc { workers: 2, budgets: Budgets::default(), watchdog_period: Duration::from_millis(20), + ..DispatchConfig::default() })) } diff --git a/crates/plugin-loader/src/tests/auth_door_tests.rs b/crates/plugin-loader/src/tests/auth_door_tests.rs index 3928fbde51..25e88c5cae 100644 --- a/crates/plugin-loader/src/tests/auth_door_tests.rs +++ b/crates/plugin-loader/src/tests/auth_door_tests.rs @@ -150,6 +150,7 @@ fn dispatcher() -> Arc { workers: 2, budgets: Budgets::default(), watchdog_period: Duration::from_millis(20), + ..DispatchConfig::default() })) } diff --git a/crates/plugin-loader/src/tests/auth_verify_conformance_tests.rs b/crates/plugin-loader/src/tests/auth_verify_conformance_tests.rs index 87034c3c09..8564bf6792 100644 --- a/crates/plugin-loader/src/tests/auth_verify_conformance_tests.rs +++ b/crates/plugin-loader/src/tests/auth_verify_conformance_tests.rs @@ -79,6 +79,7 @@ fn dispatcher() -> Arc { workers: 2, budgets: Budgets::default(), watchdog_period: Duration::from_millis(20), + ..DispatchConfig::default() })) } diff --git a/crates/plugin-loader/src/tests/conformance_suite_tests.rs b/crates/plugin-loader/src/tests/conformance_suite_tests.rs index 0a64cd9b63..f74e0a6c4f 100644 --- a/crates/plugin-loader/src/tests/conformance_suite_tests.rs +++ b/crates/plugin-loader/src/tests/conformance_suite_tests.rs @@ -911,6 +911,7 @@ mod lent_ops { ..Budgets::default() }, watchdog_period: Duration::from_millis(20), + ..DispatchConfig::default() }); let row = LinkedRow::of(plug::busbar_plugin_door).expect("the test door states itself"); let p = load_linked::(&row, bind(&d, "abandoned")).expect("it loads"); diff --git a/crates/plugin-loader/src/tests/dispatch_tests.rs b/crates/plugin-loader/src/tests/dispatch_tests.rs index 52626290dc..da91494cbf 100644 --- a/crates/plugin-loader/src/tests/dispatch_tests.rs +++ b/crates/plugin-loader/src/tests/dispatch_tests.rs @@ -290,6 +290,7 @@ fn config() -> DispatchConfig { ..Budgets::default() }, watchdog_period: Duration::from_millis(20), + ..DispatchConfig::default() } } @@ -1871,6 +1872,11 @@ fn red_a_hung_ticketless_call_faults_its_instance() { std::thread::spawn(move || hung.call(TICK, &mut frame(b"hang:ticketless")).outcome); until("the watchdog faults the instance", || p.is_faulted()); assert!(!p2.is_faulted(), "only the hung instance"); + assert_eq!( + d.stats().quarantined, + 1, + "bb_plugin_quarantined_total counts the quarantine (the default N is the first breach)" + ); assert_eq!(p.call(TICK, &mut frame(answer(1))).outcome, Outcome::Fault); assert_eq!( p2.call(TICK, &mut frame(b"unhang:ticketless")).outcome, @@ -1879,6 +1885,57 @@ fn red_a_hung_ticketless_call_faults_its_instance() { caller.join().unwrap(); } +/// THE DESIGN A.8: "on a breach of the threshold, N times, the instance is quarantined". With +/// N = 2 the first breach is counted ONCE (however many scans see the call hung), the instance +/// keeps serving, and nothing is quarantined; the second breach quarantines it, counts +/// `bb_plugin_quarantined_total`, and every new call is FAULT without a crossing. +#[test] +fn red_an_instance_is_quarantined_on_its_nth_breach_and_not_before() { + let d = Dispatcher::new(DispatchConfig { + quarantine_after: 2, + ..config() + }); + let (p, _sink) = opened(&d); + let (p2, _s2) = opened(&d); + let breaches = || p.inner.breaches.load(std::sync::atomic::Ordering::SeqCst); + + let hung = p.clone(); + let first = + std::thread::spawn(move || hung.call(TICK, &mut frame(b"hang:quarantine-n-1")).outcome); + until("the watchdog counts the first breach", || breaches() == 1); + // Several scans pass while the call stays hung; it is still one breach. + std::thread::sleep(Duration::from_millis(120)); + assert_eq!(breaches(), 1, "a hung call is one breach, not one per scan"); + assert!(!p.is_faulted(), "one breach of two quarantines nothing"); + assert_eq!(d.stats().quarantined, 0); + assert_eq!( + p2.call(TICK, &mut frame(b"unhang:quarantine-n-1")).outcome, + Outcome::Ready + ); + first.join().unwrap(); + assert_eq!( + p.call(TICK, &mut frame(answer(1))).outcome, + Outcome::Ready, + "below N the instance keeps serving" + ); + + let hung = p.clone(); + let second = + std::thread::spawn(move || hung.call(TICK, &mut frame(b"hang:quarantine-n-2")).outcome); + until("the second breach quarantines the instance", || { + p.is_faulted() + }); + assert_eq!(breaches(), 2); + assert_eq!(d.stats().quarantined, 1, "bb_plugin_quarantined_total"); + assert_eq!(p.call(TICK, &mut frame(answer(1))).outcome, Outcome::Fault); + assert!(!p2.is_faulted(), "only the breaching instance"); + assert_eq!( + p2.call(TICK, &mut frame(b"unhang:quarantine-n-2")).outcome, + Outcome::Ready + ); + second.join().unwrap(); +} + #[test] fn cancel_carries_the_disposition_and_a_fault_on_cancel_is_fault() { let d = Dispatcher::new(config()); From 05c4f1820999ab3caa58966faafdccb3c16ae863 Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Tue, 13 Oct 2026 22:41:21 -0700 Subject: [PATCH 2/2] /metrics: the dispatcher's bb_* counters are exported, one HELP/TYPE family each Coordinator ruling 2026-10-14: every bb_* counter DispatchStats holds is exported through the existing metrics path. DispatchStats::families() is the one list (bb_deadline_without_wake_total, THE DESIGN A.4.3; bb_plugin_quarantined_total, A.8). The composition root hands the kernel a reader through RootInstall::host_counters; the kernel's whole-snapshot read describes each family and sets its cumulative value before the scrape sink renders it. Additive 1.6.0 surface: docs/observability.md rows and a CHANGELOG Added line. RED: the_dispatchers_bb_counters_are_scraped_with_help_and_type (the real binary's /metrics). --- CHANGELOG.md | 5 ++ crates/busbar-kernel/src/export/scrape.rs | 5 +- crates/busbar-kernel/src/preflight.rs | 8 +++ crates/busbar-kernel/src/snapshot/mod.rs | 17 +++++ crates/busbar/src/root/dispatch.rs | 11 +++ crates/busbar/src/root/linked.rs | 1 + .../tests/metrics_scrape_boot_window.rs | 67 +++++++++++++++++++ crates/plugin-loader/src/dispatch/mod.rs | 4 +- crates/plugin-loader/src/dispatch/worker.rs | 33 ++++++++- docs/observability.md | 2 + 10 files changed, 148 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 75af599f25..ae2f7dbb62 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/crates/busbar-kernel/src/export/scrape.rs b/crates/busbar-kernel/src/export/scrape.rs index c89234bc32..0389fef620 100644 --- a/crates/busbar-kernel/src/export/scrape.rs +++ b/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) diff --git a/crates/busbar-kernel/src/preflight.rs b/crates/busbar-kernel/src/preflight.rs index 81932514c2..d9cffa411e 100644 --- a/crates/busbar-kernel/src/preflight.rs +++ b/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, } +/// 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, diff --git a/crates/busbar-kernel/src/snapshot/mod.rs b/crates/busbar-kernel/src/snapshot/mod.rs index dad65af60c..dfc3895127 100644 --- a/crates/busbar-kernel/src/snapshot/mod.rs +++ b/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. diff --git a/crates/busbar/src/root/dispatch.rs b/crates/busbar/src/root/dispatch.rs index 4e6b10242d..717c7447c4 100644 --- a/crates/busbar/src/root/dispatch.rs +++ b/crates/busbar/src/root/dispatch.rs @@ -42,6 +42,17 @@ pub fn build(workers: usize, services: Arc) -> 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] diff --git a/crates/busbar/src/root/linked.rs b/crates/busbar/src/root/linked.rs index 0a11788b11..a741b6a467 100644 --- a/crates/busbar/src/root/linked.rs +++ b/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}"); diff --git a/crates/busbar/tests/metrics_scrape_boot_window.rs b/crates/busbar/tests/metrics_scrape_boot_window.rs index cec3c26a6a..a350099cdf 100644 --- a/crates/busbar/tests/metrics_scrape_boot_window.rs +++ b/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::().is_ok())), + "{name}: no sample; exposition:\n{}", + scrape.body + ); + } + let _ = std::fs::remove_dir_all(&dir); +} + /// Extract the set of `# HELP ...` 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 { diff --git a/crates/plugin-loader/src/dispatch/mod.rs b/crates/plugin-loader/src/dispatch/mod.rs index 9207a03844..eda18b8d8d 100644 --- a/crates/plugin-loader/src/dispatch/mod.rs +++ b/crates/plugin-loader/src/dispatch/mod.rs @@ -93,8 +93,8 @@ 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, - DEFAULT_QUARANTINE_AFTER, QUARANTINED_TOTAL, + 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 diff --git a/crates/plugin-loader/src/dispatch/worker.rs b/crates/plugin-loader/src/dispatch/worker.rs index 770998949c..4a3991b01c 100644 --- a/crates/plugin-loader/src/dispatch/worker.rs +++ b/crates/plugin-loader/src/dispatch/worker.rs @@ -117,9 +117,14 @@ pub struct DispatchConfig { /// count existed. pub const DEFAULT_QUARANTINE_AFTER: u32 = 1; -/// The quarantine counter's exposition name (THE DESIGN A.8), labelled by plugin. +/// The quarantine counter's exposition name (THE DESIGN A.8): [`DispatchStats::quarantined`], the +/// process's sum over every plugin. pub const QUARANTINED_TOTAL: &str = "bb_plugin_quarantined_total"; +/// The lost-wake counter's exposition name (THE DESIGN A.4.3): +/// [`DispatchStats::deadline_without_wake`]. +pub const DEADLINE_WITHOUT_WAKE_TOTAL: &str = "bb_deadline_without_wake_total"; + impl Default for DispatchConfig { fn default() -> Self { Self { @@ -155,6 +160,32 @@ pub struct DispatchStats { pub deadline_without_wake: u64, } +/// One `bb_*` counter family the dispatcher exposes: its name, its HELP text and its value. +pub type CounterFamily = (&'static str, &'static str, u64); + +impl DispatchStats { + /// EVERY `bb_*` COUNTER these stats hold, one exposition family each (coordinator ruling + /// 2026-10-14: the dispatcher's `bb_*` counters are exported through the process's metrics + /// path, HELP and TYPE included). The one list the composition root hands the kernel's scrape. + #[must_use] + pub const fn families(&self) -> [CounterFamily; 2] { + [ + ( + 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", + self.deadline_without_wake, + ), + ( + QUARANTINED_TOTAL, + "Plugin instances the crossing watchdog quarantined for overrunning their call \ + budget, summed over every plugin", + self.quarantined, + ), + ] + } +} + #[derive(Debug, Default)] pub(crate) struct Stats { stale_wakes: AtomicU64, diff --git a/docs/observability.md b/docs/observability.md index 02853592d5..601f892a5c 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -156,6 +156,8 @@ deliberate, and it is the signal an operator most often has nothing else to see. | `busbar_journal_quarantined_total` | counter | none | Journal segments whose damaged remainder boot recovery moved into a quarantine file because a record failed its checksum with valid records after it. Incremented once per quarantine, at boot. Any non-zero value is an incident: acknowledged postings and holds are off the book, so this node's figures read lower than what it served. See [Journal corruption at boot](operations.md#journal-corruption-at-boot). A torn tail after a crash is not counted. | | `busbar_webhook_logs_dropped_total` | counter | none | A request-log webhook delivery shed because the bounded delivery pool was saturated (the endpoint is slow or unreachable). Global backpressure. A non-zero rate means logs are being dropped silently. | | `busbar_file_logs_dropped_total` | counter | none | A request-log FILE append shed because that sink's bounded in-flight append pool was saturated (a slow or stalled filesystem: full disk, hung mount). The file-sink counterpart of the webhook counter above. A non-zero rate means request-log lines are being dropped. | +| `bb_deadline_without_wake_total` | counter | none | New in 1.6.0. Plugin operations that waited for a wake and were settled by their deadline without ever being woken, summed over every plugin. It stays at 0 while every plugin wakes what it parks; a rising value names a plugin that loses wakes, so its calls end at their timeout instead of completing. | +| `bb_plugin_quarantined_total` | counter | none | New in 1.6.0. Plugin instances the crossing watchdog quarantined because a call overran its budget, summed over every plugin. A quarantined instance answers its callers with a fault (a hook goes to its `on_error`) until a timed trial call brings a fresh instance back. Any non-zero value names a wedged plugin. | **Mint labels.** Key labels attached at mint (`labels: {"team": "growth"}`) are echoed verbatim onto that key's gauge series, so Grafana can `sum by (team)` and Alertmanager can fire per team without Busbar knowing what a team is. Label keys are operator-chosen at mint (admin-plane bounded), never request bytes.