gunnar_ssh/

pipeline.rs

//! **The standing pipeline: warm machinery per class, and an operation is data.**
//!
//! # What this replaces
//!
//! Today a git operation builds its own world. It spawns what it needs,
//! allocates what it needs, opens what it needs, and throws all of it away when
//! the client disconnects. That is per-request *initialisation*, and it is the
//! largest avoidable cost in the server:
//!
//! MEASURED on the reference desktop 2026-08-14, a `linux.git` clone —
//!
//! | | |
//! |---|---:|
//! | peak RSS, gunnar | **2314 MB** |
//! | peak RSS, gitea, identical clone and identical wire bytes | 122 MB |
//! | cycles spent copying bytes that never needed to move | **~20%** |
//! | `kernel_init_pages` — the kernel ZEROING pages for per-op allocations | 4.7% of the top profile |
//!
//! (`vs_forge_clone`; and 39
//! `perf record` captures of `gunnar serve` during the mirror clone.)
//!
//! A pipeline that is already standing pays none of that. The threads exist, the
//! buffers exist, the archive is already mapped; a request adds a **descriptor**
//! to a queue and a worker that is already awake picks it up.
//!
//! ```text
//!   at BOOT, once, for the life of the process:
//!     pipeline[LsRemote] ─┐
//!     pipeline[Fetch]     ├─ N workers parked on a condvar │ arena each
//!     pipeline[Clone]     │
//!     pipeline[Push]      │
//!     pipeline[Gc]       ─┘
//!
//!   per request:
//!     classify ─► admission gate ─► submit(descriptor) ─► a parked worker wakes
//!                      │
//!                      └─ DENY "server busy", before a single allocation
//! ```
//!
//! # An operation is DATA, not a program
//!
//! Two `Clone` operations do not start two clone engines. They are two
//! descriptors in one queue, drained by the fan that is already running. This is
//! the distinction the whole design rests on, and it is why `submit` takes a
//! value rather than a closure that captures a world.
//!
//! # Why a condvar and not a channel crate
//!
//! No new dependency, and the shape is the point: a worker parked in
//! [`Condvar::wait`] is a thread in `futex_do_wait`, which is exactly what a
//! hang dump of the live server already showed — **42 threads, 40 of them
//! parked on 40 distinct futexes**. The warm fan is not a new idea being
//! introduced here; it already exists in the process. What is missing is
//! anything handing it work, which is what this module is.
//!
//! # What this module does NOT do
//!
//! It does not make one operation faster. A single `linux.git` clone gets one
//! slot and one worker, and it will take exactly as long as it took before —
//! the 25 minutes measured for that clone is serial work *inside* the operation,
//! and only the engine's own fan-out (znippy's `emit_pack` phase 1) touches
//! that. This module is concurrency **across** operations. Conflating the two
//! is how a "pipeline" gets built and then measured as having changed nothing.

use std::collections::VecDeque;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::thread::JoinHandle;

use crate::governor::OpClass;

/// How many workers a class's fan is built with.
///
/// Deliberately per-class: `LsRemote` is answered in milliseconds and wants
/// enough workers that a burst never queues, while `Gc` is long, rare, and
/// wants at most a couple so that maintenance cannot crowd out serving.
pub fn default_workers(class: OpClass, cores: usize) -> usize {
    match class {
        // ── I/O-BOUND, so sized for CONCURRENCY and not for cores ───────────
        //
        // A git transfer spends its life waiting on a socket, not computing.
        // Sizing these fans as if they were CPU-bound is a mistake I made and
        // the bench caught: at `(cores/2).clamp(2,8)` a 32-core box served
        // `multiuser_scaling`'s 102 concurrent clients with EIGHT workers, and
        // **16 of 102 clients did not complete** — not refused, which they
        // would have understood, but queued behind eight streaming packs until
        // they timed out. Zero `transfer.refused` in the whole run, which is
        // what made it look like a mystery rather than a queue.
        //
        // That was strictly worse than what it replaced: tokio's blocking pool
        // would have spawned up to 512 threads and served all 102.
        //
        // Shedding overload is the ADMISSION GATE's job — CoDel sees the queue
        // standing and refuses, in the client's own protocol. The fan's job is
        // only to stop unbounded thread growth. So it is wide enough that
        // ordinary concurrency is never artificially serialised, and the
        // governor decides when there is too much.
        OpClass::LsRemote => (cores * 4).clamp(16, 256),
        OpClass::Fetch => (cores * 4).clamp(16, 256),
        OpClass::Clone => (cores * 4).clamp(16, 256),
        OpClass::Push => (cores * 4).clamp(16, 256),
        // Maintenance yields to serving, and IS bounded by CPU: a repack is
        // real work, not a stream. Two, so one long repack cannot crowd out the
        // other repositories.
        OpClass::Gc => 2,
    }
}

