apollo_consensus_orchestrator,apollo_consensus_manager: move SNIP-35 bootstrap to pre-consensus

Author: sirandreww-starkware

crates/apollo_consensus_manager/src/consensus_manager.rs

@@ -159,14 +159,18 @@ impl ConsensusManager {
             outbound_internal_receiver,
         );
 
-        let consensus_context = self.create_sequencer_consensus_context(
+        let mut consensus_context = self.create_sequencer_consensus_context(
             &votes_broadcast_channels,
             outbound_internal_sender,
             self.config_manager_client.clone(),
         );
 
         let current_height =
             self.batcher_client.get_height().await.expect("Failed to get height from batcher");
+        consensus_context
+            .initialize_fee_proposals_window(current_height.height)
+            .await
+            .expect("Failed to bootstrap SNIP-35 fee_proposals window");
         let run_consensus_args = self.create_run_consensus_args(current_height.height);
 
         let network_task =

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};
+use std::collections::{BTreeMap, HashMap, VecDeque};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
@@ -37,7 +37,11 @@ use apollo_protobuf::consensus::{
     TransactionBatch,
     Vote,
 };
-use apollo_state_sync_types::communication::{SharedStateSyncClient, StateSyncClientError};
+use apollo_state_sync_types::communication::{
+    SharedStateSyncClient,
+    StateSyncClientError,
+    StateSyncClientResult,
+};
 use apollo_state_sync_types::errors::StateSyncError;
 use apollo_state_sync_types::state_sync_types::SyncBlock;
 use apollo_time::time::Clock;
