liminal-server 0.21.1

//! The console hub the connection services own (SRV-012 R4, R6, R7, R9).
//!
//! It holds the topology, the tap, one tap subscription per
//! `console.tap.<target>` channel with at least one subscriber, and the pump
//! that turns them into records.
//!
//! The pump runs on the publisher's thread and never publishes while it holds
//! a hub lock: it takes what is due under the lock, lets go, publishes, and
//! puts back what failed. A tap with a pending record is not drained, so its
//! ring stays the only buffer and ring eviction the only loss, which the next
//! batch's `dropped` reports. One tap's failure never stops the others: it
//! is returned once every tap has been drained, published and retired. The
//! pending records and unread ring of a tap whose last subscriber left go
//! with it, as do the unsent records of a tap attached anew before they were
//! put back; nobody is left to read them, so they are counted by name in
//! `tap_records_abandoned` and `tap_messages_abandoned`, not in
//! `console_records_lost`.

use std::collections::{BTreeMap, VecDeque};
use std::sync::{Arc, Mutex};
use std::time::Instant;

use super::counts::{ConsoleCounts, ConsoleCountsSnapshot, ConsoleFailure, add, failed, set};
use super::error::ConsoleError;
use super::model::{TopologyChange, TopologyModel};
use super::publisher::ServerPublisher;
use super::records::{canonical_bytes, tap_batch_record};
use super::wall::WallBase;
use crate::tap::error::lock;
use crate::tap::{DEFAULT_TAP_CAPACITY, Tap, TapError, TapFilter, TapSubscription, TapWaker};

/// The topology stream's subject.
pub const TOPOLOGY_SUBJECT: &str = "console.topology";
/// The prefix of every tap stream's subject; the rest names the target.
pub const TAP_SUBJECT_PREFIX: &str = "console.tap.";
/// The most taps live at once (R6); one more is `CONSOLE_TAP_LIMIT_REFUSED`.
pub const MAX_LIVE_TAPS: usize = 256;

#[derive(Debug)]
struct TopologyState {
    model: TopologyModel,
    stale: bool,
    subscribers: usize,
}

#[derive(Debug)]
struct AttachedTap {
    subscription: TapSubscription,
    subscribers: usize,
    pending: VecDeque<Vec<u8>>,
    /// Which attach of this subject it is, so the pump never puts an earlier
    /// tap's unsent records back into a later one (SRV-012 R9).
    generation: u64,
}

#[derive(Debug, Default)]
struct Taps {
    live: BTreeMap<String, AttachedTap>,
    /// New taps admitted and not yet attached, with how many subscribes hold
    /// each: they count against [`MAX_LIVE_TAPS`] from admission on.
    reserved: BTreeMap<String, u64>,
    retiring: Vec<String>,
    /// The next tap's [`AttachedTap::generation`].
    generations: u64,
}

type NotifierSlot = Arc<Mutex<Option<TapWaker>>>;

/// One of the hub's locks, named for a test that poisons it.
#[cfg(test)]
#[derive(Clone, Copy, Debug)]
pub(crate) enum HubLock {
    /// The held topology.
    Topology,
    /// The live taps.
    Taps,
    /// The publisher's notifier slot.
    Notifier,
}

/// What one pump did.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct PumpReport {
    /// `console.topology` records published.
    pub topology_records: u64,
    /// `console.tap.<target>` records published.
    pub tap_records: u64,
    /// Publishes that failed; each record is kept for the next pump.
    pub publish_failures: u64,
    /// Tap channels removed from the roster.
    pub retired: u64,
}

/// The console's state, owned by the connection services.
pub struct ConsoleHub {
    device: String,
    wall: WallBase,
    tap: Tap,
    topology: Mutex<TopologyState>,
    taps: Mutex<Taps>,
    notifier: NotifierSlot,
    counts: Arc<ConsoleCounts>,
    /// Every record the publisher took, as the encoder made it. Test-only.
    #[cfg(test)]
    sent: Mutex<Vec<(String, Vec<u8>)>>,
    /// How many topology changes reached [`Self::emit`]. Test-only.
    #[cfg(test)]
    emits: std::sync::atomic::AtomicU64,
    /// How many times the topology lock was taken. Test-only.
    #[cfg(test)]
    topology_locks: std::sync::atomic::AtomicU64,
    /// Notified after each applied topology change. Test-only.
    #[cfg(test)]
    applied: std::sync::Condvar,
    /// A tap subject whose next take fails, as a poisoned ring would.
    /// Test-only.
    #[cfg(test)]
    failing_take: Mutex<Option<String>>,
}

