fix(node): preconf watcher watch latest block + 1 (#382)

Author: kariy

crates/node/src/full/pending/mod.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 use std::time::Duration;
 
+use anyhow::{anyhow, Result};
 use katana_gateway_client::Client;
 use katana_gateway_types::{ConfirmedTransaction, ErrorCode, PreConfirmedBlock, StateDiff};
 use katana_pipeline::PipelineBlockSubscription;
@@ -45,7 +46,13 @@ impl PreconfStateFactory {
             shared_preconf_block: shared_preconf_block.clone(),
         };
 
-        tokio::spawn(async move { worker.run().await });
+        tokio::spawn(async move {
+            loop {
+                if let Err(error) = worker.run().await {
+                    error!(%error, "PreconfBlockWatcher returned with an error.");
+                }
+            }
+        });
 
         Self { gateway_client, latest_synced_block, shared_preconf_block, storage_provider }
     }
@@ -75,12 +82,12 @@ impl PreconfStateFactory {
             .map(|preconf_data| preconf_data.preconf_state_updates.clone())
     }
 
-    pub fn block(&self) -> Option<PreConfirmedBlock> {
+    pub fn block(&self) -> Option<(BlockNumber, PreConfirmedBlock)> {
         self.shared_preconf_block
             .inner
             .lock()
             .as_ref()
-            .map(|preconf_data| preconf_data.preconf_block.clone())
+            .map(|preconf_data| (preconf_data.preconf_block_id, preconf_data.preconf_block.clone()))
     }
 
     pub fn transactions(&self) -> Option<Vec<ConfirmedTransaction>> {
@@ -104,8 +111,9 @@ struct PreconfBlockData {
     preconf_state_updates: StateUpdates,
 }
 
-const DEFAULT_INTERVAL: Duration = Duration::from_millis(500);
+const DEFAULT_INTERVAL: Duration = Duration::from_millis(1000);
 
+#[derive(Debug)]
 struct PreconfBlockWatcher {
     interval: Duration,
     gateway_client: Client,
@@ -120,7 +128,7 @@ struct PreconfBlockWatcher {
 }
 
 impl PreconfBlockWatcher {
-    async fn run(&mut self) {
+    async fn run(&mut self) -> Result<()> {
         let mut current_preconf_block_num =
             self.latest_synced_block.block().map(|b| b + 1).unwrap_or(0);
 
@@ -160,32 +168,28 @@ impl PreconfBlockWatcher {
                     // chain's tip, in which case we just skip to the next
                     // iteration.
                     Err(katana_gateway_client::Error::Sequencer(error))
-                        if error.code == ErrorCode::BlockNotFound =>
-                    {
-                        continue
-                    }
+                        if error.code == ErrorCode::BlockNotFound => {}
 
-                    Err(err) => panic!("{err}"),
+                    Err(err) => return Err(anyhow!(err)),
+                }
+            } else {
+                if let Err(err) = self.latest_synced_block.changed().await {
+                    error!(error = ?err, "Error receiving latest block number.");
+                    break;
                 }
-            }
-
-            tokio::select! {
-                biased;
 
-                res = self.latest_synced_block.changed() => {
-                    if let Err(err) = res {
-                        error!(error = ?err, "Error receiving latest block number.");
-                        break;
-                    }
+                // reset preconf state
+                *self.shared_preconf_block.inner.lock() = None;
 
-                    let latest_synced_block_num = self.latest_synced_block.block().unwrap();
-                    current_preconf_block_num = latest_synced_block_num + 1;
-                }
+                let latest_synced_block_num = self.latest_synced_block.block().unwrap_or(0);
+                current_preconf_block_num = latest_synced_block_num + 1;
 
-                _ = tokio::time::sleep(self.interval) => {
-                    current_preconf_block_num += 1;
-                }
+                continue;
             }
+
+            tokio::time::sleep(self.interval).await
         }
+
+        Ok(())
     }
 }

crates/node/src/full/pending/provider.rs

@@ -15,7 +15,7 @@ impl PendingBlockProvider for PreconfStateFactory {
     fn get_pending_block_with_txs(
         &self,
     ) -> StarknetApiResult<Option<katana_rpc_types::PreConfirmedBlockWithTxs>> {
-        if let Some(block) = self.block() {
+        if let Some((block_number, block)) = self.block() {
             let transactions = block
                 .transactions
                 .clone()
@@ -26,7 +26,7 @@ impl PendingBlockProvider for PreconfStateFactory {
 
             Ok(Some(katana_rpc_types::PreConfirmedBlockWithTxs {
                 transactions,
-                block_number: 0,
+                block_number,
                 l1_da_mode: block.l1_da_mode,
                 l1_gas_price: block.l1_gas_price,
                 l2_gas_price: block.l2_gas_price,
@@ -43,10 +43,10 @@ impl PendingBlockProvider for PreconfStateFactory {
     fn get_pending_block_with_receipts(
         &self,
     ) -> StarknetApiResult<Option<katana_rpc_types::PreConfirmedBlockWithReceipts>> {
-        if let Some(block) = self.block() {
+        if let Some((block_number, block)) = self.block() {
             Ok(Some(katana_rpc_types::PreConfirmedBlockWithReceipts {
                 transactions: Vec::new(),
-                block_number: 0,
+                block_number,
                 l1_da_mode: block.l1_da_mode,
                 l1_gas_price: block.l1_gas_price,
                 l2_gas_price: block.l2_gas_price,
@@ -63,7 +63,7 @@ impl PendingBlockProvider for PreconfStateFactory {
     fn get_pending_block_with_tx_hashes(
         &self,
     ) -> StarknetApiResult<Option<katana_rpc_types::PreConfirmedBlockWithTxHashes>> {
-        if let Some(block) = self.block() {
+        if let Some((block_number, block)) = self.block() {
             let transactions = block
                 .transactions
                 .clone()
@@ -73,7 +73,7 @@ impl PendingBlockProvider for PreconfStateFactory {
 
             Ok(Some(katana_rpc_types::PreConfirmedBlockWithTxHashes {
                 transactions,
-                block_number: 0,
+                block_number,
                 l1_da_mode: block.l1_da_mode,
                 l1_gas_price: block.l1_gas_price,
                 l2_gas_price: block.l2_gas_price,
@@ -91,7 +91,7 @@ impl PendingBlockProvider for PreconfStateFactory {
         &self,
         hash: TxHash,
     ) -> StarknetApiResult<Option<katana_rpc_types::TxReceiptWithBlockInfo>> {
-        if let Some(preconf_block) = self.block() {
+        if let Some((block_number, preconf_block)) = self.block() {
             let receipt = preconf_block
                 .transaction_receipts
                 .iter()
@@ -109,7 +109,7 @@ impl PendingBlockProvider for PreconfStateFactory {
 
             let status = FinalityStatus::PreConfirmed;
             let transaction_hash = receipt.transaction_hash;
-            let block = ReceiptBlockInfo::PreConfirmed { block_number: 0 };
+            let block = ReceiptBlockInfo::PreConfirmed { block_number };
 
             let receipt = match r#type {
                 TxType::Invoke => {