[consensus] Exclude low scoring ancestors in proposals with time base…

…d inclusion/exclusion (#19605)

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] Indexer: 
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:
- [ ] REST API:

consensus/config/src/committee.rs

@@ -68,6 +68,11 @@ impl Committee {
         self.total_stake
     }
 
+    pub fn n_percent_stake_threshold(&self, n: u64) -> Stake {
+        assert!(n <= 100, "n must be between 0 and 100");
+        self.total_stake * n / 100
+    }
+
     pub fn quorum_threshold(&self) -> Stake {
         self.quorum_threshold
     }

consensus/core/src/dag_state.rs

@@ -920,7 +920,7 @@ impl DagState {
     }
 
     pub(crate) fn calculate_scoring_subdag_scores(&self) -> ReputationScores {
-        self.scoring_subdag.calculate_scores()
+        self.scoring_subdag.calculate_distributed_vote_scores()
     }
 
     pub(crate) fn scoring_subdag_commit_range(&self) -> CommitIndex {

consensus/core/src/leader_scoring.rs

@@ -129,6 +129,10 @@ impl ReputationScores {
         }
     }
 
+    pub(crate) fn highest_score(&self) -> u64 {
+        *self.scores_per_authority.iter().max().unwrap_or(&0)
+    }
+
     // Returns the authorities index with score tuples.
     pub(crate) fn authorities_by_score(&self, context: Arc<Context>) -> Vec<(AuthorityIndex, u64)> {
         self.scores_per_authority
@@ -258,17 +262,9 @@ impl ScoringSubdag {
     }
 
     // Iterate through votes and calculate scores for each authority based on
-    // scoring strategy that is used. (Vote or CertifiedVote)
-    pub(crate) fn calculate_scores(&self) -> ReputationScores {
-        let _s = self
-            .context
-            .metrics
-            .node_metrics
-            .scope_processing_time
-            .with_label_values(&["ScoringSubdag::calculate_scores"])
-            .start_timer();
-
-        let scores_per_authority = self.score_distributed_votes();
+    // distributed vote scoring strategy.
+    pub(crate) fn calculate_distributed_vote_scores(&self) -> ReputationScores {
+        let scores_per_authority = self.distributed_votes_scores();
 
         // TODO: Normalize scores
         ReputationScores::new(
@@ -283,7 +279,15 @@ impl ScoringSubdag {
     /// Instead of only giving one point for each vote that is included in 2f+1
     /// blocks. We give a score equal to the amount of stake of all blocks that
     /// included the vote.
-    fn score_distributed_votes(&self) -> Vec<u64> {
+    fn distributed_votes_scores(&self) -> Vec<u64> {
+        let _s = self
+            .context
+            .metrics
+            .node_metrics
+            .scope_processing_time
+            .with_label_values(&["ScoringSubdag::score_distributed_votes"])
+            .start_timer();
+
         let num_authorities = self.context.committee.size();
         let mut scores_per_authority = vec![0_u64; num_authorities];
 
@@ -299,29 +303,6 @@ impl ScoringSubdag {
         scores_per_authority
     }
 
-    /// This scoring strategy gives points equal to the amount of stake in blocks
-    /// that include the authority's vote, if that amount of total_stake > 2f+1.
-    /// We consider this a certified vote.
-    // TODO: This will be used for ancestor selection
-    #[allow(unused)]
-    fn score_certified_votes(&self) -> Vec<u64> {
-        let num_authorities = self.context.committee.size();
-        let mut scores_per_authority = vec![0_u64; num_authorities];
-
-        for (vote, stake_agg) in self.votes.iter() {
-            let authority = vote.author;
-            if stake_agg.reached_threshold(&self.context.committee) {
-                let stake = stake_agg.stake();
-                tracing::trace!(
-                    "[{}] scores +{stake} reputation for {authority}!",
-                    self.context.own_index,
-                );
-                scores_per_authority[authority.value()] += stake;
-            }
-        }
-        scores_per_authority
-    }
-
     pub(crate) fn scored_subdags_count(&self) -> usize {
         if let Some(commit_range) = &self.commit_range {
             commit_range.size()
@@ -555,41 +536,11 @@ mod tests {
             scoring_subdag.add_subdags(vec![sub_dag]);
         }
 
-        let scores = scoring_subdag.calculate_scores();
+        let scores = scoring_subdag.calculate_distributed_vote_scores();
         assert_eq!(scores.scores_per_authority, vec![5, 5, 5, 5]);
         assert_eq!(scores.commit_range, (1..=4).into());
     }
 
-    #[tokio::test]
-    async fn test_certified_vote_scoring_subdag() {
-        telemetry_subscribers::init_for_testing();
-        let context = Arc::new(Context::new_for_test(4).0);
-
-        // Populate fully connected test blocks for round 0 ~ 3, authorities 0 ~ 3.
-        let mut dag_builder = DagBuilder::new(context.clone());
-        dag_builder.layers(1..=3).build();
-        // Build round 4 but with just the leader block
-        dag_builder
-            .layer(4)
-            .authorities(vec![
-                AuthorityIndex::new_for_test(1),
-                AuthorityIndex::new_for_test(2),
-                AuthorityIndex::new_for_test(3),
-            ])
-            .skip_block()
-            .build();
-
-        let mut scoring_subdag = ScoringSubdag::new(context.clone());
-
-        for (sub_dag, _commit) in dag_builder.get_sub_dag_and_commits(1..=4) {
-            scoring_subdag.add_subdags(vec![sub_dag]);
-        }
-
-        let scores_per_authority = scoring_subdag.score_certified_votes();
-        assert_eq!(scores_per_authority, vec![4, 4, 4, 4]);
-        assert_eq!(scoring_subdag.commit_range.unwrap(), (1..=4).into());
-    }
-
     // TODO: Remove all tests below this when DistributedVoteScoring is enabled.
     #[tokio::test]
     async fn test_reputation_score_calculator() {

consensus/core/src/lib.rs

@@ -1,6 +1,7 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
+mod ancestor;
 mod authority_node;
 mod authority_service;
 mod base_committer;

consensus/core/src/metrics.rs

@@ -141,6 +141,10 @@ pub(crate) struct NodeMetrics {
     pub(crate) commit_round_advancement_interval: Histogram,
     pub(crate) last_decided_leader_round: IntGauge,
     pub(crate) leader_timeout_total: IntCounterVec,
+    pub(crate) smart_selection_wait: IntCounter,
+    pub(crate) ancestor_state_change_by_authority: IntCounterVec,
+    pub(crate) excluded_proposal_ancestors_count_by_authority: IntCounterVec,
+    pub(crate) included_excluded_proposal_ancestors_count_by_authority: IntCounterVec,
     pub(crate) missing_blocks_total: IntCounter,
     pub(crate) missing_blocks_after_fetch_total: IntCounter,
     pub(crate) num_of_bad_nodes: IntGauge,
@@ -441,6 +445,29 @@ impl NodeMetrics {
                 &["timeout_type"],
                 registry,
             ).unwrap(),
+            smart_selection_wait: register_int_counter_with_registry!(
+                "smart_selection_wait",
+                "Number of times we waited for smart ancestor selection.",
+                registry,
+            ).unwrap(),
+            ancestor_state_change_by_authority: register_int_counter_vec_with_registry!(
+                "ancestor_state_change_by_authority",
+                "The total number of times an ancestor state changed to EXCLUDE or INCLUDE.",
+                &["authority", "state"],
+                registry,
+            ).unwrap(),
+            excluded_proposal_ancestors_count_by_authority: register_int_counter_vec_with_registry!(
+                "excluded_proposal_ancestors_count_by_authority",
+                "Total number of excluded ancestors per authority during proposal.",
+                &["authority"],
+                registry,
+            ).unwrap(),
+            included_excluded_proposal_ancestors_count_by_authority: register_int_counter_vec_with_registry!(
+                "included_excluded_proposal_ancestors_count_by_authority",
+                "Total number of ancestors per authority with 'excluded' status that got included in proposal. Either weak or strong type.",
+                &["authority", "type"],
+                registry,
+            ).unwrap(),
             missing_blocks_total: register_int_counter_with_registry!(
                 "missing_blocks_total",
                 "Total cumulative number of missing blocks",

consensus/core/src/round_prober.rs

@@ -238,7 +238,7 @@ impl<C: NetworkClient> RoundProber<C> {
             .set_propagation_delay_and_quorum_rounds(propagation_delay, quorum_rounds.clone())
         {
             tracing::warn!(
-                "Failed to set propagation delay {propagation_delay} on Core: {:?}",
+                "Failed to set propagation delay and quorum rounds {quorum_rounds:?} on Core: {:?}",
                 e
             );
         }

crates/sui-open-rpc/spec/openrpc.json

@@ -1307,6 +1307,7 @@
                 "consensus_distributed_vote_scoring_strategy": false,
                 "consensus_order_end_of_epoch_last": true,
                 "consensus_round_prober": false,
+                "consensus_smart_ancestor_selection": false,
                 "disable_invariant_violation_check_in_swap_loc": false,
                 "disallow_adding_abilities_on_upgrade": false,
                 "disallow_change_struct_type_params_on_upgrade": false,

crates/sui-protocol-config/src/lib.rs

@@ -196,6 +196,7 @@ const MAX_PROTOCOL_VERSION: u64 = 69;
 //             Further reduce minimum number of random beacon shares.
 //             Disallow adding new modules in `deps-only` packages.
 // Version 69: Sets number of rounds allowed for fastpath voting in consensus.
+//             Enable smart ancestor selection in devnet.
 
 #[derive(Copy, Clone, Debug, Hash, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
 pub struct ProtocolVersion(u64);
@@ -565,6 +566,10 @@ struct FeatureFlags {
 
     #[serde(skip_serializing_if = "is_false")]
     disallow_new_modules_in_deps_only_packages: bool,
+
+    // Use smart ancestor selection in consensus.
+    #[serde(skip_serializing_if = "is_false")]
+    consensus_smart_ancestor_selection: bool,
 }
 
 fn is_false(b: &bool) -> bool {
@@ -1675,6 +1680,10 @@ impl ProtocolConfig {
         self.feature_flags
             .disallow_new_modules_in_deps_only_packages
     }
+
+    pub fn consensus_smart_ancestor_selection(&self) -> bool {
+        self.feature_flags.consensus_smart_ancestor_selection
+    }
 }
 
 #[cfg(not(msim))]
@@ -2944,6 +2953,11 @@ impl ProtocolConfig {
                 69 => {
                     // Sets number of rounds allowed for fastpath voting in consensus.
                     cfg.consensus_voting_rounds = Some(40);
+
+                    if chain != Chain::Mainnet && chain != Chain::Testnet {
+                        // Enable smart ancestor selection for devnet
+                        cfg.feature_flags.consensus_smart_ancestor_selection = true;
+                    }
                 }
                 // Use this template when making changes:
                 //

crates/sui-protocol-config/src/snapshots/sui_protocol_config__test__version_69.snap

@@ -72,6 +72,7 @@ feature_flags:
   relocate_event_module: true
   uncompressed_g1_group_elements: true
   disallow_new_modules_in_deps_only_packages: true
+  consensus_smart_ancestor_selection: true
 max_tx_size_bytes: 131072
 max_input_objects: 2048
 max_size_written_objects: 5000000