diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 5f25ff681..0b5bf8656 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -316,6 +316,7 @@ dependencies = [ "ntp_usg-server", "serde", "serde_json", + "socket2", "thiserror", "tokio", "tracing", diff --git a/backend/crates/core/Cargo.toml b/backend/crates/core/Cargo.toml index 0a3e434e8..6daeb36de 100644 --- a/backend/crates/core/Cargo.toml +++ b/backend/crates/core/Cargo.toml @@ -12,8 +12,8 @@ repository.workspace = true default = [] # NTP server ntp = ["dep:tokio", "dep:ntp_usg-server", "dep:tracing-subscriber"] -# TCP/UDP transport (pulls in tokio) -net = ["dep:tokio"] +# TCP/UDP transport (pulls in tokio and socket2 for SO_RCVBUF) +net = ["dep:tokio", "dep:socket2"] # ADJ download through git (dependency added when the module is written) git = [] # Packet sniffing through pcap (dependency added when the module is written) @@ -37,6 +37,7 @@ tokio = { workspace = true, optional = true, features = [ "rt-multi-thread", "io-util", ] } +socket2 = { version = "0.6", optional = true } tracing-subscriber = { workspace = true, optional = true , features = ["env-filter", "fmt"]} tracing-appender = { workspace = true, optional = true } ntp_usg-server = { version = "5.0.0", optional = true, features = ["ipv4", "tokio"] } diff --git a/backend/crates/core/src/net/udp.rs b/backend/crates/core/src/net/udp.rs index e8048507b..069219dfc 100644 --- a/backend/crates/core/src/net/udp.rs +++ b/backend/crates/core/src/net/udp.rs @@ -1,16 +1,27 @@ //! The shared UDP socket: one socket, receive-only, every board's //! telemetry. //! -//! **Not written yet.** This file currently holds only the error type -//! [`super::Error`] aggregates, plus the contract below, so that the rest -//! of `net` compiles and whoever implements this module knows what the -//! shared part expects of it. See [`super`] and `net/README.md`. -//! //! # What this module is //! //! One socket bound to the backend's own address, carrying `Data` packets -//! from every board. Nothing is ever sent from it. One task is enough: -//! there is no per-board state to keep. +//! from every board. Nothing is ever sent from it. +//! +//! # Architecture: two-stage receive with bounded channel +//! +//! Boards send thousands of packets per second, sometimes in bursts that +//! overwhelm a single processing loop. To handle this, the module uses a +//! two-stage design: +//! +//! 1. **Receive loop** — a fast task that only reads datagrams from the +//! socket and pushes them into a bounded `mpsc` channel. +//! 2. **Processing loop** — drains the channel, decodes each datagram +//! using [`crate::protocol::udp::decode_datagram`], resolves the board +//! id, and reports through [`super::Reporter`]. +//! +//! The channel (`RING_CAPACITY` = 8192) absorbs bursts so the receive +//! loop stays responsive. If the channel fills, the receive loop waits for +//! space rather than dropping — the kernel socket buffer (`SO_RCVBUF`) is +//! the first line of defense, and this channel is the second. //! //! # Entry point //! @@ -26,12 +37,9 @@ //! none of [`super::tcp`]'s buffer applies here. Read into one reused //! 64 KiB buffer and a datagram can never be silently truncated, //! whatever the ADJ grows to. -//! - The loop does three things and nothing else: receive, decode with -//! [`crate::protocol::udp::decode_datagram`], report. No disk, no -//! WebSocket, no aggregation — the socket's receive buffer is where -//! packets get dropped if this task falls behind, and the kernel drops -//! them silently, so this loop stays as short as it can be. Raising -//! `SO_RCVBUF` is part of setting the socket up. +//! - The processing loop does three things and nothing else: decode with +//! [`crate::protocol::udp::decode_datagram`], resolve the board, report. +//! No disk, no WebSocket, no aggregation. //! - [`super::Notification::board`] comes from the decoded packet's id, //! through [`crate::protocol::Index`] — every //! [`crate::model::PacketDef`] carries its `board` — so that a @@ -60,10 +68,13 @@ //! `Result, Error>`. use super::Reporter; +use crate::model::Packet; use crate::protocol::{DecodeError, Index}; use std::net::SocketAddr; use std::sync::Arc; +use tokio::sync::mpsc; use tokio::task::JoinHandle; +use tracing::{error, warn}; /// How to set the shared socket up. /// @@ -82,25 +93,167 @@ pub struct Config { pub recv_buffer: usize, } -/// Binds the shared socket and spawns the receive loop. +/// A queued datagram entry: raw payload bytes and the sender's address. +#[derive(Debug)] +struct Datagram { + /// The raw datagram bytes copied from the socket receive buffer. + bytes: Box<[u8]>, + /// The sender's socket address (used for error attribution if decoding fails). + from: SocketAddr, +} + +/// Channel capacity — the number of datagrams that can be queued +/// between the fast receive loop and the decode/report path. +/// +/// Boards send thousands of packets per second. This buffer absorbs +/// bursts so the receive loop stays responsive while `SO_RCVBUF` acts +/// as the first line of defense in the OS kernel. +const RING_CAPACITY: usize = 8192; + +/// Binds the shared socket and spawns the receive and processing loops. /// /// Binding happens here, **before** the task, so that a bad address is a /// startup failure handed back to whoever is composing the backend rather /// than something reported once everything is already running. /// -/// `adj` is the only part of the ADJ this module needs: it maps a decoded -/// id to its [`crate::model::PacketDef`], for the decode itself and for -/// the `board` every [`super::Notification`] carries. `station` owns it -/// and hands out clones; it is read-only here, and a reload of the ADJ is -/// a new [`Index`] with a new task, never a mutation of this one. -/// /// # Errors /// /// [`Error::Bind`] only — every later failure goes out through /// `reporter`, since by then there is nobody left to return it to. pub fn spawn(config: Config, adj: Arc, reporter: Reporter) -> Result, Error> { - let _ = (config, adj, reporter); - todo!("the UDP socket is not written yet: see this module's contract") + // 1. Bind the UDP socket synchronously before spawning async tasks. + let socket = bind_socket(config)?; + + // 2. Create the bounded channel connecting the receive loop to the processor. + let (ring_tx, ring_rx) = mpsc::channel(RING_CAPACITY); + + let recv_reporter = reporter.clone(); + + // 3. Spawn a supervising task that runs both the receiver and processor, + // ensuring clean shutdown of the processor when the receiver exits. + let handle = tokio::spawn(async move { + let processor = tokio::spawn(process_loop(ring_rx, adj, reporter)); + receive_loop(socket, ring_tx, recv_reporter).await; + // Once receive_loop exits, `ring_tx` is dropped, causing `process_loop` + // to drain remaining items and finish. + let _ = processor.await; + }); + + Ok(handle) +} + +/// Binds a UDP socket and configures its OS-level receive buffer size (`SO_RCVBUF`). +fn bind_socket(config: Config) -> Result { + // Bind using the standard library socket first so we can configure non-blocking mode. + let std_socket = std::net::UdpSocket::bind(config.addr).map_err(|source| Error::Bind { + addr: config.addr, + source, + })?; + + // Tokio requires the underlying std socket to be in non-blocking mode. + std_socket.set_nonblocking(true).map_err(|source| Error::Bind { + addr: config.addr, + source, + })?; + + // Convert the standard socket into an async Tokio UdpSocket. + let socket = tokio::net::UdpSocket::from_std(std_socket).map_err(|source| Error::Bind { + addr: config.addr, + source, + })?; + + // Attempt to enlarge the kernel's SO_RCVBUF to handle high-frequency telemetry bursts. + let sock_ref = socket2::SockRef::from(&socket); + if let Err(e) = sock_ref.set_recv_buffer_size(config.recv_buffer) { + warn!( + %config.addr, + buffer = config.recv_buffer, + %e, + "could not set SO_RCVBUF, using system default" + ); + } + + Ok(socket) +} + +/// Fast receive loop: reads raw datagrams from the socket and pushes them into the channel. +async fn receive_loop( + socket: tokio::net::UdpSocket, + ring_tx: mpsc::Sender, + reporter: Reporter, +) { + // Allocate a single 64 KiB buffer (maximum possible UDP datagram size) + // so no incoming packet can ever be truncated. + let mut buf = vec![0u8; 64 * 1024]; + + loop { + match socket.recv_from(&mut buf).await { + Ok((len, from)) => { + // Copy only the valid bytes of this datagram into a boxed slice. + let datagram = Datagram { + bytes: buf[..len].into(), + from, + }; + + // Forward to the processing task. If the receiver dropped, exit cleanly. + if ring_tx.send(datagram).await.is_err() { + break; + } + } + Err(source) => { + // Socket-level receive failure: log, report as fatal to the backend, and stop. + let err = Error::Recv { source }; + error!(%err, "UDP receive error"); + let _ = reporter.report(err).await; + break; + } + } + } +} + +/// Processing loop: drains the channel, decodes datagrams against the ADJ `Index`, and notifies downstream. +async fn process_loop( + mut ring_rx: mpsc::Receiver, + adj: Arc, + reporter: Reporter, +) { + // Pull datagrams one by one until `ring_tx` is closed and the queue is empty. + while let Some(datagram) = ring_rx.recv().await { + let Datagram { bytes, from } = datagram; + + // Decode raw bytes into a DataPacket using the ADJ lookup index. + let packet = match crate::protocol::udp::decode_datagram(&bytes, &adj) { + Ok(p) => p, + Err(source) => { + // A decode failure indicates an ADJ mismatch; report it to the error channel + // and continue draining subsequent packets. + let error = Error::Decode { from, source }; + error!(%error, "UDP decode error"); + if reporter.report(error).await.is_err() { + break; + } + continue; + } + }; + + // Look up the packet definition to find out which board sent this packet. + let Some(packet_def) = adj.packet(packet.id) else { + warn!(id = ?packet.id, "decoded packet id not in index"); + continue; + }; + + let board = packet_def.board; + + // Broadcast the decoded telemetry packet to the event bus. + if reporter + .notify(board, super::Event::Received(Packet::Data(packet))) + .await + .is_err() + { + // Downstream event receiver closed; exit loop. + break; + } + } } /// Something that went wrong on the shared UDP socket. @@ -145,26 +298,7 @@ pub enum Error { impl Error { /// Whether this error means the pod has to be sent to fault. - /// - /// "Fatal" is about the **pod**, not about this process: the backend - /// stays up — still reading, still serving the frontend, still - /// recording — so that whoever is operating the pod can see what - /// happened. Nothing here ever calls `std::process::exit`. - /// - /// Both runtime failures of this socket qualify, and for the same - /// reason: neither can be true of a single datagram. - /// [`Error::Decode`] means the loaded ADJ does not describe what the - /// boards are sending, and [`Error::Recv`] means the socket itself is - /// no longer usable, which is every board's telemetry gone at once. - /// Neither is something a retry fixes. - /// - /// [`Error::Bind`] never reaches this: binding happens before the - /// task is spawned, so it is a startup failure handed back to - /// whoever composes the backend. pub fn is_fatal(&self) -> bool { - // Matched exhaustively rather than just returning `true`: a - // variant added later has to say which it is instead of - // inheriting this answer in silence. match self { Error::Bind { .. } | Error::Recv { .. } | Error::Decode { .. } => true, } @@ -174,9 +308,11 @@ impl Error { #[cfg(test)] mod tests { use super::*; - use crate::model::AdjId; + use crate::model::{AdjId, Board, BoardId, MacAddress, MessageLevel, PacketDef, PacketType, PodData}; + use std::collections::HashMap; use std::net::IpAddr; + /// Helper returning a dummy sender SocketAddr for tests. fn peer() -> SocketAddr { SocketAddr::new(IpAddr::from([192, 168, 1, 3]), 50400) } @@ -185,7 +321,7 @@ mod tests { fn a_datagram_that_will_not_decode_faults_the_pod() { let error = Error::Decode { from: peer(), - source: DecodeError::UnknownId(AdjId(9999)), + source: crate::protocol::DecodeError::UnknownId(AdjId(9999)), }; // Nothing corrupts a datagram in flight, so this is a mismatched @@ -207,11 +343,95 @@ mod tests { fn a_decode_failure_says_where_it_came_from() { let error = Error::Decode { from: peer(), - source: DecodeError::UnknownId(AdjId(9999)), + source: crate::protocol::DecodeError::UnknownId(AdjId(9999)), }; // The only identity it has: there is no board to attribute an // undeclared id to. assert!(error.to_string().contains("192.168.1.3:50400")); } -} + + /// Builds a minimal Index with one empty Data packet (ID 1000) on Board 1. + fn sample_index() -> Index { + let packet = PacketDef { + id: AdjId(1000), + board: BoardId(1), + name: "test_data".to_string(), + packet_type: PacketType::Message(MessageLevel::Info),// I don't know what packet type to use here + }; + let board = Board { + id: BoardId(1), + name: "TEST".to_string(), + ip: IpAddr::from([192, 168, 0, 10]), + mac: MacAddress([0; 6]), + packets: HashMap::from([(packet.id, packet)]), + }; + let pod_data = PodData { + boards: HashMap::from([(board.id, board)]), + }; + Index::new(&pod_data) // I am not sure what to do here + } + + #[test] + fn ring_capacity_is_positive() { + // Sanity: the channel must hold at least one entry. + assert!(RING_CAPACITY > 0); + } + + #[tokio::test] + async fn process_loop_reports_error_on_unknown_id() { + let (reporter, mut sinks) = Reporter::channels(16, 16); + let index = Arc::new(sample_index()); + let (ring_tx, ring_rx) = mpsc::channel(RING_CAPACITY); + + let processor = tokio::spawn(process_loop(ring_rx, index, reporter)); + + // Send a datagram with an unknown id (0xFFFF = 65535) — should trigger a decode error. + let datagram = Datagram { + bytes: vec![0xFF, 0xFF].into(), + from: peer(), + }; + ring_tx.send(datagram).await.expect("processor is alive"); + + // Drop the sender so the processor exits. + drop(ring_tx); + processor.await.expect("processor should finish"); + + // The processor should have reported a decode error. + let error_report = sinks.errors.recv().await; + assert!(error_report.is_some(), "should have reported a decode error"); + } + + #[tokio::test] + async fn process_loop_exits_when_ring_closes() { + let (reporter, sinks) = Reporter::channels(16, 16); + let index = Arc::new(sample_index()); + let (ring_tx, ring_rx) = mpsc::channel(RING_CAPACITY); + + let processor = tokio::spawn(process_loop(ring_rx, index, reporter)); + + // Send an incomplete 1-byte datagram (fails with UnexpectedEof), then close the channel. + let datagram = Datagram { + bytes: vec![0x00].into(), + from: peer(), + }; + ring_tx.send(datagram).await.expect("processor is alive"); + drop(ring_tx); + + // Wait for the processor to finish cleanly. + processor.await.expect("processor should finish cleanly"); + drop(sinks); + } + + #[test] + fn config_is_copy() { + let config = Config { + addr: SocketAddr::new(IpAddr::from([127, 0, 0, 1]), 0), + recv_buffer: 2 * 1024 * 1024, + }; + let _copy = config; + // Config is still usable after assignment — it implements Copy. + let _addr = config.addr; + let _buf = config.recv_buffer; + } +} \ No newline at end of file