feat(code/engine): Store proposed values in WAL

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

@@ -4,7 +4,7 @@ use malachitebft_core_types::*;
 
 use crate::input::RequestId;
 use crate::types::SignedConsensusMsg;
-use crate::{ConsensusMsg, VoteExtensionError};
+use crate::{ConsensusMsg, VoteExtensionError, WalEntry};
 
 /// Provides a way to construct the appropriate [`Resume`] value to
 /// resume execution after handling an [`Effect`].
@@ -182,15 +182,10 @@ where
         resume::Continue,
     ),
 
-    /// Append a consensus message to the Write-Ahead Log for crash recovery
+    /// Append an entry to the Write-Ahead Log for crash recovery
     ///
     /// Resume with: [`resume::Continue`]`
-    WalAppendMessage(SignedConsensusMsg<Ctx>, resume::Continue),
-
-    /// Append a timeout to the Write-Ahead Log for crash recovery
-    ///
-    /// Resume with: [`resume::Continue`]`
-    WalAppendTimeout(Timeout, resume::Continue),
+    WalAppend(WalEntry<Ctx>, resume::Continue),
 
     /// Allows the application to extend the pre-commit vote with arbitrary data.
     ///

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

@@ -113,7 +113,7 @@ impl<Ctx: Context> FullProposalKeeper<Ctx> {
         Self::default()
     }
 
