Merge pull request #329 from propeller-heads/ah/ENG-4993/handle-bad-change-type-deltas

feat: Propagate db update errors

Author: kayibal

src/evm/decoder.rs

@@ -214,7 +214,7 @@ where
 
     /// Decodes a `FeedMessage` into a `BlockUpdate` containing the updated states of protocol
     /// components
-    pub async fn decode(&self, msg: FeedMessage<H>) -> Result<Update, StreamDecodeError> {
+    pub async fn decode(&self, msg: &FeedMessage<H>) -> Result<Update, StreamDecodeError> {
         // stores all states updated in this tick/msg
         let mut updated_states = HashMap::new();
         let mut new_pairs = HashMap::new();
@@ -377,7 +377,8 @@ where
                     Some(storage_by_address),
                     token_proxy_accounts,
                 )
-                .await;
+                .await
+                .map_err(|e| StreamDecodeError::Fatal(e.to_string()))?;
                 info!("Engine updated");
                 drop(state_guard);
             }
@@ -437,6 +438,8 @@ where
                                         continue 'outer;
                                     }
                                 };
+                                // TODO: Ok we deployed a proxy whenever we see a new token without
+                                // a implementation set.
                                 if !state_guard
                                     .proxy_token_addresses
                                     .contains_key(&token_address)
@@ -474,7 +477,8 @@ where
                             None,
                             new_tokens_accounts,
                         )
-                        .await;
+                        .await
+                        .map_err(|e| StreamDecodeError::Fatal(e.to_string()))?;
                     }
 
                     // collect contracts:ids mapping for states that should update on contract
@@ -562,6 +566,9 @@ where
                     .map(|(key, value)| {
                         let mut update: AccountUpdate = value.clone().into();
 
+                        // TODO: Unify this different initialisation if we receive
+                        //  state updates for the token with the usual case. Also
+                        //  switch the if cases.
                         if state_guard.tokens.contains_key(key) {
                             // If the account is a token, we need to handle it with a proxy contract
 
@@ -599,6 +606,21 @@ where
                                     .unwrap()
                             };
 
+                            // TEMP PATCH (ENG-4993)
+                            //
+                            // The indexer emits deltas without code marked as creations,
+                            // which crashes TychoDB. Until fixed, treat them as updates
+                            // (since EVM code cannot be deleted).
+                            if update.code.is_none() &&
+                                matches!(update.change, ChangeType::Creation)
+                            {
+                                error!(
+                                    update = ?update,
+                                    "FaultyCreationDelta"
+                                );
+                                update.change = ChangeType::Update;
+                            }
+
                             // assign original token contract to new address
                             update.address = proxy_addr;
                         };
@@ -617,7 +639,8 @@ where
                     None,
                     token_proxy_accounts,
                 )
-                .await;
+                .await
+                .map_err(|e| StreamDecodeError::Fatal(e.to_string()))?;
                 info!("Engine updated");
 
                 // Collect all pools related to the updated accounts
