liminal-server 0.23.1

//! The topology the server holds in memory (SRV-012 R7).
//!
//! It is built once from the roster, the responders, the live connections and
//! the membership at start, then kept current by one change per event at the
//! sites that cause them. It never reads a store row and never re-reads the roster to answer a
//! change: the latest document is always ready for a late subscriber.

use std::collections::{BTreeMap, BTreeSet};
use std::time::Instant;

use super::records::{
    Carrier, Line, LineKind, LineLife, LineState, Link, LinkCounters, LinkState, Mount,
    StationActivity,
};
mod document;
mod service_lines;
use crate::server::connection::is_console_subject;

/// One change to the topology, emitted where it happens.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum TopologyChange {
    /// A responder (a service) was registered or removed.
    Service {
        /// Its subject.
        subject: String,
        /// Whether it is registered now.
        present: bool,
    },
    /// A channel joined the roster or changed state.
    Channel {
        /// The channel.
        name: String,
        /// How long it keeps messages.
        life: LineLife,
        /// Whether it admits.
        state: LineState,
    },
    /// A connection opened.
    ConnectionOpened {
        /// The connection id (the supervisor pid).
        id: u64,
        /// The transport it arrived on.
        mount: Mount,
        /// The participant its pass names, when it has one.
        participant: Option<String>,
    },
    /// A connection closed.
    ConnectionClosed {
        /// The connection id.
        id: u64,
    },
    /// A connection registered as a worker: its identity is a station
    /// while the connection is open, carrying what it serves (SRV-027 R3).
    Worker {
        /// The connection id.
        connection: u64,
        /// The worker's identity, its station id.
        identity: String,
        /// The subjects it serves: its registration's task queue.
        subjects: Vec<String>,
        /// Its activities with their schema strings, as registered.
        activities: Vec<StationActivity>,
    },
    /// A connection began (`present`) or stopped using a line one way.
    Flow {
        /// The connection id.
        connection: u64,
        /// The line.
        line: String,
        /// Which way it uses the line.
        role: FlowRole,
        /// Whether it does so now.
        present: bool,
    },
    /// A federated link opened, ended, or changed its observations.
    Link {
        /// The canonical local-device and peer link id.
        link_id: String,
        /// The peer device id.
        peer: String,
        /// The transport and relay, when observed.
        carrier: Option<Carrier>,
        /// Whether the link is up or down.
        state: LinkState,
        /// The end cause, when observed.
        reason: Option<String>,
        /// Counts supplied by their source.
        counters: LinkCounters,
    },
    /// A cluster peer joined or left.
    Peer {
        /// The peer's node name.
        name: String,
        /// Whether it is a member now.
        present: bool,
    },
}

/// Which way a connection uses a line. Publishers order before
/// subscribers, so a line's stops run from where its messages enter to where
/// they leave.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub enum FlowRole {
    /// It publishes on the line.
    Publishes,
    /// It subscribes to the line.
    Subscribes,
}

/// What this device is, fixed at start.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ThisDevice {
    /// The device id: the cluster node name, or `local` without a cluster.
    pub id: String,
    /// The liminal version running here, from the build.
    pub liminal: String,
    /// When this server's incarnation started, for its uptime.
    pub started: Instant,
    /// Whether this server keeps a persistent store: its depot, a station
    /// that durable lines pass before the exchange (lane brief ruling 5).
    pub depot: bool,
}

/// The held topology and the revision of its latest document.
#[derive(Clone, Debug)]
pub struct TopologyModel {
    this: ThisDevice,
    rev: u64,
    stations: BTreeSet<String>,
    lines: BTreeMap<String, (LineLife, LineState)>,
    connections: BTreeMap<u64, (Mount, Option<String>)>,
    peers: BTreeSet<String>,
    links: BTreeMap<String, Link>,
    parent: Option<String>,
    /// Registered workers, by connection.
    workers: BTreeMap<u64, WorkerStation>,
    /// Each line's users, by role then connection.
    flows: BTreeMap<String, BTreeSet<(FlowRole, u64)>>,
}

