Store quote competition data

crates/autopilot/src/database/auction.rs

@@ -9,8 +9,9 @@ use {
     num::ToPrimitive,
     shared::{
         db_order_conversions::full_order_into_model_order,
-        event_storing_helpers::{create_db_search_parameters, create_quote_row},
+        event_storing_helpers::create_db_search_parameters,
         order_quoting::{QuoteCompetition, QuoteData, QuoteSearchParameters, QuoteStoring},
+        quote_storage::save_quote_competition,
     },
     std::{collections::HashMap, ops::DerefMut, sync::Arc},
 };
@@ -23,9 +24,9 @@ impl QuoteStoring for Postgres {
             .with_label_values(&["save_quote"])
             .start_timer();
 
-        let mut ex = self.pool.acquire().await?;
-        let row = create_quote_row(&data)?;
-        let id = database::quotes::save(&mut ex, &row).await?;
+        let mut tx = self.pool.begin().await?;
+        let id = save_quote_competition(&mut tx, data, &self.domain_separator).await?;
+        tx.commit().await?;
         Ok(id)
     }
 

crates/autopilot/src/database/mod.rs

@@ -1,4 +1,5 @@
 use {
+    model::DomainSeparator,
     num::ToPrimitive,
     shared::arguments::DB_MAX_CONNECTIONS_DEFAULT,
     sqlx::{Executor, PgConnection, PgPool, postgres::PgPoolOptions},
@@ -40,18 +41,30 @@ impl Default for Config {
 pub struct Postgres {
     pub pool: PgPool,
     pub config: Config,
+    /// Used to recover JIT order owners and compute their `order_uid`
+    /// before persisting fast-path quote competitions. Defaults to zero;
+    /// production callers must set it via `with_domain_separator`.
+    pub domain_separator: DomainSeparator,
 }
 
 impl Postgres {
-    pub async fn new(url: &str, config: Config) -> sqlx::Result<Self> {
+    pub async fn new(
+        url: &str,
+        config: Config,
+        domain_separator: DomainSeparator,
+    ) -> sqlx::Result<Self> {
         let pool = PgPoolOptions::new()
             .max_connections(config.max_pool_size.get())
             .connect(url)
             .await?;
 
         Self::start_db_metrics_job(pool.clone());
 
-        Ok(Self { pool, config })
+        Ok(Self {
+            pool,
+            config,
+            domain_separator,
+        })
     }
 
     fn start_db_metrics_job(pool: PgPool) {
@@ -72,8 +85,9 @@ impl Postgres {
         });
     }
 
+    #[cfg(test)]
     pub async fn with_defaults() -> sqlx::Result<Self> {
-        Self::new("postgresql://", Default::default()).await
+        Self::new("postgresql://", Default::default(), Default::default()).await
     }
 
     #[instrument(skip_all)]

crates/autopilot/src/database/onchain_order_events/mod.rs

@@ -218,6 +218,7 @@ impl<T: Send + Sync + Clone, W: Send + Sync> OnchainOrderParser<T, W> {
         order_placement_events: Vec<(ContractEvent, Log)>,
     ) -> Result<(
         Vec<W>,
+        Vec<i64>,
         Vec<Option<database::orders::Quote>>,
         Vec<(database::events::EventIndex, OnchainOrderPlacement)>,
         Vec<Order>,
@@ -265,7 +266,7 @@ impl<T: Send + Sync + Clone, W: Send + Sync> OnchainOrderParser<T, W> {
         .await;
 
         let data_tuple = onchain_order_data.into_iter().map(
-            |(event_index, quote, onchain_order_placement, order, tx_hash)| {
+            |(event_index, quote_id, quote, onchain_order_placement, order, tx_hash)| {
                 (
                     self.custom_onchain_data_parser
                         .customized_event_data_for_event_index(
@@ -274,6 +275,7 @@ impl<T: Send + Sync + Clone, W: Send + Sync> OnchainOrderParser<T, W> {
                             &custom_data_hashmap,
                             &onchain_order_placement,
                         ),
+                    quote_id,
                     quote,
                     (event_index, onchain_order_placement),
                     order,
@@ -318,9 +320,9 @@ impl<T: Send + Sync + Clone, W: Send + Sync> OnchainOrderParser<T, W> {
             .collect();
         let invalidation_events = get_invalidation_events(events)?;
         let invalided_order_uids = extract_invalidated_order_uids(invalidation_events)?;
-        let (custom_onchain_data, quotes, broadcasted_order_data, mut orders, tx_hashes) = self
-            .extract_custom_and_general_order_data(order_placement_events)
-            .await?;
+        let (custom_onchain_data, quote_ids, quotes, broadcasted_order_data, mut orders, tx_hashes) =
+            self.extract_custom_and_general_order_data(order_placement_events)
+                .await?;
 
         database::onchain_invalidations::insert_onchain_invalidations(
             transaction,
@@ -364,6 +366,33 @@ impl<T: Send + Sync + Clone, W: Send + Sync> OnchainOrderParser<T, W> {
             .await
             .context("insert_orders failed")?;
 
+        // Promote fast-path quotes for onchain orders (mirrors the trait-based
+        // path in the orderbook). For each successfully-quoted order tied to a
+        // fast-path `auction_id`, drop the transient `quotes` row and rewrite
+        // the placeholder `proposed_trade_executions.order_uid` to the real
+        // one.
+        for (quote_id, quote, order) in izip!(&quote_ids, &quotes, &orders) {
+            let Some(quote) = quote else {
+                continue;
+            };
+            let Some(auction_id) = quote.auction_id else {
+                continue;
+            };
+            // The order_quotes row already carries the auction_id (populated
+            // inline above by `insert_quotes`), so all that's left is to drop
+            // the transient `quotes` row and patch competition tables.
+            database::quotes::delete_and_return_row(transaction, *quote_id)
+                .await
+                .context("failed to delete promoted onchain quote")?;
+            database::solver_competition_v2::finalize_quote_competition(
+                transaction,
+                auction_id,
+                order.uid,
+            )
+            .await
+            .context("failed to patch competition rows for onchain order")?;
+        }
+
         for order in &invalided_order_uids {
             tracing::debug!(?order, "invalidated order");
         }
@@ -442,6 +471,7 @@ fn extract_invalidated_order_uids(
 
 type GeneralOnchainOrderPlacementData = (
     EventIndex,
+    i64,
     Option<database::orders::Quote>,
     OnchainOrderPlacement,
     Order,
@@ -513,7 +543,14 @@ where
                     None
                 }
             };
-            Ok((event_index, quote, order_data.0, order_data.1, tx_hash))
+            Ok((
+                event_index,
+                quote_id,
+                quote,
+                order_data.0,
+                order_data.1,
+                tx_hash,
+            ))
         },
     );
     let onchain_order_placement_data: Vec<Result<GeneralOnchainOrderPlacementData>> =
@@ -1269,6 +1306,7 @@ mod test {
             db: Postgres {
                 pool: PgPool::connect_lazy("postgresql://").unwrap(),
                 config: Default::default(),
+                domain_separator: Default::default(),
             },
             trampoline: HooksTrampoline::Instance::deployed(&web3.provider)
                 .await
@@ -1318,9 +1356,10 @@ mod test {
             metadata: quote.data.metadata.clone().try_into().unwrap(),
             auction_id: quote.data.auction_id,
         };
-        assert_eq!(result.1, vec![Some(expected_quote)]);
+        assert_eq!(result.1, vec![0i64]);
+        assert_eq!(result.2, vec![Some(expected_quote)]);
         assert_eq!(
-            result.2,
+            result.3,
             vec![(
                 expected_event_index,
                 OnchainOrderPlacement {

crates/autopilot/src/run.rs

@@ -189,19 +189,6 @@ pub async fn start(args: impl Iterator<Item = String>) {
 /// Assumes tracing and metrics registry have already been set up.
 pub async fn run(config: Configuration, shutdown_controller: ShutdownController) {
     assert!(config.shadow.is_none(), "cannot run in shadow mode");
-    let db_write = Postgres::new(
-        config.database.write_url.as_str(),
-        crate::database::Config {
-            insert_batch_size: config.database.insert_batch_size,
-            max_pool_size: config.database.max_connections,
-        },
-    )
-    .await
-    .unwrap();
-
-    // If the DB is in read-only mode, running ANALYZE is not possible and will
-    // trigger and error https://www.postgresql.org/docs/current/hot-standby.html
-    crate::database::run_database_metrics_work(db_write.clone());
 
     let http_factory = HttpClientFactory::from(config.http_client);
     let ethrpc_args = shared::web3::Arguments::from(&config.shared.ethrpc);
@@ -257,6 +244,22 @@ pub async fn run(config: Configuration, shutdown_controller: ShutdownController)
         .await
         .expect("Couldn't get vault relayer address");
 
+    let domain_separator = DomainSeparator::new(chain_id, *eth.contracts().settlement().address());
+    let db_write = Postgres::new(
+        config.database.write_url.as_str(),
+        crate::database::Config {
+            insert_batch_size: config.database.insert_batch_size,
+            max_pool_size: config.database.max_connections,
+        },
+        domain_separator,
+    )
+    .await
+    .unwrap();
+
+    // If the DB is in read-only mode, running ANALYZE is not possible and will
+    // trigger and error https://www.postgresql.org/docs/current/hot-standby.html
+    crate::database::run_database_metrics_work(db_write.clone());
+
     let chain = Chain::try_from(chain_id).expect("incorrect chain ID");
 
     let balance_overrider = config.price_estimation.balance_overrides.init(web3.clone());
@@ -575,7 +578,7 @@ pub async fn run(config: Configuration, shutdown_controller: ShutdownController)
             web3.clone(),
             quoter.clone(),
             Box::new(custom_ethflow_order_parser),
-            DomainSeparator::new(chain_id, *eth.contracts().settlement().address()),
+            domain_separator,
             *eth.contracts().settlement().address(),
             eth.contracts().trampoline().clone(),
         );

crates/database/src/quotes.rs

@@ -94,6 +94,19 @@ WHERE id = $1
     sqlx::query_as(QUERY).bind(id).fetch_optional(ex).await
 }
 
+/// Deletes the row from the transient `quotes` table and returns it.
+/// Used when a quote is promoted to an `order_quotes` row at order-placement
+/// time — the caller reuses the returned row's fields to build the
+/// `order_quotes` insert.
+#[instrument(skip_all)]
+pub async fn delete_and_return_row(
+    ex: &mut PgConnection,
+    id: QuoteId,
+) -> Result<Option<Quote>, sqlx::Error> {
+    const QUERY: &str = "DELETE FROM quotes WHERE id = $1 RETURNING *";
+    sqlx::query_as(QUERY).bind(id).fetch_optional(ex).await
+}
+
 /// Fields for searching stored quotes.
 #[derive(Clone)]
 pub struct QuoteSearchParameters {

crates/database/src/solver_competition_v2.rs

@@ -355,6 +355,122 @@ async fn save_jit_orders(
     Ok(())
 }
 
+/// Because the final order uid is not known when we store the quote
+/// competition data we use `0x000...000` as a sentinel value.
+/// When an order gets placed referencing a quote competition this function
+/// replaces the placeholder value with the now final order uid.
+#[instrument(skip_all)]
+pub async fn finalize_quote_competition(
+    ex: &mut PgTransaction<'_>,
+    auction_id: AuctionId,
+    order_uid: OrderUid,
+) -> Result<(), sqlx::Error> {
+    const QUERY: &str = r#"
+WITH patch_te AS (
+    UPDATE proposed_trade_executions
+    SET order_uid = $1
+    WHERE auction_id = $2 AND order_uid = $3
+)
+UPDATE competition_auctions
+SET order_uids = ARRAY[$1]
+WHERE id = $2
+"#;
+    sqlx::query(QUERY)
+        .bind(order_uid)
+        .bind(auction_id)
+        .bind(OrderUid::default())
+        .execute(ex.deref_mut())
+        .await?;
+    Ok(())
+}
+
+/// Deletes all competition rows associated with `auction_id` across
+/// `proposed_trade_executions`, `proposed_jit_orders`, `proposed_solutions`,
+/// and `competition_auctions`.
+#[instrument(skip_all)]
+pub async fn delete_by_auction_id(
+    ex: &mut PgTransaction<'_>,
+    auction_id: AuctionId,
+) -> Result<(), sqlx::Error> {
+    const QUERY: &str = r#"
+WITH
+    del_te AS (DELETE FROM proposed_trade_executions WHERE auction_id = $1),
+    del_jo AS (DELETE FROM proposed_jit_orders       WHERE auction_id = $1),
+    del_ps AS (DELETE FROM proposed_solutions        WHERE auction_id = $1)
+DELETE FROM competition_auctions WHERE id = $1
+"#;
+    sqlx::query(QUERY)
+        .bind(auction_id)
+        .execute(ex.deref_mut())
+        .await?;
+    Ok(())
+}
+
+/// A JIT order proposed by a solver as part of a quote response. Distinct
+/// from the autopilot flow, where a JIT order is identified at runtime by
+/// its `order_uid` not appearing in the `orders` table. In the quoting
+/// flow the placeholder user trade also has no matching `orders` row, so
+/// we track JIT-ness explicitly instead.
+#[derive(Clone, Debug, PartialEq)]
+pub struct QuoteJitOrder {
+    pub solution_uid: i64,
+    pub order_uid: OrderUid,
+    pub sell_token: Address,
+    pub buy_token: Address,
+    pub limit_sell: BigDecimal,
+    pub limit_buy: BigDecimal,
+    pub side: OrderKind,
+}
+
+/// Persists competition data derived from a fast-path quote response.
+/// `solutions[i].orders` should carry every trade (placeholder user trade
+/// plus each JIT trade) — these are written to `proposed_trade_executions`.
+/// `jit_orders` carries the corresponding entries for `proposed_jit_orders`
+/// and is inserted unconditionally.
+#[instrument(skip_all)]
+pub async fn save_from_quote(
+    ex: &mut PgTransaction<'_>,
+    auction_id: AuctionId,
+    solutions: &[Solution],
+    jit_orders: &[QuoteJitOrder],
+) -> Result<(), sqlx::Error> {
+    if solutions.is_empty() {
+        return Ok(());
+    }
+    save_solutions(ex, auction_id, solutions).await?;
+    save_trade_executions(ex, auction_id, solutions).await?;
+    save_jit_orders_unchecked(ex, auction_id, jit_orders).await?;
+    Ok(())
+}
+
+#[instrument(skip_all)]
+async fn save_jit_orders_unchecked(
+    ex: &mut PgTransaction<'_>,
+    auction_id: AuctionId,
+    jits: &[QuoteJitOrder],
+) -> Result<(), sqlx::Error> {
+    if jits.is_empty() {
+        return Ok(());
+    }
+    let mut builder = QueryBuilder::new(
+        r#"INSERT INTO proposed_jit_orders
+        (auction_id, solution_uid, order_uid, sell_token, buy_token, limit_sell, limit_buy, side)"#,
+    );
+    builder.push_values(jits.iter(), |mut b, j| {
+        b.push_bind(auction_id)
+            .push_bind(j.solution_uid)
+            .push_bind(j.order_uid)
+            .push_bind(j.sell_token)
+            .push_bind(j.buy_token)
+            .push_bind(&j.limit_sell)
+            .push_bind(&j.limit_buy)
+            .push_bind(j.side);
+    });
+    builder.push(" ON CONFLICT (auction_id, solution_uid, order_uid) DO NOTHING;");
+    builder.build().execute(ex.deref_mut()).await?;
+    Ok(())
+}
+
 #[derive(sqlx::FromRow)]
 struct SolutionRow {
     uid: i64,

crates/e2e/src/setup/colocation.rs

@@ -190,6 +190,7 @@ relative-slippage = "0.1"
 account = "{account}"
 merge-solutions = {merge_solutions}
 quote-using-limit-orders = {quote_using_limit_orders}
+fast-path-enabled = true
 enable-simulation-bad-token-detection = true
 enable-metrics-bad-order-detection = true
 http-time-buffer = "100ms"