impl std::fmt::Debug for ConsoleHub {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("ConsoleHub")
            .field("device", &self.device)
            .field("tap", &self.tap)
            .finish_non_exhaustive()
    }
}

/// A waker that forwards to whatever notifier is installed when it fires. A
/// poisoned slot is refused, not recovered: the wake is counted as failed.
fn forwarding_waker(slot: NotifierSlot, counts: Arc<ConsoleCounts>) -> TapWaker {
    Arc::new(move || match lock(&slot, "console notifier") {
        Ok(guard) => {
            let notifier = guard.clone();
            drop(guard);
            if let Some(notifier) = notifier {
                notifier();
            }
        }
        Err(_poisoned) => failed(
            &counts.console_wake_failures,
            1,
            ConsoleFailure::ConsoleWake,
        ),
    })
}

/// Whether the tap hook may observe a publish on `subject`: never one under
/// `console.` (the console's own records).
#[must_use]
pub fn tappable(subject: &str) -> bool {
    !crate::server::connection::is_console_subject(subject)
}

impl ConsoleHub {
    /// A hub for `device` holding `model`, timing records from `wall`, with
    /// a tap of the default capacity.
    ///
    /// # Errors
    /// Returns [`ConsoleError::EmptyDevice`] when `device` is blank.
    pub fn new(device: &str, model: TopologyModel, wall: WallBase) -> Result<Self, ConsoleError> {
        Self::with_capacity(device, model, wall, DEFAULT_TAP_CAPACITY)
    }

    /// As [`Self::new`], with each tap subscriber's ring holding `capacity`.
    ///
    /// # Errors
    /// Returns [`ConsoleError::EmptyDevice`] when `device` is blank.
    pub fn with_capacity(
        device: &str,
        model: TopologyModel,
        wall: WallBase,
        capacity: usize,
    ) -> Result<Self, ConsoleError> {
        if device.trim().is_empty() {
            return Err(ConsoleError::EmptyDevice);
        }
        let notifier = NotifierSlot::default();
        let counts = Arc::new(ConsoleCounts::default());
        let depot = model.depot();
        Ok(Self {
            device: device.to_owned(),
            wall,
            tap: Tap::with_waker(
                device,
                capacity,
                forwarding_waker(Arc::clone(&notifier), Arc::clone(&counts)),
            )
            .with_depot(depot),
            topology: Mutex::new(TopologyState {
                model,
                stale: false,
                subscribers: 0,
            }),
            taps: Mutex::new(Taps::default()),
            notifier,
            counts,
            #[cfg(test)]
            sent: Mutex::new(Vec::new()),
            #[cfg(test)]
            emits: std::sync::atomic::AtomicU64::new(0),
            #[cfg(test)]
            topology_locks: std::sync::atomic::AtomicU64::new(0),
            #[cfg(test)]
            applied: std::sync::Condvar::new(),
            #[cfg(test)]
            failing_take: Mutex::new(None),
        })
    }

    /// The device this hub publishes for (`liminal.<device>`).
    #[must_use]
    pub fn device(&self) -> &str {
        &self.device
    }

    /// The tap.
    #[must_use]
    pub const fn tap(&self) -> &Tap {
        &self.tap
    }

    /// Every console counter as it stands.
    #[must_use]
    pub fn counts(&self) -> ConsoleCountsSnapshot {
        self.counts
            .snapshot(self.tap.dropped_total(), self.tap.unsubscribe_failures())
    }

    /// Publishes one encoded record and answers whether the publisher took
    /// it; a refusal is one `console_publish` failure.
    fn send(&self, out: &dyn ServerPublisher, subject: &str, bytes: &[u8]) -> bool {
        let taken = out.publish_console(subject, bytes).is_ok();
        if !taken {
            failed(
                &self.counts.console_publish_failures,
                1,
                ConsoleFailure::ConsolePublish,
            );
        }
        #[cfg(test)]
        if taken && let Ok(mut sent) = self.sent.lock() {
            sent.push((subject.to_owned(), bytes.to_vec()));
        }
        taken
    }

    /// How many topology changes reached [`Self::emit`]. Test-only.
    #[cfg(test)]
    pub(crate) fn emits(&self) -> u64 {
        self.emits.load(std::sync::atomic::Ordering::Relaxed)
    }

