shep 0.8.0

//! `shep dog bark`: the webhook-alert dog.
//!
//! [`sinks`] holds the webhook destinations and the delivery function;
//! [`rules`] decides which bus events and poll snapshots become a
//! [`rules::Firing`]. This module has [`BarkConfig`] and [`run_loop`],
//! which subscribes to the shepherd's bus and polls the flock.
//!
//! The bus is a `tokio::sync::broadcast`, so a lagging subscriber has
//! events dropped rather than queued, and load is when an alert matters
//! most. A dropped frame triggers an immediate poll, and [`rules::Rules`]'s
//! per-subject debounce is what lets an `Errored` seen by both routes fire
//! once.

pub mod rules;
pub mod sinks;

use core::fmt;
use core::future::Future;
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use serde::Deserialize;
use shep_client::dogs::DogConfig;
use shep_client::{LinkLost, RequestError};
use shep_core::barks::{self, SinkOutcome};
use shep_core::protocol::{BusEvent, ProcessInfo};
use shep_core::values::UpDuration;
use tokio::sync::Mutex;
use tokio::time::MissedTickBehavior;

use self::rules::{Firing, Rule, Rules};
use self::sinks::Sink;
use crate::exit::ExitCode;

/// `[dog.bark]`.
///
/// `deny_unknown_fields`: a misspelled key must be a startup error naming
/// it, the same reasoning [`super::metrics::MetricsConfig`] gives for its
/// own section.
#[derive(Debug, Clone, PartialEq, Deserialize, schemars::JsonSchema, DogConfig)]
#[serde(deny_unknown_fields, default)]
pub struct BarkConfig {
    /// Named sinks, `[dog.bark.sinks]`.
    ///
    /// Marked whole, not per URL, and the difference is where the mark
    /// lands. `#[shep(secret)]` names a field of THIS type, and the URL is
    /// a field of [`Sink`] one level down, so a mark there is absent from
    /// the schema shep generates for [`BarkConfig`]: every sink would go
    /// out with its bearer token as an ordinary string. Marking the map
    /// says less than marking each URL would, and it says it about the
    /// type shep actually asks. Over-redacting is the safe direction.
    #[shep(secret)]
    pub sinks: BTreeMap<String, Sink>,
    /// Named rules, `[[dog.bark.rules]]`. Empty means
    /// [`Rules::default_rules`].
    pub rules: Vec<Rule>,
    /// How often the reconciliation poll runs when nothing has gone wrong.
    pub poll: UpDuration,
    /// Cap on `barks.jsonl`.
    pub history_bytes: u64,
    /// Per-delivery timeout.
    pub sink_timeout: UpDuration,
}

/// Hand-written: `#[serde(default)]` on the struct needs a `Default`, and
/// a derived one would give `poll`, `history_bytes` and `sink_timeout`
/// their types' zero values.
impl Default for BarkConfig {
    fn default() -> Self {
        Self {
            sinks: BTreeMap::new(),
            rules: Vec::new(),
            // 30s: the fallback cadence for when nothing has gone wrong.
            // A drop already triggers an immediate poll, so this bounds
            // steady-state cost, not responsiveness.
            poll: UpDuration::from_millis(30_000),
            // The cap `shep-daemon`'s own writer uses: one shared number
            // for the one file both append to.
            history_bytes: barks::DEFAULT_MAX_BYTES,
            // 10s: well past how fast Discord and Slack answer, well short
            // of the poll cadence above, so one stuck sink cannot absorb a
            // whole interval's deliveries.
            sink_timeout: UpDuration::from_millis(10_000),
        }
    }
}

/// One source of bus events: a frame, or a notice that frames were lost.
///
/// A trait rather than a concrete `EventStream`, so a test can drive this
/// loop from a real `tokio::sync::broadcast::Receiver` with a small
/// capacity and make the bus genuinely drop events.
pub trait EventSource: Send {
    /// The next event; `Err(count)` when the source dropped `count` frames
    /// before this one; `None` when it ends.
    fn next(&mut self) -> impl Future<Output = Option<Result<BusEvent, u64>>> + Send;

    /// Arms a fresh source against whatever shepherd is answering now,
    /// waiting a bounded time for one to be.
    ///
    /// A subscription belongs to one connection, so a handover ends it:
    /// the shepherd execs a successor on purpose and every dog is meant to
    /// cross that without restarting. What [`run_loop`] calls when
    /// [`Self::next`] returns `None`.
    ///
    /// # Errors
    /// [`Resubscribe`], which the dog exits on either way. The two arms
    /// exit differently, because a shepherd that never answered and one
    /// that answered and refused send an operator to different places.
    fn resubscribe(&mut self) -> impl Future<Output = Result<(), Resubscribe>> + Send;
}

/// Why bark could not arm a fresh subscription.
///
/// Two outcomes rather than one, so a re-subscribe reports what the first
/// subscription would have. `run_bark` exits `ExitCode::from(&err)` when
/// the shepherd refuses the opening `Subscribe`; without this a refusal
/// after a handover would exit `DaemonUnreachable` instead, naming a
/// shepherd that is running and answering.
#[derive(Debug)]
#[must_use = "which of the two it was decides how the dog exits"]
pub enum Resubscribe {
    /// No shepherd answered inside the dog's budget, or one refused this
    /// dog's protocol version at the handshake.
    Lost(LinkLost),
    /// The `Subscribe` did not succeed and waiting cannot change that.
    /// Carried whole, since `RequestError` already decides an exit code and
    /// flattening it would lose that.
    ///
    /// All four of `RequestError`'s non-`Closed` conditions arrive here,
    /// and they are not one fault:
    ///
    /// - `Rpc` is a shepherd answering and saying no.
    /// - `Wire` and `Undecodable` are the transport failing, so pointing an
    ///   operator at the shepherd's configuration sends them to the wrong
    ///   place.
    /// - `Timeout` is nobody answering in time, which means the request may
    ///   never have reached a shepherd at all.
    ///
    /// Named for the request rather than for a refusal, matching
    /// [`super::DogRunError::Request`], because only one of the four is a
    /// refusal.
    ///
    /// `Timeout` cannot arrive here today: `ClientEvents::resubscribe`
    /// bounds each attempt by what is left of `SHEPHERD_RETURN_BUDGET`,
    /// five seconds, and `Client::subscribe` carries seven, so the outer
    /// bound always fires first and reports a spent budget instead. That is
    /// a consequence of the two numbers rather than a guarantee, so the
    /// variant documents the condition rather than relying on it.
    Request(RequestError),
}

impl fmt::Display for Resubscribe {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Lost(lost) => lost.fmt(f),
            Self::Request(err) => write!(f, "the subscription request failed: {err}"),
        }
    }
}

