From 0b9775c3101e806894b24ab372a38811a4165677 Mon Sep 17 00:00:00 2001 From: Stephen Waits Date: Thu, 16 Apr 2026 15:08:07 -0600 Subject: [PATCH] feat!: bridge on DongLoRa Protocol v2 and async client MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 1.0 bridge running on the async Dongle API from donglora-client 1.0.0 and the Protocol v2 wire format. * Radio loop is now a tokio task, not a dedicated std::thread. Everything is async end to end. * [radio] config schema: tx_power renamed to tx_power_dbm (plain i8, no "max" keyword), cad field removed (CAD is a per-TX flag in v1.0), default sync_word changed 0x3444 to 0x1424 to match the SX126x register encoding of MeshCore private sync byte. * TX retry with backoff (new [tx] config section). Every TX runs through Dongle::tx_with_retry, retrying TX_DONE(CHANNEL_BUSY) and ERR(EBUSY) per PROTOCOL.md §6.10 / §C.5.5. Defaults: 3 attempts, 20-100 ms randomized backoff, doubling to 500 ms cap, 5 s per-attempt timeout. Configurable via max_attempts, backoff_min_ms, backoff_max_ms, backoff_cap_ms, backoff_multiplier, per_attempt_timeout_secs. * TUI packet log shows retry state: in-flight retries render with a cycle marker in yellow and an N/M attempt counter, terminal TX failures render in red, post-retry success shows final attempt count. * Stats gain tx_retries and tx_failures counters surfaced in StatsSnapshot. * Measured airtime from TX_DONE replaces the theoretical estimate, feeding the rate limiter with the real on-air duration. * The explicit 2 s liveness ping loop is gone. The Dongle runs a 500 ms internal keepalive, matching spec §3.4 inactivity window with 2x margin. --- CHANGELOG.md | 43 ++++ Cargo.lock | 98 +++++--- Cargo.toml | 4 +- src/config.rs | 349 +++++++++++++-------------- src/gossip.rs | 169 +++++++------ src/main.rs | 105 ++++++++- src/radio.rs | 589 +++++++++++++++++++++++----------------------- src/rate_limit.rs | 93 ++++---- src/router.rs | 139 ++++++++++- src/setup.rs | 11 +- src/tui.rs | 79 +++++-- 11 files changed, 1010 insertions(+), 669 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 488e1ad..51759a3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,48 @@ # Changelog +## [1.0.0-alpha.1] - 2026-04-21 + +### 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 diff --git a/Cargo.lock b/Cargo.lock index 95fee7c..3ff0126 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -906,7 +906,7 @@ dependencies = [ [[package]] name = "donglora-bridge" -version = "0.3.1" +version = "1.0.0-alpha.1" dependencies = [ "anyhow", "blake3", @@ -938,13 +938,22 @@ dependencies = [ [[package]] name = "donglora-client" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "343d750fe1e5dad94d4b31e6dbbb05214b39e66f3c94a2ff2a29b0031ffd2826" +version = "1.0.0-alpha.1" dependencies = [ - "anyhow", - "serialport", + "donglora-protocol", + "rand 0.9.2", + "thiserror 2.0.18", + "tokio", + "tokio-serial", + "tokio-stream", "tracing", +] + +[[package]] +name = "donglora-protocol" +version = "0.1.0" +dependencies = [ + "heapless 0.9.2", "ucobs", ] @@ -1398,6 +1407,15 @@ dependencies = [ "byteorder", ] +[[package]] +name = "hash32" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47d60b12902ba28e2730cd37e95b8c9223af2808df9e902d4df49588d1470606" +dependencies = [ + "byteorder", +] + [[package]] name = "hashbrown" version = "0.15.5" @@ -1425,13 +1443,23 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cdc6457c0eb62c71aac4bc17216026d8410337c4126773b9c5daba343f17964f" dependencies = [ "atomic-polyfill", - "hash32", + "hash32 0.2.1", "rustc_version", "serde", "spin 0.9.8", "stable_deref_trait", ] +[[package]] +name = "heapless" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2af2455f757db2b292a9b1768c4b70186d443bcb3b316252d6b540aec1cd89ed" +dependencies = [ + "hash32 0.3.1", + "stable_deref_trait", +] + [[package]] name = "heck" version = "0.5.0" @@ -2190,26 +2218,6 @@ dependencies = [ "libc", ] -[[package]] -name = "libudev" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "78b324152da65df7bb95acfcaab55e3097ceaab02fb19b228a9eb74d55f135e0" -dependencies = [ - "libc", - "libudev-sys", -] - -[[package]] -name = "libudev-sys" -version = "0.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3c8469b4a23b962c1396b9b451dda50ef5b283e8dd309d69033475fa9b334324" -dependencies = [ - "libc", - "pkg-config", -] - [[package]] name = "line-clipping" version = "0.3.7" @@ -2375,6 +2383,19 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "mio-serial" +version = "5.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "029e1f407e261176a983a6599c084efd322d9301028055c87174beac71397ba3" +dependencies = [ + "log", + "mio", + "nix 0.29.0", + "serialport", + "winapi", +] + [[package]] name = "moka" version = "0.12.15" @@ -3038,12 +3059,6 @@ dependencies = [ "spki", ] -[[package]] -name = "pkg-config" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" - [[package]] name = "plist" version = "1.8.0" @@ -3126,7 +3141,7 @@ dependencies = [ "cobs", "embedded-io 0.4.0", "embedded-io 0.6.1", - "heapless", + "heapless 0.7.17", "postcard-derive", "serde", ] @@ -3775,7 +3790,6 @@ dependencies = [ "core-foundation", "core-foundation-sys", "io-kit-sys", - "libudev", "mach2", "nix 0.26.4", "scopeguard", @@ -4284,6 +4298,20 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-serial" +version = "5.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aa1d5427f11ba7c5e6384521cfd76f2d64572ff29f3f4f7aa0f496282923fdc8" +dependencies = [ + "cfg-if", + "futures", + "log", + "mio-serial", + "serialport", + "tokio", +] + [[package]] name = "tokio-stream" version = "0.1.18" diff --git a/Cargo.toml b/Cargo.toml index 3832c8f..8e5a7d9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "donglora-bridge" -version = "0.3.1" +version = "1.0.0-alpha.1" edition = "2024" authors = ["Stephen Waits "] license = "MIT" @@ -14,7 +14,7 @@ rust-version = "1.94.1" exclude = [".jj/", ".github/"] [dependencies] -donglora-client = "0.2" +donglora-client = { path = "../client-rs", version = "1.0.0-alpha.1" } iroh = "0.97" iroh-gossip = "0.97" mainline = { version = "6", features = ["async"] } diff --git a/src/config.rs b/src/config.rs index 8654dab..d2abbef 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,12 +1,13 @@ //! TOML configuration loading and validation. //! -//! Handles the `[radio]` and `[bridge]` sections with `MeshCore` US/Canada recommended -//! defaults, and conversion to donglora-client `RadioConfig`. +//! Handles the `[radio]`, `[bridge]`, and `[tx]` sections with `MeshCore` US/Canada +//! recommended defaults, and conversion to the donglora-client `LoRaConfig`. use std::path::{Path, PathBuf}; +use std::time::Duration; use anyhow::{Context, Result, bail}; -use donglora_client::{Bandwidth, RadioConfig, TX_POWER_MAX}; +use donglora_client::{LoRaBandwidth, LoRaCodingRate, LoRaConfig, LoRaHeaderMode, RetryPolicy}; use serde::{Deserialize, Serialize}; /// Top-level bridge configuration. @@ -16,6 +17,8 @@ pub struct Config { pub radio: RadioSection, #[serde(default)] pub bridge: BridgeSection, + #[serde(default)] + pub tx: TxSection, } #[derive(Debug, Clone, Deserialize, Serialize)] @@ -33,11 +36,9 @@ pub struct RadioSection { #[serde(default = "default_sync_word")] pub sync_word: u16, #[serde(default = "default_tx_power")] - pub tx_power: String, + pub tx_power_dbm: i8, #[serde(default = "default_preamble")] pub preamble: u16, - #[serde(default = "default_true")] - pub cad: bool, } #[derive(Debug, Clone, Deserialize, Serialize)] @@ -55,6 +56,25 @@ pub struct BridgeSection { pub log_file: Option, } +/// 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 { @@ -69,24 +89,41 @@ 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 { - 0x3444 + 0x1424 } -fn default_tx_power() -> String { - "22".into() +const fn default_tx_power() -> i8 { + 20 } 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 { @@ -97,9 +134,8 @@ impl Default for RadioSection { spreading_factor: default_sf(), coding_rate: default_cr(), sync_word: default_sync_word(), - tx_power: default_tx_power(), + tx_power_dbm: default_tx_power(), preamble: default_preamble(), - cad: true, } } } @@ -116,56 +152,95 @@ impl Default for BridgeSection { } } +impl Default for TxSection { + fn default() -> Self { + Self { + max_attempts: default_max_attempts(), + backoff_min_ms: default_backoff_min_ms(), + backoff_max_ms: default_backoff_max_ms(), + backoff_cap_ms: default_backoff_cap_ms(), + backoff_multiplier: default_backoff_multiplier(), + per_attempt_timeout_secs: default_per_attempt_timeout_secs(), + } + } +} + // ── Conversion ──────────────────────────────────────────────────── -fn parse_bandwidth(s: &str) -> Result { +fn parse_bandwidth(s: &str) -> Result { match s.to_lowercase().as_str() { - "7.8khz" | "7khz" | "khz7" => Ok(Bandwidth::Khz7), - "10.4khz" | "10khz" | "khz10" => Ok(Bandwidth::Khz10), - "15.6khz" | "15khz" | "khz15" => Ok(Bandwidth::Khz15), - "20.8khz" | "20khz" | "khz20" => Ok(Bandwidth::Khz20), - "31.25khz" | "31khz" | "khz31" => Ok(Bandwidth::Khz31), - "41.7khz" | "41khz" | "khz41" => Ok(Bandwidth::Khz41), - "62.5khz" | "62khz" | "khz62" => Ok(Bandwidth::Khz62), - "125khz" | "khz125" => Ok(Bandwidth::Khz125), - "250khz" | "khz250" => Ok(Bandwidth::Khz250), - "500khz" | "khz500" => Ok(Bandwidth::Khz500), + "7.8khz" | "7khz" | "khz7" => Ok(LoRaBandwidth::Khz7), + "10.4khz" | "10khz" | "khz10" => Ok(LoRaBandwidth::Khz10), + "15.6khz" | "15khz" | "khz15" => Ok(LoRaBandwidth::Khz15), + "20.8khz" | "20khz" | "khz20" => Ok(LoRaBandwidth::Khz20), + "31.25khz" | "31khz" | "khz31" => Ok(LoRaBandwidth::Khz31), + "41.7khz" | "41khz" | "khz41" => Ok(LoRaBandwidth::Khz41), + "62.5khz" | "62khz" | "khz62" => Ok(LoRaBandwidth::Khz62), + "125khz" | "khz125" => Ok(LoRaBandwidth::Khz125), + "250khz" | "khz250" => Ok(LoRaBandwidth::Khz250), + "500khz" | "khz500" => Ok(LoRaBandwidth::Khz500), _ => bail!("unknown bandwidth: {s}"), } } -fn parse_tx_power(s: &str) -> Result { - if s.eq_ignore_ascii_case("max") { - Ok(TX_POWER_MAX) - } else { - s.parse::().context("invalid tx_power (use 'max' or a dBm integer)") +fn parse_coding_rate(v: u8) -> Result { + match v { + 5 => Ok(LoRaCodingRate::Cr4_5), + 6 => Ok(LoRaCodingRate::Cr4_6), + 7 => Ok(LoRaCodingRate::Cr4_7), + 8 => Ok(LoRaCodingRate::Cr4_8), + _ => bail!("coding_rate must be 5-8 (meaning 4/5 to 4/8), got {v}"), } } impl RadioSection { - /// Convert to the donglora-client `RadioConfig`. + /// Convert to the donglora-client `LoRaConfig`. /// /// Validates that all radio parameters are within hardware-supported ranges /// before constructing the config. - pub fn to_radio_config(&self) -> Result { + pub fn to_lora_config(&self) -> Result { if !(7..=12).contains(&self.spreading_factor) { bail!("spreading_factor must be 7-12, got {}", self.spreading_factor); } - if !(5..=8).contains(&self.coding_rate) { - bail!("coding_rate must be 5-8 (meaning 4/5 to 4/8), got {}", self.coding_rate); - } if self.frequency < 150_000_000 || self.frequency > 1_020_000_000 { bail!("frequency {:.3} MHz out of range (150 MHz to 1.02 GHz)", f64::from(self.frequency) / 1_000_000.0); } - Ok(RadioConfig { + Ok(LoRaConfig { freq_hz: self.frequency, bw: parse_bandwidth(&self.bandwidth)?, sf: self.spreading_factor, - cr: self.coding_rate, + cr: parse_coding_rate(self.coding_rate)?, sync_word: self.sync_word, - tx_power_dbm: parse_tx_power(&self.tx_power)?, + tx_power_dbm: self.tx_power_dbm, preamble_len: self.preamble, - cad: u8::from(self.cad), + header_mode: LoRaHeaderMode::Explicit, + payload_crc: true, + iq_invert: false, + }) + } +} + +impl TxSection { + /// Materialize the TOML fields into a [`RetryPolicy`]. + pub fn to_retry_policy(&self) -> Result { + if self.max_attempts == 0 { + bail!("tx.max_attempts must be >= 1"); + } + if self.backoff_min_ms > self.backoff_max_ms { + bail!( + "tx.backoff_min_ms ({}) must not exceed backoff_max_ms ({})", + self.backoff_min_ms, + self.backoff_max_ms + ); + } + Ok(RetryPolicy { + max_attempts: self.max_attempts, + backoff_ms_min: self.backoff_min_ms, + backoff_ms_max: self.backoff_max_ms, + backoff_multiplier: self.backoff_multiplier, + backoff_cap_ms: self.backoff_cap_ms, + per_attempt_timeout: Duration::from_secs(self.per_attempt_timeout_secs), + skip_cad: false, }) } } @@ -209,92 +284,97 @@ mod tests { #[test] fn parse_bandwidth_canonical() { - assert!(matches!(parse_bandwidth("7.8kHz"), Ok(Bandwidth::Khz7))); - assert!(matches!(parse_bandwidth("62.5kHz"), Ok(Bandwidth::Khz62))); - assert!(matches!(parse_bandwidth("125kHz"), Ok(Bandwidth::Khz125))); - assert!(matches!(parse_bandwidth("500kHz"), Ok(Bandwidth::Khz500))); + assert!(matches!(parse_bandwidth("7.8kHz"), Ok(LoRaBandwidth::Khz7))); + assert!(matches!(parse_bandwidth("62.5kHz"), Ok(LoRaBandwidth::Khz62))); + assert!(matches!(parse_bandwidth("125kHz"), Ok(LoRaBandwidth::Khz125))); + assert!(matches!(parse_bandwidth("500kHz"), Ok(LoRaBandwidth::Khz500))); } #[test] fn parse_bandwidth_aliases() { - assert!(matches!(parse_bandwidth("7khz"), Ok(Bandwidth::Khz7))); - assert!(matches!(parse_bandwidth("khz125"), Ok(Bandwidth::Khz125))); - assert!(matches!(parse_bandwidth("10khz"), Ok(Bandwidth::Khz10))); + assert!(matches!(parse_bandwidth("7khz"), Ok(LoRaBandwidth::Khz7))); + assert!(matches!(parse_bandwidth("khz125"), Ok(LoRaBandwidth::Khz125))); } #[test] fn parse_bandwidth_case_insensitive() { - assert!(matches!(parse_bandwidth("125KHZ"), Ok(Bandwidth::Khz125))); - assert!(matches!(parse_bandwidth("62.5KHZ"), Ok(Bandwidth::Khz62))); + assert!(matches!(parse_bandwidth("125KHZ"), Ok(LoRaBandwidth::Khz125))); } #[test] fn parse_bandwidth_rejects_unknown() { assert!(parse_bandwidth("999kHz").is_err()); assert!(parse_bandwidth("").is_err()); - assert!(parse_bandwidth("not_a_bandwidth").is_err()); } #[test] - fn parse_tx_power_numeric() { - assert_eq!(parse_tx_power("22").unwrap(), 22); - assert_eq!(parse_tx_power("-4").unwrap(), -4); - assert_eq!(parse_tx_power("0").unwrap(), 0); + fn parse_coding_rate_canonical() { + assert!(matches!(parse_coding_rate(5), Ok(LoRaCodingRate::Cr4_5))); + assert!(matches!(parse_coding_rate(8), Ok(LoRaCodingRate::Cr4_8))); } #[test] - fn parse_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); + fn parse_coding_rate_rejects_out_of_range() { + assert!(parse_coding_rate(4).is_err()); + assert!(parse_coding_rate(9).is_err()); } #[test] - fn parse_tx_power_rejects_garbage() { - assert!(parse_tx_power("abc").is_err()); - assert!(parse_tx_power("").is_err()); - } - - #[test] - fn radio_section_defaults_to_radio_config() { + fn radio_section_defaults_to_lora_config() { let section = RadioSection::default(); - let config = section.to_radio_config().unwrap(); - assert_eq!(config.freq_hz, 910_525_000); - assert_eq!(config.sf, 7); - assert_eq!(config.cr, 5); - assert_eq!(config.sync_word, 0x3444); - assert_eq!(config.preamble_len, 16); + let cfg = section.to_lora_config().unwrap(); + assert_eq!(cfg.freq_hz, 910_525_000); + assert_eq!(cfg.sf, 7); + assert!(matches!(cfg.cr, LoRaCodingRate::Cr4_5)); + assert_eq!(cfg.sync_word, 0x1424); + assert_eq!(cfg.preamble_len, 16); + assert_eq!(cfg.tx_power_dbm, 20); } #[test] - fn 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() { + fn to_lora_config_rejects_invalid_sf() { let section = RadioSection { spreading_factor: 6, ..RadioSection::default() }; - assert!(section.to_radio_config().is_err()); + assert!(section.to_lora_config().is_err()); let section = RadioSection { spreading_factor: 13, ..RadioSection::default() }; - assert!(section.to_radio_config().is_err()); + assert!(section.to_lora_config().is_err()); } #[test] - fn to_radio_config_rejects_invalid_cr() { + fn to_lora_config_rejects_invalid_cr() { let section = RadioSection { coding_rate: 4, ..RadioSection::default() }; - assert!(section.to_radio_config().is_err()); + assert!(section.to_lora_config().is_err()); let section = RadioSection { coding_rate: 9, ..RadioSection::default() }; - assert!(section.to_radio_config().is_err()); + assert!(section.to_lora_config().is_err()); } #[test] - fn to_radio_config_rejects_invalid_frequency() { + fn to_lora_config_rejects_invalid_frequency() { let section = RadioSection { frequency: 0, ..RadioSection::default() }; - assert!(section.to_radio_config().is_err()); + assert!(section.to_lora_config().is_err()); let section = RadioSection { frequency: 2_000_000_000, ..RadioSection::default() }; - assert!(section.to_radio_config().is_err()); + assert!(section.to_lora_config().is_err()); + } + + #[test] + fn tx_section_defaults_produce_valid_policy() { + let p = TxSection::default().to_retry_policy().unwrap(); + assert_eq!(p.max_attempts, 3); + assert_eq!(p.backoff_ms_min, 20); + assert_eq!(p.backoff_ms_max, 100); + assert_eq!(p.backoff_cap_ms, 500); + assert!((p.backoff_multiplier - 2.0).abs() < f32::EPSILON); + } + + #[test] + fn tx_section_rejects_zero_attempts() { + let s = TxSection { max_attempts: 0, ..TxSection::default() }; + assert!(s.to_retry_policy().is_err()); + } + + #[test] + fn tx_section_rejects_inverted_backoff_bounds() { + let s = TxSection { backoff_min_ms: 500, backoff_max_ms: 100, ..TxSection::default() }; + assert!(s.to_retry_policy().is_err()); } #[test] @@ -305,6 +385,7 @@ 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] @@ -335,100 +416,4 @@ 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); - } } diff --git a/src/gossip.rs b/src/gossip.rs index cca2b75..b65fcfe 100644 --- a/src/gossip.rs +++ b/src/gossip.rs @@ -22,7 +22,7 @@ use mainline::{Dht, MutableItem, SigningKey}; use sha2::{Digest, Sha256}; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; -use tracing::{debug, info, warn}; +use tracing::{debug, error, info, warn}; use crate::packet::GossipFrame; @@ -31,14 +31,27 @@ use crate::packet::GossipFrame; /// 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; +/// DHT heartbeat interval (base, before jitter). Each bridge periodically +/// publishes `[self] + live_neighbors` as a blind overwrite. Any live bridge +/// publishing at this cadence guarantees that the DHT record is overwritten +/// with a fresh snapshot of a real, connected sub-swarm within this window — +/// so stale entries from crashed / restarted peers age out within one cycle. +const DHT_HEARTBEAT_SECS: u64 = 60; -/// Maximum random jitter added to heartbeat interval. Prevents synchronized -/// publish storms when multiple bridges start simultaneously. -const DHT_HEARTBEAT_JITTER_SECS: u64 = 60; +/// Maximum random jitter added to heartbeat. Prevents synchronized publish +/// storms when multiple bridges start simultaneously. +const DHT_HEARTBEAT_JITTER_SECS: u64 = 15; + +/// DHT read interval while we have no neighbors. Stays aggressive so two +/// freshly-started bridges find each other quickly. +const DHT_LONELY_READ_SECS: u64 = 3; + +/// How long to try dialing DHT-discovered peers before we give up and publish +/// ourselves alone. If we connect to at least one peer in this window we +/// publish `[self, neighbors...]` right away. If we don't, we publish `[self]` +/// so newcomers find us — but we delay this so we don't overwrite a +/// potentially useful DHT record with just our own lonely entry. +const DHT_BOOTSTRAP_GRACE_SECS: u64 = 10; // ── Key derivation ─────────────────────────────────────────────── @@ -321,12 +334,23 @@ fn spawn_receiver( cancel: CancellationToken, ) { tokio::spawn(async move { - loop { + // Log a hard warning if the receiver stream ever ends — this is the + // "silent one-way gossip" failure mode. Track why it exited so the + // log tells us whether it was a stream-end (None), an error + // (Some(Err)), or a cancel. + let exit_reason: &'static str = loop { let event = tokio::select! { event = receiver.next() => event, - () = cancel.cancelled() => break, + () = cancel.cancelled() => break "cancelled", + }; + let event = match event { + None => break "stream ended (receiver closed)", + Some(Err(e)) => { + error!("gossip receiver error: {e:#}"); + break "receiver error"; + } + Some(Ok(ev)) => ev, }; - let Some(Ok(event)) = event else { break }; match event { Event::Received(msg) => { if let Some(frame) = GossipFrame::decode(&msg.content) { @@ -364,8 +388,15 @@ fn spawn_receiver( warn!("gossip receiver lagged — some messages were dropped"); } } + }; + // A shutdown cancel is expected; any other exit means we silently + // stopped receiving gossip. That's the "one-way bridge" bug — + // broadcast still works because spawn_broadcaster is independent. + if exit_reason == "cancelled" { + debug!("gossip receiver stopped ({exit_reason})"); + } else { + error!("gossip receiver STOPPED ({exit_reason}) — this bridge will no longer receive from the swarm",); } - debug!("gossip receiver stopped"); }); } @@ -391,24 +422,25 @@ fn spawn_broadcaster( }); } -/// DHT poll interval while alone. 3 seconds is aggressive but reads are cheap -/// (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. +/// Periodically publish `[self] + live_neighbors` to the DHT and read it to +/// discover new 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. +/// Strategy — deliberately simple: +/// 1. **Startup**: read DHT, try to join discovered peers. Do NOT publish +/// yet — we don't know if any of them are still alive, and publishing +/// `[self]` alone would overwrite a potentially useful record. +/// 2. **Bootstrap grace** (10 s): keep reading fast (every +/// `DHT_LONELY_READ_SECS`) looking for peers to connect to. If a +/// `NeighborUp` fires inside this window, publish immediately with the +/// live neighbor set. If nothing shows up, publish `[self]` so anyone +/// else joining finds us. +/// 3. **Steady state**: publish `[self, neighbors...]` every +/// `DHT_HEARTBEAT_SECS` (+ jitter), a blind overwrite. Any live bridge +/// doing this naturally ages out crashed/stale peers from the DHT +/// record within one cycle. +/// 4. **Re-read cadence**: fast (`DHT_LONELY_READ_SECS`) while +/// `neighbor_count == 0`, relaxed (= heartbeat interval) once +/// connected. #[allow(clippy::too_many_arguments)] fn spawn_dht_heartbeat( my_id: PublicKey, @@ -430,27 +462,29 @@ fn spawn_dht_heartbeat( }; let pub_key_bytes = signing_key.verifying_key().to_bytes(); + let mut last_publish: Option = None; let mut was_alone = true; - let mut last_publish = Instant::now(); + let started = Instant::now(); - // Initial: read to discover existing peers, then merge-publish ourselves. + // Step 1: read the DHT to find peers to dial. dht_read_and_join(&dht, &my_id, &pub_key_bytes, &salt, &gossip_sender, &state).await; - dht_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. + // Transition: alone → connected. Publish immediately with live + // neighbors so anyone who was about to dial us-as-lonely finds + // the full picture. if was_alone && !alone { 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(); + dht_publish(&dht, &my_id, &signing_key, &salt, &neighbors, &state).await; + last_publish = Some(Instant::now()); } was_alone = alone; - // Sleep: fast reads while alone, relaxed when connected. + // Sleep: fast reads while lonely, relaxed once connected. let sleep_dur = if alone { - Duration::from_secs(DHT_ALONE_READ_SECS) + Duration::from_secs(DHT_LONELY_READ_SECS) } else { let jitter = rand::random::() % DHT_HEARTBEAT_JITTER_SECS.max(1); Duration::from_secs(DHT_HEARTBEAT_SECS + jitter) @@ -461,18 +495,21 @@ fn spawn_dht_heartbeat( () = cancel.cancelled() => break, } - // Always read — cheap, discovers peers to join. + // Always read — cheap, keeps fresh peers coming in. dht_read_and_join(&dht, &my_id, &pub_key_bytes, &salt, &gossip_sender, &state).await; - // 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(); + // Steady-state: publish on every heartbeat tick once we've + // published at least once. Pre-first-publish, hold off until + // either a neighbor appears OR the bootstrap grace elapses — + // this avoids wiping a potentially-useful DHT record with + // just our lonely [self] entry on startup. + let should_publish = last_publish.map_or_else( + || !alone || started.elapsed() >= Duration::from_secs(DHT_BOOTSTRAP_GRACE_SECS), + |t| t.elapsed() >= Duration::from_secs(DHT_HEARTBEAT_SECS), + ); + if should_publish { + dht_publish(&dht, &my_id, &signing_key, &salt, &neighbors, &state).await; + last_publish = Some(Instant::now()); } } debug!("DHT heartbeat stopped"); @@ -503,44 +540,30 @@ async fn dht_read_and_join( } } +/// Publish `[self] + live_neighbors` to the DHT as a blind overwrite. +/// +/// No merge with existing DHT content. Stale peers age out naturally +/// because every live bridge publishes a snapshot of its actual live +/// connections within `DHT_HEARTBEAT_SECS`, and higher-seq records win. async fn dht_publish( dht: &AsyncDht, my_id: &PublicKey, signing_key: &SigningKey, salt: &[u8; 20], - pub_key_bytes: &[u8; 32], neighbors: &Arc>>, state: &Arc>, ) { - 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()); - } + // Snapshot current live neighbors — those we have an active gossip + // connection to right now. No inherited junk from prior DHT content. + if let Ok(n) = neighbors.lock() { + for id in n.iter() { + if peer_list.len() >= MAX_DHT_PEERS { + break; } + peer_list.push(*id.as_bytes()); } } diff --git a/src/main.rs b/src/main.rs index 7b84e49..e30e991 100644 --- a/src/main.rs +++ b/src/main.rs @@ -71,17 +71,24 @@ async fn main() -> Result<()> { info!("donglora-bridge starting, ephemeral id: {}", node_id.fmt_short()); // Parse radio config. - let radio_config = cfg.radio.to_radio_config()?; + let radio_config = cfg.radio.to_lora_config()?; + let retry_policy = cfg.tx.to_retry_policy()?; // Derive passphrase keys. 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, - radio_config.cr, + cr_denom, radio_config.preamble_len, cfg.bridge.rate_limit_pps, ); @@ -92,7 +99,7 @@ async fn main() -> Result<()> { let swarm = Arc::new(swarm); // Start radio. - let (radio_event_rx, radio_tx) = radio::spawn(radio_config, cfg.radio.port.clone())?; + let (radio_event_rx, radio_tx, _radio_handle) = radio::spawn(radio_config, cfg.radio.port.clone(), retry_policy)?; // Stats + log channel + radio config watch. let stats = Arc::new(Stats::default()); @@ -160,6 +167,13 @@ 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); @@ -182,6 +196,91 @@ 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, 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() diff --git a/src/radio.rs b/src/radio.rs index 6fe11a1..c1a9a88 100644 --- a/src/radio.rs +++ b/src/radio.rs @@ -1,13 +1,24 @@ -//! `DongLoRa` hardware communication on a blocking thread. +//! Async radio task — drives the `donglora-client` [`Dongle`] session. //! -//! Handles connection, config negotiation, RX/TX, liveness pings, and -//! exponential backoff reconnection. +//! Owns a single tokio task that connects (with exponential backoff reconnect), +//! negotiates config, starts RX, and multiplexes TX requests from the router +//! against the RX stream. TX is dispatched via [`Dongle::tx_with_retry`] so +//! `TX_DONE(CHANNEL_BUSY)` / `EBUSY` errors are retried per spec §6.10 with +//! randomized backoff, and per-attempt state is surfaced to the router (and +//! onwards to the TUI packet log) via the [`RadioEvent::Tx*`] variants. +//! +//! Keepalive pings are handled inside the [`Dongle`] itself — no explicit +//! liveness loop here. use std::time::Duration; use anyhow::{Context, Result}; -use donglora_client::{AnyTransport, RadioConfig, Response, Transport}; +use donglora_client::{ + ConnectOptions, Dongle, LoRaBandwidth, LoRaConfig, Modulation, RetryPolicy, RxPayload, connect_mux_auto_with, + connect_with, +}; use tokio::sync::mpsc; +use tokio::task::JoinHandle; use tracing::{debug, error, info, warn}; use crate::packet::{RadioPacket, snr_grade}; @@ -21,253 +32,295 @@ pub enum ConfigSource { Mux, } -/// Messages from the radio thread to the router. +/// Messages from the radio task to the router. +#[derive(Debug)] pub enum RadioEvent { /// A valid packet received from the radio (already SNR-filtered). Packet(RadioPacket), /// Radio connected (or reconnected) with the active config and device info. - Connected(RadioConfig, ConfigSource, String), + Connected(LoRaConfig, ConfigSource, String), /// Radio disconnected — attempting reconnect. Disconnected, + /// TX attempt started. `attempt` is 1-indexed. + TxAttempt { seq: u64, attempt: u8, total_attempts: u8, size: usize }, + /// Previous TX attempt failed and a retry is scheduled. + TxRetry { seq: u64, attempt_that_failed: u8, total_attempts: u8, reason: TxRetryReason, backoff_ms: u32 }, + /// TX completed successfully (perhaps after retries). + TxSucceeded { seq: u64, attempts_used: u8, final_airtime_us: u32, size: usize }, + /// TX policy exhausted or errored terminally. + TxFailed { seq: u64, attempts_used: u8, reason: String, size: usize }, } -/// Spawn the blocking radio thread. +/// Why a TX attempt was retried — carried alongside the retry event for +/// the TUI. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TxRetryReason { + /// CAD detected channel activity before TX. + ChannelBusy, + /// Firmware TX queue was full. + QueueFull, + /// Other retryable condition (future-compat). + Other, +} + +impl std::fmt::Display for TxRetryReason { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::ChannelBusy => write!(f, "channel_busy"), + Self::QueueFull => write!(f, "queue_full"), + Self::Other => write!(f, "other"), + } + } +} + +/// One outbound TX request from the router: the payload plus a sequence id so +/// the event stream can correlate back to the packet log entry. +#[derive(Debug)] +pub struct TxRequest { + pub seq: u64, + pub data: Vec, +} + +/// Spawn the async radio task. /// -/// Returns a receiver for radio events and a sender for TX payloads. -pub fn spawn(config: RadioConfig, port: Option) -> Result<(mpsc::Receiver, mpsc::Sender>)> { +/// Returns a receiver for radio events and a sender for [`TxRequest`]s. +pub fn spawn( + config: LoRaConfig, + port: Option, + retry_policy: RetryPolicy, +) -> Result<(mpsc::Receiver, mpsc::Sender, JoinHandle<()>)> { let (event_tx, event_rx) = mpsc::channel::(256); - let (tx_send, tx_recv) = mpsc::channel::>(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)) + let (tx_send, tx_recv) = mpsc::channel::(64); + let handle = tokio::spawn(radio_loop(config, port, retry_policy, event_tx, tx_recv)); + Ok((event_rx, tx_send, handle)) } -#[allow(clippy::needless_pass_by_value)] // values moved across thread boundary in spawn() -fn radio_loop( - config: RadioConfig, +async fn radio_loop( + config: LoRaConfig, port: Option, + retry_policy: RetryPolicy, event_tx: mpsc::Sender, - mut tx_recv: mpsc::Receiver>, + mut tx_recv: mpsc::Receiver, ) { let initial_backoff = Duration::from_millis(250); let max_backoff = Duration::from_secs(5); let mut backoff = initial_backoff; - let mut use_mux_only = false; + let mut stick_to_mux = false; loop { let started = std::time::Instant::now(); - match connect_and_run(&config, port.as_deref(), use_mux_only, &event_tx, &mut tx_recv) { - Ok(_was_mux) => { - info!("radio thread exiting"); - return; + match connect_and_run(&config, port.as_deref(), stick_to_mux, &retry_policy, &event_tx, &mut tx_recv).await { + Ok(was_mux) => { + if event_tx.is_closed() { + info!("radio task exiting (router closed)"); + return; + } + // We had a clean run_session exit but not shutdown — loop back. + if was_mux { + stick_to_mux = true; + } + debug!("radio session exited cleanly; reconnecting"); } Err((was_mux, e)) => { - // Session ended with error. Remember transport type for reconnect. if was_mux { - use_mux_only = true; + stick_to_mux = true; } error!("radio error: {e:#}"); - let _ = event_tx.blocking_send(RadioEvent::Disconnected); + let _ = event_tx.send(RadioEvent::Disconnected).await; - // If we were connected for a while, reset backoff so - // the next reconnect attempt is fast. if started.elapsed() > Duration::from_secs(5) { backoff = initial_backoff; } - info!("reconnecting in {backoff:?}"); - std::thread::sleep(backoff); + tokio::time::sleep(backoff).await; backoff = (backoff * 2).min(max_backoff); } } } } -/// Connect to the dongle and run the radio loop. -/// -/// Returns `Ok(was_mux)` on clean exit, `Err((was_mux, error))` on failure. -/// The `was_mux` flag lets the caller know which transport was used so it can -/// stick to the same transport type on reconnect. -fn connect_and_run( - config: &RadioConfig, +async fn connect_and_run( + config: &LoRaConfig, port: Option<&str>, - use_mux_only: bool, + stick_to_mux: bool, + retry_policy: &RetryPolicy, event_tx: &mpsc::Sender, - tx_recv: &mut mpsc::Receiver>, + tx_recv: &mut mpsc::Receiver, ) -> std::result::Result { - // Long initial timeout: the mux serializes commands through the dongle, - // so our `GetConfig` might have to wait behind other clients' commands. - let timeout = Duration::from_secs(10); - info!("connecting to dongle..."); - let mut client = if use_mux_only { - debug!("reconnecting via mux only (previously connected via mux)"); - donglora_client::connect_mux_auto(timeout).map_err(|e| (true, e))? - } else { - donglora_client::connect(port, timeout).map_err(|e| (false, e))? - }; - let is_mux = matches!(client.transport(), AnyTransport::Mux(_)); - let device = if is_mux { "mux".to_string() } else { port.map_or_else(|| "usb".to_string(), shorten_path) }; - info!("[radio] connected and validated: {device}"); - - // Run the session. Map Result<()> to include the is_mux flag. - run_session(&mut client, config, is_mux, &device, event_tx, tx_recv).map(|()| is_mux).map_err(|e| (is_mux, e)) -} - -fn run_session( - client: &mut donglora_client::Client, - config: &RadioConfig, - is_mux: bool, - device: &str, - event_tx: &mpsc::Sender, - tx_recv: &mut mpsc::Receiver>, -) -> Result<()> { - info!("[radio] negotiating config..."); - let (active_config, config_source) = match negotiate_config(client, config, is_mux) { - 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 mut opts = ConnectOptions::new().config(Modulation::LoRa(*config)); + if let Some(p) = port { + opts = opts.port(p); } - let _ = event_tx.blocking_send(RadioEvent::Connected(active_config, config_source, device.to_string())); + // Attempt mux-only if we've committed to it, otherwise the normal + // auto-discovery chain. + let dongle = if stick_to_mux { + debug!("reconnecting via mux only (previously connected via mux)"); + connect_mux_auto_with(opts).await.map_err(|e| (true, anyhow::Error::from(e)))? + } else { + 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, + tx_recv: &mut mpsc::Receiver, +) -> Result<()> { + // The Dongle's auto-configure already applied our requested config at + // connect time (via `ConnectOptions::config`). Read back the active + // config — on the mux, another client may have locked a different one. + let active_config = match dongle.config().await { + Some(Modulation::LoRa(c)) => c, + Some(other) => { + warn!("[radio] mux has a non-LoRa config locked ({other:?}); using it anyway"); + // This branch is unreachable today since the bridge only supports + // LoRa, but we surface a warning rather than hard-fail so the + // mux doesn't brick the bridge. + *config + } + None => { + warn!("[radio] Dongle::config() returned None after connect; falling back to requested"); + *config + } + }; + let source = if lora_configs_match(&active_config, config) { ConfigSource::Ours } else { ConfigSource::Mux }; + info!("[radio] config: source={source:?}, {}", format_lora_config(&active_config)); + + dongle.rx_start().await.context("rx_start failed")?; + if event_tx.send(RadioEvent::Connected(active_config, source, device.to_string())).await.is_err() { + return Ok(()); + } - // Set a short timeout so we can interleave RX and TX. - client.transport_mut().set_timeout(Duration::from_millis(100))?; info!("[radio] entering main loop"); - // Liveness: ping the mux periodically when idle to detect a dead connection. - // Without this, a killed mux causes recv() to silently return Ok(None) forever. - let liveness_interval = Duration::from_secs(2); - let mut last_activity = std::time::Instant::now(); - + // TX requests from the router and RX packets from the radio both drive + // the loop; use tokio::select! to multiplex them. loop { - // Check for TX requests (non-blocking). - while let Ok(payload) = tx_recv.try_recv() { - debug!("TX {} bytes", payload.len()); - if let Err(e) = client.transmit(&payload, None) { - warn!("transmit error: {e:#}"); - return Err(e); + tokio::select! { + tx = tx_recv.recv() => { + let Some(req) = tx else { return Ok(()); }; + handle_tx(&dongle, req, retry_policy, event_tx).await; } - last_activity = std::time::Instant::now(); - } - - // Check for RX packets. - match client.recv() { - Ok(Some(Response::RxPacket { rssi, snr, payload })) => { - last_activity = std::time::Instant::now(); - let grade = snr_grade(snr, active_config.sf); - if grade.should_forward() { - debug!("RX {} bytes rssi={rssi} snr={snr} grade={grade}", payload.len()); - let pkt = RadioPacket { rssi, snr, payload }; - if event_tx.blocking_send(RadioEvent::Packet(pkt)).is_err() { + rx = dongle.next_rx() => { + match rx { + Some(pkt) => { + handle_rx(pkt, &active_config, event_tx).await; + } + None => { + // Dongle's RX stream ended — session closed. return Ok(()); } - } else { - debug!("RX drop {} bytes rssi={rssi} snr={snr} grade={grade}", payload.len()); } } - Ok(Some(_)) => { - last_activity = std::time::Instant::now(); - } - Ok(None) => {} - Err(e) if is_timeout_error(&e) => {} - Err(e) => return Err(e), - } - - // Liveness check: if no real activity for a while, ping the mux. - if last_activity.elapsed() >= liveness_interval { - // Temporarily extend the timeout for the ping command. - let _ = client.transport_mut().set_timeout(Duration::from_secs(2)); - match client.ping() { - Ok(()) => { - last_activity = std::time::Instant::now(); - } - Err(e) => { - warn!("liveness ping failed: {e:#}"); - return Err(e); - } - } - // Restore the short timeout. - let _ = client.transport_mut().set_timeout(Duration::from_millis(100)); } if event_tx.is_closed() { return Ok(()); } + + // Pull any async errors the Dongle accumulated (EFRAME / async + // ERADIO). These get logged but don't kill the session. + for err in dongle.drain_async_errors().await { + warn!("[radio] async error: {err}"); + } + + // Intentionally unused so callers don't think is_mux matters here. + let _ = is_mux; } } -/// Negotiate radio config with the dongle/mux. -/// -/// Strategy: -/// 1. `GetConfig` first to read what's active. -/// 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( - client: &mut donglora_client::Client, - desired: &RadioConfig, - is_mux: bool, -) -> 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 is_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)) +async fn handle_tx(dongle: &Dongle, req: TxRequest, retry_policy: &RetryPolicy, event_tx: &mpsc::Sender) { + let size = req.data.len(); + let _ = event_tx + .send(RadioEvent::TxAttempt { seq: req.seq, attempt: 1, total_attempts: retry_policy.max_attempts, size }) + .await; + + let result = dongle.tx_with_retry(&req.data, retry_policy).await; + match result { + Ok(outcome) => { + // Fire a TxRetry event for every attempt beyond the first, so the + // TUI log can show the escalation. + let mut prev_attempt: u8 = 0; + for att in &outcome.attempts { + if prev_attempt != 0 { + let reason = match &outcome.attempts[(prev_attempt - 1) as usize].result { + Err(donglora_client::ClientError::ChannelBusy) => TxRetryReason::ChannelBusy, + Err(donglora_client::ClientError::Busy) => TxRetryReason::QueueFull, + _ => TxRetryReason::Other, + }; + let _ = event_tx + .send(RadioEvent::TxRetry { + seq: req.seq, + attempt_that_failed: prev_attempt, + total_attempts: retry_policy.max_attempts, + reason, + backoff_ms: retry_policy.backoff_ms_min, + }) + .await; + } + prev_attempt = att.attempt; } + let _ = event_tx + .send(RadioEvent::TxSucceeded { + seq: req.seq, + attempts_used: outcome.attempts_used(), + final_airtime_us: outcome.final_airtime_us, + size, + }) + .await; } Err(e) => { - info!("[negotiate] get_config FAILED: {e:#}"); - // No config on radio — try to set ours. - info!("[negotiate] trying set_config..."); - match client.set_config(*desired) { - Ok(()) => { - info!("[negotiate] set_config OK"); - Ok((*desired, ConfigSource::Ours)) - } - Err(e2) => { - info!("[negotiate] set_config FAILED: {e2:#}"); - anyhow::bail!("GetConfig failed ({e:#}) and SetConfig also failed ({e2:#})"); - } - } + warn!("TX #{} failed after retries: {e:#}", req.seq); + let _ = event_tx + .send(RadioEvent::TxFailed { + seq: req.seq, + attempts_used: retry_policy.max_attempts, + reason: format!("{e}"), + size, + }) + .await; } } } -/// Shorten a device path to just the filename for display. -fn shorten_path(path: &str) -> String { - std::path::Path::new(path).file_name().map_or_else(|| path.to_string(), |f| f.to_string_lossy().into_owned()) +async fn handle_rx(pkt: RxPayload, active_config: &LoRaConfig, event_tx: &mpsc::Sender) { + // Tenths-of-dB → whole dB for the existing bridge packet format. + // `rssi_tenths_dbm` and `snr_tenths_db` are i16 and the values stay + // within i16 after the /10 rounding; the `as i16` is safe. + #[allow(clippy::cast_possible_truncation)] + let rssi = (f32::from(pkt.rssi_tenths_dbm) / 10.0).round() as i16; + #[allow(clippy::cast_possible_truncation)] + let snr = (f32::from(pkt.snr_tenths_db) / 10.0).round() as i16; + + let grade = snr_grade(snr, active_config.sf); + if !grade.should_forward() { + debug!("RX drop {} bytes rssi={rssi} snr={snr} grade={grade}", pkt.data.len()); + return; + } + debug!("RX {} bytes rssi={rssi} snr={snr} grade={grade}", pkt.data.len()); + let radio_pkt = RadioPacket { rssi, snr, payload: pkt.data.to_vec() }; + let _ = event_tx.send(RadioEvent::Packet(radio_pkt)).await; } -fn configs_match(a: &RadioConfig, b: &RadioConfig) -> bool { +// ── Display helpers ─────────────────────────────────────────────── + +fn lora_configs_match(a: &LoRaConfig, b: &LoRaConfig) -> bool { a.freq_hz == b.freq_hz && a.bw == b.bw && a.sf == b.sf @@ -275,46 +328,47 @@ fn configs_match(a: &RadioConfig, b: &RadioConfig) -> bool { && a.sync_word == b.sync_word && a.tx_power_dbm == b.tx_power_dbm && a.preamble_len == b.preamble_len - && a.cad == b.cad -} - -fn is_timeout_error(e: &anyhow::Error) -> bool { - e.downcast_ref::() - .is_some_and(|io_err| matches!(io_err.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut)) + && a.header_mode == b.header_mode + && a.payload_crc == b.payload_crc + && a.iq_invert == b.iq_invert } /// Format a bandwidth value for display. #[must_use] -pub const fn format_bandwidth(bw: donglora_client::Bandwidth) -> &'static str { +pub const fn format_bandwidth(bw: LoRaBandwidth) -> &'static str { match bw { - donglora_client::Bandwidth::Khz7 => "7.8 kHz", - donglora_client::Bandwidth::Khz10 => "10.4 kHz", - donglora_client::Bandwidth::Khz15 => "15.6 kHz", - donglora_client::Bandwidth::Khz20 => "20.8 kHz", - donglora_client::Bandwidth::Khz31 => "31.25 kHz", - donglora_client::Bandwidth::Khz41 => "41.7 kHz", - donglora_client::Bandwidth::Khz62 => "62.5 kHz", - donglora_client::Bandwidth::Khz125 => "125 kHz", - donglora_client::Bandwidth::Khz250 => "250 kHz", - donglora_client::Bandwidth::Khz500 => "500 kHz", + LoRaBandwidth::Khz7 => "7.8 kHz", + LoRaBandwidth::Khz10 => "10.4 kHz", + LoRaBandwidth::Khz15 => "15.6 kHz", + LoRaBandwidth::Khz20 => "20.8 kHz", + LoRaBandwidth::Khz31 => "31.25 kHz", + LoRaBandwidth::Khz41 => "41.7 kHz", + LoRaBandwidth::Khz62 => "62.5 kHz", + LoRaBandwidth::Khz125 => "125 kHz", + LoRaBandwidth::Khz250 => "250 kHz", + LoRaBandwidth::Khz500 => "500 kHz", + LoRaBandwidth::Khz200 => "200 kHz", + LoRaBandwidth::Khz400 => "400 kHz", + LoRaBandwidth::Khz800 => "800 kHz", + LoRaBandwidth::Khz1600 => "1600 kHz", } } -/// Format a `RadioConfig` for display. +/// Format a `LoRaConfig` for display. #[must_use] -pub fn format_radio_config(config: &RadioConfig) -> String { +pub fn format_lora_config(config: &LoRaConfig) -> String { let freq_mhz = f64::from(config.freq_hz) / 1_000_000.0; let bw = format_bandwidth(config.bw); - let power = if config.tx_power_dbm == donglora_client::TX_POWER_MAX { - "max".to_string() - } else { - format!("{} dBm", config.tx_power_dbm) + let cr_denom: u8 = match config.cr { + donglora_client::LoRaCodingRate::Cr4_5 => 5, + donglora_client::LoRaCodingRate::Cr4_6 => 6, + donglora_client::LoRaCodingRate::Cr4_7 => 7, + donglora_client::LoRaCodingRate::Cr4_8 => 8, }; - let preamble = if config.preamble_len == 0 { 16 } else { config.preamble_len }; - let cad = if config.cad != 0 { "on" } else { "off" }; + let preamble = config.preamble_len; format!( - "{freq_mhz:.3}MHz SF{} BW{bw} CR4/{} SW0x{:04X} TX:{power} Pre:{preamble} CAD:{cad}", - config.sf, config.cr, config.sync_word + "{freq_mhz:.3}MHz SF{} BW{bw} CR4/{cr_denom} SW0x{:04X} TX:{} dBm Pre:{preamble}", + config.sf, config.sync_word, config.tx_power_dbm, ) } @@ -322,134 +376,79 @@ pub fn format_radio_config(config: &RadioConfig) -> String { #[allow(clippy::unwrap_used, clippy::expect_used)] mod tests { use super::*; + use donglora_client::{LoRaCodingRate, LoRaHeaderMode}; - fn default_config() -> RadioConfig { - RadioConfig { + fn default_config() -> LoRaConfig { + LoRaConfig { freq_hz: 910_525_000, - bw: donglora_client::Bandwidth::Khz62, + bw: LoRaBandwidth::Khz62, sf: 7, - cr: 5, - sync_word: 0x3444, - tx_power_dbm: 22, + cr: LoRaCodingRate::Cr4_5, + sync_word: 0x1424, + tx_power_dbm: 20, preamble_len: 16, - cad: 1, + header_mode: LoRaHeaderMode::Explicit, + payload_crc: true, + iq_invert: false, } } #[test] - fn configs_match_identical() { - let a = default_config(); - let b = default_config(); - assert!(configs_match(&a, &b)); + fn lora_configs_match_identical() { + assert!(lora_configs_match(&default_config(), &default_config())); } #[test] - fn configs_match_differs_each_field() { + fn lora_configs_match_differs_each_field() { let base = default_config(); let mut c = base; c.freq_hz = 915_000_000; - assert!(!configs_match(&base, &c)); + assert!(!lora_configs_match(&base, &c)); let mut c = base; c.sf = 12; - assert!(!configs_match(&base, &c)); + assert!(!lora_configs_match(&base, &c)); let mut c = base; - c.cr = 8; - assert!(!configs_match(&base, &c)); + c.cr = LoRaCodingRate::Cr4_8; + assert!(!lora_configs_match(&base, &c)); let mut c = base; c.sync_word = 0x1234; - assert!(!configs_match(&base, &c)); + assert!(!lora_configs_match(&base, &c)); let mut c = base; c.tx_power_dbm = 10; - assert!(!configs_match(&base, &c)); + assert!(!lora_configs_match(&base, &c)); let mut c = base; c.preamble_len = 8; - assert!(!configs_match(&base, &c)); - - let mut c = base; - c.cad = 0; - assert!(!configs_match(&base, &c)); - } - - #[test] - fn is_timeout_would_block() { - let e = anyhow::Error::from(std::io::Error::new(std::io::ErrorKind::WouldBlock, "test")); - assert!(is_timeout_error(&e)); - } - - #[test] - fn is_timeout_timed_out() { - let e = anyhow::Error::from(std::io::Error::new(std::io::ErrorKind::TimedOut, "test")); - assert!(is_timeout_error(&e)); - } - - #[test] - fn is_timeout_false_for_other() { - let e = anyhow::Error::from(std::io::Error::new(std::io::ErrorKind::ConnectionReset, "test")); - assert!(!is_timeout_error(&e)); - } - - #[test] - fn shorten_path_extracts_filename() { - assert_eq!(shorten_path("/dev/ttyACM0"), "ttyACM0"); - assert_eq!(shorten_path("ttyACM0"), "ttyACM0"); + assert!(!lora_configs_match(&base, &c)); } #[test] fn format_bandwidth_all_variants() { - assert_eq!(format_bandwidth(donglora_client::Bandwidth::Khz7), "7.8 kHz"); - assert_eq!(format_bandwidth(donglora_client::Bandwidth::Khz125), "125 kHz"); - assert_eq!(format_bandwidth(donglora_client::Bandwidth::Khz500), "500 kHz"); + assert_eq!(format_bandwidth(LoRaBandwidth::Khz7), "7.8 kHz"); + assert_eq!(format_bandwidth(LoRaBandwidth::Khz125), "125 kHz"); + assert_eq!(format_bandwidth(LoRaBandwidth::Khz500), "500 kHz"); } #[test] - fn format_radio_config_typical() { + fn format_lora_config_typical() { let cfg = default_config(); - let s = format_radio_config(&cfg); + let s = format_lora_config(&cfg); assert!(s.contains("910.525MHz")); assert!(s.contains("SF7")); assert!(s.contains("CR4/5")); - assert!(s.contains("0x3444")); - assert!(s.contains("CAD:on"), "cad=1 should show CAD:on: {s}"); + assert!(s.contains("0x1424")); + assert!(s.contains("20 dBm")); } #[test] - fn format_radio_config_cad_off() { - let mut cfg = default_config(); - cfg.cad = 0; - let s = format_radio_config(&cfg); - assert!(s.contains("CAD:off"), "cad=0 should show CAD:off: {s}"); - } - - #[test] - fn format_radio_config_max_power() { - let mut cfg = default_config(); - cfg.tx_power_dbm = donglora_client::TX_POWER_MAX; - let s = format_radio_config(&cfg); - assert!(s.contains("TX:max")); - assert!(!s.contains("dBm"), "max power should not contain 'dBm': {s}"); - } - - #[test] - fn format_radio_config_non_max_power() { - let mut cfg = default_config(); - cfg.tx_power_dbm = 22; - assert_ne!(cfg.tx_power_dbm, donglora_client::TX_POWER_MAX); - let s = format_radio_config(&cfg); - assert!(s.contains("22 dBm"), "non-max power should contain '22 dBm': {s}"); - assert!(!s.contains("max"), "non-max power should not contain 'max': {s}"); - } - - #[test] - fn format_radio_config_zero_preamble() { - let mut cfg = default_config(); - cfg.preamble_len = 0; - let s = format_radio_config(&cfg); - assert!(s.contains("Pre:16")); + 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"); } } diff --git a/src/rate_limit.rs b/src/rate_limit.rs index cd62021..40480ca 100644 --- a/src/rate_limit.rs +++ b/src/rate_limit.rs @@ -5,7 +5,7 @@ use std::time::{Duration, Instant}; -use donglora_client::Bandwidth; +use donglora_client::LoRaBandwidth; /// 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: Bandwidth, cr: u8, preamble: u16, override_pps: Option) -> Self { + pub fn from_radio_config(sf: u8, bw: LoRaBandwidth, cr: u8, preamble: u16, override_pps: Option) -> 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: Bandwidth, cr: u8, preamble: u16) -> Duration { +pub fn lora_air_time(payload_bytes: usize, sf: u8, bw: LoRaBandwidth, cr: u8, preamble: u16) -> Duration { let sf_f = f64::from(sf); let bw_hz = bandwidth_hz(bw); let preamble_symbols = if preamble == 0 { 16.0 } else { f64::from(preamble) }; @@ -108,18 +108,25 @@ pub fn lora_air_time(payload_bytes: usize, sf: u8, bw: Bandwidth, cr: u8, preamb Duration::from_secs_f64(total) } -const fn bandwidth_hz(bw: Bandwidth) -> f64 { +const fn bandwidth_hz(bw: LoRaBandwidth) -> f64 { match bw { - Bandwidth::Khz7 => 7_800.0, - Bandwidth::Khz10 => 10_400.0, - Bandwidth::Khz15 => 15_600.0, - Bandwidth::Khz20 => 20_800.0, - Bandwidth::Khz31 => 31_250.0, - Bandwidth::Khz41 => 41_700.0, - Bandwidth::Khz62 => 62_500.0, - Bandwidth::Khz125 => 125_000.0, - Bandwidth::Khz250 => 250_000.0, - Bandwidth::Khz500 => 500_000.0, + LoRaBandwidth::Khz7 => 7_800.0, + LoRaBandwidth::Khz10 => 10_400.0, + LoRaBandwidth::Khz15 => 15_600.0, + LoRaBandwidth::Khz20 => 20_800.0, + LoRaBandwidth::Khz31 => 31_250.0, + LoRaBandwidth::Khz41 => 41_700.0, + LoRaBandwidth::Khz62 => 62_500.0, + LoRaBandwidth::Khz125 => 125_000.0, + LoRaBandwidth::Khz250 => 250_000.0, + LoRaBandwidth::Khz500 => 500_000.0, + // SX128x 2.4 GHz bandwidths — not used on the bridge today, but + // plumb through so SET_CONFIG with one of these still yields a + // usable rate rather than panicking. + LoRaBandwidth::Khz200 => 200_000.0, + LoRaBandwidth::Khz400 => 400_000.0, + LoRaBandwidth::Khz800 => 800_000.0, + LoRaBandwidth::Khz1600 => 1_600_000.0, } } @@ -129,7 +136,7 @@ mod tests { #[test] fn air_time_sf7_bw125() { - let t = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16); + let t = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 16); // SF7/125kHz/CR4/5/16-preamble/50 bytes: ~100ms range assert!(t.as_millis() > 50, "air time too short: {t:?}"); assert!(t.as_millis() < 300, "air time too long: {t:?}"); @@ -137,7 +144,7 @@ mod tests { #[test] fn air_time_sf12_bw125() { - let t = lora_air_time(50, 12, Bandwidth::Khz125, 5, 16); + let t = lora_air_time(50, 12, LoRaBandwidth::Khz125, 5, 16); // SF12/125kHz: should be multiple seconds assert!(t.as_secs() >= 1, "air time too short: {t:?}"); assert!(t.as_secs() < 10, "air time too long: {t:?}"); @@ -145,7 +152,7 @@ mod tests { #[test] fn rate_limiter_allows_burst() { - let mut rl = RateLimiter::from_radio_config(7, Bandwidth::Khz125, 5, 16, None); + let mut rl = RateLimiter::from_radio_config(7, LoRaBandwidth::Khz125, 5, 16, None); // Should allow burst of 3 assert!(rl.try_acquire()); assert!(rl.try_acquire()); @@ -156,14 +163,14 @@ mod tests { #[test] fn rate_limiter_override() { - let rl = RateLimiter::from_radio_config(7, Bandwidth::Khz125, 5, 16, Some(42.0)); + let rl = RateLimiter::from_radio_config(7, LoRaBandwidth::Khz125, 5, 16, Some(42.0)); assert!((rl.rate_pps() - 42.0).abs() < f64::EPSILON); } #[test] fn air_time_zero_preamble_uses_default() { - let t_default = lora_air_time(50, 7, Bandwidth::Khz125, 5, 0); - let t_explicit = lora_air_time(50, 7, Bandwidth::Khz125, 5, 16); + let t_default = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 0); + let t_explicit = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 16); assert_eq!(t_default, t_explicit); } @@ -173,7 +180,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, Bandwidth::Khz125, 5, 16, None); + let rl = RateLimiter::from_radio_config(7, LoRaBandwidth::Khz125, 5, 16, None); let rate = rl.rate_pps(); // Must be between 0.1 and 100 (rules out fallback of 10.0 indirectly, // and catches if secs<=0 path is taken or division is wrong). @@ -189,7 +196,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, Bandwidth::Khz125, 5, 16, None); + let rl = RateLimiter::from_radio_config(12, LoRaBandwidth::Khz125, 5, 16, None); let rate = rl.rate_pps(); assert!(rate > 0.1, "rate too low: {rate}"); assert!(rate < 0.5, "rate too high: {rate}"); @@ -202,7 +209,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, Bandwidth::Khz125, 5, 16, Some(10.0)); + let mut rl = RateLimiter::from_radio_config(7, LoRaBandwidth::Khz125, 5, 16, Some(10.0)); // Exhaust all burst tokens. assert!(rl.try_acquire()); assert!(rl.try_acquire()); @@ -222,7 +229,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, Bandwidth::Khz125, 5, 16); + let t = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 16); let ms = t.as_secs_f64() * 1000.0; assert!((ms - 105.728).abs() < 0.01, "SF7/BW125 air time should be ~105.728ms, got {ms:.3}ms"); } @@ -231,7 +238,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, Bandwidth::Khz125, 5, 16); + let t = lora_air_time(50, 12, LoRaBandwidth::Khz125, 5, 16); let ms = t.as_secs_f64() * 1000.0; assert!((ms - 2564.096).abs() < 0.01, "SF12/BW125 air time should be ~2564.096ms, got {ms:.3}ms"); } @@ -240,8 +247,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, Bandwidth::Khz125, 5, 16); - let t10 = lora_air_time(50, 10, Bandwidth::Khz125, 5, 16); + let t11 = lora_air_time(50, 11, LoRaBandwidth::Khz125, 5, 16); + let t10 = lora_air_time(50, 10, LoRaBandwidth::Khz125, 5, 16); let ms11 = t11.as_secs_f64() * 1000.0; let ms10 = t10.as_secs_f64() * 1000.0; @@ -259,12 +266,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, Bandwidth::Khz250, 5, 16); + let t11_250 = lora_air_time(50, 11, LoRaBandwidth::Khz250, 5, 16); let ms = t11_250.as_secs_f64() * 1000.0; assert!((ms - 641.024).abs() < 0.01, "SF11/BW250 should be ~641.024ms (no LDRO), got {ms:.3}ms"); // Compare with SF11/BW125 which HAS LDRO. - let t11_125 = lora_air_time(50, 11, Bandwidth::Khz125, 5, 16); + let t11_125 = lora_air_time(50, 11, LoRaBandwidth::Khz125, 5, 16); // Even accounting for halved BW doubling the base time, the LDRO effect // on SF11/BW125 makes it much longer than simply 2x SF11/BW250. assert!(t11_125 > t11_250 * 2, "SF11/BW125 (LDRO on) should be >2x SF11/BW250 (LDRO off)"); @@ -274,9 +281,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, 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 t8 = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 8); + let t16 = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 16); + let t32 = lora_air_time(50, 7, LoRaBandwidth::Khz125, 5, 32); let diff_8_to_16 = t16.as_secs_f64() - t8.as_secs_f64(); let diff_16_to_32 = t32.as_secs_f64() - t16.as_secs_f64(); @@ -297,16 +304,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 = [ - Bandwidth::Khz7, - Bandwidth::Khz10, - Bandwidth::Khz15, - Bandwidth::Khz20, - Bandwidth::Khz31, - Bandwidth::Khz41, - Bandwidth::Khz62, - Bandwidth::Khz125, - Bandwidth::Khz250, - Bandwidth::Khz500, + LoRaBandwidth::Khz7, + LoRaBandwidth::Khz10, + LoRaBandwidth::Khz15, + LoRaBandwidth::Khz20, + LoRaBandwidth::Khz31, + LoRaBandwidth::Khz41, + LoRaBandwidth::Khz62, + LoRaBandwidth::Khz125, + LoRaBandwidth::Khz250, + LoRaBandwidth::Khz500, ]; for sf in 7..=12u8 { for &bw in &bandwidths { @@ -332,7 +339,7 @@ mod tests { cr in 5u8..=8u8, preamble in 0u16..=64u16, ) { - for bw in [Bandwidth::Khz7, Bandwidth::Khz62, Bandwidth::Khz125, Bandwidth::Khz500] { + for bw in [LoRaBandwidth::Khz7, LoRaBandwidth::Khz62, LoRaBandwidth::Khz125, LoRaBandwidth::Khz500] { let t = lora_air_time(payload, sf, bw, cr, preamble); prop_assert!(t > Duration::ZERO, "air time must be positive"); prop_assert!(t < Duration::from_secs(600), "air time should be < 600s"); diff --git a/src/router.rs b/src/router.rs index a43799f..fc892a1 100644 --- a/src/router.rs +++ b/src/router.rs @@ -13,14 +13,14 @@ use tracing::{debug, warn}; use crate::gossip::GossipEvent; use crate::packet::{GossipFrame, content_hash}; -use crate::radio::{ConfigSource, RadioEvent}; +use crate::radio::{ConfigSource, RadioEvent, TxRequest, TxRetryReason}; use crate::rate_limit::RateLimiter; /// Active radio config info, sent to the TUI via watch channel. #[derive(Debug, Clone)] pub struct RadioConfigInfo { - pub active: donglora_client::RadioConfig, - pub requested: donglora_client::RadioConfig, + pub active: donglora_client::LoRaConfig, + pub requested: donglora_client::LoRaConfig, pub source: ConfigSource, pub device: String, pub connected: bool, @@ -36,6 +36,29 @@ pub struct PacketLogEntry { pub snr: Option, pub rssi: Option, pub action: PacketAction, + /// When set, indicates this entry is a TX-attempt update: `Some((attempt, total))`. + pub tx_retry: Option, +} + +/// 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. @@ -83,6 +106,11 @@ 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 { @@ -98,6 +126,8 @@ 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), } } } @@ -114,6 +144,8 @@ pub struct StatsSnapshot { pub dropped_queue: u64, pub neighbor_count: u64, pub radio_connected: bool, + pub tx_retries: u64, + pub tx_failures: u64, } /// Time-bounded dedup cache. @@ -153,10 +185,10 @@ pub async fn run( our_id: PublicKey, dedup_window: Duration, tx_queue_size: usize, - requested_radio_config: donglora_client::RadioConfig, + requested_radio_config: donglora_client::LoRaConfig, mut rate_limiter: RateLimiter, mut radio_rx: mpsc::Receiver, - radio_tx: mpsc::Sender>, + radio_tx: mpsc::Sender, mut gossip_rx: mpsc::Receiver, gossip_tx: mpsc::Sender, stats: &Stats, @@ -164,6 +196,7 @@ pub async fn run( config_tx: tokio::sync::watch::Sender, ) { let mut dedup = DedupCache::new(dedup_window); + let mut next_tx_seq: u64 = 0; let mut tx_queue: VecDeque> = VecDeque::with_capacity(tx_queue_size); let mut last_cleanup = Instant::now(); let cleanup_interval = Duration::from_secs(30); @@ -195,6 +228,7 @@ pub async fn run( snr: Some(pkt.snr), rssi: Some(pkt.rssi), action: PacketAction::DroppedDedup, + tx_retry: None, }).await; continue; } @@ -212,6 +246,7 @@ pub async fn run( snr: Some(pkt.snr), rssi: Some(pkt.rssi), action: PacketAction::Bridged, + tx_retry: None, }).await; } RadioEvent::Connected(active_config, source, device) => { @@ -228,6 +263,67 @@ pub async fn run( stats.radio_connected.store(0, Ordering::Relaxed); config_tx.send_modify(|info| info.connected = false); } + RadioEvent::TxAttempt { .. } => { + // No log entry: the GossipIn Bridged row that enqueued + // this TX already represents it. Only retries / + // failures produce follow-up rows. + } + RadioEvent::TxRetry { seq, attempt_that_failed, total_attempts, reason, backoff_ms: _ } => { + stats.tx_retries.fetch_add(1, Ordering::Relaxed); + let _ = log_tx.send(PacketLogEntry { + timestamp: Instant::now(), + hash: tx_seq_hash(seq), + direction: PacketDirection::GossipIn, + size: 0, + snr: None, + rssi: None, + action: PacketAction::Bridged, + tx_retry: Some(TxRetryInfo { + attempt: attempt_that_failed.saturating_add(1), + total_attempts, + state: TxRetryState::Retrying(reason), + }), + }).await; + } + RadioEvent::TxSucceeded { seq, attempts_used, final_airtime_us: _, size } => { + stats.radio_tx.fetch_add(1, Ordering::Relaxed); + // Only emit a success row if retries were involved — + // otherwise the original GossipIn row covers it. + if attempts_used > 1 { + let _ = log_tx.send(PacketLogEntry { + timestamp: Instant::now(), + hash: tx_seq_hash(seq), + direction: PacketDirection::GossipIn, + size, + snr: None, + rssi: None, + action: PacketAction::Bridged, + tx_retry: Some(TxRetryInfo { + attempt: attempts_used, + total_attempts: attempts_used, + state: TxRetryState::Succeeded, + }), + }).await; + } + } + RadioEvent::TxFailed { seq, attempts_used, reason, size } => { + stats.tx_failures.fetch_add(1, Ordering::Relaxed); + warn!("TX #{seq} exhausted retries: {reason}"); + let _ = log_tx.send(PacketLogEntry { + timestamp: Instant::now(), + hash: tx_seq_hash(seq), + direction: PacketDirection::GossipIn, + size, + snr: None, + rssi: None, + action: PacketAction::Bridged, + tx_retry: Some(TxRetryInfo { + attempt: attempts_used, + total_attempts: attempts_used, + state: TxRetryState::Failed, + }), + }).await; + } } } @@ -250,6 +346,7 @@ pub async fn run( snr: Some(frame.snr), rssi: Some(frame.rssi), action: PacketAction::DroppedDedup, + tx_retry: None, }).await; continue; } @@ -266,6 +363,7 @@ pub async fn run( snr: Some(frame.snr), rssi: Some(frame.rssi), action: PacketAction::DroppedRateLimit, + tx_retry: None, }).await; continue; } @@ -282,6 +380,7 @@ pub async fn run( snr: Some(frame.snr), rssi: Some(frame.rssi), action: PacketAction::DroppedQueueFull, + tx_retry: None, }).await; continue; } @@ -295,6 +394,7 @@ pub async fn run( snr: Some(frame.snr), rssi: Some(frame.rssi), action: PacketAction::Bridged, + tx_retry: None, }).await; } GossipEvent::NeighborChanged(count) => { @@ -305,16 +405,39 @@ pub async fn run( } } - // Drain TX queue to radio. + // Drain TX queue to the async radio task. radio_tx is a bounded + // channel; if the radio task is busy we just wait on the next + // loop iteration rather than blocking here. while let Some(payload) = tx_queue.pop_front() { - stats.radio_tx.fetch_add(1, Ordering::Relaxed); - if radio_tx.send(payload).await.is_err() { + let seq = { + next_tx_seq = next_tx_seq.wrapping_add(1); + next_tx_seq + }; + if radio_tx.send(TxRequest { seq, data: payload }).await.is_err() { break; } } } } +/// Build a stable per-TX "hash" for packet-log correlation. Not a real +/// content hash — just a 32-byte encoding of the seq so the TUI can +/// index/update log rows by it. +const fn tx_seq_hash(seq: u64) -> [u8; 32] { + let mut out = [0u8; 32]; + out[0] = 0x01; // tag marking this as a TX-seq pseudo-hash + let bytes = seq.to_le_bytes(); + out[1] = bytes[0]; + out[2] = bytes[1]; + out[3] = bytes[2]; + out[4] = bytes[3]; + out[5] = bytes[4]; + out[6] = bytes[5]; + out[7] = bytes[6]; + out[8] = bytes[7]; + out +} + #[cfg(test)] #[allow(clippy::unwrap_used, clippy::expect_used)] mod tests { diff --git a/src/setup.rs b/src/setup.rs index 5326985..9a2de45 100644 --- a/src/setup.rs +++ b/src/setup.rs @@ -58,15 +58,12 @@ pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> { let sync_word_str = prompt_default("Sync Word", &sw_default)?; let sync_word = parse_sync_word(&sync_word_str)?; - let tx_power = prompt_default("TX Power (dBm or 'max')", &radio.tx_power)?; + let tx_power_default = format!("{}", radio.tx_power_dbm); + let tx_power_dbm: i8 = prompt_parse("TX Power (dBm)", &tx_power_default)?; let pre_default = format!("{}", radio.preamble); let preamble: u16 = prompt_parse("Preamble", &pre_default)?; - let cad_default = if radio.cad { "on" } else { "off" }; - let cad_str = prompt_default("CAD", cad_default)?; - let cad = matches!(cad_str.to_lowercase().as_str(), "on" | "true" | "yes" | "1"); - let port_default = radio.port.as_deref().unwrap_or("auto"); let port_str = prompt_default("Serial Port", port_default)?; let port = if port_str == "auto" || port_str.is_empty() { None } else { Some(port_str) }; @@ -81,9 +78,8 @@ pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> { spreading_factor, coding_rate, sync_word, - tx_power, + tx_power_dbm, preamble, - cad, }, bridge: BridgeSection { passphrase, @@ -92,6 +88,7 @@ pub fn run_wizard(config_path: &Path, existing: Option<&Config>) -> Result<()> { rate_limit_pps: bridge.rate_limit_pps, log_file: bridge.log_file, }, + tx: existing.map(|c| c.tx.clone()).unwrap_or_default(), }; write_config(config_path, &cfg)?; diff --git a/src/tui.rs b/src/tui.rs index eef9d87..3ef59bb 100644 --- a/src/tui.rs +++ b/src/tui.rs @@ -23,7 +23,9 @@ use tokio_util::sync::CancellationToken; use crate::gossip::Gossip; use crate::radio::ConfigSource; -use crate::router::{PacketAction, PacketDirection, PacketLogEntry, RadioConfigInfo, Stats, StatsSnapshot}; +use crate::router::{ + PacketAction, PacketDirection, PacketLogEntry, RadioConfigInfo, Stats, StatsSnapshot, TxRetryState, +}; const TICK_RATE: Duration = Duration::from_millis(250); const MAX_LOG_ENTRIES: usize = 200; @@ -366,13 +368,16 @@ fn draw_radio(frame: &mut ratatui::Frame, area: Rect, config_info: &RadioConfigI let freq_mhz = f64::from(a.freq_hz) / 1_000_000.0; let bw_str = crate::radio::format_bandwidth(a.bw); - let power_str = if a.tx_power_dbm == donglora_client::TX_POWER_MAX { - "max".to_string() - } else { - format!("{} dBm", a.tx_power_dbm) - }; + let power_str = 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" }; + let cr_denom = |c: donglora_client::LoRaCodingRate| -> u8 { + match c { + donglora_client::LoRaCodingRate::Cr4_5 => 5, + donglora_client::LoRaCodingRate::Cr4_6 => 6, + donglora_client::LoRaCodingRate::Cr4_7 => 7, + donglora_client::LoRaCodingRate::Cr4_8 => 8, + } + }; // Dim values when disconnected; highlight mux mismatches when connected. let val_color = if connected { C_ACCENT } else { C_LABEL }; @@ -394,11 +399,10 @@ fn draw_radio(frame: &mut ratatui::Frame, area: Rect, config_info: &RadioConfigI cf("Freq", format!("{freq_mhz:.3} MHz"), a.freq_hz == r.freq_hz), cf("BW", bw_str.to_string(), a.bw == r.bw), cf("SF", format!("{}", a.sf), a.sf == r.sf), - cf("CR", format!("4/{}", a.cr), a.cr == r.cr), + cf("CR", format!("4/{}", cr_denom(a.cr)), a.cr == r.cr), cf("Sync", format!("0x{:04X}", a.sync_word), a.sync_word == r.sync_word), cf("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); } @@ -430,6 +434,11 @@ 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); } @@ -447,6 +456,11 @@ fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry]) // Fixed column widths: Age=5 Hash=8 RF=4 Net=4 Size=6 RSSI=6 SNR=5 Act=5 // 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), @@ -488,11 +502,35 @@ fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry]) ), }; - let action_color = match e.action { - PacketAction::Bridged => C_LABEL, - PacketAction::DroppedDedup => C_DEDUP, - PacketAction::DroppedQueueFull => C_QUEUE_DROP, - PacketAction::DroppedRateLimit => C_RATE_DROP, + // 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 rssi_span = match e.rssi { @@ -504,13 +542,9 @@ fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry]) None => Span::styled(format!("{:>5}", "-"), Style::default().fg(C_LABEL)), }; - 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)); + // 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 size_str = format!("{}B", e.size); lines.push(Line::from(vec![ @@ -700,6 +734,8 @@ mod tests { dropped_queue: 0, neighbor_count: 0, radio_connected: true, + tx_retries: 0, + tx_failures: 0, } } @@ -712,6 +748,7 @@ mod tests { snr, rssi: Some(-80), action: PacketAction::Bridged, + tx_retry: None, } }