liminal_server/
link_bridge.rs

1//! Export subscriptions forward through one keyed carrier, settling each answer before the next send.
2//!
3//! Crossings are judged on the opened link resource at both ends, within its configured export prefixes.
4//! A named import refusal records the subject and reason, advances its subscription past that message, and leaves the link open for other exports.
5
6use crate::ServerError;
7use crate::link_carrier::choice::{ChoiceLink as DialedLink, Interrupt as TcpStream};
8use crate::server::connection::services::LiminalConnectionServices;
9use subscriptions::{ExportSubscriptions, PendingDelivery};
10
11mod counts;
12mod crossing;
13use crossing::cross;
14mod fault;
15mod life;
16#[cfg(feature = "test-support")]
17mod prepared;
18mod start;
19pub use fault::BridgeRegistry;
20use fault::{Observation, Wake};
21#[cfg(test)]
22mod cancellation_tests;
23#[cfg(test)]
24mod fault_tests;
25mod stored;
26pub(crate) mod subscriptions;
27use life::topology;
28use std::collections::BTreeMap;
29use std::net::Shutdown;
30use std::sync::atomic::Ordering;
31use std::sync::{Arc, Mutex};
32use std::thread::JoinHandle;
33
34/// The crossing-refusal code on the importing link's stream.
35const LINK_CROSSING_REFUSED_CODE: u16 = 0x0203;
36
37/// Observations for one link; positions belong to its local subscriptions.
38#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize)]
39pub struct BridgeCounters {
40    /// Opens and ends observed by the keyed carrier.
41    #[serde(flatten)]
42    pub lifecycle: crate::link::LinkLifeCounters,
43    /// Acknowledged crossings.
44    pub crossed: u64,
45    /// Configured dial attempts for which every carrier refused.
46    pub dial_refused: u64,
47    /// Crossings refused at either end and opens refused by the peer.
48    pub refused: u64,
49    /// Refused crossings counted for each subject, logged at its first refusal.
50    pub refused_subjects: BTreeMap<String, u64>,
51    /// Envelopes exceeding one Noise frame.
52    pub too_large: u64,
53    /// The startup durable position of each exported subject.
54    pub resumed_from: BTreeMap<String, u64>,
55    /// The last named refusal, when one was observed.
56    pub last_refusal: Option<String>,
57    /// A forwarding failure that stopped this link.
58    pub error: Option<String>,
59}
60
61fn reached(prefix: &str, subject: &str) -> bool {
62    subject == prefix
63        || subject
64            .strip_prefix(prefix)
65            .is_some_and(|rest| rest.starts_with('.'))
66}
67
68fn consumer_id(link: &str, prefix: &str) -> String {
69    let mut hash = blake3::Hasher::new();
70    hash.update(link.as_bytes());
71    hash.update(&[0]);
72    hash.update(prefix.as_bytes());
73    format!("link-bridge-{}", hash.finalize().to_hex())
74}
75
76fn failure(message: impl Into<String>) -> ServerError {
77    ServerError::ListenerAccept {
78        message: message.into(),
79    }
80}
81
82/// Owns one prefix subscription per configured export and one carrier worker.
83#[derive(Debug)]
84pub struct LinkBridge {
85    link_id: String,
86    observation: Arc<Observation>,
87    wake: Arc<Wake>,
88    interrupt: TcpStream,
89    worker: Option<JoinHandle<Result<(), ServerError>>>,
90}
91
92impl LinkBridge {
93    /// Starts an already keyed link over bounded local subscriptions.
94    ///
95    /// # Errors
96    /// Prefix validation, subscription recovery, or worker creation failed.
97    pub fn start(
98        link: impl Into<DialedLink>,
99        services: Arc<LiminalConnectionServices>,
100    ) -> Result<Self, ServerError> {
101        let link = link.into();
102        let peer = hex::encode(link.transport.peer().ed25519_public());
103        Self::start_named(link, services, peer)
104    }
105
106    /// The current counters for this link.
107    #[must_use]
108    pub fn counters(&self) -> BridgeCounters {
109        self.observation.snapshot()
110    }
111
112    /// Shares this link's live observations with a health registry.
113    pub fn install_observations(&self, registry: &BridgeRegistry) {
114        registry.install(self.link_id.clone(), &self.observation);
115    }
116
117    /// Arms a signalled test stop after the next ack, before checkpointing its subscription.
118    #[cfg(feature = "test-support")]
119    pub fn stop_before_next_acknowledge(&self) {
120        self.observation
121            .stop_before_ack
122            .store(true, Ordering::Release);
123    }
124
125    /// Bounds only a test observation of settled crossing answers.
126    #[cfg(feature = "test-support")]
127    #[must_use]
128    pub fn observe_answers(&self, count: u64, bound: std::time::Duration) -> bool {
129        self.observation.observe_answers(count, bound)
130    }
131
132    /// Stops the subscription worker and interrupts an unanswered carrier read.
133    ///
134    /// # Errors
135    /// The socket interruption or worker join failed, or forwarding had already failed.
136    pub fn finish(mut self) -> Result<(), ServerError> {
137        self.stop_worker()
138    }
139
140    fn stop_worker(&mut self) -> Result<(), ServerError> {
141        // Mark the owned stop before the carrier can observe its closed lifecycle.
142        self.wake.stop();
143        let ended = self.observation.end_life(crate::link::LinkEnd::Shutdown);
144        let interruption = self.interrupt.shutdown(Shutdown::Both);
145        if let Some(worker) = self.worker.take() {
146            worker
147                .join()
148                .map_err(|_| failure("link bridge worker panicked"))??;
149        }
150        let error = self.observation.snapshot().error;
151        if let Some(error) = error {
152            return Err(failure(error));
153        }
154        ended?;
155        match interruption {
156            Ok(()) => Ok(()),
157            Err(error) if error.kind() == std::io::ErrorKind::NotConnected => {
158                tracing::info!(link = %self.link_id,
159                    "bridge carrier already closed at stop; subscription position retained");
160                Ok(())
161            }
162            Err(error) => Err(failure(format!("link bridge interruption: {error}"))),
163        }
164    }
165}
166
167impl Drop for LinkBridge {
168    fn drop(&mut self) {
169        if self.worker.is_some() {
170            if let Err(error) = self.stop_worker() {
171                tracing::error!(link = %self.link_id, %error, "link bridge shutdown failed");
172            }
173        }
174    }
175}
176
177fn run_worker(
178    link: &mut DialedLink,
179    feeds: &ExportSubscriptions,
180    wake: &Wake,
181    observation: &Observation,
182    services: &LiminalConnectionServices,
183    peer: &str,
184) -> Result<(), ServerError> {
185    topology(services, &link.link_id, peer, observation, None);
186    let result = forward(link, feeds, wake, observation, services, peer);
187    if let Err(error) = &result {
188        wake.refuse(error);
189    }
190    observation.worker_ended(&result)?;
191    topology(
192        services,
193        &link.link_id,
194        peer,
195        observation,
196        Some(
197            result
198                .as_ref()
199                .err()
200                .map_or_else(|| "stopped".to_owned(), ToString::to_string),
201        ),
202    );
203    observation.notify_changes();
204    result
205}
206
207enum Crossing {
208    Acknowledged,
209    Refused(Option<String>),
210    TooLarge,
211    Stopped,
212}
213
214fn forward(
215    link: &mut DialedLink,
216    feeds: &ExportSubscriptions,
217    wake: &Wake,
218    observation: &Observation,
219    services: &LiminalConnectionServices,
220    peer: &str,
221) -> Result<(), ServerError> {
222    loop {
223        wake.check()?;
224        if wake.stopping() || observation.ended()? {
225            wake.check()?;
226            return Ok(());
227        }
228        if let Err(error) = link.carrier.refuse_unsolicited() {
229            if wake.stopping() || observation.ended()? {
230                wake.check()?;
231                return Ok(());
232            }
233            return Err(failure(error.to_string()));
234        }
235        let Some(mut source) = feeds.next_ready()? else {
236            if !wake.wait() {
237                wake.check()?;
238                return Ok(());
239            }
240            continue;
241        };
242        #[cfg(feature = "test-support")]
243        let answer = match prepared::take(observation, &mut source)? {
244            Some(sealed) => crossing::exchange(link, &source.subject, &sealed, wake)?,
245            None => cross(link, &mut source, wake)?,
246        };
247        #[cfg(not(feature = "test-support"))]
248        let answer = cross(link, &mut source, wake)?;
249        if matches!(answer, Crossing::Stopped) {
250            return Ok(());
251        }
252        let mut counters = lock(&observation.counters, "counters")?;
253        #[cfg(feature = "test-support")]
254        let stop_before_ack = matches!(answer, Crossing::Acknowledged)
255            && observation.stop_before_ack.swap(false, Ordering::AcqRel);
256        #[cfg(not(feature = "test-support"))]
257        let stop_before_ack = false;
258        match answer {
259            Crossing::Acknowledged => {
260                counters.crossed = counters
261                    .crossed
262                    .checked_add(1)
263                    .ok_or_else(|| failure("bridge crossing counter overflow"))?;
264            }
265            Crossing::Refused(reason) => {
266                counters.refused = counters
267                    .refused
268                    .checked_add(1)
269                    .ok_or_else(|| failure("bridge refusal counter overflow"))?;
270                let refused = counters
271                    .refused_subjects
272                    .entry(source.subject.clone())
273                    .or_default();
274                *refused = refused
275                    .checked_add(1)
276                    .ok_or_else(|| failure("bridge subject refusal counter overflow"))?;
277                if *refused == 1 {
278                    tracing::warn!(link = %link.link_id, subject = %source.subject, reason = ?reason,
279                        "link crossing refused; subscription advances without retry");
280                }
281                counters.last_refusal = reason;
282            }
283            Crossing::TooLarge => {
284                counters.too_large = counters
285                    .too_large
286                    .checked_add(1)
287                    .ok_or_else(|| failure("bridge too_large counter overflow"))?;
288                counters.last_refusal = Some("too_large".to_owned());
289                tracing::warn!(link = %link.link_id, subject = %source.subject, reason = "too_large",
290                    "link crossing refused locally");
291            }
292            Crossing::Stopped => return Ok(()),
293        }
294        drop(counters);
295        if !stop_before_ack {
296            source.acknowledge()?;
297        }
298        topology(services, &link.link_id, peer, observation, None);
299        // Pair settlement with the observer's mutex so arming cannot miss the signal.
300        let guard = lock(&observation.counters, "settlement counters")?;
301        observation
302            .settled
303            .fetch_update(Ordering::Release, Ordering::Relaxed, |count| {
304                count.checked_add(1)
305            })
306            .map_err(|_| failure("bridge settled answer counter overflow"))?;
307        drop(guard);
308        observation.notify_changes();
309        if stop_before_ack {
310            return Ok(());
311        }
312    }
313}
314
315fn lock<'a, Value>(
316    mutex: &'a Mutex<Value>,
317    state: &str,
318) -> Result<std::sync::MutexGuard<'a, Value>, ServerError> {
319    mutex
320        .lock()
321        .map_err(|_| failure(format!("link_bridge_poisoned: {state}")))
322}
liminal_server/link_bridge.rs

liminal_server/
link_bridge.rs
