From 5038c37d6edbddf1ae63e985c4d95cf09d102873 Mon Sep 17 00:00:00 2001 From: Joey Stanford Date: Sun, 2 Aug 2026 09:12:24 -0600 Subject: [PATCH 1/2] fix(reticulum): exhaust alternate paths for Nomad and LXMF Direct MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Promote ranked path-slot backups and other live hubs after link failure before Nomad give-up or Direct→PN fallback, and log tried interfaces for the next dump triage. --- docs/troubleshooting.md | 2 +- reticulum-sidecar/src/stack/live.rs | 323 +++++++++------ reticulum-sidecar/src/stack/lxmf_outbound.rs | 249 ++++++++++- reticulum-sidecar/src/stack/mod.rs | 1 + reticulum-sidecar/src/stack/path_failover.rs | 385 ++++++++++++++++++ src/renderer/stores/nomadNetworkStore.test.ts | 6 + src/renderer/stores/nomadNetworkStore.ts | 21 + src/shared/nomad-types.ts | 4 + 8 files changed, 856 insertions(+), 135 deletions(-) create mode 100644 reticulum-sidecar/src/stack/path_failover.rs diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 54fd7d309..4e301cec3 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -1034,7 +1034,7 @@ In dev, **Start stack** now rebuilds when `reticulum-sidecar/src/**/*.rs` or `Ca Unrecognized codes pass through unchanged. -TCP/network Nomad Links use path-scaled initiator hops (`link_hops = clamp(path_hops, 3, 7)`) and a LinkClient proof wait of the **remaining overall MeshChat deadline** (~45s TCP after instant pubkey recall), matching v5.25.0. Do not cap LRPROOF at hops×6 or a 30s floor — that false-failed multi-hop hub pages that still load on release. First attempts use a cached path when present (no DropPath storm); missing paths RequestPath briefly and may return `path_timeout`. Retries may DropPath + rediscover; `force_path_ok=true` means rediscovered after absence only (cache hits log `force_path_ok=false`). Failure logs (`[nomadNetworkStore] … fetch failed` and sidecar `Nomad Link query failed`) include `path_hops`, `link_hops`, `proof_budget_secs`, `force_path_ok`, `path_ensure_kind`, `elapsed_ms`, and `raw=`. UI errors distinguish cached-path vs rediscovered-path link failures. +TCP/network Nomad Links use path-scaled initiator hops (`link_hops = clamp(path_hops, 3, 7)`) and a LinkClient proof wait of the **remaining overall MeshChat deadline** (~45s TCP after instant pubkey recall), matching v5.25.0. Do not cap LRPROOF at hops×6 or a 30s floor — that false-failed multi-hop hub pages that still load on release. First attempts use a cached path when present (no DropPath storm); missing paths RequestPath briefly and may return `path_timeout`. On TCP `link_timeout`, the sidecar suppresses the dead iface, drops the failed via, promotes ranked path-slot backups / other live hubs (extra RequestPath when another TCP/RF iface is up), then retries inside the same fetch. LXMF Direct chat uses the same exhaustion before the one-shot preferred-PN fallback. `force_path_ok=true` means rediscovered after absence only (cache hits log `force_path_ok=false`). Failure logs (`[nomadNetworkStore] … fetch failed` and sidecar `Nomad Link query failed`) include `path_hops`, `link_hops`, `proof_budget_secs`, `force_path_ok`, `path_ensure_kind`, `elapsed_ms`, `tried_interfaces`, `failover_rounds`, `iface`, and `raw=`. UI errors distinguish cached-path vs rediscovered-path link failures. **Cause**: Older `LinkClient` always waited for a fresh path-response announce for the destination public key, even when Nomad announces had already cached it. Successful fetches could also deregister all `nomadnetwork.node` announce handlers. Distant/high-hop nodes can still time out at the path stage (expected RF/mesh reachability limits). diff --git a/reticulum-sidecar/src/stack/live.rs b/reticulum-sidecar/src/stack/live.rs index 9f72c4a8e..3e3dd8334 100644 --- a/reticulum-sidecar/src/stack/live.rs +++ b/reticulum-sidecar/src/stack/live.rs @@ -48,6 +48,10 @@ use super::nomad_timeouts; use super::packet_log::{ PacketLogBuffer, collect_tx_interface_names_for_egress, wire_packet_from_tap, }; +use super::path_failover::{ + self, MAX_VIA_FAILOVERS, PathSlotCandidate, active_via_hash_from_slots, live_interface_names, + push_tried_iface, remaining_live_ifaces, select_unblocked_slot, via_prefix, +}; use super::path_medium::{self, PathMediumPreferenceSetting, PathMediumSetting}; use super::path_speed; use super::persistence::PersistedState; @@ -66,7 +70,7 @@ use super::via::{ classify_path_interface_name, merge_live_interfaces_with_config, merge_observed_egress_vias, resolve_lxmf_sent_via, }; -use lxmf_outbound::LxmfOutboundDriver; +use lxmf_outbound::{LxmfOutboundDriver, PathTableRoute}; /// Settle window for PacketTap Tx correlation after LXMF enqueue. const LXMF_EGRESS_TAP_SETTLE_MS: u64 = 1500; @@ -867,6 +871,8 @@ impl LiveBridge { raw_error: None, elapsed_ms: None, tried_interfaces: None, + failover_rounds: None, + last_iface: None, })?; // Prefer path-peer cache (maintenance refreshes every ~2s). Avoid a synchronous // GetPathTable here — that control query alone can stall TCP page loads for seconds. @@ -969,6 +975,8 @@ impl LiveBridge { raw_error: Some(format!("path ensure kind={kind} (no cached path)")), elapsed_ms: Some(elapsed_ms_since(query_started)), tried_interfaces: None, + failover_rounds: None, + last_iface: None, }); } } @@ -995,6 +1003,8 @@ impl LiveBridge { raw_error: None, elapsed_ms: Some(elapsed_ms_since(query_started)), tried_interfaces: None, + failover_rounds: None, + last_iface: None, }); } let mut link_hops = nomad_timeouts::nomad_link_initiator_hops(egress, hops); @@ -1031,6 +1041,7 @@ impl LiveBridge { let mut current_iface = path_iface.clone(); let mut current_via = active_via_hash_from_slots(&path_slots_snapshot); + let live_ifaces = live_interface_names(interfaces); // One generation for this page request + all via failovers. Bumping per // Link attempt would cancel a newer request when the older one retries. let link_gen = self @@ -1045,7 +1056,7 @@ impl LiveBridge { serde_json::json!({ "round": 0, "iface": current_iface, - "via_prefix": nomad_via_prefix(current_via.as_deref()), + "via_prefix": via_prefix(current_via.as_deref()), "hops": hops, "timeout_secs": timeout_secs, }), @@ -1067,9 +1078,11 @@ impl LiveBridge { .await; // Dead next-hops often reappear on another local iface (same via_hash). - // Suppress the failed iface, DropAllVia that hop, and retry on a truly - // different via — up to two failovers inside one page fetch. + // Suppress the failed iface, DropAllVia that hop, promote ranked backups + // / other live hubs, and retry — up to MAX_VIA_FAILOVERS inside one fetch. let mut failover_round: u8 = 0; + let mut tried_interfaces: Vec = Vec::new(); + push_tried_iface(&mut tried_interfaces, current_iface.as_deref()); let mut blocked_ifaces: Vec = Vec::new(); let mut blocked_vias: Vec = Vec::new(); if let Some(iface) = current_iface.clone() { @@ -1082,7 +1095,7 @@ impl LiveBridge { .as_ref() .err() .is_some_and(|e| e.code == "link_timeout") - && failover_round < NOMAD_MAX_VIA_FAILOVERS + && failover_round < MAX_VIA_FAILOVERS { if self.nomad_link_generation.load(Ordering::SeqCst) != link_gen { break; @@ -1096,7 +1109,7 @@ impl LiveBridge { serde_json::json!({ "round": failover_round, "iface": current_iface, - "via_prefix": nomad_via_prefix(current_via.as_deref()), + "via_prefix": via_prefix(current_via.as_deref()), "hops": hops, }), ); @@ -1108,11 +1121,15 @@ impl LiveBridge { serde_json::json!({ "round": failover_round, "iface": current_iface, - "via_prefix": nomad_via_prefix(current_via.as_deref()), + "via_prefix": via_prefix(current_via.as_deref()), }), ); - let Some((failover_hops, failover_iface, failover_via)) = self - .nomad_suppress_via_and_rediscover(hash_hex, &blocked_ifaces, &blocked_vias) + let Some(PathSlotCandidate { + hops: failover_hops, + iface: failover_iface, + via: failover_via, + }) = self + .suppress_via_and_rediscover(hash_hex, &blocked_ifaces, &blocked_vias, &live_ifaces) .await else { self.emit_nomad_page_progress( @@ -1123,7 +1140,7 @@ impl LiveBridge { serde_json::json!({ "round": failover_round, "iface": current_iface, - "via_prefix": nomad_via_prefix(current_via.as_deref()), + "via_prefix": via_prefix(current_via.as_deref()), }), ); break; @@ -1131,6 +1148,7 @@ impl LiveBridge { if self.nomad_link_generation.load(Ordering::SeqCst) != link_gen { break; } + push_tried_iface(&mut tried_interfaces, failover_iface.as_deref()); if let Some(iface) = failover_iface.clone() { blocked_ifaces.push(iface); } @@ -1157,7 +1175,7 @@ impl LiveBridge { serde_json::json!({ "round": failover_round, "iface": current_iface, - "via_prefix": nomad_via_prefix(current_via.as_deref()), + "via_prefix": via_prefix(current_via.as_deref()), "hops": failover_hops, "timeout_secs": failover_timeout, }), @@ -1187,9 +1205,13 @@ impl LiveBridge { } if let Err(ref mut err) = result { - if !blocked_ifaces.is_empty() { - err.tried_interfaces = Some(blocked_ifaces.clone()); + if !tried_interfaces.is_empty() { + err.tried_interfaces = Some(tried_interfaces); + } + if failover_round > 0 { + err.failover_rounds = Some(failover_round); } + err.last_iface = current_iface.clone(); } let elapsed_ms = elapsed_ms_since(query_started); @@ -1240,6 +1262,8 @@ impl LiveBridge { raw_error: None, elapsed_ms: None, tried_interfaces: None, + failover_rounds: None, + last_iface: None, }); } let (cancel_tx, cancel_rx) = tokio::sync::oneshot::channel(); @@ -1257,6 +1281,8 @@ impl LiveBridge { raw_error: None, elapsed_ms: None, tried_interfaces: None, + failover_rounds: None, + last_iface: None, }); } if let Some(prev) = slot.take() { @@ -1281,6 +1307,8 @@ impl LiveBridge { raw_error: None, elapsed_ms: None, tried_interfaces: None, + failover_rounds: None, + last_iface: None, }); }; if self.nomad_link_generation.load(Ordering::SeqCst) != my_gen { @@ -1295,6 +1323,8 @@ impl LiveBridge { raw_error: None, elapsed_ms: None, tried_interfaces: None, + failover_rounds: None, + last_iface: None, }); } let client = LinkClient::new(self.handle.transport_tx.clone(), self.identity.clone()); @@ -1319,6 +1349,8 @@ impl LiveBridge { raw_error: None, elapsed_ms: None, tried_interfaces: None, + failover_rounds: None, + last_iface: None, }), query_result = query_fut => { query_result.map_err(|e| { @@ -1335,6 +1367,8 @@ impl LiveBridge { raw_error: Some(raw), elapsed_ms: None, tried_interfaces: None, + failover_rounds: None, + last_iface: None, } }) } @@ -1346,15 +1380,16 @@ impl LiveBridge { result } - /// After LRPROOF timeout, suppress the active iface and drop the failed - /// next-hop so rediscovery cannot reinstall the same blackhole via a - /// different local interface name (Ratspeak vs RMAP World sharing a via). - async fn nomad_suppress_via_and_rediscover( + /// After a link failure, suppress the dead iface, drop failed vias, promote + /// ranked backups / other live hubs, and RequestPath until an unblocked slot + /// appears (or the probe budget expires). + async fn suppress_via_and_rediscover( &self, hash_hex: &str, blocked_ifaces: &[String], blocked_vias: &[String], - ) -> Option<(u8, Option, Option)> { + live_ifaces: &[String], + ) -> Option { let dest = parse_hash16(hash_hex).ok()?; let slots_before = self .path_slots(hash_hex) @@ -1362,10 +1397,21 @@ impl LiveBridge { .map(|(slots, _)| slots) .unwrap_or_default(); let failed_via = active_via_hash_from_slots(&slots_before); + let prefer = remaining_live_ifaces(live_ifaces, blocked_ifaces); + + // Promote an already-known unblocked backup before waiting on rediscovery. + let known_backup = select_unblocked_slot( + &slots_before, + blocked_ifaces, + blocked_vias, + failed_via.as_deref(), + &prefer, + ); + let _ = self .query_control_timed(TransportQuery::SuppressCurrentPathInterface { dest, - duration: NOMAD_IFACE_SUPPRESS_SECS, + duration: path_failover::IFACE_SUPPRESS_SECS, }) .await; // Drop every known-bad next hop (not only the currently active slot — @@ -1389,6 +1435,40 @@ impl LiveBridge { if let Ok(mut cache) = self.peer_via_cache.lock() { cache.remove(&hash_hex.to_lowercase()); } + + if let Some(backup) = known_backup { + // Brief settle so transport can activate the promoted backup slot. + let _ = self + .handle + .transport_tx + .send(TransportMessage::RequestPath { + destination_hash: dest, + }) + .await; + let settle_deadline = + tokio::time::Instant::now() + path_failover::VIA_FAILOVER_POLL_INTERVAL * 10; + while tokio::time::Instant::now() < settle_deadline { + let _ = self.refresh_outbound_path_table().await; + let slots = self + .path_slots(hash_hex) + .await + .map(|(slots, _)| slots) + .unwrap_or_default(); + if let Some(found) = select_unblocked_slot( + &slots, + blocked_ifaces, + blocked_vias, + failed_via.as_deref(), + &prefer, + ) { + return Some(found); + } + tokio::time::sleep(path_failover::VIA_FAILOVER_POLL_INTERVAL).await; + } + // Backup was known before suppress; return it even if the table is slow. + return Some(backup); + } + let _ = self .handle .transport_tx @@ -1396,9 +1476,58 @@ impl LiveBridge { destination_hash: dest, }) .await; - // Longer wait: alternate hubs may be slower to answer path requests. - let deadline = tokio::time::Instant::now() + NOMAD_VIA_FAILOVER_PROBE_WAIT; - let mut found: Option<(u8, Option, Option)> = None; + + let mut found = self + .poll_unblocked_path_slot( + hash_hex, + blocked_ifaces, + blocked_vias, + failed_via.as_deref(), + &prefer, + path_failover::VIA_FAILOVER_PROBE_WAIT, + ) + .await; + + // When other live hubs remain, issue a second RequestPath + wait instead of + // giving up after a single short probe (TTP-only cache vs Local Pi up). + if found.is_none() && !prefer.is_empty() { + tracing::debug!( + target: "nomad", + dest = %hash_hex, + remaining = ?prefer, + "Nomad path failover: extra RequestPath toward remaining live interfaces" + ); + let _ = self + .handle + .transport_tx + .send(TransportMessage::RequestPath { + destination_hash: dest, + }) + .await; + found = self + .poll_unblocked_path_slot( + hash_hex, + blocked_ifaces, + blocked_vias, + failed_via.as_deref(), + &prefer, + path_failover::VIA_FAILOVER_EXTRA_PROBE_WAIT, + ) + .await; + } + found + } + + async fn poll_unblocked_path_slot( + &self, + hash_hex: &str, + blocked_ifaces: &[String], + blocked_vias: &[String], + failed_via: Option<&str>, + prefer_ifaces: &[String], + wait: Duration, + ) -> Option { + let deadline = tokio::time::Instant::now() + wait; while tokio::time::Instant::now() < deadline { let _ = self.refresh_outbound_path_table().await; let slots = self @@ -1406,46 +1535,18 @@ impl LiveBridge { .await .map(|(slots, _)| slots) .unwrap_or_default(); - // Prefer any live slot that is not on a blocked iface/via. - for slot in &slots { - let iface = slot - .get("interface") - .and_then(|v| v.as_str()) - .map(str::to_string); - let via = slot - .get("via_hash") - .and_then(|v| v.as_str()) - .map(str::to_string); - let hops = slot - .get("hops") - .and_then(serde_json::Value::as_u64) - .map(|h| h as u8); - let Some(h) = hops else { continue }; - let iface_blocked = iface - .as_ref() - .is_some_and(|i| blocked_ifaces.iter().any(|b| b.eq_ignore_ascii_case(i))); - let via_blocked = via - .as_ref() - .is_some_and(|v| blocked_vias.iter().any(|b| b.eq_ignore_ascii_case(v))); - if iface_blocked || via_blocked { - continue; - } - // Also reject the via we just failed on this round. - if failed_via - .as_ref() - .is_some_and(|fv| via.as_ref().is_some_and(|v| v.eq_ignore_ascii_case(fv))) - { - continue; - } - found = Some((h, iface, via)); - break; - } - if found.is_some() { - break; + if let Some(found) = select_unblocked_slot( + &slots, + blocked_ifaces, + blocked_vias, + failed_via, + prefer_ifaces, + ) { + return Some(found); } - tokio::time::sleep(Duration::from_millis(200)).await; + tokio::time::sleep(path_failover::VIA_FAILOVER_POLL_INTERVAL).await; } - found + None } pub async fn fetch_nomad_file( @@ -2284,7 +2385,13 @@ impl LiveBridge { Some( entries .iter() - .map(|e| (e.hash, e.hops, hex::encode(e.hash))) + .map(|e| PathTableRoute { + hash: e.hash, + hops: e.hops, + hex_key: hex::encode(e.hash), + interface: Some(e.interface.clone()).filter(|s| !s.is_empty()), + via: e.via.map(hex::encode), + }) .collect::>(), ) } else { @@ -2439,7 +2546,13 @@ impl LiveBridge { } let path_entries = entries .iter() - .map(|e| (e.hash, e.hops, hex::encode(e.hash))) + .map(|e| PathTableRoute { + hash: e.hash, + hops: e.hops, + hex_key: hex::encode(e.hash), + interface: Some(e.interface.clone()).filter(|s| !s.is_empty()), + via: e.via.map(hex::encode), + }) .collect::>(); if let Ok(mut driver) = self.outbound.lock() { driver.update_path_table(&path_entries); @@ -3686,10 +3799,6 @@ impl LiveBridge { } } -fn nomad_via_prefix(via: Option<&str>) -> Option { - via.map(|v| v.chars().take(8).collect()) -} - pub(super) fn lxmf_payload_from_message( msg: &LxMessage, self_lxmf_hash: &str, @@ -4111,6 +4220,10 @@ struct NomadRemoteQueryError { elapsed_ms: Option, /// Local interface names attempted (including via-aware failovers). tried_interfaces: Option>, + /// In-request via/iface failover rounds completed (0 if none). + failover_rounds: Option, + /// Last local interface used for the Link attempt. + last_iface: Option, } fn elapsed_ms_since(started: tokio::time::Instant) -> u64 { @@ -4194,6 +4307,17 @@ fn nomad_remote_error_json(err: &NomadRemoteQueryError) -> serde_json::Value { if let Some(ifaces) = err.tried_interfaces.as_ref().filter(|v| !v.is_empty()) { obj.insert("tried_interfaces".into(), serde_json::json!(ifaces)); } + if let Some(rounds) = err.failover_rounds { + obj.insert("failover_rounds".into(), serde_json::json!(rounds)); + } + if let Some(iface) = err + .last_iface + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty()) + { + obj.insert("iface".into(), serde_json::json!(iface)); + } out } @@ -4350,32 +4474,6 @@ const NOMAD_FORCE_PATH_REFRESH_WAIT: Duration = Duration::from_secs(4); /// Strict TCP/network DropPath→RequestPath wait before Link (no stale-accept fall-through). const NOMAD_TCP_PATH_PROBE_WAIT: Duration = Duration::from_secs(5); -/// How long to reject the failed Nomad path interface after LRPROOF timeout so -/// an alternate hub slot (e.g. TTP_TCP vs Ratspeak) can become active. -const NOMAD_IFACE_SUPPRESS_SECS: f64 = 120.0; - -/// Wait for a path with a different via_hash after DropAllVia + suppress. -const NOMAD_VIA_FAILOVER_PROBE_WAIT: Duration = Duration::from_secs(8); - -/// Max in-request via failovers after the first link_timeout (total Link tries = 1 + this). -const NOMAD_MAX_VIA_FAILOVERS: u8 = 2; - -fn active_via_hash_from_slots(slots: &[serde_json::Value]) -> Option { - slots.iter().find_map(|slot| { - let active = slot - .get("active") - .and_then(serde_json::Value::as_bool) - .unwrap_or(false); - if !active { - return None; - } - slot.get("via_hash") - .and_then(|v| v.as_str()) - .filter(|s| !s.is_empty()) - .map(str::to_string) - }) -} - #[allow(clippy::too_many_arguments, clippy::result_large_err)] // Nomad Link diagnostics bundle fn finish_nomad_link_result( result: Result, NomadRemoteQueryError>, @@ -4577,41 +4675,6 @@ mod announce_display_name_tests { assert_eq!(added, vec!["cc".to_string()]); } - #[test] - fn active_via_hash_from_slots_skips_inactive_and_empty() { - assert_eq!(active_via_hash_from_slots(&[]), None); - assert_eq!( - active_via_hash_from_slots(&[serde_json::json!({ - "active": true, - "via_hash": "", - })]), - None - ); - assert_eq!( - active_via_hash_from_slots(&[ - serde_json::json!({ - "active": false, - "via_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", - }), - serde_json::json!({ - "active": true, - "via_hash": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", - }), - ]), - Some("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".into()) - ); - } - - #[test] - fn nomad_via_prefix_handles_none_and_truncates() { - assert_eq!(nomad_via_prefix(None), None); - assert_eq!( - nomad_via_prefix(Some("abcdefghijklmnop")), - Some("abcdefgh".into()) - ); - assert_eq!(nomad_via_prefix(Some("abcd")), Some("abcd".into())); - } - #[test] fn force_path_refresh_rejects_stale_route_until_absent_then_accepts_refresh() { // Existing stale route still installed — must not accept yet. @@ -4692,6 +4755,8 @@ mod announce_display_name_tests { raw_error: Some("timed out waiting for link proof".into()), elapsed_ms: Some(18_250), tried_interfaces: Some(vec!["Ratspeak".into(), "RNS_Transport_US-East".into()]), + failover_rounds: Some(1), + last_iface: Some("RNS_Transport_US-East".into()), }); assert_eq!(with_diag["ok"], false); assert_eq!(with_diag["error"], "link_timeout"); @@ -4707,6 +4772,8 @@ mod announce_display_name_tests { with_diag["tried_interfaces"], serde_json::json!(["Ratspeak", "RNS_Transport_US-East"]) ); + assert_eq!(with_diag["failover_rounds"], 1); + assert_eq!(with_diag["iface"], "RNS_Transport_US-East"); let without = nomad_remote_error_json(&NomadRemoteQueryError { code: "missing_identity_hash".into(), @@ -4719,6 +4786,8 @@ mod announce_display_name_tests { raw_error: None, elapsed_ms: None, tried_interfaces: None, + failover_rounds: None, + last_iface: None, }); assert_eq!(without["ok"], false); assert_eq!(without["error"], "missing_identity_hash"); diff --git a/reticulum-sidecar/src/stack/lxmf_outbound.rs b/reticulum-sidecar/src/stack/lxmf_outbound.rs index 45bce1f8c..492d7a03f 100644 --- a/reticulum-sidecar/src/stack/lxmf_outbound.rs +++ b/reticulum-sidecar/src/stack/lxmf_outbound.rs @@ -19,11 +19,33 @@ use rns_transport::messages::{TransportMessage, TransportQuery}; use tokio::sync::broadcast; use tokio::sync::mpsc; +use super::super::path_failover::{ + IFACE_SUPPRESS_SECS, push_tried_iface, should_retry_direct_path_failover, +}; use super::{lxmf_payload_from_message, parse_hash16}; const PATH_REQUEST_BACKOFF_SECS: f64 = 20.0; const PATH_REQUEST_MAX_ATTEMPTS: u32 = 12; +/// Per-message Direct path exhaustion before preferred-PN fallback. +#[derive(Debug, Clone, Default)] +struct DirectPathFailoverState { + rounds: u8, + blocked_ifaces: Vec, + blocked_vias: Vec, + tried_interfaces: Vec, +} + +/// One GetPathTable row mirrored into the outbound driver cache. +#[derive(Debug, Clone)] +pub struct PathTableRoute { + pub hash: [u8; 16], + pub hops: u8, + pub hex_key: String, + pub interface: Option, + pub via: Option, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum PathRequestDecision { Send, @@ -99,9 +121,15 @@ pub struct LxmfOutboundDriver { /// Eviction must not remove these while Establishing. pinned_identities: HashMap, path_table_hashes: HashSet, + /// Last known path interface name per destination (from GetPathTable). + path_interfaces: HashMap<[u8; 16], String>, + /// Last known next-hop via hash hex per destination. + path_vias: HashMap<[u8; 16], String>, path_request_gate: PathRequestGate, /// Message hashes that already consumed the one-shot Direct→PN fallback. pn_fallback_attempted: HashSet<[u8; 32]>, + /// Direct link failures still exhausting alternate path slots / ifaces. + direct_path_failovers: HashMap<[u8; 32], DirectPathFailoverState>, /// When set, remote propagation sync holds a Link to this dest — do not race deposits. propagation_sync_target: Option<[u8; 16]>, self_lxmf_hash: String, @@ -127,8 +155,11 @@ impl LxmfOutboundDriver { known_identities: HashMap::new(), pinned_identities: HashMap::new(), path_table_hashes: HashSet::new(), + path_interfaces: HashMap::new(), + path_vias: HashMap::new(), path_request_gate: PathRequestGate::new(), pn_fallback_attempted: HashSet::new(), + direct_path_failovers: HashMap::new(), propagation_sync_target: None, self_lxmf_hash: self_lxmf_hash.clone(), self_display_name, @@ -193,13 +224,32 @@ impl LxmfOutboundDriver { router.set_outbound_propagation_node(hash); } - pub fn update_path_table(&mut self, entries: &[([u8; 16], u8, String)]) { + /// Refresh local path cache from transport GetPathTable rows. + pub fn update_path_table(&mut self, entries: &[PathTableRoute]) { self.route_hops.clear(); self.path_table_hashes.clear(); - for (hash, hops, hex_key) in entries { - self.route_hops.insert(*hash, (*hops).max(1)); - self.path_table_hashes.insert(hex_key.to_lowercase()); - self.path_request_gate.clear_destination(*hash); + self.path_interfaces.clear(); + self.path_vias.clear(); + for entry in entries { + self.route_hops.insert(entry.hash, entry.hops.max(1)); + self.path_table_hashes.insert(entry.hex_key.to_lowercase()); + self.path_request_gate.clear_destination(entry.hash); + if let Some(name) = entry + .interface + .as_ref() + .map(|s| s.trim()) + .filter(|s| !s.is_empty()) + { + self.path_interfaces.insert(entry.hash, name.to_string()); + } + if let Some(via_hex) = entry + .via + .as_ref() + .map(|s| s.trim()) + .filter(|s| !s.is_empty()) + { + self.path_vias.insert(entry.hash, via_hex.to_string()); + } } } @@ -209,6 +259,8 @@ impl LxmfOutboundDriver { self.path_table_hashes.remove(&key); if let Ok(dest) = parse_hash16(&key) { self.route_hops.remove(&dest); + self.path_interfaces.remove(&dest); + self.path_vias.remove(&dest); self.path_request_gate.clear_destination(dest); } } @@ -614,6 +666,7 @@ impl LxmfOutboundDriver { .pending_outbound .retain(|m| m.hash != Some(msg_hash) && m.message_id != Some(msg_hash)); self.remember_pn_fallback(msg_hash); + self.direct_path_failovers.remove(&msg_hash); message.method = DeliveryMethod::Propagated; message.delivery_attempts = 0; message.next_delivery_attempt = 0.0; @@ -697,6 +750,7 @@ impl LxmfOutboundDriver { None }; self.pn_fallback_attempted.remove(&hash); + self.direct_path_failovers.remove(&hash); let _ = router.mark_outbound_delivered(&hash); emit_outbound_status_by_hash(event_tx, &hash, "delivered", method); } @@ -741,6 +795,19 @@ impl LxmfOutboundDriver { ); return; } + // Exhaust alternate path slots / live ifaces before Direct→PN fallback. + let message = if message.method == DeliveryMethod::Direct + && is_retryable_link_delivery_failure(&reason) + { + match self.requeue_direct_after_path_failover( + router, event_tx, message, dest_hash, &reason, + ) { + Ok(()) => return, + Err(message) => *message, + } + } else { + message + }; match self.try_requeue_via_propagation(router, event_tx, message) { Ok(()) => {} Err(message) => self.emit_outbound_failed(router, event_tx, *message), @@ -749,6 +816,97 @@ impl LxmfOutboundDriver { } } + /// Suppress the dead iface/via, RequestPath, and re-queue Direct while failover + /// budget remains. `Ok(())` = re-queued; `Err(message)` = fall through to PN fallback. + fn requeue_direct_after_path_failover( + &mut self, + router: &mut LxmRouter, + event_tx: &broadcast::Sender, + mut message: LxMessage, + dest_hash: [u8; 16], + reason: &str, + ) -> Result<(), Box> { + let Some(msg_hash) = message.hash.or(message.message_id) else { + return Err(Box::new(message)); + }; + let iface = self.path_interfaces.get(&dest_hash).cloned(); + let via = self.path_vias.get(&dest_hash).cloned(); + let failover = { + let state = self.direct_path_failovers.entry(msg_hash).or_default(); + if should_retry_direct_path_failover(state.rounds) { + push_tried_iface(&mut state.tried_interfaces, iface.as_deref()); + if let Some(name) = iface.clone() { + if !state + .blocked_ifaces + .iter() + .any(|b| b.eq_ignore_ascii_case(&name)) + { + state.blocked_ifaces.push(name); + } + } + if let Some(via_hex) = via.clone() { + if !state + .blocked_vias + .iter() + .any(|b| b.eq_ignore_ascii_case(&via_hex)) + { + state.blocked_vias.push(via_hex); + } + } + state.rounds = state.rounds.saturating_add(1); + Ok(( + state.rounds, + state.tried_interfaces.clone(), + state.blocked_vias.clone(), + )) + } else { + Err((state.rounds, state.tried_interfaces.clone())) + } + }; + let (rounds, tried, vias_to_drop) = match failover { + Ok(v) => v, + Err((rounds, tried)) => { + tracing::info!( + dest = %hex::encode(dest_hash), + msg = %hex::encode(msg_hash), + rounds, + tried = ?tried, + reason, + "Direct path failover exhausted; allowing preferred-PN fallback" + ); + self.direct_path_failovers.remove(&msg_hash); + return Err(Box::new(message)); + } + }; + queue_path_failover_queries(&self.transport_tx, dest_hash, &vias_to_drop, reason); + self.clear_path_to(&hex::encode(dest_hash)); + + let now = now_f64(); + message.method = DeliveryMethod::Direct; + message.last_delivery_attempt = now; + message.next_delivery_attempt = now + f64::from(PATH_REQUEST_WAIT as u32); + tracing::info!( + dest = %hex::encode(dest_hash), + msg = %hex::encode(msg_hash), + rounds, + tried = ?tried, + reason, + "Direct path failover: suppress/drop via + RequestPath; re-queuing Direct" + ); + emit_outbound_status_detailed( + event_tx, + Some(serde_json::Value::String(hex::encode(msg_hash))), + Some(serde_json::Value::String(hex::encode(dest_hash))), + "sending", + Some("direct"), + iface, + Some(tried), + Some(rounds), + ); + router.send(message); + Ok(()) + } + /// Re-queue a Propagated deposit after a retryable link failure (lxmd parity). fn requeue_propagated_after_link_failure( &mut self, @@ -895,6 +1053,29 @@ pub fn emit_outbound_status_with_via( status: &str, delivery_method: Option<&str>, sent_via: Option, +) { + emit_outbound_status_detailed( + event_tx, + message_hash, + to_hash, + status, + delivery_method, + sent_via, + None, + None, + ); +} + +#[allow(clippy::too_many_arguments)] // status frame fields travel together +fn emit_outbound_status_detailed( + event_tx: &broadcast::Sender, + message_hash: Option, + to_hash: Option, + status: &str, + delivery_method: Option<&str>, + sent_via: Option, + tried_interfaces: Option>, + failover_rounds: Option, ) { let mut payload = serde_json::Map::new(); if let Some(h) = message_hash { @@ -913,6 +1094,12 @@ pub fn emit_outbound_status_with_via( if let Some(via) = sent_via { payload.insert("sent_via".into(), serde_json::Value::String(via)); } + if let Some(ifaces) = tried_interfaces.filter(|v| !v.is_empty()) { + payload.insert("tried_interfaces".into(), serde_json::json!(ifaces)); + } + if let Some(rounds) = failover_rounds { + payload.insert("failover_rounds".into(), serde_json::json!(rounds)); + } let frame = serde_json::json!({ "type": "lxmf_outbound_status", "payload": payload, @@ -1019,6 +1206,33 @@ fn try_queue_path_request( .is_ok() } +/// Fire-and-forget suppress + DropAllVia + DropPath + RequestPath for Direct failover. +fn queue_path_failover_queries( + transport_tx: &mpsc::Sender, + dest: [u8; 16], + vias_to_drop: &[String], + reason: &str, +) { + let (response_tx, _response_rx) = tokio::sync::oneshot::channel(); + let _ = transport_tx.try_send(TransportMessage::Rpc { + query: TransportQuery::SuppressCurrentPathInterface { + dest, + duration: IFACE_SUPPRESS_SECS, + }, + response_tx, + }); + for via_hex in vias_to_drop { + if let Ok(next_hop) = parse_hash16(via_hex) { + let (response_tx, _response_rx) = tokio::sync::oneshot::channel(); + let _ = transport_tx.try_send(TransportMessage::Rpc { + query: TransportQuery::DropAllVia { next_hop }, + response_tx, + }); + } + } + let _ = try_queue_path_request(transport_tx, dest, true, reason); +} + pub fn parse_propagation_hash(hex_str: &str) -> Option<[u8; 16]> { parse_hash16(hex_str).ok() } @@ -1088,17 +1302,38 @@ mod tests { let dest_hash = dest(0xab); let dest_hex = hex::encode(dest_hash); // Stale cached route (5 hops). - driver.update_path_table(&[(dest_hash, 5, dest_hex.clone())]); + driver.update_path_table(&[PathTableRoute { + hash: dest_hash, + hops: 5, + hex_key: dest_hex.clone(), + interface: Some("TTP_TCP".into()), + via: Some("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".into()), + }]); assert!(driver.has_path_to(&dest_hex)); + assert_eq!( + driver.path_interfaces.get(&dest_hash).map(String::as_str), + Some("TTP_TCP") + ); // Force refresh: drop local cache (transport DropPath happens in live.rs). driver.clear_path_to(&dest_hex); assert!(!driver.has_path_to(&dest_hex)); + assert!(!driver.path_interfaces.contains_key(&dest_hash)); // Fresh route response reinstalls with updated hops. - driver.update_path_table(&[(dest_hash, 2, dest_hex.clone())]); + driver.update_path_table(&[PathTableRoute { + hash: dest_hash, + hops: 2, + hex_key: dest_hex.clone(), + interface: Some("Local Transport Pi".into()), + via: Some("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".into()), + }]); assert!(driver.has_path_to(&dest_hex)); assert_eq!(driver.route_hops.get(&dest_hash).copied(), Some(2)); + assert_eq!( + driver.path_interfaces.get(&dest_hash).map(String::as_str), + Some("Local Transport Pi") + ); } #[test] diff --git a/reticulum-sidecar/src/stack/mod.rs b/reticulum-sidecar/src/stack/mod.rs index 53db23976..7b750005d 100644 --- a/reticulum-sidecar/src/stack/mod.rs +++ b/reticulum-sidecar/src/stack/mod.rs @@ -16,6 +16,7 @@ mod nomad_link_errors; mod nomad_request_payload; mod nomad_timeouts; mod packet_log; +mod path_failover; mod path_medium; mod path_speed; mod persistence; diff --git a/reticulum-sidecar/src/stack/path_failover.rs b/reticulum-sidecar/src/stack/path_failover.rs new file mode 100644 index 000000000..f666d2b9f --- /dev/null +++ b/reticulum-sidecar/src/stack/path_failover.rs @@ -0,0 +1,385 @@ +//! Shared path-slot / via failover helpers for Nomad Links and LXMF Direct. +//! +//! Exhaust ranked path slots and other live interfaces after a dead next-hop +//! before giving up (Nomad) or falling back to preferred PN (LXMF Direct). + +use std::time::Duration; + +use crate::stack::types::InterfaceRow; + +/// How long to reject the failed path interface after a link failure so an +/// alternate hub slot can become active. +pub const IFACE_SUPPRESS_SECS: f64 = 120.0; + +/// Wait for a path with a different via/iface after DropAllVia + suppress. +pub const VIA_FAILOVER_PROBE_WAIT: Duration = Duration::from_secs(8); + +/// Extra RequestPath wait when other live interfaces remain but no slot appeared. +pub const VIA_FAILOVER_EXTRA_PROBE_WAIT: Duration = Duration::from_secs(8); + +/// Max failovers after the first link failure (total tries = 1 + this). +pub const MAX_VIA_FAILOVERS: u8 = 2; + +/// Poll interval while waiting for an alternate path slot. +pub const VIA_FAILOVER_POLL_INTERVAL: Duration = Duration::from_millis(200); + +/// One usable path-table slot for the next Link / Direct attempt. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PathSlotCandidate { + pub hops: u8, + pub iface: Option, + pub via: Option, +} + +/// Active via_hash from ranked path slots (first active non-empty via). +pub fn active_via_hash_from_slots(slots: &[serde_json::Value]) -> Option { + slots.iter().find_map(|slot| { + let active = slot + .get("active") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false); + if !active { + return None; + } + slot.get("via_hash") + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()) + .map(str::to_string) + }) +} + +/// Truncate a via hash for progress / log prefixes. +pub fn via_prefix(via: Option<&str>) -> Option { + via.map(|v| v.chars().take(8).collect()) +} + +fn iface_blocked(iface: Option<&str>, blocked_ifaces: &[String]) -> bool { + iface.is_some_and(|i| blocked_ifaces.iter().any(|b| b.eq_ignore_ascii_case(i))) +} + +fn via_blocked(via: Option<&str>, blocked_vias: &[String]) -> bool { + via.is_some_and(|v| blocked_vias.iter().any(|b| b.eq_ignore_ascii_case(v))) +} + +/// Parse hops / interface / via from a path-slot JSON object. +pub fn slot_candidate(slot: &serde_json::Value) -> Option { + let hops = slot + .get("hops") + .and_then(serde_json::Value::as_u64) + .map(|h| h as u8)?; + let iface = slot + .get("interface") + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()) + .map(str::to_string); + let via = slot + .get("via_hash") + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()) + .map(str::to_string); + Some(PathSlotCandidate { hops, iface, via }) +} + +fn slot_is_unblocked( + cand: &PathSlotCandidate, + blocked_ifaces: &[String], + blocked_vias: &[String], + failed_via: Option<&str>, +) -> bool { + if iface_blocked(cand.iface.as_deref(), blocked_ifaces) { + return false; + } + if via_blocked(cand.via.as_deref(), blocked_vias) { + return false; + } + if failed_via.is_some_and(|fv| { + cand.via + .as_deref() + .is_some_and(|v| v.eq_ignore_ascii_case(fv)) + }) { + return false; + } + true +} + +fn iface_preferred(iface: Option<&str>, prefer_ifaces: &[String]) -> bool { + match iface { + Some(i) if !prefer_ifaces.is_empty() => { + prefer_ifaces.iter().any(|p| p.eq_ignore_ascii_case(i)) + } + _ => false, + } +} + +/// Pick the best unblocked path slot. +/// +/// Preference order: +/// 1. Unblocked slots on `prefer_ifaces` (other live hubs / RF) +/// 2. Any other unblocked slot (ranked backups) +/// +/// Active vs backup order from the transport is preserved within each tier +/// (slots are scanned in list order). +pub fn select_unblocked_slot( + slots: &[serde_json::Value], + blocked_ifaces: &[String], + blocked_vias: &[String], + failed_via: Option<&str>, + prefer_ifaces: &[String], +) -> Option { + let mut fallback: Option = None; + for slot in slots { + let Some(cand) = slot_candidate(slot) else { + continue; + }; + if !slot_is_unblocked(&cand, blocked_ifaces, blocked_vias, failed_via) { + continue; + } + if iface_preferred(cand.iface.as_deref(), prefer_ifaces) { + return Some(cand); + } + if fallback.is_none() { + fallback = Some(cand); + } + } + fallback +} + +/// Enabled interfaces that look live (up/connected/online/running). +pub fn live_interface_names(interfaces: &[InterfaceRow]) -> Vec { + interfaces + .iter() + .filter(|iface| iface.enabled && interface_status_live(&iface.status)) + .filter(|iface| !iface.name.trim().is_empty()) + .map(|iface| iface.name.clone()) + .collect() +} + +fn interface_status_live(status: &str) -> bool { + matches!( + status.to_ascii_lowercase().as_str(), + "up" | "connected" | "online" | "running" + ) +} + +/// Live interface names that are not in the blocked set (candidates for rediscovery). +pub fn remaining_live_ifaces(live_ifaces: &[String], blocked_ifaces: &[String]) -> Vec { + live_ifaces + .iter() + .filter(|n| !blocked_ifaces.iter().any(|b| b.eq_ignore_ascii_case(n))) + .cloned() + .collect() +} + +/// True when Direct LXMF should attempt another path before preferred-PN fallback. +pub fn should_retry_direct_path_failover(rounds_already: u8) -> bool { + rounds_already < MAX_VIA_FAILOVERS +} + +/// Merge a newly tried interface name into the diagnostics list (case-insensitive dedupe). +pub fn push_tried_iface(tried: &mut Vec, iface: Option<&str>) { + let Some(name) = iface.map(str::trim).filter(|s| !s.is_empty()) else { + return; + }; + if tried.iter().any(|t| t.eq_ignore_ascii_case(name)) { + return; + } + tried.push(name.to_string()); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::stack::types::interface_discovery_defaults; + + fn slot(active: bool, hops: u8, iface: &str, via: &str) -> serde_json::Value { + serde_json::json!({ + "active": active, + "hops": hops, + "interface": iface, + "via_hash": via, + }) + } + + fn iface_row(name: &str, enabled: bool, status: &str) -> InterfaceRow { + let ( + discoverable, + latitude, + longitude, + height, + discovery_name, + announce_interval_min, + connectable, + reachable_on, + ) = interface_discovery_defaults(); + InterfaceRow { + id: name.to_lowercase().replace(' ', "-"), + name: name.into(), + iface_type: "tcp".into(), + enabled, + status: status.into(), + host: None, + port: None, + preset: None, + serial_port: None, + frequency: None, + bandwidth: None, + txpower: None, + spreading_factor: None, + coding_rate: None, + callsign: None, + id_interval: None, + mode: None, + seed_addresses: vec![], + discoverable, + latitude, + longitude, + height, + discovery_name, + announce_interval_min, + connectable, + reachable_on, + network_name: None, + passphrase: None, + extra_config: std::collections::HashMap::default(), + } + } + + #[test] + fn active_via_hash_from_slots_skips_inactive_and_empty() { + assert_eq!(active_via_hash_from_slots(&[]), None); + assert_eq!( + active_via_hash_from_slots(&[serde_json::json!({ + "active": true, + "via_hash": "", + })]), + None + ); + assert_eq!( + active_via_hash_from_slots(&[ + slot(false, 4, "TTP_TCP", "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + slot(true, 4, "TTP_TCP", "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"), + ]), + Some("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".into()) + ); + } + + #[test] + fn via_prefix_handles_none_and_truncates() { + assert_eq!(via_prefix(None), None); + assert_eq!( + via_prefix(Some("abcdefghijklmnop")), + Some("abcdefgh".into()) + ); + assert_eq!(via_prefix(Some("abcd")), Some("abcd".into())); + } + + #[test] + fn select_unblocked_prefers_other_live_iface_over_same_hub_backup() { + let slots = [ + slot(true, 4, "TTP_TCP", "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + slot(false, 5, "TTP_TCP", "cccccccccccccccccccccccccccccccc"), + slot( + false, + 3, + "Local Transport Pi", + "dddddddddddddddddddddddddddddddd", + ), + ]; + let blocked_ifaces = vec!["TTP_TCP".into()]; + let blocked_vias = vec!["aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".into()]; + let prefer = vec!["Local Transport Pi".into()]; + let found = select_unblocked_slot( + &slots, + &blocked_ifaces, + &blocked_vias, + Some("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + &prefer, + ) + .expect("local pi slot"); + assert_eq!(found.iface.as_deref(), Some("Local Transport Pi")); + assert_eq!(found.hops, 3); + } + + #[test] + fn select_unblocked_rejects_blocked_via_even_on_other_iface() { + let via = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + let slots = [ + slot(true, 4, "TTP_TCP", via), + slot(false, 3, "Local Transport Pi", via), + ]; + let blocked_ifaces = vec!["TTP_TCP".into()]; + let blocked_vias = vec![via.into()]; + let prefer = vec!["Local Transport Pi".into()]; + assert!( + select_unblocked_slot(&slots, &blocked_ifaces, &blocked_vias, Some(via), &prefer) + .is_none() + ); + } + + #[test] + fn select_unblocked_falls_back_to_any_unblocked_slot() { + let slots = [ + slot(true, 4, "TTP_TCP", "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + slot(false, 6, "Ratspeak", "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee"), + ]; + let blocked_ifaces = vec!["TTP_TCP".into()]; + let blocked_vias = vec!["aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".into()]; + // Prefer list empty / no match — still take Ratspeak backup. + let found = select_unblocked_slot( + &slots, + &blocked_ifaces, + &blocked_vias, + Some("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + &[], + ) + .expect("backup"); + assert_eq!(found.iface.as_deref(), Some("Ratspeak")); + } + + #[test] + fn live_interface_names_filters_enabled_up() { + let rows = [ + iface_row("TTP_TCP", true, "up"), + iface_row("Ratspeak 2", false, "down"), + iface_row("Local Transport Pi", true, "connected"), + iface_row("Auto", true, "down"), + ]; + let names = live_interface_names(&rows); + assert_eq!( + names, + vec!["TTP_TCP".to_string(), "Local Transport Pi".to_string()] + ); + } + + #[test] + fn remaining_live_ifaces_excludes_blocked() { + let live = vec!["TTP_TCP".into(), "Local Transport Pi".into()]; + let blocked = vec!["TTP_TCP".into()]; + assert_eq!( + remaining_live_ifaces(&live, &blocked), + vec!["Local Transport Pi".to_string()] + ); + } + + #[test] + fn should_retry_direct_path_failover_caps_rounds() { + assert!(should_retry_direct_path_failover(0)); + assert!(should_retry_direct_path_failover(1)); + assert!(!should_retry_direct_path_failover(2)); + assert!(!should_retry_direct_path_failover(MAX_VIA_FAILOVERS)); + } + + #[test] + fn push_tried_iface_dedupes_case_insensitive() { + let mut tried = Vec::new(); + push_tried_iface(&mut tried, Some("TTP_TCP")); + push_tried_iface(&mut tried, Some("ttp_tcp")); + push_tried_iface(&mut tried, Some("Local Transport Pi")); + push_tried_iface(&mut tried, None); + push_tried_iface(&mut tried, Some(" ")); + assert_eq!( + tried, + vec!["TTP_TCP".to_string(), "Local Transport Pi".to_string()] + ); + } +} diff --git a/src/renderer/stores/nomadNetworkStore.test.ts b/src/renderer/stores/nomadNetworkStore.test.ts index bb679f248..15ac8789c 100644 --- a/src/renderer/stores/nomadNetworkStore.test.ts +++ b/src/renderer/stores/nomadNetworkStore.test.ts @@ -201,6 +201,9 @@ describe('nomadNetworkStore', () => { path_ensure_kind: 'rediscovered', elapsed_ms: 18250, raw_error: 'timed out waiting for link proof', + tried_interfaces: ['TTP_TCP', 'Local Transport Pi'], + failover_rounds: 1, + iface: 'Local Transport Pi', }); const res = await useNomadNetworkStore @@ -221,6 +224,9 @@ describe('nomadNetworkStore', () => { expect(failed).toContain('force_path_ok=true'); expect(failed).toContain('path_ensure=rediscovered'); expect(failed).toContain('elapsed_ms=18250'); + expect(failed).toContain('tried_interfaces=TTP_TCP,Local Transport Pi'); + expect(failed).toContain('failover_rounds=1'); + expect(failed).toContain('iface=Local Transport Pi'); expect(failed).toContain('raw=timed out waiting for link proof'); } finally { restore(); diff --git a/src/renderer/stores/nomadNetworkStore.ts b/src/renderer/stores/nomadNetworkStore.ts index 6d4a50e2f..54d86f467 100644 --- a/src/renderer/stores/nomadNetworkStore.ts +++ b/src/renderer/stores/nomadNetworkStore.ts @@ -65,6 +65,9 @@ interface NomadFetchLogDiag { pathEnsureKind?: string; elapsedMs?: number; rawError?: string; + triedInterfaces?: string[]; + failoverRounds?: number; + iface?: string; } function optionalFiniteNumber(value: unknown): number | undefined { @@ -85,12 +88,22 @@ function diagFieldsFromResponse(res: unknown): NomadFetchLogDiag { path_ensure_kind?: unknown; elapsed_ms?: unknown; raw_error?: unknown; + tried_interfaces?: unknown; + failover_rounds?: unknown; + iface?: unknown; }; const rawError = typeof r.raw_error === 'string' ? r.raw_error.trim() : undefined; const pathEnsureKind = typeof r.path_ensure_kind === 'string' && r.path_ensure_kind.trim() ? r.path_ensure_kind.trim() : undefined; + const triedInterfaces = Array.isArray(r.tried_interfaces) + ? r.tried_interfaces + .filter((n): n is string => typeof n === 'string') + .map((n) => n.trim()) + .filter((n) => n.length > 0) + : undefined; + const iface = typeof r.iface === 'string' && r.iface.trim() ? r.iface.trim() : undefined; return { pathHops: optionalFiniteNumber(r.path_hops), linkHops: optionalFiniteNumber(r.link_hops), @@ -100,6 +113,9 @@ function diagFieldsFromResponse(res: unknown): NomadFetchLogDiag { pathEnsureKind, elapsedMs: optionalFiniteNumber(r.elapsed_ms), rawError: rawError || undefined, + triedInterfaces: triedInterfaces?.length ? triedInterfaces : undefined, + failoverRounds: optionalFiniteNumber(r.failover_rounds), + iface, }; } @@ -111,6 +127,11 @@ function appendNomadDiagParts(parts: string[], diag: NomadFetchLogDiag): void { if (diag.forcePathOk != null) parts.push(`force_path_ok=${diag.forcePathOk}`); if (diag.pathEnsureKind) parts.push(`path_ensure=${diag.pathEnsureKind}`); if (diag.elapsedMs != null) parts.push(`elapsed_ms=${diag.elapsedMs}`); + if (diag.triedInterfaces?.length) { + parts.push(`tried_interfaces=${diag.triedInterfaces.join(',')}`); + } + if (diag.failoverRounds != null) parts.push(`failover_rounds=${diag.failoverRounds}`); + if (diag.iface) parts.push(`iface=${diag.iface}`); if (diag.rawError) { parts.push(`raw=${diag.rawError.replace(/[\r\n]+/g, ' ').slice(0, 200)}`); } diff --git a/src/shared/nomad-types.ts b/src/shared/nomad-types.ts index 4292ae26f..0ffa108d3 100644 --- a/src/shared/nomad-types.ts +++ b/src/shared/nomad-types.ts @@ -33,6 +33,10 @@ export interface NomadLinkFailureDiagnostics { raw_error?: string; /** Local interface names tried across via-aware failovers (errors only). */ tried_interfaces?: string[]; + /** In-request via/iface failover rounds completed (errors only). */ + failover_rounds?: number; + /** Last local interface used for the Link attempt (errors only). */ + iface?: string; } export interface NomadPageResponse extends NomadLinkFailureDiagnostics { From 222701af0c1c32afe2092576252c916a3cb3d5cd Mon Sep 17 00:00:00 2001 From: Joey Stanford Date: Sun, 2 Aug 2026 09:25:25 -0600 Subject: [PATCH 2/2] fix(reticulum): tighten multipath failover review feedback Avoid per-poll path-table refreshes, clear Direct failover state on terminal failure, normalize egress/iface diagnostics, and cover exhaustion in tests. --- docs/troubleshooting.md | 2 +- reticulum-sidecar/src/stack/live.rs | 66 ++++++---- reticulum-sidecar/src/stack/lxmf_outbound.rs | 119 +++++++++++++++--- src/renderer/stores/nomadNetworkStore.test.ts | 26 ++++ src/renderer/stores/nomadNetworkStore.ts | 13 +- 5 files changed, 183 insertions(+), 43 deletions(-) diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 4e301cec3..f0127daac 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -1044,7 +1044,7 @@ TCP/network Nomad Links use path-scaled initiator hops (`link_hops = clamp(path_ 2. Rebuild sidecar: `pnpm run reticulum:sidecar:build`, restart stack. 3. Prefer low-hop nodes while testing; hop count is shown in the Nomad list. 4. Match the humanized message to the table above — `path_timeout` / high hops often mean RF reachability limits, not a mesh-client bug. -5. For TCP `link_timeout`, check log fields `path_hops` / `link_hops` / `proof_budget_secs` / `raw=` — UI hop counts can lag the path table; trust `path_hops`. Persistent fails after the full proof budget usually mean the peer/hub did not return LRPROOF. +5. For TCP `link_timeout`, check log fields `tried_interfaces` / `failover_rounds` / `iface` first (primary signal after path failover), then `path_hops` / `link_hops` / `proof_budget_secs` / `raw=` — UI hop counts can lag the path table; trust `path_hops`. Persistent fails after the full proof budget usually mean the peer/hub did not return LRPROOF. ### Reticulum sidecar stops during dev (Vite HMR) diff --git a/reticulum-sidecar/src/stack/live.rs b/reticulum-sidecar/src/stack/live.rs index 3e3dd8334..192fb2d9f 100644 --- a/reticulum-sidecar/src/stack/live.rs +++ b/reticulum-sidecar/src/stack/live.rs @@ -26,7 +26,8 @@ use rns_runtime::lifecycle::ShutdownSignal; use rns_runtime::link_client::LinkClient; use rns_runtime::reticulum; use rns_transport::messages::{ - AnnounceHandlerEvent, TransportMessage, TransportQuery, TransportQueryResponse, + AnnounceHandlerEvent, PathTableRpcEntry, TransportMessage, TransportQuery, + TransportQueryResponse, }; use tokio::sync::{RwLock, broadcast}; @@ -1408,12 +1409,20 @@ impl LiveBridge { &prefer, ); - let _ = self + if self .query_control_timed(TransportQuery::SuppressCurrentPathInterface { dest, duration: path_failover::IFACE_SUPPRESS_SECS, }) - .await; + .await + .is_none() + { + tracing::debug!( + target: "nomad", + dest = %hash_hex, + "path failover: SuppressCurrentPathInterface timed out or failed" + ); + } // Drop every known-bad next hop (not only the currently active slot — // after a timeout the table may flip to another iface sharing an older via). let mut vias_to_drop: Vec = blocked_vias.to_vec(); @@ -1424,9 +1433,18 @@ impl LiveBridge { } for via_hex in &vias_to_drop { if let Ok(next_hop) = parse_hash16(via_hex) { - let _ = self + if self .query_control_timed(TransportQuery::DropAllVia { next_hop }) - .await; + .await + .is_none() + { + tracing::debug!( + target: "nomad", + dest = %hash_hex, + via = %via_hex, + "path failover: DropAllVia timed out or failed" + ); + } } } if let Ok(mut driver) = self.outbound.lock() { @@ -1448,7 +1466,6 @@ impl LiveBridge { let settle_deadline = tokio::time::Instant::now() + path_failover::VIA_FAILOVER_POLL_INTERVAL * 10; while tokio::time::Instant::now() < settle_deadline { - let _ = self.refresh_outbound_path_table().await; let slots = self .path_slots(hash_hex) .await @@ -1461,11 +1478,13 @@ impl LiveBridge { failed_via.as_deref(), &prefer, ) { + let _ = self.refresh_outbound_path_table().await; return Some(found); } tokio::time::sleep(path_failover::VIA_FAILOVER_POLL_INTERVAL).await; } // Backup was known before suppress; return it even if the table is slow. + let _ = self.refresh_outbound_path_table().await; return Some(backup); } @@ -1529,7 +1548,6 @@ impl LiveBridge { ) -> Option { let deadline = tokio::time::Instant::now() + wait; while tokio::time::Instant::now() < deadline { - let _ = self.refresh_outbound_path_table().await; let slots = self .path_slots(hash_hex) .await @@ -1542,6 +1560,8 @@ impl LiveBridge { failed_via, prefer_ifaces, ) { + // Keep LXMF outbound / peer-via caches aligned after path_slots finds a route. + let _ = self.refresh_outbound_path_table().await; return Some(found); } tokio::time::sleep(path_failover::VIA_FAILOVER_POLL_INTERVAL).await; @@ -2385,13 +2405,7 @@ impl LiveBridge { Some( entries .iter() - .map(|e| PathTableRoute { - hash: e.hash, - hops: e.hops, - hex_key: hex::encode(e.hash), - interface: Some(e.interface.clone()).filter(|s| !s.is_empty()), - via: e.via.map(hex::encode), - }) + .map(path_table_route_from_entry) .collect::>(), ) } else { @@ -2544,16 +2558,8 @@ impl LiveBridge { cache.insert(hex::encode(entry.hash), entry.interface.clone()); } } - let path_entries = entries - .iter() - .map(|e| PathTableRoute { - hash: e.hash, - hops: e.hops, - hex_key: hex::encode(e.hash), - interface: Some(e.interface.clone()).filter(|s| !s.is_empty()), - via: e.via.map(hex::encode), - }) - .collect::>(); + let path_entries: Vec = + entries.iter().map(path_table_route_from_entry).collect(); if let Ok(mut driver) = self.outbound.lock() { driver.update_path_table(&path_entries); } @@ -4474,6 +4480,16 @@ const NOMAD_FORCE_PATH_REFRESH_WAIT: Duration = Duration::from_secs(4); /// Strict TCP/network DropPath→RequestPath wait before Link (no stale-accept fall-through). const NOMAD_TCP_PATH_PROBE_WAIT: Duration = Duration::from_secs(5); +fn path_table_route_from_entry(e: &PathTableRpcEntry) -> PathTableRoute { + PathTableRoute { + hash: e.hash, + hops: e.hops, + hex_key: hex::encode(e.hash), + interface: Some(e.interface.clone()).filter(|s| !s.is_empty()), + via: e.via.map(hex::encode), + } +} + #[allow(clippy::too_many_arguments, clippy::result_large_err)] // Nomad Link diagnostics bundle fn finish_nomad_link_result( result: Result, NomadRemoteQueryError>, @@ -4794,6 +4810,8 @@ mod announce_display_name_tests { assert!(without.get("egress").is_none()); assert!(without.get("link_hops").is_none()); assert!(without.get("timeout_secs").is_none()); + assert!(without.get("failover_rounds").is_none()); + assert!(without.get("iface").is_none()); } #[test] diff --git a/reticulum-sidecar/src/stack/lxmf_outbound.rs b/reticulum-sidecar/src/stack/lxmf_outbound.rs index 492d7a03f..c827661f7 100644 --- a/reticulum-sidecar/src/stack/lxmf_outbound.rs +++ b/reticulum-sidecar/src/stack/lxmf_outbound.rs @@ -22,6 +22,7 @@ use tokio::sync::mpsc; use super::super::path_failover::{ IFACE_SUPPRESS_SECS, push_tried_iface, should_retry_direct_path_failover, }; +use super::super::via::classify_interface; use super::{lxmf_payload_from_message, parse_hash16}; const PATH_REQUEST_BACKOFF_SECS: f64 = 20.0; @@ -31,7 +32,6 @@ const PATH_REQUEST_MAX_ATTEMPTS: u32 = 12; #[derive(Debug, Clone, Default)] struct DirectPathFailoverState { rounds: u8, - blocked_ifaces: Vec, blocked_vias: Vec, tried_interfaces: Vec, } @@ -624,6 +624,7 @@ impl LxmfOutboundDriver { ); if let Some(hash) = message.hash.or(message.message_id) { self.pn_fallback_attempted.remove(&hash); + self.direct_path_failovers.remove(&hash); let _ = router.mark_outbound_failed(&hash); emit_outbound_status_by_hash(event_tx, &hash, "failed", Some(method)); } @@ -835,15 +836,6 @@ impl LxmfOutboundDriver { let state = self.direct_path_failovers.entry(msg_hash).or_default(); if should_retry_direct_path_failover(state.rounds) { push_tried_iface(&mut state.tried_interfaces, iface.as_deref()); - if let Some(name) = iface.clone() { - if !state - .blocked_ifaces - .iter() - .any(|b| b.eq_ignore_ascii_case(&name)) - { - state.blocked_ifaces.push(name); - } - } if let Some(via_hex) = via.clone() { if !state .blocked_vias @@ -893,13 +885,14 @@ impl LxmfOutboundDriver { reason, "Direct path failover: suppress/drop via + RequestPath; re-queuing Direct" ); + let sent_via = iface.as_deref().map(classify_interface).map(str::to_string); emit_outbound_status_detailed( event_tx, Some(serde_json::Value::String(hex::encode(msg_hash))), Some(serde_json::Value::String(hex::encode(dest_hash))), "sending", Some("direct"), - iface, + sent_via, Some(tried), Some(rounds), ); @@ -1214,20 +1207,35 @@ fn queue_path_failover_queries( reason: &str, ) { let (response_tx, _response_rx) = tokio::sync::oneshot::channel(); - let _ = transport_tx.try_send(TransportMessage::Rpc { + if let Err(e) = transport_tx.try_send(TransportMessage::Rpc { query: TransportQuery::SuppressCurrentPathInterface { dest, duration: IFACE_SUPPRESS_SECS, }, response_tx, - }); + }) { + tracing::debug!( + dest = %hex::encode(dest), + error = %e, + reason, + "path failover SuppressCurrentPathInterface try_send rejected" + ); + } for via_hex in vias_to_drop { if let Ok(next_hop) = parse_hash16(via_hex) { let (response_tx, _response_rx) = tokio::sync::oneshot::channel(); - let _ = transport_tx.try_send(TransportMessage::Rpc { + if let Err(e) = transport_tx.try_send(TransportMessage::Rpc { query: TransportQuery::DropAllVia { next_hop }, response_tx, - }); + }) { + tracing::debug!( + dest = %hex::encode(dest), + via = %via_hex, + error = %e, + reason, + "path failover DropAllVia try_send rejected" + ); + } } } let _ = try_queue_path_request(transport_tx, dest, true, reason); @@ -1314,11 +1322,16 @@ mod tests { driver.path_interfaces.get(&dest_hash).map(String::as_str), Some("TTP_TCP") ); + assert_eq!( + driver.path_vias.get(&dest_hash).map(String::as_str), + Some("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa") + ); // Force refresh: drop local cache (transport DropPath happens in live.rs). driver.clear_path_to(&dest_hex); assert!(!driver.has_path_to(&dest_hex)); assert!(!driver.path_interfaces.contains_key(&dest_hash)); + assert!(!driver.path_vias.contains_key(&dest_hash)); // Fresh route response reinstalls with updated hops. driver.update_path_table(&[PathTableRoute { @@ -1334,6 +1347,82 @@ mod tests { driver.path_interfaces.get(&dest_hash).map(String::as_str), Some("Local Transport Pi") ); + assert_eq!( + driver.path_vias.get(&dest_hash).map(String::as_str), + Some("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb") + ); + } + + #[test] + fn requeue_direct_after_path_failover_exhausts_then_clears_state() { + use crate::stack::path_failover::MAX_VIA_FAILOVERS; + use lxmf_core::constants::DeliveryMethod; + use lxmf_core::message::LxMessage; + use lxmf_core::router::{LxmRouter, RouterConfig}; + use tokio::sync::broadcast; + + let identity = Identity::new(); + let (tx, mut rx) = mpsc::channel(32); + let mut driver = LxmfOutboundDriver::new(tx, &identity, "aabb".repeat(8), "me".into()); + let dest_hash = dest(0xcd); + let msg_hash = [0x42u8; 32]; + driver.update_path_table(&[PathTableRoute { + hash: dest_hash, + hops: 4, + hex_key: hex::encode(dest_hash), + interface: Some("TTP_TCP".into()), + via: Some("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".into()), + }]); + + let mut router = LxmRouter::new(RouterConfig::default()); + let (event_tx, _event_rx) = broadcast::channel(8); + + let make_msg = || { + let mut msg = LxMessage::new(dest_hash, [1u8; 16], "", "hi", DeliveryMethod::Direct); + msg.hash = Some(msg_hash); + msg + }; + + for round in 1..=MAX_VIA_FAILOVERS { + // Drain queued control messages so the channel stays open. + while rx.try_recv().is_ok() {} + // Reinstall a path so each round has an iface/via to record. + driver.update_path_table(&[PathTableRoute { + hash: dest_hash, + hops: 4, + hex_key: hex::encode(dest_hash), + interface: Some(format!("Hub{round}")), + via: Some(format!("{round:032x}")), + }]); + let result = driver.requeue_direct_after_path_failover( + &mut router, + &event_tx, + make_msg(), + dest_hash, + "timed out waiting for link proof", + ); + assert!(result.is_ok(), "round {round} should re-queue"); + let state = driver + .direct_path_failovers + .get(&msg_hash) + .expect("failover state retained"); + assert_eq!(state.rounds, round); + assert_eq!(state.tried_interfaces.len(), round as usize); + } + + while rx.try_recv().is_ok() {} + let exhausted = driver.requeue_direct_after_path_failover( + &mut router, + &event_tx, + make_msg(), + dest_hash, + "timed out waiting for link proof", + ); + assert!(exhausted.is_err(), "round after MAX should exhaust"); + assert!( + !driver.direct_path_failovers.contains_key(&msg_hash), + "exhausted state must be removed" + ); } #[test] diff --git a/src/renderer/stores/nomadNetworkStore.test.ts b/src/renderer/stores/nomadNetworkStore.test.ts index 15ac8789c..9d5596af6 100644 --- a/src/renderer/stores/nomadNetworkStore.test.ts +++ b/src/renderer/stores/nomadNetworkStore.test.ts @@ -233,6 +233,32 @@ describe('nomadNetworkStore', () => { } }); + it('sanitizes newlines in tried_interfaces and iface before logging', async () => { + const { spy, restore } = mockConsoleWarn(); + try { + getStatus.mockResolvedValue({ running: true, port: 1, pid: 1 }); + fetchReticulumInterfaces.mockResolvedValue([{ type: 'tcp', enabled: true }]); + proxyGet.mockResolvedValue({ + ok: false, + error: 'link_timeout', + tried_interfaces: ['TTP\nTCP', 'Local\r\nPi'], + iface: 'Local\nPi', + }); + + await useNomadNetworkStore.getState().fetchNomadPage('abcdef12', '/page/index.mu'); + const messages = spy.mock.calls + .map((c) => c[0]) + .filter((m): m is string => typeof m === 'string'); + const failed = messages.find((m) => m.includes('[nomadNetworkStore] page fetch failed')); + expect(failed).toBeTruthy(); + expect(failed).toContain('tried_interfaces=TTP TCP,Local Pi'); + expect(failed).toContain('iface=Local Pi'); + expect(failed).not.toMatch(/tried_interfaces=[^\s]*\n/); + } finally { + restore(); + } + }); + it('logs a warning when file fetch returns ok:false', async () => { const { spy, restore } = mockConsoleWarn(); try { diff --git a/src/renderer/stores/nomadNetworkStore.ts b/src/renderer/stores/nomadNetworkStore.ts index 54d86f467..f1f2ea8b4 100644 --- a/src/renderer/stores/nomadNetworkStore.ts +++ b/src/renderer/stores/nomadNetworkStore.ts @@ -97,13 +97,20 @@ function diagFieldsFromResponse(res: unknown): NomadFetchLogDiag { typeof r.path_ensure_kind === 'string' && r.path_ensure_kind.trim() ? r.path_ensure_kind.trim() : undefined; + const sanitizeIfaceName = (value: string): string => + value + .replace(/[\r\n]+/g, ' ') + .trim() + .slice(0, 200); const triedInterfaces = Array.isArray(r.tried_interfaces) ? r.tried_interfaces .filter((n): n is string => typeof n === 'string') - .map((n) => n.trim()) + .map(sanitizeIfaceName) .filter((n) => n.length > 0) : undefined; - const iface = typeof r.iface === 'string' && r.iface.trim() ? r.iface.trim() : undefined; + const iface = + typeof r.iface === 'string' && r.iface.trim() ? sanitizeIfaceName(r.iface) : undefined; + const ifaceOrUndefined = iface && iface.length > 0 ? iface : undefined; return { pathHops: optionalFiniteNumber(r.path_hops), linkHops: optionalFiniteNumber(r.link_hops), @@ -115,7 +122,7 @@ function diagFieldsFromResponse(res: unknown): NomadFetchLogDiag { rawError: rawError || undefined, triedInterfaces: triedInterfaces?.length ? triedInterfaces : undefined, failoverRounds: optionalFiniteNumber(r.failover_rounds), - iface, + iface: ifaceOrUndefined, }; }