sync up from the latest local checkpoint

crates/beacon_state/data/src/encode.rs

@@ -13,7 +13,7 @@ use blst::min_pk::PublicKey;
 use crate::{
     BeaconState,
     decompose::FIXED_PART,
-    types::{B256, Checkpoint, Eth1Data, SyncCommittee},
+    types::{B256, Checkpoint, Eth1Data, SLOTS_PER_EPOCH, SyncCommittee},
 };
 
 // SSZ-serialised sizes of the fixed-width records (Fulu); mirror `decompose`.
@@ -258,11 +258,23 @@ impl BeaconState {
         // F11_OFF/F12_OFF validators, balances
         w_u32(w, offs[2])?;
         w_u32(w, offs[3])?;
-        // F13 randao_mixes, F14 slashings
-        write_b256_slice(w, &epoch_base.randao_mixes)?;
-        for s in epoch_base.slashings.iter() {
-            w_u64(w, *s)?;
+
+        // The current epoch's bucket is only
+        // flushed to the array tier at the epoch boundary, so mid-epoch the live
+        // value lives in the `*_current` caches (the hashing path substitutes
+        // them in `effective_{randao_mixes,slashings}_into`).
+        // F13 randao_mixes
+        let current_epoch = (sl.slot / SLOTS_PER_EPOCH) as usize;
+        let randao_curr = current_epoch % epoch_base.randao_mixes.len();
+        write_b256_slice(w, &epoch_base.randao_mixes[..randao_curr])?;
+        w.write_all(&sl.randao_mix_current)?;
+        write_b256_slice(w, &epoch_base.randao_mixes[randao_curr + 1..])?;
+        // F14 slashings.
+        let slashings_curr = current_epoch % epoch_base.slashings.len();
+        for (i, s) in epoch_base.slashings.iter().enumerate() {
+            w_u64(w, if i == slashings_curr { sl.current_epoch_slashings } else { *s })?;
         }
+
         // F15_OFF/F16_OFF previous/current participation
         w_u32(w, offs[4])?;
         w_u32(w, offs[5])?;
@@ -426,7 +438,10 @@ pub fn decode_checkpoint_pubkeys(bytes: &[u8]) -> Result<Vec<PublicKey>, Pubkeys
 
 #[cfg(test)]
 mod tests {
-    use crate::{BeaconState, SpecConfig, decode_checkpoint_pubkeys, encode::Section};
+    use crate::{
+        BeaconState, EpochGroup, EpochStateFinalized, SLOTS_PER_HISTORICAL_ROOT, SlotState,
+        SlotStateFinalized, SlotStateGroup, SpecConfig, decode_checkpoint_pubkeys, encode::Section,
+    };
 
     /// `encode → decompose → encode` must be a byte-exact fixed point, and the
     /// chunked streaming path must match the one-pass path. The pre-bootstrap
@@ -456,7 +471,47 @@ mod tests {
         assert_eq!(once, streamed, "chunk-streamed SSZ differs from one-pass");
     }
 
-    /// Pubkeys sidecar: write → decode must reproduce the decompressed keys.
+    /// Regression: a checkpoint persisted mid-epoch must encode the *live*
+    /// current-epoch randao mix / slashings (the `*_current` caches), not the
+    /// stale array bucket — that bucket is only refreshed at the epoch
+    /// boundary.
+    #[test]
+    fn encode_uses_live_current_epoch_caches() {
+        let cur = 3usize;
+        let mut bs = BeaconState::pre_bootstrap();
+
+        // Stale ring buckets, distinct from the live `*_current` caches the
+        // encode path must emit instead.
+        let mut epoch = EpochStateFinalized::default();
+        epoch.randao_mixes[cur] = [0xAA; 32];
+        epoch.slashings[cur] = 111;
+        bs.epoch = EpochGroup::new(epoch);
+
+        let zero_roots = || vec![[0u8; 32]; SLOTS_PER_HISTORICAL_ROOT].into_boxed_slice();
+        let slot = SlotState {
+            slot: 96, // epoch 3
+            randao_mix_current: [0xBB; 32],
+            current_epoch_slashings: 222,
+            ..Default::default()
+        };
+        bs.slot_states =
+            SlotStateGroup::new(SlotStateFinalized::from_parts(slot, zero_roots(), zero_roots()));
+
+        let mut ssz = Vec::with_capacity(bs.ssz_len());
+        bs.encode_ssz(&mut ssz).unwrap();
+        let decoded = BeaconState::decompose(&ssz, &SpecConfig::mainnet(), None).unwrap();
+
+        let de = decoded.epoch.finalized();
+        let ds = decoded.slot_states.finalized().state();
+        assert_eq!(de.randao_mixes[cur], [0xBB; 32], "randao current bucket = live");
+        assert_eq!(ds.randao_mix_current, [0xBB; 32]);
+        assert_eq!(de.slashings[cur], 222, "slashings current bucket = live");
+        assert_eq!(ds.current_epoch_slashings, 222);
+    }
+
+    /// Write the decompressed-pubkey sidecar, decode it, rebuild the validators
+    /// from SSZ + sidecar, and confirm the pubkeys match. Also that a corrupted
+    /// sidecar is rejected (never silently accepted).
     #[test]
     fn pubkeys_sidecar_round_trip() {
         let bs = BeaconState::pre_bootstrap();

crates/beacon_state/tile/src/lib.rs

@@ -9,6 +9,7 @@ pub mod ssz_hash;
 pub mod state_transition;
 pub mod tile;
 mod validate;
+mod weak_subjectivity;
 
 #[cfg(test)]
 pub(crate) mod test_signing;

crates/beacon_state/tile/src/tile.rs

@@ -12,8 +12,8 @@ use silver_beacon_state_data::{
 };
 use silver_common::{
     BeaconStateEvent, BlockSource, DataColumnsAvailable, GossipTopic, NewGossipMsg, P2pStreamId,
-    PeerEvent, RpcInbound, RpcResponse, RpcResponseInbound, RpcSeverity, SilverSpine, SyncUpdate,
-    TCacheRead, TRandomAccess, TRead, hex32,
+    PeerEvent, ReplayBlock, RpcInbound, RpcResponse, RpcResponseInbound, RpcSeverity, SilverSpine,
+    SyncUpdate, TCacheRead, TRandomAccess, TRead, hex32,
     ssz_view::{
         self, AttesterSlashingView, MAX_ATTESTATIONS_ELECTRA, MAX_ATTESTING_INDICES,
         PROPOSER_SLASHING_SIZE, ProposerSlashingView, SIGNED_BLS_CHANGE_SIZE,
@@ -30,6 +30,7 @@ use crate::{
     fork_choice::{BlockImport, ForkChoice, MAX_FORK_CHOICE_NODES, VoteTracker, compute_deltas},
     shuffling::{self, DOMAIN_BEACON_ATTESTER},
     ssz_hash, state_transition, validate,
+    weak_subjectivity::weak_subjectivity_period,
 };
 
 /// Survivor set re-anchored at finalization: every fork-choice node's bundle
@@ -187,19 +188,25 @@ pub struct BeaconStateTile {
 
     gossip_consumer: TRandomAccess,
     rpc_consumer: TRandomAccess,
+    replay_consumer: TRandomAccess,
+
+    verify_weak_subjectivity: bool,
 }
 
 type Producers = <SilverSpine as FluxSpine>::Producers;
 
 impl BeaconStateTile {
     /// If `checkpoint_state` is non-empty, bootstraps immediately; otherwise
     /// starts inert in `Mode::Syncing` (call `bootstrap` before the loop).
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         ticker: SlotTicker,
         spec: SpecConfig,
         mut state: BeaconStateOwner,
         gossip_consumer: TRandomAccess,
         rpc_consumer: TRandomAccess,
+        replay_consumer: TRandomAccess,
+        verify_weak_subjectivity: bool,
         checkpoint_state: &[u8],
         decompressed_pubkeys: &[u8],
     ) -> Self {
@@ -244,6 +251,8 @@ impl BeaconStateTile {
             ),
             gossip_consumer,
             rpc_consumer,
+            replay_consumer,
+            verify_weak_subjectivity,
         };
 
         if !checkpoint_state.is_empty() {
@@ -362,6 +371,8 @@ impl BeaconStateTile {
         self.fork_choice =
             ForkChoice::init(trusted, trusted, slot, block_root, anchor_state_root, anchor);
         self.state.publish_state_id(anchor);
+
+        self.assert_within_weak_subjectivity();
     }
 
     /// Index bundle of fork-choice's canonical tip. For gossip-object
@@ -513,6 +524,24 @@ impl BeaconStateTile {
         self.configured_fork_digest = Some(digest);
     }
 
+    pub fn assert_within_weak_subjectivity(&mut self) {
+        if !self.verify_weak_subjectivity {
+            return;
+        }
+        let ws_period = {
+            let view = self.state.read_view(self.last_applied);
+            weak_subjectivity_period(&view, &mut self.stf_scratch.active)
+        };
+        let checkpoint_epoch = self.head_state_slot() / SLOTS_PER_EPOCH;
+        let current_epoch = self.ticker.current_slot() / SLOTS_PER_EPOCH;
+        assert!(
+            current_epoch <= checkpoint_epoch + ws_period,
+            "checkpoint epoch {checkpoint_epoch} is outside the weak-subjectivity period \
+             ({ws_period} epochs); wall epoch {current_epoch} — refusing stale anchor \
+             (override with --disable-weak-subjectivity)",
+        );
+    }
+
     fn status_payload(&mut self) -> [u8; STATUS_V2_SIZE] {
         let fork_digest = self.fork_digest();
 
@@ -1105,6 +1134,28 @@ impl BeaconStateTile {
     fn da_boundary(&self) -> Slot {
         self.sync_finalized_slot.max(self.fork_choice.finalized_checkpoint.epoch * SLOTS_PER_EPOCH)
     }
+    fn replay_block(&mut self, data: &[u8]) {
+        if !SignedBeaconBlockView::check_size(data) {
+            tracing::error!(len = data.len(), "replayed on-disk block malformed");
+            return;
+        }
+
+        let block_slot = SignedBeaconBlockView::slot(data);
+        let feedback = self.apply_block_impl(data, false);
+        match feedback {
+            Feedback::Reject(block_root) => tracing::error!(
+                block_slot,
+                block_root = ?block_root.map(|r| hex32(&r)),
+                "replayed block rejected",
+            ),
+            _ => tracing::info!(
+                head_slot = self.head_state_slot(),
+                block_slot,
+                "replayed block: {:?}",
+                feedback
+            ),
+        }
+    }
 
     fn apply_block_impl(&mut self, data: &[u8], gate_da: bool) -> Feedback {
         let parsed = match self.precheck_block(data) {
@@ -1810,6 +1861,21 @@ impl Tile<SilverSpine> for BeaconStateTile {
         adapter.consume(|m: DataColumnsAvailable, producers| {
             self.handle_data_columns_available(m, producers);
         });
+
+        adapter.consume(|m: ReplayBlock, producers| match m {
+            ReplayBlock::Block { ssz } => {
+                let acquired = self.replay_consumer.acquire(ssz);
+                let data = acquired.buffer().ok().map(|(d, _)| d as *const [u8]);
+                if let Some(p) = data {
+                    self.replay_block(unsafe { &*p });
+                }
+            }
+            ReplayBlock::Done => {
+                producers.produce(self.status_event());
+                producers.produce(BeaconStateEvent::ReplayComplete);
+            }
+        });
+        self.replay_consumer.free();
     }
 }
 
@@ -1936,16 +2002,35 @@ mod tests {
 
     /// Tile whose ticker reports `wall_slot` as the current slot.
     fn make_tile_at_wall_slot(wall_slot: u64) -> BeaconStateTile {
+        make_tile_at_wall_slot_ws(wall_slot, true)
+    }
+
+    fn make_tile_at_wall_slot_ws(
+        wall_slot: u64,
+        verify_weak_subjectivity: bool,
+    ) -> BeaconStateTile {
         let secs_per_slot = 12u64;
         let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
         let genesis = now.saturating_sub(wall_slot * secs_per_slot + 1);
         let ticker = SlotTicker::new(genesis, Duration::from_secs(12), Duration::from_secs(4));
         let gossip_p = TCache::producer("test_gossip", 1 << 20);
         let event_p = TCache::producer("test_event", 1 << 20);
+        let replay_p = TCache::producer("test_replay", 1 << 20);
         let gossip_c = gossip_p.cache_ref().random_access("test_gossip", true).unwrap();
         let rpc_c = event_p.cache_ref().random_access("test_event", true).unwrap();
+        let replay_c = replay_p.cache_ref().random_access("test_replay", true).unwrap();
         let state = BeaconStateOwner::pre_bootstrap();
-        BeaconStateTile::new(ticker, SpecConfig::mainnet(), state, gossip_c, rpc_c, &[], &[])
+        BeaconStateTile::new(
+            ticker,
+            SpecConfig::mainnet(),
+            state,
+            gossip_c,
+            rpc_c,
+            replay_c,
+            verify_weak_subjectivity,
+            &[],
+            &[],
+        )
     }
 
     fn placeholder_pubkey(i: usize) -> BLSPubkey {

crates/beacon_state/tile/src/weak_subjectivity.rs

@@ -0,0 +1,50 @@
+use silver_beacon_state_data::{SLOTS_PER_EPOCH, StateReadView};
+
+use crate::shuffling;
+
+// Constants are deliberately the *pre-MaxEB* ones: the period's derivation
+// assumes a 32-ETH per-validator cap and count-based churn, and has not been
+// re-derived for EIP-7251. `MAX_EB_ETH` is therefore 32 — not the runtime
+// 2048-ETH `MAX_EFFECTIVE_BALANCE` — matching the spec-as-written and other
+// clients. A known approximation under MaxEB.
+const SAFETY_DECAY: u64 = 10;
+const MIN_WITHDRAWABILITY_DELAY: u64 = 256;
+const MAX_EB_ETH: u64 = 32;
+const ETH_TO_GWEI: u64 = 1_000_000_000;
+const MAX_DEPOSITS: u64 = 16;
+const MIN_PER_EPOCH_CHURN_LIMIT: u64 = 4;
+const CHURN_LIMIT_QUOTIENT: u64 = 1 << 16;
+
+pub(crate) fn weak_subjectivity_period(view: &StateReadView, scratch: &mut Vec<u32>) -> u64 {
+    let epoch = view.slot.state().slot / SLOTS_PER_EPOCH;
+    scratch.clear();
+    shuffling::get_active_validator_indices_into(&view.validators, epoch, scratch);
+    let n = scratch.len() as u64;
+    if n == 0 {
+        return MIN_WITHDRAWABILITY_DELAY;
+    }
+
+    // get_total_active_balance, floored at one increment, averaged to whole ETH.
+    let total = scratch
+        .iter()
+        .map(|&i| view.validators.effective_balance(i as usize))
+        .sum::<u64>()
+        .max(ETH_TO_GWEI);
+    let d = SAFETY_DECAY;
+    let t = total / n / ETH_TO_GWEI;
+    let cap = MAX_EB_ETH;
+    let delta = (n / CHURN_LIMIT_QUOTIENT).max(MIN_PER_EPOCH_CHURN_LIMIT);
+    let big_delta = MAX_DEPOSITS * SLOTS_PER_EPOCH;
+
+    let mut ws = MIN_WITHDRAWABILITY_DELAY;
+    // On the else side the branch guarantees `t <= cap*(200+3d)/(200+12d) < cap`,
+    // so `cap - t > 0` (no div-by-zero) and the subtraction above never wraps.
+    if cap * (200 + 3 * d) < t * (200 + 12 * d) {
+        let churn = n * (t * (200 + 12 * d) - cap * (200 + 3 * d)) / (600 * delta * (2 * t + cap));
+        let top_ups = n * (200 + 3 * d) / (600 * big_delta);
+        ws += churn.max(top_ups);
+    } else {
+        ws += 3 * n * d * t / (200 * big_delta * (cap - t));
+    }
+    ws
+}