Log original request ID when reusing shared in-flight quote requests (#4300)

## Background
Fast and optimal quoters share the same `ExternalTradeFinder` instance
(via `Arc<dyn TradeFinding>`), so `RequestSharing` already deduplicates
identical concurrent requests. However, when a hitchhiker reused an
in-flight request, there was no trace linking it back to the original
HTTP call, making it hard to debug why the optimal quoter failed to
provide solutions, while the fast one succeeded for basically the same
quote competition or vice versa.

## Changes
- Extended `RequestSharing::shared_or_else` to return a `SharedResult`
that indicates whether an existing in-flight request was reused
(`is_shared` flag), improving observability of request deduplication.
- In `ExternalTradeFinder`, the HTTP request ID (`X-REQUEST-ID`) is now
embedded in the shared future's result. When a caller (e.g. the fast or
optimal quoter) reuses an in-flight request instead of sending a new
one, a debug log is emitted with the `original_request_id` — making it
possible to trace which HTTP request produced a given quote result.
- All other `shared_or_else` callers updated to use the new `.future`
field (no behavioral change).

## How to test
Using staging, use the cowswap UI to get some quotes. Search for the new
log.

Author: squadgazzz

crates/liquidity-sources/src/recent_block_cache.rs

@@ -265,7 +265,7 @@ where
         let retries = self.maximum_retries;
         let delay = self.delay_between_retries;
         let fetcher = self.fetcher.clone();
-        let fut = self.requests.shared_or_else((key, block), |entry| {
+        let shared = self.requests.shared_or_else((key, block), |entry| {
             let (key, block) = entry.clone();
             async move {
                 for _ in 0..=retries {
@@ -280,7 +280,7 @@ where
             }
             .boxed()
         });
-        fut.await.context("could not fetch liquidity")
+        shared.await.context("could not fetch liquidity")
     }
 
     async fn fetch(&self, keys: impl IntoIterator<Item = K>, block: Block) -> Result<Vec<V>> {

crates/price-estimation/src/trade_finding/external.rs

@@ -21,19 +21,28 @@ use {
     futures::FutureExt,
     observe::tracing::distributed::headers::tracing_headers,
     request_sharing::{BoxRequestSharing, RequestSharing},
-    reqwest::{Client, header},
+    reqwest::{Client, StatusCode, header},
     tracing::instrument,
     url::Url,
 };
 
+/// Wraps a trade result with the request ID of the HTTP request that produced
+/// it, so that consumers reusing a shared in-flight request can identify the
+/// original request.
+#[derive(Clone)]
+struct SharedTradeResponse {
+    result: Result<TradeKind, PriceEstimationError>,
+    request_id: Option<String>,
+}
+
 pub struct ExternalTradeFinder {
     /// URL to call to in the driver to get a quote with call data for a trade.
     quote_endpoint: Url,
 
     /// Utility to make sure no 2 identical requests are in-flight at the same
     /// time. Instead of issuing a duplicated request this awaits the
     /// response of the in-flight request.
-    sharing: BoxRequestSharing<Query, Result<TradeKind, PriceEstimationError>>,
+    sharing: BoxRequestSharing<Query, SharedTradeResponse>,
 
     /// Client to issue http requests with.
     client: Client,
@@ -83,39 +92,55 @@ impl ExternalTradeFinder {
                     request = request.header("X-Current-Block-Hash", block_hash.to_string())
                 }
 
-                if let Some(id) = id {
-                    request = request.header("X-REQUEST-ID", id);
+                if let Some(ref id) = id {
+                    request = request.header("X-REQUEST-ID", id.clone());
                 }
 
-                let response = request
-                    .timeout(timeout)
-                    .send()
-                    .await
-                    .map_err(|err| PriceEstimationError::EstimatorInternal(anyhow!(err)))?;
-                if response.status() == 429 {
-                    return Err(PriceEstimationError::RateLimited);
+                let result = async {
+                    let response = request
+                        .timeout(timeout)
+                        .send()
+                        .await
+                        .map_err(|err| PriceEstimationError::EstimatorInternal(anyhow!(err)))?;
+                    if response.status() == StatusCode::TOO_MANY_REQUESTS {
+                        return Err(PriceEstimationError::RateLimited);
+                    }
+                    let text = response
+                        .text()
+                        .await
+                        .map_err(|err| PriceEstimationError::EstimatorInternal(anyhow!(err)))?;
+                    serde_json::from_str::<dto::QuoteKind>(&text)
+                        .map(TradeKind::from)
+                        .map_err(|err| {
+                            serde_json::from_str::<dto::Error>(&text)
+                                .map(PriceEstimationError::from)
+                                .unwrap_or_else(|_| {
+                                    PriceEstimationError::EstimatorInternal(anyhow!(err))
+                                })
+                        })
+                }
+                .await;
+
+                SharedTradeResponse {
+                    result,
+                    request_id: id,
                 }
-                let text = response
-                    .text()
-                    .await
-                    .map_err(|err| PriceEstimationError::EstimatorInternal(anyhow!(err)))?;
-                serde_json::from_str::<dto::QuoteKind>(&text)
-                    .map(TradeKind::from)
-                    .map_err(|err| {
-                        if let Ok(err) = serde_json::from_str::<dto::Error>(&text) {
-                            PriceEstimationError::from(err)
-                        } else {
-                            PriceEstimationError::EstimatorInternal(anyhow!(err))
-                        }
-                    })
             }
             .boxed()
         };
 
-        self.sharing
-            .shared_or_else(query.clone(), fut)
-            .await
-            .map_err(TradeError::from)
+        let shared = self.sharing.shared_or_else(query.clone(), fut);
+        let is_shared = shared.is_shared;
+        let response = shared.await;
+
+        if is_shared {
+            tracing::debug!(
+                original_request_id = ?response.request_id,
+                "reusing in-flight quote request"
+            );
+        }
+
+        response.result.map_err(TradeError::from)
     }
 }
 

crates/request-sharing/src/lib.rs

@@ -11,7 +11,9 @@ use {
         collections::HashMap,
         future::Future,
         hash::Hash,
+        pin::Pin,
         sync::{Arc, Mutex},
+        task::{Context, Poll},
         time::Duration,
     },
 };
@@ -39,6 +41,28 @@ pub type BoxRequestSharing<Request, Response> =
 /// A boxed shared future.
 pub type BoxShared<T> = Shared<BoxFuture<'static, T>>;
 
+/// Result of [`RequestSharing::shared_or_else`] indicating whether an
+/// already in-flight future was reused or a new one was created.
+///
+/// Implements [`Future`] so it can be awaited directly.
+pub struct SharedResult<Fut: Future> {
+    future: Shared<Fut>,
+    /// `true` when an existing in-flight request was reused instead of
+    /// starting a new one.
+    pub is_shared: bool,
+}
+
+impl<Fut: Future> Future for SharedResult<Fut>
+where
+    Fut::Output: Clone,
+{
+    type Output = Fut::Output;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        Pin::new(&mut self.future).poll(cx)
+    }
+}
+
 type Cache<Request, Response> = Arc<Mutex<HashMap<Request, WeakShared<Response>>>>;
 
 impl<Request: Send + 'static, Fut: Future + Send + 'static> RequestSharing<Request, Fut>
