liminal_server/console/
model.rs

1//! The topology the server holds in memory (SRV-012 R7).
2//!
3//! It is built once from the roster, the responders, the live connections and
4//! the membership at start, then kept current by one change per event at the
5//! sites that cause them. It never reads a store row and never re-reads the roster to answer a
6//! change: the latest document is always ready for a late subscriber.
7
8use std::collections::{BTreeMap, BTreeSet};
9use std::time::Instant;
10
11use super::records::{
12    Carrier, Line, LineKind, LineLife, LineState, Link, LinkCounters, LinkState, Mount,
13    StationActivity,
14};
15mod document;
16mod service_lines;
17use crate::server::connection::is_console_subject;
18
19/// One change to the topology, emitted where it happens.
20#[derive(Clone, Debug, PartialEq, Eq)]
21pub enum TopologyChange {
22    /// A responder (a service) was registered or removed.
23    Service {
24        /// Its subject.
25        subject: String,
26        /// Whether it is registered now.
27        present: bool,
28    },
29    /// A channel joined the roster or changed state.
30    Channel {
31        /// The channel.
32        name: String,
33        /// How long it keeps messages.
34        life: LineLife,
35        /// Whether it admits.
36        state: LineState,
37    },
38    /// A connection opened.
39    ConnectionOpened {
40        /// The connection id (the supervisor pid).
41        id: u64,
42        /// The transport it arrived on.
43        mount: Mount,
44        /// The participant its pass names, when it has one.
45        participant: Option<String>,
46    },
47    /// A connection closed.
48    ConnectionClosed {
49        /// The connection id.
50        id: u64,
51    },
52    /// A connection registered as a worker: its identity is a station
53    /// while the connection is open, carrying what it serves (SRV-027 R3).
54    Worker {
55        /// The connection id.
56        connection: u64,
57        /// The worker's identity, its station id.
58        identity: String,
59        /// The subjects it serves: its registration's task queue.
60        subjects: Vec<String>,
61        /// Its activities with their schema strings, as registered.
62        activities: Vec<StationActivity>,
63    },
64    /// A connection began (`present`) or stopped using a line one way.
65    Flow {
66        /// The connection id.
67        connection: u64,
68        /// The line.
69        line: String,
70        /// Which way it uses the line.
71        role: FlowRole,
72        /// Whether it does so now.
73        present: bool,
74    },
75    /// A federated link opened, ended, or changed its observations.
76    Link {
77        /// The canonical local-device and peer link id.
78        link_id: String,
79        /// The peer device id.
80        peer: String,
81        /// The transport and relay, when observed.
82        carrier: Option<Carrier>,
83        /// Whether the link is up or down.
84        state: LinkState,
85        /// The end cause, when observed.
86        reason: Option<String>,
87        /// Counts supplied by their source.
88        counters: LinkCounters,
89    },
90    /// A cluster peer joined or left.
91    Peer {
92        /// The peer's node name.
93        name: String,
94        /// Whether it is a member now.
95        present: bool,
96    },
97}
98
99/// Which way a connection uses a line. Publishers order before
100/// subscribers, so a line's stops run from where its messages enter to where
101/// they leave.
102#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
103pub enum FlowRole {
104    /// It publishes on the line.
105    Publishes,
106    /// It subscribes to the line.
107    Subscribes,
108}
109
110/// What this device is, fixed at start.
111#[derive(Clone, Debug, PartialEq, Eq)]
112pub struct ThisDevice {
113    /// The device id: the cluster node name, or `local` without a cluster.
114    pub id: String,
115    /// The liminal version running here, from the build.
116    pub liminal: String,
117    /// When this server's incarnation started, for its uptime.
118    pub started: Instant,
119    /// Whether this server keeps a persistent store: its depot, a station
120    /// that durable lines pass before the exchange (lane brief ruling 5).
121    pub depot: bool,
122}
123
124/// The held topology and the revision of its latest document.
125#[derive(Clone, Debug)]
126pub struct TopologyModel {
127    this: ThisDevice,
128    rev: u64,
129    stations: BTreeSet<String>,
130    lines: BTreeMap<String, (LineLife, LineState)>,
131    connections: BTreeMap<u64, (Mount, Option<String>)>,
132    peers: BTreeSet<String>,
133    links: BTreeMap<String, Link>,
134    parent: Option<String>,
135    /// Registered workers, by connection.
136    workers: BTreeMap<u64, WorkerStation>,
137    /// Each line's users, by role then connection.
138    flows: BTreeMap<String, BTreeSet<(FlowRole, u64)>>,
139}
140
141/// A registered worker as the topology holds it (SRV-027 R3).
142#[derive(Clone, Debug, PartialEq, Eq)]
143struct WorkerStation {
144    identity: String,
145    subjects: Vec<String>,
146    activities: Vec<StationActivity>,
147}
148
149impl TopologyModel {
150    /// The model at start, revision 1, from what the server already holds.
151    /// `console.*` channels are never lines.
152    #[must_use]
153    pub fn new(
154        this: ThisDevice,
155        stations: impl IntoIterator<Item = String>,
156        lines: impl IntoIterator<Item = (String, LineLife, LineState)>,
157        connections: impl IntoIterator<Item = (u64, Mount, Option<String>)>,
158        peers: impl IntoIterator<Item = String>,
159    ) -> Self {
160        Self {
161            this,
162            rev: 1,
163            stations: stations.into_iter().collect(),
164            lines: lines
165                .into_iter()
166                .filter(|(name, _, _)| !is_console_subject(name))
167                .map(|(name, life, state)| (name, (life, state)))
168                .collect(),
169            connections: connections
170                .into_iter()
171                .map(|(id, mount, participant)| (id, (mount, participant)))
172                .collect(),
173            peers: peers.into_iter().collect(),
174            links: BTreeMap::new(),
175            parent: None,
176            workers: BTreeMap::new(),
177            flows: BTreeMap::new(),
178        }
179    }
180
181    /// Reads this device's district parent from its machine configuration.
182    #[must_use]
183    pub fn with_machine(mut self, machine: &crate::config::types::MachineConfig) -> Self {
184        self.parent.clone_from(&machine.parent);
185        self
186    }
187
188    /// Whether this server keeps a persistent store, its depot.
189    #[must_use]
190    pub const fn depot(&self) -> bool {
191        self.this.depot
192    }
193
194    /// The revision of the latest document.
195    #[must_use]
196    pub const fn rev(&self) -> u64 {
197        self.rev
198    }
199
200    pub(crate) fn ensure_revision(&mut self, revision: u64) {
201        self.rev = self.rev.max(revision);
202    }
203
204    /// Applies `change`. A change that alters the topology moves the revision
205    /// on by one and answers true; one that alters nothing (a repeat, or a
206    /// `console.*` channel) leaves it and answers false, so nothing is
207    /// published for it.
208    pub fn apply(&mut self, change: TopologyChange) -> bool {
209        let changed = match change {
210            TopologyChange::Service { subject, present } => {
211                if present {
212                    self.stations.insert(subject)
213                } else {
214                    self.stations.remove(&subject)
215                }
216            }
217            TopologyChange::Channel { name, life, state } => {
218                !is_console_subject(&name)
219                    && self.lines.insert(name, (life, state)) != Some((life, state))
220            }
221            TopologyChange::ConnectionOpened {
222                id,
223                mount,
224                participant,
225            } => {
226                let entry = (mount, participant);
227                if self.connections.get(&id) == Some(&entry) {
228                    false
229                } else {
230                    self.connections.insert(id, entry);
231                    true
232                }
233            }
234            TopologyChange::ConnectionClosed { id } => {
235                let closed = self.connections.remove(&id).is_some();
236                let worker = self.workers.remove(&id).is_some();
237                for users in self.flows.values_mut() {
238                    users.retain(|(_, connection)| *connection != id);
239                }
240                self.flows.retain(|_, users| !users.is_empty());
241                closed || worker
242            }
243            TopologyChange::Worker {
244                connection,
245                identity,
246                subjects,
247                activities,
248            } => {
249                let worker = WorkerStation {
250                    identity,
251                    subjects,
252                    activities,
253                };
254                self.workers.insert(connection, worker.clone()) != Some(worker)
255            }
256            TopologyChange::Flow {
257                connection,
258                line,
259                role,
260                present,
261            } => {
262                if is_console_subject(&line) {
263                    false
264                } else {
265                    let moved = if present {
266                        self.flows
267                            .entry(line)
268                            .or_default()
269                            .insert((role, connection))
270                    } else if let Some(users) = self.flows.get_mut(&line) {
271                        let removed = users.remove(&(role, connection));
272                        if users.is_empty() {
273                            self.flows.remove(&line);
274                        }
275                        removed
276                    } else {
277                        false
278                    };
279                    moved && self.workers.contains_key(&connection)
280                }
281            }
282            TopologyChange::Link {
283                link_id,
284                peer,
285                carrier,
286                state,
287                reason,
288                counters,
289            } => {
290                let link = Link::federated(&self.this.id, &peer, carrier, state, reason, counters);
291                self.apply_link(link_id, link)
292            }
293            TopologyChange::Peer { name, present } => {
294                if present {
295                    self.peers.insert(name)
296                } else {
297                    self.peers.remove(&name)
298                }
299            }
300        };
301        if changed {
302            self.rev = self.rev.saturating_add(1);
303        }
304        changed
305    }
306
307    // Reject inconsistent endpoint identities before retaining an observation.
308    fn apply_link(&mut self, link_id: String, mut link: Link) -> bool {
309        if link.id != link_id || link.name.is_empty() || link.name == self.this.id {
310            tracing::error!(%link_id, peer = %link.name, "topology link refused: invalid endpoint or link id");
311            false
312        } else {
313            if let Some(previous) = self.links.get(&link_id) {
314                link.retain_carrier_refusals(previous);
315                link.crossed = link.crossed.or(previous.crossed);
316                link.refused = link.refused.or(previous.refused);
317                link.already_crossed = link.already_crossed.or(previous.already_crossed);
318                link.lifecycle = link.lifecycle.or(previous.lifecycle);
319            }
320            self.links.insert(link_id, link.clone()) != Some(link)
321        }
322    }
323
324    /// The stations `line` passes: its registered publishers, then this
325    /// server's depot when the line is durable and the server keeps a
326    /// persistent store (the append comes before the fan-out), then this
327    /// server's exchange, once on every line, then its registered
328    /// subscribers, each once.
329    fn stops(&self, line: &str, line_life: LineLife) -> Vec<String> {
330        let mut stops: Vec<String> = Vec::new();
331        let depot = (self.this.depot && line_life == LineLife::Kept).then(|| self.depot_id());
332        // The server's own stops, in the order a message reaches them.
333        let mut own = depot.into_iter().chain([self.this.id.clone()]);
334        let users = self.flows.get(line).into_iter().flatten();
335        for (role, connection) in users {
336            if *role == FlowRole::Subscribes {
337                stops.extend(own.by_ref());
338            }
339            if let Some(identity) = self.workers.get(connection).map(|worker| &worker.identity)
340                && self.reserved(identity).is_none()
341                && !stops.contains(identity)
342            {
343                stops.push(identity.clone());
344            }
345        }
346        stops.extend(own);
347        stops
348    }
349
350    /// The station `id` would collide with: this server's exchange (its
351    /// device id), or its depot when it keeps one. A worker or service with
352    /// such an id is kept off the map, never merged into that station
353    /// (Buttercup's V1, Hermes 04:27).
354    #[must_use]
355    pub fn reserved(&self, id: &str) -> Option<&'static str> {
356        if id == self.this.id {
357            Some("exchange")
358        } else if self.this.depot
359            && id
360                .strip_prefix(self.this.id.as_str())
361                .is_some_and(|rest| rest == ".depot")
362        {
363            Some("depot")
364        } else {
365            None
366        }
367    }
368
369    /// This server's depot's station id: `<device>.depot`.
370    fn depot_id(&self) -> String {
371        format!("{}.depot", self.this.id)
372    }
373
374    /// This server's incarnation started at `started`: its uptime runs
375    /// from there (SRV-012 R7). Answers whether that moved it.
376    pub fn incarnation_started(&mut self, started: Instant) -> bool {
377        let moved = self.this.started != started;
378        self.this.started = started;
379        moved
380    }
381
382    /// Each connection's participant, by connection id. Test-only.
383    #[cfg(test)]
384    pub(crate) fn participants(&self) -> Vec<Option<String>> {
385        self.connections
386            .values()
387            .map(|(_, participant)| participant.clone())
388            .collect()
389    }
390
391    /// How many connections the model holds. Test-only.
392    #[cfg(test)]
393    pub(crate) fn connection_count(&self) -> usize {
394        self.connections.len()
395    }
396}
liminal_server/console/model.rs

liminal_server/console/
model.rs
