liminal-server 0.25.4

//! Carrier selection and message-scoped recovery for configured links.
//!
//! A refused configured dial leaves the named link down while its server keeps serving and retries.
//! Crossings use `link_export` and `link_import` on the opened link resource and require a configured export prefix.

use crate::config::types::OutboundLinkConfig;
use crate::link_carrier::Carrier;
use std::time::{Duration, Instant};

pub mod carriers;
pub mod network;
pub use carriers::Stop;

/// Bounds and service addresses for opening a configured link.
#[derive(Clone, Debug, PartialEq, Eq, serde::Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct ChoiceConfig {
    /// Explicit security and discovery bounds for a person's device links.
    pub peers: Option<crate::config::peers::PeersConfig>,
    /// Twice the estate's 1000ms design RTT budget, including a durable flush.
    /// `docs/perf/participant-record-rtt.md` documents localhost record RTTs;
    /// it does not itself verify the estate-wide ceiling used by this default.
    pub crossing_deadline_ms: u64,
    /// The broker's admitted wire listener.
    pub broker_address: Option<String>,
    /// The broker's observed-address echo listener.
    pub broker_echo: Option<std::net::SocketAddr>,
    /// The local device's UDP listener.
    pub udp_listen: std::net::SocketAddr,
    /// The configured relaying machine's recorded holder.
    pub relay_holder: Option<String>,
    /// The relay's admitted wire listener.
    pub relay_address: Option<String>,
}

impl Default for ChoiceConfig {
    fn default() -> Self {
        Self {
            peers: None,
            crossing_deadline_ms: 2000,
            broker_address: None,
            broker_echo: None,
            udp_listen: std::net::SocketAddr::from(([0, 0, 0, 0], 0)),
            relay_holder: None,
            relay_address: None,
        }
    }
}

/// A prepared crossing retains its exact sealed bytes and Noise counter.
#[derive(Clone)]
pub struct Packet {
    pub subject: String,
    pub counter: u64,
    pub bytes: Vec<u8>,
    pub deadline: Option<Deadline>,
}

impl std::fmt::Debug for Packet {
    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        out.debug_struct("Packet")
            .field("subject", &self.subject)
            .field("counter", &self.counter)
            .field("bytes", &self.bytes.len())
            .field("deadline", &self.deadline)
            .finish()
    }
}

/// One crossing's absolute deadline, armed before its datagram is sent.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Deadline {
    expires: Instant,
    budget: Duration,
}

impl Deadline {
    /// # Errors
    /// The configured crossing budget exceeds the clock's range.
    pub fn after(budget: Duration) -> Result<Self, OpenError> {
        let expires = Instant::now()
            .checked_add(budget)
            .ok_or_else(|| OpenError("crossing_deadline_out_of_range".to_owned()))?;
        Ok(Self { expires, budget })
    }
    #[must_use]
    pub const fn budget(self) -> Duration {
        self.budget
    }
    #[must_use]
    pub const fn expires(self) -> Instant {
        self.expires
    }
}

/// A keyed carrier answers one crossing or one message deadline.
pub trait Path: Send + std::fmt::Debug {
    fn carrier(&self) -> Carrier;
    fn interrupt_handle(&self) -> Option<Stop> {
        None
    }
    fn relay_device(&self) -> Option<String>;
    fn life(&self) -> Option<crate::link::LinkLife> {
        None
    }
    /// # Errors
    /// Unsolicited traffic or an observed carrier end.
    fn refuse_unsolicited(&self) -> Result<(), OpenError> {
        Ok(())
    }
    /// The device's local direct index; streams have none.
    fn direct_index(&self) -> Option<u32> {
        None
    }
    /// # Errors
    /// Invalid input, oversized message, or exhausted Noise nonce.
    fn prepare(&mut self, subject: &str, plain: &[u8]) -> Result<Packet, OpenError>;
    /// # Errors
    /// The transport could not move this prepared packet.
    fn transmit(&mut self, packet: &Packet) -> Result<(), OpenError>;
    /// The bound exists only while this crossing is in flight.
    /// # Errors
    /// The carrier ended or the importing endpoint refused the crossing.
    fn receipt(&mut self, counter: u64, bound: Option<Deadline>) -> Result<bool, OpenError>;
}

