subc-daemon 0.21.14

//! Bounded per-module stderr capture.
//!
//! # Why this exists
//!
//! `last_exit_code` survives a respawn because the supervisor holds it in memory.
//! Stderr had no such path: it went to the daemon's inherited fd, and from there
//! to whatever rotates or evicts it. On the box this was written for, that window
//! was about three hours; on another it was bounded by a log file reaching 908 MB
//! with one module accounting for 98% of it. Two hosts, two mechanisms, the same
//! outcome -- the text explaining a crash is gone by the time anyone asks.
//!
//! So this keeps the last few lines where `last_exit` already lives: in supervisor
//! memory, immune to whatever happens to the log.
//!
//! # What it is not
//!
//! Not a log. The ring is deliberately small and lossy, and callers are expected
//! to know they are reading a tail rather than a history. The daemon log keeps
//! doing its job; this exists because that job has a time limit.

use std::collections::{BTreeMap, VecDeque};
use std::io::{self, Write};
use std::path::{Path, PathBuf};
use std::sync::{
    atomic::{AtomicBool, Ordering},
    Arc, Mutex,
};

use tokio::io::AsyncReadExt;

/// Longest single line admitted to the ring before truncation.
///
/// A module emitting a 40 MB backtrace on one line satisfies any line-count cap
/// while evicting everything else -- the pathological emitter wins twice, once by
/// filling the ring and once by being unreadable itself. Truncating on the way in
/// costs that emitter one line instead of the whole tail.
pub const DEFAULT_MAX_LINE_BYTES: usize = 2048;

/// Lines retained per module.
pub const DEFAULT_MAX_LINES: usize = 200;

/// Total bytes retained per module, across all lines.
///
/// Both this and [`DEFAULT_MAX_LINES`] apply; whichever binds first wins. A line
/// cap alone is satisfied by 200 lines of 2 KB, which is not a budget worth
/// holding for fifteen modules.
pub const DEFAULT_MAX_BYTES: usize = 64 * 1024;

/// Whether a module's stderr is being captured, and if not, why not.
///
/// Typed rather than nullable so `NotCaptured` has to be handled rather than
/// defaulted past. An empty tail and an uncaptured one send an operator in
/// opposite directions -- one says the module printed nothing before dying, the
/// other says nobody was listening -- and rendering them alike is the defect this
/// module exists to fix, reproduced one layer up.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CaptureState {
    /// A reader is attached, or was attached and reached clean EOF.
    Captured,
    /// Retained entries are valid, but capture ended before clean EOF, or the
    /// pipe of a process the supervisor has already moved on from has not
    /// reached EOF yet. The second kind clears when that pipe does reach EOF,
    /// because from then on nothing that process wrote is missing.
    Incomplete { reason: String },
    /// No reader was attached. The tail says nothing about what the module wrote.
    NotCaptured { reason: String },
}

/// One retained entry.
///
/// Boundaries are in-band rather than a separate field because their position
/// relative to the lines is the whole point: "these three lines came from the
/// process that died, those came from its replacement" is unanswerable from a
/// count.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TailEntry {
    Line {
        text: String,
        /// This line was cut at the per-line cap.
        truncated: bool,
    },
    /// The supervisor spawned a new process for this module. Lines after this
    /// entry come from the new one.
    ProcessStart,
}

impl TailEntry {
    fn cost(&self) -> usize {
        match self {
            Self::Line { text, .. } => text.len(),
            Self::ProcessStart => 0,
        }
    }
}

/// One stored entry. Unlike [`TailEntry`], a boundary remembers which process
/// generation it starts, so a line that arrives late from an older process can
/// be put back in that process's section instead of after its successor's
/// boundary.
#[derive(Debug, Clone, PartialEq, Eq)]
enum Slot {
    Line { text: String, truncated: bool },
    ProcessStart { generation: u64 },
}

impl Slot {
    fn cost(&self) -> usize {
        match self {
            Self::Line { text, .. } => text.len(),
            Self::ProcessStart { .. } => 0,
        }
    }
}

/// Where the stderr reader of one process generation stands.
#[derive(Debug, Clone, PartialEq, Eq)]
enum PumpPhase {
    /// The process is the supervisor's current concern; its lines go at the end.
    Attached,
    /// The supervisor has moved on from the process, but its pipe is still open.
    /// Lines it still delivers belong in its own section.
    Retired,
    /// Retired, and the pipe was still open when the supervisor stopped waiting
    /// for it. Reported as `Incomplete` until the pipe reaches EOF.
    Late { reason: String },
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct StderrTailConfig {
    max_lines: usize,
    max_bytes: usize,
    max_line_bytes: usize,
}

impl StderrTailConfig {
    /// Keeps every retained line within the ring's total byte budget.
    /// Clamp rather than reject so diagnostics degrade without blocking supervisor startup.
    pub const fn new(max_lines: usize, max_bytes: usize, max_line_bytes: usize) -> Self {
        Self {
            max_lines,
            max_bytes,
            max_line_bytes: if max_line_bytes > max_bytes {
                max_bytes
            } else {
                max_line_bytes
            },
        }
    }
}

impl Default for StderrTailConfig {
    fn default() -> Self {
        Self::new(DEFAULT_MAX_LINES, DEFAULT_MAX_BYTES, DEFAULT_MAX_LINE_BYTES)
    }
}

/// A module's retained stderr, oldest first.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StderrTailSnapshot {
    pub capture: CaptureState,
    pub entries: Vec<TailEntry>,
    /// Lines evicted since the module was first supervised.
    ///
    /// Non-zero means the tail starts mid-stream. That is the ring working as
    /// intended, but a reader diagnosing a crash needs to know the first retained
    /// line is not the first line the module wrote -- otherwise an absent cause
    /// reads as a module that never explained itself.
    pub dropped_lines: u64,
}

impl StderrTailSnapshot {
    /// The uncaptured case, for a module whose stderr was never piped.
    pub fn not_captured(reason: impl Into<String>) -> Self {
        Self {
            capture: CaptureState::NotCaptured {
                reason: reason.into(),
            },
            entries: Vec::new(),
            dropped_lines: 0,
        }
    }
}

