diff --git a/fynd-core/src/worker_pool/worker.rs b/fynd-core/src/worker_pool/worker.rs index 08176926..c9e3db0d 100644 --- a/fynd-core/src/worker_pool/worker.rs +++ b/fynd-core/src/worker_pool/worker.rs @@ -45,14 +45,41 @@ fn record_task_pickup_metrics(pool_name: &str, queue_wait: Duration, queue_depth } /// Records per-pool solve latency: one algorithm's own working time for one order, excluding -/// queue wait. Unlike `worker_router_solve_duration_seconds`, which times the router racing every -/// pool and so belongs to no single pool, this is attributable per pool. +/// queue wait and readiness wait. Unlike `worker_router_solve_duration_seconds`, which times the +/// router racing every pool and so belongs to no single pool, this is attributable per pool. /// -/// Successful solves only — a pool that exhausts its timeout returns before this point and is -/// counted in `worker_router_solver_failures_total{error_type="timeout"}` instead. -fn record_solve_duration(pool_name: &str, solve_time: Duration) { - metrics::histogram!("worker_pool_solve_duration_seconds", "pool" => pool_name.to_string()) - .record(solve_time.as_secs_f64()); +/// Recorded for every outcome, labelled by `outcome`. A histogram of successes alone hides the +/// cost of the failures the request still waited for, which is how a pool can show a healthy p95 +/// while setting the request's. +fn record_solve_duration(pool_name: &str, outcome: &'static str, solve_time: Duration) { + metrics::histogram!( + "worker_pool_solve_duration_seconds", + "pool" => pool_name.to_string(), + "outcome" => outcome, + ) + .record(solve_time.as_secs_f64()); +} + +/// Records how long a task waited for its algorithm's required derived data before solving. +/// +/// This sits between task pickup and the solve itself, so it is in neither +/// `worker_pool_queue_wait_seconds` nor `worker_pool_solve_duration_seconds`, yet the request +/// waits through all of it. +fn record_readiness_wait(pool_name: &str, wait: Duration) { + metrics::histogram!("worker_pool_readiness_wait_seconds", "pool" => pool_name.to_string()) + .record(wait.as_secs_f64()); +} + +/// Classifies a solve outcome for the `outcome` label on `worker_pool_solve_duration_seconds`. +fn solve_outcome(result: &Result) -> &'static str { + match result { + Ok(_) => "success", + Err(SolveError::NotReady(_)) => "not_ready", + Err(SolveError::Timeout { .. }) => "timeout", + Err(SolveError::NoRouteFound { .. }) => "no_route", + Err(SolveError::InsufficientLiquidity { .. }) => "insufficient_liquidity", + Err(_) => "error", + } } /// A solver worker instance that maintains a market graph and processes solve requests. @@ -177,13 +204,27 @@ where } /// Returns a quote for an order, optionally solved against a named state overlay. + /// + /// Times every outcome, not just the successful ones, so a pool that fails slowly is as + /// visible as one that succeeds slowly. pub async fn quote( &mut self, order: &Order, params: SolveParams, ) -> Result { let start_time = Instant::now(); + let result = self.solve(order, params).await; + let solve_time = start_time.elapsed(); + record_solve_duration(&self.pool_name, solve_outcome(&result), solve_time); + result.map(|quote| SingleOrderQuote::new(quote, solve_time.as_millis() as u64)) + } + /// Solves one order, leaving timing and metrics to [`SolverWorker::quote`]. + async fn solve( + &mut self, + order: &Order, + params: SolveParams, + ) -> Result { // Log order details once at entry debug!( order_id = %order.id(), @@ -352,10 +393,7 @@ where } }; - let solve_time = start_time.elapsed(); - record_solve_duration(&self.pool_name, solve_time); - - Ok(SingleOrderQuote::new(order_quote, solve_time.as_millis() as u64)) + Ok(order_quote) } /// Waits for required derived data to become ready, or until timeout. @@ -551,7 +589,10 @@ where // Wait for derived data readiness before solving // Use algorithm timeout as the max wait time - if let Err(e) = self.wait_until_ready(self.algorithm.timeout()).await { + let ready_start = Instant::now(); + let readiness = self.wait_until_ready(self.algorithm.timeout()).await; + record_readiness_wait(&self.pool_name, ready_start.elapsed()); + if let Err(e) = readiness { warn!( self.worker_id, task_id = %task_id, @@ -1254,7 +1295,7 @@ mod tests { let recorder = DebuggingRecorder::new(); let snapshotter = recorder.snapshotter(); metrics::with_local_recorder(&recorder, || { - record_solve_duration("test_pool", std::time::Duration::from_millis(120)); + record_solve_duration("test_pool", "success", std::time::Duration::from_millis(120)); }); let mut solve_seen = false; diff --git a/fynd-core/src/worker_pool_router/mod.rs b/fynd-core/src/worker_pool_router/mod.rs index 11a110f3..7d4b766f 100644 --- a/fynd-core/src/worker_pool_router/mod.rs +++ b/fynd-core/src/worker_pool_router/mod.rs @@ -209,29 +209,61 @@ pub(crate) struct OrderResponses { } impl OrderResponses { - /// Returns a copy keeping only candidates from public-scoped worker pools. + /// Borrows every candidate. + fn view(&self) -> ResponsesView<'_> { + self.view_filtered(|_| true) + } + + /// Borrows only the candidates from public-scoped worker pools. /// /// These form the committed reference and the ranked fallback chain (ranked by `rank_quotes`, /// consumed by the price guard); exclusive-access candidates are overlaid separately by /// `combine_with_surplus`. `failed_solvers` is retained so placeholder construction is /// unchanged. - fn public_only(&self, pool_scopes: &HashMap) -> OrderResponses { - let quotes = self - .quotes - .iter() - .filter(|wq| { - pool_scopes.get(&wq.worker_pool) != Some(&LiquidityScope::IncludeExclusive) - }) - .cloned() - .collect(); - OrderResponses { - order_id: self.order_id.clone(), - quotes, - failed_solvers: self.failed_solvers.clone(), + fn public_view(&self, pool_scopes: &HashMap) -> ResponsesView<'_> { + self.view_filtered(|pool| pool_scopes.get(pool) != Some(&LiquidityScope::IncludeExclusive)) + } + + fn view_filtered(&self, keep: impl Fn(&str) -> bool) -> ResponsesView<'_> { + ResponsesView { + order_id: &self.order_id, + quotes: self + .quotes + .iter() + .filter(|worker_pool_quote| keep(&worker_pool_quote.worker_pool)) + .collect(), + failed_solvers: &self.failed_solvers, } } } +/// A borrowed, optionally scope-narrowed view over one order's responses. +/// +/// Ranking reads candidates far more often than it keeps them, and every `OrderQuote` owns a +/// `Route` whose legs each hold a `Box`. Narrowing by borrow rather than by +/// clone keeps the single deep copy that `rank_quotes` makes of its winners the only one on the +/// request path. +struct ResponsesView<'a> { + order_id: &'a str, + quotes: Vec<&'a WorkerPoolQuote>, + failed_solvers: &'a [(String, SolveError)], +} + +/// Records how long a worker pool took to answer, as observed by the router. +/// +/// Unlike `worker_pool_solve_duration_seconds`, which a worker reports for its own successful +/// solves, this covers everything the request actually waits for — queue wait, readiness wait, +/// and the solve — and is recorded for failures too (`outcome`). The gap between the two is +/// what a request pays for a worker pool beyond its algorithm's own working time. +fn record_pool_response(worker_pool_name: &str, outcome: &'static str, elapsed: Duration) { + histogram!( + "worker_router_pool_response_seconds", + "pool" => worker_pool_name.to_string(), + "outcome" => outcome, + ) + .record(elapsed.as_secs_f64()); +} + /// Orchestrates multiple solver pools to find the best quote. pub struct WorkerPoolRouter { /// All registered solver pools. @@ -363,7 +395,7 @@ impl WorkerPoolRouter { .map(|(responses, allocation)| { if allocation.exclusive_routing_active() { let public_ranked = self.rank_quotes( - &responses.public_only(allocation.scopes()), + &responses.public_view(allocation.scopes()), request.options(), ); combine_with_surplus( @@ -374,7 +406,7 @@ impl WorkerPoolRouter { *USER_IMPROVEMENT_SHARE_BPS, ) } else { - self.rank_quotes(responses, request.options()) + self.rank_quotes(&responses.view(), request.options()) } }) .collect(); @@ -506,8 +538,7 @@ impl WorkerPoolRouter { // Timeout reached _ = tokio::time::sleep_until(deadline_instant) => { // Mark all remaining worker pools as timed out - let elapsed_ms = deadline.saturating_duration_since(Instant::now()) - .as_millis() as u64; + let elapsed_ms = start_time.elapsed().as_millis() as u64; for worker_pool_name in remaining_worker_pools.drain() { failed_solvers.push(( worker_pool_name, @@ -523,6 +554,11 @@ impl WorkerPoolRouter { Some((worker_pool_name, Ok(single_quote))) => { // Remove from remaining remaining_worker_pools.remove(&worker_pool_name); + record_pool_response( + &worker_pool_name, + "ok", + start_time.elapsed(), + ); if allocation.is_exclusive(&worker_pool_name) { has_exclusive_access_response = true; @@ -562,6 +598,11 @@ impl WorkerPoolRouter { } Some((worker_pool_name, Err(e))) => { remaining_worker_pools.remove(&worker_pool_name); + record_pool_response( + &worker_pool_name, + "error", + start_time.elapsed(), + ); // A failed exclusive-access worker pool still counts as "responded" // for gating — we know it won't produce a surplus quote, so the // public worker pools can early-return without waiting for a result @@ -619,7 +660,11 @@ impl WorkerPoolRouter { /// If no valid quotes exist, returns a single-element vec with a placeholder /// (`NoRouteFound` or `Timeout`) so that downstream always has at least one /// candidate per order. - fn rank_quotes(&self, responses: &OrderResponses, options: &QuoteOptions) -> Vec { + fn rank_quotes( + &self, + responses: &ResponsesView<'_>, + options: &QuoteOptions, + ) -> Vec { let mut valid_quotes: Vec<_> = responses .quotes .iter() @@ -645,7 +690,7 @@ impl WorkerPoolRouter { ); return valid_quotes .into_iter() - .map(|pq| pq.quote.clone()) + .map(|worker_pool_quote| worker_pool_quote.quote.clone()) .collect(); } @@ -655,7 +700,7 @@ impl WorkerPoolRouter { { counter!("worker_router_orders_total", "status" => "no_route").increment(1); let mut fallback = OrderQuote::new( - responses.order_id.clone(), + responses.order_id.to_string(), QuoteStatus::NoRouteFound, any_q.amount_in().clone(), BigUint::ZERO, @@ -716,7 +761,7 @@ impl WorkerPoolRouter { .cloned() .unwrap_or_else(|| "0".to_string()); let mut fallback = OrderQuote::new( - responses.order_id.clone(), + responses.order_id.to_string(), status, BigUint::ZERO, BigUint::ZERO, @@ -728,7 +773,7 @@ impl WorkerPoolRouter { Bytes::default(), label, ); - fallback.set_no_route_cause(aggregate_no_route_cause(&responses.failed_solvers)); + fallback.set_no_route_cause(aggregate_no_route_cause(responses.failed_solvers)); fallback }; vec![fallback] @@ -1221,10 +1266,10 @@ mod tests { ) } - /// `public_only` must carry the order identity across, or the surplus path logs and ranks + /// `public_view` must carry the order identity across, or the surplus path logs and ranks /// against a response set that has lost it. #[test] - fn test_public_only_keeps_order_id_and_failures() { + fn test_public_view_keeps_order_id_and_failures() { let responses = OrderResponses { order_id: "o1".to_string(), quotes: vec![ @@ -1237,7 +1282,7 @@ mod tests { ("public".to_string(), LiquidityScope::PublicOnly), ("excl".to_string(), LiquidityScope::IncludeExclusive), ]); - let public = responses.public_only(&scopes); + let public = responses.public_view(&scopes); assert_eq!(public.order_id, "o1"); assert_eq!(public.quotes.len(), 1); @@ -1660,7 +1705,7 @@ mod tests { let worker_router = WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder()); - let result = worker_router.rank_quotes(&responses, &options); + let result = worker_router.rank_quotes(&responses.view(), &options); if should_pass { assert_eq!(result[0].status(), QuoteStatus::Success); @@ -1706,7 +1751,7 @@ mod tests { let worker_router = WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder()); - let result = worker_router.rank_quotes(&responses, &QuoteOptions::default()); + let result = worker_router.rank_quotes(&responses.view(), &QuoteOptions::default()); assert_eq!(result.len(), 1); assert_eq!(result[0].status(), QuoteStatus::Timeout); @@ -1725,7 +1770,7 @@ mod tests { let worker_router = WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder()); - let result = worker_router.rank_quotes(&responses, &QuoteOptions::default()); + let result = worker_router.rank_quotes(&responses.view(), &QuoteOptions::default()); assert_eq!(result.len(), 1); assert_eq!(result[0].status(), QuoteStatus::NoRouteFound); @@ -1738,7 +1783,7 @@ mod tests { let worker_router = WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder()); - let result = worker_router.rank_quotes(&responses, &QuoteOptions::default()); + let result = worker_router.rank_quotes(&responses.view(), &QuoteOptions::default()); assert_eq!(result.len(), 1); assert_eq!(result[0].status(), QuoteStatus::NoRouteFound); @@ -1766,7 +1811,7 @@ mod tests { }; let worker_router = WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder()); - let result = worker_router.rank_quotes(&responses, &QuoteOptions::default()); + let result = worker_router.rank_quotes(&responses.view(), &QuoteOptions::default()); assert_eq!(result.len(), 1); assert_eq!(result[0].status(), QuoteStatus::NoRouteFound); // Token-not-in-graph wins over no_graph_path regardless of pool order. @@ -1786,7 +1831,7 @@ mod tests { }; let worker_router = WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder()); - let result = worker_router.rank_quotes(&responses, &QuoteOptions::default()); + let result = worker_router.rank_quotes(&responses.view(), &QuoteOptions::default()); assert_eq!(result[0].no_route_reason(), Some(NoPathReason::NoGraphPath)); } @@ -1880,7 +1925,7 @@ mod tests { let options = QuoteOptions::default().with_max_gas(BigUint::from(1u64)); let worker_router = WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder()); - let result = worker_router.rank_quotes(&responses, &options); + let result = worker_router.rank_quotes(&responses.view(), &options); assert_eq!(result[0].status(), QuoteStatus::NoRouteFound); assert!(matches!(result[0].no_route_cause(), Some(SolveError::MaxGasExceeded))); } @@ -1910,7 +1955,7 @@ mod tests { let options = QuoteOptions::default().with_max_gas(BigUint::from(1_000u64)); let worker_router = WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder()); - let result = worker_router.rank_quotes(&responses, &options); + let result = worker_router.rank_quotes(&responses.view(), &options); assert_eq!(result[0].status(), QuoteStatus::NoRouteFound); assert!( result[0].no_route_cause().is_none(), @@ -1928,7 +1973,7 @@ mod tests { }; let worker_router = WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder()); - let result = worker_router.rank_quotes(&responses, &QuoteOptions::default()); + let result = worker_router.rank_quotes(&responses.view(), &QuoteOptions::default()); assert_eq!(result[0].status(), QuoteStatus::Timeout); assert!(matches!(result[0].no_route_cause(), Some(SolveError::Timeout { .. }))); } @@ -1976,7 +2021,7 @@ mod tests { let worker_router = WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder()); - let result = worker_router.rank_quotes(&responses, &QuoteOptions::default()); + let result = worker_router.rank_quotes(&responses.view(), &QuoteOptions::default()); assert_eq!(result.len(), 2); assert_eq!(*result[0].amount_out_net_gas(), BigUint::from(950u64));