liminal_server/link_open/
carriers.rs

1//! Existing stream carriers and indexed direct packets behind one crossing interface.
2
3use super::{Deadline, OpenError, Packet, Path};
4use crate::link_carrier::{Carrier, CarrierReceive, LinkCarrier};
5use crate::link_dial::DialedLink;
6use crate::link_handshake::{LinkTransport, NOISE_MAX_MESSAGE, NOISE_OVERHEAD};
7use crate::link_relay::RelayedLink;
8use crate::link_udp::{Datagram, LinkCancellation, UdpCarrier};
9use liminal::protocol::{Frame, codec};
10use std::sync::Arc;
11
12/// A shutdown act over owned handles; no process names or polling.
13mod stopping;
14pub(crate) use stopping::StopSlot;
15pub type Stop = Arc<dyn Fn() -> Result<(), OpenError> + Send + Sync>;
16
17pub(super) fn named(error: impl std::fmt::Display) -> OpenError {
18    OpenError(error.to_string())
19}
20
21/// A keyed wire or relay stream retaining the endpoint's Noise transport.
22#[derive(Debug)]
23pub struct StreamPath {
24    transport: LinkTransport,
25    carrier: Box<dyn LinkCarrier>,
26    relay: Option<String>,
27    interrupt: Option<Arc<std::net::TcpStream>>,
28    life: Option<crate::link::LinkLife>,
29    position: Option<u64>,
30}
31
32impl StreamPath {
33    #[must_use]
34    pub fn wire(link: DialedLink) -> Self {
35        Self {
36            life: link.carrier.life().cloned(),
37            position: None,
38            transport: link.transport,
39            carrier: Box::new(link.carrier),
40            relay: None,
41            interrupt: None,
42        }
43    }
44    #[must_use]
45    pub fn relay(link: RelayedLink, device: String) -> Self {
46        Self {
47            transport: link.transport,
48            carrier: Box::new(link.carrier),
49            relay: Some(device),
50            interrupt: None,
51            life: None,
52            position: None,
53        }
54    }
55    #[must_use]
56    pub fn with_interrupt(mut self, interrupt: Arc<std::net::TcpStream>) -> Self {
57        self.interrupt = Some(interrupt);
58        self
59    }
60}
61
62impl Path for StreamPath {
63    fn interrupt_handle(&self) -> Option<Stop> {
64        self.interrupt.as_ref().map(|socket| {
65            let socket = Arc::clone(socket);
66            Arc::new(move || match socket.shutdown(std::net::Shutdown::Both) {
67                Ok(()) => Ok(()),
68                Err(error) if error.kind() == std::io::ErrorKind::NotConnected => Ok(()),
69                Err(error) => Err(named(error)),
70            }) as Stop
71        })
72    }
73    fn life(&self) -> Option<crate::link::LinkLife> {
74        self.life.clone()
75    }
76    fn refuse_unsolicited(&self) -> Result<(), OpenError> {
77        self.carrier.refuse_unsolicited().map_err(named)
78    }
79    fn carrier(&self) -> Carrier {
80        self.carrier.carrier()
81    }
82    fn relay_device(&self) -> Option<String> {
83        self.relay.clone()
84    }
85    fn prepare(&mut self, subject: &str, plain: &[u8]) -> Result<Packet, OpenError> {
86        if plain.len().saturating_add(NOISE_OVERHEAD) > NOISE_MAX_MESSAGE {
87            return Err(OpenError("too_large".to_owned()));
88        }
89        self.position = Some(u64::from_be_bytes(
90            plain
91                .get(..8)
92                .ok_or_else(|| OpenError("crossing_has_no_position".to_owned()))?
93                .try_into()
94                .map_err(named)?,
95        ));
96        let counter = self.transport.sending_nonce();
97        let mut bytes = vec![0; plain.len() + NOISE_OVERHEAD];
98        let length = self.transport.seal(plain, &mut bytes).map_err(named)?;
99        bytes.truncate(length);
100        Ok(Packet {
101            subject: subject.to_owned(),
102            counter,
103            bytes,
104            deadline: None,
105        })
106    }
107    fn transmit(&mut self, packet: &Packet) -> Result<(), OpenError> {
108        self.carrier
109            .send(&packet.subject, &packet.bytes)
110            .map_err(named)
111    }
112    fn receipt(&mut self, counter: u64, bound: Option<Deadline>) -> Result<bool, OpenError> {
113        if counter.checked_add(1) != Some(self.transport.sending_nonce()) {
114            return Err(OpenError("unexpected_stream_crossing_counter".to_owned()));
115        }
116        if bound.is_some() {
117            return Err(OpenError(
118                "stream_carrier_has_no_datagram_deadline".to_owned(),
119            ));
120        }
121        match self.carrier.receive().map_err(named)? {
122            CarrierReceive::Acknowledged { message_id, .. }
123                if Some(message_id) == self.position =>
124            {
125                Ok(true)
126            }
127            CarrierReceive::Acknowledged { .. } => {
128                Err(OpenError("unexpected_importing_position".to_owned()))
129            }
130            CarrierReceive::Refused {
131                reason_code,
132                reason,
133            } => Err(OpenError(format!(
134                "crossing_refused:{reason_code}:{}",
135                reason.unwrap_or_else(|| "unnamed_crossing_refusal".to_owned()),
136            ))),
137            CarrierReceive::End { reason } => Err(OpenError(reason)),
138            CarrierReceive::Sealed { .. } => Err(OpenError("unexpected_crossing_reply".to_owned())),
139        }
140    }
141}
142
143/// One indexed path on the device's shared UDP receiver.
144pub struct DirectPath {
145    carrier: Arc<UdpCarrier>,
146    index: u32,
147    inbox: crate::link_udp::streams::Inbox,
148    cancellation: LinkCancellation,
149    transport: Option<LinkTransport>,
150    position: Option<u64>,
151}
152
153impl std::fmt::Debug for DirectPath {
154    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
155        formatter
156            .debug_struct("DirectPath")
157            .field("index", &self.index)
158            .finish_non_exhaustive()
159    }
160}
161
162impl DirectPath {
163    /// Takes an already keyed and indexed path.
164    /// # Errors
165    /// The device does not hold this index.
166    pub fn new(carrier: Arc<UdpCarrier>, index: u32) -> Result<Self, OpenError> {
167        carrier.watch_acknowledgements(index).map_err(named)?;
168        let inbox = carrier.subscribe(index).map_err(named)?;
169        let cancellation = inbox.cancellation();
170        Ok(Self {
171            carrier,
172            index,
173            inbox,
174            cancellation,
175            transport: None,
176            position: None,
177        })
178    }
179}
180
181impl Path for DirectPath {
182    fn interrupt_handle(&self) -> Option<Stop> {
183        let cancellation = self.cancellation.clone();
184        Some(Arc::new(move || cancellation.cancel().map_err(named)))
185    }
186    fn refuse_unsolicited(&self) -> Result<(), OpenError> {
187        self.inbox.refuse_unsolicited().map_err(named)
188    }
189    fn carrier(&self) -> Carrier {
190        Carrier::Direct
191    }
192    fn relay_device(&self) -> Option<String> {
193        None
194    }
195    fn direct_index(&self) -> Option<u32> {
196        Some(self.index)
197    }
198    fn prepare(&mut self, subject: &str, plain: &[u8]) -> Result<Packet, OpenError> {
199        if plain.len() > crate::link_udp::MAX_DATAGRAM
200            || subject.len() > crate::link_udp::MAX_DATAGRAM
201        {
202            return Err(OpenError("too_large_for_datagram".to_owned()));
203        }
204        let frame = Frame::LinkMessage {
205            flags: 0,
206            stream_id: 1,
207            subject: subject.to_owned(),
208            sealed: match self.transport.as_mut() {
209                Some(transport) => {
210                    let mut sealed = vec![0; plain.len() + NOISE_OVERHEAD];
211                    let written = transport.seal(plain, &mut sealed).map_err(named)?;
212                    sealed.truncate(written);
213                    sealed
214                }
215                None => return Err(OpenError("udp_link_not_admitted".to_owned())),
216            },
217        };
218        let length = codec::encoded_len(&frame).map_err(named)?;
219        if length > crate::link_udp::MAX_DATAGRAM {
220            return Err(OpenError("too_large_for_datagram".to_owned()));
221        }
222        let mut bytes = vec![0; length];
223        codec::encode(&frame, &mut bytes).map_err(named)?;
224        self.position = Some(u64::from_be_bytes(
225            plain
226                .get(..8)
227                .ok_or_else(|| OpenError("crossing_has_no_position".to_owned()))?
228                .try_into()
229                .map_err(named)?,
230        ));
231        let mut request = b"LMUC".to_vec();
232        request.extend_from_slice(&u32::try_from(length).map_err(named)?.to_be_bytes());
233        request.extend_from_slice(&0_u32.to_be_bytes());
234        request.extend(bytes);
235        let datagram = self.carrier.seal(self.index, &request).map_err(named)?;
236        Ok(Packet {
237            subject: subject.to_owned(),
238            counter: datagram.counter,
239            bytes: datagram.encode().map_err(named)?,
240            deadline: None,
241        })
242    }
243    fn transmit(&mut self, packet: &Packet) -> Result<(), OpenError> {
244        self.carrier
245            .retransmit(self.index, &Datagram::decode(&packet.bytes).map_err(named)?)
246            .map_err(named)
247    }
248    fn receipt(&mut self, counter: u64, bound: Option<Deadline>) -> Result<bool, OpenError> {
249        let bound = bound.ok_or_else(|| OpenError("direct_crossing_has_no_deadline".to_owned()))?;
250        let answer = match self.answer(counter, bound.expires()) {
251            Ok(answer) => answer,
252            Err(error) if error.0 == "udp_receive_deadline" => return Ok(false),
253            Err(error) => return Err(error),
254        };
255        match answer {
256            Frame::PublishAck {
257                stream_id: 1,
258                message_id,
259                ..
260            } if Some(message_id) == self.position => Ok(true),
261            Frame::PublishError {
262                stream_id: 1,
263                reason_code,
264                message,
265                ..
266            } => Err(OpenError(format!(
267                "crossing_refused:{reason_code}:{}",
268                message.unwrap_or_else(|| "unnamed_crossing_refusal".to_owned())
269            ))),
270            _ => Err(OpenError("unexpected_importing_receipt".to_owned())),
271        }
272    }
273}
274
275mod direct;
276
277impl Drop for DirectPath {
278    fn drop(&mut self) {
279        if let Err(error) = self.carrier.remove_link(self.index) {
280            tracing::error!(%error, index = self.index, "direct carrier retirement failed");
281        }
282    }
283}
liminal_server/link_open/carriers.rs

liminal_server/link_open/
carriers.rs
