Remove auction_transaction

crates/autopilot/src/database.rs

@@ -6,7 +6,6 @@ use {
 
 mod auction;
 pub mod auction_prices;
-pub mod auction_transaction;
 pub mod competition;
 pub mod ethflow_events;
 mod events;

crates/autopilot/src/database/on_settlement_event_updater.rs

@@ -1,7 +1,8 @@
 use {
+    crate::on_settlement_event_updater::AuctionKind,
     anyhow::{Context, Result},
     database::{byte_array::ByteArray, settlement_observations::Observation},
-    ethcontract::{H160, U256},
+    ethcontract::U256,
     model::order::OrderUid,
     number::conversions::u256_to_big_decimal,
     sqlx::PgConnection,
@@ -21,12 +22,11 @@ pub struct AuctionData {
     pub order_executions: Vec<(OrderUid, ExecutedFee)>,
 }
 
-#[derive(Debug, Default, Clone)]
+#[derive(Debug, Clone)]
 pub struct SettlementUpdate {
     pub block_number: i64,
     pub log_index: i64,
-    pub tx_from: H160,
-    pub tx_nonce: i64,
+    pub auction_kind: AuctionKind,
     pub auction_data: Option<AuctionData>,
 }
 
@@ -40,32 +40,25 @@ impl super::Postgres {
             .with_label_values(&["update_settlement_details"])
             .start_timer();
 
+        let (auction_id, auction_kind) = match settlement_update.auction_kind {
+            AuctionKind::Valid { auction_id } => {
+                (Some(auction_id), database::settlements::AuctionKind::Valid)
+            }
+            AuctionKind::Invalid => (None, database::settlements::AuctionKind::Invalid),
+        };
+
         // update settlements
-        database::auction_transaction::insert_settlement_tx_info(
+        database::settlements::update_settlement_auction(
             ex,
             settlement_update.block_number,
             settlement_update.log_index,
-            &ByteArray(settlement_update.tx_from.0),
-            settlement_update.tx_nonce,
+            auction_id,
+            auction_kind,
         )
         .await
         .context("insert_settlement_tx_info")?;
 
         if let Some(auction_data) = settlement_update.auction_data {
-            // Link the `auction_id` to the settlement tx. This is needed for
-            // colocated solutions and is a no-op for centralized
-            // solutions.
-            let insert_succesful = database::auction_transaction::try_insert_auction_transaction(
-                ex,
-                auction_data.auction_id,
-                &ByteArray(settlement_update.tx_from.0),
-                settlement_update.tx_nonce,
-            )
-            .await
-            .context("failed to insert auction_transaction")?;
-
-            // in case of deep reindexing we might already have the observation, so just
-            // overwrite it
             database::settlement_observations::upsert(
                 ex,
                 Observation {
@@ -80,17 +73,15 @@ impl super::Postgres {
             .await
             .context("insert_settlement_observations")?;
 
-            if insert_succesful {
-                for (order, executed_fee) in auction_data.order_executions {
-                    database::order_execution::save(
-                        ex,
-                        &ByteArray(order.0),
-                        auction_data.auction_id,
-                        &u256_to_big_decimal(&executed_fee),
-                    )
-                    .await
-                    .context("save_order_executions")?;
-                }
+            for (order, executed_fee) in auction_data.order_executions {
+                database::order_execution::save(
+                    ex,
+                    &ByteArray(order.0),
+                    auction_data.auction_id,
+                    &u256_to_big_decimal(&executed_fee),
+                )
+                .await
+                .context("save_order_executions")?;
             }
         }
         Ok(())

crates/autopilot/src/on_settlement_event_updater.rs

@@ -39,14 +39,40 @@ use {
         decoded_settlement::{DecodedSettlement, DecodingError},
         infra,
     },
-    anyhow::{anyhow, Context, Result},
+    anyhow::{Context, Result},
     futures::StreamExt,
     primitive_types::H256,
-    shared::{event_handling::MAX_REORG_BLOCK_COUNT, external_prices::ExternalPrices},
+    shared::external_prices::ExternalPrices,
     sqlx::PgConnection,
     web3::types::Transaction,
 };
 
+#[derive(Debug, Copy, Clone)]
+pub enum AuctionKind {
+    /// This auction is regular and all the auction dependent data should be
+    /// updated.
+    Valid { auction_id: i64 },
+    /// Some possible reasons to have invalid auction are:
+    /// - This auction was created by another environment (e.g.
+    ///   production/staging)
+    /// - Failed to decode settlement calldata
+    /// - Failed to recover auction id from calldata
+    /// - Settlement transaction was submitted by solver other than the winner
+    ///
+    /// In this case, settlement event should be marked as invalid and no
+    /// auction dependent data is updated.
+    Invalid,
+}
+
+impl AuctionKind {
+    pub fn auction_id(&self) -> Option<i64> {
+        match self {
+            AuctionKind::Valid { auction_id } => Some(*auction_id),
+            AuctionKind::Invalid => None,
+        }
+    }
+}
+
 pub struct OnSettlementEventUpdater {
     pub eth: infra::Ethereum,
     pub db: Postgres,
@@ -57,7 +83,7 @@ impl OnSettlementEventUpdater {
         let mut current_block = self.eth.current_block().borrow().to_owned();
         let mut block_stream = ethrpc::current_block::into_stream(self.eth.current_block().clone());
         loop {
-            match self.update(current_block.number).await {
+            match self.update().await {
                 Ok(true) => {
                     tracing::debug!(
                         block = current_block.number,
@@ -77,31 +103,27 @@ impl OnSettlementEventUpdater {
                 }
             }
             current_block = block_stream.next().await.expect("blockchains never end");
+
+            // Wait a bit more to not race with the event indexer.
+            // Otherwise we might miss event and have to wait for next block to retrigger
+            // loop.
+            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
         }
     }
 
     /// Update database for settlement events that have not been processed yet.
     ///
     /// Returns whether an update was performed.
-    async fn update(&self, current_block: u64) -> Result<bool> {
-        let reorg_safe_block: i64 = current_block
-            .checked_sub(MAX_REORG_BLOCK_COUNT)
-            .context("no reorg safe block")?
-            .try_into()
-            .context("convert block")?;
-
+    async fn update(&self) -> Result<bool> {
         let mut ex = self
             .db
             .pool
             .begin()
             .await
             .context("acquire DB connection")?;
-        let event = match database::auction_transaction::get_settlement_event_without_tx_info(
-            &mut ex,
-            reorg_safe_block,
-        )
-        .await
-        .context("get_settlement_event_without_tx_info")?
+        let event = match database::settlements::get_settlement_without_auction(&mut ex)
+            .await
+            .context("get_settlement_without_auction")?
         {
             Some(event) => event,
             None => return Ok(false),
@@ -115,22 +137,13 @@ impl OnSettlementEventUpdater {
             .transaction(hash)
             .await?
             .with_context(|| format!("no tx {hash:?}"))?;
-        let tx_from = transaction
-            .from
-            .with_context(|| format!("no from {hash:?}"))?;
-        let tx_nonce: i64 = transaction
-            .nonce
-            .try_into()
-            .map_err(|err| anyhow!("{}", err))
-            .with_context(|| format!("convert nonce {hash:?}"))?;
 
-        let auction_id = Self::recover_auction_id_from_calldata(&mut ex, &transaction).await?;
+        let auction_kind = Self::get_auction_kind(&mut ex, &transaction).await?;
 
         let mut update = SettlementUpdate {
             block_number: event.block_number,
             log_index: event.log_index,
-            tx_from,
-            tx_nonce,
+            auction_kind,
             auction_data: None,
         };
 
@@ -140,7 +153,7 @@ impl OnSettlementEventUpdater {
         //
         // If auction_id exists, we expect all other relevant information to exist as
         // well.
-        if let Some(auction_id) = auction_id {
+        if let AuctionKind::Valid { auction_id } = auction_kind {
             let receipt = self
                 .eth
                 .transaction_receipt(hash)
@@ -236,11 +249,9 @@ impl OnSettlementEventUpdater {
     /// `auction_id` to the settlement calldata. This function tries to
     /// recover that `auction_id`. This function only returns an error if
     /// retrying the operation makes sense. If all went well and there
-    /// simply is no sensible `auction_id` `Ok(None)` will be returned.
-    async fn recover_auction_id_from_calldata(
-        ex: &mut PgConnection,
-        tx: &Transaction,
-    ) -> Result<Option<i64>> {
+    /// simply is no sensible `auction_id` `AuctionKind::Invalid` will be
+    /// returned.
+    async fn get_auction_kind(ex: &mut PgConnection, tx: &Transaction) -> Result<AuctionKind> {
         let tx_from = tx.from.context("tx is missing sender")?;
         let metadata = match DecodedSettlement::new(&tx.input.0) {
             Ok(settlement) => settlement.metadata,
@@ -250,45 +261,39 @@ impl OnSettlementEventUpdater {
                     ?err,
                     "could not decode settlement tx, unclear which auction it belongs to"
                 );
-                return Ok(None);
+                return Ok(AuctionKind::Invalid);
             }
         };
         let auction_id = match metadata {
             Some(bytes) => i64::from_be_bytes(bytes.0),
             None => {
                 tracing::warn!(?tx, "could not recover the auction_id from the calldata");
-                return Ok(None);
+                return Ok(AuctionKind::Invalid);
             }
         };
 
         let score = database::settlement_scores::fetch(ex, auction_id).await?;
-        let data_already_recorded =
-            database::auction_transaction::data_exists(ex, auction_id).await?;
-        match (score, data_already_recorded) {
-            (None, _) => {
+        match score {
+            None => {
                 tracing::debug!(
                     auction_id,
                     "calldata claims to settle auction that has no competition"
                 );
-                Ok(None)
+                Ok(AuctionKind::Invalid)
             }
-            (Some(score), _) if score.winner.0 != tx_from.0 => {
-                tracing::warn!(
-                    auction_id,
-                    ?tx_from,
-                    winner = ?score.winner,
-                    "solution submitted by solver other than the winner"
-                );
-                Ok(None)
-            }
-            (Some(_), true) => {
-                tracing::warn!(
-                    auction_id,
-                    "settlement data already recorded for this auction"
-                );
-                Ok(None)
+            Some(score) => {
+                if score.winner.0 != tx_from.0 {
+                    tracing::warn!(
+                        auction_id,
+                        ?tx_from,
+                        winner = ?score.winner,
+                        "solution submitted by solver other than the winner"
+                    );
+                    Ok(AuctionKind::Invalid)
+                } else {
+                    Ok(AuctionKind::Valid { auction_id })
+                }
             }
-            (Some(_), false) => Ok(Some(auction_id)),
         }
     }
 }