/// The existing dial, broker, punch, and relay implementations meet here.
pub trait Backend: Send + std::fmt::Debug {
    /// Interrupts the backend's owned blocking operations during shutdown.
    fn interrupt_handle(&self) -> Option<Stop> {
        None
    }
    /// # Errors
    /// Carries the dialer's own named refusal.
    fn wire(
        &mut self,
        config: &OutboundLinkConfig,
        address: &str,
    ) -> Result<Box<dyn Path>, OpenError>;
    /// # Errors
    /// The broker could not return the peer's registered addresses.
    fn lookup(&mut self, config: &OutboundLinkConfig) -> Result<Vec<String>, OpenError>;
    /// # Errors
    /// The direct KK link could not open.
    fn punch(&mut self, config: &OutboundLinkConfig) -> Result<Box<dyn Path>, OpenError>;
    /// # Errors
    /// The relay or peer refused its endpoint-keyed link.
    fn relay(&mut self, config: &OutboundLinkConfig) -> Result<Box<dyn Path>, OpenError>;
    /// Refreshes the mapping while retaining this packet's transport and counter.
    /// # Errors
    /// The broker or peer could not refresh the direct endpoint.
    fn repunch(
        &mut self,
        config: &OutboundLinkConfig,
        path: &mut dyn Path,
    ) -> Result<(), OpenError>;
    /// # Errors
    /// The topology consumer could not record the observed fact.
    fn topology(
        &mut self,
        carrier: Option<Carrier>,
        relay: Option<String>,
        refusal: Option<&str>,
    ) -> Result<(), OpenError>;
}

/// A named refusal with no credential or message bytes.
#[derive(Debug, thiserror::Error)]
#[error("{0}")]
pub struct OpenError(pub String);

/// One selected carrier and its on-demand recovery owner.
#[derive(Debug)]
pub struct Opened<B: Backend> {
    config: OutboundLinkConfig,
    choice: ChoiceConfig,
    backend: B,
    path: Option<Box<dyn Path>>,
    timers: usize,
    interrupt: carriers::StopSlot,
}

impl<B: Backend> Opened<B> {
    pub(crate) fn stopped(&self) -> bool {
        self.interrupt.stopped()
    }
    pub(crate) fn life(&self) -> Option<crate::link::LinkLife> {
        self.path.as_ref().and_then(|path| path.life())
    }
    pub(crate) fn refuse_unsolicited(&self) -> Result<(), OpenError> {
        self.path.as_ref().map_or_else(
            || Err(OpenError("link_has_no_carrier".to_owned())),
            |path| path.refuse_unsolicited(),
        )
    }
    pub(crate) const fn description(&self) -> &OutboundLinkConfig {
        &self.config
    }
    pub(crate) const fn backend(&self) -> &B {
        &self.backend
    }
    pub(crate) fn observed_carrier(&self) -> Option<crate::console::records::Carrier> {
        self.path.as_ref().and_then(|path| match path.carrier() {
            Carrier::Wire => Some(crate::console::records::Carrier::Wire),
            Carrier::Direct => Some(crate::console::records::Carrier::Direct),
            Carrier::Relay => path
                .relay_device()
                .map(crate::console::records::Carrier::Relay),
        })
    }
    /// # Errors
    /// This backend does not provide a shutdown interruption.
    pub fn interrupt_handle(&self) -> Result<Stop, OpenError> {
        self.interrupt.handle()
    }
    #[must_use]
    pub fn carrier(&self) -> Option<Carrier> {
        self.path.as_ref().map(|path| path.carrier())
    }
    #[must_use]
    pub const fn armed_timers(&self) -> usize {
        self.timers
    }
    /// Sends a crossing body already encoded by its exporting subscription.
    /// A production direct receipt is the importing publish response, never a transport ACK.
    /// # Errors
    /// Named open, carrier, crossing, or topology refusal.
    pub fn send(&mut self, subject: &str, plain: &[u8]) -> Result<(), OpenError> {
        if self.stopped() {
            return Err(OpenError("link_stopped".to_owned()));
        }
        crate::auth_lys::SubjectBinding::new(
            "wire".to_owned(),
            subject.to_owned(),
            subject.to_owned(),
        )
        .map_err(|error| OpenError(error.to_string()))?;
        if !self.config.exports.iter().any(|prefix| {
            subject == prefix
                || subject
                    .strip_prefix(prefix)
                    .is_some_and(|suffix| suffix.starts_with('.'))
        }) {
            return Err(OpenError("subject_not_exported_by_link".to_owned()));
        }
        if self.path.is_none() {
            let path = select(&self.config, &mut self.backend)?;
            self.interrupt.set(
                path.interrupt_handle()
                    .or_else(|| self.backend.interrupt_handle()),
            )?;
            self.path = Some(path);
        }
        let mut path = self
            .path
            .take()
            .ok_or_else(|| OpenError("link_has_no_carrier".to_owned()))?;
        let result = self.cross(path.as_mut(), subject, plain);
        self.timers = 0;
        match result {
            Ok(Crossing::Delivered) => {
                self.path = Some(path);
                Ok(())
            }
            Ok(Crossing::Relay) => self.cross_relay(subject, plain),
            Err(error) => {
                self.backend
                    .topology(Some(path.carrier()), path.relay_device(), Some(&error.0))?;
                if error.0.starts_with("crossing_refused:") {
                    self.path = Some(path);
                }
                Err(error)
            }
        }
    }