/// Bounded ring of a single module's stderr lines.
///
/// Survives respawn deliberately. The stderr explaining an exit is written
/// *before* that exit, so clearing on restart would discard the lines exactly
/// when they become the thing being asked for. [`TailEntry::ProcessStart`] keeps
/// the generations distinguishable instead.
///
/// A process's stderr reader can outlive the supervisor's interest in it: the
/// reader may not have been scheduled yet when the process exited, or a
/// descendant may still hold the pipe open. Lines such a reader delivers after
/// the next process started are placed before that next process's boundary, so
/// the section a line appears in always names the process that wrote it.
#[derive(Debug)]
pub struct StderrRing {
    config: StderrTailConfig,
    entries: VecDeque<Slot>,
    // Count of `TailEntry::Line` entries, kept running because eviction checks
    // it on every push and recounting would walk the whole ring under the
    // mutex each time.
    lines: usize,
    bytes: usize,
    dropped_lines: u64,
    capture: CaptureState,
    /// Generation of the newest process boundary; 0 before the first.
    generation: u64,
    /// Readers that have not reached EOF yet, by the generation they read for.
    pumps: BTreeMap<u64, PumpPhase>,
    /// Highest generation whose boundary was evicted from the front. A late line
    /// from an older generation belongs in front of that boundary, which is
    /// evicted territory, so it is counted as dropped rather than stored.
    evicted_through: u64,
}

impl StderrRing {
    pub fn new(config: StderrTailConfig) -> Self {
        Self {
            config,
            entries: VecDeque::new(),
            lines: 0,
            bytes: 0,
            dropped_lines: 0,
            // Until a reader attaches, the honest answer is that nothing is
            // listening -- not that the module has been quiet.
            capture: CaptureState::NotCaptured {
                reason: "stderr reader has not started".to_string(),
            },
            generation: 0,
            pumps: BTreeMap::new(),
            evicted_through: 0,
        }
    }

    /// Generation of the newest process boundary.
    pub fn generation(&self) -> u64 {
        self.generation
    }

    pub fn mark_captured(&mut self) {
        if matches!(self.capture, CaptureState::NotCaptured { .. }) {
            self.capture = CaptureState::Captured;
        }
    }

    pub fn mark_incomplete(&mut self, reason: impl Into<String>) {
        self.capture = CaptureState::Incomplete {
            reason: reason.into(),
        };
    }

    pub fn mark_not_captured(&mut self, reason: impl Into<String>) {
        self.capture = CaptureState::NotCaptured {
            reason: reason.into(),
        };
    }

    /// Record that a new process was spawned for this module, returning the
    /// generation number its lines are attributed to.
    ///
    /// A boundary separates output on either side of it, so one with nothing
    /// before it separates nothing: on the FIRST spawn it would make a module
    /// that printed nothing render as a marker rather than as empty, and the
    /// caller then has to decide whether a one-marker tail counts as silence.
    /// The boundary is still stored, because a late line from an older process
    /// needs it to find its section, but [`Self::snapshot`] shows it only once
    /// there is output before it to divide, which keeps "captured and empty"
    /// literally empty.
    pub fn push_process_start(&mut self) -> u64 {
        self.generation += 1;
        let generation = self.generation;
        // Two boundaries in a row mean the process between them has written
        // nothing retained. The earlier one can go only if no reader from its
        // generation onward is still open: such a reader may yet deliver a line
        // that belongs between the two. Without this a module that restarts
        // silently would grow the ring by one boundary per restart forever.
        if let Some(Slot::ProcessStart {
            generation: previous,
        }) = self.entries.back()
        {
            if self.pumps.range(*previous..).next().is_none() {
                self.entries.pop_back();
            }
        }
        self.push_entry(Slot::ProcessStart { generation });
        generation
    }

    /// [`Self::push_process_start`] for a process whose stderr reader is
    /// attached: the reader is tracked until it calls [`Self::finish_pump`].
    pub(crate) fn begin_process(&mut self) -> u64 {
        let generation = self.push_process_start();
        self.pumps.insert(generation, PumpPhase::Attached);
        generation
    }

    /// The supervisor has moved on from this generation's process. Its reader
    /// keeps running, and any line it still delivers is kept in its own section.
    pub(crate) fn retire_pump(&mut self, generation: u64) {
        if let Some(phase @ PumpPhase::Attached) = self.pumps.get_mut(&generation) {
            *phase = PumpPhase::Retired;
        }
    }

    /// The supervisor stopped waiting for this generation's reader before its
    /// pipe reached EOF. The capture reads as `Incomplete` with `reason` until
    /// the reader finishes. A reader that already finished is left alone: it
    /// got everything.
    pub(crate) fn mark_pump_late(&mut self, generation: u64, reason: impl Into<String>) {
        if let Some(phase) = self.pumps.get_mut(&generation) {
            *phase = PumpPhase::Late {
                reason: reason.into(),
            };
        }
    }

    /// This generation's reader has stopped: at EOF, or on a read error that
    /// has already been recorded with [`Self::mark_incomplete`].
    pub(crate) fn finish_pump(&mut self, generation: u64) {
        self.pumps.remove(&generation);
    }

    /// Admit one complete line, truncating it if it exceeds the per-line cap.
    ///
    /// `line` must not contain a trailing newline; the reader strips it so the
    /// stored text and the byte accounting agree.
    pub fn push_line(&mut self, line: &str) {
        self.push_line_from(self.generation, line);
    }

    /// [`Self::push_line`] for a line read from `generation`'s pipe.
    ///
    /// A line from a retired process that arrives after a newer process started
    /// goes in front of the first boundary newer than its own generation. Lines
    /// from a process the supervisor has not retired (the incumbent during a
    /// swap's overlap) go at the end, as they arrive.
    pub(crate) fn push_line_from(&mut self, generation: u64, line: &str) {
        let (text, truncated) = truncate_line(line, self.config.max_line_bytes);
        let slot = Slot::Line { text, truncated };
        let retired = matches!(
            self.pumps.get(&generation),
            Some(PumpPhase::Retired | PumpPhase::Late { .. })
        );
        if !retired || generation >= self.generation {
            self.push_entry(slot);
            return;
        }
        if self.evicted_through > generation {
            // The section this line belongs to has been evicted, so the line is
            // older than everything retained.
            self.dropped_lines += 1;
            return;
        }
        let index = self.entries.iter().position(
            |slot| matches!(slot, Slot::ProcessStart { generation: start } if *start > generation),
        );
        match index {
            Some(index) => self.insert_entry(index, slot),
            None => self.push_entry(slot),
        }
    }

    fn push_entry(&mut self, entry: Slot) {
        self.insert_entry(self.entries.len(), entry);
    }

