Skip to main content

rns_net/interface/
backbone.rs

1//! Backbone TCP mesh interface using cross-platform polling.
2//!
3//! Server mode: listens on a TCP port, accepts peer connections, spawns
4//! dynamic per-peer interfaces. Uses a single poll thread to multiplex
5//! all client sockets. HDLC framing for packet boundaries.
6//!
7//! Client mode: connects to a remote backbone server, single TCP connection
8//! with HDLC framing. Reconnects on disconnect.
9//!
10//! Matches Python `BackboneInterface.py`.
11
12use std::collections::HashMap;
13use std::hash::{BuildHasher, Hasher};
14use std::io::{self, Read, Write};
15use std::net::{IpAddr, Shutdown, TcpListener, TcpStream, ToSocketAddrs};
16use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
17use std::sync::{Arc, Mutex, OnceLock};
18use std::thread;
19use std::time::{Duration, Instant};
20
21use polling::{Event as PollEvent, Events, Poller};
22use socket2::{SockRef, TcpKeepalive};
23
24use rns_core::constants;
25use rns_core::transport::types::{IngressControlConfig, InterfaceId, InterfaceInfo};
26
27use crate::event::{Event, EventSender};
28use crate::hdlc;
29use crate::interface::{
30    lock_or_recover, DynamicInterfaceTemplate, InterfaceConfigData, InterfaceFactory, StartContext,
31    StartResult, Writer,
32};
33use crate::BackbonePeerStateEntry;
34
35/// HW_MTU: 1 MB (matches Python BackboneInterface.HW_MTU)
36const HW_MTU: usize = 1_048_576;
37
38/// Configuration for a backbone interface.
39#[derive(Debug, Clone)]
40pub struct BackboneConfig {
41    pub name: String,
42    pub listen_ip: String,
43    pub listen_port: u16,
44    pub interface_id: InterfaceId,
45    pub mode: u8,
46    pub gravity: i64,
47    pub recursive_prs: bool,
48    pub announces_from_internal: bool,
49    pub announces_to_internal: Option<bool>,
50    pub max_connections: Option<usize>,
51    pub idle_timeout: Option<Duration>,
52    pub write_stall_timeout: Option<Duration>,
53    pub abuse: BackboneAbuseConfig,
54    pub fast_flap: BackboneFastFlapConfig,
55    pub fast_flap_state: Arc<Mutex<BackboneFastFlapMonitor>>,
56    pub ingress_control: IngressControlConfig,
57    pub runtime: Arc<Mutex<BackboneServerRuntime>>,
58    pub peer_state: Arc<Mutex<BackbonePeerMonitor>>,
59}
60
61/// Configurable behavior-based abuse detection for inbound peers.
62#[derive(Debug, Clone, Default)]
63pub struct BackboneAbuseConfig {
64    pub max_penalty_duration: Option<Duration>,
65}
66
67/// Upstream-compatible automatic blocking for repeatedly short-lived inbound
68/// Backbone connections.
69#[derive(Debug, Clone, Copy, Eq, PartialEq)]
70pub struct BackboneFastFlapConfig {
71    pub enabled: bool,
72    pub connection_threshold: Duration,
73    pub grace: u64,
74    pub block_duration: Duration,
75}
76
77impl Default for BackboneFastFlapConfig {
78    fn default() -> Self {
79        Self {
80            enabled: true,
81            connection_threshold: Duration::from_secs(20),
82            grace: 5,
83            block_duration: Duration::from_secs(12 * 60 * 60),
84        }
85    }
86}
87
88#[derive(Debug, Clone)]
89struct FastFlapPeerState {
90    first_flap_at: Instant,
91    last_flap_at: Instant,
92    flap_count: u64,
93}
94
95/// Result of recording one short-lived connection.
96#[derive(Debug, Clone, Copy, Eq, PartialEq)]
97pub struct FastFlapRecord {
98    pub flap_count: u64,
99    pub blocked: bool,
100    pub newly_blocked: bool,
101    pub blocked_until: Option<Instant>,
102    pub elapsed_since_first: Duration,
103}
104
105/// Process-wide fast-flap history. Policy remains per listener, while the IP
106/// history is shared like upstream's class-level `fast_flapping` table.
107#[derive(Debug, Default)]
108pub struct BackboneFastFlapMonitor {
109    peers: HashMap<IpAddr, FastFlapPeerState>,
110}
111
112impl BackboneFastFlapMonitor {
113    pub fn new() -> Self {
114        Self::default()
115    }
116
117    pub fn record_disconnect(
118        &mut self,
119        peer_ip: IpAddr,
120        connected_for: Duration,
121        config: &BackboneFastFlapConfig,
122        now: Instant,
123    ) -> Option<FastFlapRecord> {
124        if !config.enabled || connected_for >= config.connection_threshold {
125            return None;
126        }
127
128        self.expire_peer(peer_ip, config, now);
129        let state = self.peers.entry(peer_ip).or_insert(FastFlapPeerState {
130            first_flap_at: now,
131            last_flap_at: now,
132            flap_count: 0,
133        });
134        let was_blocked = state.flap_count > config.grace;
135        state.last_flap_at = now;
136        state.flap_count = state.flap_count.saturating_add(1);
137        let blocked = state.flap_count > config.grace;
138
139        Some(FastFlapRecord {
140            flap_count: state.flap_count,
141            blocked,
142            newly_blocked: blocked && !was_blocked,
143            blocked_until: blocked
144                .then(|| now.checked_add(config.block_duration))
145                .flatten(),
146            elapsed_since_first: now.saturating_duration_since(state.first_flap_at),
147        })
148    }
149
150    pub fn is_blocked(
151        &mut self,
152        peer_ip: IpAddr,
153        config: &BackboneFastFlapConfig,
154        now: Instant,
155    ) -> bool {
156        self.blocked_until(peer_ip, config, now).is_some()
157    }
158
159    pub fn blocked_until(
160        &mut self,
161        peer_ip: IpAddr,
162        config: &BackboneFastFlapConfig,
163        now: Instant,
164    ) -> Option<Instant> {
165        if !config.enabled {
166            return None;
167        }
168        self.expire_peer(peer_ip, config, now);
169        self.peers
170            .get(&peer_ip)
171            .filter(|state| state.flap_count > config.grace)
172            .and_then(|state| state.last_flap_at.checked_add(config.block_duration))
173    }
174
175    pub fn blocked_ip_count(&mut self, config: &BackboneFastFlapConfig, now: Instant) -> usize {
176        if !config.enabled {
177            return 0;
178        }
179        self.expire_all(config, now);
180        self.peers
181            .values()
182            .filter(|state| state.flap_count > config.grace)
183            .count()
184    }
185
186    pub fn blocked_ip_list(
187        &mut self,
188        config: &BackboneFastFlapConfig,
189        now: Instant,
190    ) -> Vec<IpAddr> {
191        if !config.enabled {
192            return Vec::new();
193        }
194        self.expire_all(config, now);
195        let mut blocked: Vec<_> = self
196            .peers
197            .iter()
198            .filter(|(_, state)| state.flap_count > config.grace)
199            .map(|(peer_ip, _)| *peer_ip)
200            .collect();
201        blocked.sort_unstable();
202        blocked
203    }
204
205    pub fn flap_count(&self, peer_ip: IpAddr) -> u64 {
206        self.peers
207            .get(&peer_ip)
208            .map(|state| state.flap_count)
209            .unwrap_or(0)
210    }
211
212    pub fn clear(&mut self, peer_ip: IpAddr) -> bool {
213        self.peers.remove(&peer_ip).is_some()
214    }
215
216    fn expire_peer(&mut self, peer_ip: IpAddr, config: &BackboneFastFlapConfig, now: Instant) {
217        let expired = self.peers.get(&peer_ip).is_some_and(|state| {
218            now.saturating_duration_since(state.last_flap_at) > config.block_duration
219        });
220        if expired {
221            self.peers.remove(&peer_ip);
222        }
223    }
224
225    fn expire_all(&mut self, config: &BackboneFastFlapConfig, now: Instant) {
226        self.peers.retain(|_, state| {
227            now.saturating_duration_since(state.last_flap_at) <= config.block_duration
228        });
229    }
230}
231
232fn shared_fast_flap_monitor() -> Arc<Mutex<BackboneFastFlapMonitor>> {
233    static MONITOR: OnceLock<Arc<Mutex<BackboneFastFlapMonitor>>> = OnceLock::new();
234    Arc::clone(MONITOR.get_or_init(|| Arc::new(Mutex::new(BackboneFastFlapMonitor::new()))))
235}
236
237/// Live runtime state for a backbone server interface.
238#[derive(Debug, Clone)]
239pub struct BackboneServerRuntime {
240    pub max_connections: Option<usize>,
241    pub idle_timeout: Option<Duration>,
242    pub write_stall_timeout: Option<Duration>,
243    pub abuse: BackboneAbuseConfig,
244}
245
246impl BackboneServerRuntime {
247    pub fn from_config(config: &BackboneConfig) -> Self {
248        Self {
249            max_connections: config.max_connections,
250            idle_timeout: config.idle_timeout,
251            write_stall_timeout: config.write_stall_timeout,
252            abuse: config.abuse.clone(),
253        }
254    }
255}
256
257#[derive(Debug, Clone)]
258pub struct BackboneRuntimeConfigHandle {
259    pub interface_name: String,
260    pub runtime: Arc<Mutex<BackboneServerRuntime>>,
261    pub startup: BackboneServerRuntime,
262}
263
264#[derive(Debug, Clone)]
265pub struct BackbonePeerStateHandle {
266    pub interface_id: InterfaceId,
267    pub interface_name: String,
268    pub mode: u8,
269    pub gravity: i64,
270    pub announces_to_internal: Option<bool>,
271    pub peer_state: Arc<Mutex<BackbonePeerMonitor>>,
272    pub fast_flap: BackboneFastFlapConfig,
273    pub fast_flap_state: Arc<Mutex<BackboneFastFlapMonitor>>,
274}
275
276impl Default for BackboneConfig {
277    fn default() -> Self {
278        let mut config = BackboneConfig {
279            name: String::new(),
280            listen_ip: "0.0.0.0".into(),
281            listen_port: 0,
282            interface_id: InterfaceId(0),
283            mode: constants::MODE_FULL,
284            gravity: 0,
285            recursive_prs: false,
286            announces_from_internal: true,
287            announces_to_internal: None,
288            max_connections: None,
289            idle_timeout: None,
290            write_stall_timeout: None,
291            abuse: BackboneAbuseConfig::default(),
292            fast_flap: BackboneFastFlapConfig::default(),
293            fast_flap_state: shared_fast_flap_monitor(),
294            ingress_control: IngressControlConfig::enabled(),
295            runtime: Arc::new(Mutex::new(BackboneServerRuntime {
296                max_connections: None,
297                idle_timeout: None,
298                write_stall_timeout: None,
299                abuse: BackboneAbuseConfig::default(),
300            })),
301            peer_state: Arc::new(Mutex::new(BackbonePeerMonitor::new())),
302        };
303        let startup = BackboneServerRuntime::from_config(&config);
304        config.runtime = Arc::new(Mutex::new(startup));
305        config
306    }
307}
308
309/// Maximum pending buffer size per client (512 KB). Clients exceeding this are
310/// disconnected to prevent unbounded memory growth from slow readers.
311const MAX_PENDING_BYTES: usize = 512 * 1024;
312
313/// Writer that sends HDLC-framed data over a cloned TCP stream (server mode).
314struct BackboneWriter {
315    stream: TcpStream,
316    runtime: Arc<Mutex<BackboneServerRuntime>>,
317    interface_name: String,
318    interface_id: InterfaceId,
319    event_tx: EventSender,
320    pending: Vec<u8>,
321    stall_started: Option<Instant>,
322    disconnect_notified: bool,
323    write_stall_flag: Arc<AtomicBool>,
324}
325
326impl Writer for BackboneWriter {
327    fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
328        let write_stall_timeout =
329            lock_or_recover(&self.runtime, "backbone runtime").write_stall_timeout;
330        if !self.pending.is_empty() {
331            self.flush_pending(write_stall_timeout)?;
332            if !self.pending.is_empty() {
333                return Err(io::Error::new(
334                    io::ErrorKind::WouldBlock,
335                    "backbone writer still stalled",
336                ));
337            }
338        }
339
340        let frame = hdlc::frame(data);
341        self.write_buffer(&frame, write_stall_timeout)
342    }
343
344    fn shutdown(&mut self) {
345        let _ = self.stream.shutdown(Shutdown::Both);
346    }
347}
348
349impl BackboneWriter {
350    fn write_buffer(
351        &mut self,
352        data: &[u8],
353        write_stall_timeout: Option<Duration>,
354    ) -> io::Result<()> {
355        let mut written = 0usize;
356        while written < data.len() {
357            match self.stream.write(&data[written..]) {
358                Ok(0) => {
359                    return Err(io::Error::new(
360                        io::ErrorKind::WriteZero,
361                        "backbone writer wrote zero bytes",
362                    ))
363                }
364                Ok(n) => {
365                    written += n;
366                    self.stall_started = None;
367                }
368                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
369                    let now = Instant::now();
370                    let started = self.stall_started.get_or_insert(now);
371                    if let Some(timeout) = write_stall_timeout {
372                        if now.duration_since(*started) >= timeout {
373                            return Err(self.disconnect_for_write_stall(timeout));
374                        }
375                    }
376                    if self.pending.len() + data[written..].len() > MAX_PENDING_BYTES {
377                        return Err(self.disconnect_for_write_stall(
378                            write_stall_timeout.unwrap_or(Duration::from_secs(30)),
379                        ));
380                    }
381                    self.pending.extend_from_slice(&data[written..]);
382                    return Err(io::Error::new(
383                        io::ErrorKind::WouldBlock,
384                        "backbone writer would block",
385                    ));
386                }
387                Err(e) => return Err(e),
388            }
389        }
390        Ok(())
391    }
392
393    fn flush_pending(&mut self, write_stall_timeout: Option<Duration>) -> io::Result<()> {
394        if self.pending.is_empty() {
395            return Ok(());
396        }
397
398        let pending = std::mem::take(&mut self.pending);
399        match self.write_buffer(&pending, write_stall_timeout) {
400            Ok(()) => Ok(()),
401            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => Ok(()),
402            Err(e) => Err(e),
403        }
404    }
405
406    fn disconnect_for_write_stall(&mut self, timeout: Duration) -> io::Error {
407        if !self.disconnect_notified {
408            log::warn!(
409                "[{}] backbone client {} disconnected due to write stall timeout ({:?})",
410                self.interface_name,
411                self.interface_id.0,
412                timeout
413            );
414            self.write_stall_flag.store(true, Ordering::Relaxed);
415            let _ = self.stream.shutdown(Shutdown::Both);
416            let _ = self.event_tx.send(Event::InterfaceDown(self.interface_id));
417            self.disconnect_notified = true;
418        }
419        io::Error::new(
420            io::ErrorKind::TimedOut,
421            format!("backbone writer stalled for {:?}", timeout),
422        )
423    }
424}
425
426/// Start a backbone interface. Binds TCP listener, spawns poll thread.
427pub fn start(config: BackboneConfig, tx: EventSender, next_id: Arc<AtomicU64>) -> io::Result<()> {
428    start_with_template(config, tx, next_id, None)
429}
430
431fn start_with_template(
432    config: BackboneConfig,
433    tx: EventSender,
434    next_id: Arc<AtomicU64>,
435    dynamic_template: Option<DynamicInterfaceTemplate>,
436) -> io::Result<()> {
437    let addr = format!("{}:{}", config.listen_ip, config.listen_port);
438    let listener = TcpListener::bind(&addr)?;
439    listener.set_nonblocking(true)?;
440
441    log::info!(
442        "[{}] backbone server listening on {}",
443        config.name,
444        listener
445            .local_addr()
446            .unwrap_or_else(|_| std::net::SocketAddr::from(([0, 0, 0, 0], config.listen_port)))
447    );
448
449    let name = config.name.clone();
450    let server_interface_id = config.interface_id;
451    let runtime = Arc::clone(&config.runtime);
452    let peer_state = Arc::clone(&config.peer_state);
453    let fast_flap = config.fast_flap;
454    let fast_flap_state = Arc::clone(&config.fast_flap_state);
455    let ingress_control = config.ingress_control;
456    let accepted_peer_mode = config.mode;
457    let accepted_peer_gravity = config.gravity;
458    let accepted_peer_recursive_prs = config.recursive_prs;
459    let accepted_peer_announces_from_internal = config.announces_from_internal;
460    let accepted_peer_announces_to_internal = config.announces_to_internal;
461    let ifac_size = dynamic_template
462        .as_ref()
463        .and_then(|template| template.ifac.as_ref())
464        .map(|ifac| ifac.size)
465        .unwrap_or(0);
466    thread::Builder::new()
467        .name(format!("backbone-poll-{}", config.interface_id.0))
468        .spawn(move || {
469            if let Err(e) = poll_loop(
470                listener,
471                name,
472                server_interface_id,
473                tx,
474                next_id,
475                runtime,
476                peer_state,
477                fast_flap,
478                fast_flap_state,
479                ingress_control,
480                accepted_peer_mode,
481                accepted_peer_gravity,
482                accepted_peer_recursive_prs,
483                accepted_peer_announces_from_internal,
484                accepted_peer_announces_to_internal,
485                dynamic_template,
486                ifac_size,
487            ) {
488                log::error!("backbone poll loop error: {}", e);
489            }
490        })?;
491
492    Ok(())
493}
494
495/// Per-client tracking state.
496struct ClientState {
497    id: InterfaceId,
498    peer_ip: IpAddr,
499    peer_port: u16,
500    stream: TcpStream,
501    decoder: hdlc::Decoder,
502    connected_at: Instant,
503    has_received_data: bool,
504    write_stall_flag: Arc<AtomicBool>,
505}
506
507#[derive(Debug, Clone)]
508struct PeerBehaviorState {
509    blacklisted_until: Option<Instant>,
510    blacklist_reason: Option<String>,
511    reject_count: u64,
512    connected_count: usize,
513}
514
515impl PeerBehaviorState {
516    fn new() -> Self {
517        Self {
518            blacklisted_until: None,
519            blacklist_reason: None,
520            reject_count: 0,
521            connected_count: 0,
522        }
523    }
524}
525
526#[derive(Debug, Clone, Default)]
527pub struct BackbonePeerMonitor {
528    peers: HashMap<IpAddr, PeerBehaviorState>,
529}
530
531impl BackbonePeerMonitor {
532    pub fn new() -> Self {
533        Self {
534            peers: HashMap::new(),
535        }
536    }
537
538    fn upsert_snapshot(&mut self, peers: &HashMap<IpAddr, PeerBehaviorState>) {
539        let mut merged = self.peers.clone();
540
541        for (peer_ip, state) in peers {
542            let entry = merged
543                .entry(*peer_ip)
544                .or_insert_with(PeerBehaviorState::new);
545            entry.connected_count = state.connected_count;
546            entry.reject_count = state.reject_count;
547            entry.blacklisted_until = state.blacklisted_until;
548            entry.blacklist_reason = state.blacklist_reason.clone();
549        }
550
551        merged.retain(|peer_ip, state| {
552            peers.contains_key(peer_ip)
553                || state.blacklisted_until.is_some()
554                || state.reject_count > 0
555        });
556        self.peers = merged;
557    }
558
559    fn sync_into(&self, peers: &mut HashMap<IpAddr, PeerBehaviorState>) {
560        for (peer_ip, state) in &self.peers {
561            let entry = peers.entry(*peer_ip).or_insert_with(PeerBehaviorState::new);
562            entry.blacklisted_until = state.blacklisted_until;
563            entry.blacklist_reason = state.blacklist_reason.clone();
564            entry.reject_count = state.reject_count;
565        }
566
567        peers.retain(|peer_ip, state| {
568            if state.connected_count > 0 {
569                return true;
570            }
571            self.peers.contains_key(peer_ip)
572        });
573    }
574
575    pub fn list(&self, interface_name: &str) -> Vec<BackbonePeerStateEntry> {
576        let now = Instant::now();
577        let mut entries: Vec<BackbonePeerStateEntry> = self
578            .peers
579            .iter()
580            .map(|(peer_ip, state)| BackbonePeerStateEntry {
581                interface_name: interface_name.to_string(),
582                peer_ip: *peer_ip,
583                connected_count: state.connected_count,
584                blacklisted_remaining_secs: state
585                    .blacklisted_until
586                    .and_then(|until| (until > now).then(|| (until - now).as_secs_f64())),
587                blacklist_reason: state.blacklist_reason.clone(),
588                reject_count: state.reject_count,
589            })
590            .collect();
591        entries.sort_by(|a, b| a.peer_ip.cmp(&b.peer_ip));
592        entries
593    }
594
595    pub fn clear(&mut self, peer_ip: IpAddr) -> bool {
596        self.peers.remove(&peer_ip).is_some()
597    }
598
599    pub fn blacklist(&mut self, peer_ip: IpAddr, duration: Duration, reason: String) -> bool {
600        let state = self
601            .peers
602            .entry(peer_ip)
603            .or_insert_with(PeerBehaviorState::new);
604        state.blacklisted_until = Some(Instant::now() + duration);
605        state.blacklist_reason = Some(reason);
606        true
607    }
608
609    #[cfg(test)]
610    pub fn seed_entry(&mut self, entry: BackbonePeerStateEntry) {
611        let mut state = PeerBehaviorState::new();
612        state.connected_count = entry.connected_count;
613        state.reject_count = entry.reject_count;
614        state.blacklist_reason = entry.blacklist_reason;
615        if let Some(remaining) = entry.blacklisted_remaining_secs {
616            state.blacklisted_until = Some(Instant::now() + Duration::from_secs_f64(remaining));
617        }
618        self.peers.insert(entry.peer_ip, state);
619    }
620}
621
622#[derive(Clone, Copy)]
623enum DisconnectReason {
624    RemoteClosed,
625    IdleTimeout,
626    WriteStall,
627}
628
629/// Main poll event loop.
630fn poll_loop(
631    listener: TcpListener,
632    name: String,
633    server_interface_id: InterfaceId,
634    tx: EventSender,
635    next_id: Arc<AtomicU64>,
636    runtime: Arc<Mutex<BackboneServerRuntime>>,
637    peer_state: Arc<Mutex<BackbonePeerMonitor>>,
638    fast_flap: BackboneFastFlapConfig,
639    fast_flap_state: Arc<Mutex<BackboneFastFlapMonitor>>,
640    ingress_control: IngressControlConfig,
641    accepted_peer_mode: u8,
642    accepted_peer_gravity: i64,
643    accepted_peer_recursive_prs: bool,
644    accepted_peer_announces_from_internal: bool,
645    accepted_peer_announces_to_internal: Option<bool>,
646    dynamic_template: Option<DynamicInterfaceTemplate>,
647    ifac_size: usize,
648) -> io::Result<()> {
649    let poller = Poller::new()?;
650
651    const LISTENER_KEY: usize = 0;
652
653    // SAFETY: listener outlives its registration in the poller.
654    unsafe { poller.add(&listener, PollEvent::readable(LISTENER_KEY))? };
655
656    let mut clients: HashMap<usize, ClientState> = HashMap::new();
657    let mut peers: HashMap<IpAddr, PeerBehaviorState> = HashMap::new();
658    let mut events = Events::new();
659    let mut next_key: usize = 1;
660
661    loop {
662        let runtime_snapshot = runtime.lock().unwrap().clone();
663        let max_connections = runtime_snapshot.max_connections;
664        let idle_timeout = runtime_snapshot.idle_timeout;
665        {
666            let mut monitor = peer_state.lock().unwrap();
667            monitor.sync_into(&mut peers);
668            cleanup_peer_state(&mut peers);
669            monitor.upsert_snapshot(&peers);
670        }
671
672        events.clear();
673        poller.wait(&mut events, Some(Duration::from_secs(1)))?;
674
675        for ev in events.iter() {
676            if ev.key == LISTENER_KEY {
677                // Accept new connections
678                loop {
679                    match listener.accept() {
680                        Ok((stream, peer_addr)) => {
681                            let peer_ip = peer_addr.ip();
682                            let peer_port = peer_addr.port();
683
684                            let fast_flap_blocked_until =
685                                lock_or_recover(&fast_flap_state, "backbone fast-flap state")
686                                    .blocked_until(peer_ip, &fast_flap, Instant::now());
687                            if is_ip_blacklisted(&mut peers, peer_ip)
688                                || fast_flap_blocked_until.is_some()
689                            {
690                                let state =
691                                    peers.entry(peer_ip).or_insert_with(PeerBehaviorState::new);
692                                if let Some(blocked_until) = fast_flap_blocked_until {
693                                    state.blacklisted_until = Some(blocked_until);
694                                    state.blacklist_reason = Some("fast flapping".into());
695                                }
696                                if let Some(state) = peers.get_mut(&peer_ip) {
697                                    state.reject_count = state.reject_count.saturating_add(1);
698                                }
699                                peer_state.lock().unwrap().upsert_snapshot(&peers);
700                                log::trace!(target: crate::logging::PATHING_LOG_TARGET,
701                                    "[{}] rejecting blocked peer {}",
702                                    name,
703                                    peer_addr,
704                                );
705                                drop(stream);
706                                continue;
707                            }
708
709                            if let Some(max) = max_connections {
710                                if clients.len() >= max {
711                                    log::warn!(
712                                        "[{}] max connections ({}) reached, rejecting {}",
713                                        name,
714                                        max,
715                                        peer_addr
716                                    );
717                                    drop(stream);
718                                    continue;
719                                }
720                            }
721
722                            stream.set_nonblocking(true).ok();
723                            stream.set_nodelay(true).ok();
724                            set_tcp_keepalive(&stream).ok();
725
726                            // Prevent SIGPIPE on macOS when writing to broken pipes
727                            #[cfg(target_os = "macos")]
728                            {
729                                let sock = SockRef::from(&stream);
730                                sock.set_nosigpipe(true).ok();
731                            }
732
733                            let key = next_key;
734                            next_key += 1;
735                            let client_id = InterfaceId(next_id.fetch_add(1, Ordering::Relaxed));
736
737                            log::trace!(target: crate::logging::PATHING_LOG_TARGET,
738                                "[{}] backbone client connected: {} → id {}",
739                                name,
740                                peer_addr,
741                                client_id.0
742                            );
743
744                            // Register client with poller
745                            // SAFETY: stream is stored in ClientState and outlives registration.
746                            if let Err(e) = unsafe { poller.add(&stream, PollEvent::readable(key)) }
747                            {
748                                log::warn!("[{}] failed to add client to poller: {}", name, e);
749                                continue; // stream drops, closing socket
750                            }
751
752                            // Create writer via try_clone (cross-platform dup)
753                            let writer_stream = match stream.try_clone() {
754                                Ok(s) => s,
755                                Err(e) => {
756                                    log::warn!("[{}] failed to clone client stream: {}", name, e);
757                                    let _ = poller.delete(&stream);
758                                    continue; // stream drops
759                                }
760                            };
761                            let write_stall_flag = Arc::new(AtomicBool::new(false));
762                            let writer: Box<dyn Writer> = Box::new(BackboneWriter {
763                                stream: writer_stream,
764                                runtime: Arc::clone(&runtime),
765                                interface_name: name.clone(),
766                                interface_id: client_id,
767                                event_tx: tx.clone(),
768                                pending: Vec::new(),
769                                stall_started: None,
770                                disconnect_notified: false,
771                                write_stall_flag: Arc::clone(&write_stall_flag),
772                            });
773
774                            clients.insert(
775                                key,
776                                ClientState {
777                                    id: client_id,
778                                    peer_ip,
779                                    peer_port,
780                                    stream,
781                                    decoder: hdlc::Decoder::reticulum(HW_MTU, ifac_size),
782                                    connected_at: Instant::now(),
783                                    has_received_data: false,
784                                    write_stall_flag,
785                                },
786                            );
787                            peers
788                                .entry(peer_ip)
789                                .or_insert_with(PeerBehaviorState::new)
790                                .connected_count += 1;
791                            peer_state.lock().unwrap().upsert_snapshot(&peers);
792                            let _ = tx.send(Event::BackbonePeerConnected {
793                                server_interface_id,
794                                peer_interface_id: client_id,
795                                peer_ip,
796                                peer_port,
797                            });
798
799                            let info = InterfaceInfo {
800                                id: client_id,
801                                name: format!("BackboneInterface/{}", client_id.0),
802                                mode: accepted_peer_mode,
803                                gravity: accepted_peer_gravity,
804                                recursive_prs: accepted_peer_recursive_prs,
805                                announces_from_internal: accepted_peer_announces_from_internal,
806                                announces_to_internal: accepted_peer_announces_to_internal,
807                                out_capable: true,
808                                in_capable: true,
809                                bitrate: Some(1_000_000_000), // 1 Gbps guess
810                                airtime_profile: None,
811                                announce_rate_target: None,
812                                announce_rate_grace: 0,
813                                announce_rate_penalty: 0.0,
814                                announce_cap: constants::ANNOUNCE_CAP,
815                                is_local_client: false,
816                                wants_tunnel: false,
817                                tunnel_id: None,
818                                mtu: 65535,
819                                ia_freq: 0.0,
820                                ip_freq: 0.0,
821                                op_freq: 0.0,
822                                op_samples: 0,
823                                started: 0.0,
824                                ingress_control,
825                            };
826
827                            let event = if let Some(template) = &dynamic_template {
828                                Event::DynamicInterfaceUp {
829                                    id: client_id,
830                                    writer,
831                                    registration: template.registration(info),
832                                }
833                            } else {
834                                Event::InterfaceUp(client_id, Some(writer), Some(info))
835                            };
836                            if tx.send(event).is_err() {
837                                // Driver shut down
838                                cleanup(&poller, &clients, &listener);
839                                return Ok(());
840                            }
841                        }
842                        Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break,
843                        Err(e) => {
844                            log::warn!("[{}] accept error: {}", name, e);
845                            break;
846                        }
847                    }
848                }
849                // Re-arm listener (oneshot semantics)
850                poller.modify(&listener, PollEvent::readable(LISTENER_KEY))?;
851            } else if clients.contains_key(&ev.key) {
852                let key = ev.key;
853                let mut should_remove = false;
854                let mut client_id = InterfaceId(0);
855
856                let mut buf = [0u8; 4096];
857                let read_result = {
858                    let client = clients.get_mut(&key).unwrap();
859                    client.stream.read(&mut buf)
860                };
861
862                match read_result {
863                    Ok(0) | Err(_) => {
864                        if let Some(c) = clients.get(&key) {
865                            client_id = c.id;
866                        }
867                        should_remove = true;
868                    }
869                    Ok(n) => {
870                        let client = clients.get_mut(&key).unwrap();
871                        client_id = client.id;
872                        client.has_received_data = true;
873                        let decoded = client.decoder.feed_with_diagnostics(&buf[..n]);
874                        for frame_len in decoded.invalid_frame_lengths {
875                            log::debug!(
876                            "[{}] invalid HDLC frame of {} bytes received from client {}, dropping frame",
877                            name,
878                            frame_len,
879                            client_id.0,
880                        );
881                        }
882                        for frame in decoded.frames {
883                            if tx
884                                .send(Event::Frame {
885                                    interface_id: client_id,
886                                    data: frame,
887                                    rssi: None,
888                                    snr: None,
889                                })
890                                .is_err()
891                            {
892                                cleanup(&poller, &clients, &listener);
893                                return Ok(());
894                            }
895                        }
896                    }
897                }
898
899                if should_remove {
900                    let reason = if clients
901                        .get(&key)
902                        .is_some_and(|c| c.write_stall_flag.load(Ordering::Relaxed))
903                    {
904                        DisconnectReason::WriteStall
905                    } else {
906                        DisconnectReason::RemoteClosed
907                    };
908                    disconnect_client(
909                        &poller,
910                        &mut clients,
911                        &mut peers,
912                        &name,
913                        server_interface_id,
914                        &tx,
915                        &peer_state,
916                        &fast_flap,
917                        &fast_flap_state,
918                        key,
919                        client_id,
920                        reason,
921                    );
922                } else if let Some(client) = clients.get(&key) {
923                    // Re-arm client (oneshot semantics)
924                    poller.modify(&client.stream, PollEvent::readable(key))?;
925                }
926            }
927        }
928
929        if let Some(timeout) = idle_timeout {
930            let now = Instant::now();
931            let timed_out: Vec<(usize, InterfaceId)> = clients
932                .iter()
933                .filter_map(|(&key, client)| {
934                    if client.has_received_data || now.duration_since(client.connected_at) < timeout
935                    {
936                        None
937                    } else {
938                        Some((key, client.id))
939                    }
940                })
941                .collect();
942
943            for (key, client_id) in timed_out {
944                disconnect_client(
945                    &poller,
946                    &mut clients,
947                    &mut peers,
948                    &name,
949                    server_interface_id,
950                    &tx,
951                    &peer_state,
952                    &fast_flap,
953                    &fast_flap_state,
954                    key,
955                    client_id,
956                    DisconnectReason::IdleTimeout,
957                );
958            }
959        }
960    }
961}
962
963fn cleanup_peer_state(peers: &mut HashMap<IpAddr, PeerBehaviorState>) {
964    let now = Instant::now();
965    peers.retain(|_, state| {
966        if matches!(state.blacklisted_until, Some(until) if now >= until) {
967            state.blacklisted_until = None;
968            state.blacklist_reason = None;
969        }
970        state.blacklisted_until.is_some() || state.connected_count > 0 || state.reject_count > 0
971    });
972}
973
974fn is_ip_blacklisted(peers: &mut HashMap<IpAddr, PeerBehaviorState>, peer_ip: IpAddr) -> bool {
975    let now = Instant::now();
976    if let Some(state) = peers.get_mut(&peer_ip) {
977        if let Some(until) = state.blacklisted_until {
978            if now < until {
979                return true;
980            }
981            state.blacklisted_until = None;
982        }
983    }
984    false
985}
986
987fn disconnect_client(
988    poller: &Poller,
989    clients: &mut HashMap<usize, ClientState>,
990    peers: &mut HashMap<IpAddr, PeerBehaviorState>,
991    name: &str,
992    server_interface_id: InterfaceId,
993    tx: &EventSender,
994    peer_state: &Arc<Mutex<BackbonePeerMonitor>>,
995    fast_flap: &BackboneFastFlapConfig,
996    fast_flap_state: &Arc<Mutex<BackboneFastFlapMonitor>>,
997    key: usize,
998    client_id: InterfaceId,
999    reason: DisconnectReason,
1000) {
1001    let Some(client) = clients.remove(&key) else {
1002        return;
1003    };
1004
1005    match reason {
1006        DisconnectReason::RemoteClosed => {
1007            log::trace!(target: crate::logging::PATHING_LOG_TARGET,
1008                "[{}] backbone client {} disconnected",
1009                name,
1010                client_id.0,
1011            );
1012        }
1013        DisconnectReason::IdleTimeout => {
1014            log::info!(
1015                "[{}] backbone client {} disconnected due to idle timeout",
1016                name,
1017                client_id.0
1018            );
1019        }
1020        DisconnectReason::WriteStall => {
1021            // Already logged by BackboneWriter::disconnect_for_write_stall
1022        }
1023    }
1024
1025    let _ = poller.delete(&client.stream);
1026    // client.stream closes on drop
1027    let connected_for = client.connected_at.elapsed();
1028    if let Some(record) = lock_or_recover(fast_flap_state, "backbone fast-flap state")
1029        .record_disconnect(client.peer_ip, connected_for, fast_flap, Instant::now())
1030    {
1031        let frequency = if record.elapsed_since_first.is_zero() {
1032            None
1033        } else {
1034            Some(record.flap_count as f64 / record.elapsed_since_first.as_secs_f64())
1035        };
1036        log::debug!(
1037            "[{}] peer {} fast flapped after {:.3}s ({} flaps{})",
1038            name,
1039            client.peer_ip,
1040            connected_for.as_secs_f64(),
1041            record.flap_count,
1042            frequency
1043                .map(|value| format!(", {:.3} Hz", value))
1044                .unwrap_or_default(),
1045        );
1046        if record.blocked {
1047            let state = peers
1048                .entry(client.peer_ip)
1049                .or_insert_with(PeerBehaviorState::new);
1050            state.blacklisted_until = record.blocked_until;
1051            state.blacklist_reason = Some("fast flapping".into());
1052            if record.newly_blocked {
1053                log::warn!(
1054                    "[{}] blocking further connections from {} due to fast flapping",
1055                    name,
1056                    client.peer_ip,
1057                );
1058            }
1059        }
1060    }
1061    let _ = tx.send(Event::BackbonePeerDisconnected {
1062        server_interface_id,
1063        peer_interface_id: client.id,
1064        peer_ip: client.peer_ip,
1065        peer_port: client.peer_port,
1066        connected_for,
1067        had_received_data: client.has_received_data,
1068    });
1069    match reason {
1070        DisconnectReason::IdleTimeout => {
1071            let _ = tx.send(Event::BackbonePeerIdleTimeout {
1072                server_interface_id,
1073                peer_interface_id: client.id,
1074                peer_ip: client.peer_ip,
1075                peer_port: client.peer_port,
1076                connected_for,
1077            });
1078        }
1079        DisconnectReason::WriteStall => {
1080            let _ = tx.send(Event::BackbonePeerWriteStall {
1081                server_interface_id,
1082                peer_interface_id: client.id,
1083                peer_ip: client.peer_ip,
1084                peer_port: client.peer_port,
1085                connected_for,
1086            });
1087        }
1088        DisconnectReason::RemoteClosed => {}
1089    }
1090
1091    if let Some(state) = peers.get_mut(&client.peer_ip) {
1092        state.connected_count = state.connected_count.saturating_sub(1);
1093    }
1094    peer_state.lock().unwrap().upsert_snapshot(peers);
1095    // Writer already sent InterfaceDown for write stalls; avoid duplicate.
1096    if !matches!(reason, DisconnectReason::WriteStall) {
1097        let _ = tx.send(Event::InterfaceDown(client_id));
1098    }
1099}
1100
1101fn set_tcp_keepalive(stream: &TcpStream) -> io::Result<()> {
1102    let sock = SockRef::from(stream);
1103    let mut keepalive = TcpKeepalive::new()
1104        .with_time(Duration::from_secs(5))
1105        .with_interval(Duration::from_secs(2));
1106    #[cfg(any(target_os = "linux", target_os = "macos"))]
1107    {
1108        keepalive = keepalive.with_retries(12);
1109    }
1110    sock.set_tcp_keepalive(&keepalive)
1111}
1112
1113fn cleanup(poller: &Poller, clients: &HashMap<usize, ClientState>, listener: &TcpListener) {
1114    for (_, client) in clients {
1115        let _ = poller.delete(&client.stream);
1116    }
1117    let _ = poller.delete(listener);
1118}
1119
1120// ---------------------------------------------------------------------------
1121// Client mode
1122// ---------------------------------------------------------------------------
1123
1124/// Configuration for a backbone client interface.
1125#[derive(Debug, Clone)]
1126pub struct BackboneClientConfig {
1127    pub name: String,
1128    pub target_host: String,
1129    pub target_port: u16,
1130    pub interface_id: InterfaceId,
1131    pub reconnect_wait: Duration,
1132    pub max_reconnect_tries: Option<u32>,
1133    pub connect_timeout: Duration,
1134    pub transport_identity: Option<String>,
1135    pub runtime: Arc<Mutex<BackboneClientRuntime>>,
1136}
1137
1138#[derive(Debug, Clone)]
1139pub struct BackboneClientRuntime {
1140    pub reconnect_wait: Duration,
1141    pub max_reconnect_tries: Option<u32>,
1142    pub connect_timeout: Duration,
1143}
1144
1145impl BackboneClientRuntime {
1146    pub fn from_config(config: &BackboneClientConfig) -> Self {
1147        Self {
1148            reconnect_wait: config.reconnect_wait,
1149            max_reconnect_tries: config.max_reconnect_tries,
1150            connect_timeout: config.connect_timeout,
1151        }
1152    }
1153}
1154
1155#[derive(Debug, Clone)]
1156pub struct BackboneClientRuntimeConfigHandle {
1157    pub interface_name: String,
1158    pub runtime: Arc<Mutex<BackboneClientRuntime>>,
1159    pub startup: BackboneClientRuntime,
1160}
1161
1162impl Default for BackboneClientConfig {
1163    fn default() -> Self {
1164        let mut config = BackboneClientConfig {
1165            name: String::new(),
1166            target_host: "127.0.0.1".into(),
1167            target_port: 4242,
1168            interface_id: InterfaceId(0),
1169            reconnect_wait: Duration::from_secs(5),
1170            max_reconnect_tries: None,
1171            connect_timeout: Duration::from_secs(5),
1172            transport_identity: None,
1173            runtime: Arc::new(Mutex::new(BackboneClientRuntime {
1174                reconnect_wait: Duration::from_secs(5),
1175                max_reconnect_tries: None,
1176                connect_timeout: Duration::from_secs(5),
1177            })),
1178        };
1179        let startup = BackboneClientRuntime::from_config(&config);
1180        config.runtime = Arc::new(Mutex::new(startup));
1181        config
1182    }
1183}
1184
1185/// Writer that sends HDLC-framed data over a TCP stream (client mode).
1186struct BackboneClientWriter {
1187    stream: TcpStream,
1188}
1189
1190impl Writer for BackboneClientWriter {
1191    fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
1192        self.stream.write_all(&hdlc::frame(data))
1193    }
1194}
1195
1196/// Try to connect to the target host:port with timeout.
1197fn try_connect_client(config: &BackboneClientConfig) -> io::Result<TcpStream> {
1198    let runtime = config.runtime.lock().unwrap().clone();
1199    let addr_str = format!("{}:{}", config.target_host, config.target_port);
1200    let addr = addr_str
1201        .to_socket_addrs()?
1202        .next()
1203        .ok_or_else(|| io::Error::new(io::ErrorKind::AddrNotAvailable, "no addresses resolved"))?;
1204
1205    let stream = TcpStream::connect_timeout(&addr, runtime.connect_timeout)?;
1206    stream.set_nodelay(true)?;
1207    set_tcp_keepalive(&stream).ok();
1208
1209    // Prevent SIGPIPE on macOS when writing to broken pipes
1210    #[cfg(target_os = "macos")]
1211    {
1212        let sock = SockRef::from(&stream);
1213        sock.set_nosigpipe(true).ok();
1214    }
1215
1216    Ok(stream)
1217}
1218
1219/// Connect and start the reader thread. Returns the writer for the driver.
1220pub fn start_client(config: BackboneClientConfig, tx: EventSender) -> io::Result<Box<dyn Writer>> {
1221    start_client_with_ifac(config, tx, 0)
1222}
1223
1224pub(crate) fn start_client_with_ifac(
1225    config: BackboneClientConfig,
1226    tx: EventSender,
1227    ifac_size: usize,
1228) -> io::Result<Box<dyn Writer>> {
1229    let stream = try_connect_client(&config)?;
1230    let reader_stream = stream.try_clone()?;
1231    let writer_stream = stream.try_clone()?;
1232
1233    let id = config.interface_id;
1234    log::info!(
1235        "[{}] backbone client connected to {}:{}",
1236        config.name,
1237        config.target_host,
1238        config.target_port
1239    );
1240
1241    // Initial connect: writer is None because it's returned directly to the caller
1242    let _ = tx.send(Event::InterfaceUp(id, None, None));
1243
1244    thread::Builder::new()
1245        .name(format!("backbone-client-{}", id.0))
1246        .spawn(move || {
1247            client_reader_loop(reader_stream, config, tx, ifac_size);
1248        })?;
1249
1250    Ok(Box::new(BackboneClientWriter {
1251        stream: writer_stream,
1252    }))
1253}
1254
1255/// Reader thread: reads from socket, HDLC-decodes, sends frames to driver.
1256/// On disconnect, attempts reconnection.
1257fn client_reader_loop(
1258    mut stream: TcpStream,
1259    config: BackboneClientConfig,
1260    tx: EventSender,
1261    ifac_size: usize,
1262) {
1263    let id = config.interface_id;
1264    let mut decoder = hdlc::Decoder::reticulum(HW_MTU, ifac_size);
1265    let mut buf = [0u8; 4096];
1266
1267    loop {
1268        match stream.read(&mut buf) {
1269            Ok(0) => {
1270                log::warn!("[{}] connection closed", config.name);
1271                let _ = tx.send(Event::InterfaceDown(id));
1272                match client_reconnect(&config, &tx) {
1273                    Some(new_stream) => {
1274                        stream = new_stream;
1275                        decoder = hdlc::Decoder::reticulum(HW_MTU, ifac_size);
1276                        continue;
1277                    }
1278                    None => {
1279                        log::error!("[{}] reconnection failed, giving up", config.name);
1280                        return;
1281                    }
1282                }
1283            }
1284            Ok(n) => {
1285                let decoded = decoder.feed_with_diagnostics(&buf[..n]);
1286                for frame_len in decoded.invalid_frame_lengths {
1287                    log::debug!(
1288                        "[{}] invalid HDLC frame of {} bytes received, dropping frame",
1289                        config.name,
1290                        frame_len,
1291                    );
1292                }
1293                for frame in decoded.frames {
1294                    if tx
1295                        .send(Event::Frame {
1296                            interface_id: id,
1297                            data: frame,
1298                            rssi: None,
1299                            snr: None,
1300                        })
1301                        .is_err()
1302                    {
1303                        return;
1304                    }
1305                }
1306            }
1307            Err(e) => {
1308                log::warn!("[{}] read error: {}", config.name, e);
1309                let _ = tx.send(Event::InterfaceDown(id));
1310                match client_reconnect(&config, &tx) {
1311                    Some(new_stream) => {
1312                        stream = new_stream;
1313                        decoder = hdlc::Decoder::reticulum(HW_MTU, ifac_size);
1314                        continue;
1315                    }
1316                    None => {
1317                        log::error!("[{}] reconnection failed, giving up", config.name);
1318                        return;
1319                    }
1320                }
1321            }
1322        }
1323    }
1324}
1325
1326/// Maximum backoff multiplier: `base_delay * 2^MAX_BACKOFF_SHIFT`.
1327/// With a 5 s base this caps at 5 × 2^6 = 320 s ≈ 5 min.
1328const MAX_BACKOFF_SHIFT: u32 = 6;
1329
1330/// Attempt to reconnect with exponential backoff and jitter.
1331/// Returns the new reader stream on success.
1332/// Sends the new writer to the driver via InterfaceUp event.
1333fn client_reconnect(config: &BackboneClientConfig, tx: &EventSender) -> Option<TcpStream> {
1334    let mut attempts = 0u32;
1335    loop {
1336        let runtime = config.runtime.lock().unwrap().clone();
1337
1338        let shift = attempts.min(MAX_BACKOFF_SHIFT);
1339        let backoff = runtime.reconnect_wait * 2u32.pow(shift);
1340        // Add ±25 % jitter to avoid thundering-herd reconnects.
1341        let jitter_range = backoff / 4;
1342        let jitter = if jitter_range.as_nanos() > 0 {
1343            let offset = Duration::from_nanos(
1344                (std::hash::RandomState::new().build_hasher().finish()
1345                    % jitter_range.as_nanos() as u64)
1346                    * 2,
1347            );
1348            if offset > jitter_range {
1349                backoff + (offset - jitter_range)
1350            } else {
1351                backoff - (jitter_range - offset)
1352            }
1353        } else {
1354            backoff
1355        };
1356        thread::sleep(jitter);
1357
1358        attempts += 1;
1359
1360        if let Some(max) = runtime.max_reconnect_tries {
1361            if attempts > max {
1362                let _ = tx.send(Event::InterfaceDown(config.interface_id));
1363                return None;
1364            }
1365        }
1366
1367        log::info!(
1368            "[{}] reconnect attempt {} (backoff {:.1}s) ...",
1369            config.name,
1370            attempts,
1371            jitter.as_secs_f64(),
1372        );
1373
1374        match try_connect_client(config) {
1375            Ok(new_stream) => {
1376                let writer_stream = match new_stream.try_clone() {
1377                    Ok(s) => s,
1378                    Err(e) => {
1379                        log::warn!("[{}] failed to clone stream: {}", config.name, e);
1380                        continue;
1381                    }
1382                };
1383                log::info!(
1384                    "[{}] reconnected after {} attempt(s)",
1385                    config.name,
1386                    attempts
1387                );
1388                let new_writer: Box<dyn Writer> = Box::new(BackboneClientWriter {
1389                    stream: writer_stream,
1390                });
1391                let _ = tx.send(Event::InterfaceUp(
1392                    config.interface_id,
1393                    Some(new_writer),
1394                    None,
1395                ));
1396                return Some(new_stream);
1397            }
1398            Err(e) => {
1399                log::warn!("[{}] reconnect failed: {}", config.name, e);
1400            }
1401        }
1402    }
1403}
1404
1405// ---------------------------------------------------------------------------
1406// Factory
1407// ---------------------------------------------------------------------------
1408
1409/// Internal enum used by [`BackboneInterfaceFactory`] to carry either a
1410/// server or client config through the opaque `InterfaceConfigData` channel.
1411#[derive(Clone)]
1412pub(crate) enum BackboneMode {
1413    Server(BackboneConfig),
1414    Client(BackboneClientConfig, u8),
1415}
1416
1417/// Factory for `BackboneInterface`.
1418///
1419/// If the config params contain `"remote"` or `"target_host"` the interface
1420/// is started in client mode; otherwise it is started as a TCP listener
1421/// (server mode).
1422pub struct BackboneInterfaceFactory;
1423
1424/// Compatibility name used by Reticulum 1.3.8 for a forced client-mode
1425/// Backbone interface.
1426pub struct BackboneClientInterfaceFactory;
1427
1428impl InterfaceFactory for BackboneClientInterfaceFactory {
1429    fn type_name(&self) -> &str {
1430        "BackboneClientInterface"
1431    }
1432
1433    fn parse_config(
1434        &self,
1435        name: &str,
1436        id: InterfaceId,
1437        params: &HashMap<String, String>,
1438    ) -> Result<Box<dyn InterfaceConfigData>, String> {
1439        if !params.contains_key("remote") && !params.contains_key("target_host") {
1440            return Err("BackboneClientInterface requires 'remote' or 'target_host'".into());
1441        }
1442        BackboneInterfaceFactory.parse_config(name, id, params)
1443    }
1444
1445    fn start(
1446        &self,
1447        config: Box<dyn InterfaceConfigData>,
1448        ctx: StartContext,
1449    ) -> io::Result<StartResult> {
1450        BackboneInterfaceFactory.start(config, ctx)
1451    }
1452}
1453
1454fn parse_positive_duration_secs(params: &HashMap<String, String>, key: &str) -> Option<Duration> {
1455    params
1456        .get(key)
1457        .and_then(|v| v.parse::<f64>().ok())
1458        .filter(|v| *v > 0.0)
1459        .map(Duration::from_secs_f64)
1460}
1461
1462fn parse_nonnegative_duration(
1463    params: &HashMap<String, String>,
1464    key: &str,
1465    multiplier: f64,
1466    default: Duration,
1467) -> Result<Duration, String> {
1468    let Some(raw) = params.get(key) else {
1469        return Ok(default);
1470    };
1471    let value = raw
1472        .parse::<f64>()
1473        .map_err(|_| format!("invalid {key} '{raw}' (expected a non-negative number)"))?;
1474    let seconds = value * multiplier;
1475    if !value.is_finite() || value < 0.0 || !seconds.is_finite() || seconds > u64::MAX as f64 {
1476        return Err(format!(
1477            "invalid {key} '{raw}' (expected a finite non-negative number)"
1478        ));
1479    }
1480    let duration = Duration::from_secs_f64(seconds);
1481    if Instant::now().checked_add(duration).is_none() {
1482        return Err(format!("invalid {key} '{raw}' (duration is too large)"));
1483    }
1484    Ok(duration)
1485}
1486
1487fn parse_fast_flap_config(
1488    params: &HashMap<String, String>,
1489) -> Result<BackboneFastFlapConfig, String> {
1490    let defaults = BackboneFastFlapConfig::default();
1491    let enabled = match params.get("block_fast_flapping") {
1492        Some(raw) => crate::config::parse_bool_pub(raw)
1493            .ok_or_else(|| format!("invalid block_fast_flapping '{raw}' (expected yes or no)"))?,
1494        None => defaults.enabled,
1495    };
1496    let connection_threshold = parse_nonnegative_duration(
1497        params,
1498        "fast_flapping_threshold",
1499        1.0,
1500        defaults.connection_threshold,
1501    )?;
1502    let grace = match params.get("fast_flapping_grace") {
1503        Some(raw) => raw.parse::<u64>().map_err(|_| {
1504            format!("invalid fast_flapping_grace '{raw}' (expected a non-negative integer)")
1505        })?,
1506        None => defaults.grace,
1507    };
1508    let block_duration = parse_nonnegative_duration(
1509        params,
1510        "fast_flapping_block_time",
1511        60.0,
1512        defaults.block_duration,
1513    )?;
1514
1515    Ok(BackboneFastFlapConfig {
1516        enabled,
1517        connection_threshold,
1518        grace,
1519        block_duration,
1520    })
1521}
1522
1523impl InterfaceFactory for BackboneInterfaceFactory {
1524    fn type_name(&self) -> &str {
1525        "BackboneInterface"
1526    }
1527
1528    fn parse_config(
1529        &self,
1530        name: &str,
1531        id: InterfaceId,
1532        params: &HashMap<String, String>,
1533    ) -> Result<Box<dyn InterfaceConfigData>, String> {
1534        if let Some(target_host) = params.get("remote").or_else(|| params.get("target_host")) {
1535            // Client mode
1536            let target_host = target_host.clone();
1537            let target_port = params
1538                .get("target_port")
1539                .or_else(|| params.get("port"))
1540                .and_then(|v| v.parse().ok())
1541                .unwrap_or(4242);
1542            let transport_identity = params.get("transport_identity").cloned();
1543            let priority = match params.get("priority") {
1544                Some(value) => value.parse::<u8>().map_err(|_| {
1545                    format!(
1546                        "invalid Backbone peer priority '{}' (expected 0..100)",
1547                        value
1548                    )
1549                })?,
1550                None => crate::driver::BACKBONE_PEER_POOL_CONFIGURED_DEFAULT_PRIORITY,
1551            };
1552            if priority > 100 {
1553                return Err(format!(
1554                    "invalid Backbone peer priority '{}' (expected 0..100)",
1555                    priority
1556                ));
1557            }
1558            Ok(Box::new(BackboneMode::Client(
1559                BackboneClientConfig {
1560                    name: name.to_string(),
1561                    target_host,
1562                    target_port,
1563                    interface_id: id,
1564                    transport_identity,
1565                    ..BackboneClientConfig::default()
1566                },
1567                priority,
1568            )))
1569        } else {
1570            // Server mode
1571            let listen_ip = params
1572                .get("listen_ip")
1573                .or_else(|| params.get("device"))
1574                .cloned()
1575                .unwrap_or_else(|| "0.0.0.0".into());
1576            let listen_port = params
1577                .get("listen_port")
1578                .or_else(|| params.get("port"))
1579                .and_then(|v| v.parse().ok())
1580                .unwrap_or(4242);
1581            let max_connections = params.get("max_connections").and_then(|v| v.parse().ok());
1582            let idle_timeout = parse_positive_duration_secs(params, "idle_timeout");
1583            let write_stall_timeout = parse_positive_duration_secs(params, "write_stall_timeout");
1584            let abuse = BackboneAbuseConfig {
1585                max_penalty_duration: parse_positive_duration_secs(params, "max_penalty_duration"),
1586            };
1587            let fast_flap = parse_fast_flap_config(params)?;
1588            let mut config = BackboneConfig {
1589                name: name.to_string(),
1590                listen_ip,
1591                listen_port,
1592                interface_id: id,
1593                mode: constants::MODE_FULL,
1594                gravity: 0,
1595                recursive_prs: false,
1596                announces_from_internal: true,
1597                announces_to_internal: None,
1598                max_connections,
1599                idle_timeout,
1600                write_stall_timeout,
1601                abuse,
1602                fast_flap,
1603                fast_flap_state: shared_fast_flap_monitor(),
1604                ingress_control: IngressControlConfig::enabled(),
1605                runtime: Arc::new(Mutex::new(BackboneServerRuntime {
1606                    max_connections: None,
1607                    idle_timeout: None,
1608                    write_stall_timeout: None,
1609                    abuse: BackboneAbuseConfig::default(),
1610                })),
1611                peer_state: Arc::new(Mutex::new(BackbonePeerMonitor::new())),
1612            };
1613            let startup = BackboneServerRuntime::from_config(&config);
1614            config.runtime = Arc::new(Mutex::new(startup));
1615            Ok(Box::new(BackboneMode::Server(config)))
1616        }
1617    }
1618
1619    fn start(
1620        &self,
1621        config: Box<dyn InterfaceConfigData>,
1622        ctx: StartContext,
1623    ) -> io::Result<StartResult> {
1624        let mode = *config.into_any().downcast::<BackboneMode>().map_err(|_| {
1625            io::Error::new(
1626                io::ErrorKind::InvalidData,
1627                "wrong config type for BackboneInterface",
1628            )
1629        })?;
1630
1631        match mode {
1632            BackboneMode::Client(cfg, _) => {
1633                let id = cfg.interface_id;
1634                let name = cfg.name.clone();
1635                let info = InterfaceInfo {
1636                    id,
1637                    name,
1638                    mode: ctx.mode,
1639                    gravity: ctx.gravity,
1640                    recursive_prs: ctx.recursive_prs,
1641                    announces_from_internal: ctx.announces_from_internal,
1642                    announces_to_internal: ctx.announces_to_internal,
1643                    out_capable: true,
1644                    in_capable: true,
1645                    bitrate: Some(1_000_000_000),
1646                    airtime_profile: None,
1647                    announce_rate_target: None,
1648                    announce_rate_grace: 0,
1649                    announce_rate_penalty: 0.0,
1650                    announce_cap: constants::ANNOUNCE_CAP,
1651                    is_local_client: false,
1652                    wants_tunnel: false,
1653                    tunnel_id: None,
1654                    mtu: 65535,
1655                    ingress_control: ctx.ingress_control,
1656                    ia_freq: 0.0,
1657                    ip_freq: 0.0,
1658                    op_freq: 0.0,
1659                    op_samples: 0,
1660                    started: crate::time::now(),
1661                };
1662                let ifac_size = ctx.ifac.as_ref().map(|ifac| ifac.size).unwrap_or(0);
1663                let writer = start_client_with_ifac(cfg, ctx.tx, ifac_size)?;
1664                Ok(StartResult::Simple {
1665                    id,
1666                    info,
1667                    writer,
1668                    interface_type_name: "BackboneInterface".to_string(),
1669                })
1670            }
1671            BackboneMode::Server(mut cfg) => {
1672                cfg.ingress_control = ctx.ingress_control;
1673                cfg.mode = ctx.mode;
1674                cfg.gravity = ctx.gravity;
1675                cfg.recursive_prs = ctx.recursive_prs;
1676                cfg.announces_from_internal = ctx.announces_from_internal;
1677                cfg.announces_to_internal = ctx.announces_to_internal;
1678                let parent_id = cfg.interface_id;
1679                start_with_template(
1680                    cfg,
1681                    ctx.tx,
1682                    ctx.next_dynamic_id,
1683                    Some(DynamicInterfaceTemplate {
1684                        parent_id,
1685                        interface_type: "BackboneInterfacePeer".into(),
1686                        ifac: ctx.ifac,
1687                        mode: ctx.mode,
1688                        gravity: ctx.gravity,
1689                        recursive_prs: ctx.recursive_prs,
1690                        announces_from_internal: ctx.announces_from_internal,
1691                        announces_to_internal: ctx.announces_to_internal,
1692                    }),
1693                )?;
1694                Ok(StartResult::Listener { control: None })
1695            }
1696        }
1697    }
1698}
1699
1700pub(crate) fn runtime_handle_from_mode(mode: &BackboneMode) -> Option<BackboneRuntimeConfigHandle> {
1701    match mode {
1702        BackboneMode::Server(config) => Some(BackboneRuntimeConfigHandle {
1703            interface_name: config.name.clone(),
1704            runtime: Arc::clone(&config.runtime),
1705            startup: BackboneServerRuntime::from_config(config),
1706        }),
1707        BackboneMode::Client(_, _) => None,
1708    }
1709}
1710
1711pub(crate) fn peer_state_handle_from_mode(mode: &BackboneMode) -> Option<BackbonePeerStateHandle> {
1712    match mode {
1713        BackboneMode::Server(config) => Some(BackbonePeerStateHandle {
1714            interface_id: config.interface_id,
1715            interface_name: config.name.clone(),
1716            mode: config.mode,
1717            gravity: config.gravity,
1718            announces_to_internal: config.announces_to_internal,
1719            peer_state: Arc::clone(&config.peer_state),
1720            fast_flap: config.fast_flap,
1721            fast_flap_state: Arc::clone(&config.fast_flap_state),
1722        }),
1723        BackboneMode::Client(_, _) => None,
1724    }
1725}
1726
1727pub(crate) fn client_runtime_handle_from_mode(
1728    mode: &BackboneMode,
1729) -> Option<BackboneClientRuntimeConfigHandle> {
1730    match mode {
1731        BackboneMode::Client(config, _) => Some(BackboneClientRuntimeConfigHandle {
1732            interface_name: config.name.clone(),
1733            runtime: Arc::clone(&config.runtime),
1734            startup: BackboneClientRuntime::from_config(config),
1735        }),
1736        BackboneMode::Server(_) => None,
1737    }
1738}
1739
1740pub(crate) fn client_config_from_mode(mode: &BackboneMode) -> Option<BackboneClientConfig> {
1741    match mode {
1742        BackboneMode::Client(config, _) => Some(config.clone()),
1743        BackboneMode::Server(_) => None,
1744    }
1745}
1746
1747pub(crate) fn client_priority_from_mode(mode: &BackboneMode) -> Option<u8> {
1748    match mode {
1749        BackboneMode::Client(_, priority) => Some(*priority),
1750        BackboneMode::Server(_) => None,
1751    }
1752}
1753
1754#[cfg(test)]
1755mod tests {
1756    use super::*;
1757    use std::sync::mpsc;
1758    use std::time::Duration;
1759
1760    fn find_free_port() -> u16 {
1761        TcpListener::bind("127.0.0.1:0")
1762            .unwrap()
1763            .local_addr()
1764            .unwrap()
1765            .port()
1766    }
1767
1768    fn test_fast_flap_config() -> BackboneFastFlapConfig {
1769        BackboneFastFlapConfig {
1770            enabled: true,
1771            connection_threshold: Duration::from_secs(20),
1772            grace: 5,
1773            block_duration: Duration::from_secs(12 * 60 * 60),
1774        }
1775    }
1776
1777    #[test]
1778    fn fast_flap_defaults_match_upstream() {
1779        assert_eq!(BackboneFastFlapConfig::default(), test_fast_flap_config());
1780    }
1781
1782    #[test]
1783    fn sixth_short_connection_blocks_but_fifth_does_not() {
1784        let config = test_fast_flap_config();
1785        let mut monitor = BackboneFastFlapMonitor::new();
1786        let peer: IpAddr = "192.0.2.10".parse().unwrap();
1787        let start = Instant::now();
1788
1789        for flap in 1..=5 {
1790            let record = monitor
1791                .record_disconnect(
1792                    peer,
1793                    Duration::from_secs(19),
1794                    &config,
1795                    start + Duration::from_secs(flap),
1796                )
1797                .unwrap();
1798            assert_eq!(record.flap_count, flap as u64);
1799            assert!(!record.blocked);
1800            assert!(!monitor.is_blocked(peer, &config, start + Duration::from_secs(flap)));
1801        }
1802
1803        let sixth = monitor
1804            .record_disconnect(
1805                peer,
1806                Duration::from_secs(1),
1807                &config,
1808                start + Duration::from_secs(6),
1809            )
1810            .unwrap();
1811        assert_eq!(sixth.flap_count, 6);
1812        assert!(sixth.blocked);
1813        assert!(sixth.newly_blocked);
1814        assert!(monitor.is_blocked(peer, &config, start + Duration::from_secs(6)));
1815    }
1816
1817    #[test]
1818    fn every_short_connection_counts_even_after_receiving_data() {
1819        let config = test_fast_flap_config();
1820        let mut monitor = BackboneFastFlapMonitor::new();
1821        let peer: IpAddr = "192.0.2.11".parse().unwrap();
1822        let now = Instant::now();
1823
1824        let record = monitor
1825            .record_disconnect(peer, Duration::from_secs(2), &config, now)
1826            .unwrap();
1827        assert_eq!(record.flap_count, 1);
1828        assert_eq!(monitor.flap_count(peer), 1);
1829    }
1830
1831    #[test]
1832    fn threshold_boundary_and_long_connections_are_not_counted() {
1833        let config = test_fast_flap_config();
1834        let mut monitor = BackboneFastFlapMonitor::new();
1835        let peer: IpAddr = "192.0.2.12".parse().unwrap();
1836        let now = Instant::now();
1837
1838        assert!(monitor
1839            .record_disconnect(peer, Duration::from_secs(20), &config, now)
1840            .is_none());
1841        assert!(monitor
1842            .record_disconnect(peer, Duration::from_secs(21), &config, now)
1843            .is_none());
1844        assert_eq!(monitor.flap_count(peer), 0);
1845    }
1846
1847    #[test]
1848    fn disabled_fast_flapping_neither_tracks_nor_blocks() {
1849        let mut config = test_fast_flap_config();
1850        config.enabled = false;
1851        let mut monitor = BackboneFastFlapMonitor::new();
1852        let peer: IpAddr = "192.0.2.13".parse().unwrap();
1853        let now = Instant::now();
1854
1855        for offset in 0..10 {
1856            assert!(monitor
1857                .record_disconnect(
1858                    peer,
1859                    Duration::from_secs(1),
1860                    &config,
1861                    now + Duration::from_secs(offset),
1862                )
1863                .is_none());
1864        }
1865        assert_eq!(monitor.flap_count(peer), 0);
1866        assert!(!monitor.is_blocked(peer, &config, now + Duration::from_secs(10)));
1867        assert_eq!(monitor.blocked_ip_count(&config, now), 0);
1868    }
1869
1870    #[test]
1871    fn flap_blocks_expire_strictly_after_configured_duration() {
1872        let mut config = test_fast_flap_config();
1873        config.grace = 0;
1874        config.block_duration = Duration::from_secs(60);
1875        let mut monitor = BackboneFastFlapMonitor::new();
1876        let peer: IpAddr = "192.0.2.14".parse().unwrap();
1877        let now = Instant::now();
1878
1879        monitor
1880            .record_disconnect(peer, Duration::from_secs(1), &config, now)
1881            .unwrap();
1882        assert!(monitor.is_blocked(peer, &config, now + Duration::from_secs(60)));
1883        assert!(!monitor.is_blocked(
1884            peer,
1885            &config,
1886            now + Duration::from_secs(60) + Duration::from_nanos(1)
1887        ));
1888        assert_eq!(monitor.flap_count(peer), 0);
1889    }
1890
1891    #[test]
1892    fn fast_flap_history_is_isolated_by_ip() {
1893        let mut config = test_fast_flap_config();
1894        config.grace = 1;
1895        let mut monitor = BackboneFastFlapMonitor::new();
1896        let first: IpAddr = "192.0.2.15".parse().unwrap();
1897        let second: IpAddr = "192.0.2.16".parse().unwrap();
1898        let now = Instant::now();
1899
1900        for offset in 0..2 {
1901            monitor
1902                .record_disconnect(
1903                    first,
1904                    Duration::from_secs(1),
1905                    &config,
1906                    now + Duration::from_secs(offset),
1907                )
1908                .unwrap();
1909        }
1910        monitor
1911            .record_disconnect(second, Duration::from_secs(1), &config, now)
1912            .unwrap();
1913
1914        assert!(monitor.is_blocked(first, &config, now + Duration::from_secs(2)));
1915        assert!(!monitor.is_blocked(second, &config, now + Duration::from_secs(2)));
1916        assert_eq!(
1917            monitor.blocked_ip_count(&config, now + Duration::from_secs(2)),
1918            1
1919        );
1920    }
1921
1922    #[test]
1923    fn blocked_ip_list_is_sorted_current_and_empty_when_disabled() {
1924        let mut config = test_fast_flap_config();
1925        config.grace = 0;
1926        config.block_duration = Duration::from_secs(60);
1927        let mut monitor = BackboneFastFlapMonitor::new();
1928        let first: IpAddr = "203.0.113.20".parse().unwrap();
1929        let second: IpAddr = "192.0.2.10".parse().unwrap();
1930        let now = Instant::now();
1931        monitor
1932            .record_disconnect(first, Duration::from_secs(1), &config, now)
1933            .unwrap();
1934        monitor
1935            .record_disconnect(second, Duration::from_secs(1), &config, now)
1936            .unwrap();
1937
1938        assert_eq!(monitor.blocked_ip_list(&config, now), vec![second, first]);
1939        assert!(monitor
1940            .blocked_ip_list(&config, now + Duration::from_secs(61))
1941            .is_empty());
1942        config.enabled = false;
1943        assert!(monitor.blocked_ip_list(&config, now).is_empty());
1944    }
1945
1946    #[test]
1947    fn peer_monitor_snapshot_clears_expired_blacklist_state() {
1948        let peer: IpAddr = "192.0.2.17".parse().unwrap();
1949        let mut monitor = BackbonePeerMonitor::new();
1950        monitor.blacklist(peer, Duration::from_secs(60), "temporary".into());
1951        let mut peers = HashMap::new();
1952        monitor.sync_into(&mut peers);
1953        let state = peers.get_mut(&peer).unwrap();
1954        state.blacklisted_until = None;
1955        state.blacklist_reason = None;
1956
1957        monitor.upsert_snapshot(&peers);
1958
1959        let entry = monitor.list("test").into_iter().next().unwrap();
1960        assert!(entry.blacklisted_remaining_secs.is_none());
1961        assert!(entry.blacklist_reason.is_none());
1962    }
1963
1964    fn recv_non_peer_event(
1965        rx: &mpsc::Receiver<Event>,
1966        timeout: Duration,
1967    ) -> Result<Event, mpsc::RecvTimeoutError> {
1968        let deadline = Instant::now() + timeout;
1969        loop {
1970            let remaining = deadline.saturating_duration_since(Instant::now());
1971            if remaining.is_zero() {
1972                return Err(mpsc::RecvTimeoutError::Timeout);
1973            }
1974            let event = rx.recv_timeout(remaining)?;
1975            match event {
1976                Event::BackbonePeerConnected { .. }
1977                | Event::BackbonePeerDisconnected { .. }
1978                | Event::BackbonePeerIdleTimeout { .. }
1979                | Event::BackbonePeerWriteStall { .. }
1980                | Event::BackbonePeerPenalty { .. } => continue,
1981                other => return Ok(other),
1982            }
1983        }
1984    }
1985
1986    fn make_server_config(
1987        port: u16,
1988        interface_id: u64,
1989        max_connections: Option<usize>,
1990        idle_timeout: Option<Duration>,
1991        write_stall_timeout: Option<Duration>,
1992        abuse: BackboneAbuseConfig,
1993    ) -> BackboneConfig {
1994        let mut config = BackboneConfig {
1995            name: "test-backbone".into(),
1996            listen_ip: "127.0.0.1".into(),
1997            listen_port: port,
1998            interface_id: InterfaceId(interface_id),
1999            mode: constants::MODE_FULL,
2000            gravity: 0,
2001            recursive_prs: false,
2002            announces_from_internal: true,
2003            announces_to_internal: None,
2004            max_connections,
2005            idle_timeout,
2006            write_stall_timeout,
2007            abuse,
2008            fast_flap: BackboneFastFlapConfig::default(),
2009            fast_flap_state: Arc::new(Mutex::new(BackboneFastFlapMonitor::new())),
2010            ingress_control: IngressControlConfig::enabled(),
2011            runtime: Arc::new(Mutex::new(BackboneServerRuntime {
2012                max_connections: None,
2013                idle_timeout: None,
2014                write_stall_timeout: None,
2015                abuse: BackboneAbuseConfig::default(),
2016            })),
2017            peer_state: Arc::new(Mutex::new(BackbonePeerMonitor::new())),
2018        };
2019        let startup = BackboneServerRuntime::from_config(&config);
2020        config.runtime = Arc::new(Mutex::new(startup));
2021        config
2022    }
2023
2024    #[test]
2025    fn backbone_accept_connection() {
2026        let port = find_free_port();
2027        let (tx, rx) = crate::event::channel();
2028        let next_id = Arc::new(AtomicU64::new(8000));
2029
2030        let config = make_server_config(port, 80, None, None, None, BackboneAbuseConfig::default());
2031
2032        start(config, tx, next_id).unwrap();
2033        thread::sleep(Duration::from_millis(50));
2034
2035        let _client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2036
2037        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2038        match event {
2039            Event::InterfaceUp(id, writer, info) => {
2040                assert_eq!(id, InterfaceId(8000));
2041                assert!(writer.is_some());
2042                assert!(info.is_some());
2043                let info = info.unwrap();
2044                assert!(info.out_capable);
2045                assert!(info.in_capable);
2046            }
2047            other => panic!("expected InterfaceUp, got {:?}", other),
2048        }
2049    }
2050
2051    #[test]
2052    fn backbone_accepted_connection_inherits_server_mode() {
2053        let port = find_free_port();
2054        let (tx, rx) = crate::event::channel();
2055        let next_id = Arc::new(AtomicU64::new(8050));
2056
2057        let mut config =
2058            make_server_config(port, 85, None, None, None, BackboneAbuseConfig::default());
2059        config.mode = constants::MODE_GATEWAY;
2060
2061        start(config, tx, next_id).unwrap();
2062        thread::sleep(Duration::from_millis(50));
2063
2064        let _client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2065
2066        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2067        match event {
2068            Event::InterfaceUp(_, _, Some(info)) => {
2069                assert_eq!(info.mode, constants::MODE_GATEWAY);
2070            }
2071            other => panic!("expected InterfaceUp with info, got {:?}", other),
2072        }
2073    }
2074
2075    #[test]
2076    fn backbone_receive_frame() {
2077        let port = find_free_port();
2078        let (tx, rx) = crate::event::channel();
2079        let next_id = Arc::new(AtomicU64::new(8100));
2080
2081        let config = make_server_config(port, 81, None, None, None, BackboneAbuseConfig::default());
2082
2083        start(config, tx, next_id).unwrap();
2084        thread::sleep(Duration::from_millis(50));
2085
2086        let mut client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2087
2088        // Drain InterfaceUp
2089        let _ = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();
2090
2091        // The exact Reticulum header minimum must be dropped without
2092        // preventing recovery of the next valid frame.
2093        client
2094            .write_all(&hdlc::frame(&vec![
2095                0x11;
2096                rns_core::constants::HEADER_MINSIZE
2097            ]))
2098            .unwrap();
2099        let payload: Vec<u8> = (0..32).collect();
2100        client.write_all(&hdlc::frame(&payload)).unwrap();
2101
2102        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2103        match event {
2104            Event::Frame {
2105                interface_id,
2106                data,
2107                rssi: _,
2108                snr: _,
2109            } => {
2110                assert_eq!(interface_id, InterfaceId(8100));
2111                assert_eq!(data, payload);
2112            }
2113            other => panic!("expected Frame, got {:?}", other),
2114        }
2115    }
2116
2117    #[test]
2118    fn backbone_send_to_client() {
2119        let port = find_free_port();
2120        let (tx, rx) = crate::event::channel();
2121        let next_id = Arc::new(AtomicU64::new(8200));
2122
2123        let config = make_server_config(port, 82, None, None, None, BackboneAbuseConfig::default());
2124
2125        start(config, tx, next_id).unwrap();
2126        thread::sleep(Duration::from_millis(50));
2127
2128        let mut client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2129        client
2130            .set_read_timeout(Some(Duration::from_secs(2)))
2131            .unwrap();
2132
2133        // Get writer from InterfaceUp
2134        let event = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();
2135        let mut writer = match event {
2136            Event::InterfaceUp(_, Some(w), _) => w,
2137            other => panic!("expected InterfaceUp with writer, got {:?}", other),
2138        };
2139
2140        // Send frame via writer
2141        let payload: Vec<u8> = (0..24).collect();
2142        writer.send_frame(&payload).unwrap();
2143
2144        // Read from client
2145        let mut buf = [0u8; 256];
2146        let n = client.read(&mut buf).unwrap();
2147        let expected = hdlc::frame(&payload);
2148        assert_eq!(&buf[..n], &expected[..]);
2149    }
2150
2151    #[test]
2152    fn backbone_multiple_clients() {
2153        let port = find_free_port();
2154        let (tx, rx) = crate::event::channel();
2155        let next_id = Arc::new(AtomicU64::new(8300));
2156
2157        let config = make_server_config(port, 83, None, None, None, BackboneAbuseConfig::default());
2158
2159        start(config, tx, next_id).unwrap();
2160        thread::sleep(Duration::from_millis(50));
2161
2162        let _client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2163        let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2164
2165        let mut ids = Vec::new();
2166        for _ in 0..2 {
2167            let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2168            match event {
2169                Event::InterfaceUp(id, _, _) => ids.push(id),
2170                other => panic!("expected InterfaceUp, got {:?}", other),
2171            }
2172        }
2173
2174        assert_eq!(ids.len(), 2);
2175        assert_ne!(ids[0], ids[1]);
2176    }
2177
2178    #[test]
2179    fn backbone_client_disconnect() {
2180        let port = find_free_port();
2181        let (tx, rx) = crate::event::channel();
2182        let next_id = Arc::new(AtomicU64::new(8400));
2183
2184        let config = make_server_config(port, 84, None, None, None, BackboneAbuseConfig::default());
2185
2186        start(config, tx, next_id).unwrap();
2187        thread::sleep(Duration::from_millis(50));
2188
2189        let client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2190
2191        // Drain InterfaceUp
2192        let _ = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();
2193
2194        // Disconnect
2195        drop(client);
2196
2197        // Should receive InterfaceDown
2198        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2199        assert!(
2200            matches!(event, Event::InterfaceDown(InterfaceId(8400))),
2201            "expected InterfaceDown(8400), got {:?}",
2202            event
2203        );
2204    }
2205
2206    #[test]
2207    fn backbone_epoll_multiplexing() {
2208        let port = find_free_port();
2209        let (tx, rx) = crate::event::channel();
2210        let next_id = Arc::new(AtomicU64::new(8500));
2211
2212        let config = make_server_config(port, 85, None, None, None, BackboneAbuseConfig::default());
2213
2214        start(config, tx, next_id).unwrap();
2215        thread::sleep(Duration::from_millis(50));
2216
2217        let mut client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2218        let mut client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2219
2220        // Drain both InterfaceUp events
2221        let _ = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();
2222        let _ = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();
2223
2224        // Both clients send data simultaneously
2225        let payload1: Vec<u8> = (0..24).collect();
2226        let payload2: Vec<u8> = (100..130).collect();
2227        client1.write_all(&hdlc::frame(&payload1)).unwrap();
2228        client2.write_all(&hdlc::frame(&payload2)).unwrap();
2229
2230        // Should receive both Frame events
2231        let mut received = Vec::new();
2232        for _ in 0..2 {
2233            let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2234            match event {
2235                Event::Frame { data, .. } => received.push(data),
2236                other => panic!("expected Frame, got {:?}", other),
2237            }
2238        }
2239        assert!(received.contains(&payload1));
2240        assert!(received.contains(&payload2));
2241    }
2242
2243    #[test]
2244    fn backbone_bind_port() {
2245        let port = find_free_port();
2246        let (tx, _rx) = crate::event::channel();
2247        let next_id = Arc::new(AtomicU64::new(8600));
2248
2249        let config = make_server_config(port, 86, None, None, None, BackboneAbuseConfig::default());
2250
2251        // Should not error
2252        start(config, tx, next_id).unwrap();
2253    }
2254
2255    #[test]
2256    fn backbone_hdlc_fragmented() {
2257        let port = find_free_port();
2258        let (tx, rx) = crate::event::channel();
2259        let next_id = Arc::new(AtomicU64::new(8700));
2260
2261        let config = make_server_config(port, 87, None, None, None, BackboneAbuseConfig::default());
2262
2263        start(config, tx, next_id).unwrap();
2264        thread::sleep(Duration::from_millis(50));
2265
2266        let mut client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2267        client.set_nodelay(true).unwrap();
2268
2269        // Drain InterfaceUp
2270        let _ = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();
2271
2272        // Send HDLC frame in two fragments
2273        let payload: Vec<u8> = (0..32).collect();
2274        let framed = hdlc::frame(&payload);
2275        let mid = framed.len() / 2;
2276
2277        client.write_all(&framed[..mid]).unwrap();
2278        thread::sleep(Duration::from_millis(50));
2279        client.write_all(&framed[mid..]).unwrap();
2280
2281        // Should receive reassembled frame
2282        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2283        match event {
2284            Event::Frame { data, .. } => {
2285                assert_eq!(data, payload);
2286            }
2287            other => panic!("expected Frame, got {:?}", other),
2288        }
2289    }
2290
2291    // -----------------------------------------------------------------------
2292    // Client mode tests
2293    // -----------------------------------------------------------------------
2294
2295    fn make_client_config(port: u16, id: u64) -> BackboneClientConfig {
2296        BackboneClientConfig {
2297            name: format!("test-bb-client-{}", port),
2298            target_host: "127.0.0.1".into(),
2299            target_port: port,
2300            interface_id: InterfaceId(id),
2301            reconnect_wait: Duration::from_millis(100),
2302            max_reconnect_tries: Some(2),
2303            connect_timeout: Duration::from_secs(2),
2304            transport_identity: None,
2305            runtime: Arc::new(Mutex::new(BackboneClientRuntime {
2306                reconnect_wait: Duration::from_millis(100),
2307                max_reconnect_tries: Some(2),
2308                connect_timeout: Duration::from_secs(2),
2309            })),
2310        }
2311    }
2312
2313    #[test]
2314    fn backbone_client_connect() {
2315        let port = find_free_port();
2316        let listener = TcpListener::bind(format!("127.0.0.1:{}", port)).unwrap();
2317        let (tx, rx) = crate::event::channel();
2318
2319        let config = make_client_config(port, 9000);
2320        let _writer = start_client(config, tx).unwrap();
2321
2322        let _server_stream = listener.accept().unwrap();
2323
2324        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
2325        assert!(matches!(event, Event::InterfaceUp(InterfaceId(9000), _, _)));
2326    }
2327
2328    #[test]
2329    fn backbone_client_receive_frame() {
2330        let port = find_free_port();
2331        let listener = TcpListener::bind(format!("127.0.0.1:{}", port)).unwrap();
2332        let (tx, rx) = crate::event::channel();
2333
2334        let config = make_client_config(port, 9100);
2335        let _writer = start_client(config, tx).unwrap();
2336
2337        let (mut server_stream, _) = listener.accept().unwrap();
2338
2339        // Drain InterfaceUp
2340        let _ = rx.recv_timeout(Duration::from_secs(1)).unwrap();
2341
2342        // Client mode applies the same strict lower bound and recovers.
2343        server_stream
2344            .write_all(&hdlc::frame(&vec![
2345                0x11;
2346                rns_core::constants::HEADER_MINSIZE
2347            ]))
2348            .unwrap();
2349        let payload: Vec<u8> = (0..32).collect();
2350        server_stream.write_all(&hdlc::frame(&payload)).unwrap();
2351
2352        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
2353        match event {
2354            Event::Frame {
2355                interface_id,
2356                data,
2357                rssi: _,
2358                snr: _,
2359            } => {
2360                assert_eq!(interface_id, InterfaceId(9100));
2361                assert_eq!(data, payload);
2362            }
2363            other => panic!("expected Frame, got {:?}", other),
2364        }
2365    }
2366
2367    #[test]
2368    fn backbone_client_send_frame() {
2369        let port = find_free_port();
2370        let listener = TcpListener::bind(format!("127.0.0.1:{}", port)).unwrap();
2371        let (tx, _rx) = crate::event::channel();
2372
2373        let config = make_client_config(port, 9200);
2374        let mut writer = start_client(config, tx).unwrap();
2375
2376        let (mut server_stream, _) = listener.accept().unwrap();
2377        server_stream
2378            .set_read_timeout(Some(Duration::from_secs(2)))
2379            .unwrap();
2380
2381        let payload: Vec<u8> = (0..24).collect();
2382        writer.send_frame(&payload).unwrap();
2383
2384        let mut buf = [0u8; 256];
2385        let n = server_stream.read(&mut buf).unwrap();
2386        let expected = hdlc::frame(&payload);
2387        assert_eq!(&buf[..n], &expected[..]);
2388    }
2389
2390    #[test]
2391    fn backbone_max_connections_rejects_excess() {
2392        let port = find_free_port();
2393        let (tx, rx) = crate::event::channel();
2394        let next_id = Arc::new(AtomicU64::new(8800));
2395
2396        let config = make_server_config(
2397            port,
2398            88,
2399            Some(2),
2400            None,
2401            None,
2402            BackboneAbuseConfig::default(),
2403        );
2404
2405        start(config, tx, next_id).unwrap();
2406        thread::sleep(Duration::from_millis(50));
2407
2408        // Connect two clients (at limit)
2409        let _client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2410        let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2411
2412        // Drain both InterfaceUp events
2413        for _ in 0..2 {
2414            let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2415            assert!(matches!(event, Event::InterfaceUp(_, _, _)));
2416        }
2417
2418        // Third connection should be accepted at TCP level but immediately dropped
2419        let client3 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2420        client3
2421            .set_read_timeout(Some(Duration::from_millis(500)))
2422            .unwrap();
2423
2424        // Give server time to reject
2425        thread::sleep(Duration::from_millis(100));
2426
2427        // Should NOT receive a third InterfaceUp
2428        let result = recv_non_peer_event(&rx, Duration::from_millis(500));
2429        assert!(
2430            result.is_err(),
2431            "expected no InterfaceUp for rejected connection, got {:?}",
2432            result
2433        );
2434    }
2435
2436    #[test]
2437    fn backbone_max_connections_allows_after_disconnect() {
2438        let port = find_free_port();
2439        let (tx, rx) = crate::event::channel();
2440        let next_id = Arc::new(AtomicU64::new(8900));
2441
2442        let config = make_server_config(
2443            port,
2444            89,
2445            Some(1),
2446            None,
2447            None,
2448            BackboneAbuseConfig::default(),
2449        );
2450
2451        start(config, tx, next_id).unwrap();
2452        thread::sleep(Duration::from_millis(50));
2453
2454        // Connect first client
2455        let client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2456        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2457        assert!(matches!(event, Event::InterfaceUp(_, _, _)));
2458
2459        // Disconnect first client
2460        drop(client1);
2461
2462        // Wait for InterfaceDown
2463        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2464        assert!(matches!(event, Event::InterfaceDown(_)));
2465
2466        // Now a new connection should be accepted
2467        let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2468        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2469        assert!(
2470            matches!(event, Event::InterfaceUp(_, _, _)),
2471            "expected InterfaceUp after slot freed, got {:?}",
2472            event
2473        );
2474    }
2475
2476    #[test]
2477    fn backbone_client_reconnect() {
2478        let port = find_free_port();
2479        let listener = TcpListener::bind(format!("127.0.0.1:{}", port)).unwrap();
2480        listener.set_nonblocking(false).unwrap();
2481        let (tx, rx) = crate::event::channel();
2482
2483        let config = make_client_config(port, 9300);
2484        let _writer = start_client(config, tx).unwrap();
2485
2486        // Accept first connection and immediately close it
2487        let (server_stream, _) = listener.accept().unwrap();
2488
2489        // Drain InterfaceUp
2490        let _ = rx.recv_timeout(Duration::from_secs(1)).unwrap();
2491
2492        drop(server_stream);
2493
2494        // Should get InterfaceDown
2495        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2496        assert!(matches!(event, Event::InterfaceDown(InterfaceId(9300))));
2497
2498        // Accept the reconnection
2499        let _server_stream2 = listener.accept().unwrap();
2500
2501        // Should get InterfaceUp again
2502        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2503        assert!(matches!(event, Event::InterfaceUp(InterfaceId(9300), _, _)));
2504    }
2505
2506    #[test]
2507    fn backbone_idle_timeout_disconnects_silent_client() {
2508        let port = find_free_port();
2509        let (tx, rx) = crate::event::channel();
2510        let next_id = Arc::new(AtomicU64::new(9400));
2511
2512        let config = make_server_config(
2513            port,
2514            94,
2515            None,
2516            Some(Duration::from_millis(150)),
2517            None,
2518            BackboneAbuseConfig::default(),
2519        );
2520
2521        start(config, tx, next_id).unwrap();
2522        thread::sleep(Duration::from_millis(50));
2523
2524        let _client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2525
2526        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2527        let client_id = match event {
2528            Event::InterfaceUp(id, _, _) => id,
2529            other => panic!("expected InterfaceUp, got {:?}", other),
2530        };
2531
2532        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2533        assert!(matches!(event, Event::InterfaceDown(id) if id == client_id));
2534    }
2535
2536    #[test]
2537    fn backbone_idle_timeout_ignores_client_after_data() {
2538        let port = find_free_port();
2539        let (tx, rx) = crate::event::channel();
2540        let next_id = Arc::new(AtomicU64::new(9500));
2541
2542        let config = make_server_config(
2543            port,
2544            95,
2545            None,
2546            Some(Duration::from_millis(200)),
2547            None,
2548            BackboneAbuseConfig::default(),
2549        );
2550
2551        start(config, tx, next_id).unwrap();
2552        thread::sleep(Duration::from_millis(50));
2553
2554        let mut client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2555
2556        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2557        let client_id = match event {
2558            Event::InterfaceUp(id, _, _) => id,
2559            other => panic!("expected InterfaceUp, got {:?}", other),
2560        };
2561
2562        client.write_all(&hdlc::frame(&[1u8; 24])).unwrap();
2563
2564        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2565        match event {
2566            Event::Frame {
2567                interface_id,
2568                data,
2569                rssi: _,
2570                snr: _,
2571            } => {
2572                assert_eq!(interface_id, client_id);
2573                assert_eq!(data, vec![1u8; 24]);
2574            }
2575            other => panic!("expected Frame, got {:?}", other),
2576        }
2577
2578        let result = recv_non_peer_event(&rx, Duration::from_millis(500));
2579        assert!(
2580            result.is_err(),
2581            "expected no InterfaceDown after client sent data, got {:?}",
2582            result
2583        );
2584    }
2585
2586    #[test]
2587    fn backbone_runtime_idle_timeout_updates_live() {
2588        let port = find_free_port();
2589        let (tx, rx) = crate::event::channel();
2590        let next_id = Arc::new(AtomicU64::new(9650));
2591
2592        let config = make_server_config(port, 97, None, None, None, BackboneAbuseConfig::default());
2593        let runtime = Arc::clone(&config.runtime);
2594
2595        start(config, tx, next_id).unwrap();
2596        thread::sleep(Duration::from_millis(50));
2597
2598        let _client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2599        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2600        let client_id = match event {
2601            Event::InterfaceUp(id, _, _) => id,
2602            other => panic!("expected InterfaceUp, got {:?}", other),
2603        };
2604
2605        {
2606            let mut runtime = runtime.lock().unwrap();
2607            runtime.idle_timeout = Some(Duration::from_millis(150));
2608        }
2609
2610        let event = recv_non_peer_event(&rx, Duration::from_secs(4)).unwrap();
2611        assert!(matches!(event, Event::InterfaceDown(id) if id == client_id));
2612    }
2613
2614    #[test]
2615    fn backbone_write_stall_timeout_disconnects_unwritable_client() {
2616        let port = find_free_port();
2617        let (tx, rx) = crate::event::channel();
2618        let next_id = Arc::new(AtomicU64::new(9660));
2619
2620        let config = make_server_config(
2621            port,
2622            98,
2623            None,
2624            None,
2625            Some(Duration::from_millis(50)),
2626            BackboneAbuseConfig::default(),
2627        );
2628
2629        start(config, tx, next_id).unwrap();
2630        thread::sleep(Duration::from_millis(50));
2631
2632        let client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2633        client
2634            .set_read_timeout(Some(Duration::from_millis(100)))
2635            .unwrap();
2636        let sock = SockRef::from(&client);
2637        sock.set_recv_buffer_size(4096).ok();
2638
2639        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2640        let (client_id, mut writer) = match event {
2641            Event::InterfaceUp(id, Some(writer), _) => (id, writer),
2642            other => panic!("expected InterfaceUp with writer, got {:?}", other),
2643        };
2644
2645        let payload = vec![0x55; 512 * 1024];
2646        let deadline = Instant::now() + Duration::from_secs(3);
2647        let mut stalled = false;
2648        while Instant::now() < deadline {
2649            match writer.send_frame(&payload) {
2650                Ok(()) => {}
2651                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
2652                    thread::sleep(Duration::from_millis(10));
2653                }
2654                Err(ref e) if e.kind() == io::ErrorKind::TimedOut => {
2655                    stalled = true;
2656                    break;
2657                }
2658                Err(e) => panic!("unexpected send error: {}", e),
2659            }
2660        }
2661
2662        assert!(stalled, "expected writer to time out on persistent stall");
2663        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2664        assert!(matches!(event, Event::InterfaceDown(id) if id == client_id));
2665    }
2666
2667    #[test]
2668    fn shutting_down_backbone_writer_closes_the_client_socket() {
2669        let port = find_free_port();
2670        let (tx, rx) = crate::event::channel();
2671        let next_id = Arc::new(AtomicU64::new(9670));
2672        let config = make_server_config(port, 99, None, None, None, BackboneAbuseConfig::default());
2673
2674        start(config, tx, next_id).unwrap();
2675        thread::sleep(Duration::from_millis(50));
2676
2677        let mut client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2678        client
2679            .set_read_timeout(Some(Duration::from_secs(2)))
2680            .unwrap();
2681        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2682        let mut writer = match event {
2683            Event::InterfaceUp(_, Some(writer), _) => writer,
2684            other => panic!("expected InterfaceUp with writer, got {:?}", other),
2685        };
2686
2687        writer.shutdown();
2688
2689        let mut byte = [0u8; 1];
2690        assert_eq!(client.read(&mut byte).unwrap(), 0);
2691    }
2692
2693    /// Drain events matching a predicate, return the first match.
2694    fn wait_for<F>(rx: &mpsc::Receiver<Event>, timeout: Duration, mut pred: F) -> Option<Event>
2695    where
2696        F: FnMut(&Event) -> bool,
2697    {
2698        let deadline = Instant::now() + timeout;
2699        loop {
2700            let remaining = deadline.saturating_duration_since(Instant::now());
2701            if remaining.is_zero() {
2702                return None;
2703            }
2704            match rx.recv_timeout(remaining) {
2705                Ok(event) if pred(&event) => return Some(event),
2706                Ok(_) => continue,
2707                Err(_) => return None,
2708            }
2709        }
2710    }
2711
2712    #[test]
2713    fn backbone_write_stall_emits_peer_events() {
2714        let port = find_free_port();
2715        let (tx, rx) = crate::event::channel();
2716        let next_id = Arc::new(AtomicU64::new(9700));
2717
2718        let config = make_server_config(
2719            port,
2720            97,
2721            None,
2722            None,
2723            Some(Duration::from_millis(50)), // 50ms stall timeout
2724            BackboneAbuseConfig::default(),
2725        );
2726
2727        start(config, tx, next_id).unwrap();
2728        thread::sleep(Duration::from_millis(50));
2729
2730        // Connect a client that won't read (will cause write stall)
2731        let client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2732        client
2733            .set_read_timeout(Some(Duration::from_millis(100)))
2734            .unwrap();
2735        let sock = SockRef::from(&client);
2736        sock.set_recv_buffer_size(4096).ok();
2737
2738        // Wait for InterfaceUp and grab writer
2739        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2740        let mut writer = match event {
2741            Event::InterfaceUp(_, Some(w), _) => w,
2742            other => panic!("expected InterfaceUp with writer, got {:?}", other),
2743        };
2744
2745        // Flood until stall
2746        let payload = vec![0x55; 512 * 1024];
2747        let deadline = Instant::now() + Duration::from_secs(3);
2748        while Instant::now() < deadline {
2749            match writer.send_frame(&payload) {
2750                Ok(()) | Err(_) => {
2751                    if Instant::now() + Duration::from_millis(10) > deadline {
2752                        break;
2753                    }
2754                    thread::sleep(Duration::from_millis(10));
2755                }
2756            }
2757        }
2758
2759        // Should see BackbonePeerWriteStall event
2760        let stall_event = wait_for(&rx, Duration::from_secs(3), |e| {
2761            matches!(e, Event::BackbonePeerWriteStall { .. })
2762        });
2763        assert!(
2764            stall_event.is_some(),
2765            "expected BackbonePeerWriteStall event"
2766        );
2767    }
2768
2769    #[test]
2770    fn backbone_blacklisted_peer_rejected_on_connect() {
2771        let port = find_free_port();
2772        let (tx, rx) = crate::event::channel();
2773        let next_id = Arc::new(AtomicU64::new(9800));
2774
2775        let config = make_server_config(port, 98, None, None, None, BackboneAbuseConfig::default());
2776        let peer_state = config.peer_state.clone();
2777
2778        start(config, tx.clone(), next_id).unwrap();
2779        thread::sleep(Duration::from_millis(50));
2780
2781        // First connection should succeed
2782        let client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2783        let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2784        assert!(
2785            matches!(event, Event::InterfaceUp(_, _, _)),
2786            "first connection should succeed"
2787        );
2788        drop(client1);
2789
2790        // Drain disconnect events
2791        thread::sleep(Duration::from_millis(100));
2792        while rx.try_recv().is_ok() {}
2793
2794        // Blacklist 127.0.0.1 via the peer monitor
2795        peer_state.lock().unwrap().blacklist(
2796            "127.0.0.1".parse().unwrap(),
2797            Duration::from_secs(60),
2798            "test blacklist".into(),
2799        );
2800
2801        // Second connection from same IP should be rejected (no InterfaceUp)
2802        let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2803        // Give poll loop time to reject
2804        thread::sleep(Duration::from_millis(200));
2805
2806        // Should NOT get an InterfaceUp — connection should have been rejected
2807        let event = rx.try_recv();
2808        match event {
2809            Ok(Event::InterfaceUp(_, _, _)) => {
2810                panic!("blacklisted peer should not get InterfaceUp")
2811            }
2812            _ => {} // Expected: no InterfaceUp
2813        }
2814    }
2815
2816    #[test]
2817    fn backbone_six_fast_flaps_reject_before_allocating_seventh_interface() {
2818        let port = find_free_port();
2819        let (tx, rx) = crate::event::channel();
2820        let next_id = Arc::new(AtomicU64::new(9900));
2821        let mut config =
2822            make_server_config(port, 99, None, None, None, BackboneAbuseConfig::default());
2823        config.fast_flap = test_fast_flap_config();
2824        let fast_flap_state = Arc::clone(&config.fast_flap_state);
2825        let peer_state = Arc::clone(&config.peer_state);
2826
2827        start(config, tx, Arc::clone(&next_id)).unwrap();
2828        thread::sleep(Duration::from_millis(50));
2829
2830        for flap in 0..6 {
2831            let mut client = TcpStream::connect(format!("127.0.0.1:{port}")).unwrap();
2832            let up = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2833            assert!(matches!(up, Event::InterfaceUp(_, _, _)));
2834
2835            // Receiving traffic must not exempt a short-lived connection.
2836            if flap == 0 {
2837                let payload = vec![0x41; 32];
2838                client.write_all(&hdlc::frame(&payload)).unwrap();
2839                let frame = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2840                assert!(matches!(frame, Event::Frame { data, .. } if data == payload));
2841            }
2842
2843            drop(client);
2844            let down = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2845            assert!(matches!(down, Event::InterfaceDown(_)));
2846        }
2847
2848        let peer_ip: IpAddr = "127.0.0.1".parse().unwrap();
2849        assert!(
2850            lock_or_recover(&fast_flap_state, "test fast-flap state").is_blocked(
2851                peer_ip,
2852                &test_fast_flap_config(),
2853                Instant::now(),
2854            )
2855        );
2856        assert_eq!(next_id.load(Ordering::Relaxed), 9906);
2857        let entries = lock_or_recover(&peer_state, "test peer state").list("test-backbone");
2858        let entry = entries
2859            .iter()
2860            .find(|entry| entry.peer_ip == peer_ip)
2861            .expect("fast-flapping peer state");
2862        assert_eq!(entry.blacklist_reason.as_deref(), Some("fast flapping"));
2863        assert!(entry.blacklisted_remaining_secs.is_some());
2864
2865        let _rejected = TcpStream::connect(format!("127.0.0.1:{port}")).unwrap();
2866        thread::sleep(Duration::from_millis(200));
2867        assert_eq!(
2868            next_id.load(Ordering::Relaxed),
2869            9906,
2870            "rejection must happen before dynamic interface allocation"
2871        );
2872        while let Ok(event) = rx.try_recv() {
2873            assert!(
2874                !matches!(event, Event::InterfaceUp(_, _, _)),
2875                "blocked peer must not register a dynamic interface"
2876            );
2877        }
2878        let entries = lock_or_recover(&peer_state, "test peer state").list("test-backbone");
2879        assert_eq!(
2880            entries
2881                .iter()
2882                .find(|entry| entry.peer_ip == peer_ip)
2883                .unwrap()
2884                .reject_count,
2885            1
2886        );
2887    }
2888
2889    #[test]
2890    fn backbone_parse_config_reads_abuse_settings() {
2891        let factory = BackboneInterfaceFactory;
2892        let mut params = HashMap::new();
2893        params.insert("listen_ip".into(), "127.0.0.1".into());
2894        params.insert("listen_port".into(), "4242".into());
2895        params.insert("idle_timeout".into(), "15".into());
2896        params.insert("write_stall_timeout".into(), "45".into());
2897        params.insert("max_penalty_duration".into(), "3600".into());
2898
2899        let config = factory
2900            .parse_config("test-backbone", InterfaceId(97), &params)
2901            .unwrap();
2902        let mode = *config.into_any().downcast::<BackboneMode>().unwrap();
2903
2904        match mode {
2905            BackboneMode::Server(config) => {
2906                assert_eq!(config.listen_ip, "127.0.0.1");
2907                assert_eq!(config.listen_port, 4242);
2908                assert_eq!(config.idle_timeout, Some(Duration::from_secs(15)));
2909                assert_eq!(config.write_stall_timeout, Some(Duration::from_secs(45)));
2910                assert_eq!(
2911                    config.abuse.max_penalty_duration,
2912                    Some(Duration::from_secs(3600))
2913                );
2914            }
2915            BackboneMode::Client(_, _) => panic!("expected server config"),
2916        }
2917    }
2918
2919    #[test]
2920    fn backbone_parse_config_reads_fast_flapping_settings() {
2921        let factory = BackboneInterfaceFactory;
2922        let mut params = HashMap::new();
2923        params.insert("listen_ip".into(), "127.0.0.1".into());
2924        params.insert("listen_port".into(), "4242".into());
2925        params.insert("block_fast_flapping".into(), "no".into());
2926        params.insert("fast_flapping_threshold".into(), "2.5".into());
2927        params.insert("fast_flapping_grace".into(), "9".into());
2928        params.insert("fast_flapping_block_time".into(), "1.5".into());
2929
2930        let config = factory
2931            .parse_config("test-backbone", InterfaceId(93), &params)
2932            .unwrap();
2933        let mode = *config.into_any().downcast::<BackboneMode>().unwrap();
2934        let BackboneMode::Server(config) = mode else {
2935            panic!("expected server config");
2936        };
2937
2938        assert!(!config.fast_flap.enabled);
2939        assert_eq!(
2940            config.fast_flap.connection_threshold,
2941            Duration::from_secs_f64(2.5)
2942        );
2943        assert_eq!(config.fast_flap.grace, 9);
2944        assert_eq!(config.fast_flap.block_duration, Duration::from_secs(90));
2945    }
2946
2947    #[test]
2948    fn backbone_parse_config_rejects_invalid_fast_flapping_settings() {
2949        let factory = BackboneInterfaceFactory;
2950        for (key, value) in [
2951            ("block_fast_flapping", "perhaps"),
2952            ("fast_flapping_threshold", "-1"),
2953            ("fast_flapping_threshold", "NaN"),
2954            ("fast_flapping_grace", "-1"),
2955            ("fast_flapping_block_time", "infinite"),
2956        ] {
2957            let mut params = HashMap::new();
2958            params.insert("listen_port".into(), "4242".into());
2959            params.insert(key.into(), value.into());
2960
2961            let err = match factory.parse_config("test-backbone", InterfaceId(92), &params) {
2962                Ok(_) => panic!("{key}={value} should be rejected"),
2963                Err(err) => err,
2964            };
2965            assert!(err.contains(key), "unexpected error: {err}");
2966        }
2967    }
2968
2969    #[test]
2970    fn parsed_backbone_listeners_share_fast_flap_history() {
2971        let factory = BackboneInterfaceFactory;
2972        let mut params = HashMap::new();
2973        params.insert("listen_port".into(), "4242".into());
2974
2975        let first = factory
2976            .parse_config("first", InterfaceId(90), &params)
2977            .unwrap();
2978        let second = factory
2979            .parse_config("second", InterfaceId(91), &params)
2980            .unwrap();
2981        let BackboneMode::Server(first) = *first.into_any().downcast::<BackboneMode>().unwrap()
2982        else {
2983            panic!("expected server config");
2984        };
2985        let BackboneMode::Server(second) = *second.into_any().downcast::<BackboneMode>().unwrap()
2986        else {
2987            panic!("expected server config");
2988        };
2989
2990        assert!(Arc::ptr_eq(&first.fast_flap_state, &second.fast_flap_state));
2991    }
2992
2993    #[test]
2994    fn backbone_parse_config_reads_client_priority() {
2995        let factory = BackboneInterfaceFactory;
2996        let mut params = HashMap::new();
2997        params.insert("remote".into(), "example.com".into());
2998        params.insert("target_port".into(), "4242".into());
2999        params.insert("priority".into(), "87".into());
3000
3001        let config = factory
3002            .parse_config("test-backbone-client", InterfaceId(96), &params)
3003            .unwrap();
3004        let mode = *config.into_any().downcast::<BackboneMode>().unwrap();
3005
3006        match mode {
3007            BackboneMode::Client(_, priority) => assert_eq!(priority, 87),
3008            BackboneMode::Server(_) => panic!("expected client config"),
3009        }
3010    }
3011
3012    #[test]
3013    fn backbone_parse_config_defaults_client_priority() {
3014        let factory = BackboneInterfaceFactory;
3015        let mut params = HashMap::new();
3016        params.insert("remote".into(), "example.com".into());
3017
3018        let config = factory
3019            .parse_config("test-backbone-client", InterfaceId(95), &params)
3020            .unwrap();
3021        let mode = *config.into_any().downcast::<BackboneMode>().unwrap();
3022
3023        match mode {
3024            BackboneMode::Client(_, priority) => assert_eq!(priority, 60),
3025            BackboneMode::Server(_) => panic!("expected client config"),
3026        }
3027    }
3028
3029    #[test]
3030    fn backbone_parse_config_rejects_invalid_client_priority() {
3031        let factory = BackboneInterfaceFactory;
3032        for value in ["-1", "101", "fast"] {
3033            let mut params = HashMap::new();
3034            params.insert("remote".into(), "example.com".into());
3035            params.insert("priority".into(), value.into());
3036
3037            let err = match factory.parse_config("test-backbone-client", InterfaceId(94), &params) {
3038                Ok(_) => panic!("priority {value} should be rejected"),
3039                Err(err) => err,
3040            };
3041            assert!(
3042                err.contains("priority"),
3043                "unexpected error for {value}: {err}"
3044            );
3045        }
3046    }
3047}