    /// Each connection's participant as the model holds it. Test-only.
    #[cfg(test)]
    pub(crate) fn participants(&self) -> Vec<Option<String>> {
        self.topology
            .lock()
            .map(|state| state.model.participants())
            .unwrap_or_default()
    }

    /// How many times the topology lock was taken. Test-only.
    #[cfg(test)]
    pub(crate) fn topology_locks(&self) -> u64 {
        self.topology_locks
            .load(std::sync::atomic::Ordering::Relaxed)
    }

    /// Waits until the model holds `count` connections and answers whether
    /// it does; the guard is a minute, never a deadline. Test-only.
    #[cfg(test)]
    pub(crate) fn wait_for_connections(&self, count: usize) -> bool {
        self.topology
            .lock()
            .ok()
            .and_then(|state| {
                self.applied
                    .wait_timeout_while(state, std::time::Duration::from_secs(60), |state| {
                        state.model.connection_count() != count
                    })
                    .ok()
            })
            .is_some_and(|(_, guard)| !guard.timed_out())
    }

    /// Makes the next take of `subject`'s tap fail once. Test-only.
    #[cfg(test)]
    pub(crate) fn fail_next_take(&self, subject: &str) {
        if let Ok(mut failing) = self.failing_take.lock() {
            *failing = Some(subject.to_owned());
        }
    }

    /// Takes `subscription`'s batches, unless [`Self::fail_next_take`] named
    /// `subject`: that take fails once. Test-only.
    #[cfg(test)]
    fn take_batches(
        &self,
        subject: &str,
        subscription: &TapSubscription,
    ) -> Result<Vec<crate::tap::TapBatch>, TapError> {
        if let Ok(mut failing) = self.failing_take.lock()
            && failing.as_deref() == Some(subject)
        {
            *failing = None;
            return Err(TapError::Poisoned {
                what: "tap subscriber",
            });
        }
        subscription.take_batches()
    }

