13 Commits
Author SHA1 Message Date
swaits b4cc826e74 chore: release 2.0.0 2026-04-22 23:21:07 -06:00
swaits 63b59482fe feat!: factor DHT rendezvous into iroh-gossip-rendezvous 0.1
The partition-healing / DHT-rendezvous algorithm that used to live in
`src/gossip.rs` is now its own crate, published as
`iroh-gossip-rendezvous` 0.1.0. Bridge depends on it for swarm
discovery; the 760-line gossip module shrinks to ~75 lines of glue.

`dht-partition-healing.md` is reduced to a pointer at the new
crate's `PROTOCOL.md` — the normative spec moved with the code.

### Dependency changes

- **New:** `iroh-gossip-rendezvous = "0.1"` (path + version for local
  dev; publish strips the path).
- **Bumped:** `iroh` and `iroh-gossip` 0.97 → 0.98 (major).
- **Removed:** direct deps on `mainline`, `sha2`, `zeroize` — all
  now transitive via `iroh-gossip-rendezvous`.
- `donglora-client` spec loosened from `"1.0.0"` to `"1"` so patch
  and minor releases of the client propagate automatically.

BREAKING CHANGE: any direct users of `bridge::gossip` internals
relied on the in-bridge partition-healing types that have moved.
Import from `iroh_gossip_rendezvous::…` instead.
2026-04-22 23:20:24 -06:00
swaits 43b30952f8 chore: release 1.0.1 2026-04-22 10:47:49 -06:00
swaits 599cfd14dc chore: bump rand past RUSTSEC-2026-0097
cargo update -p rand@0.9.2 and cargo update -p rand@0.8.5 take the two pinned rand versions past the RUSTSEC-2026-0097 fix windows:


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

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


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

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

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

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

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

* Stats gain tx_retries and tx_failures counters surfaced in StatsSnapshot.

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

* The explicit 2 s liveness ping loop is gone. The Dongle runs a 500 ms internal keepalive, matching spec §3.4 inactivity window with 2x margin.
2026-04-22 09:48:01 -06:00
swaits 030280791e fix: sticky mux reconnect prevents USB port theft
Once the bridge connects via mux, all subsequent reconnects use mux-only
mode (connect_mux_auto). Previously, each reconnect re-ran the full
mux→USB fallback chain, which could steal the serial port from the mux
during a brief disconnect.
2026-04-07 18:05:56 -06:00
swaits 222802cf13 feat: upgrade to donglora-client 0.2, simplify connection, fix org URLs
- Upgrade donglora-client from 0.1 to 0.2 (ping-on-connect validation)
- Remove manual mux/serial fallback and manual ping — handled by client now
- Fix repository/homepage URLs to github.com/donglora/bridge
- Fix README links to donglora GitHub org
2026-04-07 16:11:08 -06:00
swaits 394acaee6e chore: add consistent mise.toml, justfile, and tooling config 2026-04-07 14:43:26 -06:00
swaits 943f2615f3 fix: reduce TUI minimum width from 86 to 80 columns 2026-04-06 17:08:56 -06:00
swaits ae284d571a fix: remove stale sibling-clone CI step (donglora-client is on crates.io) 2026-04-06 17:05:22 -06:00
swaits ca26368136 feat: auto-detect connection — fall back to direct USB serial when mux is unavailable 2026-04-06 16:17:39 -06:00
swaits 6bae65c464 feat: publish-ready overhaul — lints, tests, docs, CI, fuzzing, mutant testing
- Enable clippy pedantic + nursery lints; fix all warnings
- Add config validation (SF, CR, frequency range checks)
- Zeroize passphrase intermediates after key derivation
- Convert to lib+bin crate for fuzz target accessibility
- Replace Vec with VecDeque for O(1) log/bucket rotation
- Add TUI minimum terminal size guard

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

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

