Driver settle queue tests (#3133)

# Description
An attempt to properly test #3129. The idea is to send multiple
`/settle` requests to the same driver to check that some will be
discarded due to the settlement queue limit.

# Changes

- [ ] `anvil` with disabled auto-mining is used to accumulate `/settle`
requests in the queue.
- [ ] Introduced a new `TooManyPendingSettlements` API error to properly
identify the cause. I couldn't find a better way to ensure the logic is
correct. Also, it might be helpful for all the parties to understand why
a specific solution/settlement was discarded.
- [ ] `default_settle_queue_size` is reduced to `2`, since 1 ongoing
settlement(dequeued) + 2 more in the queue make it `3` pending in total,
which is high enough IMO.
- [ ] Removed lifetime params from the `Settle` struct, which stores the
`/settle` endpoint response data in order to fix a compilation issue.
More details:
#3133 (comment)

## How to test
This is the test.

crates/driver/src/domain/competition/mod.rs

@@ -362,7 +362,7 @@ impl Competition {
 
         self.settle_queue.try_send(request).map_err(|err| {
             tracing::warn!(?err, "Failed to enqueue /settle request");
-            Error::SubmissionError
+            Error::TooManyPendingSettlements
         })?;
 
         response_receiver.await.map_err(|err| {
@@ -374,7 +374,7 @@ impl Competition {
     pub fn ensure_settle_queue_capacity(&self) -> Result<(), Error> {
         if self.settle_queue.capacity() == 0 {
             tracing::warn!("settlement queue is full; auction is rejected");
-            Err(Error::SettlementQueueIsFull)
+            Err(Error::TooManyPendingSettlements)
         } else {
             Ok(())
         }
@@ -611,5 +611,5 @@ pub enum Error {
     #[error("failed to submit the solution")]
     SubmissionError,
     #[error("too many pending settlements for the same solver")]
-    SettlementQueueIsFull,
+    TooManyPendingSettlements,
 }

crates/driver/src/infra/api/error.rs

@@ -11,6 +11,7 @@ use {
 enum Kind {
     QuotingFailed,
     SolverFailed,
+    TooManyPendingSettlements,
     SolutionNotAvailable,
     DeadlineExceeded,
     Unknown,
@@ -51,6 +52,7 @@ impl From<Kind> for (hyper::StatusCode, axum::Json<Error>) {
                  or sell amount"
             }
             Kind::FailedToSubmit => "Could not submit the solution to the blockchain",
+            Kind::TooManyPendingSettlements => "Settlement queue is full",
         };
         (
             hyper::StatusCode::BAD_REQUEST,
@@ -83,7 +85,7 @@ impl From<competition::Error> for (hyper::StatusCode, axum::Json<Error>) {
             competition::Error::DeadlineExceeded(_) => Kind::DeadlineExceeded,
             competition::Error::Solver(_) => Kind::SolverFailed,
             competition::Error::SubmissionError => Kind::FailedToSubmit,
-            competition::Error::SettlementQueueIsFull => Kind::SolverFailed,
+            competition::Error::TooManyPendingSettlements => Kind::TooManyPendingSettlements,
         };
         error.into()
     }

crates/driver/src/infra/observe/mod.rs

@@ -368,7 +368,7 @@ fn competition_error(err: &competition::Error) -> &'static str {
         competition::Error::Solver(solver::Error::Deserialize(_)) => "SolverDeserializeError",
         competition::Error::Solver(solver::Error::Dto(_)) => "SolverDtoError",
         competition::Error::SubmissionError => "SubmissionError",
-        competition::Error::SettlementQueueIsFull => "SettlementQueueIsFull",
+        competition::Error::TooManyPendingSettlements => "TooManyPendingSettlements",
     }
 }
 

crates/driver/src/tests/cases/buy_eth.rs

@@ -16,5 +16,10 @@ async fn test() {
         .await;
 
     let id = test.solve().await.ok().orders(&[order]).id();
-    test.settle(id).await.ok().await.eth_order_executed().await;
+    test.settle(id)
+        .await
+        .ok()
+        .await
+        .eth_order_executed(&test)
+        .await;
 }

crates/driver/src/tests/cases/merge_settlements.rs

@@ -42,9 +42,9 @@ async fn possible() {
         // combination of the two, meaning the settlements were merged successfully.
         .ok()
         .await
-        .ab_order_executed()
+        .ab_order_executed(&test)
         .await
-        .cd_order_executed()
+        .cd_order_executed(&test)
         .await;
 }
 
@@ -92,7 +92,12 @@ async fn impossible() {
     // Only the first A-B order gets settled.
     let id = test.solve().await.ok().orders(&[order]).id();
     test.reveal(id).await.ok().calldata();
-    test.settle(id).await.ok().await.ab_order_executed().await;
+    test.settle(id)
+        .await
+        .ok()
+        .await
+        .ab_order_executed(&test)
+        .await;
 }
 
 /// Test that mergable solutions don't get merged if feature was not enabled.
@@ -115,5 +120,10 @@ async fn possible_but_forbidden() {
     // not because solution merging is not enabled by default.
     let id = test.solve().await.ok().orders(&[ab_order]).id();
     test.reveal(id).await.ok().calldata();
-    test.settle(id).await.ok().await.ab_order_executed().await;
+    test.settle(id)
+        .await
+        .ok()
+        .await
+        .ab_order_executed(&test)
+        .await;
 }

crates/driver/src/tests/cases/parallel_auctions.rs

@@ -47,7 +47,7 @@ async fn driver_handles_solutions_based_on_id() {
         .await
         .ok()
         .await
-        .eth_order_executed()
+        .eth_order_executed(&test)
         .await;
 
     // calling `/reveal` or `/settle` with for a legit solution that
@@ -89,7 +89,12 @@ async fn driver_can_settle_old_solutions() {
     // Technically this is not super convincing since all remembered solutions
     // are identical but this is the best we are going to get without needing
     // to heavily modify the testing framework.
-    test.settle(id1).await.ok().await.eth_order_executed().await;
+    test.settle(id1)
+        .await
+        .ok()
+        .await
+        .eth_order_executed(&test)
+        .await;
 }
 
 /// Tests that the driver only remembers a relatively small number of solutions.

crates/driver/src/tests/cases/settle.rs

@@ -7,6 +7,9 @@ use {
             setup::{ab_order, ab_pool, ab_solution},
         },
     },
+    futures::future::join_all,
+    itertools::Itertools,
+    std::{sync::Arc, time::Duration},
     web3::Transport,
 };
 
@@ -30,7 +33,12 @@ async fn matrix() {
                 .await;
 
             let id = test.solve().await.ok().id();
-            test.settle(id).await.ok().await.ab_order_executed().await;
+            test.settle(id)
+                .await
+                .ok()
+                .await
+                .ab_order_executed(&test)
+                .await;
         }
     }
 }
@@ -110,3 +118,174 @@ async fn high_gas_limit() {
         .unwrap();
     test.settle(id).await.ok().await;
 }
+
+#[tokio::test]
+#[ignore]
+async fn discards_excess_settle_and_solve_requests() {
+    let test = Arc::new(
+        tests::setup()
+            .allow_multiple_solve_requests()
+            .pool(ab_pool())
+            .order(ab_order())
+            .solution(ab_solution())
+            .settle_submission_deadline(6)
+            .done()
+            .await,
+    );
+
+    // MAX_SOLUTION_STORAGE = 5. Since this is hardcoded, no more solutions can be
+    // stored.
+    let solution_ids = join_all(vec![
+        test.solve(),
+        test.solve(),
+        test.solve(),
+        test.solve(),
+        test.solve(),
+    ])
+    .await
+    .into_iter()
+    .map(|res| res.ok().id())
+    .collect::<Vec<_>>();
+
+    let unique_solutions_count = solution_ids.iter().unique().count();
+    assert_eq!(unique_solutions_count, solution_ids.len());
+
+    // Disable auto mining to accumulate all the settlement requests.
+    test.set_auto_mining(false).await;
+
+    // To avoid race conditions with the settlement queue processing, a
+    // `/settle` request needs to be sent first, so it is dequeued, and it's
+    // execution is paused before any subsequent request is received.
+    let test_clone = Arc::clone(&test);
+    let first_solution_id = solution_ids[0];
+    let first_settlement_fut =
+        tokio::spawn(async move { test_clone.settle(first_solution_id).await });
+    // Make sure the first settlement gets dequeued before sending the remaining
+    // requests.
+    tokio::time::sleep(Duration::from_millis(100)).await;
+    let remaining_solutions = solution_ids[1..].to_vec();
+    let remaining_settlements = {
+        let test_clone = Arc::clone(&test);
+        remaining_solutions.into_iter().map(move |id| {
+            let test_clone = Arc::clone(&test_clone);
+            async move { test_clone.settle(id).await }
+        })
+    };
+    let remaining_settlements_fut = tokio::spawn(join_all(remaining_settlements));
+
+    // Sleep for a bit to make sure all the settlement requests are queued.
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    // While there is no room in the settlement queue, `/solve` requests must be
+    // rejected.
+    test.solve().await.err().kind("TooManyPendingSettlements");
+
+    // Enable auto mining to process all the settlement requests.
+    test.set_auto_mining(true).await;
+
+    // The first settlement must be successful.
+    let first_settlement = first_settlement_fut.await.unwrap();
+    first_settlement.ok().await.ab_order_executed(&test).await;
+
+    let remaining_settlements = remaining_settlements_fut.await.unwrap();
+    assert_eq!(remaining_settlements.len(), 4);
+
+    for (idx, result) in remaining_settlements.into_iter().enumerate() {
+        match idx {
+            // The next 2 settlements failed to submit due to the framework's limitation(unable to
+            // fulfill the same order again).
+            0 | 1 => result.err().kind("FailedToSubmit"),
+            // All the subsequent settlements rejected due to the settlement queue being full.
+            2 | 3 => result.err().kind("TooManyPendingSettlements"),
+            _ => unreachable!(),
+        }
+    }
+
+    // `/solve` works again.
+    test.solve().await.ok();
+}
+
+#[tokio::test]
+#[ignore]
+async fn accepts_new_settle_requests_after_timeout() {
+    let test = Arc::new(
+        tests::setup()
+            .allow_multiple_solve_requests()
+            .pool(ab_pool())
+            .order(ab_order())
+            .solution(ab_solution())
+            .settle_submission_deadline(6)
+            .done()
+            .await,
+    );
+
+    // MAX_SOLUTION_STORAGE = 5. Since this is hardcoded, no more solutions can be
+    // stored.
+    let solution_ids = join_all(vec![
+        test.solve(),
+        test.solve(),
+        test.solve(),
+        test.solve(),
+        test.solve(),
+    ])
+    .await
+    .into_iter()
+    .map(|res| res.ok().id())
+    .collect::<Vec<_>>();
+
+    let unique_solutions_count = solution_ids.iter().unique().count();
+    assert_eq!(unique_solutions_count, solution_ids.len());
+
+    // Disable auto mining to accumulate all the settlement requests.
+    test.set_auto_mining(false).await;
+
+    // To avoid race conditions with the settlement queue processing, a
+    // `/settle` request needs to be sent first, so it is dequeued, and it's
+    // execution is paused before any subsequent request is received.
+    let test_clone = Arc::clone(&test);
+    let first_solution_id = solution_ids[0];
+    let first_settlement_fut =
+        tokio::spawn(async move { test_clone.settle(first_solution_id).await });
+    // Make sure the first settlement gets dequeued before sending the remaining
+    // requests.
+    tokio::time::sleep(Duration::from_millis(100)).await;
+    // Send only 3 more settle requests.
+    let additional_solutions = solution_ids[1..4].to_vec();
+    let additional_settlements = {
+        let test_clone = Arc::clone(&test);
+        additional_solutions.into_iter().map(move |id| {
+            let test_clone = Arc::clone(&test_clone);
+            async move { test_clone.settle(id).await }
+        })
+    };
+    let additional_settlements_fut = tokio::spawn(join_all(additional_settlements));
+
+    // Sleep for a bit to make sure all the settlement requests are queued.
+    tokio::time::sleep(Duration::from_millis(500)).await;
+    test.set_auto_mining(true).await;
+
+    let first_settlement = first_settlement_fut.await.unwrap();
+    // The first settlement must be successful.
+    first_settlement.ok().await.ab_order_executed(&test).await;
+
+    let additional_settlements = additional_settlements_fut.await.unwrap();
+    assert_eq!(additional_settlements.len(), 3);
+
+    for (idx, result) in additional_settlements.into_iter().enumerate() {
+        match idx {
+            // The next 2 settlements failed to submit due to the framework's limitation(unable to
+            // fulfill the same order again).
+            0 | 1 => result.err().kind("FailedToSubmit"),
+            // The next request gets rejected due to the settlement queue being full.
+            2 => result.err().kind("TooManyPendingSettlements"),
+            _ => unreachable!(),
+        }
+    }
+
+    // Now we send the last settlement request. It fails due to the framework's
+    // limitation(unable to fulfill the same order again).
+    test.settle(solution_ids[4])
+        .await
+        .err()
+        .kind("FailedToSubmit");
+}

crates/driver/src/tests/setup/blockchain.rs

@@ -12,7 +12,7 @@ use {
     secp256k1::SecretKey,
     serde_json::json,
     std::collections::HashMap,
-    web3::signing::Key,
+    web3::{signing::Key, Transport},
 };
 
 // TODO Possibly might be a good idea to use an enum for tokens instead of
@@ -857,6 +857,14 @@ impl Blockchain {
             _ => self.tokens.get(token).unwrap().address(),
         }
     }
+
+    pub async fn set_auto_mining(&self, enabled: bool) {
+        self.web3
+            .transport()
+            .execute("evm_setAutomine", vec![json!(enabled)])
+            .await
+            .unwrap();
+    }
 }
 
 async fn primary_address(web3: &DynWeb3) -> ethcontract::H160 {