-    pub fn full_proposals_for_value(
+    pub fn proposals_for_value(
         &self,
         proposed_value: &ProposedValue<Ctx>,
     ) -> Vec<SignedProposal<Ctx>> {
@@ -157,7 +157,28 @@ impl<Ctx: Context> FullProposalKeeper<Ctx> {
         None
     }
 
-    #[allow(clippy::type_complexity)]
+    pub fn full_proposal_at_round_and_proposer(
+        &self,
+        height: &Ctx::Height,
+        round: Round,
+        proposer: &Ctx::Address,
+    ) -> Option<&FullProposal<Ctx>> {
+        let entries = self
+            .keeper
+            .get(&(*height, round))
+            .filter(|entries| !entries.is_empty())?;
+
+        for entry in entries {
+            if let Entry::Full(p) = entry {
+                if p.proposal.validator_address() == proposer {
+                    return Some(p);
+                }
+            }
+        }
+
+        None
+    }
+
     pub fn get_value<'a>(
         &self,
         height: &Ctx::Height,

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

@@ -8,8 +8,11 @@ use crate::handle::vote::on_vote;
 use crate::prelude::*;
 use crate::types::SignedConsensusMsg;
 use crate::util::pretty::PrettyVal;
+use crate::LocallyProposedValue;
 use crate::VoteSyncMode;
 
+use super::propose::on_propose;
+
 #[async_recursion]
 pub async fn apply_driver_input<Ctx>(
     co: &Co<Ctx>,
@@ -322,12 +325,26 @@ where
         }
 
         DriverOutput::GetValue(height, round, timeout) => {
-            info!(%height, %round, "Requesting value");
+            if let Some(full_proposal) =
+                state.full_proposal_at_round_and_proposer(&height, round, state.address())
+            {
+                info!(%height, %round, "Using already existing value");
+
+                let local_value = LocallyProposedValue {
+                    height: full_proposal.proposal.height(),
+                    round: full_proposal.proposal.round(),
+                    value: full_proposal.builder_value.clone(),
+                };
 
-            perform!(
-                co,
-                Effect::GetValue(height, round, timeout, Default::default())
-            );
+                on_propose(co, state, metrics, local_value).await?;
+            } else {
+                info!(%height, %round, "Requesting value from application");
+
+                perform!(
+                    co,
+                    Effect::GetValue(height, round, timeout, Default::default())
+                );
+            }
 
             Ok(())
         }

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

@@ -3,9 +3,8 @@ use crate::handle::signature::verify_signature;
 use crate::handle::validator_set::get_validator_set;
 use crate::input::Input;
 use crate::prelude::*;
-use crate::types::ConsensusMsg;
+use crate::types::{ConsensusMsg, ProposedValue, SignedConsensusMsg, WalEntry};
 use crate::util::pretty::PrettyProposal;
-use crate::{ProposedValue, SignedConsensusMsg};
 
 pub async fn on_proposal<Ctx>(
     co: &Co<Ctx>,
@@ -38,23 +37,14 @@ where
     }
 
     info!(
-        height = %consensus_height,
-        %proposal_height,
-        address = %proposer_address,
+        consensus.height = %consensus_height,
+        proposal.height = %proposal_height,
+        proposal.round = %proposal_round,
+        proposer = %proposer_address,
         message = %PrettyProposal::<Ctx>(&signed_proposal.message),
         "Received proposal"
     );
 
-    // Queue messages if driver is not initialized, or if they are for higher height.
-    // Process messages received for the current height.
-    // Drop all others.
-    if state.driver.round() == Round::Nil {
-        debug!("Received proposal at round -1, queuing for later");
-        state.buffer_input(signed_proposal.height(), Input::Proposal(signed_proposal));
-
-        return Ok(());
-    }
-
     if proposal_height > consensus_height {
         debug!("Received proposal for higher height, queuing for later");
         state.buffer_input(signed_proposal.height(), Input::Proposal(signed_proposal));
@@ -72,8 +62,8 @@ where
     if state.params.value_payload.include_proposal() {
         perform!(
             co,
-            Effect::WalAppendMessage(
-                SignedConsensusMsg::Proposal(signed_proposal.clone()),
+            Effect::WalAppend(
+                WalEntry::ConsensusMsg(SignedConsensusMsg::Proposal(signed_proposal.clone())),
                 Default::default()
             )
         );

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

@@ -1,52 +1,61 @@
 use crate::prelude::*;
 
 use crate::handle::driver::apply_driver_input;
-use crate::types::{LocallyProposedValue, ProposedValue};
+use crate::types::{LocallyProposedValue, ProposedValue, WalEntry};
 
 pub async fn on_propose<Ctx>(
     co: &Co<Ctx>,
     state: &mut State<Ctx>,
     metrics: &Metrics,
-    value: LocallyProposedValue<Ctx>,
+    local_value: LocallyProposedValue<Ctx>,
 ) -> Result<(), Error<Ctx>>
 where
     Ctx: Context,
 {
-    let LocallyProposedValue {
-        height,
-        round,
-        value,
-    } = value;
-
-    if state.driver.height() != height {
+    if state.driver.height() != local_value.height {
         warn!(
-            "Ignoring proposal for height {height}, current height: {}",
+            "Ignoring proposal for height {}, current height: {}",
+            local_value.height,
             state.driver.height()
         );
 
         return Ok(());
     }
 
-    if state.driver.round() != round {
+    if state.driver.round() != local_value.round {
         warn!(
-            "Ignoring propose value for round {round}, current round: {}",
+            "Ignoring proposal for round {}, current round: {}",
+            local_value.round,
             state.driver.round()
         );
 
         return Ok(());
     }
 
-    #[cfg(feature = "metrics")]
-    metrics.consensus_start();
-
-    state.store_value(&ProposedValue {
-        height,
-        round,
+    let proposed_value = ProposedValue {
+        height: local_value.height,
+        round: local_value.round,
         valid_round: Round::Nil,
         proposer: state.address().clone(),
-        value: value.clone(),
+        value: local_value.value.clone(),
         validity: Validity::Valid,
-    });
+    };
+
+    state.store_value(&proposed_value);
+
+    #[cfg(feature = "metrics")]
+    metrics.consensus_start();
 
-    apply_driver_input(co, state, metrics, DriverInput::ProposeValue(round, value)).await
+    perform!(
+        co,
+        Effect::WalAppend(WalEntry::ProposedValue(proposed_value), Default::default())
+    );
+
+    apply_driver_input(
+        co,
+        state,
+        metrics,
+        DriverInput::ProposeValue(local_value.round, local_value.value),
+    )
+    .await
 }

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

@@ -1,7 +1,7 @@
 use crate::prelude::*;
 
 use crate::handle::driver::apply_driver_input;
-use crate::types::ProposedValue;
+use crate::types::{ProposedValue, WalEntry};
 
 use super::decide::try_decide;
 use super::signature::sign_proposal;
@@ -54,7 +54,15 @@ where
         state.store_proposal(signed_proposal);
     }
 
-    let proposals = state.full_proposals_for_value(&proposed_value);
+    let validity = proposed_value.validity;
+    let proposals = state.proposals_for_value(&proposed_value);
+
+    // Append the proposed value to the WAL, so it can be used for recovery.
+    perform!(
+        co,
+        Effect::WalAppend(WalEntry::ProposedValue(proposed_value), Default::default())
+    );
+
     for signed_proposal in proposals {
         debug!(
             proposal.height = %signed_proposal.height(),
@@ -66,7 +74,7 @@ where
             co,
             state,
             metrics,
-            DriverInput::Proposal(signed_proposal, proposed_value.validity),
+            DriverInput::Proposal(signed_proposal, validity),
         )
         .await?;
     }

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

@@ -24,10 +24,10 @@ where
     debug_assert_eq!(state.height(), height);
     debug_assert_eq!(state.round(), Round::Nil);
 
-    start_height(co, state, metrics, height).await
+    on_start_height(co, state, metrics, height).await
 }
 
-pub async fn start_height<Ctx>(
+async fn on_start_height<Ctx>(
     co: &Co<Ctx>,
     state: &mut State<Ctx>,
     metrics: &Metrics,
@@ -36,6 +36,9 @@ pub async fn start_height<Ctx>(
 where
     Ctx: Context,
 {
+    debug_assert_eq!(state.height(), height);
+    debug_assert_eq!(state.round(), Round::Nil);
+
     let round = Round::new(0);
     info!(%height, "Starting new height");
 

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

@@ -3,6 +3,7 @@ use crate::handle::driver::apply_driver_input;
 use crate::handle::rebroadcast_timeout::on_rebroadcast_timeout;
 use crate::handle::step_timeout::on_step_limit_timeout;
 use crate::prelude::*;
+use crate::types::WalEntry;
 
 pub async fn on_timeout_elapsed<Ctx>(
     co: &Co<Ctx>,
@@ -41,7 +42,10 @@ where
     ) {
         // Persist the timeout in the Write-ahead Log.
         // Time-limit and rebroadcast timeouts are not persisted because they only occur when consensus is stuck.
-        perform!(co, Effect::WalAppendTimeout(timeout, Default::default()));
+        perform!(
+            co,
+            Effect::WalAppend(WalEntry::Timeout(timeout), Default::default())
+        );
     }
 
     apply_driver_input(co, state, metrics, DriverInput::TimeoutElapsed(timeout)).await?;

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

@@ -5,9 +5,8 @@ use crate::handle::signature::verify_signature;
 use crate::handle::validator_set::get_validator_set;
 use crate::input::Input;
 use crate::prelude::*;
-use crate::types::ConsensusMsg;
+use crate::types::{ConsensusMsg, SignedConsensusMsg, WalEntry};
 use crate::util::pretty::PrettyVote;
-use crate::SignedConsensusMsg;
 
 pub async fn on_vote<Ctx>(
     co: &Co<Ctx>,
@@ -82,8 +81,8 @@ where
         // Append the vote to the Write-ahead Log
         perform!(
             co,
-            Effect::WalAppendMessage(
-                SignedConsensusMsg::Vote(signed_vote.clone()),
+            Effect::WalAppend(
+                WalEntry::ConsensusMsg(SignedConsensusMsg::Vote(signed_vote.clone())),
                 Default::default()
             )
         );

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

@@ -15,7 +15,7 @@ pub enum Input<Ctx>
 where
     Ctx: Context,
 {
-    /// Start a new height with the given validator set.
+    /// Start consensus for the given height with the given validator set
     StartHeight(Ctx::Height, Ctx::ValidatorSet),
 
     /// Process a vote received over the network.

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

@@ -174,12 +174,22 @@ where
             .full_proposal_at_round_and_value(height, round, &value.id())
     }
 
-    pub fn full_proposals_for_value(
+    pub fn full_proposal_at_round_and_proposer(
+        &self,
+        height: &Ctx::Height,
+        round: Round,
+        address: &Ctx::Address,
+    ) -> Option<&FullProposal<Ctx>> {
+        self.full_proposal_keeper
+            .full_proposal_at_round_and_proposer(height, round, address)
+    }
+
+    pub fn proposals_for_value(
         &self,
         proposed_value: &ProposedValue<Ctx>,
     ) -> Vec<SignedProposal<Ctx>> {
         self.full_proposal_keeper
-            .full_proposals_for_value(proposed_value)
+            .proposals_for_value(proposed_value)
     }
 
     pub fn store_proposal(&mut self, new_proposal: SignedProposal<Ctx>) {