Use blocks to determine other environment settlements (#3053)

# Description
This PR updates how the `settlement::Observer` distinguishes between
settlements from staging and production environments.

Staging is usually deployed first, resulting in higher auction IDs,
while production lags behind. Since both environments share a settlement
contract, they must:

- Correctly process settlements from their own environment.
- Skip processing settlements from the other environment, while still
saving the settlement to auction_id relationship in the database.

Detecting staging settlements in production is straightforward and
handled by the "AuctionNotFound" error, as staging auctions have higher
IDs that don't exist in production yet.

However, detecting production settlements in staging has been
problematic. Previously, we used the "auction_has_settlement" logic, but
this wasn’t reliable. For example, if a staging auction is saved to the
database but not settled by solvers, production might settle the same
auction after 270 days, leading to incorrect detection.

This PR introduces a new approach: if a settlement refers to a VERY old
auction (in terms of block time), it’s classified as from the other
environment. While this may also catch settlements that violated auction
block deadline, from the same environment, this issue hasn't occurred in
practice.

This solution isn’t perfect but is a reasonable improvement for now.

# Changes
<!-- List of detailed changes (how the change is accomplished) -->

- [ ] Use block numbers to detect settlements with very old auction ids

## How to test
Will observe the effects on staging.

crates/autopilot/src/boundary/mod.rs

@@ -17,6 +17,7 @@ pub use {
             SellTokenSource,
         },
         signature::{EcdsaSignature, Signature, SigningScheme},
+        solver_competition::SolverCompetitionDB,
         DomainSeparator,
     },
     shared::order_validation::{is_order_outside_market_price, Amounts},

crates/autopilot/src/database/competition.rs

@@ -10,10 +10,9 @@ use {
         Address,
     },
     derivative::Derivative,
-    model::solver_competition::{SolverCompetitionAPI, SolverCompetitionDB},
+    model::solver_competition::SolverCompetitionDB,
     number::conversions::u256_to_big_decimal,
-    primitive_types::{H160, H256, U256},
-    sqlx::{types::JsonValue, PgConnection},
+    primitive_types::{H160, U256},
     std::collections::{BTreeMap, HashSet},
 };
 
@@ -141,35 +140,4 @@ impl super::Postgres {
 
         Ok(())
     }
-
-    pub async fn find_competition(
-        auction_id: AuctionId,
-        ex: &mut PgConnection,
-    ) -> anyhow::Result<Option<SolverCompetitionAPI>> {
-        database::solver_competition::load_by_id(ex, auction_id)
-            .await
-            .context("solver_competition::load_by_id")?
-            .map(|row| {
-                deserialize_solver_competition(
-                    row.json,
-                    row.id,
-                    row.tx_hashes.iter().map(|hash| H256(hash.0)).collect(),
-                )
-            })
-            .transpose()
-    }
-}
-
-fn deserialize_solver_competition(
-    json: JsonValue,
-    auction_id: model::auction::AuctionId,
-    transaction_hashes: Vec<H256>,
-) -> anyhow::Result<SolverCompetitionAPI> {
-    let common: SolverCompetitionDB =
-        serde_json::from_value(json).context("deserialize SolverCompetitionDB")?;
-    Ok(SolverCompetitionAPI {
-        auction_id,
-        transaction_hashes,
-        common,
-    })
 }

crates/autopilot/src/domain/eth/mod.rs

@@ -14,6 +14,15 @@ pub struct Address(pub H160);
 #[derive(Debug, Copy, Clone, From, PartialEq, PartialOrd, Default)]
 pub struct BlockNo(pub u64);
 
+/// Adding blocks to a block number.
+impl std::ops::Add<u64> for BlockNo {
+    type Output = BlockNo;
+
+    fn add(self, rhs: u64) -> Self::Output {
+        Self(self.0 + rhs)
+    }
+}
+
 /// A transaction ID, AKA transaction hash.
 #[derive(Debug, Copy, Clone, From, Default)]
 pub struct TxId(pub H256);

crates/autopilot/src/domain/settlement/auction.rs