@@ -305,39 +309,40 @@ impl SequencerConsensusContext {
         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
-    /// available immediately. Pre-V0_14_3 blocks return `Ok` with `fee_proposal_fri == None` and
-    /// contribute nothing; an `Err` from state_sync means the block could not be retrieved, so we
-    /// log and stop bootstrapping.
-    // TODO(AndrewL): On `Err`, state_sync may simply not have downloaded the block yet rather than
-    // the block being absent. Returning a partial window here is non-deterministic across nodes
-    // (different local sync progress → different windows → different `fee_actual`). Resolving this
-    // requires a startup gate that guarantees state_sync is caught up to `height - 1` before
-    // consensus enters `set_height_and_round`, since blocking inside this call to retry is unsafe
-    // (peers may already be voting at this height). Track and fix in a follow-up PR.
-    async fn bootstrap_fee_proposals_window(&mut self, height: BlockNumber) {
+    /// Fill `[start_height - WINDOW, start_height)` from local state_sync storage. Must be
+    /// called before `run_consensus` so the window is populated before voting begins. Blocks
+    /// state_sync has not caught up to yet are pushed to the back of the queue and revisited —
+    /// joining consensus with a partial window would make this node disagree with caught-up
+    /// peers on `fee_actual`. Other state_sync errors propagate.
+    pub async fn initialize_fee_proposals_window(
+        &mut self,
+        start_height: BlockNumber,
+    ) -> StateSyncClientResult<()> {
+        const STATE_SYNC_RETRY_INTERVAL: Duration = Duration::from_millis(500);
         let window_size =
             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 {
-            let block_number = BlockNumber(h);
+        let window_end_height = start_height.0;
+        let window_start_height = window_end_height.saturating_sub(window_size);
+        let mut pending_heights: VecDeque<BlockNumber> =
+            (window_start_height..window_end_height).map(BlockNumber).collect();
+        while let Some(block_number) = pending_heights.pop_front() {
             match self.deps.state_sync_client.get_block(block_number).await {
-                Ok(block) => {
-                    self.record_fee_proposal(
-                        block_number,
-                        block.block_header_without_hash.fee_proposal_fri,
-                    );
-                }
-                Err(e) => {
+                Ok(block) => self.record_fee_proposal(
+                    block_number,
+                    block.block_header_without_hash.fee_proposal_fri,
+                ),
+                Err(StateSyncClientError::StateSyncError(StateSyncError::BlockNotFound(_))) => {
                     warn!(
-                        "SNIP-35 bootstrap failed at block {h}: {e:?}. Window has {} / \
-                         {FEE_PROPOSAL_WINDOW_SIZE} entries.",
-                        self.fee_proposals_window.len()
+                        "State sync not ready for height {block_number}; re-queueing after \
+                         {STATE_SYNC_RETRY_INTERVAL:?}"
                     );
-                    break;
+                    pending_heights.push_back(block_number);
+                    tokio::time::sleep(STATE_SYNC_RETRY_INTERVAL).await;
                 }
+                Err(e) => return Err(e),
             }
         }
+        Ok(())
     }
 
     async fn start_stream(&mut self, stream_id: HeightAndRound) -> StreamSender {
@@ -954,11 +959,6 @@ impl ConsensusContext for SequencerConsensusContext {
                 let gas_price_u64 = u64::try_from(self.l2_gas_price.0).unwrap_or(u64::MAX);
                 CONSENSUS_L2_GAS_PRICE.set_lossy(gas_price_u64);
             }
-            // SNIP-35: on first height, bootstrap the window so fee_actual is available
-            // immediately.
-            if self.current_height.is_none() && self.fee_proposals_window.is_empty() {
-                self.bootstrap_fee_proposals_window(height).await;
-            }
             self.current_height = Some(height);
             self.current_round = round;
             self.prune_fee_proposals_window(height);

crates/apollo_consensus_orchestrator/src/sequencer_consensus_context_test.rs

@@ -1,3 +1,4 @@
+use std::collections::BTreeMap;
 use std::future::ready;
 use std::sync::Arc;
 use std::time::Duration;
@@ -1495,7 +1496,7 @@ async fn test_first_height_keeps_sync_provided_l2_gas_price() {
 
     let (mut deps, _network) = create_test_and_network_deps();
     // Specific get_block expectation must be registered before setup_default_expectations, which
-    // installs a catch-all BlockNotFound handler for SNIP-35 bootstrap.
+    // installs a catch-all handler.
     deps.state_sync_client.expect_get_block().times(1).return_once(|height| {
         let mut sync_block = SyncBlock::default();
         sync_block.block_header_without_hash.block_number = height;
@@ -1537,9 +1538,7 @@ async fn test_first_height_keeps_sync_provided_l2_gas_price() {
 }
 
 // Distinct value per height so map equality verifies identity, not just key set.
-fn window_of(
-    heights: impl IntoIterator<Item = u64>,
-) -> std::collections::BTreeMap<BlockNumber, Option<GasPrice>> {
+fn window_of(heights: impl IntoIterator<Item = u64>) -> BTreeMap<BlockNumber, Option<GasPrice>> {
     heights.into_iter().map(|h| (BlockNumber(h), Some(GasPrice(u128::from(h))))).collect()
 }
 
@@ -1552,9 +1551,9 @@ fn window_of(
 #[case::saturates_at_zero_when_height_below_window(window_of(0..=2), BlockNumber(5), window_of(0..=2))]
 #[case::no_op_when_all_entries_within_window(window_of(105..=110), BlockNumber(110), window_of(105..=110))]
 fn test_prune_fee_proposals_window(
-    #[case] initial_window: std::collections::BTreeMap<BlockNumber, Option<GasPrice>>,
+    #[case] initial_window: BTreeMap<BlockNumber, Option<GasPrice>>,
     #[case] current_height: BlockNumber,
-    #[case] expected_window: std::collections::BTreeMap<BlockNumber, Option<GasPrice>>,
+    #[case] expected_window: BTreeMap<BlockNumber, Option<GasPrice>>,
 ) {
     let (mut deps, _network) = create_test_and_network_deps();
     deps.setup_default_expectations();
@@ -1563,3 +1562,42 @@ fn test_prune_fee_proposals_window(
     context.prune_fee_proposals_window(current_height);
     assert_eq!(context.fee_proposals_window, expected_window);
 }
+
+// `initialize_fee_proposals_window` reads `[start_height - WINDOW, start_height)` from state_sync
+// and records each block's `fee_proposal_fri`. `expected_window` is the mapping the test asserts;
+// the mock answers `get_block(h)` from the same map. Genesis case (`start_height < WINDOW_SIZE`)
+// is exercised by a smaller window: the bootstrap range collapses to `[0, start_height)`.
+#[rstest]
+#[case::all_some(BlockNumber(100), window_of(90..100))]
+// v0.14.2 era: every block recorded with `fee_proposal_fri = None`.
+#[case::all_none(BlockNumber(100), (90u64..100).map(|h| (BlockNumber(h), None)).collect())]
+// Transition: heights < 95 are pre-SNIP-35 (None), heights >= 95 are SNIP-35 (Some).
+#[case::mixed_transition(
+    BlockNumber(100),
+    (90u64..100)
+        .map(|h| (BlockNumber(h), (h >= 95).then(|| GasPrice(u128::from(h)))))
+        .collect(),
+)]
+#[case::genesis_collapses_range(BlockNumber(3), window_of(0..3))]
+#[tokio::test]
+async fn test_initialize_fee_proposals_window(
+    #[case] start_height: BlockNumber,
+    #[case] expected_window: BTreeMap<BlockNumber, Option<GasPrice>>,
+) {
+    let mock_window = expected_window.clone();
+    let (mut deps, _network) = create_test_and_network_deps();
+    deps.state_sync_client.expect_get_block().times(expected_window.len()).returning(
+        move |height| {
+            let mut sync_block = SyncBlock::default();
+            sync_block.block_header_without_hash.block_number = height;
+            sync_block.block_header_without_hash.fee_proposal_fri =
+                *mock_window.get(&height).unwrap();
+            Ok(sync_block)
+        },
+    );
+    deps.setup_default_expectations();
+
+    let mut context = deps.build_context();
+    context.initialize_fee_proposals_window(start_height).await.unwrap();
+    assert_eq!(context.fee_proposals_window, expected_window);
+}

crates/apollo_consensus_orchestrator/src/test_utils.rs

@@ -326,9 +326,6 @@ impl TestDeps {
         self.l1_gas_price_provider.expect_get_rate().return_const(Ok(ETH_TO_FRI_RATE));
     }
 
-    /// Default get_block returns a pre-SNIP-35 block (no `fee_proposal_fri`). Used by SNIP-35
-    /// bootstrap on startup; mirrors real chain behavior where blocks before V0_14_3 exist but
-    /// don't carry the field.
     fn setup_default_state_sync_get_block(&mut self) {
         self.state_sync_client.expect_get_block().returning(|_| Ok(SyncBlock::default()));
     }