liminal_server/
link_open.rs

1//! Carrier selection and message-scoped recovery for configured links.
2//!
3//! A refused configured dial leaves the named link down while its server keeps serving and retries.
4//! Crossings use `link_export` and `link_import` on the opened link resource and require a configured export prefix.
5
6use crate::config::types::OutboundLinkConfig;
7use crate::link_carrier::Carrier;
8use std::time::{Duration, Instant};
9
10pub mod carriers;
11pub mod network;
12pub use carriers::Stop;
13
14/// Bounds and service addresses for opening a configured link.
15#[derive(Clone, Debug, PartialEq, Eq, serde::Deserialize)]
16#[serde(default, deny_unknown_fields)]
17pub struct ChoiceConfig {
18    /// Explicit security and discovery bounds for a person's device links.
19    pub peers: Option<crate::config::peers::PeersConfig>,
20    /// Twice the estate's 1000ms design RTT budget, including a durable flush.
21    /// `docs/perf/participant-record-rtt.md` documents localhost record RTTs;
22    /// it does not itself verify the estate-wide ceiling used by this default.
23    pub crossing_deadline_ms: u64,
24    /// The broker's admitted wire listener.
25    pub broker_address: Option<String>,
26    /// The broker's observed-address echo listener.
27    pub broker_echo: Option<std::net::SocketAddr>,
28    /// The local device's UDP listener.
29    pub udp_listen: std::net::SocketAddr,
30    /// The configured relaying machine's recorded holder.
31    pub relay_holder: Option<String>,
32    /// The relay's admitted wire listener.
33    pub relay_address: Option<String>,
34}
35
36impl Default for ChoiceConfig {
37    fn default() -> Self {
38        Self {
39            peers: None,
40            crossing_deadline_ms: 2000,
41            broker_address: None,
42            broker_echo: None,
43            udp_listen: std::net::SocketAddr::from(([0, 0, 0, 0], 0)),
44            relay_holder: None,
45            relay_address: None,
46        }
47    }
48}
49
50/// A prepared crossing retains its exact sealed bytes and Noise counter.
51#[derive(Clone)]
52pub struct Packet {
53    pub subject: String,
54    pub counter: u64,
55    pub bytes: Vec<u8>,
56    pub deadline: Option<Deadline>,
57}
58
59impl std::fmt::Debug for Packet {
60    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
61        out.debug_struct("Packet")
62            .field("subject", &self.subject)
63            .field("counter", &self.counter)
64            .field("bytes", &self.bytes.len())
65            .field("deadline", &self.deadline)
66            .finish()
67    }
68}
69
70/// One crossing's absolute deadline, armed before its datagram is sent.
71#[derive(Clone, Copy, Debug, PartialEq, Eq)]
72pub struct Deadline {
73    expires: Instant,
74    budget: Duration,
75}
76
77impl Deadline {
78    /// # Errors
79    /// The configured crossing budget exceeds the clock's range.
80    pub fn after(budget: Duration) -> Result<Self, OpenError> {
81        let expires = Instant::now()
82            .checked_add(budget)
83            .ok_or_else(|| OpenError("crossing_deadline_out_of_range".to_owned()))?;
84        Ok(Self { expires, budget })
85    }
86    #[must_use]
87    pub const fn budget(self) -> Duration {
88        self.budget
89    }
90    #[must_use]
91    pub const fn expires(self) -> Instant {
92        self.expires
93    }
94}
95
96/// A keyed carrier answers one crossing or one message deadline.
97pub trait Path: Send + std::fmt::Debug {
98    fn carrier(&self) -> Carrier;
99    fn interrupt_handle(&self) -> Option<Stop> {
100        None
101    }
102    fn relay_device(&self) -> Option<String>;
103    fn life(&self) -> Option<crate::link::LinkLife> {
104        None
105    }
106    /// # Errors
107    /// Unsolicited traffic or an observed carrier end.
108    fn refuse_unsolicited(&self) -> Result<(), OpenError> {
109        Ok(())
110    }
111    /// The device's local direct index; streams have none.
112    fn direct_index(&self) -> Option<u32> {
113        None
114    }
115    /// # Errors
116    /// Invalid input, oversized message, or exhausted Noise nonce.
117    fn prepare(&mut self, subject: &str, plain: &[u8]) -> Result<Packet, OpenError>;
118    /// # Errors
119    /// The transport could not move this prepared packet.
120    fn transmit(&mut self, packet: &Packet) -> Result<(), OpenError>;
121    /// The bound exists only while this crossing is in flight.
122    /// # Errors
123    /// The carrier ended or the importing endpoint refused the crossing.
124    fn receipt(&mut self, counter: u64, bound: Option<Deadline>) -> Result<bool, OpenError>;
125}
126
127/// The existing dial, broker, punch, and relay implementations meet here.
128pub trait Backend: Send + std::fmt::Debug {
129    /// Interrupts the backend's owned blocking operations during shutdown.
130    fn interrupt_handle(&self) -> Option<Stop> {
131        None
132    }
133    /// # Errors
134    /// Carries the dialer's own named refusal.
135    fn wire(
136        &mut self,
137        config: &OutboundLinkConfig,
138        address: &str,
139    ) -> Result<Box<dyn Path>, OpenError>;
140    /// # Errors
141    /// The broker could not return the peer's registered addresses.
142    fn lookup(&mut self, config: &OutboundLinkConfig) -> Result<Vec<String>, OpenError>;
143    /// # Errors
144    /// The direct KK link could not open.
145    fn punch(&mut self, config: &OutboundLinkConfig) -> Result<Box<dyn Path>, OpenError>;
146    /// # Errors
147    /// The relay or peer refused its endpoint-keyed link.
148    fn relay(&mut self, config: &OutboundLinkConfig) -> Result<Box<dyn Path>, OpenError>;
149    /// Refreshes the mapping while retaining this packet's transport and counter.
150    /// # Errors
151    /// The broker or peer could not refresh the direct endpoint.
152    fn repunch(
153        &mut self,
154        config: &OutboundLinkConfig,
155        path: &mut dyn Path,
156    ) -> Result<(), OpenError>;
157    /// # Errors
158    /// The topology consumer could not record the observed fact.
159    fn topology(
160        &mut self,
161        carrier: Option<Carrier>,
162        relay: Option<String>,
163        refusal: Option<&str>,
164    ) -> Result<(), OpenError>;
165}
166
167/// A named refusal with no credential or message bytes.
168#[derive(Debug, thiserror::Error)]
169#[error("{0}")]
170pub struct OpenError(pub String);
171
172/// One selected carrier and its on-demand recovery owner.
173#[derive(Debug)]
174pub struct Opened<B: Backend> {
175    config: OutboundLinkConfig,
176    choice: ChoiceConfig,
177    backend: B,
178    path: Option<Box<dyn Path>>,
179    timers: usize,
180    interrupt: carriers::StopSlot,
181}
182
183impl<B: Backend> Opened<B> {
184    pub(crate) fn stopped(&self) -> bool {
185        self.interrupt.stopped()
186    }
187    pub(crate) fn life(&self) -> Option<crate::link::LinkLife> {
188        self.path.as_ref().and_then(|path| path.life())
189    }
190    pub(crate) fn refuse_unsolicited(&self) -> Result<(), OpenError> {
191        self.path.as_ref().map_or_else(
192            || Err(OpenError("link_has_no_carrier".to_owned())),
193            |path| path.refuse_unsolicited(),
194        )
195    }
196    pub(crate) const fn description(&self) -> &OutboundLinkConfig {
197        &self.config
198    }
199    pub(crate) const fn backend(&self) -> &B {
200        &self.backend
201    }
202    pub(crate) fn observed_carrier(&self) -> Option<crate::console::records::Carrier> {
203        self.path.as_ref().and_then(|path| match path.carrier() {
204            Carrier::Wire => Some(crate::console::records::Carrier::Wire),
205            Carrier::Direct => Some(crate::console::records::Carrier::Direct),
206            Carrier::Relay => path
207                .relay_device()
208                .map(crate::console::records::Carrier::Relay),
209        })
210    }
211    /// # Errors
212    /// This backend does not provide a shutdown interruption.
213    pub fn interrupt_handle(&self) -> Result<Stop, OpenError> {
214        self.interrupt.handle()
215    }
216    #[must_use]
217    pub fn carrier(&self) -> Option<Carrier> {
218        self.path.as_ref().map(|path| path.carrier())
219    }
220    #[must_use]
221    pub const fn armed_timers(&self) -> usize {
222        self.timers
223    }
224    /// Sends a crossing body already encoded by its exporting subscription.
225    /// A production direct receipt is the importing publish response, never a transport ACK.
226    /// # Errors
227    /// Named open, carrier, crossing, or topology refusal.
228    pub fn send(&mut self, subject: &str, plain: &[u8]) -> Result<(), OpenError> {
229        if self.stopped() {
230            return Err(OpenError("link_stopped".to_owned()));
231        }
232        crate::auth_lys::SubjectBinding::new(
233            "wire".to_owned(),
234            subject.to_owned(),
235            subject.to_owned(),
236        )
237        .map_err(|error| OpenError(error.to_string()))?;
238        if !self.config.exports.iter().any(|prefix| {
239            subject == prefix
240                || subject
241                    .strip_prefix(prefix)
242                    .is_some_and(|suffix| suffix.starts_with('.'))
243        }) {
244            return Err(OpenError("subject_not_exported_by_link".to_owned()));
245        }
246        if self.path.is_none() {
247            let path = select(&self.config, &mut self.backend)?;
248            self.interrupt.set(
249                path.interrupt_handle()
250                    .or_else(|| self.backend.interrupt_handle()),
251            )?;
252            self.path = Some(path);
253        }
254        let mut path = self
255            .path
256            .take()
257            .ok_or_else(|| OpenError("link_has_no_carrier".to_owned()))?;
258        let result = self.cross(path.as_mut(), subject, plain);
259        self.timers = 0;
260        match result {
261            Ok(Crossing::Delivered) => {
262                self.path = Some(path);
263                Ok(())
264            }
265            Ok(Crossing::Relay) => self.cross_relay(subject, plain),
266            Err(error) => {
267                self.backend
268                    .topology(Some(path.carrier()), path.relay_device(), Some(&error.0))?;
269                if error.0.starts_with("crossing_refused:") {
270                    self.path = Some(path);
271                }
272                Err(error)
273            }
274        }
275    }
276
277    fn cross(
278        &mut self,
279        path: &mut dyn Path,
280        subject: &str,
281        plain: &[u8],
282    ) -> Result<Crossing, OpenError> {
283        let mut packet = match path.prepare(subject, plain) {
284            Ok(packet) => packet,
285            Err(error)
286                if path.carrier() == Carrier::Direct && error.0 == "too_large_for_datagram" =>
287            {
288                self.backend
289                    .topology(Some(Carrier::Direct), None, Some(&error.0))?;
290                return Ok(Crossing::Relay);
291            }
292            Err(error) => return Err(error),
293        };
294        if path.carrier() != Carrier::Direct {
295            path.transmit(&packet)?;
296            return if path.receipt(packet.counter, None)? {
297                Ok(Crossing::Delivered)
298            } else {
299                Err(OpenError("stream_crossing_has_no_receipt".to_owned()))
300            };
301        }
302        let bound = Duration::from_millis(self.choice.crossing_deadline_ms);
303        packet.deadline = Some(Deadline::after(bound)?);
304        self.timers = 1;
305        path.transmit(&packet)?;
306        if path.receipt(packet.counter, packet.deadline)? {
307            return Ok(Crossing::Delivered);
308        }
309        self.backend
310            .topology(Some(Carrier::Direct), None, Some("crossing_deadline"))?;
311        if let Err(error) = self.backend.repunch(&self.config, path) {
312            self.backend
313                .topology(Some(Carrier::Direct), None, Some(&error.0))?;
314            return Ok(Crossing::Relay);
315        }
316        // Reusing ciphertext is essential: sealing again would consume a second nonce.
317        packet.deadline = Some(Deadline::after(bound)?);
318        path.transmit(&packet)?;
319        if path.receipt(packet.counter, packet.deadline)? {
320            return Ok(Crossing::Delivered);
321        }
322        self.backend.topology(
323            Some(Carrier::Direct),
324            None,
325            Some("second_crossing_deadline"),
326        )?;
327        Ok(Crossing::Relay)
328    }
329
330    fn cross_relay(&mut self, subject: &str, plain: &[u8]) -> Result<(), OpenError> {
331        let mut relay = match self.backend.relay(&self.config) {
332            Ok(relay) => relay,
333            Err(error) => {
334                self.backend
335                    .topology(Some(Carrier::Relay), None, Some(&error.0))?;
336                return Err(error);
337            }
338        };
339        self.interrupt.set(
340            relay
341                .interrupt_handle()
342                .or_else(|| self.backend.interrupt_handle()),
343        )?;
344        self.backend
345            .topology(Some(relay.carrier()), relay.relay_device(), None)?;
346        let crossing = (|| {
347            let packet = relay.prepare(subject, plain)?;
348            relay.transmit(&packet)?;
349            if !relay.receipt(packet.counter, None)? {
350                return Err(OpenError("relay_crossing_has_no_receipt".to_owned()));
351            }
352            Ok(())
353        })();
354        if let Err(error) = crossing {
355            self.backend
356                .topology(Some(Carrier::Relay), relay.relay_device(), Some(&error.0))?;
357            return Err(error);
358        }
359        self.path = Some(relay);
360        Ok(())
361    }
362}
363
364enum Crossing {
365    Delivered,
366    Relay,
367}
368
369fn record<B: Backend>(
370    backend: &mut B,
371    carrier: Carrier,
372    result: Result<Box<dyn Path>, OpenError>,
373    refusals: &mut Vec<String>,
374) -> Result<Option<Box<dyn Path>>, OpenError> {
375    match result {
376        Ok(path) => {
377            backend.topology(Some(path.carrier()), path.relay_device(), None)?;
378            Ok(Some(path))
379        }
380        Err(error) => {
381            let name = match carrier {
382                Carrier::Wire => "wire",
383                Carrier::Direct => "direct",
384                Carrier::Relay => "relay",
385            };
386            refusals.push(format!("{name}={}", error.0));
387            backend.topology(Some(carrier), None, Some(&error.0))?;
388            if error.0.starts_with("link_held:")
389                || error.0.starts_with("link_refused:")
390                || error.0 == "network_stopped"
391            {
392                return Err(error);
393            }
394            Ok(None)
395        }
396    }
397}
398
399fn select<B: Backend>(
400    config: &OutboundLinkConfig,
401    backend: &mut B,
402) -> Result<Box<dyn Path>, OpenError> {
403    let mut refusals = Vec::new();
404    if let Some(address) = &config.peer_address {
405        let result = backend.wire(config, address);
406        if let Some(path) = record(backend, Carrier::Wire, result, &mut refusals)? {
407            return Ok(path);
408        }
409    } else {
410        let addresses = match backend.lookup(config) {
411            Ok(addresses) => addresses,
412            Err(error) => {
413                refusals.push(format!("lookup={}", error.0));
414                backend.topology(None, None, Some(&error.0))?;
415                Vec::new()
416            }
417        };
418        let mut tried = std::collections::BTreeSet::new();
419        for address in addresses {
420            if !tried.insert(address.clone()) {
421                continue;
422            }
423            let result = backend.wire(config, &address);
424            if let Some(path) = record(backend, Carrier::Wire, result, &mut refusals)? {
425                return Ok(path);
426            }
427        }
428    }
429    let result = backend.punch(config);
430    if let Some(path) = record(backend, Carrier::Direct, result, &mut refusals)? {
431        return Ok(path);
432    }
433    let result = backend.relay(config);
434    if let Some(path) = record(backend, Carrier::Relay, result, &mut refusals)? {
435        return Ok(path);
436    }
437    let reason = format!("all_link_carriers_refused: {}", refusals.join("; "));
438    backend.topology(None, None, Some(&reason))?;
439    Err(OpenError(reason))
440}
441
442/// Opens each carrier once, in wire, direct, relay order.
443/// # Errors
444/// Every attempted carrier refused, or the configuration was invalid.
445pub fn open<B: Backend>(
446    config: OutboundLinkConfig,
447    choice: ChoiceConfig,
448    mut backend: B,
449) -> Result<Opened<B>, OpenError> {
450    if choice.crossing_deadline_ms == 0 {
451        return Err(OpenError(
452            "crossing_deadline_ms_must_be_positive".to_owned(),
453        ));
454    }
455    if config.link_id.is_empty()
456        || config
457            .link_id
458            .chars()
459            .any(|character| character.is_whitespace() || character.is_control())
460    {
461        return Err(OpenError("invalid_link_id".to_owned()));
462    }
463    for subject in &config.exports {
464        crate::auth_lys::SubjectBinding::new("wire".to_owned(), subject.clone(), subject.clone())
465            .map_err(|error| OpenError(error.to_string()))?;
466    }
467    let path = select(&config, &mut backend)?;
468    let interrupt = carriers::StopSlot::new(
469        path.interrupt_handle()
470            .or_else(|| backend.interrupt_handle()),
471    );
472    Ok(Opened {
473        config,
474        choice,
475        backend,
476        path: Some(path),
477        timers: 0,
478        interrupt,
479    })
480}
liminal_server/link_open.rs

liminal_server/
link_open.rs