/// A registered worker as the topology holds it (SRV-027 R3).
#[derive(Clone, Debug, PartialEq, Eq)]
struct WorkerStation {
    identity: String,
    subjects: Vec<String>,
    activities: Vec<StationActivity>,
}

impl TopologyModel {
    /// The model at start, revision 1, from what the server already holds.
    /// `console.*` channels are never lines.
    #[must_use]
    pub fn new(
        this: ThisDevice,
        stations: impl IntoIterator<Item = String>,
        lines: impl IntoIterator<Item = (String, LineLife, LineState)>,
        connections: impl IntoIterator<Item = (u64, Mount, Option<String>)>,
        peers: impl IntoIterator<Item = String>,
    ) -> Self {
        Self {
            this,
            rev: 1,
            stations: stations.into_iter().collect(),
            lines: lines
                .into_iter()
                .filter(|(name, _, _)| !is_console_subject(name))
                .map(|(name, life, state)| (name, (life, state)))
                .collect(),
            connections: connections
                .into_iter()
                .map(|(id, mount, participant)| (id, (mount, participant)))
                .collect(),
            peers: peers.into_iter().collect(),
            links: BTreeMap::new(),
            parent: None,
            workers: BTreeMap::new(),
            flows: BTreeMap::new(),
        }
    }

    /// Reads this device's district parent from its machine configuration.
    #[must_use]
    pub fn with_machine(mut self, machine: &crate::config::types::MachineConfig) -> Self {
        self.parent.clone_from(&machine.parent);
        self
    }

    /// Whether this server keeps a persistent store, its depot.
    #[must_use]
    pub const fn depot(&self) -> bool {
        self.this.depot
    }

    /// The revision of the latest document.
    #[must_use]
    pub const fn rev(&self) -> u64 {
        self.rev
    }

    pub(crate) fn ensure_revision(&mut self, revision: u64) {
        self.rev = self.rev.max(revision);
    }

    /// Applies `change`. A change that alters the topology moves the revision
    /// on by one and answers true; one that alters nothing (a repeat, or a
    /// `console.*` channel) leaves it and answers false, so nothing is
    /// published for it.
    pub fn apply(&mut self, change: TopologyChange) -> bool {
        let changed = match change {
            TopologyChange::Service { subject, present } => {
                if present {
                    self.stations.insert(subject)
                } else {
                    self.stations.remove(&subject)
                }
            }
            TopologyChange::Channel { name, life, state } => {
                !is_console_subject(&name)
                    && self.lines.insert(name, (life, state)) != Some((life, state))
            }
            TopologyChange::ConnectionOpened {
                id,
                mount,
                participant,
            } => {
                let entry = (mount, participant);
                if self.connections.get(&id) == Some(&entry) {
                    false
                } else {
                    self.connections.insert(id, entry);
                    true
                }
            }
            TopologyChange::ConnectionClosed { id } => {
                let closed = self.connections.remove(&id).is_some();
                let worker = self.workers.remove(&id).is_some();
                for users in self.flows.values_mut() {
                    users.retain(|(_, connection)| *connection != id);
                }
                self.flows.retain(|_, users| !users.is_empty());
                closed || worker
            }
            TopologyChange::Worker {
                connection,
                identity,
                subjects,
                activities,
            } => {
                let worker = WorkerStation {
                    identity,
                    subjects,
                    activities,
                };
                self.workers.insert(connection, worker.clone()) != Some(worker)
            }
            TopologyChange::Flow {
                connection,
                line,
                role,
                present,
            } => {
                if is_console_subject(&line) {
                    false
                } else {
                    let moved = if present {
                        self.flows
                            .entry(line)
                            .or_default()
                            .insert((role, connection))
                    } else if let Some(users) = self.flows.get_mut(&line) {
                        let removed = users.remove(&(role, connection));
                        if users.is_empty() {
                            self.flows.remove(&line);
                        }
                        removed
                    } else {
                        false
                    };
                    moved && self.workers.contains_key(&connection)
                }
            }
            TopologyChange::Link {
                link_id,
                peer,
                carrier,
                state,
                reason,
                counters,
            } => {
                let link = Link::federated(&self.this.id, &peer, carrier, state, reason, counters);
                self.apply_link(link_id, link)
            }
            TopologyChange::Peer { name, present } => {
                if present {
                    self.peers.insert(name)
                } else {
                    self.peers.remove(&name)
                }
            }
        };
        if changed {
            self.rev = self.rev.saturating_add(1);
        }
        changed
    }

