apollo_consensus_orchestrator: introduce BTreeMap-keyed SNIP-35 fee_proposals window

Author: sirandreww-starkware

crates/apollo_consensus_orchestrator/src/sequencer_consensus_context.rs

@@ -6,7 +6,7 @@
 mod sequencer_consensus_context_test;
 
 use std::cmp::max;
-use std::collections::{BTreeMap, HashMap, VecDeque};
+use std::collections::{BTreeMap, HashMap};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
@@ -235,9 +235,8 @@ pub struct SequencerConsensusContext {
     l2_gas_price: GasPrice,
     l1_da_mode: L1DataAvailabilityMode,
     previous_proposal_init: Option<PreviousProposalInitInfo>,
-    /// SNIP-35: sliding window of recent fee_proposal values (size from
-    /// `FEE_PROPOSAL_WINDOW_SIZE`).
-    fee_proposals_window: VecDeque<GasPrice>,
+    /// SNIP-35 fee_proposals window. `None` = pre-V0_14_3.
+    fee_proposals_window: BTreeMap<BlockNumber, Option<GasPrice>>,
 }
 
 #[derive(Clone)]
@@ -288,16 +287,25 @@ impl SequencerConsensusContext {
             l2_gas_price: VersionedConstants::latest_constants().min_gas_price,
             l1_da_mode,
             previous_proposal_init: None,
-            fee_proposals_window: VecDeque::with_capacity(FEE_PROPOSAL_WINDOW_SIZE),
+            fee_proposals_window: BTreeMap::new(),
         }
     }
 
