liminal_server/
link_federated.rs

1//! Far discovery is generation-bound, prefix-filtered and never guessed.
2
3use serde_json::Value;
4use std::collections::BTreeMap;
5use std::num::NonZeroUsize;
6use std::sync::{Mutex, MutexGuard};
7
8#[path = "link_federated/projection.rs"]
9mod projection;
10
11/// Bounds checked before any peer-controlled discovery data is retained.
12#[derive(Clone, Copy, Debug)]
13pub struct ProjectionLimits {
14    /// Maximum snapshot bytes, bounding JSON parsing and retained contracts.
15    pub bytes: NonZeroUsize,
16    /// Maximum stations per snapshot, bounding directory indexing work.
17    pub stations: NonZeroUsize,
18    /// Maximum distinct links, bounding total discovery memory.
19    pub peers: NonZeroUsize,
20}
21
22/// Discovery refuses without leaking schema or message bytes.
23#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
24pub enum DiscoveryError {
25    /// Input or retained discovery exceeded an operator bound.
26    #[error("far_topology_bound")]
27    Bound,
28    /// The far document is not a whole topology from its own device.
29    #[error("far_topology_invalid")]
30    Invalid,
31    /// A stale link generation or topology revision cannot replace current data.
32    #[error("far_topology_stale")]
33    Stale,
34    /// A poisoned registry cannot supply trusted discovery.
35    #[error("far_topology_unavailable")]
36    Unavailable,
37    /// The monotonic link generation is exhausted.
38    #[error("far_topology_generation_exhausted")]
39    Exhausted,
40}
41
42#[derive(Default)]
43struct State {
44    generation: u64,
45    links: BTreeMap<String, FarLink>,
46}
47struct FarLink {
48    generation: u64,
49    peer: String,
50    prefixes: Vec<String>,
51    revision: Option<u64>,
52    document: Option<Value>,
53}
54
55/// One authoritative index used for topology, describe and service routing.
56pub struct FarTopology {
57    limits: ProjectionLimits,
58    state: Mutex<State>,
59}
60
61impl std::fmt::Debug for FarTopology {
62    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
63        out.debug_struct("FarTopology")
64            .field("limits", &self.limits)
65            .finish_non_exhaustive()
66    }
67}
68
69impl FarTopology {
70    /// Creates an empty index with required positive bounds.
71    #[must_use]
72    pub fn new(limits: ProjectionLimits) -> Self {
73        Self {
74            limits,
75            state: Mutex::default(),
76        }
77    }
78
79    fn lock(&self) -> Result<MutexGuard<'_, State>, DiscoveryError> {
80        self.state.lock().map_err(|_| DiscoveryError::Unavailable)
81    }
82
83    /// Binds this generation to the proved peer and configured export prefixes.
84    ///
85    /// # Errors
86    /// Invalid identities/prefixes, peer bound, poisoned state or exhausted ids.
87    pub fn open(
88        &self,
89        link: &str,
90        peer: &str,
91        prefixes: Vec<String>,
92    ) -> Result<u64, DiscoveryError> {
93        if link.is_empty()
94            || peer.is_empty()
95            || prefixes.is_empty()
96            || link
97                .chars()
98                .chain(peer.chars())
99                .any(|character| character.is_control() || character.is_whitespace())
100        {
101            return Err(DiscoveryError::Invalid);
102        }
103        for prefix in &prefixes {
104            crate::auth_lys::SubjectBinding::new("wire".into(), prefix.clone(), prefix.clone())
105                .map_err(|_| DiscoveryError::Invalid)?;
106            if prefix == "liminal"
107                || prefix.starts_with("liminal.")
108                || prefix == "console"
109                || prefix.starts_with("console.")
110            {
111                return Err(DiscoveryError::Invalid);
112            }
113        }
114        let mut state = self.lock()?;
115        if !state.links.contains_key(link) && state.links.len() >= self.limits.peers.get() {
116            return Err(DiscoveryError::Bound);
117        }
118        state.generation = state
119            .generation
120            .checked_add(1)
121            .ok_or(DiscoveryError::Exhausted)?;
122        let generation = state.generation;
123        state.links.insert(
124            link.into(),
125            FarLink {
126                generation,
127                peer: peer.into(),
128                prefixes,
129                revision: None,
130                document: None,
131            },
132        );
133        drop(state);
134        Ok(generation)
135    }
136
137    /// Accepts a whole far snapshot only from the current keyed generation.
138    ///
139    /// # Errors
140    /// Oversize, malformed or forged document, stale revision/generation, or poisoned state.
141    pub fn update(
142        &self,
143        link: &str,
144        generation: u64,
145        bytes: &[u8],
146    ) -> Result<bool, DiscoveryError> {
147        if bytes.len() > self.limits.bytes.get() {
148            return Err(DiscoveryError::Bound);
149        }
150        let document: Value = serde_json::from_slice(bytes).map_err(|_| DiscoveryError::Invalid)?;
151        let revision = projection::validate(&document, self.limits.stations.get())?;
152        let mut state = self.lock()?;
153        let far = state
154            .links
155            .get_mut(link)
156            .filter(|far| far.generation == generation)
157            .ok_or(DiscoveryError::Stale)?;
158        if far.revision.is_some_and(|previous| previous > revision) {
159            return Err(DiscoveryError::Stale);
160        }
161        if far.revision == Some(revision) {
162            return if far.document.as_ref() == Some(&document) {
163                Ok(false)
164            } else {
165                Err(DiscoveryError::Stale)
166            };
167        }
168        far.revision = Some(revision);
169        far.document = Some(document);
170        drop(state);
171        Ok(true)
172    }
173
174    /// Retires only this generation; a replaced carrier cannot remove its successor.
175    ///
176    /// # Errors
177    /// The registry lock is poisoned.
178    pub fn close(&self, link: &str, generation: u64) -> Result<bool, DiscoveryError> {
179        let mut state = self.lock()?;
180        if state
181            .links
182            .get(link)
183            .is_some_and(|far| far.generation == generation)
184        {
185            state.links.remove(link);
186            drop(state);
187            Ok(true)
188        } else {
189            Ok(false)
190        }
191    }
192
193    /// Number of retained peer generations, including peers awaiting discovery.
194    ///
195    /// # Errors
196    /// The registry lock is poisoned.
197    pub fn pending_peers(&self) -> Result<usize, DiscoveryError> {
198        Ok(self.lock()?.links.len())
199    }
200
201    /// Describes the actual far station serving an exported subject.
202    ///
203    /// # Errors
204    /// The registry or a retained document is invalid.
205    pub fn describe(&self, subject: &str) -> Result<Option<Value>, DiscoveryError> {
206        let state = self.lock()?;
207        for (link, far) in &state.links {
208            for station in projection::stations(link, far)? {
209                if station["subjects"].as_array().is_some_and(|subjects| {
210                    subjects.iter().any(|item| item.as_str() == Some(subject))
211                }) {
212                    return Ok(Some(station));
213                }
214            }
215        }
216        drop(state);
217        Ok(None)
218    }
219
220    /// Projects only currently up links, preserving the supplied local revision.
221    /// A down record and the removal of its far stations are one atomic document.
222    ///
223    /// # Errors
224    /// Malformed local/far topology, oversize output, encoding or registry failure.
225    pub fn project(&self, bytes: &[u8]) -> Result<Vec<u8>, DiscoveryError> {
226        projection::project(self, bytes)
227    }
228}
229
230fn reaches(prefix: &str, subject: &str) -> bool {
231    subject == prefix
232        || subject
233            .strip_prefix(prefix)
234            .is_some_and(|suffix| suffix.starts_with('.'))
235}
liminal_server/link_federated.rs

liminal_server/
link_federated.rs
