Autopilot: fetch only relevant quotes (#2967)

# Description
The latest prod release showed that quotes fetching is one of the most
expensive operations in the incremental solvable orders cache update.

# Changes
This can be slightly improved by fetching quotes for only newly created
and placed onchain orders

## How to test
Existing tests.

crates/autopilot/src/infra/persistence/mod.rs

@@ -25,7 +25,6 @@ use {
         SigningScheme as DomainSigningScheme,
     },
     futures::{StreamExt, TryStreamExt},
-    itertools::Itertools,
     number::conversions::{big_decimal_to_u256, u256_to_big_decimal, u256_to_big_uint},
     primitive_types::{H160, H256},
     shared::db_order_conversions::full_order_into_model_order,
@@ -406,7 +405,8 @@ impl Persistence {
     /// order creation timestamp, and minimum validity period.
     pub async fn solvable_orders_after(
         &self,
-        current_orders: HashMap<domain::OrderUid, model::order::Order>,
+        mut current_orders: HashMap<domain::OrderUid, model::order::Order>,
+        mut current_quotes: HashMap<domain::OrderUid, domain::Quote>,
         after_timestamp: DateTime<Utc>,
         after_block: u64,
         min_valid_to: u32,
@@ -454,59 +454,11 @@ impl Persistence {
             .await?
         };
 
-        // Fetch quotes for new orders and also update them for the cached ones since
-        // they could also be updated.
-        let updated_quotes = {
-            let _timer = Metrics::get()
-                .database_queries
-                .with_label_values(&["read_quotes"])
-                .start_timer();
-
-            let all_order_uids = next_orders
-                .keys()
-                .chain(current_orders.keys())
-                .unique()
-                .map(|uid| ByteArray(uid.0))
-                .collect::<Vec<_>>();
-
-            database::orders::read_quotes(&mut tx, &all_order_uids)
-                .await?
-                .into_iter()
-                .filter_map(|quote| {
-                    let order_uid = domain::OrderUid(quote.order_uid.0);
-                    dto::quote::into_domain(quote)
-                        .map_err(|err| {
-                            tracing::warn!(?order_uid, ?err, "failed to convert quote from db")
-                        })
-                        .ok()
-                        .map(|quote| (order_uid, quote))
-                })
-                .collect()
-        };
-
         let latest_settlement_block = database::orders::latest_settlement_block(&mut tx)
             .await?
             .to_u64()
             .context("latest_settlement_block is not u64")?;
 
-        Self::build_solvable_orders(
-            current_orders,
-            next_orders,
-            updated_quotes,
-            latest_settlement_block,
-            min_valid_to,
-            started_at,
-        )
-    }
-
-    fn build_solvable_orders(
-        mut current_orders: HashMap<domain::OrderUid, model::order::Order>,
-        next_orders: HashMap<domain::OrderUid, model::order::Order>,
-        mut next_quotes: HashMap<domain::OrderUid, domain::Quote>,
-        latest_settlement_block: u64,
-        min_valid_to: u32,
-        started_at: chrono::DateTime<chrono::Utc>,
-    ) -> anyhow::Result<boundary::SolvableOrders> {
         // Blindly insert all new orders into the cache.
         for (uid, order) in next_orders {
             current_orders.insert(uid, order);
@@ -543,12 +495,42 @@ impl Persistence {
             !expired && !invalidated && !onchain_error && !fulfilled
         });
 
-        // Keep only relevant quotes.
-        next_quotes.retain(|uid, _quote| current_orders.contains_key(uid));
+        current_quotes.retain(|uid, _| current_orders.contains_key(uid));
+
+        {
+            let _timer = Metrics::get()
+                .database_queries
+                .with_label_values(&["read_quotes"])
+                .start_timer();
+
+            // Fetch quotes only for newly created and also on-chain placed orders due to
+            // the following case: if a block containing an on-chain order
+            // (e.g., ethflow) gets reorganized, the same order with the same
+            // UID might be created in the new block, and the temporary quote
+            // associated with it may have changed in the meantime.
+            let order_uids = current_orders
+                .values()
+                .filter_map(|order| {
+                    (order.metadata.onchain_user.is_some()
+                        || order.metadata.creation_date > after_timestamp)
+                        .then_some(ByteArray(order.metadata.uid.0))
+                })
+                .collect::<Vec<_>>();
+
+            for quote in database::orders::read_quotes(&mut tx, &order_uids).await? {
+                let order_uid = domain::OrderUid(quote.order_uid.0);
+                match dto::quote::into_domain(quote) {
+                    Ok(quote) => {
+                        current_quotes.insert(order_uid, quote);
+                    }
+                    Err(err) => tracing::warn!(?order_uid, ?err, "failed to convert quote from db"),
+                }
+            }
+        };
 
         Ok(boundary::SolvableOrders {
             orders: current_orders,
-            quotes: next_quotes,
+            quotes: current_quotes,
             latest_settlement_block,
             fetched_from_db: started_at,
         })

crates/autopilot/src/solvable_orders.rs

@@ -349,6 +349,7 @@ impl SolvableOrdersCache {
                 .persistence
                 .solvable_orders_after(
                     cache.solvable_orders.orders.clone(),
+                    cache.solvable_orders.quotes.clone(),
                     cache.solvable_orders.fetched_from_db,
                     cache.solvable_orders.latest_settlement_block,
                     min_valid_to,