feat(starfish): evict pending commit votes via quorum commit index (#11392)

# Description of change

Bounds the in-memory `pending_commit_votes` tracker in `DagState` and
tightens the proposal skip paths in `Core`.

### Eviction (issue #11391)

- Switch `pending_commit_votes` from `VecDeque<CommitVote>` to
`BTreeSet<CommitVote>` so eviction uses `split_off`. Insertion order is
preserved (the linearizer produces commits monotonically); insertion
dedup is a free side benefit.
- New field on `DagState`: `last_known_quorum_commit_index: CommitIndex`
(no `Arc<CommitVoteMonitor>` reference). `Core` is the sole holder of
the monitor and pushes the value via a setter.
- `evict_pending_commit_votes()` drops votes with `index <=
last_known_quorum_commit_index - gc_depth`. Gated on
`consensus_block_restrictions`; no-op otherwise.
- Eviction is called both from `take_commit_votes` (proposal path) and
from `flush()` (mirrors the existing `pending_acknowledgments` pattern).
- `Core` refreshes `DagState::last_known_quorum_commit_index` from
`CommitVoteMonitor::quorum_commit_index()` at three places:
`Core::try_commit` after every successful
`commit_observer.handle_committed_leaders` (regular consensus path);
`Core::handle_committed_sub_dags_from_fast_sync` (fast-sync path);
`Core::should_propose` (right before we propose).

### `should_propose` cleanup (issue #11400)

- Centralised every skip reason behind a new `SkipProposalReason` enum +
`skip_proposal` helper, so the `core_skipped_proposals` metric label
space is disjoint and visible in one place.
- New skip reason `behind_quorum_commit_round` — the
`consensus_block_restrictions` gate (formerly inside `try_new_block`)
moves up to `should_propose`.
- The `clock_round <= last_proposed_round` recheck moves up too. It is
silent (no metric, no debug log) since it fires on every accepted block
in a round we have already proposed in — counting it would drown the
genuinely interesting reasons.
- `try_new_block` now reads `clock_round` only; all gates are upstream.

## Links to any relevant issues

Fixes #11391
Fixes #11400

## How the change has been tested

- [x] Basic tests (linting, compilation, formatting, unit/integration
tests)
- [x] New unit test `dag_state::test::test_evict_pending_commit_votes`
covering: quorum-driven eviction with the flag on; default-field no-op
(no setter call); flag-off no-op even with a non-zero field
- [x] Existing 280 starfish-core lib tests still pass
- [x] `cargo ci-clippy` clean across the workspace
- [ ] Patch-specific tests (correctness, functionality coverage)

Author: polinikita

crates/starfish/core/src/core.rs

@@ -155,6 +155,34 @@ impl ReasonToCreateBlock {
     }
 }
 