    fn insert_entry(&mut self, index: usize, entry: Slot) {
        self.bytes += entry.cost();
        if matches!(entry, Slot::Line { .. }) {
            self.lines += 1;
        }
        self.entries.insert(index, entry);
        self.evict_to_fit();
    }

    fn evict_to_fit(&mut self) {
        while self.lines > self.config.max_lines
            || (self.bytes > self.config.max_bytes && self.entries.len() > 1)
        {
            let Some(evicted) = self.entries.pop_front() else {
                break;
            };
            self.bytes -= evicted.cost();
            match evicted {
                Slot::Line { .. } => {
                    self.lines -= 1;
                    self.dropped_lines += 1;
                }
                Slot::ProcessStart { generation } => {
                    self.evicted_through = self.evicted_through.max(generation);
                }
            }
        }
    }

    /// The most recent entries, oldest first, bounded by the caller's limits.
    ///
    /// `max_lines`/`max_bytes` narrow the ring's own caps; they cannot widen them.
    pub fn snapshot(
        &self,
        max_lines: Option<usize>,
        max_bytes: Option<usize>,
    ) -> StderrTailSnapshot {
        let line_limit = max_lines.unwrap_or(self.config.max_lines);
        let byte_limit = max_bytes.unwrap_or(self.config.max_bytes);

        // The stored boundaries, reduced to the ones worth showing: a boundary
        // with no output before it (retained or evicted) divides nothing, and
        // two in a row say no more than one.
        let mut visible: Vec<TailEntry> = Vec::with_capacity(self.entries.len());
        let mut output_before = self.dropped_lines > 0;
        for slot in &self.entries {
            match slot {
                Slot::Line { text, truncated } => {
                    visible.push(TailEntry::Line {
                        text: text.clone(),
                        truncated: *truncated,
                    });
                    output_before = true;
                }
                Slot::ProcessStart { .. } => {
                    if output_before && !matches!(visible.last(), Some(TailEntry::ProcessStart)) {
                        visible.push(TailEntry::ProcessStart);
                    }
                }
            }
        }

        let mut taken: Vec<TailEntry> = Vec::new();
        let mut bytes = 0usize;
        let mut lines = 0usize;
        // Walk backwards: a tail is anchored at the newest end, so a caller
        // asking for 20 lines wants the last 20, not the first 20.
        for entry in visible.iter().rev() {
            match entry {
                TailEntry::Line { .. } => {
                    if lines >= line_limit {
                        break;
                    }
                    let cost = entry.cost();
                    if lines > 0 && bytes + cost > byte_limit {
                        break;
                    }
                    bytes += cost;
                    lines += 1;
                    taken.push(entry.clone());
                }
                TailEntry::ProcessStart if lines > 0 => taken.push(entry.clone()),
                TailEntry::ProcessStart => {}
            }
        }
        taken.reverse();

        let withheld = self.lines.saturating_sub(lines);

        // A retired process whose pipe the supervisor stopped waiting for may
        // still be writing; until its reader reaches EOF the tail cannot claim
        // to hold everything. A permanent state (a read failure, no pipe at
        // all) is the more specific fact and is reported as is.
        let late = self.pumps.values().find_map(|phase| match phase {
            PumpPhase::Late { reason } => Some(reason),
            _ => None,
        });
        let capture = match (&self.capture, late) {
            (CaptureState::Captured, Some(reason)) => CaptureState::Incomplete {
                reason: reason.clone(),
            },
            (capture, _) => capture.clone(),
        };

        StderrTailSnapshot {
            capture,
            entries: taken,
            // Lines the ring evicted plus lines this request's own limits held
            // back. Both mean the same thing to the reader -- the text above is
            // not the beginning -- and separating them would invite treating a
            // narrow request as evidence of a quiet module.
            dropped_lines: self.dropped_lines + withheld as u64,
        }
    }
}

/// Reassembly buffer ceiling for a line with no newline in sight.
///
/// The ring truncates what it stores, but the READER has to hold the bytes until
/// it finds a delimiter. A module emitting a gigabyte with no newline would grow
/// this buffer without bound and take the daemon down with it -- a module fault
/// escalating into a fleet fault, which is exactly what supervision exists to
/// prevent. At this ceiling the pending bytes are flushed as a line and
/// reassembly restarts.
const MAX_PENDING_LINE_BYTES: usize = 1024 * 1024;

/// Read a child's stderr to EOF, retaining the bounded crash tail and forwarding
/// every complete line to the selected capture sink.
pub async fn pump_stderr<R>(source: R, ring: Arc<Mutex<StderrRing>>)
where
    R: AsyncReadExt + Unpin,
{
    pump_stderr_into(source, ring, &mut StderrSink).await
}

/// Shared destination for a child's stdout and stderr pumps.
///
/// The file keeps the historical `.stderr.log` name even though it carries both
/// streams; the stable name is part of the operator contract. Both pumps share
/// one mutex, and cortexkit-log writes each framed line in one call, so partial
/// lines from the two pipes cannot interleave.
#[derive(Clone)]
pub(crate) enum ChildOutputSink {
    File {
        sink: Arc<Mutex<cortexkit_log::LineSink>>,
        path: Arc<PathBuf>,
        failure_reported: Arc<AtomicBool>,
    },
    Stderr,
}

impl ChildOutputSink {
    pub(crate) fn open(path: &Path, retention: cortexkit_log::Retention) -> io::Result<Self> {
        Ok(Self::File {
            sink: Arc::new(Mutex::new(cortexkit_log::LineSink::open(path, retention)?)),
            path: Arc::new(path.to_path_buf()),
            failure_reported: Arc::new(AtomicBool::new(false)),
        })
    }
}

/// Where forwarded complete lines go. It exists so tests can observe framing
/// and so production can serialize the two child pipes through one file sink.
pub trait OutputSink {
    fn write_line(&mut self, line: &[u8]);
}

struct StderrSink;

impl OutputSink for StderrSink {
    fn write_line(&mut self, line: &[u8]) {
        let stderr = std::io::stderr();
        let mut handle = stderr.lock();
        let _ = handle.write_all(line);
    }
}