impl core::error::Error for Resubscribe {
    /// Both arms wrap an error rather than describing one, so a structured
    /// logger walking the chain reaches the connection or RPC failure
    /// underneath instead of stopping here.
    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
        match self {
            Self::Lost(lost) => Some(lost),
            Self::Request(err) => Some(err),
        }
    }
}

/// What bark reads the flock through, so the loop's poll is drivable
/// without a socket.
///
/// `Sync`, not just `Send`: [`run_loop`]'s future holds `&F` across the
/// `.await` in [`reconcile`], so both its lag arm and its interval arm
/// poll the same source without moving it.
pub trait FlockSource: Send + Sync {
    /// The flock as it stands.
    ///
    /// # Errors
    /// Whatever the source failed with: in production, whatever
    /// `Request::ListFlock` failed with.
    fn flock(&self) -> impl Future<Output = Result<Vec<ProcessInfo>, RequestError>> + Send;
}

/// What bark re-asks its own `[bark]` section through, so a config change
/// is drivable without a socket.
///
/// `BusEvent::DogConfigChanged` says nothing about what changed, so the
/// frame is only a prompt: the answer is one `Request::DogConfig`.
///
/// `Sync` for the reason [`FlockSource`] is: [`run_loop`]'s future holds
/// a `&C` across the `.await` in [`reloaded_config`].
pub trait ConfigSource: Send + Sync {
    /// This dog's `[bark]` section as it stands now, empty when the file
    /// has no such section.
    ///
    /// # Errors
    /// Whatever the source failed with: in production, whatever
    /// `Request::DogConfig` failed with.
    fn section(&self) -> impl Future<Output = Result<String, RequestError>> + Send;
}

/// The rule set a `[bark]` section means: its own rules, or
/// [`rules::Rules::default_rules`] when it configured none.
///
/// Asked at startup and again on every `config.dog.bark` frame, so a
/// reloading bark cannot get different defaults from a starting one.
///
/// # Errors
/// - [`rules::RulesError`] as [`rules::Rules::new`]: a rule routing to a
///   sink the section does not define, an unknown event kind, or a sink
///   url that cannot work (an insecure webhook scheme, or credentials
///   before the host).
pub fn rules_for(config: &BarkConfig) -> Result<Rules, rules::RulesError> {
    let rule_list = if config.rules.is_empty() {
        Rules::default_rules(&config.sinks)
    } else {
        config.rules.clone()
    };
    Rules::new(rule_list, &config.sinks)
}

/// Everything a delivery needs, so the five values that travel together
/// through [`reconcile`], [`spawn_firings`] and [`deliver_and_record`]
/// travel as one.
///
/// `Clone` is what [`spawn_firings`] hands each spawned task: three
/// [`Arc`] bumps and two copies, the same clones it used to make one by
/// one. A config reload rebinds `sinks`, `sink_timeout` and `max_bytes`
/// in place; `append_lock` and `barks_path` outlive every reload.
///
/// `Debug` is safe despite `sinks` holding webhook URLs, which are bearer
/// credentials: [`Sink`]'s own `Debug` redacts them.
#[derive(Debug, Clone)]
struct Delivery {
    /// Every configured sink by name, for resolving a firing's own list.
    sinks: Arc<BTreeMap<String, Sink>>,
    /// Serializes this process's appends to `barks_path`.
    append_lock: Arc<Mutex<()>>,
    /// Where the bark trail is written.
    barks_path: Arc<PathBuf>,
    /// How long one sink delivery may take.
    sink_timeout: Duration,
    /// The trail's size ceiling.
    max_bytes: u64,
}

/// The poll timer for `period`.
///
/// `interval_at`, not `interval`: a plain `interval` fires its first tick
/// immediately, so the first poll would be attributable to the timer's
/// startup rather than to a drop or an elapsed interval.
///
/// One function, not two call sites: a reload that rebuilt the timer and
/// forgot `MissedTickBehavior::Delay` would leave a poll that ran long
/// firing a burst of catch-up ticks.
fn poll_timer(period: Duration) -> tokio::time::Interval {
    let mut interval = tokio::time::interval_at(tokio::time::Instant::now() + period, period);
    interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
    interval
}

/// Bark's loop: subscribe for speed, poll for correctness. Ends on
/// `SIGINT`/`SIGTERM` or when `events` does.
///
/// A dropped frame polls immediately, since the bus drops what a lagging
/// subscriber cannot keep up with. Firings are spawned, never awaited
/// inline, or a slow sink causes the drop this loop exists to catch.
/// Appends serialize behind an in-process [`tokio::sync::Mutex`]; the
/// cross-process race is `barks::append`'s own `flock(2)`.
///
/// A plain `fn`, not `async fn`: the returned future must borrow neither
/// `config` nor `barks_path`, so callers can spawn it.
pub fn run_loop<E: EventSource, F: FlockSource, C: ConfigSource>(
    events: E,
    flock: F,
    rules: Rules,
    config: &BarkConfig,
    barks_path: &Path,
    config_source: C,
) -> impl Future<Output = ExitCode> + Send + use<E, F, C> {
    let sinks = Arc::new(config.sinks.clone());
    let sink_timeout = config.sink_timeout.as_duration();
    let max_bytes = config.history_bytes;
    let mut poll_period = config.poll.as_duration();
    let barks_path = Arc::new(barks_path.to_path_buf());

    async move {
        let mut events = events;
        let mut rules = rules;
        let mut delivery = Delivery {
            sinks,
            append_lock: Arc::new(Mutex::new(())),
            barks_path,
            sink_timeout,
            max_bytes,
        };

        let mut sigterm = match crate::shutdown::Terminate::install() {
            Ok(sigterm) => sigterm,
            Err(err) => {
                eprintln!("shep dog bark: could not install a shutdown handler: {err}");
                return ExitCode::Failure;
            }
        };

        let mut poll_interval = poll_timer(poll_period);

        loop {
            tokio::select! {
                _ = tokio::signal::ctrl_c() => break ExitCode::Success,
                _ = sigterm.recv() => break ExitCode::Success,
                next = events.next() => {
                    match next {
                        // One connection generation ended. Usually the
                        // shepherd exec'd a successor, which is not the
                        // shepherd going away: `resubscribe` waits a
                        // bounded time for one to answer, and only a
                        // shepherd that never does ends this dog.
                        None => match events.resubscribe().await {
                            Ok(()) => {
                                // Reconcile whichever daemon answered.
                                // Frames sent with no subscription are
                                // gone, and the verdict decides nothing
                                // here: bark debounces on a sheep's name,
                                // never on an id the shepherd minted.
                                reconcile(&flock, &mut rules, &delivery).await;
                            }
                            Err(failed) => {
                                eprintln!("shep dog bark: {failed}");
                                break match &failed {
                                    Resubscribe::Lost(lost) => super::exit_for(lost),
                                    Resubscribe::Request(err) => ExitCode::from(err),
                                };
                            }
                        },
                        // Matched on the variant rather than on the dog's
                        // name: the subscription already narrows this to
                        // bark's own topic, `config.dog.<name>`.
                        Some(Ok(BusEvent::DogConfigChanged { .. })) => {
                            if let Some((next, next_rules)) = reloaded_config(&config_source).await {
                                // In place, never a restart: sinks and
                                // rules are pure data with no OS resource
                                // to rebind.
                                delivery.sinks = Arc::new(next.sinks.clone());
                                delivery.sink_timeout = next.sink_timeout.as_duration();
                                delivery.max_bytes = next.history_bytes;
                                // Rebuilt, which resets each rule's
                                // per-subject debounce: carrying it over
                                // a renumbered rule set would key state on
                                // an index that moved.
                                rules = next_rules;
                                if next.poll.as_duration() != poll_period {
                                    poll_period = next.poll.as_duration();
                                    poll_interval = poll_timer(poll_period);
                                }
                                eprintln!("shep dog bark: reloaded [bark] from dogs.toml");
                            }
                        }
                        Some(Ok(event)) => {
                            let firings = rules.on_event(&event, now_ms());
                            spawn_firings(firings, &delivery);
                        }
                        Some(Err(_dropped)) => {
                            // The drop says nothing about what was lost;
                            // only the shepherd can.
                            reconcile(&flock, &mut rules, &delivery).await;
                        }
                    }
                }
                _ = poll_interval.tick() => {
                    reconcile(&flock, &mut rules, &delivery).await;
                }
            }
        }
    }
}