Infrastructure:
- GitHub Actions CI (fmt, clippy, test, audit)
- cargo-mutants config (.cargo/mutants.toml)
- justfile for common tasks
2026-04-06 14:52:18 -06:00
27 changed files with 4438 additions and 1678 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
+10
View File
@@ -1 +1,11 @@
/target
/fuzz/target
/fuzz/corpus
/fuzz/Cargo.lock
/mutants.out
/mutants.out.old
.jj/
*.swp
*.swo
*~
.DS_Store
+148
View File
@@ -0,0 +1,148 @@
# Changelog
## [2.0.0] - 2026-04-23
### Breaking
- **Partition-healing / DHT-rendezvous extracted to its own crate.**
The 700+ lines of gossip-partition-healing logic in `src/gossip.rs`
are now the `iroh-gossip-rendezvous` crate on crates.io (pinned
here at `^0.1`). Any downstream code that reached into
`bridge::gossip::*` for partition types needs to import from
`iroh_gossip_rendezvous::*` instead. The normative protocol spec
moved with the code — `dht-partition-healing.md` at the bridge
root is now a pointer to the new crate's `PROTOCOL.md`.
- **`iroh` and `iroh-gossip` bumped 0.97 → 0.98** (major; carries
iroh's own breaking changes through to any bridge consumers that
import from those crates transitively).
### Changed
- Direct deps `mainline`, `sha2`, `zeroize` removed — all now
reachable transitively via `iroh-gossip-rendezvous`. Closure is
smaller and the rendezvous concern lives in exactly one place.
- `donglora-client` version spec loosened `"1.0.0"` → `"1"`. Patch
and minor releases of the client propagate on the next
`cargo build` / `cargo install` without a bridge re-publish;
already-released `donglora-client 1.0.1` (CH340K USB bridge
support) is picked up automatically.
## [1.0.1] - 2026-04-22
### Security
- Bump both pinned `rand` versions past [RUSTSEC-2026-0097][adv] (`rand`
is unsound with a custom `log::Log` impl that calls `rand::rng()` at
trace/warn level and hits the 64 KB reseed path):
- `rand 0.9.2` → `0.9.4` (fix `>= 0.9.3`).
- `rand 0.8.5` → `0.8.6` (fix `>= 0.8.6`).
The advisory is only reachable under a very narrow set of conditions
(custom logger invoking `rand::rng()` during a reseed while emitting
trace-level logs, or warn-level with a failed `getrandom`). We are
not known to do that, but shipping the upgrade is the right call.
[adv]: https://rustsec.org/advisories/RUSTSEC-2026-0097
## [1.0.0] - 2026-04-22
### Breaking
- **Upgraded to `donglora-client` 1.0.0-alpha.1 (DongLoRa Protocol v2 wire protocol).**
Wire-incompatible with 0.x firmware. Connect/config/TX/RX all go through
the new async `Dongle` API.
- **Radio loop is now a tokio task**, not a dedicated `std::thread`.
Everything is async end-to-end.
- **`[radio]` config schema changes:**
- `tx_power` renamed to `tx_power_dbm` and is now a plain `i8` (no
`"max"` keyword — v1.0 exposes a concrete dBm range via `GET_INFO`).
- `cad` field removed — CAD is a per-TX flag in v1.0, not a config
property. The `[tx]` section's `skip_cad` (default false) controls it.
- Default `sync_word` changed `0x3444` → `0x1424` to match the
`SX126x` register encoding of `MeshCore`'s private sync byte
(`RADIOLIB_SX126X_SYNC_WORD_PRIVATE`).
### Added
- **TX retry with backoff (new `[tx]` config section).** Every TX now
runs through `Dongle::tx_with_retry`, which retries `TX_DONE(CHANNEL_BUSY)`
and `ERR(EBUSY)` per spec `PROTOCOL.md §6.10` / §C.5.5. Defaults: 3
attempts, randomized 20–100 ms backoff, doubling up to 500 ms cap,
5 s per-attempt timeout. Configurable via `[tx]` keys `max_attempts`,
`backoff_min_ms`, `backoff_max_ms`, `backoff_cap_ms`,
`backoff_multiplier`, `per_attempt_timeout_secs`.
- **TUI packet log shows retry state.** In-flight retries render with a
`↻` marker in yellow and an `N/M` attempt counter. Terminal TX failures
render in red with `✗`. Post-retry success shows the final attempt count
(`OK 2/3`).
- **Stats: `tx_retries` and `tx_failures` counters.** Lifetime totals,
surfaced in the `StatsSnapshot`.
- **Measured airtime** from `TX_DONE` is now reported instead of a
theoretical estimate, feeding the rate limiter with the real on-air
duration.
### Removed
- The explicit 2-second liveness ping loop in the radio thread. The
`Dongle` now runs a 500 ms keepalive task internally, matching spec
§3.4's 1 s inactivity window with 2× margin.
## [0.3.1] - 2026-04-07
### Fixed
- **Sticky mux reconnect.** Once the bridge connects via mux, all subsequent
reconnects use mux-only mode (`connect_mux_auto`). Previously, each reconnect
re-ran the full mux→USB fallback chain, which could steal the serial port
from the mux during a brief disconnect.
## [0.3.0] - 2026-04-07
### Changed
- Upgraded to `donglora-client` 0.2: all connections now auto-validate via
ping-on-connect, rejecting non-DongLoRa serial devices within 200ms.
- Simplified connection logic — removed manual mux/serial fallback and manual
ping; `donglora_client::connect()` handles the full fallback chain and
validation internally.
- Config negotiation detects mux vs serial via transport type instead of
string comparison.
### Fixed
- Repository and homepage URLs now point to `github.com/donglora/bridge`
(was `swaits/donglora-bridge`).
- README links updated to `donglora` GitHub org.
## [0.2.2] - 2026-04-06
### 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
+1548 -530
View File
File diff suppressed because it is too large. Load diff
+24 -9
View File
@@ -1,23 +1,30 @@
[package]
name = "donglora-bridge"
version = "0.1.0"
version = "2.0.0"
edition = "2024"
authors = ["Stephen Waits <steve@waits.net>"]
license = "MIT"
description = "Peer-to-peer LoRa bridge using iroh and DongLoRa"
repository = "https://github.com/donglora/bridge"
homepage = "https://github.com/donglora/bridge"
keywords = ["lora", "bridge", "gossip", "p2p", "iot"]
categories = ["network-programming", "command-line-utilities"]
readme = "README.md"
rust-version = "1.94.1"
exclude = [".jj/", ".github/"]
[dependencies]
donglora-client = { path = "../donglora/clients/rust" }
iroh = "0.97"
iroh-gossip = "0.97"
mainline = { version = "6", features = ["async"] }
donglora-client = "1"
iroh = "0.98"
iroh-gossip = "0.98"
iroh-gossip-rendezvous = { version = "0.1", path = "../../iroh-gossip-rendezvous" }
tokio = { version = "1", features = ["full"] }
clap = { version = "4", features = ["derive"] }
serde = { version = "1", features = ["derive"] }
toml = "0.8"
toml = "1"
blake3 = "1"
ratatui = "0.29"
crossterm = "0.28"
ratatui = "0.30"
crossterm = "0.29"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
tracing-appender = "0.2"
@@ -27,11 +34,19 @@ 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"
[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.
+290
View File
@@ -0,0 +1,290 @@
# donglora-bridge
[![CI](https://github.com/donglora/bridge/actions/workflows/ci.yml/badge.svg)](https://github.com/donglora/bridge/actions/workflows/ci.yml)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)
![Rust](https://img.shields.io/badge/rust-1.94.1%2B-orange)
**donglora-bridge** connects LoRa radios to the internet so packets heard by one
radio can be retransmitted by another radio anywhere in the world.
[LoRa](https://en.wikipedia.org/wiki/LoRa) is a long-range, low-power wireless
protocol used by off-grid mesh networks like
[MeshCore](https://github.com/rocketgithub/meshcore-firmware). A single LoRa
radio might cover a few kilometers. This bridge removes that limit: any packet
your radio hears gets relayed over the internet to every other bridge running the
same passphrase, and their radios retransmit it locally. The result is a single
mesh network across a larger geographic area.
There is no central server. Bridges discover each other automatically using the
[Mainline DHT](https://en.wikipedia.org/wiki/Mainline_DHT) and communicate
through an encrypted peer-to-peer gossip swarm.
![TUI Screenshot](screenshot.png)
## What you need
### 1. A DongLoRa radio
[DongLoRa](https://github.com/donglora/firmware) 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/donglora/firmware) 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/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 = []
+11
View File
@@ -0,0 +1,11 @@
# Moved
The DHT partition-healing algorithm specification is now maintained as part
of the `iroh-gossip-rendezvous` crate:
**<https://github.com/swaits/iroh-gossip-rendezvous/blob/main/PROTOCOL.md>**
The bridge pins a specific `iroh-gossip-rendezvous` version; the spec that
matches whatever version you're running is in
`~/.cargo/registry/.../iroh-gossip-rendezvous-*/PROTOCOL.md` or, for path
dependencies, at `../iroh-gossip-rendezvous/PROTOCOL.md`.
+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);
});
+69
View File
@@ -0,0 +1,69 @@
set shell := ["bash", "-c"]
default: build
[private]
_ensure_tools:
@mise trust --yes . 2>/dev/null; mise install --quiet
# Run the bridge (TUI mode)
run *args: _ensure_tools
cargo run --release -- {{args}}
# Run in headless mode
run-headless *args: _ensure_tools
cargo run --release -- --log-only {{args}}
# Run the config wizard
config: _ensure_tools
cargo run --release -- config
# Run all checks (fmt, clippy, test, audit)
check: fmt-check clippy test audit
# Format code
fmt: _ensure_tools
cargo fmt
# Check formatting without changing files
fmt-check: _ensure_tools
cargo fmt --check
# Lint with clippy
clippy: _ensure_tools
cargo clippy --all-targets -- -D warnings
# Run all tests
test: test-unit
# Normal unit tests
test-unit: _ensure_tools
cargo nextest run
# Mutation testing
test-mutants *args: _ensure_tools
cargo mutants {{args}}
# Fuzz a target (requires nightly)
test-fuzz TARGET="fuzz_gossip_frame_decode" DURATION="60": _ensure_tools
cargo +nightly fuzz run {{TARGET}} -- -max_total_time={{DURATION}}
# Security + license audit
audit: _ensure_tools
cargo deny check
# Build release binary
build: _ensure_tools
cargo build --release
# Generate docs
doc: _ensure_tools
cargo doc --no-deps --open
# Check for outdated dependencies
outdated: _ensure_tools
cargo outdated
# Clean build artifacts
clean:
cargo clean
+3
View File
@@ -0,0 +1,3 @@
[tools]
rust = "1.94"
just = "latest"
+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

+277 -49
View File
@@ -1,7 +1,13 @@
//! TOML configuration loading and validation.
//!
//! Handles the `[radio]`, `[bridge]`, and `[tx]` sections with `MeshCore` US/Canada
//! recommended defaults, and conversion to the donglora-client `LoRaConfig`.
use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::{Context, Result, bail};
use donglora_client::{Bandwidth, RadioConfig, TX_POWER_MAX};
use donglora_client::{LoRaBandwidth, LoRaCodingRate, LoRaConfig, LoRaHeaderMode, RetryPolicy};
use serde::{Deserialize, Serialize};
/// Top-level bridge configuration.
@@ -11,6 +17,8 @@ pub struct Config {
pub radio: RadioSection,
#[serde(default)]
pub bridge: BridgeSection,
#[serde(default)]
pub tx: TxSection,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
@@ -28,11 +36,9 @@ pub struct RadioSection {
#[serde(default = "default_sync_word")]
pub sync_word: u16,
#[serde(default = "default_tx_power")]
pub tx_power: String,
pub tx_power_dbm: i8,
#[serde(default = "default_preamble")]
pub preamble: u16,
#[serde(default = "default_true")]
pub cad: bool,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
@@ -50,38 +56,74 @@ pub struct BridgeSection {
pub log_file: Option<PathBuf>,
}
/// Per-TX retry policy. Applied to every `tx_with_retry` call. Defaults quote
/// spec `PROTOCOL.md §C.5.5`: 3 attempts, randomized 20–100 ms backoff on the
/// first retry, doubling up to a 500 ms cap.
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct TxSection {
#[serde(default = "default_max_attempts")]
pub max_attempts: u8,
#[serde(default = "default_backoff_min_ms")]
pub backoff_min_ms: u32,
#[serde(default = "default_backoff_max_ms")]
pub backoff_max_ms: u32,
#[serde(default = "default_backoff_cap_ms")]
pub backoff_cap_ms: u32,
#[serde(default = "default_backoff_multiplier")]
pub backoff_multiplier: f32,
#[serde(default = "default_per_attempt_timeout_secs")]
pub per_attempt_timeout_secs: u64,
}
// ── Defaults (MeshCore US/Canada Recommended) ────────────────────
fn default_frequency() -> u32 {
const fn default_frequency() -> u32 {
910_525_000
}
fn default_bandwidth() -> String {
"62.5kHz".into()
}
fn default_sf() -> u8 {
const fn default_sf() -> u8 {
7
}
fn default_cr() -> u8 {
const fn default_cr() -> u8 {
5
}
fn default_sync_word() -> u16 {
0x3444
/// `MeshCore` private sync word encoded for the `SX126x` (see client-py
/// `examples/_common.py` — `RADIOLIB_SX126X_SYNC_WORD_PRIVATE`).
const fn default_sync_word() -> u16 {
0x1424
}
fn default_tx_power() -> String {
"22".into()
const fn default_tx_power() -> i8 {
20
}
fn default_preamble() -> u16 {
const fn default_preamble() -> u16 {
16
}
fn default_true() -> bool {
true
}
fn default_dedup_window() -> u64 {
const fn default_dedup_window() -> u64 {
300
}
fn default_tx_queue_size() -> usize {
const fn default_tx_queue_size() -> usize {
32
}
const fn default_max_attempts() -> u8 {
3
}
const fn default_backoff_min_ms() -> u32 {
20
}
const fn default_backoff_max_ms() -> u32 {
100
}
const fn default_backoff_cap_ms() -> u32 {
500
}
const fn default_backoff_multiplier() -> f32 {
2.0
}
const fn default_per_attempt_timeout_secs() -> u64 {
5
}
impl Default for RadioSection {
fn default() -> Self {
@@ -92,9 +134,8 @@ impl Default for RadioSection {
spreading_factor: default_sf(),
coding_rate: default_cr(),
sync_word: default_sync_word(),
tx_power: default_tx_power(),
tx_power_dbm: default_tx_power(),
preamble: default_preamble(),
cad: true,
}
}
}
@@ -111,44 +152,95 @@ impl Default for BridgeSection {
}
}
impl Default for TxSection {
fn default() -> Self {
Self {
max_attempts: default_max_attempts(),
backoff_min_ms: default_backoff_min_ms(),
backoff_max_ms: default_backoff_max_ms(),
backoff_cap_ms: default_backoff_cap_ms(),
backoff_multiplier: default_backoff_multiplier(),
per_attempt_timeout_secs: default_per_attempt_timeout_secs(),
}
}
}
// ── Conversion ────────────────────────────────────────────────────
fn parse_bandwidth(s: &str) -> Result<Bandwidth> {
fn parse_bandwidth(s: &str) -> Result<LoRaBandwidth> {
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),
"7.8khz" | "7khz" | "khz7" => Ok(LoRaBandwidth::Khz7),
"10.4khz" | "10khz" | "khz10" => Ok(LoRaBandwidth::Khz10),
"15.6khz" | "15khz" | "khz15" => Ok(LoRaBandwidth::Khz15),
"20.8khz" | "20khz" | "khz20" => Ok(LoRaBandwidth::Khz20),
"31.25khz" | "31khz" | "khz31" => Ok(LoRaBandwidth::Khz31),
"41.7khz" | "41khz" | "khz41" => Ok(LoRaBandwidth::Khz41),
"62.5khz" | "62khz" | "khz62" => Ok(LoRaBandwidth::Khz62),
"125khz" | "khz125" => Ok(LoRaBandwidth::Khz125),
"250khz" | "khz250" => Ok(LoRaBandwidth::Khz250),
"500khz" | "khz500" => Ok(LoRaBandwidth::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)")
fn parse_coding_rate(v: u8) -> Result<LoRaCodingRate> {
match v {
5 => Ok(LoRaCodingRate::Cr4_5),
6 => Ok(LoRaCodingRate::Cr4_6),
7 => Ok(LoRaCodingRate::Cr4_7),
8 => Ok(LoRaCodingRate::Cr4_8),
_ => bail!("coding_rate must be 5-8 (meaning 4/5 to 4/8), got {v}"),
}
}
impl RadioSection {
/// Convert to the donglora-client RadioConfig.
pub fn to_radio_config(&self) -> Result<RadioConfig> {
Ok(RadioConfig {
/// Convert to the donglora-client `LoRaConfig`.
///
/// Validates that all radio parameters are within hardware-supported ranges
/// before constructing the config.
pub fn to_lora_config(&self) -> Result<LoRaConfig> {
if !(7..=12).contains(&self.spreading_factor) {
bail!("spreading_factor must be 7-12, got {}", self.spreading_factor);
}
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(LoRaConfig {
freq_hz: self.frequency,
bw: parse_bandwidth(&self.bandwidth)?,
sf: self.spreading_factor,
cr: self.coding_rate,
cr: parse_coding_rate(self.coding_rate)?,
sync_word: self.sync_word,
tx_power_dbm: parse_tx_power(&self.tx_power)?,
tx_power_dbm: self.tx_power_dbm,
preamble_len: self.preamble,
cad: u8::from(self.cad),
header_mode: LoRaHeaderMode::Explicit,
payload_crc: true,
iq_invert: false,
})
}
}
impl TxSection {
/// Materialize the TOML fields into a [`RetryPolicy`].
pub fn to_retry_policy(&self) -> Result<RetryPolicy> {
if self.max_attempts == 0 {
bail!("tx.max_attempts must be >= 1");
}
if self.backoff_min_ms > self.backoff_max_ms {
bail!(
"tx.backoff_min_ms ({}) must not exceed backoff_max_ms ({})",
self.backoff_min_ms,
self.backoff_max_ms
);
}
Ok(RetryPolicy {
max_attempts: self.max_attempts,
backoff_ms_min: self.backoff_min_ms,
backoff_ms_max: self.backoff_max_ms,
backoff_multiplier: self.backoff_multiplier,
backoff_cap_ms: self.backoff_cap_ms,
per_attempt_timeout: Duration::from_secs(self.per_attempt_timeout_secs),
skip_cad: false,
})
}
}
@@ -157,9 +249,7 @@ impl RadioSection {
/// Default config directory: `$XDG_CONFIG_HOME/donglora-bridge/` or `~/.config/donglora-bridge/`.
pub fn config_dir() -> Result<PathBuf> {
let dir = dirs::config_dir()
.context("cannot determine config directory")?
.join("donglora-bridge");
let dir = dirs::config_dir().context("cannot determine config directory")?.join("donglora-bridge");
Ok(dir)
}
@@ -179,13 +269,151 @@ pub fn default_log_path() -> Result<PathBuf> {
/// Load config from a TOML file. Returns None-passphrase error if not set.
pub fn load_config(path: &Path) -> Result<Config> {
let text = std::fs::read_to_string(path)
.with_context(|| format!("reading config: {}", path.display()))?;
let text = std::fs::read_to_string(path).with_context(|| format!("reading config: {}", path.display()))?;
let cfg: Config = toml::from_str(&text).context("parsing config TOML")?;
if cfg.bridge.passphrase.is_empty() || cfg.bridge.passphrase == "change-me" {
bail!(
"passphrase is required — run `donglora-bridge config` to set it"
);
bail!("passphrase is required — run `donglora-bridge config` to set it");
}
Ok(cfg)
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
#[test]
fn parse_bandwidth_canonical() {
assert!(matches!(parse_bandwidth("7.8kHz"), Ok(LoRaBandwidth::Khz7)));
assert!(matches!(parse_bandwidth("62.5kHz"), Ok(LoRaBandwidth::Khz62)));
assert!(matches!(parse_bandwidth("125kHz"), Ok(LoRaBandwidth::Khz125)));
assert!(matches!(parse_bandwidth("500kHz"), Ok(LoRaBandwidth::Khz500)));
}
#[test]
fn parse_bandwidth_aliases() {
assert!(matches!(parse_bandwidth("7khz"), Ok(LoRaBandwidth::Khz7)));
assert!(matches!(parse_bandwidth("khz125"), Ok(LoRaBandwidth::Khz125)));
}
#[test]
fn parse_bandwidth_case_insensitive() {
assert!(matches!(parse_bandwidth("125KHZ"), Ok(LoRaBandwidth::Khz125)));
}
#[test]
fn parse_bandwidth_rejects_unknown() {
assert!(parse_bandwidth("999kHz").is_err());
assert!(parse_bandwidth("").is_err());
}
#[test]
fn parse_coding_rate_canonical() {
assert!(matches!(parse_coding_rate(5), Ok(LoRaCodingRate::Cr4_5)));
assert!(matches!(parse_coding_rate(8), Ok(LoRaCodingRate::Cr4_8)));
}
#[test]
fn parse_coding_rate_rejects_out_of_range() {
assert!(parse_coding_rate(4).is_err());
assert!(parse_coding_rate(9).is_err());
}
#[test]
fn radio_section_defaults_to_lora_config() {
let section = RadioSection::default();
let cfg = section.to_lora_config().unwrap();
assert_eq!(cfg.freq_hz, 910_525_000);
assert_eq!(cfg.sf, 7);
assert!(matches!(cfg.cr, LoRaCodingRate::Cr4_5));
assert_eq!(cfg.sync_word, 0x1424);
assert_eq!(cfg.preamble_len, 16);
assert_eq!(cfg.tx_power_dbm, 20);
}
#[test]
fn to_lora_config_rejects_invalid_sf() {
let section = RadioSection { spreading_factor: 6, ..RadioSection::default() };
assert!(section.to_lora_config().is_err());
let section = RadioSection { spreading_factor: 13, ..RadioSection::default() };
assert!(section.to_lora_config().is_err());
}
#[test]
fn to_lora_config_rejects_invalid_cr() {
let section = RadioSection { coding_rate: 4, ..RadioSection::default() };
assert!(section.to_lora_config().is_err());
let section = RadioSection { coding_rate: 9, ..RadioSection::default() };
assert!(section.to_lora_config().is_err());
}
#[test]
fn to_lora_config_rejects_invalid_frequency() {
let section = RadioSection { frequency: 0, ..RadioSection::default() };
assert!(section.to_lora_config().is_err());
let section = RadioSection { frequency: 2_000_000_000, ..RadioSection::default() };
assert!(section.to_lora_config().is_err());
}
#[test]
fn tx_section_defaults_produce_valid_policy() {
let p = TxSection::default().to_retry_policy().unwrap();
assert_eq!(p.max_attempts, 3);
assert_eq!(p.backoff_ms_min, 20);
assert_eq!(p.backoff_ms_max, 100);
assert_eq!(p.backoff_cap_ms, 500);
assert!((p.backoff_multiplier - 2.0).abs() < f32::EPSILON);
}
#[test]
fn tx_section_rejects_zero_attempts() {
let s = TxSection { max_attempts: 0, ..TxSection::default() };
assert!(s.to_retry_policy().is_err());
}
#[test]
fn tx_section_rejects_inverted_backoff_bounds() {
let s = TxSection { backoff_min_ms: 500, backoff_max_ms: 100, ..TxSection::default() };
assert!(s.to_retry_policy().is_err());
}
#[test]
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);
assert_eq!(cfg.tx.max_attempts, 3);
}
#[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());
}
}
+78 -557
View File
@@ -1,103 +1,28 @@
//! Bridge-side glue over the `iroh-gossip-rendezvous` crate.
//!
//! Starts the rendezvous, spawns a task that decodes incoming gossip events
//! into [`GossipEvent`]s the router understands, and a broadcaster task that
//! encodes outgoing [`GossipFrame`]s onto the iroh-gossip sender.
use std::collections::HashSet;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use std::sync::Arc;
use anyhow::{Context, Result};
use bytes::Bytes;
use iroh::endpoint::Incoming;
use iroh::{Endpoint, PublicKey, SecretKey};
use futures_lite::StreamExt;
use iroh::{PublicKey, SecretKey};
use iroh_gossip::api::Event;
use iroh_gossip::net::{Gossip as GossipActor, GOSSIP_ALPN};
use iroh_gossip::TopicId;
use mainline::async_dht::AsyncDht;
use mainline::{Dht, MutableItem, SigningKey};
use sha2::{Digest, Sha256};
use iroh_gossip_rendezvous::Rendezvous;
pub use iroh_gossip_rendezvous::{DhtStatus, RendezvousState};
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::{debug, info, warn};
use tracing::{debug, error, 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 (base, before jitter).
const DHT_HEARTBEAT_SECS: u64 = 300;
/// Maximum random jitter added to heartbeat interval.
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.
pub fn derive(passphrase: &str) -> Self {
// Application-specific prefix to avoid DHT collisions with other apps.
const APP_PREFIX: &str = "donglora-bridge:";
// Topic ID: blake3 hash of prefixed passphrase.
let topic_bytes = *blake3::hash(format!("{APP_PREFIX}{passphrase}").as_bytes()).as_bytes();
let topic_id = TopicId::from_bytes(topic_bytes);
// DHT signing key: ed25519 from SHA-256 of prefixed passphrase + "sign".
let sign_seed = Sha256::digest(format!("{APP_PREFIX}sign:{passphrase}").as_bytes());
let seed_bytes: [u8; 32] = sign_seed.into();
let dht_signing_key = SigningKey::from_bytes(&seed_bytes);
// DHT salt: first 20 bytes of SHA-256 of prefixed passphrase + "salt".
let salt_hash = Sha256::digest(format!("{APP_PREFIX}salt:{passphrase}").as_bytes());
let mut dht_salt = [0u8; 20];
dht_salt.copy_from_slice(&salt_hash[..20]);
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>,
}
#[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 ────────────────────────────────────────────
/// App salt for the donglora-bridge rendezvous surface. Versioned so a
/// future breaking change gets a fresh DHT meeting place — v1 and v2
/// bridges do not see each other.
pub const APP_SALT: &str = "donglora-bridge/v2";
/// Events from the gossip swarm to the router.
pub enum GossipEvent {
@@ -107,207 +32,90 @@ pub enum GossipEvent {
NeighborChanged(usize),
}
// ── Gossip handle ───────────────────────────────────────────────
/// Handle to the gossip swarm. Manages iroh endpoint, gossip subscription, and DHT.
/// Handle held by the rest of the bridge. Owns the `Rendezvous` that in turn
/// owns all DHT + gossip background work. Dropping this stops everything.
pub struct Gossip {
endpoint: Endpoint,
gossip: GossipActor,
state: Arc<Mutex<SwarmState>>,
neighbors: Arc<Mutex<HashSet<PublicKey>>>,
dht_signing_key: SigningKey,
dht_salt: [u8; 20],
rendezvous: Rendezvous,
cancel: CancellationToken,
}
impl Gossip {
/// Create and start the gossip network.
///
/// Spawns background tasks for connection acceptance, gossip event forwarding,
/// gossip broadcasting, and DHT heartbeat.
/// Spin up the gossip swarm + DHT rendezvous.
///
/// 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,
passphrase: &str,
) -> 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)
let rendezvous = Rendezvous::builder()
.passphrase(passphrase)
.app_salt(APP_SALT)
.secret_key(secret_key)
.alpns(vec![GOSSIP_ALPN.to_vec()])
.bind()
.build()
.await
.context("binding iroh endpoint")?;
.context("starting iroh-gossip rendezvous")?;
let my_id = endpoint.id();
info!("iroh endpoint bound, id: {}", my_id.fmt_short());
let my_id = rendezvous.node_id();
tracing::info!("iroh endpoint bound, id: {}", my_id.fmt_short());
tracing::info!("topic: {}", hex::encode(&rendezvous.topic_id().as_bytes()[..8]));
// Build gossip actor.
let gossip = 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 = gossip
.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(),
gossip: gossip.clone(),
state: state.clone(),
neighbors: neighbors.clone(),
dht_signing_key: keys.dht_signing_key.clone(),
dht_salt: keys.dht_salt,
cancel: cancel.clone(),
};
// Receiver task: drain the rendezvous event stream, translate into
// router-side `GossipEvent`s.
spawn_receiver(rendezvous.subscribe(), event_tx, cancel.clone());
// Spawn accept loop.
spawn_accept_loop(endpoint.clone(), gossip.clone(), cancel.clone());
// Broadcaster task: pull encoded frames from the router, hand them
// to the iroh-gossip sender.
spawn_broadcaster(rendezvous.sender().clone(), frame_rx, 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))
Ok((Self { rendezvous, cancel }, event_rx, frame_tx))
}
/// Snapshot of swarm state for TUI.
pub fn swarm_state(&self) -> SwarmState {
self.state.lock().unwrap_or_else(|e| e.into_inner()).clone()
/// Snapshot of rendezvous state for the TUI.
#[must_use]
pub fn state(&self) -> RendezvousState {
self.rendezvous.state()
}
/// Graceful shutdown. Publishes neighbors (without ourselves) to DHT
/// so other nodes can still bootstrap from live peers.
/// Graceful shutdown. Cancels our own glue tasks, then the rendezvous.
/// Idempotent.
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(|e| e.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() as i64)
.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.gossip.shutdown().await;
self.endpoint.close().await;
self.rendezvous.shutdown().await;
}
}
// ── Background tasks ────────────────────────────────────────────
// ── Internal 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,
mut rx: tokio::sync::broadcast::Receiver<Event>,
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 {
// Track our own neighbor set so we can send accurate counts to the
// router (the rendezvous crate tracks the same set internally for
// its `state()`; we duplicate here to avoid a cross-crate handle).
let mut neighbors: HashSet<PublicKey> = HashSet::new();
let exit_reason: &'static str = loop {
let event = tokio::select! {
event = receiver.next() => event,
() = cancel.cancelled() => break,
ev = rx.recv() => ev,
() = cancel.cancelled() => break "cancelled",
};
let event = match event {
Ok(ev) => ev,
Err(tokio::sync::broadcast::error::RecvError::Closed) => break "broadcast closed",
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
warn!("gossip receiver lagged — dropped {n} events");
continue;
}
};
let Some(Ok(event)) = event else { break };
match event {
Event::Received(msg) => {
if let Some(frame) = GossipFrame::decode(&msg.content) {
@@ -317,29 +125,30 @@ fn spawn_receiver(
}
}
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;
neighbors.insert(id);
tracing::info!("neighbor up: {} (total: {})", id.fmt_short(), neighbors.len());
let _ = event_tx.send(GossipEvent::NeighborChanged(neighbors.len())).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;
neighbors.remove(&id);
tracing::info!("neighbor down: {} (total: {})", id.fmt_short(), neighbors.len());
let _ = event_tx.send(GossipEvent::NeighborChanged(neighbors.len())).await;
}
Event::Lagged => {
warn!("gossip receiver lagged — some messages were dropped");
}
}
};
if exit_reason == "cancelled" {
debug!("gossip receiver stopped ({exit_reason})");
} else {
error!(
"gossip receiver STOPPED ({exit_reason}) — this bridge will no longer receive from the swarm",
);
}
debug!("gossip receiver stopped");
});
}
/// 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>,
@@ -361,299 +170,11 @@ fn spawn_broadcaster(
});
}
/// DHT poll interval while alone (read frequently, publish infrequently).
const DHT_ALONE_READ_SECS: u64 = 3;
/// How often a lonely node publishes (merge-only, never destructive).
/// Slower than reads — publishing is heavier and a lonely node only adds itself.
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().map(|n| !n.is_empty()).unwrap_or(false);
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() as i64)
.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();
match dht.get_mutable_most_recent(&pub_key_bytes, Some(&keys.dht_salt)).await {
Some(record) => {
let peers = decode_peer_list(record.value());
debug!("DHT bootstrap: found {} peer(s)", peers.len());
peers
}
None => {
debug!("DHT bootstrap: no existing record");
vec![]
}
}
}
/// Encode a list of 32-byte peer IDs into bytes for DHT storage.
/// Format: [count: u16 LE] [id_0: 32B] [id_1: 32B] ...
fn encode_peer_list(peers: &[[u8; 32]]) -> Vec<u8> {
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.
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);
// Shared via Arc<Gossip> for the TUI.
impl Gossip {
/// Wrap in `Arc` for sharing with the TUI task.
#[must_use]
pub fn into_arc(self) -> Arc<Self> {
Arc::new(self)
}
}
+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;
+148 -77
View File
@@ -1,11 +1,4 @@
mod config;
mod gossip;
mod packet;
mod radio;
mod rate_limit;
mod router;
mod setup;
mod tui;
use donglora_bridge::{config, gossip, radio, rate_limit, router, setup, tui};
use std::sync::Arc;
use std::time::Instant;
@@ -17,7 +10,6 @@ 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;
@@ -43,6 +35,7 @@ enum Commands {
}
#[tokio::main]
#[allow(clippy::too_many_lines)]
async fn main() -> Result<()> {
let cli = Cli::parse();
@@ -53,12 +46,8 @@ async fn main() -> Result<()> {
};
// Config subcommand: run wizard with existing config as defaults.
if let Some(Commands::Config) = cli.command {
let existing = if config_path.exists() {
config::load_config(&config_path).ok()
} else {
None
};
if matches!(cli.command, Some(Commands::Config)) {
let existing = if config_path.exists() { config::load_config(&config_path).ok() } else { None };
return setup::run_wizard(&config_path, existing.as_ref());
}
@@ -69,82 +58,55 @@ async fn main() -> Result<()> {
}
// Load config.
let cfg = config::load_config(&config_path)
.with_context(|| format!("loading config from {}", config_path.display()))?;
let cfg =
config::load_config(&config_path).with_context(|| format!("loading config from {}", config_path.display()))?;
// Set up logging.
if cli.log_only {
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
)
.init();
} else {
let log_path = cfg
.bridge
.log_file
.clone()
.unwrap_or_else(|| config::default_log_path().unwrap_or_else(|_| "/tmp/donglora-bridge.log".into()));
if let Some(parent) = log_path.parent() {
std::fs::create_dir_all(parent).ok();
}
let file_appender = tracing_appender::rolling::never(
log_path.parent().unwrap_or(std::path::Path::new("/tmp")),
log_path.file_name().unwrap_or(std::ffi::OsStr::new("donglora-bridge.log")),
);
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
)
.with_writer(file_appender)
.with_ansi(false)
.init();
}
setup_logging(cli.log_only, &cfg);
// Ephemeral identity — fresh keypair each launch.
let secret_key = SecretKey::generate(&mut rand::rng());
let secret_key = SecretKey::generate();
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]));
let radio_config = cfg.radio.to_lora_config()?;
let retry_policy = cfg.tx.to_retry_policy()?;
// Create rate limiter.
let cr_denom: u8 = match radio_config.cr {
donglora_client::LoRaCodingRate::Cr4_5 => 5,
donglora_client::LoRaCodingRate::Cr4_6 => 6,
donglora_client::LoRaCodingRate::Cr4_7 => 7,
donglora_client::LoRaCodingRate::Cr4_8 => 8,
};
let rate_limiter = RateLimiter::from_radio_config(
radio_config.sf,
radio_config.bw,
radio_config.cr,
cr_denom,
radio_config.preamble_len,
cfg.bridge.rate_limit_pps,
);
info!("rate limiter: {:.1} pps", rate_limiter.rate_pps());
// Create gossip network.
// Create gossip network + DHT rendezvous.
let (swarm, gossip_event_rx, gossip_frame_tx) =
gossip::Gossip::new(secret_key, &keys).await?;
gossip::Gossip::new(secret_key, &cfg.bridge.passphrase).await?;
let swarm = Arc::new(swarm);
// Start radio.
let (radio_event_rx, radio_tx) = radio::spawn(radio_config, cfg.radio.port.clone());
let (radio_event_rx, radio_tx, _radio_handle) = radio::spawn(radio_config, cfg.radio.port.clone(), retry_policy)?;
// 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,
},
);
let (config_watch_tx, config_watch_rx) = tokio::sync::watch::channel(router::RadioConfigInfo {
active: radio_config,
requested: radio_config,
source: radio::ConfigSource::Ours,
device: String::new(),
connected: false,
});
// Cancellation.
let cancel = CancellationToken::new();
@@ -180,9 +142,7 @@ async fn main() -> Result<()> {
{
let cancel = cancel.clone();
tokio::spawn(async move {
let sigterm = tokio::signal::unix::signal(
tokio::signal::unix::SignalKind::terminate(),
);
let sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate());
match sigterm {
Ok(mut sigterm) => {
tokio::select! {
@@ -203,17 +163,16 @@ async fn main() -> Result<()> {
// Run TUI or headless.
if cli.log_only {
info!("running in headless mode");
// In TUI mode the TUI drains `log_rx`; in headless mode nobody
// does unless we spawn a consumer. Without this, the router's
// bounded log channel backs up and router throughput stalls.
// Emit each entry at INFO level so operators can see what the
// bridge is bridging (RX/TX/drop/retry).
let log_cancel = cancel.clone();
tokio::spawn(headless_packet_logger(log_rx, log_cancel));
cancel.cancelled().await;
} else {
let terminal = tui::run(
config_watch_rx,
swarm.clone(),
stats,
log_rx,
cancel.clone(),
start_time,
)
.await;
let terminal = tui::run(config_watch_rx, swarm.clone(), stats, log_rx, cancel.clone(), start_time);
cancel.cancel();
swarm.shutdown().await;
@@ -232,3 +191,115 @@ async fn main() -> Result<()> {
Ok(())
}
/// Drain the router's packet-log channel in headless mode and emit
/// tracing events so `--log-only` output shows every bridged / dropped /
/// retried packet. Exits when `cancel` fires or the channel closes.
async fn headless_packet_logger(mut log_rx: mpsc::Receiver<router::PacketLogEntry>, cancel: CancellationToken) {
use router::{PacketAction, PacketDirection, TxRetryState};
loop {
tokio::select! {
() = cancel.cancelled() => return,
entry = log_rx.recv() => {
let Some(e) = entry else { return; };
let dir = match e.direction {
PacketDirection::RadioIn => "RF→NET",
PacketDirection::GossipIn => "NET→RF",
};
let hash_str = hex::encode(&e.hash[..4]);
let rssi_str = e.rssi.map_or_else(|| "-".to_string(), |r| r.to_string());
let snr_str = e.snr.map_or_else(|| "-".to_string(), |s| s.to_string());
// TX-retry lifecycle wins over the plain action label when set.
if let Some(r) = e.tx_retry {
match r.state {
TxRetryState::Retrying(reason) => {
tracing::info!(
target: "packet",
"{} retry hash={} attempt={}/{} reason={}",
dir, hash_str, r.attempt, r.total_attempts, reason,
);
}
TxRetryState::Succeeded if r.attempt > 1 => {
tracing::info!(
target: "packet",
"{} sent hash={} size={}B attempts={}/{}",
dir, hash_str, e.size, r.attempt, r.total_attempts,
);
}
TxRetryState::Failed => {
tracing::warn!(
target: "packet",
"{} FAILED hash={} size={}B attempts={}/{}",
dir, hash_str, e.size, r.attempt, r.total_attempts,
);
}
_ => {}
}
continue;
}
match e.action {
PacketAction::Bridged => {
tracing::info!(
target: "packet",
"{} bridged hash={} size={}B rssi={} snr={}",
dir, hash_str, e.size, rssi_str, snr_str,
);
}
PacketAction::DroppedDedup => {
// Log at INFO, not DEBUG — a run of dedup lines is
// proof of bidirectional gossip flow. Hiding it
// masks whether we're actually hearing peers.
tracing::info!(
target: "packet",
"{} dedup hash={} size={}B rssi={} snr={}",
dir, hash_str, e.size, rssi_str, snr_str,
);
}
PacketAction::DroppedRateLimit => {
tracing::warn!(
target: "packet",
"{} ratelim hash={} size={}B",
dir, hash_str, e.size,
);
}
PacketAction::DroppedQueueFull => {
tracing::warn!(
target: "packet",
"{} QFULL hash={} size={}B",
dir, hash_str, e.size,
);
}
}
}
}
}
}
fn setup_logging(log_only: bool, cfg: &config::Config) {
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();
}
}
+101 -18
View File
@@ -1,7 +1,12 @@
/// Gossip frame header size: sender(32) + rssi(2) + snr(2) + payload_len(2) = 38
//! Wire format definitions for radio packets and gossip frames.
//!
//! Includes SNR quality grading against `LoRa` demodulation floors and blake3
//! content hashing for deduplication.
/// Gossip frame header size: `sender(32) + rssi(2) + snr(2) + payload_len(2) = 38`
const FRAME_HEADER_SIZE: usize = 38;
/// SNR quality grade for a received LoRa packet.
/// SNR quality grade for a received `LoRa` packet.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Grade {
/// SNR well above demodulation floor (margin >= 3 dB).
@@ -16,7 +21,8 @@ pub enum Grade {
impl Grade {
/// Returns true if the packet is reliable enough to forward.
pub fn should_forward(self) -> bool {
#[must_use]
pub const fn should_forward(self) -> bool {
matches!(self, Self::Good | Self::Marginal)
}
}
@@ -34,6 +40,7 @@ impl std::fmt::Display for Grade {
/// Grade a received packet's SNR against the demodulation floor for the given
/// spreading factor.
#[must_use]
pub fn snr_grade(snr: i16, sf: u8) -> Grade {
if !(-32..=32).contains(&snr) {
return Grade::Invalid;
@@ -62,33 +69,31 @@ pub struct RadioPacket {
/// Wire format: `[sender: 32B] [rssi: i16 LE] [snr: i16 LE] [payload_len: u16 LE] [payload: NB]`
#[derive(Debug, Clone)]
pub struct GossipFrame {
/// EndpointId (PublicKey bytes) of the bridge that first heard this packet on radio.
/// `EndpointId` (`PublicKey` bytes) of the bridge that first heard this packet on radio.
pub sender: [u8; 32],
/// RSSI at the originating bridge (informational).
pub rssi: i16,
/// SNR at the originating bridge (informational).
pub snr: i16,
/// Raw LoRa payload.
/// Raw `LoRa` payload.
pub payload: Vec<u8>,
}
impl GossipFrame {
/// Create a new gossip frame from a public key and radio packet data.
#[must_use]
pub fn new(sender: &iroh::PublicKey, rssi: i16, snr: i16, payload: Vec<u8>) -> Self {
Self {
sender: *sender.as_bytes(),
rssi,
snr,
payload,
}
Self { sender: *sender.as_bytes(), rssi, snr, payload }
}
/// Encode this frame into bytes for gossip broadcast.
#[must_use]
pub fn encode(&self) -> Vec<u8> {
let mut buf = Vec::with_capacity(FRAME_HEADER_SIZE + self.payload.len());
buf.extend_from_slice(&self.sender);
buf.extend_from_slice(&self.rssi.to_le_bytes());
buf.extend_from_slice(&self.snr.to_le_bytes());
#[allow(clippy::cast_possible_truncation)] // LoRa payloads never exceed 256 bytes
let len = self.payload.len() as u16;
buf.extend_from_slice(&len.to_le_bytes());
buf.extend_from_slice(&self.payload);
@@ -96,6 +101,7 @@ impl GossipFrame {
}
/// Decode a gossip frame from bytes received from the swarm.
#[must_use]
pub fn decode(data: &[u8]) -> Option<Self> {
if data.len() < FRAME_HEADER_SIZE {
return None;
@@ -107,21 +113,18 @@ impl GossipFrame {
if data.len() != FRAME_HEADER_SIZE + payload_len {
return None;
}
Some(Self {
sender,
rssi,
snr,
payload: data[FRAME_HEADER_SIZE..].to_vec(),
})
Some(Self { sender, rssi, snr, payload: data[FRAME_HEADER_SIZE..].to_vec() })
}
}
/// Compute a 32-byte blake3 content hash for deduplication.
#[must_use]
pub fn content_hash(payload: &[u8]) -> [u8; 32] {
*blake3::hash(payload).as_bytes()
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
@@ -141,7 +144,7 @@ mod tests {
#[test]
fn gossip_frame_roundtrip() {
let key = iroh::SecretKey::generate(&mut rand::rng());
let key = iroh::SecretKey::generate();
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");
@@ -156,6 +159,17 @@ mod tests {
assert!(GossipFrame::decode(&[0u8; 10]).is_none());
}
#[test]
fn gossip_frame_decode_empty_payload() {
// Exactly FRAME_HEADER_SIZE (38) bytes with payload_len=0 must decode.
// Kills mutant: `data.len() < 38` → `data.len() <= 38`.
let mut data = vec![0u8; 38];
data[36..38].copy_from_slice(&0u16.to_le_bytes()); // payload_len = 0
let frame = GossipFrame::decode(&data);
assert!(frame.is_some(), "38-byte frame with 0-length payload must decode");
assert!(frame.unwrap().payload.is_empty());
}
#[test]
fn gossip_frame_reject_wrong_length() {
// Header says payload_len=5 but only 3 bytes follow.
@@ -164,6 +178,43 @@ mod tests {
assert!(GossipFrame::decode(&data).is_none());
}
#[test]
fn should_forward_true_for_good_and_marginal() {
assert!(Grade::Good.should_forward());
assert!(Grade::Marginal.should_forward());
}
#[test]
fn should_forward_false_for_unreliable_and_invalid() {
assert!(!Grade::Unreliable.should_forward());
assert!(!Grade::Invalid.should_forward());
}
#[test]
fn grade_display() {
assert_eq!(format!("{}", Grade::Good), "GOOD");
assert_eq!(format!("{}", Grade::Marginal), "MARG");
assert_eq!(format!("{}", Grade::Unreliable), "UNRL");
assert_eq!(format!("{}", Grade::Invalid), "INVL");
}
#[test]
fn snr_grade_exact_boundaries() {
// SF12: min_snr = -2.5 * (12 - 4) = -20.0
//
// At snr=-20, margin = -20 - (-20.0) = 0.0 exactly.
// The code uses `margin < 0.0`, so 0.0 is NOT < 0.0 → Marginal (not Unreliable).
assert_eq!(snr_grade(-20, 12), Grade::Marginal);
// One below: snr=-21, margin = -1.0 → Unreliable.
assert_eq!(snr_grade(-21, 12), Grade::Unreliable);
// At snr=-17, margin = -17 - (-20.0) = 3.0 exactly.
// The code uses `margin < 3.0`, so 3.0 is NOT < 3.0 → Good (not Marginal).
assert_eq!(snr_grade(-17, 12), Grade::Good);
// One below: snr=-18, margin = 2.0 → Marginal.
assert_eq!(snr_grade(-18, 12), Grade::Marginal);
}
#[test]
fn content_hash_deterministic() {
let h1 = content_hash(b"hello");
@@ -172,4 +223,36 @@ mod tests {
let h3 = content_hash(b"world");
assert_ne!(h1, h3);
}
mod proptests {
use super::*;
use proptest::prelude::*;
proptest! {
#[test]
fn gossip_frame_roundtrip_prop(
rssi in -200i16..=200i16,
snr in -32i16..=32i16,
payload in proptest::collection::vec(any::<u8>(), 0..256),
) {
let key = iroh::SecretKey::generate();
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);
}
}
}
}
+351 -214
View File
@@ -1,8 +1,24 @@
//! Async radio task — drives the `donglora-client` [`Dongle`] session.
//!
//! Owns a single tokio task that connects (with exponential backoff reconnect),
//! negotiates config, starts RX, and multiplexes TX requests from the router
//! against the RX stream. TX is dispatched via [`Dongle::tx_with_retry`] so
//! `TX_DONE(CHANNEL_BUSY)` / `EBUSY` errors are retried per spec §6.10 with
//! randomized backoff, and per-attempt state is surfaced to the router (and
//! onwards to the TUI packet log) via the [`RadioEvent::Tx*`] variants.
//!
//! Keepalive pings are handled inside the [`Dongle`] itself — no explicit
//! liveness loop here.
use std::time::Duration;
use anyhow::Result;
use donglora_client::{RadioConfig, Response, Transport};
use anyhow::{Context, Result};
use donglora_client::{
ConnectOptions, Dongle, LoRaBandwidth, LoRaConfig, Modulation, RetryPolicy, RxPayload, connect_mux_auto_with,
connect_with,
};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tracing::{debug, error, info, warn};
use crate::packet::{RadioPacket, snr_grade};
@@ -16,248 +32,295 @@ pub enum ConfigSource {
Mux,
}
/// Messages from the radio thread to the router.
/// Messages from the radio task to the router.
#[derive(Debug)]
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),
Connected(LoRaConfig, ConfigSource, String),
/// Radio disconnected — attempting reconnect.
Disconnected,
/// TX attempt started. `attempt` is 1-indexed.
TxAttempt { seq: u64, attempt: u8, total_attempts: u8, size: usize },
/// Previous TX attempt failed and a retry is scheduled.
TxRetry { seq: u64, attempt_that_failed: u8, total_attempts: u8, reason: TxRetryReason, backoff_ms: u32 },
/// TX completed successfully (perhaps after retries).
TxSucceeded { seq: u64, attempts_used: u8, final_airtime_us: u32, size: usize },
/// TX policy exhausted or errored terminally.
TxFailed { seq: u64, attempts_used: u8, reason: String, size: usize },
}
/// Spawn the blocking radio thread.
/// Why a TX attempt was retried — carried alongside the retry event for
/// the TUI.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TxRetryReason {
/// CAD detected channel activity before TX.
ChannelBusy,
/// Firmware TX queue was full.
QueueFull,
/// Other retryable condition (future-compat).
Other,
}
impl std::fmt::Display for TxRetryReason {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::ChannelBusy => write!(f, "channel_busy"),
Self::QueueFull => write!(f, "queue_full"),
Self::Other => write!(f, "other"),
}
}
}
/// One outbound TX request from the router: the payload plus a sequence id so
/// the event stream can correlate back to the packet log entry.
#[derive(Debug)]
pub struct TxRequest {
pub seq: u64,
pub data: Vec<u8>,
}
/// Spawn the async radio task.
///
/// Returns a receiver for radio events and a sender for TX payloads.
/// Returns a receiver for radio events and a sender for [`TxRequest`]s.
pub fn spawn(
config: RadioConfig,
config: LoRaConfig,
port: Option<String>,
) -> (mpsc::Receiver<RadioEvent>, mpsc::Sender<Vec<u8>>) {
retry_policy: RetryPolicy,
) -> Result<(mpsc::Receiver<RadioEvent>, mpsc::Sender<TxRequest>, JoinHandle<()>)> {
let (event_tx, event_rx) = mpsc::channel::<RadioEvent>(256);
let (tx_send, tx_recv) = mpsc::channel::<Vec<u8>>(64);
std::thread::Builder::new()
.name("radio".into())
.spawn(move || radio_loop(config, port, event_tx, tx_recv))
.ok();
(event_rx, tx_send)
let (tx_send, tx_recv) = mpsc::channel::<TxRequest>(64);
let handle = tokio::spawn(radio_loop(config, port, retry_policy, event_tx, tx_recv));
Ok((event_rx, tx_send, handle))
}
fn radio_loop(
config: RadioConfig,
async fn radio_loop(
config: LoRaConfig,
port: Option<String>,
retry_policy: RetryPolicy,
event_tx: mpsc::Sender<RadioEvent>,
mut tx_recv: mpsc::Receiver<Vec<u8>>,
mut tx_recv: mpsc::Receiver<TxRequest>,
) {
let initial_backoff = Duration::from_millis(250);
let max_backoff = Duration::from_secs(5);
let mut backoff = initial_backoff;
let mut stick_to_mux = false;
loop {
let started = std::time::Instant::now();
match connect_and_run(&config, &port, &event_tx, &mut tx_recv) {
Ok(()) => {
info!("radio thread exiting");
return;
match connect_and_run(&config, port.as_deref(), stick_to_mux, &retry_policy, &event_tx, &mut tx_recv).await {
Ok(was_mux) => {
if event_tx.is_closed() {
info!("radio task exiting (router closed)");
return;
}
// We had a clean run_session exit but not shutdown — loop back.
if was_mux {
stick_to_mux = true;
}
debug!("radio session exited cleanly; reconnecting");
}
Err(e) => {
Err((was_mux, e)) => {
if was_mux {
stick_to_mux = true;
}
error!("radio error: {e:#}");
let _ = event_tx.blocking_send(RadioEvent::Disconnected);
let _ = event_tx.send(RadioEvent::Disconnected).await;
// If we were connected for a while, reset backoff so
// the next reconnect attempt is fast.
if started.elapsed() > Duration::from_secs(5) {
backoff = initial_backoff;
}
info!("reconnecting in {backoff:?}");
std::thread::sleep(backoff);
tokio::time::sleep(backoff).await;
backoff = (backoff * 2).min(max_backoff);
}
}
}
}
fn connect_and_run(
config: &RadioConfig,
port: &Option<String>,
async fn connect_and_run(
config: &LoRaConfig,
port: Option<&str>,
stick_to_mux: bool,
retry_policy: &RetryPolicy,
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);
tx_recv: &mut mpsc::Receiver<TxRequest>,
) -> std::result::Result<bool, (bool, anyhow::Error)> {
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))
let mut opts = ConnectOptions::new().config(Modulation::LoRa(*config));
if let Some(p) = port {
opts = opts.port(p);
}
// Attempt mux-only if we've committed to it, otherwise the normal
// auto-discovery chain.
let dongle = if stick_to_mux {
debug!("reconnecting via mux only (previously connected via mux)");
connect_mux_auto_with(opts).await.map_err(|e| (true, anyhow::Error::from(e)))?
} else {
let c = donglora_client::connect_mux_auto(timeout)?;
(c, "mux".to_string())
};
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) {
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);
}
connect_with(opts).await.map_err(|e| (false, anyhow::Error::from(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);
// Ask the Dongle which transport it actually ended up on. That's
// authoritative — don't infer from firmware capability bits.
let kind = dongle.transport_kind().clone();
let is_mux = kind.is_mux();
let device = kind.short_label();
info!("[radio] connected: {device}");
let outcome = run_session(dongle, config, is_mux, &device, retry_policy, event_tx, tx_recv).await;
outcome.map(|()| is_mux).map_err(|e| (is_mux, e))
}
async fn run_session(
dongle: Dongle,
config: &LoRaConfig,
is_mux: bool,
device: &str,
retry_policy: &RetryPolicy,
event_tx: &mpsc::Sender<RadioEvent>,
tx_recv: &mut mpsc::Receiver<TxRequest>,
) -> Result<()> {
// The Dongle's auto-configure already applied our requested config at
// connect time (via `ConnectOptions::config`). Read back the active
// config — on the mux, another client may have locked a different one.
let active_config = match dongle.config().await {
Some(Modulation::LoRa(c)) => c,
Some(other) => {
warn!("[radio] mux has a non-LoRa config locked ({other:?}); using it anyway");
// This branch is unreachable today since the bridge only supports
// LoRa, but we surface a warning rather than hard-fail so the
// mux doesn't brick the bridge.
*config
}
None => {
warn!("[radio] Dongle::config() returned None after connect; falling back to requested");
*config
}
};
let source = if lora_configs_match(&active_config, config) { ConfigSource::Ours } else { ConfigSource::Mux };
info!("[radio] config: source={source:?}, {}", format_lora_config(&active_config));
dongle.rx_start().await.context("rx_start failed")?;
if event_tx.send(RadioEvent::Connected(active_config, source, device.to_string())).await.is_err() {
return Ok(());
}
let _ = event_tx.blocking_send(RadioEvent::Connected(active_config, config_source, device));
// Set a short timeout so we can interleave RX and TX.
client
.transport_mut()
.set_timeout(Duration::from_millis(100))?;
info!("[radio] entering main loop");
// 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();
// TX requests from the router and RX packets from the radio both drive
// the loop; use tokio::select! to multiplex them.
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);
tokio::select! {
tx = tx_recv.recv() => {
let Some(req) = tx else { return Ok(()); };
handle_tx(&dongle, req, retry_policy, event_tx).await;
}
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() {
rx = dongle.next_rx() => {
match rx {
Some(pkt) => {
handle_rx(pkt, &active_config, event_tx).await;
}
None => {
// Dongle's RX stream ended — session closed.
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(());
}
// Pull any async errors the Dongle accumulated (EFRAME / async
// ERADIO). These get logged but don't kill the session.
for err in dongle.drain_async_errors().await {
warn!("[radio] async error: {err}");
}
// Intentionally unused so callers don't think is_mux matters here.
let _ = is_mux;
}
}
/// Negotiate radio config with the dongle/mux.
///
/// Strategy:
/// 1. GetConfig first to read what's active (caches the result).
/// 2. If it matches our desired config, use it — no SetConfig needed.
/// 3. If it differs, only try SetConfig if we haven't been rejected before.
/// 4. If SetConfig is rejected, set the flag IMMEDIATELY (before the mux can
/// disconnect us and lose this state), then return the cached GetConfig result.
///
/// On reconnect, the flag ensures we never retry SetConfig.
fn negotiate_config<T: Transport>(
client: &mut donglora_client::Client<T>,
desired: &RadioConfig,
) -> Result<(RadioConfig, ConfigSource)> {
info!("[negotiate] get_config...");
match client.get_config() {
Ok(cfg) => {
let source = if configs_match(&cfg, desired) {
info!("[negotiate] get_config OK (matches desired): {}", format_radio_config(&cfg));
ConfigSource::Ours
} else {
info!("[negotiate] get_config OK (differs from desired): {}", format_radio_config(&cfg));
ConfigSource::Mux
};
Ok((cfg, source))
async fn handle_tx(dongle: &Dongle, req: TxRequest, retry_policy: &RetryPolicy, event_tx: &mpsc::Sender<RadioEvent>) {
let size = req.data.len();
let _ = event_tx
.send(RadioEvent::TxAttempt { seq: req.seq, attempt: 1, total_attempts: retry_policy.max_attempts, size })
.await;
let result = dongle.tx_with_retry(&req.data, retry_policy).await;
match result {
Ok(outcome) => {
// Fire a TxRetry event for every attempt beyond the first, so the
// TUI log can show the escalation.
let mut prev_attempt: u8 = 0;
for att in &outcome.attempts {
if prev_attempt != 0 {
let reason = match &outcome.attempts[(prev_attempt - 1) as usize].result {
Err(donglora_client::ClientError::ChannelBusy) => TxRetryReason::ChannelBusy,
Err(donglora_client::ClientError::Busy) => TxRetryReason::QueueFull,
_ => TxRetryReason::Other,
};
let _ = event_tx
.send(RadioEvent::TxRetry {
seq: req.seq,
attempt_that_failed: prev_attempt,
total_attempts: retry_policy.max_attempts,
reason,
backoff_ms: retry_policy.backoff_ms_min,
})
.await;
}
prev_attempt = att.attempt;
}
let _ = event_tx
.send(RadioEvent::TxSucceeded {
seq: req.seq,
attempts_used: outcome.attempts_used(),
final_airtime_us: outcome.final_airtime_us,
size,
})
.await;
}
Err(e) => {
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:#})");
}
}
warn!("TX #{} failed after retries: {e:#}", req.seq);
let _ = event_tx
.send(RadioEvent::TxFailed {
seq: req.seq,
attempts_used: retry_policy.max_attempts,
reason: format!("{e}"),
size,
})
.await;
}
}
}
/// Shorten a device path to just the filename for display.
fn shorten_path(path: &str) -> String {
std::path::Path::new(path)
.file_name()
.map(|f| f.to_string_lossy().into_owned())
.unwrap_or_else(|| path.to_string())
async fn handle_rx(pkt: RxPayload, active_config: &LoRaConfig, event_tx: &mpsc::Sender<RadioEvent>) {
// Tenths-of-dB → whole dB for the existing bridge packet format.
// `rssi_tenths_dbm` and `snr_tenths_db` are i16 and the values stay
// within i16 after the /10 rounding; the `as i16` is safe.
#[allow(clippy::cast_possible_truncation)]
let rssi = (f32::from(pkt.rssi_tenths_dbm) / 10.0).round() as i16;
#[allow(clippy::cast_possible_truncation)]
let snr = (f32::from(pkt.snr_tenths_db) / 10.0).round() as i16;
let grade = snr_grade(snr, active_config.sf);
if !grade.should_forward() {
debug!("RX drop {} bytes rssi={rssi} snr={snr} grade={grade}", pkt.data.len());
return;
}
debug!("RX {} bytes rssi={rssi} snr={snr} grade={grade}", pkt.data.len());
let radio_pkt = RadioPacket { rssi, snr, payload: pkt.data.to_vec() };
let _ = event_tx.send(RadioEvent::Packet(radio_pkt)).await;
}
fn configs_match(a: &RadioConfig, b: &RadioConfig) -> bool {
// ── Display helpers ───────────────────────────────────────────────
fn lora_configs_match(a: &LoRaConfig, b: &LoRaConfig) -> bool {
a.freq_hz == b.freq_hz
&& a.bw == b.bw
&& a.sf == b.sf
@@ -265,53 +328,127 @@ fn configs_match(a: &RadioConfig, b: &RadioConfig) -> bool {
&& 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 {
if let Some(io_err) = e.downcast_ref::<std::io::Error>() {
matches!(
io_err.kind(),
std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
)
} else {
false
}
&& a.header_mode == b.header_mode
&& a.payload_crc == b.payload_crc
&& a.iq_invert == b.iq_invert
}
/// Format a bandwidth value for display.
pub fn format_bandwidth(bw: donglora_client::Bandwidth) -> &'static str {
#[must_use]
pub const fn format_bandwidth(bw: LoRaBandwidth) -> &'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",
LoRaBandwidth::Khz7 => "7.8 kHz",
LoRaBandwidth::Khz10 => "10.4 kHz",
LoRaBandwidth::Khz15 => "15.6 kHz",
LoRaBandwidth::Khz20 => "20.8 kHz",
LoRaBandwidth::Khz31 => "31.25 kHz",
LoRaBandwidth::Khz41 => "41.7 kHz",
LoRaBandwidth::Khz62 => "62.5 kHz",
LoRaBandwidth::Khz125 => "125 kHz",
LoRaBandwidth::Khz250 => "250 kHz",
LoRaBandwidth::Khz500 => "500 kHz",
LoRaBandwidth::Khz200 => "200 kHz",
LoRaBandwidth::Khz400 => "400 kHz",
LoRaBandwidth::Khz800 => "800 kHz",
LoRaBandwidth::Khz1600 => "1600 kHz",
}
}
/// Format a RadioConfig for display.
pub fn format_radio_config(config: &RadioConfig) -> String {
let freq_mhz = config.freq_hz as f64 / 1_000_000.0;
/// Format a `LoRaConfig` for display.
#[must_use]
pub fn format_lora_config(config: &LoRaConfig) -> 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 cr_denom: u8 = match config.cr {
donglora_client::LoRaCodingRate::Cr4_5 => 5,
donglora_client::LoRaCodingRate::Cr4_6 => 6,
donglora_client::LoRaCodingRate::Cr4_7 => 7,
donglora_client::LoRaCodingRate::Cr4_8 => 8,
};
let preamble = if config.preamble_len == 0 {
16
} else {
config.preamble_len
};
let cad = if config.cad != 0 { "on" } else { "off" };
let preamble = config.preamble_len;
format!(
"{freq_mhz:.3}MHz SF{} BW{bw} CR4/{} SW0x{:04X} TX:{power} Pre:{preamble} CAD:{cad}",
config.sf, config.cr, config.sync_word
"{freq_mhz:.3}MHz SF{} BW{bw} CR4/{cr_denom} SW0x{:04X} TX:{} dBm Pre:{preamble}",
config.sf, config.sync_word, config.tx_power_dbm,
)
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
use donglora_client::{LoRaCodingRate, LoRaHeaderMode};
fn default_config() -> LoRaConfig {
LoRaConfig {
freq_hz: 910_525_000,
bw: LoRaBandwidth::Khz62,
sf: 7,
cr: LoRaCodingRate::Cr4_5,
sync_word: 0x1424,
tx_power_dbm: 20,
preamble_len: 16,
header_mode: LoRaHeaderMode::Explicit,
payload_crc: true,
iq_invert: false,
}
}
#[test]
fn lora_configs_match_identical() {
assert!(lora_configs_match(&default_config(), &default_config()));
}
#[test]
fn lora_configs_match_differs_each_field() {
let base = default_config();
let mut c = base;
c.freq_hz = 915_000_000;
assert!(!lora_configs_match(&base, &c));
let mut c = base;
c.sf = 12;
assert!(!lora_configs_match(&base, &c));
let mut c = base;
c.cr = LoRaCodingRate::Cr4_8;
assert!(!lora_configs_match(&base, &c));
let mut c = base;
c.sync_word = 0x1234;
assert!(!lora_configs_match(&base, &c));
let mut c = base;
c.tx_power_dbm = 10;
assert!(!lora_configs_match(&base, &c));
let mut c = base;
c.preamble_len = 8;
assert!(!lora_configs_match(&base, &c));
}
#[test]
fn format_bandwidth_all_variants() {
assert_eq!(format_bandwidth(LoRaBandwidth::Khz7), "7.8 kHz");
assert_eq!(format_bandwidth(LoRaBandwidth::Khz125), "125 kHz");
assert_eq!(format_bandwidth(LoRaBandwidth::Khz500), "500 kHz");
}
#[test]
fn format_lora_config_typical() {
let cfg = default_config();
let s = format_lora_config(&cfg);
assert!(s.contains("910.525MHz"));
assert!(s.contains("SF7"));
assert!(s.contains("CR4/5"));
assert!(s.contains("0x1424"));
assert!(s.contains("20 dBm"));
}
#[test]
fn tx_retry_reason_display() {
assert_eq!(format!("{}", TxRetryReason::ChannelBusy), "channel_busy");
assert_eq!(format!("{}", TxRetryReason::QueueFull), "queue_full");
assert_eq!(format!("{}", TxRetryReason::Other), "other");
}
}
+234 -58
View File
@@ -1,10 +1,15 @@
//! `LoRa` air-time-aware token bucket rate limiter.
//!
//! Calculates transmit rate from the Semtech `LoRa` air time formula and
//! enforces it with a configurable burst allowance.
use std::time::{Duration, Instant};
use donglora_client::Bandwidth;
use donglora_client::LoRaBandwidth;
/// Token-bucket rate limiter for radio transmissions.
///
/// Calculates a safe transmit rate from the LoRa radio configuration (spreading factor,
/// Calculates a safe transmit rate from the `LoRa` radio configuration (spreading factor,
/// bandwidth, coding rate) and enforces it with a token bucket that allows short bursts.
pub struct RateLimiter {
tokens: f64,
@@ -13,43 +18,33 @@ pub struct RateLimiter {
last_refill: Instant,
}
/// Default burst size (max tokens).
/// Default burst size (max tokens). Allows short bursts of 3 back-to-back packets
/// to handle bursty mesh traffic without triggering rate limiting on every packet.
const DEFAULT_BURST: f64 = 3.0;
/// Reference payload size (bytes) used to estimate air time for rate calculation.
/// 50 bytes is a representative `MeshCore` packet size (header + small payload).
const REFERENCE_PAYLOAD_BYTES: usize = 50;
/// Target duty cycle fraction (0.5 = 50%).
/// Target duty cycle fraction. 50% leaves headroom for other radio traffic sharing
/// the channel while still forwarding bridge traffic aggressively.
const TARGET_DUTY_CYCLE: f64 = 0.5;
impl RateLimiter {
/// Create a rate limiter derived from radio config.
///
/// If `override_pps` is `Some`, use that rate directly instead of calculating.
pub fn from_radio_config(
sf: u8,
bw: Bandwidth,
cr: u8,
preamble: u16,
override_pps: Option<f64>,
) -> Self {
let refill_rate = if let Some(pps) = override_pps {
pps.max(0.01)
} else {
let air_time = lora_air_time(REFERENCE_PAYLOAD_BYTES, sf, bw, cr, preamble);
let secs = air_time.as_secs_f64();
if secs > 0.0 {
TARGET_DUTY_CYCLE / secs
} else {
10.0 // fallback
}
};
Self {
tokens: DEFAULT_BURST,
max_tokens: DEFAULT_BURST,
refill_rate,
last_refill: Instant::now(),
}
#[must_use]
pub fn from_radio_config(sf: u8, bw: LoRaBandwidth, cr: u8, preamble: u16, override_pps: Option<f64>) -> Self {
let refill_rate = override_pps.map_or_else(
|| {
let 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.
@@ -64,19 +59,20 @@ impl RateLimiter {
}
/// Current rate in packets per second.
pub fn rate_pps(&self) -> f64 {
#[must_use]
pub const fn rate_pps(&self) -> f64 {
self.refill_rate
}
fn refill(&mut self) {
let now = Instant::now();
let elapsed = now.duration_since(self.last_refill).as_secs_f64();
self.tokens = (self.tokens + elapsed * self.refill_rate).min(self.max_tokens);
self.tokens = elapsed.mul_add(self.refill_rate, self.tokens).min(self.max_tokens);
self.last_refill = now;
}
}
/// Estimate LoRa packet air time using the Semtech SX1276 formula.
/// Estimate `LoRa` packet air time using the Semtech `LoRa` time-on-air formula.
///
/// Parameters:
/// - `payload_bytes`: payload size in bytes
@@ -84,28 +80,27 @@ impl RateLimiter {
/// - `bw`: bandwidth
/// - `cr`: coding rate denominator (5–8, meaning 4/5 to 4/8)
/// - `preamble`: preamble length in symbols (0 = firmware default of 16)
pub fn lora_air_time(payload_bytes: usize, sf: u8, bw: Bandwidth, cr: u8, preamble: u16) -> Duration {
#[must_use]
pub fn lora_air_time(payload_bytes: usize, sf: u8, bw: LoRaBandwidth, cr: u8, preamble: u16) -> Duration {
let sf_f = f64::from(sf);
let bw_hz = bandwidth_hz(bw);
let preamble_symbols = if preamble == 0 { 16.0 } else { f64::from(preamble) };
// Symbol time in seconds.
let t_sym = 2.0_f64.powf(sf_f) / bw_hz;
let t_sym = sf_f.exp2() / bw_hz;
// Preamble time.
let t_preamble = (preamble_symbols + 4.25) * t_sym;
// Low data rate optimize: enabled when SF >= 11 and BW <= 125 kHz.
let de: f64 = if sf >= 11 && bw_hz <= 125_000.0 {
1.0
} else {
0.0
};
let de: f64 = if sf >= 11 && bw_hz <= 125_000.0 { 1.0 } else { 0.0 };
// Payload symbol count (Semtech formula with CRC enabled, explicit header).
let numerator = 8.0 * payload_bytes as f64 - 4.0 * sf_f + 28.0 + 16.0;
let denominator = 4.0 * (sf_f - 2.0 * de);
let n_payload = 8.0 + (numerator / denominator).ceil().max(0.0) * f64::from(cr);
#[allow(clippy::cast_precision_loss)] // payload_bytes is always < 256
let payload_f = payload_bytes as f64;
let numerator = 8.0f64.mul_add(payload_f, (-4.0f64).mul_add(sf_f, 44.0));
let denominator = 4.0f64.mul_add(sf_f, -8.0 * de);
let n_payload = (numerator / denominator).ceil().max(0.0).mul_add(f64::from(cr), 8.0);
let t_payload = n_payload * t_sym;
let total = t_preamble + t_payload;
@@ -113,18 +108,25 @@ pub fn lora_air_time(payload_bytes: usize, sf: u8, bw: Bandwidth, cr: u8, preamb
Duration::from_secs_f64(total)
}
fn bandwidth_hz(bw: Bandwidth) -> f64 {
const fn bandwidth_hz(bw: LoRaBandwidth) -> 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,
LoRaBandwidth::Khz7 => 7_800.0,
LoRaBandwidth::Khz10 => 10_400.0,
LoRaBandwidth::Khz15 => 15_600.0,
LoRaBandwidth::Khz20 => 20_800.0,
LoRaBandwidth::Khz31 => 31_250.0,
LoRaBandwidth::Khz41 => 41_700.0,
LoRaBandwidth::Khz62 => 62_500.0,
LoRaBandwidth::Khz125 => 125_000.0,
LoRaBandwidth::Khz250 => 250_000.0,
LoRaBandwidth::Khz500 => 500_000.0,
// SX128x 2.4 GHz bandwidths — not used on the bridge today, but
// plumb through so SET_CONFIG with one of these still yields a
// usable rate rather than panicking.
LoRaBandwidth::Khz200 => 200_000.0,
LoRaBandwidth::Khz400 => 400_000.0,
LoRaBandwidth::Khz800 => 800_000.0,
LoRaBandwidth::Khz1600 => 1_600_000.0,
}
}
@@ -134,7 +136,7 @@ mod tests {
#[test]
fn air_time_sf7_bw125() {
let t = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16);
let t = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 16);
// SF7/125kHz/CR4/5/16-preamble/50 bytes: ~100ms range
assert!(t.as_millis() > 50, "air time too short: {t:?}");
assert!(t.as_millis() < 300, "air time too long: {t:?}");
@@ -142,7 +144,7 @@ mod tests {
#[test]
fn air_time_sf12_bw125() {
let t = lora_air_time(50, 12, Bandwidth::Khz125, 5, 16);
let t = lora_air_time(50, 12, LoRaBandwidth::Khz125, 5, 16);
// SF12/125kHz: should be multiple seconds
assert!(t.as_secs() >= 1, "air time too short: {t:?}");
assert!(t.as_secs() < 10, "air time too long: {t:?}");
@@ -150,7 +152,7 @@ mod tests {
#[test]
fn rate_limiter_allows_burst() {
let mut rl = RateLimiter::from_radio_config(7, Bandwidth::Khz125, 5, 16, None);
let mut rl = RateLimiter::from_radio_config(7, LoRaBandwidth::Khz125, 5, 16, None);
// Should allow burst of 3
assert!(rl.try_acquire());
assert!(rl.try_acquire());
@@ -161,14 +163,188 @@ mod tests {
#[test]
fn rate_limiter_override() {
let rl = RateLimiter::from_radio_config(7, Bandwidth::Khz125, 5, 16, Some(42.0));
let rl = RateLimiter::from_radio_config(7, LoRaBandwidth::Khz125, 5, 16, Some(42.0));
assert!((rl.rate_pps() - 42.0).abs() < f64::EPSILON);
}
#[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);
let t_default = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 0);
let t_explicit = lora_air_time(50, 7, LoRaBandwidth::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, LoRaBandwidth::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, LoRaBandwidth::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, LoRaBandwidth::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, LoRaBandwidth::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, LoRaBandwidth::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, LoRaBandwidth::Khz125, 5, 16);
let t10 = lora_air_time(50, 10, LoRaBandwidth::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, LoRaBandwidth::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, LoRaBandwidth::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, LoRaBandwidth::Khz125, 5, 8);
let t16 = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 16);
let t32 = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 32);
let diff_8_to_16 = t16.as_secs_f64() - t8.as_secs_f64();
let diff_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 = [
LoRaBandwidth::Khz7,
LoRaBandwidth::Khz10,
LoRaBandwidth::Khz15,
LoRaBandwidth::Khz20,
LoRaBandwidth::Khz31,
LoRaBandwidth::Khz41,
LoRaBandwidth::Khz62,
LoRaBandwidth::Khz125,
LoRaBandwidth::Khz250,
LoRaBandwidth::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 [LoRaBandwidth::Khz7, LoRaBandwidth::Khz62, LoRaBandwidth::Khz125, LoRaBandwidth::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");
}
}
}
}
}
+249 -13
View File
@@ -1,3 +1,8 @@
//! Central packet bus bridging radio and gossip swarm.
//!
//! Applies blake3 content-hash deduplication, token bucket rate limiting,
//! and TX queue management.
use std::collections::{HashMap, VecDeque};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
@@ -8,14 +13,14 @@ use tracing::{debug, warn};
use crate::gossip::GossipEvent;
use crate::packet::{GossipFrame, content_hash};
use crate::radio::{ConfigSource, RadioEvent};
use crate::radio::{ConfigSource, RadioEvent, TxRequest, TxRetryReason};
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 active: donglora_client::LoRaConfig,
pub requested: donglora_client::LoRaConfig,
pub source: ConfigSource,
pub device: String,
pub connected: bool,
@@ -31,6 +36,29 @@ pub struct PacketLogEntry {
pub snr: Option<i16>,
pub rssi: Option<i16>,
pub action: PacketAction,
/// When set, indicates this entry is a TX-attempt update: `Some((attempt, total))`.
pub tx_retry: Option<TxRetryInfo>,
}
/// Per-attempt retry metadata attached to a TX `PacketLogEntry`.
#[derive(Debug, Clone, Copy)]
pub struct TxRetryInfo {
pub attempt: u8,
pub total_attempts: u8,
pub state: TxRetryState,
}
/// State a TX entry is currently in.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TxRetryState {
/// First attempt in flight.
InFlight,
/// A previous attempt failed; this attempt is a retry.
Retrying(TxRetryReason),
/// Final success, possibly after retries.
Succeeded,
/// All attempts exhausted; terminal failure.
Failed,
}
/// Which side the packet arrived from.
@@ -78,9 +106,15 @@ pub struct Stats {
pub dropped_queue: AtomicU64,
pub neighbor_count: AtomicU64,
pub radio_connected: AtomicU64, // 0 = disconnected, 1 = connected
/// Cumulative TX retries across the bridge lifetime (each retry after
/// the first attempt increments by 1).
pub tx_retries: AtomicU64,
/// TX calls that exhausted the retry policy.
pub tx_failures: AtomicU64,
}
impl Stats {
#[must_use]
pub fn snapshot(&self) -> StatsSnapshot {
StatsSnapshot {
radio_rx: self.radio_rx.load(Ordering::Relaxed),
@@ -92,10 +126,13 @@ impl Stats {
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,
tx_retries: self.tx_retries.load(Ordering::Relaxed),
tx_failures: self.tx_failures.load(Ordering::Relaxed),
}
}
}
/// Point-in-time snapshot of bridge statistics, cloned from atomic counters.
#[derive(Debug, Clone)]
pub struct StatsSnapshot {
pub radio_rx: u64,
@@ -107,6 +144,8 @@ pub struct StatsSnapshot {
pub dropped_queue: u64,
pub neighbor_count: u64,
pub radio_connected: bool,
pub tx_retries: u64,
pub tx_failures: u64,
}
/// Time-bounded dedup cache.
@@ -117,10 +156,7 @@ struct DedupCache {
impl DedupCache {
fn new(ttl: Duration) -> Self {
Self {
entries: HashMap::new(),
ttl,
}
Self { entries: HashMap::new(), ttl }
}
/// Returns true if the key was already present (i.e., duplicate).
@@ -144,15 +180,15 @@ impl DedupCache {
/// Run the router — the central packet bus.
///
/// Bridges radio packets to/from the gossip swarm with deduplication and rate limiting.
#[allow(clippy::too_many_arguments)]
#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
pub async fn run(
our_id: PublicKey,
dedup_window: Duration,
tx_queue_size: usize,
requested_radio_config: donglora_client::RadioConfig,
requested_radio_config: donglora_client::LoRaConfig,
mut rate_limiter: RateLimiter,
mut radio_rx: mpsc::Receiver<RadioEvent>,
radio_tx: mpsc::Sender<Vec<u8>>,
radio_tx: mpsc::Sender<TxRequest>,
mut gossip_rx: mpsc::Receiver<GossipEvent>,
gossip_tx: mpsc::Sender<GossipFrame>,
stats: &Stats,
@@ -160,6 +196,7 @@ pub async fn run(
config_tx: tokio::sync::watch::Sender<RadioConfigInfo>,
) {
let mut dedup = DedupCache::new(dedup_window);
let mut next_tx_seq: u64 = 0;
let mut tx_queue: VecDeque<Vec<u8>> = VecDeque::with_capacity(tx_queue_size);
let mut last_cleanup = Instant::now();
let cleanup_interval = Duration::from_secs(30);
@@ -191,6 +228,7 @@ pub async fn run(
snr: Some(pkt.snr),
rssi: Some(pkt.rssi),
action: PacketAction::DroppedDedup,
tx_retry: None,
}).await;
continue;
}
@@ -208,6 +246,7 @@ pub async fn run(
snr: Some(pkt.snr),
rssi: Some(pkt.rssi),
action: PacketAction::Bridged,
tx_retry: None,
}).await;
}
RadioEvent::Connected(active_config, source, device) => {
@@ -224,6 +263,67 @@ pub async fn run(
stats.radio_connected.store(0, Ordering::Relaxed);
config_tx.send_modify(|info| info.connected = false);
}
RadioEvent::TxAttempt { .. } => {
// No log entry: the GossipIn Bridged row that enqueued
// this TX already represents it. Only retries /
// failures produce follow-up rows.
}
RadioEvent::TxRetry { seq, attempt_that_failed, total_attempts, reason, backoff_ms: _ } => {
stats.tx_retries.fetch_add(1, Ordering::Relaxed);
let _ = log_tx.send(PacketLogEntry {
timestamp: Instant::now(),
hash: tx_seq_hash(seq),
direction: PacketDirection::GossipIn,
size: 0,
snr: None,
rssi: None,
action: PacketAction::Bridged,
tx_retry: Some(TxRetryInfo {
attempt: attempt_that_failed.saturating_add(1),
total_attempts,
state: TxRetryState::Retrying(reason),
}),
}).await;
}
RadioEvent::TxSucceeded { seq, attempts_used, final_airtime_us: _, size } => {
stats.radio_tx.fetch_add(1, Ordering::Relaxed);
// Only emit a success row if retries were involved —
// otherwise the original GossipIn row covers it.
if attempts_used > 1 {
let _ = log_tx.send(PacketLogEntry {
timestamp: Instant::now(),
hash: tx_seq_hash(seq),
direction: PacketDirection::GossipIn,
size,
snr: None,
rssi: None,
action: PacketAction::Bridged,
tx_retry: Some(TxRetryInfo {
attempt: attempts_used,
total_attempts: attempts_used,
state: TxRetryState::Succeeded,
}),
}).await;
}
}
RadioEvent::TxFailed { seq, attempts_used, reason, size } => {
stats.tx_failures.fetch_add(1, Ordering::Relaxed);
warn!("TX #{seq} exhausted retries: {reason}");
let _ = log_tx.send(PacketLogEntry {
timestamp: Instant::now(),
hash: tx_seq_hash(seq),
direction: PacketDirection::GossipIn,
size,
snr: None,
rssi: None,
action: PacketAction::Bridged,
tx_retry: Some(TxRetryInfo {
attempt: attempts_used,
total_attempts: attempts_used,
state: TxRetryState::Failed,
}),
}).await;
}
}
}
@@ -246,6 +346,7 @@ pub async fn run(
snr: Some(frame.snr),
rssi: Some(frame.rssi),
action: PacketAction::DroppedDedup,
tx_retry: None,
}).await;
continue;
}
@@ -262,6 +363,7 @@ pub async fn run(
snr: Some(frame.snr),
rssi: Some(frame.rssi),
action: PacketAction::DroppedRateLimit,
tx_retry: None,
}).await;
continue;
}
@@ -278,6 +380,7 @@ pub async fn run(
snr: Some(frame.snr),
rssi: Some(frame.rssi),
action: PacketAction::DroppedQueueFull,
tx_retry: None,
}).await;
continue;
}
@@ -291,21 +394,154 @@ pub async fn run(
snr: Some(frame.snr),
rssi: Some(frame.rssi),
action: PacketAction::Bridged,
tx_retry: None,
}).await;
}
GossipEvent::NeighborChanged(count) => {
#[allow(clippy::cast_possible_truncation)]
stats.neighbor_count.store(count as u64, Ordering::Relaxed);
}
}
}
}
// Drain TX queue to radio.
// Drain TX queue to the async radio task. radio_tx is a bounded
// channel; if the radio task is busy we just wait on the next
// loop iteration rather than blocking here.
while let Some(payload) = tx_queue.pop_front() {
stats.radio_tx.fetch_add(1, Ordering::Relaxed);
if radio_tx.send(payload).await.is_err() {
let seq = {
next_tx_seq = next_tx_seq.wrapping_add(1);
next_tx_seq
};
if radio_tx.send(TxRequest { seq, data: payload }).await.is_err() {
break;
}
}
}
}
/// Build a stable per-TX "hash" for packet-log correlation. Not a real
/// content hash — just a 32-byte encoding of the seq so the TUI can
/// index/update log rows by it.
const fn tx_seq_hash(seq: u64) -> [u8; 32] {
let mut out = [0u8; 32];
out[0] = 0x01; // tag marking this as a TX-seq pseudo-hash
let bytes = seq.to_le_bytes();
out[1] = bytes[0];
out[2] = bytes[1];
out[3] = bytes[2];
out[4] = bytes[3];
out[5] = bytes[4];
out[6] = bytes[5];
out[7] = bytes[6];
out[8] = bytes[7];
out
}
#[cfg(test)]
#[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");
}
}
+45 -31
View File
@@ -17,10 +17,8 @@ const RESET: &str = "\x1b[0m";
/// Run the interactive setup wizard. If `existing` is Some, use those values as defaults.
pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> {
let defaults = existing.map_or_else(
|| (RadioSection::default(), BridgeSection::default()),
|c| (c.radio.clone(), c.bridge.clone()),
);
let defaults = existing
.map_or_else(|| (RadioSection::default(), BridgeSection::default()), |c| (c.radio.clone(), c.bridge.clone()));
let (radio, bridge) = defaults;
println!();
@@ -43,8 +41,9 @@ pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> {
// ── Radio ────────────────────────────────────────────────────
let freq_default = format!("{:.3}", radio.frequency as f64 / 1_000_000.0);
let freq_default = format!("{:.3}", f64::from(radio.frequency) / 1_000_000.0);
let freq_mhz: f64 = prompt_parse("Frequency (MHz)", &freq_default)?;
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)] // validated by prompt_parse
let frequency = (freq_mhz * 1_000_000.0) as u32;
let bandwidth = prompt_default("Bandwidth", &radio.bandwidth)?;
@@ -59,22 +58,15 @@ pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> {
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 tx_power_default = format!("{}", radio.tx_power_dbm);
let tx_power_dbm: i8 = prompt_parse("TX Power (dBm)", &tx_power_default)?;
let pre_default = format!("{}", radio.preamble);
let 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)
};
let port = if port_str == "auto" || port_str.is_empty() { None } else { Some(port_str) };
// ── Build and save ───────────────────────────────────────────
@@ -86,9 +78,8 @@ pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> {
spreading_factor,
coding_rate,
sync_word,
tx_power,
tx_power_dbm,
preamble,
cad,
},
bridge: BridgeSection {
passphrase,
@@ -97,6 +88,7 @@ pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> {
rate_limit_pps: bridge.rate_limit_pps,
log_file: bridge.log_file,
},
tx: existing.map(|c| c.tx.clone()).unwrap_or_default(),
};
write_config(config_path, &cfg)?;
@@ -111,12 +103,10 @@ pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> {
fn write_config(path: &Path, cfg: &Config) -> Result<()> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)
.with_context(|| format!("creating directory: {}", parent.display()))?;
std::fs::create_dir_all(parent).with_context(|| format!("creating directory: {}", parent.display()))?;
}
let content = toml::to_string_pretty(cfg).context("serializing config")?;
std::fs::write(path, content)
.with_context(|| format!("writing config: {}", path.display()))?;
std::fs::write(path, content).with_context(|| format!("writing config: {}", path.display()))?;
Ok(())
}
@@ -128,20 +118,12 @@ fn prompt_default(label: &str, default: &str) -> Result<String> {
let mut input = String::new();
io::stdin().read_line(&mut input)?;
let trimmed = input.trim();
if trimmed.is_empty() {
Ok(default.to_string())
} else {
Ok(trimmed.to_string())
}
if trimmed.is_empty() { Ok(default.to_string()) } else { Ok(trimmed.to_string()) }
}
fn prompt_required(label: &str, default: Option<&str>) -> Result<String> {
loop {
let prompt = if let Some(d) = default {
format!(" {label} {DIM}[{d}]{RESET}: ")
} else {
format!(" {label}: ")
};
let prompt = default.map_or_else(|| format!(" {label}: "), |d| format!(" {label} {DIM}[{d}]{RESET}: "));
print!("{prompt}");
io::stdout().flush()?;
let mut input = String::new();
@@ -175,3 +157,35 @@ fn parse_sync_word(s: &str) -> Result<u16> {
let s = s.strip_prefix("0x").unwrap_or(&s);
u16::from_str_radix(s, 16).with_context(|| format!("invalid sync word: {s}"))
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
#[test]
fn parse_sync_word_hex_prefix() {
assert_eq!(parse_sync_word("0x3444").unwrap(), 0x3444);
}
#[test]
fn parse_sync_word_no_prefix() {
assert_eq!(parse_sync_word("3444").unwrap(), 0x3444);
}
#[test]
fn parse_sync_word_case_insensitive() {
assert_eq!(parse_sync_word("0xABCD").unwrap(), 0xABCD);
}
#[test]
fn parse_sync_word_whitespace() {
assert_eq!(parse_sync_word(" 0x3444 ").unwrap(), 0x3444);
}
#[test]
fn parse_sync_word_invalid() {
assert!(parse_sync_word("ZZZZ").is_err());
assert!(parse_sync_word("").is_err());
}
}
+635 -122
View File
File diff suppressed because it is too large. Load diff