impl OutputSink for ChildOutputSink {
    fn write_line(&mut self, line: &[u8]) {
        match self {
            Self::File {
                sink,
                path,
                failure_reported,
            } => {
                let result = sink
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner())
                    .write_line(line);
                if let Err(error) = result {
                    if !failure_reported.swap(true, Ordering::Relaxed) {
                        tracing::warn!(
                            path = %path.display(),
                            error = %error,
                            "child output capture write failed; later failures are suppressed"
                        );
                    }
                }
            }
            Self::Stderr => StderrSink.write_line(line),
        }
    }
}

/// Read one process generation's stderr to EOF. `generation` is the value
/// [`StderrRing::begin_process`] returned for that process; it decides which
/// section of the ring a line lands in if the reader outlives the process.
pub(crate) async fn pump_stderr_to<R, S>(
    source: R,
    ring: Arc<Mutex<StderrRing>>,
    generation: u64,
    mut sink: S,
) where
    R: AsyncReadExt + Unpin,
    S: OutputSink,
{
    pump_lines_into(source, Some((&ring, generation)), &mut sink, "stderr").await;
}

pub(crate) async fn pump_stdout_to<R>(source: R, mut sink: ChildOutputSink)
where
    R: AsyncReadExt + Unpin,
{
    pump_lines_into(source, None, &mut sink, "stdout").await;
}

/// Read stderr for whichever process generation is newest when the reader
/// starts.
async fn pump_stderr_into<R, S>(source: R, ring: Arc<Mutex<StderrRing>>, sink: &mut S)
where
    R: AsyncReadExt + Unpin,
    S: OutputSink,
{
    let generation = lock_ring(&ring).generation();
    pump_lines_into(source, Some((&ring, generation)), sink, "stderr").await;
}

async fn pump_lines_into<R, S>(
    mut source: R,
    ring: Option<(&Arc<Mutex<StderrRing>>, u64)>,
    sink: &mut S,
    stream_name: &str,
) where
    R: AsyncReadExt + Unpin,
    S: OutputSink,
{
    if let Some((ring, _)) = ring {
        lock_ring(ring).mark_captured();
    }

    let mut pending: Vec<u8> = Vec::new();
    // Bytes before `scanned_upto` are already known to hold no newline; searching
    // them again would rescan the whole buffer on every chunk -- for a line with
    // no newline that is about 64 MiB examined per MiB of module output.
    let mut scanned_upto = 0usize;
    // Bytes before `cursor` were emitted as complete lines. They are removed in
    // one compaction per chunk rather than shifting the buffer once per line.
    let mut cursor = 0usize;
    let mut chunk = [0u8; 8192];
    loop {
        let read = match source.read(&mut chunk).await {
            Ok(0) => break,
            Ok(n) => n,
            Err(error) => {
                if let Some((ring, generation)) = ring {
                    let mut ring = lock_ring(ring);
                    ring.mark_incomplete(format!("{stream_name} read failed: {error}"));
                    ring.finish_pump(generation);
                } else {
                    tracing::warn!(stream = stream_name, error = %error, "child output capture read failed");
                }
                return;
            }
        };
        pending.extend_from_slice(&chunk[..read]);

        while let Some(relative) = find_newline(&pending[scanned_upto..]) {
            let newline = scanned_upto + relative;
            emit_line(ring, sink, &pending[cursor..newline], true);
            cursor = newline + 1;
            scanned_upto = cursor;
        }
        scanned_upto = pending.len();

        if cursor > 0 {
            pending.drain(..cursor);
            scanned_upto -= cursor;
            cursor = 0;
        }

        if pending.len() >= MAX_PENDING_LINE_BYTES {
            let line = std::mem::take(&mut pending);
            emit_line(ring, sink, &line, false);
            scanned_upto = 0;
        }
    }

    if !pending.is_empty() {
        emit_line(ring, sink, &pending, false);
    }
    if let Some((ring, generation)) = ring {
        lock_ring(ring).finish_pump(generation);
    }
}

// Bytes examined by newline searches, summed across a pump. Tests use this to
// assert the reader does not rescan bytes it already knows contain no newline.
// Per thread, not process-wide: other tests pump concurrently on their own
// threads, and a shared counter picked up their searches too, failing the
// bound by a few bytes under a parallel run. `#[tokio::test]` runs its body
// and every future it awaits on one thread, so a pump's searches land on the
// test's own counter.
#[cfg(test)]
thread_local! {
    static SCANNED_BYTES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}

#[cfg(test)]
fn take_scanned_bytes() -> usize {
    SCANNED_BYTES.with(|scanned| scanned.replace(0))
}

/// Locate the next newline in `haystack`, counting the bytes examined so a
/// test can observe how much of the pending buffer each search walks.
fn find_newline(haystack: &[u8]) -> Option<usize> {
    let found = memchr::memchr(b'\n', haystack);
    #[cfg(test)]
    SCANNED_BYTES.with(|scanned| {
        scanned.set(scanned.get() + found.map(|index| index + 1).unwrap_or(haystack.len()));
    });
    found
}

fn emit_line<S: OutputSink>(
    ring: Option<(&Arc<Mutex<StderrRing>>, u64)>,
    sink: &mut S,
    raw: &[u8],
    terminated: bool,
) {
    if let Some((ring, generation)) = ring {
        lock_ring(ring).push_line_from(generation, &String::from_utf8_lossy(raw));
    }

    // Framed and written in ONE call. Two writes would let the other pipe land
    // between the body and newline.
    if terminated {
        let mut framed = Vec::with_capacity(raw.len() + 1);
        framed.extend_from_slice(raw);
        framed.push(b'\n');
        sink.write_line(&framed);
    } else {
        sink.write_line(raw);
    }
}

fn lock_ring(ring: &Arc<Mutex<StderrRing>>) -> std::sync::MutexGuard<'_, StderrRing> {
    ring.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
}

/// Cut `line` to at most `max_bytes`, reporting whether it was shortened.
///
/// Cuts on a char boundary: slicing a multi-byte sequence would produce invalid
/// UTF-8, and a panic while capturing a crash message is the worst possible time
/// to discover that.
fn truncate_line(line: &str, max_bytes: usize) -> (String, bool) {
    if line.len() <= max_bytes {
        return (line.to_string(), false);
    }
    let mut end = max_bytes;
    while end > 0 && !line.is_char_boundary(end) {
        end -= 1;
    }
    (line[..end].to_string(), true)
}

#[cfg(test)]
mod tests {
    use std::{
        io,
        pin::Pin,
        task::{Context, Poll},
    };

    use super::*;
    use tokio::io::{AsyncRead, ReadBuf};

