fix(rust/cardano-chain-follower): fix immutable roll forward events

rust/Justfile

@@ -61,7 +61,7 @@ run-mithril-download-example-preview: code-format code-lint
 run-mithril-download-example-mainnet: code-format code-lint
     cargo build -r --package cardano-chain-follower --example follow_chains --features mimalloc
     RUST_LOG="error,follow_chains=debug,cardano_chain_follower=debug,mithril-client=debug" \
-    ./target/release/examples/follow_chains --mainnet
+    ./target/release/examples/follow_chains --mainnet --mithril-sync-workers 64 --mithril-sync-chunk-size 16 --mithril-sync-queue-ahead=6
 
 # Run long running developer test for mithril downloading.
 debug-heap-mithril-download-example: code-format code-lint

rust/cardano-chain-follower/examples/follow_chains.rs

@@ -183,6 +183,18 @@ async fn follow_for(network: Network, matches: ArgMatches) {
     while let Some(chain_update) = follower.next().await {
         updates = updates.saturating_add(1);
 
+        if chain_update.kind == Kind::ImmutableBlockRollForward {
+            let immutable_tip = u64::from(chain_update.data.point().slot_or_default());
+            let immutable = chain_update.immutable();
+            info!(
+                updates = updates,
+                immutable_tip = immutable_tip,
+                immutable = immutable,
+                "Chain Immutable Roll Forward Detected."
+            );
+            continue; // Nothing else to do, so get next sequential block.
+        }
+
         if chain_update.tip {
             reached_tip = true;
         }

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

@@ -484,10 +484,11 @@ async fn live_sync_backfill_and_purge(
         }
 
         debug!(
-            "After Purge: Size of the Live Chain is: {} Blocks",
+            "After Purge: Size of the Live Chain is: {} Blocks: Triggering Sleeping Followers.",
             live_chain_length(cfg.chain)
         );
 
+        // Trigger any sleeping followers that data has changed.
         notify_follower(
             cfg.chain,
             update_sender.as_ref(),

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

@@ -14,7 +14,7 @@ use crate::{
     mithril_snapshot_data::latest_mithril_snapshot_id,
     mithril_snapshot_iterator::MithrilSnapshotIterator,
     stats::{self},
-    Statistics,
+    Kind, Statistics,
 };
 
 /// The Chain Follower
@@ -87,6 +87,50 @@ impl ChainFollower {
     async fn next_from_mithril(&mut self) -> Option<ChainUpdate> {
         let current_mithril_tip = latest_mithril_snapshot_id(self.chain).tip();
 
+        // Get previous mithril tip, or set it and return current if a previous does not exist.
+        let previous_mithril_tip = if let Some(tip) = &self.mithril_tip {
+            tip
+        } else {
+            debug!(
+                mithril_tip = ?current_mithril_tip,
+                "Setting Initial Mithril Tip"
+            );
+            self.mithril_tip = Some(current_mithril_tip.clone());
+            &current_mithril_tip
+        };
+
+        // Return an ImmutableBlockRollForward event as soon as we can after one occurs.
+        // This is not an advancement in the followers sequential block iterating state.
+        // BUT it is a necessary status to return to a follower, so it can properly handle
+        // when immutable state advances (if it so requires)
+        if current_mithril_tip != *previous_mithril_tip {
+            debug!(
+                new_tip = ?self.mithril_tip,
+                current_tip = ?current_mithril_tip,
+                "Mithril Tip has changed"
+            );
+            // We have a new mithril tip so report Mithril Tip Roll Forward
+            if let Some(block) = self.snapshot.read_block_at(&current_mithril_tip).await {
+                // Update the snapshot in the follower state to the new snapshot.
+                let update = ChainUpdate::new(
+                    chain_update::Kind::ImmutableBlockRollForward,
+                    false, // Tip is Live chain tip, not Mithril Tip, and this is Mithril Tip.
+                    block,
+                );
+                return Some(update);
+            }
+            // This can only happen if the snapshot does not contain the tip block.
+            // So its effectively impossible/unreachable.
+            // However, IF it does happen, nothing bad (other than a delay to reporting
+            // immutable roll forward) will occur, so we log this impossible
+            // error, and continue processing.
+            error!(
+                tip = ?self.mithril_tip,
+                current = ?current_mithril_tip,
+                "Mithril Tip Block is not in snapshot. Should not happen."
+            );
+        }
+
         if current_mithril_tip > self.current {
             if self.mithril_follower.is_none() {
                 self.mithril_follower = self
@@ -97,9 +141,7 @@ impl ChainFollower {
 
             if let Some(follower) = self.mithril_follower.as_mut() {
                 if let Some(next) = follower.next().await {
-                    // debug!("Pre Previous update 3 : {:?}", self.previous);
                     self.previous = self.current.clone();
-                    // debug!("Post Previous update 3 : {:?}", self.previous);
                     self.current = next.point();
                     self.fork = Fork::IMMUTABLE; // Mithril Immutable data is always Fork 0.
                     let update = ChainUpdate::new(chain_update::Kind::Block, false, next);
@@ -108,29 +150,6 @@ impl ChainFollower {
             }
         }
 
-        let roll_forward_condition = if let Some(mithril_tip) = &self.mithril_tip {
-            current_mithril_tip > *mithril_tip && *mithril_tip > self.current
-        } else {
-            true
-        };
-
-        if roll_forward_condition {
-            let snapshot = MithrilSnapshot::new(self.chain);
-            if let Some(block) = snapshot.read_block_at(&current_mithril_tip).await {
-                // The Mithril Tip has moved forwards.
-                self.mithril_tip = Some(current_mithril_tip);
-                // Get the mithril tip block.
-                let update =
-                    ChainUpdate::new(chain_update::Kind::ImmutableBlockRollForward, false, block);
-                return Some(update);
-            }
-            error!(
-                tip = ?self.mithril_tip,
-                current = ?current_mithril_tip,
-                "Mithril Tip Block is not in snapshot. Should not happen."
-            );
-        }
-
         None
     }
 
@@ -215,6 +234,16 @@ impl ChainFollower {
     /// Update the current Point, and return `false` if this fails.
     fn update_current(&mut self, update: Option<&ChainUpdate>) -> bool {
         if let Some(update) = update {
+            if update.kind == Kind::ImmutableBlockRollForward {
+                // The ImmutableBlockRollForward includes the Mithril TIP Block.
+                // Update the mithril_tip state to the point of it.
+                self.mithril_tip = Some(update.data.point());
+                debug!(mithril_tip=?self.mithril_tip, "Updated followers current Mithril Tip");
+                // We DO NOT update anything else for this kind of update, as its informational and
+                // does not advance the state of the follower to a new block.
+                // It is still a valid update, and so return true, but don't update more state.
+                return true;
+            }
             let decoded = update.block_data().decode();
             self.current = Point::new(decoded.slot().into(), decoded.hash().into());
             return true;
@@ -235,10 +264,6 @@ impl ChainFollower {
 
         // We will loop here until we can successfully return a new block
         loop {
-            // Check if Immutable TIP has advanced, and if so, send a ChainUpdate about it.
-            // Should only happen once every ~6hrs.
-            // TODO.
-
             // Try and get the next update from the mithril chain, and return it if we are
             // successful.
             update = self.next_from_mithril().await;
@@ -254,16 +279,26 @@ impl ChainFollower {
 
             // IF we can't get a new block directly from the mithril data, or the live chain, then
             // wait for something to change which might mean we can get the next block.
-            let update = self.sync_updates.recv().await;
-            match update {
+            // Note, this is JUST a trigger, we don't process based on it other than to allow
+            // a blocked follower to continue.
+            let changed_data_trigger = self.sync_updates.recv().await;
+            match changed_data_trigger {
                 Ok(kind) => {
+                    // The KIND of event signaling changed data is not important, but we do log it
+                    // to help with debugging in case an update stops.
                     debug!("Update kind: {kind}");
                 },
                 Err(tokio::sync::broadcast::error::RecvError::Lagged(distance)) => {
+                    // The update queue is small, its possible that it fills before a task can
+                    // read from it, this will cause this Lagged error.
+                    // BUT, because we don't care what the event was, this is as good as the missed
+                    // event.  Therefore its not an error, and just log it at debug to help with
+                    // debugging the logic only.
                     debug!("Lagged by {} updates", distance);
                 },
                 Err(tokio::sync::broadcast::error::RecvError::Closed) => {
-                    // We are closed, so we need to wait for the next update.
+                    // The queue is closed, so we need to return that its no longer possible to
+                    // get data from this follower.
                     // This is not an error.
                     return None;
                 },
@@ -405,7 +440,7 @@ mod tests {
         assert_eq!(follower.current, start);
         assert_eq!(follower.fork, 1.into());
         assert!(follower.mithril_follower.is_none());
-        assert!(follower.mithril_tip.is_none());
+        // assert!(follower.mithril_tip.is_none());
     }
 
     #[tokio::test]

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

@@ -77,7 +77,7 @@ impl MithrilSnapshot {
 
     /// Read a single block from a known point.
     #[allow(clippy::indexing_slicing)]
-    #[logcall("debug")]
+    //#[logcall("debug")]
     pub(crate) async fn read_block_at(&self, point: &Point) -> Option<MultiEraBlock> {
         if let Some(iterator) = self.try_read_blocks_from_point(point).await {
             let block = iterator.next().await;

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

@@ -11,15 +11,13 @@ use cardano_blockchain_types::{MultiEraBlock, Network};
 use chrono::{TimeDelta, Utc};
 use dashmap::DashSet;
 use humantime::format_duration;
-use logcall::logcall;
 use mithril_client::{Client, MessageBuilder, MithrilCertificate, Snapshot, SnapshotListItem};
 use tokio::{
     fs::remove_dir_all,
     sync::mpsc::Sender,
     time::{sleep, Duration},
 };
 use tracing::{debug, error};
-use tracing_log::log;
 
 use crate::{
     error::{Error, Result},
@@ -251,7 +249,7 @@ pub(crate) const MITHRIL_IMMUTABLE_SUB_DIRECTORY: &str = "immutable";
 /// This function returns the tip block point, and the block point immediately proceeding
 /// it in a tuple.
 #[allow(clippy::indexing_slicing)]
-#[logcall(ok = "debug", err = "error")]
+//#[logcall(ok = "debug", err = "error")]
 pub(crate) async fn get_mithril_tip(chain: Network, path: &Path) -> Result<MultiEraBlock> {
     let mut path = path.to_path_buf();
     path.push(MITHRIL_IMMUTABLE_SUB_DIRECTORY);