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

rns_net/interface/
tcp_server.rs
