Robust gas oracles

ethers-middleware/src/gas_oracle/blocknative.rs

@@ -1,12 +1,9 @@
 use crate::gas_oracle::{GasCategory, GasOracle, GasOracleError, GWEI_TO_WEI};
 use async_trait::async_trait;
 use ethers_core::types::U256;
-use reqwest::{
-    header::{HeaderMap, HeaderValue, AUTHORIZATION},
-    Client, ClientBuilder,
-};
+use reqwest::{header::AUTHORIZATION, Client};
 use serde::Deserialize;
-use std::{collections::HashMap, convert::TryInto, iter::FromIterator};
+use std::{collections::HashMap, convert::TryInto};
 use url::Url;
 
 const BLOCKNATIVE_GAS_PRICE_ENDPOINT: &str = "https://api.blocknative.com/gasprices/blockprices";
@@ -26,6 +23,7 @@ fn gas_category_to_confidence(gas_category: &GasCategory) -> u64 {
 pub struct BlockNative {
     client: Client,
     url: Url,
+    api_key: String,
     gas_category: GasCategory,
 }
 
@@ -85,12 +83,10 @@ pub struct BaseFeeEstimate {
 
 impl BlockNative {
     /// Creates a new [BlockNative](https://www.blocknative.com/gas-estimator) gas oracle
-    pub fn new(api_key: &str) -> Self {
-        let header_value = HeaderValue::from_str(api_key).unwrap();
-        let headers = HeaderMap::from_iter([(AUTHORIZATION, header_value)]);
-        let client = ClientBuilder::new().default_headers(headers).build().unwrap();
+    pub fn new(client: Client, api_key: String) -> Self {
         Self {
             client,
+            api_key,
             url: BLOCKNATIVE_GAS_PRICE_ENDPOINT.try_into().unwrap(),
             gas_category: GasCategory::Standard,
         }
@@ -105,7 +101,15 @@ impl BlockNative {
 
     /// Perform request to Blocknative, decode response
     pub async fn request(&self) -> Result<BlockNativeGasResponse, GasOracleError> {
-        Ok(self.client.get(self.url.as_ref()).send().await?.error_for_status()?.json().await?)
+        self.client
+            .get(self.url.as_ref())
+            .header(AUTHORIZATION, &self.api_key)
+            .send()
+            .await?
+            .error_for_status()?
+            .json()
+            .await
+            .map_err(GasOracleError::HttpClientError)
     }
 }
 

ethers-middleware/src/gas_oracle/eth_gas_station.rs

@@ -61,15 +61,12 @@ pub struct EthGasStationResponse {
 
 impl EthGasStation {
     /// Creates a new [EthGasStation](https://docs.ethgasstation.info/) gas oracle
-    pub fn new(api_key: Option<&'static str>) -> Self {
-        let url = match api_key {
-            Some(key) => format!("{}?api-key={}", ETH_GAS_STATION_URL_PREFIX, key),
-            None => ETH_GAS_STATION_URL_PREFIX.to_string(),
-        };
-
-        let url = Url::parse(&url).expect("invalid url");
-
-        EthGasStation { client: Client::new(), url, gas_category: GasCategory::Standard }
+    pub fn new(client: Client, api_key: Option<&'static str>) -> Self {
+        let mut url = Url::parse(ETH_GAS_STATION_URL_PREFIX).expect("invalid url");
+        if let Some(key) = api_key {
+            url.query_pairs_mut().append_pair("api-key", key);
+        }
+        EthGasStation { client, url, gas_category: GasCategory::Standard }
     }
 
     /// Sets the gas price category to be used when fetching the gas price.
@@ -84,6 +81,12 @@ impl EthGasStation {
     }
 }
 
+impl Default for EthGasStation {
+    fn default() -> Self {
+        Self::new(Client::new(), None)
+    }
+}
+
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
 impl GasOracle for EthGasStation {

ethers-middleware/src/gas_oracle/etherchain.rs

@@ -18,12 +18,6 @@ pub struct Etherchain {
     gas_category: GasCategory,
 }
 
-impl Default for Etherchain {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
 #[derive(Clone, Debug, Deserialize, PartialEq, PartialOrd)]
 #[serde(rename_all = "camelCase")]
 pub struct EtherchainResponse {
@@ -37,10 +31,10 @@ pub struct EtherchainResponse {
 
 impl Etherchain {
     /// Creates a new [Etherchain](https://etherchain.org/tools/gasPriceOracle) gas price oracle.
-    pub fn new() -> Self {
+    pub fn new(client: Client) -> Self {
         let url = Url::parse(ETHERCHAIN_URL).expect("invalid url");
 
-        Etherchain { client: Client::new(), url, gas_category: GasCategory::Standard }
+        Etherchain { client, url, gas_category: GasCategory::Standard }
     }
 
     /// Sets the gas price category to be used when fetching the gas price.
@@ -55,6 +49,12 @@ impl Etherchain {
     }
 }
 
+impl Default for Etherchain {
+    fn default() -> Self {
+        Self::new(Client::new())
+    }
+}
+
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
 impl GasOracle for Etherchain {

ethers-middleware/src/gas_oracle/gas_now.rs

@@ -18,12 +18,6 @@ pub struct GasNow {
     gas_category: GasCategory,
 }
 
-impl Default for GasNow {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
 #[derive(Deserialize)]
 struct GasNowResponseWrapper {
     data: GasNowResponse,
@@ -39,10 +33,10 @@ pub struct GasNowResponse {
 
 impl GasNow {
     /// Creates a new [GasNow](https://gasnow.org) gas price oracle.
-    pub fn new() -> Self {
+    pub fn new(client: Client) -> Self {
         let url = Url::parse(GAS_NOW_URL).expect("invalid url");
 
-        Self { client: Client::new(), url, gas_category: GasCategory::Standard }
+        Self { url, gas_category: GasCategory::Standard }
     }
 
     /// Sets the gas price category to be used when fetching the gas price.
@@ -63,6 +57,12 @@ impl GasNow {
     }
 }
 
+impl Default for GasNow {
+    fn default() -> Self {
+        Self::new(Client::new())
+    }
+}
+
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
 impl GasOracle for GasNow {

ethers-middleware/src/gas_oracle/median.rs

@@ -0,0 +1,85 @@
+use crate::gas_oracle::{GasOracle, GasOracleError};
+use async_trait::async_trait;
+use ethers_core::types::U256;
+use futures_util::future::join_all;
+use std::fmt::Debug;
+use tracing::warn;
+
+#[derive(Debug)]
+pub struct Median<'a> {
+    oracles: Vec<Box<dyn 'a + GasOracle>>,
+}
+
+/// Computes the median gas price from a selection of oracles.
+///
+/// Don't forget to set a timeout on the source oracles. By default
+/// the reqwest based oracles will never time out.
+impl<'a> Median<'a> {
+    pub fn new(oracles: Vec<Box<dyn GasOracle>>) -> Self {
+        Self { oracles }
+    }
+
+    pub fn add<T: 'a + GasOracle>(&mut self, oracle: T) {
+        self.oracles.push(Box::new(oracle));
+    }
+}
+
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+impl GasOracle for Median<'_> {
+    async fn fetch(&self) -> Result<U256, GasOracleError> {
+        // Process the oracles in parallel
+        let futures = self.oracles.iter().map(|oracle| oracle.fetch());
+        let results = join_all(futures).await;
+
+        // Filter out any errors
+        let values = self.oracles.iter().zip(results).filter_map(|(oracle, result)| match result {
+            Ok(value) => Some(value),
+            Err(err) => {
+                warn!("Failed to fetch gas price from {:?}: {}", oracle, err);
+                None
+            }
+        });
+        let mut values = values.collect::<Vec<U256>>();
+        if values.is_empty() {
+            return Err(GasOracleError::NoValues)
+        }
+
+        // Sort the values and return the median
+        values.sort();
+        Ok(values[values.len() / 2])
+    }
+
+    async fn estimate_eip1559_fees(&self) -> Result<(U256, U256), GasOracleError> {
+        // Process the oracles in parallel
+        let futures = self.oracles.iter().map(|oracle| oracle.estimate_eip1559_fees());
+        let results = join_all(futures).await;
+
+        // Filter out any errors
+        let values = self.oracles.iter().zip(results).filter_map(|(oracle, result)| match result {
+            Ok(value) => Some(value),
+            Err(err) => {
+                warn!("Failed to fetch gas price from {:?}: {}", oracle, err);
+                None
+            }
+        });
+        let mut max_fee_per_gas = Vec::with_capacity(self.oracles.len());
+        let mut max_priority_fee_per_gas = Vec::with_capacity(self.oracles.len());
+        for (fee, priority) in values {
+            max_fee_per_gas.push(fee);
+            max_priority_fee_per_gas.push(priority);
+        }
+        assert_eq!(max_fee_per_gas.len(), max_priority_fee_per_gas.len());
+        if max_fee_per_gas.is_empty() {
+            return Err(GasOracleError::NoValues)
+        }
+
+        // Sort the values and return the median
+        max_fee_per_gas.sort();
+        max_priority_fee_per_gas.sort();
+        Ok((
+            max_fee_per_gas[max_fee_per_gas.len() / 2],
+            max_priority_fee_per_gas[max_priority_fee_per_gas.len() / 2],
+        ))
+    }
+}

ethers-middleware/src/gas_oracle/mod.rs

@@ -13,6 +13,9 @@ pub use etherscan::Etherscan;
 mod middleware;
 pub use middleware::{GasOracleMiddleware, MiddlewareError};
 
+mod median;
+pub use median::Median;
+
 use ethers_core::types::U256;
 
 use async_trait::async_trait;
@@ -58,6 +61,9 @@ pub enum GasOracleError {
 
     #[error("EIP-1559 gas estimation not supported")]
     Eip1559EstimationNotSupported,
+
+    #[error("None of the oracles returned a value")]
+    NoValues,
 }
 
 /// `GasOracle` is a trait that an underlying gas oracle needs to implement.

ethers-middleware/tests/gas_oracle.rs

@@ -57,7 +57,7 @@ async fn using_gas_oracle() {
 #[tokio::test]
 async fn eth_gas_station() {
     // initialize and fetch gas estimates from EthGasStation
-    let eth_gas_station_oracle = EthGasStation::new(None);
+    let eth_gas_station_oracle = EthGasStation::default();
     let data = eth_gas_station_oracle.fetch().await;
     assert!(data.is_ok());
 }
@@ -83,7 +83,7 @@ async fn etherscan() {
 #[tokio::test]
 async fn etherchain() {
     // initialize and fetch gas estimates from Etherchain
-    let etherchain_oracle = Etherchain::new().category(GasCategory::Fast);
+    let etherchain_oracle = Etherchain::default().category(GasCategory::Fast);
     let data = etherchain_oracle.fetch().await;
     assert!(data.is_ok());
 }