8 Commits
Author SHA1 Message Date
swaits 5c3363b307 chore: repoint URLs at git.swaits.com and bump for release
Rewrites remaining github.com / git.sr.ht references to git.swaits.com,
translating GitHub URL shapes to Gitea's (/blob/<ref>/ -> /src/branch/<ref>/,
/tree/<tag> -> /src/tag/<tag>, /actions/workflows/X -> /actions?workflow=X,
/edit/ -> /_edit/).

Left pointing at GitHub on purpose: pull-request links (PRs were not migrated)
and GitHub-only surfaces Gitea lacks (security advisories, discussions).
2026-08-31 17:58:37 -06:00
swaits 598a86f713 chore: point repository at git.swaits.com 2026-08-31 17:19:34 -06:00
swaits 0fccd5c968 chore: release 1.0.2 2026-05-11 14:58:29 -06:00
swaits d45dc945ef chore: release 1.0.1 2026-04-22 22:37:35 -06:00
swaits db0260b08d chore: release 1.0.0 2026-04-22 09:51:27 -06:00
swaits 7274293a04 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).
2026-04-22 09:50:52 -06:00
swaits 90a306853d feat: add keepalive ping and state restoration after dongle reconnect
UART boards (CP2102 bridge) keep the serial port alive across ESP32 resets,
so the mux never detected the board rebooting. Now the writer sends a
keepalive Ping every 5s of idle and declares the dongle lost after a
2s timeout.

On reconnect, saved radio config and RX state are re-sent to the dongle
so connected clients continue working without re-establishing state.
2026-04-11 02:34:05 -06:00
swaits f061a9e8a3 bump MSRV to 1.93, release v0.2.2 2026-04-08 21:47:16 -06:00
8 changed files with 1155 additions and 481 deletions

No files matched your search