@@ -8,6 +8,8 @@ use {
 #[derive(Debug)]
 pub struct Auction {
     pub id: domain::auction::Id,
+    /// The block on top of which the auction was created.
+    pub block: domain::eth::BlockNo,
     /// All orders from a competition auction. Some of them may contain fee
     /// policies.
     pub orders: HashMap<domain::OrderUid, Vec<domain::fee::Policy>>,

crates/autopilot/src/domain/settlement/mod.rs

@@ -4,7 +4,10 @@
 //! a form of settlement transaction.
 
 use {
-    crate::{domain, domain::eth, infra},
+    crate::{
+        domain::{self, eth},
+        infra,
+    },
     std::collections::HashMap,
 };
 
@@ -103,19 +106,21 @@ impl Settlement {
     pub async fn new(
         settled: Transaction,
         persistence: &infra::Persistence,
+        chain: &infra::blockchain::Id,
     ) -> Result<Self, Error> {
-        if persistence
-            .auction_has_settlement(settled.auction_id)
-            .await?
-        {
-            // This settlement has already been processed by another environment.
+        let auction = persistence.get_auction(settled.auction_id).await?;
+
+        if settled.block > auction.block + max_settlement_age(chain) {
+            // A settled transaction references a VERY old auction.
+            //
+            // A hacky way to detect processing of production settlements in the staging
+            // environment, as production is lagging with auction ids by ~270 days on
+            // Ethereum mainnet.
             //
             // TODO: remove once https://github.com/cowprotocol/services/issues/2848 is resolved and ~270 days are passed since bumping.
             return Err(Error::WrongEnvironment);
         }
 
-        let auction = persistence.get_auction(settled.auction_id).await?;
-
         let trades = settled
             .trades
             .into_iter()
@@ -133,6 +138,25 @@ impl Settlement {
     }
 }
 
+const MAINNET_BLOCK_TIME: u64 = 13_000; // ms
+const GNOSIS_BLOCK_TIME: u64 = 5_000; // ms
+const SEPOLIA_BLOCK_TIME: u64 = 13_000; // ms
+const ARBITRUM_ONE_BLOCK_TIME: u64 = 100; // ms
+
+/// How old (in terms of blocks) a settlement should be, to be considered as a
+/// settlement from another environment.
+///
+/// Currently set to ~6h
+fn max_settlement_age(chain: &infra::blockchain::Id) -> u64 {
+    const TARGET_AGE: u64 = 6 * 60 * 60 * 1000; // 6h in ms
+    match chain {
+        infra::blockchain::Id::Mainnet => TARGET_AGE / MAINNET_BLOCK_TIME,
+        infra::blockchain::Id::Gnosis => TARGET_AGE / GNOSIS_BLOCK_TIME,
+        infra::blockchain::Id::Sepolia => TARGET_AGE / SEPOLIA_BLOCK_TIME,
+        infra::blockchain::Id::ArbitrumOne => TARGET_AGE / ARBITRUM_ONE_BLOCK_TIME,
+    }
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
     #[error("failed communication with the database: {0}")]
@@ -286,6 +310,7 @@ mod tests {
         let order_uid = transaction.trades[0].uid;
 
         let auction = super::Auction {
+            block: eth::BlockNo(0),
             // prices read from https://solver-instances.s3.eu-central-1.amazonaws.com/prod/mainnet/legacy/8655372.json
             prices: auction::Prices::from([
                 (
@@ -436,6 +461,7 @@ mod tests {
 
         let order_uid = transaction.trades[0].uid;
         let auction = super::Auction {
+            block: eth::BlockNo(0),
             prices,
             surplus_capturing_jit_order_owners: Default::default(),
             id: 0,
@@ -599,6 +625,7 @@ mod tests {
         ]);
 
         let auction = super::Auction {
+            block: eth::BlockNo(0),
             prices,
             surplus_capturing_jit_order_owners: HashSet::from([eth::Address(
                 eth::H160::from_slice(&hex!("f08d4dea369c456d26a3168ff0024b904f2d8b91")),
@@ -777,6 +804,7 @@ mod tests {
         ]);
 
         let auction = super::Auction {
+            block: eth::BlockNo(0),
             prices,
             surplus_capturing_jit_order_owners: Default::default(),
             id: 0,

crates/autopilot/src/domain/settlement/observer.rs

@@ -78,8 +78,12 @@ impl Observer {
         let (auction_id, settlement) = match transaction {
             Ok(transaction) => {
                 let auction_id = transaction.auction_id;
-                let settlement = match settlement::Settlement::new(transaction, &self.persistence)
-                    .await
+                let settlement = match settlement::Settlement::new(
+                    transaction,
+                    &self.persistence,
+                    self.eth.chain(),
+                )
+                .await
                 {
                     Ok(settlement) => Some(settlement),
                     Err(err) if retryable(&err) => return Err(err.into()),

crates/autopilot/src/infra/blockchain/authenticator.rs

@@ -2,8 +2,8 @@ use {
     crate::{
         domain::{self, eth},
         infra::blockchain::{
+            self,
             contracts::{deployment_address, Contracts},
-            ChainId,
         },
     },
     ethcontract::{dyns::DynWeb3, GasPrice},
@@ -25,7 +25,7 @@ impl Manager {
     /// Creates an authenticator which can remove solvers from the allow-list
     pub async fn new(
         web3: DynWeb3,
-        chain: ChainId,
+        chain: blockchain::Id,
         contracts: Contracts,
         authenticator_pk: eth::H256,
     ) -> Self {

crates/autopilot/src/infra/blockchain/contracts.rs

@@ -1,4 +1,8 @@
-use {super::ChainId, crate::domain, ethcontract::dyns::DynWeb3, primitive_types::H160};
+use {
+    crate::{domain, infra::blockchain},
+    ethcontract::dyns::DynWeb3,
+    primitive_types::H160,
+};
 
 #[derive(Debug, Clone)]
 pub struct Contracts {
@@ -20,7 +24,7 @@ pub struct Addresses {
 }
 
 impl Contracts {
-    pub async fn new(web3: &DynWeb3, chain: &ChainId, addresses: Addresses) -> Self {
+    pub async fn new(web3: &DynWeb3, chain: &blockchain::Id, addresses: Addresses) -> Self {
         let address_for = |contract: &ethcontract::Contract, address: Option<H160>| {
             address
                 .or_else(|| deployment_address(contract, chain))
@@ -92,6 +96,12 @@ impl Contracts {
 
 /// Returns the address of a contract for the specified network, or `None` if
 /// there is no known deployment for the contract on that network.
-pub fn deployment_address(contract: &ethcontract::Contract, chain: &ChainId) -> Option<H160> {
-    Some(contract.networks.get(&chain.to_string())?.address)
+pub fn deployment_address(
+    contract: &ethcontract::Contract,
+    chain: &blockchain::Id,
+) -> Option<H160> {
+    contract
+        .networks
+        .get(chain.network_id())
+        .map(|network| network.address)
 }

crates/autopilot/src/infra/blockchain/id.rs

@@ -0,0 +1,43 @@
+use primitive_types::U256;
+
+/// A supported Ethereum Chain ID.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum Id {
+    Mainnet = 1,
+    Gnosis = 100,
+    Sepolia = 11155111,
+    ArbitrumOne = 42161,
+}
+
+impl Id {
+    pub fn new(value: U256) -> Result<Self, UnsupportedChain> {
+        // Check to avoid panics for large `U256` values, as there is no checked
+        // conversion API available and we don't support chains with IDs greater
+        // than `u64::MAX` anyway.
+        if value > U256::from(u64::MAX) {
+            return Err(UnsupportedChain);
+        }
+
+        match value.as_u64() {
+            1 => Ok(Self::Mainnet),
+            100 => Ok(Self::Gnosis),
+            11155111 => Ok(Self::Sepolia),
+            42161 => Ok(Self::ArbitrumOne),
+            _ => Err(UnsupportedChain),
+        }
+    }
+
+    /// Returns the network ID for the chain.
+    pub fn network_id(self) -> &'static str {
+        match self {
+            Id::Mainnet => "1",
+            Id::Gnosis => "100",
+            Id::Sepolia => "11155111",
+            Id::ArbitrumOne => "42161",
+        }
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+#[error("unsupported chain")]
+pub struct UnsupportedChain;

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

@@ -11,29 +11,14 @@ use {
 
 pub mod authenticator;
 pub mod contracts;
+pub mod id;
 
-/// Chain ID as defined by EIP-155.
-///
-/// https://eips.ethereum.org/EIPS/eip-155
-#[derive(Clone, Copy, Debug, Eq, PartialEq)]
-pub struct ChainId(pub U256);
-
-impl std::fmt::Display for ChainId {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", self.0)
-    }
-}
-
-impl From<U256> for ChainId {
-    fn from(value: U256) -> Self {
-        Self(value)
-    }
-}
+pub use id::Id;
 
 /// An Ethereum RPC connection.
 pub struct Rpc {
     web3: DynWeb3,
-    chain: ChainId,
+    chain: Id,
     url: Url,
 }
 
@@ -45,7 +30,7 @@ impl Rpc {
         ethrpc_args: &shared::ethrpc::Arguments,
     ) -> Result<Self, Error> {
         let web3 = boundary::web3_client(url, ethrpc_args);
-        let chain = web3.eth().chain_id().await?.into();
+        let chain = Id::new(web3.eth().chain_id().await?).map_err(|_| Error::UnsupportedChain)?;
 
         Ok(Self {
             web3,
@@ -55,7 +40,7 @@ impl Rpc {
     }
 
     /// Returns the chain id for the RPC connection.
-    pub fn chain(&self) -> ChainId {
+    pub fn chain(&self) -> Id {
         self.chain
     }
 
@@ -74,7 +59,7 @@ impl Rpc {
 #[derive(Clone)]
 pub struct Ethereum {
     web3: DynWeb3,
-    chain: ChainId,
+    chain: Id,
     current_block: CurrentBlockWatcher,
     contracts: Contracts,
 }
@@ -88,7 +73,7 @@ impl Ethereum {
     /// any initialization error.
     pub async fn new(
         web3: DynWeb3,
-        chain: ChainId,
+        chain: Id,
         url: Url,
         addresses: contracts::Addresses,
         poll_interval: Duration,
@@ -105,7 +90,7 @@ impl Ethereum {
         }
     }
 
-    pub fn network(&self) -> &ChainId {
+    pub fn chain(&self) -> &Id {
         &self.chain
     }
 
@@ -179,4 +164,6 @@ pub enum Error {
     IncompleteTransactionData(anyhow::Error),
     #[error("transaction not found")]
     TransactionNotFound,
+    #[error("unsupported chain")]
+    UnsupportedChain,
 }