/// What a submission was refused for. Both are "server busy" to the client; the
/// distinction is for the operator reading logs.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SubmitError {
    /// The class's queue is full. Backpressure working as designed.
    QueueFull,
    /// The pipeline is shutting down and will not start new work.
    ShuttingDown,
}

impl std::fmt::Display for SubmitError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            SubmitError::QueueFull => {
                f.write_str("server busy: the queue for this operation class is full")
            }
            SubmitError::ShuttingDown => f.write_str("server busy: shutting down"),
        }
    }
}

struct Queue<T> {
    items: Mutex<VecDeque<T>>,
    /// Workers park here. A parked worker is a thread in `futex_do_wait`.
    ready: Condvar,
    stopping: AtomicBool,
    depth: usize,
    /// Test-only: runs in [`Queue::pop_blocking`] with the lock held, after `stopping` was
    /// read and before the worker waits — the window a lost wake-up lives in.
    #[cfg(test)]
    before_wait: std::sync::OnceLock<Box<dyn Fn() + Send + Sync>>,
}

impl<T> Queue<T> {
    fn push(&self, item: T) -> Result<usize, SubmitError> {
        let mut q = self.items.lock().expect("pipeline queue poisoned");
        // Read under the lock `shutdown` sets it under: nothing is queued after the
        // workers were told to stop.
        if self.stopping.load(Ordering::Acquire) {
            return Err(SubmitError::ShuttingDown);
        }
        if q.len() >= self.depth {
            return Err(SubmitError::QueueFull);
        }
        q.push_back(item);
        let queued = q.len();
        drop(q);
        // One waiter, not all: waking every worker for one item is a thundering
        // herd that costs more than the work.
        self.ready.notify_one();
        Ok(queued)
    }

    fn pop_blocking(&self) -> Option<T> {
        let mut q = self.items.lock().expect("pipeline queue poisoned");
        loop {
            if let Some(item) = q.pop_front() {
                return Some(item);
            }
            if self.stopping.load(Ordering::Acquire) {
                return None;
            }
            #[cfg(test)]
            if let Some(hook) = self.before_wait.get() {
                hook();
            }
            q = self.ready.wait(q).expect("pipeline queue poisoned");
        }
    }
}

/// One class's standing fan: N workers, built once, parked until there is work.
pub struct Pipeline<T: Send + 'static> {
    class: OpClass,
    queue: Arc<Queue<T>>,
    workers: Mutex<Vec<JoinHandle<()>>>,
    /// Operations that have been handed to a worker. Applied output: it is the
    /// only thing that distinguishes "the fan ran it" from "the fan exists".
    served: Arc<AtomicU64>,
}