    fn cross(
        &mut self,
        path: &mut dyn Path,
        subject: &str,
        plain: &[u8],
    ) -> Result<Crossing, OpenError> {
        let mut packet = match path.prepare(subject, plain) {
            Ok(packet) => packet,
            Err(error)
                if path.carrier() == Carrier::Direct && error.0 == "too_large_for_datagram" =>
            {
                self.backend
                    .topology(Some(Carrier::Direct), None, Some(&error.0))?;
                return Ok(Crossing::Relay);
            }
            Err(error) => return Err(error),
        };
        if path.carrier() != Carrier::Direct {
            path.transmit(&packet)?;
            return if path.receipt(packet.counter, None)? {
                Ok(Crossing::Delivered)
            } else {
                Err(OpenError("stream_crossing_has_no_receipt".to_owned()))
            };
        }
        let bound = Duration::from_millis(self.choice.crossing_deadline_ms);
        packet.deadline = Some(Deadline::after(bound)?);
        self.timers = 1;
        path.transmit(&packet)?;
        if path.receipt(packet.counter, packet.deadline)? {
            return Ok(Crossing::Delivered);
        }
        self.backend
            .topology(Some(Carrier::Direct), None, Some("crossing_deadline"))?;
        if let Err(error) = self.backend.repunch(&self.config, path) {
            self.backend
                .topology(Some(Carrier::Direct), None, Some(&error.0))?;
            return Ok(Crossing::Relay);
        }
        // Reusing ciphertext is essential: sealing again would consume a second nonce.
        packet.deadline = Some(Deadline::after(bound)?);
        path.transmit(&packet)?;
        if path.receipt(packet.counter, packet.deadline)? {
            return Ok(Crossing::Delivered);
        }
        self.backend.topology(
            Some(Carrier::Direct),
            None,
            Some("second_crossing_deadline"),
        )?;
        Ok(Crossing::Relay)
    }

    fn cross_relay(&mut self, subject: &str, plain: &[u8]) -> Result<(), OpenError> {
        let mut relay = match self.backend.relay(&self.config) {
            Ok(relay) => relay,
            Err(error) => {
                self.backend
                    .topology(Some(Carrier::Relay), None, Some(&error.0))?;
                return Err(error);
            }
        };
        self.interrupt.set(
            relay
                .interrupt_handle()
                .or_else(|| self.backend.interrupt_handle()),
        )?;
        self.backend
            .topology(Some(relay.carrier()), relay.relay_device(), None)?;
        let crossing = (|| {
            let packet = relay.prepare(subject, plain)?;
            relay.transmit(&packet)?;
            if !relay.receipt(packet.counter, None)? {
                return Err(OpenError("relay_crossing_has_no_receipt".to_owned()));
            }
            Ok(())
        })();
        if let Err(error) = crossing {
            self.backend
                .topology(Some(Carrier::Relay), relay.relay_device(), Some(&error.0))?;
            return Err(error);
        }
        self.path = Some(relay);
        Ok(())
    }
}

