Wait for solve deadline (#2008)

# Description
Fixes #2007. The preferred solution described in that issue was not
straight forward to implement.
It expected the `autopilot` to call `/reveal` from the highest scoring
solution to the lowest scoring one to simulate it in order to avoid
cases where a solver wins the auction with a solution that would revert
by now.
The issue was that this requires the `autopilot` to know the address of
every solver which would be a significant change.

Instead this PR implements a strategy on the `driver` side that a
rational actor would be expected to follow as well.
All solvers know when they have to return a solution at the latest.
Because a solution can be more accurate the closer it was computed to
the time it is supposed to be executed it makes sense for all solvers to
delay their response as much as possible (something might change in the
mean time which might enable an even better solution).
This deadline gets propagated to the `solver` engine so ideally it would
take as much time computing the optimal solution as possible.
If the solver returns earlier (like some currently do) it still makes
sense to assume that other solvers will submit their solution at a later
point in time.
In that case the only reasonable action for the `driver` is to wait
until the deadline approaches and continuously re-simulate the computed
solution whenever a new block gets detected. If in the meantime the
solution would start to revert the `driver` would then withhold the
computed solution to not accidentally win the auction and get slashed
for not submitting it on-chain.

# Changes
`driver` waits until deadline before returning the solution on `/solve`
and checks whether it's still viable on every new block. It also updates
the score in case the solution still simulates but becomes better or
worse (gas usage might change).
Also slightly adjusted to `Ethereum::new()` to panic on any init error
since we can't handle those errors anyway because the type is essential
to the program.

## How to test
This can be tested by a new e2e test which I would like to implement in
a follow up PR when existing e2e tests don't break.

crates/autopilot/src/arguments.rs

@@ -194,6 +194,15 @@ pub struct Arguments {
     /// being specified separately.
     #[clap(long, env)]
     pub shadow: Option<Url>,
+
+    /// Time in seconds solvers have to compute a score per auction.
+    #[clap(
+        long,
+        env,
+        default_value = "15",
+        value_parser = shared::arguments::duration_from_seconds,
+    )]
+    pub solve_deadline: Duration,
 }
 
 impl std::fmt::Display for Arguments {
@@ -253,6 +262,7 @@ impl std::fmt::Display for Arguments {
         )?;
         writeln!(f, "score_cap: {}", self.score_cap)?;
         display_option(f, "shadow", &self.shadow)?;
+        writeln!(f, "solve_deadline: {:?}", self.solve_deadline)?;
         Ok(())
     }
 }

crates/autopilot/src/run.rs

@@ -609,6 +609,7 @@ pub async fn run(args: Arguments) {
             additional_deadline_for_rewards: args.additional_deadline_for_rewards as u64,
             score_cap: args.score_cap,
             max_settlement_transaction_wait: args.max_settlement_transaction_wait,
+            solve_deadline: args.solve_deadline,
         };
         run.run_forever().await;
         unreachable!("run loop exited");
@@ -662,7 +663,13 @@ async fn shadow_mode(args: Arguments) -> ! {
         .await
     };
 
-    let shadow = shadow::RunLoop::new(orderbook, drivers, trusted_tokens, args.score_cap);
+    let shadow = shadow::RunLoop::new(
+        orderbook,
+        drivers,
+        trusted_tokens,
+        args.score_cap,
+        args.solve_deadline,
+    );
     shadow.run_forever().await;
 
     unreachable!("shadow run loop exited");

crates/autopilot/src/run_loop.rs

@@ -41,8 +41,6 @@ use {
     tracing::Instrument,
 };
 
