(fast7) Kick off fast path handling in runloop

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

@@ -38,7 +38,7 @@ use {
     eth_domain_types as eth,
     futures::{StreamExt, TryStreamExt},
     number::conversions::{big_decimal_to_u256, u256_to_big_decimal, u256_to_big_uint},
-    shared::db_order_conversions::full_order_into_model_order,
+    shared::db_order_conversions::{fast_path_order_into_model, full_order_into_model_order},
     std::{
         collections::{HashMap, HashSet},
         ops::DerefMut,
@@ -1037,6 +1037,143 @@ impl Persistence {
             .map(|o| crate::domain::OrderUid(o.0))
             .collect())
     }
+
+    /// Recovers what's needed to settle a fast-path order via the driver's
+    /// `/settle`, or `None` when `uid` is not a fast-path order (its quote's
+    /// competition was not persisted).
+    pub async fn fast_path_order(
+        &self,
+        uid: domain::OrderUid,
+    ) -> anyhow::Result<Option<FastPathOrder>> {
+        let _timer = Metrics::get()
+            .database_queries
+            .with_label_values(&["fast_path_order"])
+            .start_timer();
+
+        let mut ex = self.postgres.pool.acquire().await.context("acquire")?;
+        let key = ByteArray(uid.0);
+
+        let Some(row) = database::fast_path::single_fast_path_order(&mut ex, &key).await? else {
+            return Ok(None);
+        };
+
+        let model_order = fast_path_order_into_model(&row)?;
+
+        let native_prices = row
+            .price_tokens
+            .iter()
+            .zip(&row.price_values)
+            .map(|(token, value)| {
+                let price = big_decimal_to_u256(value).context("invalid native price")?;
+                anyhow::Ok((eth::Address::from(token.0), price))
+            })
+            .collect::<anyhow::Result<HashMap<_, _>>>()?;
+
+        Ok(Some(FastPathOrder {
+            model_order,
+            auction_id: row.auction_id,
+            solution_id: row
+                .solution_id
+                .to_u64()
+                .context("solution id out of range")?,
+            solution_uid: row
+                .solution_uid
+                .to_usize()
+                .context("solution uid out of range")?,
+            solver: eth::Address::from(row.solver.0),
+            raw_sell: big_decimal_to_u256(&row.executed_sell)
+                .context("invalid executed sell amount")?,
+            raw_buy: big_decimal_to_u256(&row.executed_buy)
+                .context("invalid executed buy amount")?,
+            native_prices,
+        }))
+    }
+
+    /// Applies fees to every solver's bid on a fast-path order and stamps
+    /// the applicable fee policies. Runs after the autopilot picks up the
+    /// placed order and computes the policies via `ProtocolFees::apply`.
+    ///
+    /// Each bid's own raw `executed_sell`/`executed_buy` is adjusted by the
+    /// same volume factors so the recorded amounts stay consistent across
+    /// the whole competition — not just the winning row.
+    pub async fn record_fast_path_fees(
+        &self,
+        auction_id: database::auction::AuctionId,
+        order_uid: domain::OrderUid,
+        order_kind: model::order::OrderKind,
+        volume_factors: &[configs::fee_factor::FeeFactor],
+        fee_policies: &[domain::fee::Policy],
+    ) -> anyhow::Result<()> {
+        let _timer = Metrics::get()
+            .database_queries
+            .with_label_values(&["record_fast_path_fees"])
+            .start_timer();
+
+        let uid = ByteArray(order_uid.0);
+        let policy_rows: Vec<_> = fee_policies
+            .iter()
+            .map(|p| dto::fee_policy::from_domain(auction_id, order_uid, *p))
+            .collect();
+
+        let mut tx = self.postgres.pool.begin().await.context("begin")?;
+        let bids = database::fast_path::fast_path_bids(tx.deref_mut(), auction_id, uid)
+            .await
+            .context("fetch fast-path bids")?;
+        let adjusted_bids: Vec<_> = bids
+            .into_iter()
+            .map(|bid| {
+                let raw_sell = big_decimal_to_u256(&bid.executed_sell)
+                    .context("bid executed_sell not a U256")?;
+                let raw_buy = big_decimal_to_u256(&bid.executed_buy)
+                    .context("bid executed_buy not a U256")?;
+                let (adjusted_sell, adjusted_buy) = shared::fee::apply_volume_fees(
+                    raw_sell,
+                    raw_buy,
+                    order_kind,
+                    volume_factors.iter().copied(),
+                );
+                anyhow::Ok(database::fast_path::FastPathBid {
+                    solution_uid: bid.solution_uid,
+                    executed_sell: u256_to_big_decimal(&adjusted_sell),
+                    executed_buy: u256_to_big_decimal(&adjusted_buy),
+                })
+            })
+            .collect::<anyhow::Result<_>>()?;
+        database::fast_path::apply_fees_to_fast_path_bids(
+            tx.deref_mut(),
+            auction_id,
+            uid,
+            &adjusted_bids,
+        )
+        .await
+        .context("apply_fees_to_fast_path_bids")?;
+        database::fee_policies::insert_batch(tx.deref_mut(), policy_rows)
+            .await
+            .context("insert fast-path fee policies")?;
+        tx.commit().await.context("commit")?;
+        Ok(())
+    }
+}
+
+/// The data the autopilot needs to settle a fast-path order out of competition.
+pub struct FastPathOrder {
+    /// The order in the raw API model form. Callers pass this to
+    /// `ProtocolFees::apply` and can then convert it to `domain::Order` via
+    /// `boundary::order::to_domain` once the resulting policies are known.
+    pub model_order: model::order::Order,
+    pub auction_id: database::auction::AuctionId,
+    pub solution_id: u64,
+    pub solution_uid: usize,
+    pub solver: eth::Address,
+    /// The `proposed_trade_executions` amounts as stored at quote time
+    /// (pre-fee-adjustment; identical to the quote's amounts since nothing
+    /// rewrites them between quote time and fast-path handling). Feed these
+    /// to `apply_volume_fees` alongside the Volume-type policies to obtain
+    /// the actual limit prices to settle at.
+    pub raw_sell: eth::U256,
+    pub raw_buy: eth::U256,
+    /// Native prices (token → normalized price) from the quote's auction.
+    pub native_prices: HashMap<eth::Address, eth::U256>,
 }
 
 #[derive(prometheus_metric_storage::MetricStorage)]

