feat: weighted round-robin proposer selection

Author: qj0r9j0vc2

crates/consensus/src/context.rs

@@ -1,10 +1,13 @@
+use std::sync::Arc;
+
 use informalsystems_malachitebft_core_types::{
     Context as MalachiteContext, NilOrVal, Round, ValueId, VoteType,
 };
 
 use crate::config::ConsensusConfig;
 use crate::error::ConsensusError;
 use crate::proposal::{CutProposal, CutProposalPart};
+use crate::proposer_selector::ProposerSelector;
 use crate::signing::Ed25519SigningScheme;
 use crate::types::{ConsensusHeight, ConsensusValue};
 use crate::validator_set::{ConsensusAddress, ConsensusValidator, ConsensusValidatorSet};
@@ -32,6 +35,8 @@ pub struct CipherBftContext {
     pub validator_set: ConsensusValidatorSet,
     /// Height the engine should start from.
     pub initial_height: ConsensusHeight,
+    /// Proposer selector using Tendermint's weighted round-robin algorithm.
+    proposer_selector: Arc<ProposerSelector>,
 }
 
 impl CipherBftContext {
@@ -47,10 +52,12 @@ impl CipherBftContext {
         if validator_set.is_empty() {
             return Err(ConsensusError::EmptyValidatorSet);
         }
+        let proposer_selector = Arc::new(ProposerSelector::new(&validator_set, initial_height));
         Ok(Self {
             config,
             validator_set,
             initial_height,
+            proposer_selector,
         })
     }
 
@@ -82,19 +89,24 @@ impl CipherBftContext {
         self.config.chain_id()
     }
 
-    /// Deterministic round-robin proposer selection.
-    pub fn proposer_at_round(&self, round: Round) -> Option<ConsensusAddress> {
-        let count = self.validator_set.len();
-        if count == 0 {
+    /// Access the proposer selector for external priority updates.
+    pub fn proposer_selector(&self) -> &Arc<ProposerSelector> {
+        &self.proposer_selector
+    }
+
+    /// Weighted round-robin proposer selection for a given height and round.
+    pub fn proposer_at_round(
+        &self,
+        height: ConsensusHeight,
+        round: Round,
+    ) -> Option<ConsensusAddress> {
+        if self.validator_set.is_empty() {
             return None;
         }
-
-        // Use round index modulo validator count; nil rounds map to first validator.
-        let idx = match round.as_i64() {
-            x if x < 0 => 0,
-            x => (x as usize) % count,
-        };
-        self.validator_set.as_slice().get(idx).map(|v| v.address)
+        let proposer = self
+            .proposer_selector
+            .select_proposer(&self.validator_set, height, round);
+        Some(proposer.address)
     }
 }
 
@@ -113,18 +125,11 @@ impl MalachiteContext for CipherBftContext {
     fn select_proposer<'a>(
         &self,
         validator_set: &'a Self::ValidatorSet,
-        _height: Self::Height,
+        height: Self::Height,
         round: Round,
     ) -> &'a Self::Validator {
-        let count = validator_set.len();
-        let idx = match round.as_i64() {
-            x if x < 0 => 0,
-            x => (x as usize) % count.max(1),
-        };
-        validator_set
-            .as_slice()
-            .get(idx)
-            .expect("validator_set must not be empty")
+        self.proposer_selector
+            .select_proposer(validator_set, height, round)
     }
 
     fn new_proposal(

crates/consensus/src/host.rs

@@ -63,6 +63,17 @@ pub struct HostConfig {
     /// Higher values support longer sync windows for lagging nodes,
     /// at the cost of increased memory usage.
     pub decided_retention: usize,
+
+    /// Timeout in milliseconds to wait for the next cut after a decision (default: 50ms).
+    ///
+    /// After consensus decides on a block, the host waits for the next cut to be
+    /// available before starting the next consensus round. This prevents a race
+    /// condition where consensus requests a value before DCL has produced the cut.
+    ///
+    /// Lower values improve block throughput but may cause NIL votes if cuts
+    /// aren't ready. Higher values reduce throughput but give DCL more time.
+    /// Set to 0 to disable waiting entirely.
+    pub wait_for_cut_timeout_ms: u64,
 }
 
 impl Default for HostConfig {
@@ -76,6 +87,11 @@ impl Default for HostConfig {
             // 5 seconds. This larger window ensures peers can serve historical data for catch-up.
             // Note: Persistent storage (dcl_store) is also used for older blocks.
             decided_retention: 10000,
+            // Reduced from 500ms to 50ms to improve block throughput.
+            // Most cuts should be available within a few milliseconds if DCL is healthy.
+            // If cuts aren't ready, consensus will proceed and may vote NIL on round 0,
+            // but this is preferable to adding 500ms latency per block.
+            wait_for_cut_timeout_ms: 50,
         }
     }
 }
@@ -200,6 +216,9 @@ pub struct CipherBftHost {
     /// Host configuration.
     config: HostConfig,
 
+    /// Proposer selector for weighted round-robin (optional for backward compatibility).
+    proposer_selector: Option<Arc<crate::proposer_selector::ProposerSelector>>,
+
     /// Tracing span for this actor.
     span: tracing::Span,
 }
@@ -226,10 +245,23 @@ impl CipherBftHost {
             value_builder,
             decision_handler,
             config,
+            proposer_selector: None,
             span,
         }
     }
 
+    /// Set the proposer selector for weighted round-robin updates.
+    ///
+    /// When set, the host will update the proposer selector's priorities
+    /// when validator sets change at epoch boundaries.
+    pub fn with_proposer_selector(
+        mut self,
+        proposer_selector: Arc<crate::proposer_selector::ProposerSelector>,
+    ) -> Self {
+        self.proposer_selector = Some(proposer_selector);
+        self
+    }
+
     /// Get the validator set for a specific height.
     ///
     /// This method delegates to `ValidatorSetManager` to return the