-pub const SOLVE_TIME_LIMIT: Duration = Duration::from_secs(15);
-
 pub struct RunLoop {
     pub solvable_orders_cache: Arc<SolvableOrdersCache>,
     pub database: Postgres,
@@ -55,6 +53,7 @@ pub struct RunLoop {
     pub additional_deadline_for_rewards: u64,
     pub score_cap: U256,
     pub max_settlement_transaction_wait: Duration,
+    pub solve_deadline: Duration,
 }
 
 impl RunLoop {
@@ -302,6 +301,7 @@ impl RunLoop {
             auction,
             &self.market_makable_token_list.all(),
             self.score_cap,
+            self.solve_deadline,
         );
         let request = &request;
 
@@ -358,7 +358,7 @@ impl RunLoop {
         driver: &Driver,
         request: &solve::Request,
     ) -> Result<Vec<Result<Solution, ZeroScoreError>>, SolveError> {
-        let response = tokio::time::timeout(SOLVE_TIME_LIMIT, driver.solve(request))
+        let response = tokio::time::timeout(self.solve_deadline, driver.solve(request))
             .await
             .map_err(|_| SolveError::Timeout)?
             .map_err(SolveError::Failure)?;
@@ -448,6 +448,7 @@ pub fn solve_request(
     auction: &Auction,
     trusted_tokens: &HashSet<H160>,
     score_cap: U256,
+    time_limit: Duration,
 ) -> solve::Request {
     solve::Request {
         id,
@@ -516,7 +517,7 @@ pub fn solve_request(
             }))
             .unique_by(|token| token.address)
             .collect(),
-        deadline: Utc::now() + chrono::Duration::from_std(SOLVE_TIME_LIMIT).unwrap(),
+        deadline: Utc::now() + chrono::Duration::from_std(time_limit).unwrap(),
         score_cap,
     }
 }

crates/autopilot/src/shadow.rs

@@ -33,6 +33,7 @@ pub struct RunLoop {
     auction: AuctionId,
     block: u64,
     score_cap: U256,
+    solve_deadline: Duration,
 }
 
 impl RunLoop {
@@ -41,6 +42,7 @@ impl RunLoop {
         drivers: Vec<Driver>,
         trusted_tokens: AutoUpdatingTokenList,
         score_cap: U256,
+        solve_deadline: Duration,
     ) -> Self {
         Self {
             orderbook,
@@ -49,6 +51,7 @@ impl RunLoop {
             auction: 0,
             block: 0,
             score_cap,
+            solve_deadline,
         }
     }
 
@@ -173,8 +176,13 @@ impl RunLoop {
 
     /// Runs the solver competition, making all configured drivers participate.
     async fn competition(&self, id: AuctionId, auction: &Auction) -> Vec<Participant<'_>> {
-        let request =
-            run_loop::solve_request(id, auction, &self.trusted_tokens.all(), self.score_cap);
+        let request = run_loop::solve_request(
+            id,
+            auction,
+            &self.trusted_tokens.all(),
+            self.score_cap,
+            self.solve_deadline,
+        );
         let request = &request;
 
         futures::future::join_all(self.drivers.iter().map(|driver| async move {
@@ -190,7 +198,7 @@ impl RunLoop {
         driver: &Driver,
         request: &solve::Request,
     ) -> Result<Solution, Error> {
-        let proposed = tokio::time::timeout(run_loop::SOLVE_TIME_LIMIT, driver.solve(request))
+        let proposed = tokio::time::timeout(self.solve_deadline, driver.solve(request))
             .await
             .map_err(|_| Error::Timeout)?
             .map_err(Error::Solve)?;

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

@@ -13,7 +13,7 @@ use {
         },
         util::Bytes,
     },
-    futures::future::join_all,
+    futures::{future::join_all, StreamExt},
     itertools::Itertools,
     rand::seq::SliceRandom,
     std::{collections::HashSet, sync::Mutex},
@@ -172,13 +172,41 @@ impl Competition {
         }
 
         // Pick the best-scoring settlement.
-        let (score, settlement) = scores
+        let (mut score, settlement) = scores
             .into_iter()
             .max_by_key(|(score, _)| score.to_owned())
             .map(|(score, settlement)| (Solved { score }, settlement))
             .unzip();
 
-        *self.settlement.lock().unwrap() = settlement;
+        *self.settlement.lock().unwrap() = settlement.clone();
+
+        let settlement = match settlement {
+            Some(settlement) => settlement,
+            // Don't wait for the deadline because we can't produce a solution anyway.
+            None => return Ok(score),
+        };
+
+        // Re-simulate the solution on every new block until the deadline ends to make
+        // sure we actually submit a working solution close to when the winner
+        // gets picked by the procotol.
+        if let Ok(deadline) = auction.deadline().timeout() {
+            let score_ref = &mut score;
+            let simulate_on_new_blocks = async move {
+                let mut stream =
+                    ethrpc::current_block::into_stream(self.eth.current_block().clone());
+                while let Some(block) = stream.next().await {
+                    if let Err(err) = self.simulate_settlement(&settlement).await {
+                        tracing::warn!(block = block.number, ?err, "solution reverts on new block");
+                        *score_ref = None;
+                        *self.settlement.lock().unwrap() = None;
+                        return;
+                    }
+                }
+            };
+            let timeout = deadline.duration().to_std().unwrap_or_default();
+            let _ = tokio::time::timeout(timeout, simulate_on_new_blocks).await;
+        }
+
         Ok(score)
     }
 
@@ -245,6 +273,26 @@ impl Competition {
             .as_ref()
             .map(|s| s.auction_id)
     }
+
+    /// Returns whether the settlement can be executed or would revert.
+    async fn simulate_settlement(
+        &self,
+        settlement: &Settlement,
+    ) -> Result<(), infra::simulator::Error> {
+        self.simulator
+            .gas(eth::Tx {
+                from: self.solver.address(),
+                to: settlement.solver(),
+                value: eth::Ether(0.into()),
+                input: crate::util::Bytes(settlement.calldata(
+                    self.eth.contracts().settlement(),
+                    settlement::Internalization::Enable,
+                )),
+                access_list: settlement.access_list.clone(),
+            })
+            .await
+            .map(|_| ())
+    }
 }
 
 /// Solution information sent to the protocol by the driver before the solution

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

@@ -2,6 +2,7 @@ use {
     self::contracts::ContractAt,
     crate::{boundary, domain::eth},
     ethcontract::dyns::DynWeb3,
+    ethrpc::current_block::CurrentBlockStream,
     std::{fmt, sync::Arc},
     thiserror::Error,
     web3::Transport,
@@ -58,24 +59,38 @@ pub struct Ethereum {
     network: Network,
     contracts: Contracts,
     gas: Arc<GasPriceEstimator>,
+    current_block: CurrentBlockStream,
 }
 
 impl Ethereum {
     /// Access the Ethereum blockchain through an RPC API.
+    ///
+    /// # Panics
+    ///
+    /// Since this type is essential for the program this method will panic on
+    /// any initialization error.
     pub async fn new(
         rpc: Rpc,
         addresses: contracts::Addresses,
         gas: Arc<GasPriceEstimator>,
-    ) -> Result<Self, Error> {
+    ) -> Self {
         let Rpc { web3, network } = rpc;
-        let contracts = Contracts::new(&web3, &network.id, addresses).await?;
+        let contracts = Contracts::new(&web3, &network.id, addresses)
+            .await
+            .expect("could not initialize important smart contracts");
 
-        Ok(Self {
+        Self {
+            current_block: ethrpc::current_block::current_block_stream(
+                Arc::new(web3.clone()),
+                std::time::Duration::from_millis(500),
+            )
+            .await
+            .expect("couldn't initialize current block stream"),
             web3,
             network,
             contracts,
             gas,
-        })
+        }
     }
 
     pub fn network(&self) -> &Network {
@@ -98,6 +113,12 @@ impl Ethereum {
         Ok(!code.0.is_empty())
     }
 
+    /// Returns a type that monitors the block chain to inform about the current
+    /// block.
+    pub fn current_block(&self) -> &CurrentBlockStream {
+        &self.current_block
+    }
+
     /// Create access list used by a transaction.
     pub async fn create_access_list(&self, tx: eth::Tx) -> Result<eth::AccessList, Error> {
         let tx = web3::types::TransactionRequest {

crates/driver/src/run.rs

@@ -133,9 +133,7 @@ async fn ethereum(config: &infra::Config, ethrpc: blockchain::Rpc) -> Ethereum {
             .await
             .expect("initialize gas price estimator"),
     );
-    Ethereum::new(ethrpc, config.contracts, gas)
-        .await
-        .expect("initialize ethereum RPC API")
+    Ethereum::new(ethrpc, config.contracts, gas).await
 }
 
 fn solvers(config: &config::Config, eth: &Ethereum) -> Vec<Solver> {

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

@@ -645,7 +645,7 @@ impl Setup {
     }
 
     fn deadline(&self) -> chrono::DateTime<chrono::Utc> {
-        time::now() + chrono::Duration::days(30)
+        time::now() + chrono::Duration::seconds(2)
     }
 }
 

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

@@ -208,8 +208,7 @@ impl Solver {
             },
             gas,
         )
-        .await
-        .unwrap();
+        .await;
         let state = Arc::new(Mutex::new(StateInner { called: false }));
         let app = axum::Router::new()
         .route(

crates/e2e/src/setup/services.rs

@@ -88,6 +88,7 @@ impl<'a> Services<'a> {
             "--auction-update-interval=1".to_string(),
             format!("--ethflow-contract={:?}", self.contracts.ethflow.address()),
             "--skip-event-sync=true".to_string(),
+            "--solve-deadline=2".to_string(),
         ]
         .into_iter()
         .chain(self.api_autopilot_solver_arguments())