feat(code): Add the capability to re-run consensus for a given height (#952)

* 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

* Add and handle  ResetHeight

* More consistent method names

* Add `is_restart` to `StartedHeight` event

* Use middleware in test app to reset height

* Remove height reset from example app

* Add test for resetting height

* Only reset the height once in test

* Reset sync tip height when restarting a height

* Restore original sync inputs, add `restart` to `StartedHeight`

* Cleanup

* Add a release note

* Add warnings to the adr, tutorial and comments

* Fix spelling

---------

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

Author: ancazamfir

RELEASE_NOTES.md

@@ -1,5 +1,9 @@
 # Release Notes
 
+## Unreleased
+
+- Add the capability to re-run consensus for a given height ([#893](https://github.com/informalsystems/malachite/issues/893))
+
 ## 0.1.0
 
 *April 9th, 2025*

code/crates/app-channel/src/msgs.rs

@@ -213,6 +213,9 @@ pub enum ConsensusMsg<Ctx: Context> {
 
     /// Previousuly received value proposed by a validator
     ReceivedProposedValue(ProposedValue<Ctx>, ValueOrigin),
+
+    /// Instructs consensus to restart at a given height with the given validator set.
+    RestartHeight(Ctx::Height, Ctx::ValidatorSet),
 }
 
 impl<Ctx: Context> From<ConsensusMsg<Ctx>> for ConsensusActorMsg<Ctx> {
@@ -224,6 +227,9 @@ impl<Ctx: Context> From<ConsensusMsg<Ctx>> for ConsensusActorMsg<Ctx> {
             ConsensusMsg::ReceivedProposedValue(value, origin) => {
                 ConsensusActorMsg::ReceivedProposedValue(value, origin)
             }
+            ConsensusMsg::RestartHeight(height, validator_set) => {
+                ConsensusActorMsg::RestartHeight(height, validator_set)
+            }
         }
     }
 }

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

@@ -12,12 +12,11 @@ where
 {
     warn!(
         height = %state.driver.height(), %round,
-        "Consensus is halted in {:?} step", state.driver.step());
+        "Consensus is halted in {:?} step", state.driver.step()
+    );
 
     if state.params.vote_sync_mode == VoteSyncMode::RequestResponse {
-        warn!(
-            height = %state.driver.height(), %round,
-            "Requesting vote set");
+        warn!(height = %state.driver.height(), %round, "Requesting vote set");
 
         perform!(
             co,

code/crates/engine/src/consensus.rs

@@ -99,6 +99,16 @@ pub enum Msg<Ctx: Context> {
 
     /// Received and assembled the full value proposed by a validator
     ReceivedProposedValue(ProposedValue<Ctx>, ValueOrigin),
+
+    /// Instructs consensus to restart at a given height with the given validator set.
+    ///
+    /// On this input consensus resets the Write-Ahead Log.
+    /// # Warning
+    /// This operation should be used with extreme caution as it can lead to safety violations:
+    /// 1. The application must clean all state associated with the height for which commit has failed
+    /// 2. Since consensus resets its write-ahead log, the node may equivocate on proposals and votes
+    ///    for the restarted height, potentially violating protocol safety
+    RestartHeight(Ctx::Height, Ctx::ValidatorSet),
 }
 
 impl<Ctx: Context> From<NetworkEvent<Ctx>> for Msg<Ctx> {
@@ -298,12 +308,20 @@ where
         state: &mut State<Ctx>,
         msg: Msg<Ctx>,
     ) -> Result<(), ActorProcessingErr> {
+        let is_restart = matches!(msg, Msg::RestartHeight(_, _));
+
         match msg {
-            Msg::StartHeight(height, validator_set) => {
-                self.tx_event.send(|| Event::StartedHeight(height));
+            Msg::StartHeight(height, validator_set) | Msg::RestartHeight(height, validator_set) => {
+                self.tx_event
+                    .send(|| Event::StartedHeight(height, is_restart));
 
-                // Fetch entries from the WAL
-                let wal_entries = self.wal_fetch(height).await?;
+                // Fetch entries from the WAL or reset the WAL if this is a restart
+                let wal_entries = if is_restart {
+                    self.wal_reset(height).await?;
+                    vec![]
+                } else {
+                    self.wal_fetch(height).await?
+                };
 
                 if !wal_entries.is_empty() {
                     // Set the phase to `Recovering` while we replay the WAL
@@ -324,8 +342,7 @@ where
                 }
 
                 if !wal_entries.is_empty() {
-                    // Replay the entries from the WAL
-                    self.replay_wal(&myself, state, height, wal_entries).await;
+                    self.wal_replay(&myself, state, height, wal_entries).await;
                 }
 
                 // Set the phase to `Running` now that we have replayed the WAL
@@ -336,7 +353,7 @@ where
 
                 // Notify the sync actor that we have started a new height
                 if let Some(sync) = &self.sync {
-                    if let Err(e) = sync.cast(SyncMsg::StartedHeight(height)) {
+                    if let Err(e) = sync.cast(SyncMsg::StartedHeight(height, is_restart)) {
                         error!(%height, "Error when notifying sync of started height: {e}")
                     }
                 }
@@ -615,6 +632,24 @@ where
         Ok(())
     }
 
+    async fn wal_reset(&self, height: Ctx::Height) -> Result<(), ActorProcessingErr> {
+        let result = ractor::call!(self.wal, WalMsg::Reset, height);
+
+        match result {
+            Ok(Ok(())) => {
+                // Success
+            }
+            Ok(Err(e)) => {
+                error!("Resetting the WAL failed: {e}");
+            }
+            Err(e) => {
+                error!("Failed to send Reset command to WAL actor: {e}");
+            }
+        }
+
+        Ok(())
+    }
+
     async fn wal_fetch(
         &self,
         height: Ctx::Height,
@@ -643,7 +678,7 @@ where
         }
     }
 
-    async fn replay_wal(
+    async fn wal_replay(
         &self,
         myself: &ActorRef<Msg<Ctx>>,
         state: &mut State<Ctx>,

code/crates/engine/src/network.rs

@@ -126,14 +126,14 @@ pub enum State<Ctx: Context> {
 
 #[derive_where(Clone, Debug, PartialEq, Eq)]
 pub struct Status<Ctx: Context> {
-    pub height: Ctx::Height,
+    pub tip_height: Ctx::Height,
     pub history_min_height: Ctx::Height,
 }
 
 impl<Ctx: Context> Status<Ctx> {
-    pub fn new(height: Ctx::Height, history_min_height: Ctx::Height) -> Self {
+    pub fn new(tip_height: Ctx::Height, history_min_height: Ctx::Height) -> Self {
         Self {
-            height,
+            tip_height,
             history_min_height,
         }
     }
@@ -272,7 +272,7 @@ where
             Msg::BroadcastStatus(status) => {
                 let status = sync::Status {
                     peer_id: ctrl_handle.peer_id(),
-                    height: status.height,
+                    tip_height: status.tip_height,
                     history_min_height: status.history_min_height,
                 };
 
@@ -380,11 +380,11 @@ where
                     return Ok(());
                 }
 
-                trace!(%from, height = %status.height, "Received status");
+                trace!(%from, tip_height = %status.tip_height, "Received status");
 
                 output_port.send(NetworkEvent::Status(
                     status.peer_id,
-                    Status::new(status.height, status.history_min_height),
+                    Status::new(status.tip_height, status.history_min_height),
                 ));
             }
 

code/crates/engine/src/sync.rs

@@ -13,7 +13,7 @@ use tracing::{debug, error, info, warn, Instrument};
 
 use malachitebft_codec as codec;
 use malachitebft_core_consensus::PeerId;
-use malachitebft_core_types::{CertificateError, CommitCertificate, Context, Height, Round};
+use malachitebft_core_types::{CertificateError, CommitCertificate, Context, Round};
 use malachitebft_sync::{self as sync, InboundRequestId, OutboundRequestId, Response};
 use malachitebft_sync::{RawDecidedValue, Request};
 
@@ -82,8 +82,9 @@ pub enum Msg<Ctx: Context> {
     /// Consensus has decided on a value at the given height
     Decided(Ctx::Height),
 
-    /// Consensus has started a new height
-    StartedHeight(Ctx::Height),
+    /// Consensus has (re)started a new height.
+    /// The boolean indicates whether this is a restart or not.
+    StartedHeight(Ctx::Height, bool),
 
     /// Host has a response for the blocks request
     GotDecidedBlock(InboundRequestId, Ctx::Height, Option<RawDecidedValue<Ctx>>),
@@ -351,7 +352,7 @@ where
             Msg::NetworkEvent(NetworkEvent::Status(peer_id, status)) => {
                 let status = sync::Status {
                     peer_id,
-                    height: status.height,
+                    tip_height: status.tip_height,
                     history_min_height: status.history_min_height,
                 };
 
@@ -408,18 +409,15 @@ where
                 // Ignore other gossip events
             }
 
-            Msg::Decided(height) => {
-                self.process_input(&myself, state, sync::Input::UpdateHeight(height))
-                    .await?;
+            // (Re)Started a new height
+            Msg::StartedHeight(height, restart) => {
+                self.process_input(&myself, state, sync::Input::StartedHeight(height, restart))
+                    .await?
             }
 
-            Msg::StartedHeight(height) => {
-                if let Some(height) = height.decrement() {
-                    self.process_input(&myself, state, sync::Input::UpdateHeight(height))
-                        .await?;
-                }
-
-                self.process_input(&myself, state, sync::Input::StartHeight(height))
+            // Decided on a value
+            Msg::Decided(height) => {
+                self.process_input(&myself, state, sync::Input::Decided(height))
                     .await?;
             }
 

code/crates/engine/src/util/events.rs

@@ -42,7 +42,7 @@ impl<Ctx: Context> Default for TxEvent<Ctx> {
 
 #[derive_where(Clone, Debug)]
 pub enum Event<Ctx: Context> {
-    StartedHeight(Ctx::Height),
+    StartedHeight(Ctx::Height, bool),
     StartedRound(Ctx::Height, Round),
     Published(SignedConsensusMsg<Ctx>),
     ProposedValue(LocallyProposedValue<Ctx>),
@@ -60,7 +60,9 @@ pub enum Event<Ctx: Context> {
 impl<Ctx: Context> fmt::Display for Event<Ctx> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
-            Event::StartedHeight(height) => write!(f, "StartedHeight(height: {height})"),
+            Event::StartedHeight(height, restart) => {
+                write!(f, "StartedHeight(height: {height}, restart: {restart})")
+            }
             Event::StartedRound(height, round) => {
                 write!(f, "StartedRound(height: {height}, round: {round})")
             }

code/crates/engine/src/wal.rs

@@ -53,6 +53,7 @@ pub type WalReply<T> = RpcReplyPort<eyre::Result<T>>;
 
 pub enum Msg<Ctx: Context> {
     StartedHeight(Ctx::Height, WalReply<Option<Vec<WalEntry<Ctx>>>>),
+    Reset(Ctx::Height, WalReply<()>),
     Append(Ctx::Height, WalEntry<Ctx>, WalReply<()>),
     Flush(WalReply<()>),
     Dump,
@@ -92,6 +93,10 @@ where
                 self.started_height(state, height, reply_to).await?;
             }
 
+            Msg::Reset(height, reply_to) => {
+                self.reset(state, height, reply_to).await?;
+            }
+
             Msg::Append(height, entry, reply_to) => {
                 if height != state.height {
                     debug!("Ignoring append at height {} != {}", height, state.height);
@@ -113,6 +118,28 @@ where
         Ok(())
     }
 
+    async fn reset(
+        &self,
+        state: &mut State<Ctx>,
+        height: Ctx::Height,
+        reply_to: WalReply<()>,
+    ) -> Result<(), ActorProcessingErr> {
+        let (tx, rx) = oneshot::channel();
+
+        state
+            .wal_sender
+            .send(self::thread::WalMsg::Reset(height, tx))
+            .await?;
+
+        let result = rx.await?;
+
+        reply_to
+            .send(result)
+            .map_err(|e| eyre!("Failed to send reply: {e}"))?;
+
+        Ok(())
+    }
+
     async fn started_height(
         &self,
         state: &mut State<Ctx>,

code/crates/engine/src/wal/thread.rs

@@ -16,6 +16,7 @@ pub type ReplyTo<T> = oneshot::Sender<Result<T>>;
 
 pub enum WalMsg<Ctx: Context> {
     StartedHeight(Ctx::Height, ReplyTo<Vec<WalEntry<Ctx>>>),
+    Reset(Ctx::Height, ReplyTo<()>),
     Append(WalEntry<Ctx>, ReplyTo<()>),
     Flush(ReplyTo<()>),
     Shutdown,
@@ -91,6 +92,18 @@ where
             }
         }
 
+        WalMsg::Reset(height, reply) => {
+            let sequence = height.as_u64();
+
+            let result = log.restart(sequence).map_err(Into::into);
+
+            debug!(%height, "Reset WAL");
+
+            if reply.send(result).is_err() {
+                error!("Failed to send WAL reset reply");
+            }
+        }
+
         WalMsg::Append(entry, reply) => {
             let tpe = wal_entry_type(&entry);
 

code/crates/starknet/host/src/codec.rs

@@ -173,7 +173,7 @@ pub fn decode_sync_status(
 
     Ok(sync::Status {
         peer_id: decode_peer_id(peer_id)?,
-        height: Height::new(status.block_number, status.fork_id),
+        tip_height: Height::new(status.block_number, status.fork_id),
         history_min_height: Height::new(status.earliest_block_number, status.earliest_fork_id),
     })
 }
@@ -183,8 +183,8 @@ pub fn encode_sync_status(
 ) -> Result<proto::sync::Status, ProtoError> {
     Ok(proto::sync::Status {
         peer_id: Some(encode_peer_id(&status.peer_id)?),
-        block_number: status.height.block_number,
-        fork_id: status.height.fork_id,
+        block_number: status.tip_height.block_number,
+        fork_id: status.tip_height.fork_id,
         earliest_block_number: status.history_min_height.block_number,
         earliest_fork_id: status.history_min_height.fork_id,
     })