/// Re-asks `source` for `[bark]` and rebuilds what a config change can
/// swap, or `None` when the answer cannot be used.
///
/// Every failure is reported and dropped, never propagated: a dog that
/// exited on a bad edit would stop alerting while config is being edited.
/// stderr gets the fact, never the section: `[bark]` carries webhook URLs
/// that are bearer credentials.
async fn reloaded_config<C: ConfigSource>(source: &C) -> Option<(BarkConfig, Rules)> {
    let section = match source.section().await {
        Ok(section) => section,
        Err(err) => {
            eprintln!("shep dog bark: could not re-read [bark] from the shepherd: {err}");
            return None;
        }
    };
    // Empty means the section is gone. The default no-sink rule is
    // rejected, so bark keeps the current configuration.
    let config = if section.is_empty() {
        BarkConfig::default()
    } else {
        match toml::from_str::<BarkConfig>(&section) {
            Ok(config) => config,
            Err(_err) => {
                eprintln!("shep dog bark: [bark] in dogs.toml does not parse; see `shep dogs`");
                return None;
            }
        }
    };
    match rules_for(&config) {
        Ok(rules) => Some((config, rules)),
        Err(err) => {
            eprintln!("shep dog bark: keeping the running rules; the new ones are refused: {err}");
            None
        }
    }
}

/// One reconciliation pass: ask `flock` what the flock looks like now, run
/// it through `rules::on_poll`, and spawn a delivery for anything that
/// fires. Shared by [`run_loop`]'s lag arm and interval arm, so the two
/// polls are one code path.
///
/// A failed poll is logged and dropped: the next bus event or interval
/// tick tries again.
async fn reconcile<F: FlockSource>(flock: &F, rules: &mut Rules, delivery: &Delivery) {
    match flock.flock().await {
        Ok(snapshot) => {
            let firings = rules.on_poll(&snapshot, now_ms());
            spawn_firings(firings, delivery);
        }
        Err(err) => eprintln!("shep dog bark: reconciliation poll failed: {err}"),
    }
}

/// Spawns one delivery task per firing, so [`run_loop`]'s own `select!`
/// returns to reading the next event immediately rather than waiting on any
/// of them.
fn spawn_firings(firings: Vec<Firing>, delivery: &Delivery) {
    for firing in firings {
        let delivery = delivery.clone();
        tokio::spawn(async move {
            deliver_and_record(firing, &delivery).await;
        });
    }
}

/// Delivers `firing` to each of its named sinks, then writes the resulting
/// [`shep_core::barks::Bark`] to `barks_path`.
///
/// After delivery, since a [`Firing`]'s [`shep_core::barks::Bark::sinks`]
/// is empty until each sink has been tried. Written even when every sink
/// refused it: the local trail is what an operator reads when the page
/// never arrived.
///
/// [`Delivery::append_lock`] covers only the [`barks::append`] call, a
/// read-modify-rename against one file that several of these run at once.
/// It does not replace `append`'s own cross-process `flock(2)`.
async fn deliver_and_record(firing: Firing, delivery: &Delivery) {
    let mut bark = firing.bark;
    let mut outcomes = Vec::with_capacity(firing.sinks.len());
    for name in &firing.sinks {
        let outcome = match delivery.sinks.get(name) {
            Some(sink) => match sinks::deliver(sink, &bark, delivery.sink_timeout).await {
                Ok(()) => SinkOutcome {
                    sink: name.clone(),
                    error: None,
                },
                Err(err) => SinkOutcome {
                    sink: name.clone(),
                    error: Some(err.to_string()),
                },
            },
            // Unreachable: `Rules::new` refuses a rule routing to a sink
            // `[dog.bark.sinks]` does not define. Recorded rather than
            // panicked on.
            None => SinkOutcome {
                sink: name.clone(),
                error: Some("sink not configured".to_owned()),
            },
        };
        outcomes.push(outcome);
    }
    bark.sinks = outcomes;

    let _guard = delivery.append_lock.lock().await;
    if let Err(err) = barks::append(&delivery.barks_path, &bark, delivery.max_bytes) {
        eprintln!("shep dog bark: could not record a fired bark: {err}");
    }
}

