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

liminal_server/
link_open.rs
