apollo_consensus: fallback to sync in case of Batcher errors (#10973)

Author: dafnamatsry

Cargo.lock

@@ -121,7 +121,7 @@ pub enum BatcherResponse {
 }
 impl_debug_for_infra_requests_and_responses!(BatcherResponse);
 
-#[derive(Clone, Debug, Error)]
+#[derive(Clone, Debug, Error, PartialEq)]
 pub enum BatcherClientError {
     #[error(transparent)]
     ClientError(#[from] ClientError),

crates/apollo_batcher_types/src/communication.rs

@@ -10,6 +10,7 @@ description = "Reach consensus for Starknet"
 testing = ["mockall", "tempfile"]
 
 [dependencies]
+apollo_batcher_types.workspace = true
 apollo_config_manager_types.workspace = true
 apollo_consensus_config.workspace = true
 apollo_infra_utils.workspace = true

crates/apollo_consensus/Cargo.toml

@@ -411,13 +411,38 @@ impl<ContextT: ConsensusContext> MultiHeightManager<ContextT> {
         proposals_receiver: &mut mpsc::Receiver<mpsc::Receiver<ContextT::ProposalPart>>,
     ) -> Result<RunHeightRes, ConsensusError> {
         info!("Running consensus for height {}.", height);
-        let res =
-            self.run_height_inner(context, height, broadcast_channels, proposals_receiver).await?;
 
-        // Commit in case of decision.
-        if let RunHeightRes::Decision(decision) = &res {
-            context.decision_reached(height, decision.block).await?;
-        }
+        let consensus_result =
+            self.run_height_inner(context, height, broadcast_channels, proposals_receiver).await;
+
+        let res = match consensus_result {
+            Ok(ok) => match ok {
+                RunHeightRes::Decision(decision) => {
+                    // Commit decision to context.
+                    context.decision_reached(height, decision.block).await?;
+                    RunHeightRes::Decision(decision)
+                }
+                RunHeightRes::Sync => RunHeightRes::Sync,
+            },
+
+            Err(err) => match err {
+                e @ ConsensusError::BatcherError(_) => {
+                    error!(
+                        "Error while running consensus for height {height}, fallback to sync: {e}"
+                    );
+                    self.wait_until_sync_reaches_height(height, context).await;
+                    RunHeightRes::Sync
+                }
+                e @ ConsensusError::BlockInfoConversion(_)
+                | e @ ConsensusError::ProtobufConversionError(_)
+                | e @ ConsensusError::SendError(_)
+                | e @ ConsensusError::InternalNetworkError(_) => {
+                    // The node is missing required components/data and cannot continue
+                    // participating in the consensus. A fix and node restart are required.
+                    return Err(e);
+                }
+            },
+        };
 
         // Cleanup after height completion.
         self.cleanup_post_height(context, height, broadcast_channels, proposals_receiver).await?;
@@ -586,7 +611,7 @@ impl<ContextT: ConsensusContext> MultiHeightManager<ContextT> {
         }
 
         // Reflect initial height/round to context before executing requests.
-        context.set_height_and_round(height, shc.current_round()).await;
+        context.set_height_and_round(height, shc.current_round()).await?;
         self.execute_requests(context, height, pending_requests, shc_events, broadcast_channels)
             .await?;
         Ok(None)
@@ -635,7 +660,7 @@ impl<ContextT: ConsensusContext> MultiHeightManager<ContextT> {
                 }
             };
             // Reflect current height/round to context.
-            context.set_height_and_round(height, shc.current_round()).await;
+            context.set_height_and_round(height, shc.current_round()).await?;
             match shc_return {
                 ShcReturn::Decision(decision) => return Ok(RunHeightRes::Decision(decision)),
                 ShcReturn::Requests(requests) => {
@@ -883,7 +908,7 @@ impl<ContextT: ConsensusContext> MultiHeightManager<ContextT> {
                 // (round 0) proposal timeout for building to avoid giving the Batcher more time
                 // when proposal time is extended for consensus.
                 let timeout = timeouts.get_proposal_timeout(0);
-                let receiver = context.build_proposal(init, timeout).await;
+                let receiver = context.build_proposal(init, timeout).await?;
                 let fut = async move {
                     let proposal_id = receiver.await.ok();
                     StateMachineEvent::FinishedBuilding(proposal_id, round)

crates/apollo_consensus/src/manager.rs

@@ -2,6 +2,8 @@ use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use std::vec;
 
+use apollo_batcher_types::communication::BatcherClientError;
+use apollo_batcher_types::errors::BatcherError;
 use apollo_config_manager_types::communication::MockConfigManagerClient;
 use apollo_consensus_config::config::{
     ConsensusConfig,
@@ -39,7 +41,7 @@ use crate::test_utils::{
     NoOpHeightVotedStorage,
     TestProposalPart,
 };
-use crate::types::{Round, ValidatorId};
+use crate::types::{ConsensusError, Round, ValidatorId};
 use crate::votes_threshold::QuorumType;
 use crate::RunConsensusArguments;
 
@@ -166,7 +168,7 @@ async fn manager_multiple_heights_unordered(consensus_config: ConsensusConfig) {
     expect_validate_proposal(&mut context, Felt::ONE, 1);
     context.expect_validators().returning(move |_| vec![*PROPOSER_ID, *VALIDATOR_ID]);
     context.expect_proposer().returning(move |_, _| *PROPOSER_ID);
-    context.expect_set_height_and_round().returning(move |_, _| ());
+    context.expect_set_height_and_round().returning(move |_, _| Ok(()));
     context.expect_broadcast().returning(move |_| Ok(()));
     context
         .expect_decision_reached()
@@ -220,7 +222,7 @@ async fn run_consensus_sync(consensus_config: ConsensusConfig) {
     expect_validate_proposal(&mut context, Felt::TWO, 1);
     context.expect_validators().returning(move |_| vec![*PROPOSER_ID, *VALIDATOR_ID]);
     context.expect_proposer().returning(move |_, _| *PROPOSER_ID);
-    context.expect_set_height_and_round().returning(move |_, _| ());
+    context.expect_set_height_and_round().returning(move |_, _| Ok(()));
     context.expect_broadcast().returning(move |_| Ok(()));
     context
         .expect_decision_reached()
@@ -286,7 +288,7 @@ async fn test_timeouts(consensus_config: ConsensusConfig) {
     send(&mut sender, precommit(None, HEIGHT_1, ROUND_0, *VALIDATOR_ID_3)).await;
 
     let mut context = MockTestContext::new();
-    context.expect_set_height_and_round().returning(move |_, _| ());
+    context.expect_set_height_and_round().returning(move |_, _| Ok(()));
     expect_validate_proposal(&mut context, Felt::ONE, 2);
     context
         .expect_validators()
@@ -444,7 +446,7 @@ async fn future_height_limit_caching_and_dropping(mut consensus_config: Consensu
     expect_validate_proposal(&mut context, Felt::ONE, 1); // Height 1 validation
     context.expect_validators().returning(move |_| vec![*PROPOSER_ID, *VALIDATOR_ID]);
     context.expect_proposer().returning(move |_, _| *PROPOSER_ID);
-    context.expect_set_height_and_round().returning(move |_, _| ());
+    context.expect_set_height_and_round().returning(move |_, _| Ok(()));
     // Set up coordination to detect when node votes Nil for height 2 (indicating proposal was
     // dropped, so the node didn't received the proposal and votes Nil).
     let (height2_nil_vote_trigger, height2_nil_vote_wait) = oneshot::channel();
@@ -583,16 +585,18 @@ async fn current_height_round_limit_caching_and_dropping(mut consensus_config: C
         .times(1)
         .return_once(|_, _| {
             round1_trigger.send(()).unwrap();
+            Ok(())
         });
     context
         .expect_set_height_and_round()
         .withf(|height, round| *height == HEIGHT_1 && *round == ROUND_2)
         .times(1)
         .return_once(|_, _| {
             round2_trigger.send(()).unwrap();
+            Ok(())
         });
     // Handle all other set_height_and_round calls normally.
-    context.expect_set_height_and_round().returning(move |_, _| ());
+    context.expect_set_height_and_round().returning(move |_, _| Ok(()));
     context
         .expect_decision_reached()
         .withf(move |_, c| *c == ProposalCommitment(Felt::ONE))
@@ -667,7 +671,7 @@ async fn run_consensus_dynamic_client_updates_validator_between_heights(
     // Context with expectations: H1 we are the validator, learn height via sync; at H2 we are the
     // proposer.
     let mut context = MockTestContext::new();
-    context.expect_set_height_and_round().returning(move |_, _| ());
+    context.expect_set_height_and_round().returning(move |_, _| Ok(()));
     context.expect_validators().returning(move |h: BlockNumber| {
         if h == HEIGHT_1 { vec![*VALIDATOR_ID] } else { vec![*PROPOSER_ID] }
     });
@@ -688,7 +692,7 @@ async fn run_consensus_dynamic_client_updates_validator_between_heights(
         .returning(move |_, _| {
             let (sender, receiver) = oneshot::channel();
             sender.send(ProposalCommitment(Felt::TWO)).unwrap();
-            receiver
+            Ok(receiver)
         })
         .times(1);
     // Expect a decision at height 2.
@@ -818,7 +822,7 @@ async fn manager_runs_normally_when_height_is_greater_than_last_voted_height(
     expect_validate_proposal(&mut context, Felt::ONE, 1);
     context.expect_validators().returning(move |_| vec![*PROPOSER_ID, *VALIDATOR_ID]);
     context.expect_proposer().returning(move |_, _| *PROPOSER_ID);
-    context.expect_set_height_and_round().returning(move |_, _| ());
+    context.expect_set_height_and_round().returning(move |_, _| Ok(()));
     context.expect_broadcast().returning(move |_| Ok(()));
     context
         .expect_decision_reached()
@@ -921,7 +925,7 @@ async fn writes_voted_height_to_storage(consensus_config: ConsensusConfig) {
     context
         .expect_validators()
         .returning(move |_| vec![*PROPOSER_ID, *VALIDATOR_ID, *VALIDATOR_ID_2, *VALIDATOR_ID_3]);
-    context.expect_set_height_and_round().returning(move |_, _| ());
+    context.expect_set_height_and_round().returning(move |_, _| Ok(()));
     context.expect_try_sync().returning(|_| false);
 
     // Set up storage expectation for prevote - must happen before broadcast
@@ -1047,3 +1051,45 @@ async fn writes_voted_height_to_storage(consensus_config: ConsensusConfig) {
 
     assert_decision(decision, block_id.0, ROUND_0);
 }
+
+#[rstest]
+#[tokio::test]
+async fn manager_fallback_to_sync_on_height_level_errors(consensus_config: ConsensusConfig) {
+    let TestSubscriberChannels { mock_network: _mock_network, subscriber_channels } =
+        mock_register_broadcast_topic().unwrap();
+
+    let (mut _proposal_receiver_sender, mut proposal_receiver_receiver) =
+        mpsc::channel(CHANNEL_SIZE);
+
+    let mut context = MockTestContext::new();
+    context.expect_validators().returning(move |_| vec![*PROPOSER_ID, *VALIDATOR_ID]);
+    context.expect_proposer().returning(move |_, _| *PROPOSER_ID);
+
+    // Sync should first fail, so consensus will try to run.
+    context.expect_try_sync().times(1).returning(|_| false);
+
+    // Consensus should fail when context.set_height_and_round fails.
+    context.expect_set_height_and_round().times(1).returning(move |_, _| {
+        Err(ConsensusError::BatcherError(BatcherClientError::BatcherError(
+            BatcherError::InternalError,
+        )))
+    });
+
+    // Now sync should be called and succeed.
+    context.expect_try_sync().withf(move |height| *height == HEIGHT_1).times(1).returning(|_| true);
+
+    let mut manager = MultiHeightManager::new(
+        consensus_config,
+        QuorumType::Byzantine,
+        Arc::new(Mutex::new(NoOpHeightVotedStorage)),
+    );
+    let res = manager
+        .run_height(
+            &mut context,
+            HEIGHT_1,
+            &mut subscriber_channels.into(),
+            &mut proposal_receiver_receiver,
+        )
+        .await;
+    assert_eq!(res, Ok(RunHeightRes::Sync));
+}

crates/apollo_consensus/src/manager_test.rs

@@ -68,7 +68,7 @@ mock! {
             &mut self,
             init: ProposalInit,
             timeout: Duration,
-        ) -> oneshot::Receiver<ProposalCommitment>;
+        ) -> Result<oneshot::Receiver<ProposalCommitment>, ConsensusError>;
 
         async fn validate_proposal(
             &mut self,
@@ -97,7 +97,7 @@ mock! {
 
         async fn try_sync(&mut self, height: BlockNumber) -> bool;
 
-        async fn set_height_and_round(&mut self, height: BlockNumber, round: Round);
+        async fn set_height_and_round(&mut self, height: BlockNumber, round: Round) -> Result<(), ConsensusError>;
     }
 }
 

crates/apollo_consensus/src/test_utils.rs

@@ -2,6 +2,7 @@
 use std::fmt::Debug;
 use std::time::Duration;
 
+use apollo_batcher_types::communication::BatcherClientError;
 use apollo_network::network_manager::{
     BroadcastTopicChannels,
     BroadcastTopicClient,
@@ -57,7 +58,7 @@ pub trait ConsensusContext {
         &mut self,
         init: ProposalInit,
         timeout: Duration,
-    ) -> oneshot::Receiver<ProposalCommitment>;
+    ) -> Result<oneshot::Receiver<ProposalCommitment>, ConsensusError>;
 
     /// This function is called by consensus to validate a block. It expects that this call will
     /// return immediately and that context can then stream in the block's content in parallel to
@@ -115,7 +116,11 @@ pub trait ConsensusContext {
 
     /// Update the context with the current height and round.
     /// Must be called at the beginning of each height.
-    async fn set_height_and_round(&mut self, height: BlockNumber, round: Round);
+    async fn set_height_and_round(
+        &mut self,
+        height: BlockNumber,
+        round: Round,
+    ) -> Result<(), ConsensusError>;
 }
 
 #[derive(PartialEq, Debug)]
@@ -147,6 +152,8 @@ pub enum ConsensusError {
     ProtobufConversionError(#[from] ProtobufConversionError),
     #[error(transparent)]
     SendError(#[from] mpsc::SendError),
+    #[error(transparent)]
+    BatcherError(#[from] BatcherClientError),
     // Indicates an error in communication between consensus and the node's networking component.
     // As opposed to an error between this node and peer nodes.
     #[error("{0}")]