From 8d33b3c5ebfb7bad14fb8782c92994c5d954ae76 Mon Sep 17 00:00:00 2001 From: zancas Date: Mon, 24 Aug 2026 13:29:06 -0700 Subject: [PATCH 1/3] feat(netutils): report the bootstrap race as typed events The mobile diagnostics surface (zmobs ADR 0007) needs to name which Exit Node each bootstrap step concerns, and the prose lines from start_over cannot carry that identity. This change makes the race narrative typed at its source and keeps the prose rendering for the spawnable binary unchanged. A public BootstrapEvent enum carries five lifecycle variants: DiscoveryStarted, DiscoveryFinished with the candidate count, PullLaunched and PullFailed with the full Exit Node address (the failure text stays prose, as it already is inside RaceEvent), and Connected with the winner. NymProxy::start_observed and NymProxy::start_over_observed report them; start and start_over are now thin renderings over the same path, so the two narrations cannot drift. Internally drive_acq_race reports an AcqStep per launch, failure, and counter change, and connect_across_exit_nodes resolves arm indices to Exit Node addresses, falling back to the index form short_exit_node_name uses when a panicked pull loses its index. The planner's hedging mechanics stay private, and the existing RaceProgress narration is byte-identical. Co-Authored-By: Claude Fable 5 --- zingo-netutils/src/nym_proxy.rs | 187 +++++++++++++++++++++++++++++--- 1 file changed, 174 insertions(+), 13 deletions(-) diff --git a/zingo-netutils/src/nym_proxy.rs b/zingo-netutils/src/nym_proxy.rs index b6fe201f83..abe29f2fdc 100644 --- a/zingo-netutils/src/nym_proxy.rs +++ b/zingo-netutils/src/nym_proxy.rs @@ -61,6 +61,53 @@ fn draw_clutch(mut discovered: Vec) -> Vec { use crate::time::{DISCOVERY_TIMEOUT, NYM_LIFECYCLE_TIMEOUT, PER_ATTEMPT_CONNECT_TIMEOUT}; +/// One step of the bootstrap, carrying the Exit Node it concerns. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum BootstrapEvent { + /// The Exit Node discovery query left for the Nym directory. + DiscoveryStarted, + /// The directory answered the discovery query. + DiscoveryFinished { + /// The count of Exit Nodes the directory advertised. + candidate_count: usize, + }, + /// A pull of this Exit Node launched. + PullLaunched { + /// The Exit Node address the pull races. + exit_node: String, + }, + /// The pull of this Exit Node failed. + PullFailed { + /// The Exit Node address whose pull failed. + exit_node: String, + /// The failure rendered for a human. + error: String, + }, + /// The race kept this Exit Node and the local listener is up. + Connected { + /// The Exit Node address the proxy bound. + exit_node: String, + }, +} + +/// One step of a driven race, reported to the caller as it happens. +enum AcqStep { + /// A pull of this arm index launched. + Launched { + /// The arm index the pull races. + arm: usize, + }, + /// The pull of this arm index failed. + Failed { + /// The arm index whose pull failed. + arm: usize, + /// The failure rendered for a human. + error: String, + }, + /// The race's counters changed. + Progress(RaceProgress), +} + /// Embedded Nym SOCKS5 proxy that routes traffic through the Nym mixnet. /// /// Manages the lifecycle of an in-process Nym SOCKS5 client connected to a @@ -79,8 +126,20 @@ impl NymProxy { /// Start an embedded Nym SOCKS5 proxy over a clutch this call draws for /// itself, for a standalone run with no parent to draw one. pub async fn start() -> Result { + Self::start_observed(|_| {}).await + } + + /// [`Self::start`], reporting each bootstrap step to `on_event` as a + /// typed [`BootstrapEvent`]. + pub async fn start_observed( + mut on_event: impl FnMut(BootstrapEvent), + ) -> Result { + on_event(BootstrapEvent::DiscoveryStarted); let discovered = Self::discover_exit_nodes_at(DEFAULT_NYM_API_URL).await?; - Self::start_over(draw_clutch(discovered), |_| {}).await + on_event(BootstrapEvent::DiscoveryFinished { + candidate_count: discovered.len(), + }); + Self::start_over_observed(draw_clutch(discovered), on_event).await } /// Start over exactly `clutch`, the Exit Node Reservations the parent @@ -88,10 +147,27 @@ impl NymProxy { pub async fn start_over( clutch: Vec, on_progress: impl FnMut(String), + ) -> Result { + Self::start_bounded(clutch, |_| {}, on_progress).await + } + + /// [`Self::start_over`], reporting typed [`BootstrapEvent`]s instead of + /// prose lines. + pub async fn start_over_observed( + clutch: Vec, + on_event: impl FnMut(BootstrapEvent), + ) -> Result { + Self::start_bounded(clutch, on_event, |_| {}).await + } + + async fn start_bounded( + clutch: Vec, + on_event: impl FnMut(BootstrapEvent), + on_progress: impl FnMut(String), ) -> Result { tokio::time::timeout( NYM_LIFECYCLE_TIMEOUT, - Self::start_inner(clutch, on_progress), + Self::start_inner(clutch, on_event, on_progress), ) .await .map_err(|_| { @@ -104,13 +180,14 @@ impl NymProxy { async fn start_inner( clutch: Vec, + on_event: impl FnMut(BootstrapEvent), mut on_progress: impl FnMut(String), ) -> Result { if clutch.is_empty() { return Err(NymProxyError::NoExitNode); } on_progress(format!("racing a clutch of {} exits", clutch.len())); - let mut proxy = Self::connect_across_exit_nodes(&clutch, on_progress).await?; + let mut proxy = Self::connect_across_exit_nodes(&clutch, on_event, on_progress).await?; proxy.clutch = clutch; Ok(proxy) } @@ -121,9 +198,18 @@ impl NymProxy { /// above the SOCKS5 seam. async fn connect_across_exit_nodes( exit_nodes: &[String], + mut on_event: impl FnMut(BootstrapEvent), mut on_progress: impl FnMut(String), ) -> Result { - drive_acq_race( + // A panicked pull can lose its arm index, so the identity lookup + // falls back the way short_exit_node_name does. + let full_exit_node_name = |arm: usize| { + exit_nodes + .get(arm) + .cloned() + .unwrap_or_else(|| format!("exit node {arm}")) + }; + let proxy = drive_acq_race( exit_nodes.len(), exit_nodes.len(), acquisition_launch_policy(), @@ -158,10 +244,23 @@ impl NymProxy { text, ) }, - |progress| on_progress(progress.to_string()), + |step| match step { + AcqStep::Launched { arm } => on_event(BootstrapEvent::PullLaunched { + exit_node: full_exit_node_name(arm), + }), + AcqStep::Failed { arm, error } => on_event(BootstrapEvent::PullFailed { + exit_node: full_exit_node_name(arm), + error, + }), + AcqStep::Progress(progress) => on_progress(progress.to_string()), + }, ) .await - .map_err(acq_race_loss_error) + .map_err(acq_race_loss_error)?; + on_event(BootstrapEvent::Connected { + exit_node: proxy.exit_node().to_string(), + }); + Ok(proxy) } /// The Exit Nodes the Nym directory currently advertises, for @@ -278,7 +377,7 @@ impl NymProxy { } // Each attempt binds its own fresh port, which cannot collide with // the old client's still-bound port. - let new_proxy = Self::connect_across_exit_nodes(&clutch, |_| {}).await?; + let new_proxy = Self::connect_across_exit_nodes(&clutch, |_| {}, |_| {}).await?; // Swap only after the new client succeeded, so a failed reconnect // leaves the old client untouched. @@ -450,7 +549,7 @@ async fn drive_acq_race( launch: F, abandon: D, describe_panic: impl Fn(usize, String) -> E, - mut on_progress: impl FnMut(RaceProgress), + mut on_step: impl FnMut(AcqStep), ) -> Result> where T: Send + 'static, @@ -471,13 +570,15 @@ where pulls: &mut tokio::task::JoinSet<(usize, Result)>, pull_arms: &mut HashMap, hedge_deadline: &mut Option, - lost: &mut bool| { + lost: &mut bool, + launched: &mut Vec| { for action in actions { match action { RaceAction::Launch { arm } => { let pull = launch(arm); let handle = pulls.spawn(async move { (arm, pull.await) }); pull_arms.insert(handle.id(), arm); + launched.push(arm); } RaceAction::SetHedgeTimer(interval) => { *hedge_deadline = Some(tokio::time::Instant::now() + interval); @@ -486,6 +587,9 @@ where } } }; + // The apply closure cannot call the generic `on_step` directly, so it + // records launches and each call site drains them into steps. + let mut launched: Vec = Vec::new(); apply( acq_race.start(), @@ -493,8 +597,12 @@ where &mut pull_arms, &mut hedge_deadline, &mut lost, + &mut launched, ); - on_progress(acq_race.progress()); + for arm in launched.drain(..) { + on_step(AcqStep::Launched { arm }); + } + on_step(AcqStep::Progress(acq_race.progress())); loop { if lost { @@ -543,6 +651,10 @@ where // The planner's event wants a rendered line for its // progress narration; the typed failure is retained // whole for the terminal account. + on_step(AcqStep::Failed { + arm, + error: error.to_string(), + }); apply( acq_race.on_event(RaceEvent::PullFailed { arm, @@ -552,9 +664,13 @@ where &mut pull_arms, &mut hedge_deadline, &mut lost, + &mut launched, ); + for arm in launched.drain(..) { + on_step(AcqStep::Launched { arm }); + } failures.push(error); - on_progress(acq_race.progress()); + on_step(AcqStep::Progress(acq_race.progress())); } } } @@ -568,8 +684,12 @@ where &mut pull_arms, &mut hedge_deadline, &mut lost, + &mut launched, ); - on_progress(acq_race.progress()); + for arm in launched.drain(..) { + on_step(AcqStep::Launched { arm }); + } + on_step(AcqStep::Progress(acq_race.progress())); } } } @@ -815,7 +935,11 @@ mod tests { |arm| async move { Err::<&str, _>(format!("arm {arm} refused")) }, no_abandon, panic_text, - move |progress| sink.lock().unwrap().push(progress.to_string()), + move |step| { + if let AcqStep::Progress(progress) = step { + sink.lock().unwrap().push(progress.to_string()) + } + }, ) .await; let lines = lines.lock().unwrap(); @@ -825,6 +949,43 @@ mod tests { ); } + /// HYPOTHESIS: the driver reports every launch and every failure as a + /// typed step carrying the arm index, so the caller can name the Exit + /// Node each step concerns. Falsified if a launched arm goes + /// unreported, a failure loses its index, or a failure loses its text. + #[tokio::test(start_paused = true)] + async fn the_driver_reports_each_launch_and_failure_with_its_arm() { + let steps = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let sink = std::sync::Arc::clone(&steps); + let _ = drive_acq_race( + RACED_ARMS, + RACED_ARMS, + hedged(RESERVATION_CLUTCH_SIZE), + |arm| async move { Err::<&str, _>(format!("arm {arm} refused")) }, + no_abandon, + panic_text, + move |step| match step { + AcqStep::Launched { arm } => sink.lock().unwrap().push(format!("launched {arm}")), + AcqStep::Failed { arm, error } => { + sink.lock().unwrap().push(format!("failed {arm}: {error}")) + } + AcqStep::Progress(_) => {} + }, + ) + .await; + let steps = steps.lock().unwrap(); + for arm in 0..RACED_ARMS { + assert!( + steps.contains(&format!("launched {arm}")), + "every arm's launch is reported, got {steps:?}" + ); + assert!( + steps.contains(&format!("failed {arm}: arm {arm} refused")), + "every arm's failure carries its index and text, got {steps:?}" + ); + } + } + #[test] fn long_exit_node_names_are_shortened_for_the_summary() { let exit_nodes = vec!["a".repeat(200), "short".to_string()]; From 8fc056a945e24fa5d36d274aed5d1cf5b725fafa Mon Sep 17 00:00:00 2001 From: zancas Date: Mon, 24 Aug 2026 20:59:15 -0700 Subject: [PATCH 2/3] fix(netutils): re-export BootstrapEvent beside NymProxy The enum was public only inside the private nym_proxy module, so the standalone-workspace doc build rejected the intra-doc links and no consumer could name the variants that start_observed reports. Co-Authored-By: Claude Fable 5 --- zingo-netutils/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/zingo-netutils/src/lib.rs b/zingo-netutils/src/lib.rs index 1e711b4e1b..e22e33205e 100644 --- a/zingo-netutils/src/lib.rs +++ b/zingo-netutils/src/lib.rs @@ -93,7 +93,7 @@ pub mod provider; #[cfg(feature = "nym")] mod nym_proxy; #[cfg(feature = "nym")] -pub use nym_proxy::NymProxy; +pub use nym_proxy::{BootstrapEvent, NymProxy}; #[cfg(feature = "nym")] pub mod live_indexer_discovery; From 651bcb4455a71b0352eab5430bb7344adbbd4cf9 Mon Sep 17 00:00:00 2001 From: zancas Date: Wed, 26 Aug 2026 09:46:46 -0700 Subject: [PATCH 3/3] feat(netutils): observe reconnect races and type the pull failure NymProxy::reconnect_observed reports each step of a reconnect race through the same typed BootstrapEvent channel the bootstrap uses, and reconnect becomes a thin silent rendering over it, so the diagnostics surface no longer goes blind at the moment after a proxy death. The discovery narration is one shared discover_observed path used by both entries, and a falsifier pins that a failed directory query still reports its departure. The redraw moves into redraw_clutch, with a falsifier pinning that a reconnect never rebinds a spent exit. BootstrapEvent::PullFailed now carries the pull's whole typed zingo_net_diag::NetOpFailure rather than a line rendered for a human, so a diagnostics surface renders structure instead of parsing prose. AcqStep becomes generic over the failure so the driver hands it through untouched. The spawnable binary's prose output is byte-identical. cargo clippy --features nym --all-targets is clean and all 71 library tests pass. Co-Authored-By: Claude Fable 5 --- zingo-netutils/CHANGELOG.md | 10 +++ zingo-netutils/src/nym_proxy.rs | 110 +++++++++++++++++++++++++------- 2 files changed, 98 insertions(+), 22 deletions(-) diff --git a/zingo-netutils/CHANGELOG.md b/zingo-netutils/CHANGELOG.md index 4b84f74b07..83775e89a4 100644 --- a/zingo-netutils/CHANGELOG.md +++ b/zingo-netutils/CHANGELOG.md @@ -33,6 +33,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 parking it on a cadence whose answer cannot change. ### Added +- `nym_proxy::BootstrapEvent` and the observed lifecycle entries + `NymProxy::start_observed`, `NymProxy::start_over_observed`, and + `NymProxy::reconnect_observed`: every acquisition race — the first + bootstrap and each reconnect after a proxy death — reports its steps as + typed events naming the Exit Node each concerns, while `start`, + `start_over`, and `reconnect` remain thin silent renderings over the same + path, so the two narrations cannot drift. `PullFailed` carries the pull's + whole typed `zingo_net_diag::NetOpFailure` — the stage, the target, and + the cause chain — rather than a line rendered for a human, so a + diagnostics surface renders structure instead of parsing prose. - The `socks5-fetch` feature and the `socks5_fetch` module: one HTTP request carried through a conduit, classified into a typed `zingo_net_diag` failure. `ConduitDial::fetch_text` is the entry, and it is the first diff --git a/zingo-netutils/src/nym_proxy.rs b/zingo-netutils/src/nym_proxy.rs index abe29f2fdc..3c4e469645 100644 --- a/zingo-netutils/src/nym_proxy.rs +++ b/zingo-netutils/src/nym_proxy.rs @@ -59,6 +59,18 @@ fn draw_clutch(mut discovered: Vec) -> Vec { discovered } +/// Draw a reconnect clutch with every exit of the spent clutch excluded, +/// because a redraw that rebound the same exit would defeat the rotation it +/// exists for. +fn redraw_clutch(discovered: Vec, spent: &[String]) -> Vec { + draw_clutch( + discovered + .into_iter() + .filter(|exit_node| !spent.contains(exit_node)) + .collect(), + ) +} + use crate::time::{DISCOVERY_TIMEOUT, NYM_LIFECYCLE_TIMEOUT, PER_ATTEMPT_CONNECT_TIMEOUT}; /// One step of the bootstrap, carrying the Exit Node it concerns. @@ -80,8 +92,8 @@ pub enum BootstrapEvent { PullFailed { /// The Exit Node address whose pull failed. exit_node: String, - /// The failure rendered for a human. - error: String, + /// The pull's typed failure record. + failure: NetOpFailure, }, /// The race kept this Exit Node and the local listener is up. Connected { @@ -91,7 +103,7 @@ pub enum BootstrapEvent { } /// One step of a driven race, reported to the caller as it happens. -enum AcqStep { +enum AcqStep { /// A pull of this arm index launched. Launched { /// The arm index the pull races. @@ -101,8 +113,8 @@ enum AcqStep { Failed { /// The arm index whose pull failed. arm: usize, - /// The failure rendered for a human. - error: String, + /// The pull's failure, untouched. + error: E, }, /// The race's counters changed. Progress(RaceProgress), @@ -134,12 +146,22 @@ impl NymProxy { pub async fn start_observed( mut on_event: impl FnMut(BootstrapEvent), ) -> Result { + let discovered = Self::discover_observed(DEFAULT_NYM_API_URL, &mut on_event).await?; + Self::start_over_observed(draw_clutch(discovered), on_event).await + } + + /// Discover the directory's Exit Nodes at `nym_api_url`, reporting the + /// query's departure and its answer to `on_event`. + async fn discover_observed( + nym_api_url: &str, + on_event: &mut impl FnMut(BootstrapEvent), + ) -> Result, NymProxyError> { on_event(BootstrapEvent::DiscoveryStarted); - let discovered = Self::discover_exit_nodes_at(DEFAULT_NYM_API_URL).await?; + let discovered = Self::discover_exit_nodes_at(nym_api_url).await?; on_event(BootstrapEvent::DiscoveryFinished { candidate_count: discovered.len(), }); - Self::start_over_observed(draw_clutch(discovered), on_event).await + Ok(discovered) } /// Start over exactly `clutch`, the Exit Node Reservations the parent @@ -250,7 +272,7 @@ impl NymProxy { }), AcqStep::Failed { arm, error } => on_event(BootstrapEvent::PullFailed { exit_node: full_exit_node_name(arm), - error, + failure: error, }), AcqStep::Progress(progress) => on_progress(progress.to_string()), }, @@ -353,7 +375,16 @@ impl NymProxy { /// returned. After a successful reconnect, [`socks5_addr`](Self::socks5_addr) /// returns the new port. pub async fn reconnect(&mut self) -> Result<(), NymProxyError> { - tokio::time::timeout(NYM_LIFECYCLE_TIMEOUT, self.reconnect_inner()) + self.reconnect_observed(|_| {}).await + } + + /// [`Self::reconnect`], reporting each bootstrap step to `on_event` as a + /// typed [`BootstrapEvent`]. + pub async fn reconnect_observed( + &mut self, + on_event: impl FnMut(BootstrapEvent), + ) -> Result<(), NymProxyError> { + tokio::time::timeout(NYM_LIFECYCLE_TIMEOUT, self.reconnect_inner(on_event)) .await .map_err(|_| { NymProxyError::ConnectivityCheck(format!( @@ -363,21 +394,18 @@ impl NymProxy { })? } - async fn reconnect_inner(&mut self) -> Result<(), NymProxyError> { - // The spent clutch is redrawn rather than reused: a redraw that - // rebound the same exit would defeat the rotation it exists for. - let discovered = Self::discover_exit_nodes_at(DEFAULT_NYM_API_URL).await?; - let fresh: Vec = discovered - .into_iter() - .filter(|exit_node| !self.clutch.contains(exit_node)) - .collect(); - let clutch = draw_clutch(fresh); + async fn reconnect_inner( + &mut self, + mut on_event: impl FnMut(BootstrapEvent), + ) -> Result<(), NymProxyError> { + let discovered = Self::discover_observed(DEFAULT_NYM_API_URL, &mut on_event).await?; + let clutch = redraw_clutch(discovered, &self.clutch); if clutch.is_empty() { return Err(NymProxyError::NoExitNode); } // Each attempt binds its own fresh port, which cannot collide with // the old client's still-bound port. - let new_proxy = Self::connect_across_exit_nodes(&clutch, |_| {}, |_| {}).await?; + let new_proxy = Self::connect_across_exit_nodes(&clutch, on_event, |_| {}).await?; // Swap only after the new client succeeded, so a failed reconnect // leaves the old client untouched. @@ -549,11 +577,11 @@ async fn drive_acq_race( launch: F, abandon: D, describe_panic: impl Fn(usize, String) -> E, - mut on_step: impl FnMut(AcqStep), + mut on_step: impl FnMut(AcqStep), ) -> Result> where T: Send + 'static, - E: std::fmt::Display + Send + 'static, + E: Clone + std::fmt::Display + Send + 'static, F: Fn(usize) -> Fut, Fut: Future> + Send + 'static, D: Fn(T) -> DFut, @@ -653,7 +681,7 @@ where // whole for the terminal account. on_step(AcqStep::Failed { arm, - error: error.to_string(), + error: error.clone(), }); apply( acq_race.on_event(RaceEvent::PullFailed { @@ -762,6 +790,28 @@ mod tests { assert!(draw_clutch(Vec::new()).is_empty()); } + /// HYPOTHESIS: a reconnect redraw never rebinds a spent exit and still + /// draws a full clutch from the fresh remainder; falsified if a spent + /// exit reappears, or if a spent-out population yields anything but an + /// empty draw. + #[test] + fn a_reconnect_redraw_never_rebinds_a_spent_exit() { + let population: Vec = (0..RESERVATION_CLUTCH_SIZE * 3) + .map(|index| format!("exit-{index}")) + .collect(); + let spent: Vec = population[..RESERVATION_CLUTCH_SIZE].to_vec(); + + let redrawn = redraw_clutch(population.clone(), &spent); + assert_eq!(redrawn.len(), RESERVATION_CLUTCH_SIZE); + assert!(redrawn.iter().all(|drawn| !spent.contains(drawn))); + assert!(redrawn.iter().all(|drawn| population.contains(drawn))); + + assert!( + redraw_clutch(spent.clone(), &spent).is_empty(), + "a spent-out population leaves nothing to draw" + ); + } + fn hedged(max_parallel: usize) -> LaunchPolicy { LaunchPolicy::Hedged { max_parallel, @@ -986,6 +1036,22 @@ mod tests { } } + /// HYPOTHESIS: the shared discovery narration reports the query's + /// departure before its outcome, so a failed directory query is still + /// visible on the typed channel that both bootstrap and reconnect + /// report through; falsified if a failing discovery emits nothing or + /// claims to have finished. + #[tokio::test] + async fn a_failed_discovery_still_reports_its_departure() { + let mut events = Vec::new(); + // Port 0 is unconnectable, so the query fails without the network. + let outcome = + NymProxy::discover_observed("http://127.0.0.1:0/", &mut |event| events.push(event)) + .await; + outcome.expect_err("no directory answers on port 0"); + assert_eq!(events, vec![BootstrapEvent::DiscoveryStarted]); + } + #[test] fn long_exit_node_names_are_shortened_for_the_summary() { let exit_nodes = vec!["a".repeat(200), "short".to_string()];