1 Commits
Author SHA1 Message Date
swaits 00a8d3126e fix: reduce TUI minimum width from 86 to 80 columns 2026-04-06 17:08:23 -06:00
25 changed files with 10428 additions and 19 deletions

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
+41
View File
@@ -0,0 +1,41 @@
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@v5
- name: Install system dependencies
run: sudo apt-get update && sudo apt-get install -y libudev-dev
- 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
+11
View File
@@ -0,0 +1,11 @@
/target
/fuzz/target
/fuzz/corpus
/fuzz/Cargo.lock
/mutants.out
/mutants.out.old
.jj/
*.swp
*.swo
*~
.DS_Store
+34
View File
@@ -0,0 +1,34 @@
# Changelog
## [0.2.2] - 2026-04-06
### Fixed
- TUI now fits in 80-column terminals (was 86).
## [0.2.1] - 2026-04-06
### Fixed
- CI: removed stale sibling-clone step, added `libudev-dev` dependency,
upgraded `actions/checkout` to v5 (Node.js 24).
## [0.2.0] - 2026-04-06
### Added
- Auto-detect connection mode: when no `port` is configured, the bridge now
tries the mux daemon first and falls back to direct USB serial
(`find_port()`) if the mux is unavailable. Previously it would retry the mux
forever.
### Changed
- Config negotiation on direct serial connections now pushes `SetConfig` with
the desired radio config when it differs from the dongle's current config.
On mux connections, the existing behavior (accept the mux's config) is
unchanged.
## [0.1.0] - 2026-04-06
Initial release.
Generated
+5539
View File
File diff suppressed because it is too large. Load diff
+54
View File
@@ -0,0 +1,54 @@
[package]
name = "donglora-bridge"
version = "0.2.2"
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 = "0.1"
iroh = "0.97"
iroh-gossip = "0.97"
mainline = { version = "6", features = ["async"] }
tokio = { version = "1", features = ["full"] }
clap = { version = "4", features = ["derive"] }
serde = { version = "1", features = ["derive"] }
toml = "1"
blake3 = "1"
ratatui = "0.30"
crossterm = "0.29"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
tracing-appender = "0.2"
anyhow = "1"
dirs = "6"
bytes = "1"
futures-lite = "2"
rand = "0.9"
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.
+284 -19
View File
@@ -1,25 +1,290 @@
# donglora-bridge has moved
# donglora-bridge
This repository has moved to the
[**donglora** organization](https://git.swaits.com/donglora):
[![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](https://git.swaits.com/donglora/bridge)**
**donglora-bridge** connects LoRa radios to the internet so packets heard by one
radio can be retransmitted by another radio anywhere in the world.
donglora-bridge connects LoRa radios to the internet so packets heard by
one radio can be retransmitted by another radio anywhere in the world.
Bridges discover each other via the Mainline DHT and communicate through
an encrypted peer-to-peer gossip swarm — no central server required.
[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.
## Other DongLoRa repos
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.
| What | Location |
|------|----------|
| Firmware | [donglora/firmware](https://git.swaits.com/donglora/firmware) |
| Rust client library | [donglora/client-rs](https://git.swaits.com/donglora/client-rs) |
| Python client library | [donglora/client-py](https://git.swaits.com/donglora/client-py) |
| Rust mux daemon | [donglora/mux-rs](https://git.swaits.com/donglora/mux-rs) |
| Python mux daemon | [donglora/mux-py](https://git.swaits.com/donglora/mux-py) |
| MeshCore AI bot | [donglora/ai-bot](https://git.swaits.com/donglora/ai-bot) |
| Example scripts | [donglora/examples](https://git.swaits.com/donglora/examples) |
![TUI Screenshot](screenshot.png)
This repo is archived and will not receive further updates.
## 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
The bridge works fine without the mux -- it connects directly to the USB device.
The mux is only needed if you want multiple programs to share a single DongLoRa
radio at the same time (e.g. running two bridges with different configs, or a
bridge alongside another donglora-client program).
```sh
cargo install donglora-mux
```
Run `donglora-mux` before starting the bridge. The bridge auto-detects the mux
when it's running and falls back to direct USB serial when it's not.
### 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
```
### Running multiple bridges
To run more than one bridge on the same machine (e.g. bridging two separate
MeshCore networks), create a config file per bridge and run each with
`--config`:
```sh
donglora-bridge --config ~/bridge-network-a.toml
donglora-bridge --config ~/bridge-network-b.toml
```
Each config can specify a different passphrase, different radio settings, and
(if using the mux) share the same radio. Without the mux, each bridge needs its
own USB dongle and an explicit `port` in its config to avoid both trying to open
the same device.
## 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: tries mux, then USB)
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)
+39
View File
@@ -0,0 +1,39 @@
# 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 = []
+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

+434
View File
@@ -0,0 +1,434 @@
//! 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};
use donglora_client::{Bandwidth, RadioConfig, TX_POWER_MAX};
use serde::{Deserialize, Serialize};
/// Top-level bridge configuration.
#[derive(Debug, Deserialize, Serialize)]
pub struct Config {
#[serde(default)]
pub radio: RadioSection,
#[serde(default)]
pub bridge: BridgeSection,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct RadioSection {
/// Serial port path (auto-detect if omitted).
pub port: Option<String>,
#[serde(default = "default_frequency")]
pub frequency: u32,
#[serde(default = "default_bandwidth")]
pub bandwidth: String,
#[serde(default = "default_sf")]
pub spreading_factor: u8,
#[serde(default = "default_cr")]
pub coding_rate: u8,
#[serde(default = "default_sync_word")]
pub sync_word: u16,
#[serde(default = "default_tx_power")]
pub tx_power: String,
#[serde(default = "default_preamble")]
pub preamble: u16,
#[serde(default = "default_true")]
pub cad: bool,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct BridgeSection {
/// Shared passphrase. All bridges with the same passphrase join the same swarm.
#[serde(default)]
pub passphrase: String,
#[serde(default = "default_dedup_window")]
pub dedup_window_secs: u64,
#[serde(default = "default_tx_queue_size")]
pub tx_queue_size: usize,
/// Optional manual override for the radio TX rate limit (packets per second).
/// If omitted, a rate is calculated from the radio config.
pub rate_limit_pps: Option<f64>,
pub log_file: Option<PathBuf>,
}
// ── Defaults (MeshCore US/Canada Recommended) ────────────────────
const fn default_frequency() -> u32 {
910_525_000
}
fn default_bandwidth() -> String {
"62.5kHz".into()
}
const fn default_sf() -> u8 {
7
}
const fn default_cr() -> u8 {
5
}
const fn default_sync_word() -> u16 {
0x3444
}
fn default_tx_power() -> String {
"22".into()
}
const fn default_preamble() -> u16 {
16
}
const fn default_true() -> bool {
true
}
const fn default_dedup_window() -> u64 {
300
}
const fn default_tx_queue_size() -> usize {
32
}
impl Default for RadioSection {
fn default() -> Self {
Self {
port: None,
frequency: default_frequency(),
bandwidth: default_bandwidth(),
spreading_factor: default_sf(),
coding_rate: default_cr(),
sync_word: default_sync_word(),
tx_power: default_tx_power(),
preamble: default_preamble(),
cad: true,
}
}
}
impl Default for BridgeSection {
fn default() -> Self {
Self {
passphrase: String::new(),
dedup_window_secs: default_dedup_window(),
tx_queue_size: default_tx_queue_size(),
rate_limit_pps: None,
log_file: None,
}
}
}
// ── Conversion ────────────────────────────────────────────────────
fn parse_bandwidth(s: &str) -> Result<Bandwidth> {
match s.to_lowercase().as_str() {
"7.8khz" | "7khz" | "khz7" => Ok(Bandwidth::Khz7),
"10.4khz" | "10khz" | "khz10" => Ok(Bandwidth::Khz10),
"15.6khz" | "15khz" | "khz15" => Ok(Bandwidth::Khz15),
"20.8khz" | "20khz" | "khz20" => Ok(Bandwidth::Khz20),
"31.25khz" | "31khz" | "khz31" => Ok(Bandwidth::Khz31),
"41.7khz" | "41khz" | "khz41" => Ok(Bandwidth::Khz41),
"62.5khz" | "62khz" | "khz62" => Ok(Bandwidth::Khz62),
"125khz" | "khz125" => Ok(Bandwidth::Khz125),
"250khz" | "khz250" => Ok(Bandwidth::Khz250),
"500khz" | "khz500" => Ok(Bandwidth::Khz500),
_ => bail!("unknown bandwidth: {s}"),
}
}
fn parse_tx_power(s: &str) -> Result<i8> {
if s.eq_ignore_ascii_case("max") {
Ok(TX_POWER_MAX)
} else {
s.parse::<i8>().context("invalid tx_power (use 'max' or a dBm integer)")
}
}
impl RadioSection {
/// 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)?,
sf: self.spreading_factor,
cr: self.coding_rate,
sync_word: self.sync_word,
tx_power_dbm: parse_tx_power(&self.tx_power)?,
preamble_len: self.preamble,
cad: u8::from(self.cad),
})
}
}
// ── File I/O ──────────────────────────────────────────────────────
/// 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");
Ok(dir)
}
/// Default config file path.
pub fn config_path() -> Result<PathBuf> {
Ok(config_dir()?.join("config.toml"))
}
/// Default log file path: `$XDG_STATE_HOME/donglora-bridge/bridge.log`.
pub fn default_log_path() -> Result<PathBuf> {
let dir = dirs::state_dir()
.or_else(dirs::data_local_dir)
.context("cannot determine state directory")?
.join("donglora-bridge");
Ok(dir.join("bridge.log"))
}
/// 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 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");
}
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);
}
}
+769
View File
@@ -0,0 +1,769 @@
//! 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};
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 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};
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::{debug, info, warn};
use crate::packet::GossipFrame;
/// Maximum number of peers to store in a DHT record.
/// 28 × 32 bytes = 896 bytes raw. Bencoded: 28 × ~36 = ~1008.
/// Conservative cap to stay under BEP44's 1000-byte value limit.
const MAX_DHT_PEERS: usize = 27;
/// 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. 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.
pub struct PassphraseKeys {
pub topic_id: TopicId,
pub dht_signing_key: SigningKey,
pub dht_salt: [u8; 20],
}
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: &[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 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 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 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 }
}
}
// ── Swarm state (for TUI) ───────────────────────────────────────
/// Observable state of the gossip swarm.
#[derive(Debug, Clone)]
pub struct SwarmState {
pub topic_hash: String,
pub neighbor_count: usize,
pub dht_status: DhtStatus,
pub last_dht_publish: Option<Instant>,
}
/// Status of the mainline DHT connection.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DhtStatus {
Bootstrapping,
Ready,
PublishFailed,
}
impl std::fmt::Display for DhtStatus {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Bootstrapping => write!(f, "Bootstrapping"),
Self::Ready => write!(f, "Ready"),
Self::PublishFailed => write!(f, "Failed"),
}
}
}
// ── Events to router ────────────────────────────────────────────
/// Events from the gossip swarm to the router.
pub enum GossipEvent {
/// A gossip frame received from the swarm.
Frame(GossipFrame),
/// Neighbor count changed.
NeighborChanged(usize),
}
// ── Gossip handle ───────────────────────────────────────────────
/// Handle to the gossip swarm. Manages iroh endpoint, gossip subscription, and DHT.
pub struct Gossip {
endpoint: Endpoint,
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,
dht_salt: [u8; 20],
cancel: CancellationToken,
}
impl Gossip {
/// Create and start the gossip network.
///
/// Spawns background tasks for connection acceptance, gossip event forwarding,
/// gossip broadcasting, and DHT heartbeat.
///
/// Returns `(handle, event_rx, frame_tx)`:
/// - `event_rx`: channel of events from the swarm to the router
/// - `frame_tx`: channel for the router to send frames to broadcast
pub async fn new(
secret_key: SecretKey,
keys: &PassphraseKeys,
) -> Result<(Self, mpsc::Receiver<GossipEvent>, mpsc::Sender<GossipFrame>)> {
let cancel = CancellationToken::new();
// Build iroh endpoint with n0 discovery.
let endpoint = Endpoint::builder(iroh::endpoint::presets::N0)
.secret_key(secret_key)
.alpns(vec![GOSSIP_ALPN.to_vec()])
.bind()
.await
.context("binding iroh endpoint")?;
let my_id = endpoint.id();
info!("iroh endpoint bound, id: {}", my_id.fmt_short());
// Build gossip actor.
let actor = GossipActor::builder().spawn(endpoint.clone());
// Bootstrap from DHT.
let bootstrap_peers = dht_get_peers(keys).await;
if bootstrap_peers.is_empty() {
info!("no peers found in DHT — we are the first node");
} else {
info!("found {} peer(s) in DHT", bootstrap_peers.len());
}
// Subscribe to gossip topic.
let topic = actor.subscribe(keys.topic_id, bootstrap_peers).await.context("subscribing to gossip topic")?;
let (sender, receiver) = topic.split();
// Initial swarm state.
let topic_hash = hex::encode(&keys.topic_id.as_bytes()[..8]);
let state = Arc::new(Mutex::new(SwarmState {
topic_hash,
neighbor_count: 0,
dht_status: DhtStatus::Bootstrapping,
last_dht_publish: None,
}));
let neighbor_count = Arc::new(AtomicUsize::new(0));
let neighbors: Arc<Mutex<HashSet<PublicKey>>> = Arc::new(Mutex::new(HashSet::new()));
// Channels.
let (event_tx, event_rx) = mpsc::channel::<GossipEvent>(256);
let (frame_tx, frame_rx) = mpsc::channel::<GossipFrame>(256);
let handle = Self {
endpoint: endpoint.clone(),
actor: actor.clone(),
state: state.clone(),
neighbors: neighbors.clone(),
dht_signing_key: keys.dht_signing_key.clone(),
dht_salt: keys.dht_salt,
cancel: cancel.clone(),
};
// Spawn accept loop.
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());
// Clone sender for DHT heartbeat to feed discovered peers via join_peers.
let dht_sender = sender.clone();
// Spawn gossip broadcaster task.
spawn_broadcaster(sender, frame_rx, cancel.clone());
// Spawn DHT heartbeat.
spawn_dht_heartbeat(
my_id,
keys.dht_signing_key.clone(),
keys.dht_salt,
dht_sender,
neighbors,
state,
neighbor_count,
cancel,
);
// Try to join (non-blocking — fire and forget with timeout).
// The receiver task handles events regardless.
Ok((handle, event_rx, frame_tx))
}
/// Snapshot of swarm state for TUI.
#[must_use]
pub fn swarm_state(&self) -> SwarmState {
self.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner).clone()
}
/// Graceful shutdown. Publishes neighbors (without ourselves) to DHT
/// so other nodes can still bootstrap from live peers.
pub async fn shutdown(&self) {
// Publish neighbors-only to DHT before shutting down.
let neighbor_ids: Vec<[u8; 32]> = self
.neighbors
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.iter()
.map(|id| *id.as_bytes())
.collect();
if let Ok(dht) = Dht::client() {
let dht = dht.as_async();
let value = encode_peer_list(&neighbor_ids);
let seq = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.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()));
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:#}"),
}
}
self.cancel.cancel();
let _ = self.actor.shutdown().await;
self.endpoint.close().await;
}
}
// ── Background tasks ────────────────────────────────────────────
/// Accept incoming iroh connections and forward gossip ALPN to the gossip actor.
fn spawn_accept_loop(endpoint: Endpoint, gossip: GossipActor, cancel: CancellationToken) {
tokio::spawn(async move {
loop {
let incoming = tokio::select! {
incoming = endpoint.accept() => incoming,
() = cancel.cancelled() => break,
};
let Some(incoming) = incoming else { break };
let gossip = gossip.clone();
tokio::spawn(handle_incoming(incoming, gossip));
}
debug!("accept loop stopped");
});
}
async fn handle_incoming(incoming: Incoming, gossip: GossipActor) {
let Ok(conn) = incoming.await else { return };
let alpn = conn.alpn();
if *alpn == *GOSSIP_ALPN {
if let Err(e) = gossip.handle_connection(conn).await {
debug!("gossip handle_connection error: {e:#}");
}
} else {
debug!("unexpected ALPN: {:?}", String::from_utf8_lossy(alpn));
}
}
/// Forward gossip events to the router via channel.
fn spawn_receiver(
mut receiver: iroh_gossip::api::GossipReceiver,
event_tx: mpsc::Sender<GossipEvent>,
state: Arc<Mutex<SwarmState>>,
neighbor_count: Arc<AtomicUsize>,
neighbors: Arc<Mutex<HashSet<PublicKey>>>,
cancel: CancellationToken,
) {
tokio::spawn(async move {
loop {
let event = tokio::select! {
event = receiver.next() => event,
() = cancel.cancelled() => break,
};
let Some(Ok(event)) = event else { break };
match event {
Event::Received(msg) => {
if let Some(frame) = GossipFrame::decode(&msg.content) {
let _ = event_tx.send(GossipEvent::Frame(frame)).await;
} else {
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;
}
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;
}
let _ = event_tx.send(GossipEvent::NeighborChanged(count)).await;
}
Event::Lagged => {
warn!("gossip receiver lagged — some messages were dropped");
}
}
}
debug!("gossip receiver stopped");
});
}
/// Read frames from the router and broadcast them to the swarm.
fn spawn_broadcaster(
sender: iroh_gossip::api::GossipSender,
mut frame_rx: mpsc::Receiver<GossipFrame>,
cancel: CancellationToken,
) {
tokio::spawn(async move {
loop {
let frame = tokio::select! {
frame = frame_rx.recv() => frame,
() = cancel.cancelled() => break,
};
let Some(frame) = frame else { break };
let encoded = Bytes::from(frame.encode());
if let Err(e) = sender.broadcast(encoded).await {
debug!("gossip broadcast error: {e:#}");
}
}
debug!("gossip broadcaster stopped");
});
}
/// 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). 30 seconds
/// is much slower than reads because DHT PUT is heavier and a lonely node only
/// adds itself to the peer list.
const DHT_ALONE_PUBLISH_SECS: u64 = 30;
/// Periodically interact with the DHT to discover and share peers.
///
/// Strategy:
/// - **Alone:** Read every 3s (cheap). Publish every 30s via merge — reads
/// existing data first, appends ourselves, writes back. Two fresh nodes
/// both doing this will eventually see each other. Never blindly overwrites.
/// - **Connected:** Read + publish on relaxed 5-min schedule. We have
/// valuable neighbor info to share.
/// - **Transition to connected:** Publish immediately.
#[allow(clippy::too_many_arguments)]
fn spawn_dht_heartbeat(
my_id: PublicKey,
signing_key: SigningKey,
salt: [u8; 20],
gossip_sender: iroh_gossip::api::GossipSender,
neighbors: Arc<Mutex<HashSet<PublicKey>>>,
state: Arc<Mutex<SwarmState>>,
neighbor_count: Arc<AtomicUsize>,
cancel: CancellationToken,
) {
tokio::spawn(async move {
let dht = match Dht::client() {
Ok(d) => d.as_async(),
Err(e) => {
warn!("failed to create DHT client: {e:#}");
return;
}
};
let pub_key_bytes = signing_key.verifying_key().to_bytes();
let mut was_alone = true;
let mut last_publish = Instant::now();
// Initial: read to discover existing peers, then merge-publish ourselves.
dht_read_and_join(&dht, &my_id, &pub_key_bytes, &salt, &gossip_sender, &state).await;
dht_publish(&dht, &my_id, &signing_key, &salt, &pub_key_bytes, &neighbors, &state).await;
loop {
let alone = neighbor_count.load(Ordering::Relaxed) == 0;
// Transition: alone → connected. Publish immediately.
if was_alone && !alone {
info!("first neighbor joined — publishing to DHT");
dht_publish(&dht, &my_id, &signing_key, &salt, &pub_key_bytes, &neighbors, &state).await;
last_publish = Instant::now();
}
was_alone = alone;
// Sleep: fast reads while alone, relaxed when connected.
let sleep_dur = if alone {
Duration::from_secs(DHT_ALONE_READ_SECS)
} else {
let jitter = rand::random::<u64>() % DHT_HEARTBEAT_JITTER_SECS.max(1);
Duration::from_secs(DHT_HEARTBEAT_SECS + jitter)
};
tokio::select! {
() = tokio::time::sleep(sleep_dur) => {}
() = cancel.cancelled() => break,
}
// Always read — cheap, discovers peers to join.
dht_read_and_join(&dht, &my_id, &pub_key_bytes, &salt, &gossip_sender, &state).await;
// Publish cadence depends on state.
let publish_interval = if alone {
Duration::from_secs(DHT_ALONE_PUBLISH_SECS)
} else {
Duration::from_secs(DHT_HEARTBEAT_SECS)
};
if last_publish.elapsed() >= publish_interval {
dht_publish(&dht, &my_id, &signing_key, &salt, &pub_key_bytes, &neighbors, &state).await;
last_publish = Instant::now();
}
}
debug!("DHT heartbeat stopped");
});
}
/// Read the DHT and feed any discovered peers into the gossip swarm.
async fn dht_read_and_join(
dht: &AsyncDht,
my_id: &PublicKey,
pub_key_bytes: &[u8; 32],
salt: &[u8; 20],
gossip_sender: &iroh_gossip::api::GossipSender,
state: &Arc<Mutex<SwarmState>>,
) {
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();
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 {
debug!("join_peers error: {e:#}");
}
}
if let Ok(mut s) = state.lock() {
s.dht_status = DhtStatus::Ready;
}
}
}
async fn dht_publish(
dht: &AsyncDht,
my_id: &PublicKey,
signing_key: &SigningKey,
salt: &[u8; 20],
pub_key_bytes: &[u8; 32],
neighbors: &Arc<Mutex<HashSet<PublicKey>>>,
state: &Arc<Mutex<SwarmState>>,
) {
let has_neighbors = neighbors.lock().is_ok_and(|n| !n.is_empty());
let mut seen = HashSet::new();
let mut peer_list: Vec<[u8; 32]> = Vec::with_capacity(MAX_DHT_PEERS);
// Always include ourselves.
peer_list.push(*my_id.as_bytes());
seen.insert(*my_id.as_bytes());
if has_neighbors {
// Connected: publish authoritative data — self + live neighbors only.
// This purges dead/stale peers from the DHT. We know who's alive
// because they're our active gossip neighbors.
if let Ok(n) = neighbors.lock() {
for id in n.iter() {
if seen.insert(*id.as_bytes()) && peer_list.len() < MAX_DHT_PEERS {
peer_list.push(*id.as_bytes());
}
}
}
} else {
// Alone: merge-publish — read existing DHT data first, add ourselves,
// preserve other peers. We don't know who's alive so we can't purge.
if let Some(record) = dht.get_mutable_most_recent(pub_key_bytes, Some(salt)).await {
for existing in decode_peer_list(record.value()) {
if seen.insert(*existing.as_bytes()) && peer_list.len() < MAX_DHT_PEERS {
peer_list.push(*existing.as_bytes());
}
}
}
}
let value = encode_peer_list(&peer_list);
let seq = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs().cast_signed())
.unwrap_or(0);
let item = MutableItem::new(signing_key.clone(), &value, seq, Some(salt.as_ref()));
match dht.put_mutable(item, None).await {
Ok(_) => {
debug!("published to DHT (seq={seq}, peers={})", peer_list.len());
if let Ok(mut s) = state.lock() {
s.dht_status = DhtStatus::Ready;
s.last_dht_publish = Some(Instant::now());
}
}
Err(e) => {
warn!("DHT publish failed: {e:#}");
if let Ok(mut s) = state.lock() {
s.dht_status = DhtStatus::PublishFailed;
}
}
}
}
// ── DHT helpers ─────────────────────────────────────────────────
/// Query the DHT for existing peers (blocking bootstrap, called once on startup).
async fn dht_get_peers(keys: &PassphraseKeys) -> Vec<PublicKey> {
let dht = match Dht::client() {
Ok(d) => d.as_async(),
Err(e) => {
warn!("failed to create DHT client for bootstrap: {e:#}");
return vec![];
}
};
let pub_key_bytes = keys.dht_signing_key.verifying_key().to_bytes();
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
},
)
}
/// Encode a list of 32-byte peer IDs into bytes for DHT storage.
/// 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());
for id in peers.iter().take(MAX_DHT_PEERS) {
buf.extend_from_slice(id);
}
buf
}
/// Decode a peer list from DHT record value.
#[must_use]
pub fn decode_peer_list(data: &[u8]) -> Vec<PublicKey> {
if data.len() < 2 {
return vec![];
}
let count = u16::from_le_bytes([data[0], data[1]]) as usize;
let mut peers = Vec::with_capacity(count);
let mut offset = 2;
for _ in 0..count {
if offset + 32 > data.len() {
break;
}
let bytes: [u8; 32] = match data[offset..offset + 32].try_into() {
Ok(b) => b,
Err(_) => break,
};
if let Ok(pk) = PublicKey::from_bytes(&bytes) {
peers.push(pk);
}
offset += 32;
}
peers
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn passphrase_keys_deterministic() {
let k1 = PassphraseKeys::derive("test-passphrase");
let k2 = PassphraseKeys::derive("test-passphrase");
assert_eq!(k1.topic_id, k2.topic_id);
assert_eq!(k1.dht_signing_key.to_bytes(), k2.dht_signing_key.to_bytes());
assert_eq!(k1.dht_salt, k2.dht_salt);
}
#[test]
fn passphrase_keys_different() {
let k1 = PassphraseKeys::derive("alpha");
let k2 = PassphraseKeys::derive("beta");
assert_ne!(k1.topic_id, k2.topic_id);
assert_ne!(k1.dht_signing_key.to_bytes(), k2.dht_signing_key.to_bytes());
}
#[test]
fn peer_list_roundtrip() {
// Generate valid ed25519 public keys.
let keys: Vec<_> = (0..3)
.map(|_| {
let sk = iroh::SecretKey::generate(&mut rand::rng());
*sk.public().as_bytes()
})
.collect();
let encoded = encode_peer_list(&keys);
let decoded = decode_peer_list(&encoded);
assert_eq!(decoded.len(), 3);
for (i, key) in keys.iter().enumerate() {
assert_eq!(*decoded[i].as_bytes(), *key);
}
}
#[test]
fn peer_list_empty() {
let encoded = encode_peer_list(&[]);
let decoded = decode_peer_list(&encoded);
assert!(decoded.is_empty());
}
#[test]
fn peer_list_truncated_data() {
// Only 1 byte — too short for count.
assert!(decode_peer_list(&[0]).is_empty());
// Count says 2 but only 1 peer present.
let mut data = vec![2, 0]; // count = 2
data.extend_from_slice(&[0xAA; 32]); // only 1 peer
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;
+210
View File
@@ -0,0 +1,210 @@
use donglora_bridge::{config, gossip, radio, rate_limit, router, setup, tui};
use std::sync::Arc;
use std::time::Instant;
use anyhow::{Context, Result};
use clap::{Parser, Subcommand};
use iroh::SecretKey;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::{info, warn};
use crate::gossip::PassphraseKeys;
use crate::rate_limit::RateLimiter;
use crate::router::Stats;
#[derive(Parser)]
#[command(name = "donglora-bridge", about = "LoRa gossip bridge using iroh")]
struct Cli {
/// Path to config file (default: ~/.config/donglora-bridge/config.toml)
#[arg(long)]
config: Option<String>,
/// Run in headless mode (structured logs to stdout, no TUI)
#[arg(long)]
log_only: bool,
#[command(subcommand)]
command: Option<Commands>,
}
#[derive(Subcommand)]
enum Commands {
/// Interactive configuration wizard
Config,
}
#[tokio::main]
#[allow(clippy::too_many_lines)]
async fn main() -> Result<()> {
let cli = Cli::parse();
// Resolve config path.
let config_path = match &cli.config {
Some(p) => std::path::PathBuf::from(p),
None => config::config_path()?,
};
// Config subcommand: run wizard with existing config as defaults.
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());
}
// First-run: no config file → run wizard automatically.
if !config_path.exists() {
println!(" No config found. Let's set one up!\n");
return setup::run_wizard(&config_path, None);
}
// Load config.
let cfg =
config::load_config(&config_path).with_context(|| format!("loading config from {}", config_path.display()))?;
// Set up logging.
setup_logging(cli.log_only, &cfg);
// Ephemeral identity — fresh keypair each launch.
let secret_key = SecretKey::generate(&mut rand::rng());
let node_id = secret_key.public();
info!("donglora-bridge starting, ephemeral id: {}", node_id.fmt_short());
// Parse radio config.
let radio_config = cfg.radio.to_radio_config()?;
// Derive passphrase keys.
let keys = PassphraseKeys::derive(&cfg.bridge.passphrase);
info!("topic: {}", hex::encode(&keys.topic_id.as_bytes()[..8]));
// Create rate limiter.
let rate_limiter = RateLimiter::from_radio_config(
radio_config.sf,
radio_config.bw,
radio_config.cr,
radio_config.preamble_len,
cfg.bridge.rate_limit_pps,
);
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 = Arc::new(swarm);
// Start radio.
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 {
active: radio_config,
requested: radio_config,
source: radio::ConfigSource::Ours,
device: String::new(),
connected: false,
});
// Cancellation.
let cancel = CancellationToken::new();
let start_time = Instant::now();
// Spawn router.
let router_stats = stats.clone();
let router_cancel = cancel.clone();
let router_our_id = node_id;
let dedup_window = std::time::Duration::from_secs(cfg.bridge.dedup_window_secs);
let tx_queue_size = cfg.bridge.tx_queue_size;
tokio::spawn(async move {
router::run(
router_our_id,
dedup_window,
tx_queue_size,
radio_config,
rate_limiter,
radio_event_rx,
radio_tx,
gossip_event_rx,
gossip_frame_tx,
&router_stats,
log_tx,
config_watch_tx,
)
.await;
router_cancel.cancel();
});
// Spawn a task that cancels on ctrl-c or SIGTERM.
{
let cancel = cancel.clone();
tokio::spawn(async move {
let sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate());
match sigterm {
Ok(mut sigterm) => {
tokio::select! {
_ = tokio::signal::ctrl_c() => info!("received SIGINT"),
_ = sigterm.recv() => info!("received SIGTERM"),
}
}
Err(e) => {
warn!("failed to register SIGTERM handler: {e:#}");
let _ = tokio::signal::ctrl_c().await;
info!("received SIGINT");
}
}
cancel.cancel();
});
}
// Run TUI or headless.
if cli.log_only {
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);
cancel.cancel();
swarm.shutdown().await;
info!("donglora-bridge stopped");
if let Ok(mut term) = terminal {
tui::restore_terminal(&mut term);
}
return Ok(());
}
// Shutdown (headless path).
swarm.shutdown().await;
info!("donglora-bridge stopped");
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();
}
}
+258
View File
@@ -0,0 +1,258 @@
//! 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.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Grade {
/// SNR well above demodulation floor (margin >= 3 dB).
Good,
/// SNR near demodulation floor (0 <= margin < 3 dB).
Marginal,
/// SNR below demodulation floor — likely corrupt.
Unreliable,
/// SNR outside valid range (-32..32) — firmware/decode error.
Invalid,
}
impl Grade {
/// Returns true if the packet is reliable enough to forward.
#[must_use]
pub const fn should_forward(self) -> bool {
matches!(self, Self::Good | Self::Marginal)
}
}
impl std::fmt::Display for Grade {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Good => write!(f, "GOOD"),
Self::Marginal => write!(f, "MARG"),
Self::Unreliable => write!(f, "UNRL"),
Self::Invalid => write!(f, "INVL"),
}
}
}
/// 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;
}
let min_snr = -2.5 * (f32::from(sf) - 4.0);
let margin = f32::from(snr) - min_snr;
if margin < 0.0 {
Grade::Unreliable
} else if margin < 3.0 {
Grade::Marginal
} else {
Grade::Good
}
}
/// A packet received from the radio, before wrapping in a gossip frame.
#[derive(Debug, Clone)]
pub struct RadioPacket {
pub rssi: i16,
pub snr: i16,
pub payload: Vec<u8>,
}
/// A gossip frame — the wire format sent through the iroh-gossip swarm.
///
/// 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.
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.
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 }
}
/// 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);
buf
}
/// 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;
}
let sender: [u8; 32] = data[..32].try_into().ok()?;
let rssi = i16::from_le_bytes(data[32..34].try_into().ok()?);
let snr = i16::from_le_bytes(data[34..36].try_into().ok()?);
let payload_len = u16::from_le_bytes(data[36..38].try_into().ok()?) as usize;
if data.len() != FRAME_HEADER_SIZE + payload_len {
return None;
}
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::*;
#[test]
fn snr_grading() {
// SF7: min_snr = -2.5 * (7 - 4) = -7.5
assert_eq!(snr_grade(10, 7), Grade::Good); // margin = 17.5
assert_eq!(snr_grade(-5, 7), Grade::Marginal); // margin = 2.5 < 3
assert_eq!(snr_grade(-8, 7), Grade::Unreliable); // margin = -0.5
assert_eq!(snr_grade(-50, 7), Grade::Invalid); // out of range
// SF12: min_snr = -2.5 * (12 - 4) = -20.0
assert_eq!(snr_grade(-15, 12), Grade::Good); // margin = 5.0
assert_eq!(snr_grade(-18, 12), Grade::Marginal); // margin = 2.0
assert_eq!(snr_grade(-22, 12), Grade::Unreliable); // margin = -2.0
}
#[test]
fn gossip_frame_roundtrip() {
let key = iroh::SecretKey::generate(&mut rand::rng());
let frame = GossipFrame::new(&key.public(), -85, 8, vec![1, 2, 3, 4, 5]);
let encoded = frame.encode();
let decoded = GossipFrame::decode(&encoded).expect("decode should succeed");
assert_eq!(decoded.sender, *key.public().as_bytes());
assert_eq!(decoded.rssi, -85);
assert_eq!(decoded.snr, 8);
assert_eq!(decoded.payload, vec![1, 2, 3, 4, 5]);
}
#[test]
fn gossip_frame_reject_short() {
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.
let mut data = vec![0u8; 38 + 3];
data[36..38].copy_from_slice(&5u16.to_le_bytes());
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");
let h2 = content_hash(b"hello");
assert_eq!(h1, h2);
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);
}
}
}
}
+444
View File
@@ -0,0 +1,444 @@
//! `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::{Context, Result};
use donglora_client::{RadioConfig, Response, Transport};
use tokio::sync::mpsc;
use tracing::{debug, error, info, warn};
use crate::packet::{RadioPacket, snr_grade};
/// How the radio config was resolved.
#[derive(Debug, Clone, Copy)]
pub enum ConfigSource {
/// We set the config ourselves.
Ours,
/// The mux already had a locked config; we adopted it.
Mux,
}
/// Messages from the radio thread to the router.
pub enum RadioEvent {
/// A valid packet received from the radio (already SNR-filtered).
Packet(RadioPacket),
/// Radio connected (or reconnected) with the active config and device info.
Connected(RadioConfig, ConfigSource, String),
/// Radio disconnected — attempting reconnect.
Disconnected,
}
/// 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>) -> 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))
.context("failed to spawn radio thread")?;
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>,
event_tx: mpsc::Sender<RadioEvent>,
mut tx_recv: mpsc::Receiver<Vec<u8>>,
) {
let initial_backoff = Duration::from_millis(250);
let max_backoff = Duration::from_secs(5);
let mut backoff = initial_backoff;
loop {
let started = std::time::Instant::now();
match connect_and_run(&config, port.as_deref(), &event_tx, &mut tx_recv) {
Ok(()) => {
info!("radio thread exiting");
return;
}
Err(e) => {
error!("radio error: {e:#}");
let _ = event_tx.blocking_send(RadioEvent::Disconnected);
// If we were connected for a while, reset backoff so
// the next reconnect attempt is fast.
if started.elapsed() > Duration::from_secs(5) {
backoff = initial_backoff;
}
info!("reconnecting in {backoff:?}");
std::thread::sleep(backoff);
backoff = (backoff * 2).min(max_backoff);
}
}
}
}
fn connect_and_run(
config: &RadioConfig,
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.
let timeout = Duration::from_secs(10);
info!("connecting to dongle...");
let (mut client, device) = if let Some(port) = port {
let c = donglora_client::connect(Some(port), timeout)?;
(c, shorten_path(port))
} else if let Ok(c) = donglora_client::connect_mux_auto(timeout) {
(c, "mux".to_string())
} else {
let port_path = donglora_client::find_port().ok_or_else(|| anyhow::anyhow!("no mux or USB dongle found"))?;
info!("no mux found, connecting to serial port {port_path}");
let c = donglora_client::connect(Some(&port_path), timeout)?;
(c, shorten_path(&port_path))
};
info!("[radio] transport connected: {device}");
info!("[radio] sending ping...");
match client.ping() {
Ok(()) => info!("[radio] ping OK"),
Err(e) => {
info!("[radio] ping FAILED: {e:#}");
return Err(e);
}
}
info!("[radio] negotiating config...");
let (active_config, config_source) = match negotiate_config(&mut client, config, &device) {
Ok(result) => {
info!("[radio] config negotiated: source={:?}, config={}", result.1, format_radio_config(&result.0));
result
}
Err(e) => {
info!("[radio] config negotiation FAILED: {e:#}");
return Err(e);
}
};
info!("[radio] sending start_rx...");
match client.start_rx() {
Ok(()) => info!("[radio] start_rx OK"),
Err(e) => {
info!("[radio] start_rx FAILED: {e:#}");
return Err(e);
}
}
let _ = event_tx.blocking_send(RadioEvent::Connected(active_config, config_source, device));
// Set a short timeout so we can interleave RX and TX.
client.transport_mut().set_timeout(Duration::from_millis(100))?;
info!("[radio] entering main loop");
// Liveness: ping the mux periodically when idle to detect a dead connection.
// Without this, a killed mux causes recv() to silently return Ok(None) forever.
let liveness_interval = Duration::from_secs(2);
let mut last_activity = std::time::Instant::now();
loop {
// Check for TX requests (non-blocking).
while let Ok(payload) = tx_recv.try_recv() {
debug!("TX {} bytes", payload.len());
if let Err(e) = client.transmit(&payload, None) {
warn!("transmit error: {e:#}");
return Err(e);
}
last_activity = std::time::Instant::now();
}
// Check for RX packets.
match client.recv() {
Ok(Some(Response::RxPacket { rssi, snr, payload })) => {
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());
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());
}
}
Ok(Some(_)) => {
last_activity = std::time::Instant::now();
}
Ok(None) => {}
Err(e) if is_timeout_error(&e) => {}
Err(e) => return Err(e),
}
// Liveness check: if no real activity for a while, ping the mux.
if last_activity.elapsed() >= liveness_interval {
// Temporarily extend the timeout for the ping command.
let _ = client.transport_mut().set_timeout(Duration::from_secs(2));
match client.ping() {
Ok(()) => {
last_activity = std::time::Instant::now();
}
Err(e) => {
warn!("liveness ping failed: {e:#}");
return Err(e);
}
}
// Restore the short timeout.
let _ = client.transport_mut().set_timeout(Duration::from_millis(100));
}
if event_tx.is_closed() {
return Ok(());
}
}
}
/// Negotiate radio config with the dongle/mux.
///
/// Strategy:
/// 1. `GetConfig` first to read what's active.
/// 2. If it matches our desired config, use it — no `SetConfig` needed.
/// 3. If it differs and we're on the mux, accept the mux's config.
/// 4. If it differs and we're on direct serial, `SetConfig` to push ours.
/// 5. If `GetConfig` fails, try `SetConfig` with our desired config.
fn negotiate_config<T: Transport>(
client: &mut donglora_client::Client<T>,
desired: &RadioConfig,
device: &str,
) -> Result<(RadioConfig, ConfigSource)> {
info!("[negotiate] get_config...");
match client.get_config() {
Ok(cfg) => {
if configs_match(&cfg, desired) {
info!("[negotiate] get_config OK (matches desired): {}", format_radio_config(&cfg));
Ok((cfg, ConfigSource::Ours))
} else if device == "mux" {
info!("[negotiate] get_config OK (mux config differs, accepting): {}", format_radio_config(&cfg));
Ok((cfg, ConfigSource::Mux))
} else {
info!("[negotiate] get_config OK (serial config differs, setting ours)");
client.set_config(*desired)?;
info!("[negotiate] set_config OK");
Ok((*desired, ConfigSource::Ours))
}
}
Err(e) => {
info!("[negotiate] get_config FAILED: {e:#}");
// No config on radio — try to set ours.
info!("[negotiate] trying set_config...");
match client.set_config(*desired) {
Ok(()) => {
info!("[negotiate] set_config OK");
Ok((*desired, ConfigSource::Ours))
}
Err(e2) => {
info!("[negotiate] set_config FAILED: {e2:#}");
anyhow::bail!("GetConfig failed ({e:#}) and SetConfig also failed ({e2:#})");
}
}
}
}
}
/// Shorten a device path to just the filename for display.
fn shorten_path(path: &str) -> 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 {
a.freq_hz == b.freq_hz
&& a.bw == b.bw
&& a.sf == b.sf
&& a.cr == b.cr
&& a.sync_word == b.sync_word
&& a.tx_power_dbm == b.tx_power_dbm
&& a.preamble_len == b.preamble_len
&& a.cad == b.cad
}
fn is_timeout_error(e: &anyhow::Error) -> bool {
e.downcast_ref::<std::io::Error>()
.is_some_and(|io_err| matches!(io_err.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut))
}
/// Format a bandwidth value for display.
#[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",
donglora_client::Bandwidth::Khz15 => "15.6 kHz",
donglora_client::Bandwidth::Khz20 => "20.8 kHz",
donglora_client::Bandwidth::Khz31 => "31.25 kHz",
donglora_client::Bandwidth::Khz41 => "41.7 kHz",
donglora_client::Bandwidth::Khz62 => "62.5 kHz",
donglora_client::Bandwidth::Khz125 => "125 kHz",
donglora_client::Bandwidth::Khz250 => "250 kHz",
donglora_client::Bandwidth::Khz500 => "500 kHz",
}
}
/// Format a `RadioConfig` for display.
#[must_use]
pub fn format_radio_config(config: &RadioConfig) -> String {
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 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"));
}
}
+343
View File
@@ -0,0 +1,343 @@
//! `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,
/// bandwidth, coding rate) and enforces it with a token bucket that allows short bursts.
pub struct RateLimiter {
tokens: f64,
max_tokens: f64,
refill_rate: f64, // tokens per second
last_refill: Instant,
}
/// 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. 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.
#[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 }
},
|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.
pub fn try_acquire(&mut self) -> bool {
self.refill();
if self.tokens >= 1.0 {
self.tokens -= 1.0;
true
} else {
false
}
}
/// Current rate in packets per second.
#[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 = elapsed.mul_add(self.refill_rate, self.tokens).min(self.max_tokens);
self.last_refill = now;
}
}
/// Estimate `LoRa` packet air time using the Semtech `LoRa` time-on-air formula.
///
/// Parameters:
/// - `payload_bytes`: payload size in bytes
/// - `sf`: spreading factor (7–12)
/// - `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 = 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 };
// Payload symbol count (Semtech formula with CRC enabled, explicit header).
#[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;
Duration::from_secs_f64(total)
}
const fn bandwidth_hz(bw: Bandwidth) -> f64 {
match bw {
Bandwidth::Khz7 => 7_800.0,
Bandwidth::Khz10 => 10_400.0,
Bandwidth::Khz15 => 15_600.0,
Bandwidth::Khz20 => 20_800.0,
Bandwidth::Khz31 => 31_250.0,
Bandwidth::Khz41 => 41_700.0,
Bandwidth::Khz62 => 62_500.0,
Bandwidth::Khz125 => 125_000.0,
Bandwidth::Khz250 => 250_000.0,
Bandwidth::Khz500 => 500_000.0,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn air_time_sf7_bw125() {
let t = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16);
// SF7/125kHz/CR4/5/16-preamble/50 bytes: ~100ms range
assert!(t.as_millis() > 50, "air time too short: {t:?}");
assert!(t.as_millis() < 300, "air time too long: {t:?}");
}
#[test]
fn air_time_sf12_bw125() {
let t = lora_air_time(50, 12, Bandwidth::Khz125, 5, 16);
// SF12/125kHz: should be multiple seconds
assert!(t.as_secs() >= 1, "air time too short: {t:?}");
assert!(t.as_secs() < 10, "air time too long: {t:?}");
}
#[test]
fn rate_limiter_allows_burst() {
let mut rl = RateLimiter::from_radio_config(7, Bandwidth::Khz125, 5, 16, None);
// Should allow burst of 3
assert!(rl.try_acquire());
assert!(rl.try_acquire());
assert!(rl.try_acquire());
// 4th should fail (no time for refill)
assert!(!rl.try_acquire());
}
#[test]
fn rate_limiter_override() {
let rl = RateLimiter::from_radio_config(7, Bandwidth::Khz125, 5, 16, Some(42.0));
assert!((rl.rate_pps() - 42.0).abs() < f64::EPSILON);
}
#[test]
fn air_time_zero_preamble_uses_default() {
let t_default = lora_air_time(50, 7, Bandwidth::Khz125, 5, 0);
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");
}
}
}
}
}
+424
View File
@@ -0,0 +1,424 @@
//! 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};
use iroh::PublicKey;
use tokio::sync::mpsc;
use tracing::{debug, warn};
use crate::gossip::GossipEvent;
use crate::packet::{GossipFrame, content_hash};
use crate::radio::{ConfigSource, RadioEvent};
use crate::rate_limit::RateLimiter;
/// Active radio config info, sent to the TUI via watch channel.
#[derive(Debug, Clone)]
pub struct RadioConfigInfo {
pub active: donglora_client::RadioConfig,
pub requested: donglora_client::RadioConfig,
pub source: ConfigSource,
pub device: String,
pub connected: bool,
}
/// Packet event for the TUI log.
#[derive(Debug, Clone)]
pub struct PacketLogEntry {
pub timestamp: Instant,
pub hash: [u8; 32],
pub direction: PacketDirection,
pub size: usize,
pub snr: Option<i16>,
pub rssi: Option<i16>,
pub action: PacketAction,
}
/// Which side the packet arrived from.
#[derive(Debug, Clone, Copy)]
pub enum PacketDirection {
/// Heard on local radio → forwarded to gossip swarm.
RadioIn,
/// Received from gossip swarm → queued for radio TX.
GossipIn,
}
/// What happened to the packet.
#[derive(Debug, Clone, Copy)]
pub enum PacketAction {
/// Successfully bridged (radio→gossip or gossip→radio).
Bridged,
/// Dropped: duplicate payload.
DroppedDedup,
/// Dropped: TX queue full.
DroppedQueueFull,
/// Dropped: rate limiter.
DroppedRateLimit,
}
impl std::fmt::Display for PacketAction {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Bridged => write!(f, ""),
Self::DroppedDedup => write!(f, "DUP"),
Self::DroppedQueueFull => write!(f, "FULL"),
Self::DroppedRateLimit => write!(f, "RATE"),
}
}
}
/// Aggregate stats for the TUI.
#[derive(Debug, Default)]
pub struct Stats {
pub radio_rx: AtomicU64,
pub radio_tx: AtomicU64,
pub gossip_rx: AtomicU64,
pub gossip_tx: AtomicU64,
pub dedup_hits: AtomicU64,
pub rate_limit_drops: AtomicU64,
pub dropped_queue: AtomicU64,
pub neighbor_count: AtomicU64,
pub radio_connected: AtomicU64, // 0 = disconnected, 1 = connected
}
impl Stats {
#[must_use]
pub fn snapshot(&self) -> StatsSnapshot {
StatsSnapshot {
radio_rx: self.radio_rx.load(Ordering::Relaxed),
radio_tx: self.radio_tx.load(Ordering::Relaxed),
gossip_rx: self.gossip_rx.load(Ordering::Relaxed),
gossip_tx: self.gossip_tx.load(Ordering::Relaxed),
dedup_hits: self.dedup_hits.load(Ordering::Relaxed),
rate_limit_drops: self.rate_limit_drops.load(Ordering::Relaxed),
dropped_queue: self.dropped_queue.load(Ordering::Relaxed),
neighbor_count: self.neighbor_count.load(Ordering::Relaxed),
radio_connected: self.radio_connected.load(Ordering::Relaxed) != 0,
}
}
}
/// Point-in-time snapshot of bridge statistics, cloned from atomic counters.
#[derive(Debug, Clone)]
pub struct StatsSnapshot {
pub radio_rx: u64,
pub radio_tx: u64,
pub gossip_rx: u64,
pub gossip_tx: u64,
pub dedup_hits: u64,
pub rate_limit_drops: u64,
pub dropped_queue: u64,
pub neighbor_count: u64,
pub radio_connected: bool,
}
/// Time-bounded dedup cache.
struct DedupCache {
entries: HashMap<[u8; 32], Instant>,
ttl: Duration,
}
impl DedupCache {
fn new(ttl: Duration) -> Self {
Self { entries: HashMap::new(), ttl }
}
/// Returns true if the key was already present (i.e., duplicate).
fn check_and_insert(&mut self, key: [u8; 32]) -> bool {
let now = Instant::now();
if let Some(ts) = self.entries.get(&key)
&& now.duration_since(*ts) < self.ttl
{
return true;
}
self.entries.insert(key, now);
false
}
fn prune(&mut self) {
let now = Instant::now();
self.entries.retain(|_, ts| now.duration_since(*ts) < self.ttl);
}
}
/// 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, clippy::too_many_lines)]
pub async fn run(
our_id: PublicKey,
dedup_window: Duration,
tx_queue_size: usize,
requested_radio_config: donglora_client::RadioConfig,
mut rate_limiter: RateLimiter,
mut radio_rx: mpsc::Receiver<RadioEvent>,
radio_tx: mpsc::Sender<Vec<u8>>,
mut gossip_rx: mpsc::Receiver<GossipEvent>,
gossip_tx: mpsc::Sender<GossipFrame>,
stats: &Stats,
log_tx: mpsc::Sender<PacketLogEntry>,
config_tx: tokio::sync::watch::Sender<RadioConfigInfo>,
) {
let mut dedup = DedupCache::new(dedup_window);
let mut tx_queue: VecDeque<Vec<u8>> = VecDeque::with_capacity(tx_queue_size);
let mut last_cleanup = Instant::now();
let cleanup_interval = Duration::from_secs(30);
loop {
// Periodic dedup cleanup.
if last_cleanup.elapsed() >= cleanup_interval {
dedup.prune();
last_cleanup = Instant::now();
}
tokio::select! {
// Radio events.
event = radio_rx.recv() => {
let Some(event) = event else { break };
match event {
RadioEvent::Packet(pkt) => {
stats.radio_rx.fetch_add(1, Ordering::Relaxed);
let hash = content_hash(&pkt.payload);
if dedup.check_and_insert(hash) {
stats.dedup_hits.fetch_add(1, Ordering::Relaxed);
debug!("radio RX dedup suppressed ({} bytes)", pkt.payload.len());
let _ = log_tx.send(PacketLogEntry {
timestamp: Instant::now(),
hash,
direction: PacketDirection::RadioIn,
size: pkt.payload.len(),
snr: Some(pkt.snr),
rssi: Some(pkt.rssi),
action: PacketAction::DroppedDedup,
}).await;
continue;
}
// Wrap and broadcast to gossip.
let frame = GossipFrame::new(&our_id, pkt.rssi, pkt.snr, pkt.payload.clone());
stats.gossip_tx.fetch_add(1, Ordering::Relaxed);
let _ = gossip_tx.send(frame).await;
let _ = log_tx.send(PacketLogEntry {
timestamp: Instant::now(),
hash,
direction: PacketDirection::RadioIn,
size: pkt.payload.len(),
snr: Some(pkt.snr),
rssi: Some(pkt.rssi),
action: PacketAction::Bridged,
}).await;
}
RadioEvent::Connected(active_config, source, device) => {
stats.radio_connected.store(1, Ordering::Relaxed);
let _ = config_tx.send(RadioConfigInfo {
active: active_config,
requested: requested_radio_config,
source,
device,
connected: true,
});
}
RadioEvent::Disconnected => {
stats.radio_connected.store(0, Ordering::Relaxed);
config_tx.send_modify(|info| info.connected = false);
}
}
}
// Gossip events.
event = gossip_rx.recv() => {
let Some(event) = event else { break };
match event {
GossipEvent::Frame(frame) => {
stats.gossip_rx.fetch_add(1, Ordering::Relaxed);
let hash = content_hash(&frame.payload);
if dedup.check_and_insert(hash) {
stats.dedup_hits.fetch_add(1, Ordering::Relaxed);
debug!("gossip RX dedup suppressed ({} bytes)", frame.payload.len());
let _ = log_tx.send(PacketLogEntry {
timestamp: Instant::now(),
hash,
direction: PacketDirection::GossipIn,
size: frame.payload.len(),
snr: Some(frame.snr),
rssi: Some(frame.rssi),
action: PacketAction::DroppedDedup,
}).await;
continue;
}
// Rate limit check.
if !rate_limiter.try_acquire() {
stats.rate_limit_drops.fetch_add(1, Ordering::Relaxed);
debug!("rate limited ({} bytes)", frame.payload.len());
let _ = log_tx.send(PacketLogEntry {
timestamp: Instant::now(),
hash,
direction: PacketDirection::GossipIn,
size: frame.payload.len(),
snr: Some(frame.snr),
rssi: Some(frame.rssi),
action: PacketAction::DroppedRateLimit,
}).await;
continue;
}
// Enqueue for radio TX.
if tx_queue.len() >= tx_queue_size {
stats.dropped_queue.fetch_add(1, Ordering::Relaxed);
warn!("TX queue full, dropping packet ({} bytes)", frame.payload.len());
let _ = log_tx.send(PacketLogEntry {
timestamp: Instant::now(),
hash,
direction: PacketDirection::GossipIn,
size: frame.payload.len(),
snr: Some(frame.snr),
rssi: Some(frame.rssi),
action: PacketAction::DroppedQueueFull,
}).await;
continue;
}
tx_queue.push_back(frame.payload.clone());
let _ = log_tx.send(PacketLogEntry {
timestamp: Instant::now(),
hash,
direction: PacketDirection::GossipIn,
size: frame.payload.len(),
snr: Some(frame.snr),
rssi: Some(frame.rssi),
action: PacketAction::Bridged,
}).await;
}
GossipEvent::NeighborChanged(count) => {
#[allow(clippy::cast_possible_truncation)]
stats.neighbor_count.store(count as u64, Ordering::Relaxed);
}
}
}
}
// Drain TX queue to radio.
while let Some(payload) = tx_queue.pop_front() {
stats.radio_tx.fetch_add(1, Ordering::Relaxed);
if radio_tx.send(payload).await.is_err() {
break;
}
}
}
}
#[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");
}
}
+194
View File
@@ -0,0 +1,194 @@
//! Interactive console setup wizard for bridge configuration.
use std::io::{self, Write};
use std::path::Path;
use anyhow::{Context, Result};
use crate::config::{BridgeSection, Config, RadioSection};
/// ANSI escape helpers.
const BOLD: &str = "\x1b[1m";
const DIM: &str = "\x1b[2m";
const CYAN: &str = "\x1b[36m";
const GREEN: &str = "\x1b[32m";
const YELLOW: &str = "\x1b[33m";
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 (radio, bridge) = defaults;
println!();
println!(" {BOLD}donglora-bridge setup{RESET}");
println!(" {DIM}Configure your LoRa gossip bridge.{RESET}");
println!();
// ── Passphrase ───────────────────────────────────────────────
let default_pass = if bridge.passphrase.is_empty() || bridge.passphrase == "change-me" {
None
} else {
Some(bridge.passphrase.as_str())
};
let passphrase = prompt_required("Passphrase", default_pass)?;
println!();
println!(" {BOLD}Radio Settings{RESET} {DIM}(Enter to keep default){RESET}");
println!(" {DIM}────────────────────────────────────{RESET}");
// ── Radio ────────────────────────────────────────────────────
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)?;
let sf_default = format!("{}", radio.spreading_factor);
let spreading_factor: u8 = prompt_parse("Spreading Factor", &sf_default)?;
let cr_default = format!("{}", radio.coding_rate);
let coding_rate: u8 = prompt_parse("Coding Rate", &cr_default)?;
let sw_default = format!("0x{:04X}", radio.sync_word);
let sync_word_str = prompt_default("Sync Word", &sw_default)?;
let sync_word = parse_sync_word(&sync_word_str)?;
let tx_power = prompt_default("TX Power (dBm or 'max')", &radio.tx_power)?;
let pre_default = format!("{}", radio.preamble);
let preamble: u16 = prompt_parse("Preamble", &pre_default)?;
let cad_default = if radio.cad { "on" } else { "off" };
let cad_str = prompt_default("CAD", cad_default)?;
let cad = matches!(cad_str.to_lowercase().as_str(), "on" | "true" | "yes" | "1");
let port_default = radio.port.as_deref().unwrap_or("auto");
let port_str = prompt_default("Serial Port", port_default)?;
let port = if port_str == "auto" || port_str.is_empty() { None } else { Some(port_str) };
// ── Build and save ───────────────────────────────────────────
let cfg = Config {
radio: RadioSection {
port,
frequency,
bandwidth,
spreading_factor,
coding_rate,
sync_word,
tx_power,
preamble,
cad,
},
bridge: BridgeSection {
passphrase,
dedup_window_secs: bridge.dedup_window_secs,
tx_queue_size: bridge.tx_queue_size,
rate_limit_pps: bridge.rate_limit_pps,
log_file: bridge.log_file,
},
};
write_config(config_path, &cfg)?;
println!();
println!(" {GREEN}Config saved to {}{RESET}", config_path.display());
println!(" Run {CYAN}donglora-bridge{RESET} to start!");
println!();
Ok(())
}
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()))?;
}
let content = toml::to_string_pretty(cfg).context("serializing config")?;
std::fs::write(path, content).with_context(|| format!("writing config: {}", path.display()))?;
Ok(())
}
// ── Prompt helpers ───────────────────────────────────────────────
fn prompt_default(label: &str, default: &str) -> Result<String> {
print!(" {label} {DIM}[{default}]{RESET}: ");
io::stdout().flush()?;
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()) }
}
fn prompt_required(label: &str, default: Option<&str>) -> Result<String> {
loop {
let prompt = default.map_or_else(|| format!(" {label}: "), |d| format!(" {label} {DIM}[{d}]{RESET}: "));
print!("{prompt}");
io::stdout().flush()?;
let mut input = String::new();
io::stdin().read_line(&mut input)?;
let trimmed = input.trim();
if !trimmed.is_empty() {
return Ok(trimmed.to_string());
}
if let Some(d) = default {
return Ok(d.to_string());
}
println!(" {YELLOW}This field is required.{RESET}");
}
}
fn prompt_parse<T: std::str::FromStr>(label: &str, default: &str) -> Result<T>
where
T::Err: std::fmt::Display,
{
loop {
let input = prompt_default(label, default)?;
match input.parse::<T>() {
Ok(v) => return Ok(v),
Err(e) => println!(" {YELLOW}Invalid input: {e}{RESET}"),
}
}
}
fn parse_sync_word(s: &str) -> Result<u16> {
let s = s.trim().to_lowercase();
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());
}
}
+1153
View File
File diff suppressed because it is too large. Load diff