Fix prices from slow-updating sources getting ignored

src/price_aggregator.rs

@@ -67,26 +67,29 @@ impl PriceAggregator {
         config: Arc<OracleConfig>,
     ) -> Result<Self> {
         let mut sources = vec![
-            SourceAdapter::new(BinanceSource::new(&config)),
-            SourceAdapter::new(ByBitSource::new(&config)),
-            SourceAdapter::new(CoinbaseSource::new(&config)),
-            SourceAdapter::new(MinswapSource::new(&config)?),
-            SourceAdapter::new(SpectrumSource::new(&config)?),
+            SourceAdapter::new(BinanceSource::new(&config), &config),
+            SourceAdapter::new(ByBitSource::new(&config), &config),
+            SourceAdapter::new(CoinbaseSource::new(&config), &config),
+            SourceAdapter::new(MinswapSource::new(&config)?, &config),
+            SourceAdapter::new(SpectrumSource::new(&config)?, &config),
         ];
         if let Some(maestro_source) = MaestroSource::new(&config)? {
-            sources.push(SourceAdapter::new(maestro_source));
+            sources.push(SourceAdapter::new(maestro_source, &config));
         } else {
             warn!("Not querying maestro, because no MAESTRO_API_KEY was provided");
         }
         if let Some(fxratesapi_source) = FxRatesApiSource::new(&config)? {
-            sources.push(SourceAdapter::new(fxratesapi_source));
+            sources.push(SourceAdapter::new(fxratesapi_source, &config));
         } else {
             warn!("Not querying FXRatesAPI, because no FXRATESAPI_API_KEY was provided");
         }
         if config.sundaeswap.use_api {
-            sources.push(SourceAdapter::new(SundaeSwapSource::new(&config)?));
+            sources.push(SourceAdapter::new(SundaeSwapSource::new(&config)?, &config));
         } else {
-            sources.push(SourceAdapter::new(SundaeSwapKupoSource::new(&config)?));
+            sources.push(SourceAdapter::new(
+                SundaeSwapKupoSource::new(&config)?,
+                &config,
+            ));
         }
         Ok(Self {
             feed_sink,

src/price_aggregator/source_adapter.rs

@@ -9,6 +9,7 @@ use tokio::{
 use tracing::{info_span, instrument, warn, Instrument};
 
 use crate::{
+    config::OracleConfig,
     health::{HealthSink, HealthStatus, Origin},
     sources::source::{PriceInfoSnapshot, PriceSink, Source},
 };
@@ -21,10 +22,10 @@ pub struct SourceAdapter {
 }
 
 impl SourceAdapter {
-    pub fn new<T: Source + Send + Sync + 'static>(source: T) -> Self {
+    pub fn new<T: Source + Send + Sync + 'static>(source: T, config: &OracleConfig) -> Self {
         Self {
             name: source.name(),
-            max_time_without_updates: source.max_time_without_updates(),
+            max_time_without_updates: source.max_time_without_updates(config),
             source: Box::new(source),
             prices: Arc::new(DashMap::new()),
         }

src/sources/fxratesapi.rs

@@ -30,7 +30,7 @@ impl Source for FxRatesApiSource {
     fn name(&self) -> String {
         "FXRatesAPI".into()
     }
-    fn max_time_without_updates(&self) -> Duration {
+    fn max_time_without_updates(&self, _config: &OracleConfig) -> Duration {
         let mut runs = self.schedule.upcoming(Utc);
         let next_run = runs.next().unwrap();
         let next_next_run = runs.next().unwrap();

src/sources/source.rs

@@ -5,6 +5,8 @@ use futures::future::BoxFuture;
 use rust_decimal::Decimal;
 use tokio::sync::mpsc;
 
+use crate::config::OracleConfig;
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct PriceInfo {
     pub token: String,
@@ -48,8 +50,8 @@ impl PriceSink {
 
 pub trait Source {
     fn name(&self) -> String;
-    fn max_time_without_updates(&self) -> Duration {
-        Duration::from_secs(30)
+    fn max_time_without_updates(&self, config: &OracleConfig) -> Duration {
+        config.max_source_price_age
     }
     fn tokens(&self) -> Vec<String>;
     fn query<'a>(&'a self, sink: &'a PriceSink) -> BoxFuture<'a, Result<()>>;