Return error on order replacement when it's not safe

crates/database/src/solver_competition.rs

@@ -52,9 +52,10 @@ GROUP BY sc.id
     sqlx::query_as(QUERY).bind(id).fetch_optional(ex).await
 }
 
-pub async fn load_latest_competition(
+pub async fn load_latest_competitions(
     ex: &mut PgConnection,
-) -> Result<Option<LoadCompetition>, sqlx::Error> {
+    latest_competitions_count: u32,
+) -> Result<Vec<LoadCompetition>, sqlx::Error> {
     const QUERY: &str = r#"
 SELECT sc.json, sc.id, COALESCE(ARRAY_AGG(s.tx_hash) FILTER (WHERE so.block_number IS NOT NULL), '{}') AS tx_hashes
 FROM solver_competitions sc
@@ -66,9 +67,20 @@ LEFT OUTER JOIN settlement_observations so
     AND s.log_index = so.log_index
 GROUP BY sc.id
 ORDER BY sc.id DESC
-LIMIT 1
+LIMIT $1
     ;"#;
-    sqlx::query_as(QUERY).fetch_optional(ex).await
+    sqlx::query_as(QUERY)
+        .bind(i64::from(latest_competitions_count))
+        .fetch_all(ex)
+        .await
+}
+
+pub async fn load_latest_competition(
+    ex: &mut PgConnection,
+) -> Result<Option<LoadCompetition>, sqlx::Error> {
+    let competitions = load_latest_competitions(ex, 1).await?;
+    let latest = competitions.into_iter().next();
+    Ok(latest)
 }
 
 pub async fn load_by_tx_hash(

crates/e2e/tests/e2e/replace_order.rs

@@ -5,9 +5,11 @@ use {
         order::{OrderCreation, OrderCreationAppData, OrderKind, OrderStatus},
         signature::EcdsaSigningScheme,
     },
+    orderbook::{api::IntoWarpReply, orderbook::OrderReplacementError},
     reqwest::StatusCode,
     secp256k1::SecretKey,
     shared::ethrpc::Web3,
+    warp::reply::Reply,
     web3::signing::SecretKeyRef,
 };
 
@@ -23,6 +25,139 @@ async fn local_node_try_replace_someone_else_order() {
     run_test(try_replace_someone_else_order_test).await;
 }
 
+// TODO: The test is not ideal, as we actually want to test the replacement of
+// active orders as soon as they are being bid on, even before they are
+// executed. For that we would need the ability to mock the driver in e2e tests.
+#[tokio::test]
+#[ignore]
+async fn local_node_try_replace_executed_order() {
+    run_test(try_replace_executed_order_test).await;
+}
+
+async fn try_replace_executed_order_test(web3: Web3) {
+    let mut onchain = OnchainComponents::deploy(web3.clone()).await;
+
+    let [solver] = onchain.make_solvers(to_wei(1)).await;
+    let [trader] = onchain.make_accounts(to_wei(1)).await;
+    let [token_a, token_b] = onchain
+        .deploy_tokens_with_weth_uni_v2_pools(to_wei(1_000), to_wei(1_000))
+        .await;
+
+    // Fund trader accounts
+    token_a.mint(trader.address(), to_wei(30)).await;
+
+    // Create and fund Uniswap pool
+    token_a.mint(solver.address(), to_wei(1000)).await;
+    token_b.mint(solver.address(), to_wei(1000)).await;
+    tx!(
+        solver.account(),
+        onchain
+            .contracts()
+            .uniswap_v2_factory
+            .create_pair(token_a.address(), token_b.address())
+    );
+    tx!(
+        solver.account(),
+        token_a.approve(
+            onchain.contracts().uniswap_v2_router.address(),
+            to_wei(1000)
+        )
+    );
+    tx!(
+        solver.account(),
+        token_b.approve(
+            onchain.contracts().uniswap_v2_router.address(),
+            to_wei(1000)
+        )
+    );
+    tx!(
+        solver.account(),
+        onchain.contracts().uniswap_v2_router.add_liquidity(
+            token_a.address(),
+            token_b.address(),
+            to_wei(1000),
+            to_wei(1000),
+            0_u64.into(),
+            0_u64.into(),
+            solver.address(),
+            U256::max_value(),
+        )
+    );
+
+    // Approve GPv2 for trading
+    tx!(
+        trader.account(),
+        token_a.approve(onchain.contracts().allowance, to_wei(15))
+    );
+
+    // Place Orders
+    let services = Services::new(&onchain).await;
+    services.start_protocol(solver).await;
+
+    let order = OrderCreation {
+        sell_token: token_a.address(),
+        sell_amount: to_wei(10),
+        buy_token: token_b.address(),
+        buy_amount: to_wei(5),
+        valid_to: model::time::now_in_epoch_seconds() + 300,
+        kind: OrderKind::Sell,
+        ..Default::default()
+    }
+    .sign(
+        EcdsaSigningScheme::Eip712,
+        &onchain.contracts().domain_separator,
+        SecretKeyRef::from(&SecretKey::from_slice(trader.private_key()).unwrap()),
+    );
+    let balance_before = token_a.balance_of(trader.address()).call().await.unwrap();
+    onchain.mint_block().await;
+    let order_id = services.create_order(&order).await.unwrap();
+
+    tracing::info!("Waiting for the old order to be executed");
+    wait_for_condition(TIMEOUT, || async {
+        let balance_after = token_a.balance_of(trader.address()).call().await.unwrap();
+        balance_before.saturating_sub(balance_after) == to_wei(10)
+    })
+    .await
+    .unwrap();
+
+    // Replace order
+    let new_order = OrderCreation {
+        sell_token: token_a.address(),
+        sell_amount: to_wei(3),
+        buy_token: token_b.address(),
+        buy_amount: to_wei(1),
+        valid_to: model::time::now_in_epoch_seconds() + 300,
+        kind: OrderKind::Sell,
+        partially_fillable: false,
+        app_data: OrderCreationAppData::Full {
+            full: format!(
+                r#"{{"version":"1.1.0","metadata":{{"replacedOrder":{{"uid":"{}"}}}}}}"#,
+                order_id
+            ),
+        },
+        ..Default::default()
+    }
+    .sign(
+        EcdsaSigningScheme::Eip712,
+        &onchain.contracts().domain_separator,
+        SecretKeyRef::from(&SecretKey::from_slice(trader.private_key()).unwrap()),
+    );
+    let response = services.create_order(&new_order).await;
+    let (error_code, error_message) = response.err().unwrap();
+
+    assert_eq!(error_code, StatusCode::BAD_REQUEST);
+
+    let expected_response = OrderReplacementError::OldOrderActivelyBidOn
+        .into_warp_reply()
+        .into_response()
+        .into_body();
+    let expected_body_bytes = warp::hyper::body::to_bytes(expected_response)
+        .await
+        .unwrap();
+    let expected_body = String::from_utf8(expected_body_bytes.to_vec()).unwrap();
+    assert_eq!(error_message, expected_body);
+}
+
 async fn try_replace_someone_else_order_test(web3: Web3) {
     let mut onchain = OnchainComponents::deploy(web3.clone()).await;
 

crates/orderbook/openapi.yml

@@ -1362,6 +1362,7 @@ components:
             - InvalidAppData
             - AppDataHashMismatch
             - AppdataFromMismatch
+            - OldOrderActivelyBidOn
         description:
           type: string
       required:

crates/orderbook/src/api/post_order.rs

@@ -1,7 +1,7 @@
 use {
     crate::{
         api::{ApiReply, IntoWarpReply, error, extract_payload},
-        orderbook::{AddOrderError, Orderbook},
+        orderbook::{AddOrderError, OrderReplacementError, Orderbook},
     },
     anyhow::Result,
     model::{
@@ -258,10 +258,7 @@ impl IntoWarpReply for AddOrderError {
                 super::error("InvalidAppData", err.to_string()),
                 StatusCode::BAD_REQUEST,
             ),
-            err @ AddOrderError::InvalidReplacement => reply::with_status(
-                super::error("InvalidReplacement", err.to_string()),
-                StatusCode::UNAUTHORIZED,
-            ),
+            AddOrderError::InvalidReplacement(err) => err.into_warp_reply(),
             AddOrderError::MetadataSerializationFailed(err) => reply::with_status(
                 super::error("MetadataSerializationFailed", err.to_string()),
                 StatusCode::INTERNAL_SERVER_ERROR,
@@ -270,6 +267,32 @@ impl IntoWarpReply for AddOrderError {
     }
 }
 
+impl IntoWarpReply for OrderReplacementError {
+    fn into_warp_reply(self) -> super::ApiReply {
+        match self {
+            OrderReplacementError::InvalidSignature => with_status(
+                super::error("InvalidSignature", "Malformed signature"),
+                StatusCode::BAD_REQUEST,
+            ),
+            OrderReplacementError::WrongOwner => with_status(
+                super::error("WrongOwner", "Old and new orders have different signers"),
+                StatusCode::UNAUTHORIZED,
+            ),
+            OrderReplacementError::OldOrderActivelyBidOn => with_status(
+                super::error(
+                    "OldOrderActivelyBidOn",
+                    "The old order is actively beign bid on in recent auctions",
+                ),
+                StatusCode::BAD_REQUEST,
+            ),
+            OrderReplacementError::Other(err) => {
+                tracing::error!(?err, "replace_order");
+                crate::api::internal_error_reply()
+            }
+        }
+    }
+}
+
 pub fn create_order_response(
     result: Result<(OrderUid, Option<QuoteId>), AddOrderError>,
 ) -> ApiReply {

crates/orderbook/src/arguments.rs

@@ -137,6 +137,11 @@ pub struct Arguments {
     /// The maximum gas amount a single order can use for getting settled.
     #[clap(long, env, default_value = "8000000")]
     pub max_gas_per_order: u64,
+
+    /// The number of past solver competitions to look back at to determine
+    /// whether an order is actively being bid on.
+    #[clap(long, env, default_value = "5")]
+    pub active_order_competition_threshold: u32,
 }
 
 impl std::fmt::Display for Arguments {
@@ -167,6 +172,7 @@ impl std::fmt::Display for Arguments {
             app_data_size_limit,
             db_url,
             max_gas_per_order,
+            active_order_competition_threshold,
         } = self;
 
         write!(f, "{}", shared)?;
@@ -227,6 +233,11 @@ impl std::fmt::Display for Arguments {
         )?;
         writeln!(f, "app_data_size_limit: {}", app_data_size_limit)?;
         writeln!(f, "max_gas_per_order: {}", max_gas_per_order)?;
+        writeln!(
+            f,
+            "active_order_competition_threshold: {}",
+            active_order_competition_threshold
+        )?;
 
         Ok(())
     }

crates/orderbook/src/database/solver_competition.rs

@@ -75,7 +75,7 @@ impl SolverCompetitionStoring for Postgres {
         let mut ex = self.pool.acquire().await.map_err(anyhow::Error::from)?;
         database::solver_competition::load_latest_competition(&mut ex)
             .await
-            .context("solver_competition::load_latest")?
+            .context("solver_competition::load_latest_competition")?
             .map(|row| {
                 deserialize_solver_competition(
                     row.json,
@@ -85,6 +85,36 @@ impl SolverCompetitionStoring for Postgres {
             })
             .ok_or(LoadSolverCompetitionError::NotFound)?
     }
+
+    async fn load_latest_competitions(
+        &self,
+        latest_competitions_count: u32,
+    ) -> Result<Vec<SolverCompetitionAPI>, LoadSolverCompetitionError> {
+        let _timer = super::Metrics::get()
+            .database_queries
+            .with_label_values(&["load_latest_competitions"])
+            .start_timer();
+
+        let mut ex = self.pool.acquire().await.map_err(anyhow::Error::from)?;
+
+        let latest_competitions = database::solver_competition::load_latest_competitions(
+            &mut ex,
+            latest_competitions_count,
+        )
+        .await
+        .context("solver_competition::load_latest_competitions")?
+        .into_iter()
+        .map(|row| {
+            deserialize_solver_competition(
+                row.json,
+                row.id,
+                row.tx_hashes.iter().map(|hash| H256(hash.0)).collect(),
+            )
+        })
+        .collect::<Result<Vec<_>, _>>()?;
+
+        Ok(latest_competitions)
+    }
 }
 
 #[cfg(test)]