    fn ring(max_lines: usize, max_bytes: usize, max_line_bytes: usize) -> StderrRing {
        StderrRing::new(StderrTailConfig::new(max_lines, max_bytes, max_line_bytes))
    }

    fn lines(snapshot: &StderrTailSnapshot) -> Vec<String> {
        snapshot
            .entries
            .iter()
            .filter_map(|entry| match entry {
                TailEntry::Line { text, .. } => Some(text.clone()),
                TailEntry::ProcessStart => None,
            })
            .collect()
    }

    #[test]
    fn a_fresh_ring_reports_not_captured_rather_than_empty() {
        // The distinction this whole module exists for: "nobody was listening"
        // must not render as "the module said nothing".
        let ring = ring(10, 1024, 128);
        let snapshot = ring.snapshot(None, None);
        assert!(matches!(snapshot.capture, CaptureState::NotCaptured { .. }));
        assert!(snapshot.entries.is_empty());
    }

    #[test]
    fn a_captured_module_that_printed_nothing_is_distinguishable_from_an_uncaptured_one() {
        let mut captured = ring(10, 1024, 128);
        captured.mark_captured();
        let uncaptured = ring(10, 1024, 128);

        let captured = captured.snapshot(None, None);
        let uncaptured = uncaptured.snapshot(None, None);

        // Both are empty. Only the capture state separates them, which is the
        // point -- an assertion on emptiness alone would pass either way.
        assert!(captured.entries.is_empty());
        assert!(uncaptured.entries.is_empty());
        assert_eq!(captured.capture, CaptureState::Captured);
        assert!(matches!(
            uncaptured.capture,
            CaptureState::NotCaptured { .. }
        ));
    }

    #[test]
    fn the_line_cap_evicts_oldest_first_and_counts_what_it_dropped() {
        let mut ring = ring(3, 10_000, 128);
        ring.mark_captured();
        for i in 0..6 {
            ring.push_line(&format!("line{i}"));
        }
        let snapshot = ring.snapshot(None, None);
        assert_eq!(lines(&snapshot), vec!["line3", "line4", "line5"]);
        // Without this the tail silently becomes "the last lines that happened
        // to survive" and reads as complete.
        assert_eq!(snapshot.dropped_lines, 3);
    }

    #[test]
    fn the_byte_cap_binds_before_the_line_cap_when_lines_are_large() {
        // 100 lines allowed, but only ~30 bytes of them.
        let mut ring = ring(100, 30, 128);
        ring.mark_captured();
        for i in 0..10 {
            ring.push_line(&format!("{i}--------")); // 9 bytes each
        }
        let snapshot = ring.snapshot(None, None);
        assert!(
            snapshot.entries.len() < 10,
            "byte cap did not bind: {} entries retained",
            snapshot.entries.len()
        );
        let retained: usize = lines(&snapshot).iter().map(String::len).sum();
        assert!(
            retained <= 30,
            "retained {retained} bytes over a 30 byte cap"
        );
        assert!(snapshot.dropped_lines > 0);
    }

    #[test]
    fn one_enormous_line_is_truncated_rather_than_evicting_the_tail() {
        // The pathological-emitter case: without per-line truncation this single
        // line would evict every other line AND be unreadable itself.
        let mut ring = ring(10, 10_000, 64);
        ring.mark_captured();
        ring.push_line("context line that must survive");
        ring.push_line(&"x".repeat(40_000));

        let snapshot = ring.snapshot(None, None);
        let kept = &snapshot.entries;
        assert!(matches!(
            &kept[0],
            TailEntry::Line { text, truncated: false }
                if text == "context line that must survive"
        ));
        let TailEntry::Line { text, truncated } = &kept[1] else {
            panic!("expected a truncated line");
        };
        assert_eq!(text, &"x".repeat(64));
        assert!(*truncated);
    }

    #[test]
    fn truncation_is_visible_so_a_cut_line_is_not_mistaken_for_a_short_one() {
        let mut ring = ring(10, 10_000, 16);
        ring.mark_captured();
        ring.push_line("0123456789abcdefghij");
        ring.push_line("short");

        let snapshot = ring.snapshot(None, None);
        let TailEntry::Line { truncated, .. } = &snapshot.entries[0] else {
            panic!("expected a line");
        };
        assert!(truncated);
        let TailEntry::Line { truncated, .. } = &snapshot.entries[1] else {
            panic!("expected a line");
        };
        assert!(!truncated, "a short line must not be reported as truncated");
    }

    #[test]
    fn truncation_cuts_on_a_char_boundary_rather_than_splitting_utf8() {
        // A panic message with non-ASCII in it is not exotic, and slicing mid
        // sequence would panic while capturing a crash.
        let mut ring = ring(10, 10_000, 5);
        ring.mark_captured();
        ring.push_line("aa€€€€");
        let snapshot = ring.snapshot(None, None);
        let TailEntry::Line { text, truncated } = &snapshot.entries[0] else {
            panic!("expected a line");
        };
        assert!(truncated);
        assert!(text.starts_with("aa"));
    }

    #[test]
    fn a_restart_boundary_keeps_generations_distinguishable() {
        let mut ring = ring(10, 10_000, 128);
        ring.mark_captured();
        ring.push_line("before the crash");
        ring.push_process_start();
        ring.push_line("after the respawn");

        let snapshot = ring.snapshot(None, None);
        assert_eq!(
            snapshot.entries,
            vec![
                TailEntry::Line {
                    text: "before the crash".to_string(),
                    truncated: false
                },
                TailEntry::ProcessStart,
                TailEntry::Line {
                    text: "after the respawn".to_string(),
                    truncated: false
                },
            ]
        );
    }

    #[test]
    fn the_ring_survives_respawn_because_the_cause_is_written_before_the_exit() {
        // Clearing on restart would discard the lines at the exact moment they
        // become the thing being asked for.
        let mut ring = ring(10, 10_000, 128);
        ring.mark_captured();
        ring.push_line("Error: storage section missing");
        ring.push_process_start();

        let snapshot = ring.snapshot(None, None);
        assert!(lines(&snapshot).contains(&"Error: storage section missing".to_string()));
    }

    #[test]
    fn a_caller_limit_returns_the_newest_lines_not_the_oldest() {
        let mut ring = ring(100, 100_000, 128);
        ring.mark_captured();
        for i in 0..10 {
            ring.push_line(&format!("line{i}"));
        }
        let snapshot = ring.snapshot(Some(3), None);
        assert_eq!(lines(&snapshot), vec!["line7", "line8", "line9"]);
    }