impl<T: Send + 'static> Pipeline<T> {
    /// Build the fan and START it. Threads are spawned HERE, at boot — never on
    /// the request path. That is the entire point of the type.
    pub fn start<F>(class: OpClass, workers: usize, depth: usize, run: F) -> Pipeline<T>
    where
        F: Fn(T) + Send + Sync + 'static,
    {
        let queue = Arc::new(Queue {
            items: Mutex::new(VecDeque::with_capacity(depth)),
            ready: Condvar::new(),
            stopping: AtomicBool::new(false),
            depth,
            #[cfg(test)]
            before_wait: std::sync::OnceLock::new(),
        });
        let served = Arc::new(AtomicU64::new(0));
        let run = Arc::new(run);

        let mut handles = Vec::with_capacity(workers);
        for i in 0..workers {
            let q = Arc::clone(&queue);
            let run = Arc::clone(&run);
            let served = Arc::clone(&served);
            let name = format!("gunnar-{class:?}-{i}").to_lowercase();
            let h = std::thread::Builder::new()
                .name(name)
                .spawn(move || {
                    while let Some(item) = q.pop_blocking() {
                        served.fetch_add(1, Ordering::Relaxed);
                        run(item);
                    }
                })
                .expect("spawning a pipeline worker at boot");
            handles.push(h);
        }
        Pipeline {
            class,
            queue,
            workers: Mutex::new(handles),
            served,
        }
    }

    /// **The same fan, but each worker owns an ARENA built once at boot.**
    ///
    /// This is the half of "zero initialisation" the plain [`start`](Self::start)
    /// does not give you. A standing fan removes the thread spawn; it does not
    /// remove the buffer. If the work still does `Vec::with_capacity` per
    /// operation, the kernel still zeroes fresh pages for it — which is
    /// `kernel_init_pages`, **4.7% of the top profile** on a `linux.git` clone,
    /// the single largest kernel cost measured there.
    ///
    /// `make` runs **once per worker, at boot**. `run` receives `&mut A`, so the
    /// buffer that was grown for the first operation is still there, at capacity,
    /// for the ten-thousandth. The caller is responsible for `clear()`-ing
    /// rather than reallocating — `clear` keeps capacity, which is the whole
    /// point, and is why this takes `&mut A` instead of handing back an owned
    /// buffer per call.
    #[allow(clippy::type_complexity)]
    pub fn start_with_arena<A, M, F>(
        class: OpClass,
        workers: usize,
        depth: usize,
        make: M,
        run: F,
    ) -> Pipeline<T>
    where
        A: Send + 'static,
        M: Fn(usize) -> A + Send + Sync + 'static,
        F: Fn(&mut A, T) + Send + Sync + 'static,
    {
        let queue = Arc::new(Queue {
            items: Mutex::new(VecDeque::with_capacity(depth)),
            ready: Condvar::new(),
            stopping: AtomicBool::new(false),
            depth,
            #[cfg(test)]
            before_wait: std::sync::OnceLock::new(),
        });
        let served = Arc::new(AtomicU64::new(0));
        let run = Arc::new(run);
        let make = Arc::new(make);

        let mut handles = Vec::with_capacity(workers);
        for i in 0..workers {
            let q = Arc::clone(&queue);
            let run = Arc::clone(&run);
            let make = Arc::clone(&make);
            let served = Arc::clone(&served);
            let name = format!("gunnar-{class:?}-{i}").to_lowercase();
            let h = std::thread::Builder::new()
                .name(name)
                .spawn(move || {
                    // ONCE, at boot, before this worker takes any work.
                    let mut arena = make(i);
                    while let Some(item) = q.pop_blocking() {
                        served.fetch_add(1, Ordering::Relaxed);
                        run(&mut arena, item);
                    }
                })
                .expect("spawning a pipeline worker at boot");
            handles.push(h);
        }
        Pipeline {
            class,
            queue,
            workers: Mutex::new(handles),
            served,
        }
    }

    /// Hand an operation to the standing fan. Allocates nothing but the queue
    /// slot; spawns nothing.
    pub fn submit(&self, item: T) -> Result<usize, SubmitError> {
        self.queue.push(item)
    }

    /// The class this fan serves.
    pub fn class(&self) -> OpClass {
        self.class
    }

    /// Operations a worker has actually picked up.
    pub fn served(&self) -> u64 {
        self.served.load(Ordering::Relaxed)
    }

    /// Items waiting for a worker.
    pub fn queued(&self) -> usize {
        self.queue
            .items
            .lock()
            .expect("pipeline queue poisoned")
            .len()
    }

    /// Threads standing in this fan.
    pub fn workers(&self) -> usize {
        self.workers
            .lock()
            .expect("pipeline workers poisoned")
            .len()
    }

    /// Stop accepting work, drain what is queued, and JOIN every worker.
    ///
    /// Joining rather than detaching: a worker that outlives the pipeline holds
    /// an `Arc` on the queue and on whatever the closure captured, which on a
    /// server means a repository handle that never closes.
    pub fn shutdown(&self) {
        // `stopping` is set UNDER the queue's lock: a worker reads it with the lock held
        // and waits (releasing the lock atomically), so the store lands either before its
        // read (it returns) or after it waits (the notify wakes it). Set without the lock,
        // a store + notify between the read and the wait was lost and join() hung for good
        // (stinkies ssh-pipeline-shutdown-hang, 33 min at load ~35).
        {
            let _q = self.queue.items.lock().expect("pipeline queue poisoned");
            self.queue.stopping.store(true, Ordering::Release);
        }
        self.queue.ready.notify_all();
        let mut handles = self.workers.lock().expect("pipeline workers poisoned");
        for h in handles.drain(..) {
            let _ = h.join();
        }
    }
}

