Implement prior commit replay and improve startup synchronization in Starfish.

Fix differences between Starfish and Mysticeti in handling of previously processed commits at startup by introducing a replay mechanism. Adjust `StarfishConsensusHandler` to manage commit ranges effectively and log replay progress.

Author: piotrm50

crates/iota-core/src/consensus_handler.rs

@@ -517,6 +517,7 @@ pub struct StarfishConsensusHandler {
 
 impl StarfishConsensusHandler {
     pub fn new(
+        last_processed_commit_at_startup: starfish_core::CommitIndex,
         mut consensus_handler: ConsensusHandler<CheckpointService>,
         mut receiver: UnboundedReceiver<starfish_core::CommittedSubDag>,
         commit_consumer_monitor: Arc<starfish_core::CommitConsumerMonitor>,
@@ -526,10 +527,14 @@ impl StarfishConsensusHandler {
             // backpressure.
             while let Some(consensus_output) = receiver.recv().await {
                 let commit_index = consensus_output.commit_ref.index;
-                consensus_handler
-                    .handle_consensus_output(consensus_output)
-                    .await;
-                commit_consumer_monitor.set_highest_handled_commit(commit_index);
+                if commit_index <= last_processed_commit_at_startup {
+                    consensus_handler.handle_prior_consensus_output(consensus_output);
+                } else {
+                    consensus_handler
+                        .handle_consensus_output(consensus_output)
+                        .await;
+                    commit_consumer_monitor.set_highest_handled_commit(commit_index);
+                }
             }
         });
         Self {

crates/iota-core/src/consensus_manager/starfish_manager.rs

@@ -139,10 +139,11 @@ impl ConsensusManagerTrait for StarfishManager {
         let (commit_sender, commit_receiver) = unbounded_channel("consensus_output");
 
         let consensus_handler = consensus_handler_initializer.new_consensus_handler();
-        let consumer = CommitConsumer::new(
-            commit_sender,
-            consensus_handler.last_processed_subdag_index() as CommitIndex,
-        );
+
+        let num_prior_commits = protocol_config.consensus_num_requested_prior_commits_at_startup();
+        let last_processed_commit = consensus_handler.last_processed_subdag_index() as CommitIndex;
+        let starting_commit = last_processed_commit.saturating_sub(num_prior_commits);
+        let consumer = CommitConsumer::new(commit_sender, starting_commit);
         let monitor = consumer.monitor();
 
         // If there is a previous consumer monitor, it indicates that the consensus
@@ -201,13 +202,20 @@ impl ConsensusManagerTrait for StarfishManager {
         self.client.set(client);
 
         // spin up the new starfish consensus handler to listen for committed sub dags
-        let handler = StarfishConsensusHandler::new(consensus_handler, commit_receiver, monitor);
+        let handler = StarfishConsensusHandler::new(
+            last_processed_commit,
+            consensus_handler,
+            commit_receiver,
+            monitor,
+        );
 
         let mut consensus_handler = self.consensus_handler.lock().await;
         *consensus_handler = Some(handler);
 
         // Wait until all locally available commits have been processed
+        info!("replaying commits at startup");
         registered_authority.0.replay_complete().await;
+        info!("Startup commit replay complete");
     }
 
     async fn shutdown(&self) {

crates/iota-core/src/mysticeti_adapter.rs

@@ -90,6 +90,7 @@ impl ConsensusClient for LazyMysticetiClient {
             .iter()
             .map(|t| bcs::to_bytes(t).expect("Serializing consensus transaction cannot fail"))
             .collect::<Vec<_>>();
+
         let (block_ref, status_waiter) = client
             .as_ref()
             .expect("Client should always be returned")