majordomus_cli/mesh/
registry.rs

1//! The one registry of discovered nodes. Every provider's observation converges here,
2//! deduplicated by node id (a digest of the key — never by address), so a node seen over
3//! multicast and handed back by a rendezvous is one record with two sources, not two
4//! rows with one truth each. Every surface — CLI, HTTP, OpenAPI, MCP, Cockpit — is a
5//! projection of this map; nothing else may hold peers.
6//!
7//! In memory, bounded, and gone with the process, like the peer board beside it: a
8//! discovered node is an observation about *now*, and persisting it would only let the
9//! registry disagree with the network.
10//!
11//! ```
12//! use majordomus_cli::mesh::identity::NodeIdentity;
13//! use majordomus_cli::mesh::protocol::advertise;
14//! use majordomus_cli::mesh::registry::{MeshRegistry, MeshSource};
15//! use majordomus_cli::mesh::trust::TrustState;
16//!
17//! let registry = MeshRegistry::new();
18//! let node = NodeIdentity::ephemeral().unwrap();
19//! let heard = advertise(&node, 1, &[], &[], &[], "docs");
20//! registry.observe(&heard.adv, node.public.node_id.clone(), TrustState::Observed,
21//!     MeshSource::Synthetic, "docs", serde_json::Value::Null);
22//! assert_eq!(registry.list().len(), 1);
23//! assert_eq!(registry.tallies().accepted, 1);
24//! ```
25
26use std::collections::BTreeMap;
27use std::sync::Mutex;
28use std::time::{Duration, Instant};
29
30use schemars::JsonSchema;
31use serde::{Deserialize, Serialize};
32
33use super::identity::{InstanceId, NodeId};
34use super::protocol::Advertisement;
35use super::trust::TrustState;
36
37/// The most nodes the registry holds. When full, an insertion evicts the longest-unseen
38/// untrusted node; trusted nodes are never evicted for space.
39pub const MAX_NODES: usize = 256;
40
41/// How long a node stays `present` after its last advertisement.
42pub const PRESENCE_TTL: Duration = Duration::from_secs(60);
43
44/// How long an absent node stays listed before it expires out of the registry entirely.
45pub const RETENTION: Duration = Duration::from_secs(15 * 60);
46
47/// Where an observation came from. A record accumulates these; provenance is the answer
48/// to "why do I see this node".
49#[derive(
50    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema,
51)]
52#[serde(rename_all = "snake_case")]
53pub enum MeshSource {
54    /// A multicast datagram.
55    UdpMulticast,
56    /// A broadcast datagram.
57    UdpBroadcast,
58    /// A rendezvous answer or registration.
59    Rendezvous,
60    /// A synthetic provider (tests, self-check).
61    Synthetic,
62}
63
64impl MeshSource {
65    /// The source's wire word, as sightings and status listings carry it.
66    pub fn as_str(&self) -> &'static str {
67        match self {
68            MeshSource::UdpMulticast => "udp_multicast",
69            MeshSource::UdpBroadcast => "udp_broadcast",
70            MeshSource::Rendezvous => "rendezvous",
71            MeshSource::Synthetic => "synthetic",
72        }
73    }
74}
75
76/// Whether the node is answering now, by its own announcements.
77#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
78#[serde(rename_all = "snake_case")]
79pub enum Presence {
80    /// An advertisement arrived within [`PRESENCE_TTL`].
81    Present,
82    /// Nothing within the TTL; the node may be gone or unreachable. It stays listed
83    /// until [`RETENTION`] passes, then expires out entirely.
84    Absent,
85}
86
87/// One provenance entry: a source, and where the observation physically came from.
88#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
89pub struct Sighting {
90    /// The provider kind that saw it.
91    pub source: MeshSource,
92    /// The network path, as text: a sender address, an interface, a rendezvous URL.
93    pub path: String,
94    /// When, RFC 3339.
95    pub at: String,
96}
97
98/// A node is sighted at most once per source — a second sighting from the same source
99/// replaces the first — so the source is the whole key, and it is the explicit rank
100/// because its order is the declared precedence of the transports, not the alphabet of
101/// their names. The path and the timestamp end the key so that it stays total if that
102/// ever stops being true.
103impl crate::order::Ordered for Sighting {
104    fn order_key(&self) -> crate::order::OrderKey<'_> {
105        crate::order::OrderKey::plain(&self.path, &self.at).ranked(self.source as i64)
106    }
107}
108
109/// One discovered node: the canonical record every surface projects.
110#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
111pub struct NodeRecord {
112    /// The node id: the digest of the key below.
113    pub node_id: NodeId,
114    /// The runtime slot: which server of the node (one per checkout), 16 hex; empty for a
115    /// version-1 advertisement. A record is one runtime: two worktrees of one machine are
116    /// two records under one node id and one key.
117    #[serde(default)]
118    pub runtime: String,
119    /// The Ed25519 public key that proved this record's advertisements.
120    pub public_key: String,
121    /// The node's own display name. Presentation only.
122    pub display_name: String,
123    /// The node's current run.
124    pub instance_id: InstanceId,
125    /// What the policy decided.
126    pub trust: TrustState,
127    /// Whether the node is answering now.
128    pub presence: Presence,
129    /// Advertised `host:port` authorities.
130    pub endpoints: Vec<String>,
131    /// Advertised transports (`http`, `mcp`, `ws`).
132    pub capabilities: Vec<String>,
133    /// Advertised repository digests.
134    pub repositories: Vec<String>,
135    /// The advertised protocol version.
136    pub protocol_version: u32,
137    /// The advertised executable version.
138    pub version: String,
139    /// Every source this node was seen through, most recent sighting per source.
140    pub sources: Vec<Sighting>,
141    /// First seen, RFC 3339.
142    pub first_seen: String,
143    /// Last seen, RFC 3339.
144    pub last_seen: String,
145    /// How many instances this node id has shown (a restart increments it).
146    pub restarts: u64,
147}
148
149/// The registry's tallies, for status and health.
150#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, JsonSchema)]
151pub struct Tallies {
152    /// Records held.
153    pub nodes: usize,
154    /// Of which trusted.
155    pub trusted: usize,
156    /// Of which present.
157    pub present: usize,
158    /// Observations accepted since start.
159    pub accepted: u64,
160    /// Replayed datagrams dropped since start.
161    pub replayed: u64,
162    /// Records expired out since start.
163    pub expired: u64,
164    /// Records evicted for space since start.
165    pub evicted: u64,
166}
167
168struct Slot {
169    record: NodeRecord,
170    last_seen: Instant,
171    /// The highest sequence seen per instance: a datagram at or below it is a replay.
172    last_seq: u64,
173    /// The last envelope as heard, signature included, for rendezvous answers: a
174    /// candidate travels verbatim so its signature verifies end-to-end at the consumer.
175    raw: serde_json::Value,
176}
177
178struct Inner {
179    /// Keyed by (node, runtime): node-id order first, as every listing promises.
180    slots: BTreeMap<(NodeId, String), Slot>,
181    tallies: Tallies,
182}
183
184/// The registry. `Mutex<BTreeMap>`, like the peer board: contention is a handful of
185/// datagrams a second, and a deterministic order falls out for free.
186pub struct MeshRegistry {
187    inner: Mutex<Inner>,
188}
189
190impl Default for MeshRegistry {
191    fn default() -> Self {
192        Self::new()
193    }
194}
195
196impl MeshRegistry {
197    /// An empty registry; it fills only through [`MeshRegistry::observe`].
198    pub fn new() -> Self {
199        MeshRegistry {
200            inner: Mutex::new(Inner {
201                slots: BTreeMap::new(),
202                tallies: Tallies::default(),
203            }),
204        }
205    }
206
207    /// The key currently bound to a node id, for trust evaluation.
208    pub fn known_key(&self, node: &NodeId) -> Option<String> {
209        let inner = self.inner.lock().expect("mesh registry lock");
210        inner
211            .slots
212            .iter()
213            .find(|((n, _), _)| n == node)
214            .map(|(_, s)| s.record.public_key.clone())
215    }
216
217    /// The current trust state of a node id: the one its runtimes share, the first
218    /// rejection winning, because trust is a property of the key and not of a server.
219    pub fn trust_of(&self, node: &NodeId) -> Option<TrustState> {
220        let inner = self.inner.lock().expect("mesh registry lock");
221        let states: Vec<&TrustState> = inner
222            .slots
223            .iter()
224            .filter(|((n, _), _)| n == node)
225            .map(|(_, s)| &s.record.trust)
226            .collect();
227        states
228            .iter()
229            .find(|t| matches!(t, TrustState::Rejected(_)))
230            .or(states.first())
231            .map(|t| (*t).clone())
232    }
233
234    /// Record one verified observation; `raw` is the envelope as heard, kept verbatim
235    /// for rendezvous answers. Returns `false` when the datagram was a replay (same
236    /// instance, sequence not above the highest seen) and the record was left unchanged.
237    pub fn observe(
238        &self,
239        adv: &Advertisement,
240        node_id: NodeId,
241        trust: TrustState,
242        source: MeshSource,
243        path: &str,
244        raw: serde_json::Value,
245    ) -> bool {
246        let now = Instant::now();
247        let stamp = crate::peers::rfc3339(std::time::SystemTime::now());
248        let mut inner = self.inner.lock().expect("mesh registry lock");
249        self.expire_locked(&mut inner, now);
250        let key = (node_id.clone(), adv.rt.clone());
251        if let Some(slot) = inner.slots.get_mut(&key) {
252            let same_instance = slot.record.instance_id == adv.inst;
253            if same_instance && adv.seq <= slot.last_seq {
254                inner.tallies.replayed += 1;
255                return false;
256            }
257            if !same_instance {
258                slot.record.restarts += 1;
259                slot.record.instance_id = adv.inst.clone();
260            }
261            slot.last_seq = adv.seq;
262            slot.last_seen = now;
263            slot.record.display_name = adv.name.clone();
264            slot.record.endpoints = adv.ep.clone();
265            slot.record.capabilities = adv.caps.clone();
266            slot.record.repositories = adv.repos.clone();
267            slot.record.protocol_version = adv.v;
268            slot.record.version = adv.ver.clone();
269            slot.record.trust = trust;
270            slot.record.presence = Presence::Present;
271            slot.record.last_seen = stamp.clone();
272            slot.raw = raw;
273            let sighting = Sighting {
274                source,
275                path: path.into(),
276                at: stamp,
277            };
278            match slot.record.sources.iter_mut().find(|s| s.source == source) {
279                Some(existing) => *existing = sighting,
280                None => slot.record.sources.push(sighting),
281            }
282            crate::order::canonical(&mut slot.record.sources);
283            inner.tallies.accepted += 1;
284            return true;
285        }
286        // A new node. Make room if the table is full: the longest-unseen record not on the
287        // allowlist goes; if every record is allowlisted, the observation is dropped instead —
288        // a full table of a person's own fleet is not something an attacker gets to flush. A
289        // record trusted by TOFU is evictable: under TOFU anyone on the segment is trusted,
290        // and exempting them would let one key lock the fleet out of the table.
291        if inner.slots.len() >= MAX_NODES {
292            let victim = inner
293                .slots
294                .iter()
295                .filter(|(_, s)| !matches!(&s.record.trust, TrustState::Trusted(by) if by == "allowlist"))
296                .min_by_key(|(_, s)| s.last_seen)
297                .map(|(id, _)| id.clone());
298            match victim {
299                Some(id) => {
300                    inner.slots.remove(&id);
301                    inner.tallies.evicted += 1;
302                }
303                None => return false,
304            }
305        }
306        inner.slots.insert(
307            key,
308            Slot {
309                record: NodeRecord {
310                    node_id,
311                    runtime: adv.rt.clone(),
312                    public_key: adv.pk.clone(),
313                    display_name: adv.name.clone(),
314                    instance_id: adv.inst.clone(),
315                    trust,
316                    presence: Presence::Present,
317                    endpoints: adv.ep.clone(),
318                    capabilities: adv.caps.clone(),
319                    repositories: adv.repos.clone(),
320                    protocol_version: adv.v,
321                    version: adv.ver.clone(),
322                    sources: vec![Sighting {
323                        source,
324                        path: path.into(),
325                        at: stamp.clone(),
326                    }],
327                    first_seen: stamp.clone(),
328                    last_seen: stamp,
329                    restarts: 0,
330                },
331                last_seen: now,
332                last_seq: adv.seq,
333                raw,
334            },
335        );
336        inner.tallies.accepted += 1;
337        true
338    }
339
340    /// The envelopes of present, trusted nodes that serve one of `repos`, verbatim as heard,
341    /// at most `limit`: what a rendezvous answers a registered caller with. A caller learns
342    /// only the nodes of its own repository this node trusts — never the topology of another
343    /// repository or the strangers this node merely observed (I2133).
344    ///
345    /// ```
346    /// use majordomus_cli::mesh::MeshRegistry;
347    /// let empty = MeshRegistry::new();
348    /// assert!(empty.candidates_for(32, &["r1".into()]).is_empty());
349    /// ```
350    pub fn candidates_for(&self, limit: usize, repos: &[String]) -> Vec<serde_json::Value> {
351        let now = Instant::now();
352        let mut inner = self.inner.lock().expect("mesh registry lock");
353        self.expire_locked(&mut inner, now);
354        inner
355            .slots
356            .values()
357            .filter(|s| {
358                now.duration_since(s.last_seen) <= PRESENCE_TTL
359                    && s.record.trust.is_trusted()
360                    && s.record.repositories.iter().any(|r| repos.contains(r))
361                    && !s.raw.is_null()
362            })
363            .take(limit)
364            .map(|s| s.raw.clone())
365            .collect()
366    }
367
368    /// Every record, in node-id order, presence recomputed against now.
369    pub fn list(&self) -> Vec<NodeRecord> {
370        let now = Instant::now();
371        let mut inner = self.inner.lock().expect("mesh registry lock");
372        self.expire_locked(&mut inner, now);
373        inner
374            .slots
375            .values()
376            .map(|slot| {
377                let mut record = slot.record.clone();
378                record.presence = if now.duration_since(slot.last_seen) <= PRESENCE_TTL {
379                    Presence::Present
380                } else {
381                    Presence::Absent
382                };
383                record
384            })
385            .collect()
386    }
387
388    /// The tallies, presence recomputed against now.
389    pub fn tallies(&self) -> Tallies {
390        let now = Instant::now();
391        let mut inner = self.inner.lock().expect("mesh registry lock");
392        self.expire_locked(&mut inner, now);
393        let mut tallies = inner.tallies;
394        tallies.nodes = inner.slots.len();
395        tallies.trusted = inner
396            .slots
397            .values()
398            .filter(|s| s.record.trust.is_trusted())
399            .count();
400        tallies.present = inner
401            .slots
402            .values()
403            .filter(|s| now.duration_since(s.last_seen) <= PRESENCE_TTL)
404            .count();
405        tallies
406    }
407
408    fn expire_locked(&self, inner: &mut Inner, now: Instant) {
409        let before = inner.slots.len();
410        inner
411            .slots
412            .retain(|_, slot| now.duration_since(slot.last_seen) <= RETENTION);
413        inner.tallies.expired += (before - inner.slots.len()) as u64;
414    }
415}
416
417#[cfg(test)]
418mod tests {
419    use super::*;
420    use crate::mesh::identity::NodeIdentity;
421    use crate::mesh::protocol::advertise;
422
423    fn node() -> (NodeIdentity, Advertisement) {
424        let dir = tempfile::tempdir().unwrap();
425        let id = NodeIdentity::load_or_create(&dir.path().join("node.json")).unwrap();
426        let env = advertise(
427            &id,
428            1,
429            &["127.0.0.1:8741".into()],
430            &["http".into()],
431            &[],
432            "0.5.0",
433        );
434        (id, env.adv)
435    }
436
437    #[test]
438    fn two_sources_converge_into_one_record() {
439        let registry = MeshRegistry::new();
440        let (id, adv) = node();
441        let mut second = adv.clone();
442        second.seq = 2;
443        assert!(registry.observe(
444            &adv,
445            id.public.node_id.clone(),
446            TrustState::Observed,
447            MeshSource::UdpMulticast,
448            "192.168.1.5:7741",
449            serde_json::Value::Null
450        ));
451        assert!(registry.observe(
452            &second,
453            id.public.node_id.clone(),
454            TrustState::Observed,
455            MeshSource::Rendezvous,
456            "http://127.0.0.1:8741",
457            serde_json::Value::Null
458        ));
459        let listed = registry.list();
460        assert_eq!(listed.len(), 1, "one node, not one per source");
461        let sources: Vec<MeshSource> = listed[0].sources.iter().map(|s| s.source).collect();
462        assert_eq!(
463            sources,
464            vec![MeshSource::UdpMulticast, MeshSource::Rendezvous]
465        );
466    }
467
468    #[test]
469    fn a_replay_is_dropped_and_counted() {
470        let registry = MeshRegistry::new();
471        let (id, adv) = node();
472        assert!(registry.observe(
473            &adv,
474            id.public.node_id.clone(),
475            TrustState::Observed,
476            MeshSource::UdpMulticast,
477            "a",
478            serde_json::Value::Null
479        ));
480        assert!(!registry.observe(
481            &adv,
482            id.public.node_id.clone(),
483            TrustState::Observed,
484            MeshSource::UdpMulticast,
485            "a",
486            serde_json::Value::Null
487        ));
488        assert_eq!(registry.tallies().replayed, 1);
489    }
490
491    #[test]
492    fn a_restart_keeps_the_record_and_counts() {
493        let registry = MeshRegistry::new();
494        let (id, adv) = node();
495        registry.observe(
496            &adv,
497            id.public.node_id.clone(),
498            TrustState::Observed,
499            MeshSource::UdpMulticast,
500            "a",
501            serde_json::Value::Null,
502        );
503        // The same node key, a new instance: sequence numbering starts over.
504        let dir = tempfile::tempdir().unwrap();
505        let restarted = NodeIdentity::load_or_create(&dir.path().join("other.json")).unwrap();
506        let mut again = adv.clone();
507        again.inst = restarted.public.instance_id.clone();
508        again.seq = 1;
509        assert!(registry.observe(
510            &again,
511            id.public.node_id.clone(),
512            TrustState::Observed,
513            MeshSource::UdpMulticast,
514            "a",
515            serde_json::Value::Null
516        ));
517        let listed = registry.list();
518        assert_eq!(listed.len(), 1, "a restart is not a second node");
519        assert_eq!(listed[0].restarts, 1);
520    }
521
522    #[test]
523    fn the_table_is_bounded_and_trusted_nodes_survive_pressure() {
524        let registry = MeshRegistry::new();
525        let (id, adv) = node();
526        registry.observe(
527            &adv,
528            id.public.node_id.clone(),
529            TrustState::Trusted("allowlist".into()),
530            MeshSource::Synthetic,
531            "t",
532            serde_json::Value::Null,
533        );
534        for i in 0..(MAX_NODES + 10) {
535            let (other_id, mut other) = node();
536            other.seq = 1;
537            other.name = format!("n{i}");
538            registry.observe(
539                &other,
540                other_id.public.node_id.clone(),
541                TrustState::Observed,
542                MeshSource::Synthetic,
543                "s",
544                serde_json::Value::Null,
545            );
546        }
547        let listed = registry.list();
548        assert!(listed.len() <= MAX_NODES);
549        assert!(
550            listed.iter().any(|r| r.node_id == id.public.node_id),
551            "the trusted node was not evicted by untrusted pressure"
552        );
553        assert!(registry.tallies().evicted > 0);
554    }
555
556    #[test]
557    fn listing_is_in_node_id_order() {
558        let registry = MeshRegistry::new();
559        for _ in 0..5 {
560            let (id, adv) = node();
561            registry.observe(
562                &adv,
563                id.public.node_id.clone(),
564                TrustState::Observed,
565                MeshSource::Synthetic,
566                "s",
567                serde_json::Value::Null,
568            );
569        }
570        let listed = registry.list();
571        let mut ids: Vec<String> = listed.iter().map(|r| r.node_id.to_string()).collect();
572        let sorted = {
573            let mut s = ids.clone();
574            s.sort();
575            s
576        };
577        assert_eq!(ids, sorted, "the registry lists in one deterministic order");
578        ids.dedup();
579        assert_eq!(ids.len(), listed.len());
580    }
581}
majordomus_cli/mesh/registry.rs

majordomus_cli/mesh/
registry.rs