impl<T: Send + 'static> Drop for Pipeline<T> {
    fn drop(&mut self) {
        self.shutdown();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashSet;
    use std::sync::mpsc;
    use std::time::Duration;

    /// ★ THREADS ARE SPAWNED AT BOOT, NOT PER OPERATION.
    ///
    /// The whole claim of this module in one guard: submit N operations and the
    /// set of threads that ran them is bounded by the fan's size, not by N.
    ///
    /// Seen RED by removing the standing fan entirely — `start` spawns nothing
    /// and `submit` runs the work on a fresh thread, which is the shape this
    /// module replaces: **"12 operations ran on 12 distinct threads — the fan is
    /// being built per request"**.
    ///
    /// A FIRST attempt at that break spawned an idle thread per submit while
    /// leaving the work on the fan, and the guard stayed green — correctly, and
    /// it is recorded here because it is the trap: breaking something *near* the
    /// property is not breaking the property.
    #[test]
    fn operations_reuse_the_standing_fan_rather_than_a_thread_each() {
        let (tx, rx) = mpsc::channel();
        let p = Pipeline::start(OpClass::Clone, 3, 64, move |_op: u32| {
            let id = std::thread::current().id();
            tx.send(id).expect("collector alive");
        });
        for i in 0..12 {
            p.submit(i).expect("queue has room");
        }
        let mut seen: HashSet<std::thread::ThreadId> = HashSet::new();
        for _ in 0..12 {
            seen.insert(rx.recv_timeout(Duration::from_secs(5)).expect("op ran"));
        }
        assert!(
            seen.len() <= 3,
            "{} operations ran on {} distinct threads — the fan is being built per request",
            12,
            seen.len()
        );
        assert_eq!(p.served(), 12);
    }

    /// The workers exist BEFORE any operation arrives. A fan that is created
    /// lazily on first submit is not standing, it is merely cached.
    ///
    /// Seen RED by the same break: **"the fan has 0 workers before any
    /// operation was submitted"**.
    #[test]
    fn the_fan_is_running_before_the_first_operation() {
        let p: Pipeline<u32> = Pipeline::start(OpClass::Fetch, 4, 16, |_| {});
        assert_eq!(
            p.workers(),
            4,
            "the fan has {} workers before any operation was submitted",
            p.workers()
        );
        assert_eq!(p.served(), 0, "nothing has been served yet");
    }

    /// ★ A FULL QUEUE REFUSES. Backpressure is the point of a bounded queue; an
    /// unbounded one converts overload into unbounded memory and a slower
    /// collapse.
    ///
    /// Seen RED by removing the `q.len() >= self.depth` check: **"the queue
    /// accepted 40 items into a 4-deep queue"**.
    #[test]
    fn a_full_queue_refuses_rather_than_growing() {
        // A worker that blocks, so the queue genuinely fills.
        // The worker closure must be `Sync`, and a `Receiver` is not — so it
        // is shared behind a mutex. One worker, so there is no contention.
        let (release, held) = mpsc::channel::<()>();
        let held = Arc::new(Mutex::new(held));
        let p = Pipeline::start(OpClass::Clone, 1, 4, move |_op: u32| {
            // BOUNDED, and that is not tidiness. A bare `recv()` here deadlocks
            // the whole test binary when an assertion fires before the release
            // loop: the worker blocks forever, `Drop` calls `shutdown()`, which
            // JOINS it, and nothing ever returns. Seen — the red proof for the
            // guard below hung instead of failing, which is strictly worse than
            // a failure because CI reports it as a timeout with no message.
            let _ = held
                .lock()
                .expect("held")
                .recv_timeout(Duration::from_secs(10));
        });
        let mut accepted = 0usize;
        let mut refused = 0usize;
        for i in 0..40 {
            match p.submit(i) {
                Ok(_) => accepted += 1,
                Err(SubmitError::QueueFull) => refused += 1,
                Err(e) => panic!("unexpected refusal: {e}"),
            }
        }
        assert!(
            refused > 0,
            "the queue accepted {accepted} items into a 4-deep queue"
        );
        assert!(
            accepted <= 4 + 1,
            "accepted {accepted}, which is more than the depth plus the one in service"
        );
        for _ in 0..accepted {
            let _ = release.send(());
        }
    }

    /// Shutdown JOINS. A detached worker holds an `Arc` on whatever the closure
    /// captured — on a server, a repository handle that never closes.
    ///
    /// Seen RED by replacing `h.join()` with `drop(h)`: the guard below observed
    /// **"a worker outlived the pipeline: 3 strong refs on the queue after
    /// shutdown"**.
    #[test]
    fn shutdown_joins_every_worker() {
        let p: Pipeline<u32> = Pipeline::start(OpClass::Gc, 3, 8, |_| {});
        let q = Arc::clone(&p.queue);
        assert!(
            Arc::strong_count(&q) > 1,
            "workers hold the queue while running"
        );
        p.shutdown();
        assert_eq!(p.workers(), 0, "handles were not drained");
        assert_eq!(
            Arc::strong_count(&q),
            2,
            "a worker outlived the pipeline: {} strong refs on the queue after shutdown",
            Arc::strong_count(&q)
        );
    }

    /// **A shutdown that lands between a worker's `stopping` check and its wait is not
    /// lost** (stinkies ssh-pipeline-shutdown-hang: the suite at load ~35 hung 33 min in
    /// `shutdown_joins_every_worker`, the worker parked in `futex_do_wait` for good). The
    /// worker is held in exactly that window (lock held, `stopping` read false); shutdown
    /// is asked meanwhile; the worker then waits. Shutdown must still end.
    #[test]
    fn a_shutdown_in_the_window_before_a_worker_waits_still_joins_it() {
        let p: Pipeline<u32> = Pipeline::start(OpClass::Gc, 1, 8, |_| {});
        let (in_window, window) = mpsc::channel::<()>();
        let (go, gate) = mpsc::channel::<()>();
        let gate = Mutex::new(Some(gate));
        let in_window = Mutex::new(in_window);
        let _ = p.queue.before_wait.set(Box::new(move || {
            // Only the first pass after the item is held; every later wait goes straight on.
            if let Some(g) = gate.lock().unwrap().take() {
                let _ = in_window.lock().unwrap().send(());
                let _ = g.recv_timeout(Duration::from_secs(10));
            }
        }));
        p.submit(1).expect("queued");
        window
            .recv_timeout(Duration::from_secs(10))
            .expect("the worker reached the window");
        let (done, ended) = mpsc::channel::<()>();
        std::thread::spawn(move || {
            p.shutdown();
            let _ = done.send(());
        });
        // Let the shutdown run its course (or block, if it must take the lock) while the
        // worker is still in the window, then let the worker go on to wait.
        std::thread::sleep(Duration::from_millis(300));
        let _ = go.send(());
        assert!(
            ended.recv_timeout(Duration::from_secs(10)).is_ok(),
            "shutdown never joined: the worker waits on a wake-up that came before it waited"
        );
    }

    /// ★ THE ARENA IS BUILT ONCE PER WORKER, NOT ONCE PER OPERATION.
    ///
    /// A standing fan removes the thread spawn but not the buffer. This asserts
    /// the buffer survives between operations at full capacity, which is what
    /// makes `kernel_init_pages` — 4.7% of the top profile on a `linux.git`
    /// clone — stop happening rather than merely shrink.
    ///
    /// Seen RED by moving the allocation inside `run` (`let mut arena =
    /// make(i);` per item, the shape this replaces): **"the arena was rebuilt:
    /// 20 constructions for 20 operations across 2 workers"**.
    #[test]
    fn the_arena_is_built_once_per_worker_and_reused_by_every_operation() {
        let built = Arc::new(AtomicU64::new(0));
        let b = Arc::clone(&built);
        let (tx, rx) = mpsc::channel();
        let p = Pipeline::start_with_arena(
            OpClass::Clone,
            2,
            64,
            move |_i| {
                b.fetch_add(1, Ordering::Relaxed);
                Vec::<u8>::with_capacity(4096)
            },
            move |arena: &mut Vec<u8>, _op: u32| {
                let before = arena.capacity();
                arena.clear();
                arena.extend_from_slice(&[7u8; 1024]);
                tx.send((before, arena.capacity()))
                    .expect("collector alive");
            },
        );
        for i in 0..20 {
            p.submit(i).expect("queue has room");
        }
        let mut reused = 0usize;
        for _ in 0..20 {
            let (before, _after) = rx.recv_timeout(Duration::from_secs(5)).expect("op ran");
            if before >= 4096 {
                reused += 1;
            }
        }
        // `<= 2`, not `== 2`: a worker builds its arena when the THREAD starts,
        // and 20 fast operations can be drained by the first worker before the
        // second is even scheduled. Asserting `== 2` made this guard flake on a
        // loaded box — the property is "not one per operation", and pinning it
        // to an exact thread count tests the scheduler instead.
        let n = built.load(Ordering::Relaxed);
        assert!(
            (1..=2).contains(&n),
            "the arena was rebuilt: {n} constructions for 20 operations across 2 workers"
        );
        assert_eq!(
            reused, 20,
            "only {reused} of 20 operations saw a warm arena — capacity is being thrown away"
        );
    }

    /// ★ THE TRANSFER FANS ARE SIZED FOR CONCURRENCY, NOT FOR CORES.
    ///
    /// Seen RED — in the BENCH, not here — by `(cores/2).clamp(2,8)`:
    /// **"multiuser_scaling: 16 of 102 user(s) did not complete their ops"**,
    /// with **zero** `transfer.refused` in the entire run. Eight workers, 102
    /// I/O-bound clients, a 64-deep queue: nobody was turned away, they were
    /// made to wait until they timed out. Shedding load is the governor's job;
    /// the fan must not serialise ordinary concurrency on its own.
    ///
    /// `gc` is the exception and stays small: a repack is real CPU work, so it
    /// is bounded by cores and must not crowd out serving.
    #[test]
    fn transfer_fans_are_wide_and_gc_stays_small() {
        for cores in [1usize, 4, 32, 256] {
            assert_eq!(
                default_workers(OpClass::Gc, cores),
                2,
                "gc at {cores} cores"
            );
            for class in [
                OpClass::LsRemote,
                OpClass::Fetch,
                OpClass::Clone,
                OpClass::Push,
            ] {
                let n = default_workers(class, cores);
                assert!(
                    n >= 16,
                    "{class:?} at {cores} cores got {n} workers — 102 concurrent \
                     clients against a fan this small queue until they time out"
                );
                assert!(n <= 256, "{class:?} at {cores} cores is unbounded: {n}");
            }
        }
        // The 32-core box the bench runs on: comfortably over 102 clients.
        assert!(
            default_workers(OpClass::Clone, 32) >= 102,
            "the fan must hold multiuser_scaling's 102 clients without queueing"
        );
    }
}

