diff --git a/backend/crates/core/Cargo.toml b/backend/crates/core/Cargo.toml index 83985c825..ee8dc7a88 100644 --- a/backend/crates/core/Cargo.toml +++ b/backend/crates/core/Cargo.toml @@ -29,7 +29,14 @@ serde.workspace = true serde_json.workspace = true thiserror.workspace = true tracing.workspace = true -tokio = { workspace = true, optional = true, features = ["macros", "rt-multi-thread"] } +# `io-util` is what `net` needs on top of the workspace's own list +# (`net`, `sync`, `time`...): `AsyncReadExt`/`AsyncWriteExt`, i.e. reading +# a TCP stream into a buffer at all. +tokio = { workspace = true, optional = true, features = [ + "macros", + "rt-multi-thread", + "io-util", +] } tracing-subscriber = { workspace = true, optional = true , features = ["env-filter", "fmt"]} tracing-appender = { workspace = true, optional = true } gix = { version = "0.87.1", optional = true, features = ["blocking-http-transport-reqwest-rust-tls", "blocking-network-client", "worktree-mutation"] } @@ -37,6 +44,10 @@ dirs = { version = "7.0.0", optional = true } tempfile = { version = "3.27", optional = true } ntp_usg-server = { version = "5.0.0", optional = true, features = ["ipv4", "tokio"] } +[dev-dependencies] +# Virtual time makes the writer's per-frame deadlines deterministic in tests. +tokio = { workspace = true, features = ["test-util"] } + # `workspace = true` can't be combined with extra lints in the same table, # so the workspace list is repeated here with `missing_docs` added: this is # the crate meant to be reused by other projects, so its public API must be diff --git a/backend/crates/core/src/lib.rs b/backend/crates/core/src/lib.rs index fdf802844..26b937a84 100644 --- a/backend/crates/core/src/lib.rs +++ b/backend/crates/core/src/lib.rs @@ -8,13 +8,12 @@ pub mod hello_world; #[cfg(feature = "git")] pub mod git; -/// NTP server -#[cfg(feature = "ntp")] -pub mod ntp; - /// Basic domain types: identifiers, values, measurements, protections, /// packets and boards. pub mod model; +/// NTP server +#[cfg(feature = "ntp")] +pub mod ntp; /// Parses an ADJ (Architecture Description JSON) directory into /// [`model::PodData`], plus backend-session metadata not modeled as @@ -25,3 +24,8 @@ pub mod adj; /// [`adj`] declares as the schema. See [`protocol::wire`], /// [`protocol::tcp`], [`protocol::udp`]. pub mod protocol; + +/// Owns the real tokio sockets and reports what arrives on them, calling +/// into [`protocol`] to decode. See [`net::tcp`], [`net::udp`]. +#[cfg(feature = "net")] +pub mod net; diff --git a/backend/crates/core/src/net.rs b/backend/crates/core/src/net.rs new file mode 100644 index 000000000..ff2824e13 --- /dev/null +++ b/backend/crates/core/src/net.rs @@ -0,0 +1,45 @@ +//! Real sockets, on top of tokio (feature `net`). +//! +//! Two submodules do the actual I/O: [`crate::net::tcp`], a client that +//! keeps one task per board and recovers frames out of a byte stream, +//! and [`crate::net::udp`], a single shared receive-only socket. This module opens nothing. It defines only what +//! both of them need to agree on: [`crate::net::Event`] (what happened on +//! a link, carried by a [`crate::net::Notification`]) and +//! [`crate::net::Reporter`] (the two sinks they report through — one for +//! notifications, one for errors). +//! +//! Everything that reaches a [`crate::net::Reporter`] has already been +//! decoded by [`crate::protocol`], which this module layers on top of: +//! `protocol` turns bytes into packets and never touches a socket, `net` +//! owns the socket and calls into it. A sniffer or a simulator can use +//! `protocol` on its own, against a capture, without any of this. +//! +//! # Identity +//! +//! Everything that travels down the notification channel carries its +//! board: [`crate::net::Notification`] holds a +//! [`crate::model::BoardId`] next to the timestamp, whatever the event +//! is. No consumer ever has to go and look up whose packet it just got. +//! +//! It is never a second source of truth, though — it is resolved from +//! what the ADJ already says, once, here. A TCP task was spawned for one +//! board and knows it without asking; [`crate::net::udp`] resolves it +//! from the decoded packet's id through [`crate::protocol::Index`] (every +//! [`crate::model::PacketDef`] carries its `board`), which is one +//! `HashMap` lookup per datagram instead of one per consumer downstream. +//! +//! A board id is all the identity there is: one connection per board, and +//! one shared socket for UDP, so nothing needs telling apart beyond it. +//! +//! Addresses are for the operator, not for the domain types: they show +//! up interpolated in each submodule's error messages (`tcp::Error`, +//! `udp::Error`), which is where somebody actually reads them. + +mod error; +mod report; + +pub mod tcp; +pub mod udp; + +pub use error::Error; +pub use report::{Closed, ErrorReport, Event, Notification, Reporter, Sinks}; diff --git a/backend/crates/core/src/net/README.md b/backend/crates/core/src/net/README.md new file mode 100644 index 000000000..f5f9cd5f3 --- /dev/null +++ b/backend/crates/core/src/net/README.md @@ -0,0 +1,175 @@ +# `net` — Real Sockets (feature `net`) + +Owns the actual tokio sockets. `protocol` turns bytes into packets and never +touches a socket; `net` owns the socket and calls into `protocol`. Nothing above +`net` (WebSocket, disk, aggregation) is known here. + +Two submodules do the I/O, and **this directory's shared part opens nothing**: + +``` +net/ +├── net.rs # module root: docs and re-exports +├── report.rs # Event, Notification, ErrorReport, Reporter — what is reported +├── error.rs # Error — aggregates tcp::Error / udp::Error, nothing else +├── tcp.rs # the TCP client (not written yet: error type + contract) +└── udp.rs # the shared UDP socket (not written yet: error type + contract) +``` + +## The shared part + +### Identity: the board travels with everything + +Every notification carries its board, whatever the event: + +```rust +Notification { board: BoardId, at: SystemTime, event: Event } +``` + +It is never a second source of truth, though — it is resolved from what the ADJ +already says, once, inside `net`. A TCP task was spawned for one board and knows +it without asking; `udp` resolves it from the decoded packet's id through `Index` +(every `PacketDef` carries its `board`), which is one `HashMap` lookup per +datagram instead of one per consumer downstream. + +And the board is all the identity there is: one connection per board, one shared +socket for UDP, so nothing needs telling apart beyond it. + +Addresses are for the operator, not for the domain types: they show up +interpolated in `tcp::Error`/`udp::Error` messages, which is where somebody +actually reads them. + +### `Event` and `Notification` — what happened, and when + +```rust +enum Event { + Connected, // TCP only + Disconnected, // TCP only + Received(model::Packet), +} +``` + +`Event::Received` leans on the existing `model::Packet` umbrella +(`Data | Order | Protection | Message`) rather than repeating that split with one +variant per packet type. Its `Order` variant is never produced while receiving — +it is there for an eventual "the order really did go out on the wire" report, +which only `net` can make. + +`board` and `at` sit in the envelope rather than inside the variants so that +**everything** that travels down the channel carries them, and so a notification +**cannot be built without a timestamp**. `Reporter::notify` is the only thing that +stamps it. That +matters: a queue adds delay under load, so a date taken when the record is +*written* would be wrong by exactly that delay, precisely when the data matters +most. It is also the only full date a record can have — the board clock on +protections and messages (`ShortTimestamp`) is a time of day with no date. + +### `Reporter` — the two sinks + +```rust +let (reporter, sinks) = Reporter::channels(notifications, errors); + +reporter.notify(board, event).await?; // notification channel +reporter.report(tcp_or_udp_error).await?; // error channel +``` + +Cloneable, and meant to be cloned: one copy per TCP connection task, one for the +UDP loop, all feeding the same two channels. + +Two channels, not one, because the rules differ. **A notification is never +dropped**: `notify` waits when the channel is full, so the backpressure reaches +the read loop and the kernel's receive buffer becomes the only place anything can +be lost — and whatever is lost there is lost *before* being decoded, so +"everything decoded gets through" stays literally true. An error is rare enough +that it can afford to queue behind telemetry. + +`ErrorReport` carries **no board**, unlike a notification: the identity of what +failed lives inside the `Error` variant, which is the only place that always has +it. A datagram that fails to decode resolves to no board at all — its id is not in +the ADJ, so there is no `PacketDef` to ask — and all that can be said about it is +the address it came from, which is what `udp::Error::Decode` holds. + +### `Error` — aggregation only + +`net::Error` adds no identity and no message of its own; both variants are +`#[error(transparent)]`, so `to_string()` is the submodule's own message with its +identity already interpolated, and `source()` still reaches the original +`io::Error` or `DecodeError`. Each submodule owns its error type and puts the +board or address **inside the variant** — the Go backend lost exactly +that when an error crossed its channel. `is_fatal()`, if needed, belongs to each +submodule's own error: nobody outside knows which of its variants are fatal. + +## The contract for the two submodules + +| | `tcp` | `udp` | +|---|---|---| +| Entry point | `spawn(Config, Arc, Reporter) -> Handles` | `spawn(Config, Arc, Reporter) -> Result, Error>` (binds first) | +| Inputs | its `Config` (boards, timeouts, backoff, keep-alive) | its `Config` (bind address, `SO_RCVBUF`) | +| Role | client only; one connection per board, one task each | one shared socket, receive-only, one task | +| Framing | buffer + `&mut Cursor` + advance by `cursor.position()`; bail out past 374 bytes | none — `recv_from` delivers a whole datagram into a reused 64 KiB buffer | +| Writing | yes (orders, keep-alive); a single writer task per connection | no | +| On a decode error | report it; **fatal**, and close/reopen the connection (the stream cannot be resynchronised) | report it; **fatal**, and keep receiving | +| `Connected`/`Disconnected` | per board | never | +| Fault | writes it: `Handles::propagate_fault()`, id 0 to every board | cannot write; only reports | + +Both take the same two things besides their own `Config`: an `Arc` — the +only part of the ADJ `net` needs, owned by `station` and read-only here — and a +`Reporter`. Neither reads a config file, and neither reads the ADJ itself: the +addresses `tcp` connects to are resolved by `station` and arrive in `Config`, +which is also what lets a test point it at a fake peer. + +`net` never decides to fault the pod, but `tcp` is what can write one: `station` +reads `is_fatal()` and calls `Handles::propagate_fault()`, which puts the order +with id 0 into every board's writer channel (the Go backend's `SendFault`). A +*received* packet with id 0 is the same action requested by a board, and +`station` answers it the same way. + +Three rules both follow: + +1. The timestamp comes from `Reporter`, never by hand. +2. Decode and report, nothing else — no disk, no WebSocket, no aggregation. +3. What to do about an error is decided locally; only the report goes up. + +### What "fatal" means + +`tcp::Error::is_fatal`/`udp::Error::is_fatal` are about the **pod**, not about +this process: it means the pod has to be sent to fault, and the backend +**stays up** — still reading, still serving the frontend, still recording — so the +operator can see why. Nothing in `net` ever calls `std::process::exit`. + +Only a decode failure is fatal, on either transport, and this is a deliberate +departure from Go. Nothing corrupts a frame or a datagram in flight (both +transports checksum, TCP retransmits, and the pod's network does not fragment), so +something that will not decode means the loaded ADJ does not describe what the +firmware is sending — which reconnecting cannot fix. A failed connect or write is +*not* fatal: that is what the backoff is for. + +`net` never raises the fault itself — neither transport does. It logs, it reports, +and `is_fatal()` is the classification whoever consumes the error channel acts on. +That is deliberate: a mismatched ADJ fails on every frame that follows, thousands +a second, so the fault has to be broadcast **once**, and the guard for that lives +in one place instead of one per transport. + +## Why TCP needs a buffer and UDP does not + +TCP is a byte stream and preserves no write boundaries, so one read can deliver +half a frame or two frames glued together — both are normal, and the glued case is +the frequent one. Our frames carry no length prefix: a protection's length is +fixed but derivable only from the ADJ (`2 + 4 + 2 + the watched measurement's wire +width`), and a message ends at the second NUL byte of its two variable-length +strings. So a frame's end is never known in advance, only deduced by parsing it — +which is why the loop tries, and retries with more bytes when it fails. + +The Go backend never faced this because it read straight off the connection, where +a blocking read simply waits mid-frame. That state has to live somewhere: in Go it +lived in the blocked goroutine's stack, and here it lives in an explicit buffer — +which is also what lets `protocol` stay synchronous and socket-free, reusable by a +sniffer against a capture. + +UDP has none of it: the kernel hands over a whole datagram or nothing. + +## Open questions + +- Who raises the fault on `is_fatal()`, and where its "only once" guard lives. + Not in `net`: both transports only report. +- Whether an order's write is reported back as + `Event::Received(Packet::Order(..))`, or `station` just assumes it went out. diff --git a/backend/crates/core/src/net/error.rs b/backend/crates/core/src/net/error.rs new file mode 100644 index 000000000..64109a954 --- /dev/null +++ b/backend/crates/core/src/net/error.rs @@ -0,0 +1,28 @@ +//! The one error type an [`super::ErrorReport`] can carry. + +use super::{tcp, udp}; + +/// Anything that went wrong in `net`. +/// +/// Aggregation only: this enum adds no identity and no message of its +/// own. Each submodule owns its own error type and puts the identity of +/// whatever failed — the board, or the peer address — +/// *inside* the variant, not just in its text, so that the identity +/// survives being sent down a channel. The Go backend lost it exactly +/// there: the log line before the send had the board, the error value +/// that travelled did not. +/// +/// Both variants are `#[error(transparent)]`, so `to_string()` on this +/// enum is the submodule's own message, identity interpolated and all — +/// which is what gets shown to the operator — and `source()` still walks +/// down to the original `io::Error` or +/// [`crate::protocol::DecodeError`]. +#[derive(Debug, thiserror::Error)] +pub enum Error { + /// On a board's TCP connection. + #[error(transparent)] + Tcp(#[from] tcp::Error), + /// On the shared UDP socket. + #[error(transparent)] + Udp(#[from] udp::Error), +} diff --git a/backend/crates/core/src/net/report.rs b/backend/crates/core/src/net/report.rs new file mode 100644 index 000000000..35f7dc16c --- /dev/null +++ b/backend/crates/core/src/net/report.rs @@ -0,0 +1,367 @@ +//! What [`super::tcp`]/[`super::udp`] report, and the sinks they report +//! through. +//! +//! Two separate channels, both bounded. Notifications +//! ([`Notification`]) carry everything that happened on a link, errors +//! ([`ErrorReport`]) everything that went wrong. They are separate +//! because their rules differ: a notification must never be dropped, +//! while an error is rare enough that it can afford to wait behind a +//! queue of telemetry. +//! +//! Both sinks apply backpressure rather than dropping: see +//! [`Reporter::notify`]. + +use super::Error; +use crate::model::{BoardId, Packet}; +use std::time::SystemTime; +use tokio::sync::mpsc; + +/// Something that happened on a link: which board it concerns, when, +/// and what. +/// +/// Both `board` and `at` live in this envelope rather than inside the +/// variants of [`Event`], so that **everything** that travels down the +/// channel carries them — there is no event whose board the consumer has +/// to go and look up, and none that can be built without a timestamp. +/// [`Reporter::notify`] is the only thing that stamps `at`. +#[derive(Debug)] +pub struct Notification { + /// The board this concerns. + /// + /// Always known by the time anything is reported: a TCP task was + /// spawned for one board and knows it without asking, and [`super::udp`] + /// resolves it from the decoded packet's id through + /// [`crate::protocol::Index`] — `index.packet(id)` gives a + /// [`crate::model::PacketDef`], which carries its `board` — so that a + /// notification looks the same whichever transport brought it. + pub board: BoardId, + /// The instant the bytes came off the socket. + /// + /// Stamped when the event is reported, never by whoever writes it to + /// disk later: a queue adds delay under load, so a date taken at + /// write time would be wrong by exactly that delay, at exactly the + /// moment the data matters most. The frames that carry a board clock + /// of their own ([`crate::model::ShortTimestamp`], on protections and + /// messages) only carry a time of day, with no date, so this is also + /// the only full date a record can have. + pub at: SystemTime, + /// What happened. + pub event: Event, +} + +/// Something that went wrong, and when. +/// +/// Carries no identity of its own: whose failure it was travels inside +/// the [`Error`]'s own variant, which is the only place that always has +/// it. A datagram that fails to decode, for instance, resolves to no +/// board at all — its id is not in the ADJ, so there is no +/// [`crate::model::PacketDef`] to ask — and all that can be said about +/// it is the address it came from, which is exactly what +/// [`super::udp::Error::Decode`] holds. +#[derive(Debug)] +pub struct ErrorReport { + /// The instant the failure was reported, on the same clock as + /// [`Notification::at`]. + pub at: SystemTime, + /// What went wrong, identity included. + pub error: Error, +} + +/// What happened on a link. +#[derive(Debug)] +pub enum Event { + /// The board's connection came up, and it can now be talked to. + /// Which board is in the [`Notification`] around this, which is all + /// the identity there is: one connection per board. + /// + /// TCP only: there is nothing to connect in UDP, whose socket is + /// bound once and belongs to no single board. + Connected, + /// The board's connection went down, whatever the reason. If the + /// cause was worth reporting, it already went out as an + /// [`ErrorReport`] — but on the other channel, so nothing orders the + /// two against each other. + /// + /// TCP only, same as [`Event::Connected`]. + Disconnected, + /// A packet arrived and was decoded. Whose it is, is in the + /// [`Notification`] around it. + /// + /// [`Packet`] is the whole umbrella, so this one variant covers data, + /// protections and messages alike, with no second enum repeating that + /// split. Its [`Packet::Order`] variant is never produced while + /// receiving; it is what an eventual "the order really did go out on + /// the wire" report would use, which only `net` is in a position to + /// make. + Received(Packet), +} + +/// The receiving ends of a [`Reporter`]'s two channels, handed to +/// whoever consumes them (`station`). +#[derive(Debug)] +pub struct Sinks { + /// Everything that happened, in the order it happened. + pub notifications: mpsc::Receiver, + /// Everything that went wrong. + pub errors: mpsc::Receiver, +} + +/// The sinks [`super::tcp`] and [`super::udp`] report through. +/// +/// Cloneable, and meant to be cloned: every task gets its own copy — one +/// per TCP connection, one for the UDP loop — and they all feed the same +/// two channels. +#[derive(Debug, Clone)] +pub struct Reporter { + notifications: mpsc::Sender, + errors: mpsc::Sender, +} + +/// The consumer is gone, so there is nobody left to report to. +/// +/// The only failure either sink has, and a signal to stop rather than +/// something to retry. +#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] +#[error("the receiving end is gone")] +pub struct Closed; + +impl Reporter { + /// Builds both channels with the given capacities, and returns the + /// sending half to hand to the submodules plus the receiving halves + /// to hand to the consumer. + /// + /// Both are bounded on purpose: see [`Reporter::notify`] for what + /// happens when the notification channel fills up. + pub fn channels(notifications: usize, errors: usize) -> (Self, Sinks) { + let (notification_tx, notification_rx) = mpsc::channel(notifications); + let (error_tx, error_rx) = mpsc::channel(errors); + + ( + Self { + notifications: notification_tx, + errors: error_tx, + }, + Sinks { + notifications: notification_rx, + errors: error_rx, + }, + ) + } + + /// Stamps the instant and queues `event`. + /// + /// **Waits** if the channel is full, rather than dropping anything: + /// a packet that has already been decoded always reaches the + /// consumer. What that costs is backpressure — the wait reaches back + /// to the read loop, which stops reading, and the kernel's own + /// receive buffer becomes the only place anything can be lost. That + /// is deliberate: whatever is dropped there is dropped *before* being + /// decoded, so "everything decoded gets through" stays literally + /// true. + /// + /// The instant is taken before the wait, not after, so a full queue + /// delays the notification without distorting its timestamp. + pub async fn notify(&self, board: BoardId, event: Event) -> Result<(), Closed> { + let notification = Notification { + board, + at: SystemTime::now(), + event, + }; + + self.notifications + .send(notification) + .await + .map_err(|_| Closed) + } + + /// Stamps the instant and queues an error, on the error channel. + /// + /// Takes anything that converts into [`Error`], so a submodule + /// reports its own error type directly: + /// `reporter.report(tcp::Error::Connect { .. }).await`. + /// + /// Reporting is not deciding: by the time an error gets here its + /// submodule has already done whatever it was going to do about it + /// (reconnect, carry on with the next datagram), and has already + /// logged it through `tracing`. This channel is only for telling the + /// operator, through the frontend. + pub async fn report(&self, error: impl Into) -> Result<(), Closed> { + let report = ErrorReport { + at: SystemTime::now(), + error: error.into(), + }; + + self.errors.send(report).await.map_err(|_| Closed) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::model::{AdjId, DataPacket}; + use crate::net::udp; + use std::net::{IpAddr, SocketAddr}; + use std::time::Duration; + + fn a_packet() -> Event { + Event::Received(Packet::Data(DataPacket::new(AdjId(1000), vec![]))) + } + + fn an_error() -> udp::Error { + udp::Error::Recv { + source: std::io::Error::from(std::io::ErrorKind::ConnectionReset), + } + } + + #[tokio::test] + async fn a_notification_arrives_stamped() { + let (reporter, mut sinks) = Reporter::channels(4, 4); + + let before = SystemTime::now(); + let Ok(()) = reporter.notify(BoardId(3), a_packet()).await else { + unreachable!("the receiver is still alive"); + }; + let after = SystemTime::now(); + + let Some(notification) = sinks.notifications.recv().await else { + unreachable!("exactly one notification was sent"); + }; + assert!(notification.at >= before && notification.at <= after); + assert_eq!( + notification.board, + BoardId(3), + "the board travels with every notification, received packets included" + ); + } + + #[tokio::test] + async fn a_connection_event_names_its_board() { + let (reporter, mut sinks) = Reporter::channels(4, 4); + + // The one identity that cannot be derived from a packet: there is + // no packet here at all. + let Ok(()) = reporter.notify(BoardId(3), Event::Disconnected).await else { + unreachable!("the receiver is still alive"); + }; + + let Some(notification) = sinks.notifications.recv().await else { + unreachable!("exactly one notification was sent"); + }; + // The board is the whole identity: one connection per board. + assert_eq!(notification.board, BoardId(3)); + let Event::Disconnected = notification.event else { + unreachable!("a disconnection should arrive as Event::Disconnected"); + }; + } + + #[tokio::test] + async fn a_full_channel_makes_the_reporter_wait_instead_of_dropping() { + let (reporter, mut sinks) = Reporter::channels(1, 1); + + // Fill the single slot, then start a second report that cannot + // fit yet. + let Ok(()) = reporter.notify(BoardId(3), a_packet()).await else { + unreachable!("the first one fits"); + }; + let blocked = tokio::spawn({ + let reporter = reporter.clone(); + async move { reporter.notify(BoardId(3), a_packet()).await } + }); + + // It is still waiting: nothing was dropped, and nothing got + // through either. + tokio::time::sleep(Duration::from_millis(50)).await; + assert!( + !blocked.is_finished(), + "the second report should be waiting" + ); + + // Draining one slot lets it through, so both are delivered. + let Some(_first) = sinks.notifications.recv().await else { + unreachable!("the first notification was sent"); + }; + let Ok(Ok(())) = blocked.await else { + unreachable!("once a slot frees up, the second one goes through"); + }; + let Some(_second) = sinks.notifications.recv().await else { + unreachable!("the second notification went through"); + }; + } + + #[tokio::test] + async fn waiting_on_a_full_channel_does_not_delay_the_timestamp() { + let (reporter, mut sinks) = Reporter::channels(1, 1); + + let Ok(()) = reporter.notify(BoardId(3), a_packet()).await else { + unreachable!("the first one fits"); + }; + let blocked = tokio::spawn({ + let reporter = reporter.clone(); + async move { reporter.notify(BoardId(3), a_packet()).await } + }); + + // Let the second report sit in the queue for a while before + // anything drains it. + tokio::time::sleep(Duration::from_millis(50)).await; + let after_the_wait = SystemTime::now(); + + let Some(_first) = sinks.notifications.recv().await else { + unreachable!("the first notification was sent"); + }; + let Ok(Ok(())) = blocked.await else { + unreachable!("the second one goes through once there is room"); + }; + let Some(second) = sinks.notifications.recv().await else { + unreachable!("the second notification went through"); + }; + + // Stamped when it was reported, not when it finally fit. + assert!( + second.at <= after_the_wait, + "the timestamp should predate the end of the wait" + ); + } + + #[tokio::test] + async fn an_error_arrives_on_its_own_channel() { + let (reporter, mut sinks) = Reporter::channels(4, 4); + + let Ok(()) = reporter.report(an_error()).await else { + unreachable!("the receiver is still alive"); + }; + + let Some(report) = sinks.errors.recv().await else { + unreachable!("exactly one error was reported"); + }; + let Error::Udp(udp::Error::Recv { .. }) = report.error else { + unreachable!("a udp error should arrive as Error::Udp"); + }; + // Nothing reached the notification channel. + assert!(sinks.notifications.try_recv().is_err()); + } + + #[test] + fn a_udp_decode_error_keeps_the_address_it_cannot_resolve_to_a_board() { + let addr = SocketAddr::new(IpAddr::from([192, 168, 1, 3]), 50400); + let error = udp::Error::Decode { + from: addr, + source: crate::protocol::DecodeError::UnknownId(AdjId(9999)), + }; + + // The whole reason an error report carries no board: this one has + // an address, and no board to attribute it to. + let Error::Udp(udp::Error::Decode { from, .. }) = Error::from(error) else { + unreachable!("a decode error should convert into Error::Udp"); + }; + assert_eq!(from, addr); + } + + #[tokio::test] + async fn reporting_once_the_consumer_is_gone_says_so() { + let (reporter, sinks) = Reporter::channels(4, 4); + drop(sinks); + + assert_eq!(reporter.notify(BoardId(3), a_packet()).await, Err(Closed)); + assert_eq!(reporter.report(an_error()).await, Err(Closed)); + } +} diff --git a/backend/crates/core/src/net/tcp.rs b/backend/crates/core/src/net/tcp.rs new file mode 100644 index 000000000..8ac30c427 --- /dev/null +++ b/backend/crates/core/src/net/tcp.rs @@ -0,0 +1,308 @@ +//! The TCP client: one connection per board, one task each. +//! +//! **Session supervision not written yet.** Handles support board lookup, +//! routing, order encoding and byte queueing. In-memory framing, connection +//! attempts, session reading with reporting and queued frame/keep-alive writing are +//! implemented. Task startup and reconnection remain unimplemented. This module +//! exposes the error type [`super::Error`] aggregates and defines the transport contract. +//! See [`super`] and `net/README.md`. +//! +//! # What this module is +//! +//! The backend is the **client**: it connects out to each board's +//! `TCP_SERVER` port, and never listens. One connection per board, each +//! one its own task with its own framing buffer and its own reconnection +//! state. That connection carries protections and messages inbound, and +//! orders and the keep-alive outbound. +//! +//! # Entry point +//! +//! [`spawn`], taking [`Config`], an `Arc` and a +//! [`super::Reporter`], and returning [`Handles`] — the API is written +//! out below, with its reasoning on the items themselves. One call opens +//! **every** connection: one task per entry in [`Config::boards`], so the +//! per-board bookkeeping stays here and `station` only ever asks for +//! "write this to that board". +//! +//! # Contract +//! +//! - **Framing is this module's job.** TCP is a byte stream: a frame can +//! arrive split across reads, and two frames can arrive glued together +//! in one read — both are normal, and the second is the common case. +//! So: accumulate into a per-connection buffer, run +//! [`crate::protocol::tcp::decode_next`] against a `&mut Cursor` over +//! it, and on success advance the buffer by `cursor.position()`. Never +//! consume from the buffer unless a frame parsed in full, or the stream +//! is permanently misaligned. A failed parse means "not enough bytes +//! yet"; the one safeguard is that no frame can exceed 374 bytes +//! (`2 + 4 + 48 + 320`), so a buffer that grows past that without +//! yielding a frame is out of step rather than waiting, and the +//! connection should be dropped and reopened. +//! - **One writer task per connection.** Orders, the keep-alive and the +//! fault all share the socket, so a single task must own the write +//! half, or two half-written frames can interleave on the wire. That +//! task is what [`super::Reporter`]-side callers reach through the +//! board's channel in `Handles`. +//! - **Be ready to be told to propagate the fault.** `station` is what +//! decides (it is the one consuming the error channel and reading +//! [`Error::is_fatal`]); this module is what can actually write, so it +//! exposes `Handles::propagate_fault`, which puts **the order with id +//! 0** — the fault, same as the Go backend's `SendFault` — into every +//! board's writer channel at once, bounded by `write_timeout` per +//! write. It reports back which boards it could not reach instead of +//! collapsing them into one opaque error, since a fault that only +//! half-landed is exactly what the operator has to know. The same path +//! covers the other case the Go backend had under `propagate_fault`: a +//! *received* packet with id 0 is a board announcing a fault, and +//! `station` answers it by calling this, so that logic does not have to +//! exist twice. +//! - Reconnection with exponential backoff and unlimited retries, the +//! OS-level keep-alive disabled (there is an application-level one), +//! `linger = 0` so a dropped connection frees its fixed local port +//! immediately instead of sitting in `TIME_WAIT`. +//! - Report through [`super::Reporter`] only: [`super::Event::Connected`] +//! and [`super::Event::Disconnected`] per connection, +//! [`super::Event::Received`] per frame, and [`Error`] on the error +//! channel. Never stamp a timestamp by hand — [`super::Reporter::notify`] +//! does it. Never touch disk, WebSocket or aggregation. +//! - **A terminal decode failure is fatal**: see [`Error::is_fatal`]. An +//! unsuccessful attempt on at most 374 pending bytes is not terminal; +//! it only means "wait for more" under the framing rule above. Log a +//! terminal failure and report it on the error channel, which is all this +//! module does about it. Sending the pod to fault is not `net`'s call to make: +//! whoever consumes the error channel reads [`Error::is_fatal`] and +//! decides, which also keeps the "fault once" guard in one place +//! instead of one per transport. The stream cannot be resynchronised +//! mid-frame, though, so this module does close the connection and let +//! the reconnection loop reopen it — reconnecting will not fix a +//! mismatched ADJ, so expect the same frame to fail again. +//! - Everything else is decided here, locally: a failed connect or write +//! is what the backoff is for, and only the report goes up. + +use super::Reporter; +use crate::{model::BoardId, protocol::Index}; +use std::{cmp::min, net::SocketAddr, sync::Arc, time::Duration}; + +mod board_handle; +mod connection; +mod error; +mod framing; +mod handles; +mod reader; +mod writer; + +pub use board_handle::BoardHandle; +pub use error::{Error, SendError, Unreachable}; +pub use handles::Handles; + +/// One board to keep a connection to. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct BoardLink { + /// Whose connection this is. Every report about it names this board, + /// which is the whole identity of a link. + pub board: BoardId, + /// Where to connect: the board's `ip` + its `ports["TCP_SERVER"]`. + pub addr: SocketAddr, + /// The fixed local port to bind the client side to (`TCP_CLIENT + i` + /// in the Go backend). Being fixed is why `linger = 0` matters: a + /// dropped connection must not leave this port in `TIME_WAIT`. + /// Zero requests an OS-assigned port, useful for isolated local tests. + pub local_port: u16, +} + +/// How long to wait between reconnection attempts. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Backoff { + // The first wait after a connection fails or drops. + initial: u64, + // Times tried to backoff + count: u64, + // The ceiling the doubling stops at. Retries are unlimited: a board + // that is not up yet is the normal state before a test, not an + // error to give up on. + max: u64, +} + +impl Backoff { + /// Creates an instance of Backoff + /// Arguments: + /// - initial: value for first backoff atempt + /// - max: max value of backoff + pub fn new(initial: u64, max: u64) -> Self { + Self { + initial, + count: 0, + max, + } + } + + /// Returns the backoff interval for each attempt + pub fn get_interval(&mut self) -> Duration { + let res = min(self.max, self.initial * 2u64.pow(self.count as u32)); + // Increments the count for the next get_interval + self.count += 1; + Duration::from_millis(res) + } +} + +/// How to run the TCP client. +/// +/// Everything here is resolved by `station` before `net` is reached: this +/// module reads no config file, and it does not read the ADJ to find its +/// peers either. That is also what lets a test point it at a fake peer. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Config { + /// One entry per board to connect to, already filtered by what the + /// config file lists: a board the ADJ declares but the config does + /// not name gets no client at all. + pub boards: Vec, + /// Bound on each connection attempt. + pub connect_timeout: Duration, + /// What to wait between attempts. + pub backoff: Backoff, + /// Period of the application-level keep-alive: an empty order with ID 1. + /// Zero disables it. The first send waits one period; subsequent periods + /// start when the previous keep-alive finishes. Ordinary orders do not reset + /// the timer, and overdue sends never accumulate into a catch-up burst. + /// The session writer implements this; startup through `spawn` is still pending. + /// OS-level keep-alive remains disabled in favor of this application packet. + pub keep_alive: Duration, + /// Bound on each write, so that a peer that stops reading cannot park + /// a writer task for good (1 s in the Go backend). + pub write_timeout: Duration, +} + +/// Opens a connection to every board in `config.boards`. +/// +/// One call, one task per board: the per-connection bookkeeping stays in +/// this module, and `station` only ever asks for "write this to that +/// board" through the [`Handles`] that come back. +/// +/// `adj` is the only part of the ADJ this module needs: it decodes +/// inbound frames and encodes outbound orders against it. `station` owns +/// it and hands out clones; it is read-only here, and a reload of the ADJ +/// is a new [`Index`] with new tasks, never a mutation of this one. +/// +/// Connecting is *not* done before spawning, unlike [`super::udp`]: a +/// board that is not up yet is normal, so a failed attempt is a report +/// and a retry, never a startup failure. +pub fn spawn(config: Config, adj: Arc, reporter: Reporter) -> Handles { + let _ = (config, adj, reporter); + todo!("the TCP client is not written yet: see this module's contract") +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::protocol::DecodeError; + use std::path::Path; + + fn io_error() -> std::io::Error { + std::io::Error::from(std::io::ErrorKind::ConnectionRefused) + } + + #[test] + fn a_frame_that_will_not_decode_faults_the_pod() { + let error = Error::Decode { + board: BoardId(3), + source: DecodeError::UnknownId(crate::model::AdjId(9999)), + }; + + // Not a desynchronised stream to retry: the ADJ does not match + // the firmware, and reconnecting cannot fix that. + assert!(error.is_fatal()); + } + + #[test] + fn a_board_that_is_not_up_yet_does_not_fault_the_pod() { + let connect = Error::Connect { + board: BoardId(3), + source: io_error(), + }; + let write = Error::Write { + board: BoardId(3), + source: io_error(), + }; + + // That is what the reconnection backoff is for. + assert!(!connect.is_fatal()); + assert!(!write.is_fatal()); + } + + #[test] + fn an_error_says_which_board_it_was() { + let error = Error::Write { + board: BoardId(3), + source: io_error(), + }; + + // The identity survives being formatted, which is what reaches + // the operator. + assert!(error.to_string().contains("BoardId(3)")); + } + + #[tokio::test] + #[ignore = "manual test: start the documented nc listener on 127.0.0.1:50500 first"] + // THIS TEST IS SPECIAL: + // + // To try it, you need two terminals. Run these commands in this order, + // one in each terminal: + // 1. `{ + // printf '\x03\x00\x0c\x1e\x00\x00nc\x00One\x00' + // printf '\x03\x00\x0c\x1e\x00\x00nc\x00Two\x00' + // printf '\x03\x00\x0c\x1e\x00\x00n' + // sleep 3 + // printf 'c\x00Three\x00' + // } | nc -lN 127.0.0.1 50500` + // 2. `cargo test -p software-core -F net manual_tcp_receive -- --ignored --nocapture` + // + // The first two frames are sent whole. The third frame is split across two + // writes with a three-second delay. TCP may split or combine reads differently; + // the framing code must not depend on write/read boundaries matching. + // The test prints each complete frame, then finishes when nc closes the connection. + // + // This test checks connection, framing and reporting by displaying notifications. + async fn manual_tcp_receive() -> Result<(), Box> { + // Load the fixture ADJ so the real protocol decoder can resolve message ID 3. + let adj_path = Path::new(env!("CARGO_MANIFEST_DIR")).join("src/adj/test/adj"); + let adj = crate::adj::load_from_dir(&adj_path)?; + let index = Arc::new(Index::new(&adj.pod_data, Some(&adj.info.message_levels))); + let addr = "127.0.0.1:50500".parse()?; + let link = BoardLink { + board: BoardId(1), + addr, + local_port: 0, + }; + + // Connect to netcat; this test is ignored by the normal automated suite. + let stream = connection::connect(&link, Duration::from_secs(5)).await?; + let (read_half, _write_half) = stream.into_split(); + let (reporter, mut sinks) = Reporter::channels(1, 1); + let receiving = async { + let result = reader::receive(read_half, link.board, &index, &reporter).await; + // Close the notification channel after either success or failure so printing ends. + drop(reporter); + result + }; + let displaying = async { + let mut frames = 0; + while let Some(notification) = sinks.notifications.recv().await { + println!("Received notification: {notification:?}"); + frames += 1; + } + frames + }; + + // Drain concurrently to respect backpressure; only the manual test has a deadline. + let (end, frames) = tokio::time::timeout(Duration::from_secs(30), async { + tokio::join!(receiving, displaying) + }) + .await?; + assert_eq!(end?, reader::ReceiveEnd::PeerClosed); + assert!( + frames > 0, + "the peer closed without sending a complete frame" + ); + Ok(()) + } +} diff --git a/backend/crates/core/src/net/tcp/board_handle.rs b/backend/crates/core/src/net/tcp/board_handle.rs new file mode 100644 index 000000000..d0c39180b --- /dev/null +++ b/backend/crates/core/src/net/tcp/board_handle.rs @@ -0,0 +1,360 @@ +use super::SendError; +use crate::model::{BoardId, OrderPacket}; +use crate::protocol::{Index, tcp}; +use std::sync::Arc; +use tokio::sync::mpsc; + +/// Access to one board's TCP client, independent of its current socket. +/// +/// Orders are validated and encoded against the shared index before their bytes +/// are queued through a bounded channel. The future connection task owns the +/// receiving end and the socket; this handle owns neither of them. +/// A session writer is implemented; task startup and supervision are still pending. +#[derive(Debug)] +pub struct BoardHandle { + pub(super) board: BoardId, + pub(super) index: Arc, + pub(super) orders: mpsc::Sender>, +} + +impl BoardHandle { + /// The board this handle belongs to. + #[must_use] + pub fn board(&self) -> BoardId { + self.board + } + + /// Validates, encodes and queues an order, waiting if the channel is full. + /// + /// Returns once the encoded bytes are queued, not once they are written to + /// the socket or received by the board. Encoding runs before waiting for + /// capacity and uses the existing protocol checks, not application safety + /// rules. This method does not resolve or check the destination: callers + /// select the board. + /// Use [`super::Handles::send`] to resolve the destination by order ID. + /// + /// # Errors + /// + /// Returns [`SendError::Encode`] if the protocol encoder rejects the ID, + /// packet type or parameters. No bytes are queued on encoding failure, + /// even if the channel is full or closed. + /// Returns [`SendError::Gone`] if encoding succeeds but the receiver has been + /// closed or dropped, including while this call was waiting for queue capacity. + /// This indicates an unavailable consumer, not a TCP disconnection. + pub async fn send(&self, order: OrderPacket) -> Result<(), SendError> { + let bytes = tcp::encode_order(&order, &self.index).map_err(|source| SendError::Encode { + board: self.board, + source, + })?; + self.orders + .send(bytes) + .await + .map_err(|_| SendError::Gone(self.board)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + model::{ + AdjId, Board, MacAddress, Measurement, MeasurementKind, NumericKind, PacketDef, + PacketType, PodData, Value, + }, + protocol::EncodeError, + }; + use std::{ + collections::HashMap, + error::Error as _, + future::{Future, poll_fn}, + net::Ipv4Addr, + task::Poll, + }; + + /// Builds a socket-free channel with empty orders, one u8 order and a data ID. + fn board_channel(capacity: usize) -> (BoardHandle, mpsc::Receiver>) { + let parameter = Measurement { + id: AdjId(100), + alias: "value".to_string(), + name: "Value".to_string(), + display_units: String::new(), + kind: MeasurementKind::Numeric { + wire_type: NumericKind::U8, + }, + protections: vec![], + }; + let packets = [ + (600, PacketType::Order(vec![])), + (601, PacketType::Order(vec![])), + (602, PacketType::Order(vec![parameter])), + (1600, PacketType::Data(vec![])), + ] + .into_iter() + .map(|(id, packet_type)| { + let packet = PacketDef { + id: AdjId(id), + board: BoardId(3), + name: "sample".to_string(), + packet_type, + }; + (packet.id, packet) + }) + .collect(); + let board = Board { + id: BoardId(3), + name: "MOCK".to_string(), + ip: Ipv4Addr::LOCALHOST.into(), + mac: MacAddress([0; 6]), + packets, + }; + let index = Arc::new(Index::new( + &PodData { + boards: HashMap::from([(board.id, board)]), + }, + None, + )); + let (orders, receiver) = mpsc::channel(capacity); + ( + BoardHandle { + board: BoardId(3), + index, + orders, + }, + receiver, + ) + } + + #[tokio::test] + async fn queues_encoded_orders_in_send_order() { + let (handle, mut receiver) = board_channel(2); + let first = OrderPacket::new(AdjId(602), vec![Value::Number(42.0)]); + let second = OrderPacket::new(AdjId(601), vec![]); + + // Check exact little-endian IDs and payload bytes, not just encoder equality. + assert_eq!(handle.send(first).await, Ok(())); + assert_eq!(handle.send(second).await, Ok(())); + + assert_eq!(receiver.try_recv(), Ok(vec![0x5a, 0x02, 42])); + assert_eq!(receiver.try_recv(), Ok(vec![0x59, 0x02])); + assert_eq!(receiver.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + } + + #[tokio::test] + async fn waits_for_capacity_without_dropping_orders() { + let (handle, mut receiver) = board_channel(1); + let first = OrderPacket::new(AdjId(600), vec![]); + let second = OrderPacket::new(AdjId(601), vec![]); + assert_eq!(handle.send(first).await, Ok(())); + + let blocked = handle.send(second); + tokio::pin!(blocked); + + // Poll explicitly so the test proves backpressure without timing sleeps. + poll_fn(|cx| { + assert!(blocked.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + + assert_eq!(receiver.try_recv(), Ok(vec![0x58, 0x02])); + assert_eq!(blocked.await, Ok(())); + assert_eq!(receiver.try_recv(), Ok(vec![0x59, 0x02])); + } + + #[tokio::test] + async fn reports_the_board_when_the_receiver_is_dropped() { + let (handle, receiver) = board_channel(1); + // A valid order must report consumer loss rather than an encoding failure. + drop(receiver); + + let order = OrderPacket::new(AdjId(600), vec![]); + assert_eq!(handle.send(order).await, Err(SendError::Gone(BoardId(3)))); + } + + #[tokio::test] + async fn a_waiting_send_fails_when_the_receiver_closes() { + let (handle, mut receiver) = board_channel(1); + let first = OrderPacket::new(AdjId(600), vec![]); + // Closing the receiver must retain bytes queued before the blocked send. + assert_eq!(handle.send(first).await, Ok(())); + + let blocked = handle.send(OrderPacket::new(AdjId(601), vec![])); + tokio::pin!(blocked); + poll_fn(|cx| { + assert!(blocked.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + + receiver.close(); + assert_eq!(blocked.await, Err(SendError::Gone(BoardId(3)))); + assert_eq!(receiver.try_recv(), Ok(vec![0x58, 0x02])); + assert_eq!( + receiver.try_recv(), + Err(mpsc::error::TryRecvError::Disconnected) + ); + } + + #[tokio::test] + async fn rejects_unknown_and_non_order_ids_when_called_directly() { + let (handle, mut receiver) = board_channel(1); + + // Bypassing Handles must not bypass the protocol's ID and packet-type checks. + for (id, source) in [ + (AdjId(999), EncodeError::UnknownId(AdjId(999))), + ( + AdjId(1600), + EncodeError::WrongPacketType { id: AdjId(1600) }, + ), + ] { + assert_eq!( + handle.send(OrderPacket::new(id, vec![])).await, + Err(SendError::Encode { + board: BoardId(3), + source + }) + ); + assert_eq!(receiver.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + } + } + + #[tokio::test] + async fn rejects_missing_and_extra_parameters_without_queueing_bytes() { + let (handle, mut receiver) = board_channel(1); + + // Both missing and extra parameters must fail before any frame enters the queue. + for values in [vec![], vec![Value::Number(1.0), Value::Number(2.0)]] { + let got = values.len(); + assert_eq!( + handle.send(OrderPacket::new(AdjId(602), values)).await, + Err(SendError::Encode { + board: BoardId(3), + source: EncodeError::WrongFieldCount { expected: 1, got }, + }) + ); + assert_eq!(receiver.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + } + } + + #[tokio::test] + async fn rejects_a_wrong_parameter_type_without_queueing_bytes() { + let (handle, mut receiver) = board_channel(1); + + // A boolean cannot stand in for the numeric field declared in the index. + assert_eq!( + handle + .send(OrderPacket::new(AdjId(602), vec![Value::Bool(true)])) + .await, + Err(SendError::Encode { + board: BoardId(3), + source: EncodeError::TypeMismatch, + }) + ); + assert_eq!(receiver.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + } + + #[tokio::test] + async fn preserves_numeric_range_errors_without_queueing_bytes() { + let (handle, mut receiver) = board_channel(1); + + // Preserve the encoder's bounds and rejected value, including NaN and infinity. + for number in [-1.0, 256.0, f64::NAN, f64::INFINITY] { + let result = handle + .send(OrderPacket::new(AdjId(602), vec![Value::Number(number)])) + .await; + let Err(SendError::Encode { + board, + source: EncodeError::OutOfRange { value, min, max }, + }) = result + else { + unreachable!("these values cannot be encoded as u8"); + }; + assert_eq!(board, BoardId(3)); + assert!(value == number || (value.is_nan() && number.is_nan())); + assert_eq!((min, max), (0.0, 255.0)); + assert_eq!(receiver.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + } + } + + #[tokio::test] + async fn rejects_an_invalid_order_before_waiting_for_capacity() { + let (handle, mut receiver) = board_channel(1); + assert_eq!( + handle.send(OrderPacket::new(AdjId(600), vec![])).await, + Ok(()) + ); + + // Explicit polling proves validation returns immediately despite a full queue. + let invalid = handle.send(OrderPacket::new(AdjId(602), vec![])); + tokio::pin!(invalid); + let result = poll_fn(|cx| match invalid.as_mut().poll(cx) { + Poll::Ready(result) => Poll::Ready(result), + Poll::Pending => unreachable!("invalid orders must not wait for queue capacity"), + }) + .await; + assert_eq!( + result, + Err(SendError::Encode { + board: BoardId(3), + source: EncodeError::WrongFieldCount { + expected: 1, + got: 0 + }, + }) + ); + assert_eq!(receiver.try_recv(), Ok(vec![0x58, 0x02])); + assert_eq!(receiver.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + } + + #[tokio::test] + async fn reports_encoding_failure_before_a_closed_receiver() { + let (handle, receiver) = board_channel(1); + drop(receiver); + + // Encoding precedes channel access, so invalid input remains an input error. + assert_eq!( + handle.send(OrderPacket::new(AdjId(602), vec![])).await, + Err(SendError::Encode { + board: BoardId(3), + source: EncodeError::WrongFieldCount { + expected: 1, + got: 0 + }, + }) + ); + } + + #[tokio::test] + async fn encoding_errors_expose_the_original_protocol_source() { + let (handle, _receiver) = board_channel(1); + + // Callers can inspect the original error through the standard error chain. + let Err(error) = handle.send(OrderPacket::new(AdjId(602), vec![])).await else { + unreachable!("the numeric parameter is missing"); + }; + let source = error + .source() + .and_then(|source| source.downcast_ref::()); + assert_eq!( + source, + Some(&EncodeError::WrongFieldCount { + expected: 1, + got: 0 + }) + ); + assert!(error.to_string().contains("BoardId(3)")); + } + + #[tokio::test] + async fn direct_sends_keep_the_callers_selected_board() { + let (mut handle, mut receiver) = board_channel(1); + + // Direct sends validate encoding, not destination ownership; routing stays in Handles. + handle.board = BoardId(9); + assert_eq!( + handle.send(OrderPacket::new(AdjId(600), vec![])).await, + Ok(()) + ); + assert_eq!(receiver.try_recv(), Ok(vec![0x58, 0x02])); + } +} diff --git a/backend/crates/core/src/net/tcp/connection.rs b/backend/crates/core/src/net/tcp/connection.rs new file mode 100644 index 000000000..ceac1417f --- /dev/null +++ b/backend/crates/core/src/net/tcp/connection.rs @@ -0,0 +1,168 @@ +//! A single TCP connection attempt, independent of session supervision. + +use super::{BoardLink, Error}; +use std::{ + io, + net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}, + time::Duration, +}; +use tokio::{ + net::{TcpSocket, TcpStream}, + time::timeout, +}; + +/// Opens one board's connection, binding its local port before connecting. +/// +/// The remote address selects IPv4 or IPv6. The local address is unspecified +/// so routing selects the interface; local port zero requests an ephemeral port. +/// Kernel keep-alive is disabled: the application keep-alive is an empty order +/// with ID 1, sent later by the session's single writer. TCP_NODELAY is enabled. +/// Zero linger makes closing abortive and may discard unsent data. +/// +/// The timeout bounds the asynchronous connect, not the synchronous socket +/// setup. This function makes no retries, emits no events and performs no +/// application I/O or handshake. The caller owns the returned stream. +/// +/// # Errors +/// +/// Returns [`Error::Connect`] with the board ID and original I/O error if socket +/// creation, configuration, binding or connecting fails. An elapsed timeout is +/// represented by [`io::ErrorKind::TimedOut`]; its pending socket is dropped. +// The connection supervisor will use this once it is implemented. +#[cfg_attr(not(test), allow(dead_code))] +pub(super) async fn connect( + link: &BoardLink, + connect_timeout: Duration, +) -> Result { + let result = async { + let (socket, local_ip) = match link.addr { + SocketAddr::V4(_) => (TcpSocket::new_v4()?, IpAddr::V4(Ipv4Addr::UNSPECIFIED)), + SocketAddr::V6(_) => (TcpSocket::new_v6()?, IpAddr::V6(Ipv6Addr::UNSPECIFIED)), + }; + socket.set_keepalive(false)?; + socket.set_zero_linger()?; + socket.set_nodelay(true)?; + socket.bind(SocketAddr::new(local_ip, link.local_port))?; + + timeout(connect_timeout, socket.connect(link.addr)) + .await + .map_err(io::Error::from)? + } + .await; + + result.map_err(|source| Error::Connect { + board: link.board, + source, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::model::BoardId; + use tokio::net::TcpListener; + + const TEST_TIMEOUT: Duration = Duration::from_secs(2); + + async fn check_loopback_connection( + listener: TcpListener, + ) -> Result<(), Box> { + let link = BoardLink { + board: BoardId(3), + addr: listener.local_addr()?, + local_port: 0, + }; + + // Port zero lets parallel tests obtain distinct client ports from the OS. + let stream = connect(&link, TEST_TIMEOUT).await?; + let (_peer, source) = timeout(TEST_TIMEOUT, listener.accept()).await??; + assert_eq!(stream.peer_addr()?, link.addr); + assert_eq!(source, stream.local_addr()?); + assert_ne!(source.port(), 0); + + // Inspect the connected socket, not just the options requested during setup. + assert!(stream.nodelay()?); + assert_eq!(stream.linger()?, Some(Duration::ZERO)); + Ok(()) + } + + #[tokio::test] + async fn connects_to_an_ipv4_server_with_the_requested_options() + -> Result<(), Box> { + // A loopback listener acts as firmware without requiring a physical board. + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?; + check_loopback_connection(listener).await + } + + #[tokio::test] + async fn connects_to_an_ipv6_server_with_the_requested_options() + -> Result<(), Box> { + // Skip only when the host cannot provide IPv6 loopback; other failures matter. + // Linux EAFNOSUPPORT is not mapped to ErrorKind::Unsupported by std::io. + const LINUX_EAFNOSUPPORT: i32 = 97; + let listener = match TcpListener::bind((Ipv6Addr::LOCALHOST, 0)).await { + Ok(listener) => listener, + Err(error) + if matches!( + error.kind(), + io::ErrorKind::AddrNotAvailable | io::ErrorKind::Unsupported + ) || (cfg!(target_os = "linux") + && error.raw_os_error() == Some(LINUX_EAFNOSUPPORT)) => + { + eprintln!("IPv6 loopback is unavailable; skipping IPv6 connection test: {error}"); + return Ok(()); + } + Err(error) => return Err(error.into()), + }; + check_loopback_connection(listener).await + } + + #[tokio::test] + async fn an_occupied_local_port_reports_the_board_and_bind_error() + -> Result<(), Box> { + // Keep the port occupied throughout the attempt, avoiding a free-port race. + let occupied = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?; + let link = BoardLink { + board: BoardId(7), + addr: occupied.local_addr()?, + local_port: occupied.local_addr()?.port(), + }; + let Err(error) = connect(&link, TEST_TIMEOUT).await else { + unreachable!("binding an occupied local port must fail"); + }; + + // Setup failures must retain both identity and cause, without faulting the pod. + assert!(!error.is_fatal()); + let Error::Connect { board, source } = error else { + unreachable!("a bind failure belongs to the connection attempt"); + }; + assert_eq!(board, BoardId(7)); + assert_eq!(source.kind(), io::ErrorKind::AddrInUse); + Ok(()) + } + + #[tokio::test] + async fn a_non_listening_peer_reports_the_board_and_connect_error() + -> Result<(), Box> { + // Reserve the remote port without listening, so no other test can claim it. + let unavailable = TcpSocket::new_v4()?; + unavailable.bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)))?; + let link = BoardLink { + board: BoardId(9), + addr: unavailable.local_addr()?, + local_port: 0, + }; + let Err(error) = connect(&link, TEST_TIMEOUT).await else { + unreachable!("a bound but non-listening peer cannot accept the connection"); + }; + + // A refused connection is a retryable network failure, not a decode fault. + assert!(!error.is_fatal()); + let Error::Connect { board, source } = error else { + unreachable!("a refused connection must return Error::Connect"); + }; + assert_eq!(board, BoardId(9)); + assert_eq!(source.kind(), io::ErrorKind::ConnectionRefused); + Ok(()) + } +} diff --git a/backend/crates/core/src/net/tcp/error.rs b/backend/crates/core/src/net/tcp/error.rs new file mode 100644 index 000000000..3732ac3e7 --- /dev/null +++ b/backend/crates/core/src/net/tcp/error.rs @@ -0,0 +1,127 @@ +use crate::{ + model::{AdjId, BoardId}, + protocol::{DecodeError, EncodeError}, +}; + +/// Why an order could not be handed to a writer. +/// +/// Returned to the caller, not reported through [`crate::net::Reporter`]: it +/// says something about the order that was asked for, not about the state +/// of a connection. [`Error`] is the one that travels on the error +/// channel. +#[derive(Debug, PartialEq, thiserror::Error)] +pub enum SendError { + /// No order with this id is declared in the loaded ADJ: the id is absent + /// or belongs to another packet type. Returned by [`super::Handles::send`] + /// during routing; direct board sends retain the encoder's error in `Encode`. + #[error("no order is declared with id {0:?}")] + UnknownId(AdjId), + /// The order resolves to a board the config did not list, so no + /// client was ever opened for it. + #[error("board {0:?} has no TCP client: the config does not list it")] + NoClient(BoardId), + /// The order could not be encoded against the shared index. + /// No bytes were queued; this is not a connection failure. + #[error("board {board:?}: could not encode order")] + Encode { + /// The board selected for this send. + board: BoardId, + /// The original protocol validation or encoding error. + #[source] + source: EncodeError, + }, + /// The order channel's receiver has been closed or dropped. + /// This is not the same as a disconnected TCP socket. + #[error("board {0:?}: its connection task is gone")] + Gone(BoardId), +} + +/// The boards a fault broadcast did not reach. +/// +/// Never empty: a broadcast that reached everybody returns `Ok`. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +#[error("the fault did not reach these boards: {0:?}")] +pub struct Unreachable(pub Vec); + +/// Something that went wrong on a board's TCP connection. +/// +/// Every variant names its board, which is the whole identity of a link: +/// one connection per board. +#[derive(Debug, thiserror::Error)] +pub enum Error { + /// The connection could not be opened. + #[error("board {board:?}: could not connect")] + Connect { + /// The board that was being connected to. + board: BoardId, + /// The underlying I/O failure. + #[source] + source: std::io::Error, + }, + + /// A socket read failed, or the peer closed in the middle of a frame. + #[error("board {board:?}: could not read")] + Read { + /// The board whose connection could not be read. + board: BoardId, + /// The underlying I/O failure, including an incomplete-frame EOF. + #[source] + source: std::io::Error, + }, + + /// The pending frame could not be decoded after exceeding the framing + /// limit. Under the protocol assumptions this indicates an ADJ mismatch. + /// Fatal: see [`Error::is_fatal`]. + #[error("board {board:?}: could not decode a frame")] + Decode { + /// The board whose frame failed. + board: BoardId, + /// What [`crate::protocol`] said about it. + #[source] + source: DecodeError, + }, + + /// An order or a keep-alive could not be written. + #[error("board {board:?}: could not write")] + Write { + /// The board that was being written to. + board: BoardId, + /// The underlying I/O failure, including `TimedOut` for an elapsed frame deadline. + #[source] + source: std::io::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 its other connections, still serving the + /// frontend, still recording — so that whoever is operating the pod + /// can see what happened. Nothing here ever calls + /// `std::process::exit`. + /// + /// Only [`Error::Decode`] qualifies, and this is the one place where + /// the behaviour deliberately departs from the Go backend. Go treated + /// an undecodable frame as a connection that had fallen out of step, + /// and reconnecting was the remedy. But nothing corrupts a frame in + /// flight — TCP checksums and retransmits — so a frame that will not + /// decode means the loaded ADJ does not match what the firmware is + /// sending, and reopening the connection cannot fix that. The stream + /// still cannot be resynchronised mid-frame, so the connection is + /// closed and reopened anyway; what changes is that the pod is also + /// sent to fault instead of carrying on as if a retry would help. + /// + /// [`Error::Connect`], [`Error::Read`] and [`Error::Write`] are not fatal: a board + /// that is not up yet, or a write that failed on a connection that + /// just dropped, is exactly what the reconnection loop with its + /// backoff is for. Losing a board for good is reported through + /// [`crate::net::Event::Disconnected`], not through this. + /// EOF within a partial frame is a read failure, not proof of an ADJ mismatch. + pub fn is_fatal(&self) -> bool { + match self { + Error::Decode { .. } => true, + Error::Connect { .. } | Error::Read { .. } | Error::Write { .. } => false, + } + } +} diff --git a/backend/crates/core/src/net/tcp/framing.rs b/backend/crates/core/src/net/tcp/framing.rs new file mode 100644 index 000000000..8a1f57994 --- /dev/null +++ b/backend/crates/core/src/net/tcp/framing.rs @@ -0,0 +1,342 @@ +//! Transactional frame extraction from one TCP session's accumulated bytes. + +use crate::protocol::{ + DecodeError, Index, + tcp::{self, TcpFrame}, +}; +use bytes::{Buf, BytesMut}; +use std::io::Cursor; + +/// Extracts one complete frame, advancing `pending` only after a successful decode. +/// +/// Append incoming bytes with `BytesMut::extend_from_slice`, then call this +/// repeatedly until it returns `Ok(None)`. That result means the buffer is empty +/// or the next frame needs more bytes. Each connection session must own its own +/// buffer; never carry pending bytes into a reconnected session. +/// +/// All decode errors are deferred while the pending suffix is at most 374 bytes, +/// assuming well-formed firmware output. The limit is checked after decoding: +/// a larger buffer may contain many valid frames that must be extracted first. +/// +/// # Errors +/// +/// Returns the decoder's error if no frame can be extracted and more than 374 +/// bytes remain. The buffer stays unchanged; the socket owner must discard the +/// session and reconnect rather than try to resynchronise within the stream. +pub(super) fn next_frame( + pending: &mut BytesMut, + index: &Index, +) -> Result, DecodeError> { + // Max length of a message frame + const MAX_FRAME_LEN: usize = 374; + + if pending.is_empty() { + return Ok(None); + } + + let mut cursor = Cursor::new(&pending[..]); + match tcp::decode_next(&mut cursor, index) { + Ok(frame) => { + let consumed = cursor.position() as usize; + pending.advance(consumed); + // Everything ok, returns the frame + Ok(Some(frame)) + } + // if the function returns an error and the pending bytes + // are greater than the max frame sneds an error + Err(error) if pending.len() > MAX_FRAME_LEN => Err(error), + Err(_) => Ok(None), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::model::{ + AdjId, Board, BoardId, MacAddress, Measurement, MeasurementKind, MessageLevel, + MessagePacket, NumericKind, PacketDef, PacketType, PodData, ProtectionPacket, + ShortTimestamp, Value, + }; + use crate::protocol::packet::{message, protection}; + use std::{collections::HashMap, net::IpAddr}; + + fn sample_frames() -> (Index, Vec<(TcpFrame, Vec)>) { + let packets = [ + (100, MeasurementKind::Boolean), + ( + 101, + MeasurementKind::Numeric { + wire_type: NumericKind::F64, + }, + ), + ] + .into_iter() + .map(|(id, kind)| { + let id = AdjId(id); + let definition = PacketDef { + id: AdjId((1 << 13) | id.0), + board: BoardId(1), + name: "protection".to_string(), + packet_type: PacketType::Protection(Measurement { + id, + alias: "watched".to_string(), + name: "Watched".to_string(), + display_units: String::new(), + kind, + protections: vec![], + }), + }; + (definition.id, definition) + }) + .collect(); + let board = Board { + id: BoardId(1), + name: "MOCK".to_string(), + ip: IpAddr::from([127, 0, 0, 1]), + mac: MacAddress([0; 6]), + packets, + }; + let pod = PodData { + boards: HashMap::from([(board.id, board)]), + }; + let levels = HashMap::from([(AdjId(10), MessageLevel::Fault)]); + let index = Index::new(&pod, Some(&levels)); + let timestamp = ShortTimestamp { + hour: 12, + minute: 30, + second: 0, + subsecond: 0, + }; + let frames = [ + TcpFrame::Protection(ProtectionPacket { + id: AdjId((1 << 13) | 100), + timestamp, + value: Value::Bool(true), + }), + TcpFrame::Message(MessagePacket { + id: AdjId(10), + timestamp, + origin: "LCU".to_string(), + message: "temperature \u{00b0}C".to_string(), + }), + TcpFrame::Protection(ProtectionPacket { + id: AdjId((1 << 13) | 101), + timestamp, + value: Value::Number(42.25), + }), + TcpFrame::Message(MessagePacket { + id: AdjId(10), + timestamp, + origin: String::new(), + message: String::new(), + }), + ] + .into_iter() + .map(|frame| { + let encoded = match &frame { + TcpFrame::Protection(packet) => protection::encode(packet, &index), + TcpFrame::Message(packet) => message::encode(packet, &index), + }; + let Ok(bytes) = encoded else { + unreachable!("the sample frames match their index definitions"); + }; + (frame, bytes) + }) + .collect(); + + (index, frames) + } + + #[test] + fn an_empty_buffer_needs_more_bytes() { + let mut pending = BytesMut::new(); + assert_eq!(next_frame(&mut pending, &Index::default()), Ok(None)); + assert!(pending.is_empty()); + } + + #[test] + fn extracts_mixed_frames_with_adj_dependent_lengths() { + let (index, frames) = sample_frames(); + assert_eq!(frames[0].1.len(), 9); + assert_eq!(frames[2].1.len(), 16); + let mut pending = BytesMut::new(); + for (_, bytes) in &frames { + pending.extend_from_slice(bytes); + } + + for (frame, bytes) in &frames { + let before = pending.len(); + assert_eq!(next_frame(&mut pending, &index), Ok(Some(frame.clone()))); + assert_eq!(pending.len(), before - bytes.len()); + } + assert_eq!(next_frame(&mut pending, &index), Ok(None)); + assert!(pending.is_empty()); + } + + #[test] + fn preserves_every_partial_frame_until_its_remaining_bytes_arrive() { + let (index, frames) = sample_frames(); + for (frame, bytes) in &frames { + for split in 0..bytes.len() { + let mut pending = BytesMut::from(&bytes[..split]); + assert_eq!(next_frame(&mut pending, &index), Ok(None)); + assert_eq!(&pending[..], &bytes[..split]); + + pending.extend_from_slice(&bytes[split..]); + assert_eq!(next_frame(&mut pending, &index), Ok(Some(frame.clone()))); + assert!(pending.is_empty()); + } + } + } + + #[test] + fn extracts_a_mixed_stream_received_one_byte_at_a_time() -> Result<(), DecodeError> { + let (index, frames) = sample_frames(); + let mut pending = BytesMut::new(); + let mut received = Vec::new(); + for (_, bytes) in &frames { + for byte in bytes { + pending.extend_from_slice(&[*byte]); + while let Some(frame) = next_frame(&mut pending, &index)? { + received.push(frame); + } + } + } + + let expected: Vec<_> = frames.into_iter().map(|(frame, _)| frame).collect(); + assert_eq!(received, expected); + assert!(pending.is_empty()); + Ok(()) + } + + #[test] + fn extracts_a_complete_frame_and_keeps_the_next_partial_frame() { + let (index, frames) = sample_frames(); + let (first, first_bytes) = &frames[0]; + let (second, second_bytes) = &frames[1]; + let split = second_bytes.len() - 1; + let mut pending = BytesMut::from(&first_bytes[..]); + pending.extend_from_slice(&second_bytes[..split]); + + assert_eq!(next_frame(&mut pending, &index), Ok(Some(first.clone()))); + assert_eq!(next_frame(&mut pending, &index), Ok(None)); + assert_eq!(&pending[..], &second_bytes[..split]); + + pending.extend_from_slice(&second_bytes[split..]); + assert_eq!(next_frame(&mut pending, &index), Ok(Some(second.clone()))); + assert!(pending.is_empty()); + } + + #[test] + fn defers_non_eof_errors_until_the_pending_suffix_exceeds_the_limit() { + let (index, _) = sample_frames(); + let mut invalid_utf8 = AdjId(10).0.to_le_bytes().to_vec(); + invalid_utf8.extend([0, 0, 0, 0, 0xff, 0, 0]); + let mut unterminated = AdjId(10).0.to_le_bytes().to_vec(); + unterminated.extend([0; 4]); + unterminated.extend([b'x'; 48]); + + for (bytes, expected_error) in [ + (vec![0xff, 0xff], DecodeError::UnknownId(AdjId(u16::MAX))), + (invalid_utf8, DecodeError::InvalidUtf8(vec![0xff])), + (unterminated, DecodeError::UnterminatedString { max: 48 }), + ] { + let mut pending = BytesMut::from(&bytes[..]); + assert_eq!(next_frame(&mut pending, &index), Ok(None)); + assert_eq!(&pending[..], &bytes); + + pending.resize(374, 0); + let before = pending.clone(); + assert_eq!(next_frame(&mut pending, &index), Ok(None)); + assert_eq!(pending, before); + + pending.extend_from_slice(&[0]); + let before = pending.clone(); + assert_eq!(next_frame(&mut pending, &index), Err(expected_error)); + assert_eq!(pending, before); + } + } + + #[test] + fn accepts_a_maximum_length_message_once_its_last_byte_arrives() { + let (index, _) = sample_frames(); + let packet = MessagePacket { + id: AdjId(10), + timestamp: ShortTimestamp { + hour: 0, + minute: 0, + second: 0, + subsecond: 0, + }, + origin: "o".repeat(47), + message: "m".repeat(319), + }; + let Ok(bytes) = message::encode(&packet, &index) else { + unreachable!("both strings fit their wire limits including NUL"); + }; + assert_eq!(bytes.len(), 374); + let mut pending = BytesMut::from(&bytes[..373]); + assert_eq!(next_frame(&mut pending, &index), Ok(None)); + assert_eq!(&pending[..], &bytes[..373]); + + pending.extend_from_slice(&bytes[373..]); + assert_eq!( + next_frame(&mut pending, &index), + Ok(Some(TcpFrame::Message(packet))) + ); + assert!(pending.is_empty()); + } + + #[test] + fn drains_a_valid_burst_larger_than_the_frame_limit() { + let (index, frames) = sample_frames(); + let mut pending = BytesMut::new(); + for _ in 0..32 { + for (_, bytes) in &frames { + pending.extend_from_slice(bytes); + } + } + assert!(pending.len() > 374); + + for _ in 0..32 { + for (frame, _) in &frames { + assert_eq!(next_frame(&mut pending, &index), Ok(Some(frame.clone()))); + } + } + assert_eq!(next_frame(&mut pending, &index), Ok(None)); + assert!(pending.is_empty()); + } + + #[test] + fn delivers_valid_frames_before_rejecting_an_oversized_invalid_suffix() { + let (index, frames) = sample_frames(); + let (frame, bytes) = &frames[0]; + let mut pending = BytesMut::from(&bytes[..]); + let invalid = vec![0xff; 375]; + pending.extend_from_slice(&invalid); + + assert_eq!(next_frame(&mut pending, &index), Ok(Some(frame.clone()))); + assert_eq!( + next_frame(&mut pending, &index), + Err(DecodeError::UnknownId(AdjId(u16::MAX))) + ); + assert_eq!(&pending[..], &invalid); + } + + #[test] + fn independent_buffers_do_not_share_partial_frames() { + let (index, frames) = sample_frames(); + let (frame, bytes) = &frames[0]; + let mut first = BytesMut::from(&bytes[..3]); + let mut second = BytesMut::from(&bytes[..]); + + assert_eq!(next_frame(&mut first, &index), Ok(None)); + assert_eq!(next_frame(&mut second, &index), Ok(Some(frame.clone()))); + assert!(second.is_empty()); + assert_eq!(&first[..], &bytes[..3]); + + first.extend_from_slice(&bytes[3..]); + assert_eq!(next_frame(&mut first, &index), Ok(Some(frame.clone()))); + assert!(first.is_empty()); + } +} diff --git a/backend/crates/core/src/net/tcp/handles.rs b/backend/crates/core/src/net/tcp/handles.rs new file mode 100644 index 000000000..571b7b531 --- /dev/null +++ b/backend/crates/core/src/net/tcp/handles.rs @@ -0,0 +1,279 @@ +use super::{BoardHandle, SendError}; +use crate::model::{BoardId, OrderPacket, PacketType}; +use crate::protocol::Index; +use std::{collections::HashMap, sync::Arc}; + +/// Access to the configured boards' TCP clients. +/// +/// A board remains in this collection while disconnected. Connection state +/// belongs to its task, not to the presence of its handle in this map. +/// The shared, read-only index resolves order IDs to their destination boards. +#[derive(Debug)] +pub struct Handles { + boards: HashMap, + index: Arc, +} + +impl Handles { + /// Returns the handle for a configured board, regardless of connection state. + #[must_use] + pub fn board(&self, board: BoardId) -> Option<&BoardHandle> { + self.boards.get(&board) + } + + /// Resolves an order's destination through the index and queues it there. + /// + /// Only IDs declared as orders are accepted. The destination handle validates + /// and encodes the parameters before queueing the bytes. Like + /// [`BoardHandle::send`], this waits for queue capacity and returns once the + /// bytes are queued, not written to TCP. + /// It does not broadcast reserved IDs or send anything to other boards. + /// + /// # Errors + /// + /// Returns [`SendError::UnknownId`] if the ID is absent or declared as a + /// different packet type, [`SendError::NoClient`] if the destination board + /// has no configured handle, [`SendError::Encode`] if encoding fails, or + /// [`SendError::Gone`] if encoding succeeds but its receiver is closed. + pub async fn send(&self, order: OrderPacket) -> Result<(), SendError> { + let definition = self + .index + .packet(order.id) + .filter(|definition| matches!(&definition.packet_type, PacketType::Order(_))) + .ok_or(SendError::UnknownId(order.id))?; + let handle = self + .board(definition.board) + .ok_or(SendError::NoClient(definition.board))?; + + handle.send(order).await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + model::{ + AdjId, Board, MacAddress, Measurement, MeasurementKind, MessageLevel, NumericKind, + PacketDef, PodData, Value, + }, + protocol::EncodeError, + }; + use std::net::IpAddr; + use tokio::sync::mpsc; + + /// Shares one index between routing and both boards' encoding channels. + fn sample_handles() -> (Handles, mpsc::Receiver>, mpsc::Receiver>) { + let boards = [(1, 600), (3, 601)] + .into_iter() + .map(|(board_id, order_id)| { + let order = PacketDef { + id: AdjId(order_id), + board: BoardId(board_id), + name: "order".to_string(), + packet_type: PacketType::Order(if board_id == 1 { + vec![Measurement { + id: AdjId(100), + alias: "value".to_string(), + name: "Value".to_string(), + display_units: String::new(), + kind: MeasurementKind::Numeric { + wire_type: NumericKind::U8, + }, + protections: vec![], + }] + } else { + vec![] + }), + }; + let data = PacketDef { + id: AdjId(order_id + 1000), + board: BoardId(board_id), + name: "data".to_string(), + packet_type: PacketType::Data(vec![]), + }; + let board = Board { + id: BoardId(board_id), + name: "MOCK".to_string(), + ip: IpAddr::from([127, 0, 0, 1]), + mac: MacAddress([0; 6]), + packets: HashMap::from([(order.id, order), (data.id, data)]), + }; + (board.id, board) + }) + .collect(); + let message_levels = HashMap::from([(AdjId(2), MessageLevel::Fault)]); + let index = Arc::new(Index::new(&PodData { boards }, Some(&message_levels))); + let (first_orders, first_receiver) = mpsc::channel(2); + let (third_orders, third_receiver) = mpsc::channel(2); + let handles = Handles { + index: Arc::clone(&index), + boards: HashMap::from([ + ( + BoardId(1), + BoardHandle { + board: BoardId(1), + index: Arc::clone(&index), + orders: first_orders, + }, + ), + ( + BoardId(3), + BoardHandle { + board: BoardId(3), + index: Arc::clone(&index), + orders: third_orders, + }, + ), + ]), + }; + + (handles, first_receiver, third_receiver) + } + + #[test] + fn finds_each_configured_board_without_a_socket() { + let (handles, _first_receiver, _third_receiver) = sample_handles(); + + // Lookup and encoding share one immutable index, without opening sockets. + assert_eq!( + handles.board(BoardId(1)).map(BoardHandle::board), + Some(BoardId(1)) + ); + assert_eq!( + handles.board(BoardId(3)).map(BoardHandle::board), + Some(BoardId(3)) + ); + for handle in handles.boards.values() { + assert!(Arc::ptr_eq(&handles.index, &handle.index)); + } + } + + #[test] + fn returns_none_for_an_unconfigured_board() { + let (handles, _first_receiver, _third_receiver) = sample_handles(); + + // The index alone does not create handles for unconfigured boards. + assert!(handles.board(BoardId(2)).is_none()); + } + + #[tokio::test] + async fn routes_each_order_only_to_its_declared_board() { + let (handles, mut first_receiver, mut third_receiver) = sample_handles(); + let first = OrderPacket::new(AdjId(600), vec![Value::Number(42.0)]); + let third = OrderPacket::new(AdjId(601), vec![]); + + // Routing must deliver encoded bytes only to the declared destination. + assert_eq!(handles.send(first).await, Ok(())); + assert_eq!(first_receiver.try_recv(), Ok(vec![0x58, 0x02, 42])); + assert_eq!( + third_receiver.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + ); + + assert_eq!(handles.send(third).await, Ok(())); + assert_eq!(third_receiver.try_recv(), Ok(vec![0x59, 0x02])); + assert_eq!( + first_receiver.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + ); + } + + #[tokio::test] + async fn rejects_an_unknown_id_without_queueing_anything() { + let (handles, mut first_receiver, mut third_receiver) = sample_handles(); + let order = OrderPacket::new(AdjId(999), vec![]); + + // Unknown IDs fail during routing, before any board channel is touched. + assert_eq!( + handles.send(order).await, + Err(SendError::UnknownId(AdjId(999))) + ); + assert_eq!( + first_receiver.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + ); + assert_eq!( + third_receiver.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + ); + } + + #[tokio::test] + async fn rejects_non_order_ids_without_queueing_anything() { + let (handles, mut first_receiver, mut third_receiver) = sample_handles(); + + // Message IDs also resolve through Index, but are not outbound orders. + for id in [AdjId(1600), AdjId(1601), AdjId(2)] { + let order = OrderPacket::new(id, vec![]); + assert_eq!(handles.send(order).await, Err(SendError::UnknownId(id))); + } + + assert_eq!( + first_receiver.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + ); + assert_eq!( + third_receiver.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + ); + } + + #[tokio::test] + async fn reports_an_unconfigured_destination_without_using_another_board() { + let (mut handles, mut first_receiver, mut third_receiver) = sample_handles(); + handles.boards.remove(&BoardId(3)); + let order = OrderPacket::new(AdjId(601), vec![]); + + // A missing destination must not fall back to an unrelated board's queue. + assert_eq!( + handles.send(order).await, + Err(SendError::NoClient(BoardId(3))) + ); + assert_eq!( + first_receiver.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + ); + assert_eq!( + third_receiver.try_recv(), + Err(mpsc::error::TryRecvError::Disconnected) + ); + } + + #[tokio::test] + async fn reports_the_destination_when_its_receiver_is_gone() { + let (handles, mut first_receiver, third_receiver) = sample_handles(); + drop(third_receiver); + let order = OrderPacket::new(AdjId(601), vec![]); + + // Valid encoding must preserve Gone from the selected board's closed channel. + assert_eq!(handles.send(order).await, Err(SendError::Gone(BoardId(3)))); + assert_eq!( + first_receiver.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + ); + } + + #[tokio::test] + async fn returns_the_destination_encoding_error_without_queueing_anything() { + let (handles, mut first_receiver, mut third_receiver) = sample_handles(); + + // Handles must propagate protocol validation failures from the destination handle. + let order = OrderPacket::new(AdjId(600), vec![Value::Bool(true)]); + assert_eq!( + handles.send(order).await, + Err(SendError::Encode { + board: BoardId(1), + source: EncodeError::TypeMismatch, + }) + ); + assert_eq!( + first_receiver.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + ); + assert_eq!( + third_receiver.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + ); + } +} diff --git a/backend/crates/core/src/net/tcp/reader.rs b/backend/crates/core/src/net/tcp/reader.rs new file mode 100644 index 000000000..cffe91c49 --- /dev/null +++ b/backend/crates/core/src/net/tcp/reader.rs @@ -0,0 +1,489 @@ +//! Reading and reporting ordinary frames from one established TCP session. + +use super::{Error, framing}; +use crate::{ + model::{BoardId, Packet}, + net::{Event, Reporter}, + protocol::{Index, tcp::TcpFrame}, +}; +use bytes::BytesMut; +use std::io; +use tokio::{io::AsyncReadExt, net::tcp::OwnedReadHalf}; + +/// Why a session reader stopped without a transport error. +#[derive(Debug, PartialEq, Eq)] +pub(super) enum ReceiveEnd { + /// The peer closed its write side at a frame boundary. + PeerClosed, + /// Publishing a frame failed because the notification consumer is gone. + /// The owner must stop rather than reconnect to an absent consumer. + ConsumerClosed, +} + +/// Reads ordinary frames and publishes each one through `reporter` in order. +/// +/// The caller must complete any application handshake before handing over the +/// read half. The buffer is local to this invocation and is never reused across +/// sessions. The connection supplies board identity, including for global +/// message IDs whose index definitions do not belong to a real board. +/// +/// A full notification channel applies backpressure instead of dropping frames. +/// Consumer closure is detected when publishing, not while an idle socket read +/// is pending. There is no read timeout: silence is not a protocol violation. +/// +/// Only `Received` notifications are emitted. The session owner handles logging, +/// error reporting, connection events and disposal of both socket halves. This +/// function does not reconnect, write orders or decide to fault the pod. +/// +/// # Errors +/// +/// Returns [`Error::Read`] for socket failures or EOF in a partial frame, and +/// [`Error::Decode`] for a terminal framing error. Completed preceding frames +/// have already been published. The read half is dropped when the function ends. +// The session supervisor will use this once it is implemented. +#[cfg_attr(not(test), allow(dead_code))] +pub(super) async fn receive( + mut reader: OwnedReadHalf, + board: BoardId, + index: &Index, + reporter: &Reporter, +) -> Result { + // pending bytes to be processed by framing function + let mut pending = BytesMut::new(); + // Chunk of the maximus size of a TCP stream (of bytes) + let mut chunk = [0; 1024]; + + loop { + // returns the number of bytes read + let read = reader + // modifies the chunk with the readed data + .read(&mut chunk) + .await + .map_err(|source| Error::Read { board, source })?; + + // if no bytes are readed + if read == 0 { + return if pending.is_empty() { + Ok(ReceiveEnd::PeerClosed) + } else { + Err(Error::Read { + board, + source: io::Error::new( + io::ErrorKind::UnexpectedEof, + "peer closed with an incomplete TCP frame", + ), + }) + }; + } + + // adds only the bytes readed (modified) of the chunk to the pending ones + pending.extend_from_slice(&chunk[..read]); + while let Some(frame) = framing::next_frame(&mut pending, index) + .map_err(|source| Error::Decode { board, source })? + { + let packet = match frame { + TcpFrame::Message(message) => Packet::Message(message), + TcpFrame::Protection(protection) => Packet::Protection(protection), + }; + + // if the reporter succeeds notifying the event + if reporter + .notify(board, Event::Received(packet)) + .await + .is_err() + { + return Ok(ReceiveEnd::ConsumerClosed); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + model::{ + AdjId, Board, MacAddress, Measurement, MeasurementKind, MessageLevel, MessagePacket, + PacketDef, PacketType, PodData, ProtectionPacket, ShortTimestamp, Value, + }, + protocol::{ + DecodeError, + packet::{message, protection}, + }, + }; + use std::{ + collections::HashMap, + future::{Future, poll_fn}, + net::Ipv4Addr, + task::Poll, + time::Duration, + }; + use tokio::{ + io::AsyncWriteExt, + net::{TcpListener, TcpStream, tcp::OwnedWriteHalf}, + time::timeout, + }; + + const TEST_TIMEOUT: Duration = Duration::from_secs(2); + + fn sample_frames() -> (Index, Vec<(Packet, Vec)>) { + let definition = PacketDef { + id: AdjId((1 << 13) | 100), + board: BoardId(7), + name: "protection".to_string(), + packet_type: PacketType::Protection(Measurement { + id: AdjId(100), + alias: "watched".to_string(), + name: "Watched".to_string(), + display_units: String::new(), + kind: MeasurementKind::Boolean, + protections: vec![], + }), + }; + let board = Board { + id: BoardId(7), + name: "MOCK".to_string(), + ip: Ipv4Addr::LOCALHOST.into(), + mac: MacAddress([0; 6]), + packets: HashMap::from([(definition.id, definition)]), + }; + let pod = PodData { + boards: HashMap::from([(board.id, board)]), + }; + let levels = HashMap::from([(AdjId(3), MessageLevel::Warning)]); + let index = Index::new(&pod, Some(&levels)); + let timestamp = ShortTimestamp { + hour: 12, + minute: 30, + second: 0, + subsecond: 0, + }; + let frames = [ + Packet::Message(MessagePacket { + id: AdjId(3), + timestamp, + origin: "mock".to_string(), + message: "hello".to_string(), + }), + Packet::Protection(ProtectionPacket { + id: AdjId((1 << 13) | 100), + timestamp, + value: Value::Bool(true), + }), + ] + .into_iter() + .map(|packet| { + let encoded = match &packet { + Packet::Message(packet) => message::encode(packet, &index), + Packet::Protection(packet) => protection::encode(packet, &index), + _ => unreachable!("the fixture contains only inbound TCP frames"), + }; + let Ok(bytes) = encoded else { + unreachable!("the sample packets match their index definitions"); + }; + (packet, bytes) + }) + .collect(); + (index, frames) + } + + async fn connected_pair() + -> Result<(OwnedReadHalf, OwnedWriteHalf, TcpStream), Box> { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?; + let client = timeout(TEST_TIMEOUT, TcpStream::connect(listener.local_addr()?)).await??; + let (peer, _) = timeout(TEST_TIMEOUT, listener.accept()).await??; + let (reader, writer) = client.into_split(); + Ok((reader, writer, peer)) + } + + #[tokio::test] + async fn reports_a_mixed_burst_in_order_with_the_connection_board() + -> Result<(), Box> { + let (index, frames) = sample_frames(); + let (reader, _writer, mut peer) = connected_pair().await?; + let (reporter, mut sinks) = Reporter::channels(1, 1); + let mut bytes = Vec::new(); + for _ in 0..64 { + for (_, encoded) in &frames { + bytes.extend_from_slice(encoded); + } + } + assert!(bytes.len() > 1024); + + // Consume concurrently: a one-slot channel must not stall the whole test. + let (sent, received, ()) = timeout(TEST_TIMEOUT, async { + tokio::join!( + async { + peer.write_all(&bytes).await?; + peer.shutdown().await + }, + receive(reader, BoardId(42), &index, &reporter), + async { + for _ in 0..64 { + for (expected, _) in &frames { + let Some(notification) = sinks.notifications.recv().await else { + unreachable!("the reporter remains alive while receiving"); + }; + assert_eq!(notification.board, BoardId(42)); + let Event::Received(packet) = notification.event else { + unreachable!("the reader emits only Received notifications"); + }; + assert_eq!(&packet, expected); + } + } + } + ) + }) + .await?; + sent?; + assert_eq!(received?, ReceiveEnd::PeerClosed); + assert!(sinks.notifications.try_recv().is_err()); + assert!(sinks.errors.try_recv().is_err()); + Ok(()) + } + + #[tokio::test] + async fn waits_for_more_bytes_without_publishing_a_partial_frame() + -> Result<(), Box> { + let (index, frames) = sample_frames(); + let (mut reader, _writer, mut peer) = connected_pair().await?; + let (reporter, mut sinks) = Reporter::channels(1, 1); + let (expected, bytes) = &frames[0]; + + // Make a prefix readable before polling the reader, then keep the peer open. + peer.write_all(&bytes[..7]).await?; + timeout(TEST_TIMEOUT, reader.peek(&mut [0; 1])).await??; + let reading = receive(reader, BoardId(42), &index, &reporter); + tokio::pin!(reading); + poll_fn(|cx| { + assert!(reading.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + assert!(sinks.notifications.try_recv().is_err()); + + // Completing the frame and closing the write side must produce exactly one packet. + peer.write_all(&bytes[7..]).await?; + peer.shutdown().await?; + assert_eq!( + timeout(TEST_TIMEOUT, reading).await??, + ReceiveEnd::PeerClosed + ); + let Ok(notification) = sinks.notifications.try_recv() else { + unreachable!("the completed frame was published"); + }; + let Event::Received(packet) = notification.event else { + unreachable!("the reader emits only Received notifications"); + }; + assert_eq!(&packet, expected); + assert!(sinks.notifications.try_recv().is_err()); + Ok(()) + } + + #[tokio::test] + async fn an_empty_peer_closure_is_normal_and_emits_no_events() + -> Result<(), Box> { + let (reader, _writer, mut peer) = connected_pair().await?; + let (reporter, mut sinks) = Reporter::channels(1, 1); + + // Connection lifecycle notifications belong to the future supervisor, not this reader. + peer.shutdown().await?; + let end = timeout( + TEST_TIMEOUT, + receive(reader, BoardId(42), &Index::default(), &reporter), + ) + .await??; + assert_eq!(end, ReceiveEnd::PeerClosed); + assert!(sinks.notifications.try_recv().is_err()); + assert!(sinks.errors.try_recv().is_err()); + Ok(()) + } + + #[tokio::test] + async fn eof_in_a_partial_frame_is_a_nonfatal_read_error_after_the_valid_prefix() + -> Result<(), Box> { + let (index, frames) = sample_frames(); + let (reader, _writer, mut peer) = connected_pair().await?; + let (reporter, mut sinks) = Reporter::channels(1, 1); + + // A complete message must survive a following protection cut short by EOF. + peer.write_all(&frames[0].1).await?; + peer.write_all(&frames[1].1[..8]).await?; + peer.shutdown().await?; + let Err(error) = timeout( + TEST_TIMEOUT, + receive(reader, BoardId(42), &index, &reporter), + ) + .await? + else { + unreachable!("EOF cannot complete the pending protection value"); + }; + assert!(!error.is_fatal()); + let Error::Read { board, source } = error else { + unreachable!("a truncated connection is a read failure, not an ADJ mismatch"); + }; + assert_eq!(board, BoardId(42)); + assert_eq!(source.kind(), io::ErrorKind::UnexpectedEof); + let Ok(notification) = sinks.notifications.try_recv() else { + unreachable!("the complete message preceding EOF was published"); + }; + let Event::Received(packet) = notification.event else { + unreachable!("the reader emits only Received notifications"); + }; + assert_eq!(packet, frames[0].0); + assert!(sinks.errors.try_recv().is_err()); + Ok(()) + } + + #[tokio::test] + async fn an_oversized_invalid_suffix_is_fatal_after_publishing_the_valid_prefix() + -> Result<(), Box> { + let (index, frames) = sample_frames(); + let (reader, _writer, mut peer) = connected_pair().await?; + let (reporter, mut sinks) = Reporter::channels(1, 1); + + // A terminal suffix must not hide a valid message that arrived before it. + let mut bytes = frames[0].1.clone(); + bytes.extend([0xff; 375]); + peer.write_all(&bytes).await?; + let Err(error) = timeout( + TEST_TIMEOUT, + receive(reader, BoardId(42), &index, &reporter), + ) + .await? + else { + unreachable!("an undecodable suffix above 374 bytes must terminate the session"); + }; + assert!(error.is_fatal()); + let Error::Decode { board, source } = error else { + unreachable!("framing errors must retain the protocol decoder error"); + }; + assert_eq!(board, BoardId(42)); + assert_eq!(source, DecodeError::UnknownId(AdjId(u16::MAX))); + let Ok(notification) = sinks.notifications.try_recv() else { + unreachable!("the complete message preceding the invalid suffix was published"); + }; + let Event::Received(packet) = notification.event else { + unreachable!("the reader emits only Received notifications"); + }; + assert_eq!(packet, frames[0].0); + assert!(sinks.errors.try_recv().is_err()); + Ok(()) + } + + #[tokio::test] + async fn a_socket_reset_preserves_the_io_error_and_is_not_fatal() + -> Result<(), Box> { + let (reader, _writer, peer) = connected_pair().await?; + let (reporter, _sinks) = Reporter::channels(1, 1); + + // Abort the peer rather than half-closing it, to exercise the actual read-error path. + peer.set_zero_linger()?; + drop(peer); + let Err(error) = timeout( + TEST_TIMEOUT, + receive(reader, BoardId(42), &Index::default(), &reporter), + ) + .await? + else { + unreachable!("an abortive peer close must fail the socket read"); + }; + assert!(!error.is_fatal()); + let Error::Read { board, source } = error else { + unreachable!("a connection reset is a read failure"); + }; + assert_eq!(board, BoardId(42)); + assert_eq!(source.kind(), io::ErrorKind::ConnectionReset); + Ok(()) + } + + #[tokio::test] + async fn a_closed_notification_consumer_stops_instead_of_requesting_reconnection() + -> Result<(), Box> { + let (index, frames) = sample_frames(); + let (reader, _writer, mut peer) = connected_pair().await?; + let (reporter, mut sinks) = Reporter::channels(1, 1); + + // Keep the peer connected: this outcome comes from notify, not from socket EOF. + sinks.notifications.close(); + peer.write_all(&frames[0].1).await?; + let end = timeout( + TEST_TIMEOUT, + receive(reader, BoardId(42), &index, &reporter), + ) + .await??; + assert_eq!(end, ReceiveEnd::ConsumerClosed); + assert!(sinks.errors.try_recv().is_err()); + Ok(()) + } + + #[tokio::test] + async fn backpressure_waits_for_capacity_or_consumer_closure() + -> Result<(), Box> { + let (index, frames) = sample_frames(); + for close_consumer in [false, true] { + let (mut reader, _writer, mut peer) = connected_pair().await?; + let (reporter, mut sinks) = Reporter::channels(1, 1); + let bytes: Vec<_> = frames + .iter() + .flat_map(|(_, bytes)| bytes.iter().copied()) + .collect(); + peer.write_all(&bytes).await?; + peer.shutdown().await?; + + // Peek without consuming until this tiny burst is available. This makes + // one reader poll reach notify, without sleeps or assumptions about writes. + let mut preview = vec![0; bytes.len()]; + timeout(TEST_TIMEOUT, async { + while reader.peek(&mut preview).await? < bytes.len() { + tokio::task::yield_now().await; + } + Ok::<_, io::Error>(()) + }) + .await??; + + // Fill the only notification slot before the reader attempts to publish. + reporter.notify(BoardId(42), Event::Connected).await?; + let reading = receive(reader, BoardId(42), &index, &reporter); + tokio::pin!(reading); + poll_fn(|cx| { + assert!(reading.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + + if close_consumer { + // Closing a full channel must release the pending publication. + sinks.notifications.close(); + assert_eq!( + timeout(TEST_TIMEOUT, reading).await??, + ReceiveEnd::ConsumerClosed + ); + } else { + let Some(initial) = sinks.notifications.recv().await else { + unreachable!("the test prefilled the notification slot"); + }; + assert!(matches!(initial.event, Event::Connected)); + + // Drain while the reader advances, proving no decoded packet was dropped. + let (end, ()) = timeout(TEST_TIMEOUT, async { + tokio::join!(reading, async { + for (expected, _) in &frames { + let Some(notification) = sinks.notifications.recv().await else { + unreachable!("the reporter remains alive"); + }; + let Event::Received(packet) = notification.event else { + unreachable!("the prefilled lifecycle event was already drained"); + }; + assert_eq!(&packet, expected); + } + }) + }) + .await?; + assert_eq!(end?, ReceiveEnd::PeerClosed); + assert!(sinks.notifications.try_recv().is_err()); + } + } + Ok(()) + } +} diff --git a/backend/crates/core/src/net/tcp/writer.rs b/backend/crates/core/src/net/tcp/writer.rs new file mode 100644 index 000000000..093592a05 --- /dev/null +++ b/backend/crates/core/src/net/tcp/writer.rs @@ -0,0 +1,974 @@ +//! Serial writing of queued orders and periodic keep-alives for one TCP session. + +use super::Error; +use crate::model::BoardId; +use std::{future::pending, io, time::Duration}; +use tokio::{ + io::{AsyncWrite, AsyncWriteExt}, + net::tcp::OwnedWriteHalf, + sync::mpsc, + time::{Instant, sleep, timeout}, +}; + +/// The application heartbeat is only the reserved ID 1, little-endian. +const KEEP_ALIVE: [u8; 2] = [0x01, 0x00]; + +/// Writes queued frames in order, ending normally when the channel is closed and drained. +/// +/// Owns the write half but only borrows the receiver: ending a session must not +/// destroy the queue. The caller must finish any handshake before starting this +/// writer and must not write frames through another path on the same connection. +/// No index is needed because the queued frames are already encoded. +/// +/// `keep_alive` is the configured period; zero disables the timer. The first +/// heartbeat waits one period. Normal orders do not reset it, and a due heartbeat +/// takes priority at frame boundaries, never interrupting a frame in progress. +/// After a heartbeat completes, its timer restarts for a full period: delayed +/// writes produce no catch-up burst, including when the heartbeat itself is slow. +/// A closed and drained channel ends the writer even when a heartbeat is due. +/// +/// Each complete frame gets one `write_timeout`, including all partial writes. +/// Waiting for an order has no deadline. Success means the socket accepted all +/// queued bytes, not that firmware received or executed them. Dropping the owned +/// write half shuts down the socket's write side; it is not a delivery guarantee. +/// +/// On failure or cancellation during an order write, its dequeued frame is lost +/// to the caller and may have been partly or fully transmitted. A heartbeat write +/// does not dequeue an order. In either case, never automatically retry the frame +/// or continue using that session. Orders not yet dequeued remain in `orders`; +/// their disconnected-order policy belongs to the future session supervisor. +/// That owner must also dispose of the read half and report failures once. +/// This function does not log, report events, reconnect, or broadcast faults. +/// +/// # Errors +/// +/// Returns non-fatal [`Error::Write`] with the original I/O error, or an I/O +/// error of kind [`io::ErrorKind::TimedOut`] when a frame deadline elapses. +// The session supervisor will use this once it is implemented. +#[cfg_attr(not(test), allow(dead_code))] +pub(super) async fn send( + writer: OwnedWriteHalf, + board: BoardId, + orders: &mut mpsc::Receiver>, + write_timeout: Duration, + keep_alive: Duration, +) -> Result<(), Error> { + write_orders(writer, board, orders, write_timeout, keep_alive).await +} + +/// Drives the same loop over TCP or a controlled test writer, without socket-specific behavior. +async fn write_orders( + mut writer: W, + board: BoardId, + orders: &mut mpsc::Receiver>, + write_timeout: Duration, + keep_alive: Duration, +) -> Result<(), Error> { + let mut heartbeat = (!keep_alive.is_zero()).then(|| Box::pin(sleep(keep_alive))); + + loop { + // Some owns one queued order; None selects the static heartbeat bytes. + // Choose only between frames, so neither branch can cancel write_all. + let order = tokio::select! { + biased; + _ = async { + match &mut heartbeat { + Some(timer) => timer.as_mut().await, + None => pending::<()>().await, + } + } => { + // Closure wins over a due timer once the queue is empty. recv + // also waits for outstanding reserved permits before declaring + // the channel drained; is_closed + is_empty alone is not enough. + if orders.is_closed() && orders.is_empty() { + let Some(bytes) = orders.recv().await else { return Ok(()) }; + Some(bytes) + } else { + None + } + } + order = orders.recv() => { + let Some(bytes) = order else { return Ok(()) }; + Some(bytes) + } + }; + let bytes = order.as_deref().unwrap_or(&KEEP_ALIVE); + // Keep one deadline around the whole frame, not one deadline per partial write. + timeout(write_timeout, writer.write_all(bytes)) + .await + .map_err(|_| Error::Write { + board, + source: io::Error::new(io::ErrorKind::TimedOut, "TCP frame write timed out"), + })? + .map_err(|source| Error::Write { board, source })?; + + if order.is_none() + && let Some(timer) = &mut heartbeat + { + // Restart only after heartbeat completion, not after ordinary traffic. + timer.as_mut().reset(Instant::now() + keep_alive); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::{ + collections::VecDeque, + future::{Future, poll_fn}, + net::Ipv4Addr, + pin::Pin, + task::{Context, Poll}, + }; + use tokio::{ + io::AsyncReadExt, + net::{TcpListener, TcpStream}, + time::{Instant, Sleep, advance}, + }; + + const TEST_TIMEOUT: Duration = Duration::from_secs(2); + const FRAME_TIMEOUT: Duration = Duration::from_millis(10); + const FIRST: &[u8] = &[0x58, 0x02, 42]; + const SECOND: &[u8] = &[0x59, 0x02]; + + /// One action at the next write attempt, independent of kernel buffer sizes. + #[derive(Debug)] + enum Step { + Write(usize), + Wait(Duration), + Fail(io::ErrorKind), + Zero, + Pending, + } + + /// Records accepted bytes while scripting partial progress, failures and stalls. + #[derive(Debug)] + struct ControlledWriter { + steps: VecDeque, + delay: Option>>, + written: Vec, + write_times: Vec, + } + + impl ControlledWriter { + // New empty ControlledWriter + fn new(steps: impl IntoIterator) -> Self { + Self { + steps: steps.into_iter().collect(), + delay: None, + written: vec![], + write_times: vec![], + } + } + } + + impl AsyncWrite for ControlledWriter { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + bytes: &[u8], + ) -> Poll> { + loop { + if let Some(delay) = &mut self.delay { + // Start each delay when reached, rather than at fixture construction. + if delay.as_mut().poll(cx).is_pending() { + return Poll::Pending; + } + self.delay = None; + } + match self.steps.pop_front() { + Some(Step::Wait(duration)) => { + self.delay = Some(Box::pin(tokio::time::sleep(duration))); + } + Some(Step::Fail(kind)) => { + return Poll::Ready(Err(io::Error::new(kind, "scripted write failure"))); + } + Some(Step::Zero) => return Poll::Ready(Ok(0)), + Some(Step::Pending) => { + // This sink never becomes writable; only the deadline can finish it. + self.steps.push_front(Step::Pending); + return Poll::Pending; + } + step => { + let count = match step { + Some(Step::Write(limit)) => limit.min(bytes.len()), + None => bytes.len(), + _ => unreachable!("all non-write actions are handled above"), + }; + self.written.extend_from_slice(&bytes[..count]); + self.write_times.push(Instant::now()); + return Poll::Ready(Ok(count)); + } + } + } + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + } + + /// Preloads two complete frames and keeps the sender alive for failure-path checks. + fn queued_frames() -> (mpsc::Sender>, mpsc::Receiver>) { + let (sender, receiver) = mpsc::channel(2); + assert!(sender.try_send(FIRST.to_vec()).is_ok()); + assert!(sender.try_send(SECOND.to_vec()).is_ok()); + (sender, receiver) + } + + /// Checks the public transport classification as well as the underlying failure. + fn assert_write_error(error: Error, kind: io::ErrorKind) -> io::Error { + assert!(!error.is_fatal()); + let Error::Write { board, source } = error else { + unreachable!("writer failures must retain their board and I/O cause"); + }; + assert_eq!(board, BoardId(3)); + assert_eq!(source.kind(), kind); + source + } + + /// Polls once without sleeping, proving the writer is still waiting for work or time. + async fn assert_pending_once(mut future: Pin<&mut F>) { + poll_fn(|cx| { + assert!(future.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + } + + /// Expresses recorded write times relative to the scenario's virtual start. + fn elapsed_writes(writer: &ControlledWriter, started: Instant) -> Vec { + writer + .write_times + .iter() + .map(|at| at.duration_since(started)) + .collect() + } + + #[tokio::test] + async fn writes_queued_frames_to_tcp_in_order_and_closes_its_write_side() + -> Result<(), Box> { + // Use ordinary loopback sockets, not zero linger, to observe bytes followed by EOF. + // Assert the concatenated stream: TCP need not preserve individual write boundaries. + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?; + let client = timeout(TEST_TIMEOUT, TcpStream::connect(listener.local_addr()?)).await??; + let (mut peer, _) = timeout(TEST_TIMEOUT, listener.accept()).await??; + let (_read_half, write_half) = client.into_split(); + let (sender, mut receiver) = queued_frames(); + drop(sender); + let mut received = vec![]; + + let (sent, read) = timeout(TEST_TIMEOUT, async { + tokio::join!( + send( + write_half, + BoardId(3), + &mut receiver, + TEST_TIMEOUT, + Duration::ZERO + ), + peer.read_to_end(&mut received), + ) + }) + .await?; + sent?; + read?; + assert_eq!(received, [FIRST, SECOND].concat()); + assert_eq!( + receiver.try_recv(), + Err(mpsc::error::TryRecvError::Disconnected) + ); + Ok(()) + } + + #[tokio::test] + async fn an_empty_closed_channel_finishes_without_attempting_a_write() + -> Result<(), Box> { + // The failure action is a sentinel: a drained channel must not touch the writer. + let (sender, mut receiver) = mpsc::channel(1); + drop(sender); + let mut writer = ControlledWriter::new([Step::Fail(io::ErrorKind::BrokenPipe)]); + timeout( + TEST_TIMEOUT, + write_orders( + &mut writer, + BoardId(3), + &mut receiver, + FRAME_TIMEOUT, + Duration::ZERO, + ), + ) + .await??; + assert!(writer.written.is_empty()); + assert_eq!(writer.steps.len(), 1); + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn waiting_for_an_order_has_no_write_deadline() -> Result<(), Box> + { + // With keep-alive disabled, advance well beyond the frame timeout while idle. + // The writer must remain pending, then accept a later order normally. + let (sender, mut receiver) = mpsc::channel(1); + let mut writer = ControlledWriter::new([]); + { + let sending = write_orders( + &mut writer, + BoardId(3), + &mut receiver, + FRAME_TIMEOUT, + Duration::ZERO, + ); + tokio::pin!(sending); + poll_fn(|cx| { + assert!(sending.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + advance(FRAME_TIMEOUT * 10).await; + poll_fn(|cx| { + assert!(sending.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + sender.try_send(FIRST.to_vec())?; + drop(sender); + timeout(TEST_TIMEOUT, sending).await??; + } + assert_eq!(writer.written, FIRST); + Ok(()) + } + + #[tokio::test] + async fn completes_partial_writes_before_starting_the_next_frame() + -> Result<(), Box> { + // Accept only one byte per write call. write_all must finish FIRST before SECOND, + // even though each frame requires several successful partial writes. + let (sender, mut receiver) = queued_frames(); + drop(sender); + let mut writer = ControlledWriter::new((0..5).map(|_| Step::Write(1))); + timeout( + TEST_TIMEOUT, + write_orders( + &mut writer, + BoardId(3), + &mut receiver, + FRAME_TIMEOUT, + Duration::ZERO, + ), + ) + .await??; + assert_eq!(writer.written, [FIRST, SECOND].concat()); + assert!(writer.steps.is_empty()); + Ok(()) + } + + #[tokio::test] + async fn io_failure_stops_without_retrying_or_dequeueing_the_next_frame() + -> Result<(), Box> { + // Fail both before any progress and after a prefix. In either case FIRST is + // not requeued, SECOND stays untouched, and the borrowed receiver remains open. + for prefix in [0, 2] { + let (sender, mut receiver) = queued_frames(); + let mut steps = vec![]; + if prefix > 0 { + steps.push(Step::Write(prefix)); + } + steps.push(Step::Fail(io::ErrorKind::BrokenPipe)); + let mut writer = ControlledWriter::new(steps); + let Err(error) = timeout( + TEST_TIMEOUT, + write_orders( + &mut writer, + BoardId(3), + &mut receiver, + FRAME_TIMEOUT, + Duration::ZERO, + ), + ) + .await? + else { + unreachable!("the scripted I/O failure must terminate this session"); + }; + let source = assert_write_error(error, io::ErrorKind::BrokenPipe); + assert_eq!(source.to_string(), "scripted write failure"); + assert_eq!(writer.written, FIRST[..prefix]); + assert_eq!(receiver.try_recv(), Ok(SECOND.to_vec())); + assert_eq!(receiver.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + sender.try_send(vec![0xaa])?; + } + Ok(()) + } + + #[tokio::test] + async fn a_zero_byte_write_is_a_write_zero_error_not_success() + -> Result<(), Box> { + // AsyncWrite can return zero without an I/O error; write_all must treat that + // as inability to complete a nonempty frame and leave SECOND in the queue. + let (_sender, mut receiver) = queued_frames(); + let mut writer = ControlledWriter::new([Step::Zero]); + let Err(error) = timeout( + TEST_TIMEOUT, + write_orders( + &mut writer, + BoardId(3), + &mut receiver, + FRAME_TIMEOUT, + Duration::ZERO, + ), + ) + .await? + else { + unreachable!("zero progress cannot complete a nonempty frame"); + }; + assert_write_error(error, io::ErrorKind::WriteZero); + assert!(writer.written.is_empty()); + assert_eq!(receiver.try_recv(), Ok(SECOND.to_vec())); + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn a_stalled_write_times_out_without_touching_the_next_frame() + -> Result<(), Box> { + // A permanently pending sink deterministically exercises the timeout path, + // both before any bytes and after partial progress, without filling a TCP buffer. + for prefix in [0, 2] { + let (sender, mut receiver) = queued_frames(); + let mut steps = vec![]; + if prefix > 0 { + steps.push(Step::Write(prefix)); + } + steps.push(Step::Pending); + let mut writer = ControlledWriter::new(steps); + let started = Instant::now(); + let Err(error) = timeout( + TEST_TIMEOUT, + write_orders( + &mut writer, + BoardId(3), + &mut receiver, + FRAME_TIMEOUT, + Duration::ZERO, + ), + ) + .await? + else { + unreachable!("a stalled frame must hit its deadline"); + }; + assert_write_error(error, io::ErrorKind::TimedOut); + assert_eq!(started.elapsed(), FRAME_TIMEOUT); + assert_eq!(writer.written, FIRST[..prefix]); + assert_eq!(receiver.try_recv(), Ok(SECOND.to_vec())); + assert_eq!(receiver.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + sender.try_send(vec![0xaa])?; + } + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn partial_progress_does_not_restart_the_frame_deadline() + -> Result<(), Box> { + // Each delayed write takes less than 10 ms, but the whole frame takes longer. + // A per-write timeout would incorrectly succeed; the frame deadline must expire. + let (_sender, mut receiver) = queued_frames(); + let mut writer = ControlledWriter::new([ + Step::Wait(Duration::from_millis(6)), + Step::Write(1), + Step::Wait(Duration::from_millis(6)), + Step::Write(1), + Step::Wait(Duration::from_millis(6)), + Step::Write(1), + ]); + let started = Instant::now(); + let Err(error) = timeout( + TEST_TIMEOUT, + write_orders( + &mut writer, + BoardId(3), + &mut receiver, + FRAME_TIMEOUT, + Duration::ZERO, + ), + ) + .await? + else { + unreachable!("partial progress must not extend the frame's deadline"); + }; + assert_write_error(error, io::ErrorKind::TimedOut); + assert_eq!(started.elapsed(), FRAME_TIMEOUT); + assert_eq!(writer.written, FIRST[..1]); + assert_eq!(receiver.try_recv(), Ok(SECOND.to_vec())); + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn each_frame_receives_a_fresh_write_deadline() -> Result<(), Box> + { + // Two 6 ms frames exceed 10 ms together but each fits its own deadline. + // This rules out accidentally timing the entire queue-draining loop. + let (sender, mut receiver) = queued_frames(); + drop(sender); + let mut writer = ControlledWriter::new([ + Step::Wait(Duration::from_millis(6)), + Step::Write(usize::MAX), + Step::Wait(Duration::from_millis(6)), + Step::Write(usize::MAX), + ]); + let started = Instant::now(); + timeout( + TEST_TIMEOUT, + write_orders( + &mut writer, + BoardId(3), + &mut receiver, + FRAME_TIMEOUT, + Duration::ZERO, + ), + ) + .await??; + assert_eq!(started.elapsed(), Duration::from_millis(12)); + assert_eq!(writer.written, [FIRST, SECOND].concat()); + Ok(()) + } + + #[tokio::test] + async fn cancellation_does_not_requeue_the_current_frame_or_destroy_the_receiver() + -> Result<(), Box> { + // Cancel after a prefix reached the sink. The remaining queue survives, but + // FIRST is no longer recoverable: a supervisor must discard this session. + let (sender, mut receiver) = queued_frames(); + let mut writer = ControlledWriter::new([Step::Write(1), Step::Pending]); + { + let sending = write_orders( + &mut writer, + BoardId(3), + &mut receiver, + FRAME_TIMEOUT, + Duration::ZERO, + ); + tokio::pin!(sending); + poll_fn(|cx| { + assert!(sending.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + // Leaving this scope drops the in-flight future, not the borrowed receiver. + } + assert_eq!(writer.written, FIRST[..1]); + assert_eq!(receiver.try_recv(), Ok(SECOND.to_vec())); + assert_eq!(receiver.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + sender.try_send(vec![0xaa])?; + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn idle_keep_alives_wait_a_full_configured_period_and_repeat() + -> Result<(), Box> { + // Different periods must produce their first two heartbeats at the configured + // times, never immediately. An open, empty order channel must stay alive. + for period in [Duration::from_millis(10), Duration::from_millis(24)] { + let (sender, mut receiver) = mpsc::channel(1); + let mut writer = ControlledWriter::new([]); + let started = Instant::now(); + { + let sending = write_orders( + &mut writer, + BoardId(3), + &mut receiver, + FRAME_TIMEOUT, + period, + ); + tokio::pin!(sending); + assert_pending_once(sending.as_mut()).await; + advance(period / 2).await; + assert_pending_once(sending.as_mut()).await; + advance(period / 2).await; + assert_pending_once(sending.as_mut()).await; + advance(period).await; + assert_pending_once(sending.as_mut()).await; + drop(sender); + timeout(TEST_TIMEOUT, sending).await??; + } + assert_eq!(writer.written, KEEP_ALIVE.repeat(2)); + assert_eq!(elapsed_writes(&writer, started), vec![period, period * 2]); + } + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn ordinary_traffic_does_not_reset_or_starve_the_keep_alive() + -> Result<(), Box> { + // Each queued order takes 12 ms, crossing the 10 ms heartbeat period. + // The due heartbeat must win before the next queued order, not wait for idle time. + let period = Duration::from_millis(10); + let order_time = Duration::from_millis(12); + let (sender, mut receiver) = queued_frames(); + let mut writer = ControlledWriter::new([ + Step::Wait(order_time), + Step::Write(usize::MAX), + Step::Write(usize::MAX), + Step::Wait(order_time), + Step::Write(usize::MAX), + ]); + let started = Instant::now(); + { + let sending = + write_orders(&mut writer, BoardId(3), &mut receiver, TEST_TIMEOUT, period); + tokio::pin!(sending); + assert_pending_once(sending.as_mut()).await; + advance(order_time).await; + assert_pending_once(sending.as_mut()).await; + advance(order_time).await; + assert_pending_once(sending.as_mut()).await; + drop(sender); + timeout(TEST_TIMEOUT, sending).await??; + } + assert_eq!( + writer.written, + [FIRST, &KEEP_ALIVE, SECOND, &KEEP_ALIVE].concat() + ); + assert_eq!( + elapsed_writes(&writer, started), + vec![order_time, order_time, order_time * 2, order_time * 2] + ); + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn a_due_keep_alive_never_interrupts_a_partially_written_order() + -> Result<(), Box> { + // FIRST pauses after one byte while several heartbeat periods elapse. + // Only its remaining bytes may follow that prefix; the heartbeat comes afterward. + let period = Duration::from_millis(10); + let delay = Duration::from_millis(35); + let (sender, mut receiver) = queued_frames(); + let mut writer = + ControlledWriter::new([Step::Write(1), Step::Wait(delay), Step::Write(usize::MAX)]); + { + let sending = + write_orders(&mut writer, BoardId(3), &mut receiver, TEST_TIMEOUT, period); + tokio::pin!(sending); + assert_pending_once(sending.as_mut()).await; + advance(delay).await; + assert_pending_once(sending.as_mut()).await; + drop(sender); + timeout(TEST_TIMEOUT, sending).await??; + } + assert_eq!(writer.written, [FIRST, &KEEP_ALIVE, SECOND].concat()); + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn missed_keep_alive_periods_produce_one_send_not_a_catch_up_burst() + -> Result<(), Box> { + // Delay an order for seven periods. Resuming must send one heartbeat, then + // wait a new period. Include 1 ms to catch timer policies with short-delay bursts. + for period in [Duration::from_millis(1), Duration::from_millis(10)] { + let delay = period * 7; + let (sender, mut receiver) = mpsc::channel(1); + sender.try_send(FIRST.to_vec())?; + let mut writer = ControlledWriter::new([Step::Wait(delay), Step::Write(usize::MAX)]); + let started = Instant::now(); + { + let sending = + write_orders(&mut writer, BoardId(3), &mut receiver, TEST_TIMEOUT, period); + tokio::pin!(sending); + assert_pending_once(sending.as_mut()).await; + advance(delay).await; + assert_pending_once(sending.as_mut()).await; + advance(period).await; + assert_pending_once(sending.as_mut()).await; + drop(sender); + timeout(TEST_TIMEOUT, sending).await??; + } + assert_eq!(writer.written, [FIRST, &KEEP_ALIVE, &KEEP_ALIVE].concat()); + assert_eq!( + elapsed_writes(&writer, started), + vec![delay, delay, delay + period] + ); + } + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn a_slow_partial_keep_alive_finishes_before_orders_and_rearms_after_completion() + -> Result<(), Box> { + // Stall between the two heartbeat bytes, then queue an ordinary order. + // Neither that order nor another overdue heartbeat may split or follow the + // slow heartbeat as a catch-up send; its next period starts at completion. + let period = Duration::from_millis(10); + let delay = Duration::from_millis(35); + let (sender, mut receiver) = mpsc::channel(1); + let mut writer = ControlledWriter::new([Step::Write(1), Step::Wait(delay), Step::Write(1)]); + let started = Instant::now(); + { + let sending = + write_orders(&mut writer, BoardId(3), &mut receiver, TEST_TIMEOUT, period); + tokio::pin!(sending); + assert_pending_once(sending.as_mut()).await; + advance(period).await; + assert_pending_once(sending.as_mut()).await; + sender.try_send(FIRST.to_vec())?; + advance(delay).await; + assert_pending_once(sending.as_mut()).await; + advance(period).await; + assert_pending_once(sending.as_mut()).await; + drop(sender); + timeout(TEST_TIMEOUT, sending).await??; + } + assert_eq!(writer.written, [&KEEP_ALIVE, FIRST, &KEEP_ALIVE].concat()); + assert_eq!( + elapsed_writes(&writer, started), + vec![period, period + delay, period + delay, period * 2 + delay] + ); + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn keep_alive_io_failures_preserve_the_cause_and_leave_orders_queued() + -> Result<(), Box> { + // Make the timer and an order ready together. The due heartbeat wins, but + // failing before/after a prefix must not dequeue or silently discard that order. + for prefix in [0, 1] { + let period = Duration::from_millis(10); + let (sender, mut receiver) = mpsc::channel(1); + let mut steps = vec![]; + if prefix > 0 { + steps.push(Step::Write(prefix)); + } + steps.push(Step::Fail(io::ErrorKind::BrokenPipe)); + let mut writer = ControlledWriter::new(steps); + let error; + { + let sending = write_orders( + &mut writer, + BoardId(3), + &mut receiver, + FRAME_TIMEOUT, + period, + ); + tokio::pin!(sending); + assert_pending_once(sending.as_mut()).await; + advance(period).await; + sender.try_send(FIRST.to_vec())?; + let Err(failure) = timeout(TEST_TIMEOUT, sending).await? else { + unreachable!("a failed heartbeat must terminate this session"); + }; + error = failure; + } + let source = assert_write_error(error, io::ErrorKind::BrokenPipe); + assert_eq!(source.to_string(), "scripted write failure"); + assert_eq!(writer.written, KEEP_ALIVE[..prefix]); + assert_eq!(receiver.try_recv(), Ok(FIRST.to_vec())); + assert_eq!(receiver.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + } + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn keep_alive_timeouts_use_the_frame_deadline_and_leave_orders_queued() + -> Result<(), Box> { + // A heartbeat has the same whole-frame deadline as an order. Stalling + // before or after its first byte must return TimedOut without consuming FIRST. + for prefix in [0, 1] { + let period = Duration::from_millis(10); + let (sender, mut receiver) = mpsc::channel(1); + let mut steps = vec![]; + if prefix > 0 { + steps.push(Step::Write(prefix)); + } + steps.push(Step::Pending); + let mut writer = ControlledWriter::new(steps); + let started = Instant::now(); + let error; + { + let sending = write_orders( + &mut writer, + BoardId(3), + &mut receiver, + FRAME_TIMEOUT, + period, + ); + tokio::pin!(sending); + assert_pending_once(sending.as_mut()).await; + advance(period).await; + sender.try_send(FIRST.to_vec())?; + assert_pending_once(sending.as_mut()).await; + let Err(failure) = timeout(TEST_TIMEOUT, sending).await? else { + unreachable!("a stalled heartbeat must hit the frame deadline"); + }; + error = failure; + } + assert_write_error(error, io::ErrorKind::TimedOut); + assert_eq!(started.elapsed(), period + FRAME_TIMEOUT); + assert_eq!(writer.written, KEEP_ALIVE[..prefix]); + assert_eq!(receiver.try_recv(), Ok(FIRST.to_vec())); + assert_eq!(receiver.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + } + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn cancelling_a_partial_keep_alive_preserves_the_order_queue() + -> Result<(), Box> { + // Cancellation must drop the partial heartbeat, not move or requeue an order. + // The connection must be discarded afterward, just as for partial order writes. + let period = Duration::from_millis(10); + let (sender, mut receiver) = mpsc::channel(1); + let mut writer = ControlledWriter::new([Step::Write(1), Step::Pending]); + { + let sending = write_orders( + &mut writer, + BoardId(3), + &mut receiver, + FRAME_TIMEOUT, + period, + ); + tokio::pin!(sending); + assert_pending_once(sending.as_mut()).await; + advance(period).await; + sender.try_send(FIRST.to_vec())?; + assert_pending_once(sending.as_mut()).await; + } + assert_eq!(writer.written, KEEP_ALIVE[..1]); + assert_eq!(receiver.try_recv(), Ok(FIRST.to_vec())); + assert_eq!(receiver.try_recv(), Err(mpsc::error::TryRecvError::Empty)); + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn a_closed_drained_channel_wins_even_when_the_keep_alive_is_due() + -> Result<(), Box> { + // Close while the writer waits, then wake it with both closure and a due + // timer. It must finish without sending a final heartbeat or waiting forever. + let period = Duration::from_millis(10); + let (sender, mut receiver) = mpsc::channel(1); + let mut writer = ControlledWriter::new([Step::Fail(io::ErrorKind::BrokenPipe)]); + { + let sending = write_orders( + &mut writer, + BoardId(3), + &mut receiver, + FRAME_TIMEOUT, + period, + ); + tokio::pin!(sending); + assert_pending_once(sending.as_mut()).await; + drop(sender); + advance(period).await; + timeout(TEST_TIMEOUT, sending).await??; + } + assert!(writer.written.is_empty()); + assert_eq!(writer.steps.len(), 1); + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn closing_drains_orders_but_suppresses_a_keep_alive_after_the_last_frame() + -> Result<(), Box> { + // Keep-alives still fit between pending orders during draining, but the + // timer must not send again once the last queued frame has completed. + let period = Duration::from_millis(10); + let order_time = Duration::from_millis(12); + let (sender, mut receiver) = queued_frames(); + drop(sender); + let mut writer = ControlledWriter::new([ + Step::Wait(order_time), + Step::Write(usize::MAX), + Step::Write(usize::MAX), + Step::Wait(order_time), + Step::Write(usize::MAX), + ]); + timeout( + TEST_TIMEOUT, + write_orders(&mut writer, BoardId(3), &mut receiver, TEST_TIMEOUT, period), + ) + .await??; + assert_eq!(writer.written, [FIRST, &KEEP_ALIVE, SECOND].concat()); + assert_eq!( + receiver.try_recv(), + Err(mpsc::error::TryRecvError::Disconnected) + ); + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn closing_waits_for_reserved_orders_without_emitting_keep_alives_on_an_empty_queue() + -> Result<(), Box> { + // A reserved permit can still enqueue after Receiver::close. Closed plus + // empty is therefore not proof that recv is finished; do not lose that order. + let period = Duration::from_millis(10); + let (sender, mut receiver) = mpsc::channel(1); + let permit = sender.reserve().await?; + receiver.close(); + let mut writer = ControlledWriter::new([]); + { + let sending = write_orders( + &mut writer, + BoardId(3), + &mut receiver, + FRAME_TIMEOUT, + period, + ); + tokio::pin!(sending); + assert_pending_once(sending.as_mut()).await; + advance(period).await; + assert_pending_once(sending.as_mut()).await; + permit.send(FIRST.to_vec()); + timeout(TEST_TIMEOUT, sending).await??; + } + assert_eq!(writer.written, FIRST); + assert_eq!( + receiver.try_recv(), + Err(mpsc::error::TryRecvError::Disconnected) + ); + Ok(()) + } + + #[tokio::test] + async fn the_tcp_writer_sends_the_exact_keep_alive_bytes_until_the_channel_closes() + -> Result<(), Box> { + // Exercise the concrete socket adapter, not only the controlled loop. + // Close after observing one heartbeat; any bytes already sent afterward + // must still consist solely of whole ID-1 frames before write-side EOF. + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?; + let client = timeout(TEST_TIMEOUT, TcpStream::connect(listener.local_addr()?)).await??; + let (mut peer, _) = timeout(TEST_TIMEOUT, listener.accept()).await??; + let (_read_half, write_half) = client.into_split(); + let (sender, mut receiver) = mpsc::channel(1); + let mut first = [0; 2]; + let mut rest = vec![]; + let (sent, received) = timeout(TEST_TIMEOUT, async { + tokio::join!( + send( + write_half, + BoardId(3), + &mut receiver, + TEST_TIMEOUT, + Duration::from_millis(10) + ), + async { + peer.read_exact(&mut first).await?; + drop(sender); + peer.read_to_end(&mut rest).await?; + Ok::<_, io::Error>(()) + }, + ) + }) + .await?; + sent?; + received?; + assert_eq!(first, KEEP_ALIVE); + assert_eq!(rest.len() % KEEP_ALIVE.len(), 0); + assert!( + rest.chunks(KEEP_ALIVE.len()) + .all(|frame| frame == KEEP_ALIVE) + ); + Ok(()) + } +} diff --git a/backend/crates/core/src/net/udp.rs b/backend/crates/core/src/net/udp.rs new file mode 100644 index 000000000..e8048507b --- /dev/null +++ b/backend/crates/core/src/net/udp.rs @@ -0,0 +1,217 @@ +//! 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. +//! +//! # Entry point +//! +//! [`spawn`], taking [`Config`], an `Arc` and a +//! [`super::Reporter`] — written out with its reasoning on the items +//! themselves. No handle comes back beyond the `JoinHandle`: there is no +//! writing side. +//! +//! # Contract +//! +//! - **No framing.** A datagram arrives whole or not at all, so +//! `recv_from` already delivers exactly one packet and its length — +//! 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. +//! - [`super::Notification::board`] comes from the decoded packet's id, +//! through [`crate::protocol::Index`] — every +//! [`crate::model::PacketDef`] carries its `board` — so that a +//! notification looks the same whichever transport brought it. That is +//! one `HashMap` lookup per datagram, done here once rather than +//! repeated by every consumer downstream. +//! - **A decode failure is fatal**: see [`Error::is_fatal`]. Nothing +//! corrupts a datagram in flight — UDP checksums it and the pod's +//! network does not fragment it — so a datagram that will not decode +//! means the ADJ this backend loaded does not describe what the boards +//! are actually sending, and every datagram after it is suspect for the +//! same reason. So: log it and report it on the error channel, exactly +//! as [`super::tcp`] does — and that is all. This socket writes +//! nothing, and sending the pod to fault is not `net`'s call anyway: +//! whoever consumes the error channel reads [`Error::is_fatal`] and +//! decides, which also keeps the "fault once" guard in one place +//! rather than one per transport. Keep receiving meanwhile: a +//! mismatched ADJ fails on every datagram that follows, several +//! thousand a second, and each one is reported, not acted on. +//! - [`super::Event::Connected`]/[`super::Event::Disconnected`] are never +//! emitted: there is no connection to come up or go down. +//! - Binding happens **before** the task is spawned, so a bad address is +//! a startup failure returned to whoever is composing the backend +//! (`station::vehicle::bootstrap`), not something reported once +//! everything is already running. Hence the entry point returns +//! `Result, Error>`. + +use super::Reporter; +use crate::protocol::{DecodeError, Index}; +use std::net::SocketAddr; +use std::sync::Arc; +use tokio::task::JoinHandle; + +/// How to set the shared socket up. +/// +/// Everything here is resolved by `station` before `net` is reached: this +/// module reads no config file and no ADJ of its own. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Config { + /// Where to bind: the backend's own address, `backend_ip` + + /// `ports["UDP"]` in the ADJ's general info. + pub addr: SocketAddr, + /// `SO_RCVBUF`, in bytes. + /// + /// The only thing absorbing a stall of the receive loop, since the + /// kernel drops the tail silently once it is full: 2 MiB is about + /// 0.8 s of margin at 20 Mbit/s. + pub recv_buffer: usize, +} + +/// Binds the shared socket and spawns the receive loop. +/// +/// 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") +} + +/// Something that went wrong on the shared UDP socket. +/// +/// No [`crate::model::BoardId`] in any variant, on purpose: the socket +/// belongs to no board, and a datagram that fails to decode may not +/// resolve to one at all — [`Error::Decode`] carries the address it came +/// from, which is the only identity that always exists. +#[derive(Debug, thiserror::Error)] +pub enum Error { + /// The socket could not be bound, so no telemetry will arrive at all. + #[error("could not bind the shared UDP socket to {addr}")] + Bind { + /// The address it was being bound to. + addr: SocketAddr, + /// The underlying I/O failure. + #[source] + source: std::io::Error, + }, + + /// Receiving on the socket failed. + #[error("could not receive on the shared UDP socket")] + Recv { + /// The underlying I/O failure. + #[source] + source: std::io::Error, + }, + + /// A datagram could not be decoded, which means the loaded ADJ does + /// not describe what the boards are sending. Fatal: see + /// [`Error::is_fatal`]. + #[error("could not decode a datagram from {from}")] + Decode { + /// Where the datagram came from — there may be no board to + /// attribute it to. + from: SocketAddr, + /// What [`crate::protocol`] said about it. + #[source] + source: DecodeError, + }, +} + +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, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::model::AdjId; + use std::net::IpAddr; + + fn peer() -> SocketAddr { + SocketAddr::new(IpAddr::from([192, 168, 1, 3]), 50400) + } + + #[test] + fn a_datagram_that_will_not_decode_faults_the_pod() { + let error = Error::Decode { + from: peer(), + source: DecodeError::UnknownId(AdjId(9999)), + }; + + // Nothing corrupts a datagram in flight, so this is a mismatched + // ADJ, not noise to skip over. + assert!(error.is_fatal()); + } + + #[test] + fn losing_the_socket_faults_the_pod_too() { + let error = Error::Recv { + source: std::io::Error::from(std::io::ErrorKind::ConnectionReset), + }; + + // Every board's telemetry at once. + assert!(error.is_fatal()); + } + + #[test] + fn a_decode_failure_says_where_it_came_from() { + let error = Error::Decode { + from: peer(), + source: 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")); + } +}