From b4026ace34ea281e17a3cceaaec514ae43216a13 Mon Sep 17 00:00:00 2001 From: Stephen Waits Date: Sun, 5 Apr 2026 19:03:06 -0600 Subject: [PATCH] feat: overhaul radio reconnection, TUI layout, DHT strategy, and signal handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Radio reconnection: - Add connect_mux_auto() to donglora-client — mux-only connect that never silently falls back to USB serial (matches Python client's sticky mux) - Fix codec.rs read_frame() to handle WouldBlock alongside TimedOut - Add liveness ping (2s) to detect dead mux connections - Treat transmit errors as fatal (trigger reconnect) - Faster backoff: 250ms initial, 5s max, resets after stable connection TUI improvements: - Redesign packet log: RF+Net columns show full packet path (RX→TX), hash column with dedup ellipsis, fixed-width columns throughout - Move activity charts to right column (fills remaining width) - 1-second sparkline buckets instead of 10-second - Radio panel shows "(disconnected)" with red border when mux is down - Header: status pills left, clock with TZ right-justified, Ctrl-C quits - Local time via jiff crate with proper timezone abbreviation DHT strategy: - Alone: read every 3s, merge-publish every 30s (preserves existing peers) - Connected: publish authoritative [self + neighbors], purging stale peers - Publish immediately on first neighbor join Signal handling: - SIGINT and SIGTERM both trigger graceful shutdown via cancellation token - All paths go through swarm.shutdown() to update DHT before exit --- Cargo.lock | 51 ++++++++++++ Cargo.toml | 1 + src/gossip.rs | 135 ++++++++++++++++++++---------- src/main.rs | 27 +++--- src/radio.rs | 44 +++++++++- src/router.rs | 66 +++++++-------- src/tui.rs | 224 ++++++++++++++++++++++++++++++++------------------ 7 files changed, 373 insertions(+), 175 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 135e7c9..0d2b34c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -862,6 +862,7 @@ dependencies = [ "hex", "iroh", "iroh-gossip", + "jiff", "mainline", "rand", "ratatui", @@ -1984,6 +1985,47 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "jiff" +version = "0.2.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a3546dc96b6d42c5f24902af9e2538e82e39ad350b0c766eb3fbf2d8f3d8359" +dependencies = [ + "jiff-static", + "jiff-tzdb-platform", + "log", + "portable-atomic", + "portable-atomic-util", + "serde_core", + "windows-sys 0.61.2", +] + +[[package]] +name = "jiff-static" +version = "0.2.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a8c8b344124222efd714b73bb41f8b5120b27a7cc1c75593a6ff768d9d05aa4" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "jiff-tzdb" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c900ef84826f1338a557697dc8fc601df9ca9af4ac137c7fb61d4c6f2dfd3076" + +[[package]] +name = "jiff-tzdb-platform" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "875a5a69ac2bab1a891711cf5eccbec1ce0341ea805560dcd90b7a2e925132e8" +dependencies = [ + "jiff-tzdb", +] + [[package]] name = "js-sys" version = "0.3.94" @@ -2748,6 +2790,15 @@ dependencies = [ "serde", ] +[[package]] +name = "portable-atomic-util" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "091397be61a01d4be58e7841595bd4bfedb15f1cd54977d79b8271e94ed799a3" +dependencies = [ + "portable-atomic", +] + [[package]] name = "portmapper" version = "0.15.0" diff --git a/Cargo.toml b/Cargo.toml index 82c45ee..1765912 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,6 +29,7 @@ rand = "0.9" hex = "0.4.3" sha2 = "0.10" tokio-util = { version = "0.7.18", features = ["rt"] } +jiff = "0.2.23" [lints.clippy] unwrap_used = "deny" diff --git a/src/gossip.rs b/src/gossip.rs index 2beb654..95bbaed 100644 --- a/src/gossip.rs +++ b/src/gossip.rs @@ -361,12 +361,22 @@ fn spawn_broadcaster( }); } -/// Fast DHT interval while we have no neighbors. -const DHT_FAST_INTERVAL_SECS: u64 = 5; +/// DHT poll interval while alone (read frequently, publish infrequently). +const DHT_ALONE_READ_SECS: u64 = 3; -/// Periodically publish our peer list to the DHT and discover new peers. +/// How often a lonely node publishes (merge-only, never destructive). +/// Slower than reads — publishing is heavier and a lonely node only adds itself. +const DHT_ALONE_PUBLISH_SECS: u64 = 30; + +/// Periodically interact with the DHT to discover and share peers. /// -/// Interval is adaptive: fast (5s) while alone, slow (5 min) once connected. +/// 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, @@ -388,49 +398,82 @@ fn spawn_dht_heartbeat( }; let pub_key_bytes = signing_key.verifying_key().to_bytes(); + let mut was_alone = true; + let mut last_publish = Instant::now(); - // Initial publish immediately. + // 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 { - // Fast while alone, slow once we have neighbors. let alone = neighbor_count.load(Ordering::Relaxed) == 0; - let base_secs = if alone { - DHT_FAST_INTERVAL_SECS + + // 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 { - DHT_HEARTBEAT_SECS + let jitter = rand::random::() % DHT_HEARTBEAT_JITTER_SECS.max(1); + Duration::from_secs(DHT_HEARTBEAT_SECS + jitter) }; - let max_jitter = if alone { 2 } else { DHT_HEARTBEAT_JITTER_SECS }; - let jitter = rand::random::() % max_jitter.max(1); - let sleep_dur = Duration::from_secs(base_secs + jitter); tokio::select! { () = tokio::time::sleep(sleep_dur) => {} () = cancel.cancelled() => break, } - // GET: discover new peers and feed to gossip. - 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:#}"); - } - } - } + // Always read — cheap, discovers peers to join. + dht_read_and_join(&dht, &my_id, &pub_key_bytes, &salt, &gossip_sender, &state).await; - // PUT: merge-publish our info + neighbors + existing DHT peers. - dht_publish(&dht, &my_id, &signing_key, &salt, &pub_key_bytes, &neighbors, &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>, +) { + 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, @@ -440,30 +483,34 @@ async fn dht_publish( neighbors: &Arc>>, state: &Arc>, ) { - // Merge: ourselves + live neighbors + existing DHT peers (preserve others' data). + let has_neighbors = neighbors.lock().map(|n| !n.is_empty()).unwrap_or(false); + let mut seen = HashSet::new(); let mut peer_list: Vec<[u8; 32]> = Vec::with_capacity(MAX_DHT_PEERS); - // 1. Always include ourselves first. + // Always include ourselves. peer_list.push(*my_id.as_bytes()); seen.insert(*my_id.as_bytes()); - // 2. Add our live 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()); + 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()); + } } } - } - - // 3. Preserve existing DHT peers we don't already know about. - if peer_list.len() < MAX_DHT_PEERS - && 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()); + } 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()); + } } } } diff --git a/src/main.rs b/src/main.rs index 88495ab..f37c97a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -142,6 +142,7 @@ async fn main() -> Result<()> { requested: radio_config, source: radio::ConfigSource::Ours, device: String::new(), + connected: false, }, ); @@ -175,18 +176,25 @@ async fn main() -> Result<()> { router_cancel.cancel(); }); + // Spawn a task that cancels on ctrl-c or SIGTERM. + { + let cancel = cancel.clone(); + tokio::spawn(async move { + let mut sigterm = tokio::signal::unix::signal( + tokio::signal::unix::SignalKind::terminate(), + ).expect("failed to register SIGTERM handler"); + tokio::select! { + _ = tokio::signal::ctrl_c() => info!("received SIGINT"), + _ = sigterm.recv() => info!("received SIGTERM"), + } + cancel.cancel(); + }); + } + // Run TUI or headless. if cli.log_only { info!("running in headless mode"); - tokio::select! { - () = cancel.cancelled() => {}, - result = tokio::signal::ctrl_c() => { - if let Err(e) = result { - tracing::error!("signal handler error: {e}"); - } - info!("received ctrl-c, shutting down"); - } - } + cancel.cancelled().await; } else { let terminal = tui::run( config_watch_rx, @@ -210,7 +218,6 @@ async fn main() -> Result<()> { } // Shutdown (headless path). - cancel.cancel(); swarm.shutdown().await; info!("donglora-bridge stopped"); diff --git a/src/radio.rs b/src/radio.rs index dad813d..a952a4a 100644 --- a/src/radio.rs +++ b/src/radio.rs @@ -50,10 +50,12 @@ fn radio_loop( event_tx: mpsc::Sender, mut tx_recv: mpsc::Receiver>, ) { - let mut backoff = Duration::from_secs(1); - let max_backoff = Duration::from_secs(30); + let initial_backoff = Duration::from_millis(250); + let max_backoff = Duration::from_secs(5); + let mut backoff = initial_backoff; loop { + let started = std::time::Instant::now(); match connect_and_run(&config, &port, &event_tx, &mut tx_recv) { Ok(()) => { info!("radio thread exiting"); @@ -62,6 +64,13 @@ fn radio_loop( Err(e) => { error!("radio error: {e:#}"); let _ = event_tx.blocking_send(RadioEvent::Disconnected); + + // If we were connected for a while, reset backoff so + // the next reconnect attempt is fast. + if started.elapsed() > Duration::from_secs(5) { + backoff = initial_backoff; + } + info!("reconnecting in {backoff:?}"); std::thread::sleep(backoff); backoff = (backoff * 2).min(max_backoff); @@ -85,7 +94,7 @@ fn connect_and_run( let c = donglora_client::connect(Some(port), timeout)?; (c, shorten_path(port)) } else { - let c = donglora_client::connect_default()?; + let c = donglora_client::connect_mux_auto(timeout)?; (c, "mux".to_string()) }; info!("[radio] transport connected: {device}"); @@ -128,18 +137,26 @@ fn connect_and_run( .set_timeout(Duration::from_millis(100))?; info!("[radio] entering main loop"); + // Liveness: ping the mux periodically when idle to detect a dead connection. + // Without this, a killed mux causes recv() to silently return Ok(None) forever. + let liveness_interval = Duration::from_secs(2); + let mut last_activity = std::time::Instant::now(); + loop { // Check for TX requests (non-blocking). while let Ok(payload) = tx_recv.try_recv() { debug!("TX {} bytes", payload.len()); if let Err(e) = client.transmit(&payload, None) { warn!("transmit error: {e:#}"); + return Err(e); } + last_activity = std::time::Instant::now(); } // Check for RX packets. match client.recv() { Ok(Some(Response::RxPacket { rssi, snr, payload })) => { + last_activity = std::time::Instant::now(); let grade = snr_grade(snr, active_config.sf); if grade.should_forward() { debug!( @@ -157,12 +174,31 @@ fn connect_and_run( ); } } - Ok(Some(_)) => {} + 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(()); } diff --git a/src/router.rs b/src/router.rs index a3d9401..59d81e0 100644 --- a/src/router.rs +++ b/src/router.rs @@ -18,12 +18,14 @@ pub struct RadioConfigInfo { pub requested: donglora_client::RadioConfig, pub source: ConfigSource, pub device: String, + pub connected: bool, } /// Packet event for the TUI log. #[derive(Debug, Clone)] pub struct PacketLogEntry { pub timestamp: Instant, + pub hash: [u8; 32], pub direction: PacketDirection, pub size: usize, pub snr: Option, @@ -31,37 +33,32 @@ pub struct PacketLogEntry { pub action: PacketAction, } +/// Which side the packet arrived from. #[derive(Debug, Clone, Copy)] pub enum PacketDirection { - RadioRx, - RadioTx, - GossipRx, + /// Heard on local radio → forwarded to gossip swarm. + RadioIn, + /// Received from gossip swarm → queued for radio TX. + GossipIn, } +/// What happened to the packet. #[derive(Debug, Clone, Copy)] pub enum PacketAction { - Forwarded, - Transmitted, + /// Successfully bridged (radio→gossip or gossip→radio). + Bridged, + /// Dropped: duplicate payload. DroppedDedup, + /// Dropped: TX queue full. DroppedQueueFull, + /// Dropped: rate limiter. DroppedRateLimit, } -impl std::fmt::Display for PacketDirection { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::RadioRx => write!(f, "RX "), - Self::RadioTx => write!(f, "TX "), - Self::GossipRx => write!(f, "GSP"), - } - } -} - impl std::fmt::Display for PacketAction { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - Self::Forwarded => write!(f, "FWD"), - Self::Transmitted => write!(f, "TX"), + Self::Bridged => write!(f, ""), Self::DroppedDedup => write!(f, "DUP"), Self::DroppedQueueFull => write!(f, "FULL"), Self::DroppedRateLimit => write!(f, "RATE"), @@ -188,7 +185,8 @@ pub async fn run( debug!("radio RX dedup suppressed ({} bytes)", pkt.payload.len()); let _ = log_tx.send(PacketLogEntry { timestamp: Instant::now(), - direction: PacketDirection::RadioRx, + hash, + direction: PacketDirection::RadioIn, size: pkt.payload.len(), snr: Some(pkt.snr), rssi: Some(pkt.rssi), @@ -204,11 +202,12 @@ pub async fn run( let _ = log_tx.send(PacketLogEntry { timestamp: Instant::now(), - direction: PacketDirection::RadioRx, + hash, + direction: PacketDirection::RadioIn, size: pkt.payload.len(), snr: Some(pkt.snr), rssi: Some(pkt.rssi), - action: PacketAction::Forwarded, + action: PacketAction::Bridged, }).await; } RadioEvent::Connected(active_config, source, device) => { @@ -218,10 +217,12 @@ pub async fn run( requested: requested_radio_config, source, device, + connected: true, }); } RadioEvent::Disconnected => { stats.radio_connected.store(0, Ordering::Relaxed); + config_tx.send_modify(|info| info.connected = false); } } } @@ -239,7 +240,8 @@ pub async fn run( debug!("gossip RX dedup suppressed ({} bytes)", frame.payload.len()); let _ = log_tx.send(PacketLogEntry { timestamp: Instant::now(), - direction: PacketDirection::GossipRx, + hash, + direction: PacketDirection::GossipIn, size: frame.payload.len(), snr: Some(frame.snr), rssi: Some(frame.rssi), @@ -254,7 +256,8 @@ pub async fn run( debug!("rate limited ({} bytes)", frame.payload.len()); let _ = log_tx.send(PacketLogEntry { timestamp: Instant::now(), - direction: PacketDirection::GossipRx, + hash, + direction: PacketDirection::GossipIn, size: frame.payload.len(), snr: Some(frame.snr), rssi: Some(frame.rssi), @@ -269,7 +272,8 @@ pub async fn run( warn!("TX queue full, dropping packet ({} bytes)", frame.payload.len()); let _ = log_tx.send(PacketLogEntry { timestamp: Instant::now(), - direction: PacketDirection::GossipRx, + hash, + direction: PacketDirection::GossipIn, size: frame.payload.len(), snr: Some(frame.snr), rssi: Some(frame.rssi), @@ -281,11 +285,12 @@ pub async fn run( tx_queue.push_back(frame.payload.clone()); let _ = log_tx.send(PacketLogEntry { timestamp: Instant::now(), - direction: PacketDirection::GossipRx, + hash, + direction: PacketDirection::GossipIn, size: frame.payload.len(), snr: Some(frame.snr), rssi: Some(frame.rssi), - action: PacketAction::Transmitted, + action: PacketAction::Bridged, }).await; } GossipEvent::NeighborChanged(count) => { @@ -297,18 +302,7 @@ pub async fn run( // Drain TX queue to radio. while let Some(payload) = tx_queue.pop_front() { - let size = payload.len(); stats.radio_tx.fetch_add(1, Ordering::Relaxed); - let _ = log_tx - .send(PacketLogEntry { - timestamp: Instant::now(), - direction: PacketDirection::RadioTx, - size, - snr: None, - rssi: None, - action: PacketAction::Transmitted, - }) - .await; if radio_tx.send(payload).await.is_err() { break; } diff --git a/src/tui.rs b/src/tui.rs index 1fa54b5..c5af0a8 100644 --- a/src/tui.rs +++ b/src/tui.rs @@ -23,8 +23,8 @@ use crate::router::{PacketAction, PacketDirection, PacketLogEntry, RadioConfigIn const TICK_RATE: Duration = Duration::from_millis(250); const MAX_LOG_ENTRIES: usize = 200; -const BUCKET_SECS: u64 = 10; -const MAX_BUCKETS: usize = 60; +const BUCKET_SECS: u64 = 1; +const MAX_BUCKETS: usize = 300; // ── Color palette (consistent across panels) ───────────────────── @@ -202,7 +202,11 @@ async fn tui_loop( if event::poll(timeout)? && let Event::Key(key) = event::read()? && key.kind == KeyEventKind::Press - && key.code == KeyCode::Char('q') + && matches!( + (key.code, key.modifiers), + (KeyCode::Char('q'), _) + | (KeyCode::Char('c'), crossterm::event::KeyModifiers::CONTROL) + ) { cancel.cancel(); return Ok(()); @@ -226,18 +230,22 @@ fn draw_ui( start_time: Instant, series: &TimeSeries, ) { - let outer = Layout::default() + // Header row + body. + let rows = Layout::default() .direction(Direction::Vertical) - .constraints([Constraint::Length(1), Constraint::Percentage(65), Constraint::Percentage(35)]) + .constraints([Constraint::Length(1), Constraint::Min(0)]) .split(area); - draw_header(frame, outer[0], stats); + draw_header(frame, rows[0], stats); + // Body: 3 columns — left panels | packet log (fixed) | activity charts (fill). + // Packet log columns: Age=5 Hash=8 RF=4 Net=4 Size=6 RSSI=6 SNR=5 Act=5 = 43 + 2 border + 1 pad = 46 let cols = Layout::default() .direction(Direction::Horizontal) - .constraints([Constraint::Length(30), Constraint::Min(40)]) - .split(outer[1]); + .constraints([Constraint::Length(30), Constraint::Length(46), Constraint::Min(10)]) + .split(rows[1]); + // Left column: Network, Radio, Stats stacked. let left = Layout::default() .direction(Direction::Vertical) .constraints([ @@ -251,23 +259,35 @@ fn draw_ui( draw_radio(frame, left[1], config_info); draw_stats(frame, left[2], stats, start_time); draw_log(frame, cols[1], log_entries); - draw_activity(frame, outer[2], series); + draw_activity(frame, cols[2], series); } // ── Header ─────────────────────────────────────────────────────── fn draw_header(frame: &mut ratatui::Frame, area: Rect, stats: &StatsSnapshot) { - let now = chrono_time(); - let line = Line::from(vec![ - Span::styled(" donglora-bridge ", Style::default().fg(C_TITLE).add_modifier(Modifier::BOLD)), + let cols = Layout::default() + .direction(Direction::Horizontal) + .constraints([ + Constraint::Length(14), // pills + Constraint::Min(0), // fill + ]) + .split(area); + + // Left: status pills. + let pills = Line::from(vec![ + Span::raw(" "), status_pill("RADIO", stats.radio_connected), Span::raw(" "), status_pill("NET", stats.neighbor_count > 0), - Span::raw(" "), - Span::styled(now, Style::default().fg(C_LABEL)), - Span::styled(" q=quit", Style::default().fg(C_LABEL)), ]); - frame.render_widget(Paragraph::new(line), area); + frame.render_widget(Paragraph::new(pills), cols[0]); + + // Right: clock with timezone. + let clock = Line::from(Span::styled( + format!("{} ", chrono_time()), + Style::default().fg(C_LABEL), + )); + frame.render_widget(Paragraph::new(clock).alignment(Alignment::Right), cols[1]); } fn status_pill(label: &str, ok: bool) -> Span<'static> { @@ -307,11 +327,28 @@ fn draw_network(frame: &mut ratatui::Frame, area: Rect, state: &crate::gossip::S // ── Radio panel ────────────────────────────────────────────────── fn draw_radio(frame: &mut ratatui::Frame, area: Rect, config_info: &RadioConfigInfo) { + let connected = config_info.connected; let is_mux = matches!(config_info.source, ConfigSource::Mux); - let border_color = if is_mux { C_RATE_DROP } else { C_BORDER }; + + let border_color = if !connected { + C_QUEUE_DROP + } else if is_mux { + C_RATE_DROP + } else { + C_BORDER + }; + + let title = if !connected { + " Radio (disconnected) " + } else if is_mux { + " Radio (mux) " + } else { + " Radio " + }; + let block = Block::default() .title(Span::styled( - if is_mux { " Radio (mux) " } else { " Radio " }, + title, Style::default().fg(C_TITLE).add_modifier(Modifier::BOLD), )) .borders(Borders::ALL) @@ -330,15 +367,23 @@ fn draw_radio(frame: &mut ratatui::Frame, area: Rect, config_info: &RadioConfigI 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 }; let cf = |label: &str, value: String, matches: bool| -> Line<'static> { - let color = if !is_mux || matches { C_ACCENT } else { C_RATE_DROP }; - let mark = if is_mux && !matches { " !" } else { "" }; + let color = if !connected { + C_LABEL + } else if !is_mux || matches { + C_ACCENT + } else { + C_RATE_DROP + }; + let mark = if connected && is_mux && !matches { " !" } else { "" }; kv_line(label, &format!("{value}{mark}"), color) }; let dev = if config_info.device.is_empty() { "-" } else { &config_info.device }; let lines = vec![ - kv_line("Device", dev, C_ACCENT), + kv_line("Device", dev, val_color), 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), @@ -392,41 +437,59 @@ fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry]) let start = entries.len().saturating_sub(data_height); let visible = &entries[start..]; - // Column header with underline effect via bold. + // 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. + let hdr = Style::default().fg(Color::Gray).add_modifier(Modifier::BOLD); let header = Line::from(vec![ - Span::styled(" Age", Style::default().fg(C_LABEL).add_modifier(Modifier::BOLD)), - Span::styled(" RF", Style::default().fg(C_LABEL).add_modifier(Modifier::BOLD)), - Span::styled(" Net", Style::default().fg(C_LABEL).add_modifier(Modifier::BOLD)), - Span::styled(" Size", Style::default().fg(C_LABEL).add_modifier(Modifier::BOLD)), - Span::styled(" RSSI", Style::default().fg(C_LABEL).add_modifier(Modifier::BOLD)), - Span::styled(" SNR", Style::default().fg(C_LABEL).add_modifier(Modifier::BOLD)), - Span::styled(" Act", Style::default().fg(C_LABEL).add_modifier(Modifier::BOLD)), + Span::styled(format!("{:>5}", "Age"), hdr), + Span::styled(format!("{:>8}", "Hash"), hdr), + Span::styled(format!("{:>4}", "RF"), hdr), + Span::styled(format!("{:>4}", "Net"), hdr), + Span::styled(format!("{:>6}", "Size"), hdr), + Span::styled(format!("{:>6}", "RSSI"), hdr), + Span::styled(format!("{:>5}", "SNR"), hdr), + Span::styled(format!("{:>5}", "Act"), hdr), ]); let mut lines = vec![header]; + let mut prev_hash: Option<[u8; 32]> = None; for e in visible { let age = format_short_duration(e.timestamp.elapsed()); + let bridged = matches!(e.action, PacketAction::Bridged); - let dash = " ───"; + // Show short hash, or ellipsis if same as previous line. + let hash_span = if prev_hash == Some(e.hash) { + Span::styled(format!("{:>8}", "··"), Style::default().fg(C_BORDER)) + } else { + Span::styled(format!("{:>8}", hex::encode(&e.hash[..3])), Style::default().fg(C_LABEL)) + }; + prev_hash = Some(e.hash); + + // Each row shows both columns. Bridged packets light up both sides. + // Dropped packets only show the input side. + let dash = Span::styled(format!("{:>4}", "──"), Style::default().fg(C_BORDER)); let (rf_col, net_col) = match e.direction { - PacketDirection::RadioRx => ( - Span::styled(" RX", Style::default().fg(C_RADIO_RX)), - Span::styled(dash, Style::default().fg(C_BORDER)), + PacketDirection::RadioIn => ( + Span::styled(format!("{:>4}", "RX"), Style::default().fg(C_RADIO_RX)), + if bridged { + Span::styled(format!("{:>4}", "TX"), Style::default().fg(C_NET_TX)) + } else { + dash + }, ), - PacketDirection::RadioTx => ( - Span::styled(" TX", Style::default().fg(C_RADIO_TX)), - Span::styled(dash, Style::default().fg(C_BORDER)), - ), - PacketDirection::GossipRx => ( - Span::styled(dash, Style::default().fg(C_BORDER)), - Span::styled(" RX", Style::default().fg(C_NET_RX)), + PacketDirection::GossipIn => ( + if bridged { + Span::styled(format!("{:>4}", "TX"), Style::default().fg(C_RADIO_TX)) + } else { + dash + }, + Span::styled(format!("{:>4}", "RX"), Style::default().fg(C_NET_RX)), ), }; let action_color = match e.action { - PacketAction::Forwarded => C_NET_TX, - PacketAction::Transmitted => C_RADIO_TX, + PacketAction::Bridged => C_LABEL, PacketAction::DroppedDedup => C_DEDUP, PacketAction::DroppedQueueFull => C_QUEUE_DROP, PacketAction::DroppedRateLimit => C_RATE_DROP, @@ -434,21 +497,31 @@ fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry]) let rssi_span = match e.rssi { Some(r) => Span::styled(format!("{r:>6}"), Style::default().fg(rssi_color(r))), - None => Span::styled(" -", Style::default().fg(C_LABEL)), + None => Span::styled(format!("{:>6}", "-"), Style::default().fg(C_LABEL)), }; let snr_span = match e.snr { Some(s) => Span::styled(format!("{s:>5}"), Style::default().fg(snr_color(s))), - None => Span::styled(" -", Style::default().fg(C_LABEL)), + 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!(" {:>4}", action_str), Style::default().fg(action_color)); + + let size_str = format!("{}B", e.size); lines.push(Line::from(vec![ - Span::styled(format!(" {age:>3} "), Style::default().fg(C_LABEL)), + Span::styled(format!("{:>5}", age), Style::default().fg(C_LABEL)), + hash_span, rf_col, net_col, - Span::styled(format!("{:>5}B", e.size), Style::default().fg(C_VALUE)), + Span::styled(format!("{:>6}", size_str), Style::default().fg(C_VALUE)), rssi_span, snr_span, - Span::styled(format!(" {}", e.action), Style::default().fg(action_color)), + action_span, ])); } @@ -458,11 +531,11 @@ fn draw_log(frame: &mut ratatui::Frame, area: Rect, entries: &[PacketLogEntry]) // ── Activity chart ─────────────────────────────────────────────── fn draw_activity(frame: &mut ratatui::Frame, area: Rect, series: &TimeSeries) { - let block = panel_block(" Activity (10s buckets) "); + let block = panel_block(" Activity (1s) "); let inner = block.inner(area); frame.render_widget(block, area); - if inner.height < 3 || inner.width < 20 { + if inner.height < 6 || inner.width < 6 { return; } @@ -475,61 +548,53 @@ fn draw_activity(frame: &mut ratatui::Frame, area: Rect, series: &TimeSeries) { ]) .split(inner); - let rx_data = series.radio_rx(); - let tx_data = series.radio_tx(); - let snr_data = series.avg_snr(); + let spark_w = inner.width as usize; - let label_w = 12u16; - let spark_w = inner.width.saturating_sub(label_w) as usize; + let rx_data = pad_right_align(&series.radio_rx(), spark_w); + let tx_data = pad_right_align(&series.radio_tx(), spark_w); + let snr_data = pad_right_align(&series.avg_snr(), spark_w); let peak_rx = series.peak_rx(); let peak_tx = series.peak_tx(); let snr_label = series .latest_avg_snr() - .map(|s| format!("avg:{s}dB")) + .map(|s| format!("{s}dB")) .unwrap_or_else(|| "-".into()); - let rx_padded = pad_right_align(&rx_data, spark_w); - let tx_padded = pad_right_align(&tx_data, spark_w); - let snr_padded = pad_right_align(&snr_data, spark_w); - - draw_sparkline_row(frame, rows[0], "RF RX", &format!("pk:{peak_rx}"), &rx_padded, C_RADIO_RX, label_w); - draw_sparkline_row(frame, rows[1], "RF TX", &format!("pk:{peak_tx}"), &tx_padded, C_RADIO_TX, label_w); - draw_sparkline_row(frame, rows[2], "SNR", &snr_label, &snr_padded, C_SNR, label_w); + draw_sparkline_section(frame, rows[0], "RX", &format!("pk:{peak_rx}"), &rx_data, C_RADIO_RX); + draw_sparkline_section(frame, rows[1], "TX", &format!("pk:{peak_tx}"), &tx_data, C_RADIO_TX); + draw_sparkline_section(frame, rows[2], "SNR", &snr_label, &snr_data, C_SNR); } -fn draw_sparkline_row( +/// Draw a label line on top, sparkline filling the rest below. +fn draw_sparkline_section( frame: &mut ratatui::Frame, area: Rect, label: &str, detail: &str, data: &[u64], color: Color, - label_width: u16, ) { - if area.width < label_width + 5 || area.height < 1 { + if area.height < 2 || area.width < 4 { return; } - let cols = Layout::default() - .direction(Direction::Horizontal) - .constraints([Constraint::Length(label_width), Constraint::Min(1)]) + let rows = Layout::default() + .direction(Direction::Vertical) + .constraints([Constraint::Length(1), Constraint::Min(1)]) .split(area); - // Label + detail stacked if room, otherwise just label. - let mut label_lines = vec![ - Line::from(Span::styled(format!(" {label}"), Style::default().fg(color).add_modifier(Modifier::BOLD))), - ]; - if cols[0].height > 1 { - label_lines.push(Line::from(Span::styled(format!(" {detail}"), Style::default().fg(C_LABEL)))); - } - frame.render_widget(Paragraph::new(label_lines), cols[0]); + let label_line = Line::from(vec![ + Span::styled(format!(" {label} "), Style::default().fg(color).add_modifier(Modifier::BOLD)), + Span::styled(detail.to_string(), Style::default().fg(C_LABEL)), + ]); + frame.render_widget(Paragraph::new(label_line), rows[0]); let spark = Sparkline::default() .data(data) .bar_set(symbols::bar::NINE_LEVELS) .style(Style::default().fg(color)); - frame.render_widget(spark, cols[1]); + frame.render_widget(spark, rows[1]); } /// Pad data with leading zeros so values are right-aligned in `width` columns. @@ -609,8 +674,5 @@ fn format_short_duration(d: Duration) -> String { } fn chrono_time() -> String { - let now = std::time::SystemTime::now(); - let since_epoch = now.duration_since(std::time::UNIX_EPOCH).unwrap_or_default(); - let secs = since_epoch.as_secs(); - format!("{:02}:{:02}:{:02}", (secs / 3600) % 24, (secs % 3600) / 60, secs % 60) + jiff::Zoned::now().strftime("%H:%M:%S %Z").to_string() }