// ════════════════════════════════════════════════════════════════════════════
// The five fans, built at boot
// ════════════════════════════════════════════════════════════════════════════

/// One unit of work handed to a standing fan.
///
/// A boxed `FnOnce` and not a pure descriptor, and that is a compromise worth
/// naming: a descriptor would be `operate(op)` in its ideal form, but the git
/// transfer it has to carry is an async future over a live socket, and the
/// honest bridge to an OS thread is a closure that owns it. One box per
/// operation is one allocation — against the ~27.6 M this design exists to
/// remove, and against tokio's blocking pool spawning a THREAD for the same
/// work. It is not zero; it is four orders of magnitude off the thing that hurt.
pub type Job = Box<dyn FnOnce() + Send + 'static>;

/// **Every class's fan, standing, for the life of the process.**
///
/// Built once in [`crate::server::SshServer::new`]. A request classifies, passes
/// the governor, and is submitted here; the thread that runs it was already
/// alive and parked before the client connected.
pub struct Pipelines {
    ls_remote: Pipeline<Job>,
    fetch: Pipeline<Job>,
    clone: Pipeline<Job>,
    push: Pipeline<Job>,
    gc: Pipeline<Job>,
}

impl Pipelines {
    /// Build all five fans. Call this at BOOT, once.
    ///
    /// `depth` is per class, not shared: a queue of clones filling up must not
    /// refuse an ls-remote, which is the same head-of-line reasoning the
    /// governor's per-class sojourn windows exist for.
    pub fn start(cores: usize, depth: usize) -> Pipelines {
        fn fan(class: OpClass, cores: usize, depth: usize) -> Pipeline<Job> {
            Pipeline::start(class, default_workers(class, cores), depth, |job: Job| {
                job()
            })
        }
        Pipelines {
            ls_remote: fan(OpClass::LsRemote, cores, depth),
            fetch: fan(OpClass::Fetch, cores, depth),
            clone: fan(OpClass::Clone, cores, depth),
            push: fan(OpClass::Push, cores, depth),
            gc: fan(OpClass::Gc, cores, depth),
        }
    }

