Propagate the buffered CoinGecko implementation to the configuration …

…3/3 (#2909)

# Description
Instantiate the CoinGecko buffered and propagate the configuration.

I set the default configuration to reasonable parameters, so we can try
them out without needing to change the infra PR.

# Changes
- Instantiate the CoinGecko buffered instead of normal CoinGecko
"object"
- Propagate the configuration accordingly

## How to test
1. Regression tests

## Related Issues

Fixes #2814

crates/shared/src/price_estimation.rs

@@ -183,17 +183,9 @@ pub struct Arguments {
     #[clap(long, env, default_value = "https://api.1inch.dev/")]
     pub one_inch_url: Url,
 
-    /// The API key for the CoinGecko API.
-    #[clap(long, env)]
-    pub coin_gecko_api_key: Option<String>,
-
-    /// The base URL for the CoinGecko API.
-    #[clap(
-        long,
-        env,
-        default_value = "https://api.coingecko.com/api/v3/simple/token_price"
-    )]
-    pub coin_gecko_url: Url,
+    /// The CoinGecko native price configuration
+    #[clap(flatten)]
+    pub coin_gecko: CoinGecko,
 
     /// How inaccurate a quote must be before it gets discarded provided as a
     /// factor.
@@ -222,6 +214,52 @@ pub struct Arguments {
     pub quote_timeout: Duration,
 }
 
+#[derive(clap::Parser)]
+pub struct CoinGecko {
+    /// The API key for the CoinGecko API.
+    #[clap(long, env)]
+    pub coin_gecko_api_key: Option<String>,
+
+    /// The base URL for the CoinGecko API.
+    #[clap(
+        long,
+        env,
+        default_value = "https://api.coingecko.com/api/v3/simple/token_price"
+    )]
+    pub coin_gecko_url: Url,
+
+    #[clap(flatten)]
+    pub coin_gecko_buffered: Option<CoinGeckoBuffered>,
+}
+
+#[derive(clap::Parser)]
+#[clap(group(
+    clap::ArgGroup::new("coin_gecko_buffered")
+    .requires_all(&[
+        "coin_gecko_debouncing_time",
+        "coin_gecko_result_ready_timeout",
+        "coin_gecko_broadcast_channel_capacity"
+    ])
+    .multiple(true)
+    .required(false),
+))]
+pub struct CoinGeckoBuffered {
+    /// An additional minimum delay to wait for collecting CoinGecko requests.
+    ///
+    /// The delay to start counting after receiving the first request.
+    #[clap(long, env, value_parser = humantime::parse_duration, group = "coin_gecko_buffered")]
+    pub coin_gecko_debouncing_time: Option<Duration>,
+
+    /// The timeout to wait for the result to be ready
+    #[clap(long, env, value_parser = humantime::parse_duration, group = "coin_gecko_buffered")]
+    pub coin_gecko_result_ready_timeout: Option<Duration>,
+
+    /// Maximum capacity of the broadcast channel to store the CoinGecko native
+    /// prices results
+    #[clap(long, env, group = "coin_gecko_buffered")]
+    pub coin_gecko_broadcast_channel_capacity: Option<usize>,
+}
+
 /// Controls which level of quote verification gets applied.
 #[derive(Copy, Clone, Debug, clap::ValueEnum)]
 #[clap(rename_all = "kebab-case")]
