fix(code): Handle multiple commits for same height (#921)

* Add assert for non empty commit certificate

* Let full node store and write votes to WAL

* Round state machine check for previous decision.

Store the proposal round and value in the round state machine.

Remove decided proposal from the consensus state, get it from state machine instead.

Add flag to consensus state to indicate if Effect::Decide was sent to the app.

Cleanup test app.

* Fix mux typo

* Cleanup

* Change test to ensure the syncing node was a proposer at least once.

* Process the commit certificate first

* Panic in test app on multiple commits

* Correct comment

* Increase history for channel app

* Make channel app propose different values when restart from initial height

* Update comments, rename decided to decided_sent

* Check step in commit timeout as after WAL replay we may not be in commit
step.

* Clear all full proposals when moving to the next height

* Fix unused import

* Small reorg

* Cleanup

* Review comments

* Assert certificate is correct

Co-authored-by: Nenad Milosevic <[email protected]>

* Review comments

---------

Signed-off-by: Romain Ruetschi <[email protected]>
Co-authored-by: Romain Ruetschi <[email protected]>
Co-authored-by: Nenad Milosevic <[email protected]>

Author: 3 people

code/crates/core-consensus/src/error.rs

@@ -23,9 +23,17 @@ where
     #[error("Proposer not found at height {0} and round {1}")]
     ProposerNotFound(Ctx::Height, Round),
 
-    /// Decided value not found after commit timeout.
-    #[error("Decided value not found after commit timeout")]
-    DecidedValueNotFound(Ctx::Height, Round),
+    /// State machine has no decision in commit step.
+    #[error("State machine has no decision in commit step")]
+    DecisionNotFound(Ctx::Height, Round),
+
+    /// Driver proposal not found in commit step.
+    #[error("Driver proposal not found in commit step")]
+    DriverProposalNotFound(Ctx::Height, Round),
+
+    /// Full proposal not found in commit step.
+    #[error("Full proposal not found in commit step")]
+    FullProposalNotFound(Ctx::Height, Round),
 
     /// The driver failed to process an input.
     #[error("Driver failed to process input, reason: {0}")]

code/crates/core-consensus/src/full_proposal.rs

@@ -1,9 +1,8 @@
 use std::collections::BTreeMap;
 
 use derive_where::derive_where;
-use tracing::debug;
 
-use malachitebft_core_types::{Context, Height, Proposal, Round, SignedProposal, Validity, Value};
+use malachitebft_core_types::{Context, Proposal, Round, SignedProposal, Validity, Value};
 
 use crate::ProposedValue;
 
@@ -64,7 +63,7 @@ impl<Ctx: Context> Default for Entry<Ctx> {
     }
 }
 
-/// Keeper for collecting proposed values and consensus proposal messages for a given height and round.
+/// Keeper for collecting proposed values and consensus proposals for a given height and round.
 ///
 /// When a new_value is received from the value builder the following entry is stored:
 /// `Entry::ValueOnly(new_value.value, new_value.validity)`