@@ -679,7 +711,7 @@ impl Actor for CipherBftHost {
                 }
 
                 // Check for epoch transition
-                let _epoch_transition = match self.on_block_committed(height) {
+                let epoch_transition = match self.on_block_committed(height) {
                     Ok(true) => {
                         info!(
                             parent: &self.span,
@@ -701,10 +733,38 @@ impl Actor for CipherBftHost {
                     }
                 };
 
+                // Update proposer selector if validator set changed
+                if epoch_transition {
+                    if let Some(ref selector) = self.proposer_selector {
+                        // Get the new validator set for the next height
+                        if let Ok(new_validator_set) = self.get_validator_set(height.next()) {
+                            selector.update_validator_set(&new_validator_set);
+                            debug!(
+                                parent: &self.span,
+                                height = height.0,
+                                "Updated proposer selector for new validator set"
+                            );
+                        }
+                    }
+                }
+
+                // Periodic priority maintenance to prevent overflow
+                if let Some(ref selector) = self.proposer_selector {
+                    if height.0 % 100 == 0 {
+                        selector.center_priorities();
+                        selector.scale_if_needed();
+                        trace!(
+                            parent: &self.span,
+                            height = height.0,
+                            "Centered proposer selector priorities"
+                        );
+                    }
+                }
+
                 // Reply with the next height to start
                 let next_height = height.next();
 
-                // CRITICAL: Wait for the next cut to be available before starting next height.
+                // Wait for the next cut to be available before starting next height.
                 //
                 // This fixes a race condition where:
                 // 1. on_decided() sends the decision to the node's event loop
@@ -716,21 +776,23 @@ impl Actor for CipherBftHost {
                 // By waiting here, we ensure the node has processed the decision and
                 // Primary has formed the next Cut before consensus starts requesting it.
                 //
-                // Timeout of 500ms should be sufficient for the node to process the
-                // decision and form the Cut. If it times out, consensus will still
-                // start but may vote NIL on the first round.
-                let wait_timeout = std::time::Duration::from_millis(500);
-                if !self
-                    .value_builder
-                    .wait_for_cut(next_height, wait_timeout)
-                    .await
-                {
-                    warn!(
-                        parent: &self.span,
-                        next_height = next_height.0,
-                        "Cut not ready for next height after {:?}, proceeding anyway",
-                        wait_timeout
-                    );
+                // The timeout is configurable via HostConfig::wait_for_cut_timeout_ms.
+                // If it times out, consensus will still start but may vote NIL on round 0.
+                let wait_timeout_ms = self.config.wait_for_cut_timeout_ms;
+                if wait_timeout_ms > 0 {
+                    let wait_timeout = std::time::Duration::from_millis(wait_timeout_ms);
+                    if !self
+                        .value_builder
+                        .wait_for_cut(next_height, wait_timeout)
+                        .await
+                    {
+                        warn!(
+                            parent: &self.span,
+                            next_height = next_height.0,
+                            "Cut not ready for next height after {:?}, proceeding anyway",
+                            wait_timeout
+                        );
+                    }
                 }
 
                 match self.get_validator_set(next_height) {
@@ -1748,6 +1810,7 @@ impl DecisionHandler for ChannelDecisionHandler {
 /// * `decided_tx` - Channel to send decided events
 /// * `network` - Optional network reference for publishing proposal parts
 /// * `dcl_store` - Optional persistent storage for decisions (enables sync support)
+/// * `wait_for_cut_timeout_ms` - Timeout in ms to wait for next cut after decision (0 to disable)
 ///
 /// # Returns
 ///
@@ -1776,6 +1839,7 @@ pub async fn spawn_host(
     decided_tx: Option<mpsc::Sender<(ConsensusHeight, Cut)>>,
     network: Option<NetworkRef<CipherBftContext>>,
     dcl_store: Option<Arc<dyn cipherbft_storage::DclStore>>,
+    wait_for_cut_timeout_ms: u64,
 ) -> anyhow::Result<HostRef<CipherBftContext>> {
     // Extract validators from context
     let validators: Vec<_> = ctx.validator_set.as_slice().to_vec();
@@ -1787,8 +1851,11 @@ pub async fn spawn_host(
             .map_err(|e| anyhow::anyhow!("Failed to create validator set manager: {}", e))?,
     );
 
-    // Create config with default retention values
-    let config = HostConfig::default();
+    // Create config with provided wait_for_cut_timeout
+    let config = HostConfig {
+        wait_for_cut_timeout_ms,
+        ..Default::default()
+    };
 
     // Create channel-based handlers with config values
     let value_builder = if let Some(network_ref) = network {

crates/consensus/src/lib.rs

@@ -37,6 +37,9 @@ pub mod vote;
 #[cfg(feature = "malachite")]
 pub mod wal;
 
+#[cfg(feature = "malachite")]
+pub mod proposer_selector;
+
 #[cfg(feature = "malachite")]
 pub mod execution_handler;
 
@@ -83,6 +86,9 @@ pub use vote::ConsensusVote;
 #[cfg(feature = "malachite")]
 pub use wal::spawn_wal;
 
+#[cfg(feature = "malachite")]
+pub use proposer_selector::ProposerSelector;
+
 #[cfg(feature = "malachite")]
 pub use execution_handler::{
     ExecutingDecisionHandler, ExecutionCallback, NoOpReceiptStore, ReceiptStore,