Persist streamed quotes

crates/database/src/quotes.rs

@@ -123,7 +123,9 @@ WHERE
     order_kind = $6 AND
     expiration_timestamp >= $7 AND
     quote_kind = $8
-ORDER BY gas_amount * gas_price * sell_token_price ASC
+-- Return the best quote for the user: the highest buy/sell exchange rate net of
+-- the (sell-token-denominated) fee.
+ORDER BY buy_amount / (sell_amount + gas_amount * gas_price * sell_token_price) DESC
 LIMIT 1
     "#;
     sqlx::query_as(QUERY)
@@ -395,6 +397,130 @@ mod tests {
         assert_eq!(find(&mut db, &search_b).await.unwrap(), None);
     }
 
+    #[tokio::test]
+    #[ignore]
+    async fn postgres_find_quote_picks_best_net_of_fee_rate_sell_order() {
+        let mut db = PgConnection::connect("postgresql://").await.unwrap();
+        let mut db = db.begin().await.unwrap();
+        crate::clear_DANGER_(&mut db).await.unwrap();
+
+        let now = low_precision_now();
+        // All candidates quote the same sell order (same sell amount); they only
+        // differ in buy amount and fee.
+        let base = Quote {
+            id: Default::default(),
+            sell_token: ByteArray([1; 20]),
+            buy_token: ByteArray([2; 20]),
+            sell_amount: 1000.into(),
+            buy_amount: Default::default(),
+            gas_amount: 0.,
+            gas_price: 1.,
+            sell_token_price: 1.,
+            order_kind: OrderKind::Sell,
+            expiration_timestamp: now,
+            quote_kind: QuoteKind::Standard,
+            solver: ByteArray([1; 20]),
+            verified: false,
+            metadata: Default::default(),
+        };
+
+        // Highest absolute buy amount, but an expensive fee.
+        // net rate = 210 / (1000 + 100) ≈ 0.1909
+        let mut high_buy_high_fee = Quote {
+            buy_amount: 210.into(),
+            gas_amount: 100.,
+            solver: ByteArray([1; 20]),
+            ..base.clone()
+        };
+        high_buy_high_fee.id = save(&mut db, &high_buy_high_fee).await.unwrap();
+
+        // Lower absolute buy amount, but a negligible fee -> best net-of-fee rate.
+        // net rate = 200 / (1000 + 1) ≈ 0.1998
+        let mut best_rate = Quote {
+            buy_amount: 200.into(),
+            gas_amount: 1.,
+            solver: ByteArray([2; 20]),
+            ..base.clone()
+        };
+        best_rate.id = save(&mut db, &best_rate).await.unwrap();
+
+        let search = QuoteSearchParameters {
+            sell_token: base.sell_token,
+            buy_token: base.buy_token,
+            sell_amount_0: base.sell_amount.clone(),
+            sell_amount_1: base.sell_amount.clone(),
+            buy_amount: Default::default(),
+            kind: OrderKind::Sell,
+            expiration: now,
+            quote_kind: QuoteKind::Standard,
+        };
+        assert_eq!(find(&mut db, &search).await.unwrap().unwrap(), best_rate);
+    }
+
+    #[tokio::test]
+    #[ignore]
+    async fn postgres_find_quote_picks_best_net_of_fee_rate_buy_order() {
+        let mut db = PgConnection::connect("postgresql://").await.unwrap();
+        let mut db = db.begin().await.unwrap();
+        crate::clear_DANGER_(&mut db).await.unwrap();
+
+        let now = low_precision_now();
+        // All candidates quote the same buy order (same buy amount); they only
+        // differ in sell amount and fee.
+        let base = Quote {
+            id: Default::default(),
+            sell_token: ByteArray([1; 20]),
+            buy_token: ByteArray([2; 20]),
+            sell_amount: Default::default(),
+            buy_amount: 100.into(),
+            gas_amount: 0.,
+            gas_price: 1.,
+            sell_token_price: 1.,
+            order_kind: OrderKind::Buy,
+            expiration_timestamp: now,
+            quote_kind: QuoteKind::Standard,
+            solver: ByteArray([1; 20]),
+            verified: false,
+            metadata: Default::default(),
+        };
+
+        // Lowest absolute sell amount, but an expensive fee -> total spend 1200.
+        // net rate = 100 / (1000 + 200) ≈ 0.0833
+        let mut low_sell_high_fee = Quote {
+            sell_amount: 1000.into(),
+            gas_amount: 200.,
+            solver: ByteArray([1; 20]),
+            ..base.clone()
+        };
+        low_sell_high_fee.id = save(&mut db, &low_sell_high_fee).await.unwrap();
+
+        // Higher absolute sell amount, but a negligible fee -> total spend 1101.
+        // net rate = 100 / (1100 + 1) ≈ 0.0908 -> best (least total spend)
+        let mut high_sell_low_fee = Quote {
+            sell_amount: 1100.into(),
+            gas_amount: 1.,
+            solver: ByteArray([2; 20]),
+            ..base.clone()
+        };
+        high_sell_low_fee.id = save(&mut db, &high_sell_low_fee).await.unwrap();
+
+        let search = QuoteSearchParameters {
+            sell_token: base.sell_token,
+            buy_token: base.buy_token,
+            sell_amount_0: Default::default(),
+            sell_amount_1: Default::default(),
+            buy_amount: base.buy_amount.clone(),
+            kind: OrderKind::Buy,
+            expiration: now,
+            quote_kind: QuoteKind::Standard,
+        };
+
+        assert_eq!(
+            find(&mut db, &search).await.unwrap().unwrap(),
+            high_sell_low_fee
+        );
+    }
+
     #[tokio::test]
     #[ignore]
     async fn postgres_save_and_find_quote_and_differentiates_by_signing_scheme() {

crates/e2e/tests/e2e/quoting.rs

@@ -26,6 +26,7 @@ use {
     shared::web3::Web3,
     solvers_dto::solution::{SolverError, SolverErrorCode, SolverResponse},
     std::{
+        ops::DerefMut,
         sync::Arc,
         time::{Duration, Instant},
     },
@@ -951,17 +952,37 @@ async fn volume_fee(web3: Web3) {
 
 // Smoke test for the SSE streaming quote endpoint. Posts a quote request to
 // /api/v1/quote/stream and asserts that at least one SSE data line parses as
-// a valid OrderQuoteResponse with id == None.
+// a valid OrderQuoteResponse carrying a persisted quote id.
 async fn quote_stream_smoke(web3: Web3) {
     tracing::info!("Setting up chain state.");
     let mut onchain = OnchainComponents::deploy(web3).await;
 
     let [solver] = onchain.make_solvers(10u64.eth()).await;
-    let [trader] = onchain.make_accounts(2u64.eth()).await;
+    let [trader] = onchain.make_accounts(10u64.eth()).await;
     let [token] = onchain
         .deploy_tokens_with_weth_uni_v2_pools(1_000u64.eth(), 1_000u64.eth())
         .await;
 
+    // Wrap and approve WETH so the trader can later place an order referencing
+    // one of the streamed quotes.
+    onchain
+        .contracts()
+        .weth
+        .approve(onchain.contracts().allowance, 2u64.eth())
+        .from(trader.address())
+        .send_and_watch()
+        .await
+        .unwrap();
+    onchain
+        .contracts()
+        .weth
+        .deposit()
+        .from(trader.address())
+        .value(2u64.eth())
+        .send_and_watch()
+        .await
+        .unwrap();
+
     tracing::info!("Starting services.");
     let services = Services::new(&onchain).await;
     services.start_protocol(solver).await;
@@ -1026,8 +1047,8 @@ async fn quote_stream_smoke(web3: Web3) {
     let buy_token = *token.address();
     for response in &parsed {
         assert!(
-            response.id.is_none(),
-            "streaming quotes should have id == null, got {:?}",
+            response.id.is_some(),
+            "streaming quotes should be persisted and carry an id, got {:?}",
             response.id
         );
         assert_eq!(response.from, trader.address());
@@ -1038,4 +1059,46 @@ async fn quote_stream_smoke(web3: Web3) {
             "streamed quote should have a non-zero buy amount, got {response:?}"
         );
     }
+
+    // Prove that placing an order with a streamed quote id reuses the persisted
+    // quote instead of re-quoting: the quote attached to the order must be the
+    // exact one we stored while streaming (same metadata).
+    let streamed = parsed
+        .iter()
+        .find(|response| response.id.is_some())
+        .expect("at least one streamed quote should carry an id");
+    let quote_id = streamed.id.unwrap();
+    let (streamed_metadata,) = crate::database::quote_metadata(services.db(), quote_id)
+        .await
+        .expect("streamed quote should be persisted");
+
+    tracing::info!("Placing order referencing a streamed quote id.");
+    let order = OrderCreation {
+        quote_id: Some(quote_id),
+        sell_token: weth,
+        sell_amount: 1u64.eth(),
+        buy_token,
+        buy_amount: streamed.quote.buy_amount,
+        valid_to: model::time::now_in_epoch_seconds() + 300,
+        kind: OrderKind::Sell,
+        ..Default::default()
+    }
+    .sign(
+        EcdsaSigningScheme::Eip712,
+        &onchain.contracts().domain_separator,
+        &trader.signer,
+    );
+    let order_uid = services.create_order(&order).await.unwrap();
+
+    let order_quote = database::orders::read_quote(
+        services.db().acquire().await.unwrap().deref_mut(),
+        &database::byte_array::ByteArray(order_uid.0),
+    )
+    .await
+    .unwrap()
+    .expect("order should reference a stored quote");
+    assert_eq!(
+        streamed_metadata, order_quote.metadata,
+        "order should reuse the streamed quote, not trigger a re-quote"
+    );
 }

crates/orderbook/openapi.yml

@@ -538,10 +538,9 @@ paths:
         endpoint opens a Server-Sent Events stream and emits one event per quote
         as solvers respond. Solvers without a usable quote emit no event, so you
         may receive fewer events than there are solvers. The stream closes when
-        the quote timeout elapses or all solvers have responded. Each event's
-        `id` is always `null`. Clients can use this to show progressive quote
-        updates in real time.
-
+        the quote timeout elapses or all solvers have responded. Each emitted
+        quote carries an `id` that can be referenced when placing an order.
+        Clients can use this to show progressive quote updates in real time.
 
         The `priceQuality` field of the request body is ignored: this endpoint
         always queries all solvers and attempts verification, emitting each

crates/orderbook/src/quoter.rs

@@ -189,7 +189,7 @@ impl QuoteHandler {
                 )
                 .map_err(|err| OrderQuoteError::CalculateQuote(err.into()))
                 .and_then(|adjusted| {
-                    build_order_quote_response(&request, &quote, &adjusted, None, valid_to)
+                    build_order_quote_response(&request, &quote, &adjusted, quote.id, valid_to)
                 });
                 match response {
                     Ok(response) => {

crates/shared/src/order_quoting.rs

@@ -608,7 +608,9 @@ impl OrderQuoting for OrderQuoter {
 #[async_trait::async_trait]
 pub trait StreamingQuoting: Send + Sync {
     /// Fetches gas and native prices once, then yields one `Quote` per solver
-    /// result as it arrives. Stores nothing.
+    /// result as it arrives. Each yielded quote is persisted (just like the
+    /// one-shot quote endpoint) so it can be referenced by id during a later
+    /// order placement.
     async fn calculate_quote_stream(
         &self,
         parameters: QuoteParameters,
@@ -664,6 +666,7 @@ impl StreamingQuoting for OrderQuoter {
         };
 
         let additional_cost = parameters.additional_cost();
+        let storage = self.storage.clone();
 
         let stream = async_stream::stream! {
             let inner = estimator.estimate_stream(trade_query);
@@ -702,6 +705,14 @@ impl StreamingQuoting for OrderQuoter {
                     }
                     quote = quote.with_scaled_sell_amount(sell_amount);
                 }
+                // Persist the quote so it can be referenced by id when the
+                // order is later placed, mirroring the one-shot endpoint. A
+                // failed write must not turn an otherwise good quote into an
+                // error event.
+                match storage.save(quote.data.clone()).await {
+                    Ok(id) => quote.id = Some(id),
+                    Err(err) => tracing::error!(?err, "failed to persist streamed quote"),
+                }
                 yield Ok(quote);
             }
         };
@@ -1999,11 +2010,18 @@ mod tests {
         gas_price: alloy::eips::eip1559::Eip1559Estimation,
         now: chrono::DateTime<Utc>,
     ) -> OrderQuoter {
+        let next_id = std::sync::atomic::AtomicI64::new(1);
+        let mut storage = MockQuoteStoring::new();
+        storage
+            .expect_save()
+            .times(0..)
+            .returning(move |_| Ok(next_id.fetch_add(1, std::sync::atomic::Ordering::SeqCst)));
+
         OrderQuoter {
             price_estimator: Arc::new(MockPriceEstimating::new()),
             native_price_estimator: Arc::new(native_price_estimator),
             gas_estimator: Arc::new(FakeGasPriceEstimator::new(gas_price)),
-            storage: Arc::new(MockQuoteStoring::new()),
+            storage: Arc::new(storage),
             now: Arc::new(now),
             validity: Validity::default(),
             default_quote_timeout: HEALTHY_PRICE_ESTIMATION_TIME,
@@ -2091,8 +2109,8 @@ mod tests {
         let q2 = stream.next().await.expect("second quote").expect("ok");
         assert!(stream.next().await.is_none());
 
-        assert_eq!(q1.id, None);
-        assert_eq!(q2.id, None);
+        assert_eq!(q1.id, Some(1));
+        assert_eq!(q2.id, Some(2));
         assert_eq!(q1.data.quoted_buy_amount, AlloyU256::from(500));
         assert_eq!(q2.data.quoted_buy_amount, AlloyU256::from(600));
     }