From 9610626817952eceeeaa6bfeff48b30d87bd617c Mon Sep 17 00:00:00 2001 From: Michael Neale Date: Fri, 17 Jul 2026 16:55:51 +1000 Subject: [PATCH 1/4] fix(desktop): keep mesh allowlist on transient roster query failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A transient relay failure (e.g. 503) during the periodic roster reconcile was collapsed into an empty roster, which reconcile_roster read as 'roster changed to empty' and restarted the mesh node down to self-only — de-admitting every other member and flapping the node (restart loop) on each relay blip. Remote consumers hit either a self-only allowlist or a node restarting mid-inference, so 'nobody else can use my shared compute'. - resolve_trusted_owner_ids now returns Result: a failed query is Err, distinct from Ok(empty) (a genuinely empty community). - Start/restore paths use resolve_trusted_owner_ids_or_self_only (fail closed to self-only is correct only at start, with no live allowlist to preserve). - reconcile_roster: on query Err, keep the current allowlist and retry next poll instead of restarting. Extracted pure roster_reconcile_action for tests. - Regression tests for the failed-query, unchanged, changed, and genuinely empty cases. Not a revert of #2000 (that consumer-admission fix stays); this hardens the same path against transient relay errors. --- desktop/src-tauri/src/commands/mesh_llm.rs | 31 +++++-- desktop/src-tauri/src/mesh_llm/coordinator.rs | 84 ++++++++++++++++++- 2 files changed, 102 insertions(+), 13 deletions(-) diff --git a/desktop/src-tauri/src/commands/mesh_llm.rs b/desktop/src-tauri/src/commands/mesh_llm.rs index 03fc7c0ac4c..141d80fec8f 100644 --- a/desktop/src-tauri/src/commands/mesh_llm.rs +++ b/desktop/src-tauri/src/commands/mesh_llm.rs @@ -88,12 +88,25 @@ async fn query_mesh_discovery_events(state: &AppState) -> Result Vec { - match query_mesh_discovery_events(state).await { - Ok(events) => mesh_llm::owner_ids_from_events(&events), +/// reporters with the current NIP-43 direct-member list. +/// +/// Returns `Err` when the relay query fails. Callers MUST distinguish this from +/// an `Ok(empty)` roster (a genuinely empty community): a failed query must +/// never be collapsed into "self-only", or a transient relay blip de-admits +/// every other member. `reconcile_roster` relies on this to keep the current +/// allowlist on error instead of restarting the node down to self-only. +pub(crate) async fn resolve_trusted_owner_ids(state: &AppState) -> Result, String> { + let events = query_mesh_discovery_events(state).await?; + Ok(mesh_llm::owner_ids_from_events(&events)) +} + +/// Resolve the roster for an initial node *start*, failing closed to self-only +/// (an empty roster) when the relay query fails. This is safe only at start: +/// there is no established allowlist to preserve yet. The periodic +/// `reconcile_roster` path must NOT use this — it has a live roster to keep. +pub(crate) async fn resolve_trusted_owner_ids_or_self_only(state: &AppState) -> Vec { + match resolve_trusted_owner_ids(state).await { + Ok(owners) => owners, Err(error) => { eprintln!("buzz-mesh: roster query failed; allowing only this node: {error}"); Vec::new() @@ -117,7 +130,7 @@ pub(crate) async fn restore_mesh_sharing(app: &AppHandle, state: &AppState) -> C model_id: Some(config.model_id), max_vram_gb: config.max_vram_gb, join_token: None, - trusted_owner_ids: Some(resolve_trusted_owner_ids(state).await), + trusted_owner_ids: Some(resolve_trusted_owner_ids_or_self_only(state).await), }; let started = mesh_llm::DesktopMeshRuntime::start(request) .await @@ -137,7 +150,7 @@ pub async fn mesh_start_node( // Frontend requests never carry a roster; resolve it here so every // UI-started node enforces the member allowlist. if request.trusted_owner_ids.is_none() { - request.trusted_owner_ids = Some(resolve_trusted_owner_ids(&state).await); + request.trusted_owner_ids = Some(resolve_trusted_owner_ids_or_self_only(&state).await); } let mut runtime = state.mesh_llm_runtime.lock().await; if runtime.is_some() { @@ -268,7 +281,7 @@ pub(crate) async fn ensure_client_node_for_model( model_id: None, max_vram_gb: None, join_token: Some(join_token), - trusted_owner_ids: Some(resolve_trusted_owner_ids(state).await), + trusted_owner_ids: Some(resolve_trusted_owner_ids_or_self_only(state).await), }; let mut runtime = state.mesh_llm_runtime.lock().await; if runtime.is_some() { diff --git a/desktop/src-tauri/src/mesh_llm/coordinator.rs b/desktop/src-tauri/src/mesh_llm/coordinator.rs index ad67fd62f5d..5f59ced8210 100644 --- a/desktop/src-tauri/src/mesh_llm/coordinator.rs +++ b/desktop/src-tauri/src/mesh_llm/coordinator.rs @@ -73,6 +73,37 @@ pub async fn start_coordinator(app: AppHandle) { } } +/// Outcome of a roster reconcile decision: either keep the running allowlist +/// untouched, or restart the node with a freshly resolved one. +#[derive(Debug, PartialEq, Eq)] +enum RosterReconcileAction { + Keep, + Restart(Vec), +} + +/// Pure decision for `reconcile_roster`, extracted so the transient-failure +/// invariant is unit-testable without a live relay. +/// +/// Rules: +/// - query failed (`Err`) → `Keep` (never de-admit on a relay blip) +/// - resolved roster == current → `Keep` (no-op) +/// - resolved roster != current → `Restart` with the new roster +fn roster_reconcile_action( + current_owners: &[String], + query: Result, String>, +) -> RosterReconcileAction { + match query { + Err(error) => { + eprintln!( + "buzz-mesh: roster reconcile query failed; keeping current allowlist: {error}" + ); + RosterReconcileAction::Keep + } + Ok(fresh) if fresh == current_owners => RosterReconcileAction::Keep, + Ok(fresh) => RosterReconcileAction::Restart(fresh), + } +} + async fn reconcile_roster(state: &AppState) -> Result<(), String> { let current_request = { let runtime = state.mesh_llm_runtime.lock().await; @@ -84,10 +115,15 @@ async fn reconcile_roster(state: &AppState) -> Result<(), String> { let Some(current_owners) = current_request.trusted_owner_ids.as_ref() else { return Ok(()); }; - let fresh = crate::commands::mesh_llm::resolve_trusted_owner_ids(state).await; - if &fresh == current_owners { - return Ok(()); - } + // A failed roster query must NOT be treated as "the roster became empty": + // doing so would restart the node down to self-only and de-admit every + // other member on a transient relay blip (the flapping restart loop). Keep + // the current allowlist and try again on the next poll. + let query = crate::commands::mesh_llm::resolve_trusted_owner_ids(state).await; + let fresh = match roster_reconcile_action(current_owners, query) { + RosterReconcileAction::Keep => return Ok(()), + RosterReconcileAction::Restart(fresh) => fresh, + }; let mut request = current_request; request.trusted_owner_ids = Some(fresh); @@ -227,6 +263,46 @@ mod tests { use super::*; + // Regression: a transient roster-query failure must never restart the node + // down to self-only. Before the fix, `resolve_trusted_owner_ids` returned + // an empty Vec on error, which `reconcile_roster` read as "roster changed + // to empty" and restarted — de-admitting every other member and flapping + // the node on each relay blip. See #2000 follow-up. + #[test] + fn failed_roster_query_keeps_current_allowlist() { + let current = vec!["owner-a".to_string(), "owner-b".to_string()]; + let action = roster_reconcile_action(¤t, Err("relay returned 503".to_string())); + assert_eq!( + action, + RosterReconcileAction::Keep, + "a failed query must keep the running allowlist, never de-admit members" + ); + } + + #[test] + fn unchanged_roster_is_a_noop() { + let current = vec!["owner-a".to_string()]; + let action = roster_reconcile_action(¤t, Ok(vec!["owner-a".to_string()])); + assert_eq!(action, RosterReconcileAction::Keep); + } + + #[test] + fn genuinely_changed_roster_restarts_with_fresh_owners() { + let current = vec!["owner-a".to_string()]; + let fresh = vec!["owner-a".to_string(), "owner-c".to_string()]; + let action = roster_reconcile_action(¤t, Ok(fresh.clone())); + assert_eq!(action, RosterReconcileAction::Restart(fresh)); + } + + // An `Ok(empty)` — a genuinely empty community, distinct from a failed + // query — is still allowed to shrink the allowlist to self-only. + #[test] + fn genuinely_empty_roster_restarts_to_self_only() { + let current = vec!["owner-a".to_string()]; + let action = roster_reconcile_action(¤t, Ok(Vec::new())); + assert_eq!(action, RosterReconcileAction::Restart(Vec::new())); + } + #[test] fn member_heartbeat_leaves_room_before_admission_status_expires() { assert!( From 7b47e1267ec358415fb1f1828224d720944a46a5 Mon Sep 17 00:00:00 2001 From: Michael Neale Date: Fri, 17 Jul 2026 17:16:19 +1000 Subject: [PATCH 2/4] fix(desktop): harden mesh roster reconcile against transient relay flake Addresses review feedback on the 503-flap fix (#1 + #2): #1 Missing membership snapshot no longer collapses to self-only. query_mesh_discovery_events returned Ok(empty) both when the relay returned an authoritative zero-member snapshot AND when no kind:13534 snapshot came back at all (a transient gap / replication lag). The relay publishes an explicit 13534 event even for a zero-member community, so absence means the query was incomplete. Added has_membership_snapshot(); a missing snapshot is now surfaced as Err so reconcile keeps the current allowlist instead of flapping the node down to self-only on a successful-but-empty response. #2 Shrink hysteresis. reconcile_roster restarted the node on the first observation of any smaller roster, so a single short-read could drop a member mid-inference. RosterReconcileAction gains AwaitConfirm: a shrink (or empty) must be observed with the same reduced roster on two consecutive polls before teardown; growth (pure additions) still applies immediately. The watcher loop carries the pending-shrink state across polls. Tests: has_membership_snapshot present/empty/missing; growth-immediate, shrink-await, shrink-confirm, shrink-reconfirm-on-change, empty-await-confirm, shrink-then-recovery. 44 mesh tests pass; clippy + fmt clean. #3 (restart rollback + mid-poll runtime-change race) tracked as fast-follow. --- desktop/src-tauri/src/commands/mesh_llm.rs | 11 ++ desktop/src-tauri/src/mesh_llm/coordinator.rs | 151 +++++++++++++++--- desktop/src-tauri/src/mesh_llm/discovery.rs | 14 ++ desktop/src-tauri/src/mesh_llm/mod.rs | 4 +- desktop/src-tauri/src/mesh_llm/mod_tests.rs | 28 ++++ 5 files changed, 181 insertions(+), 27 deletions(-) diff --git a/desktop/src-tauri/src/commands/mesh_llm.rs b/desktop/src-tauri/src/commands/mesh_llm.rs index 141d80fec8f..ad55d950601 100644 --- a/desktop/src-tauri/src/commands/mesh_llm.rs +++ b/desktop/src-tauri/src/commands/mesh_llm.rs @@ -64,6 +64,17 @@ async fn query_mesh_discovery_events(state: &AppState) -> Result> = None; loop { tokio::time::sleep(ROSTER_POLL_INTERVAL).await; let state = roster_app.state::(); - if let Err(error) = reconcile_roster(&state).await { + if let Err(error) = reconcile_roster(&state, &mut pending_shrink).await { eprintln!("buzz-mesh: roster reconcile failed: {error}"); } } @@ -73,56 +76,107 @@ pub async fn start_coordinator(app: AppHandle) { } } -/// Outcome of a roster reconcile decision: either keep the running allowlist -/// untouched, or restart the node with a freshly resolved one. +/// Outcome of a roster reconcile decision. #[derive(Debug, PartialEq, Eq)] enum RosterReconcileAction { + /// Keep the running allowlist untouched (no-op, or a failure we ride out). Keep, + /// Restart the node with a freshly resolved roster. Restart(Vec), + /// Observed a *shrink* (or empty) once. Hold the current allowlist and + /// require the same reduced roster on the next poll before tearing down, + /// so a single transient short-read never drops a member mid-inference. + AwaitConfirm(Vec), +} + +/// Whether `fresh` removes any owner present in `current` (a shrink), as +/// opposed to purely adding owners or leaving the set unchanged. +fn roster_shrinks(current: &[String], fresh: &[String]) -> bool { + current.iter().any(|owner| !fresh.contains(owner)) } /// Pure decision for `reconcile_roster`, extracted so the transient-failure -/// invariant is unit-testable without a live relay. +/// and hysteresis invariants are unit-testable without a live relay. +/// +/// `pending_shrink` is the reduced roster we are waiting to re-confirm (from a +/// prior poll's [`RosterReconcileAction::AwaitConfirm`]), if any. /// /// Rules: -/// - query failed (`Err`) → `Keep` (never de-admit on a relay blip) -/// - resolved roster == current → `Keep` (no-op) -/// - resolved roster != current → `Restart` with the new roster +/// - query failed (`Err`) → `Keep` (never de-admit on a relay blip) +/// - resolved roster == current → `Keep` (no-op) +/// - grows (only additions) → `Restart` immediately (fast admission) +/// - shrinks/empties, first observation → `AwaitConfirm` (hold, re-check next poll) +/// - shrinks/empties, confirmed → `Restart` (same reduced roster twice) fn roster_reconcile_action( current_owners: &[String], + pending_shrink: Option<&[String]>, query: Result, String>, ) -> RosterReconcileAction { - match query { + let fresh = match query { Err(error) => { eprintln!( "buzz-mesh: roster reconcile query failed; keeping current allowlist: {error}" ); - RosterReconcileAction::Keep + return RosterReconcileAction::Keep; } - Ok(fresh) if fresh == current_owners => RosterReconcileAction::Keep, - Ok(fresh) => RosterReconcileAction::Restart(fresh), + Ok(fresh) => fresh, + }; + + if fresh == current_owners { + return RosterReconcileAction::Keep; + } + + // Growth (pure additions) is safe to apply immediately. + if !roster_shrinks(current_owners, &fresh) { + return RosterReconcileAction::Restart(fresh); + } + + // A shrink (including down to empty) must be confirmed across two + // consecutive polls with the *same* reduced roster before we tear down. + match pending_shrink { + Some(pending) if pending == fresh => RosterReconcileAction::Restart(fresh), + _ => RosterReconcileAction::AwaitConfirm(fresh), } } -async fn reconcile_roster(state: &AppState) -> Result<(), String> { +async fn reconcile_roster( + state: &AppState, + pending_shrink: &mut Option>, +) -> Result<(), String> { let current_request = { let runtime = state.mesh_llm_runtime.lock().await; match runtime.as_ref() { Some(runtime) => runtime.start_request().clone(), - None => return Ok(()), + None => { + *pending_shrink = None; + return Ok(()); + } } }; let Some(current_owners) = current_request.trusted_owner_ids.as_ref() else { + *pending_shrink = None; return Ok(()); }; // A failed roster query must NOT be treated as "the roster became empty": // doing so would restart the node down to self-only and de-admit every // other member on a transient relay blip (the flapping restart loop). Keep - // the current allowlist and try again on the next poll. + // the current allowlist and try again on the next poll. A shrink is held + // for one extra poll (hysteresis) so a single short-read never tears down. let query = crate::commands::mesh_llm::resolve_trusted_owner_ids(state).await; - let fresh = match roster_reconcile_action(current_owners, query) { - RosterReconcileAction::Keep => return Ok(()), - RosterReconcileAction::Restart(fresh) => fresh, + let fresh = match roster_reconcile_action(current_owners, pending_shrink.as_deref(), query) { + RosterReconcileAction::Keep => { + *pending_shrink = None; + return Ok(()); + } + RosterReconcileAction::AwaitConfirm(reduced) => { + eprintln!("buzz-mesh: roster shrink observed; awaiting confirmation before restart"); + *pending_shrink = Some(reduced); + return Ok(()); + } + RosterReconcileAction::Restart(fresh) => { + *pending_shrink = None; + fresh + } }; let mut request = current_request; @@ -271,7 +325,7 @@ mod tests { #[test] fn failed_roster_query_keeps_current_allowlist() { let current = vec!["owner-a".to_string(), "owner-b".to_string()]; - let action = roster_reconcile_action(¤t, Err("relay returned 503".to_string())); + let action = roster_reconcile_action(¤t, None, Err("relay returned 503".to_string())); assert_eq!( action, RosterReconcileAction::Keep, @@ -282,25 +336,70 @@ mod tests { #[test] fn unchanged_roster_is_a_noop() { let current = vec!["owner-a".to_string()]; - let action = roster_reconcile_action(¤t, Ok(vec!["owner-a".to_string()])); + let action = roster_reconcile_action(¤t, None, Ok(vec!["owner-a".to_string()])); assert_eq!(action, RosterReconcileAction::Keep); } + // Growth (pure additions) applies immediately — fast admission is fine. #[test] - fn genuinely_changed_roster_restarts_with_fresh_owners() { + fn roster_growth_restarts_immediately() { let current = vec!["owner-a".to_string()]; let fresh = vec!["owner-a".to_string(), "owner-c".to_string()]; - let action = roster_reconcile_action(¤t, Ok(fresh.clone())); + let action = roster_reconcile_action(¤t, None, Ok(fresh.clone())); assert_eq!(action, RosterReconcileAction::Restart(fresh)); } - // An `Ok(empty)` — a genuinely empty community, distinct from a failed - // query — is still allowed to shrink the allowlist to self-only. + // A shrink is NOT applied on first observation — it must be confirmed. #[test] - fn genuinely_empty_roster_restarts_to_self_only() { + fn roster_shrink_awaits_confirmation_first() { + let current = vec!["owner-a".to_string(), "owner-b".to_string()]; + let reduced = vec!["owner-a".to_string()]; + let action = roster_reconcile_action(¤t, None, Ok(reduced.clone())); + assert_eq!(action, RosterReconcileAction::AwaitConfirm(reduced)); + } + + // The same reduced roster on two consecutive polls confirms the shrink. + #[test] + fn roster_shrink_restarts_once_confirmed() { + let current = vec!["owner-a".to_string(), "owner-b".to_string()]; + let reduced = vec!["owner-a".to_string()]; + let action = roster_reconcile_action(¤t, Some(&reduced), Ok(reduced.clone())); + assert_eq!(action, RosterReconcileAction::Restart(reduced)); + } + + // A shrink that changes between polls is not confirmed — it re-holds with + // the newly observed reduced roster instead of tearing down. + #[test] + fn roster_shrink_reconfirms_when_it_changes() { + let current = vec!["a".to_string(), "b".to_string(), "c".to_string()]; + let first_reduced = vec!["a".to_string(), "b".to_string()]; + let second_reduced = vec!["a".to_string()]; + let action = + roster_reconcile_action(¤t, Some(&first_reduced), Ok(second_reduced.clone())); + assert_eq!(action, RosterReconcileAction::AwaitConfirm(second_reduced)); + } + + // A genuinely empty community (Ok(empty), distinct from a failed query) + // still shrinks to self-only — but only after confirmation. + #[test] + fn genuinely_empty_roster_awaits_then_restarts_to_self_only() { let current = vec!["owner-a".to_string()]; - let action = roster_reconcile_action(¤t, Ok(Vec::new())); - assert_eq!(action, RosterReconcileAction::Restart(Vec::new())); + let first = roster_reconcile_action(¤t, None, Ok(Vec::new())); + assert_eq!(first, RosterReconcileAction::AwaitConfirm(Vec::new())); + let empty: Vec = Vec::new(); + let confirmed = roster_reconcile_action(¤t, Some(&empty), Ok(Vec::new())); + assert_eq!(confirmed, RosterReconcileAction::Restart(Vec::new())); + } + + // A shrink followed by recovery to the full roster cancels the teardown. + #[test] + fn roster_shrink_then_recovery_keeps_allowlist() { + let current = vec!["owner-a".to_string(), "owner-b".to_string()]; + let reduced = vec!["owner-a".to_string()]; + let held = roster_reconcile_action(¤t, None, Ok(reduced.clone())); + assert_eq!(held, RosterReconcileAction::AwaitConfirm(reduced.clone())); + let recovered = roster_reconcile_action(¤t, Some(&reduced), Ok(current.clone())); + assert_eq!(recovered, RosterReconcileAction::Keep); } #[test] diff --git a/desktop/src-tauri/src/mesh_llm/discovery.rs b/desktop/src-tauri/src/mesh_llm/discovery.rs index b4624ca7dc1..e4f4fb07f9a 100644 --- a/desktop/src-tauri/src/mesh_llm/discovery.rs +++ b/desktop/src-tauri/src/mesh_llm/discovery.rs @@ -93,6 +93,20 @@ pub(crate) fn current_member_pubkeys(events: &[nostr::Event]) -> Vec { .unwrap_or_default() } +/// Whether the relay actually returned a NIP-43 membership snapshot (kind +/// 13534) in `events`. +/// +/// The relay publishes an explicit membership event even for a zero-member +/// community, so its presence is what makes an empty roster *authoritative*. +/// Callers use this to distinguish "the community genuinely has no members" +/// (snapshot present, zero `member` tags) from "no snapshot came back at all" +/// (a transient relay gap / replication lag). Only the former may shrink the +/// admission roster; the latter must be surfaced as an error so the reconcile +/// loop keeps the current allowlist instead of restarting to self-only. +pub(crate) fn has_membership_snapshot(events: &[nostr::Event]) -> bool { + events.iter().any(|event| event.kind.as_u16() == 13_534) +} + fn owner_id_from_status_event(event: &nostr::Event) -> Option { let content = serde_json::from_str::(&event.content).ok()?; let owner_id = content diff --git a/desktop/src-tauri/src/mesh_llm/mod.rs b/desktop/src-tauri/src/mesh_llm/mod.rs index 59e46492223..bd59716bdf0 100644 --- a/desktop/src-tauri/src/mesh_llm/mod.rs +++ b/desktop/src-tauri/src/mesh_llm/mod.rs @@ -8,7 +8,9 @@ mod discovery; pub use discovery::{ availability_from_events, mesh_status_filter, owner_ids_from_events, relay_membership_filter, }; -pub(crate) use discovery::{current_member_pubkeys, MESH_STATUS_PAGE_SIZE}; +pub(crate) use discovery::{ + current_member_pubkeys, has_membership_snapshot, MESH_STATUS_PAGE_SIZE, +}; use discovery::{device_name_from_status, endpoint_id_from_status, enrich_status_payload_identity}; mod catalog; diff --git a/desktop/src-tauri/src/mesh_llm/mod_tests.rs b/desktop/src-tauri/src/mesh_llm/mod_tests.rs index 32cfae113a1..e169af88864 100644 --- a/desktop/src-tauri/src/mesh_llm/mod_tests.rs +++ b/desktop/src-tauri/src/mesh_llm/mod_tests.rs @@ -214,6 +214,34 @@ fn signed_membership_event(members: &[String]) -> nostr::Event { .expect("test membership event signs") } +#[test] +fn has_membership_snapshot_distinguishes_empty_from_missing() { + // A zero-member community still publishes an explicit kind:13534 event, so + // its presence — not the member count — is what makes an empty roster + // authoritative. No snapshot at all means the query was incomplete. + let zero_member_snapshot = signed_membership_event(&[]); + assert!( + super::has_membership_snapshot(std::slice::from_ref(&zero_member_snapshot)), + "an explicit zero-member snapshot counts as present" + ); + + let member = nostr::Keys::parse(&"1".repeat(64)) + .unwrap() + .public_key() + .to_hex(); + let populated = signed_membership_event(std::slice::from_ref(&member)); + assert!(super::has_membership_snapshot(std::slice::from_ref( + &populated + ))); + + // A response with only status events (or nothing) has no snapshot. + assert!(!super::has_membership_snapshot(&[])); + let status_only = signed_reporter_status(&"2".repeat(64), "owner-x"); + assert!(!super::has_membership_snapshot(std::slice::from_ref( + &status_only + ))); +} + #[test] fn owner_ids_from_events_collects_sorted_deduped_roster() { let secret_a = "1".repeat(64); From 09028bbde113b26660defcba2d7c4985bd999492 Mon Sep 17 00:00:00 2001 From: Michael Neale Date: Fri, 17 Jul 2026 18:01:06 +1000 Subject: [PATCH 3/4] fix(desktop): accept mesh-llm default relays + partial-usable endpoints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Shared compute consumers rejected every stock mesh-llm serving node, failing with "no live member is serving this model" even though discovery found it. Two independent transport-policy bugs, both on the CONSUMER side (a serving node never validates its own advertised endpoint, so only consumers need this): 1. Relay allowlist: IrohRelayMode::Default only accepted relays in iroh's own prod relay map. mesh-llm serving nodes advertise endpoints on mesh-llm's default public relays (*.relay.michaelneale.mesh-llm.iroh.link, see mesh-llm-host-runtime effective_relay_urls / RelayPolicy::DefaultPublic). The two sets never overlap, so relay_allowed() returned false and the serve target was filtered out of availability_from_events. Default mode now accepts iroh prod OR mesh-llm's default relays. 2. All-or-nothing endpoint validation: validate_endpoint_addr bailed on the FIRST unusable transport candidate. A stock mesh-llm endpoint advertises a relay alongside a port-0 direct IP; the port-0 candidate rejected the whole endpoint even though the relay route was usable. Transport candidates are alternatives, not all required — now accept the endpoint if at least one candidate is usable, recording why the rest were dropped. Adds regression tests using real mesh-llm relay URLs and the real relay+port-0 multi-candidate endpoint shape (the existing tests were hermetic: fake relay.example URLs and single clean-address endpoints, so neither bug was observable). --- .../src/mesh_llm/transport_policy.rs | 125 +++++++++++++++++- 1 file changed, 120 insertions(+), 5 deletions(-) diff --git a/desktop/src-tauri/src/mesh_llm/transport_policy.rs b/desktop/src-tauri/src/mesh_llm/transport_policy.rs index e4e7ca9fe1e..037237eceb1 100644 --- a/desktop/src-tauri/src/mesh_llm/transport_policy.rs +++ b/desktop/src-tauri/src/mesh_llm/transport_policy.rs @@ -128,23 +128,73 @@ fn validate_endpoint_addr(addr: &EndpointAddr, mode: &IrohRelayMode) -> anyhow:: "mesh endpoint must contain 1..={MAX_ENDPOINT_TRANSPORT_ADDRS} transport addresses" ); } + // An advertised endpoint routinely carries several transport candidates + // (relay + one or more direct IPs). They are *alternative* routes, not all + // required, so a single unusable candidate (e.g. a port-0 direct IP, which + // mesh-llm advertises alongside a valid relay) must NOT reject the whole + // endpoint — otherwise "find it any which way" collapses the moment one + // candidate is junk. Accept the endpoint if AT LEAST ONE candidate is + // usable; record why the rest were dropped for diagnosis. + let mut usable = 0usize; + let mut rejections: Vec = Vec::new(); for transport in &addr.addrs { match transport { - TransportAddr::Relay(relay) if relay_allowed(relay, mode) => {} + TransportAddr::Relay(relay) if relay_allowed(relay, mode) => usable += 1, TransportAddr::Relay(relay) => { - anyhow::bail!("mesh endpoint advertises unapproved relay URL {relay}") + rejections.push(format!("unapproved relay URL {relay}")); } - TransportAddr::Ip(socket) => validate_direct_socket(*socket)?, - _ => anyhow::bail!("mesh endpoint contains an unsupported transport address"), + TransportAddr::Ip(socket) => match validate_direct_socket(*socket) { + Ok(()) => usable += 1, + Err(error) => rejections.push(error.to_string()), + }, + _ => rejections.push("unsupported transport address".to_string()), } } + if usable == 0 { + anyhow::bail!( + "mesh endpoint has no usable transport address (all {} rejected: {})", + addr.addrs.len(), + rejections.join("; ") + ); + } Ok(()) } +/// mesh-llm's default public relay set (`RelayPolicy::DefaultPublic` in +/// `mesh-llm-host-runtime`). A stock mesh-llm server with no custom relay +/// config advertises endpoints on exactly these relays, so buzz's `Default` +/// mode MUST accept them — otherwise shared compute rejects every out-of-the-box +/// mesh-llm serving node (they are not in iroh's own prod relay map). +/// +/// Kept in sync with `effective_relay_urls(RelayPolicy::DefaultPublic, &[])`. +const MESH_LLM_DEFAULT_RELAYS: &[&str] = &[ + "https://usw1-2.relay.michaelneale.mesh-llm.iroh.link./", + "https://aps1-1.relay.michaelneale.mesh-llm.iroh.link./", +]; + +/// Whether `relay` is one of mesh-llm's baked-in default public relays. +/// Parses each known URL to a `RelayUrl` so comparison is normalization-safe +/// (matches regardless of trailing-dot / trailing-slash formatting). +fn is_mesh_llm_default_relay(relay: &RelayUrl) -> bool { + MESH_LLM_DEFAULT_RELAYS.iter().any(|candidate| { + candidate + .parse::() + .map(|known| &known == relay) + .unwrap_or(false) + }) +} + fn relay_allowed(relay: &RelayUrl, mode: &IrohRelayMode) -> bool { match mode { IrohRelayMode::Disabled => false, - IrohRelayMode::Default => iroh::defaults::prod::default_relay_map().contains(relay), + // `Default` covers both iroh's own production relays AND mesh-llm's + // default public relays. Without the latter, a stock mesh-llm serving + // node is unreachable by default and shared compute silently fails with + // "no live member is serving this model" even though discovery found it. + IrohRelayMode::Default => { + iroh::defaults::prod::default_relay_map().contains(relay) + || is_mesh_llm_default_relay(relay) + } IrohRelayMode::Custom(urls) => urls.contains(relay), } } @@ -213,6 +263,71 @@ mod tests { assert!(validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Disabled).is_err()); } + #[test] + fn default_mode_accepts_meshllm_default_relays() { + // Regression: a stock mesh-llm serving node advertises endpoints on + // mesh-llm's OWN default public relays (not iroh's prod relay map). + // Under `Default` mode these MUST be accepted, or shared compute rejects + // every out-of-the-box mesh-llm server with "no live member is serving + // this model" even though discovery found it. See mesh-llm + // effective_relay_urls(RelayPolicy::DefaultPublic, &[]). + for relay_url in MESH_LLM_DEFAULT_RELAYS { + let relay: RelayUrl = relay_url + .parse() + .unwrap_or_else(|e| panic!("mesh-llm default relay {relay_url:?} must parse: {e}")); + assert!( + relay_allowed(&relay, &IrohRelayMode::Default), + "Default mode must accept mesh-llm default relay {relay_url}" + ); + + // And end-to-end through the advertised-endpoint validator. + let token = endpoint_token_for_test([TransportAddr::Relay(relay)]); + assert!( + validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).is_ok(), + "Default mode must validate an endpoint on mesh-llm default relay {relay_url}" + ); + } + } + + #[test] + fn endpoint_with_one_good_and_one_junk_candidate_is_accepted() { + // Regression: a mesh-llm endpoint advertises a usable relay alongside a + // port-0 direct IP. The junk candidate must NOT reject the whole + // endpoint — as long as one route is usable the endpoint is valid. + let good_relay: RelayUrl = MESH_LLM_DEFAULT_RELAYS[0].parse().unwrap(); + let token = endpoint_token_for_test([ + TransportAddr::Relay(good_relay), + TransportAddr::Ip("180.181.228.108:0".parse().unwrap()), // port 0 = junk + ]); + assert!( + validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).is_ok(), + "endpoint with one usable candidate must be accepted despite a junk one" + ); + } + + #[test] + fn endpoint_with_all_junk_candidates_is_rejected() { + // Guard: if EVERY candidate is unusable, the endpoint must still fail. + let token = endpoint_token_for_test([ + TransportAddr::Ip("180.181.228.108:0".parse().unwrap()), // port 0 + TransportAddr::Ip("127.0.0.1:9337".parse().unwrap()), // loopback + ]); + assert!( + validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).is_err(), + "endpoint with no usable candidate must be rejected" + ); + } + + #[test] + fn default_mode_still_rejects_unknown_relay() { + // Guard the fix doesn't over-open: a relay that is neither iroh-prod nor + // a mesh-llm default must still be rejected under Default mode. + let unknown: RelayUrl = "https://not-a-real-relay.example".parse().unwrap(); + assert!(!relay_allowed(&unknown, &IrohRelayMode::Default)); + let token = endpoint_token_for_test([TransportAddr::Relay(unknown)]); + assert!(validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).is_err()); + } + #[test] fn remote_endpoint_rejects_unsafe_direct_targets_and_oversized_tokens() { for socket in ["127.0.0.1:9337", "169.254.169.254:80", "0.0.0.0:1"] { From ec12c28242351a2f852a1b4843c8480b016934b0 Mon Sep 17 00:00:00 2001 From: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co> Date: Fri, 17 Jul 2026 07:21:44 -0400 Subject: [PATCH 4/4] fix(desktop): sanitize mesh endpoint dial candidates Co-authored-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co> Signed-off-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co> --- desktop/src-tauri/src/mesh_llm/discovery.rs | 9 +- desktop/src-tauri/src/mesh_llm/mod.rs | 14 +- .../src/mesh_llm/transport_policy.rs | 214 +++++++++++++----- 3 files changed, 166 insertions(+), 71 deletions(-) diff --git a/desktop/src-tauri/src/mesh_llm/discovery.rs b/desktop/src-tauri/src/mesh_llm/discovery.rs index e4f4fb07f9a..d7dca6b11bd 100644 --- a/desktop/src-tauri/src/mesh_llm/discovery.rs +++ b/desktop/src-tauri/src/mesh_llm/discovery.rs @@ -214,11 +214,12 @@ pub fn availability_from_events(events: Vec) -> MeshAvailability { .unwrap_or_default() .into_iter() .filter_map(|mut target| { - let endpoint_id = + let validated = super::transport_policy::validate_advertised_endpoint(&target.endpoint_addr) .ok()?; + target.endpoint_addr = validated.join_token; if target.endpoint_id.is_none() { - target.endpoint_id = Some(endpoint_id); + target.endpoint_id = Some(validated.endpoint_id); } if target.device_id.is_none() { target.device_id = target.endpoint_id.clone(); @@ -336,7 +337,9 @@ pub(super) fn device_name_from_status( } fn endpoint_id_from_invite_token(invite_token: &str) -> Option { - super::transport_policy::validate_advertised_endpoint(invite_token).ok() + super::transport_policy::validate_advertised_endpoint(invite_token) + .ok() + .map(|validated| validated.endpoint_id) } fn string_value(value: &serde_json::Value, key: &str) -> Option { diff --git a/desktop/src-tauri/src/mesh_llm/mod.rs b/desktop/src-tauri/src/mesh_llm/mod.rs index bd59716bdf0..80d887bec7d 100644 --- a/desktop/src-tauri/src/mesh_llm/mod.rs +++ b/desktop/src-tauri/src/mesh_llm/mod.rs @@ -272,8 +272,8 @@ async fn ensure_model_downloaded(model: &str) -> anyhow::Result<()> { } impl DesktopMeshRuntime { - pub async fn start(request: StartMeshNodeRequest) -> anyhow::Result { - validate_no_leak_request(&request)?; + pub async fn start(mut request: StartMeshNodeRequest) -> anyhow::Result { + sanitize_no_leak_request(&mut request)?; initialize_mesh_native_runtime().await?; let model_id = request .model_id @@ -422,8 +422,8 @@ impl DesktopMeshRuntime { pub async fn dial_endpoint_addr(&self, endpoint_addr: impl Into) -> anyhow::Result<()> { let endpoint_addr = endpoint_addr.into(); - validate_advertised_endpoint(&endpoint_addr)?; - self.handle.join_token(endpoint_addr).await + let validated = validate_advertised_endpoint(&endpoint_addr)?; + self.handle.join_token(validated.join_token).await } pub async fn installed_models(&self) -> anyhow::Result> { @@ -505,9 +505,9 @@ fn normalized_roster( Some(owners) } -fn validate_no_leak_request(request: &StartMeshNodeRequest) -> anyhow::Result<()> { - if let Some(join_token) = request.join_token.as_deref() { - validate_advertised_endpoint(join_token)?; +fn sanitize_no_leak_request(request: &mut StartMeshNodeRequest) -> anyhow::Result<()> { + if let Some(join_token) = request.join_token.as_mut() { + *join_token = validate_advertised_endpoint(join_token)?.join_token; } Ok(()) } diff --git a/desktop/src-tauri/src/mesh_llm/transport_policy.rs b/desktop/src-tauri/src/mesh_llm/transport_policy.rs index 037237eceb1..122bffd60e4 100644 --- a/desktop/src-tauri/src/mesh_llm/transport_policy.rs +++ b/desktop/src-tauri/src/mesh_llm/transport_policy.rs @@ -72,7 +72,15 @@ fn parse_configured_relay_url(raw: &str) -> anyhow::Result { .map_err(|error| anyhow::anyhow!("invalid iroh relay URL {raw:?}: {error}")) } -pub(super) fn validate_advertised_endpoint(invite_token: &str) -> anyhow::Result { +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) struct ValidatedEndpoint { + pub endpoint_id: String, + pub join_token: String, +} + +pub(super) fn validate_advertised_endpoint( + invite_token: &str, +) -> anyhow::Result { let mode = iroh_relay_mode()?; validate_advertised_endpoint_with_mode(invite_token, &mode) } @@ -80,7 +88,7 @@ pub(super) fn validate_advertised_endpoint(invite_token: &str) -> anyhow::Result pub(super) fn validate_advertised_endpoint_with_mode( invite_token: &str, mode: &IrohRelayMode, -) -> anyhow::Result { +) -> anyhow::Result { let token = invite_token.trim(); if token.is_empty() { anyhow::bail!("mesh invite token is empty"); @@ -92,74 +100,104 @@ pub(super) fn validate_advertised_endpoint_with_mode( .decode(token) .map_err(|error| anyhow::anyhow!("invalid mesh invite encoding: {error}"))?; - let addrs = if let Ok(addr) = serde_json::from_slice::(&payload) { - vec![addr] - } else { - let signed = serde_json::from_slice::(&payload) - .map_err(|error| anyhow::anyhow!("invalid mesh invite payload: {error}"))?; - signed - .verify() - .map_err(|reason| anyhow::anyhow!("invalid signed mesh invite: {}", reason.code()))?; - if signed.serialized_addrs.is_empty() || signed.serialized_addrs.len() > MAX_BOOTSTRAP_ADDRS - { - anyhow::bail!( - "signed mesh invite must contain 1..={MAX_BOOTSTRAP_ADDRS} endpoint addresses" - ); - } - signed - .serialized_addrs - .iter() - .map(|bytes| { - serde_json::from_slice::(bytes) - .map_err(|error| anyhow::anyhow!("invalid signed endpoint address: {error}")) - }) - .collect::>>()? - }; + if let Ok(mut addr) = serde_json::from_slice::(&payload) { + retain_usable_transports(&mut addr, mode)?; + let endpoint_id = addr.id.to_string(); + let join_token = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&addr)?); + return Ok(ValidatedEndpoint { + endpoint_id, + join_token, + }); + } + let signed = serde_json::from_slice::(&payload) + .map_err(|error| anyhow::anyhow!("invalid mesh invite payload: {error}"))?; + signed + .verify() + .map_err(|reason| anyhow::anyhow!("invalid signed mesh invite: {}", reason.code()))?; + if signed.serialized_addrs.is_empty() || signed.serialized_addrs.len() > MAX_BOOTSTRAP_ADDRS { + anyhow::bail!( + "signed mesh invite must contain 1..={MAX_BOOTSTRAP_ADDRS} endpoint addresses" + ); + } + let addrs = signed + .serialized_addrs + .iter() + .map(|bytes| { + serde_json::from_slice::(bytes) + .map_err(|error| anyhow::anyhow!("invalid signed endpoint address: {error}")) + }) + .collect::>>()?; + + // Rewriting a signed token would invalidate its signature. Keep the signed + // envelope intact only when every advertised transport is policy-approved; + // mixed signed tokens fail closed rather than leaking rejected dial targets. for addr in &addrs { - validate_endpoint_addr(addr, mode)?; + validate_signed_transports(addr, mode)?; } - Ok(addrs[0].id.to_string()) + Ok(ValidatedEndpoint { + endpoint_id: addrs[0].id.to_string(), + join_token: token.to_string(), + }) } -fn validate_endpoint_addr(addr: &EndpointAddr, mode: &IrohRelayMode) -> anyhow::Result<()> { - if addr.addrs.is_empty() || addr.addrs.len() > MAX_ENDPOINT_TRANSPORT_ADDRS { +fn retain_usable_transports(addr: &mut EndpointAddr, mode: &IrohRelayMode) -> anyhow::Result<()> { + validate_transport_count(addr)?; + let mut rejections = Vec::new(); + addr.addrs + .retain(|transport| match validate_transport(transport, mode) { + Ok(()) => true, + Err(error) => { + rejections.push(error.to_string()); + false + } + }); + if addr.addrs.is_empty() { anyhow::bail!( - "mesh endpoint must contain 1..={MAX_ENDPOINT_TRANSPORT_ADDRS} transport addresses" + "mesh endpoint has no usable transport address (all rejected: {})", + rejections.join("; ") ); } - // An advertised endpoint routinely carries several transport candidates - // (relay + one or more direct IPs). They are *alternative* routes, not all - // required, so a single unusable candidate (e.g. a port-0 direct IP, which - // mesh-llm advertises alongside a valid relay) must NOT reject the whole - // endpoint — otherwise "find it any which way" collapses the moment one - // candidate is junk. Accept the endpoint if AT LEAST ONE candidate is - // usable; record why the rest were dropped for diagnosis. + Ok(()) +} + +fn validate_signed_transports(addr: &EndpointAddr, mode: &IrohRelayMode) -> anyhow::Result<()> { + validate_transport_count(addr)?; let mut usable = 0usize; - let mut rejections: Vec = Vec::new(); for transport in &addr.addrs { - match transport { - TransportAddr::Relay(relay) if relay_allowed(relay, mode) => usable += 1, - TransportAddr::Relay(relay) => { - rejections.push(format!("unapproved relay URL {relay}")); - } - TransportAddr::Ip(socket) => match validate_direct_socket(*socket) { - Ok(()) => usable += 1, - Err(error) => rejections.push(error.to_string()), - }, - _ => rejections.push("unsupported transport address".to_string()), + match validate_transport(transport, mode) { + Ok(()) => usable += 1, + // mesh-llm currently signs its own port-0 placeholder alongside a + // valid relay. It is non-dialable, so preserving it is safe and + // necessary for stock signed tokens to remain usable. + Err(_) if matches!(transport, TransportAddr::Ip(socket) if socket.port() == 0) => {} + Err(error) => return Err(error), } } if usable == 0 { + anyhow::bail!("signed mesh endpoint has no usable transport address"); + } + Ok(()) +} + +fn validate_transport_count(addr: &EndpointAddr) -> anyhow::Result<()> { + if addr.addrs.is_empty() || addr.addrs.len() > MAX_ENDPOINT_TRANSPORT_ADDRS { anyhow::bail!( - "mesh endpoint has no usable transport address (all {} rejected: {})", - addr.addrs.len(), - rejections.join("; ") + "mesh endpoint must contain 1..={MAX_ENDPOINT_TRANSPORT_ADDRS} transport addresses" ); } Ok(()) } +fn validate_transport(transport: &TransportAddr, mode: &IrohRelayMode) -> anyhow::Result<()> { + match transport { + TransportAddr::Relay(relay) if relay_allowed(relay, mode) => Ok(()), + TransportAddr::Relay(relay) => anyhow::bail!("unapproved relay URL {relay}"), + TransportAddr::Ip(socket) => validate_direct_socket(*socket), + _ => anyhow::bail!("unsupported transport address"), + } +} + /// mesh-llm's default public relay set (`RelayPolicy::DefaultPublic` in /// `mesh-llm-host-runtime`). A stock mesh-llm server with no custom relay /// config advertises endpoints on exactly these relays, so buzz's `Default` @@ -290,18 +328,23 @@ mod tests { } #[test] - fn endpoint_with_one_good_and_one_junk_candidate_is_accepted() { - // Regression: a mesh-llm endpoint advertises a usable relay alongside a - // port-0 direct IP. The junk candidate must NOT reject the whole - // endpoint — as long as one route is usable the endpoint is valid. + fn endpoint_with_one_good_and_one_junk_candidate_is_sanitized() { + // A mesh-llm endpoint can advertise a usable relay alongside an + // unusable direct IP. Keep the endpoint reachable, but never pass the + // rejected candidate through to iroh's parallel dialer. let good_relay: RelayUrl = MESH_LLM_DEFAULT_RELAYS[0].parse().unwrap(); + let unsafe_socket = "169.254.169.254:80".parse().unwrap(); let token = endpoint_token_for_test([ - TransportAddr::Relay(good_relay), - TransportAddr::Ip("180.181.228.108:0".parse().unwrap()), // port 0 = junk + TransportAddr::Relay(good_relay.clone()), + TransportAddr::Ip(unsafe_socket), ]); - assert!( - validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).is_ok(), - "endpoint with one usable candidate must be accepted despite a junk one" + let validated = + validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).unwrap(); + let payload = URL_SAFE_NO_PAD.decode(validated.join_token).unwrap(); + let sanitized: EndpointAddr = serde_json::from_slice(&payload).unwrap(); + assert_eq!( + sanitized.addrs, + [TransportAddr::Relay(good_relay)].into_iter().collect() ); } @@ -378,10 +421,59 @@ mod tests { .expect("sign test bootstrap token"); let token = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&signed).unwrap()); assert_eq!( - validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).unwrap(), + validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default) + .unwrap() + .endpoint_id, endpoint.id.to_string() ); + let good_relay: RelayUrl = MESH_LLM_DEFAULT_RELAYS[0].parse().unwrap(); + let placeholder_endpoint = EndpointAddr { + id: endpoint.id, + addrs: [ + TransportAddr::Relay(good_relay.clone()), + TransportAddr::Ip("180.181.228.108:0".parse().unwrap()), + ] + .into_iter() + .collect(), + }; + let placeholder_signed = SignedBootstrapToken::sign( + vec![serde_json::to_vec(&placeholder_endpoint).unwrap()], + &signed_policy, + None, + &owner, + ) + .expect("sign placeholder test bootstrap token"); + let placeholder_token = + URL_SAFE_NO_PAD.encode(serde_json::to_vec(&placeholder_signed).unwrap()); + assert!( + validate_advertised_endpoint_with_mode(&placeholder_token, &IrohRelayMode::Default) + .is_ok(), + "signed stock token with a port-0 placeholder must remain usable" + ); + + let mixed_endpoint = EndpointAddr { + id: endpoint.id, + addrs: [ + TransportAddr::Relay(good_relay), + TransportAddr::Ip("169.254.169.254:80".parse().unwrap()), + ] + .into_iter() + .collect(), + }; + let mixed_signed = SignedBootstrapToken::sign( + vec![serde_json::to_vec(&mixed_endpoint).unwrap()], + &signed_policy, + None, + &owner, + ) + .expect("sign mixed test bootstrap token"); + let mixed_token = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&mixed_signed).unwrap()); + assert!( + validate_advertised_endpoint_with_mode(&mixed_token, &IrohRelayMode::Default).is_err(), + "signed mixed-candidate tokens must fail closed because they cannot be rewritten" + ); + let mut tampered = signed; tampered.serialized_addrs[0] = serde_json::to_vec(&EndpointAddr { id: endpoint.id,