liminal-server 0.23.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, OnceLock};
mod placements;
mod pump;
mod taps;
#[cfg(test)]
mod test_support;
mod topology;
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,
    published: Option<(u64, std::collections::BTreeSet<String>)>,
}

#[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>,
    placements: OnceLock<Arc<super::tap_placement::TapPlacements>>,
    /// 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,
                published: None,
            }),
            taps: Mutex::new(Taps::default()),
            notifier,
            counts,
            placements: OnceLock::new(),
            #[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
    }

    /// The station serving a registered local responder subject.
    ///
    /// # Errors
    /// The topology lock is poisoned.
    pub fn serving_station(&self, subject: &str) -> Result<Option<String>, TapError> {
        Ok(self.lock_topology()?.model.serving_station(subject))
    }

    /// 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
    }

    /// 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");
        }
        if colliding.is_none() {
            self.offer_change(&change);
        }
        // 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(())
    }

    /// 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 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()
}