Shared native price cache

crates/autopilot/src/arguments.rs

@@ -96,6 +96,11 @@ pub struct Arguments {
     #[clap(long, env)]
     pub native_price_estimators: NativePriceEstimators,
 
+    /// Estimators for the API endpoint. Falls back to
+    /// `--native-price-estimators` if unset.
+    #[clap(long, env)]
+    pub api_native_price_estimators: Option<NativePriceEstimators>,
+
     /// How many successful price estimates for each order will cause a native
     /// price estimation to return its result early. It's possible to pass
     /// values greater than the total number of enabled estimators but that
@@ -276,6 +281,26 @@ pub struct Arguments {
     /// further.
     #[clap(long, env, default_value = "5s", value_parser = humantime::parse_duration)]
     pub max_maintenance_timeout: Duration,
+
+    /// How often the native price estimator should refresh its cache.
+    #[clap(
+        long,
+        env,
+        default_value = "1s",
+        value_parser = humantime::parse_duration,
+    )]
+    pub native_price_cache_refresh: Duration,
+
+    /// How long before expiry the native price cache should try to update the
+    /// price in the background. This value has to be smaller than
+    /// `--native-price-cache-max-age`.
+    #[clap(
+        long,
+        env,
+        default_value = "80s",
+        value_parser = humantime::parse_duration,
+    )]
+    pub native_price_prefetch_time: Duration,
 }
 
 impl std::fmt::Display for Arguments {
@@ -296,6 +321,7 @@ impl std::fmt::Display for Arguments {
             allowed_tokens,
             unsupported_tokens,
             native_price_estimators,
+            api_native_price_estimators,
             min_order_validity_period,
             banned_users,
             banned_users_max_cache_size,
@@ -327,6 +353,8 @@ impl std::fmt::Display for Arguments {
             disable_1271_order_sig_filter,
             enable_leader_lock,
             max_maintenance_timeout,
+            native_price_cache_refresh,
+            native_price_prefetch_time,
         } = self;
 
         write!(f, "{shared}")?;
@@ -345,6 +373,11 @@ impl std::fmt::Display for Arguments {
         writeln!(f, "allowed_tokens: {allowed_tokens:?}")?;
         writeln!(f, "unsupported_tokens: {unsupported_tokens:?}")?;
         writeln!(f, "native_price_estimators: {native_price_estimators}")?;
+        display_option(
+            f,
+            "api_native_price_estimators",
+            api_native_price_estimators,
+        )?;
         writeln!(
             f,
             "min_order_validity_period: {min_order_validity_period:?}"
@@ -408,6 +441,14 @@ impl std::fmt::Display for Arguments {
         )?;
         writeln!(f, "enable_leader_lock: {enable_leader_lock}")?;
         writeln!(f, "max_maintenance_timeout: {max_maintenance_timeout:?}")?;
+        writeln!(
+            f,
+            "native_price_cache_refresh: {native_price_cache_refresh:?}"
+        )?;
+        writeln!(
+            f,
+            "native_price_prefetch_time: {native_price_prefetch_time:?}"
+        )?;
         Ok(())
     }
 }

crates/autopilot/src/run.rs

@@ -47,7 +47,10 @@ use {
         code_fetching::CachedCodeFetcher,
         http_client::HttpClientFactory,
         order_quoting::{self, OrderQuoter},
-        price_estimation::factory::{self, PriceEstimatorFactory},
+        price_estimation::{
+            factory::{self, PriceEstimatorFactory},
+            native::NativePriceEstimating,
+        },
         signature_validator,
         sources::{BaselineSource, uniswap_v2::UniV2BaselineSourceParameters},
         token_info::{CachedTokenInfoFetcher, TokenInfoFetcher},
@@ -386,17 +389,44 @@ pub async fn run(args: Arguments, shutdown_controller: ShutdownController) {
     .await
     .expect("failed to initialize price estimator factory");
 
-    let native_price_estimator = price_estimator_factory
-        .native_price_estimator(
-            args.native_price_estimators.as_slice(),
-            args.native_price_estimation_results_required,
-            eth.contracts().weth().clone(),
-        )
-        .instrument(info_span!("native_price_estimator"))
-        .await
-        .unwrap();
+    let weth = eth.contracts().weth().clone();
     let prices = db_write.fetch_latest_prices().await.unwrap();
-    native_price_estimator.initialize_cache(prices);
+    let shared_cache = shared::price_estimation::native_price_cache::Cache::new(
+        args.price_estimation.native_price_cache_max_age,
+        prices,
+    );
+    let api_sources = args
+        .api_native_price_estimators
+        .as_ref()
+        .unwrap_or(&args.native_price_estimators);
+    let api_native_price_estimator: Arc<dyn NativePriceEstimating> = Arc::new(
+        price_estimator_factory
+            .caching_native_price_estimator(
+                api_sources.as_slice(),
+                args.native_price_estimation_results_required,
+                &weth,
+                shared_cache.clone(),
+            )
+            .instrument(info_span!("api_native_price_estimator"))
+            .await,
+    );
+
+    let competition_native_price_updater = {
+        let caching = price_estimator_factory
+            .caching_native_price_estimator(
+                args.native_price_estimators.as_slice(),
+                args.native_price_estimation_results_required,
+                &weth,
+                shared_cache.clone(),
+            )
+            .instrument(info_span!("competition_native_price_updater"))
+            .await;
+        shared::price_estimation::native_price_cache::NativePriceUpdater::new(
+            caching,
+            args.native_price_cache_refresh,
+            args.native_price_prefetch_time,
+        )
+    };
 
     let price_estimator = price_estimator_factory
         .price_estimator(
@@ -406,7 +436,7 @@ pub async fn run(args: Arguments, shutdown_controller: ShutdownController) {
                 .iter()
                 .map(|price_estimator_driver| price_estimator_driver.clone().into())
                 .collect::<Vec<_>>(),
-            native_price_estimator.clone(),
+            api_native_price_estimator.clone(),
             gas_price_estimator.clone(),
         )
         .unwrap();
@@ -470,7 +500,7 @@ pub async fn run(args: Arguments, shutdown_controller: ShutdownController) {
 
     let quoter = Arc::new(OrderQuoter::new(
         price_estimator,
-        native_price_estimator.clone(),
+        api_native_price_estimator.clone(),
         gas_price_estimator,
         Arc::new(db_write.clone()),
         order_quoting::Validity {
@@ -502,7 +532,7 @@ pub async fn run(args: Arguments, shutdown_controller: ShutdownController) {
         ),
         balance_fetcher.clone(),
         bad_token_detector.clone(),
-        native_price_estimator.clone(),
+        competition_native_price_updater.clone(),
         signature_validator.clone(),
         *eth.contracts().weth().address(),
         args.limit_order_price_factor
@@ -527,7 +557,7 @@ pub async fn run(args: Arguments, shutdown_controller: ShutdownController) {
     let (api_shutdown_sender, api_shutdown_receiver) = tokio::sync::oneshot::channel();
     let api_task = tokio::spawn(infra::api::serve(
         args.api_address,
-        native_price_estimator.clone(),
+        api_native_price_estimator,
         args.price_estimation.quote_timeout,
         api_shutdown_receiver,
     ));