rns_net/interface/
tcp_server.rs

1//! TCP server interface with HDLC framing.
2//!
3//! Accepts client connections and spawns per-client reader threads.
4//! Each client gets a dynamically allocated InterfaceId.
5//! Matches Python `TCPServerInterface` from `TCPInterface.py`.
6
7use std::io::{self, Write};
8use std::net::{TcpListener, TcpStream};
9use std::os::unix::io::AsRawFd;
10use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
11use std::sync::{Arc, Mutex};
12
13use rns_core::constants;
14use rns_core::transport::types::{IngressControlConfig, InterfaceId, InterfaceInfo};
15
16use crate::event::{Event, EventSender, LoopContext, LoopSource, SourceStatus};
17use crate::hdlc;
18use crate::interface::tcp::{read_stream_frames, StreamRead};
19use crate::interface::{lock_or_recover, ListenerControl, Writer};
20
21/// Upstream `TCPInterface.HW_MTU` inherited by spawned server clients.
22const HW_MTU: usize = 262_144;
23
24/// Configuration for a TCP server interface.
25#[derive(Debug, Clone)]
26pub struct TcpServerConfig {
27    pub name: String,
28    pub listen_ip: String,
29    pub listen_port: u16,
30    pub interface_id: InterfaceId,
31    pub max_connections: Option<usize>,
32    pub ingress_control: IngressControlConfig,
33    pub runtime: Arc<Mutex<TcpServerRuntime>>,
34}
35
36#[derive(Debug, Clone)]
37pub struct TcpServerRuntime {
38    pub max_connections: Option<usize>,
39}
40
41impl TcpServerRuntime {
42    pub fn from_config(config: &TcpServerConfig) -> Self {
43        Self {
44            max_connections: config.max_connections,
45        }
46    }
47}
48
49#[derive(Debug, Clone)]
50pub struct TcpServerRuntimeConfigHandle {
51    pub interface_name: String,
52    pub runtime: Arc<Mutex<TcpServerRuntime>>,
53    pub startup: TcpServerRuntime,
54}
55
56impl Default for TcpServerConfig {
57    fn default() -> Self {
58        let mut config = TcpServerConfig {
59            name: String::new(),
60            listen_ip: "0.0.0.0".into(),
61            listen_port: 4242,
62            interface_id: InterfaceId(0),
63            max_connections: None,
64            ingress_control: IngressControlConfig::enabled(),
65            runtime: Arc::new(Mutex::new(TcpServerRuntime {
66                max_connections: None,
67            })),
68        };
69        let startup = TcpServerRuntime::from_config(&config);
70        config.runtime = Arc::new(Mutex::new(startup));
71        config
72    }
73}
74
75/// Writer that sends HDLC-framed data over a TCP stream.
76struct TcpServerWriter {
77    stream: TcpStream,
78    /// Reused encoding buffer for coalesced writes.
79    scratch: Vec<u8>,
80}
81
82impl Writer for TcpServerWriter {
83    fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
84        self.stream.write_all(&hdlc::frame(data))
85    }
86
87    fn send_frames(&mut self, frames: &[crate::interface::FrameBytes]) -> io::Result<()> {
88        hdlc::write_frames(&mut self.stream, frames, &mut self.scratch)
89    }
90
91    fn loop_socket(&self) -> Option<std::os::fd::RawFd> {
92        Some(self.stream.as_raw_fd())
93    }
94}
95
96/// Start a TCP server. Spawns a listener thread that accepts connections
97/// and per-client reader threads. Returns immediately.
98///
99/// `next_id` is shared with the node for allocating unique InterfaceIds
100/// for each connected client.
101pub fn start(
102    config: TcpServerConfig,
103    tx: EventSender,
104    next_id: Arc<AtomicU64>,
105) -> io::Result<ListenerControl> {
106    start_with_template(config, tx, next_id, None, None)
107}
108
109fn start_with_template(
110    config: TcpServerConfig,
111    tx: EventSender,
112    next_id: Arc<AtomicU64>,
113    dynamic_template: Option<super::DynamicInterfaceTemplate>,
114    underlay_mark: Option<u32>,
115) -> io::Result<ListenerControl> {
116    let addr = format!("{}:{}", config.listen_ip, config.listen_port);
117    let listener = TcpListener::bind(&addr)?;
118    super::apply_underlay_mark(listener.as_raw_fd(), underlay_mark)?;
119    listener.set_nonblocking(true)?;
120
121    log::info!("[{}] TCP server listening on {}", config.name, addr);
122
123    let name = config.name.clone();
124    let ifac_size = dynamic_template
125        .as_ref()
126        .and_then(|template| template.ifac.as_ref())
127        .map(|ifac| ifac.size)
128        .unwrap_or(0);
129    let runtime = Arc::clone(&config.runtime);
130    let ingress_control = config.ingress_control;
131    let active_connections = Arc::new(AtomicUsize::new(0));
132    let control = ListenerControl::new();
133    let listener_key = crate::interface::loop_writer::read_key(config.interface_id);
134    tx.register_source(
135        listener_key,
136        Box::new(ListenerSource {
137            listener,
138            name,
139            tx: tx.clone(),
140            next_id,
141            runtime,
142            ingress_control,
143            active_connections,
144            control: control.clone(),
145            dynamic_template,
146            ifac_size,
147            underlay_mark,
148        }),
149    )?;
150
151    Ok(control)
152}
153
154/// Connections accepted per readiness before other sources get a turn.
155const ACCEPT_BURST: usize = 64;
156
157/// The listening socket, served on the driver thread: accepts clients,
158/// registers them with the driver and adds a source per client.
159struct ListenerSource {
160    listener: TcpListener,
161    name: String,
162    tx: EventSender,
163    next_id: Arc<AtomicU64>,
164    runtime: Arc<Mutex<TcpServerRuntime>>,
165    ingress_control: IngressControlConfig,
166    active_connections: Arc<AtomicUsize>,
167    control: ListenerControl,
168    dynamic_template: Option<super::DynamicInterfaceTemplate>,
169    ifac_size: usize,
170    underlay_mark: Option<u32>,
171}
172
173impl LoopSource for ListenerSource {
174    fn fd(&self) -> std::os::fd::RawFd {
175        self.listener.as_raw_fd()
176    }
177
178    fn stopped(&self) -> bool {
179        self.control.should_stop()
180    }
181
182    fn on_removed(&mut self) {
183        log::info!("[{}] listener stopping", self.name);
184    }
185
186    fn on_ready(&mut self, cx: &mut LoopContext<'_>) -> SourceStatus {
187        if self.control.should_stop() {
188            return SourceStatus::Remove;
189        }
190        for _ in 0..ACCEPT_BURST {
191            let stream = match self.listener.accept() {
192                Ok((stream, _)) => stream,
193                Err(e) => match accept_error_step(&e) {
194                    AcceptStep::Drained => break,
195                    AcceptStep::Next => {
196                        log::debug!("[{}] accept failed: {}", self.name, e);
197                        continue;
198                    }
199                    AcceptStep::Pause => {
200                        // The pending connection stays queued, so the
201                        // listener would be ready again at once.
202                        log::warn!(
203                            "[{}] accept failed: {}; accepting resumes on the next tick",
204                            self.name,
205                            e
206                        );
207                        return SourceStatus::Pause;
208                    }
209                },
210            };
211            match self.admit(stream) {
212                Admitted::Client(id, source) => {
213                    cx.add_source(crate::interface::loop_writer::read_key(id), source)
214                }
215                Admitted::Rejected => {}
216                Admitted::DriverGone => return SourceStatus::Remove,
217            }
218        }
219        SourceStatus::Rearm
220    }
221}
222
223/// What the accept loop does after `accept` returns an error.
224#[derive(Debug, Clone, Copy, PartialEq, Eq)]
225pub(crate) enum AcceptStep {
226    /// No connection is pending: wait for readiness.
227    Drained,
228    /// The error concerned one pending connection only: try the next one.
229    Next,
230    /// The error is likely to persist (e.g. out of descriptors or memory):
231    /// stop watching the listener until the next tick.
232    Pause,
233}
234
235pub(crate) fn accept_error_step(error: &io::Error) -> AcceptStep {
236    match error.kind() {
237        io::ErrorKind::WouldBlock => AcceptStep::Drained,
238        io::ErrorKind::Interrupted
239        | io::ErrorKind::ConnectionAborted
240        | io::ErrorKind::ConnectionReset
241        | io::ErrorKind::NetworkDown
242        | io::ErrorKind::NetworkUnreachable
243        | io::ErrorKind::HostUnreachable => AcceptStep::Next,
244        // Linux reports some network errors of a pending connection
245        // through accept; they concern that connection only.
246        _ if error.raw_os_error() == Some(libc::EPROTO) => AcceptStep::Next,
247        _ => AcceptStep::Pause,
248    }
249}
250
251enum Admitted {
252    Client(InterfaceId, Box<dyn LoopSource>),
253    Rejected,
254    DriverGone,
255}
256
257impl ListenerSource {
258    fn admit(&self, stream: TcpStream) -> Admitted {
259        let name = &self.name;
260        if let Err(error) = super::apply_underlay_mark(stream.as_raw_fd(), self.underlay_mark) {
261            log::error!(
262                "[{}] failed to mark accepted underlay socket: {}",
263                name,
264                error
265            );
266            return Admitted::Rejected;
267        }
268
269        // Keepalive and user timeout notice a peer that went silent or
270        // stopped reading; both descriptors of the client share them.
271        if let Err(error) = super::tcp::set_socket_options(&stream) {
272            log::warn!(
273                "[{}] failed to set socket options for {:?}: {}",
274                name,
275                stream.peer_addr().ok(),
276                error
277            );
278        }
279
280        let max_connections = lock_or_recover(&self.runtime, "tcp server runtime").max_connections;
281        if let Some(max) = max_connections {
282            if self.active_connections.load(Ordering::Relaxed) >= max {
283                let peer = stream.peer_addr().ok();
284                log::warn!(
285                    "[{}] max connections ({}) reached, rejecting {:?}",
286                    name,
287                    max,
288                    peer
289                );
290                return Admitted::Rejected;
291            }
292        }
293
294        let client_id = InterfaceId(self.next_id.fetch_add(1, Ordering::Relaxed));
295        let peer_addr = stream.peer_addr().ok();
296
297        log::info!(
298            "[{}] client connected: {:?} → id {}",
299            name,
300            peer_addr,
301            client_id.0
302        );
303
304        // Clone stream for writer
305        let writer_stream = match stream.try_clone() {
306            Ok(s) => s,
307            Err(e) => {
308                log::warn!("[{}] failed to clone stream: {}", name, e);
309                return Admitted::Rejected;
310            }
311        };
312
313        let writer: Box<dyn Writer> = Box::new(TcpServerWriter {
314            stream: writer_stream,
315            scratch: Vec::new(),
316        });
317
318        let info = InterfaceInfo {
319            id: client_id,
320            name: format!("TCPServerInterface/Client-{}", client_id.0),
321            mode: constants::MODE_FULL,
322            gravity: 0,
323            recursive_prs: false,
324            announces_from_internal: true,
325            announces_to_internal: None,
326            out_capable: true,
327            in_capable: true,
328            bitrate: None,
329            airtime_profile: None,
330            announce_rate_target: None,
331            announce_rate_grace: 0,
332            announce_rate_penalty: 0.0,
333            announce_cap: constants::ANNOUNCE_CAP,
334            is_local_client: false,
335            wants_tunnel: false,
336            tunnel_id: None,
337            mtu: 65535,
338            ia_freq: 0.0,
339            ip_freq: 0.0,
340            op_freq: 0.0,
341            op_samples: 0,
342            started: 0.0,
343            ingress_control: self.ingress_control,
344        };
345
346        // Register the client with the driver before any of its frames: the
347        // queue never delivers a frame ahead of an older control event.
348        let registration = if let Some(template) = &self.dynamic_template {
349            Event::DynamicInterfaceUp {
350                id: client_id,
351                writer,
352                registration: Box::new(template.registration(info)),
353            }
354        } else {
355            Event::InterfaceUp(client_id, Some(writer), Some(Box::new(info)))
356        };
357        if self.tx.push_from_loop([registration]).is_err() {
358            return Admitted::DriverGone;
359        }
360
361        self.active_connections.fetch_add(1, Ordering::Relaxed);
362        Admitted::Client(
363            client_id,
364            Box::new(ClientSource {
365                stream,
366                id: client_id,
367                name: name.clone(),
368                tx: self.tx.clone(),
369                active_connections: Arc::clone(&self.active_connections),
370                decoder: hdlc::Decoder::reticulum(HW_MTU, self.ifac_size),
371                control: self.control.clone(),
372            }),
373        )
374    }
375}
376
377/// One accepted client, read on the driver thread.
378struct ClientSource {
379    stream: TcpStream,
380    id: InterfaceId,
381    name: String,
382    tx: EventSender,
383    active_connections: Arc<AtomicUsize>,
384    decoder: hdlc::Decoder,
385    control: ListenerControl,
386}
387
388impl LoopSource for ClientSource {
389    fn fd(&self) -> std::os::fd::RawFd {
390        self.stream.as_raw_fd()
391    }
392
393    fn stopped(&self) -> bool {
394        self.control.should_stop()
395    }
396
397    fn on_removed(&mut self) {
398        self.active_connections.fetch_sub(1, Ordering::Relaxed);
399    }
400
401    fn on_abandoned(&mut self) {
402        // The driver was told about this client; it will never be read.
403        let _ = self.tx.push_from_loop([Event::InterfaceDown(self.id)]);
404        self.on_removed();
405    }
406
407    fn on_ready(&mut self, cx: &mut LoopContext<'_>) -> SourceStatus {
408        if self.control.should_stop() {
409            return SourceStatus::Remove;
410        }
411        match read_stream_frames(
412            self.stream.as_raw_fd(),
413            &mut self.decoder,
414            cx,
415            &self.tx,
416            self.id,
417            &self.name,
418            |_, _| {},
419        ) {
420            StreamRead::Rearm => SourceStatus::Rearm,
421            StreamRead::Suspend => SourceStatus::Suspend,
422            StreamRead::Closed => {
423                log::info!("[{}] client {} disconnected", self.name, self.id.0);
424                let _ = self.tx.push_from_loop([Event::InterfaceDown(self.id)]);
425                SourceStatus::Remove
426            }
427            StreamRead::Failed(e) => {
428                log::warn!("[{}] client {} read error: {}", self.name, self.id.0, e);
429                let _ = self.tx.push_from_loop([Event::InterfaceDown(self.id)]);
430                SourceStatus::Remove
431            }
432            StreamRead::DriverGone => SourceStatus::Remove,
433        }
434    }
435}
436
437// --- Factory implementation ---
438
439use super::{InterfaceConfigData, InterfaceFactory, StartContext, StartResult};
440use std::collections::HashMap;
441
442/// Factory for `TCPServerInterface`.
443pub struct TcpServerFactory;
444
445impl InterfaceFactory for TcpServerFactory {
446    fn type_name(&self) -> &str {
447        "TCPServerInterface"
448    }
449
450    fn parse_config(
451        &self,
452        name: &str,
453        id: InterfaceId,
454        params: &HashMap<String, String>,
455    ) -> Result<Box<dyn InterfaceConfigData>, String> {
456        let listen_ip = params
457            .get("listen_ip")
458            .cloned()
459            .unwrap_or_else(|| "0.0.0.0".into());
460        let listen_port = params
461            .get("listen_port")
462            .and_then(|v| v.parse().ok())
463            .unwrap_or(4242);
464        let max_connections = params.get("max_connections").and_then(|v| v.parse().ok());
465        let mut config = TcpServerConfig {
466            name: name.to_string(),
467            listen_ip,
468            listen_port,
469            interface_id: id,
470            max_connections,
471            ingress_control: IngressControlConfig::enabled(),
472            runtime: Arc::new(Mutex::new(TcpServerRuntime {
473                max_connections: None,
474            })),
475        };
476        let startup = TcpServerRuntime::from_config(&config);
477        config.runtime = Arc::new(Mutex::new(startup));
478        Ok(Box::new(config))
479    }
480
481    fn start(
482        &self,
483        config: Box<dyn InterfaceConfigData>,
484        ctx: StartContext,
485    ) -> io::Result<StartResult> {
486        let mut cfg = *config
487            .into_any()
488            .downcast::<TcpServerConfig>()
489            .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "wrong config type"))?;
490        cfg.ingress_control = ctx.ingress_control;
491        let parent_id = cfg.interface_id;
492        let control = start_with_template(
493            cfg,
494            ctx.tx,
495            ctx.next_dynamic_id,
496            Some(super::DynamicInterfaceTemplate {
497                parent_id,
498                interface_type: "TCPServerClientInterface".into(),
499                ifac: ctx.ifac,
500                mode: ctx.mode,
501                gravity: ctx.gravity,
502                recursive_prs: ctx.recursive_prs,
503                announces_from_internal: ctx.announces_from_internal,
504                announces_to_internal: ctx.announces_to_internal,
505            }),
506            ctx.underlay_mark,
507        )?;
508        Ok(StartResult::Listener {
509            control: Some(control),
510        })
511    }
512}
513
514pub(crate) fn runtime_handle_from_config(config: &TcpServerConfig) -> TcpServerRuntimeConfigHandle {
515    TcpServerRuntimeConfigHandle {
516        interface_name: config.name.clone(),
517        runtime: Arc::clone(&config.runtime),
518        startup: TcpServerRuntime::from_config(config),
519    }
520}
521
522#[cfg(test)]
523mod tests {
524    use super::*;
525    use std::io::Read;
526    use std::net::TcpStream;
527    use std::sync::mpsc::RecvTimeoutError;
528    use std::thread;
529    use std::time::Duration;
530
531    fn find_free_port() -> u16 {
532        crate::test_support::port()
533    }
534
535    fn make_server_config(
536        port: u16,
537        interface_id: u64,
538        max_connections: Option<usize>,
539    ) -> TcpServerConfig {
540        let mut config = TcpServerConfig {
541            name: "test-server".into(),
542            listen_ip: "127.0.0.1".into(),
543            listen_port: port,
544            interface_id: InterfaceId(interface_id),
545            max_connections,
546            ingress_control: IngressControlConfig::enabled(),
547            runtime: Arc::new(Mutex::new(TcpServerRuntime {
548                max_connections: None,
549            })),
550        };
551        let startup = TcpServerRuntime::from_config(&config);
552        config.runtime = Arc::new(Mutex::new(startup));
553        config
554    }
555
556    #[cfg(target_os = "linux")]
557    #[test]
558    fn stop_resume_preserves_partial_server_frame() {
559        use std::io::{BufRead, BufReader};
560        use std::process::{Child, Command, Stdio};
561        use std::sync::mpsc;
562
563        const CHILD_PORT: &str = "RNS_TEST_INTERRUPTED_TCP_SERVER_PORT";
564        const ROUNDS: usize = 3;
565        let payload = vec![0x7e; constants::HEADER_MINSIZE + 16];
566        if let Ok(port) = std::env::var(CHILD_PORT) {
567            let (tx, rx) = crate::event::channel();
568            let control = start(
569                make_server_config(port.parse().unwrap(), 1, None),
570                tx,
571                Arc::new(AtomicU64::new(1200)),
572            )
573            .unwrap();
574            println!("EINTR_READY");
575            let registration = rx.recv_timeout(Duration::from_secs(5)).unwrap();
576            assert!(matches!(
577                registration,
578                Event::InterfaceUp(InterfaceId(1200), _, _)
579            ));
580            println!("EINTR_CONNECTED");
581            for round in 0..ROUNDS {
582                let event = rx.recv_timeout(Duration::from_secs(5)).unwrap();
583                assert!(
584                    matches!(event, Event::Frame { interface_id: InterfaceId(1200), data, .. } if data == payload),
585                    "read interruption must preserve the connection and partial frame"
586                );
587                println!("EINTR_FRAME_{round}");
588            }
589            control.request_stop();
590            return;
591        }
592
593        // Signals affect only the isolated child, never the parallel test runner.
594        struct ChildGuard(Child);
595        impl Drop for ChildGuard {
596            fn drop(&mut self) {
597                if matches!(self.0.try_wait(), Ok(None)) {
598                    unsafe { libc::kill(self.0.id() as i32, libc::SIGCONT) };
599                    let _ = self.0.kill();
600                    let _ = self.0.wait();
601                }
602            }
603        }
604        let port = find_free_port();
605        let mut child = ChildGuard(
606            Command::new(std::env::current_exe().unwrap())
607                .args([
608                    "interface::tcp_server::tests::stop_resume_preserves_partial_server_frame",
609                    "--exact",
610                    "--nocapture",
611                ])
612                .env(CHILD_PORT, port.to_string())
613                .stdout(Stdio::piped())
614                .spawn()
615                .unwrap(),
616        );
617        let stdout = child.0.stdout.take().unwrap();
618        let (lines_tx, lines_rx) = mpsc::channel();
619        let output = thread::spawn(move || {
620            for line in BufReader::new(stdout).lines() {
621                if lines_tx.send(line.unwrap()).is_err() {
622                    break;
623                }
624            }
625        });
626        let wait_for = |expected: &str| {
627            let deadline = std::time::Instant::now() + Duration::from_secs(5);
628            loop {
629                let line = lines_rx
630                    .recv_timeout(deadline.saturating_duration_since(std::time::Instant::now()))
631                    .unwrap_or_else(|error| panic!("waiting for {expected}: {error}"));
632                if line == expected {
633                    break;
634                }
635            }
636        };
637        wait_for("EINTR_READY");
638        let mut client = TcpStream::connect(("127.0.0.1", port)).unwrap();
639        wait_for("EINTR_CONNECTED");
640        let frame = hdlc::frame(&payload);
641        // Interrupt with an escape byte pending in the HDLC decoder.
642        let split = frame.iter().position(|b| *b == 0x7d).unwrap() + 1;
643        for round in 0..ROUNDS {
644            client.write_all(&frame[..split]).unwrap();
645            thread::sleep(Duration::from_millis(30));
646            assert_eq!(unsafe { libc::kill(child.0.id() as i32, libc::SIGSTOP) }, 0);
647            thread::sleep(Duration::from_millis(200));
648            assert_eq!(unsafe { libc::kill(child.0.id() as i32, libc::SIGCONT) }, 0);
649            client.write_all(&frame[split..]).unwrap();
650            wait_for(&format!("EINTR_FRAME_{round}"));
651        }
652        assert!(child.0.wait().unwrap().success());
653        output.join().unwrap();
654    }
655
656    #[test]
657    fn accept_connection() {
658        let port = find_free_port();
659        let (tx, rx) = crate::event::channel();
660        let next_id = Arc::new(AtomicU64::new(1000));
661
662        let config = make_server_config(port, 1, None);
663
664        start(config, tx, next_id).unwrap();
665
666        // Give server time to start listening
667        thread::sleep(Duration::from_millis(50));
668
669        // Connect a client
670        let _client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
671
672        // Should receive InterfaceUp with InterfaceInfo (dynamic)
673        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
674        match event {
675            Event::InterfaceUp(id, writer, info) => {
676                assert_eq!(id, InterfaceId(1000));
677                assert!(writer.is_some());
678                assert!(info.is_some());
679            }
680            other => panic!("expected InterfaceUp, got {:?}", other),
681        }
682    }
683
684    #[test]
685    fn spawned_client_drops_exact_header_minimum_and_recovers() {
686        let port = find_free_port();
687        let (tx, rx) = crate::event::channel();
688        let next_id = Arc::new(AtomicU64::new(1100));
689        start(make_server_config(port, 1, None), tx, next_id).unwrap();
690        thread::sleep(Duration::from_millis(50));
691
692        let mut client = TcpStream::connect(format!("127.0.0.1:{port}")).unwrap();
693        assert!(matches!(
694            rx.recv_timeout(Duration::from_secs(2)).unwrap(),
695            Event::InterfaceUp(InterfaceId(1100), _, _)
696        ));
697        client
698            .write_all(&hdlc::frame(&[0x11; rns_core::constants::HEADER_MINSIZE]))
699            .unwrap();
700        let valid = vec![0x22; rns_core::constants::HEADER_MINSIZE + 1];
701        client.write_all(&hdlc::frame(&valid)).unwrap();
702
703        assert!(matches!(
704            rx.recv_timeout(Duration::from_secs(2)).unwrap(),
705            Event::Frame { interface_id: InterfaceId(1100), data, .. } if data == valid
706        ));
707    }
708
709    #[test]
710    fn typed_registration_carries_exact_type_mode_parent_and_ifac() {
711        let port = find_free_port();
712        let (tx, rx) = crate::event::channel();
713        let ifac = crate::ifac::derive_ifac(Some("network"), Some("key"), 8).unwrap();
714        let template = crate::interface::DynamicInterfaceTemplate {
715            parent_id: InterfaceId(77),
716            interface_type: "TCPServerClientInterface".into(),
717            ifac: Some(ifac),
718            mode: constants::MODE_GATEWAY,
719            gravity: -6,
720            recursive_prs: true,
721            announces_from_internal: false,
722            announces_to_internal: None,
723        };
724        start_with_template(
725            make_server_config(port, 77, None),
726            tx,
727            Arc::new(AtomicU64::new(1200)),
728            Some(template),
729            None,
730        )
731        .unwrap();
732        thread::sleep(Duration::from_millis(50));
733        let _client = TcpStream::connect(format!("127.0.0.1:{port}")).unwrap();
734        match rx.recv_timeout(Duration::from_secs(2)).unwrap() {
735            Event::DynamicInterfaceUp { registration, .. } => {
736                assert_eq!(registration.parent_id, InterfaceId(77));
737                assert_eq!(registration.interface_type, "TCPServerClientInterface");
738                assert_eq!(registration.info.mode, constants::MODE_GATEWAY);
739                assert_eq!(registration.info.gravity, -6);
740                assert!(registration.info.recursive_prs);
741                assert!(!registration.info.announces_from_internal);
742                assert_eq!(registration.ifac.as_ref().map(|state| state.size), Some(8));
743            }
744            other => panic!("expected typed dynamic registration, got {other:?}"),
745        }
746    }
747
748    #[test]
749    fn spawned_client_inherits_ingress_control_config() {
750        let port = find_free_port();
751        let (tx, rx) = crate::event::channel();
752        let next_id = Arc::new(AtomicU64::new(1100));
753
754        let mut config = make_server_config(port, 11, None);
755        config.ingress_control = IngressControlConfig::disabled();
756        config.ingress_control.max_held_announces = 17;
757        config.ingress_control.burst_hold = 1.5;
758        config.ingress_control.burst_freq_new = 2.5;
759        config.ingress_control.burst_freq = 3.5;
760        config.ingress_control.new_time = 4.5;
761        config.ingress_control.burst_penalty = 5.5;
762        config.ingress_control.held_release_interval = 6.5;
763
764        start(config, tx, next_id).unwrap();
765        thread::sleep(Duration::from_millis(50));
766
767        let _client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
768
769        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
770        match event {
771            Event::InterfaceUp(_, _, Some(info)) => {
772                assert!(!info.ingress_control.enabled);
773                assert_eq!(info.ingress_control.max_held_announces, 17);
774                assert_eq!(info.ingress_control.burst_hold, 1.5);
775                assert_eq!(info.ingress_control.burst_freq_new, 2.5);
776                assert_eq!(info.ingress_control.burst_freq, 3.5);
777                assert_eq!(info.ingress_control.new_time, 4.5);
778                assert_eq!(info.ingress_control.burst_penalty, 5.5);
779                assert_eq!(info.ingress_control.held_release_interval, 6.5);
780            }
781            other => panic!("expected InterfaceUp with InterfaceInfo, got {:?}", other),
782        }
783    }
784
785    #[test]
786    fn listener_stop_prevents_new_accepts() {
787        let port = find_free_port();
788        let (tx, rx) = crate::event::channel();
789        let next_id = Arc::new(AtomicU64::new(1500));
790
791        let config = make_server_config(port, 15, None);
792        let control = start(config, tx, next_id).unwrap();
793
794        thread::sleep(Duration::from_millis(50));
795        control.request_stop();
796        thread::sleep(Duration::from_millis(120));
797
798        let connect_result = TcpStream::connect(format!("127.0.0.1:{}", port));
799        if let Ok(stream) = connect_result {
800            drop(stream);
801        }
802
803        match rx.recv_timeout(Duration::from_millis(200)) {
804            Err(RecvTimeoutError::Timeout) | Err(RecvTimeoutError::Disconnected) => {}
805            other => panic!(
806                "expected no InterfaceUp after listener stop, got {:?}",
807                other
808            ),
809        }
810    }
811
812    #[test]
813    fn receive_frame_from_client() {
814        let port = find_free_port();
815        let (tx, rx) = crate::event::channel();
816        let next_id = Arc::new(AtomicU64::new(2000));
817
818        let config = make_server_config(port, 2, None);
819
820        start(config, tx, next_id).unwrap();
821        thread::sleep(Duration::from_millis(50));
822
823        let mut client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
824
825        // Drain InterfaceUp
826        let _ = rx.recv_timeout(Duration::from_secs(1)).unwrap();
827
828        // Send an HDLC frame (>= 19 bytes)
829        let payload: Vec<u8> = (0..32).collect();
830        let framed = hdlc::frame(&payload);
831        client.write_all(&framed).unwrap();
832
833        // Should receive Frame event
834        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
835        match event {
836            Event::Frame {
837                interface_id,
838                data,
839                rssi: _,
840                snr: _,
841            } => {
842                assert_eq!(interface_id, InterfaceId(2000));
843                assert_eq!(data, payload);
844            }
845            other => panic!("expected Frame, got {:?}", other),
846        }
847    }
848
849    #[test]
850    fn a_flood_into_a_small_queue_suspends_reading_without_losing_frames() {
851        const FRAMES: u32 = 3000;
852        let port = find_free_port();
853        let (tx, rx) = crate::event::channel_with_capacity(4);
854        let snapshot_tx = tx.clone();
855        start(
856            make_server_config(port, 4, None),
857            tx,
858            Arc::new(AtomicU64::new(4000)),
859        )
860        .unwrap();
861        let mut client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
862        assert!(matches!(
863            rx.recv_timeout(Duration::from_secs(2)).unwrap(),
864            Event::InterfaceUp(InterfaceId(4000), _, _)
865        ));
866
867        let writer = thread::spawn(move || {
868            let mut wire = Vec::new();
869            for seq in 0..FRAMES {
870                let mut payload = vec![0u8; 32];
871                payload[2..18].copy_from_slice(&[0x5A; 16]);
872                payload[20..24].copy_from_slice(&seq.to_be_bytes());
873                wire.extend(hdlc::frame(&payload));
874            }
875            client.write_all(&wire).unwrap();
876            client
877        });
878
879        for expected in 0..FRAMES {
880            match rx.recv_timeout(Duration::from_secs(5)).unwrap() {
881                Event::Frame {
882                    interface_id, data, ..
883                } => {
884                    assert_eq!(interface_id, InterfaceId(4000));
885                    let seq = u32::from_be_bytes(data[20..24].try_into().unwrap());
886                    assert_eq!(seq, expected, "frames lost or reordered");
887                }
888                other => panic!("expected Frame, got {:?}", other),
889            }
890        }
891        let _client = writer.join().unwrap();
892        assert_eq!(snapshot_tx.inbound_queue_snapshot().dropped, [0, 0, 0, 0]);
893    }
894
895    #[test]
896    fn dropping_the_receiver_closes_the_listener() {
897        let port = find_free_port();
898        let (tx, rx) = crate::event::channel();
899        start(
900            make_server_config(port, 6, None),
901            tx,
902            Arc::new(AtomicU64::new(6000)),
903        )
904        .unwrap();
905        drop(rx);
906        let refused = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap_err();
907        assert_eq!(refused.kind(), io::ErrorKind::ConnectionRefused);
908    }
909
910    #[test]
911    fn a_stopped_listener_is_closed_at_the_next_sweep() {
912        let port = find_free_port();
913        let (tx, rx) = crate::event::channel();
914        let control = start(
915            make_server_config(port, 5, None),
916            tx,
917            Arc::new(AtomicU64::new(5000)),
918        )
919        .unwrap();
920        control.request_stop();
921        rx.sweep_stopped_sources();
922        let refused = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap_err();
923        assert_eq!(refused.kind(), io::ErrorKind::ConnectionRefused);
924    }
925
926    #[test]
927    fn accept_errors_map_to_drain_next_or_pause() {
928        use io::ErrorKind;
929        let step = |kind: ErrorKind| accept_error_step(&io::Error::from(kind));
930        assert_eq!(step(ErrorKind::WouldBlock), AcceptStep::Drained);
931        for kind in [
932            ErrorKind::Interrupted,
933            ErrorKind::ConnectionAborted,
934            ErrorKind::ConnectionReset,
935            ErrorKind::NetworkDown,
936            ErrorKind::NetworkUnreachable,
937            ErrorKind::HostUnreachable,
938        ] {
939            assert_eq!(step(kind), AcceptStep::Next, "{kind:?}");
940        }
941        assert_eq!(
942            accept_error_step(&io::Error::from_raw_os_error(libc::ECONNABORTED)),
943            AcceptStep::Next
944        );
945        assert_eq!(
946            accept_error_step(&io::Error::from_raw_os_error(libc::EPROTO)),
947            AcceptStep::Next
948        );
949        for errno in [libc::EMFILE, libc::ENFILE, libc::ENOBUFS, libc::ENOMEM] {
950            assert_eq!(
951                accept_error_step(&io::Error::from_raw_os_error(errno)),
952                AcceptStep::Pause,
953                "errno {errno}"
954            );
955        }
956    }
957
958    /// Pauses on its first pass without accepting, as the listener does
959    /// after an accept error, then serves the real listener.
960    struct PauseOnce {
961        inner: ListenerSource,
962        paused: bool,
963    }
964
965    impl LoopSource for PauseOnce {
966        fn fd(&self) -> std::os::fd::RawFd {
967            self.inner.fd()
968        }
969        fn on_ready(&mut self, cx: &mut LoopContext<'_>) -> SourceStatus {
970            if !std::mem::replace(&mut self.paused, true) {
971                return SourceStatus::Pause;
972            }
973            self.inner.on_ready(cx)
974        }
975    }
976
977    fn serve_pass(rx: &crate::event::EventReceiver) {
978        let mut ready = polling::Events::new();
979        rx.wait(&mut ready, Some(Duration::from_millis(50)))
980            .unwrap();
981        rx.serve_sources(&ready);
982    }
983
984    #[test]
985    fn connections_made_while_the_listener_is_paused_are_accepted_after_the_resume() {
986        let (tx, rx) = crate::event::channel();
987        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
988        listener.set_nonblocking(true).unwrap();
989        let addr = listener.local_addr().unwrap();
990        let config = make_server_config(addr.port(), 8, None);
991        let inner = ListenerSource {
992            listener,
993            name: config.name.clone(),
994            tx: tx.clone(),
995            next_id: Arc::new(AtomicU64::new(8000)),
996            runtime: Arc::clone(&config.runtime),
997            ingress_control: config.ingress_control,
998            active_connections: Arc::new(AtomicUsize::new(0)),
999            control: ListenerControl::new(),
1000            dynamic_template: None,
1001            ifac_size: 0,
1002            underlay_mark: None,
1003        };
1004        tx.register_source(
1005            crate::interface::loop_writer::read_key(config.interface_id),
1006            Box::new(PauseOnce {
1007                inner,
1008                paused: false,
1009            }),
1010        )
1011        .unwrap();
1012
1013        let _first = TcpStream::connect(addr).unwrap();
1014        serve_pass(&rx);
1015        let _second = TcpStream::connect(addr).unwrap();
1016        for _ in 0..3 {
1017            serve_pass(&rx);
1018        }
1019        assert!(
1020            rx.try_recv().is_err(),
1021            "a paused listener accepted a connection"
1022        );
1023
1024        rx.resume_paused_sources();
1025        let mut up = Vec::new();
1026        for _ in 0..20 {
1027            serve_pass(&rx);
1028            while let Ok(event) = rx.try_recv() {
1029                match event {
1030                    Event::InterfaceUp(id, _, _) => up.push(id),
1031                    other => panic!("expected InterfaceUp, got {:?}", other),
1032                }
1033            }
1034            if up.len() == 2 {
1035                break;
1036            }
1037        }
1038        assert_eq!(up, [InterfaceId(8000), InterfaceId(8001)]);
1039    }
1040
1041    fn sockopt_int(fd: std::os::fd::RawFd, level: libc::c_int, name: libc::c_int) -> libc::c_int {
1042        let mut value: libc::c_int = 0;
1043        let mut len = std::mem::size_of::<libc::c_int>() as libc::socklen_t;
1044        // SAFETY: `value` and `len` are valid for the call and sized for an int.
1045        let rc = unsafe {
1046            libc::getsockopt(
1047                fd,
1048                level,
1049                name,
1050                &mut value as *mut _ as *mut libc::c_void,
1051                &mut len,
1052            )
1053        };
1054        assert_eq!(rc, 0, "getsockopt: {}", io::Error::last_os_error());
1055        value
1056    }
1057
1058    #[test]
1059    fn accepted_sockets_get_keepalive_and_timeout_options() {
1060        let port = find_free_port();
1061        let (tx, rx) = crate::event::channel();
1062        start(
1063            make_server_config(port, 9, None),
1064            tx,
1065            Arc::new(AtomicU64::new(9000)),
1066        )
1067        .unwrap();
1068        let _client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
1069        let writer = match rx.recv_timeout(Duration::from_secs(2)).unwrap() {
1070            Event::InterfaceUp(_, Some(writer), _) => writer,
1071            other => panic!("expected InterfaceUp with writer, got {:?}", other),
1072        };
1073        let fd = writer
1074            .loop_socket()
1075            .expect("server writer exposes its socket");
1076        assert_eq!(sockopt_int(fd, libc::SOL_SOCKET, libc::SO_KEEPALIVE), 1);
1077        assert_eq!(sockopt_int(fd, libc::IPPROTO_TCP, libc::TCP_NODELAY), 1);
1078        #[cfg(target_os = "linux")]
1079        assert_eq!(
1080            sockopt_int(fd, libc::IPPROTO_TCP, libc::TCP_USER_TIMEOUT),
1081            24_000
1082        );
1083    }
1084
1085    #[test]
1086    fn send_frame_to_client() {
1087        let port = find_free_port();
1088        let (tx, rx) = crate::event::channel();
1089        let next_id = Arc::new(AtomicU64::new(3000));
1090
1091        let config = make_server_config(port, 3, None);
1092
1093        start(config, tx, next_id).unwrap();
1094        thread::sleep(Duration::from_millis(50));
1095
1096        let mut client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
1097        client
1098            .set_read_timeout(Some(Duration::from_secs(2)))
1099            .unwrap();
1100
1101        // Get the writer from InterfaceUp
1102        let event = rx.recv_timeout(Duration::from_secs(1)).unwrap();
1103        let mut writer = match event {
1104            Event::InterfaceUp(_, Some(w), _) => w,
1105            other => panic!("expected InterfaceUp with writer, got {:?}", other),
1106        };
1107
1108        // Send a frame via writer
1109        let payload: Vec<u8> = (0..24).collect();
1110        writer.send_frame(&payload).unwrap();
1111
1112        // Read from client side
1113        let mut buf = [0u8; 256];
1114        let n = client.read(&mut buf).unwrap();
1115        let expected = hdlc::frame(&payload);
1116        assert_eq!(&buf[..n], &expected[..]);
1117    }
1118
1119    #[test]
1120    fn multiple_clients() {
1121        let port = find_free_port();
1122        let (tx, rx) = crate::event::channel();
1123        let next_id = Arc::new(AtomicU64::new(4000));
1124
1125        let config = make_server_config(port, 4, None);
1126
1127        start(config, tx, next_id).unwrap();
1128        thread::sleep(Duration::from_millis(50));
1129
1130        // Connect two clients
1131        let _client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
1132        let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
1133
1134        // Collect InterfaceUp events
1135        let mut ids = Vec::new();
1136        for _ in 0..2 {
1137            let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
1138            match event {
1139                Event::InterfaceUp(id, _, _) => ids.push(id),
1140                other => panic!("expected InterfaceUp, got {:?}", other),
1141            }
1142        }
1143
1144        // Should have unique IDs
1145        assert_eq!(ids.len(), 2);
1146        assert_ne!(ids[0], ids[1]);
1147    }
1148
1149    #[test]
1150    fn client_disconnect() {
1151        let port = find_free_port();
1152        let (tx, rx) = crate::event::channel();
1153        let next_id = Arc::new(AtomicU64::new(5000));
1154
1155        let config = make_server_config(port, 5, None);
1156
1157        start(config, tx, next_id).unwrap();
1158        thread::sleep(Duration::from_millis(50));
1159
1160        let client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
1161
1162        // Drain InterfaceUp
1163        let _ = rx.recv_timeout(Duration::from_secs(1)).unwrap();
1164
1165        // Disconnect
1166        drop(client);
1167
1168        // Should receive InterfaceDown
1169        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
1170        assert!(
1171            matches!(event, Event::InterfaceDown(InterfaceId(5000))),
1172            "expected InterfaceDown(5000), got {:?}",
1173            event
1174        );
1175    }
1176
1177    #[test]
1178    fn server_bind_port() {
1179        let port = find_free_port();
1180        let (tx, _rx) = crate::event::channel();
1181        let next_id = Arc::new(AtomicU64::new(6000));
1182
1183        let config = make_server_config(port, 6, None);
1184
1185        // Should not error
1186        start(config, tx, next_id).unwrap();
1187    }
1188
1189    #[test]
1190    fn max_connections_rejects_excess() {
1191        let port = find_free_port();
1192        let (tx, rx) = crate::event::channel();
1193        let next_id = Arc::new(AtomicU64::new(7000));
1194
1195        let config = make_server_config(port, 7, Some(2));
1196
1197        start(config, tx, next_id).unwrap();
1198        thread::sleep(Duration::from_millis(50));
1199
1200        // Connect two clients (at limit)
1201        let _client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
1202        let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
1203
1204        // Drain both InterfaceUp events
1205        for _ in 0..2 {
1206            let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
1207            assert!(matches!(event, Event::InterfaceUp(_, _, _)));
1208        }
1209
1210        // Third connection should be accepted at TCP level but immediately dropped by server
1211        let client3 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
1212        client3
1213            .set_read_timeout(Some(Duration::from_millis(500)))
1214            .unwrap();
1215
1216        // Give server time to reject
1217        thread::sleep(Duration::from_millis(100));
1218
1219        // Should NOT receive a third InterfaceUp
1220        let result = rx.recv_timeout(Duration::from_millis(500));
1221        assert!(
1222            result.is_err(),
1223            "expected no InterfaceUp for rejected connection, got {:?}",
1224            result
1225        );
1226    }
1227
1228    #[test]
1229    fn max_connections_allows_after_disconnect() {
1230        let port = find_free_port();
1231        let (tx, rx) = crate::event::channel();
1232        let next_id = Arc::new(AtomicU64::new(7100));
1233
1234        let config = make_server_config(port, 71, Some(1));
1235
1236        start(config, tx, next_id).unwrap();
1237        thread::sleep(Duration::from_millis(50));
1238
1239        // Connect first client
1240        let client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
1241        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
1242        assert!(matches!(event, Event::InterfaceUp(_, _, _)));
1243
1244        // Disconnect first client
1245        drop(client1);
1246
1247        // Wait for InterfaceDown
1248        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
1249        assert!(matches!(event, Event::InterfaceDown(_)));
1250
1251        // Now a new connection should be accepted
1252        let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
1253        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
1254        assert!(
1255            matches!(event, Event::InterfaceUp(_, _, _)),
1256            "expected InterfaceUp after slot freed, got {:?}",
1257            event
1258        );
1259    }
1260
1261    #[test]
1262    fn runtime_max_connections_updates_live() {
1263        let port = find_free_port();
1264        let (tx, rx) = crate::event::channel();
1265        let next_id = Arc::new(AtomicU64::new(7200));
1266
1267        let config = make_server_config(port, 72, None);
1268        let runtime = Arc::clone(&config.runtime);
1269
1270        start(config, tx, next_id).unwrap();
1271        thread::sleep(Duration::from_millis(50));
1272
1273        let _client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
1274        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
1275        assert!(matches!(event, Event::InterfaceUp(_, _, _)));
1276
1277        {
1278            let mut runtime = runtime.lock().unwrap();
1279            runtime.max_connections = Some(1);
1280        }
1281
1282        let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
1283        let result = rx.recv_timeout(Duration::from_millis(400));
1284        assert!(
1285            result.is_err(),
1286            "expected no InterfaceUp after lowering max_connections, got {:?}",
1287            result
1288        );
1289    }
1290}
rns_net/interface/tcp_server.rs

rns_net/interface/
tcp_server.rs