enum Crossing {
    Delivered,
    Relay,
}

fn record<B: Backend>(
    backend: &mut B,
    carrier: Carrier,
    result: Result<Box<dyn Path>, OpenError>,
    refusals: &mut Vec<String>,
) -> Result<Option<Box<dyn Path>>, OpenError> {
    match result {
        Ok(path) => {
            backend.topology(Some(path.carrier()), path.relay_device(), None)?;
            Ok(Some(path))
        }
        Err(error) => {
            let name = match carrier {
                Carrier::Wire => "wire",
                Carrier::Direct => "direct",
                Carrier::Relay => "relay",
            };
            refusals.push(format!("{name}={}", error.0));
            backend.topology(Some(carrier), None, Some(&error.0))?;
            if error.0.starts_with("link_held:")
                || error.0.starts_with("link_refused:")
                || error.0 == "network_stopped"
            {
                return Err(error);
            }
            Ok(None)
        }
    }
}

fn select<B: Backend>(
    config: &OutboundLinkConfig,
    backend: &mut B,
) -> Result<Box<dyn Path>, OpenError> {
    let mut refusals = Vec::new();
    if let Some(address) = &config.peer_address {
        let result = backend.wire(config, address);
        if let Some(path) = record(backend, Carrier::Wire, result, &mut refusals)? {
            return Ok(path);
        }
    } else {
        let addresses = match backend.lookup(config) {
            Ok(addresses) => addresses,
            Err(error) => {
                refusals.push(format!("lookup={}", error.0));
                backend.topology(None, None, Some(&error.0))?;
                Vec::new()
            }
        };
        let mut tried = std::collections::BTreeSet::new();
        for address in addresses {
            if !tried.insert(address.clone()) {
                continue;
            }
            let result = backend.wire(config, &address);
            if let Some(path) = record(backend, Carrier::Wire, result, &mut refusals)? {
                return Ok(path);
            }
        }
    }
    let result = backend.punch(config);
    if let Some(path) = record(backend, Carrier::Direct, result, &mut refusals)? {
        return Ok(path);
    }
    let result = backend.relay(config);
    if let Some(path) = record(backend, Carrier::Relay, result, &mut refusals)? {
        return Ok(path);
    }
    let reason = format!("all_link_carriers_refused: {}", refusals.join("; "));
    backend.topology(None, None, Some(&reason))?;
    Err(OpenError(reason))
}

/// Opens each carrier once, in wire, direct, relay order.
/// # Errors
/// Every attempted carrier refused, or the configuration was invalid.
pub fn open<B: Backend>(
    config: OutboundLinkConfig,
    choice: ChoiceConfig,
    mut backend: B,
) -> Result<Opened<B>, OpenError> {
    if choice.crossing_deadline_ms == 0 {
        return Err(OpenError(
            "crossing_deadline_ms_must_be_positive".to_owned(),
        ));
    }
    if config.link_id.is_empty()
        || config
            .link_id
            .chars()
            .any(|character| character.is_whitespace() || character.is_control())
    {
        return Err(OpenError("invalid_link_id".to_owned()));
    }
    for subject in &config.exports {
        crate::auth_lys::SubjectBinding::new("wire".to_owned(), subject.clone(), subject.clone())
            .map_err(|error| OpenError(error.to_string()))?;
    }
    let path = select(&config, &mut backend)?;
    let interrupt = carriers::StopSlot::new(
        path.interrupt_handle()
            .or_else(|| backend.interrupt_handle()),
    );
    Ok(Opened {
        config,
        choice,
        backend,
        path: Some(path),
        timers: 0,
        interrupt,
    })
}