7 Commits
Author SHA1 Message Date
swaits 43b30952f8 chore: release 1.0.1 2026-04-22 10:47:49 -06:00
swaits 599cfd14dc chore: bump rand past RUSTSEC-2026-0097
cargo update -p rand@0.9.2 and cargo update -p rand@0.8.5 take the two pinned rand versions past the RUSTSEC-2026-0097 fix windows:


* rand 0.9.2 -> 0.9.4 (fix >= 0.9.3).

* rand 0.8.5 -> 0.8.6 (fix >= 0.8.6).


The advisory is reachable only with a custom logger that calls rand::rng() at trace/warn level and hits the 64 KB reseed path. We are not known to do that, but unblocking CI is reason enough.
2026-04-22 10:34:26 -06:00
swaits 436ab155e7 chore: release 1.0.0 2026-04-22 09:48:32 -06:00
swaits 0b9775c310 feat!: bridge on DongLoRa Protocol v2 and async client
The 1.0 bridge running on the async Dongle API from donglora-client 1.0.0 and the Protocol v2 wire format.

* Radio loop is now a tokio task, not a dedicated std::thread. Everything is async end to end.

* [radio] config schema: tx_power renamed to tx_power_dbm (plain i8, no "max" keyword), cad field removed (CAD is a per-TX flag in v1.0), default sync_word changed 0x3444 to 0x1424 to match the SX126x register encoding of MeshCore private sync byte.

* TX retry with backoff (new [tx] config section). Every TX runs through Dongle::tx_with_retry, retrying TX_DONE(CHANNEL_BUSY) and ERR(EBUSY) per PROTOCOL.md §6.10 / §C.5.5. Defaults: 3 attempts, 20-100 ms randomized backoff, doubling to 500 ms cap, 5 s per-attempt timeout. Configurable via max_attempts, backoff_min_ms, backoff_max_ms, backoff_cap_ms, backoff_multiplier, per_attempt_timeout_secs.

* TUI packet log shows retry state: in-flight retries render with a cycle marker in yellow and an N/M attempt counter, terminal TX failures render in red, post-retry success shows final attempt count.

* Stats gain tx_retries and tx_failures counters surfaced in StatsSnapshot.

* Measured airtime from TX_DONE replaces the theoretical estimate, feeding the rate limiter with the real on-air duration.

* The explicit 2 s liveness ping loop is gone. The Dongle runs a 500 ms internal keepalive, matching spec §3.4 inactivity window with 2x margin.
2026-04-22 09:48:01 -06:00
swaits 030280791e fix: sticky mux reconnect prevents USB port theft
Once the bridge connects via mux, all subsequent reconnects use mux-only
mode (connect_mux_auto). Previously, each reconnect re-ran the full
mux→USB fallback chain, which could steal the serial port from the mux
during a brief disconnect.
2026-04-07 18:05:56 -06:00
swaits 222802cf13 feat: upgrade to donglora-client 0.2, simplify connection, fix org URLs
- Upgrade donglora-client from 0.1 to 0.2 (ping-on-connect validation)
- Remove manual mux/serial fallback and manual ping — handled by client now
- Fix repository/homepage URLs to github.com/donglora/bridge
- Fix README links to donglora GitHub org
2026-04-07 16:11:08 -06:00
swaits 394acaee6e chore: add consistent mise.toml, justfile, and tooling config 2026-04-07 14:43:26 -06:00
14 changed files with 1139 additions and 737 deletions

No files matched your search

