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

rns_net/interface/
tcp_server.rs