    // Reject inconsistent endpoint identities before retaining an observation.
    fn apply_link(&mut self, link_id: String, mut link: Link) -> bool {
        if link.id != link_id || link.name.is_empty() || link.name == self.this.id {
            tracing::error!(%link_id, peer = %link.name, "topology link refused: invalid endpoint or link id");
            false
        } else {
            if let Some(previous) = self.links.get(&link_id) {
                link.retain_carrier_refusals(previous);
                link.crossed = link.crossed.or(previous.crossed);
                link.refused = link.refused.or(previous.refused);
                link.already_crossed = link.already_crossed.or(previous.already_crossed);
                link.lifecycle = link.lifecycle.or(previous.lifecycle);
            }
            self.links.insert(link_id, link.clone()) != Some(link)
        }
    }

    /// The stations `line` passes: its registered publishers, then this
    /// server's depot when the line is durable and the server keeps a
    /// persistent store (the append comes before the fan-out), then this
    /// server's exchange, once on every line, then its registered
    /// subscribers, each once.
    fn stops(&self, line: &str, line_life: LineLife) -> Vec<String> {
        let mut stops: Vec<String> = Vec::new();
        let depot = (self.this.depot && line_life == LineLife::Kept).then(|| self.depot_id());
        // The server's own stops, in the order a message reaches them.
        let mut own = depot.into_iter().chain([self.this.id.clone()]);
        let users = self.flows.get(line).into_iter().flatten();
        for (role, connection) in users {
            if *role == FlowRole::Subscribes {
                stops.extend(own.by_ref());
            }
            if let Some(identity) = self.workers.get(connection).map(|worker| &worker.identity)
                && self.reserved(identity).is_none()
                && !stops.contains(identity)
            {
                stops.push(identity.clone());
            }
        }
        stops.extend(own);
        stops
    }

    /// The station `id` would collide with: this server's exchange (its
    /// device id), or its depot when it keeps one. A worker or service with
    /// such an id is kept off the map, never merged into that station
    /// (Buttercup's V1, Hermes 04:27).
    #[must_use]
    pub fn reserved(&self, id: &str) -> Option<&'static str> {
        if id == self.this.id {
            Some("exchange")
        } else if self.this.depot
            && id
                .strip_prefix(self.this.id.as_str())
                .is_some_and(|rest| rest == ".depot")
        {
            Some("depot")
        } else {
            None
        }
    }

    /// This server's depot's station id: `<device>.depot`.
    fn depot_id(&self) -> String {
        format!("{}.depot", self.this.id)
    }

    /// This server's incarnation started at `started`: its uptime runs
    /// from there (SRV-012 R7). Answers whether that moved it.
    pub fn incarnation_started(&mut self, started: Instant) -> bool {
        let moved = self.this.started != started;
        self.this.started = started;
        moved
    }

    /// Each connection's participant, by connection id. Test-only.
    #[cfg(test)]
    pub(crate) fn participants(&self) -> Vec<Option<String>> {
        self.connections
            .values()
            .map(|(_, participant)| participant.clone())
            .collect()
    }

    /// How many connections the model holds. Test-only.
    #[cfg(test)]
    pub(crate) fn connection_count(&self) -> usize {
        self.connections.len()
    }
}