From 7274293a04d2ea2d4da0e986398e5d51b1332d7c Mon Sep 17 00:00:00 2001 From: Stephen Waits Date: Thu, 16 Apr 2026 15:08:38 -0600 Subject: [PATCH] feat!: mux on DongLoRa Protocol v2 with tag-aware dispatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 1.0 mux, running on the Protocol v2 wire format and full §13 multi-client semantics. * Tag-aware dispatch. Each inbound client frame is re-tagged with a mux-allocated device tag; responses are routed back to the original client using the tag mapping. The single-oneshot response slot is gone; multiple clients can have commands in flight concurrently. * Full §13.2 multi-client SET_CONFIG semantics. Tracks the (client, modulation) lock owner and synthesizes OK(APPLIED/MINE), OK(ALREADY_MATCHED/OTHER), or OK(LOCKED_MISMATCH/OTHER) based on lock ownership and parameter match. Lock releases automatically on client disconnect. * TX loopback fan-out (§13.4). On TX_DONE(TRANSMITTED) the mux synthesizes an RX frame with origin = LocalLoopback and sends it to every connected client except the transmitter. * Async ERR(EFRAME) fan-out. If the dongle emits async framing errors (tag = 0), they are broadcast to all clients rather than swallowed. * TagMapper with skip-zero allocation and per-client bulk drop on disconnect. * Per-client TX cache on ClientSession so the loopback fan-out does not need the sender to re-buffer. * 500 ms keepalive cadence matching PROTOCOL.md §3.4 inactivity window (replacing the old 5 s loop). --- CHANGELOG.md | 43 ++- Cargo.lock | 179 ++++++++-- Cargo.toml | 9 +- src/daemon.rs | 848 ++++++++++++++++++++++++++++++----------------- src/intercept.rs | 464 +++++++++++++++++--------- src/main.rs | 13 +- src/session.rs | 47 ++- 7 files changed, 1081 insertions(+), 522 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 33ed41d..6649c6a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,14 +1,45 @@ # Changelog -## Unreleased +## 1.0.0-alpha.1 — 2026-04-21 + +### Breaking + +- **Upgraded to DongLoRa Protocol v2 wire format.** Wire-incompatible with 0.x + clients and firmware. Now depends directly on `donglora-protocol` for + frame encode/decode. +- **Tag-aware dispatch.** Each inbound client frame is re-tagged with a + mux-allocated device tag; responses are routed back to the original + client using the tag mapping. No more single-oneshot response slot — + multiple clients can have commands in flight concurrently. ### Added -- Keepalive ping: detects unresponsive dongles (e.g. UART boards behind a - CP2102 bridge where the serial port stays open across ESP32 resets). Pings - every 5 seconds of idle, declares dongle lost after 2-second timeout. -- Radio state restoration after reconnect: saved SetConfig and StartRx are - re-sent to the dongle so clients don't need to know a reset happened. +- **Full `§13.2` multi-client `SET_CONFIG` semantics.** Tracks the + `(client, modulation)` lock owner. Cross-client calls synthesize: + - `OK(APPLIED / MINE, params)` when the calling client owns the lock + (forwarded to dongle). + - `OK(ALREADY_MATCHED / OTHER, current)` when another client holds + the lock and the requested params byte-match. + - `OK(LOCKED_MISMATCH / OTHER, current)` when another client holds + the lock and the requested params differ. + - Lock is released automatically when the owning client disconnects. +- **TX loopback fan-out (spec §13.4).** On `TX_DONE(TRANSMITTED)`, the + mux synthesizes an `RX` frame with `origin = LocalLoopback` and sends + it to every connected client except the transmitter. +- **Async `ERR(EFRAME)` fan-out.** If the dongle emits async framing + errors (tag = 0), they're broadcast to all clients rather than + swallowed. +- **Per-client TX cache** on `ClientSession` so the loopback fan-out + doesn't need the sender to re-buffer. +- **`TagMapper`** with skip-zero allocation and per-client bulk drop on + disconnect. + +### Removed + +- The single `pending_response: oneshot::Sender` slot in the daemon + (fundamentally incompatible with v1.0's per-tag concurrency). +- The separate 5-second keepalive timer — replaced with a 500 ms + internal cadence matching the spec §3.4 inactivity window. ## 0.2.2 — 2026-04-08 diff --git a/Cargo.lock b/Cargo.lock index 04b0e5a..7a7998c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -79,6 +79,12 @@ version = "2.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af" +[[package]] +name = "byteorder" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" + [[package]] name = "bytes" version = "1.11.1" @@ -161,24 +167,27 @@ checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" [[package]] name = "donglora-client" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "343d750fe1e5dad94d4b31e6dbbb05214b39e66f3c94a2ff2a29b0031ffd2826" +version = "1.0.0-alpha.1" dependencies = [ - "anyhow", - "serialport", + "donglora-protocol", + "rand", + "thiserror", + "tokio", + "tokio-serial", + "tokio-stream", "tracing", - "ucobs", ] [[package]] name = "donglora-mux" -version = "0.2.2" +version = "1.0.0-alpha.1" dependencies = [ "anyhow", "clap", "donglora-client", + "donglora-protocol", "fd-lock", + "heapless", "tokio", "tokio-serial", "tokio-util", @@ -186,6 +195,14 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "donglora-protocol" +version = "0.1.0" +dependencies = [ + "heapless", + "ucobs", +] + [[package]] name = "errno" version = "0.3.14" @@ -295,6 +312,37 @@ dependencies = [ "slab", ] +[[package]] +name = "getrandom" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "wasip2", +] + +[[package]] +name = "hash32" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47d60b12902ba28e2730cd37e95b8c9223af2808df9e902d4df49588d1470606" +dependencies = [ + "byteorder", +] + +[[package]] +name = "heapless" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2af2455f757db2b292a9b1768c4b70186d443bcb3b316252d6b540aec1cd89ed" +dependencies = [ + "hash32", + "stable_deref_trait", +] + [[package]] name = "heck" version = "0.5.0" @@ -329,26 +377,6 @@ version = "0.2.184" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "48f5d2a454e16a5ea0f4ced81bd44e4cfc7bd3a507b61887c99fd3538b28e4af" -[[package]] -name = "libudev" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "78b324152da65df7bb95acfcaab55e3097ceaab02fb19b228a9eb74d55f135e0" -dependencies = [ - "libc", - "libudev-sys", -] - -[[package]] -name = "libudev-sys" -version = "0.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3c8469b4a23b962c1396b9b451dda50ef5b283e8dd309d69033475fa9b334324" -dependencies = [ - "libc", - "pkg-config", -] - [[package]] name = "linux-raw-sys" version = "0.12.1" @@ -493,10 +521,13 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" [[package]] -name = "pkg-config" -version = "0.3.32" +name = "ppv-lite86" +version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] [[package]] name = "proc-macro2" @@ -516,6 +547,41 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + +[[package]] +name = "rand" +version = "0.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" +dependencies = [ + "rand_chacha", + "rand_core", +] + +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -572,7 +638,6 @@ dependencies = [ "core-foundation", "core-foundation-sys", "io-kit-sys", - "libudev", "mach2", "nix 0.26.4", "scopeguard", @@ -621,6 +686,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "stable_deref_trait" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" + [[package]] name = "strsim" version = "0.11.1" @@ -709,6 +780,17 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-stream" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + [[package]] name = "tokio-util" version = "0.7.18" @@ -822,6 +904,15 @@ version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" +[[package]] +name = "wasip2" +version = "1.0.3+wasi-0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "20064672db26d7cdc89c7798c48a0fdfac8213434a1186e5ef29fd560ae223d6" +dependencies = [ + "wit-bindgen", +] + [[package]] name = "winapi" version = "0.3.9" @@ -940,3 +1031,29 @@ name = "windows_x86_64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + +[[package]] +name = "wit-bindgen" +version = "0.57.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" + +[[package]] +name = "zerocopy" +version = "0.8.48" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eed437bf9d6692032087e337407a86f04cd8d6a16a37199ed57949d415bd68e9" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.48" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "70e3cd084b1788766f53af483dd21f93881ff30d7320490ec3ef7526d203bad4" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] diff --git a/Cargo.toml b/Cargo.toml index 46dd6fc..59f98ac 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "donglora-mux" -version = "0.2.2" +version = "1.0.0-alpha.1" edition = "2024" rust-version = "1.93" authors = ["Stephen Waits "] @@ -13,7 +13,11 @@ categories = ["hardware-support"] exclude = [".jj/", "deny.toml", "justfile", "mise.toml", "rustfmt.toml"] [dependencies] -donglora-client = "0.2" +# Mux speaks raw DongLoRa Protocol frames — pull the wire codec directly from the +# protocol crate rather than routing through the high-level Dongle API. +donglora-protocol = { path = "../protocol", default-features = false, features = ["std"] } +# Still used for USB discovery helpers (`find_port`, `wait_for_device`). +donglora-client = { path = "../client-rs", version = "1.0.0-alpha.1" } anyhow = "1" clap = { version = "4", features = ["derive"] } tokio = { version = "1", features = ["full"] } @@ -22,6 +26,7 @@ tokio-util = "0.7" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } fd-lock = "4" +heapless = "0.9" [lints.clippy] unwrap_used = "deny" diff --git a/src/daemon.rs b/src/daemon.rs index e56781f..008b658 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -1,35 +1,37 @@ -//! Core mux daemon: dongle I/O, client management, reconnect loop. +//! Core mux daemon — DongLoRa Protocol v2 tag-aware frame routing. //! -//! The daemon owns the USB serial connection and exposes Unix domain socket -//! and optional TCP listeners. It routes commands from clients to the dongle -//! and broadcasts/routes responses back. +//! Owns the USB serial connection and exposes Unix-domain + TCP listeners. +//! Every inbound client frame goes through a [`TagMapper`] that allocates a +//! fresh **device tag** and remembers the owning `(client, upstream_tag)` so +//! the dongle's reply can be routed back. Async frames from the dongle +//! (`tag == 0`: `RX`, async `ERR(EFRAME)`) fan out to every client. +//! +//! Additional spec-mandated behaviour: +//! +//! - **§13.2** — `SET_CONFIG` arbitration with `APPLIED` / `ALREADY_MATCHED` +//! / `LOCKED_MISMATCH` results and owner tracking. Handled in +//! [`crate::intercept`]. +//! - **§13.4** — TX loopback: on `TX_DONE(TRANSMITTED)` the mux synthesizes +//! an RX frame with `origin = LocalLoopback` and fans it out to every +//! client that didn't send the original TX. use std::collections::HashMap; use std::path::Path; use std::sync::Arc; use std::time::Duration; -use donglora_client::protocol::{ - CMD_TAG_SET_CONFIG, CMD_TAG_START_RX, CMD_TAG_STOP_RX, RADIO_CONFIG_SIZE, RESP_TAG_OK, RESP_TAG_RX_PACKET, +use donglora_protocol::{ + Command, ErrorCode, FrameDecoder, FrameResult, MAX_WIRE_FRAME, OkPayload, Owner, RxOrigin, RxPayload, + SetConfigResultCode, TxDonePayload, TxResult, commands, encode_frame, events, }; -use donglora_client::{FrameReader, encode_frame}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::{TcpListener, UnixListener}; -use tokio::sync::{Mutex, mpsc, oneshot}; - -/// A queued command: (client_id, raw_cmd_bytes). -type QueuedCmd = (u64, Vec); - -/// Shared command receiver, wrapped for use across reconnect cycles. -type SharedCmdRx = Arc>>; +use tokio::sync::{Mutex, mpsc}; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; -use crate::intercept::{self, MuxState}; -use crate::session::ClientSession; - -/// Command timeout waiting for dongle response. -const COMMAND_TIMEOUT: Duration = Duration::from_secs(10); +use crate::intercept::{self, Decision, MuxState}; +use crate::session::{ClientId, ClientSession}; /// Delay between dongle reconnect attempts. const RECONNECT_DELAY: Duration = Duration::from_secs(2); @@ -37,21 +39,86 @@ const RECONNECT_DELAY: Duration = Duration::from_secs(2); /// Max consecutive dongle read errors before declaring disconnect. const MAX_CONSECUTIVE_ERRORS: u32 = 3; -/// Sentinel client ID for synthetic commands (e.g. auto StopRx). -const SYNTHETIC_CLIENT_ID: u64 = u64::MAX; +/// Keepalive cadence for the mux → dongle connection. Matches the +/// client-rs `KEEPALIVE_INTERVAL` with 2× margin on the 1 s inactivity +/// window (spec §3.4). +const KEEPALIVE_INTERVAL: Duration = Duration::from_millis(500); -/// Keepalive ping interval: if no commands flow for this long, ping the dongle. -const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(5); +/// Queued inbound work for the dongle writer. +#[derive(Debug)] +enum WriterWork { + /// Forward a client frame. Payload is the pre-wire `(type_id, payload)` + /// — writer allocates a fresh device tag, records the mapping, and + /// re-encodes. + Forward { client_id: ClientId, type_id: u8, upstream_tag: u16, payload: Vec }, + /// Periodic mux-originated keepalive ping. No client owns this. + Keepalive, +} -/// How long to wait for a keepalive Pong before declaring the dongle lost. -const KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(2); +/// Maps mux-allocated device tags back to `(client, upstream_tag, cmd_type)`. +struct TagMapper { + next: u16, + pending: HashMap, +} -/// Shared mux state, protected by async mutex. +struct Pending { + client_id: ClientId, + upstream_tag: u16, + cmd_type: u8, +} + +impl TagMapper { + fn new() -> Self { + Self { next: 1, pending: HashMap::new() } + } + + /// Allocate a fresh device tag. Skips 0 and wraps at `u16::MAX`. + fn alloc(&mut self, client_id: ClientId, upstream_tag: u16, cmd_type: u8) -> u16 { + // Linear probe for an unused tag. In practice the pending map is + // small (tens of entries max) so this is cheap. + loop { + let tag = self.next; + self.next = self.next.wrapping_add(1); + if self.next == 0 { + self.next = 1; + } + if tag == 0 || self.pending.contains_key(&tag) { + continue; + } + self.pending.insert(tag, Pending { client_id, upstream_tag, cmd_type }); + return tag; + } + } + + /// Final-resolve and remove the mapping. + fn take(&mut self, device_tag: u16) -> Option { + self.pending.remove(&device_tag) + } + + /// Peek without removing (used for TX two-phase completion where the + /// OK is observed but TX_DONE will still arrive under the same tag). + fn peek(&self, device_tag: u16) -> Option<&Pending> { + self.pending.get(&device_tag) + } + + /// Drop every mapping owned by this client (on disconnect). + fn drop_client(&mut self, client_id: ClientId) { + self.pending.retain(|_, p| p.client_id != client_id); + } + + #[cfg(test)] + fn len(&self) -> usize { + self.pending.len() + } +} + +/// Shared mux state. type SharedState = Arc>; struct MuxInner { - sessions: HashMap, + sessions: HashMap, mux_state: MuxState, + tag_mapper: TagMapper, } /// The multiplexer daemon. @@ -74,23 +141,20 @@ impl MuxDaemon { /// Run the mux daemon. Returns when shutdown is signalled. pub async fn run(&self) -> anyhow::Result<()> { - let state: SharedState = - Arc::new(Mutex::new(MuxInner { sessions: HashMap::new(), mux_state: MuxState::new() })); + let state: SharedState = Arc::new(Mutex::new(MuxInner { + sessions: HashMap::new(), + mux_state: MuxState::new(), + tag_mapper: TagMapper::new(), + })); - // Command channel: client handlers → dongle writer - let (cmd_tx, cmd_rx) = mpsc::channel::<(u64, Vec)>(64); - let cmd_rx = Arc::new(Mutex::new(cmd_rx)); + // Work queue: client handlers → dongle writer. + let (work_tx, work_rx) = mpsc::channel::(64); + let work_rx = Arc::new(Mutex::new(work_rx)); - // Pending response slot: dongle writer installs, dongle reader resolves - let pending_response: Arc>>>> = Arc::new(Mutex::new(None)); - - // Prepare socket directory let sock_path = Path::new(&self.socket_path); if let Some(parent) = sock_path.parent() { tokio::fs::create_dir_all(parent).await.ok(); } - // Acquire an exclusive lock to prevent multiple mux instances. - // The lock is held for the process lifetime and auto-released on exit (even SIGKILL). let lock_path = format!("{}.lock", self.socket_path); let lock_file = std::fs::OpenOptions::new() .create(true) @@ -105,17 +169,14 @@ impl MuxDaemon { }; if sock_path.exists() { - // Lock acquired but socket file exists — stale from a crashed instance. info!("removing stale socket {}", self.socket_path); tokio::fs::remove_file(sock_path).await.ok(); } - // Start Unix socket server let unix_listener = UnixListener::bind(&self.socket_path) .map_err(|e| anyhow::anyhow!("failed to bind Unix socket {}: {e}", self.socket_path))?; info!("Unix socket listening on {}", self.socket_path); - // Start TCP server (optional) let tcp_listener = if let Some((ref host, port)) = self.tcp_addr { let addr = format!("{host}:{port}"); let listener = TcpListener::bind(&addr).await.map_err(|e| { @@ -131,16 +192,13 @@ impl MuxDaemon { None }; - // Spawn client accept tasks - tokio::spawn(accept_unix_clients(unix_listener, state.clone(), cmd_tx.clone(), self.shutdown.clone())); + tokio::spawn(accept_unix_clients(unix_listener, state.clone(), work_tx.clone(), self.shutdown.clone())); if let Some(listener) = tcp_listener { - tokio::spawn(accept_tcp_clients(listener, state.clone(), cmd_tx.clone(), self.shutdown.clone())); + tokio::spawn(accept_tcp_clients(listener, state.clone(), work_tx.clone(), self.shutdown.clone())); } - // Connect/reconnect loop - self.reconnect_loop(state, cmd_tx, cmd_rx, pending_response).await; + self.reconnect_loop(state, work_tx, work_rx).await; - // Cleanup info!("shutting down..."); if Path::new(&self.socket_path).exists() { tokio::fs::remove_file(&self.socket_path).await.ok(); @@ -152,18 +210,13 @@ impl MuxDaemon { async fn reconnect_loop( &self, state: SharedState, - cmd_tx: mpsc::Sender<(u64, Vec)>, - cmd_rx: SharedCmdRx, - pending_response: Arc>>>>, + work_tx: mpsc::Sender, + work_rx: Arc>>, ) { - let mut saved_config: Option<[u8; RADIO_CONFIG_SIZE]> = None; - let mut had_rx_interest = false; - loop { if self.shutdown.is_cancelled() { break; } - info!("opening dongle on {}", self.serial_port); let (serial, _serial_lock_guard) = loop { if self.shutdown.is_cancelled() { @@ -183,38 +236,24 @@ impl MuxDaemon { info!("dongle connected"); let (serial_read, serial_write) = tokio::io::split(serial); - let dongle_lost = CancellationToken::new(); - let reader_handle = tokio::spawn(dongle_reader( - serial_read, - state.clone(), - pending_response.clone(), - dongle_lost.clone(), - self.shutdown.clone(), - )); + let reader_handle = + tokio::spawn(dongle_reader(serial_read, state.clone(), dongle_lost.clone(), self.shutdown.clone())); let writer_handle = tokio::spawn(dongle_writer( serial_write, - cmd_rx.clone(), + work_rx.clone(), state.clone(), - pending_response.clone(), dongle_lost.clone(), self.shutdown.clone(), )); - // Re-establish radio state from previous session - if let Some(config) = saved_config.take() { - info!("restoring radio config after reconnect"); - let mut cmd = vec![CMD_TAG_SET_CONFIG]; - cmd.extend_from_slice(&config); - let _ = cmd_tx.send((SYNTHETIC_CLIENT_ID, cmd)).await; - if had_rx_interest { - let _ = cmd_tx.send((SYNTHETIC_CLIENT_ID, vec![CMD_TAG_START_RX])).await; - } - } + // Drive keepalive from a dedicated timer task so the writer + // loop doesn't need to interleave. + let keepalive_handle = + tokio::spawn(keepalive_loop(work_tx.clone(), dongle_lost.clone(), self.shutdown.clone())); - // Wait for dongle loss or shutdown tokio::select! { () = self.shutdown.cancelled() => {}, () = dongle_lost.cancelled() => {}, @@ -222,20 +261,26 @@ impl MuxDaemon { reader_handle.abort(); writer_handle.abort(); + keepalive_handle.abort(); let _ = reader_handle.await; let _ = writer_handle.await; + let _ = keepalive_handle.await; - // Drain pending commands + // Drain any queued work. Mappings will get cleared on reconnect + // because the lock owner's client may still be live. { - let mut rx = cmd_rx.lock().await; + let mut rx = work_rx.lock().await; while rx.try_recv().is_ok() {} } - // Save radio state for restoration after reconnect + // Wake any clients that had commands in flight by clearing + // the pending map. Their session handlers notice the writer + // isn't producing replies and eventually time out — but the + // cleaner thing is to let the dongle-writer reinstall pending + // mappings as their clients retry. { let mut inner = state.lock().await; - saved_config = inner.mux_state.locked_config.take(); - had_rx_interest = intercept::rx_interest_count(&inner.sessions) > 0; + inner.tag_mapper.pending.clear(); } if !self.shutdown.is_cancelled() { @@ -257,15 +302,7 @@ fn serial_lock_path(port: &str) -> String { } /// Open the serial port with an exclusive lock and verify the dongle responds to ping. -/// -/// Returns the serial stream and the lock. The caller must keep `_serial_lock` alive -/// for the duration of the dongle session. Dropping it releases the exclusive flock, -/// allowing other processes to claim the port. -async fn open_and_ping( - port: &str, -) -> anyhow::Result<(tokio_serial::SerialStream, fd_lock::RwLock)> { - // Acquire exclusive flock before opening the serial port. This prevents bridges - // or other processes from directly opening the same device while the mux owns it. +async fn open_and_ping(port: &str) -> anyhow::Result<(tokio_serial::SerialStream, fd_lock::RwLock)> { let lock_path = serial_lock_path(port); let lock_file = std::fs::OpenOptions::new() .create(true) @@ -274,27 +311,23 @@ async fn open_and_ping( .open(&lock_path) .map_err(|e| anyhow::anyhow!("failed to open serial lock file {lock_path}: {e}"))?; let mut lock = fd_lock::RwLock::new(lock_file); - // Acquire exclusive flock (non-blocking). Fails if another process holds it. - let guard = lock - .try_write() - .map_err(|_| anyhow::anyhow!("serial port {port} is locked by another process"))?; - // Forget the guard so its Drop doesn't call flock(UNLOCK). The underlying flock - // stays held as long as the file descriptor (owned by `lock`) remains open. - // When `lock` is dropped at the end of the dongle session, the fd closes and - // the OS releases the flock automatically. + let guard = lock.try_write().map_err(|_| anyhow::anyhow!("serial port {port} is locked by another process"))?; std::mem::forget(guard); let builder = tokio_serial::new(port, 115_200).timeout(Duration::from_secs(2)); let mut serial = tokio_serial::SerialStream::open(&builder).map_err(|e| anyhow::anyhow!("failed to open {port}: {e}"))?; - let ping_frame = encode_frame(&[0]); - AsyncWriteExt::write_all(&mut serial, &ping_frame) + // PING the dongle (tag=1) to verify it's alive. + let mut wire = [0u8; MAX_WIRE_FRAME]; + let n = encode_frame(commands::TYPE_PING, 1, &[], &mut wire) + .map_err(|e| anyhow::anyhow!("failed to encode PING: {e:?}"))?; + AsyncWriteExt::write_all(&mut serial, &wire[..n]) .await .map_err(|e| anyhow::anyhow!("failed to write ping: {e}"))?; - let mut reader = FrameReader::new(); - let mut buf = [0u8; 64]; + let mut decoder = FrameDecoder::new(); + let mut buf = [0u8; 256]; let deadline = tokio::time::Instant::now() + Duration::from_secs(2); loop { @@ -302,25 +335,43 @@ async fn open_and_ping( if remaining.is_zero() { anyhow::bail!("ping timeout"); } - let n = tokio::select! { result = serial.read(&mut buf) => { result.map_err(|e| anyhow::anyhow!("serial read error: {e}"))? } - () = tokio::time::sleep(remaining) => { - anyhow::bail!("ping timeout"); - } + () = tokio::time::sleep(remaining) => anyhow::bail!("ping timeout"), }; - if n == 0 { anyhow::bail!("serial port closed during ping"); } - - for frame in reader.feed(&buf[..n]) { - if frame.first() == Some(&0) { - info!("dongle responded to Ping"); - return Ok((serial, lock)); + let mut got = false; + decoder.feed(&buf[..n], |res| { + if let FrameResult::Ok { type_id, tag: _, payload: _ } = res + && type_id == events::TYPE_OK + { + got = true; } + }); + if got { + info!("dongle responded to PING"); + return Ok((serial, lock)); + } + } +} + +async fn keepalive_loop( + work_tx: mpsc::Sender, + dongle_lost: CancellationToken, + shutdown: CancellationToken, +) { + loop { + tokio::select! { + () = shutdown.cancelled() => return, + () = dongle_lost.cancelled() => return, + () = tokio::time::sleep(KEEPALIVE_INTERVAL) => {}, + } + if work_tx.send(WriterWork::Keepalive).await.is_err() { + return; } } } @@ -328,11 +379,10 @@ async fn open_and_ping( async fn dongle_reader( mut serial: tokio::io::ReadHalf, state: SharedState, - pending_response: Arc>>>>, dongle_lost: CancellationToken, shutdown: CancellationToken, ) { - let mut reader = FrameReader::new(); + let mut decoder = FrameDecoder::new(); let mut buf = [0u8; 512]; let mut consecutive_errors: u32 = 0; @@ -340,31 +390,44 @@ async fn dongle_reader( if shutdown.is_cancelled() || dongle_lost.is_cancelled() { return; } - let read_result = tokio::select! { result = serial.read(&mut buf) => result, () = shutdown.cancelled() => return, () = dongle_lost.cancelled() => return, }; - match read_result { Ok(0) => { error!("serial port closed"); - resolve_pending_with_error(&pending_response).await; dongle_lost.cancel(); return; } Ok(n) => { consecutive_errors = 0; - for frame in reader.feed(&buf[..n]) { - dispatch_frame(&frame, &state, &pending_response).await; + // Collect frames up-front because the decoder's callback + // borrows the payload from its internal buffer. + let mut works: Vec<(u8, u16, Vec)> = Vec::new(); + let mut frame_errs = 0usize; + decoder.feed(&buf[..n], |res| match res { + FrameResult::Ok { type_id, tag, payload } => { + works.push((type_id, tag, payload.to_vec())); + } + FrameResult::Err(_) => frame_errs += 1, + }); + for (type_id, tag, payload) in works { + dispatch_dongle_frame(type_id, tag, &payload, &state).await; + } + if frame_errs > 0 { + warn!("{frame_errs} corrupt frame(s) from dongle (CRC or COBS)"); + // Forward synthetic async ERR(EFRAME) to all clients — + // §2.5 allows the device to emit these and we want clients + // to be able to observe framing issues on the link. + fanout_async_err(&state, ErrorCode::EFrame).await; } } Err(e) => { consecutive_errors += 1; if consecutive_errors >= MAX_CONSECUTIVE_ERRORS { error!("dongle read error (giving up after {consecutive_errors}): {e}"); - resolve_pending_with_error(&pending_response).await; dongle_lost.cancel(); return; } @@ -375,176 +438,215 @@ async fn dongle_reader( } } -async fn dispatch_frame( - frame: &[u8], - state: &SharedState, - pending_response: &Arc>>>>, -) { - let Some(&tag) = frame.first() else { +async fn fanout_async_err(state: &SharedState, code: ErrorCode) { + let mut buf = [0u8; 2]; + if events::encode_err_payload(code, &mut buf).is_err() { + return; + } + let mut wire = [0u8; MAX_WIRE_FRAME]; + let n = match encode_frame(events::TYPE_ERR, 0, &buf, &mut wire) { + Ok(n) => n, + Err(_) => return, + }; + let frame = wire[..n].to_vec(); + let inner = state.lock().await; + for session in inner.sessions.values() { + session.enqueue(frame.clone()); + } +} + +async fn dispatch_dongle_frame(type_id: u8, tag: u16, payload: &[u8], state: &SharedState) { + // Async events (tag == 0): fan out to all clients verbatim. + if tag == 0 { + let mut wire = [0u8; MAX_WIRE_FRAME]; + let n = match encode_frame(type_id, 0, payload, &mut wire) { + Ok(n) => n, + Err(e) => { + warn!("re-encode of async frame failed: {e:?}"); + return; + } + }; + let frame = wire[..n].to_vec(); + let inner = state.lock().await; + for session in inner.sessions.values() { + session.enqueue(frame.clone()); + } + return; + } + + // Tag-correlated: look up pending mapping. + let mut inner = state.lock().await; + // TX has a two-phase completion (OK -> TX_DONE). On the intermediate + // OK, keep the mapping alive; remove only on the terminal frame. + let pending_snapshot = inner.tag_mapper.peek(tag).map(|p| (p.client_id, p.upstream_tag, p.cmd_type)); + let Some((client_id, upstream_tag, cmd_type)) = pending_snapshot else { + warn!("unsolicited frame type={type_id:#04x} tag={tag} — no pending mapping"); return; }; - if tag == RESP_TAG_RX_PACKET { - let cobs_frame = encode_frame(frame); - let inner = state.lock().await; - for session in inner.sessions.values() { - session.enqueue(cobs_frame.clone()); - } - } else { - let mut slot = pending_response.lock().await; - if let Some(sender) = slot.take() { - let _ = sender.send(frame.to_vec()); - } else { - warn!("unsolicited non-RxPacket response (tag {tag}) — dropped"); - } + // Decide if this frame is terminal for the mapping. + let terminal = !(cmd_type == commands::TYPE_TX && type_id == events::TYPE_OK); + if terminal { + inner.tag_mapper.take(tag); } -} -async fn resolve_pending_with_error(pending_response: &Arc>>>>) { - let mut slot = pending_response.lock().await; - if let Some(sender) = slot.take() { - let _ = sender.send(vec![5, 1]); // Error(RadioBusy) - } -} - -/// Send a Ping and wait for Pong. Returns `true` if the dongle responded. -async fn keepalive_ping( - serial: &mut tokio::io::WriteHalf, - pending_response: &Arc>>>>, -) -> bool { - let (resp_tx, resp_rx) = oneshot::channel(); + // Lock-installation: on a successful SET_CONFIG APPLIED, record the + // `(client, modulation)` lock so subsequent clients see synthesized + // ALREADY_MATCHED / LOCKED_MISMATCH. + if cmd_type == commands::TYPE_SET_CONFIG + && type_id == events::TYPE_OK + && let Ok(OkPayload::SetConfig(result)) = OkPayload::parse_for(cmd_type, payload) + && result.result == SetConfigResultCode::Applied + && result.owner == Owner::Mine { - let mut slot = pending_response.lock().await; - *slot = Some(resp_tx); + inner.mux_state.locked = Some((client_id, result.current)); + info!("SET_CONFIG applied by client {client_id} — lock installed"); } - let ping_frame = encode_frame(&[0]); - if AsyncWriteExt::write_all(serial, &ping_frame).await.is_err() { - return false; + // Drop from TX cache on terminal TX_DONE, regardless of outcome. + let mut tx_loopback: Option> = None; + if cmd_type == commands::TYPE_TX + && type_id == events::TYPE_TX_DONE + && let Some(session) = inner.sessions.get_mut(&client_id) + && let Some(bytes) = session.tx_cache.remove(&upstream_tag) + && let Ok(td) = TxDonePayload::decode(payload) + && td.result == TxResult::Transmitted + { + tx_loopback = Some(bytes); } - match tokio::time::timeout(KEEPALIVE_TIMEOUT, resp_rx).await { - Ok(Ok(_)) => true, - _ => { - let mut slot = pending_response.lock().await; - *slot = None; - false + // Track RX interest on RX_START/RX_STOP OK replies so state matches. + if type_id == events::TYPE_OK { + match cmd_type { + commands::TYPE_RX_START => { + if let Some(s) = inner.sessions.get_mut(&client_id) { + s.rx_interested = true; + } + } + commands::TYPE_RX_STOP => { + if let Some(s) = inner.sessions.get_mut(&client_id) { + s.rx_interested = false; + } + } + _ => {} } } + + // Re-encode with the upstream tag and route to the owning client. + let mut wire = [0u8; MAX_WIRE_FRAME]; + let n = match encode_frame(type_id, upstream_tag, payload, &mut wire) { + Ok(n) => n, + Err(e) => { + warn!("re-encode of downstream frame failed: {e:?}"); + return; + } + }; + let frame = wire[..n].to_vec(); + if let Some(session) = inner.sessions.get(&client_id) { + session.enqueue(frame); + } + + // TX loopback fan-out: synthesize an RX frame to every OTHER client. + if let Some(data) = tx_loopback { + let mut rx_data = heapless::Vec::::new(); + if rx_data.extend_from_slice(&data).is_ok() { + let rx = RxPayload { + rssi_tenths_dbm: 0, + snr_tenths_db: 0, + freq_err_hz: 0, + timestamp_us: now_us(), + crc_valid: true, + packets_dropped: 0, + origin: RxOrigin::LocalLoopback, + data: rx_data, + }; + let mut rxbuf = [0u8; RxPayload::METADATA_SIZE + donglora_protocol::MAX_OTA_PAYLOAD]; + if let Ok(n) = rx.encode(&mut rxbuf) { + let mut wirebuf = [0u8; MAX_WIRE_FRAME]; + if let Ok(n2) = encode_frame(events::TYPE_RX, 0, &rxbuf[..n], &mut wirebuf) { + let loop_frame = wirebuf[..n2].to_vec(); + for (id, session) in &inner.sessions { + if *id != client_id { + session.enqueue(loop_frame.clone()); + } + } + } + } + } + } +} + +fn now_us() -> u64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now().duration_since(UNIX_EPOCH).map_or(0, |d| d.as_micros().min(u128::from(u64::MAX)) as u64) } async fn dongle_writer( mut serial: tokio::io::WriteHalf, - cmd_rx: SharedCmdRx, + work_rx: Arc>>, state: SharedState, - pending_response: Arc>>>>, dongle_lost: CancellationToken, shutdown: CancellationToken, ) { - let keepalive = tokio::time::sleep(KEEPALIVE_INTERVAL); - tokio::pin!(keepalive); - loop { - // Pull next command, or fire keepalive if idle too long - let cmd = { - let mut rx = cmd_rx.lock().await; + let work = { + let mut rx = work_rx.lock().await; tokio::select! { - cmd = rx.recv() => cmd, - () = &mut keepalive => None, + w = rx.recv() => w, () = shutdown.cancelled() => return, () = dongle_lost.cancelled() => return, } }; - - keepalive.as_mut().reset(tokio::time::Instant::now() + KEEPALIVE_INTERVAL); - - let (client_id, raw_cmd) = match cmd { - Some(c) => c, - None => { - debug!("keepalive ping"); - if !keepalive_ping(&mut serial, &pending_response).await { - error!("keepalive ping failed - dongle lost"); + let Some(work) = work else { return }; + match work { + WriterWork::Keepalive => { + // Allocate a synthetic mapping so the reply is absorbed + // cleanly. Use client_id = u64::MAX as the "synthetic" + // marker; there's never a real client with that id. + let device_tag = { + let mut inner = state.lock().await; + inner.tag_mapper.alloc(u64::MAX, 0, commands::TYPE_PING) + }; + let mut wire = [0u8; MAX_WIRE_FRAME]; + let n = match encode_frame(commands::TYPE_PING, device_tag, &[], &mut wire) { + Ok(n) => n, + Err(e) => { + warn!("keepalive encode failed: {e:?}"); + continue; + } + }; + if let Err(e) = AsyncWriteExt::write_all(&mut serial, &wire[..n]).await { + error!("dongle write error (keepalive): {e}"); dongle_lost.cancel(); return; } - continue; } - }; - - // Check for interception - let intercepted = { - let mut inner = state.lock().await; - let locked = inner.mux_state.locked_config; - intercept::maybe_intercept(&raw_cmd, client_id, &mut inner.sessions, &locked) - }; - if let Some(synthetic) = intercepted { - let inner = state.lock().await; - if let Some(session) = inner.sessions.get(&client_id) { - session.enqueue(encode_frame(&synthetic)); - } - continue; - } - - // Install oneshot for the response - let (resp_tx, resp_rx) = oneshot::channel(); - { - let mut slot = pending_response.lock().await; - *slot = Some(resp_tx); - } - - // Send command to dongle - let cobs_frame = encode_frame(&raw_cmd); - if let Err(e) = AsyncWriteExt::write_all(&mut serial, &cobs_frame).await { - error!("dongle write error: {e}"); - resolve_pending_with_error(&pending_response).await; - dongle_lost.cancel(); - return; - } - - // Wait for response - let response = match tokio::time::timeout(COMMAND_TIMEOUT, resp_rx).await { - Ok(Ok(resp)) => resp, - Ok(Err(_)) => { - warn!("response channel dropped"); - vec![5, 1] - } - Err(_) => { - error!("dongle response timeout"); - let mut slot = pending_response.lock().await; - *slot = None; - vec![5, 1] - } - }; - - // Track state on successful forward - let cmd_tag = raw_cmd.first().copied().unwrap_or(0xFF); - let resp_tag = response.first().copied().unwrap_or(0xFF); - { - let mut inner = state.lock().await; - if cmd_tag == CMD_TAG_SET_CONFIG && resp_tag == RESP_TAG_OK { - if let Some(config_bytes) = raw_cmd.get(1..1 + RADIO_CONFIG_SIZE) { - let mut locked = [0u8; RADIO_CONFIG_SIZE]; - locked.copy_from_slice(config_bytes); - inner.mux_state.locked_config = Some(locked); - info!("radio config locked"); + WriterWork::Forward { client_id, type_id, upstream_tag, payload } => { + let device_tag = { + let mut inner = state.lock().await; + // If this is a TX, stash the payload so the reader + // can fan out on TX_DONE(TRANSMITTED). + if type_id == commands::TYPE_TX + && payload.len() > 1 + && let Some(session) = inner.sessions.get_mut(&client_id) + { + session.tx_cache.insert(upstream_tag, payload[1..].to_vec()); + } + inner.tag_mapper.alloc(client_id, upstream_tag, type_id) + }; + let mut wire = [0u8; MAX_WIRE_FRAME]; + let n = match encode_frame(type_id, device_tag, &payload, &mut wire) { + Ok(n) => n, + Err(e) => { + warn!("forward encode failed: {e:?}"); + continue; + } + }; + if let Err(e) = AsyncWriteExt::write_all(&mut serial, &wire[..n]).await { + error!("dongle write error: {e}"); + dongle_lost.cancel(); + return; } - } else if cmd_tag == CMD_TAG_START_RX && resp_tag == RESP_TAG_OK { - if let Some(session) = inner.sessions.get_mut(&client_id) { - session.rx_interested = true; - } - } else if cmd_tag == CMD_TAG_STOP_RX - && resp_tag == RESP_TAG_OK - && let Some(session) = inner.sessions.get_mut(&client_id) - { - session.rx_interested = false; - } - } - - // Route response to commanding client - if client_id != SYNTHETIC_CLIENT_ID { - let inner = state.lock().await; - if let Some(session) = inner.sessions.get(&client_id) { - session.enqueue(encode_frame(&response)); } } } @@ -555,7 +657,7 @@ async fn dongle_writer( async fn accept_unix_clients( listener: UnixListener, state: SharedState, - cmd_tx: mpsc::Sender<(u64, Vec)>, + work_tx: mpsc::Sender, shutdown: CancellationToken, ) { loop { @@ -563,11 +665,10 @@ async fn accept_unix_clients( result = listener.accept() => result, () = shutdown.cancelled() => return, }; - match accept_result { Ok((stream, _addr)) => { let (read_half, write_half) = stream.into_split(); - spawn_client(read_half, write_half, state.clone(), cmd_tx.clone(), shutdown.clone()); + spawn_client(read_half, write_half, state.clone(), work_tx.clone(), shutdown.clone()); } Err(e) => warn!("Unix accept error: {e}"), } @@ -577,7 +678,7 @@ async fn accept_unix_clients( async fn accept_tcp_clients( listener: TcpListener, state: SharedState, - cmd_tx: mpsc::Sender<(u64, Vec)>, + work_tx: mpsc::Sender, shutdown: CancellationToken, ) { loop { @@ -585,14 +686,12 @@ async fn accept_tcp_clients( result = listener.accept() => result, () = shutdown.cancelled() => return, }; - match accept_result { Ok((stream, addr)) => { debug!("TCP connection from {addr}"); - // Low-latency: disable Nagle (COBS frames are small) let _ = stream.set_nodelay(true); let (read_half, write_half) = stream.into_split(); - spawn_client(read_half, write_half, state.clone(), cmd_tx.clone(), shutdown.clone()); + spawn_client(read_half, write_half, state.clone(), work_tx.clone(), shutdown.clone()); } Err(e) => warn!("TCP accept error: {e}"), } @@ -603,7 +702,7 @@ fn spawn_client( read_half: R, write_half: W, state: SharedState, - cmd_tx: mpsc::Sender<(u64, Vec)>, + work_tx: mpsc::Sender, shutdown: CancellationToken, ) where R: AsyncReadExt + Unpin + Send + 'static, @@ -614,7 +713,7 @@ fn spawn_client( let label = session.label(); let state_clone = state.clone(); - let cmd_tx_clone = cmd_tx.clone(); + let work_tx_clone = work_tx.clone(); tokio::spawn(async move { { @@ -622,44 +721,113 @@ fn spawn_client( inner.sessions.insert(client_id, session); info!("{label} connected ({} total)", inner.sessions.len()); } - let writer_shutdown = shutdown.clone(); - let writer_handle = tokio::spawn(client_writer(write_half, send_rx, writer_shutdown)); - - client_reader(read_half, client_id, cmd_tx_clone.clone(), shutdown).await; - + let writer_handle = tokio::spawn(client_writer(write_half, send_rx, client_id, writer_shutdown)); + client_reader(read_half, client_id, work_tx_clone.clone(), state_clone.clone(), shutdown).await; writer_handle.abort(); let _ = writer_handle.await; - - remove_client(client_id, &state_clone, &label, &cmd_tx_clone).await; + remove_client(client_id, &state_clone, &label, &work_tx_clone).await; }); } async fn client_reader( mut read_half: R, - client_id: u64, - cmd_tx: mpsc::Sender<(u64, Vec)>, + client_id: ClientId, + work_tx: mpsc::Sender, + state: SharedState, shutdown: CancellationToken, ) { - let mut reader = FrameReader::new(); + let mut decoder = FrameDecoder::new(); let mut buf = [0u8; 4096]; loop { let read_result = tokio::select! { result = read_half.read(&mut buf) => result, - () = shutdown.cancelled() => return, + () = shutdown.cancelled() => { + debug!("client-{client_id} reader exit: mux shutdown"); + return; + } }; - match read_result { - Ok(0) => return, + Ok(0) => { + debug!("client-{client_id} reader exit: EOF"); + return; + } Ok(n) => { - for frame in reader.feed(&buf[..n]) { - if cmd_tx.send((client_id, frame)).await.is_err() { + let mut to_process: Vec<(u8, u16, Vec)> = Vec::new(); + decoder.feed(&buf[..n], |res| match res { + FrameResult::Ok { type_id, tag, payload } => { + to_process.push((type_id, tag, payload.to_vec())); + } + FrameResult::Err(_) => { + // Corrupt upstream frame — best-effort reply with + // async ERR(EFRAME) (tag=0). + } + }); + for (type_id, upstream_tag, payload) in to_process { + if !handle_client_frame(type_id, upstream_tag, payload, client_id, &work_tx, &state).await { return; } } } - Err(_) => return, + Err(e) => { + debug!("client-{client_id} reader exit: read error: {e}"); + return; + } + } + } +} + +/// Returns `true` to keep reading, `false` on terminal error. +async fn handle_client_frame( + type_id: u8, + upstream_tag: u16, + payload: Vec, + client_id: ClientId, + work_tx: &mpsc::Sender, + state: &SharedState, +) -> bool { + // Decide whether to intercept or forward. + let decision = { + let mut inner = state.lock().await; + let MuxInner { mux_state, sessions, .. } = &mut *inner; + intercept::decide(type_id, &payload, client_id, mux_state, sessions) + }; + match decision { + Decision::Synthesize { type_id: out_type, payload: out_payload } => { + let mut wire = [0u8; MAX_WIRE_FRAME]; + let n = match encode_frame(out_type, upstream_tag, &out_payload, &mut wire) { + Ok(n) => n, + Err(e) => { + warn!("synthesized frame encode failed: {e:?}"); + return true; + } + }; + let inner = state.lock().await; + if let Some(session) = inner.sessions.get(&client_id) { + session.enqueue(wire[..n].to_vec()); + } + true + } + Decision::Drop => true, + Decision::Forward => { + // Unknown/malformed commands: short-circuit with a synthesized + // ERR(EUNKNOWN_CMD) that echoes the upstream tag, so the + // client's per-tag correlation still resolves. Done here so + // the mux's dispatcher stays symmetric with the spec. + if Command::parse(type_id, &payload).is_err() { + let (err_type, err_payload) = intercept::synthesize_err(ErrorCode::EUnknownCmd); + let mut wire = [0u8; MAX_WIRE_FRAME]; + if let Ok(n) = encode_frame(err_type, upstream_tag, &err_payload, &mut wire) { + let inner = state.lock().await; + if let Some(session) = inner.sessions.get(&client_id) { + session.enqueue(wire[..n].to_vec()); + } + } + return true; + } + let send = work_tx.send(WriterWork::Forward { client_id, type_id, upstream_tag, payload }).await; + send.is_ok() } } } @@ -667,45 +835,113 @@ async fn client_reader( async fn client_writer( mut write_half: W, mut send_rx: mpsc::Receiver>, + client_id: ClientId, shutdown: CancellationToken, ) { loop { let frame = tokio::select! { - f = send_rx.recv() => { - match f { - Some(f) => f, - None => return, + f = send_rx.recv() => match f { + Some(f) => f, + None => { + debug!("client-{client_id} writer exit: queue closed"); + return; } + }, + () = shutdown.cancelled() => { + debug!("client-{client_id} writer exit: shutdown"); + return; } - () = shutdown.cancelled() => return, }; - - if AsyncWriteExt::write_all(&mut write_half, &frame).await.is_err() { + if let Err(e) = AsyncWriteExt::write_all(&mut write_half, &frame).await { + debug!("client-{client_id} writer exit: {e}"); return; } } } -async fn remove_client(client_id: u64, state: &SharedState, label: &str, cmd_tx: &mpsc::Sender<(u64, Vec)>) { +async fn remove_client(client_id: ClientId, state: &SharedState, label: &str, work_tx: &mpsc::Sender) { let mut inner = state.lock().await; - - let was_interested = inner.sessions.get(&client_id).is_some_and(|s| s.rx_interested); - + let (was_interested, drops) = + inner.sessions.get(&client_id).map_or((false, 0), |s| (s.rx_interested, s.drop_count())); inner.sessions.remove(&client_id); - info!("{label} disconnected ({} remain)", inner.sessions.len()); + inner.tag_mapper.drop_client(client_id); - // If last RX-interested client gone, send synthetic StopRx - if was_interested && intercept::rx_interest_count(&inner.sessions) == 0 { - info!("last RX-interested client gone — sending StopRx"); - // Drop the lock before sending to avoid deadlock - drop(inner); - let _ = cmd_tx.send((SYNTHETIC_CLIENT_ID, vec![CMD_TAG_STOP_RX])).await; - return; + // Release the config lock if this client held it. + if let Some((owner, _)) = inner.mux_state.locked + && owner == client_id + { + inner.mux_state.locked = None; + info!("config lock released (owner {label} disconnected)"); } - // Reset locked config when all clients are gone - if inner.sessions.is_empty() && inner.mux_state.locked_config.is_some() { - info!("all clients gone — config lock released"); - inner.mux_state.locked_config = None; + if drops > 0 { + info!("{label} disconnected ({} remain, {drops} dropped frames)", inner.sessions.len()); + } else { + info!("{label} disconnected ({} remain)", inner.sessions.len()); + } + + // If last RX-interested client is gone, synthesize an RX_STOP. + if was_interested && MuxState::rx_interest_count(&inner.sessions) == 0 { + drop(inner); + let _ = work_tx + .send(WriterWork::Forward { + client_id: u64::MAX, + type_id: commands::TYPE_RX_STOP, + upstream_tag: 0, + payload: vec![], + }) + .await; + } +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use super::*; + + #[test] + fn tag_mapper_skips_zero() { + let mut m = TagMapper::new(); + // Start at 1; after many allocations we should never see 0. + let tag1 = m.alloc(1, 100, commands::TYPE_PING); + assert_ne!(tag1, 0); + m.take(tag1); + for _ in 0..1024 { + let t = m.alloc(1, 100, commands::TYPE_PING); + assert_ne!(t, 0); + m.take(t); + } + } + + #[test] + fn tag_mapper_take_returns_mapping() { + let mut m = TagMapper::new(); + let tag = m.alloc(42, 7, commands::TYPE_TX); + let p = m.take(tag).unwrap(); + assert_eq!(p.client_id, 42); + assert_eq!(p.upstream_tag, 7); + assert_eq!(p.cmd_type, commands::TYPE_TX); + assert_eq!(m.len(), 0); + } + + #[test] + fn tag_mapper_peek_keeps_mapping() { + let mut m = TagMapper::new(); + let tag = m.alloc(42, 7, commands::TYPE_TX); + let p = m.peek(tag).unwrap(); + assert_eq!(p.client_id, 42); + assert_eq!(m.len(), 1); + } + + #[test] + fn tag_mapper_drop_client_removes_only_that_clients_tags() { + let mut m = TagMapper::new(); + let t_a1 = m.alloc(1, 10, commands::TYPE_PING); + let t_a2 = m.alloc(1, 11, commands::TYPE_PING); + let t_b = m.alloc(2, 12, commands::TYPE_PING); + m.drop_client(1); + assert!(m.take(t_a1).is_none()); + assert!(m.take(t_a2).is_none()); + assert!(m.take(t_b).is_some()); } } diff --git a/src/intercept.rs b/src/intercept.rs index 07efdf0..4e84db1 100644 --- a/src/intercept.rs +++ b/src/intercept.rs @@ -1,269 +1,413 @@ -//! Smart multiplexing: SetConfig locking, StartRx/StopRx reference counting. +//! Mux-level command interception (spec §13.2 / §13.4). //! -//! The mux intercepts certain commands to avoid redundant dongle operations -//! when multiple clients are connected. Intercepted commands get synthetic -//! responses without hitting the physical dongle. +//! When multiple clients share a single dongle, certain commands must be +//! handled by the mux rather than blindly forwarded. The interesting +//! cases: +//! +//! - **`SET_CONFIG`**: the first successful call from any client +//! "locks" the config. Subsequent calls from that same client pass +//! through; calls from *other* clients receive a synthesized +//! `OK(SetConfigResult)` — `ALREADY_MATCHED` if params byte-match the +//! lock, `LOCKED_MISMATCH` otherwise — so other clients never clobber +//! the lock owner's configuration. +//! - **`RX_START` / `RX_STOP`**: reference-counted. The mux keeps RX +//! running until the last interested client stops. use std::collections::HashMap; -use donglora_client::protocol::{ - CMD_TAG_SET_CONFIG, CMD_TAG_START_RX, CMD_TAG_STOP_RX, ERROR_INVALID_CONFIG, RADIO_CONFIG_SIZE, RESP_TAG_ERROR, - RESP_TAG_OK, +use donglora_protocol::{ + Command, Modulation, OkPayload, Owner, SetConfigResult, SetConfigResultCode, + events::{TYPE_ERR, TYPE_OK}, }; -use tracing::{debug, warn}; +use tracing::debug; -use crate::session::ClientSession; +use crate::session::{ClientId, ClientSession}; -/// Mux-level state for interception logic. +/// What to do with a client command. +/// +/// - [`Decision::Forward`] — pass it through to the dongle (after the +/// caller rewrites the tag to a mux-allocated device tag). +/// - [`Decision::Synthesize`] — reply directly to the client with the +/// attached `(type_id, payload_bytes)`; do not touch the dongle. +/// - [`Decision::Drop`] — silently swallow (reserved for corrupt input). +#[derive(Debug, Clone)] +pub enum Decision { + Forward, + Synthesize { + type_id: u8, + payload: Vec, + }, + #[allow(dead_code)] // reserved for future shape-rejection paths + Drop, +} + +/// Mux-level state tracked across all client sessions. pub struct MuxState { - /// Raw 13-byte RadioConfig, locked once set by any client with others connected. - pub locked_config: Option<[u8; RADIO_CONFIG_SIZE]>, + /// Active config lock. When `Some`, identifies the client that most + /// recently applied `SET_CONFIG` and the exact modulation they set. + pub locked: Option<(ClientId, Modulation)>, } impl MuxState { pub fn new() -> Self { - Self { locked_config: None } + Self { locked: None } + } + + /// Number of clients that have called `RX_START`. + pub fn rx_interest_count(sessions: &HashMap) -> usize { + sessions.values().filter(|s| s.rx_interested).count() } } -/// Check if a command should be intercepted (synthetic response) vs forwarded to the dongle. +/// Decide what to do with a decoded command. /// -/// Returns `Some(response_bytes)` if intercepted, `None` to forward normally. -pub fn maybe_intercept( - raw_cmd: &[u8], - client_id: u64, - sessions: &mut HashMap, - locked_config: &Option<[u8; RADIO_CONFIG_SIZE]>, -) -> Option> { - let cmd_tag = *raw_cmd.first()?; - - match cmd_tag { - CMD_TAG_SET_CONFIG => intercept_set_config(raw_cmd, client_id, sessions, locked_config), - CMD_TAG_START_RX => intercept_start_rx(client_id, sessions), - CMD_TAG_STOP_RX => intercept_stop_rx(client_id, sessions), - _ => None, - } -} - -fn intercept_set_config( - raw_cmd: &[u8], - client_id: u64, - sessions: &HashMap, - locked_config: &Option<[u8; RADIO_CONFIG_SIZE]>, -) -> Option> { - let config_bytes = raw_cmd.get(1..1 + RADIO_CONFIG_SIZE)?; - - // Single client — allow free config changes (scanner mode) - if sessions.len() <= 1 { - return None; - } - - // First SetConfig with multiple clients — forward and lock on success - let Some(locked) = locked_config else { - return None; +/// `type_id` + `payload` together come from the FrameDecoder on the +/// client's incoming frame (pre-tag-rewrite). The returned `Decision` +/// says whether the mux should absorb it locally or forward it to the +/// dongle. +pub fn decide( + type_id: u8, + payload: &[u8], + client_id: ClientId, + state: &mut MuxState, + sessions: &mut HashMap, +) -> Decision { + // Parse the command — if it's unparseable, let it through and let + // the firmware return ERR(EUNKNOWN_CMD) with the echoed tag. Keeps + // the mux forward-compatible with minor-version command additions. + let Ok(cmd) = Command::parse(type_id, payload) else { + return Decision::Forward; }; - // Same config — reply Ok without hitting the dongle - if config_bytes == locked.as_slice() { - let label = sessions.get(&client_id).map(ClientSession::label).unwrap_or_else(|| format!("id-{client_id}")); - debug!("{label}: SetConfig matches locked config — Ok"); - return Some(vec![RESP_TAG_OK]); + match cmd { + Command::SetConfig(modulation) => decide_set_config(modulation, client_id, state, sessions), + Command::RxStart => decide_rx_start(client_id, sessions), + Command::RxStop => decide_rx_stop(client_id, sessions), + _ => Decision::Forward, } - - // Different config — reject - let label = sessions.get(&client_id).map(ClientSession::label).unwrap_or_else(|| format!("id-{client_id}")); - warn!("{label}: SetConfig rejected (conflicts with locked config)"); - Some(vec![RESP_TAG_ERROR, ERROR_INVALID_CONFIG]) } -fn intercept_start_rx(client_id: u64, sessions: &mut HashMap) -> Option> { - let client = sessions.get(&client_id)?; - - // Already interested — no-op - if client.rx_interested { - return Some(vec![RESP_TAG_OK]); +fn decide_set_config( + modulation: Modulation, + client_id: ClientId, + state: &mut MuxState, + sessions: &HashMap, +) -> Decision { + // Single client, no lock: unconditionally forward. + if sessions.len() <= 1 { + return Decision::Forward; } - // Others already receiving — just mark this client - if rx_interest_count(sessions) > 0 { + match &state.locked { + None => { + // First SetConfig with multiple clients — forward; the writer + // will install the lock when APPLIED comes back. + Decision::Forward + } + Some((owner, current)) if *owner == client_id => { + // The lock owner is reconfiguring — forward. + let _ = current; + Decision::Forward + } + Some((_, current)) => { + // Another client holds the lock. Synthesize a response. + let matches = modulations_match(&modulation, current); + let code = if matches { SetConfigResultCode::AlreadyMatched } else { SetConfigResultCode::LockedMismatch }; + let result = SetConfigResult { result: code, owner: Owner::Other, current: *current }; + let mut buf = [0u8; donglora_protocol::MAX_SETCONFIG_OK_PAYLOAD]; + let n = match OkPayload::SetConfig(result).encode(&mut buf) { + Ok(n) => n, + Err(_) => return Decision::Forward, // unreachable in practice + }; + let label = sessions.get(&client_id).map_or_else(|| format!("id-{client_id}"), ClientSession::label); + debug!("{label}: SetConfig {:?} (owner=OTHER) — synthesized", code); + Decision::Synthesize { type_id: TYPE_OK, payload: buf[..n].to_vec() } + } + } +} + +fn decide_rx_start(client_id: ClientId, sessions: &mut HashMap) -> Decision { + let Some(client) = sessions.get(&client_id) else { + return Decision::Forward; + }; + + if client.rx_interested { + // Already in RX — synthesize empty OK. + return Decision::Synthesize { type_id: TYPE_OK, payload: vec![] }; + } + + if MuxState::rx_interest_count(sessions) > 0 { + // Others already have RX running on the dongle — just mark this + // client as interested and synthesize OK. if let Some(c) = sessions.get_mut(&client_id) { c.rx_interested = true; } - return Some(vec![RESP_TAG_OK]); + return Decision::Synthesize { type_id: TYPE_OK, payload: vec![] }; } - // First interested client — forward to dongle - None + // First interested client — actually turn RX on. + Decision::Forward } -fn intercept_stop_rx(client_id: u64, sessions: &mut HashMap) -> Option> { - let client = sessions.get(&client_id)?; - - // Wasn't interested — no-op +fn decide_rx_stop(client_id: ClientId, sessions: &mut HashMap) -> Decision { + let Some(client) = sessions.get(&client_id) else { + return Decision::Forward; + }; if !client.rx_interested { - return Some(vec![RESP_TAG_OK]); + // Wasn't listening — nothing to do. + return Decision::Synthesize { type_id: TYPE_OK, payload: vec![] }; } - - // Mark as not interested if let Some(c) = sessions.get_mut(&client_id) { c.rx_interested = false; } - - // Others still interested — don't stop the dongle - if rx_interest_count(sessions) > 0 { - return Some(vec![RESP_TAG_OK]); + if MuxState::rx_interest_count(sessions) > 0 { + // Others still interested — don't shut RX off. + return Decision::Synthesize { type_id: TYPE_OK, payload: vec![] }; } - - // Last interested client — forward StopRx to dongle - None + Decision::Forward } -/// Count how many sessions have rx_interested set. -pub fn rx_interest_count(sessions: &HashMap) -> usize { - sessions.values().filter(|s| s.rx_interested).count() +fn modulations_match(a: &Modulation, b: &Modulation) -> bool { + // Byte-wise comparison via encode (spec definition of "matches" is + // "same params on the wire"). + let mut ba = [0u8; donglora_protocol::MAX_SETCONFIG_OK_PAYLOAD]; + let mut bb = [0u8; donglora_protocol::MAX_SETCONFIG_OK_PAYLOAD]; + let na = match a.encode(&mut ba) { + Ok(n) => n, + Err(_) => return false, + }; + let nb = match b.encode(&mut bb) { + Ok(n) => n, + Err(_) => return false, + }; + ba[..na] == bb[..nb] } -// ── Tests ────────────────────────────────────────────────────────── +/// Synthesize an `ERR(code)` payload. +#[must_use] +pub fn synthesize_err(code: donglora_protocol::ErrorCode) -> (u8, Vec) { + let mut buf = [0u8; 2]; + // This never fails: buf is large enough. + let _ = donglora_protocol::events::encode_err_payload(code, &mut buf); + (TYPE_ERR, buf.to_vec()) +} #[cfg(test)] -#[allow(clippy::unwrap_used)] +#[allow(clippy::unwrap_used, clippy::panic)] mod tests { use super::*; + use donglora_protocol::{LoRaBandwidth, LoRaCodingRate, LoRaConfig, LoRaHeaderMode}; - fn make_sessions(n: usize) -> HashMap { + fn lora(freq: u32) -> Modulation { + Modulation::LoRa(LoRaConfig { + freq_hz: freq, + sf: 7, + bw: LoRaBandwidth::Khz62, + cr: LoRaCodingRate::Cr4_5, + preamble_len: 16, + sync_word: 0x1424, + tx_power_dbm: 20, + header_mode: LoRaHeaderMode::Explicit, + payload_crc: true, + iq_invert: false, + }) + } + + fn make_sessions(n: usize) -> HashMap { let mut map = HashMap::new(); for _ in 0..n { let (session, _rx) = ClientSession::new(); - let id = session.id; - map.insert(id, session); + map.insert(session.id, session); } map } - fn first_id(sessions: &HashMap) -> u64 { + fn first_id(sessions: &HashMap) -> ClientId { *sessions.keys().min().unwrap() } - - fn second_id(sessions: &HashMap) -> u64 { - let mut ids: Vec = sessions.keys().copied().collect(); - ids.sort(); + fn second_id(sessions: &HashMap) -> ClientId { + let mut ids: Vec<_> = sessions.keys().copied().collect(); + ids.sort_unstable(); ids[1] } - // ── SetConfig ──────────────────────────────���─────────────────── + fn encode_set_config(modulation: Modulation) -> Vec { + let mut buf = [0u8; 64]; + let n = Command::SetConfig(modulation).encode_payload(&mut buf).unwrap(); + buf[..n].to_vec() + } #[test] fn set_config_single_client_forwards() { let mut sessions = make_sessions(1); - let locked = None; + let mut state = MuxState::new(); let id = first_id(&sessions); - let cmd = [CMD_TAG_SET_CONFIG, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]; - assert!(maybe_intercept(&cmd, id, &mut sessions, &locked).is_none()); + let payload = encode_set_config(lora(910_525_000)); + assert!(matches!( + decide(donglora_protocol::commands::TYPE_SET_CONFIG, &payload, id, &mut state, &mut sessions), + Decision::Forward + )); } #[test] - fn set_config_first_with_multiple_forwards() { + fn set_config_first_multi_client_forwards() { let mut sessions = make_sessions(2); - let locked = None; + let mut state = MuxState::new(); let id = first_id(&sessions); - let cmd = [CMD_TAG_SET_CONFIG, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]; - assert!(maybe_intercept(&cmd, id, &mut sessions, &locked).is_none()); + let payload = encode_set_config(lora(910_525_000)); + assert!(matches!( + decide(donglora_protocol::commands::TYPE_SET_CONFIG, &payload, id, &mut state, &mut sessions), + Decision::Forward + )); } #[test] - fn set_config_matching_returns_ok() { + fn set_config_matching_synthesizes_already_matched() { let mut sessions = make_sessions(2); - let config = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]; - let locked = Some(config); - let id = first_id(&sessions); - let mut cmd = vec![CMD_TAG_SET_CONFIG]; - cmd.extend_from_slice(&config); - assert_eq!(maybe_intercept(&cmd, id, &mut sessions, &locked), Some(vec![RESP_TAG_OK])); + let id_a = first_id(&sessions); + let id_b = second_id(&sessions); + let mut state = MuxState::new(); + state.locked = Some((id_a, lora(910_525_000))); + let payload = encode_set_config(lora(910_525_000)); + match decide(donglora_protocol::commands::TYPE_SET_CONFIG, &payload, id_b, &mut state, &mut sessions) { + Decision::Synthesize { type_id, payload } => { + assert_eq!(type_id, TYPE_OK); + let parsed = OkPayload::parse_for(donglora_protocol::commands::TYPE_SET_CONFIG, &payload).unwrap(); + let OkPayload::SetConfig(result) = parsed else { panic!("expected SetConfig") }; + assert_eq!(result.result, SetConfigResultCode::AlreadyMatched); + assert_eq!(result.owner, Owner::Other); + } + other => panic!("unexpected decision: {other:?}"), + } } #[test] - fn set_config_conflicting_returns_error() { + fn set_config_conflict_synthesizes_locked_mismatch() { let mut sessions = make_sessions(2); - let locked = Some([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]); - let id = first_id(&sessions); - let mut cmd = vec![CMD_TAG_SET_CONFIG]; - cmd.extend_from_slice(&[99, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]); - assert_eq!(maybe_intercept(&cmd, id, &mut sessions, &locked), Some(vec![RESP_TAG_ERROR, ERROR_INVALID_CONFIG])); + let id_a = first_id(&sessions); + let id_b = second_id(&sessions); + let mut state = MuxState::new(); + state.locked = Some((id_a, lora(910_525_000))); + let payload = encode_set_config(lora(915_000_000)); + match decide(donglora_protocol::commands::TYPE_SET_CONFIG, &payload, id_b, &mut state, &mut sessions) { + Decision::Synthesize { type_id, payload } => { + assert_eq!(type_id, TYPE_OK); + let parsed = OkPayload::parse_for(donglora_protocol::commands::TYPE_SET_CONFIG, &payload).unwrap(); + let OkPayload::SetConfig(result) = parsed else { panic!("expected SetConfig") }; + assert_eq!(result.result, SetConfigResultCode::LockedMismatch); + assert_eq!(result.owner, Owner::Other); + // `current` should reflect the locked modulation (910.525 MHz). + if let Modulation::LoRa(c) = result.current { + assert_eq!(c.freq_hz, 910_525_000); + } else { + panic!("expected LoRa"); + } + } + other => panic!("unexpected decision: {other:?}"), + } } - // ── StartRx ──────────────────────────────────────────────────── + #[test] + fn set_config_owner_reconfig_forwards() { + let mut sessions = make_sessions(2); + let id_a = first_id(&sessions); + let mut state = MuxState::new(); + state.locked = Some((id_a, lora(910_525_000))); + let payload = encode_set_config(lora(915_000_000)); + assert!(matches!( + decide(donglora_protocol::commands::TYPE_SET_CONFIG, &payload, id_a, &mut state, &mut sessions), + Decision::Forward + )); + } #[test] - fn start_rx_first_client_forwards() { + fn rx_start_first_forwards() { let mut sessions = make_sessions(1); - let locked = None; + let mut state = MuxState::new(); let id = first_id(&sessions); - assert!(maybe_intercept(&[CMD_TAG_START_RX], id, &mut sessions, &locked).is_none()); + assert!(matches!( + decide(donglora_protocol::commands::TYPE_RX_START, &[], id, &mut state, &mut sessions), + Decision::Forward + )); } #[test] - fn start_rx_already_interested_returns_ok() { + fn rx_start_when_others_listening_synthesizes_ok() { + let mut sessions = make_sessions(2); + let mut state = MuxState::new(); + let id_a = first_id(&sessions); + let id_b = second_id(&sessions); + sessions.get_mut(&id_a).unwrap().rx_interested = true; + let res = decide(donglora_protocol::commands::TYPE_RX_START, &[], id_b, &mut state, &mut sessions); + assert!(matches!(res, Decision::Synthesize { type_id: TYPE_OK, .. })); + assert!(sessions.get(&id_b).unwrap().rx_interested); + } + + #[test] + fn rx_start_already_interested_synthesizes_ok() { let mut sessions = make_sessions(1); - let locked = None; + let mut state = MuxState::new(); let id = first_id(&sessions); sessions.get_mut(&id).unwrap().rx_interested = true; - assert_eq!(maybe_intercept(&[CMD_TAG_START_RX], id, &mut sessions, &locked), Some(vec![RESP_TAG_OK])); + let res = decide(donglora_protocol::commands::TYPE_RX_START, &[], id, &mut state, &mut sessions); + assert!(matches!(res, Decision::Synthesize { type_id: TYPE_OK, .. })); } #[test] - fn start_rx_others_interested_marks_and_returns_ok() { - let mut sessions = make_sessions(2); - let locked = None; - let id1 = first_id(&sessions); - let id2 = second_id(&sessions); - sessions.get_mut(&id1).unwrap().rx_interested = true; - let result = maybe_intercept(&[CMD_TAG_START_RX], id2, &mut sessions, &locked); - assert_eq!(result, Some(vec![RESP_TAG_OK])); - assert!(sessions.get(&id2).unwrap().rx_interested); - } - - // ── StopRx ───────────────────────────────────────────────────── - - #[test] - fn stop_rx_not_interested_returns_ok() { + fn rx_stop_not_interested_synthesizes_ok() { let mut sessions = make_sessions(1); - let locked = None; + let mut state = MuxState::new(); let id = first_id(&sessions); - assert_eq!(maybe_intercept(&[CMD_TAG_STOP_RX], id, &mut sessions, &locked), Some(vec![RESP_TAG_OK])); + let res = decide(donglora_protocol::commands::TYPE_RX_STOP, &[], id, &mut state, &mut sessions); + assert!(matches!(res, Decision::Synthesize { type_id: TYPE_OK, .. })); } #[test] - fn stop_rx_others_remain_returns_ok() { + fn rx_stop_with_others_listening_synthesizes_ok_and_clears() { let mut sessions = make_sessions(2); - let locked = None; - let id1 = first_id(&sessions); - let id2 = second_id(&sessions); - sessions.get_mut(&id1).unwrap().rx_interested = true; - sessions.get_mut(&id2).unwrap().rx_interested = true; - let result = maybe_intercept(&[CMD_TAG_STOP_RX], id1, &mut sessions, &locked); - assert_eq!(result, Some(vec![RESP_TAG_OK])); - assert!(!sessions.get(&id1).unwrap().rx_interested); + let mut state = MuxState::new(); + let id_a = first_id(&sessions); + let id_b = second_id(&sessions); + sessions.get_mut(&id_a).unwrap().rx_interested = true; + sessions.get_mut(&id_b).unwrap().rx_interested = true; + let res = decide(donglora_protocol::commands::TYPE_RX_STOP, &[], id_a, &mut state, &mut sessions); + assert!(matches!(res, Decision::Synthesize { type_id: TYPE_OK, .. })); + assert!(!sessions.get(&id_a).unwrap().rx_interested); + assert!(sessions.get(&id_b).unwrap().rx_interested); } #[test] - fn stop_rx_last_interested_forwards() { + fn rx_stop_last_forwards_and_clears() { let mut sessions = make_sessions(1); - let locked = None; + let mut state = MuxState::new(); let id = first_id(&sessions); sessions.get_mut(&id).unwrap().rx_interested = true; - assert!(maybe_intercept(&[CMD_TAG_STOP_RX], id, &mut sessions, &locked).is_none()); + let res = decide(donglora_protocol::commands::TYPE_RX_STOP, &[], id, &mut state, &mut sessions); + assert!(matches!(res, Decision::Forward)); assert!(!sessions.get(&id).unwrap().rx_interested); } - // ── Other commands ───────────────────────────────────────────── + #[test] + fn other_commands_forward() { + let mut sessions = make_sessions(1); + let mut state = MuxState::new(); + let id = first_id(&sessions); + // PING, GET_INFO forward unchanged. + assert!(matches!( + decide(donglora_protocol::commands::TYPE_PING, &[], id, &mut state, &mut sessions), + Decision::Forward + )); + assert!(matches!( + decide(donglora_protocol::commands::TYPE_GET_INFO, &[], id, &mut state, &mut sessions), + Decision::Forward + )); + } #[test] - fn other_commands_not_intercepted() { - let mut sessions = make_sessions(1); - let locked = None; - let id = first_id(&sessions); - for tag in [0u8, 1, 5, 6, 7, 8] { - assert!(maybe_intercept(&[tag], id, &mut sessions, &locked).is_none()); - } + fn synthesize_err_encodes_wire_value() { + let (type_id, payload) = synthesize_err(donglora_protocol::ErrorCode::EBusy); + assert_eq!(type_id, TYPE_ERR); + let code = donglora_protocol::events::decode_err_payload(&payload).unwrap(); + assert_eq!(code, donglora_protocol::ErrorCode::EBusy); } } diff --git a/src/main.rs b/src/main.rs index 414445b..8011312 100644 --- a/src/main.rs +++ b/src/main.rs @@ -50,14 +50,13 @@ async fn main() -> anyhow::Result<()> { // Resolve serial port let port = match args.port { Some(p) => p, - None => tokio::task::spawn_blocking(|| { - donglora_client::find_port().unwrap_or_else(|| { + None => match donglora_client::find_port() { + Some(p) => p, + None => { info!("waiting for DongLoRa device..."); - donglora_client::wait_for_device() - }) - }) - .await - .map_err(|e| anyhow::anyhow!("device discovery task failed: {e}"))?, + donglora_client::wait_for_device().await + } + }, }; // Resolve socket path diff --git a/src/session.rs b/src/session.rs index 369277f..126e09e 100644 --- a/src/session.rs +++ b/src/session.rs @@ -1,8 +1,12 @@ //! Per-client session state and bounded send queue. //! //! Each connected client gets a [`ClientSession`] that tracks its ID, -//! RX interest flag, and a bounded channel for outgoing COBS frames. +//! RX-interest flag, an optional TX-data cache (for the TX-loopback +//! fan-out spec §13.4), and a bounded channel for outgoing wire-encoded +//! frames. +use std::collections::HashMap; +use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use tokio::sync::mpsc; @@ -11,17 +15,31 @@ use tracing::warn; /// Bounded send queue capacity per client. const SEND_QUEUE_CAP: usize = 256; +/// Emit a drop-count summary after this many drops for a given client. +const DROP_LOG_EVERY: u64 = 32; + /// Global client ID counter. static NEXT_CLIENT_ID: AtomicU64 = AtomicU64::new(0); +pub type ClientId = u64; + /// Per-client state tracked by the mux daemon. pub struct ClientSession { /// Unique client identifier. - pub id: u64, - /// Whether this client has called StartRx. + pub id: ClientId, + /// Whether this client has called `RX_START`. pub rx_interested: bool, + /// In-flight TX payloads keyed by the client's upstream tag. Needed + /// so the mux can fan out a `TX_DONE(TRANSMITTED)` to other clients + /// as an origin=LocalLoopback RX frame (spec §13.4) without asking + /// the sender to re-transmit the bytes. + pub tx_cache: HashMap>, /// Sender half of the bounded channel to the client's writer task. tx: mpsc::Sender>, + /// Running total of frames dropped for this client because its + /// queue was full. Shared with the writer task so metrics survive + /// the session. + drops: Arc, } impl ClientSession { @@ -30,7 +48,8 @@ impl ClientSession { pub fn new() -> (Self, mpsc::Receiver>) { let id = NEXT_CLIENT_ID.fetch_add(1, Ordering::Relaxed); let (tx, rx) = mpsc::channel(SEND_QUEUE_CAP); - let session = Self { id, rx_interested: false, tx }; + let session = + Self { id, rx_interested: false, tx_cache: HashMap::new(), tx, drops: Arc::new(AtomicU64::new(0)) }; (session, rx) } @@ -39,15 +58,23 @@ impl ClientSession { format!("client-{}", self.id) } - /// Best-effort enqueue of a COBS frame for sending to this client. + /// Running total of frames dropped for this client. + pub fn drop_count(&self) -> u64 { + self.drops.load(Ordering::Relaxed) + } + + /// Best-effort enqueue of a wire frame for sending to this client. /// - /// If the channel is full (slow client), the frame is dropped with a warning. - /// This prevents backpressure from one slow client blocking the entire mux. + /// If the channel is full (slow client), the frame is dropped and + /// counted. Drop counts are summarized every `DROP_LOG_EVERY` drops. pub fn enqueue(&self, frame: Vec) { if let Err(mpsc::error::TrySendError::Full(_)) = self.tx.try_send(frame) { - warn!("{}: send queue full, dropping frame", self.label()); + let prev = self.drops.fetch_add(1, Ordering::Relaxed); + let total = prev + 1; + if total == 1 || total.is_multiple_of(DROP_LOG_EVERY) { + warn!("{}: send queue full, dropped {total} frames so far", self.label()); + } } - // Closed channel (client disconnected) is silently ignored — the session - // will be cleaned up when the client handler task exits. + // Closed channel (client disconnected) is silently ignored. } }