diff --git a/crates/price-estimation/src/native_price_cache.rs b/crates/price-estimation/src/native_price_cache.rs index c6da967733..c117c89705 100644 --- a/crates/price-estimation/src/native_price_cache.rs +++ b/crates/price-estimation/src/native_price_cache.rs @@ -7,6 +7,7 @@ use { futures::{FutureExt, StreamExt}, prometheus::{IntCounter, IntCounterVec, IntGauge}, rand::Rng, + request_sharing::{BoxRequestSharing, RequestSharing}, std::{ collections::{HashMap, HashSet}, sync::Arc, @@ -287,6 +288,7 @@ struct CachingInner { /// After startup this is a read only value. approximation_tokens: HashMap, quote_timeout: Duration, + requests_in_flight: BoxRequestSharing, } impl CachingNativePriceEstimator { @@ -303,6 +305,7 @@ impl CachingNativePriceEstimator { concurrent_requests, approximation_tokens, quote_timeout, + requests_in_flight: RequestSharing::labelled("native_price".to_string()), }); Self(inner) } @@ -312,6 +315,10 @@ impl CachingNativePriceEstimator { /// estimation request gets issued. We check the cache before each /// request because they can take a long time and some other task might /// have fetched some requested price in the meantime. + /// + /// Concurrent requests for the same token are de-duplicated, so only a + /// single underlying quote is issued per token even when it's requested + /// from multiple places at once. fn estimate_prices_and_update_cache<'a, I>( &'a self, tokens: I, @@ -323,32 +330,44 @@ impl CachingNativePriceEstimator { I::IntoIter: Send + 'a, { let estimates = tokens.into_iter().map(move |token| async move { - // check if the price is cached by now - let now = Instant::now(); + // The price may already be cached if let Some(cached) = - Cache::get_cached_price(token, now, &self.0.cache.0.data, &max_age) + Cache::get_cached_price(token, Instant::now(), &self.0.cache.0.data, &max_age) { return (token, cached.result); } - let approximation = self - .0 - .approximation_tokens - .get(&token) - .copied() - .unwrap_or(ApproximationToken::same_decimals(token)); - + // Coalesce concurrent in-flight requests for the same token so the + // underlying estimator is queried at most once let result = self .0 - .estimator - .estimate_native_price(approximation.address, request_timeout) - .await - .map(|price| approximation.normalize_price(price)); - - // update price in cache - if should_cache(&result) { - self.0.cache.insert(token, result.clone()); - }; + .requests_in_flight + .shared_or_else(token, move |&token| { + let this = self.clone(); + async move { + let approximation = this + .0 + .approximation_tokens + .get(&token) + .copied() + .unwrap_or(ApproximationToken::same_decimals(token)); + + let result = this + .0 + .estimator + .estimate_native_price(approximation.address, request_timeout) + .await + .map(|price| approximation.normalize_price(price)); + + if should_cache(&result) { + this.0.cache.insert(token, result.clone()); + } + + result + } + .boxed() + }) + .await; (token, result) }); @@ -636,6 +655,38 @@ mod tests { } } + #[tokio::test] + async fn deduplicates_concurrent_requests_for_same_token() { + let mut inner = MockNativePriceEstimating::new(); + // Even though many callers request the same uncached token at once, the + // underlying estimator must be queried exactly once (sleep keeps the leader's + // request in flight while others arrive). + inner + .expect_estimate_native_price() + .times(1) + .returning(|_, _| { + async { + tokio::time::sleep(Duration::from_millis(50)).await; + Ok(1.0) + } + .boxed() + }); + + let estimator = + create_caching_estimator(inner, Duration::from_secs(10), 10, Default::default()); + + let results = futures::future::join_all( + (0..10) + .map(|_| estimator.estimate_native_price(token(0), HEALTHY_PRICE_ESTIMATION_TIME)), + ) + .await; + + assert_eq!(results.len(), 10); + for result in results { + assert_eq!(result.as_ref().unwrap().to_i64().unwrap(), 1); + } + } + #[tokio::test] async fn caches_approximated_estimates_use() { let mut inner = MockNativePriceEstimating::new();