liminal-server 0.25.4

//! Existing stream carriers and indexed direct packets behind one crossing interface.

use super::{Deadline, OpenError, Packet, Path};
use crate::link_carrier::{Carrier, CarrierReceive, LinkCarrier};
use crate::link_dial::DialedLink;
use crate::link_handshake::{LinkTransport, NOISE_MAX_MESSAGE, NOISE_OVERHEAD};
use crate::link_relay::RelayedLink;
use crate::link_udp::{Datagram, LinkCancellation, UdpCarrier};
use liminal::protocol::{Frame, codec};
use std::sync::Arc;

/// A shutdown act over owned handles; no process names or polling.
mod stopping;
pub(crate) use stopping::StopSlot;
pub type Stop = Arc<dyn Fn() -> Result<(), OpenError> + Send + Sync>;

pub(super) fn named(error: impl std::fmt::Display) -> OpenError {
    OpenError(error.to_string())
}

/// A keyed wire or relay stream retaining the endpoint's Noise transport.
#[derive(Debug)]
pub struct StreamPath {
    transport: LinkTransport,
    carrier: Box<dyn LinkCarrier>,
    relay: Option<String>,
    interrupt: Option<Arc<std::net::TcpStream>>,
    life: Option<crate::link::LinkLife>,
    position: Option<u64>,
}

impl StreamPath {
    #[must_use]
    pub fn wire(link: DialedLink) -> Self {
        Self {
            life: link.carrier.life().cloned(),
            position: None,
            transport: link.transport,
            carrier: Box::new(link.carrier),
            relay: None,
            interrupt: None,
        }
    }
    #[must_use]
    pub fn relay(link: RelayedLink, device: String) -> Self {
        Self {
            transport: link.transport,
            carrier: Box::new(link.carrier),
            relay: Some(device),
            interrupt: None,
            life: None,
            position: None,
        }
    }
    #[must_use]
    pub fn with_interrupt(mut self, interrupt: Arc<std::net::TcpStream>) -> Self {
        self.interrupt = Some(interrupt);
        self
    }
}

impl Path for StreamPath {
    fn interrupt_handle(&self) -> Option<Stop> {
        self.interrupt.as_ref().map(|socket| {
            let socket = Arc::clone(socket);
            Arc::new(move || match socket.shutdown(std::net::Shutdown::Both) {
                Ok(()) => Ok(()),
                Err(error) if error.kind() == std::io::ErrorKind::NotConnected => Ok(()),
                Err(error) => Err(named(error)),
            }) as Stop
        })
    }
    fn life(&self) -> Option<crate::link::LinkLife> {
        self.life.clone()
    }
    fn refuse_unsolicited(&self) -> Result<(), OpenError> {
        self.carrier.refuse_unsolicited().map_err(named)
    }
    fn carrier(&self) -> Carrier {
        self.carrier.carrier()
    }
    fn relay_device(&self) -> Option<String> {
        self.relay.clone()
    }
    fn prepare(&mut self, subject: &str, plain: &[u8]) -> Result<Packet, OpenError> {
        if plain.len().saturating_add(NOISE_OVERHEAD) > NOISE_MAX_MESSAGE {
            return Err(OpenError("too_large".to_owned()));
        }
        self.position = Some(u64::from_be_bytes(
            plain
                .get(..8)
                .ok_or_else(|| OpenError("crossing_has_no_position".to_owned()))?
                .try_into()
                .map_err(named)?,
        ));
        let counter = self.transport.sending_nonce();
        let mut bytes = vec![0; plain.len() + NOISE_OVERHEAD];
        let length = self.transport.seal(plain, &mut bytes).map_err(named)?;
        bytes.truncate(length);
        Ok(Packet {
            subject: subject.to_owned(),
            counter,
            bytes,
            deadline: None,
        })
    }
    fn transmit(&mut self, packet: &Packet) -> Result<(), OpenError> {
        self.carrier
            .send(&packet.subject, &packet.bytes)
            .map_err(named)
    }
    fn receipt(&mut self, counter: u64, bound: Option<Deadline>) -> Result<bool, OpenError> {
        if counter.checked_add(1) != Some(self.transport.sending_nonce()) {
            return Err(OpenError("unexpected_stream_crossing_counter".to_owned()));
        }
        if bound.is_some() {
            return Err(OpenError(
                "stream_carrier_has_no_datagram_deadline".to_owned(),
            ));
        }
        match self.carrier.receive().map_err(named)? {
            CarrierReceive::Acknowledged { message_id, .. }
                if Some(message_id) == self.position =>
            {
                Ok(true)
            }
            CarrierReceive::Acknowledged { .. } => {
                Err(OpenError("unexpected_importing_position".to_owned()))
            }
            CarrierReceive::Refused {
                reason_code,
                reason,
            } => Err(OpenError(format!(
                "crossing_refused:{reason_code}:{}",
                reason.unwrap_or_else(|| "unnamed_crossing_refusal".to_owned()),
            ))),
            CarrierReceive::End { reason } => Err(OpenError(reason)),
            CarrierReceive::Sealed { .. } => Err(OpenError("unexpected_crossing_reply".to_owned())),
        }
    }
}