@@ -771,7 +794,7 @@ where
         // Send the tick with all updated states
         Ok(Update::new(header.block_number_or_timestamp(), updated_states, new_pairs)
             .set_removed_pairs(removed_pairs)
-            .set_sync_states(msg.sync_states))
+            .set_sync_states(msg.sync_states.clone()))
     }
 
     fn apply_update(
@@ -975,12 +998,12 @@ mod tests {
 
         let msg = load_test_msg("uniswap_v2_snapshot");
         let res1 = decoder
-            .decode(msg)
+            .decode(&msg)
             .await
             .expect("decode failure");
         let msg = load_test_msg("uniswap_v2_delta");
         let res2 = decoder
-            .decode(msg)
+            .decode(&msg)
             .await
             .expect("decode failure");
 
@@ -1007,7 +1030,7 @@ mod tests {
 
         let msg = load_test_msg("uniswap_v2_snapshot");
         let res1 = decoder
-            .decode(msg)
+            .decode(&msg)
             .await
             .expect("decode failure");
 
@@ -1023,7 +1046,7 @@ mod tests {
         decoder.skip_state_decode_failures = skip_failures;
 
         let msg = load_test_msg("uniswap_v2_snapshot_broken_id");
-        match decoder.decode(msg).await {
+        match decoder.decode(&msg).await {
             Err(StreamDecodeError::Fatal(msg)) => {
                 if !skip_failures {
                     assert_eq!(
@@ -1053,7 +1076,7 @@ mod tests {
         decoder.skip_state_decode_failures = skip_failures;
 
         let msg = load_test_msg("uniswap_v2_snapshot_broken_state");
-        match decoder.decode(msg).await {
+        match decoder.decode(&msg).await {
             Err(StreamDecodeError::Fatal(msg)) => {
                 if !skip_failures {
                     assert_eq!(msg, "Missing attributes reserve0");
@@ -1119,7 +1142,7 @@ mod tests {
 
         // Decode the message
         let _ = decoder
-            .decode(msg)
+            .decode(&msg)
             .await
             .expect("decode failure");
 

src/evm/engine_db/mod.rs

@@ -7,7 +7,10 @@ use tycho_client::feed::BlockHeader;
 use tycho_common::simulation::errors::SimulationError;
 
 use crate::evm::{
-    engine_db::{engine_db_interface::EngineDatabaseInterface, tycho_db::PreCachedDB},
+    engine_db::{
+        engine_db_interface::EngineDatabaseInterface,
+        tycho_db::{PreCachedDB, PreCachedDBError},
+    },
     simulation::SimulationEngine,
     tycho_models::{AccountUpdate, ChangeType, ResponseAccount},
 };
@@ -70,7 +73,7 @@ pub async fn update_engine(
     block: Option<BlockHeader>,
     vm_storage: Option<HashMap<Address, ResponseAccount>>,
     account_updates: HashMap<Address, AccountUpdate>,
-) -> Vec<AccountUpdate> {
+) -> Result<Vec<AccountUpdate>, PreCachedDBError> {
     if let Some(block) = block {
         let mut vm_updates: Vec<AccountUpdate> = Vec::new();
 
@@ -93,11 +96,11 @@ pub async fn update_engine(
         }
 
         if !vm_updates.is_empty() {
-            db.update(vm_updates.clone(), Some(block));
+            db.update(vm_updates.clone(), Some(block))?;
         }
 
-        vm_updates
+        Ok(vm_updates)
     } else {
-        vec![]
+        Ok(vec![])
     }
 }

src/evm/engine_db/tycho_db.rs

@@ -36,6 +36,8 @@ pub enum TychoClientError {
 pub enum PreCachedDBError {
     #[error("Account {0} not found")]
     MissingAccount(Address),
+    #[error("Bad account update: {0} - {1:?}")]
+    BadUpdate(String, Box<AccountUpdate>),
     #[error("Block needs to be set")]
     BlockNotSet(),
     #[error("Tycho Client error: {0}")]
@@ -76,10 +78,17 @@ impl PreCachedDB {
     }
 
     #[instrument(skip_all)]
-    pub fn update(&self, account_updates: Vec<AccountUpdate>, block: Option<BlockHeader>) {
+    pub fn update(
+        &self,
+        account_updates: Vec<AccountUpdate>,
+        block: Option<BlockHeader>,
+    ) -> Result<(), PreCachedDBError> {
         // Hold the write lock for the duration of the function so that no other thread can
         // write to the storage.
-        let mut write_guard = self.inner.write().unwrap();
+        let mut write_guard = self
+            .inner
+            .write()
+            .expect("Fatal tycho state db lock poisoned");
 
         write_guard.block = block;
 
@@ -108,10 +117,13 @@ impl PreCachedDB {
 
                     // We expect the code to be present.
                     let code = Bytecode::new_raw(AlloyBytes::from(
-                        update.code.clone().unwrap_or_else(|| {
+                        update.code.clone().ok_or_else(|| {
                             error!(%update.address, "MissingCode");
-                            Vec::new()
-                        }),
+                            PreCachedDBError::BadUpdate(
+                                "MissingCode".into(),
+                                Box::new(update.clone()),
+                            )
+                        })?,
                     ));
                     // If the balance is not present, we set it to zero.
                     let balance = update.balance.unwrap_or(U256::ZERO);
@@ -130,6 +142,7 @@ impl PreCachedDB {
                 }
             }
         }
+        Ok(())
     }
 
     /// Retrieves the storage value at the specified index for the given account, if it exists.
@@ -552,7 +565,9 @@ mod tests {
             ..Default::default()
         };
 
-        mock_db.update(vec![account_update], Some(new_block));
+        mock_db
+            .update(vec![account_update], Some(new_block))
+            .unwrap();
 
         let account_info = mock_db
             .basic_ref(Address::from_str("0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D").unwrap())

src/evm/protocol/vm/state.rs

@@ -790,7 +790,8 @@ mod tests {
                 false,
             );
         }
-        db.update(accounts, Some(block));
+        db.update(accounts, Some(block))
+            .unwrap();
 
         let tokens = vec![dai().address, bal().address];
         for token in &tokens {

src/evm/protocol/vm/tycho_decoder.rs

@@ -311,7 +311,8 @@ mod tests {
                 false,
             );
         }
-        db.update(accounts, Some(block));
+        db.update(accounts, Some(block))
+            .unwrap();
         let account_balances = HashMap::from([(
             Bytes::from("0xBA12222222228d8Ba445958a75a0704d566BF2C8"),
             HashMap::from([

src/evm/stream.rs

@@ -2,7 +2,7 @@ use std::{collections::HashMap, sync::Arc};
 
 use futures::{Stream, StreamExt};
 use tokio_stream::wrappers::ReceiverStream;
-use tracing::warn;
+use tracing::{debug, warn};
 use tycho_client::{
     feed::{component_tracker::ComponentFilter, synchronizer::ComponentWithState, BlockHeader},
     stream::{StreamError, TychoStreamBuilder},
@@ -153,7 +153,12 @@ impl ProtocolStreamBuilder {
             let decoder = decoder.clone(); // Clone the decoder for the closure
             move |msg| {
                 let decoder = decoder.clone(); // Clone again for the async block
-                async move { decoder.decode(msg).await }
+                async move {
+                    decoder.decode(&msg).await.map_err(|e| {
+                        debug!(msg=?msg, "Decode error: {}", e);
+                        e
+                    })
+                }
             }
         })))
     }

src/rfq/stream.rs

@@ -68,7 +68,7 @@ impl RFQStreamBuilder {
                 Ok((provider, msg)) => {
                     let update = self
                         .decoder
-                        .decode(FeedMessage {
+                        .decode(&FeedMessage {
                             state_msgs: HashMap::from([(provider.clone(), msg)]),
                             sync_states: HashMap::new(),
                         })

tycho_simulation_py/src/structs_py.rs

@@ -537,6 +537,7 @@ impl TychoDB {
 
         self_
             .inner
-            .update(account_updates, block);
+            .update(account_updates, block)
+            .unwrap();
     }
 }