feat: publish-ready overhaul — lints, tests, docs, CI, fuzzing, mutant testing

- Enable clippy pedantic + nursery lints; fix all warnings
- Add config validation (SF, CR, frequency range checks)
- Zeroize passphrase intermediates after key derivation
- Convert to lib+bin crate for fuzz target accessibility
- Replace Vec with VecDeque for O(1) log/bucket rotation
- Add TUI minimum terminal size guard

Testing:
- 145 unit + property tests (was 15)
- 6 proptest groups (frame roundtrip, hash, SNR, air time, peer list, padding)
- 2 cargo-fuzz targets (gossip frame decode, peer list decode)
- cargo-mutants: 0 missed, 241 caught, 54 unviable

Documentation:
- README with prerequisites, quick start, config reference, architecture
- Module-level and item-level rustdoc on all public items
- LICENSE (MIT), rustfmt.toml, .gitignore

Infrastructure:
- GitHub Actions CI (fmt, clippy, test, audit)
- cargo-mutants config (.cargo/mutants.toml)
- justfile for common tasks
This commit is contained in:
swaits committed 2026-04-06 11:39:43 -06:00
1 parent b3bd92c5f8
commit 5dbec82aa9
24 files changed
+3040 -555

No files matched your search

+55
View File
@@ -0,0 +1,55 @@
# cargo-mutants configuration
# Run with: cargo mutants
# Install with: cargo install cargo-mutants
# Exclude entire files that are untestable (main requires full app, setup requires interactive stdin)
exclude_globs = [
"src/main.rs",
"src/setup.rs",
]
# Exclude specific functions that require hardware, network, terminal, or full async infrastructure
exclude_re = [
# Equivalent mutants: Instant arithmetic never produces exact equality,
# so `< ttl` and `<= ttl` behave identically in practice.
"replace < with <= in DedupCache",
"replace < with <= in decode_peer_list",
# rate_limit.rs: equivalent mutant (secs is always > 0 from lora_air_time)
"replace > with >= in RateLimiter::from_radio_config",
# radio.rs: hardware-dependent functions
"radio_loop",
"connect_and_run",
# gossip.rs: async network functions
"Gossip::new",
"Gossip::shutdown",
"spawn_accept_loop",
"handle_incoming",
"spawn_receiver",
"spawn_broadcaster",
"spawn_dht_heartbeat",
"dht_read_and_join",
"dht_publish",
"dht_get_peers",
# router.rs: async event loop
"^src/router\\.rs:.* replace run ",
"^src/router\\.rs:.* in run$",
# tui.rs: terminal rendering functions
"tui::run",
"tui_loop",
"restore_terminal",
"draw_ui",
"draw_header",
"draw_network",
"draw_radio",
"draw_stats",
"draw_log",
"draw_activity",
"draw_sparkline_section",
"draw_shutdown_overlay",
"panel_block",
"status_pill",
"kv_line",
]
# Timeout per mutant test (seconds)
timeout_multiplier = 3.0
+44
View File
@@ -0,0 +1,44 @@
name: CI
on:
push:
branches: [main]
pull_request:
branches: [main]
env:
CARGO_TERM_COLOR: always
jobs:
check:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Clone donglora (sibling dependency)
uses: actions/checkout@v4
with:
repository: swaits/donglora
path: ../donglora
- uses: dtolnay/rust-toolchain@master
with:
toolchain: "1.94.1"
components: rustfmt, clippy
- uses: Swatinem/rust-cache@v2
- name: Format check
run: cargo fmt --check
- name: Clippy
run: cargo clippy --all-targets -- -D warnings
- name: Test
run: cargo test
- name: Install cargo-deny
run: cargo install cargo-deny --locked
- name: Audit (advisories + licenses)
run: cargo deny check
+10
View File
@@ -1 +1,11 @@
/target
/fuzz/target
/fuzz/corpus
/fuzz/Cargo.lock
/mutants.out
/mutants.out.old
.jj/
*.swp
*.swo
*~
.DS_Store
Generated
+785 -188
View File
File diff suppressed because it is too large. Load diff
+20 -3
View File
@@ -5,6 +5,13 @@ edition = "2024"
authors = ["Stephen Waits <steve@waits.net>"]
license = "MIT"
description = "Peer-to-peer LoRa bridge using iroh and DongLoRa"
repository = "https://github.com/swaits/donglora-bridge"
homepage = "https://github.com/swaits/donglora-bridge"
keywords = ["lora", "bridge", "gossip", "p2p", "iot"]
categories = ["network-programming", "command-line-utilities"]
readme = "README.md"
rust-version = "1.94.1"
exclude = [".jj/", ".github/"]
[dependencies]
donglora-client = { path = "../donglora/clients/rust" }
@@ -14,10 +21,10 @@ mainline = { version = "6", features = ["async"] }
tokio = { version = "1", features = ["full"] }
clap = { version = "4", features = ["derive"] }
serde = { version = "1", features = ["derive"] }
toml = "0.8"
toml = "1"
blake3 = "1"
ratatui = "0.29"
crossterm = "0.28"
ratatui = "0.30"
crossterm = "0.29"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
tracing-appender = "0.2"
@@ -30,8 +37,18 @@ hex = "0.4.3"
sha2 = "0.10"
tokio-util = { version = "0.7.18", features = ["rt"] }
jiff = "0.2.23"
zeroize = "1"
[dev-dependencies]
proptest = "1"
tempfile = "3"
[lints.clippy]
pedantic = { level = "warn", priority = -1 }
nursery = { level = "warn", priority = -1 }
unwrap_used = "deny"
expect_used = "deny"
panic = "deny"
similar_names = "allow"
module_name_repetitions = "allow"
missing_errors_doc = "allow"
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 Stephen Waits <steve@waits.net>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
+273
View File
@@ -0,0 +1,273 @@
# 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)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)
![Rust](https://img.shields.io/badge/rust-1.94.1%2B-orange)
**donglora-bridge** connects LoRa radios to the internet so packets heard by one
radio can be retransmitted by another radio anywhere in the world.
[LoRa](https://en.wikipedia.org/wiki/LoRa) is a long-range, low-power wireless
protocol used by off-grid mesh networks like
[MeshCore](https://github.com/rocketgithub/meshcore-firmware). A single LoRa
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
[Mainline DHT](https://en.wikipedia.org/wiki/Mainline_DHT) and communicate
through an encrypted peer-to-peer gossip swarm.
![TUI Screenshot](screenshot.png)
## What you need
### 1. A DongLoRa radio
[DongLoRa](https://github.com/swaits/donglora) is open-source firmware that
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:
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
the full list.
2. **Flash the DongLoRa firmware** onto the board. Instructions are in the
DongLoRa repository.
3. **Plug it in via USB**. The board appears as a serial device
(e.g. `/dev/ttyACM0` on Linux).
### 2. (Optional) donglora-mux
If you want multiple programs to share a single DongLoRa radio at the same time,
install the multiplexer daemon:
```sh
cargo install donglora-mux
```
Run `donglora-mux` before starting the bridge. The bridge will auto-detect and
connect to the mux. Without the mux, the bridge talks to the USB device
directly, which means only one program can use the radio at a time.
### 3. donglora-bridge (this project)
Install via cargo:
```sh
cargo install donglora-bridge
```
Or via [mise](https://mise.jdx.dev/):
```sh
mise use -g cargo:donglora-bridge
```
Or build from source:
```sh
git clone https://github.com/swaits/donglora-bridge.git
cd donglora-bridge
cargo build --release
```
Requires Linux. macOS may work but is untested.
## Quick start
### 1. Plug in your DongLoRa board and (optionally) start the mux
```sh
donglora-mux # optional -- skip if not sharing the radio
```
### 2. Run the bridge
```sh
donglora-bridge
```
On first run, an interactive wizard walks you through configuration:
- **Passphrase** (required) -- every bridge using the same passphrase joins the
same network. Pick something unique to your group.
- **Radio settings** -- frequency, bandwidth, spreading factor, etc. The
defaults match MeshCore US/Canada recommended values. If you're bridging a
MeshCore network, just press Enter through these.
The config is saved to `~/.config/donglora-bridge/config.toml`. To change it
later, run `donglora-bridge config`.
### 3. That's it
The TUI shows real-time stats. On another machine (or continent), repeat the
same steps with the same passphrase. The two bridges will discover each other
via DHT and start relaying packets within seconds.
```sh
donglora-bridge # TUI mode (default)
donglora-bridge --log-only # headless mode (structured logs to stdout)
donglora-bridge --config path.toml # custom config file
donglora-bridge config # re-run setup wizard
```
## Features
- **Gossip swarm** -- broadcast packets via [iroh-gossip](https://github.com/n0-computer/iroh), a QUIC-based overlay network
- **Zero-config peer discovery** -- automatic bootstrapping via BitTorrent mainline DHT ([BEP44](http://bittorrent.org/beps/bep_0044.html) mutable items)
- **Content-hash deduplication** -- blake3 hashing suppresses duplicate packets across all bridges
- **Air-time-aware rate limiting** -- token bucket calculated from LoRa radio parameters (Semtech LoRa time-on-air formula)
- **Real-time terminal UI** -- network status, radio config, packet log, activity sparklines
- **Resilient radio connection** -- exponential backoff reconnection with liveness pings
- **Headless mode** -- structured logs to stdout for servers (`--log-only`)
- **Interactive setup wizard** -- guided first-run configuration
- **Shared passphrase security** -- all cryptographic keys derived deterministically from a single passphrase
## TUI overview
| Panel | Content |
|-------|---------|
| **Network** | Topic ID, peer count, DHT status, last publish time |
| **Radio** | Device, frequency, bandwidth, SF, CR, sync word, TX power, preamble, CAD |
| **Stats** | Uptime, RF/Net RX/TX, dedup hits, rate limit drops, queue drops |
| **Packet Log** | Scrolling log: age, hash, RF/Net direction, size, RSSI, SNR, action |
| **Activity** | Sparkline charts: RX rate, TX rate, average SNR (1-second buckets, 5-minute window) |
Press `q` or `Ctrl+C` to shut down gracefully.
## Configuration
Default location: `~/.config/donglora-bridge/config.toml`
```toml
[radio]
# port = "/dev/ttyACM0" # serial port (omit for auto-detect via mux)
frequency = 910525000 # Hz (default: 910.525 MHz)
bandwidth = "62.5kHz" # 7.8kHz, 10.4kHz, 15.6kHz, 20.8kHz, 31.25kHz,
# 41.7kHz, 62.5kHz, 125kHz, 250kHz, 500kHz
spreading_factor = 7 # 7-12
coding_rate = 5 # 5-8 (meaning 4/5 to 4/8)
sync_word = 0x3444 # hex, 2 bytes
tx_power = "22" # dBm integer or "max"
preamble = 16 # symbols
cad = true # channel activity detection
[bridge]
passphrase = "your-shared-secret" # REQUIRED: all bridges with same passphrase join the same swarm
dedup_window_secs = 300 # duplicate suppression window (default: 5 minutes)
tx_queue_size = 32 # max pending radio TX packets
# rate_limit_pps = 5.0 # optional: override calculated TX rate (packets/sec)
# log_file = "/path/to/bridge.log" # optional: override log path
```
Default radio settings match MeshCore US/Canada recommended parameters.
## How it works
### Passphrase key derivation
A single passphrase deterministically derives three values:
1. **Topic ID** -- `blake3(APP_PREFIX + passphrase)` -- identifies the gossip swarm
2. **DHT signing key** -- ed25519 from `SHA-256(APP_PREFIX + "sign:" + passphrase)` -- authenticates DHT records
3. **DHT salt** -- first 20 bytes of `SHA-256(APP_PREFIX + "salt:" + passphrase)` -- namespaces the DHT key
All nodes with the same passphrase compute identical keys and automatically join
the same network.
### Gossip swarm
Each bridge creates an ephemeral iroh endpoint (fresh ed25519 identity per
launch) and subscribes to the derived topic. Radio packets are wrapped in a
`GossipFrame` (32-byte sender ID + RSSI + SNR + payload) and broadcast to all
neighbors. Received frames are decoded and queued for radio transmission.
### DHT peer discovery
Peers find each other through BitTorrent's mainline DHT using BEP44 mutable
items. Peer lists are encoded as `[count:u16 LE][id:32B]...` (max 27 peers per
record to stay under the 1000-byte value limit).
**When alone** (no neighbors): reads DHT every 3 seconds, merge-publishes every
30 seconds (preserves existing peers, never destructive).
**When connected**: relaxed 5-minute heartbeat with random jitter. Publishes
self + live neighbors only (purges stale peers).
**On shutdown**: publishes neighbors without self, so other nodes can still
bootstrap from live peers.
### Deduplication
Every packet payload is hashed with blake3 (32 bytes). A time-bounded cache
(default 300s, pruned every 30s) suppresses duplicates from both radio and
gossip directions.
### Rate limiting
A token bucket rate limiter prevents radio channel saturation. The rate is
calculated from the LoRa air time formula (Semtech SX1276):
```
air_time = preamble_time + payload_symbol_count * symbol_time
rate = 0.5 / air_time (50% duty cycle target)
```
Default burst allowance is 3 packets. The rate can be manually overridden with
`rate_limit_pps` in the config.
### Radio communication
A dedicated blocking thread communicates with the DongLoRa hardware via
`donglora-client`. Features:
- **Config negotiation**: reads existing config first; only sets if needed
- **Exponential backoff reconnection**: 250ms to 5s, resets after 5s of uptime
- **Liveness pings**: every 2s of idle, detects silently-dead connections
- **SNR filtering**: drops packets below the demodulation floor
### Architecture
```
┌─────────────────────────────────────┐
│ donglora-bridge │
│ │
DongLoRa ┌────────┐│ ┌────────┐ ┌──────┐ ┌───────┐ │ Internet
USB dongle◄─┤ Radio ├┤◄─┤ Router ├──►│Gossip├──►│ iroh ├──┤◄──► Other
│ Thread ││─►│ (dedup ││ │Swarm │ │ QUIC │ │ Bridges
└────────┘│ │ rate ││ └──┬───┘ └───────┘ │
│ │ limit)│| │ │
│ └────────┘│ ┌──┴───┐ │
│ ▲ │ │ DHT │ │
│ │ │ │(peer │ │
│ ┌──┴──┐ │ │disc.)│ │
│ │ TUI │ │ └──────┘ │
│ └─────┘ │ │
└─────────────────────────────────────┘
```
## Testing
```sh
cargo test # unit tests + property tests (proptest)
cargo clippy --all-targets # lint check (pedantic + nursery)
cargo fmt --check # format check
cargo mutants # mutation testing (install: cargo install cargo-mutants)
cargo +nightly fuzz run fuzz_gossip_frame_decode # fuzz testing (requires nightly)
cargo +nightly fuzz run fuzz_decode_peer_list
```
## Troubleshooting
| Symptom | Cause | Fix |
|---------|-------|-----|
| "passphrase is required" | Empty or placeholder passphrase | Run `donglora-bridge config` |
| No peers found | Passphrase mismatch or firewall | Check passphrase matches; iroh uses QUIC (UDP), DHT uses UDP |
| Radio disconnected | USB dongle unplugged or mux crash | Check USB connection; bridge will auto-reconnect |
| Yellow `!` on radio fields | Mux has different radio config | Another client set different params; bridge adopts mux config |
| Log file location | Default path | `~/.local/state/donglora-bridge/bridge.log` |
## License
[MIT](LICENSE)
+41
View File
@@ -0,0 +1,41 @@
# cargo-deny configuration (v0.19+)
# Run with: cargo deny check
[licenses]
version = 2
# Permissive licenses only. No copyleft.
allow = [
"MIT",
"MIT-0",
"Apache-2.0",
"BSD-1-Clause",
"BSD-2-Clause",
"BSD-3-Clause",
"BSL-1.0",
"ISC",
"Zlib",
"Unlicense",
"CC0-1.0",
"CDLA-Permissive-2.0",
"Unicode-3.0",
"MPL-2.0",
]
[advisories]
# Unmaintained transitive deps from upstream (iroh dep tree) — not actionable by us.
ignore = [
"RUSTSEC-2023-0089", # atomic-polyfill: unmaintained (transitive via iroh)
"RUSTSEC-2024-0436", # paste: unmaintained (transitive via iroh)
]
[bans]
multiple-versions = "warn"
wildcards = "warn"
[sources]
unknown-registry = "deny"
unknown-git = "warn"
allow-git = [
"https://github.com/swaits/donglora",
]
+26
View File
@@ -0,0 +1,26 @@
[package]
name = "donglora-bridge-fuzz"
version = "0.0.0"
publish = false
edition = "2024"
[package.metadata]
cargo-fuzz = true
[dependencies]
libfuzzer-sys = "0.4"
[dependencies.donglora-bridge]
path = ".."
[[bin]]
name = "fuzz_gossip_frame_decode"
path = "fuzz_targets/fuzz_gossip_frame_decode.rs"
doc = false
[[bin]]
name = "fuzz_decode_peer_list"
path = "fuzz_targets/fuzz_decode_peer_list.rs"
doc = false
[workspace]
@@ -0,0 +1,7 @@
#![no_main]
use libfuzzer_sys::fuzz_target;
fuzz_target!(|data: &[u8]| {
// decode_peer_list must never panic on arbitrary input.
let _ = donglora_bridge::gossip::decode_peer_list(data);
});
@@ -0,0 +1,7 @@
#![no_main]
use libfuzzer_sys::fuzz_target;
fuzz_target!(|data: &[u8]| {
// GossipFrame::decode must never panic on arbitrary input.
let _ = donglora_bridge::packet::GossipFrame::decode(data);
});
+60
View File
@@ -0,0 +1,60 @@
# Default: run all checks
default: build
# Run all checks (fmt, clippy, test)
check: fmt-check clippy test audit
# Format code
fmt:
cargo fmt
# Check formatting without changing files
fmt-check:
cargo fmt --check
# Lint with clippy (pedantic + nursery)
clippy:
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
test: test-unit test-fuzz test-mutants
# Normal unit tests
test-unit:
cargo nextest run
# Mutation testing
test-mutants *ARGS:
cargo mutants {{ARGS}}
# Fuzz a target (requires nightly)
test-fuzz TARGET="fuzz_gossip_frame_decode" DURATION="60":
cargo +nightly fuzz run {{TARGET}} -- -max_total_time={{DURATION}}
# Clean build artifacts
clean:
cargo clean
+3
View File
@@ -0,0 +1,3 @@
edition = "2024"
max_width = 120
use_small_heuristics = "Max"
BIN
View File
Binary file not shown.

After

Width:  |  Height:  |  Size: 187 KiB

+260 -17
View File
@@ -1,3 +1,8 @@
//! TOML configuration loading and validation.
//!
//! Handles the `[radio]` and `[bridge]` sections with `MeshCore` US/Canada recommended
//! defaults, and conversion to donglora-client `RadioConfig`.
use std::path::{Path, PathBuf};
use anyhow::{Context, Result, bail};
@@ -52,34 +57,34 @@ pub struct BridgeSection {
// ── Defaults (MeshCore US/Canada Recommended) ────────────────────
fn default_frequency() -> u32 {
const fn default_frequency() -> u32 {
910_525_000
}
fn default_bandwidth() -> String {
"62.5kHz".into()
}
fn default_sf() -> u8 {
const fn default_sf() -> u8 {
7
}
fn default_cr() -> u8 {
const fn default_cr() -> u8 {
5
}
fn default_sync_word() -> u16 {
const fn default_sync_word() -> u16 {
0x3444
}
fn default_tx_power() -> String {
"22".into()
}
fn default_preamble() -> u16 {
const fn default_preamble() -> u16 {
16
}
fn default_true() -> bool {
const fn default_true() -> bool {
true
}
fn default_dedup_window() -> u64 {
const fn default_dedup_window() -> u64 {
300
}
fn default_tx_queue_size() -> usize {
const fn default_tx_queue_size() -> usize {
32
}
@@ -138,8 +143,20 @@ fn parse_tx_power(s: &str) -> Result<i8> {
}
impl RadioSection {
/// Convert to the donglora-client RadioConfig.
/// Convert to the donglora-client `RadioConfig`.
///
/// Validates that all radio parameters are within hardware-supported ranges
/// before constructing the config.
pub fn to_radio_config(&self) -> Result<RadioConfig> {
if !(7..=12).contains(&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 {
bail!("frequency {:.3} MHz out of range (150 MHz to 1.02 GHz)", f64::from(self.frequency) / 1_000_000.0);
}
Ok(RadioConfig {
freq_hz: self.frequency,
bw: parse_bandwidth(&self.bandwidth)?,
@@ -157,9 +174,7 @@ impl RadioSection {
/// Default config directory: `$XDG_CONFIG_HOME/donglora-bridge/` or `~/.config/donglora-bridge/`.
pub fn config_dir() -> Result<PathBuf> {
let dir = dirs::config_dir()
.context("cannot determine config directory")?
.join("donglora-bridge");
let dir = dirs::config_dir().context("cannot determine config directory")?.join("donglora-bridge");
Ok(dir)
}
@@ -179,13 +194,241 @@ pub fn default_log_path() -> Result<PathBuf> {
/// Load config from a TOML file. Returns None-passphrase error if not set.
pub fn load_config(path: &Path) -> Result<Config> {
let text = std::fs::read_to_string(path)
.with_context(|| format!("reading config: {}", path.display()))?;
let text = std::fs::read_to_string(path).with_context(|| format!("reading config: {}", path.display()))?;
let cfg: Config = toml::from_str(&text).context("parsing config TOML")?;
if cfg.bridge.passphrase.is_empty() || cfg.bridge.passphrase == "change-me" {
bail!(
"passphrase is required — run `donglora-bridge config` to set it"
);
bail!("passphrase is required — run `donglora-bridge config` to set it");
}
Ok(cfg)
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
#[test]
fn parse_bandwidth_canonical() {
assert!(matches!(parse_bandwidth("7.8kHz"), Ok(Bandwidth::Khz7)));
assert!(matches!(parse_bandwidth("62.5kHz"), Ok(Bandwidth::Khz62)));
assert!(matches!(parse_bandwidth("125kHz"), Ok(Bandwidth::Khz125)));
assert!(matches!(parse_bandwidth("500kHz"), Ok(Bandwidth::Khz500)));
}
#[test]
fn parse_bandwidth_aliases() {
assert!(matches!(parse_bandwidth("7khz"), Ok(Bandwidth::Khz7)));
assert!(matches!(parse_bandwidth("khz125"), Ok(Bandwidth::Khz125)));
assert!(matches!(parse_bandwidth("10khz"), Ok(Bandwidth::Khz10)));
}
#[test]
fn parse_bandwidth_case_insensitive() {
assert!(matches!(parse_bandwidth("125KHZ"), Ok(Bandwidth::Khz125)));
assert!(matches!(parse_bandwidth("62.5KHZ"), Ok(Bandwidth::Khz62)));
}
#[test]
fn parse_bandwidth_rejects_unknown() {
assert!(parse_bandwidth("999kHz").is_err());
assert!(parse_bandwidth("").is_err());
assert!(parse_bandwidth("not_a_bandwidth").is_err());
}
#[test]
fn parse_tx_power_numeric() {
assert_eq!(parse_tx_power("22").unwrap(), 22);
assert_eq!(parse_tx_power("-4").unwrap(), -4);
assert_eq!(parse_tx_power("0").unwrap(), 0);
}
#[test]
fn parse_tx_power_max_keyword() {
assert_eq!(parse_tx_power("max").unwrap(), TX_POWER_MAX);
assert_eq!(parse_tx_power("MAX").unwrap(), TX_POWER_MAX);
assert_eq!(parse_tx_power("Max").unwrap(), TX_POWER_MAX);
}
#[test]
fn parse_tx_power_rejects_garbage() {
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 config = section.to_radio_config().unwrap();
assert_eq!(config.freq_hz, 910_525_000);
assert_eq!(config.sf, 7);
assert_eq!(config.cr, 5);
assert_eq!(config.sync_word, 0x3444);
assert_eq!(config.preamble_len, 16);
}
#[test]
fn radio_section_max_power() {
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() };
assert!(section.to_radio_config().is_err());
let section = RadioSection { spreading_factor: 13, ..RadioSection::default() };
assert!(section.to_radio_config().is_err());
}
#[test]
fn to_radio_config_rejects_invalid_cr() {
let section = RadioSection { coding_rate: 4, ..RadioSection::default() };
assert!(section.to_radio_config().is_err());
let section = RadioSection { coding_rate: 9, ..RadioSection::default() };
assert!(section.to_radio_config().is_err());
}
#[test]
fn to_radio_config_rejects_invalid_frequency() {
let section = RadioSection { frequency: 0, ..RadioSection::default() };
assert!(section.to_radio_config().is_err());
let section = RadioSection { frequency: 2_000_000_000, ..RadioSection::default() };
assert!(section.to_radio_config().is_err());
}
#[test]
fn load_config_valid_minimal() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.toml");
std::fs::write(&path, "[bridge]\npassphrase = \"my-secret\"\n").unwrap();
let cfg = load_config(&path).unwrap();
assert_eq!(cfg.bridge.passphrase, "my-secret");
assert_eq!(cfg.radio.frequency, 910_525_000);
}
#[test]
fn load_config_empty_passphrase_rejected() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.toml");
std::fs::write(&path, "[bridge]\npassphrase = \"\"\n").unwrap();
assert!(load_config(&path).is_err());
}
#[test]
fn load_config_change_me_rejected() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.toml");
std::fs::write(&path, "[bridge]\npassphrase = \"change-me\"\n").unwrap();
assert!(load_config(&path).is_err());
}
#[test]
fn load_config_missing_file() {
assert!(load_config(std::path::Path::new("/nonexistent/config.toml")).is_err());
}
#[test]
fn load_config_invalid_toml() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.toml");
std::fs::write(&path, "not valid toml {{{{").unwrap();
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);
}
}
+169 -59
View File
@@ -1,3 +1,9 @@
//! iroh-gossip swarm management and mainline DHT peer discovery.
//!
//! Derives cryptographic keys from a shared passphrase, manages the gossip
//! subscription, and maintains a DHT heartbeat for zero-configuration peer
//! bootstrapping.
use std::collections::HashSet;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
@@ -5,12 +11,12 @@ use std::time::{Duration, Instant};
use anyhow::{Context, Result};
use bytes::Bytes;
use futures_lite::StreamExt;
use iroh::endpoint::Incoming;
use iroh::{Endpoint, PublicKey, SecretKey};
use futures_lite::StreamExt;
use iroh_gossip::api::Event;
use iroh_gossip::net::{Gossip as GossipActor, GOSSIP_ALPN};
use iroh_gossip::TopicId;
use iroh_gossip::api::Event;
use iroh_gossip::net::{GOSSIP_ALPN, Gossip as GossipActor};
use mainline::async_dht::AsyncDht;
use mainline::{Dht, MutableItem, SigningKey};
use sha2::{Digest, Sha256};
@@ -25,13 +31,15 @@ use crate::packet::GossipFrame;
/// Conservative cap to stay under BEP44's 1000-byte value limit.
const MAX_DHT_PEERS: usize = 27;
/// DHT heartbeat interval (base, before jitter).
/// DHT heartbeat interval when connected (base, before jitter). 5 minutes
/// balances freshness against DHT write amplification — connected nodes already
/// have live gossip links and only need DHT for newcomer discovery.
const DHT_HEARTBEAT_SECS: u64 = 300;
/// Maximum random jitter added to heartbeat interval.
/// Maximum random jitter added to heartbeat interval. Prevents synchronized
/// publish storms when multiple bridges start simultaneously.
const DHT_HEARTBEAT_JITTER_SECS: u64 = 60;
// ── Key derivation ───────────────────────────────────────────────
/// Cryptographic keys derived from a shared passphrase.
@@ -43,29 +51,44 @@ pub struct PassphraseKeys {
impl PassphraseKeys {
/// Derive all keys from a passphrase.
///
/// Intermediate buffers containing the passphrase are zeroized after use to
/// minimize time sensitive material spends in memory.
#[must_use]
pub fn derive(passphrase: &str) -> Self {
// Application-specific prefix to avoid DHT collisions with other apps.
const APP_PREFIX: &str = "donglora-bridge:";
const APP_PREFIX: &[u8] = b"donglora-bridge:";
// Helper: build a zeroized buffer of [prefix | suffix | passphrase].
let keyed_buf = |suffix: &[u8]| -> Vec<u8> {
let mut buf = Vec::with_capacity(APP_PREFIX.len() + suffix.len() + passphrase.len());
buf.extend_from_slice(APP_PREFIX);
buf.extend_from_slice(suffix);
buf.extend_from_slice(passphrase.as_bytes());
buf
};
// Topic ID: blake3 hash of prefixed passphrase.
let topic_bytes = *blake3::hash(format!("{APP_PREFIX}{passphrase}").as_bytes()).as_bytes();
let mut buf = keyed_buf(b"");
let topic_bytes = *blake3::hash(&buf).as_bytes();
let topic_id = TopicId::from_bytes(topic_bytes);
zeroize::Zeroize::zeroize(&mut buf);
// DHT signing key: ed25519 from SHA-256 of prefixed passphrase + "sign".
let sign_seed = Sha256::digest(format!("{APP_PREFIX}sign:{passphrase}").as_bytes());
// DHT signing key: ed25519 from SHA-256 of prefixed passphrase + "sign:".
let mut buf = keyed_buf(b"sign:");
let sign_seed = Sha256::digest(&buf);
let seed_bytes: [u8; 32] = sign_seed.into();
let dht_signing_key = SigningKey::from_bytes(&seed_bytes);
zeroize::Zeroize::zeroize(&mut buf);
// DHT salt: first 20 bytes of SHA-256 of prefixed passphrase + "salt".
let salt_hash = Sha256::digest(format!("{APP_PREFIX}salt:{passphrase}").as_bytes());
// DHT salt: first 20 bytes of SHA-256 of prefixed passphrase + "salt:".
let mut buf = keyed_buf(b"salt:");
let salt_hash = Sha256::digest(&buf);
let mut dht_salt = [0u8; 20];
dht_salt.copy_from_slice(&salt_hash[..20]);
zeroize::Zeroize::zeroize(&mut buf);
Self {
topic_id,
dht_signing_key,
dht_salt,
}
Self { topic_id, dht_signing_key, dht_salt }
}
}
@@ -80,6 +103,7 @@ pub struct SwarmState {
pub last_dht_publish: Option<Instant>,
}
/// Status of the mainline DHT connection.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DhtStatus {
Bootstrapping,
@@ -112,7 +136,9 @@ pub enum GossipEvent {
/// Handle to the gossip swarm. Manages iroh endpoint, gossip subscription, and DHT.
pub struct Gossip {
endpoint: Endpoint,
gossip: GossipActor,
actor: GossipActor,
// std::sync::Mutex (not tokio::sync::Mutex) because locks are never held across
// .await points. This avoids async-aware mutex overhead for brief critical sections.
state: Arc<Mutex<SwarmState>>,
neighbors: Arc<Mutex<HashSet<PublicKey>>>,
dht_signing_key: SigningKey,
@@ -147,7 +173,7 @@ impl Gossip {
info!("iroh endpoint bound, id: {}", my_id.fmt_short());
// Build gossip actor.
let gossip = GossipActor::builder().spawn(endpoint.clone());
let actor = GossipActor::builder().spawn(endpoint.clone());
// Bootstrap from DHT.
let bootstrap_peers = dht_get_peers(keys).await;
@@ -158,10 +184,7 @@ impl Gossip {
}
// Subscribe to gossip topic.
let topic = gossip
.subscribe(keys.topic_id, bootstrap_peers)
.await
.context("subscribing to gossip topic")?;
let topic = actor.subscribe(keys.topic_id, bootstrap_peers).await.context("subscribing to gossip topic")?;
let (sender, receiver) = topic.split();
@@ -182,7 +205,7 @@ impl Gossip {
let handle = Self {
endpoint: endpoint.clone(),
gossip: gossip.clone(),
actor: actor.clone(),
state: state.clone(),
neighbors: neighbors.clone(),
dht_signing_key: keys.dht_signing_key.clone(),
@@ -191,7 +214,7 @@ impl Gossip {
};
// Spawn accept loop.
spawn_accept_loop(endpoint.clone(), gossip.clone(), cancel.clone());
spawn_accept_loop(endpoint.clone(), actor.clone(), cancel.clone());
// Spawn gossip receiver task.
spawn_receiver(receiver, event_tx, state.clone(), neighbor_count.clone(), neighbors.clone(), cancel.clone());
@@ -221,8 +244,9 @@ impl Gossip {
}
/// Snapshot of swarm state for TUI.
#[must_use]
pub fn swarm_state(&self) -> SwarmState {
self.state.lock().unwrap_or_else(|e| e.into_inner()).clone()
self.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner).clone()
}
/// Graceful shutdown. Publishes neighbors (without ourselves) to DHT
@@ -232,7 +256,7 @@ impl Gossip {
let neighbor_ids: Vec<[u8; 32]> = self
.neighbors
.lock()
.unwrap_or_else(|e| e.into_inner())
.unwrap_or_else(std::sync::PoisonError::into_inner)
.iter()
.map(|id| *id.as_bytes())
.collect();
@@ -242,14 +266,9 @@ impl Gossip {
let value = encode_peer_list(&neighbor_ids);
let seq = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.map(|d| d.as_secs().cast_signed())
.unwrap_or(0);
let item = MutableItem::new(
self.dht_signing_key.clone(),
&value,
seq,
Some(self.dht_salt.as_ref()),
);
let item = MutableItem::new(self.dht_signing_key.clone(), &value, seq, Some(self.dht_salt.as_ref()));
match dht.put_mutable(item, None).await {
Ok(_) => info!("shutdown: published {} neighbor(s) to DHT (removed self)", neighbor_ids.len()),
Err(e) => debug!("shutdown: DHT publish failed: {e:#}"),
@@ -257,7 +276,7 @@ impl Gossip {
}
self.cancel.cancel();
let _ = self.gossip.shutdown().await;
let _ = self.actor.shutdown().await;
self.endpoint.close().await;
}
}
@@ -316,18 +335,29 @@ fn spawn_receiver(
debug!("received malformed gossip frame ({} bytes)", msg.content.len());
}
}
// Neighbor count is tracked in both AtomicUsize (for lock-free reads in DHT
// heartbeat) and SwarmState (for TUI snapshots). Brief inconsistency between
// the two is acceptable — the TUI polls at 250ms intervals.
Event::NeighborUp(id) => {
let count = neighbor_count.fetch_add(1, Ordering::Relaxed) + 1;
info!("neighbor up: {} (total: {count})", id.fmt_short());
if let Ok(mut n) = neighbors.lock() { n.insert(id); }
if let Ok(mut s) = state.lock() { s.neighbor_count = count; }
if let Ok(mut n) = neighbors.lock() {
n.insert(id);
}
if let Ok(mut s) = state.lock() {
s.neighbor_count = count;
}
let _ = event_tx.send(GossipEvent::NeighborChanged(count)).await;
}
Event::NeighborDown(id) => {
let count = neighbor_count.fetch_sub(1, Ordering::Relaxed).saturating_sub(1);
info!("neighbor down: {} (total: {count})", id.fmt_short());
if let Ok(mut n) = neighbors.lock() { n.remove(&id); }
if let Ok(mut s) = state.lock() { s.neighbor_count = count; }
if let Ok(mut n) = neighbors.lock() {
n.remove(&id);
}
if let Ok(mut s) = state.lock() {
s.neighbor_count = count;
}
let _ = event_tx.send(GossipEvent::NeighborChanged(count)).await;
}
Event::Lagged => {
@@ -361,11 +391,13 @@ fn spawn_broadcaster(
});
}
/// DHT poll interval while alone (read frequently, publish infrequently).
/// DHT poll interval while alone. 3 seconds is aggressive but reads are cheap
/// (single DHT GET) and fast discovery matters when two nodes start together.
const DHT_ALONE_READ_SECS: u64 = 3;
/// How often a lonely node publishes (merge-only, never destructive).
/// Slower than reads — publishing is heavier and a lonely node only adds itself.
/// 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.
@@ -458,10 +490,7 @@ async fn dht_read_and_join(
) {
if let Some(record) = dht.get_mutable_most_recent(pub_key_bytes, Some(salt)).await {
let peers = decode_peer_list(record.value());
let new_peers: Vec<_> = peers
.into_iter()
.filter(|id| *id != *my_id)
.collect();
let new_peers: Vec<_> = peers.into_iter().filter(|id| *id != *my_id).collect();
if !new_peers.is_empty() {
info!("DHT discovered {} peer(s), feeding to gossip", new_peers.len());
if let Err(e) = gossip_sender.join_peers(new_peers).await {
@@ -483,7 +512,7 @@ async fn dht_publish(
neighbors: &Arc<Mutex<HashSet<PublicKey>>>,
state: &Arc<Mutex<SwarmState>>,
) {
let has_neighbors = neighbors.lock().map(|n| !n.is_empty()).unwrap_or(false);
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);
@@ -518,7 +547,7 @@ async fn dht_publish(
let value = encode_peer_list(&peer_list);
let seq = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.map(|d| d.as_secs().cast_signed())
.unwrap_or(0);
let item = MutableItem::new(signing_key.clone(), &value, seq, Some(salt.as_ref()));
@@ -552,22 +581,24 @@ async fn dht_get_peers(keys: &PassphraseKeys) -> Vec<PublicKey> {
};
let pub_key_bytes = keys.dht_signing_key.verifying_key().to_bytes();
match dht.get_mutable_most_recent(&pub_key_bytes, Some(&keys.dht_salt)).await {
Some(record) => {
dht.get_mutable_most_recent(&pub_key_bytes, Some(&keys.dht_salt)).await.map_or_else(
|| {
debug!("DHT bootstrap: no existing record");
vec![]
},
|record| {
let peers = decode_peer_list(record.value());
debug!("DHT bootstrap: found {} peer(s)", peers.len());
peers
}
None => {
debug!("DHT bootstrap: no existing record");
vec![]
}
}
},
)
}
/// Encode a list of 32-byte peer IDs into bytes for DHT storage.
/// Format: [count: u16 LE] [id_0: 32B] [id_1: 32B] ...
fn encode_peer_list(peers: &[[u8; 32]]) -> Vec<u8> {
/// Format: `[count: u16 LE] [id_0: 32B] [id_1: 32B] ...`
#[must_use]
pub fn encode_peer_list(peers: &[[u8; 32]]) -> Vec<u8> {
#[allow(clippy::cast_possible_truncation)] // MAX_DHT_PEERS (27) fits in u16
let count = peers.len().min(MAX_DHT_PEERS) as u16;
let mut buf = Vec::with_capacity(2 + (count as usize) * 32);
buf.extend_from_slice(&count.to_le_bytes());
@@ -578,7 +609,8 @@ fn encode_peer_list(peers: &[[u8; 32]]) -> Vec<u8> {
}
/// Decode a peer list from DHT record value.
fn decode_peer_list(data: &[u8]) -> Vec<PublicKey> {
#[must_use]
pub fn decode_peer_list(data: &[u8]) -> Vec<PublicKey> {
if data.len() < 2 {
return vec![];
}
@@ -656,4 +688,82 @@ mod tests {
let decoded = decode_peer_list(&data);
assert_eq!(decoded.len(), 1);
}
#[test]
fn dht_status_display() {
assert_eq!(format!("{}", DhtStatus::Bootstrapping), "Bootstrapping");
assert_eq!(format!("{}", DhtStatus::Ready), "Ready");
assert_eq!(format!("{}", DhtStatus::PublishFailed), "Failed");
}
#[test]
fn decode_peer_list_exact_boundary() {
// Build data with count=2 and exactly 2 + 32 + 32 = 66 bytes.
// Both peers should decode successfully.
let sk0 = iroh::SecretKey::generate(&mut rand::rng());
let sk1 = iroh::SecretKey::generate(&mut rand::rng());
let id0 = *sk0.public().as_bytes();
let id1 = *sk1.public().as_bytes();
let mut data = vec![2, 0]; // count = 2 (u16 LE)
data.extend_from_slice(&id0);
data.extend_from_slice(&id1);
assert_eq!(data.len(), 66);
let decoded = decode_peer_list(&data);
assert_eq!(decoded.len(), 2);
assert_eq!(*decoded[0].as_bytes(), id0);
assert_eq!(*decoded[1].as_bytes(), id1);
}
#[test]
fn decode_peer_list_one_byte_short() {
// Build data with count=2 but only 2 + 32 + 31 = 65 bytes.
// First peer decodes, second is truncated and skipped.
let sk0 = iroh::SecretKey::generate(&mut rand::rng());
let sk1 = iroh::SecretKey::generate(&mut rand::rng());
let id0 = *sk0.public().as_bytes();
let id1 = *sk1.public().as_bytes();
let mut data = vec![2, 0]; // count = 2 (u16 LE)
data.extend_from_slice(&id0);
data.extend_from_slice(&id1[..31]); // only 31 bytes of second peer
assert_eq!(data.len(), 65);
let decoded = decode_peer_list(&data);
assert_eq!(decoded.len(), 1);
assert_eq!(*decoded[0].as_bytes(), id0);
}
#[test]
fn decode_peer_list_count_header_only() {
// Data with exactly 2 bytes (count=0). Must not be rejected by the length check.
// Kills mutant: `data.len() < 2` → `data.len() <= 2`.
let data = [0u8, 0]; // count = 0
let decoded = decode_peer_list(&data);
assert!(decoded.is_empty());
// Data with 2 bytes but count=1 and no peer data — decodes 0 (truncated).
let data = [1u8, 0]; // count = 1 but no peer bytes
let decoded = decode_peer_list(&data);
assert!(decoded.is_empty());
}
mod proptests {
use super::*;
use proptest::prelude::*;
proptest! {
#[test]
fn peer_list_roundtrip_prop(count in 0usize..=30) {
let keys: Vec<[u8; 32]> = (0..count)
.map(|_| *iroh::SecretKey::generate(&mut rand::rng()).public().as_bytes())
.collect();
let encoded = encode_peer_list(&keys);
let decoded = decode_peer_list(&encoded);
let expected = count.min(MAX_DHT_PEERS);
prop_assert_eq!(decoded.len(), expected);
}
}
}
}
+18
View File
@@ -0,0 +1,18 @@
//! Peer-to-peer `LoRa` bridge connecting `DongLoRa` USB dongles across the internet.
//!
//! Uses iroh-gossip for swarm-based packet broadcast and mainline DHT for
//! zero-configuration peer discovery. Nodes sharing a passphrase automatically
//! discover each other and relay radio packets through a gossip swarm.
pub mod config;
pub mod gossip;
pub mod packet;
pub mod radio;
pub mod rate_limit;
pub mod router;
pub mod setup;
// TUI is public only so its types are visible to the binary; not intended for
// external consumption.
#[doc(hidden)]
pub mod tui;
+40 -64
View File
@@ -1,11 +1,4 @@
mod config;
mod gossip;
mod packet;
mod radio;
mod rate_limit;
mod router;
mod setup;
mod tui;
use donglora_bridge::{config, gossip, radio, rate_limit, router, setup, tui};
use std::sync::Arc;
use std::time::Instant;
@@ -43,6 +36,7 @@ enum Commands {
}
#[tokio::main]
#[allow(clippy::too_many_lines)]
async fn main() -> Result<()> {
let cli = Cli::parse();
@@ -53,12 +47,8 @@ async fn main() -> Result<()> {
};
// Config subcommand: run wizard with existing config as defaults.
if let Some(Commands::Config) = cli.command {
let existing = if config_path.exists() {
config::load_config(&config_path).ok()
} else {
None
};
if matches!(cli.command, Some(Commands::Config)) {
let existing = if config_path.exists() { config::load_config(&config_path).ok() } else { None };
return setup::run_wizard(&config_path, existing.as_ref());
}
@@ -69,39 +59,11 @@ async fn main() -> Result<()> {
}
// Load config.
let cfg = config::load_config(&config_path)
.with_context(|| format!("loading config from {}", config_path.display()))?;
let cfg =
config::load_config(&config_path).with_context(|| format!("loading config from {}", config_path.display()))?;
// Set up logging.
if cli.log_only {
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
)
.init();
} else {
let log_path = cfg
.bridge
.log_file
.clone()
.unwrap_or_else(|| config::default_log_path().unwrap_or_else(|_| "/tmp/donglora-bridge.log".into()));
if let Some(parent) = log_path.parent() {
std::fs::create_dir_all(parent).ok();
}
let file_appender = tracing_appender::rolling::never(
log_path.parent().unwrap_or(std::path::Path::new("/tmp")),
log_path.file_name().unwrap_or(std::ffi::OsStr::new("donglora-bridge.log")),
);
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
)
.with_writer(file_appender)
.with_ansi(false)
.init();
}
setup_logging(cli.log_only, &cfg);
// Ephemeral identity — fresh keypair each launch.
let secret_key = SecretKey::generate(&mut rand::rng());
@@ -126,25 +88,22 @@ async fn main() -> Result<()> {
info!("rate limiter: {:.1} pps", rate_limiter.rate_pps());
// Create gossip network.
let (swarm, gossip_event_rx, gossip_frame_tx) =
gossip::Gossip::new(secret_key, &keys).await?;
let (swarm, gossip_event_rx, gossip_frame_tx) = gossip::Gossip::new(secret_key, &keys).await?;
let swarm = Arc::new(swarm);
// Start radio.
let (radio_event_rx, radio_tx) = radio::spawn(radio_config, cfg.radio.port.clone());
let (radio_event_rx, radio_tx) = radio::spawn(radio_config, cfg.radio.port.clone())?;
// Stats + log channel + radio config watch.
let stats = Arc::new(Stats::default());
let (log_tx, log_rx) = mpsc::channel(512);
let (config_watch_tx, config_watch_rx) = tokio::sync::watch::channel(
router::RadioConfigInfo {
let (config_watch_tx, config_watch_rx) = tokio::sync::watch::channel(router::RadioConfigInfo {
active: radio_config,
requested: radio_config,
source: radio::ConfigSource::Ours,
device: String::new(),
connected: false,
},
);
});
// Cancellation.
let cancel = CancellationToken::new();
@@ -180,9 +139,7 @@ async fn main() -> Result<()> {
{
let cancel = cancel.clone();
tokio::spawn(async move {
let sigterm = tokio::signal::unix::signal(
tokio::signal::unix::SignalKind::terminate(),
);
let sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate());
match sigterm {
Ok(mut sigterm) => {
tokio::select! {
@@ -205,15 +162,7 @@ async fn main() -> Result<()> {
info!("running in headless mode");
cancel.cancelled().await;
} else {
let terminal = tui::run(
config_watch_rx,
swarm.clone(),
stats,
log_rx,
cancel.clone(),
start_time,
)
.await;
let terminal = tui::run(config_watch_rx, swarm.clone(), stats, log_rx, cancel.clone(), start_time);
cancel.cancel();
swarm.shutdown().await;
@@ -232,3 +181,30 @@ async fn main() -> Result<()> {
Ok(())
}
fn setup_logging(log_only: bool, cfg: &config::Config) {
let env_filter = || {
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info"))
};
if log_only {
tracing_subscriber::fmt().with_env_filter(env_filter()).init();
} else {
let log_path = cfg
.bridge
.log_file
.clone()
.unwrap_or_else(|| config::default_log_path().unwrap_or_else(|_| "/tmp/donglora-bridge.log".into()));
if let Some(parent) = log_path.parent()
&& let Err(e) = std::fs::create_dir_all(parent)
{
eprintln!("warning: failed to create log directory {}: {e}", parent.display());
}
let file_appender = tracing_appender::rolling::never(
log_path.parent().unwrap_or_else(|| std::path::Path::new("/tmp")),
log_path.file_name().unwrap_or_else(|| std::ffi::OsStr::new("donglora-bridge.log")),
);
tracing_subscriber::fmt().with_env_filter(env_filter()).with_writer(file_appender).with_ansi(false).init();
}
}
+100 -17
View File
@@ -1,7 +1,12 @@
/// Gossip frame header size: sender(32) + rssi(2) + snr(2) + payload_len(2) = 38
//! Wire format definitions for radio packets and gossip frames.
//!
//! Includes SNR quality grading against `LoRa` demodulation floors and blake3
//! content hashing for deduplication.
/// Gossip frame header size: `sender(32) + rssi(2) + snr(2) + payload_len(2) = 38`
const FRAME_HEADER_SIZE: usize = 38;
/// SNR quality grade for a received LoRa packet.
/// SNR quality grade for a received `LoRa` packet.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Grade {
/// SNR well above demodulation floor (margin >= 3 dB).
@@ -16,7 +21,8 @@ pub enum Grade {
impl Grade {
/// Returns true if the packet is reliable enough to forward.
pub fn should_forward(self) -> bool {
#[must_use]
pub const fn should_forward(self) -> bool {
matches!(self, Self::Good | Self::Marginal)
}
}
@@ -34,6 +40,7 @@ impl std::fmt::Display for Grade {
/// Grade a received packet's SNR against the demodulation floor for the given
/// spreading factor.
#[must_use]
pub fn snr_grade(snr: i16, sf: u8) -> Grade {
if !(-32..=32).contains(&snr) {
return Grade::Invalid;
@@ -62,33 +69,31 @@ pub struct RadioPacket {
/// Wire format: `[sender: 32B] [rssi: i16 LE] [snr: i16 LE] [payload_len: u16 LE] [payload: NB]`
#[derive(Debug, Clone)]
pub struct GossipFrame {
/// EndpointId (PublicKey bytes) of the bridge that first heard this packet on radio.
/// `EndpointId` (`PublicKey` bytes) of the bridge that first heard this packet on radio.
pub sender: [u8; 32],
/// RSSI at the originating bridge (informational).
pub rssi: i16,
/// SNR at the originating bridge (informational).
pub snr: i16,
/// Raw LoRa payload.
/// Raw `LoRa` payload.
pub payload: Vec<u8>,
}
impl GossipFrame {
/// Create a new gossip frame from a public key and radio packet data.
#[must_use]
pub fn new(sender: &iroh::PublicKey, rssi: i16, snr: i16, payload: Vec<u8>) -> Self {
Self {
sender: *sender.as_bytes(),
rssi,
snr,
payload,
}
Self { sender: *sender.as_bytes(), rssi, snr, payload }
}
/// Encode this frame into bytes for gossip broadcast.
#[must_use]
pub fn encode(&self) -> Vec<u8> {
let mut buf = Vec::with_capacity(FRAME_HEADER_SIZE + self.payload.len());
buf.extend_from_slice(&self.sender);
buf.extend_from_slice(&self.rssi.to_le_bytes());
buf.extend_from_slice(&self.snr.to_le_bytes());
#[allow(clippy::cast_possible_truncation)] // LoRa payloads never exceed 256 bytes
let len = self.payload.len() as u16;
buf.extend_from_slice(&len.to_le_bytes());
buf.extend_from_slice(&self.payload);
@@ -96,6 +101,7 @@ impl GossipFrame {
}
/// Decode a gossip frame from bytes received from the swarm.
#[must_use]
pub fn decode(data: &[u8]) -> Option<Self> {
if data.len() < FRAME_HEADER_SIZE {
return None;
@@ -107,21 +113,18 @@ impl GossipFrame {
if data.len() != FRAME_HEADER_SIZE + payload_len {
return None;
}
Some(Self {
sender,
rssi,
snr,
payload: data[FRAME_HEADER_SIZE..].to_vec(),
})
Some(Self { sender, rssi, snr, payload: data[FRAME_HEADER_SIZE..].to_vec() })
}
}
/// Compute a 32-byte blake3 content hash for deduplication.
#[must_use]
pub fn content_hash(payload: &[u8]) -> [u8; 32] {
*blake3::hash(payload).as_bytes()
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
@@ -156,6 +159,17 @@ mod tests {
assert!(GossipFrame::decode(&[0u8; 10]).is_none());
}
#[test]
fn gossip_frame_decode_empty_payload() {
// Exactly FRAME_HEADER_SIZE (38) bytes with payload_len=0 must decode.
// Kills mutant: `data.len() < 38` → `data.len() <= 38`.
let mut data = vec![0u8; 38];
data[36..38].copy_from_slice(&0u16.to_le_bytes()); // payload_len = 0
let frame = GossipFrame::decode(&data);
assert!(frame.is_some(), "38-byte frame with 0-length payload must decode");
assert!(frame.unwrap().payload.is_empty());
}
#[test]
fn gossip_frame_reject_wrong_length() {
// Header says payload_len=5 but only 3 bytes follow.
@@ -164,6 +178,43 @@ mod tests {
assert!(GossipFrame::decode(&data).is_none());
}
#[test]
fn should_forward_true_for_good_and_marginal() {
assert!(Grade::Good.should_forward());
assert!(Grade::Marginal.should_forward());
}
#[test]
fn should_forward_false_for_unreliable_and_invalid() {
assert!(!Grade::Unreliable.should_forward());
assert!(!Grade::Invalid.should_forward());
}
#[test]
fn grade_display() {
assert_eq!(format!("{}", Grade::Good), "GOOD");
assert_eq!(format!("{}", Grade::Marginal), "MARG");
assert_eq!(format!("{}", Grade::Unreliable), "UNRL");
assert_eq!(format!("{}", Grade::Invalid), "INVL");
}
#[test]
fn snr_grade_exact_boundaries() {
// SF12: min_snr = -2.5 * (12 - 4) = -20.0
//
// At snr=-20, margin = -20 - (-20.0) = 0.0 exactly.
// The code uses `margin < 0.0`, so 0.0 is NOT < 0.0 → Marginal (not Unreliable).
assert_eq!(snr_grade(-20, 12), Grade::Marginal);
// One below: snr=-21, margin = -1.0 → Unreliable.
assert_eq!(snr_grade(-21, 12), Grade::Unreliable);
// At snr=-17, margin = -17 - (-20.0) = 3.0 exactly.
// The code uses `margin < 3.0`, so 3.0 is NOT < 3.0 → Good (not Marginal).
assert_eq!(snr_grade(-17, 12), Grade::Good);
// One below: snr=-18, margin = 2.0 → Marginal.
assert_eq!(snr_grade(-18, 12), Grade::Marginal);
}
#[test]
fn content_hash_deterministic() {
let h1 = content_hash(b"hello");
@@ -172,4 +223,36 @@ mod tests {
let h3 = content_hash(b"world");
assert_ne!(h1, h3);
}
mod proptests {
use super::*;
use proptest::prelude::*;
proptest! {
#[test]
fn gossip_frame_roundtrip_prop(
rssi in -200i16..=200i16,
snr in -32i16..=32i16,
payload in proptest::collection::vec(any::<u8>(), 0..256),
) {
let key = iroh::SecretKey::generate(&mut rand::rng());
let frame = GossipFrame::new(&key.public(), rssi, snr, payload.clone());
let encoded = frame.encode();
let decoded = GossipFrame::decode(&encoded).unwrap();
prop_assert_eq!(decoded.rssi, rssi);
prop_assert_eq!(decoded.snr, snr);
prop_assert_eq!(decoded.payload, payload);
}
#[test]
fn content_hash_deterministic_prop(data in proptest::collection::vec(any::<u8>(), 0..1024)) {
prop_assert_eq!(content_hash(&data), content_hash(&data));
}
#[test]
fn snr_grade_never_panics(snr in i16::MIN..=i16::MAX, sf in 5u8..=12u8) {
let _ = snr_grade(snr, sf);
}
}
}
}
+168 -48
View File
@@ -1,6 +1,11 @@
//! `DongLoRa` hardware communication on a blocking thread.
//!
//! Handles connection, config negotiation, RX/TX, liveness pings, and
//! exponential backoff reconnection.
use std::time::Duration;
use anyhow::Result;
use anyhow::{Context, Result};
use donglora_client::{RadioConfig, Response, Transport};
use tokio::sync::mpsc;
use tracing::{debug, error, info, warn};
@@ -29,21 +34,19 @@ pub enum RadioEvent {
/// Spawn the blocking radio thread.
///
/// Returns a receiver for radio events and a sender for TX payloads.
pub fn spawn(
config: RadioConfig,
port: Option<String>,
) -> (mpsc::Receiver<RadioEvent>, mpsc::Sender<Vec<u8>>) {
pub fn spawn(config: RadioConfig, port: Option<String>) -> Result<(mpsc::Receiver<RadioEvent>, mpsc::Sender<Vec<u8>>)> {
let (event_tx, event_rx) = mpsc::channel::<RadioEvent>(256);
let (tx_send, tx_recv) = mpsc::channel::<Vec<u8>>(64);
std::thread::Builder::new()
.name("radio".into())
.spawn(move || radio_loop(config, port, event_tx, tx_recv))
.ok();
.context("failed to spawn radio thread")?;
(event_rx, tx_send)
Ok((event_rx, tx_send))
}
#[allow(clippy::needless_pass_by_value)] // values moved across thread boundary in spawn()
fn radio_loop(
config: RadioConfig,
port: Option<String>,
@@ -56,7 +59,7 @@ fn radio_loop(
loop {
let started = std::time::Instant::now();
match connect_and_run(&config, &port, &event_tx, &mut tx_recv) {
match connect_and_run(&config, port.as_deref(), &event_tx, &mut tx_recv) {
Ok(()) => {
info!("radio thread exiting");
return;
@@ -81,16 +84,16 @@ fn radio_loop(
fn connect_and_run(
config: &RadioConfig,
port: &Option<String>,
port: Option<&str>,
event_tx: &mpsc::Sender<RadioEvent>,
tx_recv: &mut mpsc::Receiver<Vec<u8>>,
) -> Result<()> {
// Long initial timeout: the mux serializes commands through the dongle,
// so our GetConfig might have to wait behind other clients' commands.
// so our `GetConfig` might have to wait behind other clients' commands.
let timeout = Duration::from_secs(10);
info!("connecting to dongle...");
let (mut client, device) = if let Some(port) = &port {
let (mut client, device) = if let Some(port) = port {
let c = donglora_client::connect(Some(port), timeout)?;
(c, shorten_path(port))
} else {
@@ -132,9 +135,7 @@ fn connect_and_run(
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))?;
client.transport_mut().set_timeout(Duration::from_millis(100))?;
info!("[radio] entering main loop");
// Liveness: ping the mux periodically when idle to detect a dead connection.
@@ -159,19 +160,13 @@ fn connect_and_run(
last_activity = std::time::Instant::now();
let grade = snr_grade(snr, active_config.sf);
if grade.should_forward() {
debug!(
"RX {} bytes rssi={rssi} snr={snr} grade={grade}",
payload.len()
);
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(());
}
} else {
debug!(
"RX drop {} bytes rssi={rssi} snr={snr} grade={grade}",
payload.len()
);
debug!("RX drop {} bytes rssi={rssi} snr={snr} grade={grade}", payload.len());
}
}
Ok(Some(_)) => {
@@ -208,13 +203,13 @@ fn connect_and_run(
/// Negotiate radio config with the dongle/mux.
///
/// Strategy:
/// 1. GetConfig first to read what's active (caches the result).
/// 2. If it matches our desired config, use it — no SetConfig needed.
/// 3. If it differs, only try SetConfig if we haven't been rejected before.
/// 4. If SetConfig is rejected, set the flag IMMEDIATELY (before the mux can
/// disconnect us and lose this state), then return the cached GetConfig result.
/// 1. `GetConfig` first to read what's active (caches the result).
/// 2. If it matches our desired config, use it — no `SetConfig` needed.
/// 3. If it differs, only try `SetConfig` if we haven't been rejected before.
/// 4. If `SetConfig` is rejected, set the flag IMMEDIATELY (before the mux can
/// disconnect us and lose this state), then return the cached `GetConfig` result.
///
/// On reconnect, the flag ensures we never retry SetConfig.
/// On reconnect, the flag ensures we never retry `SetConfig`.
fn negotiate_config<T: Transport>(
client: &mut donglora_client::Client<T>,
desired: &RadioConfig,
@@ -251,10 +246,7 @@ fn negotiate_config<T: Transport>(
/// Shorten a device path to just the filename for display.
fn shorten_path(path: &str) -> String {
std::path::Path::new(path)
.file_name()
.map(|f| f.to_string_lossy().into_owned())
.unwrap_or_else(|| path.to_string())
std::path::Path::new(path).file_name().map_or_else(|| path.to_string(), |f| f.to_string_lossy().into_owned())
}
fn configs_match(a: &RadioConfig, b: &RadioConfig) -> bool {
@@ -269,18 +261,13 @@ fn configs_match(a: &RadioConfig, b: &RadioConfig) -> bool {
}
fn is_timeout_error(e: &anyhow::Error) -> bool {
if let Some(io_err) = e.downcast_ref::<std::io::Error>() {
matches!(
io_err.kind(),
std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
)
} else {
false
}
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.
pub fn format_bandwidth(bw: donglora_client::Bandwidth) -> &'static str {
#[must_use]
pub const fn format_bandwidth(bw: donglora_client::Bandwidth) -> &'static str {
match bw {
donglora_client::Bandwidth::Khz7 => "7.8 kHz",
donglora_client::Bandwidth::Khz10 => "10.4 kHz",
@@ -295,23 +282,156 @@ pub fn format_bandwidth(bw: donglora_client::Bandwidth) -> &'static str {
}
}
/// Format a RadioConfig for display.
/// Format a `RadioConfig` for display.
#[must_use]
pub fn format_radio_config(config: &RadioConfig) -> String {
let freq_mhz = config.freq_hz as f64 / 1_000_000.0;
let freq_mhz = f64::from(config.freq_hz) / 1_000_000.0;
let bw = format_bandwidth(config.bw);
let power = if config.tx_power_dbm == donglora_client::TX_POWER_MAX {
"max".to_string()
} else {
format!("{} dBm", config.tx_power_dbm)
};
let preamble = if config.preamble_len == 0 {
16
} else {
config.preamble_len
};
let preamble = if config.preamble_len == 0 { 16 } else { config.preamble_len };
let cad = if config.cad != 0 { "on" } else { "off" };
format!(
"{freq_mhz:.3}MHz SF{} BW{bw} CR4/{} SW0x{:04X} TX:{power} Pre:{preamble} CAD:{cad}",
config.sf, config.cr, config.sync_word
)
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
fn default_config() -> RadioConfig {
RadioConfig {
freq_hz: 910_525_000,
bw: donglora_client::Bandwidth::Khz62,
sf: 7,
cr: 5,
sync_word: 0x3444,
tx_power_dbm: 22,
preamble_len: 16,
cad: 1,
}
}
#[test]
fn configs_match_identical() {
let a = default_config();
let b = default_config();
assert!(configs_match(&a, &b));
}
#[test]
fn configs_match_differs_each_field() {
let base = default_config();
let mut c = base;
c.freq_hz = 915_000_000;
assert!(!configs_match(&base, &c));
let mut c = base;
c.sf = 12;
assert!(!configs_match(&base, &c));
let mut c = base;
c.cr = 8;
assert!(!configs_match(&base, &c));
let mut c = base;
c.sync_word = 0x1234;
assert!(!configs_match(&base, &c));
let mut c = base;
c.tx_power_dbm = 10;
assert!(!configs_match(&base, &c));
let mut c = base;
c.preamble_len = 8;
assert!(!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]
fn format_bandwidth_all_variants() {
assert_eq!(format_bandwidth(donglora_client::Bandwidth::Khz7), "7.8 kHz");
assert_eq!(format_bandwidth(donglora_client::Bandwidth::Khz125), "125 kHz");
assert_eq!(format_bandwidth(donglora_client::Bandwidth::Khz500), "500 kHz");
}
#[test]
fn format_radio_config_typical() {
let cfg = default_config();
let s = format_radio_config(&cfg);
assert!(s.contains("910.525MHz"));
assert!(s.contains("SF7"));
assert!(s.contains("CR4/5"));
assert!(s.contains("0x3444"));
assert!(s.contains("CAD:on"), "cad=1 should show CAD:on: {s}");
}
#[test]
fn format_radio_config_cad_off() {
let mut cfg = default_config();
cfg.cad = 0;
let s = format_radio_config(&cfg);
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"));
}
}
+207 -38
View File
@@ -1,10 +1,15 @@
//! `LoRa` air-time-aware token bucket rate limiter.
//!
//! Calculates transmit rate from the Semtech `LoRa` air time formula and
//! enforces it with a configurable burst allowance.
use std::time::{Duration, Instant};
use donglora_client::Bandwidth;
/// Token-bucket rate limiter for radio transmissions.
///
/// Calculates a safe transmit rate from the LoRa radio configuration (spreading factor,
/// Calculates a safe transmit rate from the `LoRa` radio configuration (spreading factor,
/// bandwidth, coding rate) and enforces it with a token bucket that allows short bursts.
pub struct RateLimiter {
tokens: f64,
@@ -13,43 +18,33 @@ pub struct RateLimiter {
last_refill: Instant,
}
/// Default burst size (max tokens).
/// Default burst size (max tokens). Allows short bursts of 3 back-to-back packets
/// to handle bursty mesh traffic without triggering rate limiting on every packet.
const DEFAULT_BURST: f64 = 3.0;
/// Reference payload size (bytes) used to estimate air time for rate calculation.
/// 50 bytes is a representative `MeshCore` packet size (header + small payload).
const REFERENCE_PAYLOAD_BYTES: usize = 50;
/// Target duty cycle fraction (0.5 = 50%).
/// Target duty cycle fraction. 50% leaves headroom for other radio traffic sharing
/// the channel while still forwarding bridge traffic aggressively.
const TARGET_DUTY_CYCLE: f64 = 0.5;
impl RateLimiter {
/// Create a rate limiter derived from radio config.
///
/// If `override_pps` is `Some`, use that rate directly instead of calculating.
pub fn from_radio_config(
sf: u8,
bw: Bandwidth,
cr: u8,
preamble: u16,
override_pps: Option<f64>,
) -> Self {
let refill_rate = if let Some(pps) = override_pps {
pps.max(0.01)
} else {
#[must_use]
pub fn from_radio_config(sf: u8, bw: Bandwidth, cr: u8, preamble: u16, override_pps: Option<f64>) -> Self {
let refill_rate = override_pps.map_or_else(
|| {
let air_time = lora_air_time(REFERENCE_PAYLOAD_BYTES, sf, bw, cr, preamble);
let secs = air_time.as_secs_f64();
if secs > 0.0 {
TARGET_DUTY_CYCLE / secs
} else {
10.0 // fallback
}
};
Self {
tokens: DEFAULT_BURST,
max_tokens: DEFAULT_BURST,
refill_rate,
last_refill: Instant::now(),
}
if secs > 0.0 { TARGET_DUTY_CYCLE / secs } else { 10.0 }
},
|pps| pps.max(0.01),
);
Self { tokens: DEFAULT_BURST, max_tokens: DEFAULT_BURST, refill_rate, last_refill: Instant::now() }
}
/// Try to consume one token. Returns `true` if the transmission is allowed.
@@ -64,19 +59,20 @@ impl RateLimiter {
}
/// Current rate in packets per second.
pub fn rate_pps(&self) -> f64 {
#[must_use]
pub const fn rate_pps(&self) -> f64 {
self.refill_rate
}
fn refill(&mut self) {
let now = Instant::now();
let elapsed = now.duration_since(self.last_refill).as_secs_f64();
self.tokens = (self.tokens + elapsed * self.refill_rate).min(self.max_tokens);
self.tokens = elapsed.mul_add(self.refill_rate, self.tokens).min(self.max_tokens);
self.last_refill = now;
}
}
/// Estimate LoRa packet air time using the Semtech SX1276 formula.
/// Estimate `LoRa` packet air time using the Semtech `LoRa` time-on-air formula.
///
/// Parameters:
/// - `payload_bytes`: payload size in bytes
@@ -84,28 +80,27 @@ impl RateLimiter {
/// - `bw`: bandwidth
/// - `cr`: coding rate denominator (5–8, meaning 4/5 to 4/8)
/// - `preamble`: preamble length in symbols (0 = firmware default of 16)
#[must_use]
pub fn lora_air_time(payload_bytes: usize, sf: u8, bw: Bandwidth, cr: u8, preamble: u16) -> Duration {
let sf_f = f64::from(sf);
let bw_hz = bandwidth_hz(bw);
let preamble_symbols = if preamble == 0 { 16.0 } else { f64::from(preamble) };
// Symbol time in seconds.
let t_sym = 2.0_f64.powf(sf_f) / bw_hz;
let t_sym = sf_f.exp2() / bw_hz;
// Preamble time.
let t_preamble = (preamble_symbols + 4.25) * t_sym;
// Low data rate optimize: enabled when SF >= 11 and BW <= 125 kHz.
let de: f64 = if sf >= 11 && bw_hz <= 125_000.0 {
1.0
} else {
0.0
};
let de: f64 = if sf >= 11 && bw_hz <= 125_000.0 { 1.0 } else { 0.0 };
// Payload symbol count (Semtech formula with CRC enabled, explicit header).
let numerator = 8.0 * payload_bytes as f64 - 4.0 * sf_f + 28.0 + 16.0;
let denominator = 4.0 * (sf_f - 2.0 * de);
let n_payload = 8.0 + (numerator / denominator).ceil().max(0.0) * f64::from(cr);
#[allow(clippy::cast_precision_loss)] // payload_bytes is always < 256
let payload_f = payload_bytes as f64;
let numerator = 8.0f64.mul_add(payload_f, (-4.0f64).mul_add(sf_f, 44.0));
let denominator = 4.0f64.mul_add(sf_f, -8.0 * de);
let n_payload = (numerator / denominator).ceil().max(0.0).mul_add(f64::from(cr), 8.0);
let t_payload = n_payload * t_sym;
let total = t_preamble + t_payload;
@@ -113,7 +108,7 @@ pub fn lora_air_time(payload_bytes: usize, sf: u8, bw: Bandwidth, cr: u8, preamb
Duration::from_secs_f64(total)
}
fn bandwidth_hz(bw: Bandwidth) -> f64 {
const fn bandwidth_hz(bw: Bandwidth) -> f64 {
match bw {
Bandwidth::Khz7 => 7_800.0,
Bandwidth::Khz10 => 10_400.0,
@@ -171,4 +166,178 @@ mod tests {
let t_explicit = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16);
assert_eq!(t_default, t_explicit);
}
// ── Mutant-killing tests for from_radio_config (items 1-5) ────────
#[test]
fn from_radio_config_calculates_reasonable_rate() {
// SF7/BW125/CR5/preamble16 with 50-byte reference payload:
// 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 rate = rl.rate_pps();
// 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).
assert!(rate > 0.1, "rate too low: {rate}");
assert!(rate < 100.0, "rate too high: {rate}");
// Specifically verify it is NOT the 10.0 fallback, proving secs > 0.0 was true.
assert!((rate - 10.0).abs() > 0.5, "rate should not be the 10.0 fallback: {rate}");
// Tight check around the expected ~4.73 pps.
assert!(rate > 4.0, "rate should be near 4.73: {rate}");
assert!(rate < 5.5, "rate should be near 4.73: {rate}");
}
#[test]
fn from_radio_config_sf12_rate_much_slower() {
// 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 rate = rl.rate_pps();
assert!(rate > 0.1, "rate too low: {rate}");
assert!(rate < 0.5, "rate too high: {rate}");
// Definitely not 10.0 fallback.
assert!((rate - 10.0).abs() > 1.0, "rate should not be the fallback: {rate}");
}
// ── Mutant-killing tests for refill (item 6) ───────────────────
#[test]
fn rate_limiter_refills_after_waiting() {
// 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));
// Exhaust all burst tokens.
assert!(rl.try_acquire());
assert!(rl.try_acquire());
assert!(rl.try_acquire());
assert!(!rl.try_acquire(), "burst should be exhausted");
// Sleep enough for at least 1 token to refill (10 pps -> 100ms per token).
std::thread::sleep(Duration::from_millis(200));
// After sleeping, refill should have added tokens, so acquire succeeds.
assert!(rl.try_acquire(), "should succeed after refill");
}
// ── Mutant-killing tests for lora_air_time (items 7-14) ────────
#[test]
fn air_time_exact_sf7_bw125() {
// SF7, BW125kHz, CR4/5, preamble=16, payload=50 bytes.
// Expected: 105.728 ms (computed from Semtech formula).
let t = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16);
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");
}
#[test]
fn air_time_exact_sf12_bw125() {
// SF12, BW125kHz, CR4/5, preamble=16, payload=50 bytes.
// Expected: 2564.096 ms.
let t = lora_air_time(50, 12, Bandwidth::Khz125, 5, 16);
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");
}
#[test]
fn air_time_ldro_sf11_bw125_vs_sf10_bw125() {
// SF>=11 && BW<=125kHz triggers LDRO. SF10 does not.
// 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 t10 = lora_air_time(50, 10, Bandwidth::Khz125, 5, 16);
let ms11 = t11.as_secs_f64() * 1000.0;
let ms10 = t10.as_secs_f64() * 1000.0;
// Verify exact values.
assert!((ms11 - 1445.888).abs() < 0.01, "SF11/BW125 should be ~1445.888ms, got {ms11:.3}ms");
assert!((ms10 - 681.984).abs() < 0.01, "SF10/BW125 should be ~681.984ms, got {ms10:.3}ms");
// SF11 must be more than double SF10 due to LDRO + SF increase.
// (Kills mutant: replace >= with < in sf >= 11 check.)
assert!(t11 > t10 * 2, "SF11 with LDRO should be >2x SF10");
}
#[test]
fn air_time_ldro_not_triggered_high_bw() {
// SF11/BW250: LDRO should NOT be active (BW > 125kHz).
// Expected: 641.024 ms (de=0).
let t11_250 = lora_air_time(50, 11, Bandwidth::Khz250, 5, 16);
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");
// Compare with SF11/BW125 which HAS LDRO.
let t11_125 = lora_air_time(50, 11, Bandwidth::Khz125, 5, 16);
// Even accounting for halved BW doubling the base time, the LDRO effect
// 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)");
}
#[test]
fn air_time_preamble_scaling() {
// Changing preamble by N symbols changes air time by N * t_sym.
// t_sym for SF7/BW125 = 2^7 / 125000 = 1.024 ms.
let t8 = lora_air_time(50, 7, Bandwidth::Khz125, 5, 8);
let t16 = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16);
let t32 = lora_air_time(50, 7, Bandwidth::Khz125, 5, 32);
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 t_sym = 128.0 / 125_000.0; // 1.024 ms
// 8 -> 16: 8 symbols difference.
let expected_8 = 8.0f64.mul_add(t_sym, -diff_8_to_16).abs();
assert!(expected_8 < 1e-9, "preamble 8->16 should add 8*t_sym, got {diff_8_to_16:.6}s");
// 16 -> 32: 16 symbols difference.
let expected_16 = 16.0f64.mul_add(t_sym, -diff_16_to_32).abs();
assert!(expected_16 < 1e-9, "preamble 16->32 should add 16*t_sym, got {diff_16_to_32:.6}s");
}
#[test]
fn air_time_sensible_range_all_sf_bw() {
// Verify reasonable bounds for all SF/BW combos with 50-byte payload.
// This catches gross formula errors (e.g. division replaced with modulo).
let bandwidths = [
Bandwidth::Khz7,
Bandwidth::Khz10,
Bandwidth::Khz15,
Bandwidth::Khz20,
Bandwidth::Khz31,
Bandwidth::Khz41,
Bandwidth::Khz62,
Bandwidth::Khz125,
Bandwidth::Khz250,
Bandwidth::Khz500,
];
for sf in 7..=12u8 {
for &bw in &bandwidths {
let t = lora_air_time(50, sf, bw, 5, 16);
let ms = t.as_secs_f64() * 1000.0;
// Fastest: SF7/BW500 ~26ms. Slowest: SF12/BW7.8kHz ~many seconds.
// All should be > 1ms and < 300s.
assert!(ms > 1.0, "SF{sf}/BW{bw:?}: air time too short: {ms:.3}ms");
assert!(ms < 300_000.0, "SF{sf}/BW{bw:?}: air time too long: {ms:.3}ms");
}
}
}
mod proptests {
use super::*;
use proptest::prelude::*;
proptest! {
#[test]
fn air_time_always_positive(
payload in 0usize..256,
sf in 7u8..=12u8,
cr in 5u8..=8u8,
preamble in 0u16..=64u16,
) {
for bw in [Bandwidth::Khz7, Bandwidth::Khz62, Bandwidth::Khz125, Bandwidth::Khz500] {
let t = lora_air_time(payload, sf, bw, cr, preamble);
prop_assert!(t > Duration::ZERO, "air time must be positive");
prop_assert!(t < Duration::from_secs(600), "air time should be < 600s");
}
}
}
}
}
+118 -5
View File
@@ -1,3 +1,8 @@
//! Central packet bus bridging radio and gossip swarm.
//!
//! Applies blake3 content-hash deduplication, token bucket rate limiting,
//! and TX queue management.
use std::collections::{HashMap, VecDeque};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
@@ -81,6 +86,7 @@ pub struct Stats {
}
impl Stats {
#[must_use]
pub fn snapshot(&self) -> StatsSnapshot {
StatsSnapshot {
radio_rx: self.radio_rx.load(Ordering::Relaxed),
@@ -96,6 +102,7 @@ impl Stats {
}
}
/// Point-in-time snapshot of bridge statistics, cloned from atomic counters.
#[derive(Debug, Clone)]
pub struct StatsSnapshot {
pub radio_rx: u64,
@@ -117,10 +124,7 @@ struct DedupCache {
impl DedupCache {
fn new(ttl: Duration) -> Self {
Self {
entries: HashMap::new(),
ttl,
}
Self { entries: HashMap::new(), ttl }
}
/// Returns true if the key was already present (i.e., duplicate).
@@ -144,7 +148,7 @@ impl DedupCache {
/// Run the router — the central packet bus.
///
/// Bridges radio packets to/from the gossip swarm with deduplication and rate limiting.
#[allow(clippy::too_many_arguments)]
#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
pub async fn run(
our_id: PublicKey,
dedup_window: Duration,
@@ -294,6 +298,7 @@ pub async fn run(
}).await;
}
GossipEvent::NeighborChanged(count) => {
#[allow(clippy::cast_possible_truncation)]
stats.neighbor_count.store(count as u64, Ordering::Relaxed);
}
}
@@ -309,3 +314,111 @@ pub async fn run(
}
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
#[test]
fn dedup_cache_new_key_not_duplicate() {
let mut cache = DedupCache::new(Duration::from_secs(10));
assert!(!cache.check_and_insert([0u8; 32]));
}
#[test]
fn dedup_cache_duplicate_detected() {
let mut cache = DedupCache::new(Duration::from_secs(10));
cache.check_and_insert([1u8; 32]);
assert!(cache.check_and_insert([1u8; 32]));
}
#[test]
fn dedup_cache_different_keys_independent() {
let mut cache = DedupCache::new(Duration::from_secs(10));
cache.check_and_insert([1u8; 32]);
assert!(!cache.check_and_insert([2u8; 32]));
}
#[test]
fn dedup_cache_expired_entry_not_duplicate() {
let mut cache = DedupCache::new(Duration::from_millis(1));
cache.check_and_insert([1u8; 32]);
std::thread::sleep(Duration::from_millis(5));
assert!(!cache.check_and_insert([1u8; 32]));
}
#[test]
fn dedup_cache_prune_removes_expired() {
let mut cache = DedupCache::new(Duration::from_millis(1));
cache.check_and_insert([1u8; 32]);
cache.check_and_insert([2u8; 32]);
std::thread::sleep(Duration::from_millis(5));
cache.prune();
assert!(cache.entries.is_empty());
}
#[test]
fn dedup_cache_prune_keeps_fresh() {
let mut cache = DedupCache::new(Duration::from_secs(10));
cache.check_and_insert([1u8; 32]);
cache.prune();
assert_eq!(cache.entries.len(), 1);
}
#[test]
fn stats_snapshot_reflects_increments() {
let stats = Stats::default();
stats.radio_rx.fetch_add(5, Ordering::Relaxed);
stats.gossip_tx.fetch_add(3, Ordering::Relaxed);
stats.radio_connected.store(1, Ordering::Relaxed);
let snap = stats.snapshot();
assert_eq!(snap.radio_rx, 5);
assert_eq!(snap.gossip_tx, 3);
assert!(snap.radio_connected);
}
#[test]
fn stats_default_is_zeroed() {
let snap = Stats::default().snapshot();
assert_eq!(snap.radio_rx, 0);
assert_eq!(snap.gossip_rx, 0);
assert!(!snap.radio_connected);
}
#[test]
fn packet_action_display() {
assert_eq!(format!("{}", PacketAction::Bridged), "");
assert_eq!(format!("{}", PacketAction::DroppedDedup), "DUP");
assert_eq!(format!("{}", PacketAction::DroppedQueueFull), "FULL");
assert_eq!(format!("{}", PacketAction::DroppedRateLimit), "RATE");
}
#[test]
fn dedup_cache_ttl_boundary_check_and_insert() {
// Force an entry's timestamp to be exactly TTL ago. With `< ttl`, the entry
// at exactly TTL is NOT a duplicate. With `<= ttl` (the mutant), it would
// wrongly be treated as still live.
let ttl = Duration::from_secs(10);
let mut cache = DedupCache::new(ttl);
cache.check_and_insert([9u8; 32]);
// Backdate the entry to exactly TTL ago.
let exactly_ttl_ago = Instant::now().checked_sub(ttl).unwrap();
cache.entries.insert([9u8; 32], exactly_ttl_ago);
// duration_since(exactly_ttl_ago) == ttl, so `< ttl` is false → not duplicate.
// The mutant `<= ttl` would say true → duplicate. This catches it.
assert!(!cache.check_and_insert([9u8; 32]), "entry at exact TTL must be treated as expired");
}
#[test]
fn dedup_cache_ttl_boundary_prune() {
let ttl = Duration::from_secs(10);
let mut cache = DedupCache::new(ttl);
cache.check_and_insert([8u8; 32]);
// Backdate to exactly TTL ago.
let exactly_ttl_ago = Instant::now().checked_sub(ttl).unwrap();
cache.entries.insert([8u8; 32], exactly_ttl_ago);
cache.prune();
assert!(cache.entries.is_empty(), "entry at exact TTL must be pruned");
}
}
+41 -24
View File
@@ -17,10 +17,8 @@ const RESET: &str = "\x1b[0m";
/// Run the interactive setup wizard. If `existing` is Some, use those values as defaults.
pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> {
let defaults = existing.map_or_else(
|| (RadioSection::default(), BridgeSection::default()),
|c| (c.radio.clone(), c.bridge.clone()),
);
let defaults = existing
.map_or_else(|| (RadioSection::default(), BridgeSection::default()), |c| (c.radio.clone(), c.bridge.clone()));
let (radio, bridge) = defaults;
println!();
@@ -43,8 +41,9 @@ pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> {
// ── Radio ────────────────────────────────────────────────────
let freq_default = format!("{:.3}", radio.frequency as f64 / 1_000_000.0);
let freq_default = format!("{:.3}", f64::from(radio.frequency) / 1_000_000.0);
let freq_mhz: f64 = prompt_parse("Frequency (MHz)", &freq_default)?;
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)] // validated by prompt_parse
let frequency = (freq_mhz * 1_000_000.0) as u32;
let bandwidth = prompt_default("Bandwidth", &radio.bandwidth)?;
@@ -70,11 +69,7 @@ pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> {
let port_default = radio.port.as_deref().unwrap_or("auto");
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) };
// ── Build and save ───────────────────────────────────────────
@@ -111,12 +106,10 @@ pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> {
fn write_config(path: &Path, cfg: &Config) -> Result<()> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)
.with_context(|| format!("creating directory: {}", parent.display()))?;
std::fs::create_dir_all(parent).with_context(|| format!("creating directory: {}", parent.display()))?;
}
let content = toml::to_string_pretty(cfg).context("serializing config")?;
std::fs::write(path, content)
.with_context(|| format!("writing config: {}", path.display()))?;
std::fs::write(path, content).with_context(|| format!("writing config: {}", path.display()))?;
Ok(())
}
@@ -128,20 +121,12 @@ fn prompt_default(label: &str, default: &str) -> Result<String> {
let mut input = String::new();
io::stdin().read_line(&mut input)?;
let trimmed = input.trim();
if trimmed.is_empty() {
Ok(default.to_string())
} else {
Ok(trimmed.to_string())
}
if trimmed.is_empty() { Ok(default.to_string()) } else { Ok(trimmed.to_string()) }
}
fn prompt_required(label: &str, default: Option<&str>) -> Result<String> {
loop {
let prompt = if let Some(d) = default {
format!(" {label} {DIM}[{d}]{RESET}: ")
} else {
format!(" {label}: ")
};
let prompt = default.map_or_else(|| format!(" {label}: "), |d| format!(" {label} {DIM}[{d}]{RESET}: "));
print!("{prompt}");
io::stdout().flush()?;
let mut input = String::new();
@@ -175,3 +160,35 @@ fn parse_sync_word(s: &str) -> Result<u16> {
let s = s.strip_prefix("0x").unwrap_or(&s);
u16::from_str_radix(s, 16).with_context(|| format!("invalid sync word: {s}"))
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
#[test]
fn parse_sync_word_hex_prefix() {
assert_eq!(parse_sync_word("0x3444").unwrap(), 0x3444);
}
#[test]
fn parse_sync_word_no_prefix() {
assert_eq!(parse_sync_word("3444").unwrap(), 0x3444);
}
#[test]
fn parse_sync_word_case_insensitive() {
assert_eq!(parse_sync_word("0xABCD").unwrap(), 0xABCD);
}
#[test]
fn parse_sync_word_whitespace() {
assert_eq!(parse_sync_word(" 0x3444 ").unwrap(), 0x3444);
}
#[test]
fn parse_sync_word_invalid() {
assert!(parse_sync_word("ZZZZ").is_err());
assert!(parse_sync_word("").is_err());
}
}
+567 -92
View File
@@ -1,19 +1,23 @@
//! Ratatui terminal user interface.
//!
//! Displays network status, radio configuration, aggregate statistics,
//! a scrolling packet log, and real-time activity sparklines.
use std::collections::VecDeque;
use std::io::{self, Stdout};
use std::sync::Arc;
use std::time::{Duration, Instant};
use crossterm::event::{self, Event, KeyCode, KeyEventKind};
use crossterm::terminal::{
EnterAlternateScreen, LeaveAlternateScreen, disable_raw_mode, enable_raw_mode,
};
use crossterm::ExecutableCommand;
use crossterm::event::{self, Event, KeyCode, KeyEventKind};
use crossterm::terminal::{EnterAlternateScreen, LeaveAlternateScreen, disable_raw_mode, enable_raw_mode};
use ratatui::Terminal;
use ratatui::backend::CrosstermBackend;
use ratatui::layout::{Alignment, Constraint, Direction, Layout, Rect};
use ratatui::style::{Color, Modifier, Style};
use ratatui::symbols;
use ratatui::text::{Line, Span};
use ratatui::widgets::{Block, Borders, Clear, Paragraph, Sparkline};
use ratatui::Terminal;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
@@ -53,7 +57,7 @@ struct Bucket {
}
struct TimeSeries {
buckets: Vec<Bucket>,
buckets: VecDeque<Bucket>,
last_snap: StatsSnapshot,
last_bucket_time: Instant,
}
@@ -61,7 +65,7 @@ struct TimeSeries {
impl TimeSeries {
fn new(snap: &StatsSnapshot) -> Self {
Self {
buckets: vec![Bucket::default(); MAX_BUCKETS],
buckets: VecDeque::from(vec![Bucket::default(); MAX_BUCKETS]),
last_snap: snap.clone(),
last_bucket_time: Instant::now(),
}
@@ -78,9 +82,9 @@ impl TimeSeries {
snr_sum,
snr_count,
};
self.buckets.push(bucket);
self.buckets.push_back(bucket);
if self.buckets.len() > MAX_BUCKETS {
self.buckets.remove(0);
self.buckets.pop_front();
}
self.last_snap = snap.clone();
self.last_bucket_time = now;
@@ -98,8 +102,8 @@ impl TimeSeries {
.iter()
.map(|b| {
if b.snr_count > 0 {
let avg = b.snr_sum / b.snr_count as i64;
(avg + 30).max(0) as u64
let avg = b.snr_sum / b.snr_count.cast_signed();
(avg + 30).max(0).cast_unsigned()
} else {
0
}
@@ -114,7 +118,7 @@ impl TimeSeries {
self.buckets.iter().map(|b| b.radio_tx).max().unwrap_or(0)
}
fn latest_avg_snr(&self) -> Option<i64> {
self.buckets.iter().rev().find(|b| b.snr_count > 0).map(|b| b.snr_sum / b.snr_count as i64)
self.buckets.iter().rev().find(|b| b.snr_count > 0).map(|b| b.snr_sum / b.snr_count.cast_signed())
}
}
@@ -137,7 +141,9 @@ fn snr_stats_since(entries: &[PacketLogEntry], since: Instant) -> (i64, u64) {
type Term = Terminal<CrosstermBackend<Stdout>>;
pub async fn run(
/// Start the terminal UI event loop. Returns the terminal handle for cleanup on exit.
#[allow(clippy::needless_pass_by_value)] // Arc/CancellationToken are cheaply cloned and owned by this scope
pub fn run(
config_watch: tokio::sync::watch::Receiver<RadioConfigInfo>,
gossip: Arc<Gossip>,
stats: Arc<Stats>,
@@ -150,26 +156,34 @@ pub async fn run(
let backend = CrosstermBackend::new(io::stdout());
let mut terminal = Terminal::new(backend)?;
let mut log_entries: Vec<PacketLogEntry> = Vec::with_capacity(MAX_LOG_ENTRIES);
let mut log_entries: VecDeque<PacketLogEntry> = VecDeque::with_capacity(MAX_LOG_ENTRIES);
let mut series = TimeSeries::new(&stats.snapshot());
let _result = tui_loop(
&mut terminal, config_watch, &gossip, &stats,
&mut log_rx, &cancel, start_time, &mut log_entries, &mut series,
).await;
&mut terminal,
config_watch,
&gossip,
&stats,
&mut log_rx,
&cancel,
start_time,
&mut log_entries,
&mut series,
);
terminal.draw(|frame| draw_shutdown_overlay(frame, frame.area()))?;
Ok(terminal)
}
/// Restore the terminal to its normal state (disable raw mode, leave alternate screen).
pub fn restore_terminal(terminal: &mut Term) {
let _ = disable_raw_mode();
let _ = io::stdout().execute(LeaveAlternateScreen);
let _ = terminal.show_cursor();
}
#[allow(clippy::too_many_arguments)]
async fn tui_loop(
#[allow(clippy::too_many_arguments, clippy::needless_pass_by_value)]
fn tui_loop(
terminal: &mut Term,
config_watch: tokio::sync::watch::Receiver<RadioConfigInfo>,
gossip: &Gossip,
@@ -177,26 +191,27 @@ async fn tui_loop(
log_rx: &mut mpsc::Receiver<PacketLogEntry>,
cancel: &CancellationToken,
start_time: Instant,
log_entries: &mut Vec<PacketLogEntry>,
log_entries: &mut VecDeque<PacketLogEntry>,
series: &mut TimeSeries,
) -> anyhow::Result<()> {
let mut last_tick = Instant::now();
loop {
while let Ok(entry) = log_rx.try_recv() {
if log_entries.len() >= MAX_LOG_ENTRIES {
log_entries.remove(0);
log_entries.pop_front();
}
log_entries.push(entry);
log_entries.push_back(entry);
}
if cancel.is_cancelled() {
return Ok(());
}
let snap = stats.snapshot();
series.update(&snap, log_entries);
let log_slice = log_entries.make_contiguous();
series.update(&snap, log_slice);
let swarm_state = gossip.swarm_state();
let config_info = config_watch.borrow().clone();
terminal.draw(|frame| {
draw_ui(frame, frame.area(), &config_info, &snap, &swarm_state, log_entries, start_time, series);
draw_ui(frame, frame.area(), &config_info, &snap, &swarm_state, log_slice, start_time, series);
})?;
let timeout = TICK_RATE.checked_sub(last_tick.elapsed()).unwrap_or(Duration::ZERO);
if event::poll(timeout)?
@@ -204,8 +219,7 @@ async fn tui_loop(
&& key.kind == KeyEventKind::Press
&& matches!(
(key.code, key.modifiers),
(KeyCode::Char('q'), _)
| (KeyCode::Char('c'), crossterm::event::KeyModifiers::CONTROL)
(KeyCode::Char('q'), _) | (KeyCode::Char('c'), crossterm::event::KeyModifiers::CONTROL)
)
{
cancel.cancel();
@@ -230,6 +244,16 @@ fn draw_ui(
start_time: Instant,
series: &TimeSeries,
) {
// Minimum terminal size for the 3-column layout.
if area.width < 86 || area.height < 12 {
let msg = format!("Terminal too small ({}x{}), need 86x12+", area.width, area.height);
frame.render_widget(
Paragraph::new(msg).alignment(Alignment::Center).style(Style::default().fg(C_RATE_DROP)),
area,
);
return;
}
// Header row + body.
let rows = Layout::default()
.direction(Direction::Vertical)
@@ -248,11 +272,7 @@ fn draw_ui(
// Left column: Network, Radio, Stats stacked.
let left = Layout::default()
.direction(Direction::Vertical)
.constraints([
Constraint::Length(6),
Constraint::Length(12),
Constraint::Min(5),
])
.constraints([Constraint::Length(6), Constraint::Length(12), Constraint::Min(5)])
.split(cols[0]);
draw_network(frame, left[0], swarm_state);
@@ -283,19 +303,12 @@ fn draw_header(frame: &mut ratatui::Frame, area: Rect, stats: &StatsSnapshot) {
frame.render_widget(Paragraph::new(pills), cols[0]);
// Right: clock with timezone.
let clock = Line::from(Span::styled(
format!("{} ", chrono_time()),
Style::default().fg(C_LABEL),
));
let clock = Line::from(Span::styled(format!("{} ", chrono_time()), Style::default().fg(C_LABEL)));
frame.render_widget(Paragraph::new(clock).alignment(Alignment::Right), cols[1]);
}
fn status_pill(label: &str, ok: bool) -> Span<'static> {
let (fg, bg) = if ok {
(Color::Black, Color::Green)
} else {
(Color::White, C_LABEL)
};
let (fg, bg) = if ok { (Color::Black, Color::Green) } else { (Color::White, C_LABEL) };
Span::styled(format!(" {label} "), Style::default().fg(fg).bg(bg))
}
@@ -310,10 +323,7 @@ fn draw_network(frame: &mut ratatui::Frame, area: Rect, state: &crate::gossip::S
crate::gossip::DhtStatus::PublishFailed => C_QUEUE_DROP,
};
let neighbor_color = if state.neighbor_count > 0 { Color::Green } else { C_RATE_DROP };
let last_pub = state
.last_dht_publish
.map(|t| format!("{}s ago", t.elapsed().as_secs()))
.unwrap_or_else(|| "-".into());
let last_pub = state.last_dht_publish.map_or_else(|| "-".into(), |t| format!("{}s ago", t.elapsed().as_secs()));
let lines = vec![
kv_line("ID", &state.topic_hash, C_ACCENT),
@@ -347,17 +357,14 @@ fn draw_radio(frame: &mut ratatui::Frame, area: Rect, config_info: &RadioConfigI
};
let block = Block::default()
.title(Span::styled(
title,
Style::default().fg(C_TITLE).add_modifier(Modifier::BOLD),
))
.title(Span::styled(title, Style::default().fg(C_TITLE).add_modifier(Modifier::BOLD)))
.borders(Borders::ALL)
.border_style(Style::default().fg(border_color));
let a = &config_info.active;
let r = &config_info.requested;
let freq_mhz = a.freq_hz as f64 / 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 power_str = if a.tx_power_dbm == donglora_client::TX_POWER_MAX {
"max".to_string()
@@ -429,6 +436,7 @@ fn draw_stats(frame: &mut ratatui::Frame, area: Rect, stats: &StatsSnapshot, sta
// ── Packet log ───────────────────────────────────────────────────
#[allow(clippy::option_if_let_else)] // match is more readable for RSSI/SNR formatting
fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry]) {
let block = panel_block(" Packet Log ");
@@ -472,18 +480,10 @@ fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry])
let (rf_col, net_col) = match e.direction {
PacketDirection::RadioIn => (
Span::styled(format!("{:>4}", "RX"), Style::default().fg(C_RADIO_RX)),
if bridged {
Span::styled(format!("{:>4}", "TX"), Style::default().fg(C_NET_TX))
} else {
dash
},
if bridged { Span::styled(format!("{:>4}", "TX"), Style::default().fg(C_NET_TX)) } else { dash },
),
PacketDirection::GossipIn => (
if bridged {
Span::styled(format!("{:>4}", "TX"), Style::default().fg(C_RADIO_TX))
} else {
dash
},
if bridged { Span::styled(format!("{:>4}", "TX"), Style::default().fg(C_RADIO_TX)) } else { dash },
Span::styled(format!("{:>4}", "RX"), Style::default().fg(C_NET_RX)),
),
};
@@ -510,15 +510,15 @@ fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry])
PacketAction::DroppedQueueFull => "FULL",
PacketAction::DroppedRateLimit => "RATE",
};
let action_span = Span::styled(format!(" {:>4}", action_str), Style::default().fg(action_color));
let action_span = Span::styled(format!(" {action_str:>4}"), Style::default().fg(action_color));
let size_str = format!("{}B", e.size);
lines.push(Line::from(vec![
Span::styled(format!("{:>5}", age), Style::default().fg(C_LABEL)),
Span::styled(format!("{age:>5}"), Style::default().fg(C_LABEL)),
hash_span,
rf_col,
net_col,
Span::styled(format!("{:>6}", size_str), Style::default().fg(C_VALUE)),
Span::styled(format!("{size_str:>6}"), Style::default().fg(C_VALUE)),
rssi_span,
snr_span,
action_span,
@@ -541,11 +541,7 @@ fn draw_activity(frame: &mut ratatui::Frame, area: Rect, series: &TimeSeries) {
let rows = Layout::default()
.direction(Direction::Vertical)
.constraints([
Constraint::Ratio(1, 3),
Constraint::Ratio(1, 3),
Constraint::Ratio(1, 3),
])
.constraints([Constraint::Ratio(1, 3), Constraint::Ratio(1, 3), Constraint::Ratio(1, 3)])
.split(inner);
let spark_w = inner.width as usize;
@@ -556,10 +552,7 @@ fn draw_activity(frame: &mut ratatui::Frame, area: Rect, series: &TimeSeries) {
let peak_rx = series.peak_rx();
let peak_tx = series.peak_tx();
let snr_label = series
.latest_avg_snr()
.map(|s| format!("{s}dB"))
.unwrap_or_else(|| "-".into());
let snr_label = series.latest_avg_snr().map_or_else(|| "-".into(), |s| format!("{s}dB"));
draw_sparkline_section(frame, rows[0], "RX", &format!("pk:{peak_rx}"), &rx_data, C_RADIO_RX);
draw_sparkline_section(frame, rows[1], "TX", &format!("pk:{peak_tx}"), &tx_data, C_RADIO_TX);
@@ -590,10 +583,7 @@ fn draw_sparkline_section(
]);
frame.render_widget(Paragraph::new(label_line), rows[0]);
let spark = Sparkline::default()
.data(data)
.bar_set(symbols::bar::NINE_LEVELS)
.style(Style::default().fg(color));
let spark = Sparkline::default().data(data).bar_set(symbols::bar::NINE_LEVELS).style(Style::default().fg(color));
frame.render_widget(spark, rows[1]);
}
@@ -619,9 +609,7 @@ fn draw_shutdown_overlay(frame: &mut ratatui::Frame, area: Rect) {
let popup = Rect::new(x, y, w.min(area.width), h.min(area.height));
frame.render_widget(Clear, popup);
let block = Block::default()
.borders(Borders::ALL)
.border_style(Style::default().fg(C_BORDER));
let block = Block::default().borders(Borders::ALL).border_style(Style::default().fg(C_BORDER));
let text = Paragraph::new(Line::from(vec![
Span::styled(" Shutting down", Style::default().fg(C_RATE_DROP)),
Span::styled(" — updating DHT...", Style::default().fg(C_LABEL)),
@@ -635,20 +623,29 @@ fn draw_shutdown_overlay(frame: &mut ratatui::Frame, area: Rect) {
fn panel_block(title: &str) -> Block<'static> {
Block::default()
.title(Span::styled(
title.to_string(),
Style::default().fg(C_TITLE).add_modifier(Modifier::BOLD),
))
.title(Span::styled(title.to_string(), Style::default().fg(C_TITLE).add_modifier(Modifier::BOLD)))
.borders(Borders::ALL)
.border_style(Style::default().fg(C_BORDER))
}
fn rssi_color(rssi: i16) -> Color {
if rssi >= -70 { Color::Green } else if rssi >= -90 { C_RATE_DROP } else { C_QUEUE_DROP }
const fn rssi_color(rssi: i16) -> Color {
if rssi >= -70 {
Color::Green
} else if rssi >= -90 {
C_RATE_DROP
} else {
C_QUEUE_DROP
}
}
fn snr_color(snr: i16) -> Color {
if snr >= 5 { Color::Green } else if snr >= 0 { C_RATE_DROP } else { C_QUEUE_DROP }
const fn snr_color(snr: i16) -> Color {
if snr >= 5 {
Color::Green
} else if snr >= 0 {
C_RATE_DROP
} else {
C_QUEUE_DROP
}
}
fn kv_line(label: &str, value: &str, color: Color) -> Line<'static> {
@@ -661,18 +658,496 @@ fn kv_line(label: &str, value: &str, color: Color) -> Line<'static> {
fn format_duration(d: Duration) -> String {
let secs = d.as_secs();
if secs < 60 { format!("{secs}s") }
else if secs < 3600 { format!("{}m{}s", secs / 60, secs % 60) }
else { format!("{}h{}m", secs / 3600, (secs % 3600) / 60) }
if secs < 60 {
format!("{secs}s")
} else if secs < 3600 {
format!("{}m{}s", secs / 60, secs % 60)
} else {
format!("{}h{}m", secs / 3600, (secs % 3600) / 60)
}
}
fn format_short_duration(d: Duration) -> String {
let secs = d.as_secs();
if secs < 60 { format!("{secs}s") }
else if secs < 3600 { format!("{}m", secs / 60) }
else { format!("{}h", secs / 3600) }
if secs < 60 {
format!("{secs}s")
} else if secs < 3600 {
format!("{}m", secs / 60)
} else {
format!("{}h", secs / 3600)
}
}
fn chrono_time() -> String {
jiff::Zoned::now().strftime("%H:%M:%S %Z").to_string()
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
// ── Helpers ──────────────────────────────────────────────────
fn make_snapshot(radio_rx: u64, radio_tx: u64) -> StatsSnapshot {
StatsSnapshot {
radio_rx,
radio_tx,
gossip_rx: 0,
gossip_tx: 0,
dedup_hits: 0,
rate_limit_drops: 0,
dropped_queue: 0,
neighbor_count: 0,
radio_connected: true,
}
}
fn make_log_entry(timestamp: Instant, snr: Option<i16>) -> PacketLogEntry {
PacketLogEntry {
timestamp,
hash: [0u8; 32],
direction: PacketDirection::RadioIn,
size: 10,
snr,
rssi: Some(-80),
action: PacketAction::Bridged,
}
}
// ── format_duration tests ───────────────────────────────────
#[test]
fn format_duration_seconds() {
assert_eq!(format_duration(Duration::from_secs(45)), "45s");
}
#[test]
fn format_duration_minutes() {
assert_eq!(format_duration(Duration::from_secs(125)), "2m5s");
}
#[test]
fn format_duration_hours() {
assert_eq!(format_duration(Duration::from_secs(7500)), "2h5m");
}
#[test]
fn format_duration_zero() {
assert_eq!(format_duration(Duration::ZERO), "0s");
}
#[test]
fn format_duration_boundary_59s() {
assert_eq!(format_duration(Duration::from_secs(59)), "59s");
}
#[test]
fn format_duration_boundary_60s() {
assert_eq!(format_duration(Duration::from_secs(60)), "1m0s");
}
#[test]
fn format_duration_boundary_3599s() {
assert_eq!(format_duration(Duration::from_secs(3599)), "59m59s");
}
#[test]
fn format_duration_boundary_3600s() {
assert_eq!(format_duration(Duration::from_secs(3600)), "1h0m");
}
// ── format_short_duration tests ─────────────────────────────
#[test]
fn format_short_duration_rounds() {
assert_eq!(format_short_duration(Duration::from_secs(30)), "30s");
assert_eq!(format_short_duration(Duration::from_secs(90)), "1m");
assert_eq!(format_short_duration(Duration::from_secs(7500)), "2h");
}
#[test]
fn format_short_duration_boundary_59s() {
assert_eq!(format_short_duration(Duration::from_secs(59)), "59s");
}
#[test]
fn format_short_duration_boundary_60s() {
assert_eq!(format_short_duration(Duration::from_secs(60)), "1m");
}
#[test]
fn format_short_duration_boundary_3599s() {
assert_eq!(format_short_duration(Duration::from_secs(3599)), "59m");
}
#[test]
fn format_short_duration_boundary_3600s() {
assert_eq!(format_short_duration(Duration::from_secs(3600)), "1h");
}
// ── chrono_time tests ───────────────────────────────────────
#[test]
fn chrono_time_returns_non_empty_string() {
let t = chrono_time();
assert!(!t.is_empty(), "chrono_time() should not be empty");
}
#[test]
fn chrono_time_contains_colons() {
// HH:MM:SS format should have at least two colons
let t = chrono_time();
assert!(t.matches(':').count() >= 2, "chrono_time() should contain HH:MM:SS, got: {t}");
}
// ── rssi_color tests ────────────────────────────────────────
#[test]
fn rssi_color_green_at_minus_70() {
assert_eq!(rssi_color(-70), Color::Green);
}
#[test]
fn rssi_color_yellow_at_minus_71() {
assert_eq!(rssi_color(-71), C_RATE_DROP);
}
#[test]
fn rssi_color_yellow_at_minus_90() {
assert_eq!(rssi_color(-90), C_RATE_DROP);
}
#[test]
fn rssi_color_red_at_minus_91() {
assert_eq!(rssi_color(-91), C_QUEUE_DROP);
}
#[test]
fn rssi_color_green_at_zero() {
assert_eq!(rssi_color(0), Color::Green);
}
// ── snr_color tests ─────────────────────────────────────────
#[test]
fn snr_color_green_at_5() {
assert_eq!(snr_color(5), Color::Green);
}
#[test]
fn snr_color_yellow_at_4() {
assert_eq!(snr_color(4), C_RATE_DROP);
}
#[test]
fn snr_color_yellow_at_0() {
assert_eq!(snr_color(0), C_RATE_DROP);
}
#[test]
fn snr_color_red_at_minus_1() {
assert_eq!(snr_color(-1), C_QUEUE_DROP);
}
#[test]
fn snr_color_green_at_high_value() {
assert_eq!(snr_color(20), Color::Green);
}
// ── snr_stats_since tests ───────────────────────────────────
#[test]
fn snr_stats_since_empty() {
assert_eq!(snr_stats_since(&[], Instant::now()), (0, 0));
}
#[test]
fn snr_stats_since_sums_and_counts_correctly() {
let now = Instant::now();
let entries = vec![
make_log_entry(now + Duration::from_secs(1), Some(10)),
make_log_entry(now + Duration::from_secs(2), Some(20)),
make_log_entry(now + Duration::from_secs(3), Some(-5)),
];
let (sum, count) = snr_stats_since(&entries, now);
assert_eq!(sum, 25); // 10 + 20 + (-5)
assert_eq!(count, 3);
}
#[test]
fn snr_stats_since_respects_cutoff() {
let now = Instant::now();
let entries = vec![
make_log_entry(now + Duration::from_secs(1), Some(10)),
make_log_entry(now + Duration::from_secs(2), Some(20)),
make_log_entry(now + Duration::from_secs(5), Some(30)),
];
// Use a cutoff after the first two entries
let cutoff = now + Duration::from_secs(3);
let (sum, count) = snr_stats_since(&entries, cutoff);
assert_eq!(sum, 30);
assert_eq!(count, 1);
}
#[test]
fn snr_stats_since_skips_none_snr() {
let now = Instant::now();
let entries = vec![
make_log_entry(now + Duration::from_secs(1), Some(10)),
make_log_entry(now + Duration::from_secs(2), None),
make_log_entry(now + Duration::from_secs(3), Some(20)),
];
let (sum, count) = snr_stats_since(&entries, now);
assert_eq!(sum, 30); // 10 + 20, skipping None
assert_eq!(count, 2);
}
#[test]
fn snr_stats_since_all_before_cutoff() {
let now = Instant::now();
let entries = vec![make_log_entry(now, Some(10)), make_log_entry(now, Some(20))];
// Cutoff at `now` means entries at exactly `now` are <= since, so excluded
let (sum, count) = snr_stats_since(&entries, now);
assert_eq!(sum, 0);
assert_eq!(count, 0);
}
#[test]
fn snr_stats_since_all_none_snr() {
let now = Instant::now();
let entries = vec![
make_log_entry(now + Duration::from_secs(1), None),
make_log_entry(now + Duration::from_secs(2), None),
];
let (sum, count) = snr_stats_since(&entries, now);
assert_eq!(sum, 0);
assert_eq!(count, 0);
}
// ── TimeSeries tests ────────────────────────────────────────
#[test]
fn time_series_new_has_max_buckets() {
let snap = make_snapshot(0, 0);
let ts = TimeSeries::new(&snap);
assert_eq!(ts.buckets.len(), MAX_BUCKETS);
}
#[test]
fn time_series_radio_rx_returns_bucket_data() {
let snap = make_snapshot(0, 0);
let mut ts = TimeSeries::new(&snap);
// Manually push a bucket with known data
ts.buckets.push_back(Bucket { radio_rx: 42, radio_tx: 0, snr_sum: 0, snr_count: 0 });
let rx = ts.radio_rx();
assert_eq!(*rx.last().unwrap(), 42);
}
#[test]
fn time_series_radio_tx_returns_bucket_data() {
let snap = make_snapshot(0, 0);
let mut ts = TimeSeries::new(&snap);
ts.buckets.push_back(Bucket { radio_rx: 0, radio_tx: 99, snr_sum: 0, snr_count: 0 });
let tx = ts.radio_tx();
assert_eq!(*tx.last().unwrap(), 99);
}
#[test]
fn time_series_avg_snr_with_data() {
let snap = make_snapshot(0, 0);
let mut ts = TimeSeries::new(&snap);
// Push bucket: snr_sum=10, snr_count=2 -> avg=5 -> shifted = 5+30 = 35
ts.buckets.push_back(Bucket { radio_rx: 0, radio_tx: 0, snr_sum: 10, snr_count: 2 });
let snr = ts.avg_snr();
assert_eq!(*snr.last().unwrap(), 35);
}
#[test]
fn time_series_avg_snr_zero_count() {
let snap = make_snapshot(0, 0);
let ts = TimeSeries::new(&snap);
// All default buckets have snr_count=0, so avg_snr should be all zeros
let snr = ts.avg_snr();
assert!(snr.iter().all(|&v| v == 0));
}
#[test]
fn time_series_avg_snr_negative_clamps_to_zero() {
let snap = make_snapshot(0, 0);
let mut ts = TimeSeries::new(&snap);
// avg = -50, shifted = -50+30 = -20, clamped to 0
ts.buckets.push_back(Bucket { radio_rx: 0, radio_tx: 0, snr_sum: -50, snr_count: 1 });
let snr = ts.avg_snr();
assert_eq!(*snr.last().unwrap(), 0);
}
#[test]
fn time_series_peak_rx() {
let snap = make_snapshot(0, 0);
let mut ts = TimeSeries::new(&snap);
ts.buckets.push_back(Bucket { radio_rx: 5, radio_tx: 0, snr_sum: 0, snr_count: 0 });
ts.buckets.push_back(Bucket { radio_rx: 15, radio_tx: 0, snr_sum: 0, snr_count: 0 });
ts.buckets.push_back(Bucket { radio_rx: 3, radio_tx: 0, snr_sum: 0, snr_count: 0 });
assert_eq!(ts.peak_rx(), 15);
}
#[test]
fn time_series_peak_tx() {
let snap = make_snapshot(0, 0);
let mut ts = TimeSeries::new(&snap);
ts.buckets.push_back(Bucket { radio_rx: 0, radio_tx: 7, snr_sum: 0, snr_count: 0 });
ts.buckets.push_back(Bucket { radio_rx: 0, radio_tx: 22, snr_sum: 0, snr_count: 0 });
assert_eq!(ts.peak_tx(), 22);
}
#[test]
fn time_series_peak_rx_default_is_zero() {
let snap = make_snapshot(0, 0);
let ts = TimeSeries::new(&snap);
assert_eq!(ts.peak_rx(), 0);
}
#[test]
fn time_series_peak_tx_default_is_zero() {
let snap = make_snapshot(0, 0);
let ts = TimeSeries::new(&snap);
assert_eq!(ts.peak_tx(), 0);
}
#[test]
fn time_series_latest_avg_snr_with_data() {
let snap = make_snapshot(0, 0);
let mut ts = TimeSeries::new(&snap);
// Push an older bucket with different SNR
ts.buckets.push_back(Bucket { radio_rx: 0, radio_tx: 0, snr_sum: 100, snr_count: 10 });
// Push the latest bucket with known SNR: sum=20, count=4 -> avg=5
ts.buckets.push_back(Bucket { radio_rx: 0, radio_tx: 0, snr_sum: 20, snr_count: 4 });
assert_eq!(ts.latest_avg_snr(), Some(5));
}
#[test]
fn time_series_latest_avg_snr_none_when_no_data() {
let snap = make_snapshot(0, 0);
let ts = TimeSeries::new(&snap);
// All default buckets have snr_count=0
assert_eq!(ts.latest_avg_snr(), None);
}
#[test]
fn time_series_latest_avg_snr_skips_trailing_empty() {
let snap = make_snapshot(0, 0);
let mut ts = TimeSeries::new(&snap);
// A bucket with data followed by an empty bucket
ts.buckets.push_back(Bucket { radio_rx: 0, radio_tx: 0, snr_sum: -12, snr_count: 3 });
ts.buckets.push_back(Bucket::default()); // snr_count = 0
// latest_avg_snr scans backwards, skips the empty one, finds -12/3 = -4
assert_eq!(ts.latest_avg_snr(), Some(-4));
}
#[test]
fn time_series_update_creates_bucket() {
let snap0 = make_snapshot(10, 5);
let mut ts = TimeSeries::new(&snap0);
// Force the last_bucket_time to be old enough to trigger a new bucket
ts.last_bucket_time = Instant::now().checked_sub(Duration::from_secs(BUCKET_SECS + 1)).unwrap();
let initial_len = ts.buckets.len();
let snap1 = make_snapshot(15, 8);
let entries = vec![make_log_entry(Instant::now(), Some(10))];
ts.update(&snap1, &entries);
// A new bucket should have been pushed
assert!(ts.buckets.len() >= initial_len);
let last = ts.buckets.back().unwrap();
assert_eq!(last.radio_rx, 5); // 15 - 10
assert_eq!(last.radio_tx, 3); // 8 - 5
}
#[test]
fn time_series_update_does_not_create_bucket_when_interval_not_elapsed() {
let snap0 = make_snapshot(10, 5);
let mut ts = TimeSeries::new(&snap0);
// last_bucket_time is now, so interval has NOT elapsed
let len_before = ts.buckets.len();
let snap1 = make_snapshot(20, 10);
ts.update(&snap1, &[]);
// No new bucket should be added
assert_eq!(ts.buckets.len(), len_before);
}
#[test]
fn time_series_update_rotates_when_exceeding_max() {
let snap0 = make_snapshot(0, 0);
let mut ts = TimeSeries::new(&snap0);
assert_eq!(ts.buckets.len(), MAX_BUCKETS);
// Force old timestamp so update triggers
ts.last_bucket_time = Instant::now().checked_sub(Duration::from_secs(BUCKET_SECS + 1)).unwrap();
let snap1 = make_snapshot(1, 1);
ts.update(&snap1, &[]);
// Should still be at MAX_BUCKETS (the front was popped)
assert_eq!(ts.buckets.len(), MAX_BUCKETS);
// The last bucket should be our new one
let last = ts.buckets.back().unwrap();
assert_eq!(last.radio_rx, 1);
}
#[test]
fn time_series_update_no_pop_at_exactly_max() {
// Start with MAX_BUCKETS - 1 entries. Push one to reach exactly MAX_BUCKETS.
// With `> MAX_BUCKETS`: 300 is NOT > 300, so no pop → len stays 300. (correct)
// With `>= MAX_BUCKETS`: 300 IS >= 300, so pop → len becomes 299. (wrong)
let snap0 = make_snapshot(0, 0);
let mut ts = TimeSeries::new(&snap0);
ts.buckets.pop_front(); // now MAX_BUCKETS - 1
assert_eq!(ts.buckets.len(), MAX_BUCKETS - 1);
ts.last_bucket_time = Instant::now().checked_sub(Duration::from_secs(BUCKET_SECS + 1)).unwrap();
let snap1 = make_snapshot(1, 0);
ts.update(&snap1, &[]);
assert_eq!(ts.buckets.len(), MAX_BUCKETS, "should be exactly MAX_BUCKETS, not popped");
}
// ── pad_right_align tests ───────────────────────────────────
#[test]
fn pad_right_align_shorter_than_width() {
assert_eq!(pad_right_align(&[1, 2, 3], 5), vec![0, 0, 1, 2, 3]);
}
#[test]
fn pad_right_align_longer_than_width() {
assert_eq!(pad_right_align(&[1, 2, 3, 4, 5], 3), vec![3, 4, 5]);
}
#[test]
fn pad_right_align_exact() {
assert_eq!(pad_right_align(&[1, 2, 3], 3), vec![1, 2, 3]);
}
#[test]
fn pad_right_align_empty() {
assert_eq!(pad_right_align(&[], 3), vec![0, 0, 0]);
}
mod proptests {
use super::*;
use proptest::prelude::*;
proptest! {
#[test]
fn pad_right_align_output_length(
data in proptest::collection::vec(0u64..100, 0..500),
width in 1usize..200,
) {
let result = pad_right_align(&data, width);
prop_assert_eq!(result.len(), width);
}
}
}
}