+73
View File
@@ -1,5 +1,78 @@
# Changelog # Changelog
## 1.0.2 — 2026-05-11
### Changed
- Refreshed `Cargo.lock` to pull in the latest compatible patch/minor
releases of dependencies:
- `donglora-client` 1.0.2 → 1.1.1
- `donglora-protocol` 1.0.0 → 1.2.0
- `tokio` 1.51.0 → 1.52.3
- `clap` 4.6.0 → 4.6.1
- `heapless` 0.9.2 → 0.9.3
- `libc` 0.2.184 → 0.2.186
- `bitflags` 2.11.0 → 2.11.1
No source changes; CLI and behavior are unchanged.
## 1.0.1 — 2026-04-22
### Dependencies
- Bumped `donglora-client` to `1.0.1` — adds the WCH CH340K USB-UART
bridge (`1a86:7522`) to the port-discovery list. The Elecrow
ThinkNode-M2 board is now detected automatically; previously mux
refused to attach to it because the bridge VID:PID wasn't
recognized.
## 1.0.0 — 2026-04-22
### 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
- **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
### Fixed
- Bumped MSRV from 1.85 to 1.93 to satisfy `ucobs` dependency requirement.
`cargo install donglora-mux` now works on toolchains >= 1.93.
## 0.2.1 — 2026-04-07 ## 0.2.1 — 2026-04-07
### Fixed ### Fixed
Generated
+164 -45
View File
@@ -75,9 +75,15 @@ checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
[[package]] [[package]]
name = "bitflags" name = "bitflags"
version = "2.11.0" version = "2.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af" checksum = "c4512299f36f043ab09a583e57bceb5a5aab7a73db1805848e8fef3c9e8c78b3"
[[package]]
name = "byteorder"
version = "1.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
[[package]] [[package]]
name = "bytes" name = "bytes"
@@ -99,9 +105,9 @@ checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"
[[package]] [[package]]
name = "clap" name = "clap"
version = "4.6.0" version = "4.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b193af5b67834b676abd72466a96c1024e6a6ad978a1f484bd90b85c94041351" checksum = "1ddb117e43bbf7dacf0a4190fef4d345b9bad68dfc649cb349e7d17d28428e51"
dependencies = [ dependencies = [
"clap_builder", "clap_builder",
"clap_derive", "clap_derive",
@@ -121,9 +127,9 @@ dependencies = [
[[package]] [[package]]
name = "clap_derive" name = "clap_derive"
version = "4.6.0" version = "4.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1110bd8a634a1ab8cb04345d8d878267d57c3cf1b38d91b71af6686408bbca6a" checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9"
dependencies = [ dependencies = [
"heck", "heck",
"proc-macro2", "proc-macro2",
@@ -161,24 +167,27 @@ checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
[[package]] [[package]]
name = "donglora-client" name = "donglora-client"
version = "0.2.1" version = "1.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "343d750fe1e5dad94d4b31e6dbbb05214b39e66f3c94a2ff2a29b0031ffd2826"
dependencies = [ dependencies = [
"anyhow", "donglora-protocol",
"serialport", "rand",
"thiserror",
"tokio",
"tokio-serial",
"tokio-stream",
"tracing", "tracing",
"ucobs",
] ]
[[package]] [[package]]
name = "donglora-mux" name = "donglora-mux"
version = "0.2.1" version = "1.0.2"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"clap", "clap",
"donglora-client", "donglora-client",
"donglora-protocol",
"fd-lock", "fd-lock",
"heapless",
"tokio", "tokio",
"tokio-serial", "tokio-serial",
"tokio-util", "tokio-util",
@@ -186,6 +195,16 @@ dependencies = [
"tracing-subscriber", "tracing-subscriber",
] ]
[[package]]
name = "donglora-protocol"
version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0eea2a87a5ad31629140accc2bae6c931434bf1946706f2184dec9b3d356505d"
dependencies = [
"heapless",
"ucobs",
]
[[package]] [[package]]
name = "errno" name = "errno"
version = "0.3.14" version = "0.3.14"
@@ -295,6 +314,37 @@ dependencies = [
"slab", "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.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "25ba4bd83f9415b58b4ed8dc5714c76e626a105be4646c02630ad730ad3b5aa4"
dependencies = [
"hash32",
"stable_deref_trait",
]
[[package]] [[package]]
name = "heck" name = "heck"
version = "0.5.0" version = "0.5.0"
@@ -325,29 +375,9 @@ checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
[[package]] [[package]]
name = "libc" name = "libc"
version = "0.2.184" version = "0.2.186"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "48f5d2a454e16a5ea0f4ced81bd44e4cfc7bd3a507b61887c99fd3538b28e4af" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66"
[[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]] [[package]]
name = "linux-raw-sys" name = "linux-raw-sys"
@@ -436,7 +466,7 @@ version = "0.29.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "71e2746dc3a24dd78b3cfcb7be93368c6de9963d30f43a6a73998a9cf4b17b46" checksum = "71e2746dc3a24dd78b3cfcb7be93368c6de9963d30f43a6a73998a9cf4b17b46"
dependencies = [ dependencies = [
"bitflags 2.11.0", "bitflags 2.11.1",
"cfg-if", "cfg-if",
"cfg_aliases", "cfg_aliases",
"libc", "libc",
@@ -493,10 +523,13 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
[[package]] [[package]]
name = "pkg-config" name = "ppv-lite86"
version = "0.3.32" version = "0.2.21"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9"
dependencies = [
"zerocopy",
]
[[package]] [[package]]
name = "proc-macro2" name = "proc-macro2"
@@ -516,13 +549,48 @@ dependencies = [
"proc-macro2", "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]] [[package]]
name = "redox_syscall" name = "redox_syscall"
version = "0.5.18" version = "0.5.18"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d"
dependencies = [ dependencies = [
"bitflags 2.11.0", "bitflags 2.11.1",
] ]
[[package]] [[package]]
@@ -548,7 +616,7 @@ version = "1.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190" checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190"
dependencies = [ dependencies = [
"bitflags 2.11.0", "bitflags 2.11.1",
"errno", "errno",
"libc", "libc",
"linux-raw-sys", "linux-raw-sys",
@@ -567,12 +635,11 @@ version = "4.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a4d91116f97173694f1642263b2ff837f80d933aa837e2314969f6728f661df3" checksum = "a4d91116f97173694f1642263b2ff837f80d933aa837e2314969f6728f661df3"
dependencies = [ dependencies = [
"bitflags 2.11.0", "bitflags 2.11.1",
"cfg-if", "cfg-if",
"core-foundation", "core-foundation",
"core-foundation-sys", "core-foundation-sys",
"io-kit-sys", "io-kit-sys",
"libudev",
"mach2", "mach2",
"nix 0.26.4", "nix 0.26.4",
"scopeguard", "scopeguard",
@@ -621,6 +688,12 @@ dependencies = [
"windows-sys 0.61.2", "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]] [[package]]
name = "strsim" name = "strsim"
version = "0.11.1" version = "0.11.1"
@@ -669,9 +742,9 @@ dependencies = [
[[package]] [[package]]
name = "tokio" name = "tokio"
version = "1.51.0" version = "1.52.3"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2bd1c4c0fc4a7ab90fc15ef6daaa3ec3b893f004f915f2392557ed23237820cd" checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe"
dependencies = [ dependencies = [
"bytes", "bytes",
"libc", "libc",
@@ -709,6 +782,17 @@ dependencies = [
"tokio", "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]] [[package]]
name = "tokio-util" name = "tokio-util"
version = "0.7.18" version = "0.7.18"
@@ -822,6 +906,15 @@ version = "0.11.1+wasi-snapshot-preview1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" 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]] [[package]]
name = "winapi" name = "winapi"
version = "0.3.9" version = "0.3.9"
@@ -940,3 +1033,29 @@ name = "windows_x86_64_msvc"
version = "0.52.6" version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" 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",
]
+14 -5
View File
@@ -1,19 +1,27 @@
[package] [package]
name = "donglora-mux" name = "donglora-mux"
version = "0.2.1" version = "1.0.3"
edition = "2024" edition = "2024"
rust-version = "1.85" rust-version = "1.93"
authors = ["Stephen Waits <steve@waits.net>"] authors = ["Stephen Waits <steve@waits.net>"]
license = "MIT" license = "MIT"
description = "DongLoRa USB Multiplexer — share one dongle with multiple applications" description = "DongLoRa USB Multiplexer — share one dongle with multiple applications"
repository = "https://github.com/donglora/mux-rs" repository = "https://git.swaits.com/donglora/mux-rs"
homepage = "https://github.com/donglora/mux-rs" homepage = "https://git.swaits.com/donglora/mux-rs"
keywords = ["lora", "radio", "usb", "multiplexer"] keywords = ["lora", "radio", "usb", "multiplexer"]
categories = ["hardware-support"] categories = ["hardware-support"]
exclude = [".jj/", "deny.toml", "justfile", "mise.toml", "rustfmt.toml"] exclude = [".jj/", "deny.toml", "justfile", "mise.toml", "rustfmt.toml"]
[dependencies] [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 = { version = "1", default-features = false, features = ["std"] }
# Still used for USB discovery helpers (`find_port`, `wait_for_device`).
# The `path` makes local builds use the sibling checkout; `cargo publish`
# strips it and publishes against `version` from crates.io. Spec is
# major-only so patch/minor releases of the client propagate without
# a mux-rs re-publish.
donglora-client = { version = "1", path = "../client-rs" }
anyhow = "1" anyhow = "1"
clap = { version = "4", features = ["derive"] } clap = { version = "4", features = ["derive"] }
tokio = { version = "1", features = ["full"] } tokio = { version = "1", features = ["full"] }
@@ -22,6 +30,7 @@ tokio-util = "0.7"
tracing = "0.1" tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] } tracing-subscriber = { version = "0.3", features = ["env-filter"] }
fd-lock = "4" fd-lock = "4"
heapless = "0.9"
[lints.clippy] [lints.clippy]
unwrap_used = "deny" unwrap_used = "deny"
+1 -1
View File
@@ -1,3 +1,3 @@
[tools] [tools]
rust = "1.94" rust = "1.93"
just = "latest" just = "latest"
+556 -253
View File
File diff suppressed because it is too large. Load diff
+304 -160
View File
@@ -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 share a single dongle, certain commands must be
//! when multiple clients are connected. Intercepted commands get synthetic //! handled by the mux rather than blindly forwarded. The interesting
//! responses without hitting the physical dongle. //! 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 std::collections::HashMap;
use donglora_client::protocol::{ use donglora_protocol::{
CMD_TAG_SET_CONFIG, CMD_TAG_START_RX, CMD_TAG_STOP_RX, ERROR_INVALID_CONFIG, RADIO_CONFIG_SIZE, RESP_TAG_ERROR, Command, Modulation, OkPayload, Owner, SetConfigResult, SetConfigResultCode,
RESP_TAG_OK, 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 { pub struct MuxState {
/// Raw 13-byte RadioConfig, locked once set by any client with others connected. /// Active config lock. When `Some`, identifies the client that most
pub locked_config: Option<[u8; RADIO_CONFIG_SIZE]>, /// recently applied `SET_CONFIG` and the exact modulation they set.
pub locked: Option<(ClientId, Modulation)>,
} }
impl MuxState { impl MuxState {
pub fn new() -> Self { 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. /// `type_id` + `payload` together come from the FrameDecoder on the
pub fn maybe_intercept( /// client's incoming frame (pre-tag-rewrite). The returned `Decision`
raw_cmd: &[u8], /// says whether the mux should absorb it locally or forward it to the
client_id: u64, /// dongle.
sessions: &mut HashMap<u64, ClientSession>, pub fn decide(
locked_config: &Option<[u8; RADIO_CONFIG_SIZE]>, type_id: u8,
) -> Option<Vec<u8>> { payload: &[u8],
let cmd_tag = *raw_cmd.first()?; client_id: ClientId,
state: &mut MuxState,
match cmd_tag { sessions: &mut HashMap<ClientId, ClientSession>,
CMD_TAG_SET_CONFIG => intercept_set_config(raw_cmd, client_id, sessions, locked_config), ) -> Decision {
CMD_TAG_START_RX => intercept_start_rx(client_id, sessions), // Parse the command — if it's unparseable, let it through and let
CMD_TAG_STOP_RX => intercept_stop_rx(client_id, sessions), // the firmware return ERR(EUNKNOWN_CMD) with the echoed tag. Keeps
_ => None, // the mux forward-compatible with minor-version command additions.
} let Ok(cmd) = Command::parse(type_id, payload) else {
} return Decision::Forward;
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;
}; };
// Same config — reply Ok without hitting the dongle match cmd {
if config_bytes == locked.as_slice() { Command::SetConfig(modulation) => decide_set_config(modulation, client_id, state, sessions),
let label = sessions.get(&client_id).map(ClientSession::label).unwrap_or_else(|| format!("id-{client_id}")); Command::RxStart => decide_rx_start(client_id, sessions),
debug!("{label}: SetConfig matches locked config — Ok"); Command::RxStop => decide_rx_stop(client_id, sessions),
return Some(vec![RESP_TAG_OK]); _ => 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<u64, ClientSession>) -> Option<Vec<u8>> { fn decide_set_config(
let client = sessions.get(&client_id)?; modulation: Modulation,
client_id: ClientId,
// Already interested — no-op state: &mut MuxState,
if client.rx_interested { sessions: &HashMap<ClientId, ClientSession>,
return Some(vec![RESP_TAG_OK]); ) -> Decision {
// Single client, no lock: unconditionally forward.
if sessions.len() <= 1 {
return Decision::Forward;
} }
// Others already receiving — just mark this client match &state.locked {
if rx_interest_count(sessions) > 0 { 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;
};
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) { if let Some(c) = sessions.get_mut(&client_id) {
c.rx_interested = true; 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 // First interested client — actually turn RX on.
None Decision::Forward
} }
fn intercept_stop_rx(client_id: u64, sessions: &mut HashMap<u64, ClientSession>) -> Option<Vec<u8>> { fn decide_rx_stop(client_id: ClientId, sessions: &mut HashMap<ClientId, ClientSession>) -> Decision {
let client = sessions.get(&client_id)?; let Some(client) = sessions.get(&client_id) else {
return Decision::Forward;
// Wasn't interested — no-op };
if !client.rx_interested { 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) { if let Some(c) = sessions.get_mut(&client_id) {
c.rx_interested = false; c.rx_interested = false;
} }
if MuxState::rx_interest_count(sessions) > 0 {
// Others still interested — don't stop the dongle // Others still interested — don't shut RX off.
if rx_interest_count(sessions) > 0 { return Decision::Synthesize { type_id: TYPE_OK, payload: vec![] };
return Some(vec![RESP_TAG_OK]);
} }
Decision::Forward
// Last interested client — forward StopRx to dongle
None
} }
/// Count how many sessions have rx_interested set. fn modulations_match(a: &Modulation, b: &Modulation) -> bool {
pub fn rx_interest_count(sessions: &HashMap<u64, ClientSession>) -> usize { // Byte-wise comparison via encode (spec definition of "matches" is
sessions.values().filter(|s| s.rx_interested).count() // "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<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())
}
#[cfg(test)] #[cfg(test)]
#[allow(clippy::unwrap_used)] #[allow(clippy::unwrap_used, clippy::panic)]
mod tests { mod tests {
use super::*; 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(); let mut map = HashMap::new();
for _ in 0..n { for _ in 0..n {
let (session, _rx) = ClientSession::new(); let (session, _rx) = ClientSession::new();
let id = session.id; map.insert(session.id, session);
map.insert(id, session);
} }
map map
} }
fn first_id(sessions: &HashMap<u64, ClientSession>) -> u64 { fn first_id(sessions: &HashMap<ClientId, ClientSession>) -> ClientId {
*sessions.keys().min().unwrap() *sessions.keys().min().unwrap()
} }
fn second_id(sessions: &HashMap<ClientId, ClientSession>) -> ClientId {
fn second_id(sessions: &HashMap<u64, ClientSession>) -> u64 { let mut ids: Vec<_> = sessions.keys().copied().collect();
let mut ids: Vec<u64> = sessions.keys().copied().collect(); ids.sort_unstable();
ids.sort();
ids[1] 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] #[test]
fn set_config_single_client_forwards() { fn set_config_single_client_forwards() {
let mut sessions = make_sessions(1); let mut sessions = make_sessions(1);
let locked = None; let mut state = MuxState::new();
let id = first_id(&sessions); let id = first_id(&sessions);
let cmd = [CMD_TAG_SET_CONFIG, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]; let payload = encode_set_config(lora(910_525_000));
assert!(maybe_intercept(&cmd, id, &mut sessions, &locked).is_none()); assert!(matches!(
decide(donglora_protocol::commands::TYPE_SET_CONFIG, &payload, id, &mut state, &mut sessions),
Decision::Forward
));
} }
#[test] #[test]
fn set_config_first_with_multiple_forwards() { fn set_config_first_multi_client_forwards() {
let mut sessions = make_sessions(2); let mut sessions = make_sessions(2);
let locked = None; let mut state = MuxState::new();
let id = first_id(&sessions); let id = first_id(&sessions);
let cmd = [CMD_TAG_SET_CONFIG, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]; let payload = encode_set_config(lora(910_525_000));
assert!(maybe_intercept(&cmd, id, &mut sessions, &locked).is_none()); assert!(matches!(
decide(donglora_protocol::commands::TYPE_SET_CONFIG, &payload, id, &mut state, &mut sessions),
Decision::Forward
));
} }
#[test] #[test]
fn set_config_matching_returns_ok() { fn set_config_matching_synthesizes_already_matched() {
let mut sessions = make_sessions(2); let mut sessions = make_sessions(2);
let config = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]; let id_a = first_id(&sessions);
let locked = Some(config); let id_b = second_id(&sessions);
let id = first_id(&sessions); let mut state = MuxState::new();
let mut cmd = vec![CMD_TAG_SET_CONFIG]; state.locked = Some((id_a, lora(910_525_000)));
cmd.extend_from_slice(&config); let payload = encode_set_config(lora(910_525_000));
assert_eq!(maybe_intercept(&cmd, id, &mut sessions, &locked), Some(vec![RESP_TAG_OK])); 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] #[test]
fn set_config_conflicting_returns_error() { fn set_config_conflict_synthesizes_locked_mismatch() {
let mut sessions = make_sessions(2); let mut sessions = make_sessions(2);
let locked = Some([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]); let id_a = first_id(&sessions);
let id = first_id(&sessions); let id_b = second_id(&sessions);
let mut cmd = vec![CMD_TAG_SET_CONFIG]; let mut state = MuxState::new();
cmd.extend_from_slice(&[99, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]); state.locked = Some((id_a, lora(910_525_000)));
assert_eq!(maybe_intercept(&cmd, id, &mut sessions, &locked), Some(vec![RESP_TAG_ERROR, ERROR_INVALID_CONFIG])); 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] #[test]
fn start_rx_first_client_forwards() { fn rx_start_first_forwards() {
let mut sessions = make_sessions(1); let mut sessions = make_sessions(1);
let locked = None; let mut state = MuxState::new();
let id = first_id(&sessions); 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] #[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 mut sessions = make_sessions(1);
let locked = None; let mut state = MuxState::new();
let id = first_id(&sessions); let id = first_id(&sessions);
sessions.get_mut(&id).unwrap().rx_interested = true; 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] #[test]
fn start_rx_others_interested_marks_and_returns_ok() { fn rx_stop_not_interested_synthesizes_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() {
let mut sessions = make_sessions(1); let mut sessions = make_sessions(1);
let locked = None; let mut state = MuxState::new();
let id = first_id(&sessions); 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] #[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 mut sessions = make_sessions(2);
let locked = None; let mut state = MuxState::new();
let id1 = first_id(&sessions); let id_a = first_id(&sessions);
let id2 = second_id(&sessions); let id_b = second_id(&sessions);
sessions.get_mut(&id1).unwrap().rx_interested = true; sessions.get_mut(&id_a).unwrap().rx_interested = true;
sessions.get_mut(&id2).unwrap().rx_interested = true; sessions.get_mut(&id_b).unwrap().rx_interested = true;
let result = maybe_intercept(&[CMD_TAG_STOP_RX], id1, &mut sessions, &locked); let res = decide(donglora_protocol::commands::TYPE_RX_STOP, &[], id_a, &mut state, &mut sessions);
assert_eq!(result, Some(vec![RESP_TAG_OK])); assert!(matches!(res, Decision::Synthesize { type_id: TYPE_OK, .. }));
assert!(!sessions.get(&id1).unwrap().rx_interested); assert!(!sessions.get(&id_a).unwrap().rx_interested);
assert!(sessions.get(&id_b).unwrap().rx_interested);
} }
#[test] #[test]
fn stop_rx_last_interested_forwards() { fn rx_stop_last_forwards_and_clears() {
let mut sessions = make_sessions(1); let mut sessions = make_sessions(1);
let locked = None; let mut state = MuxState::new();
let id = first_id(&sessions); let id = first_id(&sessions);
sessions.get_mut(&id).unwrap().rx_interested = true; 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); 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] #[test]
fn other_commands_not_intercepted() { fn synthesize_err_encodes_wire_value() {
let mut sessions = make_sessions(1); let (type_id, payload) = synthesize_err(donglora_protocol::ErrorCode::EBusy);
let locked = None; assert_eq!(type_id, TYPE_ERR);
let id = first_id(&sessions); let code = donglora_protocol::events::decode_err_payload(&payload).unwrap();
for tag in [0u8, 1, 5, 6, 7, 8] { assert_eq!(code, donglora_protocol::ErrorCode::EBusy);
assert!(maybe_intercept(&[tag], id, &mut sessions, &locked).is_none());
}
} }
} }
+6 -7
View File
@@ -50,14 +50,13 @@ async fn main() -> anyhow::Result<()> {
// Resolve serial port // Resolve serial port
let port = match args.port { let port = match args.port {
Some(p) => p, Some(p) => p,
None => tokio::task::spawn_blocking(|| { None => match donglora_client::find_port() {
donglora_client::find_port().unwrap_or_else(|| { Some(p) => p,
None => {
info!("waiting for DongLoRa device..."); info!("waiting for DongLoRa device...");
donglora_client::wait_for_device() donglora_client::wait_for_device().await
}) }
}) },
.await
.map_err(|e| anyhow::anyhow!("device discovery task failed: {e}"))?,
}; };
// Resolve socket path // Resolve socket path
+37 -10
View File
@@ -1,8 +1,12 @@
//! Per-client session state and bounded send queue. //! Per-client session state and bounded send queue.
//! //!
//! Each connected client gets a [`ClientSession`] that tracks its ID, //! 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 std::sync::atomic::{AtomicU64, Ordering};
use tokio::sync::mpsc; use tokio::sync::mpsc;
@@ -11,17 +15,31 @@ use tracing::warn;
/// Bounded send queue capacity per client. /// Bounded send queue capacity per client.
const SEND_QUEUE_CAP: usize = 256; 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. /// Global client ID counter.
static NEXT_CLIENT_ID: AtomicU64 = AtomicU64::new(0); static NEXT_CLIENT_ID: AtomicU64 = AtomicU64::new(0);
pub type ClientId = u64;
/// Per-client state tracked by the mux daemon. /// Per-client state tracked by the mux daemon.
pub struct ClientSession { pub struct ClientSession {
/// Unique client identifier. /// Unique client identifier.
pub id: u64, pub id: ClientId,
/// Whether this client has called StartRx. /// Whether this client has called `RX_START`.
pub rx_interested: bool, 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. /// Sender half of the bounded channel to the client's writer task.
tx: mpsc::Sender<Vec<u8>>, 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 { impl ClientSession {
@@ -30,7 +48,8 @@ impl ClientSession {
pub fn new() -> (Self, mpsc::Receiver<Vec<u8>>) { pub fn new() -> (Self, mpsc::Receiver<Vec<u8>>) {
let id = NEXT_CLIENT_ID.fetch_add(1, Ordering::Relaxed); let id = NEXT_CLIENT_ID.fetch_add(1, Ordering::Relaxed);
let (tx, rx) = mpsc::channel(SEND_QUEUE_CAP); 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) (session, rx)
} }
@@ -39,15 +58,23 @@ impl ClientSession {
format!("client-{}", self.id) 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. /// If the channel is full (slow client), the frame is dropped and
/// This prevents backpressure from one slow client blocking the entire mux. /// counted. Drop counts are summarized every `DROP_LOG_EVERY` drops.
pub fn enqueue(&self, frame: Vec<u8>) { pub fn enqueue(&self, frame: Vec<u8>) {
if let Err(mpsc::error::TrySendError::Full(_)) = self.tx.try_send(frame) { 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 // Closed channel (client disconnected) is silently ignored.
// will be cleaned up when the client handler task exits.
} }
} }