[new-protocol] garbage collect consensus data for views less than decided view (#4415)

* gc for view < decided view

* fix snapshot

* gc on every decide

* fix test

Author: imabdulbasit

crates/espresso/node/src/context.rs

@@ -2,7 +2,6 @@ use std::{
     fmt::{Debug, Display},
     future::Future,
     marker::PhantomData,
-    num::NonZeroU64,
     sync::Arc,
     time::{Duration, Instant},
 };
@@ -126,7 +125,6 @@ where
         event_consumer: impl PersistenceEventConsumer + 'static,
         proposal_fetcher_cfg: ProposalFetcherConfig,
         bootstrap_epoch_catchup_timeout: Duration,
-        new_protocol_consensus_gc_interval: NonZeroU64,
     ) -> anyhow::Result<Self> {
         let config = &network_config.config;
         let pub_key = validator_config.public_key;
@@ -213,7 +211,6 @@ where
             .stake_table_capacity(stake_table_capacity)
             .timeout_duration(Duration::from_secs(10))
             .storage(Arc::clone(&persistence))
-            .garbage_collection_interval(new_protocol_consensus_gc_interval.get())
             .metrics(metrics)
             .make();
 

crates/espresso/node/src/lib.rs

@@ -19,7 +19,7 @@ pub mod state_cert;
 pub mod state_signature;
 pub mod util;
 
-use std::{fmt::Debug, marker::PhantomData, num::NonZeroU64, sync::Arc, time::Duration};
+use std::{fmt::Debug, marker::PhantomData, sync::Arc, time::Duration};
 
 use alloy::primitives::U256;
 use anyhow::Context;
@@ -136,8 +136,6 @@ pub struct NetworkParams {
     /// Per-step timeout for the startup stake-table catchup walk
     /// (`bootstrap_epoch_window`).
     pub bootstrap_epoch_catchup_timeout: Duration,
-    /// Number of blocks between new-protocol consensus garbage collection passes.
-    pub new_protocol_consensus_gc_interval: NonZeroU64,
     /// The address to advertise as our public API's URL
     pub public_api_url: Option<Url>,
     /// Cliquenet network address.
@@ -837,7 +835,6 @@ where
         event_consumer,
         proposal_fetcher_config,
         network_params.bootstrap_epoch_catchup_timeout,
-        network_params.new_protocol_consensus_gc_interval,
     )
     .await?;
 
@@ -1667,7 +1664,6 @@ pub mod testing {
                 event_consumer,
                 Default::default(),
                 Duration::from_secs(2),
-                NonZeroU64::new(100).unwrap(),
             )
             .await
             .unwrap()

crates/espresso/node/src/options.rs

@@ -6,7 +6,6 @@ use std::{
     collections::HashSet,
     fmt::{self, Formatter},
     iter::once,
-    num::NonZeroU64,
     path::PathBuf,
     time::Duration,
 };
@@ -341,18 +340,6 @@ pub struct Options {
     #[clap(long, env = "ESPRESSO_NODE_BOOTSTRAP_EPOCH_CATCHUP_TIMEOUT", default_value = "30s", value_parser = parse_duration)]
     pub bootstrap_epoch_catchup_timeout: Duration,
 
-    /// Number of blocks between new-protocol garbage collection passes.
-    ///
-    /// Controls how often the new-protocol coordinator triggers garbage
-    /// collection of decided views/epochs across consensus state, VID,
-    /// vote collectors, and storage.
-    #[clap(
-        long,
-        env = "ESPRESSO_NODE_NEW_PROTOCOL_CONSENSUS_GC_INTERVAL",
-        default_value = "100"
-    )]
-    pub new_protocol_consensus_gc_interval: NonZeroU64,
-
     #[clap(flatten)]
     pub logging: logging::Config,
 