/// One indexed path on the device's shared UDP receiver.
pub struct DirectPath {
    carrier: Arc<UdpCarrier>,
    index: u32,
    inbox: crate::link_udp::streams::Inbox,
    cancellation: LinkCancellation,
    transport: Option<LinkTransport>,
    position: Option<u64>,
}

impl std::fmt::Debug for DirectPath {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("DirectPath")
            .field("index", &self.index)
            .finish_non_exhaustive()
    }
}

impl DirectPath {
    /// Takes an already keyed and indexed path.
    /// # Errors
    /// The device does not hold this index.
    pub fn new(carrier: Arc<UdpCarrier>, index: u32) -> Result<Self, OpenError> {
        carrier.watch_acknowledgements(index).map_err(named)?;
        let inbox = carrier.subscribe(index).map_err(named)?;
        let cancellation = inbox.cancellation();
        Ok(Self {
            carrier,
            index,
            inbox,
            cancellation,
            transport: None,
            position: None,
        })
    }
}

impl Path for DirectPath {
    fn interrupt_handle(&self) -> Option<Stop> {
        let cancellation = self.cancellation.clone();
        Some(Arc::new(move || cancellation.cancel().map_err(named)))
    }
    fn refuse_unsolicited(&self) -> Result<(), OpenError> {
        self.inbox.refuse_unsolicited().map_err(named)
    }
    fn carrier(&self) -> Carrier {
        Carrier::Direct
    }
    fn relay_device(&self) -> Option<String> {
        None
    }
    fn direct_index(&self) -> Option<u32> {
        Some(self.index)
    }
    fn prepare(&mut self, subject: &str, plain: &[u8]) -> Result<Packet, OpenError> {
        if plain.len() > crate::link_udp::MAX_DATAGRAM
            || subject.len() > crate::link_udp::MAX_DATAGRAM
        {
            return Err(OpenError("too_large_for_datagram".to_owned()));
        }
        let frame = Frame::LinkMessage {
            flags: 0,
            stream_id: 1,
            subject: subject.to_owned(),
            sealed: match self.transport.as_mut() {
                Some(transport) => {
                    let mut sealed = vec![0; plain.len() + NOISE_OVERHEAD];
                    let written = transport.seal(plain, &mut sealed).map_err(named)?;
                    sealed.truncate(written);
                    sealed
                }
                None => return Err(OpenError("udp_link_not_admitted".to_owned())),
            },
        };
        let length = codec::encoded_len(&frame).map_err(named)?;
        if length > crate::link_udp::MAX_DATAGRAM {
            return Err(OpenError("too_large_for_datagram".to_owned()));
        }
        let mut bytes = vec![0; length];
        codec::encode(&frame, &mut bytes).map_err(named)?;
        self.position = Some(u64::from_be_bytes(
            plain
                .get(..8)
                .ok_or_else(|| OpenError("crossing_has_no_position".to_owned()))?
                .try_into()
                .map_err(named)?,
        ));
        let mut request = b"LMUC".to_vec();
        request.extend_from_slice(&u32::try_from(length).map_err(named)?.to_be_bytes());
        request.extend_from_slice(&0_u32.to_be_bytes());
        request.extend(bytes);
        let datagram = self.carrier.seal(self.index, &request).map_err(named)?;
        Ok(Packet {
            subject: subject.to_owned(),
            counter: datagram.counter,
            bytes: datagram.encode().map_err(named)?,
            deadline: None,
        })
    }
    fn transmit(&mut self, packet: &Packet) -> Result<(), OpenError> {
        self.carrier
            .retransmit(self.index, &Datagram::decode(&packet.bytes).map_err(named)?)
            .map_err(named)
    }
    fn receipt(&mut self, counter: u64, bound: Option<Deadline>) -> Result<bool, OpenError> {
        let bound = bound.ok_or_else(|| OpenError("direct_crossing_has_no_deadline".to_owned()))?;
        let answer = match self.answer(counter, bound.expires()) {
            Ok(answer) => answer,
            Err(error) if error.0 == "udp_receive_deadline" => return Ok(false),
            Err(error) => return Err(error),
        };
        match answer {
            Frame::PublishAck {
                stream_id: 1,
                message_id,
                ..
            } if Some(message_id) == self.position => Ok(true),
            Frame::PublishError {
                stream_id: 1,
                reason_code,
                message,
                ..
            } => Err(OpenError(format!(
                "crossing_refused:{reason_code}:{}",
                message.unwrap_or_else(|| "unnamed_crossing_refusal".to_owned())
            ))),
            _ => Err(OpenError("unexpected_importing_receipt".to_owned())),
        }
    }
}

mod direct;

impl Drop for DirectPath {
    fn drop(&mut self) {
        if let Err(error) = self.carrier.remove_link(self.index) {
            tracing::error!(%error, index = self.index, "direct carrier retirement failed");
        }
    }
}