crates/autopilot/src/infra/solvers/dto/settle.rs

@@ -1,6 +1,10 @@
 use {
+    crate::infra::persistence::dto::order::Order,
+    alloy::primitives::{Address, U256},
+    number::serialization::HexOrDecimalU256,
     serde::Serialize,
     serde_with::{serde_as, skip_serializing_none},
+    std::collections::HashMap,
 };
 
 #[serde_as]
@@ -15,4 +19,30 @@ pub struct Request {
     /// Auction ID in which the specified solution ID is competing.
     #[serde_as(as = "serde_with::DisplayFromStr")]
     pub auction_id: i64,
+    /// Fast-path (out-of-competition) inputs. Present only when settling a
+    /// cached quote solution against the real signed order.
+    pub fast_path: Option<FastPath>,
+}
+
+#[serde_as]
+#[derive(Clone, Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct FastPath {
+    /// The real signed order the cached solution is re-encoded against.
+    pub order: Order,
+    /// The sell/buy amounts the order must fill at exactly.
+    pub limit_prices: LimitPrices,
+    /// Native prices (wei per 10**18) for the order's tokens.
+    #[serde_as(as = "HashMap<_, HexOrDecimalU256>")]
+    pub native_prices: HashMap<Address, U256>,
+}
+
+#[serde_as]
+#[derive(Clone, Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct LimitPrices {
+    #[serde_as(as = "HexOrDecimalU256")]
+    pub sell: U256,
+    #[serde_as(as = "HexOrDecimalU256")]
+    pub buy: U256,
 }

crates/autopilot/src/run.rs

@@ -491,6 +491,20 @@ pub async fn run(config: Configuration, shutdown_controller: ShutdownController)
         None => None,
     };
 
+    let protocol_fees = Arc::new(domain::ProtocolFees::new(
+        &config.fee_policies,
+        config
+            .shared
+            .volume_fee_bucket_overrides
+            .iter()
+            .map(Into::into)
+            .collect(),
+        config.shared.enable_sell_equals_buy_volume_fee,
+        *eth.contracts().weth().address(),
+    ));
+    let surplus_capturing_jit_order_owners =
+        Arc::new(config.surplus_capturing_jit_order_owners.clone());
+
     let solvable_orders_cache = SolvableOrdersCache::new(
         config.min_order_validity_period,
         persistence.clone(),
@@ -499,19 +513,9 @@ pub async fn run(config: Configuration, shutdown_controller: ShutdownController)
         deny_listed_tokens.clone(),
         competition_native_price_updater.clone(),
         *eth.contracts().weth().address(),
-        domain::ProtocolFees::new(
-            &config.fee_policies,
-            config
-                .shared
-                .volume_fee_bucket_overrides
-                .iter()
-                .map(Into::into)
-                .collect(),
-            config.shared.enable_sell_equals_buy_volume_fee,
-            *eth.contracts().weth().address(),
-        ),
+        protocol_fees.clone(),
         penalty_cap_calculator,
-        config.surplus_capturing_jit_order_owners,
+        surplus_capturing_jit_order_owners.clone(),
         config.native_price_timeout,
         *eth.contracts().settlement().address(),
         config.disable_order_balance_filter,
@@ -662,6 +666,8 @@ pub async fn run(config: Configuration, shutdown_controller: ShutdownController)
         },
         awaiter,
         new_orders_receiver,
