fix(cardano-chain-follower): use Network, MultiEraBlock, Point from cardano-blockchain-types

Signed-off-by: bkioshn <[email protected]>

Author: bkioshn

rust/cardano-chain-follower/Cargo.toml

@@ -21,6 +21,7 @@ mithril-client = { version = "0.10.4", default-features = false, features = [
 ] }
 
 rbac-registration = { version = "0.0.2", git = "https://github.com/input-output-hk/catalyst-libs.git", tag = "v0.0.8" }
+cardano-blockchain-types = { git = "https://github.com/input-output-hk/catalyst-libs.git", branch = "feat/cardano-blockchain-types" }
 
 thiserror = "1.0.64"
 tokio = { version = "1.40.0", features = [

rust/cardano-chain-follower/src/chain_sync.rs

@@ -6,6 +6,7 @@
 use std::time::Duration;
 
 use anyhow::Context;
+use cardano_blockchain_types::{Fork, MultiEraBlock, Network, Point};
 use pallas::{
     ledger::traverse::MultiEraHeader,
     network::{
@@ -32,8 +33,7 @@ use crate::{
     error::{Error, Result},
     mithril_snapshot_config::MithrilUpdateMessage,
     mithril_snapshot_data::latest_mithril_snapshot_id,
-    point::{TIP_POINT, UNKNOWN_POINT},
-    stats, ChainSyncConfig, MultiEraBlock, Network, Point, ORIGIN_POINT,
+    stats, ChainSyncConfig,
 };
 
 /// The maximum number of seconds we wait for a node to connect.
@@ -88,7 +88,7 @@ async fn retry_connect(
 
 /// Purge the live chain, and intersect with TIP.
 async fn purge_and_intersect_tip(client: &mut PeerClient, chain: Network) -> Result<Point> {
-    if let Err(error) = purge_live_chain(chain, &TIP_POINT) {
+    if let Err(error) = purge_live_chain(chain, &Point::TIP) {
         // Shouldn't happen.
         error!("failed to purge live chain: {error}");
     }
@@ -122,16 +122,16 @@ async fn resync_live_tip(client: &mut PeerClient, chain: Network) -> Result<Poin
 
 /// Fetch a single block from the Peer, and Decode it.
 async fn fetch_block_from_peer(
-    peer: &mut PeerClient, chain: Network, point: Point, previous_point: Point, fork_count: u64,
+    peer: &mut PeerClient, chain: Network, point: Point, previous_point: Point, fork: Fork,
 ) -> anyhow::Result<MultiEraBlock> {
     let block_data = peer
         .blockfetch()
         .fetch_single(point.clone().into())
         .await
         .with_context(|| "Fetching block data")?;
 
-    debug!("{chain}, {previous_point}, {fork_count}");
-    let live_block_data = MultiEraBlock::new(chain, block_data, &previous_point, fork_count)?;
+    debug!("{chain}, {previous_point}, {fork:?}");
+    let live_block_data = MultiEraBlock::new(chain, block_data, &previous_point, fork)?;
 
     Ok(live_block_data)
 }
@@ -141,16 +141,16 @@ async fn fetch_block_from_peer(
 /// Fetch the rollback block, and try and insert it into the live-chain.
 /// If its a real rollback, it will purge the chain ahead of the block automatically.
 async fn process_rollback_actual(
-    peer: &mut PeerClient, chain: Network, point: Point, tip: &Tip, fork_count: &mut u64,
+    peer: &mut PeerClient, chain: Network, point: Point, tip: &Tip, fork: &mut Fork,
 ) -> anyhow::Result<Point> {
     debug!("RollBackward: {:?} {:?}", point, tip);
 
     // Check if the block is in the live chain, if it is, re-add it, which auto-purges the
     // rest of live chain tip. And increments the fork count.
     if let Some(mut block) = get_live_block(chain, &point, 0, true) {
         // Even though we are re-adding the known block, increase the fork count.
-        block.set_fork(*fork_count);
-        live_chain_add_block_to_tip(chain, block, fork_count, tip.0.clone().into())?;
+        block.set_fork(*fork);
+        live_chain_add_block_to_tip(chain, block, fork, tip.0.clone().into())?;
         return Ok(point);
     }
 
@@ -165,7 +165,7 @@ async fn process_rollback_actual(
     let previous_point = if let Some(previous_block) = previous_block {
         let previous = previous_block.previous();
         debug!("Previous block: {:?}", previous);
-        if previous == ORIGIN_POINT {
+        if previous == Point::ORIGIN {
             latest_mithril_snapshot_id(chain).tip()
         } else {
             previous
@@ -175,9 +175,8 @@ async fn process_rollback_actual(
         latest_mithril_snapshot_id(chain).tip()
     };
     debug!("Previous point: {:?}", previous_point);
-    let block =
-        fetch_block_from_peer(peer, chain, point.clone(), previous_point, *fork_count).await?;
-    live_chain_add_block_to_tip(chain, block, fork_count, tip.0.clone().into())?;
+    let block = fetch_block_from_peer(peer, chain, point.clone(), previous_point, *fork).await?;
+    live_chain_add_block_to_tip(chain, block, fork, tip.0.clone().into())?;
 
     // Next block we receive is a rollback.
     Ok(point)
@@ -186,7 +185,7 @@ async fn process_rollback_actual(
 /// Process a rollback detected from the peer.
 async fn process_rollback(
     peer: &mut PeerClient, chain: Network, point: Point, tip: &Tip, previous_point: &Point,
-    fork_count: &mut u64,
+    fork: &mut Fork,
 ) -> anyhow::Result<Point> {
     let rollback_slot = point.slot_or_default();
     let head_slot = previous_point.slot_or_default();
@@ -199,7 +198,7 @@ async fn process_rollback(
     };
 
     // We actually do the work here...
-    let response = process_rollback_actual(peer, chain, point, tip, fork_count).await?;
+    let response = process_rollback_actual(peer, chain, point, tip, fork).await?;
 
     // We never really know how many blocks are rolled back when advised by the peer, but we
     // can work out how many slots. This function wraps the real work, so we can properly
@@ -212,7 +211,7 @@ async fn process_rollback(
 /// Process a rollback detected from the peer.
 async fn process_next_block(
     peer: &mut PeerClient, chain: Network, header: HeaderContent, tip: &Tip,
-    previous_point: &Point, fork_count: &mut u64,
+    previous_point: &Point, fork: &mut Fork,
 ) -> anyhow::Result<Point> {
     // Decode the Header of the block so we know what to fetch.
     let decoded_header = MultiEraHeader::decode(
@@ -222,7 +221,7 @@ async fn process_next_block(
     )
     .with_context(|| "Decoding Block Header")?;
 
-    let block_point = Point::new(decoded_header.slot(), decoded_header.hash().to_vec());
+    let block_point = Point::new(decoded_header.slot().into(), decoded_header.hash().into());
 
     debug!("RollForward: {block_point:?} {tip:?}");
 
@@ -231,20 +230,20 @@ async fn process_next_block(
         chain,
         block_point.clone(),
         previous_point.clone(),
-        *fork_count,
+        *fork,
     )
     .await?;
 
     let block_point = block.point();
 
     // We can't store this block because we don't know the previous one so the chain
     // would break, so just use it for previous.
-    if *previous_point == UNKNOWN_POINT {
+    if *previous_point == Point::UNKNOWN {
         // Nothing else we can do with the first block when we don't know the previous
         // one.  Just return it's point.
         debug!("Not storing the block, because we did not know the previous point.");
     } else {
-        live_chain_add_block_to_tip(chain, block, fork_count, tip.0.clone().into())?;
+        live_chain_add_block_to_tip(chain, block, fork, tip.0.clone().into())?;
     }
 
     Ok(block_point)
@@ -255,10 +254,10 @@ async fn process_next_block(
 ///
 /// We take ownership of the client because of that.
 async fn follow_chain(
-    peer: &mut PeerClient, chain: Network, fork_count: &mut u64,
+    peer: &mut PeerClient, chain: Network, fork: &mut Fork,
 ) -> anyhow::Result<()> {
     let mut update_sender = get_chain_update_tx_queue(chain).await;
-    let mut previous_point = UNKNOWN_POINT;
+    let mut previous_point = Point::UNKNOWN;
 
     loop {
         // debug!("Waiting for data from Cardano Peer Node:");
@@ -286,16 +285,15 @@ async fn follow_chain(
                 // subtracting current block height and the tip block height.
                 // IF the TIP is <= the current block height THEN we are at tip.
                 previous_point =
-                    process_next_block(peer, chain, header, &tip, &previous_point, fork_count)
-                        .await?;
+                    process_next_block(peer, chain, header, &tip, &previous_point, fork).await?;
 
                 // This update is just for followers to know to look again at their live chains for
                 // new data.
                 notify_follower(chain, &update_sender, &chain_update::Kind::Block);
             },
             chainsync::NextResponse::RollBackward(point, tip) => {
                 previous_point =
-                    process_rollback(peer, chain, point.into(), &tip, &previous_point, fork_count)
+                    process_rollback(peer, chain, point.into(), &tip, &previous_point, fork)
                         .await?;
                 // This update is just for followers to know to look again at their live chains for
                 // new data.
@@ -367,8 +365,8 @@ async fn live_sync_backfill(
 
     while let Some(block_data) = peer.blockfetch().recv_while_streaming().await? {
         // Backfilled blocks get placed in the oldest fork currently on the live-chain.
-        let block =
-            MultiEraBlock::new(cfg.chain, block_data, &previous_point, 1).with_context(|| {
+        let block = MultiEraBlock::new(cfg.chain, block_data, &previous_point, 1.into())
+            .with_context(|| {
                 format!(
                     "Failed to decode block data. previous: {previous_point:?}, range: {range_msg}"
                 )
@@ -533,7 +531,7 @@ pub(crate) async fn chain_sync(cfg: ChainSyncConfig, rx: mpsc::Receiver<MithrilU
     // Live Fill data starts at fork 1.
     // Immutable data from a mithril snapshot is fork 0.
     // Live backfill is always Fork 1.
-    let mut fork_count: u64 = 2;
+    let mut fork: Fork = Fork::from_saturating(2);
 
     loop {
         // We never have a connection if we end up around the loop, so make a new one.
@@ -551,7 +549,7 @@ pub(crate) async fn chain_sync(cfg: ChainSyncConfig, rx: mpsc::Receiver<MithrilU
         }
 
         // Note: This can ONLY return with an error, otherwise it will sync indefinitely.
-        if let Err(error) = follow_chain(&mut peer, cfg.chain, &mut fork_count).await {
+        if let Err(error) = follow_chain(&mut peer, cfg.chain, &mut fork).await {
             error!(
                 "Cardano Client {} failed to follow chain: {}: Reconnecting.",
                 cfg.relay_address, error
@@ -560,7 +558,7 @@ pub(crate) async fn chain_sync(cfg: ChainSyncConfig, rx: mpsc::Receiver<MithrilU
         }
 
         // If this returns, we are on a new fork (or assume we are)
-        fork_count += 1;
+        fork.incr();
     }
 }
 

rust/cardano-chain-follower/src/chain_sync_config.rs

@@ -6,6 +6,7 @@
 
 use std::sync::LazyLock;
 
+use cardano_blockchain_types::Network;
 use dashmap::DashMap;
 use strum::IntoEnumIterator;
 use tokio::{sync::Mutex, task::JoinHandle};
@@ -15,7 +16,6 @@ use crate::{
     chain_sync::chain_sync,
     error::{Error, Result},
     mithril_snapshot_config::MithrilSnapshotConfig,
-    network::Network,
     stats,
 };
 

rust/cardano-chain-follower/src/chain_sync_live_chains.rs

@@ -6,6 +6,7 @@ use std::{
     time::Duration,
 };
 
+use cardano_blockchain_types::{Fork, MultiEraBlock, Network, Point};
 use crossbeam_skiplist::SkipMap;
 use rayon::prelude::*;
 use strum::IntoEnumIterator;
@@ -14,8 +15,7 @@ use tracing::{debug, error};
 use crate::{
     error::{Error, Result},
     mithril_snapshot_data::latest_mithril_snapshot_id,
-    point::UNKNOWN_POINT,
-    stats, MultiEraBlock, Network, Point, TIP_POINT,
+    stats,
 };
 
 /// Type we use to manage the Sync Task handle map.
@@ -40,7 +40,7 @@ static LIVE_CHAINS: LazyLock<SkipMap<Network, ProtectedLiveChainBlockList>> = La
 static PEER_TIP: LazyLock<SkipMap<Network, Point>> = LazyLock::new(|| {
     let map = SkipMap::new();
     for network in Network::iter() {
-        map.insert(network, UNKNOWN_POINT);
+        map.insert(network, Point::UNKNOWN);
     }
     map
 });
@@ -52,7 +52,7 @@ fn update_peer_tip(chain: Network, tip: Point) {
 
 /// Set the last TIP received from the peer.
 pub(crate) fn get_peer_tip(chain: Network) -> Point {
-    (*PEER_TIP.get_or_insert(chain, UNKNOWN_POINT).value()).clone()
+    (*PEER_TIP.get_or_insert(chain, Point::UNKNOWN).value()).clone()
 }
 
 /// Number of seconds to wait if we detect a `SyncReady` race condition.
@@ -122,7 +122,7 @@ impl ProtectedLiveChainBlockList {
         let Some(check_last_live_entry) = live_chain.back() else {
             // Its not an error if we can't get a latest block because the chain is empty,
             // so report that we don't know...
-            return UNKNOWN_POINT;
+            return Point::UNKNOWN;
         };
         let check_last_live_block = check_last_live_entry.value();
         check_last_live_block.point()
@@ -192,7 +192,7 @@ impl ProtectedLiveChainBlockList {
     /// would be lost due to rollback. Will REFUSE to add a block which does NOT have
     /// a proper "previous" point defined.
     fn add_block_to_tip(
-        &self, chain: Network, block: MultiEraBlock, fork_count: &mut u64, tip: Point,
+        &self, chain: Network, block: MultiEraBlock, fork: &mut Fork, tip: Point,
     ) -> Result<()> {
         let live_chain = self.0.write().map_err(|_| Error::Internal)?;
 
@@ -202,11 +202,11 @@ impl ProtectedLiveChainBlockList {
         let last_live_point = Self::get_last_live_point(&live_chain);
         if !previous_point.strict_eq(&last_live_point) {
             // Detected a rollback, so increase the fork count.
-            *fork_count += 1;
+            fork.incr();
             let mut rollback_size: u64 = 0;
 
             // We are NOT contiguous, so check if we can become contiguous with a rollback.
-            debug!("Detected non-contiguous block, rolling back. Fork: {fork_count}");
+            debug!("Detected non-contiguous block, rolling back. Fork: {fork:?}");
 
             // First check if the previous is >= the earliest block in the live chain.
             // This is because when we start syncing we could rollback earlier than our
@@ -265,7 +265,7 @@ impl ProtectedLiveChainBlockList {
         // Ensures we are properly connected to the Mithril Chain.
         // But don't check this if we are about to purge the entire chain.
         // We do this before we bother locking the chain for update.
-        if *point != TIP_POINT {
+        if *point != Point::TIP {
             let latest_mithril_tip = latest_mithril_snapshot_id(chain).tip();
             if !point.strict_eq(&latest_mithril_tip) {
                 return Err(Error::LiveSync(format!(
@@ -278,7 +278,7 @@ impl ProtectedLiveChainBlockList {
 
         // Special Case.
         // If the Purge Point == TIP_POINT, then we purge the entire chain.
-        if *point == TIP_POINT {
+        if *point == Point::TIP {
             live_chain.clear();
         } else {
             // If the block we want to purge upto must be in the chain.
@@ -345,7 +345,7 @@ impl ProtectedLiveChainBlockList {
 
         // Loop until we exhaust probe slots, OR we would step past genesis.
         while slot_age < reference_slot {
-            let ref_point = Point::fuzzy(reference_slot - slot_age);
+            let ref_point = Point::fuzzy((reference_slot - slot_age).into());
             let Some(entry) = chain.lower_bound(Bound::Included(&ref_point)) else {
                 break;
             };
@@ -363,15 +363,15 @@ impl ProtectedLiveChainBlockList {
     /// Given a known point on the live chain, and a fork count, find the best block we
     /// have.
     fn find_best_fork_block(
-        &self, point: &Point, previous_point: &Point, fork: u64,
+        &self, point: &Point, previous_point: &Point, fork: Fork,
     ) -> Option<(MultiEraBlock, u64)> {
         let mut rollback_depth: u64 = 0;
         let Ok(chain) = self.0.read() else {
             return None;
         };
 
         // Get the block <= the current slot.
-        let ref_point = Point::fuzzy(point.slot_or_default());
+        let ref_point = Point::fuzzy(point.slot_or_default().into());
         let mut entry = chain.upper_bound(Bound::Included(&ref_point))?;
 
         let mut this_block = entry.value().clone();
@@ -400,7 +400,7 @@ impl ProtectedLiveChainBlockList {
         let live_chain = self.0.read().map_err(|_| Error::Internal).ok()?;
 
         let head_point = Self::get_last_live_point(&live_chain);
-        if head_point == UNKNOWN_POINT {
+        if head_point == Point::UNKNOWN {
             return None;
         }
 
@@ -457,7 +457,7 @@ pub(crate) fn get_live_block(
 ///
 /// Note: It MAY change between calling this function and actually backfilling.
 /// This is expected and normal behavior.
-pub(crate) async fn get_fill_to_point(chain: Network) -> (Point, u64) {
+pub(crate) async fn get_fill_to_point(chain: Network) -> (Point, Fork) {
     let live_chain = get_live_chain(chain);
 
     loop {
@@ -474,10 +474,10 @@ pub(crate) async fn get_fill_to_point(chain: Network) -> (Point, u64) {
 /// `rollback_count` should be set to 1 on the very first connection, after that,
 /// it is maintained by this function, and MUST not be modified elsewhere.
 pub(crate) fn live_chain_add_block_to_tip(
-    chain: Network, block: MultiEraBlock, fork_count: &mut u64, tip: Point,
+    chain: Network, block: MultiEraBlock, fork: &mut Fork, tip: Point,
 ) -> Result<()> {
     let live_chain = get_live_chain(chain);
-    live_chain.add_block_to_tip(chain, block, fork_count, tip)
+    live_chain.add_block_to_tip(chain, block, fork, tip)
 }
 
 /// Backfill the live chain with the block set provided.
@@ -509,7 +509,7 @@ pub(crate) fn get_intersect_points(chain: Network) -> Vec<pallas::network::minip
 
 /// Find best block from a fork relative to a point.
 pub(crate) fn find_best_fork_block(
-    chain: Network, point: &Point, previous_point: &Point, fork: u64,
+    chain: Network, point: &Point, previous_point: &Point, fork: Fork,
 ) -> Option<(MultiEraBlock, u64)> {
     let live_chain = get_live_chain(chain);
     live_chain.find_best_fork_block(point, previous_point, fork)