    #[test]
    fn a_caller_line_limit_keeps_the_boundary_before_the_selected_line() {
        let mut ring = ring(100, 100_000, 128);
        ring.mark_captured();
        ring.push_line("before restart");
        ring.push_process_start();
        ring.push_line("after restart");

        let snapshot = ring.snapshot(Some(1), None);
        assert_eq!(
            snapshot.entries,
            vec![
                TailEntry::ProcessStart,
                TailEntry::Line {
                    text: "after restart".to_string(),
                    truncated: false,
                },
            ]
        );
    }

    #[test]
    fn a_caller_line_limit_omits_a_trailing_boundary_after_the_selected_line() {
        let mut ring = ring(100, 100_000, 128);
        ring.mark_captured();
        ring.push_line("before restart");
        ring.push_process_start();

        let snapshot = ring.snapshot(Some(1), None);
        assert_eq!(
            snapshot.entries,
            vec![TailEntry::Line {
                text: "before restart".to_string(),
                truncated: false,
            }]
        );
    }

    #[test]
    fn a_caller_limit_reports_what_it_withheld_rather_than_looking_complete() {
        let mut ring = ring(100, 100_000, 128);
        ring.mark_captured();
        for i in 0..10 {
            ring.push_line(&format!("line{i}"));
        }
        // Nothing was evicted; the narrowing is the caller's own. It still has to
        // be reported, or a 3-line request reads as a module that wrote 3 lines.
        assert_eq!(ring.snapshot(Some(3), None).dropped_lines, 7);
        assert_eq!(ring.snapshot(None, None).dropped_lines, 0);
    }

    #[test]
    fn a_caller_limit_cannot_widen_the_rings_own_caps() {
        let mut ring = ring(2, 10_000, 128);
        ring.mark_captured();
        for i in 0..5 {
            ring.push_line(&format!("line{i}"));
        }
        let snapshot = ring.snapshot(Some(1000), Some(1_000_000));
        assert_eq!(lines(&snapshot).len(), 2);
    }

    fn shared(max_lines: usize, max_bytes: usize, max_line_bytes: usize) -> Arc<Mutex<StderrRing>> {
        Arc::new(Mutex::new(ring(max_lines, max_bytes, max_line_bytes)))
    }

    /// Records each forwarded write separately, so a test can tell one write of
    /// `b"abc\n"` from two writes of `b"abc"` and `b"\n"`.
    #[derive(Default)]
    struct RecordingSink {
        writes: Vec<Vec<u8>>,
    }

    impl OutputSink for RecordingSink {
        fn write_line(&mut self, line: &[u8]) {
            self.writes.push(line.to_vec());
        }
    }

    /// Yields predetermined chunks, one per read, so a test controls exactly
    /// where the byte stream is split.
    struct ChunkedReader {
        chunks: VecDeque<Vec<u8>>,
    }

    impl AsyncRead for ChunkedReader {
        fn poll_read(
            mut self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            buf: &mut ReadBuf<'_>,
        ) -> Poll<io::Result<()>> {
            match self.chunks.pop_front() {
                None => Poll::Ready(Ok(())),
                Some(chunk) => {
                    buf.put_slice(&chunk);
                    Poll::Ready(Ok(()))
                }
            }
        }
    }

    struct FailingReader {
        bytes: Vec<u8>,
        emitted: bool,
    }

    impl AsyncRead for FailingReader {
        fn poll_read(
            mut self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            buf: &mut ReadBuf<'_>,
        ) -> Poll<io::Result<()>> {
            if self.emitted {
                return Poll::Ready(Err(io::Error::other("reader failed")));
            }
            self.emitted = true;
            buf.put_slice(&self.bytes);
            Poll::Ready(Ok(()))
        }
    }

    #[tokio::test]
    async fn the_pump_splits_on_newlines_and_keeps_a_trailing_fragment() {
        let ring = shared(10, 10_000, 128);
        // No trailing newline on the last line: a crashing process routinely dies
        // mid-line, and that fragment is often the message worth reading.
        let source = std::io::Cursor::new(b"one\ntwo\nthree".to_vec());
        let mut sink = RecordingSink::default();
        pump_stderr_into(source, Arc::clone(&ring), &mut sink).await;

        let snapshot = lock_ring(&ring).snapshot(None, None);
        assert_eq!(lines(&snapshot), vec!["one", "two", "three"]);
        assert_eq!(snapshot.capture, CaptureState::Captured);
        assert_eq!(
            sink.writes,
            vec![b"one\n".to_vec(), b"two\n".to_vec(), b"three".to_vec()]
        );
    }

    #[tokio::test]
    async fn a_read_failure_keeps_prior_lines_and_marks_the_capture_incomplete() {
        let ring = shared(10, 10_000, 128);
        let source = FailingReader {
            bytes: b"crash cause\n".to_vec(),
            emitted: false,
        };
        let mut sink = RecordingSink::default();
        pump_stderr_into(source, Arc::clone(&ring), &mut sink).await;

        let snapshot = lock_ring(&ring).snapshot(None, None);
        assert_eq!(lines(&snapshot), vec!["crash cause"]);
        assert!(matches!(
            snapshot.capture,
            CaptureState::Incomplete { ref reason } if reason.contains("reader failed")
        ));
        assert_eq!(sink.writes, vec![b"crash cause\n".to_vec()]);
    }

    #[tokio::test]
    async fn every_captured_line_is_also_forwarded() {
        // Forwarding is not optional. The daemon log is overwhelmingly module
        // output; a tap that captured without forwarding would leave it nearly
        // empty and every existing reader would report clean on nothing.
        let ring = shared(10, 10_000, 128);
        let source = std::io::Cursor::new(b"alpha\nbeta\n".to_vec());
        let mut sink = RecordingSink::default();
        pump_stderr_into(source, Arc::clone(&ring), &mut sink).await;

        assert_eq!(sink.writes, vec![b"alpha\n".to_vec(), b"beta\n".to_vec()]);
    }