    fn of(&self, class: OpClass) -> &Pipeline<Job> {
        match class {
            OpClass::LsRemote => &self.ls_remote,
            OpClass::Fetch => &self.fetch,
            OpClass::Clone => &self.clone,
            OpClass::Push => &self.push,
            OpClass::Gc => &self.gc,
        }
    }

    /// Hand an operation to its class's standing fan.
    pub fn submit(&self, class: OpClass, job: Job) -> Result<usize, SubmitError> {
        self.of(class).submit(job)
    }

    /// Operations this class's fan has picked up. Applied output — the only
    /// thing that says the fan is CARRYING traffic rather than merely existing.
    pub fn served(&self, class: OpClass) -> u64 {
        self.of(class).served()
    }

    /// Items waiting for a worker in this class.
    pub fn queued(&self, class: OpClass) -> usize {
        self.of(class).queued()
    }

    /// Threads standing in this class's fan.
    pub fn workers(&self, class: OpClass) -> usize {
        self.of(class).workers()
    }
}

/// **The process's fans.** One set, for the whole server, forever.
///
/// A global and not a field, for two reasons that only became visible once the
/// SSH path was wired and the HTTP path was not:
///
///  * **Both transports must use the SAME fans.** `gunnar-http` holds its own
///    `Arc<dyn GitTransport>` and calls `serve()` itself, so a set of fans owned
///    by `SshServer` covered SSH transfers and silently missed every HTTP one —
///    which is most of the `population_*` and `multiuser_scaling` load.
///  * **Two listeners must not mean two fans.** Standing machinery sized to the
///    box is the whole point; one set per listener doubles the threads and
///    halves the meaning of the limit.
///
/// Built on first use, sized from the box, and never rebuilt.
static FANS: std::sync::OnceLock<Pipelines> = std::sync::OnceLock::new();

