1 Commits
16 changed files with 1880 additions and 2015 deletions

No files matched your search

+3 -183
View File
@@ -1,191 +1,11 @@
# Changelog
## [3.0.1] - 2026-05-11
### Fixed
- **DHT `PublishFailing` no longer wedges the swarm for minutes at a
time.** Pulls in `iroh-gossip-rendezvous` 0.2.1, which retries
failed publishes on a 15 s cadence (default) and bypasses the
`1/(n+1)` write-probability gate while the DHT is unhealthy. Prior
to this, a single failed publish parked the bridge in
`PublishFailing` for at least `write_period` (5 min) and on
average `(n + 1) × write_period` — half an hour with a populated
active view — before the next retry attempt. Recovery is now
bounded by `failure_retry_period × consecutive_failures` and
flips back to `Ready` the instant a write lands.
## [3.0.0] - 2026-05-11
### Breaking
- **`iroh` bumped 0.98 → `=1.0.0-rc.0`** and **`iroh-gossip` bumped
0.98 → 0.99**. The bridge is now built on iroh's first 1.0 release
candidate. iroh-gossip 0.99 hard-pins `iroh = "=1.0.0-rc.0"`, so the
bridge does too until iroh ships a final 1.0 — once it does, this
pin will relax to plain `"1"`. Any downstream consumer that pulls
iroh transitively must move in lockstep.
- **`iroh-gossip-rendezvous` bumped 0.1 → 0.2** (now consumed from
crates.io; the local path-dep is gone). The 0.2 release of the
rendezvous crate carries the same iroh 1.0-rc.0 bump and pins
`ed25519-dalek = "=3.0.0-pre.7"` to match iroh-base's pre-release.
## [0.2.3] - 2026-04-06
### Changed
- `donglora-client` resolves to 1.1.1 (was 1.0.x). Adds the
`connect()`-time auto-clamp of `tx_power_dbm` against
`GET_INFO.tx_power_min/max` and pre-flight rejection of
out-of-range `freq_hz` / `sf` / `bw` with
`ClientError::ConfigNotSupported` — both surface as ordinary
bridge connection errors today.
- `Cargo.lock` refreshed end-to-end: `mainline` 6.0 → 6.1.2, plus
the usual tokio/rustls/hyper patch sweep.
### Notes
- **Pre-v2 config files were silently mis-loaded.** A stale
`~/.config/donglora-bridge/config.toml` from v0.x / v1.x — with
`sync_word = 13380` (`0x3444`, the old LoRaWAN-public default),
`tx_power = "22"` (old string field), or `cad = true` (removed in
v2) — would deserialize without error: serde silently ignored the
unknown `tx_power` / `cad` fields and the bridge fell back to its
per-field defaults, producing a "requested config" that didn't
match what the user had ever entered. On a shared mux the TUI
flagged this as a `!` against the active config. Re-run
`donglora-bridge config` (or update the file by hand) if your
config still has any of `tx_power`, `cad`, or
`sync_word = 13380`. A future minor will tighten the loader so
this fails loudly instead.
## [2.0.1] - 2026-04-25
### Fixed
- **TUI quit no longer hangs on a misleading "updating DHT" screen.**
The 2.0.0 migration to `iroh-gossip-rendezvous` removed the
synchronous mainline-DHT publish that the old `Gossip::shutdown()`
performed (rendezvous entries age out instead), but the TUI flow
was unchanged: pressing `q` drew an "updating DHT…" overlay,
awaited `swarm.shutdown()`, and only then restored the terminal —
so the user sat on the stale message for as long as
`endpoint.close()` took to flush in-flight connections. Restore
the terminal the moment `q` is pressed, drop the overlay, and
cap the graceful-shutdown wait at 2 s so a slow endpoint close
can't stall the process.
## [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.
- README: clarify that the bridge relays raw LoRa packets and works with any
protocol (Meshtastic, MeshCore, etc.) simultaneously.
## [0.2.2] - 2026-04-06
Generated
+550 -843
View File
File diff suppressed because it is too large. Load diff
+9 -7
View File
@@ -1,12 +1,12 @@
[package]
name = "donglora-bridge"
version = "3.0.1"
version = "0.2.3"
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"
repository = "https://github.com/swaits/donglora-bridge"
homepage = "https://github.com/swaits/donglora-bridge"
keywords = ["lora", "bridge", "gossip", "p2p", "iot"]
categories = ["network-programming", "command-line-utilities"]
readme = "README.md"
@@ -14,10 +14,10 @@ rust-version = "1.94.1"
exclude = [".jj/", ".github/"]
[dependencies]
donglora-client = "1"
iroh = "=1.0.0-rc.0"
iroh-gossip = "0.99"
iroh-gossip-rendezvous = "0.2.1"
donglora-client = "0.1"
iroh = "0.97"
iroh-gossip = "0.97"
mainline = { version = "6", features = ["async"] }
tokio = { version = "1", features = ["full"] }
clap = { version = "4", features = ["derive"] }
serde = { version = "1", features = ["derive"] }
@@ -34,8 +34,10 @@ bytes = "1"
futures-lite = "2"
rand = "0.9"
hex = "0.4.3"
sha2 = "0.10"
tokio-util = { version = "0.7.18", features = ["rt"] }
jiff = "0.2.23"
zeroize = "1"
[dev-dependencies]
proptest = "1"
+15 -12
View File
@@ -1,19 +1,22 @@
# donglora-bridge
[![CI](https://github.com/donglora/bridge/actions/workflows/ci.yml/badge.svg)](https://github.com/donglora/bridge/actions/workflows/ci.yml)
[![CI](https://github.com/swaits/donglora-bridge/actions/workflows/ci.yml/badge.svg)](https://github.com/swaits/donglora-bridge/actions/workflows/ci.yml)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)
![Rust](https://img.shields.io/badge/rust-1.94.1%2B-orange)
**donglora-bridge** connects LoRa radios to the internet so packets heard by one
radio can be retransmitted by another radio anywhere in the world.
**donglora-bridge** relays raw LoRa packets over the internet. Every packet your
radio hears is forwarded to every other bridge running the same passphrase, and
their radios retransmit it locally. Because it operates on raw packets, it works
with any protocol built on LoRa --
[Meshtastic](https://meshtastic.org/),
[MeshCore](https://github.com/rocketgithub/meshcore-firmware),
or anything else -- simultaneously and without configuration. If your radio can
hear it, the bridge will relay it.
[LoRa](https://en.wikipedia.org/wiki/LoRa) is a long-range, low-power wireless
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.
modulation used by off-grid mesh networks. A single radio might cover a few
kilometers. This bridge removes that limit, extending a local LoRa 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
@@ -25,12 +28,12 @@ through an encrypted peer-to-peer gossip swarm.
### 1. A DongLoRa radio
[DongLoRa](https://github.com/donglora/firmware) is open-source firmware that
[DongLoRa](https://github.com/swaits/donglora) is open-source firmware that
turns a supported LoRa development board (such as a Heltec WiFi LoRa 32 or
similar ESP32+SX1262 board) into a USB LoRa dongle. You will need to:
1. **Get a board** -- any ESP32 board with an SX1262 LoRa radio that DongLoRa
supports. See the [DongLoRa README](https://github.com/donglora/firmware) for
supports. See the [DongLoRa README](https://github.com/swaits/donglora) for
the full list.
2. **Flash the DongLoRa firmware** onto the board. Instructions are in the
DongLoRa repository.
@@ -68,7 +71,7 @@ mise use -g cargo:donglora-bridge
Or build from source:
```sh
git clone https://github.com/donglora/bridge.git
git clone https://github.com/swaits/donglora-bridge.git
cd donglora-bridge
cargo build --release
```
-11
View File
@@ -1,11 +0,0 @@
# 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`.
+38 -47
View File
@@ -1,68 +1,59 @@
set shell := ["bash", "-c"]
# Default: run all checks
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)
# Run all checks (fmt, clippy, test)
check: fmt-check clippy test audit
# Format code
fmt: _ensure_tools
fmt:
cargo fmt
# Check formatting without changing files
fmt-check: _ensure_tools
fmt-check:
cargo fmt --check
# Lint with clippy
clippy: _ensure_tools
# Lint with clippy (pedantic + nursery)
clippy:
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
build:
cargo build --release
# Generate docs
doc: _ensure_tools
doc:
cargo doc --no-deps --open
# Check for outdated dependencies
outdated: _ensure_tools
cargo outdated
# Run the bridge (TUI mode)
run *ARGS:
cargo run --release -- {{ARGS}}
# Run in headless mode
run-headless *ARGS:
cargo run --release -- --log-only {{ARGS}}
# Run the config wizard
config:
cargo run --release -- config
# Security + license audit
audit:
cargo deny check
# Run all tests
test: test-unit test-fuzz test-mutants
# Normal unit tests
test-unit:
cargo nextest run
# Mutation testing
test-mutants *ARGS:
cargo mutants {{ARGS}}
# Fuzz a target (requires nightly)
test-fuzz TARGET="fuzz_gossip_frame_decode" DURATION="60":
cargo +nightly fuzz run {{TARGET}} -- -max_total_time={{DURATION}}
# Clean build artifacts
clean:
-3
View File
@@ -1,3 +0,0 @@
[tools]
rust = "1.94"
just = "latest"
+182 -167
View File
@@ -1,13 +1,12 @@
//! 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`.
//! Handles the `[radio]` and `[bridge]` sections with `MeshCore` US/Canada recommended
//! defaults, and conversion to donglora-client `RadioConfig`.
use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::{Context, Result, bail};
use donglora_client::{LoRaBandwidth, LoRaCodingRate, LoRaConfig, LoRaHeaderMode, RetryPolicy};
use donglora_client::{Bandwidth, RadioConfig, TX_POWER_MAX};
use serde::{Deserialize, Serialize};
/// Top-level bridge configuration.
@@ -17,8 +16,6 @@ pub struct Config {
pub radio: RadioSection,
#[serde(default)]
pub bridge: BridgeSection,
#[serde(default)]
pub tx: TxSection,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
@@ -36,9 +33,11 @@ pub struct RadioSection {
#[serde(default = "default_sync_word")]
pub sync_word: u16,
#[serde(default = "default_tx_power")]
pub tx_power_dbm: i8,
pub tx_power: String,
#[serde(default = "default_preamble")]
pub preamble: u16,
#[serde(default = "default_true")]
pub cad: bool,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
@@ -56,25 +55,6 @@ 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) ────────────────────
const fn default_frequency() -> u32 {
@@ -89,41 +69,24 @@ const fn default_sf() -> u8 {
const fn default_cr() -> u8 {
5
}
/// `MeshCore` private sync word encoded for the `SX126x` (see client-py
/// `examples/_common.py` — `RADIOLIB_SX126X_SYNC_WORD_PRIVATE`).
const fn default_sync_word() -> u16 {
0x1424
0x3444
}
const fn default_tx_power() -> i8 {
20
fn default_tx_power() -> String {
"22".into()
}
const fn default_preamble() -> u16 {
16
}
const fn default_true() -> bool {
true
}
const fn default_dedup_window() -> u64 {
300
}
const fn default_tx_queue_size() -> usize {
32
}
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 {
@@ -134,8 +97,9 @@ impl Default for RadioSection {
spreading_factor: default_sf(),
coding_rate: default_cr(),
sync_word: default_sync_word(),
tx_power_dbm: default_tx_power(),
tx_power: default_tx_power(),
preamble: default_preamble(),
cad: true,
}
}
}
@@ -152,95 +116,56 @@ 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<LoRaBandwidth> {
fn parse_bandwidth(s: &str) -> Result<Bandwidth> {
match s.to_lowercase().as_str() {
"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),
"7.8khz" | "7khz" | "khz7" => Ok(Bandwidth::Khz7),
"10.4khz" | "10khz" | "khz10" => Ok(Bandwidth::Khz10),
"15.6khz" | "15khz" | "khz15" => Ok(Bandwidth::Khz15),
"20.8khz" | "20khz" | "khz20" => Ok(Bandwidth::Khz20),
"31.25khz" | "31khz" | "khz31" => Ok(Bandwidth::Khz31),
"41.7khz" | "41khz" | "khz41" => Ok(Bandwidth::Khz41),
"62.5khz" | "62khz" | "khz62" => Ok(Bandwidth::Khz62),
"125khz" | "khz125" => Ok(Bandwidth::Khz125),
"250khz" | "khz250" => Ok(Bandwidth::Khz250),
"500khz" | "khz500" => Ok(Bandwidth::Khz500),
_ => bail!("unknown bandwidth: {s}"),
}
}
fn parse_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}"),
fn parse_tx_power(s: &str) -> Result<i8> {
if s.eq_ignore_ascii_case("max") {
Ok(TX_POWER_MAX)
} else {
s.parse::<i8>().context("invalid tx_power (use 'max' or a dBm integer)")
}
}
impl RadioSection {
/// Convert to the donglora-client `LoRaConfig`.
/// Convert to the donglora-client `RadioConfig`.
///
/// Validates that all radio parameters are within hardware-supported ranges
/// before constructing the config.
pub fn to_lora_config(&self) -> Result<LoRaConfig> {
pub fn to_radio_config(&self) -> Result<RadioConfig> {
if !(7..=12).contains(&self.spreading_factor) {
bail!("spreading_factor must be 7-12, got {}", self.spreading_factor);
}
if !(5..=8).contains(&self.coding_rate) {
bail!("coding_rate must be 5-8 (meaning 4/5 to 4/8), got {}", self.coding_rate);
}
if self.frequency < 150_000_000 || self.frequency > 1_020_000_000 {
bail!("frequency {:.3} MHz out of range (150 MHz to 1.02 GHz)", f64::from(self.frequency) / 1_000_000.0);
}
Ok(LoRaConfig {
Ok(RadioConfig {
freq_hz: self.frequency,
bw: parse_bandwidth(&self.bandwidth)?,
sf: self.spreading_factor,
cr: parse_coding_rate(self.coding_rate)?,
cr: self.coding_rate,
sync_word: self.sync_word,
tx_power_dbm: self.tx_power_dbm,
tx_power_dbm: parse_tx_power(&self.tx_power)?,
preamble_len: self.preamble,
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,
cad: u8::from(self.cad),
})
}
}
@@ -284,97 +209,92 @@ mod tests {
#[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)));
assert!(matches!(parse_bandwidth("7.8kHz"), Ok(Bandwidth::Khz7)));
assert!(matches!(parse_bandwidth("62.5kHz"), Ok(Bandwidth::Khz62)));
assert!(matches!(parse_bandwidth("125kHz"), Ok(Bandwidth::Khz125)));
assert!(matches!(parse_bandwidth("500kHz"), Ok(Bandwidth::Khz500)));
}
#[test]
fn parse_bandwidth_aliases() {
assert!(matches!(parse_bandwidth("7khz"), Ok(LoRaBandwidth::Khz7)));
assert!(matches!(parse_bandwidth("khz125"), Ok(LoRaBandwidth::Khz125)));
assert!(matches!(parse_bandwidth("7khz"), Ok(Bandwidth::Khz7)));
assert!(matches!(parse_bandwidth("khz125"), Ok(Bandwidth::Khz125)));
assert!(matches!(parse_bandwidth("10khz"), Ok(Bandwidth::Khz10)));
}
#[test]
fn parse_bandwidth_case_insensitive() {
assert!(matches!(parse_bandwidth("125KHZ"), Ok(LoRaBandwidth::Khz125)));
assert!(matches!(parse_bandwidth("125KHZ"), Ok(Bandwidth::Khz125)));
assert!(matches!(parse_bandwidth("62.5KHZ"), Ok(Bandwidth::Khz62)));
}
#[test]
fn parse_bandwidth_rejects_unknown() {
assert!(parse_bandwidth("999kHz").is_err());
assert!(parse_bandwidth("").is_err());
assert!(parse_bandwidth("not_a_bandwidth").is_err());
}
#[test]
fn parse_coding_rate_canonical() {
assert!(matches!(parse_coding_rate(5), Ok(LoRaCodingRate::Cr4_5)));
assert!(matches!(parse_coding_rate(8), Ok(LoRaCodingRate::Cr4_8)));
fn parse_tx_power_numeric() {
assert_eq!(parse_tx_power("22").unwrap(), 22);
assert_eq!(parse_tx_power("-4").unwrap(), -4);
assert_eq!(parse_tx_power("0").unwrap(), 0);
}
#[test]
fn parse_coding_rate_rejects_out_of_range() {
assert!(parse_coding_rate(4).is_err());
assert!(parse_coding_rate(9).is_err());
fn parse_tx_power_max_keyword() {
assert_eq!(parse_tx_power("max").unwrap(), TX_POWER_MAX);
assert_eq!(parse_tx_power("MAX").unwrap(), TX_POWER_MAX);
assert_eq!(parse_tx_power("Max").unwrap(), TX_POWER_MAX);
}
#[test]
fn radio_section_defaults_to_lora_config() {
fn parse_tx_power_rejects_garbage() {
assert!(parse_tx_power("abc").is_err());
assert!(parse_tx_power("").is_err());
}
#[test]
fn radio_section_defaults_to_radio_config() {
let section = RadioSection::default();
let 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);
let config = section.to_radio_config().unwrap();
assert_eq!(config.freq_hz, 910_525_000);
assert_eq!(config.sf, 7);
assert_eq!(config.cr, 5);
assert_eq!(config.sync_word, 0x3444);
assert_eq!(config.preamble_len, 16);
}
#[test]
fn to_lora_config_rejects_invalid_sf() {
fn radio_section_max_power() {
let section = RadioSection { tx_power: "max".into(), ..RadioSection::default() };
let config = section.to_radio_config().unwrap();
assert_eq!(config.tx_power_dbm, TX_POWER_MAX);
}
#[test]
fn to_radio_config_rejects_invalid_sf() {
let section = RadioSection { spreading_factor: 6, ..RadioSection::default() };
assert!(section.to_lora_config().is_err());
assert!(section.to_radio_config().is_err());
let section = RadioSection { spreading_factor: 13, ..RadioSection::default() };
assert!(section.to_lora_config().is_err());
assert!(section.to_radio_config().is_err());
}
#[test]
fn to_lora_config_rejects_invalid_cr() {
fn to_radio_config_rejects_invalid_cr() {
let section = RadioSection { coding_rate: 4, ..RadioSection::default() };
assert!(section.to_lora_config().is_err());
assert!(section.to_radio_config().is_err());
let section = RadioSection { coding_rate: 9, ..RadioSection::default() };
assert!(section.to_lora_config().is_err());
assert!(section.to_radio_config().is_err());
}
#[test]
fn to_lora_config_rejects_invalid_frequency() {
fn to_radio_config_rejects_invalid_frequency() {
let section = RadioSection { frequency: 0, ..RadioSection::default() };
assert!(section.to_lora_config().is_err());
assert!(section.to_radio_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());
assert!(section.to_radio_config().is_err());
}
#[test]
@@ -385,7 +305,6 @@ mod tests {
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]
@@ -416,4 +335,100 @@ mod tests {
std::fs::write(&path, "not valid toml {{{{").unwrap();
assert!(load_config(&path).is_err());
}
// ── Mutant killers: default values (items 1-5) ─────────────────
#[test]
fn default_radio_cad_is_true() {
let radio = RadioSection::default();
assert!(radio.cad);
// Also test via serde (exercises the default_true function used by #[serde(default)])
let radio: RadioSection = toml::from_str("").unwrap();
assert!(radio.cad);
}
#[test]
fn default_dedup_window_is_300() {
let bridge = BridgeSection::default();
assert_eq!(bridge.dedup_window_secs, 300);
}
#[test]
fn default_tx_queue_size_is_32() {
let bridge = BridgeSection::default();
assert_eq!(bridge.tx_queue_size, 32);
}
// ── Mutant killers: missing bandwidth variants (items 6-10) ──
#[test]
fn parse_bandwidth_all_canonical_variants() {
assert!(matches!(parse_bandwidth("15.6kHz"), Ok(Bandwidth::Khz15)));
assert!(matches!(parse_bandwidth("20.8kHz"), Ok(Bandwidth::Khz20)));
assert!(matches!(parse_bandwidth("31.25kHz"), Ok(Bandwidth::Khz31)));
assert!(matches!(parse_bandwidth("41.7kHz"), Ok(Bandwidth::Khz41)));
assert!(matches!(parse_bandwidth("250kHz"), Ok(Bandwidth::Khz250)));
}
// ── Mutant killers: frequency boundary (items 11-12) ─────────
#[test]
fn to_radio_config_accepts_boundary_frequencies() {
let low = RadioSection { frequency: 150_000_000, ..RadioSection::default() };
assert!(low.to_radio_config().is_ok());
let high = RadioSection { frequency: 1_020_000_000, ..RadioSection::default() };
assert!(high.to_radio_config().is_ok());
}
// ── Mutant killers: path helpers (items 13-15) ───────────────
#[test]
fn config_dir_is_non_empty() {
let dir = config_dir().unwrap();
assert!(dir.components().count() > 0);
}
#[test]
fn config_path_ends_with_config_toml() {
let path = config_path().unwrap();
assert_eq!(path.file_name().unwrap(), "config.toml");
}
#[test]
fn default_log_path_ends_with_bridge_log() {
let path = default_log_path().unwrap();
assert_eq!(path.file_name().unwrap(), "bridge.log");
}
#[test]
fn load_config_full_round_trip() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.toml");
std::fs::write(
&path,
r#"
[radio]
frequency = 915000000
bandwidth = "125kHz"
spreading_factor = 12
coding_rate = 8
sync_word = 0x1234
tx_power = "max"
preamble = 8
cad = false
[bridge]
passphrase = "test-phrase"
dedup_window_secs = 600
tx_queue_size = 64
"#,
)
.unwrap();
let cfg = load_config(&path).unwrap();
assert_eq!(cfg.radio.frequency, 915_000_000);
assert_eq!(cfg.radio.spreading_factor, 12);
assert_eq!(cfg.bridge.dedup_window_secs, 600);
assert_eq!(cfg.bridge.tx_queue_size, 64);
}
}
+664 -73
View File
@@ -1,28 +1,127 @@
//! Bridge-side glue over the `iroh-gossip-rendezvous` crate.
//! iroh-gossip swarm management and mainline DHT peer discovery.
//!
//! 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.
//! Derives cryptographic keys from a shared passphrase, manages the gossip
//! subscription, and maintains a DHT heartbeat for zero-configuration peer
//! bootstrapping.
use std::collections::HashSet;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use anyhow::{Context, Result};
use bytes::Bytes;
use iroh::{PublicKey, SecretKey};
use futures_lite::StreamExt;
use iroh::endpoint::Incoming;
use iroh::{Endpoint, PublicKey, SecretKey};
use iroh_gossip::TopicId;
use iroh_gossip::api::Event;
use iroh_gossip_rendezvous::Rendezvous;
pub use iroh_gossip_rendezvous::{DhtStatus, RendezvousState};
use iroh_gossip::net::{GOSSIP_ALPN, Gossip as GossipActor};
use mainline::async_dht::AsyncDht;
use mainline::{Dht, MutableItem, SigningKey};
use sha2::{Digest, Sha256};
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, warn};
use tracing::{debug, info, warn};
use crate::packet::GossipFrame;
/// 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";
/// Maximum number of peers to store in a DHT record.
/// 28 × 32 bytes = 896 bytes raw. Bencoded: 28 × ~36 = ~1008.
/// Conservative cap to stay under BEP44's 1000-byte value limit.
const MAX_DHT_PEERS: usize = 27;
/// DHT heartbeat interval when connected (base, before jitter). 5 minutes
/// balances freshness against DHT write amplification — connected nodes already
/// have live gossip links and only need DHT for newcomer discovery.
const DHT_HEARTBEAT_SECS: u64 = 300;
/// Maximum random jitter added to heartbeat interval. Prevents synchronized
/// publish storms when multiple bridges start simultaneously.
const DHT_HEARTBEAT_JITTER_SECS: u64 = 60;
// ── Key derivation ───────────────────────────────────────────────
/// Cryptographic keys derived from a shared passphrase.
pub struct PassphraseKeys {
pub topic_id: TopicId,
pub dht_signing_key: SigningKey,
pub dht_salt: [u8; 20],
}
impl PassphraseKeys {
/// Derive all keys from a passphrase.
///
/// Intermediate buffers containing the passphrase are zeroized after use to
/// minimize time sensitive material spends in memory.
#[must_use]
pub fn derive(passphrase: &str) -> Self {
// Application-specific prefix to avoid DHT collisions with other apps.
const APP_PREFIX: &[u8] = b"donglora-bridge:";
// Helper: build a zeroized buffer of [prefix | suffix | passphrase].
let keyed_buf = |suffix: &[u8]| -> Vec<u8> {
let mut buf = Vec::with_capacity(APP_PREFIX.len() + suffix.len() + passphrase.len());
buf.extend_from_slice(APP_PREFIX);
buf.extend_from_slice(suffix);
buf.extend_from_slice(passphrase.as_bytes());
buf
};
// Topic ID: blake3 hash of prefixed passphrase.
let mut buf = keyed_buf(b"");
let topic_bytes = *blake3::hash(&buf).as_bytes();
let topic_id = TopicId::from_bytes(topic_bytes);
zeroize::Zeroize::zeroize(&mut buf);
// DHT signing key: ed25519 from SHA-256 of prefixed passphrase + "sign:".
let mut buf = keyed_buf(b"sign:");
let sign_seed = Sha256::digest(&buf);
let seed_bytes: [u8; 32] = sign_seed.into();
let dht_signing_key = SigningKey::from_bytes(&seed_bytes);
zeroize::Zeroize::zeroize(&mut buf);
// DHT salt: first 20 bytes of SHA-256 of prefixed passphrase + "salt:".
let mut buf = keyed_buf(b"salt:");
let salt_hash = Sha256::digest(&buf);
let mut dht_salt = [0u8; 20];
dht_salt.copy_from_slice(&salt_hash[..20]);
zeroize::Zeroize::zeroize(&mut buf);
Self { topic_id, dht_signing_key, dht_salt }
}
}
// ── Swarm state (for TUI) ───────────────────────────────────────
/// Observable state of the gossip swarm.
#[derive(Debug, Clone)]
pub struct SwarmState {
pub topic_hash: String,
pub neighbor_count: usize,
pub dht_status: DhtStatus,
pub last_dht_publish: Option<Instant>,
}
/// Status of the mainline DHT connection.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DhtStatus {
Bootstrapping,
Ready,
PublishFailed,
}
impl std::fmt::Display for DhtStatus {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Bootstrapping => write!(f, "Bootstrapping"),
Self::Ready => write!(f, "Ready"),
Self::PublishFailed => write!(f, "Failed"),
}
}
}
// ── Events to router ────────────────────────────────────────────
/// Events from the gossip swarm to the router.
pub enum GossipEvent {
@@ -32,90 +131,202 @@ pub enum GossipEvent {
NeighborChanged(usize),
}
/// Handle held by the rest of the bridge. Owns the `Rendezvous` that in turn
/// owns all DHT + gossip background work. Dropping this stops everything.
// ── Gossip handle ───────────────────────────────────────────────
/// Handle to the gossip swarm. Manages iroh endpoint, gossip subscription, and DHT.
pub struct Gossip {
rendezvous: Rendezvous,
endpoint: Endpoint,
actor: GossipActor,
// std::sync::Mutex (not tokio::sync::Mutex) because locks are never held across
// .await points. This avoids async-aware mutex overhead for brief critical sections.
state: Arc<Mutex<SwarmState>>,
neighbors: Arc<Mutex<HashSet<PublicKey>>>,
dht_signing_key: SigningKey,
dht_salt: [u8; 20],
cancel: CancellationToken,
}
impl Gossip {
/// Spin up the gossip swarm + DHT rendezvous.
/// Create and start the gossip network.
///
/// Spawns background tasks for connection acceptance, gossip event forwarding,
/// gossip broadcasting, and DHT heartbeat.
///
/// Returns `(handle, event_rx, frame_tx)`:
/// - `event_rx`: channel of events from the swarm to the router
/// - `frame_tx`: channel for the router to send frames to broadcast
pub async fn new(
secret_key: SecretKey,
passphrase: &str,
keys: &PassphraseKeys,
) -> Result<(Self, mpsc::Receiver<GossipEvent>, mpsc::Sender<GossipFrame>)> {
let cancel = CancellationToken::new();
let rendezvous = Rendezvous::builder()
.passphrase(passphrase)
.app_salt(APP_SALT)
// Build iroh endpoint with n0 discovery.
let endpoint = Endpoint::builder(iroh::endpoint::presets::N0)
.secret_key(secret_key)
.build()
.alpns(vec![GOSSIP_ALPN.to_vec()])
.bind()
.await
.context("starting iroh-gossip rendezvous")?;
.context("binding iroh endpoint")?;
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]));
let my_id = endpoint.id();
info!("iroh endpoint bound, id: {}", my_id.fmt_short());
// Build gossip actor.
let actor = GossipActor::builder().spawn(endpoint.clone());
// Bootstrap from DHT.
let bootstrap_peers = dht_get_peers(keys).await;
if bootstrap_peers.is_empty() {
info!("no peers found in DHT — we are the first node");
} else {
info!("found {} peer(s) in DHT", bootstrap_peers.len());
}
// Subscribe to gossip topic.
let topic = actor.subscribe(keys.topic_id, bootstrap_peers).await.context("subscribing to gossip topic")?;
let (sender, receiver) = topic.split();
// Initial swarm state.
let topic_hash = hex::encode(&keys.topic_id.as_bytes()[..8]);
let state = Arc::new(Mutex::new(SwarmState {
topic_hash,
neighbor_count: 0,
dht_status: DhtStatus::Bootstrapping,
last_dht_publish: None,
}));
let neighbor_count = Arc::new(AtomicUsize::new(0));
let neighbors: Arc<Mutex<HashSet<PublicKey>>> = Arc::new(Mutex::new(HashSet::new()));
// Channels.
let (event_tx, event_rx) = mpsc::channel::<GossipEvent>(256);
let (frame_tx, frame_rx) = mpsc::channel::<GossipFrame>(256);
// Receiver task: drain the rendezvous event stream, translate into
// router-side `GossipEvent`s.
spawn_receiver(rendezvous.subscribe(), event_tx, cancel.clone());
let handle = Self {
endpoint: endpoint.clone(),
actor: actor.clone(),
state: state.clone(),
neighbors: neighbors.clone(),
dht_signing_key: keys.dht_signing_key.clone(),
dht_salt: keys.dht_salt,
cancel: cancel.clone(),
};
// 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 accept loop.
spawn_accept_loop(endpoint.clone(), actor.clone(), cancel.clone());
Ok((Self { rendezvous, cancel }, event_rx, frame_tx))
// Spawn gossip receiver task.
spawn_receiver(receiver, event_tx, state.clone(), neighbor_count.clone(), neighbors.clone(), cancel.clone());
// Clone sender for DHT heartbeat to feed discovered peers via join_peers.
let dht_sender = sender.clone();
// Spawn gossip broadcaster task.
spawn_broadcaster(sender, frame_rx, cancel.clone());
// Spawn DHT heartbeat.
spawn_dht_heartbeat(
my_id,
keys.dht_signing_key.clone(),
keys.dht_salt,
dht_sender,
neighbors,
state,
neighbor_count,
cancel,
);
// Try to join (non-blocking — fire and forget with timeout).
// The receiver task handles events regardless.
Ok((handle, event_rx, frame_tx))
}
/// Snapshot of rendezvous state for the TUI.
/// Snapshot of swarm state for TUI.
#[must_use]
pub fn state(&self) -> RendezvousState {
self.rendezvous.state()
pub fn swarm_state(&self) -> SwarmState {
self.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner).clone()
}
/// Graceful shutdown. Cancels our own glue tasks, then the rendezvous.
/// Idempotent.
/// Graceful shutdown. Publishes neighbors (without ourselves) to DHT
/// so other nodes can still bootstrap from live peers.
pub async fn shutdown(&self) {
// Publish neighbors-only to DHT before shutting down.
let neighbor_ids: Vec<[u8; 32]> = self
.neighbors
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.iter()
.map(|id| *id.as_bytes())
.collect();
if let Ok(dht) = Dht::client() {
let dht = dht.as_async();
let value = encode_peer_list(&neighbor_ids);
let seq = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs().cast_signed())
.unwrap_or(0);
let item = MutableItem::new(self.dht_signing_key.clone(), &value, seq, Some(self.dht_salt.as_ref()));
match dht.put_mutable(item, None).await {
Ok(_) => info!("shutdown: published {} neighbor(s) to DHT (removed self)", neighbor_ids.len()),
Err(e) => debug!("shutdown: DHT publish failed: {e:#}"),
}
}
self.cancel.cancel();
self.rendezvous.shutdown().await;
let _ = self.actor.shutdown().await;
self.endpoint.close().await;
}
}
// ── Internal tasks ──────────────────────────────────────────────────────
// ── Background tasks ────────────────────────────────────────────
/// Accept incoming iroh connections and forward gossip ALPN to the gossip actor.
fn spawn_accept_loop(endpoint: Endpoint, gossip: GossipActor, cancel: CancellationToken) {
tokio::spawn(async move {
loop {
let incoming = tokio::select! {
incoming = endpoint.accept() => incoming,
() = cancel.cancelled() => break,
};
let Some(incoming) = incoming else { break };
let gossip = gossip.clone();
tokio::spawn(handle_incoming(incoming, gossip));
}
debug!("accept loop stopped");
});
}
async fn handle_incoming(incoming: Incoming, gossip: GossipActor) {
let Ok(conn) = incoming.await else { return };
let alpn = conn.alpn();
if *alpn == *GOSSIP_ALPN {
if let Err(e) = gossip.handle_connection(conn).await {
debug!("gossip handle_connection error: {e:#}");
}
} else {
debug!("unexpected ALPN: {:?}", String::from_utf8_lossy(alpn));
}
}
/// Forward gossip events to the router via channel.
fn spawn_receiver(
mut rx: tokio::sync::broadcast::Receiver<Event>,
mut receiver: iroh_gossip::api::GossipReceiver,
event_tx: mpsc::Sender<GossipEvent>,
state: Arc<Mutex<SwarmState>>,
neighbor_count: Arc<AtomicUsize>,
neighbors: Arc<Mutex<HashSet<PublicKey>>>,
cancel: CancellationToken,
) {
tokio::spawn(async move {
// 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 {
loop {
let event = tokio::select! {
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;
}
event = receiver.next() => event,
() = cancel.cancelled() => break,
};
let Some(Ok(event)) = event else { break };
match event {
Event::Received(msg) => {
if let Some(frame) = GossipFrame::decode(&msg.content) {
@@ -124,29 +335,41 @@ fn spawn_receiver(
debug!("received malformed gossip frame ({} bytes)", msg.content.len());
}
}
// Neighbor count is tracked in both AtomicUsize (for lock-free reads in DHT
// heartbeat) and SwarmState (for TUI snapshots). Brief inconsistency between
// the two is acceptable — the TUI polls at 250ms intervals.
Event::NeighborUp(id) => {
neighbors.insert(id);
tracing::info!("neighbor up: {} (total: {})", id.fmt_short(), neighbors.len());
let _ = event_tx.send(GossipEvent::NeighborChanged(neighbors.len())).await;
let count = neighbor_count.fetch_add(1, Ordering::Relaxed) + 1;
info!("neighbor up: {} (total: {count})", id.fmt_short());
if let Ok(mut n) = neighbors.lock() {
n.insert(id);
}
if let Ok(mut s) = state.lock() {
s.neighbor_count = count;
}
let _ = event_tx.send(GossipEvent::NeighborChanged(count)).await;
}
Event::NeighborDown(id) => {
neighbors.remove(&id);
tracing::info!("neighbor down: {} (total: {})", id.fmt_short(), neighbors.len());
let _ = event_tx.send(GossipEvent::NeighborChanged(neighbors.len())).await;
let count = neighbor_count.fetch_sub(1, Ordering::Relaxed).saturating_sub(1);
info!("neighbor down: {} (total: {count})", id.fmt_short());
if let Ok(mut n) = neighbors.lock() {
n.remove(&id);
}
if let Ok(mut s) = state.lock() {
s.neighbor_count = count;
}
let _ = event_tx.send(GossipEvent::NeighborChanged(count)).await;
}
Event::Lagged => {
warn!("gossip receiver lagged — some messages were dropped");
}
}
};
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>,
@@ -168,11 +391,379 @@ fn spawn_broadcaster(
});
}
// 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)
/// DHT poll interval while alone. 3 seconds is aggressive but reads are cheap
/// (single DHT GET) and fast discovery matters when two nodes start together.
const DHT_ALONE_READ_SECS: u64 = 3;
/// How often a lonely node publishes (merge-only, never destructive). 30 seconds
/// is much slower than reads because DHT PUT is heavier and a lonely node only
/// adds itself to the peer list.
const DHT_ALONE_PUBLISH_SECS: u64 = 30;
/// Periodically interact with the DHT to discover and share peers.
///
/// Strategy:
/// - **Alone:** Read every 3s (cheap). Publish every 30s via merge — reads
/// existing data first, appends ourselves, writes back. Two fresh nodes
/// both doing this will eventually see each other. Never blindly overwrites.
/// - **Connected:** Read + publish on relaxed 5-min schedule. We have
/// valuable neighbor info to share.
/// - **Transition to connected:** Publish immediately.
#[allow(clippy::too_many_arguments)]
fn spawn_dht_heartbeat(
my_id: PublicKey,
signing_key: SigningKey,
salt: [u8; 20],
gossip_sender: iroh_gossip::api::GossipSender,
neighbors: Arc<Mutex<HashSet<PublicKey>>>,
state: Arc<Mutex<SwarmState>>,
neighbor_count: Arc<AtomicUsize>,
cancel: CancellationToken,
) {
tokio::spawn(async move {
let dht = match Dht::client() {
Ok(d) => d.as_async(),
Err(e) => {
warn!("failed to create DHT client: {e:#}");
return;
}
};
let pub_key_bytes = signing_key.verifying_key().to_bytes();
let mut was_alone = true;
let mut last_publish = Instant::now();
// Initial: read to discover existing peers, then merge-publish ourselves.
dht_read_and_join(&dht, &my_id, &pub_key_bytes, &salt, &gossip_sender, &state).await;
dht_publish(&dht, &my_id, &signing_key, &salt, &pub_key_bytes, &neighbors, &state).await;
loop {
let alone = neighbor_count.load(Ordering::Relaxed) == 0;
// Transition: alone → connected. Publish immediately.
if was_alone && !alone {
info!("first neighbor joined — publishing to DHT");
dht_publish(&dht, &my_id, &signing_key, &salt, &pub_key_bytes, &neighbors, &state).await;
last_publish = Instant::now();
}
was_alone = alone;
// Sleep: fast reads while alone, relaxed when connected.
let sleep_dur = if alone {
Duration::from_secs(DHT_ALONE_READ_SECS)
} else {
let jitter = rand::random::<u64>() % DHT_HEARTBEAT_JITTER_SECS.max(1);
Duration::from_secs(DHT_HEARTBEAT_SECS + jitter)
};
tokio::select! {
() = tokio::time::sleep(sleep_dur) => {}
() = cancel.cancelled() => break,
}
// Always read — cheap, discovers peers to join.
dht_read_and_join(&dht, &my_id, &pub_key_bytes, &salt, &gossip_sender, &state).await;
// Publish cadence depends on state.
let publish_interval = if alone {
Duration::from_secs(DHT_ALONE_PUBLISH_SECS)
} else {
Duration::from_secs(DHT_HEARTBEAT_SECS)
};
if last_publish.elapsed() >= publish_interval {
dht_publish(&dht, &my_id, &signing_key, &salt, &pub_key_bytes, &neighbors, &state).await;
last_publish = Instant::now();
}
}
debug!("DHT heartbeat stopped");
});
}
/// Read the DHT and feed any discovered peers into the gossip swarm.
async fn dht_read_and_join(
dht: &AsyncDht,
my_id: &PublicKey,
pub_key_bytes: &[u8; 32],
salt: &[u8; 20],
gossip_sender: &iroh_gossip::api::GossipSender,
state: &Arc<Mutex<SwarmState>>,
) {
if let Some(record) = dht.get_mutable_most_recent(pub_key_bytes, Some(salt)).await {
let peers = decode_peer_list(record.value());
let new_peers: Vec<_> = peers.into_iter().filter(|id| *id != *my_id).collect();
if !new_peers.is_empty() {
info!("DHT discovered {} peer(s), feeding to gossip", new_peers.len());
if let Err(e) = gossip_sender.join_peers(new_peers).await {
debug!("join_peers error: {e:#}");
}
}
if let Ok(mut s) = state.lock() {
s.dht_status = DhtStatus::Ready;
}
}
}
async fn dht_publish(
dht: &AsyncDht,
my_id: &PublicKey,
signing_key: &SigningKey,
salt: &[u8; 20],
pub_key_bytes: &[u8; 32],
neighbors: &Arc<Mutex<HashSet<PublicKey>>>,
state: &Arc<Mutex<SwarmState>>,
) {
let has_neighbors = neighbors.lock().is_ok_and(|n| !n.is_empty());
let mut seen = HashSet::new();
let mut peer_list: Vec<[u8; 32]> = Vec::with_capacity(MAX_DHT_PEERS);
// Always include ourselves.
peer_list.push(*my_id.as_bytes());
seen.insert(*my_id.as_bytes());
if has_neighbors {
// Connected: publish authoritative data — self + live neighbors only.
// This purges dead/stale peers from the DHT. We know who's alive
// because they're our active gossip neighbors.
if let Ok(n) = neighbors.lock() {
for id in n.iter() {
if seen.insert(*id.as_bytes()) && peer_list.len() < MAX_DHT_PEERS {
peer_list.push(*id.as_bytes());
}
}
}
} else {
// Alone: merge-publish — read existing DHT data first, add ourselves,
// preserve other peers. We don't know who's alive so we can't purge.
if let Some(record) = dht.get_mutable_most_recent(pub_key_bytes, Some(salt)).await {
for existing in decode_peer_list(record.value()) {
if seen.insert(*existing.as_bytes()) && peer_list.len() < MAX_DHT_PEERS {
peer_list.push(*existing.as_bytes());
}
}
}
}
let value = encode_peer_list(&peer_list);
let seq = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs().cast_signed())
.unwrap_or(0);
let item = MutableItem::new(signing_key.clone(), &value, seq, Some(salt.as_ref()));
match dht.put_mutable(item, None).await {
Ok(_) => {
debug!("published to DHT (seq={seq}, peers={})", peer_list.len());
if let Ok(mut s) = state.lock() {
s.dht_status = DhtStatus::Ready;
s.last_dht_publish = Some(Instant::now());
}
}
Err(e) => {
warn!("DHT publish failed: {e:#}");
if let Ok(mut s) = state.lock() {
s.dht_status = DhtStatus::PublishFailed;
}
}
}
}
// ── DHT helpers ─────────────────────────────────────────────────
/// Query the DHT for existing peers (blocking bootstrap, called once on startup).
async fn dht_get_peers(keys: &PassphraseKeys) -> Vec<PublicKey> {
let dht = match Dht::client() {
Ok(d) => d.as_async(),
Err(e) => {
warn!("failed to create DHT client for bootstrap: {e:#}");
return vec![];
}
};
let pub_key_bytes = keys.dht_signing_key.verifying_key().to_bytes();
dht.get_mutable_most_recent(&pub_key_bytes, Some(&keys.dht_salt)).await.map_or_else(
|| {
debug!("DHT bootstrap: no existing record");
vec![]
},
|record| {
let peers = decode_peer_list(record.value());
debug!("DHT bootstrap: found {} peer(s)", peers.len());
peers
},
)
}
/// Encode a list of 32-byte peer IDs into bytes for DHT storage.
/// Format: `[count: u16 LE] [id_0: 32B] [id_1: 32B] ...`
#[must_use]
pub fn encode_peer_list(peers: &[[u8; 32]]) -> Vec<u8> {
#[allow(clippy::cast_possible_truncation)] // MAX_DHT_PEERS (27) fits in u16
let count = peers.len().min(MAX_DHT_PEERS) as u16;
let mut buf = Vec::with_capacity(2 + (count as usize) * 32);
buf.extend_from_slice(&count.to_le_bytes());
for id in peers.iter().take(MAX_DHT_PEERS) {
buf.extend_from_slice(id);
}
buf
}
/// Decode a peer list from DHT record value.
#[must_use]
pub fn decode_peer_list(data: &[u8]) -> Vec<PublicKey> {
if data.len() < 2 {
return vec![];
}
let count = u16::from_le_bytes([data[0], data[1]]) as usize;
let mut peers = Vec::with_capacity(count);
let mut offset = 2;
for _ in 0..count {
if offset + 32 > data.len() {
break;
}
let bytes: [u8; 32] = match data[offset..offset + 32].try_into() {
Ok(b) => b,
Err(_) => break,
};
if let Ok(pk) = PublicKey::from_bytes(&bytes) {
peers.push(pk);
}
offset += 32;
}
peers
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn passphrase_keys_deterministic() {
let k1 = PassphraseKeys::derive("test-passphrase");
let k2 = PassphraseKeys::derive("test-passphrase");
assert_eq!(k1.topic_id, k2.topic_id);
assert_eq!(k1.dht_signing_key.to_bytes(), k2.dht_signing_key.to_bytes());
assert_eq!(k1.dht_salt, k2.dht_salt);
}
#[test]
fn passphrase_keys_different() {
let k1 = PassphraseKeys::derive("alpha");
let k2 = PassphraseKeys::derive("beta");
assert_ne!(k1.topic_id, k2.topic_id);
assert_ne!(k1.dht_signing_key.to_bytes(), k2.dht_signing_key.to_bytes());
}
#[test]
fn peer_list_roundtrip() {
// Generate valid ed25519 public keys.
let keys: Vec<_> = (0..3)
.map(|_| {
let sk = iroh::SecretKey::generate(&mut rand::rng());
*sk.public().as_bytes()
})
.collect();
let encoded = encode_peer_list(&keys);
let decoded = decode_peer_list(&encoded);
assert_eq!(decoded.len(), 3);
for (i, key) in keys.iter().enumerate() {
assert_eq!(*decoded[i].as_bytes(), *key);
}
}
#[test]
fn peer_list_empty() {
let encoded = encode_peer_list(&[]);
let decoded = decode_peer_list(&encoded);
assert!(decoded.is_empty());
}
#[test]
fn peer_list_truncated_data() {
// Only 1 byte — too short for count.
assert!(decode_peer_list(&[0]).is_empty());
// Count says 2 but only 1 peer present.
let mut data = vec![2, 0]; // count = 2
data.extend_from_slice(&[0xAA; 32]); // only 1 peer
let decoded = decode_peer_list(&data);
assert_eq!(decoded.len(), 1);
}
#[test]
fn dht_status_display() {
assert_eq!(format!("{}", DhtStatus::Bootstrapping), "Bootstrapping");
assert_eq!(format!("{}", DhtStatus::Ready), "Ready");
assert_eq!(format!("{}", DhtStatus::PublishFailed), "Failed");
}
#[test]
fn decode_peer_list_exact_boundary() {
// Build data with count=2 and exactly 2 + 32 + 32 = 66 bytes.
// Both peers should decode successfully.
let sk0 = iroh::SecretKey::generate(&mut rand::rng());
let sk1 = iroh::SecretKey::generate(&mut rand::rng());
let id0 = *sk0.public().as_bytes();
let id1 = *sk1.public().as_bytes();
let mut data = vec![2, 0]; // count = 2 (u16 LE)
data.extend_from_slice(&id0);
data.extend_from_slice(&id1);
assert_eq!(data.len(), 66);
let decoded = decode_peer_list(&data);
assert_eq!(decoded.len(), 2);
assert_eq!(*decoded[0].as_bytes(), id0);
assert_eq!(*decoded[1].as_bytes(), id1);
}
#[test]
fn decode_peer_list_one_byte_short() {
// Build data with count=2 but only 2 + 32 + 31 = 65 bytes.
// First peer decodes, second is truncated and skipped.
let sk0 = iroh::SecretKey::generate(&mut rand::rng());
let sk1 = iroh::SecretKey::generate(&mut rand::rng());
let id0 = *sk0.public().as_bytes();
let id1 = *sk1.public().as_bytes();
let mut data = vec![2, 0]; // count = 2 (u16 LE)
data.extend_from_slice(&id0);
data.extend_from_slice(&id1[..31]); // only 31 bytes of second peer
assert_eq!(data.len(), 65);
let decoded = decode_peer_list(&data);
assert_eq!(decoded.len(), 1);
assert_eq!(*decoded[0].as_bytes(), id0);
}
#[test]
fn decode_peer_list_count_header_only() {
// Data with exactly 2 bytes (count=0). Must not be rejected by the length check.
// Kills mutant: `data.len() < 2` → `data.len() <= 2`.
let data = [0u8, 0]; // count = 0
let decoded = decode_peer_list(&data);
assert!(decoded.is_empty());
// Data with 2 bytes but count=1 and no peer data — decodes 0 (truncated).
let data = [1u8, 0]; // count = 1 but no peer bytes
let decoded = decode_peer_list(&data);
assert!(decoded.is_empty());
}
mod proptests {
use super::*;
use proptest::prelude::*;
proptest! {
#[test]
fn peer_list_roundtrip_prop(count in 0usize..=30) {
let keys: Vec<[u8; 32]> = (0..count)
.map(|_| *iroh::SecretKey::generate(&mut rand::rng()).public().as_bytes())
.collect();
let encoded = encode_peer_list(&keys);
let decoded = decode_peer_list(&encoded);
let expected = count.min(MAX_DHT_PEERS);
prop_assert_eq!(decoded.len(), expected);
}
}
}
}
+16 -115
View File
@@ -1,7 +1,7 @@
use donglora_bridge::{config, gossip, radio, rate_limit, router, setup, tui};
use std::sync::Arc;
use std::time::{Duration, Instant};
use std::time::Instant;
use anyhow::{Context, Result};
use clap::{Parser, Subcommand};
@@ -10,6 +10,7 @@ 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;
@@ -65,36 +66,33 @@ async fn main() -> Result<()> {
setup_logging(cli.log_only, &cfg);
// Ephemeral identity — fresh keypair each launch.
let secret_key = SecretKey::generate();
let secret_key = SecretKey::generate(&mut rand::rng());
let node_id = secret_key.public();
info!("donglora-bridge starting, ephemeral id: {}", node_id.fmt_short());
// Parse radio config.
let radio_config = cfg.radio.to_lora_config()?;
let retry_policy = cfg.tx.to_retry_policy()?;
let radio_config = cfg.radio.to_radio_config()?;
// Derive passphrase keys.
let keys = PassphraseKeys::derive(&cfg.bridge.passphrase);
info!("topic: {}", hex::encode(&keys.topic_id.as_bytes()[..8]));
// Create rate limiter.
let 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,
cr_denom,
radio_config.cr,
radio_config.preamble_len,
cfg.bridge.rate_limit_pps,
);
info!("rate limiter: {:.1} pps", rate_limiter.rate_pps());
// Create gossip network + DHT rendezvous.
let (swarm, gossip_event_rx, gossip_frame_tx) = gossip::Gossip::new(secret_key, &cfg.bridge.passphrase).await?;
// Create gossip network.
let (swarm, gossip_event_rx, gossip_frame_tx) = gossip::Gossip::new(secret_key, &keys).await?;
let swarm = Arc::new(swarm);
// Start radio.
let (radio_event_rx, radio_tx, _radio_handle) = radio::spawn(radio_config, cfg.radio.port.clone(), retry_policy)?;
let (radio_event_rx, radio_tx) = radio::spawn(radio_config, cfg.radio.port.clone())?;
// Stats + log channel + radio config watch.
let stats = Arc::new(Stats::default());
@@ -162,29 +160,17 @@ 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);
// Restore the terminal first so the user's shell reappears the moment
// they hit `q`. iroh-gossip-rendezvous needs no synchronous DHT work
// on shutdown — entries age out — so the only remaining wait is
// `endpoint.close()` flushing in-flight connections. Bound it.
cancel.cancel();
swarm.shutdown().await;
info!("donglora-bridge stopped");
if let Ok(mut term) = terminal {
tui::restore_terminal(&mut term);
}
cancel.cancel();
if tokio::time::timeout(Duration::from_secs(2), swarm.shutdown()).await.is_err() {
warn!("graceful shutdown timed out — exiting anyway");
}
info!("donglora-bridge stopped");
return Ok(());
}
@@ -196,91 +182,6 @@ 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()
+2 -2
View File
@@ -144,7 +144,7 @@ mod tests {
#[test]
fn gossip_frame_roundtrip() {
let key = iroh::SecretKey::generate();
let key = iroh::SecretKey::generate(&mut rand::rng());
let frame = GossipFrame::new(&key.public(), -85, 8, vec![1, 2, 3, 4, 5]);
let encoded = frame.encode();
let decoded = GossipFrame::decode(&encoded).expect("decode should succeed");
@@ -235,7 +235,7 @@ mod tests {
snr in -32i16..=32i16,
payload in proptest::collection::vec(any::<u8>(), 0..256),
) {
let key = iroh::SecretKey::generate();
let key = iroh::SecretKey::generate(&mut rand::rng());
let frame = GossipFrame::new(&key.public(), rssi, snr, payload.clone());
let encoded = frame.encode();
let decoded = GossipFrame::decode(&encoded).unwrap();
+288 -298
View File
@@ -1,24 +1,13 @@
//! Async radio task — drives the `donglora-client` [`Dongle`] session.
//! `DongLoRa` hardware communication on a blocking thread.
//!
//! 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.
//! Handles connection, config negotiation, RX/TX, liveness pings, and
//! exponential backoff reconnection.
use std::time::Duration;
use anyhow::{Context, Result};
use donglora_client::{
ConnectOptions, Dongle, LoRaBandwidth, LoRaConfig, Modulation, RetryPolicy, RxPayload, connect_mux_auto_with,
connect_with,
};
use donglora_client::{RadioConfig, Response, Transport};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tracing::{debug, error, info, warn};
use crate::packet::{RadioPacket, snr_grade};
@@ -32,295 +21,242 @@ pub enum ConfigSource {
Mux,
}
/// Messages from the radio task to the router.
#[derive(Debug)]
/// Messages from the radio thread to the router.
pub enum RadioEvent {
/// A valid packet received from the radio (already SNR-filtered).
Packet(RadioPacket),
/// Radio connected (or reconnected) with the active config and device info.
Connected(LoRaConfig, ConfigSource, String),
Connected(RadioConfig, 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 },
}
/// 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.
/// Spawn the blocking radio thread.
///
/// Returns a receiver for radio events and a sender for [`TxRequest`]s.
pub fn spawn(
config: LoRaConfig,
port: Option<String>,
retry_policy: RetryPolicy,
) -> Result<(mpsc::Receiver<RadioEvent>, mpsc::Sender<TxRequest>, JoinHandle<()>)> {
/// Returns a receiver for radio events and a sender for TX payloads.
pub fn spawn(config: RadioConfig, port: Option<String>) -> Result<(mpsc::Receiver<RadioEvent>, mpsc::Sender<Vec<u8>>)> {
let (event_tx, event_rx) = mpsc::channel::<RadioEvent>(256);
let (tx_send, tx_recv) = mpsc::channel::<TxRequest>(64);
let handle = tokio::spawn(radio_loop(config, port, retry_policy, event_tx, tx_recv));
Ok((event_rx, tx_send, handle))
let (tx_send, tx_recv) = mpsc::channel::<Vec<u8>>(64);
std::thread::Builder::new()
.name("radio".into())
.spawn(move || radio_loop(config, port, event_tx, tx_recv))
.context("failed to spawn radio thread")?;
Ok((event_rx, tx_send))
}
async fn radio_loop(
config: LoRaConfig,
#[allow(clippy::needless_pass_by_value)] // values moved across thread boundary in spawn()
fn radio_loop(
config: RadioConfig,
port: Option<String>,
retry_policy: RetryPolicy,
event_tx: mpsc::Sender<RadioEvent>,
mut tx_recv: mpsc::Receiver<TxRequest>,
mut tx_recv: mpsc::Receiver<Vec<u8>>,
) {
let initial_backoff = Duration::from_millis(250);
let max_backoff = Duration::from_secs(5);
let mut backoff = initial_backoff;
let mut stick_to_mux = false;
loop {
let started = std::time::Instant::now();
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");
match connect_and_run(&config, port.as_deref(), &event_tx, &mut tx_recv) {
Ok(()) => {
info!("radio thread exiting");
return;
}
Err((was_mux, e)) => {
if was_mux {
stick_to_mux = true;
}
Err(e) => {
error!("radio error: {e:#}");
let _ = event_tx.send(RadioEvent::Disconnected).await;
let _ = event_tx.blocking_send(RadioEvent::Disconnected);
// If we were connected for a while, reset backoff so
// the next reconnect attempt is fast.
if started.elapsed() > Duration::from_secs(5) {
backoff = initial_backoff;
}
info!("reconnecting in {backoff:?}");
tokio::time::sleep(backoff).await;
std::thread::sleep(backoff);
backoff = (backoff * 2).min(max_backoff);
}
}
}
}
async fn connect_and_run(
config: &LoRaConfig,
fn connect_and_run(
config: &RadioConfig,
port: Option<&str>,
stick_to_mux: bool,
retry_policy: &RetryPolicy,
event_tx: &mpsc::Sender<RadioEvent>,
tx_recv: &mut mpsc::Receiver<TxRequest>,
) -> std::result::Result<bool, (bool, anyhow::Error)> {
info!("connecting to dongle...");
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 {
connect_with(opts).await.map_err(|e| (false, anyhow::Error::from(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>,
tx_recv: &mut mpsc::Receiver<Vec<u8>>,
) -> 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));
// 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);
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(());
info!("connecting to dongle...");
let (mut client, device) = if let Some(port) = port {
let c = donglora_client::connect(Some(port), timeout)?;
(c, shorten_path(port))
} else if let Ok(c) = donglora_client::connect_mux_auto(timeout) {
(c, "mux".to_string())
} else {
let port_path = donglora_client::find_port().ok_or_else(|| anyhow::anyhow!("no mux or USB dongle found"))?;
info!("no mux found, connecting to serial port {port_path}");
let c = donglora_client::connect(Some(&port_path), timeout)?;
(c, shorten_path(&port_path))
};
info!("[radio] transport connected: {device}");
info!("[radio] sending ping...");
match client.ping() {
Ok(()) => info!("[radio] ping OK"),
Err(e) => {
info!("[radio] ping FAILED: {e:#}");
return Err(e);
}
}
info!("[radio] negotiating config...");
let (active_config, config_source) = match negotiate_config(&mut client, config, &device) {
Ok(result) => {
info!("[radio] config negotiated: source={:?}, config={}", result.1, format_radio_config(&result.0));
result
}
Err(e) => {
info!("[radio] config negotiation FAILED: {e:#}");
return Err(e);
}
};
info!("[radio] sending start_rx...");
match client.start_rx() {
Ok(()) => info!("[radio] start_rx OK"),
Err(e) => {
info!("[radio] start_rx FAILED: {e:#}");
return Err(e);
}
}
let _ = event_tx.blocking_send(RadioEvent::Connected(active_config, config_source, device));
// Set a short timeout so we can interleave RX and TX.
client.transport_mut().set_timeout(Duration::from_millis(100))?;
info!("[radio] entering main loop");
// TX requests from the router and RX packets from the radio both drive
// the loop; use tokio::select! to multiplex them.
// Liveness: ping the mux periodically when idle to detect a dead connection.
// Without this, a killed mux causes recv() to silently return Ok(None) forever.
let liveness_interval = Duration::from_secs(2);
let mut last_activity = std::time::Instant::now();
loop {
tokio::select! {
tx = tx_recv.recv() => {
let Some(req) = tx else { return Ok(()); };
handle_tx(&dongle, req, retry_policy, event_tx).await;
// 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);
}
rx = dongle.next_rx() => {
match rx {
Some(pkt) => {
handle_rx(pkt, &active_config, event_tx).await;
}
None => {
// Dongle's RX stream ended — session closed.
last_activity = std::time::Instant::now();
}
// Check for RX packets.
match client.recv() {
Ok(Some(Response::RxPacket { rssi, snr, payload })) => {
last_activity = std::time::Instant::now();
let grade = snr_grade(snr, active_config.sf);
if grade.should_forward() {
debug!("RX {} bytes rssi={rssi} snr={snr} grade={grade}", payload.len());
let pkt = RadioPacket { rssi, snr, payload };
if event_tx.blocking_send(RadioEvent::Packet(pkt)).is_err() {
return Ok(());
}
} else {
debug!("RX drop {} bytes rssi={rssi} snr={snr} grade={grade}", payload.len());
}
}
Ok(Some(_)) => {
last_activity = std::time::Instant::now();
}
Ok(None) => {}
Err(e) if is_timeout_error(&e) => {}
Err(e) => return Err(e),
}
// Liveness check: if no real activity for a while, ping the mux.
if last_activity.elapsed() >= liveness_interval {
// Temporarily extend the timeout for the ping command.
let _ = client.transport_mut().set_timeout(Duration::from_secs(2));
match client.ping() {
Ok(()) => {
last_activity = std::time::Instant::now();
}
Err(e) => {
warn!("liveness ping failed: {e:#}");
return Err(e);
}
}
// Restore the short timeout.
let _ = client.transport_mut().set_timeout(Duration::from_millis(100));
}
if event_tx.is_closed() {
return Ok(());
}
// 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;
}
}
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;
/// Negotiate radio config with the dongle/mux.
///
/// Strategy:
/// 1. `GetConfig` first to read what's active.
/// 2. If it matches our desired config, use it — no `SetConfig` needed.
/// 3. If it differs and we're on the mux, accept the mux's config.
/// 4. If it differs and we're on direct serial, `SetConfig` to push ours.
/// 5. If `GetConfig` fails, try `SetConfig` with our desired config.
fn negotiate_config<T: Transport>(
client: &mut donglora_client::Client<T>,
desired: &RadioConfig,
device: &str,
) -> Result<(RadioConfig, ConfigSource)> {
info!("[negotiate] get_config...");
match client.get_config() {
Ok(cfg) => {
if configs_match(&cfg, desired) {
info!("[negotiate] get_config OK (matches desired): {}", format_radio_config(&cfg));
Ok((cfg, ConfigSource::Ours))
} else if device == "mux" {
info!("[negotiate] get_config OK (mux config differs, accepting): {}", format_radio_config(&cfg));
Ok((cfg, ConfigSource::Mux))
} else {
info!("[negotiate] get_config OK (serial config differs, setting ours)");
client.set_config(*desired)?;
info!("[negotiate] set_config OK");
Ok((*desired, ConfigSource::Ours))
}
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) => {
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;
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:#})");
}
}
}
}
}
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;
/// Shorten a device path to just the filename for display.
fn shorten_path(path: &str) -> String {
std::path::Path::new(path).file_name().map_or_else(|| path.to_string(), |f| f.to_string_lossy().into_owned())
}
// ── Display helpers ───────────────────────────────────────────────
fn lora_configs_match(a: &LoRaConfig, b: &LoRaConfig) -> bool {
fn configs_match(a: &RadioConfig, b: &RadioConfig) -> bool {
a.freq_hz == b.freq_hz
&& a.bw == b.bw
&& a.sf == b.sf
@@ -328,47 +264,46 @@ fn lora_configs_match(a: &LoRaConfig, b: &LoRaConfig) -> bool {
&& a.sync_word == b.sync_word
&& a.tx_power_dbm == b.tx_power_dbm
&& a.preamble_len == b.preamble_len
&& a.header_mode == b.header_mode
&& a.payload_crc == b.payload_crc
&& a.iq_invert == b.iq_invert
&& a.cad == b.cad
}
fn is_timeout_error(e: &anyhow::Error) -> bool {
e.downcast_ref::<std::io::Error>()
.is_some_and(|io_err| matches!(io_err.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut))
}
/// Format a bandwidth value for display.
#[must_use]
pub const fn format_bandwidth(bw: LoRaBandwidth) -> &'static str {
pub const fn format_bandwidth(bw: donglora_client::Bandwidth) -> &'static str {
match bw {
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",
donglora_client::Bandwidth::Khz7 => "7.8 kHz",
donglora_client::Bandwidth::Khz10 => "10.4 kHz",
donglora_client::Bandwidth::Khz15 => "15.6 kHz",
donglora_client::Bandwidth::Khz20 => "20.8 kHz",
donglora_client::Bandwidth::Khz31 => "31.25 kHz",
donglora_client::Bandwidth::Khz41 => "41.7 kHz",
donglora_client::Bandwidth::Khz62 => "62.5 kHz",
donglora_client::Bandwidth::Khz125 => "125 kHz",
donglora_client::Bandwidth::Khz250 => "250 kHz",
donglora_client::Bandwidth::Khz500 => "500 kHz",
}
}
/// Format a `LoRaConfig` for display.
/// Format a `RadioConfig` for display.
#[must_use]
pub fn format_lora_config(config: &LoRaConfig) -> String {
pub fn format_radio_config(config: &RadioConfig) -> String {
let freq_mhz = f64::from(config.freq_hz) / 1_000_000.0;
let bw = format_bandwidth(config.bw);
let 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 power = if config.tx_power_dbm == donglora_client::TX_POWER_MAX {
"max".to_string()
} else {
format!("{} dBm", config.tx_power_dbm)
};
let preamble = config.preamble_len;
let preamble = if config.preamble_len == 0 { 16 } else { config.preamble_len };
let cad = if config.cad != 0 { "on" } else { "off" };
format!(
"{freq_mhz:.3}MHz SF{} BW{bw} CR4/{cr_denom} SW0x{:04X} TX:{} dBm Pre:{preamble}",
config.sf, config.sync_word, config.tx_power_dbm,
"{freq_mhz:.3}MHz SF{} BW{bw} CR4/{} SW0x{:04X} TX:{power} Pre:{preamble} CAD:{cad}",
config.sf, config.cr, config.sync_word
)
}
@@ -376,79 +311,134 @@ pub fn format_lora_config(config: &LoRaConfig) -> String {
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
use donglora_client::{LoRaCodingRate, LoRaHeaderMode};
fn default_config() -> LoRaConfig {
LoRaConfig {
fn default_config() -> RadioConfig {
RadioConfig {
freq_hz: 910_525_000,
bw: LoRaBandwidth::Khz62,
bw: donglora_client::Bandwidth::Khz62,
sf: 7,
cr: LoRaCodingRate::Cr4_5,
sync_word: 0x1424,
tx_power_dbm: 20,
cr: 5,
sync_word: 0x3444,
tx_power_dbm: 22,
preamble_len: 16,
header_mode: LoRaHeaderMode::Explicit,
payload_crc: true,
iq_invert: false,
cad: 1,
}
}
#[test]
fn lora_configs_match_identical() {
assert!(lora_configs_match(&default_config(), &default_config()));
fn configs_match_identical() {
let a = default_config();
let b = default_config();
assert!(configs_match(&a, &b));
}
#[test]
fn lora_configs_match_differs_each_field() {
fn 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));
assert!(!configs_match(&base, &c));
let mut c = base;
c.sf = 12;
assert!(!lora_configs_match(&base, &c));
assert!(!configs_match(&base, &c));
let mut c = base;
c.cr = LoRaCodingRate::Cr4_8;
assert!(!lora_configs_match(&base, &c));
c.cr = 8;
assert!(!configs_match(&base, &c));
let mut c = base;
c.sync_word = 0x1234;
assert!(!lora_configs_match(&base, &c));
assert!(!configs_match(&base, &c));
let mut c = base;
c.tx_power_dbm = 10;
assert!(!lora_configs_match(&base, &c));
assert!(!configs_match(&base, &c));
let mut c = base;
c.preamble_len = 8;
assert!(!lora_configs_match(&base, &c));
assert!(!configs_match(&base, &c));
let mut c = base;
c.cad = 0;
assert!(!configs_match(&base, &c));
}
#[test]
fn is_timeout_would_block() {
let e = anyhow::Error::from(std::io::Error::new(std::io::ErrorKind::WouldBlock, "test"));
assert!(is_timeout_error(&e));
}
#[test]
fn is_timeout_timed_out() {
let e = anyhow::Error::from(std::io::Error::new(std::io::ErrorKind::TimedOut, "test"));
assert!(is_timeout_error(&e));
}
#[test]
fn is_timeout_false_for_other() {
let e = anyhow::Error::from(std::io::Error::new(std::io::ErrorKind::ConnectionReset, "test"));
assert!(!is_timeout_error(&e));
}
#[test]
fn shorten_path_extracts_filename() {
assert_eq!(shorten_path("/dev/ttyACM0"), "ttyACM0");
assert_eq!(shorten_path("ttyACM0"), "ttyACM0");
}
#[test]
fn format_bandwidth_all_variants() {
assert_eq!(format_bandwidth(LoRaBandwidth::Khz7), "7.8 kHz");
assert_eq!(format_bandwidth(LoRaBandwidth::Khz125), "125 kHz");
assert_eq!(format_bandwidth(LoRaBandwidth::Khz500), "500 kHz");
assert_eq!(format_bandwidth(donglora_client::Bandwidth::Khz7), "7.8 kHz");
assert_eq!(format_bandwidth(donglora_client::Bandwidth::Khz125), "125 kHz");
assert_eq!(format_bandwidth(donglora_client::Bandwidth::Khz500), "500 kHz");
}
#[test]
fn format_lora_config_typical() {
fn format_radio_config_typical() {
let cfg = default_config();
let s = format_lora_config(&cfg);
let s = format_radio_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"));
assert!(s.contains("0x3444"));
assert!(s.contains("CAD:on"), "cad=1 should show CAD:on: {s}");
}
#[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");
fn format_radio_config_cad_off() {
let mut cfg = default_config();
cfg.cad = 0;
let s = format_radio_config(&cfg);
assert!(s.contains("CAD:off"), "cad=0 should show CAD:off: {s}");
}
#[test]
fn format_radio_config_max_power() {
let mut cfg = default_config();
cfg.tx_power_dbm = donglora_client::TX_POWER_MAX;
let s = format_radio_config(&cfg);
assert!(s.contains("TX:max"));
assert!(!s.contains("dBm"), "max power should not contain 'dBm': {s}");
}
#[test]
fn format_radio_config_non_max_power() {
let mut cfg = default_config();
cfg.tx_power_dbm = 22;
assert_ne!(cfg.tx_power_dbm, donglora_client::TX_POWER_MAX);
let s = format_radio_config(&cfg);
assert!(s.contains("22 dBm"), "non-max power should contain '22 dBm': {s}");
assert!(!s.contains("max"), "non-max power should not contain 'max': {s}");
}
#[test]
fn format_radio_config_zero_preamble() {
let mut cfg = default_config();
cfg.preamble_len = 0;
let s = format_radio_config(&cfg);
assert!(s.contains("Pre:16"));
}
}
+43 -50
View File
@@ -5,7 +5,7 @@
use std::time::{Duration, Instant};
use donglora_client::LoRaBandwidth;
use donglora_client::Bandwidth;
/// Token-bucket rate limiter for radio transmissions.
///
@@ -35,7 +35,7 @@ impl RateLimiter {
///
/// If `override_pps` is `Some`, use that rate directly instead of calculating.
#[must_use]
pub fn from_radio_config(sf: u8, bw: LoRaBandwidth, cr: u8, preamble: u16, override_pps: Option<f64>) -> Self {
pub fn from_radio_config(sf: u8, bw: Bandwidth, cr: u8, preamble: u16, override_pps: Option<f64>) -> Self {
let refill_rate = override_pps.map_or_else(
|| {
let air_time = lora_air_time(REFERENCE_PAYLOAD_BYTES, sf, bw, cr, preamble);
@@ -81,7 +81,7 @@ impl RateLimiter {
/// - `cr`: coding rate denominator (5–8, meaning 4/5 to 4/8)
/// - `preamble`: preamble length in symbols (0 = firmware default of 16)
#[must_use]
pub fn lora_air_time(payload_bytes: usize, sf: u8, bw: LoRaBandwidth, cr: u8, preamble: u16) -> Duration {
pub fn lora_air_time(payload_bytes: usize, sf: u8, bw: Bandwidth, cr: u8, preamble: u16) -> Duration {
let sf_f = f64::from(sf);
let bw_hz = bandwidth_hz(bw);
let preamble_symbols = if preamble == 0 { 16.0 } else { f64::from(preamble) };
@@ -108,25 +108,18 @@ pub fn lora_air_time(payload_bytes: usize, sf: u8, bw: LoRaBandwidth, cr: u8, pr
Duration::from_secs_f64(total)
}
const fn bandwidth_hz(bw: LoRaBandwidth) -> f64 {
const fn bandwidth_hz(bw: Bandwidth) -> f64 {
match bw {
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,
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,
}
}
@@ -136,7 +129,7 @@ mod tests {
#[test]
fn air_time_sf7_bw125() {
let t = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 16);
let t = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16);
// SF7/125kHz/CR4/5/16-preamble/50 bytes: ~100ms range
assert!(t.as_millis() > 50, "air time too short: {t:?}");
assert!(t.as_millis() < 300, "air time too long: {t:?}");
@@ -144,7 +137,7 @@ mod tests {
#[test]
fn air_time_sf12_bw125() {
let t = lora_air_time(50, 12, LoRaBandwidth::Khz125, 5, 16);
let t = lora_air_time(50, 12, Bandwidth::Khz125, 5, 16);
// SF12/125kHz: should be multiple seconds
assert!(t.as_secs() >= 1, "air time too short: {t:?}");
assert!(t.as_secs() < 10, "air time too long: {t:?}");
@@ -152,7 +145,7 @@ mod tests {
#[test]
fn rate_limiter_allows_burst() {
let mut rl = RateLimiter::from_radio_config(7, LoRaBandwidth::Khz125, 5, 16, None);
let mut rl = RateLimiter::from_radio_config(7, Bandwidth::Khz125, 5, 16, None);
// Should allow burst of 3
assert!(rl.try_acquire());
assert!(rl.try_acquire());
@@ -163,14 +156,14 @@ mod tests {
#[test]
fn rate_limiter_override() {
let rl = RateLimiter::from_radio_config(7, LoRaBandwidth::Khz125, 5, 16, Some(42.0));
let rl = RateLimiter::from_radio_config(7, Bandwidth::Khz125, 5, 16, Some(42.0));
assert!((rl.rate_pps() - 42.0).abs() < f64::EPSILON);
}
#[test]
fn air_time_zero_preamble_uses_default() {
let t_default = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 0);
let t_explicit = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 16);
let t_default = lora_air_time(50, 7, Bandwidth::Khz125, 5, 0);
let t_explicit = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16);
assert_eq!(t_default, t_explicit);
}
@@ -180,7 +173,7 @@ mod tests {
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 rl = RateLimiter::from_radio_config(7, Bandwidth::Khz125, 5, 16, None);
let rate = rl.rate_pps();
// Must be between 0.1 and 100 (rules out fallback of 10.0 indirectly,
// and catches if secs<=0 path is taken or division is wrong).
@@ -196,7 +189,7 @@ mod tests {
#[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 rl = RateLimiter::from_radio_config(12, Bandwidth::Khz125, 5, 16, None);
let rate = rl.rate_pps();
assert!(rate > 0.1, "rate too low: {rate}");
assert!(rate < 0.5, "rate too high: {rate}");
@@ -209,7 +202,7 @@ mod tests {
#[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));
let mut rl = RateLimiter::from_radio_config(7, Bandwidth::Khz125, 5, 16, Some(10.0));
// Exhaust all burst tokens.
assert!(rl.try_acquire());
assert!(rl.try_acquire());
@@ -229,7 +222,7 @@ mod tests {
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 t = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16);
let ms = t.as_secs_f64() * 1000.0;
assert!((ms - 105.728).abs() < 0.01, "SF7/BW125 air time should be ~105.728ms, got {ms:.3}ms");
}
@@ -238,7 +231,7 @@ mod tests {
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 t = lora_air_time(50, 12, Bandwidth::Khz125, 5, 16);
let ms = t.as_secs_f64() * 1000.0;
assert!((ms - 2564.096).abs() < 0.01, "SF12/BW125 air time should be ~2564.096ms, got {ms:.3}ms");
}
@@ -247,8 +240,8 @@ mod tests {
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 t11 = lora_air_time(50, 11, Bandwidth::Khz125, 5, 16);
let t10 = lora_air_time(50, 10, Bandwidth::Khz125, 5, 16);
let ms11 = t11.as_secs_f64() * 1000.0;
let ms10 = t10.as_secs_f64() * 1000.0;
@@ -266,12 +259,12 @@ mod tests {
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 t11_250 = lora_air_time(50, 11, Bandwidth::Khz250, 5, 16);
let ms = t11_250.as_secs_f64() * 1000.0;
assert!((ms - 641.024).abs() < 0.01, "SF11/BW250 should be ~641.024ms (no LDRO), got {ms:.3}ms");
// Compare with SF11/BW125 which HAS LDRO.
let t11_125 = lora_air_time(50, 11, LoRaBandwidth::Khz125, 5, 16);
let t11_125 = lora_air_time(50, 11, Bandwidth::Khz125, 5, 16);
// Even accounting for halved BW doubling the base time, the LDRO effect
// on SF11/BW125 makes it much longer than simply 2x SF11/BW250.
assert!(t11_125 > t11_250 * 2, "SF11/BW125 (LDRO on) should be >2x SF11/BW250 (LDRO off)");
@@ -281,9 +274,9 @@ mod tests {
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 t8 = lora_air_time(50, 7, Bandwidth::Khz125, 5, 8);
let t16 = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16);
let t32 = lora_air_time(50, 7, Bandwidth::Khz125, 5, 32);
let diff_8_to_16 = t16.as_secs_f64() - t8.as_secs_f64();
let diff_16_to_32 = t32.as_secs_f64() - t16.as_secs_f64();
@@ -304,16 +297,16 @@ mod tests {
// 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,
Bandwidth::Khz7,
Bandwidth::Khz10,
Bandwidth::Khz15,
Bandwidth::Khz20,
Bandwidth::Khz31,
Bandwidth::Khz41,
Bandwidth::Khz62,
Bandwidth::Khz125,
Bandwidth::Khz250,
Bandwidth::Khz500,
];
for sf in 7..=12u8 {
for &bw in &bandwidths {
@@ -339,7 +332,7 @@ mod tests {
cr in 5u8..=8u8,
preamble in 0u16..=64u16,
) {
for bw in [LoRaBandwidth::Khz7, LoRaBandwidth::Khz62, LoRaBandwidth::Khz125, LoRaBandwidth::Khz500] {
for bw in [Bandwidth::Khz7, Bandwidth::Khz62, Bandwidth::Khz125, Bandwidth::Khz500] {
let t = lora_air_time(payload, sf, bw, cr, preamble);
prop_assert!(t > Duration::ZERO, "air time must be positive");
prop_assert!(t < Duration::from_secs(600), "air time should be < 600s");
+8 -131
View File
@@ -13,14 +13,14 @@ use tracing::{debug, warn};
use crate::gossip::GossipEvent;
use crate::packet::{GossipFrame, content_hash};
use crate::radio::{ConfigSource, RadioEvent, TxRequest, TxRetryReason};
use crate::radio::{ConfigSource, RadioEvent};
use crate::rate_limit::RateLimiter;
/// Active radio config info, sent to the TUI via watch channel.
#[derive(Debug, Clone)]
pub struct RadioConfigInfo {
pub active: donglora_client::LoRaConfig,
pub requested: donglora_client::LoRaConfig,
pub active: donglora_client::RadioConfig,
pub requested: donglora_client::RadioConfig,
pub source: ConfigSource,
pub device: String,
pub connected: bool,
@@ -36,29 +36,6 @@ 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.
@@ -106,11 +83,6 @@ 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 {
@@ -126,8 +98,6 @@ 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),
}
}
}
@@ -144,8 +114,6 @@ 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.
@@ -185,10 +153,10 @@ pub async fn run(
our_id: PublicKey,
dedup_window: Duration,
tx_queue_size: usize,
requested_radio_config: donglora_client::LoRaConfig,
requested_radio_config: donglora_client::RadioConfig,
mut rate_limiter: RateLimiter,
mut radio_rx: mpsc::Receiver<RadioEvent>,
radio_tx: mpsc::Sender<TxRequest>,
radio_tx: mpsc::Sender<Vec<u8>>,
mut gossip_rx: mpsc::Receiver<GossipEvent>,
gossip_tx: mpsc::Sender<GossipFrame>,
stats: &Stats,
@@ -196,7 +164,6 @@ 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);
@@ -228,7 +195,6 @@ pub async fn run(
snr: Some(pkt.snr),
rssi: Some(pkt.rssi),
action: PacketAction::DroppedDedup,
tx_retry: None,
}).await;
continue;
}
@@ -246,7 +212,6 @@ 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) => {
@@ -263,67 +228,6 @@ 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;
}
}
}
@@ -346,7 +250,6 @@ pub async fn run(
snr: Some(frame.snr),
rssi: Some(frame.rssi),
action: PacketAction::DroppedDedup,
tx_retry: None,
}).await;
continue;
}
@@ -363,7 +266,6 @@ pub async fn run(
snr: Some(frame.snr),
rssi: Some(frame.rssi),
action: PacketAction::DroppedRateLimit,
tx_retry: None,
}).await;
continue;
}
@@ -380,7 +282,6 @@ pub async fn run(
snr: Some(frame.snr),
rssi: Some(frame.rssi),
action: PacketAction::DroppedQueueFull,
tx_retry: None,
}).await;
continue;
}
@@ -394,7 +295,6 @@ pub async fn run(
snr: Some(frame.snr),
rssi: Some(frame.rssi),
action: PacketAction::Bridged,
tx_retry: None,
}).await;
}
GossipEvent::NeighborChanged(count) => {
@@ -405,39 +305,16 @@ pub async fn run(
}
}
// 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.
// Drain TX queue to radio.
while let Some(payload) = tx_queue.pop_front() {
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() {
stats.radio_tx.fetch_add(1, Ordering::Relaxed);
if radio_tx.send(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 {
+7 -4
View File
@@ -58,12 +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_default = format!("{}", radio.tx_power_dbm);
let tx_power_dbm: i8 = prompt_parse("TX Power (dBm)", &tx_power_default)?;
let tx_power = prompt_default("TX Power (dBm or 'max')", &radio.tx_power)?;
let pre_default = format!("{}", radio.preamble);
let preamble: u16 = prompt_parse("Preamble", &pre_default)?;
let cad_default = if radio.cad { "on" } else { "off" };
let cad_str = prompt_default("CAD", cad_default)?;
let cad = matches!(cad_str.to_lowercase().as_str(), "on" | "true" | "yes" | "1");
let port_default = radio.port.as_deref().unwrap_or("auto");
let port_str = prompt_default("Serial Port", port_default)?;
let port = if port_str == "auto" || port_str.is_empty() { None } else { Some(port_str) };
@@ -78,8 +81,9 @@ pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> {
spreading_factor,
coding_rate,
sync_word,
tx_power_dbm,
tx_power,
preamble,
cad,
},
bridge: BridgeSection {
passphrase,
@@ -88,7 +92,6 @@ 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)?;
+55 -69
View File
@@ -17,15 +17,13 @@ use ratatui::layout::{Alignment, Constraint, Direction, Layout, Rect};
use ratatui::style::{Color, Modifier, Style};
use ratatui::symbols;
use ratatui::text::{Line, Span};
use ratatui::widgets::{Block, Borders, Paragraph, Sparkline};
use ratatui::widgets::{Block, Borders, Clear, Paragraph, Sparkline};
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use crate::gossip::{DhtStatus, Gossip, RendezvousState};
use crate::gossip::Gossip;
use crate::radio::ConfigSource;
use crate::router::{
PacketAction, PacketDirection, PacketLogEntry, RadioConfigInfo, Stats, StatsSnapshot, TxRetryState,
};
use crate::router::{PacketAction, PacketDirection, PacketLogEntry, RadioConfigInfo, Stats, StatsSnapshot};
const TICK_RATE: Duration = Duration::from_millis(250);
const MAX_LOG_ENTRIES: usize = 200;
@@ -173,6 +171,7 @@ pub fn run(
&mut series,
);
terminal.draw(|frame| draw_shutdown_overlay(frame, frame.area()))?;
Ok(terminal)
}
@@ -209,7 +208,7 @@ fn tui_loop(
let snap = stats.snapshot();
let log_slice = log_entries.make_contiguous();
series.update(&snap, log_slice);
let swarm_state = gossip.state();
let swarm_state = gossip.swarm_state();
let config_info = config_watch.borrow().clone();
terminal.draw(|frame| {
draw_ui(frame, frame.area(), &config_info, &snap, &swarm_state, log_slice, start_time, series);
@@ -240,7 +239,7 @@ fn draw_ui(
area: Rect,
config_info: &RadioConfigInfo,
stats: &StatsSnapshot,
swarm_state: &RendezvousState,
swarm_state: &crate::gossip::SwarmState,
log_entries: &[PacketLogEntry],
start_time: Instant,
series: &TimeSeries,
@@ -315,20 +314,19 @@ fn status_pill(label: &str, ok: bool) -> Span<'static> {
// ── Network panel (was Swarm) ────────────────────────────────────
fn draw_network(frame: &mut ratatui::Frame, area: Rect, state: &RendezvousState) {
fn draw_network(frame: &mut ratatui::Frame, area: Rect, state: &crate::gossip::SwarmState) {
let block = panel_block(" Network ");
let dht_color = match state.dht_status {
DhtStatus::Ready => Color::Green,
DhtStatus::Bootstrapping => C_RATE_DROP,
DhtStatus::PublishFailing => C_QUEUE_DROP,
crate::gossip::DhtStatus::Ready => Color::Green,
crate::gossip::DhtStatus::Bootstrapping => C_RATE_DROP,
crate::gossip::DhtStatus::PublishFailed => C_QUEUE_DROP,
};
let neighbor_color = if state.neighbor_count > 0 { Color::Green } else { C_RATE_DROP };
let last_pub = state.last_publish.map_or_else(|| "-".into(), |t| format!("{}s ago", t.elapsed().as_secs()));
let topic_hash = hex::encode(&state.topic_id.as_bytes()[..8]);
let last_pub = state.last_dht_publish.map_or_else(|| "-".into(), |t| format!("{}s ago", t.elapsed().as_secs()));
let lines = vec![
kv_line("ID", &topic_hash, C_ACCENT),
kv_line("ID", &state.topic_hash, C_ACCENT),
kv_line("Peers", &format!("{}", state.neighbor_count), neighbor_color),
kv_line("DHT", &format!("{}", state.dht_status), dht_color),
kv_line("Published", &last_pub, C_LABEL),
@@ -368,16 +366,13 @@ fn draw_radio(frame: &mut ratatui::Frame, area: Rect, config_info: &RadioConfigI
let freq_mhz = f64::from(a.freq_hz) / 1_000_000.0;
let bw_str = crate::radio::format_bandwidth(a.bw);
let power_str = format!("{} dBm", a.tx_power_dbm);
let preamble = if a.preamble_len == 0 { 16 } else { a.preamble_len };
let cr_denom = |c: donglora_client::LoRaCodingRate| -> u8 {
match c {
donglora_client::LoRaCodingRate::Cr4_5 => 5,
donglora_client::LoRaCodingRate::Cr4_6 => 6,
donglora_client::LoRaCodingRate::Cr4_7 => 7,
donglora_client::LoRaCodingRate::Cr4_8 => 8,
}
let power_str = if a.tx_power_dbm == donglora_client::TX_POWER_MAX {
"max".to_string()
} else {
format!("{} dBm", a.tx_power_dbm)
};
let preamble = if a.preamble_len == 0 { 16 } else { a.preamble_len };
let cad = if a.cad != 0 { "on" } else { "off" };
// Dim values when disconnected; highlight mux mismatches when connected.
let val_color = if connected { C_ACCENT } else { C_LABEL };
@@ -399,10 +394,11 @@ fn draw_radio(frame: &mut ratatui::Frame, area: Rect, config_info: &RadioConfigI
cf("Freq", format!("{freq_mhz:.3} MHz"), a.freq_hz == r.freq_hz),
cf("BW", bw_str.to_string(), a.bw == r.bw),
cf("SF", format!("{}", a.sf), a.sf == r.sf),
cf("CR", format!("4/{}", cr_denom(a.cr)), a.cr == r.cr),
cf("CR", format!("4/{}", a.cr), a.cr == r.cr),
cf("Sync", format!("0x{:04X}", a.sync_word), a.sync_word == r.sync_word),
cf("TX", power_str, a.tx_power_dbm == r.tx_power_dbm),
cf("Preamble", format!("{preamble}"), a.preamble_len == r.preamble_len),
cf("CAD", cad.to_string(), a.cad == r.cad),
];
frame.render_widget(Paragraph::new(lines).block(block), area);
}
@@ -434,11 +430,6 @@ fn draw_stats(frame: &mut ratatui::Frame, area: Rect, stats: &StatsSnapshot, sta
s("Deduped", stats.dedup_hits, C_DEDUP),
s("Rate lim", stats.rate_limit_drops, C_RATE_DROP),
s("Q drop", stats.dropped_queue, C_QUEUE_DROP),
// CAD-busy retries tally (per spec §6.10). Yellow when accumulating,
// dim when quiet.
s("TX retry", stats.tx_retries, if stats.tx_retries > 0 { C_RATE_DROP } else { C_LABEL }),
// TX calls that exhausted the retry policy and gave up.
s("TX fail", stats.tx_failures, if stats.tx_failures > 0 { C_QUEUE_DROP } else { C_LABEL }),
];
frame.render_widget(Paragraph::new(lines).block(block), area);
}
@@ -456,11 +447,6 @@ fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry])
// Fixed column widths: Age=5 Hash=8 RF=4 Net=4 Size=6 RSSI=6 SNR=5 Act=5
// Every span (header and data) uses the same width per column.
//
// `Act` stays at 5 chars so the full row fits inside the Packet Log
// panel. Retry labels compress to 4 chars (`↻3/3`, `✓3/3`, `✗3/3`) —
// plain drop labels (`DUP`, `FULL`, `RATE`) right-align within the
// same width.
let hdr = Style::default().fg(Color::Gray).add_modifier(Modifier::BOLD);
let header = Line::from(vec![
Span::styled(format!("{:>5}", "Age"), hdr),
@@ -502,35 +488,11 @@ fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry])
),
};
// TX retry state wins over the plain action label when present —
// TX-log entries use the retry tag to carry through retry / success /
// failure state. Otherwise fall back to the generic action.
let (action_text, action_color) = if let Some(r) = e.tx_retry {
// Compact `↻N/M` / `✓N/M` / `✗N/M` — no space between glyph
// and counter, so it fits in the 4-char data slot.
match r.state {
TxRetryState::Retrying(_) => (format!("↻{}/{}", r.attempt, r.total_attempts), C_RATE_DROP),
TxRetryState::Succeeded => (format!("✓{}/{}", r.attempt, r.total_attempts), C_NET_TX),
TxRetryState::Failed => (format!("✗{}/{}", r.attempt, r.total_attempts), C_QUEUE_DROP),
// `InFlight` isn't currently emitted by the router, but keep
// a graceful fallback so any future debug wiring doesn't
// break rendering.
TxRetryState::InFlight => (String::new(), C_LABEL),
}
} else {
let color = match e.action {
PacketAction::Bridged => C_LABEL,
PacketAction::DroppedDedup => C_DEDUP,
PacketAction::DroppedQueueFull => C_QUEUE_DROP,
PacketAction::DroppedRateLimit => C_RATE_DROP,
};
let label = match e.action {
PacketAction::Bridged => "",
PacketAction::DroppedDedup => "DUP",
PacketAction::DroppedQueueFull => "FULL",
PacketAction::DroppedRateLimit => "RATE",
};
(label.to_string(), color)
let action_color = match e.action {
PacketAction::Bridged => C_LABEL,
PacketAction::DroppedDedup => C_DEDUP,
PacketAction::DroppedQueueFull => C_QUEUE_DROP,
PacketAction::DroppedRateLimit => C_RATE_DROP,
};
let rssi_span = match e.rssi {
@@ -542,9 +504,13 @@ fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry])
None => Span::styled(format!("{:>5}", "-"), Style::default().fg(C_LABEL)),
};
// Leading space + right-aligned 4-char data = 5 total, matching
// the `Act` header.
let action_span = Span::styled(format!(" {action_text:>4}"), Style::default().fg(action_color));
let action_str = match e.action {
PacketAction::Bridged => "",
PacketAction::DroppedDedup => "DUP",
PacketAction::DroppedQueueFull => "FULL",
PacketAction::DroppedRateLimit => "RATE",
};
let action_span = Span::styled(format!(" {action_str:>4}"), Style::default().fg(action_color));
let size_str = format!("{}B", e.size);
lines.push(Line::from(vec![
@@ -630,6 +596,29 @@ fn pad_right_align(data: &[u64], width: usize) -> Vec<u64> {
padded
}
// ── Shutdown overlay ─────────────────────────────────────────────
fn draw_shutdown_overlay(frame: &mut ratatui::Frame, area: Rect) {
let dim = Paragraph::new("").style(Style::default().bg(Color::Black));
frame.render_widget(dim, area);
let w = 36u16;
let h = 3u16;
let x = area.x + area.width.saturating_sub(w) / 2;
let y = area.y + area.height.saturating_sub(h) / 2;
let popup = Rect::new(x, y, w.min(area.width), h.min(area.height));
frame.render_widget(Clear, popup);
let block = Block::default().borders(Borders::ALL).border_style(Style::default().fg(C_BORDER));
let text = Paragraph::new(Line::from(vec![
Span::styled(" Shutting down", Style::default().fg(C_RATE_DROP)),
Span::styled(" — updating DHT...", Style::default().fg(C_LABEL)),
]))
.alignment(Alignment::Center)
.block(block);
frame.render_widget(text, popup);
}
// ── Shared helpers ───────────────────────────────────────────────
fn panel_block(title: &str) -> Block<'static> {
@@ -711,8 +700,6 @@ mod tests {
dropped_queue: 0,
neighbor_count: 0,
radio_connected: true,
tx_retries: 0,
tx_failures: 0,
}
}
@@ -725,7 +712,6 @@ mod tests {
snr,
rssi: Some(-80),
action: PacketAction::Bridged,
tx_retry: None,
}
}