    #[tokio::test]
    async fn each_forwarded_line_is_exactly_one_write() {
        // Inheriting the fd gave line atomicity for free. Reading a pipe and
        // re-emitting can split a line that used to be atomic, so the framing
        // must be one syscall per complete line -- asserted as one write per
        // line, not merely as correct bytes.
        let ring = shared(10, 10_000, 128);
        let source = std::io::Cursor::new(b"first\nsecond\nthird\n".to_vec());
        let mut sink = RecordingSink::default();
        pump_stderr_into(source, Arc::clone(&ring), &mut sink).await;

        assert_eq!(sink.writes.len(), 3);
        for write in &sink.writes {
            assert_eq!(
                write.iter().filter(|byte| **byte == b'\n').count(),
                1,
                "a write carried something other than exactly one complete line"
            );
            assert_eq!(*write.last().unwrap(), b'\n');
        }
    }

    #[test]
    fn the_first_process_start_is_not_recorded_because_it_divides_nothing() {
        // Otherwise a module that printed nothing renders as a lone boundary
        // marker, and every caller has to decide whether that counts as silence.
        let mut ring = ring(10, 10_000, 128);
        ring.push_process_start();
        assert!(ring.snapshot(None, None).entries.is_empty());

        ring.push_line("first process said this");
        ring.push_process_start();
        assert!(
            matches!(ring.entries.back(), Some(Slot::ProcessStart { .. })),
            "a boundary with output before it must be recorded"
        );
        ring.push_line("second process said this");
        assert_eq!(
            ring.snapshot(None, None).entries,
            vec![
                TailEntry::Line {
                    text: "first process said this".to_string(),
                    truncated: false
                },
                TailEntry::ProcessStart,
                TailEntry::Line {
                    text: "second process said this".to_string(),
                    truncated: false
                },
            ],
            "only the boundary with output before it may be shown"
        );
    }

    #[test]
    fn a_process_start_is_recorded_when_only_dropped_lines_precede_it() {
        // The ring can be non-empty in the sense that matters -- lines were
        // written and evicted -- while `entries` is empty. Suppressing the
        // boundary there would attribute surviving output to the wrong process.
        let mut ring = ring(1, 10_000, 128);
        ring.push_line("evicted");
        ring.push_line("also evicted");
        // Emptying `entries` by hand must also zero the running totals kept
        // beside it, or the ring holds counts for lines it no longer has and
        // any later eviction decision is made against the stale numbers.
        ring.entries.clear();
        ring.lines = 0;
        ring.bytes = 0;
        ring.push_process_start();
        ring.push_line("survivor");
        assert_eq!(
            ring.snapshot(None, None).entries,
            vec![
                TailEntry::ProcessStart,
                TailEntry::Line {
                    text: "survivor".to_string(),
                    truncated: false
                },
            ]
        );
    }

    fn line(text: &str) -> TailEntry {
        TailEntry::Line {
            text: text.to_string(),
            truncated: false,
        }
    }

    #[test]
    fn a_late_line_from_a_retired_process_lands_in_that_processs_section() {
        // The reader of a process that already exited may deliver its last
        // lines after the next process started. Appending them would put the
        // crash's own explanation under its successor's boundary.
        let mut ring = ring(10, 10_000, 128);
        ring.mark_captured();
        let old = ring.begin_process();
        ring.push_line_from(old, "old: booting");
        ring.retire_pump(old);
        let new = ring.begin_process();
        ring.push_line_from(new, "new: booting");
        ring.push_line_from(old, "old: config error");

        assert_eq!(
            ring.snapshot(None, None).entries,
            vec![
                line("old: booting"),
                line("old: config error"),
                TailEntry::ProcessStart,
                line("new: booting"),
            ]
        );
    }

    #[test]
    fn a_line_from_a_process_that_was_not_retired_is_appended_as_it_arrives() {
        // A swap runs the incumbent alongside its candidate; until the
        // supervisor retires it, the incumbent is live and its lines are news.
        let mut ring = ring(10, 10_000, 128);
        ring.mark_captured();
        let incumbent = ring.begin_process();
        ring.push_line_from(incumbent, "incumbent: before");
        let candidate = ring.begin_process();
        ring.push_line_from(candidate, "candidate: booting");
        ring.push_line_from(incumbent, "incumbent: still serving");

        assert_eq!(
            ring.snapshot(None, None).entries,
            vec![
                line("incumbent: before"),
                TailEntry::ProcessStart,
                line("candidate: booting"),
                line("incumbent: still serving"),
            ]
        );
    }

    #[test]
    fn a_late_line_keeps_its_section_when_the_process_had_printed_nothing_before() {
        // A process whose reader had delivered nothing when its successor
        // started has a boundary with nothing after it. Dropping that boundary
        // as redundant would file the late line under the process before.
        let mut ring = ring(10, 10_000, 128);
        ring.mark_captured();
        let first = ring.begin_process();
        ring.push_line_from(first, "first: done");
        ring.finish_pump(first);
        let old = ring.begin_process();
        ring.retire_pump(old);
        let new = ring.begin_process();
        ring.push_line_from(new, "new: booting");
        ring.push_line_from(old, "old: config error");

        assert_eq!(
            ring.snapshot(None, None).entries,
            vec![
                line("first: done"),
                TailEntry::ProcessStart,
                line("old: config error"),
                TailEntry::ProcessStart,
                line("new: booting"),
            ]
        );
    }

    #[test]
    fn a_late_line_whose_section_was_evicted_counts_as_dropped() {
        let mut ring = ring(2, 10_000, 128);
        ring.mark_captured();
        let old = ring.begin_process();
        ring.push_line_from(old, "old");
        ring.retire_pump(old);
        let new = ring.begin_process();
        for text in ["new 1", "new 2", "new 3"] {
            ring.push_line_from(new, text);
        }
        // The old section and the boundary after it are gone; the late line
        // belongs in front of everything retained.
        ring.push_line_from(old, "old, late");

        let snapshot = ring.snapshot(None, None);
        assert_eq!(snapshot.entries, vec![line("new 2"), line("new 3")]);
        assert_eq!(snapshot.dropped_lines, 3);
    }

    #[test]
    fn a_late_reader_reads_incomplete_until_its_pipe_reaches_eof() {
        let mut ring = ring(10, 10_000, 128);
        ring.mark_captured();
        let old = ring.begin_process();
        ring.retire_pump(old);
        ring.mark_pump_late(old, "still open");
        ring.begin_process();
        assert_eq!(
            ring.snapshot(None, None).capture,
            CaptureState::Incomplete {
                reason: "still open".to_string()
            }
        );

        ring.finish_pump(old);
        assert_eq!(ring.snapshot(None, None).capture, CaptureState::Captured);
    }