/// The process's standing fans, building them on first call.
pub fn fans() -> &'static Pipelines {
    FANS.get_or_init(|| {
        let cores = std::thread::available_parallelism()
            .map(|n| n.get())
            .unwrap_or(4);
        // Queue depth per class. Deep enough that a burst waits rather than
        // being refused; the governor's CoDel window decides whether that wait
        // is acceptable, not this number.
        Pipelines::start(cores, 64)
    })
}

#[cfg(test)]
mod pipelines_tests {
    use super::*;
    use std::time::Duration;

    /// ★ ALL FIVE FANS ARE STANDING BEFORE ANY REQUEST ARRIVES, and gc's is the
    /// small one.
    ///
    /// Seen RED by building only the class asked for on first submit: **"the
    /// clone fan has 0 workers before any request"**.
    #[test]
    fn every_class_has_a_standing_fan_at_boot() {
        let p = Pipelines::start(32, 16);
        for class in [
            OpClass::LsRemote,
            OpClass::Fetch,
            OpClass::Clone,
            OpClass::Push,
            OpClass::Gc,
        ] {
            assert!(
                p.workers(class) > 0,
                "the {class:?} fan has 0 workers before any request"
            );
            assert_eq!(p.served(class), 0, "{class:?} served something at boot");
        }
        assert_eq!(p.workers(OpClass::Gc), 2, "gc must stay small");
        assert!(
            p.workers(OpClass::LsRemote) > p.workers(OpClass::Gc),
            "ls-remote must out-scale gc"
        );
    }