@@ -77,7 +76,7 @@ impl<Ctx: Context> Default for Entry<Ctx> {
 ///
 /// It is possible that a proposer sends two (builder_value, proposal) pairs for same `(height, round)`.
 /// In this case both are stored, and we consider that the proposer is equivocating.
-/// Currently, the actual equivocation is caught deeper in the consensus crate, through consensus actor
+/// Currently, the actual equivocation is caught in the driver, through consensus actor
 /// propagating both proposals.
 ///
 /// When a new_proposal is received at most one complete proposal can be created. If a value at
@@ -88,9 +87,8 @@ impl<Ctx: Context> Default for Entry<Ctx> {
 /// at higher round with pol_round equal to the value round (L28). Therefore when a value is added
 /// multiple complete proposals may form.
 ///
-/// Note: In the future when we support implicit proposal message:
-/// - [`FullProposalKeeper::store_proposal()`] will never be called
-/// - [`FullProposalKeeper::full_proposal_at_round_and_value()`] should only check the presence of `builder_value`
+/// Note: For `parts_only` mode there is no explicit proposal wire message, instead
+/// one is synthesized by the caller (`on_proposed_value` handler) before it invokes the `store_proposal` method.
 #[derive_where(Clone, Debug, Default)]
 pub struct FullProposalKeeper<Ctx: Context> {
     keeper: BTreeMap<(Ctx::Height, Round), Vec<Entry<Ctx>>>,
@@ -352,10 +350,7 @@ impl<Ctx: Context> FullProposalKeeper<Ctx> {
         }
     }
 
-    pub fn remove_full_proposals(&mut self, last_height: Ctx::Height) {
-        // Keep last two decided heights
-        debug!(%last_height, "Removing proposals, keep the last two");
-        self.keeper
-            .retain(|(height, _), _| height.increment() >= last_height);
+    pub fn clear(&mut self) {
+        self.keeper.clear();
     }
 }

code/crates/core-consensus/src/handle/decide.rs

@@ -1,25 +1,76 @@
-use crate::prelude::*;
+use crate::{handle::signature::verify_certificate, prelude::*};
 
 #[cfg_attr(not(feature = "metrics"), allow(unused_variables))]
-pub async fn decide<Ctx>(
+pub async fn try_decide<Ctx>(
     co: &Co<Ctx>,
     state: &mut State<Ctx>,
     metrics: &Metrics,
-    consensus_round: Round,
-    proposal: SignedProposal<Ctx>,
 ) -> Result<(), Error<Ctx>>
 where
     Ctx: Context,
 {
-    let height = proposal.height();
-    let proposal_round = proposal.round();
-    let value = proposal.value();
+    if !state.driver.step_is_commit() {
+        return Ok(());
+    }
+
+    let height = state.driver.height();
+    let consensus_round = state.driver.round();
+
+    let Some((proposal_round, decided_value)) = state.decided_value() else {
+        return Err(Error::DecisionNotFound(height, consensus_round));
+    };
+
+    let decided_id = decided_value.id();
+
+    // Look for an existing certificate
+    let (certificate, extensions) = state
+        .driver
+        .commit_certificate(proposal_round, decided_id.clone())
+        .cloned()
+        .map(|certificate| (certificate, VoteExtensions::default()))
+        .unwrap_or_else(|| {
+            // Restore the commits. Note that they will be removed from `state`
+            let mut commits = state.restore_precommits(height, proposal_round, &decided_value);
 
-    // We only decide proposals for the current height
-    assert_eq!(height, state.driver.height());
+            let extensions = extract_vote_extensions(&mut commits);
 
-    // Clean proposals and values
-    state.remove_full_proposals(height);
+            let certificate =
+                CommitCertificate::new(height, proposal_round, decided_id.clone(), commits);
+
+            (certificate, extensions)
+        });
+
+    let Some((proposal, _)) = state.driver.proposal_and_validity_for_round(proposal_round) else {
+        return Err(Error::DriverProposalNotFound(height, proposal_round));
+    };
+
+    let Some(full_proposal) =
+        state.full_proposal_at_round_and_value(&height, proposal_round, &decided_value)
+    else {
+        return Err(Error::FullProposalNotFound(height, proposal_round));
+    };
+
+    if proposal.value().id() != decided_id {
+        info!(
+            "Decide: driver proposal value id {} does not match the decided value id {}, this may happen if consensus and value sync run in parallel",
+            proposal.value().id(),
+            decided_id
+        );
+    }
+
+    assert_eq!(full_proposal.builder_value.id(), decided_id);
+    assert_eq!(full_proposal.proposal.value().id(), decided_id);
+    assert_eq!(full_proposal.validity, Validity::Valid);
+
+    // The certificate must be valid if state is Commit
+    assert!(verify_certificate(
+        co,
+        certificate.clone(),
+        state.driver.validator_set().clone(),
+        state.params.threshold_params,
+    )
+    .await
+    .is_ok());
 
     // Update metrics
     #[cfg(feature = "metrics")]
@@ -48,32 +99,19 @@ where
         }
     }
 
-    // Look for an existing certificate
-    let (certificate, extensions) = state
-        .driver
-        .commit_certificate(proposal_round, value.id())
-        .cloned()
-        .map(|certificate| (certificate, VoteExtensions::default()))
-        .unwrap_or_else(|| {
-            // Restore the commits. Note that they will be removed from `state`
-            let mut commits = state.restore_precommits(height, proposal_round, value);
-
-            let extensions = extract_vote_extensions(&mut commits);
-
-            // TODO: Should we verify we have 2/3rd commits?
-            let certificate = CommitCertificate::new(height, proposal_round, value.id(), commits);
-
-            (certificate, extensions)
-        });
-
     perform!(
         co,
-        Effect::Decide(certificate, extensions, Default::default())
+        Effect::CancelTimeout(Timeout::commit(state.driver.round()), Default::default())
     );
 
-    // Reinitialize to remove any previous round or equivocating precommits.
-    // TODO: Revise when evidence module is added.
-    state.signed_precommits.clear();
+    if !state.decided_sent {
+        state.decided_sent = true;
+
+        perform!(
+            co,
+            Effect::Decide(certificate, extensions, Default::default())
+        );
+    }
 
     Ok(())
 }
@@ -92,28 +130,3 @@ pub fn extract_vote_extensions<Ctx: Context>(votes: &mut [SignedVote<Ctx>]) -> V
 
     VoteExtensions::new(extensions)
 }
-
-/// Decide on the current proposal without waiting for Commit timeout.
-pub async fn decide_current_no_timeout<Ctx>(
-    co: &Co<Ctx>,
-    state: &mut State<Ctx>,
-    metrics: &Metrics,
-) -> Result<(), Error<Ctx>>
-where
-    Ctx: Context,
-{
-    let height = state.driver.height();
-    let round = state.driver.round();
-
-    perform!(
-        co,
-        Effect::CancelTimeout(Timeout::commit(round), Default::default())
-    );
-
-    let proposal = state
-        .decision
-        .remove(&(height, round))
-        .ok_or_else(|| Error::DecidedValueNotFound(height, round))?;
-
-    decide(co, state, metrics, round, proposal).await
-}