+83 -3
View File
@@ -1,11 +1,91 @@
# Changelog # Changelog
## [0.2.3] - 2026-04-06 ## [1.0.1] - 2026-04-22
### Security
- Bump both pinned `rand` versions past [RUSTSEC-2026-0097][adv] (`rand`
is unsound with a custom `log::Log` impl that calls `rand::rng()` at
trace/warn level and hits the 64 KB reseed path):
- `rand 0.9.2` → `0.9.4` (fix `>= 0.9.3`).
- `rand 0.8.5` → `0.8.6` (fix `>= 0.8.6`).
The advisory is only reachable under a very narrow set of conditions
(custom logger invoking `rand::rng()` during a reseed while emitting
trace-level logs, or warn-level with a failed `getrandom`). We are
not known to do that, but shipping the upgrade is the right call.
[adv]: https://rustsec.org/advisories/RUSTSEC-2026-0097
## [1.0.0] - 2026-04-22
### Breaking
- **Upgraded to `donglora-client` 1.0.0-alpha.1 (DongLoRa Protocol v2 wire protocol).**
Wire-incompatible with 0.x firmware. Connect/config/TX/RX all go through
the new async `Dongle` API.
- **Radio loop is now a tokio task**, not a dedicated `std::thread`.
Everything is async end-to-end.
- **`[radio]` config schema changes:**
- `tx_power` renamed to `tx_power_dbm` and is now a plain `i8` (no
`"max"` keyword — v1.0 exposes a concrete dBm range via `GET_INFO`).
- `cad` field removed — CAD is a per-TX flag in v1.0, not a config
property. The `[tx]` section's `skip_cad` (default false) controls it.
- Default `sync_word` changed `0x3444` → `0x1424` to match the
`SX126x` register encoding of `MeshCore`'s private sync byte
(`RADIOLIB_SX126X_SYNC_WORD_PRIVATE`).
### Added
- **TX retry with backoff (new `[tx]` config section).** Every TX now
runs through `Dongle::tx_with_retry`, which retries `TX_DONE(CHANNEL_BUSY)`
and `ERR(EBUSY)` per spec `PROTOCOL.md §6.10` / §C.5.5. Defaults: 3
attempts, randomized 20–100 ms backoff, doubling up to 500 ms cap,
5 s per-attempt timeout. Configurable via `[tx]` keys `max_attempts`,
`backoff_min_ms`, `backoff_max_ms`, `backoff_cap_ms`,
`backoff_multiplier`, `per_attempt_timeout_secs`.
- **TUI packet log shows retry state.** In-flight retries render with a
`↻` marker in yellow and an `N/M` attempt counter. Terminal TX failures
render in red with `✗`. Post-retry success shows the final attempt count
(`OK 2/3`).
- **Stats: `tx_retries` and `tx_failures` counters.** Lifetime totals,
surfaced in the `StatsSnapshot`.
- **Measured airtime** from `TX_DONE` is now reported instead of a
theoretical estimate, feeding the rate limiter with the real on-air
duration.
### Removed
- The explicit 2-second liveness ping loop in the radio thread. The
`Dongle` now runs a 500 ms keepalive task internally, matching spec
§3.4's 1 s inactivity window with 2× margin.
## [0.3.1] - 2026-04-07
### Fixed
- **Sticky mux reconnect.** Once the bridge connects via mux, all subsequent
reconnects use mux-only mode (`connect_mux_auto`). Previously, each reconnect
re-ran the full mux→USB fallback chain, which could steal the serial port
from the mux during a brief disconnect.
## [0.3.0] - 2026-04-07
### Changed ### Changed
- README: clarify that the bridge relays raw LoRa packets and works with any - Upgraded to `donglora-client` 0.2: all connections now auto-validate via
protocol (Meshtastic, MeshCore, etc.) simultaneously. ping-on-connect, rejecting non-DongLoRa serial devices within 200ms.
- Simplified connection logic — removed manual mux/serial fallback and manual
ping; `donglora_client::connect()` handles the full fallback chain and
validation internally.
- Config negotiation detects mux vs serial via transport type instead of
string comparison.
### Fixed
- Repository and homepage URLs now point to `github.com/donglora/bridge`
(was `swaits/donglora-bridge`).
- README links updated to `donglora` GitHub org.
## [0.2.2] - 2026-04-06 ## [0.2.2] - 2026-04-06
Generated
+90 -58
View File
@@ -906,7 +906,7 @@ dependencies = [
[[package]] [[package]]
name = "donglora-bridge" name = "donglora-bridge"
version = "0.2.2" version = "1.0.1"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"blake3", "blake3",
@@ -922,7 +922,7 @@ dependencies = [
"jiff", "jiff",
"mainline", "mainline",
"proptest", "proptest",
"rand 0.9.2", "rand 0.9.4",
"ratatui", "ratatui",
"serde", "serde",
"sha2 0.10.9", "sha2 0.10.9",
@@ -938,13 +938,26 @@ dependencies = [
[[package]] [[package]]
name = "donglora-client" name = "donglora-client"
version = "0.1.0" version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "162468289ac189a798c4ecd961ed92c5dde2d9c5709d779ae324e0bb9a5ac4dc" checksum = "e2d2eb852b012cd692ab270397cdc81ac52c95a7b6543e8d8f6ee1e98537c461"
dependencies = [ dependencies = [
"anyhow", "donglora-protocol",
"serialport", "rand 0.9.4",
"thiserror 2.0.18",
"tokio",
"tokio-serial",
"tokio-stream",
"tracing", "tracing",
]
[[package]]
name = "donglora-protocol"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c8f2743ac7668845500e3686890917cdf5b0a09d17ec099bde2db4d3fae2a4a4"
dependencies = [
"heapless 0.9.2",
"ucobs", "ucobs",
] ]
@@ -1065,15 +1078,15 @@ checksum = "4e7f34442dbe69c60fe8eaf58a8cafff81a1f278816d8ab4db255b3bef4ac3c4"
dependencies = [ dependencies = [
"getrandom 0.3.4", "getrandom 0.3.4",
"libm", "libm",
"rand 0.9.2", "rand 0.9.4",
"siphasher", "siphasher",
] ]
[[package]] [[package]]
name = "fastrand" name = "fastrand"
version = "2.4.0" version = "2.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a043dc74da1e37d6afe657061213aa6f425f855399a11d3463c6ecccc4dfda1f" checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6"
[[package]] [[package]]
name = "fiat-crypto" name = "fiat-crypto"
@@ -1398,6 +1411,15 @@ dependencies = [
"byteorder", "byteorder",
] ]
[[package]]
name = "hash32"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "47d60b12902ba28e2730cd37e95b8c9223af2808df9e902d4df49588d1470606"
dependencies = [
"byteorder",
]
[[package]] [[package]]
name = "hashbrown" name = "hashbrown"
version = "0.15.5" version = "0.15.5"
@@ -1425,13 +1447,23 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cdc6457c0eb62c71aac4bc17216026d8410337c4126773b9c5daba343f17964f" checksum = "cdc6457c0eb62c71aac4bc17216026d8410337c4126773b9c5daba343f17964f"
dependencies = [ dependencies = [
"atomic-polyfill", "atomic-polyfill",
"hash32", "hash32 0.2.1",
"rustc_version", "rustc_version",
"serde", "serde",
"spin 0.9.8", "spin 0.9.8",
"stable_deref_trait", "stable_deref_trait",
] ]
[[package]]
name = "heapless"
version = "0.9.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2af2455f757db2b292a9b1768c4b70186d443bcb3b316252d6b540aec1cd89ed"
dependencies = [
"hash32 0.3.1",
"stable_deref_trait",
]
[[package]] [[package]]
name = "heck" name = "heck"
version = "0.5.0" version = "0.5.0"
@@ -1463,7 +1495,7 @@ dependencies = [
"idna", "idna",
"ipnet", "ipnet",
"once_cell", "once_cell",
"rand 0.9.2", "rand 0.9.4",
"ring", "ring",
"rustls", "rustls",
"thiserror 2.0.18", "thiserror 2.0.18",
@@ -1487,7 +1519,7 @@ dependencies = [
"moka", "moka",
"once_cell", "once_cell",
"parking_lot", "parking_lot",
"rand 0.9.2", "rand 0.9.4",
"resolv-conf", "resolv-conf",
"rustls", "rustls",
"smallvec", "smallvec",
@@ -1773,7 +1805,7 @@ dependencies = [
"hyper", "hyper",
"hyper-util", "hyper-util",
"log", "log",
"rand 0.9.2", "rand 0.9.4",
"tokio", "tokio",
"url", "url",
"xmltree", "xmltree",
@@ -1894,7 +1926,7 @@ dependencies = [
"pkcs8", "pkcs8",
"portable-atomic", "portable-atomic",
"portmapper", "portmapper",
"rand 0.9.2", "rand 0.9.4",
"reqwest", "reqwest",
"rustc-hash", "rustc-hash",
"rustls", "rustls",
@@ -1957,7 +1989,7 @@ dependencies = [
"n0-error", "n0-error",
"n0-future", "n0-future",
"postcard", "postcard",
"rand 0.9.2", "rand 0.9.4",
"serde", "serde",
"tokio", "tokio",
"tokio-util", "tokio-util",
@@ -2020,7 +2052,7 @@ dependencies = [
"pin-project", "pin-project",
"pkarr", "pkarr",
"postcard", "postcard",
"rand 0.9.2", "rand 0.9.4",
"reqwest", "reqwest",
"rustls", "rustls",
"rustls-pki-types", "rustls-pki-types",
@@ -2190,26 +2222,6 @@ dependencies = [
"libc", "libc",
] ]
[[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 = "line-clipping" name = "line-clipping"
version = "0.3.7" version = "0.3.7"
@@ -2375,6 +2387,19 @@ dependencies = [
"windows-sys 0.61.2", "windows-sys 0.61.2",
] ]
[[package]]
name = "mio-serial"
version = "5.0.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "029e1f407e261176a983a6599c084efd322d9301028055c87174beac71397ba3"
dependencies = [
"log",
"mio",
"nix 0.29.0",
"serialport",
"winapi",
]
[[package]] [[package]]
name = "moka" name = "moka"
version = "0.12.15" version = "0.12.15"
@@ -2621,7 +2646,7 @@ dependencies = [
"getrandom 0.3.4", "getrandom 0.3.4",
"identity-hash", "identity-hash",
"lru-slab", "lru-slab",
"rand 0.9.2", "rand 0.9.4",
"ring", "ring",
"rustc-hash", "rustc-hash",
"rustls", "rustls",
@@ -2958,7 +2983,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3c80231409c20246a13fddb31776fb942c38553c51e871f8cbd687a4cfb5843d" checksum = "3c80231409c20246a13fddb31776fb942c38553c51e871f8cbd687a4cfb5843d"
dependencies = [ dependencies = [
"phf_shared", "phf_shared",
"rand 0.8.5", "rand 0.8.6",
] ]
[[package]] [[package]]
@@ -3038,12 +3063,6 @@ dependencies = [
"spki", "spki",
] ]
[[package]]
name = "pkg-config"
version = "0.3.32"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c"
[[package]] [[package]]
name = "plist" name = "plist"
version = "1.8.0" version = "1.8.0"
@@ -3105,7 +3124,7 @@ dependencies = [
"n0-error", "n0-error",
"netwatch", "netwatch",
"num_enum", "num_enum",
"rand 0.9.2", "rand 0.9.4",
"serde", "serde",
"smallvec", "smallvec",
"socket2", "socket2",
@@ -3126,7 +3145,7 @@ dependencies = [
"cobs", "cobs",
"embedded-io 0.4.0", "embedded-io 0.4.0",
"embedded-io 0.6.1", "embedded-io 0.6.1",
"heapless", "heapless 0.7.17",
"postcard-derive", "postcard-derive",
"serde", "serde",
] ]
@@ -3204,7 +3223,7 @@ dependencies = [
"bit-vec 0.8.0", "bit-vec 0.8.0",
"bitflags 2.11.0", "bitflags 2.11.0",
"num-traits", "num-traits",
"rand 0.9.2", "rand 0.9.4",
"rand_chacha", "rand_chacha",
"rand_xorshift", "rand_xorshift",
"regex-syntax", "regex-syntax",
@@ -3257,7 +3276,7 @@ dependencies = [
"bytes", "bytes",
"getrandom 0.3.4", "getrandom 0.3.4",
"lru-slab", "lru-slab",
"rand 0.9.2", "rand 0.9.4",
"ring", "ring",
"rustc-hash", "rustc-hash",
"rustls", "rustls",
@@ -3306,18 +3325,18 @@ checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"
[[package]] [[package]]
name = "rand" name = "rand"
version = "0.8.5" version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "34af8d1a0e25924bc5b7c43c079c942339d8f0a8b57c39049bef581b46327404" checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a"
dependencies = [ dependencies = [
"rand_core 0.6.4", "rand_core 0.6.4",
] ]
[[package]] [[package]]
name = "rand" name = "rand"
version = "0.9.2" version = "0.9.4"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6db2770f06117d490610c7488547d543617b21bfa07796d7a12f6f1bd53850d1" checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea"
dependencies = [ dependencies = [
"rand_chacha", "rand_chacha",
"rand_core 0.9.5", "rand_core 0.9.5",
@@ -3607,9 +3626,9 @@ dependencies = [
[[package]] [[package]]
name = "rustls-webpki" name = "rustls-webpki"
version = "0.103.10" version = "0.103.13"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "df33b2b81ac578cabaf06b89b0631153a3f416b0a886e8a7a1707fb51abbd1ef" checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e"
dependencies = [ dependencies = [
"ring", "ring",
"rustls-pki-types", "rustls-pki-types",
@@ -3775,7 +3794,6 @@ dependencies = [
"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",
@@ -4284,6 +4302,20 @@ dependencies = [
"tokio", "tokio",
] ]
[[package]]
name = "tokio-serial"
version = "5.4.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "aa1d5427f11ba7c5e6384521cfd76f2d64572ff29f3f4f7aa0f496282923fdc8"
dependencies = [
"cfg-if",
"futures",
"log",
"mio-serial",
"serialport",
"tokio",
]
[[package]] [[package]]
name = "tokio-stream" name = "tokio-stream"
version = "0.1.18" version = "0.1.18"
@@ -4323,7 +4355,7 @@ dependencies = [
"getrandom 0.3.4", "getrandom 0.3.4",
"http", "http",
"httparse", "httparse",
"rand 0.9.2", "rand 0.9.4",
"ring", "ring",
"rustls-pki-types", "rustls-pki-types",
"simdutf8", "simdutf8",
@@ -4358,9 +4390,9 @@ dependencies = [
[[package]] [[package]]
name = "toml_edit" name = "toml_edit"
version = "0.25.10+spec-1.1.0" version = "0.25.11+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a82418ca169e235e6c399a84e395ab6debeb3bc90edc959bf0f48647c6a32d1b" checksum = "0b59c4d22ed448339746c59b905d24568fcbb3ab65a500494f7b8c3e97739f2b"
dependencies = [ dependencies = [
"indexmap", "indexmap",
"toml_datetime", "toml_datetime",
+4 -4
View File
@@ -1,12 +1,12 @@
[package] [package]
name = "donglora-bridge" name = "donglora-bridge"
version = "0.2.3" version = "1.0.1"
edition = "2024" edition = "2024"
authors = ["Stephen Waits <steve@waits.net>"] authors = ["Stephen Waits <steve@waits.net>"]
license = "MIT" license = "MIT"
description = "Peer-to-peer LoRa bridge using iroh and DongLoRa" description = "Peer-to-peer LoRa bridge using iroh and DongLoRa"
repository = "https://github.com/swaits/donglora-bridge" repository = "https://github.com/donglora/bridge"
homepage = "https://github.com/swaits/donglora-bridge" homepage = "https://github.com/donglora/bridge"
keywords = ["lora", "bridge", "gossip", "p2p", "iot"] keywords = ["lora", "bridge", "gossip", "p2p", "iot"]
categories = ["network-programming", "command-line-utilities"] categories = ["network-programming", "command-line-utilities"]
readme = "README.md" readme = "README.md"
@@ -14,7 +14,7 @@ rust-version = "1.94.1"
exclude = [".jj/", ".github/"] exclude = [".jj/", ".github/"]
[dependencies] [dependencies]
donglora-client = "0.1" donglora-client = "1.0.0"
iroh = "0.97" iroh = "0.97"
iroh-gossip = "0.97" iroh-gossip = "0.97"
mainline = { version = "6", features = ["async"] } mainline = { version = "6", features = ["async"] }
+12 -15
View File
@@ -1,22 +1,19 @@
# donglora-bridge # donglora-bridge
[![CI](https://github.com/swaits/donglora-bridge/actions/workflows/ci.yml/badge.svg)](https://github.com/swaits/donglora-bridge/actions/workflows/ci.yml) [![CI](https://github.com/donglora/bridge/actions/workflows/ci.yml/badge.svg)](https://github.com/donglora/bridge/actions/workflows/ci.yml)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE) [![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)
![Rust](https://img.shields.io/badge/rust-1.94.1%2B-orange) ![Rust](https://img.shields.io/badge/rust-1.94.1%2B-orange)
**donglora-bridge** relays raw LoRa packets over the internet. Every packet your **donglora-bridge** connects LoRa radios to the internet so packets heard by one
radio hears is forwarded to every other bridge running the same passphrase, and radio can be retransmitted by another radio anywhere in the world.
their radios retransmit it locally. Because it operates on raw packets, it works
with any protocol built on LoRa --
[Meshtastic](https://meshtastic.org/),
[MeshCore](https://github.com/rocketgithub/meshcore-firmware),
or anything else -- simultaneously and without configuration. If your radio can
hear it, the bridge will relay it.
[LoRa](https://en.wikipedia.org/wiki/LoRa) is a long-range, low-power wireless [LoRa](https://en.wikipedia.org/wiki/LoRa) is a long-range, low-power wireless
modulation used by off-grid mesh networks. A single radio might cover a few protocol used by off-grid mesh networks like
kilometers. This bridge removes that limit, extending a local LoRa network [MeshCore](https://github.com/rocketgithub/meshcore-firmware). A single LoRa
across a larger geographic area. radio might cover a few kilometers. This bridge removes that limit: any packet
your radio hears gets relayed over the internet to every other bridge running the
same passphrase, and their radios retransmit it locally. The result is a single
mesh network across a larger geographic area.
There is no central server. Bridges discover each other automatically using the There is no central server. Bridges discover each other automatically using the
[Mainline DHT](https://en.wikipedia.org/wiki/Mainline_DHT) and communicate [Mainline DHT](https://en.wikipedia.org/wiki/Mainline_DHT) and communicate
@@ -28,12 +25,12 @@ through an encrypted peer-to-peer gossip swarm.
### 1. A DongLoRa radio ### 1. A DongLoRa radio
[DongLoRa](https://github.com/swaits/donglora) is open-source firmware that [DongLoRa](https://github.com/donglora/firmware) is open-source firmware that
turns a supported LoRa development board (such as a Heltec WiFi LoRa 32 or turns a supported LoRa development board (such as a Heltec WiFi LoRa 32 or
similar ESP32+SX1262 board) into a USB LoRa dongle. You will need to: similar ESP32+SX1262 board) into a USB LoRa dongle. You will need to:
1. **Get a board** -- any ESP32 board with an SX1262 LoRa radio that DongLoRa 1. **Get a board** -- any ESP32 board with an SX1262 LoRa radio that DongLoRa
supports. See the [DongLoRa README](https://github.com/swaits/donglora) for supports. See the [DongLoRa README](https://github.com/donglora/firmware) for
the full list. the full list.
2. **Flash the DongLoRa firmware** onto the board. Instructions are in the 2. **Flash the DongLoRa firmware** onto the board. Instructions are in the
DongLoRa repository. DongLoRa repository.
@@ -71,7 +68,7 @@ mise use -g cargo:donglora-bridge
Or build from source: Or build from source:
```sh ```sh
git clone https://github.com/swaits/donglora-bridge.git git clone https://github.com/donglora/bridge.git
cd donglora-bridge cd donglora-bridge
cargo build --release cargo build --release
``` ```
+44 -35
View File
@@ -1,60 +1,69 @@
# Default: run all checks set shell := ["bash", "-c"]
default: build default: build
# Run all checks (fmt, clippy, test) [private]
_ensure_tools:
@mise trust --yes . 2>/dev/null; mise install --quiet
# Run the bridge (TUI mode)
run *args: _ensure_tools
cargo run --release -- {{args}}
# Run in headless mode
run-headless *args: _ensure_tools
cargo run --release -- --log-only {{args}}
# Run the config wizard
config: _ensure_tools
cargo run --release -- config
# Run all checks (fmt, clippy, test, audit)
check: fmt-check clippy test audit check: fmt-check clippy test audit
# Format code # Format code
fmt: fmt: _ensure_tools
cargo fmt cargo fmt
# Check formatting without changing files # Check formatting without changing files
fmt-check: fmt-check: _ensure_tools
cargo fmt --check cargo fmt --check
# Lint with clippy (pedantic + nursery) # Lint with clippy
clippy: clippy: _ensure_tools
cargo clippy --all-targets -- -D warnings cargo clippy --all-targets -- -D warnings
# Build release binary
build:
cargo build --release
# Generate docs
doc:
cargo doc --no-deps --open
# Run the bridge (TUI mode)
run *ARGS:
cargo run --release -- {{ARGS}}
# Run in headless mode
run-headless *ARGS:
cargo run --release -- --log-only {{ARGS}}
# Run the config wizard
config:
cargo run --release -- config
# Security + license audit
audit:
cargo deny check
# Run all tests # Run all tests
test: test-unit test-fuzz test-mutants test: test-unit
# Normal unit tests # Normal unit tests
test-unit: test-unit: _ensure_tools
cargo nextest run cargo nextest run
# Mutation testing # Mutation testing
test-mutants *ARGS: test-mutants *args: _ensure_tools
cargo mutants {{ARGS}} cargo mutants {{args}}
# Fuzz a target (requires nightly) # Fuzz a target (requires nightly)
test-fuzz TARGET="fuzz_gossip_frame_decode" DURATION="60": test-fuzz TARGET="fuzz_gossip_frame_decode" DURATION="60": _ensure_tools
cargo +nightly fuzz run {{TARGET}} -- -max_total_time={{DURATION}} cargo +nightly fuzz run {{TARGET}} -- -max_total_time={{DURATION}}
# Security + license audit
audit: _ensure_tools
cargo deny check
# Build release binary
build: _ensure_tools
cargo build --release
# Generate docs
doc: _ensure_tools
cargo doc --no-deps --open
# Check for outdated dependencies
outdated: _ensure_tools
cargo outdated
# Clean build artifacts # Clean build artifacts
clean: clean:
cargo clean cargo clean
+3
View File
@@ -0,0 +1,3 @@
[tools]
rust = "1.94"
just = "latest"
+167 -182
View File
@@ -1,12 +1,13 @@
//! TOML configuration loading and validation. //! TOML configuration loading and validation.
//! //!
//! Handles the `[radio]` and `[bridge]` sections with `MeshCore` US/Canada recommended //! Handles the `[radio]`, `[bridge]`, and `[tx]` sections with `MeshCore` US/Canada
//! defaults, and conversion to donglora-client `RadioConfig`. //! recommended defaults, and conversion to the donglora-client `LoRaConfig`.
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::{Context, Result, bail}; use anyhow::{Context, Result, bail};
use donglora_client::{Bandwidth, RadioConfig, TX_POWER_MAX}; use donglora_client::{LoRaBandwidth, LoRaCodingRate, LoRaConfig, LoRaHeaderMode, RetryPolicy};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
/// Top-level bridge configuration. /// Top-level bridge configuration.
@@ -16,6 +17,8 @@ pub struct Config {
pub radio: RadioSection, pub radio: RadioSection,
#[serde(default)] #[serde(default)]
pub bridge: BridgeSection, pub bridge: BridgeSection,
#[serde(default)]
pub tx: TxSection,
} }
#[derive(Debug, Clone, Deserialize, Serialize)] #[derive(Debug, Clone, Deserialize, Serialize)]
@@ -33,11 +36,9 @@ pub struct RadioSection {
#[serde(default = "default_sync_word")] #[serde(default = "default_sync_word")]
pub sync_word: u16, pub sync_word: u16,
#[serde(default = "default_tx_power")] #[serde(default = "default_tx_power")]
pub tx_power: String, pub tx_power_dbm: i8,
#[serde(default = "default_preamble")] #[serde(default = "default_preamble")]
pub preamble: u16, pub preamble: u16,
#[serde(default = "default_true")]
pub cad: bool,
} }
#[derive(Debug, Clone, Deserialize, Serialize)] #[derive(Debug, Clone, Deserialize, Serialize)]
@@ -55,6 +56,25 @@ pub struct BridgeSection {
pub log_file: Option<PathBuf>, pub log_file: Option<PathBuf>,
} }
/// Per-TX retry policy. Applied to every `tx_with_retry` call. Defaults quote
/// spec `PROTOCOL.md §C.5.5`: 3 attempts, randomized 20–100 ms backoff on the
/// first retry, doubling up to a 500 ms cap.
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct TxSection {
#[serde(default = "default_max_attempts")]
pub max_attempts: u8,
#[serde(default = "default_backoff_min_ms")]
pub backoff_min_ms: u32,
#[serde(default = "default_backoff_max_ms")]
pub backoff_max_ms: u32,
#[serde(default = "default_backoff_cap_ms")]
pub backoff_cap_ms: u32,
#[serde(default = "default_backoff_multiplier")]
pub backoff_multiplier: f32,
#[serde(default = "default_per_attempt_timeout_secs")]
pub per_attempt_timeout_secs: u64,
}
// ── Defaults (MeshCore US/Canada Recommended) ──────────────────── // ── Defaults (MeshCore US/Canada Recommended) ────────────────────
const fn default_frequency() -> u32 { const fn default_frequency() -> u32 {
@@ -69,24 +89,41 @@ const fn default_sf() -> u8 {
const fn default_cr() -> u8 { const fn default_cr() -> u8 {
5 5
} }
/// `MeshCore` private sync word encoded for the `SX126x` (see client-py
/// `examples/_common.py` — `RADIOLIB_SX126X_SYNC_WORD_PRIVATE`).
const fn default_sync_word() -> u16 { const fn default_sync_word() -> u16 {
0x3444 0x1424
} }
fn default_tx_power() -> String { const fn default_tx_power() -> i8 {
"22".into() 20
} }
const fn default_preamble() -> u16 { const fn default_preamble() -> u16 {
16 16
} }
const fn default_true() -> bool {
true
}
const fn default_dedup_window() -> u64 { const fn default_dedup_window() -> u64 {
300 300
} }
const fn default_tx_queue_size() -> usize { const fn default_tx_queue_size() -> usize {
32 32
} }
const fn default_max_attempts() -> u8 {
3
}
const fn default_backoff_min_ms() -> u32 {
20
}
const fn default_backoff_max_ms() -> u32 {
100
}
const fn default_backoff_cap_ms() -> u32 {
500
}
const fn default_backoff_multiplier() -> f32 {
2.0
}
const fn default_per_attempt_timeout_secs() -> u64 {
5
}
impl Default for RadioSection { impl Default for RadioSection {
fn default() -> Self { fn default() -> Self {
@@ -97,9 +134,8 @@ impl Default for RadioSection {
spreading_factor: default_sf(), spreading_factor: default_sf(),
coding_rate: default_cr(), coding_rate: default_cr(),
sync_word: default_sync_word(), sync_word: default_sync_word(),
tx_power: default_tx_power(), tx_power_dbm: default_tx_power(),
preamble: default_preamble(), preamble: default_preamble(),
cad: true,
} }
} }
} }
@@ -116,56 +152,95 @@ impl Default for BridgeSection {
} }
} }
impl Default for TxSection {
fn default() -> Self {
Self {
max_attempts: default_max_attempts(),
backoff_min_ms: default_backoff_min_ms(),
backoff_max_ms: default_backoff_max_ms(),
backoff_cap_ms: default_backoff_cap_ms(),
backoff_multiplier: default_backoff_multiplier(),
per_attempt_timeout_secs: default_per_attempt_timeout_secs(),
}
}
}
// ── Conversion ──────────────────────────────────────────────────── // ── Conversion ────────────────────────────────────────────────────
fn parse_bandwidth(s: &str) -> Result<Bandwidth> { fn parse_bandwidth(s: &str) -> Result<LoRaBandwidth> {
match s.to_lowercase().as_str() { match s.to_lowercase().as_str() {
"7.8khz" | "7khz" | "khz7" => Ok(Bandwidth::Khz7), "7.8khz" | "7khz" | "khz7" => Ok(LoRaBandwidth::Khz7),
"10.4khz" | "10khz" | "khz10" => Ok(Bandwidth::Khz10), "10.4khz" | "10khz" | "khz10" => Ok(LoRaBandwidth::Khz10),
"15.6khz" | "15khz" | "khz15" => Ok(Bandwidth::Khz15), "15.6khz" | "15khz" | "khz15" => Ok(LoRaBandwidth::Khz15),
"20.8khz" | "20khz" | "khz20" => Ok(Bandwidth::Khz20), "20.8khz" | "20khz" | "khz20" => Ok(LoRaBandwidth::Khz20),
"31.25khz" | "31khz" | "khz31" => Ok(Bandwidth::Khz31), "31.25khz" | "31khz" | "khz31" => Ok(LoRaBandwidth::Khz31),
"41.7khz" | "41khz" | "khz41" => Ok(Bandwidth::Khz41), "41.7khz" | "41khz" | "khz41" => Ok(LoRaBandwidth::Khz41),
"62.5khz" | "62khz" | "khz62" => Ok(Bandwidth::Khz62), "62.5khz" | "62khz" | "khz62" => Ok(LoRaBandwidth::Khz62),
"125khz" | "khz125" => Ok(Bandwidth::Khz125), "125khz" | "khz125" => Ok(LoRaBandwidth::Khz125),
"250khz" | "khz250" => Ok(Bandwidth::Khz250), "250khz" | "khz250" => Ok(LoRaBandwidth::Khz250),
"500khz" | "khz500" => Ok(Bandwidth::Khz500), "500khz" | "khz500" => Ok(LoRaBandwidth::Khz500),
_ => bail!("unknown bandwidth: {s}"), _ => bail!("unknown bandwidth: {s}"),
} }
} }
fn parse_tx_power(s: &str) -> Result<i8> { fn parse_coding_rate(v: u8) -> Result<LoRaCodingRate> {
if s.eq_ignore_ascii_case("max") { match v {
Ok(TX_POWER_MAX) 5 => Ok(LoRaCodingRate::Cr4_5),
} else { 6 => Ok(LoRaCodingRate::Cr4_6),
s.parse::<i8>().context("invalid tx_power (use 'max' or a dBm integer)") 7 => Ok(LoRaCodingRate::Cr4_7),
8 => Ok(LoRaCodingRate::Cr4_8),
_ => bail!("coding_rate must be 5-8 (meaning 4/5 to 4/8), got {v}"),
} }
} }
impl RadioSection { impl RadioSection {
/// Convert to the donglora-client `RadioConfig`. /// Convert to the donglora-client `LoRaConfig`.
/// ///
/// Validates that all radio parameters are within hardware-supported ranges /// Validates that all radio parameters are within hardware-supported ranges
/// before constructing the config. /// before constructing the config.
pub fn to_radio_config(&self) -> Result<RadioConfig> { pub fn to_lora_config(&self) -> Result<LoRaConfig> {
if !(7..=12).contains(&self.spreading_factor) { if !(7..=12).contains(&self.spreading_factor) {
bail!("spreading_factor must be 7-12, got {}", self.spreading_factor); bail!("spreading_factor must be 7-12, got {}", self.spreading_factor);
} }
if !(5..=8).contains(&self.coding_rate) {
bail!("coding_rate must be 5-8 (meaning 4/5 to 4/8), got {}", self.coding_rate);
}
if self.frequency < 150_000_000 || self.frequency > 1_020_000_000 { if self.frequency < 150_000_000 || self.frequency > 1_020_000_000 {
bail!("frequency {:.3} MHz out of range (150 MHz to 1.02 GHz)", f64::from(self.frequency) / 1_000_000.0); bail!("frequency {:.3} MHz out of range (150 MHz to 1.02 GHz)", f64::from(self.frequency) / 1_000_000.0);
} }
Ok(RadioConfig { Ok(LoRaConfig {
freq_hz: self.frequency, freq_hz: self.frequency,
bw: parse_bandwidth(&self.bandwidth)?, bw: parse_bandwidth(&self.bandwidth)?,
sf: self.spreading_factor, sf: self.spreading_factor,
cr: self.coding_rate, cr: parse_coding_rate(self.coding_rate)?,
sync_word: self.sync_word, sync_word: self.sync_word,
tx_power_dbm: parse_tx_power(&self.tx_power)?, tx_power_dbm: self.tx_power_dbm,
preamble_len: self.preamble, preamble_len: self.preamble,
cad: u8::from(self.cad), header_mode: LoRaHeaderMode::Explicit,
payload_crc: true,
iq_invert: false,
})
}
}
impl TxSection {
/// Materialize the TOML fields into a [`RetryPolicy`].
pub fn to_retry_policy(&self) -> Result<RetryPolicy> {
if self.max_attempts == 0 {
bail!("tx.max_attempts must be >= 1");
}
if self.backoff_min_ms > self.backoff_max_ms {
bail!(
"tx.backoff_min_ms ({}) must not exceed backoff_max_ms ({})",
self.backoff_min_ms,
self.backoff_max_ms
);
}
Ok(RetryPolicy {
max_attempts: self.max_attempts,
backoff_ms_min: self.backoff_min_ms,
backoff_ms_max: self.backoff_max_ms,
backoff_multiplier: self.backoff_multiplier,
backoff_cap_ms: self.backoff_cap_ms,
per_attempt_timeout: Duration::from_secs(self.per_attempt_timeout_secs),
skip_cad: false,
}) })
} }
} }
@@ -209,92 +284,97 @@ mod tests {
#[test] #[test]
fn parse_bandwidth_canonical() { fn parse_bandwidth_canonical() {
assert!(matches!(parse_bandwidth("7.8kHz"), Ok(Bandwidth::Khz7))); assert!(matches!(parse_bandwidth("7.8kHz"), Ok(LoRaBandwidth::Khz7)));
assert!(matches!(parse_bandwidth("62.5kHz"), Ok(Bandwidth::Khz62))); assert!(matches!(parse_bandwidth("62.5kHz"), Ok(LoRaBandwidth::Khz62)));
assert!(matches!(parse_bandwidth("125kHz"), Ok(Bandwidth::Khz125))); assert!(matches!(parse_bandwidth("125kHz"), Ok(LoRaBandwidth::Khz125)));
assert!(matches!(parse_bandwidth("500kHz"), Ok(Bandwidth::Khz500))); assert!(matches!(parse_bandwidth("500kHz"), Ok(LoRaBandwidth::Khz500)));
} }
#[test] #[test]
fn parse_bandwidth_aliases() { fn parse_bandwidth_aliases() {
assert!(matches!(parse_bandwidth("7khz"), Ok(Bandwidth::Khz7))); assert!(matches!(parse_bandwidth("7khz"), Ok(LoRaBandwidth::Khz7)));
assert!(matches!(parse_bandwidth("khz125"), Ok(Bandwidth::Khz125))); assert!(matches!(parse_bandwidth("khz125"), Ok(LoRaBandwidth::Khz125)));
assert!(matches!(parse_bandwidth("10khz"), Ok(Bandwidth::Khz10)));
} }
#[test] #[test]
fn parse_bandwidth_case_insensitive() { fn parse_bandwidth_case_insensitive() {
assert!(matches!(parse_bandwidth("125KHZ"), Ok(Bandwidth::Khz125))); assert!(matches!(parse_bandwidth("125KHZ"), Ok(LoRaBandwidth::Khz125)));
assert!(matches!(parse_bandwidth("62.5KHZ"), Ok(Bandwidth::Khz62)));
} }
#[test] #[test]
fn parse_bandwidth_rejects_unknown() { fn parse_bandwidth_rejects_unknown() {
assert!(parse_bandwidth("999kHz").is_err()); assert!(parse_bandwidth("999kHz").is_err());
assert!(parse_bandwidth("").is_err()); assert!(parse_bandwidth("").is_err());
assert!(parse_bandwidth("not_a_bandwidth").is_err());
} }
#[test] #[test]
fn parse_tx_power_numeric() { fn parse_coding_rate_canonical() {
assert_eq!(parse_tx_power("22").unwrap(), 22); assert!(matches!(parse_coding_rate(5), Ok(LoRaCodingRate::Cr4_5)));
assert_eq!(parse_tx_power("-4").unwrap(), -4); assert!(matches!(parse_coding_rate(8), Ok(LoRaCodingRate::Cr4_8)));
assert_eq!(parse_tx_power("0").unwrap(), 0);
} }
#[test] #[test]
fn parse_tx_power_max_keyword() { fn parse_coding_rate_rejects_out_of_range() {
assert_eq!(parse_tx_power("max").unwrap(), TX_POWER_MAX); assert!(parse_coding_rate(4).is_err());
assert_eq!(parse_tx_power("MAX").unwrap(), TX_POWER_MAX); assert!(parse_coding_rate(9).is_err());
assert_eq!(parse_tx_power("Max").unwrap(), TX_POWER_MAX);
} }
#[test] #[test]
fn parse_tx_power_rejects_garbage() { fn radio_section_defaults_to_lora_config() {
assert!(parse_tx_power("abc").is_err());
assert!(parse_tx_power("").is_err());
}
#[test]
fn radio_section_defaults_to_radio_config() {
let section = RadioSection::default(); let section = RadioSection::default();
let config = section.to_radio_config().unwrap(); let cfg = section.to_lora_config().unwrap();
assert_eq!(config.freq_hz, 910_525_000); assert_eq!(cfg.freq_hz, 910_525_000);
assert_eq!(config.sf, 7); assert_eq!(cfg.sf, 7);
assert_eq!(config.cr, 5); assert!(matches!(cfg.cr, LoRaCodingRate::Cr4_5));
assert_eq!(config.sync_word, 0x3444); assert_eq!(cfg.sync_word, 0x1424);
assert_eq!(config.preamble_len, 16); assert_eq!(cfg.preamble_len, 16);
assert_eq!(cfg.tx_power_dbm, 20);
} }
#[test] #[test]
fn radio_section_max_power() { fn to_lora_config_rejects_invalid_sf() {
let section = RadioSection { tx_power: "max".into(), ..RadioSection::default() };
let config = section.to_radio_config().unwrap();
assert_eq!(config.tx_power_dbm, TX_POWER_MAX);
}
#[test]
fn to_radio_config_rejects_invalid_sf() {
let section = RadioSection { spreading_factor: 6, ..RadioSection::default() }; let section = RadioSection { spreading_factor: 6, ..RadioSection::default() };
assert!(section.to_radio_config().is_err()); assert!(section.to_lora_config().is_err());
let section = RadioSection { spreading_factor: 13, ..RadioSection::default() }; let section = RadioSection { spreading_factor: 13, ..RadioSection::default() };
assert!(section.to_radio_config().is_err()); assert!(section.to_lora_config().is_err());
} }
#[test] #[test]
fn to_radio_config_rejects_invalid_cr() { fn to_lora_config_rejects_invalid_cr() {
let section = RadioSection { coding_rate: 4, ..RadioSection::default() }; let section = RadioSection { coding_rate: 4, ..RadioSection::default() };
assert!(section.to_radio_config().is_err()); assert!(section.to_lora_config().is_err());
let section = RadioSection { coding_rate: 9, ..RadioSection::default() }; let section = RadioSection { coding_rate: 9, ..RadioSection::default() };
assert!(section.to_radio_config().is_err()); assert!(section.to_lora_config().is_err());
} }
#[test] #[test]
fn to_radio_config_rejects_invalid_frequency() { fn to_lora_config_rejects_invalid_frequency() {
let section = RadioSection { frequency: 0, ..RadioSection::default() }; let section = RadioSection { frequency: 0, ..RadioSection::default() };
assert!(section.to_radio_config().is_err()); assert!(section.to_lora_config().is_err());
let section = RadioSection { frequency: 2_000_000_000, ..RadioSection::default() }; let section = RadioSection { frequency: 2_000_000_000, ..RadioSection::default() };
assert!(section.to_radio_config().is_err()); assert!(section.to_lora_config().is_err());
}
#[test]
fn tx_section_defaults_produce_valid_policy() {
let p = TxSection::default().to_retry_policy().unwrap();
assert_eq!(p.max_attempts, 3);
assert_eq!(p.backoff_ms_min, 20);
assert_eq!(p.backoff_ms_max, 100);
assert_eq!(p.backoff_cap_ms, 500);
assert!((p.backoff_multiplier - 2.0).abs() < f32::EPSILON);
}
#[test]
fn tx_section_rejects_zero_attempts() {
let s = TxSection { max_attempts: 0, ..TxSection::default() };
assert!(s.to_retry_policy().is_err());
}
#[test]
fn tx_section_rejects_inverted_backoff_bounds() {
let s = TxSection { backoff_min_ms: 500, backoff_max_ms: 100, ..TxSection::default() };
assert!(s.to_retry_policy().is_err());
} }
#[test] #[test]
@@ -305,6 +385,7 @@ mod tests {
let cfg = load_config(&path).unwrap(); let cfg = load_config(&path).unwrap();
assert_eq!(cfg.bridge.passphrase, "my-secret"); assert_eq!(cfg.bridge.passphrase, "my-secret");
assert_eq!(cfg.radio.frequency, 910_525_000); assert_eq!(cfg.radio.frequency, 910_525_000);
assert_eq!(cfg.tx.max_attempts, 3);
} }
#[test] #[test]
@@ -335,100 +416,4 @@ mod tests {
std::fs::write(&path, "not valid toml {{{{").unwrap(); std::fs::write(&path, "not valid toml {{{{").unwrap();
assert!(load_config(&path).is_err()); assert!(load_config(&path).is_err());
} }
// ── Mutant killers: default values (items 1-5) ─────────────────
#[test]
fn default_radio_cad_is_true() {
let radio = RadioSection::default();
assert!(radio.cad);
// Also test via serde (exercises the default_true function used by #[serde(default)])
let radio: RadioSection = toml::from_str("").unwrap();
assert!(radio.cad);
}
#[test]
fn default_dedup_window_is_300() {
let bridge = BridgeSection::default();
assert_eq!(bridge.dedup_window_secs, 300);
}
#[test]
fn default_tx_queue_size_is_32() {
let bridge = BridgeSection::default();
assert_eq!(bridge.tx_queue_size, 32);
}
// ── Mutant killers: missing bandwidth variants (items 6-10) ──
#[test]
fn parse_bandwidth_all_canonical_variants() {
assert!(matches!(parse_bandwidth("15.6kHz"), Ok(Bandwidth::Khz15)));
assert!(matches!(parse_bandwidth("20.8kHz"), Ok(Bandwidth::Khz20)));
assert!(matches!(parse_bandwidth("31.25kHz"), Ok(Bandwidth::Khz31)));
assert!(matches!(parse_bandwidth("41.7kHz"), Ok(Bandwidth::Khz41)));
assert!(matches!(parse_bandwidth("250kHz"), Ok(Bandwidth::Khz250)));
}
// ── Mutant killers: frequency boundary (items 11-12) ─────────
#[test]
fn to_radio_config_accepts_boundary_frequencies() {
let low = RadioSection { frequency: 150_000_000, ..RadioSection::default() };
assert!(low.to_radio_config().is_ok());
let high = RadioSection { frequency: 1_020_000_000, ..RadioSection::default() };
assert!(high.to_radio_config().is_ok());
}
// ── Mutant killers: path helpers (items 13-15) ───────────────
#[test]
fn config_dir_is_non_empty() {
let dir = config_dir().unwrap();
assert!(dir.components().count() > 0);
}
#[test]
fn config_path_ends_with_config_toml() {
let path = config_path().unwrap();
assert_eq!(path.file_name().unwrap(), "config.toml");
}
#[test]
fn default_log_path_ends_with_bridge_log() {
let path = default_log_path().unwrap();
assert_eq!(path.file_name().unwrap(), "bridge.log");
}
#[test]
fn load_config_full_round_trip() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.toml");
std::fs::write(
&path,
r#"
[radio]
frequency = 915000000
bandwidth = "125kHz"
spreading_factor = 12
coding_rate = 8
sync_word = 0x1234
tx_power = "max"
preamble = 8
cad = false
[bridge]
passphrase = "test-phrase"
dedup_window_secs = 600
tx_queue_size = 64
"#,
)
.unwrap();
let cfg = load_config(&path).unwrap();
assert_eq!(cfg.radio.frequency, 915_000_000);
assert_eq!(cfg.radio.spreading_factor, 12);
assert_eq!(cfg.bridge.dedup_window_secs, 600);
assert_eq!(cfg.bridge.tx_queue_size, 64);
}
} }
+96 -73
View File
@@ -22,7 +22,7 @@ use mainline::{Dht, MutableItem, SigningKey};
use sha2::{Digest, Sha256}; use sha2::{Digest, Sha256};
use tokio::sync::mpsc; use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken; use tokio_util::sync::CancellationToken;
use tracing::{debug, info, warn}; use tracing::{debug, error, info, warn};
use crate::packet::GossipFrame; use crate::packet::GossipFrame;
@@ -31,14 +31,27 @@ use crate::packet::GossipFrame;
/// Conservative cap to stay under BEP44's 1000-byte value limit. /// Conservative cap to stay under BEP44's 1000-byte value limit.
const MAX_DHT_PEERS: usize = 27; const MAX_DHT_PEERS: usize = 27;
/// DHT heartbeat interval when connected (base, before jitter). 5 minutes /// DHT heartbeat interval (base, before jitter). Each bridge periodically
/// balances freshness against DHT write amplification — connected nodes already /// publishes `[self] + live_neighbors` as a blind overwrite. Any live bridge
/// have live gossip links and only need DHT for newcomer discovery. /// publishing at this cadence guarantees that the DHT record is overwritten
const DHT_HEARTBEAT_SECS: u64 = 300; /// with a fresh snapshot of a real, connected sub-swarm within this window —
/// so stale entries from crashed / restarted peers age out within one cycle.
const DHT_HEARTBEAT_SECS: u64 = 60;
/// Maximum random jitter added to heartbeat interval. Prevents synchronized /// Maximum random jitter added to heartbeat. Prevents synchronized publish
/// publish storms when multiple bridges start simultaneously. /// storms when multiple bridges start simultaneously.
const DHT_HEARTBEAT_JITTER_SECS: u64 = 60; const DHT_HEARTBEAT_JITTER_SECS: u64 = 15;
/// DHT read interval while we have no neighbors. Stays aggressive so two
/// freshly-started bridges find each other quickly.
const DHT_LONELY_READ_SECS: u64 = 3;
/// How long to try dialing DHT-discovered peers before we give up and publish
/// ourselves alone. If we connect to at least one peer in this window we
/// publish `[self, neighbors...]` right away. If we don't, we publish `[self]`
/// so newcomers find us — but we delay this so we don't overwrite a
/// potentially useful DHT record with just our own lonely entry.
const DHT_BOOTSTRAP_GRACE_SECS: u64 = 10;
// ── Key derivation ─────────────────────────────────────────────── // ── Key derivation ───────────────────────────────────────────────
@@ -321,12 +334,23 @@ fn spawn_receiver(
cancel: CancellationToken, cancel: CancellationToken,
) { ) {
tokio::spawn(async move { tokio::spawn(async move {
loop { // Log a hard warning if the receiver stream ever ends — this is the
// "silent one-way gossip" failure mode. Track why it exited so the
// log tells us whether it was a stream-end (None), an error
// (Some(Err)), or a cancel.
let exit_reason: &'static str = loop {
let event = tokio::select! { let event = tokio::select! {
event = receiver.next() => event, event = receiver.next() => event,
() = cancel.cancelled() => break, () = cancel.cancelled() => break "cancelled",
};
let event = match event {
None => break "stream ended (receiver closed)",
Some(Err(e)) => {
error!("gossip receiver error: {e:#}");
break "receiver error";
}
Some(Ok(ev)) => ev,
}; };
let Some(Ok(event)) = event else { break };
match event { match event {
Event::Received(msg) => { Event::Received(msg) => {
if let Some(frame) = GossipFrame::decode(&msg.content) { if let Some(frame) = GossipFrame::decode(&msg.content) {
@@ -364,8 +388,15 @@ fn spawn_receiver(
warn!("gossip receiver lagged — some messages were dropped"); warn!("gossip receiver lagged — some messages were dropped");
} }
} }
};
// A shutdown cancel is expected; any other exit means we silently
// stopped receiving gossip. That's the "one-way bridge" bug —
// broadcast still works because spawn_broadcaster is independent.
if exit_reason == "cancelled" {
debug!("gossip receiver stopped ({exit_reason})");
} else {
error!("gossip receiver STOPPED ({exit_reason}) — this bridge will no longer receive from the swarm",);
} }
debug!("gossip receiver stopped");
}); });
} }
@@ -391,24 +422,25 @@ fn spawn_broadcaster(
}); });
} }
/// DHT poll interval while alone. 3 seconds is aggressive but reads are cheap /// Periodically publish `[self] + live_neighbors` to the DHT and read it to
/// (single DHT GET) and fast discovery matters when two nodes start together. /// discover new peers.
const DHT_ALONE_READ_SECS: u64 = 3;
/// How often a lonely node publishes (merge-only, never destructive). 30 seconds
/// is much slower than reads because DHT PUT is heavier and a lonely node only
/// adds itself to the peer list.
const DHT_ALONE_PUBLISH_SECS: u64 = 30;
/// Periodically interact with the DHT to discover and share peers.
/// ///
/// Strategy: /// Strategy — deliberately simple:
/// - **Alone:** Read every 3s (cheap). Publish every 30s via merge — reads /// 1. **Startup**: read DHT, try to join discovered peers. Do NOT publish
/// existing data first, appends ourselves, writes back. Two fresh nodes /// yet — we don't know if any of them are still alive, and publishing
/// both doing this will eventually see each other. Never blindly overwrites. /// `[self]` alone would overwrite a potentially useful record.
/// - **Connected:** Read + publish on relaxed 5-min schedule. We have /// 2. **Bootstrap grace** (10 s): keep reading fast (every
/// valuable neighbor info to share. /// `DHT_LONELY_READ_SECS`) looking for peers to connect to. If a
/// - **Transition to connected:** Publish immediately. /// `NeighborUp` fires inside this window, publish immediately with the
/// live neighbor set. If nothing shows up, publish `[self]` so anyone
/// else joining finds us.
/// 3. **Steady state**: publish `[self, neighbors...]` every
/// `DHT_HEARTBEAT_SECS` (+ jitter), a blind overwrite. Any live bridge
/// doing this naturally ages out crashed/stale peers from the DHT
/// record within one cycle.
/// 4. **Re-read cadence**: fast (`DHT_LONELY_READ_SECS`) while
/// `neighbor_count == 0`, relaxed (= heartbeat interval) once
/// connected.
#[allow(clippy::too_many_arguments)] #[allow(clippy::too_many_arguments)]
fn spawn_dht_heartbeat( fn spawn_dht_heartbeat(
my_id: PublicKey, my_id: PublicKey,
@@ -430,27 +462,29 @@ fn spawn_dht_heartbeat(
}; };
let pub_key_bytes = signing_key.verifying_key().to_bytes(); let pub_key_bytes = signing_key.verifying_key().to_bytes();
let mut last_publish: Option<Instant> = None;
let mut was_alone = true; let mut was_alone = true;
let mut last_publish = Instant::now(); let started = Instant::now();
// Initial: read to discover existing peers, then merge-publish ourselves. // Step 1: read the DHT to find peers to dial.
dht_read_and_join(&dht, &my_id, &pub_key_bytes, &salt, &gossip_sender, &state).await; dht_read_and_join(&dht, &my_id, &pub_key_bytes, &salt, &gossip_sender, &state).await;
dht_publish(&dht, &my_id, &signing_key, &salt, &pub_key_bytes, &neighbors, &state).await;
loop { loop {
let alone = neighbor_count.load(Ordering::Relaxed) == 0; let alone = neighbor_count.load(Ordering::Relaxed) == 0;
// Transition: alone → connected. Publish immediately. // Transition: alone → connected. Publish immediately with live
// neighbors so anyone who was about to dial us-as-lonely finds
// the full picture.
if was_alone && !alone { if was_alone && !alone {
info!("first neighbor joined — publishing to DHT"); info!("first neighbor joined — publishing to DHT");
dht_publish(&dht, &my_id, &signing_key, &salt, &pub_key_bytes, &neighbors, &state).await; dht_publish(&dht, &my_id, &signing_key, &salt, &neighbors, &state).await;
last_publish = Instant::now(); last_publish = Some(Instant::now());
} }
was_alone = alone; was_alone = alone;
// Sleep: fast reads while alone, relaxed when connected. // Sleep: fast reads while lonely, relaxed once connected.
let sleep_dur = if alone { let sleep_dur = if alone {
Duration::from_secs(DHT_ALONE_READ_SECS) Duration::from_secs(DHT_LONELY_READ_SECS)
} else { } else {
let jitter = rand::random::<u64>() % DHT_HEARTBEAT_JITTER_SECS.max(1); let jitter = rand::random::<u64>() % DHT_HEARTBEAT_JITTER_SECS.max(1);
Duration::from_secs(DHT_HEARTBEAT_SECS + jitter) Duration::from_secs(DHT_HEARTBEAT_SECS + jitter)
@@ -461,18 +495,21 @@ fn spawn_dht_heartbeat(
() = cancel.cancelled() => break, () = cancel.cancelled() => break,
} }
// Always read — cheap, discovers peers to join. // Always read — cheap, keeps fresh peers coming in.
dht_read_and_join(&dht, &my_id, &pub_key_bytes, &salt, &gossip_sender, &state).await; dht_read_and_join(&dht, &my_id, &pub_key_bytes, &salt, &gossip_sender, &state).await;
// Publish cadence depends on state. // Steady-state: publish on every heartbeat tick once we've
let publish_interval = if alone { // published at least once. Pre-first-publish, hold off until
Duration::from_secs(DHT_ALONE_PUBLISH_SECS) // either a neighbor appears OR the bootstrap grace elapses —
} else { // this avoids wiping a potentially-useful DHT record with
Duration::from_secs(DHT_HEARTBEAT_SECS) // just our lonely [self] entry on startup.
}; let should_publish = last_publish.map_or_else(
if last_publish.elapsed() >= publish_interval { || !alone || started.elapsed() >= Duration::from_secs(DHT_BOOTSTRAP_GRACE_SECS),
dht_publish(&dht, &my_id, &signing_key, &salt, &pub_key_bytes, &neighbors, &state).await; |t| t.elapsed() >= Duration::from_secs(DHT_HEARTBEAT_SECS),
last_publish = Instant::now(); );
if should_publish {
dht_publish(&dht, &my_id, &signing_key, &salt, &neighbors, &state).await;
last_publish = Some(Instant::now());
} }
} }
debug!("DHT heartbeat stopped"); debug!("DHT heartbeat stopped");
@@ -503,44 +540,30 @@ async fn dht_read_and_join(
} }
} }
/// Publish `[self] + live_neighbors` to the DHT as a blind overwrite.
///
/// No merge with existing DHT content. Stale peers age out naturally
/// because every live bridge publishes a snapshot of its actual live
/// connections within `DHT_HEARTBEAT_SECS`, and higher-seq records win.
async fn dht_publish( async fn dht_publish(
dht: &AsyncDht, dht: &AsyncDht,
my_id: &PublicKey, my_id: &PublicKey,
signing_key: &SigningKey, signing_key: &SigningKey,
salt: &[u8; 20], salt: &[u8; 20],
pub_key_bytes: &[u8; 32],
neighbors: &Arc<Mutex<HashSet<PublicKey>>>, neighbors: &Arc<Mutex<HashSet<PublicKey>>>,
state: &Arc<Mutex<SwarmState>>, state: &Arc<Mutex<SwarmState>>,
) { ) {
let has_neighbors = neighbors.lock().is_ok_and(|n| !n.is_empty());
let mut seen = HashSet::new();
let mut peer_list: Vec<[u8; 32]> = Vec::with_capacity(MAX_DHT_PEERS); let mut peer_list: Vec<[u8; 32]> = Vec::with_capacity(MAX_DHT_PEERS);
// Always include ourselves.
peer_list.push(*my_id.as_bytes()); peer_list.push(*my_id.as_bytes());
seen.insert(*my_id.as_bytes());
if has_neighbors { // Snapshot current live neighbors — those we have an active gossip
// Connected: publish authoritative data — self + live neighbors only. // connection to right now. No inherited junk from prior DHT content.
// This purges dead/stale peers from the DHT. We know who's alive if let Ok(n) = neighbors.lock() {
// because they're our active gossip neighbors. for id in n.iter() {
if let Ok(n) = neighbors.lock() { if peer_list.len() >= MAX_DHT_PEERS {
for id in n.iter() { break;
if seen.insert(*id.as_bytes()) && peer_list.len() < MAX_DHT_PEERS {
peer_list.push(*id.as_bytes());
}
}
}
} else {
// Alone: merge-publish — read existing DHT data first, add ourselves,
// preserve other peers. We don't know who's alive so we can't purge.
if let Some(record) = dht.get_mutable_most_recent(pub_key_bytes, Some(salt)).await {
for existing in decode_peer_list(record.value()) {
if seen.insert(*existing.as_bytes()) && peer_list.len() < MAX_DHT_PEERS {
peer_list.push(*existing.as_bytes());
}
} }
peer_list.push(*id.as_bytes());
} }
} }
+102 -3
View File
@@ -71,17 +71,24 @@ async fn main() -> Result<()> {
info!("donglora-bridge starting, ephemeral id: {}", node_id.fmt_short()); info!("donglora-bridge starting, ephemeral id: {}", node_id.fmt_short());
// Parse radio config. // Parse radio config.
let radio_config = cfg.radio.to_radio_config()?; let radio_config = cfg.radio.to_lora_config()?;
let retry_policy = cfg.tx.to_retry_policy()?;
// Derive passphrase keys. // Derive passphrase keys.
let keys = PassphraseKeys::derive(&cfg.bridge.passphrase); let keys = PassphraseKeys::derive(&cfg.bridge.passphrase);
info!("topic: {}", hex::encode(&keys.topic_id.as_bytes()[..8])); info!("topic: {}", hex::encode(&keys.topic_id.as_bytes()[..8]));
// Create rate limiter. // Create rate limiter.
let cr_denom: u8 = match radio_config.cr {
donglora_client::LoRaCodingRate::Cr4_5 => 5,
donglora_client::LoRaCodingRate::Cr4_6 => 6,
donglora_client::LoRaCodingRate::Cr4_7 => 7,
donglora_client::LoRaCodingRate::Cr4_8 => 8,
};
let rate_limiter = RateLimiter::from_radio_config( let rate_limiter = RateLimiter::from_radio_config(
radio_config.sf, radio_config.sf,
radio_config.bw, radio_config.bw,
radio_config.cr, cr_denom,
radio_config.preamble_len, radio_config.preamble_len,
cfg.bridge.rate_limit_pps, cfg.bridge.rate_limit_pps,
); );
@@ -92,7 +99,7 @@ async fn main() -> Result<()> {
let swarm = Arc::new(swarm); let swarm = Arc::new(swarm);
// Start radio. // Start radio.
let (radio_event_rx, radio_tx) = radio::spawn(radio_config, cfg.radio.port.clone())?; let (radio_event_rx, radio_tx, _radio_handle) = radio::spawn(radio_config, cfg.radio.port.clone(), retry_policy)?;
// Stats + log channel + radio config watch. // Stats + log channel + radio config watch.
let stats = Arc::new(Stats::default()); let stats = Arc::new(Stats::default());
@@ -160,6 +167,13 @@ async fn main() -> Result<()> {
// Run TUI or headless. // Run TUI or headless.
if cli.log_only { if cli.log_only {
info!("running in headless mode"); info!("running in headless mode");
// In TUI mode the TUI drains `log_rx`; in headless mode nobody
// does unless we spawn a consumer. Without this, the router's
// bounded log channel backs up and router throughput stalls.
// Emit each entry at INFO level so operators can see what the
// bridge is bridging (RX/TX/drop/retry).
let log_cancel = cancel.clone();
tokio::spawn(headless_packet_logger(log_rx, log_cancel));
cancel.cancelled().await; cancel.cancelled().await;
} else { } else {
let terminal = tui::run(config_watch_rx, swarm.clone(), stats, log_rx, cancel.clone(), start_time); let terminal = tui::run(config_watch_rx, swarm.clone(), stats, log_rx, cancel.clone(), start_time);
@@ -182,6 +196,91 @@ async fn main() -> Result<()> {
Ok(()) Ok(())
} }
/// Drain the router's packet-log channel in headless mode and emit
/// tracing events so `--log-only` output shows every bridged / dropped /
/// retried packet. Exits when `cancel` fires or the channel closes.
async fn headless_packet_logger(mut log_rx: mpsc::Receiver<router::PacketLogEntry>, cancel: CancellationToken) {
use router::{PacketAction, PacketDirection, TxRetryState};
loop {
tokio::select! {
() = cancel.cancelled() => return,
entry = log_rx.recv() => {
let Some(e) = entry else { return; };
let dir = match e.direction {
PacketDirection::RadioIn => "RF→NET",
PacketDirection::GossipIn => "NET→RF",
};
let hash_str = hex::encode(&e.hash[..4]);
let rssi_str = e.rssi.map_or_else(|| "-".to_string(), |r| r.to_string());
let snr_str = e.snr.map_or_else(|| "-".to_string(), |s| s.to_string());
// TX-retry lifecycle wins over the plain action label when set.
if let Some(r) = e.tx_retry {
match r.state {
TxRetryState::Retrying(reason) => {
tracing::info!(
target: "packet",
"{} retry hash={} attempt={}/{} reason={}",
dir, hash_str, r.attempt, r.total_attempts, reason,
);
}
TxRetryState::Succeeded if r.attempt > 1 => {
tracing::info!(
target: "packet",
"{} sent hash={} size={}B attempts={}/{}",
dir, hash_str, e.size, r.attempt, r.total_attempts,
);
}
TxRetryState::Failed => {
tracing::warn!(
target: "packet",
"{} FAILED hash={} size={}B attempts={}/{}",
dir, hash_str, e.size, r.attempt, r.total_attempts,
);
}
_ => {}
}
continue;
}
match e.action {
PacketAction::Bridged => {
tracing::info!(
target: "packet",
"{} bridged hash={} size={}B rssi={} snr={}",
dir, hash_str, e.size, rssi_str, snr_str,
);
}
PacketAction::DroppedDedup => {
// Log at INFO, not DEBUG — a run of dedup lines is
// proof of bidirectional gossip flow. Hiding it
// masks whether we're actually hearing peers.
tracing::info!(
target: "packet",
"{} dedup hash={} size={}B rssi={} snr={}",
dir, hash_str, e.size, rssi_str, snr_str,
);
}
PacketAction::DroppedRateLimit => {
tracing::warn!(
target: "packet",
"{} ratelim hash={} size={}B",
dir, hash_str, e.size,
);
}
PacketAction::DroppedQueueFull => {
tracing::warn!(
target: "packet",
"{} QFULL hash={} size={}B",
dir, hash_str, e.size,
);
}
}
}
}
}
}
fn setup_logging(log_only: bool, cfg: &config::Config) { fn setup_logging(log_only: bool, cfg: &config::Config) {
let env_filter = || { let env_filter = || {
tracing_subscriber::EnvFilter::try_from_default_env() tracing_subscriber::EnvFilter::try_from_default_env()
+295 -285
View File
@@ -1,13 +1,24 @@
//! `DongLoRa` hardware communication on a blocking thread. //! Async radio task — drives the `donglora-client` [`Dongle`] session.
//! //!
//! Handles connection, config negotiation, RX/TX, liveness pings, and //! Owns a single tokio task that connects (with exponential backoff reconnect),
//! exponential backoff reconnection. //! negotiates config, starts RX, and multiplexes TX requests from the router
//! against the RX stream. TX is dispatched via [`Dongle::tx_with_retry`] so
//! `TX_DONE(CHANNEL_BUSY)` / `EBUSY` errors are retried per spec §6.10 with
//! randomized backoff, and per-attempt state is surfaced to the router (and
//! onwards to the TUI packet log) via the [`RadioEvent::Tx*`] variants.
//!
//! Keepalive pings are handled inside the [`Dongle`] itself — no explicit
//! liveness loop here.
use std::time::Duration; use std::time::Duration;
use anyhow::{Context, Result}; use anyhow::{Context, Result};
use donglora_client::{RadioConfig, Response, Transport}; use donglora_client::{
ConnectOptions, Dongle, LoRaBandwidth, LoRaConfig, Modulation, RetryPolicy, RxPayload, connect_mux_auto_with,
connect_with,
};
use tokio::sync::mpsc; use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tracing::{debug, error, info, warn}; use tracing::{debug, error, info, warn};
use crate::packet::{RadioPacket, snr_grade}; use crate::packet::{RadioPacket, snr_grade};
@@ -21,242 +32,295 @@ pub enum ConfigSource {
Mux, Mux,
} }
/// Messages from the radio thread to the router. /// Messages from the radio task to the router.
#[derive(Debug)]
pub enum RadioEvent { pub enum RadioEvent {
/// A valid packet received from the radio (already SNR-filtered). /// A valid packet received from the radio (already SNR-filtered).
Packet(RadioPacket), Packet(RadioPacket),
/// Radio connected (or reconnected) with the active config and device info. /// Radio connected (or reconnected) with the active config and device info.
Connected(RadioConfig, ConfigSource, String), Connected(LoRaConfig, ConfigSource, String),
/// Radio disconnected — attempting reconnect. /// Radio disconnected — attempting reconnect.
Disconnected, Disconnected,
/// TX attempt started. `attempt` is 1-indexed.
TxAttempt { seq: u64, attempt: u8, total_attempts: u8, size: usize },
/// Previous TX attempt failed and a retry is scheduled.
TxRetry { seq: u64, attempt_that_failed: u8, total_attempts: u8, reason: TxRetryReason, backoff_ms: u32 },
/// TX completed successfully (perhaps after retries).
TxSucceeded { seq: u64, attempts_used: u8, final_airtime_us: u32, size: usize },
/// TX policy exhausted or errored terminally.
TxFailed { seq: u64, attempts_used: u8, reason: String, size: usize },
} }
/// Spawn the blocking radio thread. /// Why a TX attempt was retried — carried alongside the retry event for
/// the TUI.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TxRetryReason {
/// CAD detected channel activity before TX.
ChannelBusy,
/// Firmware TX queue was full.
QueueFull,
/// Other retryable condition (future-compat).
Other,
}
impl std::fmt::Display for TxRetryReason {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::ChannelBusy => write!(f, "channel_busy"),
Self::QueueFull => write!(f, "queue_full"),
Self::Other => write!(f, "other"),
}
}
}
/// One outbound TX request from the router: the payload plus a sequence id so
/// the event stream can correlate back to the packet log entry.
#[derive(Debug)]
pub struct TxRequest {
pub seq: u64,
pub data: Vec<u8>,
}
/// Spawn the async radio task.
/// ///
/// Returns a receiver for radio events and a sender for TX payloads. /// Returns a receiver for radio events and a sender for [`TxRequest`]s.
pub fn spawn(config: RadioConfig, port: Option<String>) -> Result<(mpsc::Receiver<RadioEvent>, mpsc::Sender<Vec<u8>>)> { pub fn spawn(
config: LoRaConfig,
port: Option<String>,
retry_policy: RetryPolicy,
) -> Result<(mpsc::Receiver<RadioEvent>, mpsc::Sender<TxRequest>, JoinHandle<()>)> {
let (event_tx, event_rx) = mpsc::channel::<RadioEvent>(256); let (event_tx, event_rx) = mpsc::channel::<RadioEvent>(256);
let (tx_send, tx_recv) = mpsc::channel::<Vec<u8>>(64); let (tx_send, tx_recv) = mpsc::channel::<TxRequest>(64);
let handle = tokio::spawn(radio_loop(config, port, retry_policy, event_tx, tx_recv));
std::thread::Builder::new() Ok((event_rx, tx_send, handle))
.name("radio".into())
.spawn(move || radio_loop(config, port, event_tx, tx_recv))
.context("failed to spawn radio thread")?;
Ok((event_rx, tx_send))
} }
#[allow(clippy::needless_pass_by_value)] // values moved across thread boundary in spawn() async fn radio_loop(
fn radio_loop( config: LoRaConfig,
config: RadioConfig,
port: Option<String>, port: Option<String>,
retry_policy: RetryPolicy,
event_tx: mpsc::Sender<RadioEvent>, event_tx: mpsc::Sender<RadioEvent>,
mut tx_recv: mpsc::Receiver<Vec<u8>>, mut tx_recv: mpsc::Receiver<TxRequest>,
) { ) {
let initial_backoff = Duration::from_millis(250); let initial_backoff = Duration::from_millis(250);
let max_backoff = Duration::from_secs(5); let max_backoff = Duration::from_secs(5);
let mut backoff = initial_backoff; let mut backoff = initial_backoff;
let mut stick_to_mux = false;
loop { loop {
let started = std::time::Instant::now(); let started = std::time::Instant::now();
match connect_and_run(&config, port.as_deref(), &event_tx, &mut tx_recv) { match connect_and_run(&config, port.as_deref(), stick_to_mux, &retry_policy, &event_tx, &mut tx_recv).await {
Ok(()) => { Ok(was_mux) => {
info!("radio thread exiting"); if event_tx.is_closed() {
return; info!("radio task exiting (router closed)");
return;
}
// We had a clean run_session exit but not shutdown — loop back.
if was_mux {
stick_to_mux = true;
}
debug!("radio session exited cleanly; reconnecting");
} }
Err(e) => { Err((was_mux, e)) => {
if was_mux {
stick_to_mux = true;
}
error!("radio error: {e:#}"); error!("radio error: {e:#}");
let _ = event_tx.blocking_send(RadioEvent::Disconnected); let _ = event_tx.send(RadioEvent::Disconnected).await;
// If we were connected for a while, reset backoff so
// the next reconnect attempt is fast.
if started.elapsed() > Duration::from_secs(5) { if started.elapsed() > Duration::from_secs(5) {
backoff = initial_backoff; backoff = initial_backoff;
} }
info!("reconnecting in {backoff:?}"); info!("reconnecting in {backoff:?}");
std::thread::sleep(backoff); tokio::time::sleep(backoff).await;
backoff = (backoff * 2).min(max_backoff); backoff = (backoff * 2).min(max_backoff);
} }
} }
} }
} }
fn connect_and_run( async fn connect_and_run(
config: &RadioConfig, config: &LoRaConfig,
port: Option<&str>, port: Option<&str>,
stick_to_mux: bool,
retry_policy: &RetryPolicy,
event_tx: &mpsc::Sender<RadioEvent>, event_tx: &mpsc::Sender<RadioEvent>,
tx_recv: &mut mpsc::Receiver<Vec<u8>>, tx_recv: &mut mpsc::Receiver<TxRequest>,
) -> Result<()> { ) -> std::result::Result<bool, (bool, anyhow::Error)> {
// Long initial timeout: the mux serializes commands through the dongle,
// so our `GetConfig` might have to wait behind other clients' commands.
let timeout = Duration::from_secs(10);
info!("connecting to dongle..."); info!("connecting to dongle...");
let (mut client, device) = if let Some(port) = port { let mut opts = ConnectOptions::new().config(Modulation::LoRa(*config));
let c = donglora_client::connect(Some(port), timeout)?; if let Some(p) = port {
(c, shorten_path(port)) opts = opts.port(p);
} else if let Ok(c) = donglora_client::connect_mux_auto(timeout) { }
(c, "mux".to_string())
// Attempt mux-only if we've committed to it, otherwise the normal
// auto-discovery chain.
let dongle = if stick_to_mux {
debug!("reconnecting via mux only (previously connected via mux)");
connect_mux_auto_with(opts).await.map_err(|e| (true, anyhow::Error::from(e)))?
} else { } else {
let port_path = donglora_client::find_port().ok_or_else(|| anyhow::anyhow!("no mux or USB dongle found"))?; connect_with(opts).await.map_err(|e| (false, anyhow::Error::from(e)))?
info!("no mux found, connecting to serial port {port_path}");
let c = donglora_client::connect(Some(&port_path), timeout)?;
(c, shorten_path(&port_path))
};
info!("[radio] transport connected: {device}");
info!("[radio] sending ping...");
match client.ping() {
Ok(()) => info!("[radio] ping OK"),
Err(e) => {
info!("[radio] ping FAILED: {e:#}");
return Err(e);
}
}
info!("[radio] negotiating config...");
let (active_config, config_source) = match negotiate_config(&mut client, config, &device) {
Ok(result) => {
info!("[radio] config negotiated: source={:?}, config={}", result.1, format_radio_config(&result.0));
result
}
Err(e) => {
info!("[radio] config negotiation FAILED: {e:#}");
return Err(e);
}
}; };
info!("[radio] sending start_rx..."); // Ask the Dongle which transport it actually ended up on. That's
match client.start_rx() { // authoritative — don't infer from firmware capability bits.
Ok(()) => info!("[radio] start_rx OK"), let kind = dongle.transport_kind().clone();
Err(e) => { let is_mux = kind.is_mux();
info!("[radio] start_rx FAILED: {e:#}"); let device = kind.short_label();
return Err(e); info!("[radio] connected: {device}");
let outcome = run_session(dongle, config, is_mux, &device, retry_policy, event_tx, tx_recv).await;
outcome.map(|()| is_mux).map_err(|e| (is_mux, e))
}
async fn run_session(
dongle: Dongle,
config: &LoRaConfig,
is_mux: bool,
device: &str,
retry_policy: &RetryPolicy,
event_tx: &mpsc::Sender<RadioEvent>,
tx_recv: &mut mpsc::Receiver<TxRequest>,
) -> Result<()> {
// The Dongle's auto-configure already applied our requested config at
// connect time (via `ConnectOptions::config`). Read back the active
// config — on the mux, another client may have locked a different one.
let active_config = match dongle.config().await {
Some(Modulation::LoRa(c)) => c,
Some(other) => {
warn!("[radio] mux has a non-LoRa config locked ({other:?}); using it anyway");
// This branch is unreachable today since the bridge only supports
// LoRa, but we surface a warning rather than hard-fail so the
// mux doesn't brick the bridge.
*config
} }
None => {
warn!("[radio] Dongle::config() returned None after connect; falling back to requested");
*config
}
};
let source = if lora_configs_match(&active_config, config) { ConfigSource::Ours } else { ConfigSource::Mux };
info!("[radio] config: source={source:?}, {}", format_lora_config(&active_config));
dongle.rx_start().await.context("rx_start failed")?;
if event_tx.send(RadioEvent::Connected(active_config, source, device.to_string())).await.is_err() {
return Ok(());
} }
let _ = event_tx.blocking_send(RadioEvent::Connected(active_config, config_source, device));
// Set a short timeout so we can interleave RX and TX.
client.transport_mut().set_timeout(Duration::from_millis(100))?;
info!("[radio] entering main loop"); info!("[radio] entering main loop");
// Liveness: ping the mux periodically when idle to detect a dead connection. // TX requests from the router and RX packets from the radio both drive
// Without this, a killed mux causes recv() to silently return Ok(None) forever. // the loop; use tokio::select! to multiplex them.
let liveness_interval = Duration::from_secs(2);
let mut last_activity = std::time::Instant::now();
loop { loop {
// Check for TX requests (non-blocking). tokio::select! {
while let Ok(payload) = tx_recv.try_recv() { tx = tx_recv.recv() => {
debug!("TX {} bytes", payload.len()); let Some(req) = tx else { return Ok(()); };
if let Err(e) = client.transmit(&payload, None) { handle_tx(&dongle, req, retry_policy, event_tx).await;
warn!("transmit error: {e:#}");
return Err(e);
} }
last_activity = std::time::Instant::now(); rx = dongle.next_rx() => {
} match rx {
Some(pkt) => {
// Check for RX packets. handle_rx(pkt, &active_config, event_tx).await;
match client.recv() { }
Ok(Some(Response::RxPacket { rssi, snr, payload })) => { None => {
last_activity = std::time::Instant::now(); // Dongle's RX stream ended — session closed.
let grade = snr_grade(snr, active_config.sf);
if grade.should_forward() {
debug!("RX {} bytes rssi={rssi} snr={snr} grade={grade}", payload.len());
let pkt = RadioPacket { rssi, snr, payload };
if event_tx.blocking_send(RadioEvent::Packet(pkt)).is_err() {
return Ok(()); return Ok(());
} }
} else {
debug!("RX drop {} bytes rssi={rssi} snr={snr} grade={grade}", payload.len());
} }
} }
Ok(Some(_)) => {
last_activity = std::time::Instant::now();
}
Ok(None) => {}
Err(e) if is_timeout_error(&e) => {}
Err(e) => return Err(e),
}
// Liveness check: if no real activity for a while, ping the mux.
if last_activity.elapsed() >= liveness_interval {
// Temporarily extend the timeout for the ping command.
let _ = client.transport_mut().set_timeout(Duration::from_secs(2));
match client.ping() {
Ok(()) => {
last_activity = std::time::Instant::now();
}
Err(e) => {
warn!("liveness ping failed: {e:#}");
return Err(e);
}
}
// Restore the short timeout.
let _ = client.transport_mut().set_timeout(Duration::from_millis(100));
} }
if event_tx.is_closed() { if event_tx.is_closed() {
return Ok(()); return Ok(());
} }
// Pull any async errors the Dongle accumulated (EFRAME / async
// ERADIO). These get logged but don't kill the session.
for err in dongle.drain_async_errors().await {
warn!("[radio] async error: {err}");
}
// Intentionally unused so callers don't think is_mux matters here.
let _ = is_mux;
} }
} }
/// Negotiate radio config with the dongle/mux. async fn handle_tx(dongle: &Dongle, req: TxRequest, retry_policy: &RetryPolicy, event_tx: &mpsc::Sender<RadioEvent>) {
/// let size = req.data.len();
/// Strategy: let _ = event_tx
/// 1. `GetConfig` first to read what's active. .send(RadioEvent::TxAttempt { seq: req.seq, attempt: 1, total_attempts: retry_policy.max_attempts, size })
/// 2. If it matches our desired config, use it — no `SetConfig` needed. .await;
/// 3. If it differs and we're on the mux, accept the mux's config.
/// 4. If it differs and we're on direct serial, `SetConfig` to push ours. let result = dongle.tx_with_retry(&req.data, retry_policy).await;
/// 5. If `GetConfig` fails, try `SetConfig` with our desired config. match result {
fn negotiate_config<T: Transport>( Ok(outcome) => {
client: &mut donglora_client::Client<T>, // Fire a TxRetry event for every attempt beyond the first, so the
desired: &RadioConfig, // TUI log can show the escalation.
device: &str, let mut prev_attempt: u8 = 0;
) -> Result<(RadioConfig, ConfigSource)> { for att in &outcome.attempts {
info!("[negotiate] get_config..."); if prev_attempt != 0 {
match client.get_config() { let reason = match &outcome.attempts[(prev_attempt - 1) as usize].result {
Ok(cfg) => { Err(donglora_client::ClientError::ChannelBusy) => TxRetryReason::ChannelBusy,
if configs_match(&cfg, desired) { Err(donglora_client::ClientError::Busy) => TxRetryReason::QueueFull,
info!("[negotiate] get_config OK (matches desired): {}", format_radio_config(&cfg)); _ => TxRetryReason::Other,
Ok((cfg, ConfigSource::Ours)) };
} else if device == "mux" { let _ = event_tx
info!("[negotiate] get_config OK (mux config differs, accepting): {}", format_radio_config(&cfg)); .send(RadioEvent::TxRetry {
Ok((cfg, ConfigSource::Mux)) seq: req.seq,
} else { attempt_that_failed: prev_attempt,
info!("[negotiate] get_config OK (serial config differs, setting ours)"); total_attempts: retry_policy.max_attempts,
client.set_config(*desired)?; reason,
info!("[negotiate] set_config OK"); backoff_ms: retry_policy.backoff_ms_min,
Ok((*desired, ConfigSource::Ours)) })
.await;
}
prev_attempt = att.attempt;
} }
let _ = event_tx
.send(RadioEvent::TxSucceeded {
seq: req.seq,
attempts_used: outcome.attempts_used(),
final_airtime_us: outcome.final_airtime_us,
size,
})
.await;
} }
Err(e) => { Err(e) => {
info!("[negotiate] get_config FAILED: {e:#}"); warn!("TX #{} failed after retries: {e:#}", req.seq);
// No config on radio — try to set ours. let _ = event_tx
info!("[negotiate] trying set_config..."); .send(RadioEvent::TxFailed {
match client.set_config(*desired) { seq: req.seq,
Ok(()) => { attempts_used: retry_policy.max_attempts,
info!("[negotiate] set_config OK"); reason: format!("{e}"),
Ok((*desired, ConfigSource::Ours)) size,
} })
Err(e2) => { .await;
info!("[negotiate] set_config FAILED: {e2:#}");
anyhow::bail!("GetConfig failed ({e:#}) and SetConfig also failed ({e2:#})");
}
}
} }
} }
} }
/// Shorten a device path to just the filename for display. async fn handle_rx(pkt: RxPayload, active_config: &LoRaConfig, event_tx: &mpsc::Sender<RadioEvent>) {
fn shorten_path(path: &str) -> String { // Tenths-of-dB → whole dB for the existing bridge packet format.
std::path::Path::new(path).file_name().map_or_else(|| path.to_string(), |f| f.to_string_lossy().into_owned()) // `rssi_tenths_dbm` and `snr_tenths_db` are i16 and the values stay
// within i16 after the /10 rounding; the `as i16` is safe.
#[allow(clippy::cast_possible_truncation)]
let rssi = (f32::from(pkt.rssi_tenths_dbm) / 10.0).round() as i16;
#[allow(clippy::cast_possible_truncation)]
let snr = (f32::from(pkt.snr_tenths_db) / 10.0).round() as i16;
let grade = snr_grade(snr, active_config.sf);
if !grade.should_forward() {
debug!("RX drop {} bytes rssi={rssi} snr={snr} grade={grade}", pkt.data.len());
return;
}
debug!("RX {} bytes rssi={rssi} snr={snr} grade={grade}", pkt.data.len());
let radio_pkt = RadioPacket { rssi, snr, payload: pkt.data.to_vec() };
let _ = event_tx.send(RadioEvent::Packet(radio_pkt)).await;
} }
fn configs_match(a: &RadioConfig, b: &RadioConfig) -> bool { // ── Display helpers ───────────────────────────────────────────────
fn lora_configs_match(a: &LoRaConfig, b: &LoRaConfig) -> bool {
a.freq_hz == b.freq_hz a.freq_hz == b.freq_hz
&& a.bw == b.bw && a.bw == b.bw
&& a.sf == b.sf && a.sf == b.sf
@@ -264,46 +328,47 @@ fn configs_match(a: &RadioConfig, b: &RadioConfig) -> bool {
&& a.sync_word == b.sync_word && a.sync_word == b.sync_word
&& a.tx_power_dbm == b.tx_power_dbm && a.tx_power_dbm == b.tx_power_dbm
&& a.preamble_len == b.preamble_len && a.preamble_len == b.preamble_len
&& a.cad == b.cad && a.header_mode == b.header_mode
} && a.payload_crc == b.payload_crc
&& a.iq_invert == b.iq_invert
fn is_timeout_error(e: &anyhow::Error) -> bool {
e.downcast_ref::<std::io::Error>()
.is_some_and(|io_err| matches!(io_err.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut))
} }
/// Format a bandwidth value for display. /// Format a bandwidth value for display.
#[must_use] #[must_use]
pub const fn format_bandwidth(bw: donglora_client::Bandwidth) -> &'static str { pub const fn format_bandwidth(bw: LoRaBandwidth) -> &'static str {
match bw { match bw {
donglora_client::Bandwidth::Khz7 => "7.8 kHz", LoRaBandwidth::Khz7 => "7.8 kHz",
donglora_client::Bandwidth::Khz10 => "10.4 kHz", LoRaBandwidth::Khz10 => "10.4 kHz",
donglora_client::Bandwidth::Khz15 => "15.6 kHz", LoRaBandwidth::Khz15 => "15.6 kHz",
donglora_client::Bandwidth::Khz20 => "20.8 kHz", LoRaBandwidth::Khz20 => "20.8 kHz",
donglora_client::Bandwidth::Khz31 => "31.25 kHz", LoRaBandwidth::Khz31 => "31.25 kHz",
donglora_client::Bandwidth::Khz41 => "41.7 kHz", LoRaBandwidth::Khz41 => "41.7 kHz",
donglora_client::Bandwidth::Khz62 => "62.5 kHz", LoRaBandwidth::Khz62 => "62.5 kHz",
donglora_client::Bandwidth::Khz125 => "125 kHz", LoRaBandwidth::Khz125 => "125 kHz",
donglora_client::Bandwidth::Khz250 => "250 kHz", LoRaBandwidth::Khz250 => "250 kHz",
donglora_client::Bandwidth::Khz500 => "500 kHz", LoRaBandwidth::Khz500 => "500 kHz",
LoRaBandwidth::Khz200 => "200 kHz",
LoRaBandwidth::Khz400 => "400 kHz",
LoRaBandwidth::Khz800 => "800 kHz",
LoRaBandwidth::Khz1600 => "1600 kHz",
} }
} }
/// Format a `RadioConfig` for display. /// Format a `LoRaConfig` for display.
#[must_use] #[must_use]
pub fn format_radio_config(config: &RadioConfig) -> String { pub fn format_lora_config(config: &LoRaConfig) -> String {
let freq_mhz = f64::from(config.freq_hz) / 1_000_000.0; let freq_mhz = f64::from(config.freq_hz) / 1_000_000.0;
let bw = format_bandwidth(config.bw); let bw = format_bandwidth(config.bw);
let power = if config.tx_power_dbm == donglora_client::TX_POWER_MAX { let cr_denom: u8 = match config.cr {
"max".to_string() donglora_client::LoRaCodingRate::Cr4_5 => 5,
} else { donglora_client::LoRaCodingRate::Cr4_6 => 6,
format!("{} dBm", config.tx_power_dbm) donglora_client::LoRaCodingRate::Cr4_7 => 7,
donglora_client::LoRaCodingRate::Cr4_8 => 8,
}; };
let preamble = if config.preamble_len == 0 { 16 } else { config.preamble_len }; let preamble = config.preamble_len;
let cad = if config.cad != 0 { "on" } else { "off" };
format!( format!(
"{freq_mhz:.3}MHz SF{} BW{bw} CR4/{} SW0x{:04X} TX:{power} Pre:{preamble} CAD:{cad}", "{freq_mhz:.3}MHz SF{} BW{bw} CR4/{cr_denom} SW0x{:04X} TX:{} dBm Pre:{preamble}",
config.sf, config.cr, config.sync_word config.sf, config.sync_word, config.tx_power_dbm,
) )
} }
@@ -311,134 +376,79 @@ pub fn format_radio_config(config: &RadioConfig) -> String {
#[allow(clippy::unwrap_used, clippy::expect_used)] #[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests { mod tests {
use super::*; use super::*;
use donglora_client::{LoRaCodingRate, LoRaHeaderMode};
fn default_config() -> RadioConfig { fn default_config() -> LoRaConfig {
RadioConfig { LoRaConfig {
freq_hz: 910_525_000, freq_hz: 910_525_000,
bw: donglora_client::Bandwidth::Khz62, bw: LoRaBandwidth::Khz62,
sf: 7, sf: 7,
cr: 5, cr: LoRaCodingRate::Cr4_5,
sync_word: 0x3444, sync_word: 0x1424,
tx_power_dbm: 22, tx_power_dbm: 20,
preamble_len: 16, preamble_len: 16,
cad: 1, header_mode: LoRaHeaderMode::Explicit,
payload_crc: true,
iq_invert: false,
} }
} }
#[test] #[test]
fn configs_match_identical() { fn lora_configs_match_identical() {
let a = default_config(); assert!(lora_configs_match(&default_config(), &default_config()));
let b = default_config();
assert!(configs_match(&a, &b));
} }
#[test] #[test]
fn configs_match_differs_each_field() { fn lora_configs_match_differs_each_field() {
let base = default_config(); let base = default_config();
let mut c = base; let mut c = base;
c.freq_hz = 915_000_000; c.freq_hz = 915_000_000;
assert!(!configs_match(&base, &c)); assert!(!lora_configs_match(&base, &c));
let mut c = base; let mut c = base;
c.sf = 12; c.sf = 12;
assert!(!configs_match(&base, &c)); assert!(!lora_configs_match(&base, &c));
let mut c = base; let mut c = base;
c.cr = 8; c.cr = LoRaCodingRate::Cr4_8;
assert!(!configs_match(&base, &c)); assert!(!lora_configs_match(&base, &c));
let mut c = base; let mut c = base;
c.sync_word = 0x1234; c.sync_word = 0x1234;
assert!(!configs_match(&base, &c)); assert!(!lora_configs_match(&base, &c));
let mut c = base; let mut c = base;
c.tx_power_dbm = 10; c.tx_power_dbm = 10;
assert!(!configs_match(&base, &c)); assert!(!lora_configs_match(&base, &c));
let mut c = base; let mut c = base;
c.preamble_len = 8; c.preamble_len = 8;
assert!(!configs_match(&base, &c)); assert!(!lora_configs_match(&base, &c));
let mut c = base;
c.cad = 0;
assert!(!configs_match(&base, &c));
}
#[test]
fn is_timeout_would_block() {
let e = anyhow::Error::from(std::io::Error::new(std::io::ErrorKind::WouldBlock, "test"));
assert!(is_timeout_error(&e));
}
#[test]
fn is_timeout_timed_out() {
let e = anyhow::Error::from(std::io::Error::new(std::io::ErrorKind::TimedOut, "test"));
assert!(is_timeout_error(&e));
}
#[test]
fn is_timeout_false_for_other() {
let e = anyhow::Error::from(std::io::Error::new(std::io::ErrorKind::ConnectionReset, "test"));
assert!(!is_timeout_error(&e));
}
#[test]
fn shorten_path_extracts_filename() {
assert_eq!(shorten_path("/dev/ttyACM0"), "ttyACM0");
assert_eq!(shorten_path("ttyACM0"), "ttyACM0");
} }
#[test] #[test]
fn format_bandwidth_all_variants() { fn format_bandwidth_all_variants() {
assert_eq!(format_bandwidth(donglora_client::Bandwidth::Khz7), "7.8 kHz"); assert_eq!(format_bandwidth(LoRaBandwidth::Khz7), "7.8 kHz");
assert_eq!(format_bandwidth(donglora_client::Bandwidth::Khz125), "125 kHz"); assert_eq!(format_bandwidth(LoRaBandwidth::Khz125), "125 kHz");
assert_eq!(format_bandwidth(donglora_client::Bandwidth::Khz500), "500 kHz"); assert_eq!(format_bandwidth(LoRaBandwidth::Khz500), "500 kHz");
} }
#[test] #[test]
fn format_radio_config_typical() { fn format_lora_config_typical() {
let cfg = default_config(); let cfg = default_config();
let s = format_radio_config(&cfg); let s = format_lora_config(&cfg);
assert!(s.contains("910.525MHz")); assert!(s.contains("910.525MHz"));
assert!(s.contains("SF7")); assert!(s.contains("SF7"));
assert!(s.contains("CR4/5")); assert!(s.contains("CR4/5"));
assert!(s.contains("0x3444")); assert!(s.contains("0x1424"));
assert!(s.contains("CAD:on"), "cad=1 should show CAD:on: {s}"); assert!(s.contains("20 dBm"));
} }
#[test] #[test]
fn format_radio_config_cad_off() { fn tx_retry_reason_display() {
let mut cfg = default_config(); assert_eq!(format!("{}", TxRetryReason::ChannelBusy), "channel_busy");
cfg.cad = 0; assert_eq!(format!("{}", TxRetryReason::QueueFull), "queue_full");
let s = format_radio_config(&cfg); assert_eq!(format!("{}", TxRetryReason::Other), "other");
assert!(s.contains("CAD:off"), "cad=0 should show CAD:off: {s}");
}
#[test]
fn format_radio_config_max_power() {
let mut cfg = default_config();
cfg.tx_power_dbm = donglora_client::TX_POWER_MAX;
let s = format_radio_config(&cfg);
assert!(s.contains("TX:max"));
assert!(!s.contains("dBm"), "max power should not contain 'dBm': {s}");
}
#[test]
fn format_radio_config_non_max_power() {
let mut cfg = default_config();
cfg.tx_power_dbm = 22;
assert_ne!(cfg.tx_power_dbm, donglora_client::TX_POWER_MAX);
let s = format_radio_config(&cfg);
assert!(s.contains("22 dBm"), "non-max power should contain '22 dBm': {s}");
assert!(!s.contains("max"), "non-max power should not contain 'max': {s}");
}
#[test]
fn format_radio_config_zero_preamble() {
let mut cfg = default_config();
cfg.preamble_len = 0;
let s = format_radio_config(&cfg);
assert!(s.contains("Pre:16"));
} }
} }
+50 -43
View File
@@ -5,7 +5,7 @@
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
use donglora_client::Bandwidth; use donglora_client::LoRaBandwidth;
/// Token-bucket rate limiter for radio transmissions. /// Token-bucket rate limiter for radio transmissions.
/// ///
@@ -35,7 +35,7 @@ impl RateLimiter {
/// ///
/// If `override_pps` is `Some`, use that rate directly instead of calculating. /// If `override_pps` is `Some`, use that rate directly instead of calculating.
#[must_use] #[must_use]
pub fn from_radio_config(sf: u8, bw: Bandwidth, cr: u8, preamble: u16, override_pps: Option<f64>) -> Self { pub fn from_radio_config(sf: u8, bw: LoRaBandwidth, cr: u8, preamble: u16, override_pps: Option<f64>) -> Self {
let refill_rate = override_pps.map_or_else( let refill_rate = override_pps.map_or_else(
|| { || {
let air_time = lora_air_time(REFERENCE_PAYLOAD_BYTES, sf, bw, cr, preamble); let air_time = lora_air_time(REFERENCE_PAYLOAD_BYTES, sf, bw, cr, preamble);
@@ -81,7 +81,7 @@ impl RateLimiter {
/// - `cr`: coding rate denominator (5–8, meaning 4/5 to 4/8) /// - `cr`: coding rate denominator (5–8, meaning 4/5 to 4/8)
/// - `preamble`: preamble length in symbols (0 = firmware default of 16) /// - `preamble`: preamble length in symbols (0 = firmware default of 16)
#[must_use] #[must_use]
pub fn lora_air_time(payload_bytes: usize, sf: u8, bw: Bandwidth, cr: u8, preamble: u16) -> Duration { pub fn lora_air_time(payload_bytes: usize, sf: u8, bw: LoRaBandwidth, cr: u8, preamble: u16) -> Duration {
let sf_f = f64::from(sf); let sf_f = f64::from(sf);
let bw_hz = bandwidth_hz(bw); let bw_hz = bandwidth_hz(bw);
let preamble_symbols = if preamble == 0 { 16.0 } else { f64::from(preamble) }; let preamble_symbols = if preamble == 0 { 16.0 } else { f64::from(preamble) };
@@ -108,18 +108,25 @@ pub fn lora_air_time(payload_bytes: usize, sf: u8, bw: Bandwidth, cr: u8, preamb
Duration::from_secs_f64(total) Duration::from_secs_f64(total)
} }
const fn bandwidth_hz(bw: Bandwidth) -> f64 { const fn bandwidth_hz(bw: LoRaBandwidth) -> f64 {
match bw { match bw {
Bandwidth::Khz7 => 7_800.0, LoRaBandwidth::Khz7 => 7_800.0,
Bandwidth::Khz10 => 10_400.0, LoRaBandwidth::Khz10 => 10_400.0,
Bandwidth::Khz15 => 15_600.0, LoRaBandwidth::Khz15 => 15_600.0,
Bandwidth::Khz20 => 20_800.0, LoRaBandwidth::Khz20 => 20_800.0,
Bandwidth::Khz31 => 31_250.0, LoRaBandwidth::Khz31 => 31_250.0,
Bandwidth::Khz41 => 41_700.0, LoRaBandwidth::Khz41 => 41_700.0,
Bandwidth::Khz62 => 62_500.0, LoRaBandwidth::Khz62 => 62_500.0,
Bandwidth::Khz125 => 125_000.0, LoRaBandwidth::Khz125 => 125_000.0,
Bandwidth::Khz250 => 250_000.0, LoRaBandwidth::Khz250 => 250_000.0,
Bandwidth::Khz500 => 500_000.0, LoRaBandwidth::Khz500 => 500_000.0,
// SX128x 2.4 GHz bandwidths — not used on the bridge today, but
// plumb through so SET_CONFIG with one of these still yields a
// usable rate rather than panicking.
LoRaBandwidth::Khz200 => 200_000.0,
LoRaBandwidth::Khz400 => 400_000.0,
LoRaBandwidth::Khz800 => 800_000.0,
LoRaBandwidth::Khz1600 => 1_600_000.0,
} }
} }
@@ -129,7 +136,7 @@ mod tests {
#[test] #[test]
fn air_time_sf7_bw125() { fn air_time_sf7_bw125() {
let t = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16); let t = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 16);
// SF7/125kHz/CR4/5/16-preamble/50 bytes: ~100ms range // SF7/125kHz/CR4/5/16-preamble/50 bytes: ~100ms range
assert!(t.as_millis() > 50, "air time too short: {t:?}"); assert!(t.as_millis() > 50, "air time too short: {t:?}");
assert!(t.as_millis() < 300, "air time too long: {t:?}"); assert!(t.as_millis() < 300, "air time too long: {t:?}");
@@ -137,7 +144,7 @@ mod tests {
#[test] #[test]
fn air_time_sf12_bw125() { fn air_time_sf12_bw125() {
let t = lora_air_time(50, 12, Bandwidth::Khz125, 5, 16); let t = lora_air_time(50, 12, LoRaBandwidth::Khz125, 5, 16);
// SF12/125kHz: should be multiple seconds // SF12/125kHz: should be multiple seconds
assert!(t.as_secs() >= 1, "air time too short: {t:?}"); assert!(t.as_secs() >= 1, "air time too short: {t:?}");
assert!(t.as_secs() < 10, "air time too long: {t:?}"); assert!(t.as_secs() < 10, "air time too long: {t:?}");
@@ -145,7 +152,7 @@ mod tests {
#[test] #[test]
fn rate_limiter_allows_burst() { fn rate_limiter_allows_burst() {
let mut rl = RateLimiter::from_radio_config(7, Bandwidth::Khz125, 5, 16, None); let mut rl = RateLimiter::from_radio_config(7, LoRaBandwidth::Khz125, 5, 16, None);
// Should allow burst of 3 // Should allow burst of 3
assert!(rl.try_acquire()); assert!(rl.try_acquire());
assert!(rl.try_acquire()); assert!(rl.try_acquire());
@@ -156,14 +163,14 @@ mod tests {
#[test] #[test]
fn rate_limiter_override() { fn rate_limiter_override() {
let rl = RateLimiter::from_radio_config(7, Bandwidth::Khz125, 5, 16, Some(42.0)); let rl = RateLimiter::from_radio_config(7, LoRaBandwidth::Khz125, 5, 16, Some(42.0));
assert!((rl.rate_pps() - 42.0).abs() < f64::EPSILON); assert!((rl.rate_pps() - 42.0).abs() < f64::EPSILON);
} }
#[test] #[test]
fn air_time_zero_preamble_uses_default() { fn air_time_zero_preamble_uses_default() {
let t_default = lora_air_time(50, 7, Bandwidth::Khz125, 5, 0); let t_default = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 0);
let t_explicit = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16); let t_explicit = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 16);
assert_eq!(t_default, t_explicit); assert_eq!(t_default, t_explicit);
} }
@@ -173,7 +180,7 @@ mod tests {
fn from_radio_config_calculates_reasonable_rate() { fn from_radio_config_calculates_reasonable_rate() {
// SF7/BW125/CR5/preamble16 with 50-byte reference payload: // SF7/BW125/CR5/preamble16 with 50-byte reference payload:
// air time ~105.7ms, so rate = 0.5 / 0.1057 ~= 4.73 pps. // air time ~105.7ms, so rate = 0.5 / 0.1057 ~= 4.73 pps.
let rl = RateLimiter::from_radio_config(7, Bandwidth::Khz125, 5, 16, None); let rl = RateLimiter::from_radio_config(7, LoRaBandwidth::Khz125, 5, 16, None);
let rate = rl.rate_pps(); let rate = rl.rate_pps();
// Must be between 0.1 and 100 (rules out fallback of 10.0 indirectly, // Must be between 0.1 and 100 (rules out fallback of 10.0 indirectly,
// and catches if secs<=0 path is taken or division is wrong). // and catches if secs<=0 path is taken or division is wrong).
@@ -189,7 +196,7 @@ mod tests {
#[test] #[test]
fn from_radio_config_sf12_rate_much_slower() { fn from_radio_config_sf12_rate_much_slower() {
// SF12/BW125: air time ~2564ms, rate = 0.5 / 2.564 ~= 0.195 pps. // SF12/BW125: air time ~2564ms, rate = 0.5 / 2.564 ~= 0.195 pps.
let rl = RateLimiter::from_radio_config(12, Bandwidth::Khz125, 5, 16, None); let rl = RateLimiter::from_radio_config(12, LoRaBandwidth::Khz125, 5, 16, None);
let rate = rl.rate_pps(); let rate = rl.rate_pps();
assert!(rate > 0.1, "rate too low: {rate}"); assert!(rate > 0.1, "rate too low: {rate}");
assert!(rate < 0.5, "rate too high: {rate}"); assert!(rate < 0.5, "rate too high: {rate}");
@@ -202,7 +209,7 @@ mod tests {
#[test] #[test]
fn rate_limiter_refills_after_waiting() { fn rate_limiter_refills_after_waiting() {
// Use override_pps for predictable timing: 10 pps = 1 token per 100ms. // Use override_pps for predictable timing: 10 pps = 1 token per 100ms.
let mut rl = RateLimiter::from_radio_config(7, Bandwidth::Khz125, 5, 16, Some(10.0)); let mut rl = RateLimiter::from_radio_config(7, LoRaBandwidth::Khz125, 5, 16, Some(10.0));
// Exhaust all burst tokens. // Exhaust all burst tokens.
assert!(rl.try_acquire()); assert!(rl.try_acquire());
assert!(rl.try_acquire()); assert!(rl.try_acquire());
@@ -222,7 +229,7 @@ mod tests {
fn air_time_exact_sf7_bw125() { fn air_time_exact_sf7_bw125() {
// SF7, BW125kHz, CR4/5, preamble=16, payload=50 bytes. // SF7, BW125kHz, CR4/5, preamble=16, payload=50 bytes.
// Expected: 105.728 ms (computed from Semtech formula). // Expected: 105.728 ms (computed from Semtech formula).
let t = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16); let t = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 16);
let ms = t.as_secs_f64() * 1000.0; let ms = t.as_secs_f64() * 1000.0;
assert!((ms - 105.728).abs() < 0.01, "SF7/BW125 air time should be ~105.728ms, got {ms:.3}ms"); assert!((ms - 105.728).abs() < 0.01, "SF7/BW125 air time should be ~105.728ms, got {ms:.3}ms");
} }
@@ -231,7 +238,7 @@ mod tests {
fn air_time_exact_sf12_bw125() { fn air_time_exact_sf12_bw125() {
// SF12, BW125kHz, CR4/5, preamble=16, payload=50 bytes. // SF12, BW125kHz, CR4/5, preamble=16, payload=50 bytes.
// Expected: 2564.096 ms. // Expected: 2564.096 ms.
let t = lora_air_time(50, 12, Bandwidth::Khz125, 5, 16); let t = lora_air_time(50, 12, LoRaBandwidth::Khz125, 5, 16);
let ms = t.as_secs_f64() * 1000.0; let ms = t.as_secs_f64() * 1000.0;
assert!((ms - 2564.096).abs() < 0.01, "SF12/BW125 air time should be ~2564.096ms, got {ms:.3}ms"); assert!((ms - 2564.096).abs() < 0.01, "SF12/BW125 air time should be ~2564.096ms, got {ms:.3}ms");
} }
@@ -240,8 +247,8 @@ mod tests {
fn air_time_ldro_sf11_bw125_vs_sf10_bw125() { fn air_time_ldro_sf11_bw125_vs_sf10_bw125() {
// SF>=11 && BW<=125kHz triggers LDRO. SF10 does not. // SF>=11 && BW<=125kHz triggers LDRO. SF10 does not.
// SF11/BW125: 1445.888 ms (de=1), SF10/BW125: 681.984 ms (de=0). // SF11/BW125: 1445.888 ms (de=1), SF10/BW125: 681.984 ms (de=0).
let t11 = lora_air_time(50, 11, Bandwidth::Khz125, 5, 16); let t11 = lora_air_time(50, 11, LoRaBandwidth::Khz125, 5, 16);
let t10 = lora_air_time(50, 10, Bandwidth::Khz125, 5, 16); let t10 = lora_air_time(50, 10, LoRaBandwidth::Khz125, 5, 16);
let ms11 = t11.as_secs_f64() * 1000.0; let ms11 = t11.as_secs_f64() * 1000.0;
let ms10 = t10.as_secs_f64() * 1000.0; let ms10 = t10.as_secs_f64() * 1000.0;
@@ -259,12 +266,12 @@ mod tests {
fn air_time_ldro_not_triggered_high_bw() { fn air_time_ldro_not_triggered_high_bw() {
// SF11/BW250: LDRO should NOT be active (BW > 125kHz). // SF11/BW250: LDRO should NOT be active (BW > 125kHz).
// Expected: 641.024 ms (de=0). // Expected: 641.024 ms (de=0).
let t11_250 = lora_air_time(50, 11, Bandwidth::Khz250, 5, 16); let t11_250 = lora_air_time(50, 11, LoRaBandwidth::Khz250, 5, 16);
let ms = t11_250.as_secs_f64() * 1000.0; let ms = t11_250.as_secs_f64() * 1000.0;
assert!((ms - 641.024).abs() < 0.01, "SF11/BW250 should be ~641.024ms (no LDRO), got {ms:.3}ms"); assert!((ms - 641.024).abs() < 0.01, "SF11/BW250 should be ~641.024ms (no LDRO), got {ms:.3}ms");
// Compare with SF11/BW125 which HAS LDRO. // Compare with SF11/BW125 which HAS LDRO.
let t11_125 = lora_air_time(50, 11, Bandwidth::Khz125, 5, 16); let t11_125 = lora_air_time(50, 11, LoRaBandwidth::Khz125, 5, 16);
// Even accounting for halved BW doubling the base time, the LDRO effect // Even accounting for halved BW doubling the base time, the LDRO effect
// on SF11/BW125 makes it much longer than simply 2x SF11/BW250. // on SF11/BW125 makes it much longer than simply 2x SF11/BW250.
assert!(t11_125 > t11_250 * 2, "SF11/BW125 (LDRO on) should be >2x SF11/BW250 (LDRO off)"); assert!(t11_125 > t11_250 * 2, "SF11/BW125 (LDRO on) should be >2x SF11/BW250 (LDRO off)");
@@ -274,9 +281,9 @@ mod tests {
fn air_time_preamble_scaling() { fn air_time_preamble_scaling() {
// Changing preamble by N symbols changes air time by N * t_sym. // Changing preamble by N symbols changes air time by N * t_sym.
// t_sym for SF7/BW125 = 2^7 / 125000 = 1.024 ms. // t_sym for SF7/BW125 = 2^7 / 125000 = 1.024 ms.
let t8 = lora_air_time(50, 7, Bandwidth::Khz125, 5, 8); let t8 = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 8);
let t16 = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16); let t16 = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 16);
let t32 = lora_air_time(50, 7, Bandwidth::Khz125, 5, 32); let t32 = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 32);
let diff_8_to_16 = t16.as_secs_f64() - t8.as_secs_f64(); let diff_8_to_16 = t16.as_secs_f64() - t8.as_secs_f64();
let diff_16_to_32 = t32.as_secs_f64() - t16.as_secs_f64(); let diff_16_to_32 = t32.as_secs_f64() - t16.as_secs_f64();
@@ -297,16 +304,16 @@ mod tests {
// Verify reasonable bounds for all SF/BW combos with 50-byte payload. // Verify reasonable bounds for all SF/BW combos with 50-byte payload.
// This catches gross formula errors (e.g. division replaced with modulo). // This catches gross formula errors (e.g. division replaced with modulo).
let bandwidths = [ let bandwidths = [
Bandwidth::Khz7, LoRaBandwidth::Khz7,
Bandwidth::Khz10, LoRaBandwidth::Khz10,
Bandwidth::Khz15, LoRaBandwidth::Khz15,
Bandwidth::Khz20, LoRaBandwidth::Khz20,
Bandwidth::Khz31, LoRaBandwidth::Khz31,
Bandwidth::Khz41, LoRaBandwidth::Khz41,
Bandwidth::Khz62, LoRaBandwidth::Khz62,
Bandwidth::Khz125, LoRaBandwidth::Khz125,
Bandwidth::Khz250, LoRaBandwidth::Khz250,
Bandwidth::Khz500, LoRaBandwidth::Khz500,
]; ];
for sf in 7..=12u8 { for sf in 7..=12u8 {
for &bw in &bandwidths { for &bw in &bandwidths {
@@ -332,7 +339,7 @@ mod tests {
cr in 5u8..=8u8, cr in 5u8..=8u8,
preamble in 0u16..=64u16, preamble in 0u16..=64u16,
) { ) {
for bw in [Bandwidth::Khz7, Bandwidth::Khz62, Bandwidth::Khz125, Bandwidth::Khz500] { for bw in [LoRaBandwidth::Khz7, LoRaBandwidth::Khz62, LoRaBandwidth::Khz125, LoRaBandwidth::Khz500] {
let t = lora_air_time(payload, sf, bw, cr, preamble); let t = lora_air_time(payload, sf, bw, cr, preamble);
prop_assert!(t > Duration::ZERO, "air time must be positive"); prop_assert!(t > Duration::ZERO, "air time must be positive");
prop_assert!(t < Duration::from_secs(600), "air time should be < 600s"); prop_assert!(t < Duration::from_secs(600), "air time should be < 600s");
+131 -8
View File
@@ -13,14 +13,14 @@ use tracing::{debug, warn};
use crate::gossip::GossipEvent; use crate::gossip::GossipEvent;
use crate::packet::{GossipFrame, content_hash}; use crate::packet::{GossipFrame, content_hash};
use crate::radio::{ConfigSource, RadioEvent}; use crate::radio::{ConfigSource, RadioEvent, TxRequest, TxRetryReason};
use crate::rate_limit::RateLimiter; use crate::rate_limit::RateLimiter;
/// Active radio config info, sent to the TUI via watch channel. /// Active radio config info, sent to the TUI via watch channel.
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct RadioConfigInfo { pub struct RadioConfigInfo {
pub active: donglora_client::RadioConfig, pub active: donglora_client::LoRaConfig,
pub requested: donglora_client::RadioConfig, pub requested: donglora_client::LoRaConfig,
pub source: ConfigSource, pub source: ConfigSource,
pub device: String, pub device: String,
pub connected: bool, pub connected: bool,
@@ -36,6 +36,29 @@ pub struct PacketLogEntry {
pub snr: Option<i16>, pub snr: Option<i16>,
pub rssi: Option<i16>, pub rssi: Option<i16>,
pub action: PacketAction, pub action: PacketAction,
/// When set, indicates this entry is a TX-attempt update: `Some((attempt, total))`.
pub tx_retry: Option<TxRetryInfo>,
}
/// Per-attempt retry metadata attached to a TX `PacketLogEntry`.
#[derive(Debug, Clone, Copy)]
pub struct TxRetryInfo {
pub attempt: u8,
pub total_attempts: u8,
pub state: TxRetryState,
}
/// State a TX entry is currently in.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TxRetryState {
/// First attempt in flight.
InFlight,
/// A previous attempt failed; this attempt is a retry.
Retrying(TxRetryReason),
/// Final success, possibly after retries.
Succeeded,
/// All attempts exhausted; terminal failure.
Failed,
} }
/// Which side the packet arrived from. /// Which side the packet arrived from.
@@ -83,6 +106,11 @@ pub struct Stats {
pub dropped_queue: AtomicU64, pub dropped_queue: AtomicU64,
pub neighbor_count: AtomicU64, pub neighbor_count: AtomicU64,
pub radio_connected: AtomicU64, // 0 = disconnected, 1 = connected pub radio_connected: AtomicU64, // 0 = disconnected, 1 = connected
/// Cumulative TX retries across the bridge lifetime (each retry after
/// the first attempt increments by 1).
pub tx_retries: AtomicU64,
/// TX calls that exhausted the retry policy.
pub tx_failures: AtomicU64,
} }
impl Stats { impl Stats {
@@ -98,6 +126,8 @@ impl Stats {
dropped_queue: self.dropped_queue.load(Ordering::Relaxed), dropped_queue: self.dropped_queue.load(Ordering::Relaxed),
neighbor_count: self.neighbor_count.load(Ordering::Relaxed), neighbor_count: self.neighbor_count.load(Ordering::Relaxed),
radio_connected: self.radio_connected.load(Ordering::Relaxed) != 0, radio_connected: self.radio_connected.load(Ordering::Relaxed) != 0,
tx_retries: self.tx_retries.load(Ordering::Relaxed),
tx_failures: self.tx_failures.load(Ordering::Relaxed),
} }
} }
} }
@@ -114,6 +144,8 @@ pub struct StatsSnapshot {
pub dropped_queue: u64, pub dropped_queue: u64,
pub neighbor_count: u64, pub neighbor_count: u64,
pub radio_connected: bool, pub radio_connected: bool,
pub tx_retries: u64,
pub tx_failures: u64,
} }
/// Time-bounded dedup cache. /// Time-bounded dedup cache.
@@ -153,10 +185,10 @@ pub async fn run(
our_id: PublicKey, our_id: PublicKey,
dedup_window: Duration, dedup_window: Duration,
tx_queue_size: usize, tx_queue_size: usize,
requested_radio_config: donglora_client::RadioConfig, requested_radio_config: donglora_client::LoRaConfig,
mut rate_limiter: RateLimiter, mut rate_limiter: RateLimiter,
mut radio_rx: mpsc::Receiver<RadioEvent>, mut radio_rx: mpsc::Receiver<RadioEvent>,
radio_tx: mpsc::Sender<Vec<u8>>, radio_tx: mpsc::Sender<TxRequest>,
mut gossip_rx: mpsc::Receiver<GossipEvent>, mut gossip_rx: mpsc::Receiver<GossipEvent>,
gossip_tx: mpsc::Sender<GossipFrame>, gossip_tx: mpsc::Sender<GossipFrame>,
stats: &Stats, stats: &Stats,
@@ -164,6 +196,7 @@ pub async fn run(
config_tx: tokio::sync::watch::Sender<RadioConfigInfo>, config_tx: tokio::sync::watch::Sender<RadioConfigInfo>,
) { ) {
let mut dedup = DedupCache::new(dedup_window); let mut dedup = DedupCache::new(dedup_window);
let mut next_tx_seq: u64 = 0;
let mut tx_queue: VecDeque<Vec<u8>> = VecDeque::with_capacity(tx_queue_size); let mut tx_queue: VecDeque<Vec<u8>> = VecDeque::with_capacity(tx_queue_size);
let mut last_cleanup = Instant::now(); let mut last_cleanup = Instant::now();
let cleanup_interval = Duration::from_secs(30); let cleanup_interval = Duration::from_secs(30);
@@ -195,6 +228,7 @@ pub async fn run(
snr: Some(pkt.snr), snr: Some(pkt.snr),
rssi: Some(pkt.rssi), rssi: Some(pkt.rssi),
action: PacketAction::DroppedDedup, action: PacketAction::DroppedDedup,
tx_retry: None,
}).await; }).await;
continue; continue;
} }
@@ -212,6 +246,7 @@ pub async fn run(
snr: Some(pkt.snr), snr: Some(pkt.snr),
rssi: Some(pkt.rssi), rssi: Some(pkt.rssi),
action: PacketAction::Bridged, action: PacketAction::Bridged,
tx_retry: None,
}).await; }).await;
} }
RadioEvent::Connected(active_config, source, device) => { RadioEvent::Connected(active_config, source, device) => {
@@ -228,6 +263,67 @@ pub async fn run(
stats.radio_connected.store(0, Ordering::Relaxed); stats.radio_connected.store(0, Ordering::Relaxed);
config_tx.send_modify(|info| info.connected = false); config_tx.send_modify(|info| info.connected = false);
} }
RadioEvent::TxAttempt { .. } => {
// No log entry: the GossipIn Bridged row that enqueued
// this TX already represents it. Only retries /
// failures produce follow-up rows.
}
RadioEvent::TxRetry { seq, attempt_that_failed, total_attempts, reason, backoff_ms: _ } => {
stats.tx_retries.fetch_add(1, Ordering::Relaxed);
let _ = log_tx.send(PacketLogEntry {
timestamp: Instant::now(),
hash: tx_seq_hash(seq),
direction: PacketDirection::GossipIn,
size: 0,
snr: None,
rssi: None,
action: PacketAction::Bridged,
tx_retry: Some(TxRetryInfo {
attempt: attempt_that_failed.saturating_add(1),
total_attempts,
state: TxRetryState::Retrying(reason),
}),
}).await;
}
RadioEvent::TxSucceeded { seq, attempts_used, final_airtime_us: _, size } => {
stats.radio_tx.fetch_add(1, Ordering::Relaxed);
// Only emit a success row if retries were involved —
// otherwise the original GossipIn row covers it.
if attempts_used > 1 {
let _ = log_tx.send(PacketLogEntry {
timestamp: Instant::now(),
hash: tx_seq_hash(seq),
direction: PacketDirection::GossipIn,
size,
snr: None,
rssi: None,
action: PacketAction::Bridged,
tx_retry: Some(TxRetryInfo {
attempt: attempts_used,
total_attempts: attempts_used,
state: TxRetryState::Succeeded,
}),
}).await;
}
}
RadioEvent::TxFailed { seq, attempts_used, reason, size } => {
stats.tx_failures.fetch_add(1, Ordering::Relaxed);
warn!("TX #{seq} exhausted retries: {reason}");
let _ = log_tx.send(PacketLogEntry {
timestamp: Instant::now(),
hash: tx_seq_hash(seq),
direction: PacketDirection::GossipIn,
size,
snr: None,
rssi: None,
action: PacketAction::Bridged,
tx_retry: Some(TxRetryInfo {
attempt: attempts_used,
total_attempts: attempts_used,
state: TxRetryState::Failed,
}),
}).await;
}
} }
} }
@@ -250,6 +346,7 @@ pub async fn run(
snr: Some(frame.snr), snr: Some(frame.snr),
rssi: Some(frame.rssi), rssi: Some(frame.rssi),
action: PacketAction::DroppedDedup, action: PacketAction::DroppedDedup,
tx_retry: None,
}).await; }).await;
continue; continue;
} }
@@ -266,6 +363,7 @@ pub async fn run(
snr: Some(frame.snr), snr: Some(frame.snr),
rssi: Some(frame.rssi), rssi: Some(frame.rssi),
action: PacketAction::DroppedRateLimit, action: PacketAction::DroppedRateLimit,
tx_retry: None,
}).await; }).await;
continue; continue;
} }
@@ -282,6 +380,7 @@ pub async fn run(
snr: Some(frame.snr), snr: Some(frame.snr),
rssi: Some(frame.rssi), rssi: Some(frame.rssi),
action: PacketAction::DroppedQueueFull, action: PacketAction::DroppedQueueFull,
tx_retry: None,
}).await; }).await;
continue; continue;
} }
@@ -295,6 +394,7 @@ pub async fn run(
snr: Some(frame.snr), snr: Some(frame.snr),
rssi: Some(frame.rssi), rssi: Some(frame.rssi),
action: PacketAction::Bridged, action: PacketAction::Bridged,
tx_retry: None,
}).await; }).await;
} }
GossipEvent::NeighborChanged(count) => { GossipEvent::NeighborChanged(count) => {
@@ -305,16 +405,39 @@ pub async fn run(
} }
} }
// Drain TX queue to radio. // Drain TX queue to the async radio task. radio_tx is a bounded
// channel; if the radio task is busy we just wait on the next
// loop iteration rather than blocking here.
while let Some(payload) = tx_queue.pop_front() { while let Some(payload) = tx_queue.pop_front() {
stats.radio_tx.fetch_add(1, Ordering::Relaxed); let seq = {
if radio_tx.send(payload).await.is_err() { next_tx_seq = next_tx_seq.wrapping_add(1);
next_tx_seq
};
if radio_tx.send(TxRequest { seq, data: payload }).await.is_err() {
break; break;
} }
} }
} }
} }
/// Build a stable per-TX "hash" for packet-log correlation. Not a real
/// content hash — just a 32-byte encoding of the seq so the TUI can
/// index/update log rows by it.
const fn tx_seq_hash(seq: u64) -> [u8; 32] {
let mut out = [0u8; 32];
out[0] = 0x01; // tag marking this as a TX-seq pseudo-hash
let bytes = seq.to_le_bytes();
out[1] = bytes[0];
out[2] = bytes[1];
out[3] = bytes[2];
out[4] = bytes[3];
out[5] = bytes[4];
out[6] = bytes[5];
out[7] = bytes[6];
out[8] = bytes[7];
out
}
#[cfg(test)] #[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)] #[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests { mod tests {
+4 -7
View File
@@ -58,15 +58,12 @@ pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> {
let sync_word_str = prompt_default("Sync Word", &sw_default)?; let sync_word_str = prompt_default("Sync Word", &sw_default)?;
let sync_word = parse_sync_word(&sync_word_str)?; let sync_word = parse_sync_word(&sync_word_str)?;
let tx_power = prompt_default("TX Power (dBm or 'max')", &radio.tx_power)?; let tx_power_default = format!("{}", radio.tx_power_dbm);
let tx_power_dbm: i8 = prompt_parse("TX Power (dBm)", &tx_power_default)?;
let pre_default = format!("{}", radio.preamble); let pre_default = format!("{}", radio.preamble);
let preamble: u16 = prompt_parse("Preamble", &pre_default)?; let preamble: u16 = prompt_parse("Preamble", &pre_default)?;
let cad_default = if radio.cad { "on" } else { "off" };
let cad_str = prompt_default("CAD", cad_default)?;
let cad = matches!(cad_str.to_lowercase().as_str(), "on" | "true" | "yes" | "1");
let port_default = radio.port.as_deref().unwrap_or("auto"); let port_default = radio.port.as_deref().unwrap_or("auto");
let port_str = prompt_default("Serial Port", port_default)?; let port_str = prompt_default("Serial Port", port_default)?;
let port = if port_str == "auto" || port_str.is_empty() { None } else { Some(port_str) }; let port = if port_str == "auto" || port_str.is_empty() { None } else { Some(port_str) };
@@ -81,9 +78,8 @@ pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> {
spreading_factor, spreading_factor,
coding_rate, coding_rate,
sync_word, sync_word,
tx_power, tx_power_dbm,
preamble, preamble,
cad,
}, },
bridge: BridgeSection { bridge: BridgeSection {
passphrase, passphrase,
@@ -92,6 +88,7 @@ pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> {
rate_limit_pps: bridge.rate_limit_pps, rate_limit_pps: bridge.rate_limit_pps,
log_file: bridge.log_file, log_file: bridge.log_file,
}, },
tx: existing.map(|c| c.tx.clone()).unwrap_or_default(),
}; };
write_config(config_path, &cfg)?; write_config(config_path, &cfg)?;
+58 -21
View File
@@ -23,7 +23,9 @@ use tokio_util::sync::CancellationToken;
use crate::gossip::Gossip; use crate::gossip::Gossip;
use crate::radio::ConfigSource; use crate::radio::ConfigSource;
use crate::router::{PacketAction, PacketDirection, PacketLogEntry, RadioConfigInfo, Stats, StatsSnapshot}; use crate::router::{
PacketAction, PacketDirection, PacketLogEntry, RadioConfigInfo, Stats, StatsSnapshot, TxRetryState,
};
const TICK_RATE: Duration = Duration::from_millis(250); const TICK_RATE: Duration = Duration::from_millis(250);
const MAX_LOG_ENTRIES: usize = 200; const MAX_LOG_ENTRIES: usize = 200;
@@ -366,13 +368,16 @@ fn draw_radio(frame: &mut ratatui::Frame, area: Rect, config_info: &RadioConfigI
let freq_mhz = f64::from(a.freq_hz) / 1_000_000.0; let freq_mhz = f64::from(a.freq_hz) / 1_000_000.0;
let bw_str = crate::radio::format_bandwidth(a.bw); let bw_str = crate::radio::format_bandwidth(a.bw);
let power_str = if a.tx_power_dbm == donglora_client::TX_POWER_MAX { let power_str = format!("{} dBm", a.tx_power_dbm);
"max".to_string()
} else {
format!("{} dBm", a.tx_power_dbm)
};
let preamble = if a.preamble_len == 0 { 16 } else { a.preamble_len }; let preamble = if a.preamble_len == 0 { 16 } else { a.preamble_len };
let cad = if a.cad != 0 { "on" } else { "off" }; let cr_denom = |c: donglora_client::LoRaCodingRate| -> u8 {
match c {
donglora_client::LoRaCodingRate::Cr4_5 => 5,
donglora_client::LoRaCodingRate::Cr4_6 => 6,
donglora_client::LoRaCodingRate::Cr4_7 => 7,
donglora_client::LoRaCodingRate::Cr4_8 => 8,
}
};
// Dim values when disconnected; highlight mux mismatches when connected. // Dim values when disconnected; highlight mux mismatches when connected.
let val_color = if connected { C_ACCENT } else { C_LABEL }; let val_color = if connected { C_ACCENT } else { C_LABEL };
@@ -394,11 +399,10 @@ fn draw_radio(frame: &mut ratatui::Frame, area: Rect, config_info: &RadioConfigI
cf("Freq", format!("{freq_mhz:.3} MHz"), a.freq_hz == r.freq_hz), cf("Freq", format!("{freq_mhz:.3} MHz"), a.freq_hz == r.freq_hz),
cf("BW", bw_str.to_string(), a.bw == r.bw), cf("BW", bw_str.to_string(), a.bw == r.bw),
cf("SF", format!("{}", a.sf), a.sf == r.sf), cf("SF", format!("{}", a.sf), a.sf == r.sf),
cf("CR", format!("4/{}", a.cr), a.cr == r.cr), cf("CR", format!("4/{}", cr_denom(a.cr)), a.cr == r.cr),
cf("Sync", format!("0x{:04X}", a.sync_word), a.sync_word == r.sync_word), cf("Sync", format!("0x{:04X}", a.sync_word), a.sync_word == r.sync_word),
cf("TX", power_str, a.tx_power_dbm == r.tx_power_dbm), cf("TX", power_str, a.tx_power_dbm == r.tx_power_dbm),
cf("Preamble", format!("{preamble}"), a.preamble_len == r.preamble_len), cf("Preamble", format!("{preamble}"), a.preamble_len == r.preamble_len),
cf("CAD", cad.to_string(), a.cad == r.cad),
]; ];
frame.render_widget(Paragraph::new(lines).block(block), area); frame.render_widget(Paragraph::new(lines).block(block), area);
} }
@@ -430,6 +434,11 @@ fn draw_stats(frame: &mut ratatui::Frame, area: Rect, stats: &StatsSnapshot, sta
s("Deduped", stats.dedup_hits, C_DEDUP), s("Deduped", stats.dedup_hits, C_DEDUP),
s("Rate lim", stats.rate_limit_drops, C_RATE_DROP), s("Rate lim", stats.rate_limit_drops, C_RATE_DROP),
s("Q drop", stats.dropped_queue, C_QUEUE_DROP), s("Q drop", stats.dropped_queue, C_QUEUE_DROP),
// CAD-busy retries tally (per spec §6.10). Yellow when accumulating,
// dim when quiet.
s("TX retry", stats.tx_retries, if stats.tx_retries > 0 { C_RATE_DROP } else { C_LABEL }),
// TX calls that exhausted the retry policy and gave up.
s("TX fail", stats.tx_failures, if stats.tx_failures > 0 { C_QUEUE_DROP } else { C_LABEL }),
]; ];
frame.render_widget(Paragraph::new(lines).block(block), area); frame.render_widget(Paragraph::new(lines).block(block), area);
} }
@@ -447,6 +456,11 @@ fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry])
// Fixed column widths: Age=5 Hash=8 RF=4 Net=4 Size=6 RSSI=6 SNR=5 Act=5 // Fixed column widths: Age=5 Hash=8 RF=4 Net=4 Size=6 RSSI=6 SNR=5 Act=5
// Every span (header and data) uses the same width per column. // Every span (header and data) uses the same width per column.
//
// `Act` stays at 5 chars so the full row fits inside the Packet Log
// panel. Retry labels compress to 4 chars (`↻3/3`, `✓3/3`, `✗3/3`) —
// plain drop labels (`DUP`, `FULL`, `RATE`) right-align within the
// same width.
let hdr = Style::default().fg(Color::Gray).add_modifier(Modifier::BOLD); let hdr = Style::default().fg(Color::Gray).add_modifier(Modifier::BOLD);
let header = Line::from(vec![ let header = Line::from(vec![
Span::styled(format!("{:>5}", "Age"), hdr), Span::styled(format!("{:>5}", "Age"), hdr),
@@ -488,11 +502,35 @@ fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry])
), ),
}; };
let action_color = match e.action { // TX retry state wins over the plain action label when present —
PacketAction::Bridged => C_LABEL, // TX-log entries use the retry tag to carry through retry / success /
PacketAction::DroppedDedup => C_DEDUP, // failure state. Otherwise fall back to the generic action.
PacketAction::DroppedQueueFull => C_QUEUE_DROP, let (action_text, action_color) = if let Some(r) = e.tx_retry {
PacketAction::DroppedRateLimit => C_RATE_DROP, // Compact `↻N/M` / `✓N/M` / `✗N/M` — no space between glyph
// and counter, so it fits in the 4-char data slot.
match r.state {
TxRetryState::Retrying(_) => (format!("↻{}/{}", r.attempt, r.total_attempts), C_RATE_DROP),
TxRetryState::Succeeded => (format!("✓{}/{}", r.attempt, r.total_attempts), C_NET_TX),
TxRetryState::Failed => (format!("✗{}/{}", r.attempt, r.total_attempts), C_QUEUE_DROP),
// `InFlight` isn't currently emitted by the router, but keep
// a graceful fallback so any future debug wiring doesn't
// break rendering.
TxRetryState::InFlight => (String::new(), C_LABEL),
}
} else {
let color = match e.action {
PacketAction::Bridged => C_LABEL,
PacketAction::DroppedDedup => C_DEDUP,
PacketAction::DroppedQueueFull => C_QUEUE_DROP,
PacketAction::DroppedRateLimit => C_RATE_DROP,
};
let label = match e.action {
PacketAction::Bridged => "",
PacketAction::DroppedDedup => "DUP",
PacketAction::DroppedQueueFull => "FULL",
PacketAction::DroppedRateLimit => "RATE",
};
(label.to_string(), color)
}; };
let rssi_span = match e.rssi { let rssi_span = match e.rssi {
@@ -504,13 +542,9 @@ fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry])
None => Span::styled(format!("{:>5}", "-"), Style::default().fg(C_LABEL)), None => Span::styled(format!("{:>5}", "-"), Style::default().fg(C_LABEL)),
}; };
let action_str = match e.action { // Leading space + right-aligned 4-char data = 5 total, matching
PacketAction::Bridged => "", // the `Act` header.
PacketAction::DroppedDedup => "DUP", let action_span = Span::styled(format!(" {action_text:>4}"), Style::default().fg(action_color));
PacketAction::DroppedQueueFull => "FULL",
PacketAction::DroppedRateLimit => "RATE",
};
let action_span = Span::styled(format!(" {action_str:>4}"), Style::default().fg(action_color));
let size_str = format!("{}B", e.size); let size_str = format!("{}B", e.size);
lines.push(Line::from(vec![ lines.push(Line::from(vec![
@@ -700,6 +734,8 @@ mod tests {
dropped_queue: 0, dropped_queue: 0,
neighbor_count: 0, neighbor_count: 0,
radio_connected: true, radio_connected: true,
tx_retries: 0,
tx_failures: 0,
} }
} }
@@ -712,6 +748,7 @@ mod tests {
snr, snr,
rssi: Some(-80), rssi: Some(-80),
action: PacketAction::Bridged, action: PacketAction::Bridged,
tx_retry: None,
} }
} }