Fix change state to following when we are not + limit cache size of epoch precompute

crates/beacon_state/tile/src/tile.rs

@@ -496,11 +496,13 @@ impl BeaconStateTile {
     /// later one, else `from`.
     #[timed]
     fn epoch_start_state(&mut self, root: B256, from: StateId, slot: Slot) -> StateId {
-        let from_slot = self.slot_state_at(from).slot;
-        let (state, spec, scratch) = (&mut self.state, &self.spec, &mut self.stf_scratch);
         let epoch = slot / SLOTS_PER_EPOCH;
-        self.precomputed_epochs.get_or_advance(root, from, from_slot, epoch, |from, to| {
-            Self::process_slots_advance(state, spec, scratch, from, to)
+        if epoch <= self.slot_state_at(from).slot / SLOTS_PER_EPOCH {
+            return from;
+        }
+        let (state, spec, scratch) = (&mut self.state, &self.spec, &mut self.stf_scratch);
+        self.precomputed_epochs.get_or_insert(root, epoch, || {
+            Self::process_slots_advance(state, spec, scratch, from, epoch * SLOTS_PER_EPOCH)
         })
     }
 

crates/beacon_state/tile/src/tile/precomputed_epochs.rs

@@ -1,52 +1,63 @@
-use rustc_hash::FxHashMap;
-use silver_beacon_state_data::{B256, Epoch, SLOTS_PER_EPOCH, Slot, StateId};
+use silver_beacon_state_data::{B256, Epoch, StateId};
 
 use crate::fork_choice::ForkChoice;
 
-/// A block's post-state at each later epoch's first slot, shared by the tick,
+// Steady state holds the head's next epoch plus reorg/precompute transients.
+const MAX_PRECOMPUTED_EPOCHS: usize = 8;
+
+struct Entry {
+    root: B256,
+    epoch: Epoch,
+    state: StateId,
+}
+
+/// A block's post-state at a later epoch's first slot, shared by the tick,
 /// the block path and the precompute. The spec's `store.checkpoint_states`.
 #[derive(Default)]
-pub(super) struct PrecomputedEpochs(FxHashMap<(B256, Epoch), StateId>);
+pub(super) struct PrecomputedEpochs([Option<Entry>; MAX_PRECOMPUTED_EPOCHS]);
 
 impl PrecomputedEpochs {
-    /// Advances from the latest cached epoch, else `from`, caching every
-    /// boundary crossed. Spec `store_target_checkpoint_state`.
-    pub(super) fn get_or_advance(
+    pub(super) fn get_or_insert(
         &mut self,
         root: B256,
-        mut from: StateId,
-        from_slot: Slot,
         epoch: Epoch,
-        mut advance: impl FnMut(StateId, Slot) -> StateId,
+        compute: impl FnOnce() -> StateId,
     ) -> StateId {
-        let mut next = from_slot / SLOTS_PER_EPOCH + 1;
-        for cached in (next..=epoch).rev() {
-            if let Some(id) = self.get(root, cached) {
-                (from, next) = (id, cached + 1);
-                break;
-            }
-        }
-        for crossed in next..=epoch {
-            from = advance(from, crossed * SLOTS_PER_EPOCH);
-            self.0.insert((root, crossed), from);
+        if let Some(e) = self.0.iter().flatten().find(|e| e.root == root && e.epoch == epoch) {
+            return e.state;
         }
-        from
+
+        let state = compute();
+        let slot = self.find_slot();
+        self.0[slot] = Some(Entry { root, epoch, state });
+        state
     }
 
-    fn get(&self, root: B256, epoch: Epoch) -> Option<StateId> {
-        self.0.get(&(root, epoch)).copied()
+    /// Empty slot first, otherwise the lowest-epoch entry.
+    fn find_slot(&self) -> usize {
+        if let Some(empty) = self.0.iter().position(Option::is_none) {
+            return empty;
+        }
+        self.0
+            .iter()
+            .enumerate()
+            .filter_map(|(slot, e)| e.as_ref().map(|e| (slot, e.epoch)))
+            .min_by_key(|&(_, epoch)| epoch)
+            .map_or(0, |(slot, _)| slot)
     }
 
     /// Entries off pruned blocks, or at or before the finalized epoch, can no
     /// longer be asked for.
     pub(super) fn drop_outdated(&mut self, fork_choice: &ForkChoice) {
         let finalized_epoch = fork_choice.finalized_checkpoint.epoch;
-        self.0.retain(|(root, epoch), _| {
-            *epoch > finalized_epoch && fork_choice.find_node_idx(root).is_some()
-        });
+        for slot in &mut self.0 {
+            slot.take_if(|e| {
+                e.epoch <= finalized_epoch || fork_choice.find_node_idx(&e.root).is_none()
+            });
+        }
     }
 
     pub(super) fn state_ids_mut(&mut self) -> impl Iterator<Item = &mut StateId> {
-        self.0.values_mut()
+        self.0.iter_mut().flatten().map(|e| &mut e.state)
     }
 }

crates/beacon_state/tile/src/tile/tests.rs

@@ -354,6 +354,24 @@ fn precomputed_epoch_survives_finalization() {
     assert_eq!(ssz_hash::hash_tree_root_state(&forks.tile.state.read_view(after)), before_root);
 }
 
+/// A multi-epoch advance in one fork matches stepping one epoch at a time.
+#[test]
+fn long_gap_advance_matches_stepping_epoch_by_epoch() {
+    let mut stepped = make_tile();
+    seed_tile(&mut stepped, 4, 31);
+    let mut id = stepped.last_applied;
+    for epoch in 1..=5 {
+        id = stepped.epoch_start_state(ANCHOR_ROOT, id, epoch * SLOTS_PER_EPOCH);
+    }
+    let stepped_root = ssz_hash::hash_tree_root_state(&stepped.state.read_view(id));
+
+    let mut jumped = make_tile();
+    seed_tile(&mut jumped, 4, 31);
+    let id = jumped.epoch_start_state(ANCHOR_ROOT, jumped.last_applied, 5 * SLOTS_PER_EPOCH);
+
+    assert_eq!(ssz_hash::hash_tree_root_state(&jumped.state.read_view(id)), stepped_root);
+}
+
 /// Sustained non-finality: slot advances far past the rings' initial
 /// capacity (slot tiers past `SLOTS_RING_N` twice over, the epoch tier
 /// past `EPOCHS_RING_N`) must grow the rings instead of panicking on

crates/control/src/sync_engine/mod.rs

@@ -547,9 +547,12 @@ impl SyncEngine {
         if !chosen.is_following() {
             return Some(chosen);
         }
-        let comparable = self.ctx.local.have_status &&
+        let local = &self.ctx.local;
+        let comparable = local.have_status &&
             (self.phase.target().is_some() || self.ctx.peers.received_statuses());
-        comparable.then_some(SyncUpdate::Following)
+        let peers_are_ahead =
+            self.ctx.peers.any_peer_ahead_of(local.head_imported_slot, &self.ctx.cfg);
+        (comparable && !peers_are_ahead).then_some(SyncUpdate::Following)
     }
 
     pub fn drive_requests(&mut self, now: Instant, emit: &mut impl FnMut(SyncAction) -> bool) {

crates/control/src/sync_engine/peers.rs

@@ -45,6 +45,10 @@ impl PeerView {
         !self.claims.is_empty()
     }
 
+    pub(super) fn any_peer_ahead_of(&self, our_head_slot: u64, cfg: &SyncingConfig) -> bool {
+        self.claims.values().any(|c| c.head_slot > our_head_slot + cfg.head_lag_threshold_slots)
+    }
+
     pub(super) fn claims_span(&self, peer: usize, span: RangeInclusive<Slot>) -> bool {
         self.claims.get(&peer).is_some_and(|c| {
             c.head_slot >= *span.end() && c.earliest_available_slot <= *span.start()

crates/control/src/sync_engine/tests.rs

@@ -240,6 +240,25 @@ fn quiet_chain_is_caught_up_not_behind() {
     );
 }
 
+/// The only offered chain went unavailable while its peer still claims a
+/// head far past ours. That is "behind with nothing selectable", so the
+/// engine must not fall back to `Following`: the following tick would
+/// advance the head state across every slot up to the wall clock.
+#[test]
+fn unselectable_chains_with_peers_ahead_stay_idle_not_following() {
+    let mut e = engine();
+    peer_status(&mut e, PEER, HEAD_ROOT, 20_000);
+    local_status(&mut e, 0, 20_000);
+    advance(&mut e);
+    assert!(matches!(e.current_target(), Some(SyncUpdate::SyncingHead { .. })));
+
+    e.ctx.peers.mark_unavailable(HEAD_ROOT);
+    e.mark_dirty();
+    assert_eq!(e.advance(), None, "no target published");
+    assert!(matches!(e.phase, Phase::Idle), "behind a peer: Idle, not Following");
+    assert_ne!(e.current_target(), Some(SyncUpdate::Following));
+}
+
 #[test]
 fn tail_advances_only_on_coverage() {
     let now = Instant::now();