Persist streamed quotes (#4579)

# Description
Streamed quotes currently don't carry a quote ID making it impossible to
reuse them during order placement. This is problematic for a couple of
reasons
1. It increases the amount of quote requests we are sending to solvers
by a non trivial amount.
2. Re-quoting when the order is placed may mess with intended slippage,
market order classification and other things that are needed for a
reliable user and integrator experience.
3. Quotes without an ID can never receive a quote reward. If solvers
find a way to segregate the flow, they would be better off selectively
not responding to streamed quote requests, which are expected to become
the standard way for users to make the decision whether they want to
trade (therefore they need to be extra competitive)

# Changes
* Persist every quote candidate before it gets sent out in the stream
* Update the documentation and test cases around this.

## How to test
Existing tests

## Alternatives considered

https://app.notion.com/p/cownation/RFC-Quote-Persistence-for-Streaming-Quotes-38f8da5f04ca81a98f0fc016a3a37e81
contains two alternative AI generated options:

1. Mint the id at completion, send it as a final event.
2. Reserve one id up front, store the best under it (recommended).

The first option leads to having no quote id unless we wait for the full
timeout. This defeats the purpose of being able to quickly place orders
as soon as we have a quote that is "good enough". Especially with the
high deadlines suggested in cowprotocol/docs#638
I think we would not end up getting the final event in a good amount of
cases.

In the second option, if the client terminates the connection
prematurely there may be subjectivity around which quotes have been
received (and therefore what the client saw as "best").

The main downside of the approach in this PR is that it adds a small
amount of latency on the critical path (for the storage write), which in
the grand scheme of current quote latency is likely irrelevant. It will
also create more pressure on the DB, however since unused quotes are
very short lived I don't expect this to be a concern in practice. It
strikes me as the simplest approach therefore.

---------

Co-authored-by: Martin Magnus <martin.beckmann@protonmail.com>

Author: fleupold

crates/database/src/quotes.rs

@@ -123,7 +123,12 @@ 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, mirroring the price-estimation
+-- competition: prefer verified quotes over unverified ones, then take the
+-- highest buy/sell exchange rate net of the sell-token-denominated fee.
+ORDER BY
+    verified DESC,
+    buy_amount / (sell_amount + gas_amount * gas_price / sell_token_price) DESC
 LIMIT 1
     "#;
     sqlx::query_as(QUERY)
@@ -395,6 +400,191 @@ 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: 0.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*1/0.1) = 210/2000 ≈ 0.105
+        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*1/0.1) = 200/1010 ≈ 0.198
+        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: 0.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 3000.
+        // net rate = 100 / (1000 + 200*1/0.1) = 100/3000 ≈ 0.033
+        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 1110.
+        // net rate = 100 / (1100 + 1*1/0.1) = 100/1110 ≈ 0.090
+        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_find_quote_prefers_verified_over_better_rate() {
+        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();
+        // Two candidates for the same sell order, no fee; they differ only in
+        // buy amount (rate) and verification status.
+        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(),
+        };
+
+        // Unverified but strictly better rate (more buy for the same sell).
+        let mut unverified_better = Quote {
+            buy_amount: 210.into(),
+            verified: false,
+            solver: ByteArray([1; 20]),
+            ..base.clone()
+        };
+        unverified_better.id = save(&mut db, &unverified_better).await.unwrap();
+
+        // Verified with a worse rate -> should still win
+        let mut verified_worse = Quote {
+            buy_amount: 200.into(),
+            verified: true,
+            solver: ByteArray([2; 20]),
+            ..base.clone()
+        };
+        verified_worse.id = save(&mut db, &verified_worse).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(),
+            verified_worse
+        );
+    }
+
     #[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 (as opposed to
+    /// just the final quote in 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));
     }