-    /// FIFO append into the SNIP-35 sliding window.
-    fn push_fee_proposal(&mut self, fee_proposal: GasPrice) {
-        if self.fee_proposals_window.len() >= FEE_PROPOSAL_WINDOW_SIZE {
-            self.fee_proposals_window.pop_front();
-        }
-        self.fee_proposals_window.push_back(fee_proposal);
+    fn record_fee_proposal(&mut self, height: BlockNumber, fee_proposal_fri: Option<GasPrice>) {
+        self.fee_proposals_window.insert(height, fee_proposal_fri);
+    }
+
+    /// Drop fee_proposals older than the SNIP-35 window so only heights in
+    /// `[current_height - FEE_PROPOSAL_WINDOW_SIZE, ..)` remain.
+    ///
+    /// `BTreeMap::split_off(&cutoff)` is inclusive on `cutoff`: it returns the entries with
+    /// key `>= cutoff` and leaves keys `< cutoff` in `self`. Reassigning the returned half
+    /// back to the field keeps `[cutoff, ..)` and drops everything strictly below.
+    fn prune_fee_proposals_window(&mut self, current_height: BlockNumber) {
+        let window_size =
+            u64::try_from(FEE_PROPOSAL_WINDOW_SIZE).expect("FEE_PROPOSAL_WINDOW_SIZE fits in u64");
+        let cutoff = BlockNumber(current_height.0.saturating_sub(window_size));
+        self.fee_proposals_window = self.fee_proposals_window.split_off(&cutoff);
     }
 
     /// SNIP-35: bootstrap the fee_proposals window from state_sync on startup so `fee_actual` is
@@ -315,11 +323,13 @@ impl SequencerConsensusContext {
             u64::try_from(FEE_PROPOSAL_WINDOW_SIZE).expect("FEE_PROPOSAL_WINDOW_SIZE fits in u64");
         let start = height.0.saturating_sub(window_size);
         for h in start..height.0 {
-            match self.deps.state_sync_client.get_block(BlockNumber(h)).await {
+            let block_number = BlockNumber(h);
+            match self.deps.state_sync_client.get_block(block_number).await {
                 Ok(block) => {
-                    if let Some(fee_proposal) = block.block_header_without_hash.fee_proposal_fri {
-                        self.push_fee_proposal(fee_proposal);
-                    }
+                    self.record_fee_proposal(
+                        block_number,
+                        block.block_header_without_hash.fee_proposal_fri,
+                    );
                 }
                 Err(e) => {
                     warn!(
@@ -436,9 +446,7 @@ impl SequencerConsensusContext {
         let DecisionReachedResponse { state_diff, central_objects } = decision_reached_response;
 
         self.update_l2_gas_price(height, l2_gas_used);
-        if let Some(fee_proposal) = init.fee_proposal_fri {
-            self.push_fee_proposal(fee_proposal);
-        }
+        self.record_fee_proposal(height, init.fee_proposal_fri);
 
         // A hash map of (possibly failed) transactions, where the key is the transaction hash
         // and the value is the transaction itself.
@@ -891,9 +899,10 @@ impl ConsensusContext for SequencerConsensusContext {
             sync_block.block_header_without_hash.next_l2_gas_price,
             VersionedConstants::latest_constants().min_gas_price,
         );
-        if let Some(fee_proposal) = sync_block.block_header_without_hash.fee_proposal_fri {
-            self.push_fee_proposal(fee_proposal);
-        }
+        self.record_fee_proposal(
+            sync_block.block_header_without_hash.block_number,
+            sync_block.block_header_without_hash.fee_proposal_fri,
+        );
 
         // TODO(Asmaa): validate starknet_version and parent_hash when they are stored.
         let block_number = sync_block.block_header_without_hash.block_number;
@@ -958,6 +967,7 @@ impl ConsensusContext for SequencerConsensusContext {
             }
             self.current_height = Some(height);
             self.current_round = round;
+            self.prune_fee_proposals_window(height);
             self.queued_proposals.clear();
             // The Batcher must be told when we begin to work on a new height. The implicit model is
             // that consensus works on a given height until it is done (either a decision is reached

crates/apollo_consensus_orchestrator/src/sequencer_consensus_context_test.rs

@@ -1535,3 +1535,53 @@ async fn test_first_height_keeps_sync_provided_l2_gas_price() {
         expected_price, context.l2_gas_price.0
     );
 }
+
+#[rstest]
+// FEE_PROPOSAL_WINDOW_SIZE = 10, prune at 110: cutoff = 100, keep heights >= 100.
+#[case::drops_entries_below_cutoff(
+    (90u64..=110).collect::<Vec<_>>(),
+    BlockNumber(110),
+    (100u64..=110).collect::<Vec<_>>(),
+)]
+#[case::empty_when_all_entries_below_cutoff(
+    vec![50u64, 60, 70],
+    BlockNumber(200),
+    vec![],
+)]
+// `current_height < FEE_PROPOSAL_WINDOW_SIZE`: cutoff would underflow; saturating_sub clamps
+// to 0, so every entry is retained.
+#[case::saturates_at_zero_when_height_below_window(
+    vec![0u64, 1, 2],
+    BlockNumber(5),
+    vec![0u64, 1, 2],
+)]
+#[case::no_op_when_all_entries_within_window(
+    (105u64..=110).collect::<Vec<_>>(),
+    BlockNumber(110),
+    (105u64..=110).collect::<Vec<_>>(),
+)]
+fn test_prune_fee_proposals_window(
+    #[case] inserted_heights: Vec<u64>,
+    #[case] current_height: BlockNumber,
+    #[case] expected_kept_heights: Vec<u64>,
+) {
+    let (mut deps, _network) = create_test_and_network_deps();
+    deps.setup_default_expectations();
+    let mut context = deps.build_context();
+
+    // Distinct value per height so the assertion checks identity, not just count.
+    for h in &inserted_heights {
+        context.record_fee_proposal(BlockNumber(*h), Some(GasPrice(u128::from(*h))));
+    }
+
+    context.prune_fee_proposals_window(current_height);
+
+    let kept_heights: Vec<u64> = context.fee_proposals_window.keys().map(|k| k.0).collect();
+    assert_eq!(kept_heights, expected_kept_heights);
+    for h in &expected_kept_heights {
+        assert_eq!(
+            context.fee_proposals_window.get(&BlockNumber(*h)),
+            Some(&Some(GasPrice(u128::from(*h)))),
+        );
+    }
+}