fix(cat-gateway): Log relevant chain indexer info and update live tip metric (#2752)

* fix(cat-gateway): log relevant chain indexer info and update live tip metric

* fix(cat-gateway): fix update sync task count

Author: saibatizoku

catalyst-gateway/bin/src/cardano/mod.rs

@@ -39,10 +39,10 @@ async fn start_sync_for(cfg: &chain_follower::EnvVars) -> anyhow::Result<()> {
 
     let mut cfg = ChainSyncConfig::default_for(chain);
     cfg.mithril_cfg = cfg.mithril_cfg.with_dl_config(dl_config);
-    info!(chain = %chain, "Starting Blockchain Sync");
+    info!(chain = %chain, "Starting Chain Sync Task");
 
     if let Err(error) = cfg.run().await {
-        error!(chain=%chain, error=%error, "Failed to start chain sync task");
+        error!(chain=%chain, error=%error, "Failed to start Chain Sync Task");
         Err(error)?;
     }
 
@@ -237,14 +237,12 @@ fn sync_subchain(
     params: SyncParams, event_sender: broadcast::Sender<event::ChainIndexerEvent>,
 ) -> tokio::task::JoinHandle<SyncParams> {
     tokio::spawn(async move {
-        info!(chain = %params.chain, params=%params, "Indexing Blockchain");
-
         // Backoff hitting the database if we need to.
         params.backoff().await;
 
         // Wait for indexing DB to be ready before continuing.
         drop(CassandraSession::wait_until_ready(INDEXING_DB_READY_WAIT_INTERVAL, true).await);
-        info!(chain=%params.chain, params=%params,"Indexing DB is ready");
+        info!(chain=%params.chain, params=%params,"Starting Chain Indexing");
 
         let mut first_indexed_block = params.first_indexed_block.clone();
         let mut first_immutable = params.first_is_immutable;
@@ -255,6 +253,18 @@ fn sync_subchain(
         let mut follower =
             ChainFollower::new(params.chain, params.actual_start(), params.end.clone()).await;
         while let Some(chain_update) = follower.next().await {
+            let tips = ChainFollower::get_tips(params.chain).await;
+            let immutable_slot = tips.0.slot_or_default();
+            let live_slot = tips.1.slot_or_default();
+            let event = event::ChainIndexerEvent::LiveTipSlotChanged {
+                immutable_slot,
+                live_slot,
+            };
+            if let Err(err) = event_sender.send(event) {
+                error!(error=%err, "Unable to send event.");
+            } else {
+                debug!(live_tip_slot=?live_slot, "Chain Indexer update");
+            }
             match chain_update.kind {
                 cardano_chain_follower::Kind::ImmutableBlockRollForward => {
                     // We only process these on the follower tracking the TIP.
@@ -437,6 +447,30 @@ impl SyncTask {
         }
     }
 
+    /// Add a new `SyncTask` to the queue.
+    fn add_sync_task(
+        &mut self, params: SyncParams, event_sender: broadcast::Sender<event::ChainIndexerEvent>,
+    ) {
+        self.sync_tasks.push(sync_subchain(params, event_sender));
+        self.current_sync_tasks = self.current_sync_tasks.saturating_add(1);
+        debug!(current_sync_tasks=%self.current_sync_tasks, "Added new Sync Task");
+        self.dispatch_event(event::ChainIndexerEvent::SyncTasksChanged {
+            current_sync_tasks: self.current_sync_tasks,
+        });
+    }
+
+    /// Update `SyncTask` count.
+    fn sync_task_finished(&mut self) {
+        self.current_sync_tasks = self.current_sync_tasks.checked_sub(1).unwrap_or_else(|| {
+            error!("current_sync_tasks -= 1 overflow");
+            0
+        });
+        debug!(current_sync_tasks=%self.current_sync_tasks, "Finished Sync Task");
+        self.dispatch_event(event::ChainIndexerEvent::SyncTasksChanged {
+            current_sync_tasks: self.current_sync_tasks,
+        });
+    }
+
     /// Primary Chain Follower task.
     ///
     /// This continuously runs in the background, and never terminates.
@@ -452,7 +486,7 @@ impl SyncTask {
         let tips = ChainFollower::get_tips(self.cfg.chain).await;
         self.immutable_tip_slot = tips.0.slot_or_default();
         self.live_tip_slot = tips.1.slot_or_default();
-        info!(chain=%self.cfg.chain, immutable_tip=?self.immutable_tip_slot, live_tip=?self.live_tip_slot, "Blockchain ready to sync from.");
+        info!(chain=%self.cfg.chain, immutable_tip=?self.immutable_tip_slot, live_tip=?self.live_tip_slot, "Running the primary blockchain follower task.");
 
         self.dispatch_event(event::ChainIndexerEvent::ImmutableTipSlotChanged {
             immutable_slot: self.immutable_tip_slot,
@@ -470,20 +504,20 @@ impl SyncTask {
         // After waiting, we set the liveness flag to true if it is not already set.
         drop(CassandraSession::wait_until_ready(INDEXING_DB_READY_WAIT_INTERVAL, true).await);
 
-        info!(chain=%self.cfg.chain, "Indexing DB is ready - Getting recovery state");
+        info!(chain=%self.cfg.chain, "Indexing DB is ready - Getting recovery state for indexing");
         self.sync_status = get_sync_status().await;
         debug!(chain=%self.cfg.chain, "Sync Status: {:?}", self.sync_status);
 
         // Start the Live Chain sync task - This can never end because it is syncing to TIP.
         // So, if it fails, it will automatically be restarted.
-        self.sync_tasks.push(sync_subchain(
+        self.add_sync_task(
             SyncParams::new(
                 self.cfg.chain,
                 Point::fuzzy(self.immutable_tip_slot),
                 Point::TIP,
             ),
             self.event_channel.0.clone(),
-        ));
+        );
 
         self.start_immutable_followers();
 
@@ -495,8 +529,15 @@ impl SyncTask {
         // They will return from this iterator in the order they complete.
         // This iterator actually never ends, because the live sync task is always restarted.
         while let Some(completed) = self.sync_tasks.next().await {
+            // update sync task count
+            self.sync_task_finished();
+
             match completed {
                 Ok(finished) => {
+                    let tips = ChainFollower::get_tips(self.cfg.chain).await;
+                    let immutable_tip_slot = tips.0.slot_or_default();
+                    let live_tip_slot = tips.1.slot_or_default();
+                    info!(immutable_tip_slot=?immutable_tip_slot, live_tip_slot=?live_tip_slot, "Chain Indexer task finished");
                     // Sync task finished.  Check if it completed OK or had an error.
                     // If it failed, we need to reschedule it.
 
@@ -516,26 +557,20 @@ impl SyncTask {
                                     live_slot: self.live_tip_slot,
                                 },
                             );
+                            info!(chain=%self.cfg.chain, report=%finished,
+                            "Chain Indexer finished reaching TIP.");
 
                             self.start_immutable_followers();
                         } else {
                             error!(chain=%self.cfg.chain, report=%finished,
-                            "The TIP follower failed, restarting it.");
+                            "Chain Indexer finished without to reach TIP.");
                         }
 
                         // Start the Live Chain sync task again from where it left off.
-                        self.sync_tasks.push(sync_subchain(
-                            finished.retry(),
-                            self.event_channel.0.clone(),
-                        ));
+                        self.add_sync_task(finished.retry(), self.event_channel.0.clone());
                     } else if let Some(result) = finished.result.as_ref() {
                         match result {
                             Ok(()) => {
-                                self.current_sync_tasks =
-                                    self.current_sync_tasks.checked_sub(1).unwrap_or_else(|| {
-                                        error!("current_sync_tasks -= 1 overflow");
-                                        0
-                                    });
                                 info!(chain=%self.cfg.chain, report=%finished,
                                     "The Immutable follower completed successfully.");
 
@@ -546,9 +581,6 @@ impl SyncTask {
                                         },
                                     );
                                 });
-                                self.dispatch_event(event::ChainIndexerEvent::SyncTasksChanged {
-                                    current_sync_tasks: self.current_sync_tasks,
-                                });
 
                                 // If we need more immutable chain followers to sync the block
                                 // chain, we can now start them.
@@ -559,10 +591,7 @@ impl SyncTask {
                                         "An Immutable follower failed, restarting it.");
                                 // Restart the Immutable Chain sync task again from where it left
                                 // off.
-                                self.sync_tasks.push(sync_subchain(
-                                    finished.retry(),
-                                    self.event_channel.0.clone(),
-                                ));
+                                self.add_sync_task(finished.retry(), self.event_channel.0.clone());
                             },
                         }
                     } else {
@@ -575,6 +604,8 @@ impl SyncTask {
                 },
             }
 
+            let sync_task_count = self.sync_tasks.len();
+
             // IF there is only 1 chain follower left in sync_tasks, then all
             // immutable followers have finished.
             // When this happens we need to purge the live index of any records that exist
@@ -583,7 +614,7 @@ impl SyncTask {
             // want to put a gap in this, so that there are X slots of overlap
             // between the live chain and immutable chain.  This gap should be
             // a parameter.
-            if self.sync_tasks.len() == 1 {
+            if sync_task_count == 1 {
                 self.dispatch_event(event::ChainIndexerEvent::SyncCompleted);
 
                 // Purge data up to this slot
@@ -622,15 +653,10 @@ impl SyncTask {
                 if let Some((first_point, last_point)) =
                     self.get_syncable_range(self.start_slot, end_slot)
                 {
-                    self.sync_tasks.push(sync_subchain(
+                    self.add_sync_task(
                         SyncParams::new(self.cfg.chain, first_point, last_point.clone()),
                         self.event_channel.0.clone(),
-                    ));
-                    self.current_sync_tasks = self.current_sync_tasks.saturating_add(1);
-
-                    self.dispatch_event(event::ChainIndexerEvent::SyncTasksChanged {
-                        current_sync_tasks: self.current_sync_tasks,
-                    });
+                    );
                 }
 
                 // The one slot overlap is deliberate, it doesn't hurt anything and prevents all off
@@ -691,6 +717,7 @@ impl event::EventTarget<event::ChainIndexerEvent> for SyncTask {
     }
 
     fn dispatch_event(&self, message: event::ChainIndexerEvent) {
+        debug!(event = ?message, "Chain Indexer Event");
         let _ = self.event_channel.0.send(message);
     }
 }
@@ -704,7 +731,6 @@ pub(crate) async fn start_followers() -> anyhow::Result<()> {
 
     // Start Syncing the blockchain, so we can consume its data as required.
     start_sync_for(&cfg).await?;
-    info!(chain=%cfg.chain,"Chain Sync is started.");
 
     tokio::spawn(async move {
         use self::event::ChainIndexerEvent as Event;