    #[test]
    fn silent_restarts_do_not_grow_the_ring() {
        let mut ring = ring(10, 10_000, 128);
        ring.mark_captured();
        ring.push_line("once");
        for _ in 0..100 {
            let generation = ring.begin_process();
            ring.finish_pump(generation);
        }
        assert_eq!(ring.entries.len(), 2);
    }

    #[tokio::test]
    async fn the_pump_marks_captured_even_when_the_module_writes_nothing() {
        // Clean EOF with no output is a module that was quiet, not one nobody
        // listened to -- and the two must not render alike.
        let ring = shared(10, 10_000, 128);
        let source = std::io::Cursor::new(Vec::new());
        let mut sink = RecordingSink::default();
        pump_stderr_into(source, Arc::clone(&ring), &mut sink).await;

        let snapshot = lock_ring(&ring).snapshot(None, None);
        assert!(snapshot.entries.is_empty());
        assert_eq!(snapshot.capture, CaptureState::Captured);
        assert!(sink.writes.is_empty());
    }

    #[tokio::test]
    async fn a_line_with_no_newline_cannot_grow_the_reader_without_bound() {
        // A module fault must not become a daemon fault: without the pending
        // ceiling this buffer grows to whatever the module writes.
        let ring = shared(10, 10_000_000, 4 * 1024 * 1024);
        let source = std::io::Cursor::new(vec![b'x'; MAX_PENDING_LINE_BYTES + 4096]);
        let mut sink = RecordingSink::default();
        pump_stderr_into(source, Arc::clone(&ring), &mut sink).await;

        let snapshot = lock_ring(&ring).snapshot(None, None);
        assert_eq!(
            lines(&snapshot).len(),
            2,
            "expected a forced flush at the ceiling plus the remainder"
        );
        assert_eq!(
            sink.writes,
            vec![vec![b'x'; MAX_PENDING_LINE_BYTES], vec![b'x'; 4096],],
            "forced flushes and EOF fragments must not invent delimiters"
        );
    }

    #[tokio::test]
    async fn boundaries_truncation_and_framing_do_not_depend_on_chunk_splits() {
        // The same stream split at hostile boundaries -- mid-line, between a CR
        // and its LF, and a line sitting exactly on the per-line cap -- must
        // produce the same ring entries and forwarded bytes as any other split.
        let ring = shared(100, 100_000, 8);
        let source = ChunkedReader {
            chunks: vec![
                b"fir".to_vec(),
                b"st\nsec".to_vec(),
                b"ond\ncarry\r".to_vec(),
                b"\nover\n".to_vec(),
                b"12345678\n".to_vec(),
                b"1234567".to_vec(),
                b"89\n".to_vec(),
                b"tail".to_vec(),
            ]
            .into_iter()
            .collect(),
        };
        let mut sink = RecordingSink::default();
        pump_stderr_into(source, Arc::clone(&ring), &mut sink).await;

        let snapshot = lock_ring(&ring).snapshot(None, None);
        assert_eq!(snapshot.capture, CaptureState::Captured);
        assert_eq!(
            snapshot.entries,
            vec![
                TailEntry::Line {
                    text: "first".to_string(),
                    truncated: false
                },
                TailEntry::Line {
                    text: "second".to_string(),
                    truncated: false
                },
                // The pump delimits on '\n' alone; a CR belongs to the line body.
                TailEntry::Line {
                    text: "carry\r".to_string(),
                    truncated: false
                },
                TailEntry::Line {
                    text: "over".to_string(),
                    truncated: false
                },
                // Exactly at the per-line cap: kept whole.
                TailEntry::Line {
                    text: "12345678".to_string(),
                    truncated: false
                },
                // One byte past the cap: cut, and marked as cut.
                TailEntry::Line {
                    text: "12345678".to_string(),
                    truncated: true
                },
                TailEntry::Line {
                    text: "tail".to_string(),
                    truncated: false
                },
            ]
        );
        assert_eq!(
            sink.writes,
            vec![
                b"first\n".to_vec(),
                b"second\n".to_vec(),
                b"carry\r\n".to_vec(),
                b"over\n".to_vec(),
                b"12345678\n".to_vec(),
                b"123456789\n".to_vec(),
                b"tail".to_vec(),
            ]
        );
    }

    #[tokio::test]
    async fn a_line_with_no_newline_is_not_rescanned_from_byte_zero_on_every_chunk() {
        // One 1 MiB line arrives in 8192-byte reads. Searching the whole
        // pending buffer for a newline on every chunk scans each byte once per
        // chunk that arrived after it -- about 64 MiB examined per MiB of
        // output. Searching only the bytes that arrived since the last search
        // scans each byte once.
        let input = vec![b'x'; MAX_PENDING_LINE_BYTES + 4096];
        let ring = shared(10, 10_000_000, 4 * 1024 * 1024);
        let source = std::io::Cursor::new(input.clone());
        let mut sink = RecordingSink::default();

        take_scanned_bytes();
        pump_stderr_into(source, Arc::clone(&ring), &mut sink).await;
        let scanned = take_scanned_bytes();

        assert!(
            scanned <= 2 * input.len(),
            "newline searches examined {scanned} bytes for {} bytes of input; \
             each chunk must search only newly arrived bytes",
            input.len()
        );
    }

    #[test]
    fn a_byte_limit_smaller_than_one_line_still_returns_that_line() {
        // Returning nothing would be indistinguishable from a quiet module, which
        // is the failure this module exists to prevent.
        let mut ring = ring(10, 10_000, 128);
        ring.mark_captured();
        ring.push_line("a line considerably longer than the request limit");
        let snapshot = ring.snapshot(None, Some(4));
        assert_eq!(snapshot.entries.len(), 1);
    }

    #[test]
    fn an_incoherent_config_clamps_the_line_cap_and_keeps_its_restart_boundary() {
        let config = StderrTailConfig::new(2, 10, 100);
        assert_eq!(config.max_line_bytes, config.max_bytes);
        let mut ring = StderrRing::new(config);
        ring.mark_captured();
        ring.push_line("old");
        ring.push_process_start();
        ring.push_line("new process line longer than the ring byte cap");

        assert_eq!(
            ring.snapshot(None, None).entries,
            vec![
                TailEntry::ProcessStart,
                TailEntry::Line {
                    text: "new proces".to_string(),
                    truncated: true,
                },
            ]
        );
    }
}