    /// ★ ONE set of fans for the process, however many listeners ask for them.
    ///
    /// Seen RED by returning `Pipelines::start(..)` fresh on each call instead
    /// of the `OnceLock`: **"two callers got two different fan sets — SSH and
    /// HTTP would each have their own"**.
    #[test]
    fn every_caller_gets_the_same_process_wide_fans() {
        let a = fans();
        let b = fans();
        assert!(
            std::ptr::eq(a, b),
            "two callers got two different fan sets — SSH and HTTP would each have their own"
        );
    }

    /// ★ THE COUNTERS REPORT REAL OCCUPANCY, not a constant.
    ///
    /// `served`, `queued` and `workers` existed from the start and NOTHING read
    /// them, so the only evidence the fans carried traffic was the absence of
    /// refusals — silence, not evidence. Now `server.rs` logs depth on every
    /// operation that actually queues, and a counter that is wired to a log is
    /// a counter that can lie.
    ///
    /// So this asserts the queue is SEEN NON-EMPTY under back-pressure, which
    /// is the only reading that distinguishes "the fan is carrying work" from
    /// "the fan exists". A guard that merely called `queued()` and accepted
    /// whatever came back would pass against a function returning 0 forever —
    /// which is exactly the shape of four dead guards found in this suite
    /// today, including a control satisfied by `0 == 0`.
    ///
    /// Seen RED with `queued()` stubbed to `0`: **"the queue never reported a
    /// waiting item across 24 submissions against 1 worker — the depth this
    /// logs is a constant"**.
    #[test]
    fn the_occupancy_counters_are_not_constants() {
        // One worker, held, so submissions must pile up behind it.
        let (release, held) = std::sync::mpsc::channel::<()>();
        let held = Arc::new(Mutex::new(held));
        let p: Pipeline<u32> = Pipeline::start(OpClass::Clone, 1, 64, move |_| {
            // BOUNDED, and that is not tidiness. A bare `recv()` here deadlocks
            // the whole test binary when an assertion fires before the release
            // loop: the worker blocks forever, `Drop` calls `shutdown()`, which
            // JOINS it, and nothing ever returns. Seen — the red proof for the
            // guard below hung instead of failing, which is strictly worse than
            // a failure because CI reports it as a timeout with no message.
            let _ = held
                .lock()
                .expect("held")
                .recv_timeout(Duration::from_secs(10));
        });

        let mut deepest = 0usize;
        for i in 0..24 {
            if let Ok(depth) = p.submit(i) {
                deepest = deepest.max(depth);
            }
        }
        assert!(
            deepest > 1,
            "the queue never reported a waiting item across 24 submissions against \
             1 worker — the depth this logs is a constant"
        );
        assert!(
            p.queued() > 0,
            "queued() says {} with {deepest} known to have piled up",
            p.queued()
        );
        assert_eq!(p.workers(), 1, "workers() lost count of its own fan");

        // Let them through, then the served counter must MOVE — a `served` that
        // never rises would make the log line above report an idle fan forever.
        for _ in 0..24 {
            let _ = release.send(());
        }
        let mut waited = 0;
        while p.served() < 24 && waited < 200 {
            std::thread::sleep(Duration::from_millis(25));
            waited += 1;
        }
        assert_eq!(
            p.served(),
            24,
            "served() reports {} after 24 jobs were released",
            p.served()
        );
    }

    /// A submitted job RUNS on the fan, and lands in the right class.
    #[test]
    fn a_job_runs_on_its_own_class_fan() {
        let p = Pipelines::start(8, 16);
        let (tx, rx) = std::sync::mpsc::channel();
        p.submit(OpClass::Push, Box::new(move || tx.send(()).unwrap()))
            .expect("queue has room");
        rx.recv_timeout(std::time::Duration::from_secs(5))
            .expect("the job ran on the fan");
        assert_eq!(p.served(OpClass::Push), 1);
        assert_eq!(p.served(OpClass::Clone), 0, "it ran on the wrong fan");
    }
}