perf(beacon): keep inactivity scores and the ring buffers on the tree

crates/blockchain/state_transition/src/beacon/block_production.rs

@@ -594,6 +594,10 @@ pub fn assemble_block(
         &ActiveBalanceCache::default(),
         options,
     )?;
+    // Fold the block's writes into their trees first, so the root is computed
+    // on (and cached in) the state's own nodes rather than a throwaway copy,
+    // as `state_transition` does before it checks a block's state root.
+    post.apply_pending_mutations();
     block.state_root = post.hash_tree_root();
     Ok(block)
 }

crates/blockchain/state_transition/src/beacon/stf/epoch/altair.rs

@@ -104,7 +104,7 @@ pub(super) fn weigh_with_totals(
 /// epoch's participation and genesis has none.
 ///
 /// Shared by every fork from altair on, gloas included: reading
-/// [`crate::beacon::containers::BeaconState::inactivity_scores_mut`] rather than
+/// [`crate::beacon::containers::BeaconState::inactivity_score_mut`] rather than
 /// [`crate::beacon::containers::BeaconState::altair_validator_lists_mut`] is what
 /// makes that so, since this only ever writes a score in place and never
 /// grows or replaces the list, and that narrower accessor extends to a gloas
@@ -129,26 +129,40 @@ fn update_inactivity_scores(
 ) -> Result<()> {
     let leaking = is_in_inactivity_leak(state)?;
 
-    let inactivity_scores = state.inactivity_scores_mut()?;
-    let score_count = inactivity_scores.len();
-    for (index, (flags, _)) in summary.iter().enumerate() {
-        if !flags.is_eligible() {
-            continue;
-        }
-        let score = inactivity_scores
-            .get_mut(index)
-            .ok_or(Error::IndexOutOfBounds {
+    // Decided over an in-order walk of the scores and written afterwards: a
+    // tree-backed list is read with an iterator (one descent per element would
+    // dominate), and only a score that changes is written, since outside a
+    // leak almost every score stays zero and a write of an equal value would
+    // still unshare its leaf from the parent state's.
+    let changes = {
+        let (_, _, inactivity_scores) = state.altair_validator_lists()?;
+        let score_count = inactivity_scores.len();
+        let mut score_iter = inactivity_scores.iter();
+        let mut changes: Vec<(usize, u64)> = Vec::new();
+        for (index, (flags, _)) in summary.iter().enumerate() {
+            // Advanced for ineligible validators too, to stay in step with `index`;
+            // a score list shorter than the registry is an error only at the first
+            // eligible validator without one.
+            let score = score_iter.next().copied();
+            if !flags.is_eligible() {
+                continue;
+            }
+            let score = score.ok_or(Error::IndexOutOfBounds {
                 index,
                 len: score_count,
             })?;
 
-        let updated = next_inactivity_score(*score, flags, leaking, config)?;
-
-        if updated != *score {
-            *score = updated;
+            let updated = next_inactivity_score(score, flags, leaking, config)?;
+            if updated != score {
+                changes.push((index, updated));
+            }
         }
-    }
+        changes
+    };
 
+    for (index, updated) in changes {
+        *state.inactivity_score_mut(index)? = updated;
+    }
     Ok(())
 }
 
@@ -515,7 +529,7 @@ mod tests {
         process_inactivity_updates(&mut state, &Config::mainnet()).unwrap();
 
         let (_, _, scores) = state.altair_validator_lists().unwrap();
-        assert_eq!(scores, before);
+        assert_eq!(scores.to_vec(), before);
     }
 
     #[test]
@@ -570,6 +584,58 @@ mod tests {
         assert_eq!(scores[0], 0);
     }
 
+    /// Outside a leak a missed epoch adds the bias and recovery takes it back
+    /// to zero, so every score ends where it started. The pass must then
+    /// leave the list untouched, not just equal: a write of an equal value
+    /// would still unshare the tree from the parent state's.
+    #[test]
+    fn inactivity_updates_that_change_nothing_leave_the_tree_shared() {
+        let mut state = altair_state_with_validators(4);
+        state.apply_pending_mutations();
+        let before_state = state.clone();
+        let (_, _, before) = before_state.altair_validator_lists().unwrap();
+
+        process_inactivity_updates(&mut state, &Config::mainnet()).unwrap();
+
+        let (_, _, scores) = state.altair_validator_lists().unwrap();
+        assert!(!scores.has_pending_updates(), "no write was buffered");
+        assert!(scores.ptr_eq(&before));
+        assert_eq!(scores.to_vec(), vec![0; 4]);
+    }
+
+    #[test]
+    fn inactivity_updates_buffer_only_the_scores_that_change() {
+        let mut state = altair_state_with_validators(4);
+        {
+            let (previous_epoch_participation, _, scores) =
+                state.altair_validator_lists_mut().unwrap();
+            scores[2] = 3;
+            // Validator 2 participates, so its score falls to zero; the rest
+            // stay at zero.
+            previous_epoch_participation[2] = add_flag(0, constants::TIMELY_TARGET_FLAG_INDEX);
+        }
+        state.apply_pending_mutations();
+
+        process_inactivity_updates(&mut state, &Config::mainnet()).unwrap();
+
+        let (_, _, scores) = state.altair_validator_lists().unwrap();
+        assert!(scores.has_pending_updates());
+        assert_eq!(scores.to_vec(), vec![0; 4]);
+    }
+
+    #[test]
+    fn inactivity_updates_still_reject_an_overflowing_score() {
+        let mut state = altair_state_with_validators(4);
+        {
+            let (_, _, scores) = state.altair_validator_lists_mut().unwrap();
+            scores[1] = u64::MAX;
+        }
+
+        let result = process_inactivity_updates(&mut state, &Config::mainnet());
+
+        assert!(matches!(result, Err(Error::ArithmeticOverflow(_))));
+    }
+
     // -----------------------------------------------------------------------
     // process_rewards_and_penalties
     // -----------------------------------------------------------------------

crates/blockchain/state_transition/src/beacon/stf/epoch/gloas.rs

@@ -31,7 +31,7 @@
 //!   [`super::altair::process_rewards_and_penalties`]) reach
 //!   `previous_epoch_participation`/`current_epoch_participation`/`inactivity_scores`
 //!   as plain slices, through [`crate::beacon::containers::BeaconState::altair_validator_lists`]
-//!   and [`crate::beacon::containers::BeaconState::inactivity_scores_mut`]: both
+//!   and [`crate::beacon::containers::BeaconState::inactivity_score_mut`]: both
 //!   `SszList` and `ProgressiveList` `Deref` to a plain `[T]`.
 //! - **[`super::electra::process_pending_consolidations`]** takes the queue
 //!   as a `Vec` and writes it back, through

crates/blockchain/state_transition/src/beacon/stf/mod.rs

@@ -357,6 +357,7 @@ mod tests {
     use super::*;
     use crate::beacon::fork::ForkName;
     use crate::beacon::helpers::test_state;
+    use crate::beacon::primitives::Root;
 
     #[test]
     fn process_slot_unsets_next_slots_payload_availability_on_gloas() {
@@ -433,7 +434,11 @@ mod tests {
 
         process_slot(&mut state).unwrap();
 
-        assert!(!state.has_pending_mutations());
+        assert!(!state.registry_has_pending_updates());
+        // Its own two roots writes stay buffered until the next flush, which
+        // the following slot (or `process_slots`' last step) performs.
+        assert!(state.state_roots().has_pending_updates());
+        assert!(state.block_roots().has_pending_updates());
     }
 
     /// Epoch processing (`process_rewards_and_penalties` here) writes every
@@ -452,4 +457,112 @@ mod tests {
 
         assert!(!state.has_pending_mutations());
     }
+
+    #[test]
+    fn process_slots_flushes_the_roots_it_wrote_in_its_last_slot() {
+        for fork in BEACON_FORKS {
+            let mut state = test_state::with_validators_at(fork, 4);
+            let target_slot = state.slot() + 1;
+
+            process_slots(&mut state, target_slot, &Config::mainnet()).unwrap();
+
+            assert!(!state.has_pending_mutations(), "{fork:?}");
+        }
+    }
+
+    const BEACON_FORKS: [ForkName; 8] = [
+        ForkName::Phase0,
+        ForkName::Altair,
+        ForkName::Bellatrix,
+        ForkName::Capella,
+        ForkName::Deneb,
+        ForkName::Electra,
+        ForkName::Fulu,
+        ForkName::Gloas,
+    ];
+
+    /// Writes one element of the field a case names, or answers `false` if the
+    /// fork does not have it.
+    type Write = fn(&mut BeaconState) -> bool;
+
+    fn push_historical_summary(state: &mut BeaconState) -> bool {
+        let summary = containers::HistoricalSummary::default();
+        match state {
+            BeaconState::Capella(s) => s.historical_summaries.push(summary).unwrap(),
+            BeaconState::Deneb(s) => s.historical_summaries.push(summary).unwrap(),
+            BeaconState::Electra(s) => s.historical_summaries.push(summary).unwrap(),
+            BeaconState::Fulu(s) => s.historical_summaries.push(summary).unwrap(),
+            BeaconState::Gloas(s) => s.historical_summaries.push(summary).unwrap(),
+            _ => return false,
+        }
+        true
+    }
+
+    /// One write to every tree-backed field of every fork: the flush and the
+    /// pending check must both see all of them, since both come from one
+    /// field list.
+    #[test]
+    fn every_tree_field_of_every_fork_is_flushed_and_checked() {
+        let cases: [(&str, Write); 11] = [
+            ("validators", |s| {
+                s.validator_mut(0).unwrap().effective_balance -= 1;
+                true
+            }),
+            ("balances", |s| {
+                *s.balance_mut(0).unwrap() += 1;
+                true
+            }),
+            ("block_roots", |s| {
+                s.block_roots_mut()[0] = Root::repeat_byte(1);
+                true
+            }),
+            ("state_roots", |s| {
+                s.state_roots_mut()[0] = Root::repeat_byte(1);
+                true
+            }),
+            ("historical_roots", |s| {
+                s.historical_roots_mut().push(Root::repeat_byte(1)).unwrap();
+                true
+            }),
+            ("eth1_data_votes", |s| {
+                s.eth1_data_votes_mut().push(Default::default()).unwrap();
+                true
+            }),
+            ("randao_mixes", |s| {
+                s.randao_mixes_mut()[0] = Root::repeat_byte(1);
+                true
+            }),
+            ("slashings", |s| {
+                s.slashings_mut()[0] += 1;
+                true
+            }),
+            // Gloas keeps its scores in a flat progressive list that buffers
+            // nothing, so a write there leaves nothing pending.
+            ("inactivity_scores", |s| {
+                let buffered = !matches!(s, BeaconState::Gloas(_));
+                match s.inactivity_score_mut(0) {
+                    Ok(score) => {
+                        *score += 1;
+                        buffered
+                    }
+                    Err(_) => false,
+                }
+            }),
+            ("historical_summaries", push_historical_summary),
+            // A read-only pass writes nothing and must leave nothing pending.
+            ("none", |_| false),
+        ];
+
+        for fork in BEACON_FORKS {
+            let mut state = test_state::with_validators_at(fork, 4);
+            assert!(!state.has_pending_mutations(), "{fork:?} starts flushed");
+
+            for (name, write) in cases {
+                let wrote = write(&mut state);
+                assert_eq!(state.has_pending_mutations(), wrote, "{fork:?} {name}");
+                state.apply_pending_mutations();
+                assert!(!state.has_pending_mutations(), "{fork:?} {name} flushed");
+            }
+        }
+    }
 }

crates/blockchain/state_transition/src/beacon/upgrade.rs

@@ -1411,8 +1411,8 @@ mod tests {
         // place rather than clearing it, since `historical_summaries` is
         // where new history accumulates from here on.
         assert_eq!(
-            post.historical_roots.into_inner(),
-            bellatrix_state.historical_roots.into_inner()
+            post.historical_roots.to_vec(),
+            bellatrix_state.historical_roots.to_vec()
         );
         assert_eq!(post.next_withdrawal_index, 0);
         assert_eq!(post.next_withdrawal_validator_index, 0);

crates/blockchain/state_transition/tests/beacon_spec/epoch_processing.rs

@@ -185,15 +185,15 @@ fn apply(
         // module's own doc. `inactivity_updates` reads gloas too, below: its
         // participation and inactivity lists are progressive (EIP-7688), a
         // different Rust type from the bounded `SszList` altair's own copy
-        // used to need, but `BeaconState::inactivity_scores_mut` and
-        // `altair_validator_lists` both now reach either list kind through a
-        // shared slice type. `participation_flag_updates`, just below, is the
+        // used to need, but `BeaconState::inactivity_score_mut` and
+        // `altair_validator_lists` both now reach either list kind (the latter
+        // through `InactivityScoresRef`). `participation_flag_updates`, just below, is the
         // one exception: it *replaces* the whole list rather than writing an
         // element, which only the fork's own concrete container type can do,
         // so gloas keeps its own copy of that one step; see
         // `epoch::gloas`'s own module doc.
         "inactivity_updates" => match fork {
-            // `BeaconState::inactivity_scores_mut` reaches a gloas state's
+            // `BeaconState::inactivity_score_mut` reaches a gloas state's
             // scores too (element writes only, which both list kinds allow
             // through `DerefMut`; see that accessor's own doc), so this one
             // copy serves gloas as well.

crates/common/ssz-tree/src/lib.rs

@@ -91,6 +91,19 @@ pub use vector::Vector;
 use libssz::{SszDecode, SszEncode};
 use libssz_merkle::HashTreeRoot;
 
+/// The buffered-write surface of a [`List`] or [`Vector`], independent of its
+/// element type and update map.
+///
+/// Object safe, so a container can hold one `&dyn Buffered` per tree field of
+/// different element types and flush or check them all from one field list.
+pub trait Buffered {
+    /// Folds every buffered write into the tree.
+    fn apply_updates(&mut self);
+
+    /// Whether any write is buffered and not yet folded into the tree.
+    fn has_pending_updates(&self) -> bool;
+}
+
 /// A 32-byte Merkle node.
 pub(crate) type Hash256 = libssz_merkle::Node;
 

crates/common/ssz-tree/src/list.rs

@@ -137,6 +137,16 @@ impl<T: Value, const N: usize, U: UpdateMap<T>> List<T, N, U> {
     }
 }
 
+impl<T: Value, const N: usize, U: UpdateMap<T>> crate::Buffered for List<T, N, U> {
+    fn apply_updates(&mut self) {
+        List::apply_updates(self);
+    }
+
+    fn has_pending_updates(&self) -> bool {
+        List::has_pending_updates(self)
+    }
+}
+
 impl<T: Value, const N: usize, U: UpdateMap<T>> Default for List<T, N, U> {
     fn default() -> Self {
         Self::empty()

crates/common/ssz-tree/src/progressive_list.rs

@@ -230,6 +230,16 @@ impl<T: Value, U: UpdateMap<T>> ProgressiveIterCow<'_, T, U> {
     }
 }
 
+impl<T: Value, U: UpdateMap<T>> crate::Buffered for ProgressiveList<T, U> {
+    fn apply_updates(&mut self) {
+        ProgressiveList::apply_updates(self);
+    }
+
+    fn has_pending_updates(&self) -> bool {
+        ProgressiveList::has_pending_updates(self)
+    }
+}
+
 impl<T: Value, U: UpdateMap<T>> Default for ProgressiveList<T, U> {
     fn default() -> Self {
         Self::empty()

crates/common/ssz-tree/src/vector.rs

@@ -108,6 +108,16 @@ impl<T: Value, const N: usize, U: UpdateMap<T>> Vector<T, N, U> {
     }
 }
 
+impl<T: Value, const N: usize, U: UpdateMap<T>> crate::Buffered for Vector<T, N, U> {
+    fn apply_updates(&mut self) {
+        Vector::apply_updates(self);
+    }
+
+    fn has_pending_updates(&self) -> bool {
+        Vector::has_pending_updates(self)
+    }
+}
+
 impl<T: Value + Default, const N: usize, U: UpdateMap<T>> Default for Vector<T, N, U> {
     /// A vector of `N` default-valued elements.
     fn default() -> Self {