Ignore failed transactions (#19)

* (fix) Ignore failed transactions.
(feat) An `end_block` option to limit history replay for special situations

* (fix) Limit to_block by the latest finalized block

* (fix) Fix formatting

bridge-validators/config.toml

@@ -7,6 +7,13 @@
 [[chain_configs]]
 ## The block at which history replay should start.
 chain_gateway_block_deployed = 36696045
+
+## When replaying history, we replay dispatches first, then relays. In order to do this
+## on old blocks in a reasonable time, we should not only start just before the message to be
+## replayed, but stop just after. This allows us to stop at a particular block when catching
+## up
+## to_block_number = 4000000
+
 ## RPC URL to talk to for this chain.
 rpc_url = "https://bsc-prebsc-dataseed.bnbchain.org"
 ## The (EVM) address of the chain gateway contract

bridge-validators/src/block.rs

@@ -11,7 +11,7 @@ use ethers::{
 use ethers_contract::{parse_log, EthEvent};
 use futures::{Stream, StreamExt, TryStreamExt};
 use tokio::time::interval;
-use tracing::{debug, info, warn};
+use tracing::{info, warn};
 
 use crate::client::{ChainClient, LogStrategy};
 
@@ -50,7 +50,7 @@ impl ChainClient {
                 // go through all the transactions
                 for txn_hash in block.transactions {
                     // We have a transaction. Did it have any logs?
-                    debug!("block {} txn {:#x}", block_number, txn_hash);
+                    info!("block {} txn {:#x}", block_number, txn_hash);
                     // Get the receipt
                     let maybe_receipt = self
                         .client
@@ -59,6 +59,16 @@ impl ChainClient {
                         .await?;
                     if let Some(receipt) = maybe_receipt {
                         // Yay!
+                        if let Some(v) = &receipt.status {
+                            info!("[1] txn {:#x} has status {v}", txn_hash);
+                            if *v != U64::from(1) {
+                                info!("[1] txn failed - skipping");
+                                continue;
+                            }
+                        } else {
+                            info!("[1] txn {:#x} has no status - ignoring", txn_hash);
+                            continue;
+                        }
                         info!("Got receipt for txn {:#x}", txn_hash);
                         for log in receipt.logs {
                             // Because FML, the filter doesn't actually include the address.
@@ -101,6 +111,7 @@ impl ChainClient {
                                 // If there's no filter element for this topic, we're fine.
                             }
                             if matches {
+                                info!("Event matches; pushing for transit");
                                 result.push(log);
                             }
                         }
@@ -153,6 +164,19 @@ impl BlockPolling for ChainClient {
                     .request("eth_getLogs", [event])
                     .await?;
                 logs.into_iter()
+                    // Zilliqa 1 will generate logs for failed txns.
+                    .filter(|log| {
+                        log.get("status")
+                            .and_then(|v| v.as_i64())
+                            .map_or(false, |s| {
+                                if s != 1 {
+                                    info!("txn failed: status = {s:#x}");
+                                    false
+                                } else {
+                                    true
+                                }
+                            })
+                    })
                     .filter(|log| {
                         log.get("address")
                             .and_then(|val| val.as_str())

bridge-validators/src/bridge_node.rs

@@ -86,34 +86,43 @@ impl BridgeNode {
     }
 
     pub async fn sync_historic_events(&mut self) -> Result<()> {
-        let to_block = if self.chain_client.block_instant_finality {
+        let max_block_specifier = if self.chain_client.block_instant_finality {
             BlockNumber::Latest
         } else {
             BlockNumber::Finalized
         };
-        info!(
-            "Getting Historic Events for chainId#{}: {}",
-            self.chain_client.chain_id, to_block
-        );
-
-        let to_block_number = self
+        let finalized_block_number = self
             .chain_client
             .client
-            .get_block(to_block)
+            .get_block(max_block_specifier)
             .await?
             .expect("Latest finalized block should be retrieved")
             .number
             .expect("Number should be here");
 
+        let to_block_number = if let Some(v) = self.chain_client.to_block_number {
+            if v > finalized_block_number.as_u64() {
+                warn!("to_block in config file {} was greater than latest finalized block {} - will terminate at {}",
+                      v, finalized_block_number, v);
+            }
+            std::cmp::min(v, finalized_block_number.as_u64())
+        } else {
+            finalized_block_number.as_u64()
+        };
+
+        info!(
+            "Getting Historic Events for chainId#{} from blk# {} to blk# {}",
+            self.chain_client.chain_id,
+            self.chain_client.chain_gateway_block_deployed,
+            to_block_number
+        );
+
         dbg!(to_block_number);
 
         let chain_gateway: ChainGateway<Client> = self.chain_client.get_contract();
 
         let dispatch_events = self
-            .get_historic_events(
-                chain_gateway.event::<DispatchedFilter>(),
-                to_block_number.as_u64(),
-            )
+            .get_historic_events(chain_gateway.event::<DispatchedFilter>(), to_block_number)
             .await?;
         info!("  .. dispatch_events: {}", dispatch_events.len());
 
@@ -122,10 +131,7 @@ impl BridgeNode {
         }
 
         let relay_events = self
-            .get_historic_events(
-                chain_gateway.event::<RelayedFilter>(),
-                to_block_number.as_u64(),
-            )
+            .get_historic_events(chain_gateway.event::<RelayedFilter>(), to_block_number)
             .await?;
 
         info!("  .. relay_events: {}", relay_events.len());

bridge-validators/src/client.rs

@@ -35,6 +35,7 @@ pub struct ChainClient {
     pub legacy_gas_estimation_percent: Option<u64>,
     pub scan_behind_blocks: u64,
     pub log_strategy: LogStrategy,
+    pub to_block_number: Option<u64>,
 }
 
 impl fmt::Display for ChainClient {
@@ -80,6 +81,7 @@ impl ChainClient {
             legacy_gas_estimation_percent: config.legacy_gas_estimation_percent,
             scan_behind_blocks: config.scan_behind_blocks.unwrap_or_default(),
             log_strategy: strategy,
+            to_block_number: config.to_block_number,
         })
     }
 }

bridge-validators/src/main.rs

@@ -34,6 +34,7 @@ pub struct ChainConfig {
     pub legacy_gas_estimation_percent: Option<u64>,
     pub scan_behind_blocks: Option<u64>,
     pub use_get_transactions: Option<bool>,
+    pub to_block_number: Option<u64>,
 }
 
 #[derive(Debug, Clone, Deserialize)]