Incremental solvable orders cache update

crates/autopilot/src/boundary/mod.rs

@@ -36,7 +36,7 @@ pub fn web3_client(ethrpc: &Url, ethrpc_args: &shared::ethrpc::Arguments) -> Web
 }
 
 pub struct SolvableOrders {
-    pub orders: Vec<model::order::Order>,
+    pub orders: HashMap<domain::OrderUid, model::order::Order>,
     pub quotes: HashMap<domain::OrderUid, domain::Quote>,
     pub latest_settlement_block: u64,
 }

crates/autopilot/src/database/auction.rs

@@ -5,12 +5,13 @@ use {
     chrono::{DateTime, Utc},
     futures::{StreamExt, TryStreamExt},
     model::{order::Order, quote::QuoteId},
+    num::ToPrimitive,
     shared::{
         db_order_conversions::full_order_into_model_order,
         event_storing_helpers::{create_db_search_parameters, create_quote_row},
         order_quoting::{QuoteData, QuoteSearchParameters, QuoteStoring},
     },
-    std::ops::DerefMut,
+    std::{collections::HashMap, ops::DerefMut},
 };
 
 #[async_trait::async_trait]
@@ -60,7 +61,7 @@ impl QuoteStoring for Postgres {
 }
 
 impl Postgres {
-    pub async fn solvable_orders(&self, min_valid_to: u32) -> Result<boundary::SolvableOrders> {
+    pub async fn all_solvable_orders(&self, min_valid_to: u32) -> Result<boundary::SolvableOrders> {
         let _timer = super::Metrics::get()
             .database_queries
             .with_label_values(&["solvable_orders"])
@@ -73,24 +74,20 @@ impl Postgres {
         sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
             .execute(ex.deref_mut())
             .await?;
-        let orders: Vec<Order> = database::orders::solvable_orders(&mut ex, min_valid_to as i64)
-            .map(|result| match result {
-                Ok(order) => full_order_into_model_order(order),
-                Err(err) => Err(anyhow::Error::from(err)),
-            })
-            .try_collect()
-            .await?;
-        let latest_settlement_block =
-            database::orders::latest_settlement_block(&mut ex).await? as u64;
-        let quotes = self
-            .read_quotes(
-                orders
-                    .iter()
-                    .map(|order| domain::OrderUid(order.metadata.uid.0))
-                    .collect::<Vec<_>>()
-                    .iter(),
-            )
-            .await?;
+        let orders: HashMap<domain::OrderUid, Order> =
+            database::orders::solvable_orders(&mut ex, i64::from(min_valid_to))
+                .map(|result| match result {
+                    Ok(order) => full_order_into_model_order(order)
+                        .map(|order| (domain::OrderUid(order.metadata.uid.0), order)),
+                    Err(err) => Err(anyhow::Error::from(err)),
+                })
+                .try_collect()
+                .await?;
+        let latest_settlement_block = database::orders::latest_settlement_block(&mut ex)
+            .await?
+            .to_u64()
+            .context("latest_settlement_block is not u64")?;
+        let quotes = self.read_quotes(orders.keys()).await?;
         Ok(boundary::SolvableOrders {
             orders,
             quotes,

crates/autopilot/src/database/quotes.rs

@@ -30,6 +30,11 @@ impl Postgres {
         &self,
         orders: impl Iterator<Item = &domain::OrderUid>,
     ) -> Result<HashMap<domain::OrderUid, domain::Quote>, sqlx::Error> {
+        let _timer = super::Metrics::get()
+            .database_queries
+            .with_label_values(&["read_quotes"])
+            .start_timer();
+
         let mut ex = self.pool.acquire().await?;
         let order_uids: Vec<_> = orders.map(|uid| ByteArray(uid.0)).collect();
         let quotes: HashMap<_, _> = database::orders::read_quotes(&mut ex, &order_uids)

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

@@ -6,8 +6,9 @@ use {
         infra::persistence::dto::AuctionId,
     },
     anyhow::Context,
+    bigdecimal::ToPrimitive,
     boundary::database::byte_array::ByteArray,
-    chrono::Utc,
+    chrono::{DateTime, Utc},
     database::{
         order_events::OrderEventLabel,
         orders::{
@@ -22,10 +23,14 @@ use {
         SellTokenSource as DomainSellTokenSource,
         SigningScheme as DomainSigningScheme,
     },
-    number::conversions::{big_decimal_to_u256, u256_to_big_decimal},
+    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,
     std::{
         collections::{HashMap, HashSet},
+        ops::DerefMut,
         sync::Arc,
     },
     tracing::Instrument,
@@ -71,12 +76,13 @@ impl Persistence {
             .map_err(DatabaseError)
     }
 
-    pub async fn solvable_orders(
+    /// Finds solvable orders based on the order's min validity period.
+    pub async fn all_solvable_orders(
         &self,
         min_valid_to: u32,
     ) -> Result<boundary::SolvableOrders, DatabaseError> {
         self.postgres
-            .solvable_orders(min_valid_to)
+            .all_solvable_orders(min_valid_to)
             .await
             .map_err(DatabaseError)
     }
@@ -395,6 +401,139 @@ impl Persistence {
         Ok(solution)
     }
 
+    /// Computes solvable orders based on the latest observed block number,
+    /// order creation timestamp, and minimum validity period.
+    pub async fn solvable_orders_after(
+        &self,
+        current_orders: HashMap<domain::OrderUid, model::order::Order>,
+        after_timestamp: DateTime<Utc>,
+        after_block: u64,
+        min_valid_to: u32,
+    ) -> anyhow::Result<boundary::SolvableOrders> {
+        let after_block = i64::try_from(after_block).context("block number value exceeds i64")?;
+        let mut tx = self.postgres.pool.begin().await.context("begin")?;
+        // Set the transaction isolation level to REPEATABLE READ
+        // so all the SELECT queries below are executed in the same database snapshot
+        // taken at the moment before the first query is executed.
+        sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
+            .execute(tx.deref_mut())
+            .await?;
+
+        // Fetch the orders that were updated after the given block and were created or
+        // cancelled after the given timestamp.
+        let next_orders: HashMap<domain::OrderUid, model::order::Order> = {
+            let _timer = Metrics::get()
+                .database_queries
+                .with_label_values(&["open_orders_after"])
+                .start_timer();
+
+            database::orders::open_orders_after(&mut tx, after_block, after_timestamp)
+                .map(|result| match result {
+                    Ok(order) => full_order_into_model_order(order)
+                        .map(|order| (domain::OrderUid(order.metadata.uid.0), order)),
+                    Err(err) => Err(anyhow::Error::from(err)),
+                })
+                .try_collect()
+                .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,
+        )
+    }
+
+    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,
+    ) -> anyhow::Result<boundary::SolvableOrders> {
+        // Blindly insert all new orders into the cache.
+        for (uid, order) in next_orders {
+            current_orders.insert(uid, order);
+        }
+
+        // Filter out all the invalid orders.
+        current_orders.retain(|_uid, order| {
+            let expired = order.data.valid_to < min_valid_to
+                || order
+                    .metadata
+                    .ethflow_data
+                    .as_ref()
+                    .is_some_and(|data| data.user_valid_to < i64::from(min_valid_to));
+
+            let invalidated = order.metadata.invalidated;
+            let onchain_error = order
+                .metadata
+                .onchain_order_data
+                .as_ref()
+                .is_some_and(|data| data.placement_error.is_some());
+            let fulfilled = {
+                match order.data.kind {
+                    model::order::OrderKind::Sell => {
+                        order.metadata.executed_sell_amount
+                            >= u256_to_big_uint(&order.data.sell_amount)
+                    }
+                    model::order::OrderKind::Buy => {
+                        order.metadata.executed_buy_amount
+                            >= u256_to_big_uint(&order.data.buy_amount)
+                    }
+                }
+            };
+
+            !expired && !invalidated && !onchain_error && !fulfilled
+        });
+
+        // Keep only relevant quotes.
+        next_quotes.retain(|uid, _quote| current_orders.contains_key(uid));
+
+        Ok(boundary::SolvableOrders {
+            orders: current_orders,
+            quotes: next_quotes,
+            latest_settlement_block,
+        })
+    }
+
     /// Returns the oldest settlement event for which the accociated auction is
     /// not yet populated in the database.
     pub async fn get_settlement_without_auction(

crates/autopilot/src/run_loop.rs

@@ -150,7 +150,7 @@ impl RunLoop {
     }
 
     async fn cut_auction(&self) -> Option<domain::AuctionWithId> {
-        let auction = match self.solvable_orders_cache.current_auction() {
+        let auction = match self.solvable_orders_cache.current_auction().await {
             Some(auction) => auction,
             None => {
                 tracing::debug!("no current auction");