1use 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
35const HW_MTU: usize = 1_048_576;
37
38#[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#[derive(Debug, Clone, Default)]
63pub struct BackboneAbuseConfig {
64 pub max_penalty_duration: Option<Duration>,
65}
66
67#[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#[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#[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#[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
309const MAX_PENDING_BYTES: usize = 512 * 1024;
312
313struct 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
426pub 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
495struct 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
629fn 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 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 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 #[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 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; }
751
752 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; }
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), 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 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 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 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 }
1023 }
1024
1025 let _ = poller.delete(&client.stream);
1026 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 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#[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
1185struct 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
1196fn 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 #[cfg(target_os = "macos")]
1211 {
1212 let sock = SockRef::from(&stream);
1213 sock.set_nosigpipe(true).ok();
1214 }
1215
1216 Ok(stream)
1217}
1218
1219pub 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 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
1255fn 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
1326const MAX_BACKOFF_SHIFT: u32 = 6;
1329
1330fn 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 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#[derive(Clone)]
1412pub(crate) enum BackboneMode {
1413 Server(BackboneConfig),
1414 Client(BackboneClientConfig, u8),
1415}
1416
1417pub struct BackboneInterfaceFactory;
1423
1424pub 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 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 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 let _ = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();
2090
2091 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 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 let payload: Vec<u8> = (0..24).collect();
2142 writer.send_frame(&payload).unwrap();
2143
2144 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 let _ = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();
2193
2194 drop(client);
2196
2197 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 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 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 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 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 let _ = recv_non_peer_event(&rx, Duration::from_secs(1)).unwrap();
2271
2272 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 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 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 let _ = rx.recv_timeout(Duration::from_secs(1)).unwrap();
2341
2342 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 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 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 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 thread::sleep(Duration::from_millis(100));
2426
2427 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 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 drop(client1);
2461
2462 let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2464 assert!(matches!(event, Event::InterfaceDown(_)));
2465
2466 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 let (server_stream, _) = listener.accept().unwrap();
2488
2489 let _ = rx.recv_timeout(Duration::from_secs(1)).unwrap();
2491
2492 drop(server_stream);
2493
2494 let event = recv_non_peer_event(&rx, Duration::from_secs(2)).unwrap();
2496 assert!(matches!(event, Event::InterfaceDown(InterfaceId(9300))));
2497
2498 let _server_stream2 = listener.accept().unwrap();
2500
2501 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 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)), BackboneAbuseConfig::default(),
2725 );
2726
2727 start(config, tx, next_id).unwrap();
2728 thread::sleep(Duration::from_millis(50));
2729
2730 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 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 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 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 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 thread::sleep(Duration::from_millis(100));
2792 while rx.try_recv().is_ok() {}
2793
2794 peer_state.lock().unwrap().blacklist(
2796 "127.0.0.1".parse().unwrap(),
2797 Duration::from_secs(60),
2798 "test blacklist".into(),
2799 );
2800
2801 let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
2803 thread::sleep(Duration::from_millis(200));
2805
2806 let event = rx.try_recv();
2808 match event {
2809 Ok(Event::InterfaceUp(_, _, _)) => {
2810 panic!("blacklisted peer should not get InterfaceUp")
2811 }
2812 _ => {} }
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 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), ¶ms)
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), ¶ms)
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), ¶ms) {
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), ¶ms)
2977 .unwrap();
2978 let second = factory
2979 .parse_config("second", InterfaceId(91), ¶ms)
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), ¶ms)
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), ¶ms)
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), ¶ms) {
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}