@@ -691,7 +678,6 @@ pub struct PublicNodeConfig {
     pub catchup_base_timeout: Duration,
     pub local_catchup_timeout: Duration,
     pub bootstrap_epoch_catchup_timeout: Duration,
-    pub new_protocol_consensus_gc_interval: NonZeroU64,
     pub catchup_backoff: BackoffParams,
     pub proposal_fetcher: ProposalFetcherConfig,
     pub libp2p: Libp2pTuning,
@@ -1031,7 +1017,6 @@ impl PublicNodeConfig {
             catchup_base_timeout: opt.catchup_base_timeout,
             local_catchup_timeout: opt.local_catchup_timeout,
             bootstrap_epoch_catchup_timeout: opt.bootstrap_epoch_catchup_timeout,
-            new_protocol_consensus_gc_interval: opt.new_protocol_consensus_gc_interval,
             catchup_backoff: opt.catchup_backoff,
             proposal_fetcher: opt.proposal_fetcher_config,
             libp2p: Libp2pTuning::from(opt),

crates/espresso/node/src/run.rs

@@ -103,7 +103,6 @@ where
         catchup_base_timeout: opt.catchup_base_timeout,
         local_catchup_timeout: opt.local_catchup_timeout,
         bootstrap_epoch_catchup_timeout: opt.bootstrap_epoch_catchup_timeout,
-        new_protocol_consensus_gc_interval: opt.new_protocol_consensus_gc_interval,
         libp2p_history_gossip: opt.libp2p_history_gossip,
         libp2p_history_length: opt.libp2p_history_length,
         libp2p_max_ihave_length: opt.libp2p_max_ihave_length,

crates/espresso/node/src/snapshots/espresso_node__options__tests__config_node_response_postgres.snap

@@ -48,7 +48,6 @@ local_catchup_timeout:
 bootstrap_epoch_catchup_timeout:
   secs: 30
   nanos: 0
-new_protocol_consensus_gc_interval: 100
 catchup_backoff:
   factor: 4
   base:

crates/hotshot/new-protocol/bench/src/node.rs

@@ -131,14 +131,12 @@ async fn build_coordinator(
         upgrade_lock.clone(),
         genesis_leaf.clone(),
         epoch_height,
-        100,
     );
 
     let vote1_collector = VoteCollector::new(membership.clone(), upgrade_lock.clone());
     let vote2_collector = VoteCollector::new(membership.clone(), upgrade_lock.clone());
     let timeout_collector = VoteCollector::new(membership.clone(), upgrade_lock.clone());
     let timeout_one_honest_collector = VoteCollector::new(membership.clone(), upgrade_lock.clone());
-    let checkpoint_collector = VoteCollector::new(membership.clone(), upgrade_lock.clone());
     let epoch_root_collector =
         EpochRootVoteCollector::new(membership.clone(), upgrade_lock.clone());
 
@@ -198,7 +196,6 @@ async fn build_coordinator(
         .vote2_collector(vote2_collector)
         .timeout_collector(timeout_collector)
         .timeout_one_honest_collector(timeout_one_honest_collector)
-        .checkpoint_collector(checkpoint_collector)
         .epoch_root_collector(epoch_root_collector)
         .vid_disperser(vid_disperser)
         .vid_reconstructor(vid_reconstructor)

crates/hotshot/new-protocol/src/consensus.rs

@@ -21,8 +21,8 @@ use hotshot_types::{
         check_qc_state_cert_correspondence,
     },
     simple_vote::{
-        CheckpointData, HasEpoch, LightClientStateUpdateVote2, QuorumData2, SimpleVote,
-        TimeoutData2, TimeoutVote2, Vote2Data,
+        HasEpoch, LightClientStateUpdateVote2, QuorumData2, SimpleVote, TimeoutData2, TimeoutVote2,
+        Vote2Data,
     },
     stake_table::{HSStakeTable, StakeTableEntries},
     traits::{
@@ -42,8 +42,8 @@ use crate::{
     helpers::proposal_commitment,
     logging::KeyPrefix,
     message::{
-        Certificate1, Certificate2, CheckpointVote, EpochChangeMessage, Proposal,
-        ProposalFetchRequest, ProposalMessage, Validated, VidShareMessage, Vote1, Vote2,
+        Certificate1, Certificate2, EpochChangeMessage, Proposal, ProposalFetchRequest,
+        ProposalMessage, Validated, VidShareMessage, Vote1, Vote2,
     },
     outbox::Outbox,
     state::{StateRequest, StateResponse},
@@ -115,7 +115,6 @@ pub enum ConsensusOutput<T: NodeType> {
     RequestDrbResult(EpochNumber),
     SendProposal(SignedProposal<T, Proposal<T>>),
     SendVidShares(Vec<VidShareMessage<T>>),
-    SendCheckpointVote(CheckpointVote<T>),
     SendTimeoutVote(TimeoutVote2<T>, Option<Certificate1<T>>),
     SendVote1(Vote1<T>),
     SendVote2(Vote2<T>),
@@ -199,7 +198,6 @@ pub struct Consensus<T: NodeType> {
     node_id: KeyPrefix,
     upgrade_lock: UpgradeLock<T>,
 
-    garbage_collection_interval: BlockNumber,
     pub(crate) epoch_height: BlockNumber,
 }
 
@@ -232,7 +230,6 @@ impl<T: NodeType> Consensus<T> {
         upgrade_lock: UpgradeLock<T>,
         genesis_leaf: Leaf2<T>,
         epoch_height: B,
-        garbage_collection_interval: B,
     ) -> Self
     where
         B: Into<BlockNumber>,
@@ -270,7 +267,6 @@ impl<T: NodeType> Consensus<T> {
             state_certs: BTreeMap::new(),
             upgrade_lock,
             vid_shares: BTreeMap::new(),
-            garbage_collection_interval: garbage_collection_interval.into(),
             epoch_height: epoch_height.into(),
         }
     }
@@ -1360,10 +1356,6 @@ impl<T: NodeType> Consensus<T> {
         }
         let new_decided_view = max(self.last_decided_view, leaf.view_number());
         let last_decided_leaf = leaf.clone();
-        let mut gc = None;
-        if leaf.block_header().block_number() % *self.garbage_collection_interval == 0 {
-            gc = Some((leaf.view_number(), leaf.justify_qc().epoch()));
-        }
         let mut decided = vec![leaf];
         let mut vid_shares = vec![self.signed_vid_share(view)];
 
@@ -1381,11 +1373,6 @@ impl<T: NodeType> Consensus<T> {
             if let Some(payload) = self.blocks.get(&parent_view) {
                 leaf.fill_block_payload_unchecked(payload.clone());
             }
-            if gc.is_none()
-                && leaf.block_header().block_number() % *self.garbage_collection_interval == 0
-            {
-                gc = Some((leaf.view_number(), leaf.justify_qc().epoch()));
-            }
             vid_shares.push(self.signed_vid_share(parent_view));
             decided.push(leaf);
             parent_view = proposal.justify_qc.view_number();
@@ -1403,26 +1390,6 @@ impl<T: NodeType> Consensus<T> {
             cert2: Some(cert2.clone()),
             vid_shares,
         });
-        if let Some(gc) = gc {
-            let gc_data = CheckpointData {
-                view: gc.0,
-                epoch: gc.1.unwrap_or_default(),
-            };
-            let vote = match SimpleVote::create_signed_vote(
-                gc_data,
-                view,
-                &self.public_key,
-                &self.private_key,
-                &self.upgrade_lock,
-            ) {
-                Ok(vote) => vote,
-                Err(err) => {
-                    warn!(%view, %err, "failed to create signed checkpoint vote");
-                    return;
-                },
-            };
-            outbox.push_back(ConsensusOutput::SendCheckpointVote(vote));
-        }
     }
 
     /// Build a `LightClientStateUpdateVote2` for an epoch-root leaf.

crates/hotshot/new-protocol/src/coordinator.rs

@@ -42,9 +42,9 @@ use crate::{
     helpers::proposal_commitment,
     logging::KeyPrefix,
     message::{
-        self, BlockMessage, Certificate2, CheckpointCertificate, CheckpointVote, ConsensusMessage,
-        Message, MessageType, Proposal, ProposalFetchMessage, ProposalMessage, TimeoutOneHonest,
-        TransactionMessage, Unchecked, Vote2,
+        self, BlockMessage, Certificate2, ConsensusMessage, Message, MessageType, Proposal,
+        ProposalFetchMessage, ProposalMessage, TimeoutOneHonest, TransactionMessage, Unchecked,
+        Vote2,
     },
     network::Network,
     outbox::Outbox,
@@ -78,7 +78,6 @@ pub struct Coordinator<T: NodeType, N, S> {
     vote2_collector: VoteCollector<T, Vote2<T>, Certificate2<T>>,
     timeout_collector: VoteCollector<T, TimeoutVote2<T>, TimeoutCertificate2<T>>,
     timeout_one_honest_collector: VoteCollector<T, TimeoutVote2<T>, TimeoutOneHonest<T>>,
-    checkpoint_collector: VoteCollector<T, CheckpointVote<T>, CheckpointCertificate<T>>,
     epoch_root_collector: EpochRootVoteCollector<T>,
     epoch_manager: EpochManager<T>,
     block_builder: BlockBuilder<T>,
@@ -122,7 +121,6 @@ where
         stake_table_capacity: usize,
         timeout_duration: Duration,
         storage: S,
-        garbage_collection_interval: u64,
         metrics: &dyn Metrics,
     ) -> Self {
         let mut consensus = Consensus::new(
@@ -134,7 +132,6 @@ where
             upgrade_lock.clone(),
             initializer.anchor_leaf.clone(),
             initializer.epoch_height,
-            garbage_collection_interval,
         );
 
         let genesis_cert1 = initializer.high_qc.clone();
@@ -193,10 +190,6 @@ where
                 membership_coordinator.clone(),
                 lock.clone(),
             ))
-            .checkpoint_collector(VoteCollector::new(
-                membership_coordinator.clone(),
-                lock.clone(),
-            ))
             .epoch_root_collector(EpochRootVoteCollector::new(
                 membership_coordinator.clone(),
                 lock,
@@ -425,14 +418,6 @@ where
                         return Err(CoordinatorError::regular(e).context("proposal validation"))
                     }
                 },
-                Some(cert) = self.checkpoint_collector.next() => {
-                    finish_measurement(next_input);
-                    let Some(epoch) = cert.epoch() else {
-                        let msg = format!("missing epoch in view {}", cert.view_number());
-                        return Err(CoordinatorError::critical(msg).context("gc certificate"))
-                    };
-                    self.gc(cert.view_number(), epoch);
-                }
                 Some(item) = self.block_builder.next() => match item {
                     Ok(block) => {
                         finish_measurement(next_input);
@@ -571,20 +556,6 @@ where
                 debug!(%node, %epoch, "request drb result");
                 self.epoch_manager.request_drb_result(epoch);
             },
-            ConsensusOutput::SendCheckpointVote(checkpoint_vote) => {
-                let view = checkpoint_vote.view_number();
-                let epoch = checkpoint_vote.data.epoch;
-                debug!(%node, %view, %epoch, "send checkpoint vote");
-                let message = Message {
-                    sender: self.public_key.clone(),
-                    message_type: MessageType::Consensus(ConsensusMessage::Checkpoint(
-                        checkpoint_vote,
-                    )),
-                };
-                self.network
-                    .broadcast(message.view_number(), &message)
-                    .map_err(|e| CoordinatorError::from(e).context("broadcast checkpoint vote"))?
-            },
             ConsensusOutput::LeafDecided {
                 leaves,
                 cert1,
@@ -601,6 +572,13 @@ where
                 if let Some(cert2) = cert2 {
                     self.storage.append_cert2(cert2.view_number, cert2.clone());
                 }
+                // `leaves` is ordered newest first
+                //  garbage collect the data for views < decided view
+                if let Some(newest) = leaves.first() {
+                    let gc_view = newest.view_number();
+                    let gc_epoch = newest.justify_qc().epoch().unwrap_or_default();
+                    self.gc(gc_view, gc_epoch);
+                }
                 for leaf in leaves {
                     self.epoch_manager.handle_leaf_decided(leaf);
                 }
@@ -940,16 +918,6 @@ where
                     );
                     Some(ConsensusInput::EpochChange(epoch_change))
                 },
-                ConsensusMessage::Checkpoint(checkpoint) => {
-                    debug!(
-                        %node, %sender,
-                        view = %checkpoint.view_number(),
-                        epoch = %checkpoint.data.epoch,
-                        "recv checkpoint vote"
-                    );
-                    self.checkpoint_collector.accumulate_vote(checkpoint).await;
-                    None
-                },
             },
             MessageType::Block(msg) => {
                 match msg {
@@ -1401,7 +1369,6 @@ where
     fn gc(&mut self, view: ViewNumber, epoch: EpochNumber) {
         info!(node = %self.node_id, %view, "garbage collecting");
         self.consensus.gc(view, epoch);
-        self.checkpoint_collector.gc(view, epoch);
         let _ = self.network.gc(view); // TODO
         self.state_manager.gc(view);
         self.vid_disperser.gc(view);