@@ -249,8 +287,7 @@ impl Display for Arguments {
             balancer_sor_url,
             one_inch_api_key,
             one_inch_url,
-            coin_gecko_api_key,
-            coin_gecko_url,
+            coin_gecko,
             quote_inaccuracy_limit,
             quote_verification,
             quote_timeout,
@@ -294,8 +331,36 @@ impl Display for Arguments {
         display_option(f, "balancer_sor_url", balancer_sor_url)?;
         display_secret_option(f, "one_inch_spot_price_api_key: {:?}", one_inch_api_key)?;
         writeln!(f, "one_inch_spot_price_api_url: {}", one_inch_url)?;
-        display_secret_option(f, "coin_gecko_api_key: {:?}", coin_gecko_api_key)?;
-        writeln!(f, "coin_gecko_api_url: {}", coin_gecko_url)?;
+        display_secret_option(
+            f,
+            "coin_gecko_api_key: {:?}",
+            &coin_gecko.coin_gecko_api_key,
+        )?;
+        writeln!(f, "coin_gecko_api_url: {}", coin_gecko.coin_gecko_url)?;
+        writeln!(f, "coin_gecko_api_url: {}", coin_gecko.coin_gecko_url)?;
+        writeln!(
+            f,
+            "coin_gecko_result_ready_timeout: {:?}",
+            coin_gecko
+                .coin_gecko_buffered
+                .as_ref()
+                .map(|coin_gecko_buffered| coin_gecko_buffered.coin_gecko_result_ready_timeout),
+        )?;
+        writeln!(
+            f,
+            "coin_gecko_debouncing_time: {:?}",
+            coin_gecko
+                .coin_gecko_buffered
+                .as_ref()
+                .map(|coin_gecko_buffered| coin_gecko_buffered.coin_gecko_debouncing_time),
+        )?;
+        writeln!(
+            f,
+            "coin_gecko_broadcast_channel_capacity: {:?}",
+            coin_gecko.coin_gecko_buffered.as_ref().map(
+                |coin_gecko_buffered| coin_gecko_buffered.coin_gecko_broadcast_channel_capacity
+            ),
+        )?;
         writeln!(f, "quote_inaccuracy_limit: {}", quote_inaccuracy_limit)?;
         writeln!(f, "quote_verification: {:?}", quote_verification)?;
         writeln!(f, "quote_timeout: {:?}", quote_timeout)?;
@@ -501,7 +566,7 @@ pub mod mocks {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
+    use {super::*, clap::Parser};
 
     #[test]
     fn string_repr_round_trip_native_price_estimators() {
@@ -526,4 +591,68 @@ mod tests {
             assert_eq!(stringified(&parsed(repr).unwrap()), repr);
         }
     }
+
+    #[test]
+    fn enable_coin_gecko_buffered() {
+        let args = vec![
+            "test", // Program name
+            "--coin-gecko-api-key",
+            "someapikey",
+            "--coin-gecko-url",
+            "https://api.coingecko.com/api/v3/simple/token_price",
+            "--coin-gecko-debouncing-time",
+            "300ms",
+            "--coin-gecko-result-ready-timeout",
+            "600ms",
+            "--coin-gecko-broadcast-channel-capacity",
+            "50",
+        ];
+
+        let coin_gecko = CoinGecko::parse_from(args);
+
+        assert!(coin_gecko.coin_gecko_buffered.is_some());
+
+        let buffered = coin_gecko.coin_gecko_buffered.unwrap();
+        assert_eq!(
+            buffered.coin_gecko_debouncing_time.unwrap(),
+            Duration::from_millis(300)
+        );
+        assert_eq!(
+            buffered.coin_gecko_result_ready_timeout.unwrap(),
+            Duration::from_millis(600)
+        );
+        assert_eq!(buffered.coin_gecko_broadcast_channel_capacity.unwrap(), 50);
+    }
+
+    #[test]
+    fn test_without_buffered_present() {
+        let args = vec![
+            "test", // Program name
+            "--coin-gecko-api-key",
+            "someapikey",
+            "--coin-gecko-url",
+            "https://api.coingecko.com/api/v3/simple/token_price",
+        ];
+
+        let coin_gecko = CoinGecko::parse_from(args);
+
+        assert!(coin_gecko.coin_gecko_buffered.is_none());
+    }
+
+    #[test]
+    fn test_invalid_partial_buffered_present() {
+        let args = vec![
+            "test", // Program name
+            "--coin-gecko-api-key",
+            "someapikey",
+            "--coin-gecko-url",
+            "https://api.coingecko.com/api/v3/simple/token_price",
+            "--coin-gecko-debouncing-time",
+            "300ms",
+        ];
+
+        let result = CoinGecko::try_parse_from(args);
+
+        assert!(result.is_err());
+    }
 }

crates/shared/src/price_estimation/factory.rs

@@ -19,7 +19,11 @@ use {
         code_simulation::{self, CodeSimulating, TenderlyCodeSimulator},
         ethrpc::Web3,
         http_client::HttpClientFactory,
-        price_estimation::{competition::PriceRanking, native::NativePriceEstimating},
+        price_estimation::{
+            buffered::{self, BufferedRequest, NativePriceBatchFetching},
+            competition::PriceRanking,
+            native::NativePriceEstimating,
+        },
         token_info::TokenInfoFetching,
     },
     anyhow::{Context as _, Result},
@@ -206,20 +210,46 @@ impl<'a> PriceEstimatorFactory<'a> {
                     self.components.tokens.clone(),
                 )),
             )),
-            NativePriceEstimatorSource::CoinGecko => Ok((
-                "CoinGecko".into(),
-                Arc::new(
-                    native::CoinGecko::new(
-                        self.components.http_factory.create(),
-                        self.args.coin_gecko_url.clone(),
-                        self.args.coin_gecko_api_key.clone(),
-                        self.network.chain_id,
-                        weth.address(),
-                        self.components.tokens.clone(),
-                    )
-                    .await?,
-                ),
-            )),
+            NativePriceEstimatorSource::CoinGecko => {
+                let coin_gecko = native::CoinGecko::new(
+                    self.components.http_factory.create(),
+                    self.args.coin_gecko.coin_gecko_url.clone(),
+                    self.args.coin_gecko.coin_gecko_api_key.clone(),
+                    self.network.chain_id,
+                    weth.address(),
+                    self.components.tokens.clone(),
+                )
+                .await?;
+
+                let coin_gecko: Arc<dyn NativePriceEstimating> =
+                    if let Some(coin_gecko_buffered_configuration) =
+                        &self.args.coin_gecko.coin_gecko_buffered
+                    {
+                        let configuration = buffered::Configuration {
+                            max_concurrent_requests: Some(
+                                coin_gecko
+                                    .max_batch_size()
+                                    .try_into()
+                                    .context("invalid CoinGecko max batch size")?,
+                            ),
+                            debouncing_time: coin_gecko_buffered_configuration
+                                .coin_gecko_debouncing_time
+                                .unwrap(),
+                            result_ready_timeout: coin_gecko_buffered_configuration
+                                .coin_gecko_result_ready_timeout
+                                .unwrap(),
+                            broadcast_channel_capacity: coin_gecko_buffered_configuration
+                                .coin_gecko_broadcast_channel_capacity
+                                .unwrap(),
+                        };
+
+                        Arc::new(BufferedRequest::with_config(coin_gecko, configuration))
+                    } else {
+                        Arc::new(coin_gecko)
+                    };
+
+                Ok(("CoinGecko".into(), coin_gecko))
+            }
         }
     }