+        protocol_fees,
+        surplus_capturing_jit_order_owners,
     );
     run.run_forever(shutdown_controller).await;
 

crates/autopilot/src/run_loop.rs

@@ -1,7 +1,11 @@
+pub mod fast_path;
 pub mod settle_call_coordinator;
 
 use {
-    self::settle_call_coordinator::{SettleCallCoordinator, SettleError},
+    self::{
+        fast_path::FastPathHandler,
+        settle_call_coordinator::{SettleCallCoordinator, SettleError},
+    },
     crate::{
         domain::{
             self,
@@ -170,8 +174,10 @@ pub struct RunLoop {
     /// Drivers that do NOT support delta auctions
     drivers: Vec<Arc<infra::Driver>>,
     /// Sends `/settle` calls to drivers and waits for the resulting
-    /// transaction to be mined.
+    /// transaction to be mined. Shared with the fast-path handler.
     settle_coordinator: Arc<SettleCallCoordinator>,
+    /// Handles fast-path orders on the side.
+    fast_path: Arc<FastPathHandler>,
 }
 
 impl RunLoop {
@@ -186,6 +192,8 @@ impl RunLoop {
         probes: Probes,
         maintenance: MaintenanceSync,
         new_orders_listener: mpsc::UnboundedReceiver<OrderUid>,
+        protocol_fees: Arc<crate::domain::ProtocolFees>,
+        surplus_capturing_jit_order_owners: Arc<Vec<alloy::primitives::Address>>,
     ) -> Arc<Self> {
         let max_winners = config.max_winners_per_auction.get();
         let weth = eth.contracts().wrapped_native_token();
@@ -200,6 +208,16 @@ impl RunLoop {
             config.max_settlement_transaction_wait,
         ));
 
+        let fast_path = FastPathHandler::new(
+            eth.clone(),
+            persistence.clone(),
+            drivers.clone(),
+            protocol_fees,
+            surplus_capturing_jit_order_owners,
+            settle_coordinator.clone(),
+            config.submission_deadline,
+        );
+
         let self_ = Arc::new(Self {
             delta_state: std::sync::Mutex::new(DeltaState::new(
                 config.auction_delta_checkpoint_interval,
@@ -215,17 +233,34 @@ impl RunLoop {
             wake_notify: wake_runloop,
             drivers,
             settle_coordinator,
+            fast_path,
         });
         Self::spawn_order_listener(self_.clone(), new_orders_listener);
         self_
     }
 
-    /// Spawns a background task that listens to the creation of new orders
-    /// and wakes the run loop for each incoming order.
+    /// Spawns a background task that listens to the creation of new orders and
+    /// notifies the runloop to kick it off if necessary and initiates the
+    /// fast path handling if an order needs it.
     fn spawn_order_listener(self: Arc<Self>, mut receiver: mpsc::UnboundedReceiver<OrderUid>) {
         tokio::spawn(async move {
-            while let Some(_order_uid) = receiver.next().await {
+            while let Some(order_uid) = receiver.next().await {
                 self.wake_notify.notify_one();
+                let persistence = self.persistence.clone();
+                let fast_path = self.fast_path.clone();
+                // immediately spawn separate task to never delay processing
+                // fast path orders
+                tokio::spawn(
+                    async move {
+                        match persistence.fast_path_order(order_uid).await {
+                            // not a fast path order -> do nothing
+                            Ok(None) => {}
+                            Err(err) => tracing::error!(?err, "failed to look up fast path order"),
+                            Ok(Some(order)) => fast_path.handle(order).await,
+                        };
+                    }
+                    .instrument(tracing::info_span!("fast_path", ?order_uid)),
+                );
             }
         });
     }
@@ -518,6 +553,7 @@ impl RunLoop {
                 solution_id,
                 submission_deadline_latest_block: block_deadline,
                 auction_id,
+                fast_path: None,
             };
 
             match self_