    /// The topology lock, counted under test.
    fn lock_topology(&self) -> Result<std::sync::MutexGuard<'_, TopologyState>, TapError> {
        #[cfg(test)]
        self.topology_locks
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        lock(&self.topology, "console topology")
    }

    /// This server's incarnation started at `started`; the topology's uptime
    /// runs from there and the next document says so.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the topology lock is poisoned.
    pub fn incarnation_started(&self, started: Instant) -> Result<(), TapError> {
        let mut state = self.lock_topology()?;
        if state.model.incarnation_started(started) {
            state.stale = true;
        }
        drop(state);
        Ok(())
    }

    /// Counts a dedup-suppressed re-publish, which is not tapped.
    pub fn suppressed(&self) {
        add(&self.counts.tap_suppressed_total, 1);
    }

    /// Whether the taps lock is held at this moment. Test-only.
    #[cfg(test)]
    pub(crate) fn taps_held(&self) -> bool {
        self.taps.try_lock().is_err()
    }

    /// How many tap channels hold an admission reservation. Test-only.
    #[cfg(test)]
    pub(crate) fn reserved_taps(&self) -> Option<usize> {
        self.taps.lock().ok().map(|taps| taps.reserved.len())
    }

    /// Every record the publisher took, oldest first, as the encoder made
    /// it; `None` when the capture is poisoned. Test-only.
    #[cfg(test)]
    pub(crate) fn sent(&self) -> Option<Vec<(String, Vec<u8>)>> {
        self.sent.lock().ok().map(|sent| sent.clone())
    }

    /// Poisons one of the hub's locks, as a panic on a thread holding it
    /// would, and answers whether it is now poisoned. Test-only.
    #[cfg(test)]
    pub(crate) fn poison(&self, lock: HubLock) -> bool {
        std::thread::scope(|scope| {
            scope
                .spawn(|| {
                    let _held = (
                        matches!(lock, HubLock::Topology).then(|| self.topology.lock()),
                        matches!(lock, HubLock::Taps).then(|| self.taps.lock()),
                        matches!(lock, HubLock::Notifier).then(|| self.notifier.lock()),
                    );
                    std::panic::resume_unwind(Box::new("poisoned for a test"));
                })
                .join()
                .is_err()
        }) && match lock {
            HubLock::Topology => self.topology.is_poisoned(),
            HubLock::Taps => self.taps.is_poisoned(),
            HubLock::Notifier => self.notifier.is_poisoned(),
        }
    }

    /// Counts a publish the tap could not observe (its lock was poisoned).
    /// The publish itself answered as it would without the console.
    pub fn observe_failed(&self) {
        failed(
            &self.counts.tap_observe_failures,
            1,
            ConsoleFailure::TapObserve,
        );
    }

    /// Installs the function that tells the publisher there is work. The
    /// publisher calls [`Self::pump`] on its own thread when told.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the slot is poisoned.
    pub fn set_notifier(&self, notifier: TapWaker) -> Result<(), TapError> {
        *lock(&self.notifier, "console notifier")? = Some(notifier);
        Ok(())
    }

    fn wake(&self) {
        forwarding_waker(Arc::clone(&self.notifier), Arc::clone(&self.counts))();
    }

    /// Wakes the publisher when work is already waiting: a topology
    /// subscriber, a live tap or a tap to retire. The start of the console
    /// streams calls it, so a subscriber that arrived before the publisher
    /// ran is served, and an idle server is never woken.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when a hub lock is poisoned.
    pub fn wake_if_due(&self) -> Result<(), TapError> {
        let topology = self.lock_topology()?.subscribers > 0;
        let taps = {
            let taps = lock(&self.taps, "console taps")?;
            !taps.live.is_empty() || !taps.retiring.is_empty()
        };
        if topology || taps {
            self.wake();
        }
        Ok(())
    }

    /// Applies `change` to the held topology. A change that alters it makes
    /// the document due, and wakes the publisher when someone subscribes.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the topology lock is poisoned; the
    /// change is counted in `topology_emit_failures`.
    pub fn emit(&self, change: TopologyChange) -> Result<(), TapError> {
        #[cfg(test)]
        self.emits
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let emit_failed = |_: &TapError| {
            failed(
                &self.counts.topology_emit_failures,
                1,
                ConsoleFailure::TopologyEmit,
            );
        };
        // A worker identity or service subject that is this server's exchange
        // or depot id is kept off the map and counted (Buttercup's V1, Hermes
        // 04:27): the registration stands, merged into nothing.
        let colliding = match &change {
            TopologyChange::Worker { identity, .. } => Some(identity),
            TopologyChange::Service {
                subject,
                present: true,
            } => Some(subject),
            _ => None,
        }
        .and_then(|id| self.tap.reserved(id).map(|station| (id, station)));
        if let Some((id, station)) = colliding {
            failed(
                &self.counts.station_collisions,
                1,
                ConsoleFailure::StationCollision,
            );
            tracing::warn!(id = %id, station, "STATION_COLLISION");
        }
        // A registered worker's tapped publishes name its station from here
        // to its connection's close.
        match &change {
            TopologyChange::Worker {
                connection,
                identity,
                ..
            } if colliding.is_none() => self.tap.name_station(*connection, identity),
            TopologyChange::ConnectionClosed { id } => self
                .tap
                .forget_station(*id)
                .and_then(|()| self.tap.forget_receivers_of(*id)),
            _ => Ok(()),
        }
        .inspect_err(emit_failed)?;
        let mut state = self.lock_topology().inspect_err(emit_failed)?;
        let changed = state.model.apply(change);
        if changed {
            state.stale = true;
        }
        let due = changed && state.subscribers > 0;
        drop(state);
        #[cfg(test)]
        self.applied.notify_all();
        if due {
            self.wake();
        }
        Ok(())
    }

    /// Whether a new subscription to `channel` is within the live-tap limit:
    /// always for a channel that is not a tap or whose tap is live.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the tap table is poisoned.
    pub fn tap_room(&self, channel: &str) -> Result<bool, TapError> {
        if !channel.starts_with(TAP_SUBJECT_PREFIX) {
            return Ok(true);
        }
        let taps = lock(&self.taps, "console taps")?;
        Ok(taps.live.contains_key(channel)
            || taps.reserved.contains_key(channel)
            || taps.live.len() + taps.reserved.len() < MAX_LIVE_TAPS)
    }

    /// Admits a subscribe to `channel` against [`MAX_LIVE_TAPS`] and, for a
    /// tap not yet live, reserves its slot in the same lock, so two racing
    /// subscribes cannot both take the last slot. Answers false when no slot
    /// is left. The attach that follows takes the reservation up;
    /// [`Self::release_tap`] gives it back when the subscribe fails.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the tap table is poisoned.
    pub fn reserve_tap(&self, channel: &str) -> Result<bool, TapError> {
        if !channel.starts_with(TAP_SUBJECT_PREFIX) {
            return Ok(true);
        }
        let mut taps = lock(&self.taps, "console taps")?;
        let admitted = if taps.live.contains_key(channel) {
            true
        } else if let Some(holders) = taps.reserved.get_mut(channel) {
            *holders = holders.saturating_add(1);
            true
        } else if taps.live.len() + taps.reserved.len() < MAX_LIVE_TAPS {
            taps.reserved.insert(channel.to_owned(), 1);
            true
        } else {
            false
        };
        drop(taps);
        Ok(admitted)
    }

    /// Gives back one reservation of `channel` for a subscribe that failed
    /// after admission. When it was the last and no tap is live there, the
    /// tap channel admission registered is taken off the roster by `retire`,
    /// under the taps lock, so no subscribe reserving the tap again can see
    /// its entry go: a refused subscribe leaves the roster as it found it. A
    /// failed `retire` queues the subject for the pump; a poisoned table is
    /// counted as a detach failure.
    pub fn release_tap(
        &self,
        channel: &str,
        retire: impl FnOnce(&str) -> Result<(), ConsoleError>,
    ) {
        if !channel.starts_with(TAP_SUBJECT_PREFIX) {
            return;
        }
        let Ok(mut taps) = lock(&self.taps, "console taps") else {
            failed(
                &self.counts.tap_detach_failures,
                1,
                ConsoleFailure::TapDetach,
            );
            return;
        };
        let last = match taps.reserved.get_mut(channel) {
            Some(holders) if *holders > 1 => {
                *holders -= 1;
                false
            }
            Some(_) => {
                taps.reserved.remove(channel);
                true
            }
            None => false,
        };
        if !last || taps.live.contains_key(channel) || retire(channel).is_ok() {
            return;
        }
        taps.retiring.push(channel.to_owned());
        drop(taps);
        failed(
            &self.counts.tap_detach_failures,
            1,
            ConsoleFailure::TapDetach,
        );
        self.wake();
    }

    /// Counts a subscription to `channel`. A new `console.topology`
    /// subscriber makes the latest document due again, so it is sent the
    /// whole topology (R7); the first to a `console.tap.<target>` starts that
    /// target's tap. Any other channel is ignored.
    ///
    /// # Errors
    /// Returns [`ConsoleError::TapLimit`] when a new tap would pass
    /// [`MAX_LIVE_TAPS`], or [`ConsoleError::Tap`] when a lock is poisoned;
    /// either is counted in `tap_attach_failures`.
    pub fn attach(&self, channel: &str) -> Result<(), ConsoleError> {
        if channel == TOPOLOGY_SUBJECT {
            let mut state = self.lock_topology()?;
            state.subscribers = state.subscribers.saturating_add(1);
            state.stale = true;
            drop(state);
            self.wake();
            return Ok(());
        }
        let Some(target) = channel.strip_prefix(TAP_SUBJECT_PREFIX) else {
            return Ok(());
        };
        self.attach_tap(channel, target).inspect_err(|_| {
            failed(
                &self.counts.tap_attach_failures,
                1,
                ConsoleFailure::TapAttach,
            );
        })
    }

    fn attach_tap(&self, channel: &str, target: &str) -> Result<(), ConsoleError> {
        let mut taps = lock(&self.taps, "console taps")?;
        match taps.reserved.get_mut(channel) {
            Some(holders) if *holders > 1 => *holders -= 1,
            Some(_) => {
                taps.reserved.remove(channel);
            }
            None => {}
        }
        if let Some(entry) = taps.live.get_mut(channel) {
            entry.subscribers = entry.subscribers.saturating_add(1);
            return Ok(());
        }
        if taps.live.len() >= MAX_LIVE_TAPS {
            return Err(ConsoleError::TapLimit {
                subject: channel.to_owned(),
                limit: MAX_LIVE_TAPS,
            });
        }
        let subscription = self.tap.subscribe(TapFilter {
            subject: target.to_owned(),
        })?;
        taps.retiring.retain(|subject| subject != channel);
        let generation = taps.generations;
        taps.generations = taps.generations.wrapping_add(1);
        taps.live.insert(
            channel.to_owned(),
            AttachedTap {
                subscription,
                subscribers: 1,
                pending: VecDeque::new(),
                generation,
            },
        );
        set(&self.counts.taps_attached, len_u64(taps.live.len()));
        drop(taps);
        Ok(())
    }

    /// Uncounts a subscription to `channel`. The last one to leave a
    /// `console.tap.<target>` stops that target's tap and queues the channel
    /// for roster removal by the next pump. Safe on the channel actor's
    /// thread: it takes only hub locks and wakes without waiting.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when a lock is poisoned; counted in
    /// `tap_detach_failures`.
    pub fn detach(&self, channel: &str) -> Result<(), TapError> {
        self.detach_inner(channel).inspect_err(|_| {
            failed(
                &self.counts.tap_detach_failures,
                1,
                ConsoleFailure::TapDetach,
            );
        })
    }

    fn detach_inner(&self, channel: &str) -> Result<(), TapError> {
        if channel == TOPOLOGY_SUBJECT {
            let mut state = self.lock_topology()?;
            state.subscribers = state.subscribers.saturating_sub(1);
            drop(state);
            return Ok(());
        }
        if !channel.starts_with(TAP_SUBJECT_PREFIX) {
            return Ok(());
        }
        let mut taps = lock(&self.taps, "console taps")?;
        let last = taps.live.get_mut(channel).is_some_and(|entry| {
            entry.subscribers = entry.subscribers.saturating_sub(1);
            entry.subscribers == 0
        });
        if !last {
            return Ok(());
        }
        // R9: its pending records and unread ring leave with the tap, since
        // nobody is left to read them; each is counted by name.
        if let Some(entry) = taps.live.remove(channel) {
            add(
                &self.counts.tap_records_abandoned,
                len_u64(entry.pending.len()),
            );
            add(
                &self.counts.tap_messages_abandoned,
                len_u64(entry.subscription.held().unwrap_or_default()),
            );
        }
        taps.retiring.push(channel.to_owned());
        set(&self.counts.taps_attached, len_u64(taps.live.len()));
        let pending = pending_total(&taps.live);
        drop(taps);
        set(&self.counts.console_pending_records, pending);
        self.wake();
        Ok(())
    }

    /// Subjects with a live tap, sorted.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the tap table is poisoned.
    pub fn attached_subjects(&self) -> Result<Vec<String>, TapError> {
        Ok(lock(&self.taps, "console taps")?
            .live
            .keys()
            .cloned()
            .collect())
    }

    /// Publishes what is due through `out`: the topology document when it
    /// changed and someone subscribes, then each live tap's records, a
    /// tap's kept records first and its ring drained only when none are
    /// kept; then removes retired tap channels from the roster. A failed
    /// publish keeps that record and the tap's later ones for the next pump,
    /// and the pump goes on to the other taps.
    ///
    /// # Errors
    /// Returns [`ConsoleError`] when a lock is poisoned or a record cannot be
    /// encoded; publish failures are counted and reported, not returned.
    pub fn pump(&self, out: &dyn ServerPublisher) -> Result<PumpReport, ConsoleError> {
        add(&self.counts.console_pump_wakes, 1);
        let mut report = PumpReport::default();
        // The taps pump whatever the topology did, so one failure there
        // never holds the feed.
        let topology = self.pump_topology(out, &mut report);
        let tapped = self.pump_taps(out, &mut report);
        self.retire(out, &mut report)?;
        topology.and(tapped).map(|()| report)
    }

    fn pump_topology(
        &self,
        out: &dyn ServerPublisher,
        report: &mut PumpReport,
    ) -> Result<(), ConsoleError> {
        let mut state = self.lock_topology()?;
        if !state.stale || state.subscribers == 0 {
            return Ok(());
        }
        let (document, fault) = state.model.document(&self.wall);
        state.stale = false;
        drop(state);
        if let Some(error) = fault {
            // Named on the record itself; the pump goes on.
            failed(
                &self.counts.topology_clock_failures,
                1,
                ConsoleFailure::TopologyClock,
            );
            tracing::warn!(%error, "CONSOLE_TOPOLOGY_CLOCK");
        }
        let bytes = canonical_bytes(&document, "topology")?;
        if self.send(out, TOPOLOGY_SUBJECT, &bytes) {
            add(&self.counts.topology_records_published, 1);
            report.topology_records += 1;
        } else {
            report.publish_failures += 1;
            // Due again: the next pump sends the document as it stands then.
            self.lock_topology()?.stale = true;
        }
        Ok(())
    }

    fn pump_taps(
        &self,
        out: &dyn ServerPublisher,
        report: &mut PumpReport,
    ) -> Result<(), ConsoleError> {
        let mut due = Vec::new();
        // R9: one tap's failure is kept and returned after every other tap
        // has been drained and published; it never drops what they took.
        let mut first_error = None;
        let mut taps = lock(&self.taps, "console taps")?;
        for (subject, entry) in &mut taps.live {
            if entry.pending.is_empty() {
                let target = subject.strip_prefix(TAP_SUBJECT_PREFIX).unwrap_or(subject);
                #[cfg(test)]
                let taken = self.take_batches(subject, &entry.subscription);
                #[cfg(not(test))]
                let taken = entry.subscription.take_batches();
                match taken {
                    Ok(batches) => {
                        for batch in batches {
                            let encoded =
                                tap_batch_record(&self.device, target, &batch, &self.wall)
                                    .and_then(|record| canonical_bytes(&record, "tap.batch"));
                            match encoded {
                                Ok(bytes) => entry.pending.push_back(bytes),
                                Err(error) => {
                                    // Taken from the ring and not encodable:
                                    // the one loss besides eviction, counted.
                                    failed(
                                        &self.counts.console_records_lost,
                                        1,
                                        ConsoleFailure::ConsoleRecordsLost,
                                    );
                                    first_error.get_or_insert(error);
                                }
                            }
                        }
                    }
                    Err(error) => {
                        first_error.get_or_insert_with(|| error.into());
                    }
                }
            }
            if !entry.pending.is_empty() {
                due.push((
                    subject.clone(),
                    entry.generation,
                    std::mem::take(&mut entry.pending),
                ));
            }
        }
        drop(taps);

        let mut unsent = Vec::new();
        for (subject, generation, mut records) in due {
            while let Some(bytes) = records.front() {
                if !self.send(out, &subject, bytes) {
                    report.publish_failures += 1;
                    unsent.push((subject, generation, records));
                    break;
                }
                records.pop_front();
                add(&self.counts.tap_records_published, 1);
                report.tap_records += 1;
            }
        }

        let mut taps = lock(&self.taps, "console taps")?;
        for (subject, generation, records) in unsent {
            // Only to the tap they were taken from: one that detached
            // meanwhile, or was attached anew, has no reader for them.
            match taps.live.get_mut(&subject) {
                Some(entry) if entry.generation == generation => entry.pending = records,
                _ => add(&self.counts.tap_records_abandoned, len_u64(records.len())),
            }
        }
        let pending = pending_total(&taps.live);
        drop(taps);
        set(&self.counts.console_pending_records, pending);
        first_error.map_or(Ok(()), Err)
    }

    fn retire(
        &self,
        out: &dyn ServerPublisher,
        report: &mut PumpReport,
    ) -> Result<(), ConsoleError> {
        let mut retiring = std::mem::take(&mut lock(&self.taps, "console taps")?.retiring);
        while let Some(subject) = retiring.first() {
            // The check and the roster removal share the taps lock, taken
            // before the roster's, so no subscribe can reserve or attach the
            // tap between them and find its channel gone (SRV-012 R6).
            let mut taps = lock(&self.taps, "console taps")?;
            // Attached or reserved again since it was queued: it stays.
            if taps.live.contains_key(subject) || taps.reserved.contains_key(subject) {
                drop(taps);
                retiring.remove(0);
                continue;
            }
            if let Err(error) = out.retire_console_channel(subject) {
                // Kept for the next pump unless it was attached again since.
                retiring.retain(|subject| !taps.live.contains_key(subject));
                retiring.append(&mut taps.retiring);
                taps.retiring = retiring;
                drop(taps);
                failed(
                    &self.counts.tap_detach_failures,
                    1,
                    ConsoleFailure::TapDetach,
                );
                return Err(error);
            }
            drop(taps);
            retiring.remove(0);
            report.retired += 1;
        }
        Ok(())
    }
}

fn len_u64(len: usize) -> u64 {
    u64::try_from(len).unwrap_or(u64::MAX)
}

fn pending_total(live: &BTreeMap<String, AttachedTap>) -> u64 {
    live.values()
        .map(|entry| len_u64(entry.pending.len()))
        .sum()
}