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                failure: std::sync::OnceLock::new(),
48            },
49        })
50    }
51}
52impl From<DialedLink> for ChoiceLink {
53    fn from(link: DialedLink) -> Self {
54        let life = link.carrier.life().cloned();
55        Self {
56            link_id: link.link_id,
57            exports: link.exports,
58            transport: Preparation::Wire(link.transport),
59            carrier: ChoiceCarrier {
60                path: Path::Wire(link.carrier),
61                last: Carrier::Wire,
62                life,
63                observed: Arc::new(Mutex::new(Some(ObservedCarrier::Wire))),
64                interrupt: None,
65                answer: None,
66                failure: std::sync::OnceLock::new(),
67            },
68        }
69    }
70}
71#[derive(Debug)]
72pub(crate) enum Preparation {
73    Wire(LinkTransport),
74    Chosen(MachineIdentity),
75}
76impl Preparation {
77    pub(crate) const fn peer(&self) -> &MachineIdentity {
78        match self {
79            Self::Wire(transport) => transport.peer(),
80            Self::Chosen(peer) => peer,
81        }
82    }
83    pub(crate) fn prepare(&mut self, plain: &[u8], output: &mut [u8]) -> Result<usize, OpenError> {
84        match self {
85            Self::Wire(transport) => transport
86                .seal(plain, output)
87                .map_err(|error| OpenError(error.to_string())),
88            Self::Chosen(_) => {
89                let target = output
90                    .get_mut(..plain.len())
91                    .ok_or_else(|| OpenError("crossing_buffer_too_small".to_owned()))?;
92                target.copy_from_slice(plain);
93                Ok(plain.len())
94            }
95        }
96    }
97}
98#[derive(Debug)]
99enum Path {
100    Wire(WireCarrier),
101    Chosen(Box<Opened<NetworkBackend>>),
102}
103#[derive(Debug)]
104pub(crate) struct ChoiceCarrier {
105    path: Path,
106    last: Carrier,
107    life: Option<LinkLife>,
108    observed: Arc<Mutex<Option<ObservedCarrier>>>,
109    interrupt: Option<Interrupt>,
110    answer: Option<CarrierReceive>,
111    /// The first failure the chosen path answered, by name.
112    failure: std::sync::OnceLock<String>,
113}
114impl ChoiceCarrier {
115    pub(crate) fn interrupt_handle(&mut self) -> Result<Interrupt, CarrierError> {
116        match &self.path {
117            Path::Wire(wire) => wire.interrupt_handle().map(Interrupt::wire),
118            Path::Chosen(_) => self.interrupt.take().ok_or(CarrierError::Ended),
119        }
120    }
121    pub(crate) const fn life(&self) -> Option<&LinkLife> {
122        self.life.as_ref()
123    }
124    pub(crate) fn observe_carrier(&self) -> Arc<Mutex<Option<ObservedCarrier>>> {
125        Arc::clone(&self.observed)
126    }
127    pub(crate) fn refuse_unsolicited(&self) -> Result<(), CarrierError> {
128        match &self.path {
129            Path::Wire(wire) => wire.refuse_unsolicited(),
130            Path::Chosen(link) => link
131                .refuse_unsolicited()
132                .map_err(|error| CarrierError::Lifecycle(error.0)),
133        }
134    }
135}
136impl LinkCarrier for ChoiceCarrier {
137    fn carrier(&self) -> Carrier {
138        self.last
139    }
140    fn ended(&self) -> Option<String> {
141        match &self.path {
142            Path::Wire(wire) => wire.ended(),
143            Path::Chosen(_) => self.failure.get().cloned().or_else(|| {
144                self.life
145                    .as_ref()
146                    .and_then(|life| life.reason().ok().flatten())
147            }),
148        }
149    }
150    fn send(&mut self, subject: &str, sealed: &[u8]) -> Result<(), CarrierError> {
151        match &mut self.path {
152            Path::Wire(wire) => wire.send(subject, sealed),
153            Path::Chosen(link) => {
154                let result = link.send(subject, sealed);
155                if let Some(carrier) = link.carrier() {
156                    self.last = carrier;
157                }
158                *self.observed.lock().map_err(|_| {
159                    CarrierError::Lifecycle("carrier_observation_poisoned".to_owned())
160                })? = link.observed_carrier();
161                match result {
162                    Ok(()) => {
163                        let message_id = u64::from_be_bytes(
164                            sealed
165                                .get(..8)
166                                .ok_or(CarrierError::UnexpectedFrame)?
167                                .try_into()
168                                .map_err(|_| CarrierError::UnexpectedFrame)?,
169                        );
170                        self.answer = Some(CarrierReceive::Acknowledged {
171                            message_id,
172                            durable_position: None,
173                        });
174                        Ok(())
175                    }
176                    Err(error) if error.0.starts_with("crossing_refused:") => {
177                        let parts = error
178                            .0
179                            .strip_prefix("crossing_refused:")
180                            .and_then(|value| value.split_once(':'))
181                            .ok_or(CarrierError::UnexpectedFrame)?;
182                        self.answer = Some(CarrierReceive::Refused {
183                            reason_code: parts
184                                .0
185                                .parse()
186                                .map_err(|_| CarrierError::UnexpectedFrame)?,
187                            reason: Some(parts.1.to_owned()),
188                        });
189                        Ok(())
190                    }
191                    Err(error)
192                        if link.stopped()
193                            || matches!(
194                                error.0.as_str(),
195                                "network_stopped" | "udp_receiver_stopped"
196                            ) =>
197                    {
198                        self.failure.get_or_init(|| error.0.clone());
199                        Err(std::io::Error::new(std::io::ErrorKind::Interrupted, error.0).into())
200                    }
201                    Err(error) => {
202                        self.failure.get_or_init(|| error.0.clone());
203                        if let Some(life) = &self.life {
204                            life.end(LinkEnd::Refused(error.0.clone()))
205                                .map_err(CarrierError::Lifecycle)?;
206                        }
207                        Err(CarrierError::Lifecycle(error.0))
208                    }
209                }
210            }
211        }
212    }
213    fn receive(&mut self) -> Result<CarrierReceive, CarrierError> {
214        match &mut self.path {
215            Path::Wire(wire) => wire.receive(),
216            Path::Chosen(_) => self.answer.take().ok_or(CarrierError::UnexpectedFrame),
217        }
218    }
219}
liminal_server/link_carrier/choice.rs

liminal_server/link_carrier/
choice.rs