@@ -100,7 +124,7 @@ where
 {
     /// Returns an existing in flight future or creates and uses a new future
     /// from the specified closure.
-    pub fn shared_or_else<F>(&self, request: Request, future: F) -> Shared<Fut>
+    pub fn shared_or_else<F>(&self, request: Request, future: F) -> SharedResult<Fut>
     where
         F: FnOnce(&Request) -> Fut,
     {
@@ -113,7 +137,10 @@ where
                 .request_sharing_access
                 .with_label_values(&[self.request_label.as_str(), "hits"])
                 .inc();
-            return existing;
+            return SharedResult {
+                future: existing,
+                is_shared: true,
+            };
         }
 
         Metrics::get()
@@ -129,7 +156,10 @@ where
             .request_sharing_cached_items
             .with_label_values(&[&self.request_label])
             .set(in_flight.len() as u64);
-        shared
+        SharedResult {
+            future: shared,
+            is_shared: false,
+        }
     }
 }
 
@@ -165,30 +195,26 @@ mod tests {
             request_label: label.clone(),
         };
 
-        let shared0 = sharing.shared_or_else(0, |_| futures::future::ready(0).boxed());
-        let shared1 = sharing.shared_or_else(0, |_| async { panic!() }.boxed());
+        let result0 = sharing.shared_or_else(0, |_| futures::future::ready(0).boxed());
+        let result1 = sharing.shared_or_else(0, |_| async { panic!() }.boxed());
 
-        assert!(shared0.ptr_eq(&shared1));
-        assert_eq!(shared0.strong_count().unwrap(), 2);
-        assert_eq!(shared1.strong_count().unwrap(), 2);
-        assert_eq!(shared0.weak_count().unwrap(), 1);
+        assert!(!result0.is_shared);
+        assert!(result1.is_shared);
 
-        // complete first shared
-        assert_eq!(shared0.now_or_never().unwrap(), 0);
-        assert_eq!(shared1.strong_count().unwrap(), 1);
-        assert_eq!(shared1.weak_count().unwrap(), 1);
+        // Complete first shared — result1 still holds a reference.
+        assert_eq!(result0.await, 0);
 
-        // GC does not delete any keys because some tasks still use the future
+        // GC does not delete because result1 still references the future.
         RequestSharing::collect_garbage(&sharing.in_flight, &label);
         assert_eq!(sharing.in_flight.lock().unwrap().len(), 1);
         assert!(sharing.in_flight.lock().unwrap().get(&0).is_some());
 
-        // complete second shared
-        assert_eq!(shared1.now_or_never().unwrap(), 0);
+        // Complete second shared — proves sharing since its factory would panic.
+        assert_eq!(result1.await, 0);
 
         RequestSharing::collect_garbage(&sharing.in_flight, &label);
 
-        // GC deleted all now unused futures
+        // GC deleted all now unused futures.
         assert!(sharing.in_flight.lock().unwrap().is_empty());
     }
 }