diff --git a/zingo-netutils/CHANGELOG.md b/zingo-netutils/CHANGELOG.md index 5d5c4edaf8..786b56b3e7 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/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; diff --git a/zingo-netutils/src/nym_proxy.rs b/zingo-netutils/src/nym_proxy.rs index b6fe201f83..3c4e469645 100644 --- a/zingo-netutils/src/nym_proxy.rs +++ b/zingo-netutils/src/nym_proxy.rs @@ -59,8 +59,67 @@ 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. +#[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 pull's typed failure record. + failure: NetOpFailure, + }, + /// 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 pull's failure, untouched. + error: E, + }, + /// 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 +138,30 @@ 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 { - let discovered = Self::discover_exit_nodes_at(DEFAULT_NYM_API_URL).await?; - Self::start_over(draw_clutch(discovered), |_| {}).await + 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 { + 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(nym_api_url).await?; + on_event(BootstrapEvent::DiscoveryFinished { + candidate_count: discovered.len(), + }); + Ok(discovered) } /// Start over exactly `clutch`, the Exit Node Reservations the parent @@ -88,10 +169,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 +202,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 +220,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 +266,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), + failure: 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 @@ -254,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!( @@ -264,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. @@ -450,11 +577,11 @@ 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, - 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, @@ -471,13 +598,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 +615,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 +625,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 +679,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.clone(), + }); apply( acq_race.on_event(RaceEvent::PullFailed { arm, @@ -552,9 +692,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 +712,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())); } } } @@ -642,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, @@ -815,7 +985,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 +999,59 @@ 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:?}" + ); + } + } + + /// 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()];