feat(consensus): set active/obserever height based on batcher height

crates/sequencing/papyrus_consensus/src/types.rs

@@ -178,4 +178,6 @@ pub enum ConsensusError {
     InternalNetworkError(String),
     #[error("{0}")]
     SyncError(String),
+    #[error("{0}")]
+    Other(String),
 }

crates/starknet_consensus_manager/Cargo.toml

@@ -17,6 +17,7 @@ papyrus_consensus_orchestrator.workspace = true
 papyrus_network.workspace = true
 papyrus_protobuf.workspace = true
 serde.workspace = true
+starknet_api.workspace = true
 starknet_batcher_types.workspace = true
 starknet_sequencer_infra.workspace = true
 starknet_state_sync_types.workspace = true

crates/starknet_consensus_manager/src/consensus_manager.rs

@@ -8,6 +8,7 @@ use papyrus_consensus_orchestrator::sequencer_consensus_context::SequencerConsen
 use papyrus_network::gossipsub_impl::Topic;
 use papyrus_network::network_manager::{BroadcastTopicChannels, NetworkManager};
 use papyrus_protobuf::consensus::{ConsensusMessage, ProposalPart, StreamMessage};
+use starknet_api::block::BlockNumber;
 use starknet_batcher_types::communication::SharedBatcherClient;
 use starknet_sequencer_infra::component_definitions::ComponentStarter;
 use starknet_sequencer_infra::errors::ComponentError;
@@ -63,6 +64,21 @@ impl ConsensusManager {
         let (outbound_internal_sender, inbound_internal_receiver, mut stream_handler_task_handle) =
             StreamHandler::get_channels(inbound_network_receiver, outbound_network_sender);
 
+        let observer_height =
+            self.batcher_client.get_height().await.map(|h| h.height).map_err(|e| {
+                error!("Failed to get height from batcher: {:?}", e);
+                ConsensusError::Other("Failed to get height from batcher".to_string())
+            })?;
+        let active_height = if self.config.consensus_config.start_height == observer_height {
+            // Setting `start_height` is only used to enable consensus starting immediately without
+            // observing the first height. This means consensus may return to a height
+            // it has already voted on, risking equivocation. This is only safe to do if we
+            // restart all nodes at this height.
+            observer_height
+        } else {
+            BlockNumber(observer_height.0 + 1)
+        };
+
         let context = SequencerConsensusContext::new(
             Arc::clone(&self.batcher_client),
             outbound_internal_sender,
@@ -74,9 +90,8 @@ impl ConsensusManager {
         let mut network_handle = tokio::task::spawn(network_manager.run());
         let consensus_task = papyrus_consensus::run_consensus(
             context,
-            self.config.consensus_config.start_height,
-            // TODO(Asmaa): replace with the correct value.
-            self.config.consensus_config.start_height,
+            active_height,
+            observer_height,
             self.config.consensus_config.validator_id,
             self.config.consensus_config.consensus_delay,
             self.config.consensus_config.timeouts.clone(),