liminal-server 0.19.0

//! 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::error::ConsoleError;
use super::records::{
    Connection, Device, DeviceBox, DeviceKind, Line, LineKind, LineLife, LineState, Link, Mount,
    Station, StationKind, StationState, TopologyDocument,
};
use super::wall::WallBase;
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.
    Worker {
        /// The connection id.
        connection: u64,
        /// The worker's identity, its station id.
        identity: String,
    },
    /// 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 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>,
    /// Registered workers, by connection.
    workers: BTreeMap<u64, String>,
    /// Each line's users, by role then connection.
    flows: BTreeMap<String, BTreeSet<(FlowRole, u64)>>,
}

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(),
            workers: BTreeMap::new(),
            flows: BTreeMap::new(),
        }
    }

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

    /// 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,
            } => self.workers.insert(connection, identity.clone()) != Some(identity),
            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::Peer { name, present } => {
                if present {
                    self.peers.insert(name)
                } else {
                    self.peers.remove(&name)
                }
            }
        };
        if changed {
            self.rev = self.rev.saturating_add(1);
        }
        changed
    }

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

    /// The whole document as it stands, always. This device's start is wall
    /// time from `wall`, the same whenever the document is built, so a reader
    /// derives the uptime on its own clock and a document never goes stale.
    /// When the start falls outside u64 milliseconds since the epoch, the
    /// document names that fault where the start would be, and the fault
    /// comes back beside it for the caller to count: one bad reading never
    /// withholds the topology.
    #[must_use]
    pub fn document(&self, wall: &WallBase) -> (TopologyDocument, Option<ConsoleError>) {
        let this = &self.this.id;
        let (started, fault) = match wall.unix_ms_of(self.this.started) {
            Ok(started) => (Some(started), None),
            Err(error) => (None, Some(error)),
        };
        let mut devices = vec![Device {
            id: this.clone(),
            name: this.clone(),
            kind: DeviceKind::Device,
            liminal: Some(self.this.liminal.clone()),
            started,
            started_fault: fault.as_ref().map(ToString::to_string),
            zone: Some(DeviceBox::THIS),
        }];
        devices.extend(self.peers.iter().map(|peer| Device {
            id: peer.clone(),
            name: peer.clone(),
            kind: DeviceKind::Device,
            liminal: None,
            started: None,
            started_fault: None,
            zone: None,
        }));
        // A registered responder is a service; a registered worker that is
        // not also one is a worker. Neither takes this server's exchange or
        // depot id: one that would is kept off the map.
        let depot = self.this.depot.then(|| self.depot_id());
        let mut stations: BTreeMap<&String, StationKind> = self
            .workers
            .values()
            .filter(|identity| self.reserved(identity).is_none())
            .map(|identity| (identity, StationKind::Worker))
            .collect();
        stations.extend(
            self.stations
                .iter()
                .filter(|subject| self.reserved(subject).is_none())
                .map(|subject| (subject, StationKind::Service)),
        );
        stations.extend(depot.iter().map(|id| (id, StationKind::Depot)));
        stations.insert(this, StationKind::Exchange);
        let document = TopologyDocument::new(
            self.rev,
            this.clone(),
            devices,
            stations
                .into_iter()
                .map(|(subject, kind)| Station {
                    id: subject.clone(),
                    name: subject.clone(),
                    device: this.clone(),
                    kind,
                    state: StationState::Live,
                })
                .collect(),
            self.lines
                .iter()
                .map(|(name, (life, state))| Line {
                    id: name.clone(),
                    kind: LineKind::Stream,
                    life: *life,
                    state: *state,
                    stops: self.stops(name, *life),
                })
                .collect(),
            self.peers
                .iter()
                .map(|peer| Link::to_peer(this, peer))
                .collect(),
            self.connections
                .iter()
                .map(|(id, (mount, participant))| Connection {
                    id: id.to_string(),
                    mount: *mount,
                    participant: participant.clone(),
                })
                .collect(),
        );
        (document, fault)
    }
}