feat!: mux on DongLoRa Protocol v2 with tag-aware dispatch
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).
This commit is contained in:
7 files changed
+1067
-508
No files matched your search
+37
-6
@@ -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
|
||||
|
||||
|
||||
Generated
+148
-31
@@ -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",
|
||||
]
|
||||
+7
-2
@@ -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 <steve@waits.net>"]
|
||||
@@ -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"
|
||||
|
||||
+531
-295
File diff suppressed because it is too large.
Load diff
+301
-157
@@ -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<u8>,
|
||||
},
|
||||
#[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<ClientId, ClientSession>) -> 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<u64, ClientSession>,
|
||||
locked_config: &Option<[u8; RADIO_CONFIG_SIZE]>,
|
||||
) -> Option<Vec<u8>> {
|
||||
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<u64, ClientSession>,
|
||||
locked_config: &Option<[u8; RADIO_CONFIG_SIZE]>,
|
||||
) -> Option<Vec<u8>> {
|
||||
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<ClientId, ClientSession>,
|
||||
) -> 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 decide_set_config(
|
||||
modulation: Modulation,
|
||||
client_id: ClientId,
|
||||
state: &mut MuxState,
|
||||
sessions: &HashMap<ClientId, ClientSession>,
|
||||
) -> Decision {
|
||||
// Single client, no lock: unconditionally forward.
|
||||
if sessions.len() <= 1 {
|
||||
return Decision::Forward;
|
||||
}
|
||||
|
||||
fn intercept_start_rx(client_id: u64, sessions: &mut HashMap<u64, ClientSession>) -> Option<Vec<u8>> {
|
||||
let client = sessions.get(&client_id)?;
|
||||
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<ClientId, ClientSession>) -> Decision {
|
||||
let Some(client) = sessions.get(&client_id) else {
|
||||
return Decision::Forward;
|
||||
};
|
||||
|
||||
// Already interested — no-op
|
||||
if client.rx_interested {
|
||||
return Some(vec![RESP_TAG_OK]);
|
||||
// Already in RX — synthesize empty OK.
|
||||
return Decision::Synthesize { type_id: TYPE_OK, payload: vec![] };
|
||||
}
|
||||
|
||||
// Others already receiving — just mark this client
|
||||
if rx_interest_count(sessions) > 0 {
|
||||
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<u64, ClientSession>) -> Option<Vec<u8>> {
|
||||
let client = sessions.get(&client_id)?;
|
||||
|
||||
// Wasn't interested — no-op
|
||||
fn decide_rx_stop(client_id: ClientId, sessions: &mut HashMap<ClientId, ClientSession>) -> 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![] };
|
||||
}
|
||||
Decision::Forward
|
||||
}
|
||||
|
||||
// Last interested client — forward StopRx to dongle
|
||||
None
|
||||
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]
|
||||
}
|
||||
|
||||
/// Count how many sessions have rx_interested set.
|
||||
pub fn rx_interest_count(sessions: &HashMap<u64, ClientSession>) -> usize {
|
||||
sessions.values().filter(|s| s.rx_interested).count()
|
||||
/// Synthesize an `ERR(code)` payload.
|
||||
#[must_use]
|
||||
pub fn synthesize_err(code: donglora_protocol::ErrorCode) -> (u8, Vec<u8>) {
|
||||
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())
|
||||
}
|
||||
|
||||
// ── Tests ──────────────────────────────────────────────────────────
|
||||
|
||||
#[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<u64, ClientSession> {
|
||||
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<ClientId, ClientSession> {
|
||||
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, ClientSession>) -> u64 {
|
||||
fn first_id(sessions: &HashMap<ClientId, ClientSession>) -> ClientId {
|
||||
*sessions.keys().min().unwrap()
|
||||
}
|
||||
|
||||
fn second_id(sessions: &HashMap<u64, ClientSession>) -> u64 {
|
||||
let mut ids: Vec<u64> = sessions.keys().copied().collect();
|
||||
ids.sort();
|
||||
fn second_id(sessions: &HashMap<ClientId, ClientSession>) -> ClientId {
|
||||
let mut ids: Vec<_> = sessions.keys().copied().collect();
|
||||
ids.sort_unstable();
|
||||
ids[1]
|
||||
}
|
||||
|
||||
// ── SetConfig ──────────────────────────────���───────────────────
|
||||
fn encode_set_config(modulation: Modulation) -> Vec<u8> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
+6
-7
@@ -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
|
||||
|
||||
+37
-10
@@ -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<u16, Vec<u8>>,
|
||||
/// Sender half of the bounded channel to the client's writer task.
|
||||
tx: mpsc::Sender<Vec<u8>>,
|
||||
/// 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<AtomicU64>,
|
||||
}
|
||||
|
||||
impl ClientSession {
|
||||
@@ -30,7 +48,8 @@ impl ClientSession {
|
||||
pub fn new() -> (Self, mpsc::Receiver<Vec<u8>>) {
|
||||
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<u8>) {
|
||||
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.
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user