Add L2 block_id monitoring

Node/src/node/l2_block_id.rs

@@ -0,0 +1,78 @@
+use std::sync::atomic::{AtomicU64, Ordering};
+
+pub struct L2BlockId {
+    block_id: AtomicU64,
+}
+
+impl L2BlockId {
+    pub fn new() -> Self {
+        Self {
+            block_id: AtomicU64::new(0),
+        }
+    }
+
+    pub fn update(&self, block_id: u64) {
+        let mut current_block_id = self.block_id.load(Ordering::Acquire);
+
+        while block_id > current_block_id {
+            match self.block_id.compare_exchange(
+                current_block_id,
+                block_id,
+                Ordering::Release,
+                Ordering::Acquire,
+            ) {
+                Ok(_) => break,
+                Err(previous) => current_block_id = previous,
+            }
+        }
+    }
+
+    pub fn next(&self, block_id: u64) -> u64 {
+        let mut current_block_id = self.block_id.load(Ordering::Acquire);
+        let mut new_block_id = std::cmp::max(block_id + 1, current_block_id + 1);
+        while new_block_id > current_block_id {
+            match self.block_id.compare_exchange(
+                current_block_id,
+                new_block_id,
+                Ordering::Release,
+                Ordering::Acquire,
+            ) {
+                Ok(_) => break,
+                Err(previous) => {
+                    current_block_id = previous;
+                    new_block_id = std::cmp::max(new_block_id, current_block_id + 1);
+                }
+            }
+        }
+        new_block_id
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_next() {
+        let l2_block_id = L2BlockId::new();
+
+        assert_eq!(l2_block_id.next(1), 2);
+        assert_eq!(l2_block_id.next(0), 3);
+        assert_eq!(l2_block_id.next(1), 4);
+        assert_eq!(l2_block_id.next(7), 8);
+        assert_eq!(l2_block_id.next(8), 9);
+        assert_eq!(l2_block_id.next(4), 10);
+    }
+
+    #[test]
+    fn test_update() {
+        let l2_block_id = L2BlockId::new();
+
+        l2_block_id.update(1);
+        assert_eq!(l2_block_id.block_id.load(Ordering::SeqCst), 1);
+        l2_block_id.update(10);
+        assert_eq!(l2_block_id.block_id.load(Ordering::SeqCst), 10);
+        l2_block_id.update(5);
+        assert_eq!(l2_block_id.block_id.load(Ordering::SeqCst), 10);
+    }
+}

Node/src/node/mod.rs

@@ -1,5 +1,6 @@
 pub mod block_proposed_receiver;
 mod commit;
+mod l2_block_id;
 pub mod lookahead_monitor;
 pub mod lookahead_updated_receiver;
 mod operator;
@@ -16,6 +17,7 @@ use crate::{
 };
 use anyhow::Error;
 use commit::L2TxListsCommit;
+use l2_block_id::L2BlockId;
 use operator::{Operator, Status as OperatorStatus};
 use preconfirmation_helper::PreconfirmationHelper;
 use preconfirmation_message::PreconfirmationMessage;
@@ -32,7 +34,7 @@ use tokio::sync::{
     Mutex,
 };
 use tokio::time::{sleep, Duration};
-use tracing::{debug, error, info};
+use tracing::{debug, error, info, warn};
 
 const OLDEST_BLOCK_DISTANCE: u64 = 256;
 
@@ -54,6 +56,7 @@ pub struct Node {
     preconfirmation_helper: PreconfirmationHelper,
     bls_service: Arc<BLSService>,
     always_push_lookahead: bool,
+    l2_block_id: Arc<L2BlockId>,
 }
 
 impl Node {
@@ -86,6 +89,7 @@ impl Node {
             preconfirmation_helper: PreconfirmationHelper::new(),
             bls_service,
             always_push_lookahead,
+            l2_block_id: Arc::new(L2BlockId::new()),
         })
     }
 
@@ -104,6 +108,7 @@ impl Node {
         let taiko = self.taiko.clone();
         let is_preconfer_now = self.is_preconfer_now.clone();
         let preconfirmation_txs = self.preconfirmation_txs.clone();
+        let l2_block_id = self.l2_block_id.clone();
         if let (Some(node_rx), Some(p2p_to_node_rx)) = (
             self.node_block_proposed_rx.take(),
             self.p2p_to_node_rx.take(),
@@ -117,6 +122,7 @@ impl Node {
                     taiko,
                     is_preconfer_now,
                     preconfirmation_txs,
+                    l2_block_id,
                 )
                 .await;
             });
@@ -133,6 +139,7 @@ impl Node {
         taiko: Arc<Taiko>,
         is_preconfer_now: Arc<AtomicBool>,
         preconfirmation_txs: Arc<Mutex<HashMap<u64, Vec<u8>>>>,
+        l2_block_id: Arc<L2BlockId>,
     ) {
         loop {
             tokio::select! {
@@ -154,9 +161,10 @@ impl Node {
                     if !is_preconfer_now.load(Ordering::Acquire) {
                         debug!("Received Message from p2p!");
                         let msg: PreconfirmationMessage = p2p_message.into();
+                        l2_block_id.update(msg.block_height);
                         Self::advance_l2_head(msg, &preconfirmed_blocks, ethereum_l1.clone(), taiko.clone()).await;
                     } else {
-                        debug!("Node is Preconfer and received message from p2p: {:?}", p2p_message);
+                        warn!("Node is Preconfer and received message from p2p: {:?}", p2p_message);
                     }
                 }
             }
@@ -480,9 +488,7 @@ impl Node {
             pending_tx_lists.tx_list_bytes[0].clone() // TODO: handle multiple tx lists
         };
 
-        let new_block_height = self
-            .preconfirmation_helper
-            .get_new_block_id(pending_tx_lists.parent_block_id);
+        let new_block_height = self.l2_block_id.next(pending_tx_lists.parent_block_id);
         debug!("Preconfirming block with the height: {}", new_block_height);
 
         let (commit_hash, signature) =

Node/src/node/preconfirmation_helper.rs

@@ -1,14 +1,10 @@
 pub struct PreconfirmationHelper {
     nonce: u64,
-    last_block_id: u64,
 }
 
 impl PreconfirmationHelper {
     pub fn new() -> Self {
-        Self {
-            nonce: 0,
-            last_block_id: 0,
-        }
+        Self { nonce: 0 }
     }
 
     pub fn init(&mut self, nonce: u64) {
@@ -24,31 +20,4 @@ impl PreconfirmationHelper {
     pub fn increment_nonce(&mut self) {
         self.nonce += 1;
     }
-
-    pub fn get_new_block_id(&mut self, parent_block_id: u64) -> u64 {
-        let mut new_block_id = parent_block_id + 1;
-        if self.last_block_id >= new_block_id {
-            new_block_id = self.last_block_id + 1;
-        }
-        self.last_block_id = new_block_id;
-        new_block_id
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_get_new_block_id() {
-        let mut helper = PreconfirmationHelper::new();
-        assert_eq!(helper.get_new_block_id(0), 1);
-        assert_eq!(helper.get_new_block_id(0), 2);
-        assert_eq!(helper.get_new_block_id(0), 3);
-        assert_eq!(helper.get_new_block_id(0), 4);
-        assert_eq!(helper.get_new_block_id(4), 5);
-        assert_eq!(helper.get_new_block_id(4), 6);
-        assert_eq!(helper.get_new_block_id(4), 7);
-        assert_eq!(helper.get_new_block_id(4), 8);
-    }
 }