Skip to content

refactor: improve exclusive liquidity request latency #430

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
louise-poole wants to merge 2 commits into
base: main
Choose a base branch
from
+135 −49
Open

refactor: improve exclusive liquidity request latency #430

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
95e153b
refactor: rank quotes from borrowed responses
louise-poole Aug 17, 2026
11fd640
feat: record per-pool response latency at the router
louise-poole Aug 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 54 additions & 13 deletions fynd-core/src/worker_pool/worker.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<OrderQuote, SolveError>) -> &'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",
Comment on lines +77 to +81

@Troshchk Troshchk Aug 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why not to use solver_error_label for the error matching?
Is it ok, that we wrap some errors into simple "error" string?

}
}

/// A solver worker instance that maintains a market graph and processes solve requests.
Expand Down Expand Up @@ -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<SingleOrderQuote, SolveError> {
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<OrderQuote, SolveError> {
// Log order details once at entry
debug!(
order_id = %order.id(),
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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;
Expand Down
117 changes: 81 additions & 36 deletions fynd-core/src/worker_pool_router/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String, LiquidityScope>) -> 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<String, LiquidityScope>) -> 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<dyn ProtocolSim>`. 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.

@Troshchk Troshchk Aug 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would it make sense to add that the pools are worker pools?

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.
Expand Down Expand Up @@ -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(
Expand All @@ -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();
Expand Down Expand Up @@ -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;

@Troshchk Troshchk Aug 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice fix!

for worker_pool_name in remaining_worker_pools.drain() {
failed_solvers.push((
worker_pool_name,
Expand All @@ -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;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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<OrderQuote> {
fn rank_quotes(
&self,
responses: &ResponsesView<'_>,
options: &QuoteOptions,
) -> Vec<OrderQuote> {
let mut valid_quotes: Vec<_> = responses
.quotes
.iter()
Expand All @@ -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();
}

Expand All @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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]
Expand Down Expand Up @@ -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![
Expand All @@ -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);
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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);
Expand All @@ -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);
Expand All @@ -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);
Expand Down Expand Up @@ -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.
Expand All @@ -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));
}

Expand Down Expand Up @@ -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)));
}
Expand Down Expand Up @@ -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(),
Expand All @@ -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 { .. })));
}
Expand Down Expand Up @@ -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));
Expand Down
Loading