/// Wall-clock milliseconds since the Unix epoch.
///
/// [`Rules::on_event`] and [`Rules::on_poll`] take a caller-supplied
/// timestamp so a test can fix it; this is the production caller.
fn now_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::SystemTime::UNIX_EPOCH)
        .map(|elapsed| elapsed.as_millis() as u64)
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use std::net::SocketAddr;

    use shep_core::barks::Bark;
    use shep_core::protocol::{ProcessEventKind, RpcError, RpcErrorCode};
    use shep_core::status::ProcStatus;
    use tokio::sync::{broadcast, oneshot};

    use super::*;
    use crate::http::{HttpRequest, read_request, write_response};

    /// The sinks map carries the credential marker, and the rules beside
    /// it do not.
    ///
    /// Marked at the map rather than at each `Sink`'s `url`, because
    /// `#[shep(secret)]` names a field of the type being asked and the URL
    /// belongs to a type one level down. `Rule` is checked under `$defs`,
    /// the one place a marker could land on `Rule::sinks`.
    #[test]
    fn the_bark_schema_marks_the_sinks_map_and_leaves_the_rules_plain() {
        let schema = shep_client::dogs::config_schema::<BarkConfig>()
            .expect("`sinks` is a property of this type");
        let schema = schema.as_value();

        assert_eq!(
            schema.pointer("/properties/sinks/x-shep-secret"),
            Some(&serde_json::Value::Bool(true)),
            "every sink holds a webhook URL, which is a bearer credential"
        );
        assert_eq!(
            schema.pointer("/properties/rules/x-shep-secret"),
            None,
            "a rule names sinks and holds no credential of its own"
        );
        assert!(
            schema.pointer("/$defs/Rule/properties/sinks").is_some(),
            "the pointer below is only a check while `Rule` still has this \
             shape, so a rename that moved it must fail here first"
        );
        assert_eq!(
            schema.pointer("/$defs/Rule/properties/sinks/x-shep-secret"),
            None,
            "a rule's sinks are names, and a name is not a credential"
        );
    }

    /// [`EventSource`] over the real thing bark's subscription lags on: a
    /// `tokio::sync::broadcast::Receiver`. The production path implements
    /// the trait for [`shep_client::EventStream`] in `dog/mod.rs`.
    impl EventSource for broadcast::Receiver<BusEvent> {
        async fn next(&mut self) -> Option<Result<BusEvent, u64>> {
            match self.recv().await {
                Ok(event) => Some(Ok(event)),
                Err(broadcast::error::RecvError::Lagged(count)) => Some(Err(count)),
                Err(broadcast::error::RecvError::Closed) => None,
            }
        }

        /// A bare receiver has no shepherd behind it to ask again, so this
        /// stands for the shepherd that never came back.
        async fn resubscribe(&mut self) -> Result<(), Resubscribe> {
            Err(Resubscribe::Lost(LinkLost::Budget {
                waited: Duration::ZERO,
            }))
        }
    }

    /// An [`EventSource`] that ends the way a handover ends a real
    /// subscription, and hands out the next generation when asked.
    ///
    /// The whole shape of a handover from a dog's side: the stream stops,
    /// and whether the dog lives depends on there being a successor to
    /// subscribe to.
    struct HandoverSource {
        current: Option<broadcast::Receiver<BusEvent>>,
        /// Generations still to come, oldest first. Empty means no
        /// shepherd answered before the budget ran out.
        later: std::collections::VecDeque<broadcast::Receiver<BusEvent>>,
        /// What [`Self::resubscribe`] reports when `later` is empty.
        when_gone: LinkLost,
        /// Set instead of `when_gone` when the shepherd answers and
        /// refuses, which exits on the error's own code rather than on the
        /// budget's.
        refuses: Option<RpcErrorCode>,
        resubscribes: Arc<std::sync::atomic::AtomicU32>,
    }

    impl HandoverSource {
        fn across(
            generations: Vec<broadcast::Receiver<BusEvent>>,
            when_gone: LinkLost,
        ) -> (Self, Arc<std::sync::atomic::AtomicU32>) {
            Self::across_inner(generations, when_gone, None)
        }

        /// A source whose shepherd answers the re-subscribe and refuses it.
        fn refusing(
            generations: Vec<broadcast::Receiver<BusEvent>>,
            code: RpcErrorCode,
        ) -> (Self, Arc<std::sync::atomic::AtomicU32>) {
            Self::across_inner(
                generations,
                LinkLost::Budget {
                    waited: Duration::ZERO,
                },
                Some(code),
            )
        }

        fn across_inner(
            generations: Vec<broadcast::Receiver<BusEvent>>,
            when_gone: LinkLost,
            refuses: Option<RpcErrorCode>,
        ) -> (Self, Arc<std::sync::atomic::AtomicU32>) {
            let mut later: std::collections::VecDeque<_> = generations.into();
            let current = later.pop_front();
            let resubscribes = Arc::new(std::sync::atomic::AtomicU32::new(0));
            (
                Self {
                    current,
                    later,
                    when_gone,
                    refuses,
                    resubscribes: Arc::clone(&resubscribes),
                },
                resubscribes,
            )
        }
    }

    impl EventSource for HandoverSource {
        async fn next(&mut self) -> Option<Result<BusEvent, u64>> {
            let stream = self.current.as_mut()?;
            match stream.recv().await {
                Ok(event) => Some(Ok(event)),
                Err(broadcast::error::RecvError::Lagged(count)) => Some(Err(count)),
                Err(broadcast::error::RecvError::Closed) => {
                    self.current = None;
                    None
                }
            }
        }

        async fn resubscribe(&mut self) -> Result<(), Resubscribe> {
            self.resubscribes
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            if let Some(code) = self.refuses {
                return Err(Resubscribe::Request(RequestError::Rpc(RpcError {
                    code,
                    message: "this shepherd does not serve that topic".into(),
                    daemon_version: None,
                })));
            }
            match self.later.pop_front() {
                Some(stream) => {
                    self.current = Some(stream);
                    Ok(())
                }
                None => Err(Resubscribe::Lost(self.when_gone.clone())),
            }
        }
    }

    /// A [`FlockSource`] answering one fixed listing, counting how many
    /// times it was asked.
    #[derive(Clone)]
    struct ScriptedFlock {
        answer: Arc<Vec<ProcessInfo>>,
        calls: Arc<std::sync::atomic::AtomicU32>,
    }

    impl ScriptedFlock {
        fn answering(answer: Vec<ProcessInfo>) -> Self {
            Self {
                answer: Arc::new(answer),
                calls: Arc::new(std::sync::atomic::AtomicU32::new(0)),
            }
        }

        fn calls(&self) -> u32 {
            self.calls.load(std::sync::atomic::Ordering::SeqCst)
        }
    }

    impl FlockSource for ScriptedFlock {
        async fn flock(&self) -> Result<Vec<ProcessInfo>, RequestError> {
            self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            Ok((*self.answer).clone())
        }
    }

    /// A [`ConfigSource`] answering one fixed `[bark]` section, counting
    /// how many times it was asked. The count is what proves the loop
    /// re-asks on a `config.dog.bark` frame rather than acting on it.
    #[derive(Clone)]
    struct ScriptedConfig {
        section: Arc<String>,
        calls: Arc<std::sync::atomic::AtomicU32>,
    }

    impl ScriptedConfig {
        fn answering(section: String) -> Self {
            Self {
                section: Arc::new(section),
                calls: Arc::new(std::sync::atomic::AtomicU32::new(0)),
            }
        }

        fn calls(&self) -> u32 {
            self.calls.load(std::sync::atomic::Ordering::SeqCst)
        }
    }

    impl ConfigSource for ScriptedConfig {
        async fn section(&self) -> Result<String, RequestError> {
            self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            Ok((*self.section).clone())
        }
    }

    /// Binds an ephemeral port, accepts exactly one connection, answers
    /// `status`/`body`, and hands the captured request back through the
    /// returned receiver.
    async fn one_shot_sink(
        status: u16,
        body: &str,
    ) -> (SocketAddr, oneshot::Receiver<HttpRequest>) {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let (tx, rx) = oneshot::channel();
        let body = body.to_string();
        tokio::spawn(async move {
            let (mut stream, _peer) = listener.accept().await.unwrap();
            let req = read_request(&mut stream, Duration::from_secs(5))
                .await
                .unwrap();
            write_response(&mut stream, status, "application/json", body.as_bytes())
                .await
                .unwrap();
            let _ = tx.send(req);
        });
        (addr, rx)
    }

    /// A sink that accepts one connection and then never answers, plus a
    /// signal that fires the moment it has accepted.
    ///
    /// The signal lets a caller assert an order rather than a duration,
    /// which would be a claim about how a runner schedules two tasks.
    async fn slow_sink() -> (SocketAddr, tokio::sync::oneshot::Receiver<()>) {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let (connected, rx) = tokio::sync::oneshot::channel();
        tokio::spawn(async move {
            let (_stream, _peer) = listener.accept().await.unwrap();
            // Ignored: a caller that does not care drops the receiver.
            let _ = connected.send(());
            core::future::pending::<()>().await;
        });
        (addr, rx)
    }

    fn base_info(name: &str, status: ProcStatus, restarts: u32) -> ProcessInfo {
        ProcessInfo::builder(1, name, status)
            .pid(Some(4242))
            .restarts(restarts)
            .uptime_ms(1_000)
            .build()
    }

    fn errored_info(name: &str, restarts: u32) -> ProcessInfo {
        base_info(name, ProcStatus::Errored, restarts)
    }

    fn process_event(name: &str, kind: ProcessEventKind) -> BusEvent {
        BusEvent::Process {
            event: kind,
            info: base_info(name, ProcStatus::Online, 0),
            manually: false,
            at_ms: 0,
        }
    }

    fn errored_event(name: &str) -> BusEvent {
        process_event(name, ProcessEventKind::Errored)
    }

    /// A JSON sink POSTing to `url` with the default body. Every sink
    /// these tests configure is this one.
    fn json_sink(url: String) -> Sink {
        Sink::Json { url, body: None }
    }

    /// A cheap bus event no rule below fires on: filler for overflowing
    /// the broadcast channel's small capacity.
    fn log_event(i: u32) -> BusEvent {
        BusEvent::LogOut {
            id: i,
            line: format!("log line {i}"),
        }
    }

    /// One `gave_up` rule routed to the sink named `"ops"`, the name
    /// [`config_with_sink`] defines.
    ///
    /// The debounce is a real five minutes, not zero: the reconciliation
    /// test's channel yields `errored_event("web")` again as an ordinary
    /// item after the lag notice, and a zero debounce suppresses nothing.
    fn gave_up_rules() -> Rules {
        let mut sinks = BTreeMap::new();
        sinks.insert(
            "ops".to_owned(),
            json_sink("http://127.0.0.1:1/hook".to_owned()),
        );
        Rules::new(
            vec![rules::Rule {
                when: rules::Trigger::GaveUp {},
                sinks: vec!["ops".to_owned()],
                debounce: UpDuration::from_millis(5 * 60_000),
            }],
            &sinks,
        )
        .unwrap()
    }

    /// A [`BarkConfig`] with one sink, `"ops"`, POSTing to `addr`. `poll`
    /// is 60s, past every timeout these tests bound themselves by, so a
    /// poll that fires is attributable to the lag path.
    fn config_with_sink(addr: SocketAddr, _barks_path: &Path) -> BarkConfig {
        let mut sinks = BTreeMap::new();
        sinks.insert("ops".to_owned(), json_sink(format!("http://{addr}/hook")));
        BarkConfig {
            sinks,
            rules: Vec::new(),
            poll: UpDuration::from_millis(60_000),
            history_bytes: barks::DEFAULT_MAX_BYTES,
            sink_timeout: UpDuration::from_millis(5_000),
        }
    }

    /// Fails if the poll is only ever driven by its interval: `web`'s
    /// `errored` frame is genuinely dropped by a real broadcast channel,
    /// and the fixture's interval is 60s.
    ///
    /// A real clock, not `start_paused`: a deadline awaited through a
    /// `spawn_blocking` bridge cannot elapse under a paused one, so a
    /// regression hangs the suite instead of failing it.
    #[tokio::test]
    async fn a_dropped_frame_makes_bark_poll_and_catch_up() {
        let (tx, mut rx) = tokio::sync::broadcast::channel(4);
        for i in 0..64 {
            tx.send(log_event(i)).unwrap();
        }
        tx.send(errored_event("web")).unwrap();

        // The drop is real, or this test proves nothing.
        assert!(
            matches!(rx.recv().await, Err(broadcast::error::RecvError::Lagged(n)) if n > 0),
            "the fixture must actually overflow the channel"
        );

        let (tx2, rx2) = tokio::sync::broadcast::channel(4);
        for i in 0..64 {
            tx2.send(log_event(i)).unwrap();
        }
        tx2.send(errored_event("web")).unwrap();

        let (addr, captured) = one_shot_sink(200, "").await;
        let dir = tempfile::tempdir().unwrap();
        let barks_path = dir.path().join("barks.jsonl");
        let flock = ScriptedFlock::answering(vec![errored_info("web", 16)]);

        let loop_handle = tokio::spawn(run_loop(
            rx2,
            flock.clone(),
            gave_up_rules(),
            &config_with_sink(addr, &barks_path),
            &barks_path,
            // Never asked: no `config.dog.bark` frame is sent here.
            ScriptedConfig::answering(String::new()),
        ));

        let req = tokio::time::timeout(Duration::from_secs(5), captured)
            .await
            .expect("a dropped frame must produce a delivered bark")
            .unwrap();
        assert!(String::from_utf8_lossy(&req.body).contains("web"));

        // `captured` resolves when the sink server finishes writing its
        // response, concurrently with the delivery task's own tail
        // (outcome, append). A bounded poll covers that gap.
        let recorded = tokio::time::timeout(Duration::from_secs(5), async {
            loop {
                let records = shep_core::barks::read(&barks_path).unwrap();
                if !records.is_empty() {
                    break records;
                }
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("the delivered bark must be recorded promptly after delivery");
        assert_eq!(recorded.len(), 1);
        assert_eq!(recorded[0].subject, "web");
        assert_eq!(recorded[0].sinks[0].error, None);

        assert_eq!(
            flock.calls(),
            1,
            "the poll ran because of the lag, not because an interval elapsed: \
             the interval is 60s and this test is milliseconds old"
        );

        loop_handle.abort();
    }

    /// fails if a handover restarts the bark dog, which is what it used to
    /// do: the stream ends with the connection, and a loop that took that
    /// for its shepherd going away exited 0 once per `shep daemon reload`.
    /// The restart is not free. `restarts` is the one column an operator
    /// reads to judge a dog's health, and the dog drops every rule's
    /// per-subject debounce on the way out, so a sheep already alerted on
    /// can be alerted on again.
    #[tokio::test]
    async fn a_handover_re_subscribes_instead_of_ending_the_dog() {
        let (first_tx, first_rx) = tokio::sync::broadcast::channel(8);
        let (second_tx, second_rx) = tokio::sync::broadcast::channel(8);
        let (source, resubscribes) = HandoverSource::across(
            vec![first_rx, second_rx],
            LinkLost::Budget {
                waited: Duration::ZERO,
            },
        );

        let (addr, captured) = one_shot_sink(200, "").await;
        let dir = tempfile::tempdir().unwrap();
        let barks_path = dir.path().join("barks.jsonl");
        // Empty on purpose. A re-subscribe reconciles, so a flock with
        // anything firing in it delivers a bark from that reconcile and
        // satisfies `captured` whether or not the second generation was
        // ever armed. The bark below has to come from the stream.
        let flock = ScriptedFlock::answering(Vec::new());

        let loop_handle = tokio::spawn(run_loop(
            source,
            flock.clone(),
            gave_up_rules(),
            &config_with_sink(addr, &barks_path),
            &barks_path,
            ScriptedConfig::answering(String::new()),
        ));

        // The handover: the first generation's sender goes, exactly as an
        // `execve` takes the accepted connection with it.
        drop(first_tx);

        // An event only the SECOND generation could have carried. A loop
        // that ended on the handover never sees this one.
        let delivered = tokio::time::timeout(Duration::from_secs(5), async {
            loop {
                // Sent repeatedly: the second generation is armed inside
                // the loop, and a broadcast sent before anyone subscribed
                // reaches nobody.
                let _ = second_tx.send(errored_event("web"));
                tokio::time::sleep(Duration::from_millis(20)).await;
                if loop_handle.is_finished() {
                    panic!("the loop ended on a handover instead of re-subscribing");
                }
            }
        });
        let captured = tokio::time::timeout(Duration::from_secs(5), captured);
        let request = tokio::select! {
            // Reached only when the timeout above expires, which is this
            // test failing rather than an impossible state: the sender
            // loop itself never returns.
            () = async { delivered.await.ok(); } => panic!(
                "no bark was delivered within 5s of the handover, so the \
                 second generation never reached the loop"
            ),
            request = captured => request.expect("a bark must be delivered after the handover"),
        };
        assert!(String::from_utf8_lossy(&request.unwrap().body).contains("web"));

        assert_eq!(
            resubscribes.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "the handover must have been answered by exactly one re-subscribe"
        );
        assert!(
            flock.calls() >= 1,
            "a re-subscribe must reconcile: frames sent while there was no \
             subscription are gone, and only the shepherd knows the flock now"
        );
        assert!(
            !loop_handle.is_finished(),
            "the dog must still be running after a handover"
        );
        loop_handle.abort();
    }

    /// fails if a dog survives one handover and not the next.
    ///
    /// The single-handover test proves the arm runs once. A reload is not a
    /// one-off, and a rolling restart is several in a row, so the property
    /// that matters is that the second and third cost no more than the
    /// first. A `resubscribe` that double-advanced its queue or left the
    /// previous generation in place would pass with one handover.
    #[tokio::test]
    async fn a_dog_rides_out_three_shepherds_in_a_row() {
        let (first_tx, first_rx) = tokio::sync::broadcast::channel(8);
        let (second_tx, second_rx) = tokio::sync::broadcast::channel(8);
        let (third_tx, third_rx) = tokio::sync::broadcast::channel(8);
        let (source, resubscribes) = HandoverSource::across(
            vec![first_rx, second_rx, third_rx],
            LinkLost::Budget {
                waited: Duration::ZERO,
            },
        );

        let (addr, captured) = one_shot_sink(200, "").await;
        let dir = tempfile::tempdir().unwrap();
        let barks_path = dir.path().join("barks.jsonl");
        // Empty on purpose. A re-subscribe reconciles, and a flock with
        // anything firing in it would deliver a bark from that reconcile,
        // which satisfies `captured` without the third generation ever
        // being read. The only bark this test can produce has to come from
        // the stream.
        let flock = ScriptedFlock::answering(Vec::new());

        let loop_handle = tokio::spawn(run_loop(
            source,
            flock.clone(),
            gave_up_rules(),
            &config_with_sink(addr, &barks_path),
            &barks_path,
            ScriptedConfig::answering(String::new()),
        ));

        // Two handovers back to back, each ending a generation the dog is
        // already on.
        drop(first_tx);
        let two = tokio::time::timeout(Duration::from_secs(5), async {
            while resubscribes.load(std::sync::atomic::Ordering::SeqCst) < 1 {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
            drop(second_tx);
            while resubscribes.load(std::sync::atomic::Ordering::SeqCst) < 2 {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await;
        assert!(two.is_ok(), "both handovers must be answered within 5s");

        // An event only the THIRD generation could carry.
        let delivered = tokio::time::timeout(Duration::from_secs(5), async {
            loop {
                let _ = third_tx.send(errored_event("web"));
                tokio::time::sleep(Duration::from_millis(20)).await;
                assert!(
                    !loop_handle.is_finished(),
                    "the dog ended on one of the two handovers"
                );
            }
        });
        let captured = tokio::time::timeout(Duration::from_secs(5), captured);
        let request = tokio::select! {
            () = async { delivered.await.ok(); } => panic!(
                "no bark was delivered on the third generation within 5s"
            ),
            request = captured => request.expect("a bark must be delivered after two handovers"),
        };
        assert!(String::from_utf8_lossy(&request.unwrap().body).contains("web"));

        assert_eq!(
            resubscribes.load(std::sync::atomic::Ordering::SeqCst),
            2,
            "two handovers, two re-subscribes, no more"
        );
        loop_handle.abort();
    }

    /// fails if a dog whose shepherd is gone for good lingers. A lingering
    /// dog attaches itself to whatever shepherd next binds that socket,
    /// beside that shepherd's own dog of the same kind, and doubles its
    /// alerts quietly.
    #[tokio::test]
    async fn a_shepherd_that_never_comes_back_ends_the_dog() {
        let (tx, rx) = tokio::sync::broadcast::channel(8);
        let (source, resubscribes) = HandoverSource::across(
            vec![rx],
            LinkLost::Budget {
                waited: Duration::from_secs(5),
            },
        );

        let (addr, _captured) = one_shot_sink(200, "").await;
        let dir = tempfile::tempdir().unwrap();
        let barks_path = dir.path().join("barks.jsonl");

        let loop_handle = tokio::spawn(run_loop(
            source,
            ScriptedFlock::answering(Vec::new()),
            gave_up_rules(),
            &config_with_sink(addr, &barks_path),
            &barks_path,
            ScriptedConfig::answering(String::new()),
        ));

        drop(tx);

        let code = tokio::time::timeout(Duration::from_secs(5), loop_handle)
            .await
            .expect("a dog whose shepherd is gone must exit, not linger")
            .unwrap();
        assert_eq!(
            code,
            ExitCode::DaemonUnreachable,
            "exiting 0 would read as a dog that finished its work"
        );
        assert_eq!(
            resubscribes.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "it must have tried once before giving up"
        );
    }

    /// fails if a shepherd that answers and refuses the subscription is
    /// reported as a shepherd that never answered.
    ///
    /// `run_bark` exits `ExitCode::from(&err)` when the opening `Subscribe`
    /// is refused. A re-subscribe meeting the same refusal has to exit the
    /// same code, or the same fault reports differently depending on
    /// whether a handover happened to have run first.
    #[tokio::test]
    async fn a_refused_re_subscribe_exits_on_the_refusals_own_code() {
        let (tx, rx) = tokio::sync::broadcast::channel(8);
        let (source, _resubscribes) = HandoverSource::refusing(vec![rx], RpcErrorCode::Unsupported);

        let (addr, _captured) = one_shot_sink(200, "").await;
        let dir = tempfile::tempdir().unwrap();
        let barks_path = dir.path().join("barks.jsonl");

        let loop_handle = tokio::spawn(run_loop(
            source,
            ScriptedFlock::answering(Vec::new()),
            gave_up_rules(),
            &config_with_sink(addr, &barks_path),
            &barks_path,
            ScriptedConfig::answering(String::new()),
        ));

        drop(tx);

        let code = tokio::time::timeout(Duration::from_secs(5), loop_handle)
            .await
            .expect("a refused subscription must end the dog, not be waited out")
            .unwrap();
        assert_eq!(
            code,
            ExitCode::Unsupported,
            "exiting DaemonUnreachable would name a shepherd that answered"
        );
    }

    /// fails if a dog that cannot speak the shepherd's protocol exits as
    /// though nothing answered. The shepherd that refused is running and
    /// answering, and an operator told it was unreachable goes looking for
    /// the wrong thing.
    #[tokio::test]
    async fn a_refused_dog_exits_saying_it_was_refused() {
        let (tx, rx) = tokio::sync::broadcast::channel(8);
        let (source, _resubscribes) = HandoverSource::across(
            vec![rx],
            LinkLost::Refused {
                daemon_version: Some("0.9.9".into()),
                message: "daemon speaks protocol 9, client speaks 8".into(),
            },
        );

        let (addr, _captured) = one_shot_sink(200, "").await;
        let dir = tempfile::tempdir().unwrap();
        let barks_path = dir.path().join("barks.jsonl");

        let loop_handle = tokio::spawn(run_loop(
            source,
            ScriptedFlock::answering(Vec::new()),
            gave_up_rules(),
            &config_with_sink(addr, &barks_path),
            &barks_path,
            ScriptedConfig::answering(String::new()),
        ));

        drop(tx);

        let code = tokio::time::timeout(Duration::from_secs(5), loop_handle)
            .await
            .expect("a refused dog must exit rather than retry")
            .unwrap();
        assert_eq!(code, ExitCode::ProtocolMismatch);
    }

    /// Drives `deliver_and_record` directly rather than through
    /// `run_loop`: the property belongs to that function, and the loop's
    /// event plumbing would need a second synchronization mechanism to
    /// know when a failed delivery finished.
    #[tokio::test]
    async fn a_bark_is_recorded_even_when_every_sink_refuses_it() {
        let (addr, _captured) = one_shot_sink(500, "refused").await;
        let dir = tempfile::tempdir().unwrap();
        let barks_path = dir.path().join("barks.jsonl");

        let mut sinks = BTreeMap::new();
        sinks.insert("ops".to_owned(), json_sink(format!("http://{addr}/hook")));
        let firing = Firing {
            bark: Bark {
                at_ms: 1_000,
                rule: "gave_up".to_owned(),
                subject: "web".to_owned(),
                message: "web gave up: restart budget exhausted".to_owned(),
                sinks: Vec::new(),
            },
            sinks: vec!["ops".to_owned()],
        };

        deliver_and_record(
            firing,
            &Delivery {
                sinks: Arc::new(sinks),
                append_lock: Arc::new(Mutex::new(())),
                barks_path: Arc::new(barks_path.clone()),
                sink_timeout: Duration::from_secs(5),
                max_bytes: barks::DEFAULT_MAX_BYTES,
            },
        )
        .await;

        let recorded = shep_core::barks::read(&barks_path).unwrap();
        assert_eq!(
            recorded.len(),
            1,
            "a refused delivery must still be recorded"
        );
        assert_eq!(recorded[0].subject, "web");
        assert!(
            recorded[0].sinks[0].error.is_some(),
            "the 500 must be recorded as a failed delivery, not silently dropped"
        );
    }

    /// Fails if a slow sink stalls the loop: a bark dog that stops reading
    /// the bus while it waits drops the frames it exists to catch.
    ///
    /// The proof is an order, not a duration. The slow sink signals that
    /// it has accepted and parked, and the fast sink is reached after
    /// that; an inline-awaiting loop would still be parked. `sink_timeout`
    /// is ten minutes and both timeouts are failure guards, so their
    /// values change how long a broken loop takes to report, never what
    /// passes.
    #[tokio::test]
    async fn a_slow_sink_never_stalls_the_loop() {
        let (slow_addr, slow_connected) = slow_sink().await;
        let (fast_addr, fast_captured) = one_shot_sink(200, "").await;
        let dir = tempfile::tempdir().unwrap();
        let barks_path = dir.path().join("barks.jsonl");

        let mut sinks = BTreeMap::new();
        sinks.insert(
            "slow".to_owned(),
            json_sink(format!("http://{slow_addr}/hook")),
        );
        sinks.insert(
            "fast".to_owned(),
            json_sink(format!("http://{fast_addr}/hook")),
        );
        let rules = Rules::new(
            vec![
                rules::Rule {
                    when: rules::Trigger::GaveUp {},
                    sinks: vec!["slow".to_owned()],
                    debounce: UpDuration::from_millis(0),
                },
                rules::Rule {
                    when: rules::Trigger::Event {
                        kinds: vec!["online".to_owned()],
                    },
                    sinks: vec!["fast".to_owned()],
                    debounce: UpDuration::from_millis(0),
                },
            ],
            &sinks,
        )
        .unwrap();
        let config = BarkConfig {
            sinks,
            rules: Vec::new(),
            poll: UpDuration::from_millis(60_000),
            history_bytes: barks::DEFAULT_MAX_BYTES,
            sink_timeout: UpDuration::from_millis(600_000),
        };

        let (tx, rx) = tokio::sync::broadcast::channel(8);
        tx.send(errored_event("web")).unwrap();
        tx.send(process_event("api", ProcessEventKind::Online))
            .unwrap();

        let flock = ScriptedFlock::answering(Vec::new());
        let loop_handle = tokio::spawn(run_loop(
            rx,
            flock,
            rules,
            &config,
            &barks_path,
            ScriptedConfig::answering(String::new()),
        ));

        // The order is the assertion: the slow sink has a connection and
        // is parked on it, so the loop is provably mid-delivery.
        tokio::time::timeout(Duration::from_secs(30), slow_connected)
            .await
            .expect("the slow sink must be reached at all, or this test proves nothing")
            .unwrap();

        // Then the fast sink is reached anyway. A loop awaiting firings
        // inline would still be parked on the delivery above.
        let req = tokio::time::timeout(Duration::from_secs(30), fast_captured)
            .await
            .expect(
                "the fast sink must be reached while a slow sink is still in \
                 flight; a slow sink must not stall the loop",
            )
            .unwrap();
        assert_eq!(req.method, "POST");

        loop_handle.abort();
    }

    /// Fails if an unconfigured `[dog.bark]` polls in a hot loop, keeps no
    /// history, or times every delivery out instantly: what
    /// `#[derive(Default)]` would give this struct.
    #[test]
    fn an_empty_section_gets_sane_defaults_not_zeros() {
        let parsed: BarkConfig = toml::from_str("").unwrap();
        assert_eq!(parsed, BarkConfig::default());
        assert_eq!(BarkConfig::default().poll.as_millis(), 30_000);
        assert_eq!(
            BarkConfig::default().history_bytes,
            barks::DEFAULT_MAX_BYTES
        );
        assert_eq!(BarkConfig::default().sink_timeout.as_millis(), 10_000);
    }

    /// Fails if `[dog.bark]` cannot parse the fragment `docs/dogs.md` and
    /// `web/src/pages/docs/dogs.astro` publish, copy-pasted here relative
    /// to `[dog.bark]` the way `runtime.config::<BarkConfig>()` sees it,
    /// so `[sinks]`/`[[rules]]` rather than the full paths.
    ///
    /// The only other `toml::from_str::<BarkConfig>` in this module parses
    /// an empty document, which never deserializes a [`rules::Rule`].
    #[test]
    fn the_documented_bark_config_parses_from_toml() {
        let toml_str = r#"
[sinks]
oncall = { kind = "discord", url = "https://discord.com/api/webhooks/..." }
audit = { kind = "json", url = "https://example.internal/hook" }

[[rules]]
on = "gave_up"
sinks = ["oncall", "audit"]

[[rules]]
on = "restart_rate"
restarts = 5
within = "2m"
sinks = ["oncall"]
"#;
        let config: BarkConfig =
            toml::from_str(toml_str).expect("the documented [dog.bark] example must parse");
        assert_eq!(config.sinks.len(), 2);
        assert_eq!(config.rules.len(), 2);
        assert_eq!(config.rules[0].when, rules::Trigger::GaveUp {});
        assert_eq!(config.rules[0].sinks, vec!["oncall", "audit"]);
        assert_eq!(
            config.rules[1].when,
            rules::Trigger::RestartRate {
                restarts: 5,
                within: UpDuration::from_millis(2 * 60_000),
            }
        );
        assert_eq!(config.rules[1].sinks, vec!["oncall"]);
        // Parsing is necessary but not sufficient: the dog starts only if
        // `Rules::new` accepts both rules against the sinks beside them.
        Rules::new(config.rules, &config.sinks).expect("both documented rules route to real sinks");
    }

    /// Fails if a `config.dog.bark` frame does not reach bark's sinks: the
    /// next firing is delivered to the sink the new section names, over a
    /// real socket, while the old sink's server is still up.
    ///
    /// The loop is the same loop throughout, which is the other half of
    /// what this pins: bark's config is pure data with no OS resource
    /// attached, so it swaps in place instead of restarting.
    ///
    /// A real clock, not `start_paused`, or a broken swap hangs the suite
    /// rather than failing it.
    #[tokio::test]
    async fn a_config_change_swaps_barks_sinks_in_place() {
        let (old_addr, mut old_captured) = one_shot_sink(200, "").await;
        let (new_addr, new_captured) = one_shot_sink(200, "").await;
        let dir = tempfile::tempdir().unwrap();
        let barks_path = dir.path().join("barks.jsonl");
        let source = ScriptedConfig::answering(format!(
            "[sinks.ops]\nkind = \"json\"\nurl = \"http://{new_addr}/hook\"\n"
        ));

        let (tx, rx) = broadcast::channel(16);
        let loop_handle = tokio::spawn(run_loop(
            rx,
            ScriptedFlock::answering(Vec::new()),
            gave_up_rules(),
            &config_with_sink(old_addr, &barks_path),
            &barks_path,
            source.clone(),
        ));

        // One receiver, one queue: the loop awaits the re-ask before it
        // takes the second event, so the delivery below is attributable
        // to the new section.
        tx.send(BusEvent::DogConfigChanged {
            dog: "bark".to_owned(),
        })
        .unwrap();
        tx.send(errored_event("web")).unwrap();

        let req = tokio::time::timeout(Duration::from_secs(5), new_captured)
            .await
            .expect("the bark must reach the sink the new section names")
            .unwrap();
        assert!(String::from_utf8_lossy(&req.body).contains("web"));

        assert_eq!(source.calls(), 1, "one frame, one re-ask");
        assert!(
            !loop_handle.is_finished(),
            "bark swaps its config in place; it does not exit to pick one up"
        );
        // `try_recv`, never an await: the old sink's server is still
        // parked in `accept`, so it holds its sender alive and an await
        // would never return. The delivery above has already happened.
        assert!(
            old_captured.try_recv().is_err(),
            "the sink the old section named must be left alone"
        );

        loop_handle.abort();
    }
}