Fetch appdata in background (cowprotocol#3710)

xdecentralix · xdecentralix · commit f949b9a59b1c · 2025-11-17T16:08:59.000+01:00
@marcovc reported that enabling appdata fetching in the driver leads to very slow driver restarts. Since we used `join_all` the driver will only continuing the auction pre-processing if ALL futures finished. Because drivers running in the same k8s cluster as the orderbook have significantly lower latency we never saw that issue. In order to not block the pre-processing for an unreasonable amount of time I adjusted the logic to only await new appdatas for 500ms. That should give it enough time to fetch completely new appdatas we've never seen before (basically every new auction introduces at most a couple of new orders + appdatas). If that were the only change we would likely need MANY auctions to completely fill the cache with only 500ms per auction. To address that issue I adjusted the appdata fetcher to spawn 1 tokio task per appdata that needs to be fetched. That way the caller can await however many requests they want and all the ones they didn't wait for will still make progress while their solver already computes a solution for the current auction. In the next auction all (or at least many) of the missing appdata values should already be available in the cache. e2e tests should already cover that appdatas become available eventually.
diff --git a/crates/driver/src/domain/competition/order/app_data.rs b/crates/driver/src/domain/competition/order/app_data.rs
@@ -3,10 +3,8 @@ use {
     anyhow::Context,
     app_data::AppDataDocument,
     derive_more::From,
-    futures::FutureExt,
     moka::future::Cache,
     reqwest::StatusCode,
-    shared::request_sharing::BoxRequestSharing,
     std::{collections::HashMap, sync::Arc},
     thiserror::Error,
     url::Url,
@@ -30,10 +28,6 @@ pub struct AppDataRetriever(Arc<Inner>);
 struct Inner {
     client: reqwest::Client,
     base_url: Url,
-    request_sharing: BoxRequestSharing<
-        AppDataHash,
-        Result<Option<Arc<app_data::ValidatedAppData>>, FetchingError>,
-    >,
     cache: Cache<AppDataHash, Option<Arc<app_data::ValidatedAppData>>>,
 }
 
@@ -42,7 +36,6 @@ impl AppDataRetriever {
         Self(Arc::new(Inner {
             client: reqwest::Client::new(),
             base_url: orderbook_url,
-            request_sharing: BoxRequestSharing::labelled("app_data".to_string()),
             cache: Cache::new(cache_size),
         }))
     }
@@ -57,6 +50,9 @@ impl AppDataRetriever {
     }
 
     /// Retrieves the full app-data for the given `app_data` hash, if it exists.
+    /// HTTP requests needed to fetch the data are spawned in background tasks
+    /// such that they eventually populate the cache even in case the caller
+    /// stops awaiting the returned future.
     pub async fn get_cached_or_fetch(
         &self,
         app_data: &AppDataHash,
@@ -65,47 +61,38 @@ impl AppDataRetriever {
             return Ok(app_data.clone());
         }
 
-        let app_data_fut = move |app_data: &AppDataHash| {
-            let app_data = *app_data;
-            let self_ = self.clone();
-
-            async move {
-                let url = self_
-                    .0
-                    .base_url
-                    .join(&format!("api/v1/app_data/{:?}", app_data.0))?;
-                let response = self_.0.client.get(url).send().await?;
-                let validated_app_data = match response.status() {
-                    StatusCode::NOT_FOUND => None,
-                    _ => {
-                        let appdata: AppDataDocument =
-                            serde_json::from_str(&response.text().await?)
-                                .context("invalid app data document")?;
-                        match appdata.full_app_data == app_data::EMPTY {
-                            true => None, // empty app data
-                            false => Some(Arc::new(app_data::ValidatedAppData {
-                                hash: app_data::AppDataHash(app_data.0.0),
-                                protocol: app_data::parse(appdata.full_app_data.as_bytes())?,
-                                document: appdata.full_app_data,
-                            })),
-                        }
+        let inner = self.0.clone();
+        let app_data = *app_data;
+
+        let fut = async move {
+            let url = inner
+                .base_url
+                .join(&format!("api/v1/app_data/{:?}", app_data.0))?;
+            let response = inner.client.get(url).send().await?;
+            let validated_app_data = match response.status() {
+                StatusCode::NOT_FOUND => None,
+                _ => {
+                    let appdata: AppDataDocument = serde_json::from_str(&response.text().await?)
+                        .context("invalid app data document")?;
+                    match appdata.full_app_data == app_data::EMPTY {
+                        true => None, // empty app data
+                        false => Some(Arc::new(app_data::ValidatedAppData {
+                            hash: app_data::AppDataHash(app_data.0.0),
+                            protocol: app_data::parse(appdata.full_app_data.as_bytes())?,
+                            document: appdata.full_app_data,
+                        })),
                     }
-                };
-                self_
-                    .0
-                    .cache
-                    .insert(app_data, validated_app_data.clone())
-                    .await;
-
-                Ok(validated_app_data)
-            }
-            .boxed()
+                }
+            };
+            inner
+                .cache
+                .insert(app_data, validated_app_data.clone())
+                .await;
+
+            Ok(validated_app_data)
         };
 
-        self.0
-            .request_sharing
-            .shared_or_else(*app_data, app_data_fut)
-            .await
+        tokio::task::spawn(fut).await?
     }
 }
 
@@ -182,6 +169,8 @@ pub enum FetchingError {
     InvalidAppData(#[from] anyhow::Error),
     #[error("internal error: {0}")]
     Internal(String),
+    #[error("failed to join task: {0}")]
+    TaskJoinFailed(#[from] tokio::task::JoinError),
 }
 
 impl From<reqwest::Error> for FetchingError {
@@ -195,13 +184,3 @@ impl From<url::ParseError> for FetchingError {
         FetchingError::Internal(err.to_string())
     }
 }
-
-impl Clone for FetchingError {
-    fn clone(&self) -> Self {
-        match self {
-            Self::Http(message) => Self::Http(message.clone()),
-            Self::InvalidAppData(err) => Self::InvalidAppData(shared::clone_anyhow_error(err)),
-            Self::Internal(message) => Self::Internal(message.clone()),
-        }
-    }
-}
diff --git a/crates/driver/src/domain/competition/pre_processing.rs b/crates/driver/src/domain/competition/pre_processing.rs
@@ -11,10 +11,7 @@ use {
     },
     anyhow::{Context, Result},
     chrono::Utc,
-    futures::{
-        FutureExt,
-        future::{BoxFuture, join_all},
-    },
+    futures::{FutureExt, StreamExt, future::BoxFuture, stream::FuturesUnordered},
     itertools::Itertools,
     model::{
         interaction::InteractionData,
@@ -26,8 +23,7 @@ use {
         price_estimation::trade_verifier::balance_overrides::BalanceOverrideRequest,
         signature_validator::SignatureValidating,
     },
-    std::{collections::HashMap, future::Future, sync::Arc},
-    tap::TapFallible,
+    std::{collections::HashMap, future::Future, sync::Arc, time::Duration},
     tokio::sync::Mutex,
     tracing::Instrument,
 };
@@ -339,36 +335,43 @@ impl Utilities {
         let _timer2 =
             observe::metrics::metrics().on_auction_overhead_start("driver", "fetch_app_data");
 
-        let app_data = join_all(
-            auction
-                .orders
-                .iter()
-                .flat_map(|order| match order.app_data {
-                    AppData::Full(_) => None,
-                    // only fetch appdata we don't already have in full
-                    AppData::Hash(hash) => Some(hash),
-                })
-                .unique()
-                .map(|app_data_hash| {
-                    let app_data_retriever = app_data_retriever.clone();
-                    async move {
-                        let fetched_app_data = app_data_retriever
-                            .get_cached_or_fetch(&app_data_hash)
-                            .await
-                            .tap_err(|err| {
-                                tracing::warn!(?app_data_hash, ?err, "failed to fetch app data");
-                            })
-                            .ok()
-                            .flatten();
+        let futures: FuturesUnordered<_> = auction
+            .orders
+            .iter()
+            .flat_map(|order| match order.app_data {
+                AppData::Full(_) => None,
+                // only fetch appdata we don't already have in full
+                AppData::Hash(hash) => Some(hash),
+            })
+            .unique()
+            .map(async move |app_data_hash| {
+                let fetched_app_data = app_data_retriever
+                    .get_cached_or_fetch(&app_data_hash)
+                    .await
+                    .inspect_err(|err| {
+                        tracing::warn!(?app_data_hash, ?err, "failed to fetch app data");
+                    })
+                    .ok()
+                    .flatten();
+
+                (app_data_hash, fetched_app_data)
+            })
+            .collect();
 
-                        (app_data_hash, fetched_app_data)
-                    }
-                }),
-        )
-        .await
-        .into_iter()
-        .filter_map(|(app_data_hash, app_data)| app_data.map(|app_data| (app_data_hash, app_data)))
-        .collect::<HashMap<_, _>>();
+        // Only await responses for a short amount of time. Even if we don't await
+        // all futures fully the remaining appdata requests will finish in background
+        // tasks. That way we should have enough time to immediately fetch appdatas
+        // of new orders (once the cache is filled). But we also don't run the risk
+        // of stalling the driver completely until everything is fetched.
+        // In practice that means the solver will only see a few appdatas in the first
+        // auction after a restart. But on subsequent auctions everything should be
+        // available.
+        const MAX_APP_DATA_WAIT: Duration = Duration::from_millis(500);
+        let app_data: HashMap<_, _> = futures
+            .take_until(tokio::time::sleep(MAX_APP_DATA_WAIT))
+            .filter_map(async move |(hash, json)| Some((hash, json?)))
+            .collect()
+            .await;
 
         Arc::new(app_data)
     }
diff --git a/crates/liquidity-driver/src/domain/competition/order/app_data.rs b/crates/liquidity-driver/src/domain/competition/order/app_data.rs
@@ -3,10 +3,8 @@ use {
     anyhow::Context,
     app_data::AppDataDocument,
     derive_more::From,
-    futures::FutureExt,
     moka::future::Cache,
     reqwest::StatusCode,
-    shared::request_sharing::BoxRequestSharing,
     std::{collections::HashMap, sync::Arc},
     thiserror::Error,
     url::Url,
@@ -30,10 +28,6 @@ pub struct AppDataRetriever(Arc<Inner>);
 struct Inner {
     client: reqwest::Client,
     base_url: Url,
-    request_sharing: BoxRequestSharing<
-        AppDataHash,
-        Result<Option<Arc<app_data::ValidatedAppData>>, FetchingError>,
-    >,
     cache: Cache<AppDataHash, Option<Arc<app_data::ValidatedAppData>>>,
 }
 
@@ -42,7 +36,6 @@ impl AppDataRetriever {
         Self(Arc::new(Inner {
             client: reqwest::Client::new(),
             base_url: orderbook_url,
-            request_sharing: BoxRequestSharing::labelled("app_data".to_string()),
             cache: Cache::new(cache_size),
         }))
     }
@@ -57,6 +50,9 @@ impl AppDataRetriever {
     }
 
     /// Retrieves the full app-data for the given `app_data` hash, if it exists.
+    /// HTTP requests needed to fetch the data are spawned in background tasks
+    /// such that they eventually populate the cache even in case the caller
+    /// stops awaiting the returned future.
     pub async fn get_cached_or_fetch(
         &self,
         app_data: &AppDataHash,
@@ -65,47 +61,38 @@ impl AppDataRetriever {
             return Ok(app_data.clone());
         }
 
-        let app_data_fut = move |app_data: &AppDataHash| {
-            let app_data = *app_data;
-            let self_ = self.clone();
-
-            async move {
-                let url = self_
-                    .0
-                    .base_url
-                    .join(&format!("api/v1/app_data/{:?}", app_data.0))?;
-                let response = self_.0.client.get(url).send().await?;
-                let validated_app_data = match response.status() {
-                    StatusCode::NOT_FOUND => None,
-                    _ => {
-                        let appdata: AppDataDocument =
-                            serde_json::from_str(&response.text().await?)
-                                .context("invalid app data document")?;
-                        match appdata.full_app_data == app_data::EMPTY {
-                            true => None, // empty app data
-                            false => Some(Arc::new(app_data::ValidatedAppData {
-                                hash: app_data::AppDataHash(app_data.0.0),
-                                protocol: app_data::parse(appdata.full_app_data.as_bytes())?,
-                                document: appdata.full_app_data,
-                            })),
-                        }
+        let inner = self.0.clone();
+        let app_data = *app_data;
+
+        let fut = async move {
+            let url = inner
+                .base_url
+                .join(&format!("api/v1/app_data/{:?}", app_data.0))?;
+            let response = inner.client.get(url).send().await?;
+            let validated_app_data = match response.status() {
+                StatusCode::NOT_FOUND => None,
+                _ => {
+                    let appdata: AppDataDocument = serde_json::from_str(&response.text().await?)
+                        .context("invalid app data document")?;
+                    match appdata.full_app_data == app_data::EMPTY {
+                        true => None, // empty app data
+                        false => Some(Arc::new(app_data::ValidatedAppData {
+                            hash: app_data::AppDataHash(app_data.0.0),
+                            protocol: app_data::parse(appdata.full_app_data.as_bytes())?,
+                            document: appdata.full_app_data,
+                        })),
                     }
-                };
-                self_
-                    .0
-                    .cache
-                    .insert(app_data, validated_app_data.clone())
-                    .await;
-
-                Ok(validated_app_data)
-            }
-            .boxed()
+                }
+            };
+            inner
+                .cache
+                .insert(app_data, validated_app_data.clone())
+                .await;
+
+            Ok(validated_app_data)
         };
 
-        self.0
-            .request_sharing
-            .shared_or_else(*app_data, app_data_fut)
-            .await
+        tokio::task::spawn(fut).await?
     }
 }
 
@@ -182,6 +169,8 @@ pub enum FetchingError {
     InvalidAppData(#[from] anyhow::Error),
     #[error("internal error: {0}")]
     Internal(String),
+    #[error("failed to join task: {0}")]
+    TaskJoinFailed(#[from] tokio::task::JoinError),
 }
 
 impl From<reqwest::Error> for FetchingError {
@@ -195,13 +184,3 @@ impl From<url::ParseError> for FetchingError {
         FetchingError::Internal(err.to_string())
     }
 }
-
-impl Clone for FetchingError {
-    fn clone(&self) -> Self {
-        match self {
-            Self::Http(message) => Self::Http(message.clone()),
-            Self::InvalidAppData(err) => Self::InvalidAppData(shared::clone_anyhow_error(err)),
-            Self::Internal(message) => Self::Internal(message.clone()),
-        }
-    }
-}
diff --git a/crates/liquidity-driver/src/domain/competition/pre_processing.rs b/crates/liquidity-driver/src/domain/competition/pre_processing.rs