code/crates/core-consensus/src/handle/driver.rs

@@ -305,9 +305,6 @@ where
                 "Decided",
             );
 
-            // Store value decided on for retrieval when timeout commit elapses
-            state.store_decision(state.driver.height(), consensus_round, proposal.clone());
-
             perform!(
                 co,
                 Effect::ScheduleTimeout(Timeout::commit(consensus_round), Default::default())

code/crates/core-consensus/src/handle/proposed_value.rs

@@ -3,7 +3,7 @@ use crate::prelude::*;
 use crate::handle::driver::apply_driver_input;
 use crate::types::ProposedValue;
 
-use super::decide::decide_current_no_timeout;
+use super::decide::try_decide;
 use super::signature::sign_proposal;
 
 pub async fn on_proposed_value<Ctx>(
@@ -71,8 +71,10 @@ where
         .await?;
     }
 
-    if origin == ValueOrigin::Sync && state.driver.step_is_commit() {
-        decide_current_no_timeout(co, state, metrics).await?;
+    if origin == ValueOrigin::Sync {
+        // The proposed value was provided by Sync, try to decide immediately, without waiting for the Commit timeout.
+        // `try_decide` will check that we are in the commit step after applying the proposed value to the state machine.
+        try_decide(co, state, metrics).await?;
     }
 
     Ok(())

code/crates/core-consensus/src/handle/rebroadcast_timeout.rs

@@ -18,11 +18,11 @@ where
 
     let (maybe_vote, timeout) = match timeout.kind {
         TimeoutKind::PrevoteRebroadcast => (
-            state.last_prevote.as_ref(),
+            state.last_signed_prevote.as_ref(),
             Timeout::prevote_rebroadcast(round),
         ),
         TimeoutKind::PrecommitRebroadcast => (
-            state.last_precommit.as_ref(),
+            state.last_signed_precommit.as_ref(),
             Timeout::precommit_rebroadcast(round),
         ),
         _ => return Ok(()),

code/crates/core-consensus/src/handle/start_height.rs

@@ -19,10 +19,10 @@ where
     #[cfg(feature = "metrics")]
     metrics.step_end(state.driver.step());
 
-    state.driver.move_to_height(height, validator_set);
+    state.reset_and_start_height(height, validator_set);
 
-    debug_assert_eq!(state.driver.height(), height);
-    debug_assert_eq!(state.driver.round(), Round::Nil);
+    debug_assert_eq!(state.height(), height);
+    debug_assert_eq!(state.round(), Round::Nil);
 
     start_height(co, state, metrics, height).await
 }

code/crates/core-consensus/src/handle/sync.rs

@@ -1,4 +1,4 @@
-use crate::handle::decide::decide_current_no_timeout;
+use crate::handle::decide::try_decide;
 use crate::handle::driver::apply_driver_input;
 use crate::handle::signature::verify_certificate;
 use crate::handle::validator_set::get_validator_set;
@@ -42,9 +42,9 @@ where
     )
     .await?;
 
-    if state.driver.step_is_commit() {
-        decide_current_no_timeout(co, state, metrics).await?;
-    }
+    // The CommitCertificate is provided by Value Sync, try to decide immediately, without waiting for the Commit timeout.
+    // `try_decide` will check that we are in the commit step after applying the certificate to the state machine.
+    try_decide(co, state, metrics).await?;
 
     Ok(())
 }

code/crates/core-consensus/src/handle/timeout.rs

@@ -1,4 +1,4 @@
-use crate::handle::decide::decide;
+use crate::handle::decide::try_decide;
 use crate::handle::driver::apply_driver_input;
 use crate::handle::rebroadcast_timeout::on_rebroadcast_timeout;
 use crate::handle::step_timeout::on_step_limit_timeout;
@@ -53,14 +53,12 @@ where
         TimeoutKind::PrevoteTimeLimit | TimeoutKind::PrecommitTimeLimit => {
             on_step_limit_timeout(co, state, metrics, timeout.round).await
         }
-        TimeoutKind::Commit => {
-            let proposal = state
-                .decision
-                .remove(&(height, round))
-                .ok_or_else(|| Error::DecidedValueNotFound(height, round))?;
-
-            decide(co, state, metrics, round, proposal).await
-        }
+        // Decide if the timeout is a commit timeout and the step is commit.
+        // `try_decide` will check that we are in the commit step. This is necessary because the timeout can be triggered
+        // by WAL replay before the step is `Commit`, e.g. when the node hasn't replayed the full value and
+        // the step is still `Propose`.
+        // For the Propose, Prevote and Precommit timeouts, the step is checked in the state machine.
+        TimeoutKind::Commit => try_decide(co, state, metrics).await,
         _ => Ok(()),
     }
 }