+/// Reason a `Core::should_propose` call returned `false`. Used as the `reason`
+/// label on the `core_skipped_proposals` metric. Keeping the variants in one
+/// enum makes the label space disjoint and centrally visible. Variant data is
+/// the per-reason context interpolated into the corresponding `debug!` line.
+///
+/// The "already proposed at this round" branch is intentionally not modeled
+/// here: it fires on every block accepted in a round we have already proposed
+/// in, which is a normal high-rate event. Counting and logging it would drown
+/// the genuinely interesting reasons below.
+#[derive(Clone, Copy)]
+pub(crate) enum SkipProposalReason {
+    NoQuorumSubscriber,
+    NoLastKnownProposedRound,
+    HigherLastKnownProposedRound { last_known: Round },
+    BehindQuorumCommitRound { approx_quorum: Round },
+}
+
+impl SkipProposalReason {
+    fn label(self) -> &'static str {
+        match self {
+            Self::NoQuorumSubscriber => "no_quorum_subscriber",
+            Self::NoLastKnownProposedRound => "no_last_known_proposed_round",
+            Self::HigherLastKnownProposedRound { .. } => "higher_last_known_proposed_round",
+            Self::BehindQuorumCommitRound { .. } => "behind_quorum_commit_round",
+        }
+    }
+}
+
 impl Core {
     pub(crate) fn new(
         context: Arc<Context>,
@@ -588,10 +616,18 @@ impl Core {
             dag_state.set_fast_sync_ongoing_flag(true);
         }
 
+        // After a positive commit advance, refresh the quorum commit index
+        // on `DagState` so the eviction inside `flush()` is bounded.
+        let quorum_commit_index = self.commit_vote_monitor.quorum_commit_index();
+
         // Flush commits to storage so they're available for
         // get_block_refs_for_recent_commits when close-to-quorum mode
         // triggers header fetching.
-        self.dag_state.write().flush();
+        {
+            let mut dag_state = self.dag_state.write();
+            dag_state.set_last_known_quorum_commit_index(quorum_commit_index);
+            dag_state.flush();
+        }
 
         // Then process subdags as usual
         self.commit_observer.finalize_and_send_solid_subdags(
@@ -738,9 +774,9 @@ impl Core {
         Ok((None, BTreeMap::new()))
     }
 
-    /// Attempts to propose a new block for the next round. If a block has
-    /// already proposed for latest or earlier round, then no block is
-    /// created and None is returned.
+    /// Attempts to propose a new block at the current clock round. Eligibility
+    /// (round bounds, block restrictions) is enforced upstream in
+    /// `should_propose`.
     #[instrument(level = "trace", skip_all)]
     fn try_new_block(&mut self, reason: ReasonToCreateBlock) -> Option<VerifiedBlock> {
         let _s = self
@@ -751,31 +787,9 @@ impl Core {
             .with_label_values(&["Core::try_new_block"])
             .start_timer();
 
-        // Ensure the new block has a higher round than the last proposed block
-        // and, under `consensus_block_restrictions`, also the approximate quorum commit
-        // round. Blocks at or below the quorum commit round will not improve
-        // the commit rule.
-        let clock_round = {
-            let dag_state = self.dag_state.read();
-            let clock_round = dag_state.threshold_clock_round();
-            let last_proposed_round = dag_state.get_last_proposed_block_header().round();
-            let min_round = if self.context.protocol_config.consensus_block_restrictions() {
-                let quorum_commit_index = self.commit_vote_monitor.quorum_commit_index();
-                let local_commit_index = dag_state.last_commit_index();
-                let local_commit_round = dag_state.last_commit_round();
-                // Lower bound on the quorum commit round: at least 1 round per
-                // commit, so the gap in indices maps to at least that many rounds.
-                let approx_quorum_round =
-                    local_commit_round + quorum_commit_index.saturating_sub(local_commit_index);
-                last_proposed_round.max(approx_quorum_round)
-            } else {
-                last_proposed_round
-            };
-            if clock_round <= min_round {
-                return None;
-            }
-            clock_round
-        };
+        // Eligibility checks (round bounds, block restrictions) are enforced
+        // upstream in `should_propose`.
+        let clock_round = self.dag_state.read().threshold_clock_round();
 
         // There must be a quorum of blocks from the previous round.
         let quorum_round = clock_round.saturating_sub(1);
@@ -1123,10 +1137,16 @@ impl Core {
             );
             all_missing_committed_txns.extend(missing_transactions_refs);
 
+            // After a positive commit advance, refresh the quorum commit index
+            // on `DagState` so the eviction of `pending_commit_votes` is bounded.
             // Both pending and solid sub DAGs should be added to scoring subdags.
-            self.dag_state
-                .write()
-                .add_scoring_subdags(subdags.iter().map(|s| s.base.clone()).collect());
+            {
+                let mut dag_state = self.dag_state.write();
+                dag_state.set_last_known_quorum_commit_index(
+                    self.commit_vote_monitor.quorum_commit_index(),
+                );
+                dag_state.add_scoring_subdags(subdags.iter().map(|s| s.base.clone()).collect());
+            }
 
             committed_sub_dags.extend(subdags);
 
@@ -1198,41 +1218,104 @@ impl Core {
         info!("Last known proposed round set to {round}");
     }
 
-    /// Whether the core should propose new blocks.
+    /// Returns true when Core should propose at the current clock round. As a
+    /// side effect, when proposal is greenlit under
+    /// `consensus_block_restrictions`, refreshes `DagState`'s last-known
+    /// quorum commit index to enable eviction for commit votes
     pub(crate) fn should_propose(&self) -> bool {
-        let clock_round = self.dag_state.read().threshold_clock_round();
-        let core_skipped_proposals = &self.context.metrics.node_metrics.core_skipped_proposals;
+        let (clock_round, last_proposed_round, local_commit_index, local_commit_round) = {
+            let dag_state = self.dag_state.read();
+            (
+                dag_state.threshold_clock_round(),
+                dag_state.get_last_proposed_block_header().round(),
+                dag_state.last_commit_index(),
+                dag_state.last_commit_round(),
+            )
+        };
 
         if !self.quorum_subscribers_exists {
-            debug!("Skip proposing for round {clock_round}, don't have a quorum of subscribers.");
-            core_skipped_proposals
-                .with_label_values(&["no_quorum_subscriber"])
-                .inc();
-            return false;
+            return self.skip_proposal(clock_round, SkipProposalReason::NoQuorumSubscriber);
         }
 
         let Some(last_known_proposed_round) = self.last_known_proposed_round else {
-            debug!(
-                "Skip proposing for round {clock_round}, last known proposed round has not been synced yet."
-            );
-            core_skipped_proposals
-                .with_label_values(&["no_last_known_proposed_round"])
-                .inc();
-            return false;
+            return self.skip_proposal(clock_round, SkipProposalReason::NoLastKnownProposedRound);
         };
         if clock_round <= last_known_proposed_round {
-            debug!(
-                "Skip proposing for round {clock_round} as last known proposed round is {last_known_proposed_round}"
+            return self.skip_proposal(
+                clock_round,
+                SkipProposalReason::HigherLastKnownProposedRound {
+                    last_known: last_known_proposed_round,
+                },
             );
-            core_skipped_proposals
-                .with_label_values(&["higher_last_known_proposed_round"])
-                .inc();
+        }
+
+        // Silently skip when we already proposed at or above `clock_round`.
+        // This branch fires on every accepted block within the same clock round
+        if clock_round <= last_proposed_round {
             return false;
         }
 
+        // Under `consensus_block_restrictions`, skip if the candidate round
+        // does not exceed an approximation of the quorum commit round. Blocks
+        // at or below it cannot improve the commit rule.
+        if self.context.protocol_config.consensus_block_restrictions() {
+            let quorum_commit_index = self.commit_vote_monitor.quorum_commit_index();
+            let approx_quorum_round =
+                local_commit_round + quorum_commit_index.saturating_sub(local_commit_index);
+            if clock_round <= approx_quorum_round {
+                return self.skip_proposal(
+                    clock_round,
+                    SkipProposalReason::BehindQuorumCommitRound {
+                        approx_quorum: approx_quorum_round,
+                    },
+                );
+            }
+
+            // We are about to propose: refresh DagState's known quorum commit
+            // index so the eviction of `pending_commit_votes` is bounded.
+            self.dag_state
+                .write()
+                .set_last_known_quorum_commit_index(quorum_commit_index);
+        }
+
         true
     }
 
+    /// Records a skipped proposal: emits the per-reason `debug!` line and
+    /// increments `core_skipped_proposals` with the matching label. Always
+    /// returns `false` so call sites can `return self.skip_proposal(...)`.
+    fn skip_proposal(&self, clock_round: Round, reason: SkipProposalReason) -> bool {
+        match reason {
+            SkipProposalReason::NoQuorumSubscriber => {
+                debug!(
+                    "Skip proposing for round {clock_round}, don't have a quorum of subscribers."
+                );
+            }
+            SkipProposalReason::NoLastKnownProposedRound => {
+                debug!(
+                    "Skip proposing for round {clock_round}, last known proposed round has not been synced yet."
+                );
+            }
+            SkipProposalReason::HigherLastKnownProposedRound { last_known } => {
+                debug!(
+                    "Skip proposing for round {clock_round} as last known proposed round is {last_known}"
+                );
+            }
+            SkipProposalReason::BehindQuorumCommitRound { approx_quorum } => {
+                debug!(
+                    "Skip proposing for round {clock_round}, behind approximate quorum commit round {approx_quorum}"
+                );
+            }
+        }
+        self.context
+            .metrics
+            .node_metrics
+            .core_skipped_proposals
+            .with_label_values(&[reason.label()])
+            .inc();
+        false
+    }
+
     /// Retrieves the next ancestors to propose to form a block at `clock_round`
     /// round.
     fn ancestors_to_propose(&mut self, clock_round: Round) -> Vec<VerifiedBlockHeader> {