liminal_server/link_carrier/
choice.rs

1//! The bridge prepares a crossing once; the chosen path owns its encryption and recovery.
2use super::{Carrier, CarrierError, CarrierReceive, LinkCarrier, WireCarrier};
3use crate::console::records::Carrier as ObservedCarrier;
4use crate::link::{LinkEnd, LinkLife};
5use crate::link_dial::DialedLink;
6use crate::link_handshake::{LinkTransport, MachineIdentity};
7use crate::link_open::network::NetworkBackend;
8use crate::link_open::{OpenError, Opened};
9use std::sync::{Arc, Mutex};
10mod stop;
11pub use stop::Interrupt;
12
13/// A bridge's configured prefixes and one common carrier owner.
14#[derive(Debug)]
15pub struct ChoiceLink {
16    pub(crate) link_id: String,
17    pub(crate) exports: Vec<String>,
18    pub(crate) transport: Preparation,
19    pub(crate) carrier: ChoiceCarrier,
20}
21impl ChoiceLink {
22    /// Retains the selected carrier and its message-scoped recovery.
23    /// # Errors
24    /// The carrier cannot be interrupted or its stop adapter cannot start.
25    pub fn chosen(link: Opened<NetworkBackend>) -> Result<Self, OpenError> {
26        let peer = link.backend().peer();
27        let link_id = link.description().link_id.clone();
28        let exports = link.description().exports.clone();
29        let stop = link.interrupt_handle()?;
30        let interrupt = Interrupt::chosen(stop.clone())?;
31        let last = link
32            .carrier()
33            .ok_or_else(|| OpenError("link_has_no_carrier".to_owned()))?;
34        let life = link.life().unwrap_or_else(|| LinkLife::new(peer));
35        let observed = Arc::new(Mutex::new(link.observed_carrier()));
36        Ok(Self {
37            link_id,
38            exports,
39            transport: Preparation::Chosen(peer),
40            carrier: ChoiceCarrier {
41                path: Path::Chosen(Box::new(link)),
42                last,
43                life: Some(life),
44                observed,
45                interrupt: Some(interrupt),
46                answer: None,
47            },
48        })
49    }
50}
51impl From<DialedLink> for ChoiceLink {
52    fn from(link: DialedLink) -> Self {
53        let life = link.carrier.life().cloned();
54        Self {
55            link_id: link.link_id,
56            exports: link.exports,
57            transport: Preparation::Wire(link.transport),
58            carrier: ChoiceCarrier {
59                path: Path::Wire(link.carrier),
60                last: Carrier::Wire,
61                life,
62                observed: Arc::new(Mutex::new(Some(ObservedCarrier::Wire))),
63                interrupt: None,
64                answer: None,
65            },
66        }
67    }
68}
69#[derive(Debug)]
70pub(crate) enum Preparation {
71    Wire(LinkTransport),
72    Chosen(MachineIdentity),
73}
74impl Preparation {
75    pub(crate) const fn peer(&self) -> &MachineIdentity {
76        match self {
77            Self::Wire(transport) => transport.peer(),
78            Self::Chosen(peer) => peer,
79        }
80    }
81    pub(crate) fn prepare(&mut self, plain: &[u8], output: &mut [u8]) -> Result<usize, OpenError> {
82        match self {
83            Self::Wire(transport) => transport
84                .seal(plain, output)
85                .map_err(|error| OpenError(error.to_string())),
86            Self::Chosen(_) => {
87                let target = output
88                    .get_mut(..plain.len())
89                    .ok_or_else(|| OpenError("crossing_buffer_too_small".to_owned()))?;
90                target.copy_from_slice(plain);
91                Ok(plain.len())
92            }
93        }
94    }
95}
96#[derive(Debug)]
97enum Path {
98    Wire(WireCarrier),
99    Chosen(Box<Opened<NetworkBackend>>),
100}
101#[derive(Debug)]
102pub(crate) struct ChoiceCarrier {
103    path: Path,
104    last: Carrier,
105    life: Option<LinkLife>,
106    observed: Arc<Mutex<Option<ObservedCarrier>>>,
107    interrupt: Option<Interrupt>,
108    answer: Option<CarrierReceive>,
109}
110impl ChoiceCarrier {
111    pub(crate) fn interrupt_handle(&mut self) -> Result<Interrupt, CarrierError> {
112        match &self.path {
113            Path::Wire(wire) => wire.interrupt_handle().map(Interrupt::wire),
114            Path::Chosen(_) => self.interrupt.take().ok_or(CarrierError::Ended),
115        }
116    }
117    pub(crate) const fn life(&self) -> Option<&LinkLife> {
118        self.life.as_ref()
119    }
120    pub(crate) fn observe_carrier(&self) -> Arc<Mutex<Option<ObservedCarrier>>> {
121        Arc::clone(&self.observed)
122    }
123    pub(crate) fn refuse_unsolicited(&self) -> Result<(), CarrierError> {
124        match &self.path {
125            Path::Wire(wire) => wire.refuse_unsolicited(),
126            Path::Chosen(link) => link
127                .refuse_unsolicited()
128                .map_err(|error| CarrierError::Lifecycle(error.0)),
129        }
130    }
131}
132impl LinkCarrier for ChoiceCarrier {
133    fn carrier(&self) -> Carrier {
134        self.last
135    }
136    fn send(&mut self, subject: &str, sealed: &[u8]) -> Result<(), CarrierError> {
137        match &mut self.path {
138            Path::Wire(wire) => wire.send(subject, sealed),
139            Path::Chosen(link) => {
140                let result = link.send(subject, sealed);
141                if let Some(carrier) = link.carrier() {
142                    self.last = carrier;
143                }
144                *self.observed.lock().map_err(|_| {
145                    CarrierError::Lifecycle("carrier_observation_poisoned".to_owned())
146                })? = link.observed_carrier();
147                match result {
148                    Ok(()) => {
149                        let message_id = u64::from_be_bytes(
150                            sealed
151                                .get(..8)
152                                .ok_or(CarrierError::UnexpectedFrame)?
153                                .try_into()
154                                .map_err(|_| CarrierError::UnexpectedFrame)?,
155                        );
156                        self.answer = Some(CarrierReceive::Acknowledged {
157                            message_id,
158                            durable_position: None,
159                        });
160                        Ok(())
161                    }
162                    Err(error) if error.0.starts_with("crossing_refused:") => {
163                        let parts = error
164                            .0
165                            .strip_prefix("crossing_refused:")
166                            .and_then(|value| value.split_once(':'))
167                            .ok_or(CarrierError::UnexpectedFrame)?;
168                        self.answer = Some(CarrierReceive::Refused {
169                            reason_code: parts
170                                .0
171                                .parse()
172                                .map_err(|_| CarrierError::UnexpectedFrame)?,
173                            reason: Some(parts.1.to_owned()),
174                        });
175                        Ok(())
176                    }
177                    Err(error)
178                        if link.stopped()
179                            || matches!(
180                                error.0.as_str(),
181                                "network_stopped" | "udp_receiver_stopped"
182                            ) =>
183                    {
184                        Err(std::io::Error::new(std::io::ErrorKind::Interrupted, error.0).into())
185                    }
186                    Err(error) => {
187                        if let Some(life) = &self.life {
188                            life.end(LinkEnd::Refused(error.0.clone()))
189                                .map_err(CarrierError::Lifecycle)?;
190                        }
191                        Err(CarrierError::Lifecycle(error.0))
192                    }
193                }
194            }
195        }
196    }
197    fn receive(&mut self) -> Result<CarrierReceive, CarrierError> {
198        match &mut self.path {
199            Path::Wire(wire) => wire.receive(),
200            Path::Chosen(_) => self.answer.take().ok_or(CarrierError::UnexpectedFrame),
201        }
202    }
203}
liminal_server/link_carrier/choice.rs

liminal_server/link_carrier/
choice.rs
