eth_state_diff/

participation.rs

//! Compact delta encoding for Ethereum epoch participation flags.
//!
//! This module computes and applies deltas between validator participation
//! vectors.
//!
//! Participation changes are typically sparse: during an epoch transition,
//! most validators retain their existing participation flags while only a
//! subset of validators receive new values. The delta therefore stores only
//! modified indices and their replacement values rather than the complete
//! target vector.
//!
//! ## Encoding
//!
//! The delta has two representations:
//!
//! - [`ParticipationDiff::AllZeros`] represents a target vector containing
//!   only zero-valued participation flags.
//! - [`ParticipationDiff::Sparse`] stores only changed entries.
//!
//! For the sparse representation, modified indices are encoded as
//! delta-varint gaps between successive modified indices. The corresponding
//! replacement values are stored separately in `new_values`. This avoids
//! storing unchanged participation flags and makes sequences of nearby
//! changes particularly compact.
//!
//! Validators present in the target vector but not in the base vector are
//! stored separately in `extension` and appended during application.
//!
//! ## APIs
//!
//! [`diff_participation`] and [`apply_participation`] operate on contiguous
//! vectors and provide the specialized [`ParticipationDiff::AllZeros`] fast
//! path.
//!
//! [`diff_participation_iter`] and [`apply_participation_iter`] operate through
//! iterators and [`crate::ListMutTarget`], allowing consensus clients with
//! tree-backed or otherwise non-contiguous state representations to compute
//! and apply participation deltas without first materializing the complete
//! vector.
//!
//! The iterator-based diff API always produces the sparse representation.
//! The slice-based API can additionally detect an all-zero target and use the
//! more compact [`ParticipationDiff::AllZeros`] representation.
//!
//! ## Reconstruction
//!
//! Applying a sparse delta updates only the modified indices of the existing
//! collection and then appends any values in `extension`.
//!
//! Applying an [`ParticipationDiff::AllZeros`] delta replaces the destination
//! with a vector of the specified length containing only zero values.
//!
//! ## Complexity
//!
//! Diff generation is `O(n)` time and `O(m)` additional space, where `n` is
//! the number of participation flags examined and `m` is the number of
//! modified entries.
//!
//! Sparse delta application is `O(m + e)` time, where `m` is the number of
//! modified entries and `e` is the number of appended participation flags.
//!
//! The contiguous all-zero fast path performs `O(n)` work to initialize the
//! resulting vector.
//!
//! ## Serialization
//!
//! [`ParticipationDiff`] is designed to be serialized with `rkyv` and can be
//! subsequently compressed with a general-purpose compressor such as `zstd`.

use crate::{
    balances::{read_varint, write_varint},
    types::{ArchivedParticipationDiff, ParticipationDiff},
    Error,
};

/// Computes a compact delta between two participation flag slices.
///
/// The returned [`ParticipationDiff`] contains the information required to
/// reconstruct `target` from `base`.
///
/// If every flag in `target` is zero, the function uses the specialized
/// [`ParticipationDiff::AllZeros`] representation. Otherwise it produces a
/// sparse delta containing only modified flags.
///
/// This is the contiguous-slice convenience API. Clients whose participation
/// flags are stored in a non-contiguous representation can use
/// [`diff_participation_iter`] instead.
///
/// # Complexity
///
/// `O(n)` time and `O(m)` additional space, where `n` is the number of flags
/// examined and `m` is the number of modified flags.
pub fn diff_participation(base: &[u8], target: &[u8]) -> ParticipationDiff {
    if target.iter().all(|&value| value == 0) {
        return ParticipationDiff::AllZeros(
            target
                .len()
                .try_into()
                .expect("target length exceeds u32::MAX"),
        );
    }

    diff_participation_iter(base.iter().copied(), target.iter().copied())
}

/// Applies a participation delta to a contiguous vector in place.
///
/// This is the contiguous-vector convenience API. Sparse deltas are delegated
/// to [`apply_participation_iter`], while [`ParticipationDiff::AllZeros`] is
/// handled directly by replacing the destination with a zero-filled vector of
/// the encoded length.
///
/// After successful application, `base` contains the target participation
/// vector from which `delta` was produced.
///
/// # Errors
///
/// Returns [`Error::InvalidDelta`] if an encoded [`ParticipationDiff::AllZeros`]
/// length cannot be represented by the target vector or if a sparse delta is
/// internally inconsistent.
///
/// Returns [`Error::MalformedDelta`] if a sparse delta contains invalid
/// serialized payload data, such as a truncated or overflowing varint.
///
/// # Complexity
///
/// Sparse deltas require `O(m + e)` work, where `m` is the number of modified
/// entries and `e` is the number of appended flags.
///
/// An [`ParticipationDiff::AllZeros`] delta requires `O(n)` work to construct
/// the resulting zero-filled vector of length `n`.
pub fn apply_participation(
    base: &mut Vec<u8>,
    delta: &ArchivedParticipationDiff,
) -> Result<(), Error> {
    match delta {
        ArchivedParticipationDiff::AllZeros(len) => {
            let len = usize::try_from(len.to_native()).map_err(|_| {
                Error::InvalidDelta("all-zero participation length does not fit in usize".into())
            })?;

            base.clear();
            base.resize(len, 0);

            Ok(())
        }
        ArchivedParticipationDiff::Sparse { .. } => apply_participation_iter(base, delta),
    }
}

/// Computes a compact sparse delta between two participation flag iterators.
///
/// This API is intended for consensus clients whose participation flags are
/// stored in tree-backed or otherwise non-contiguous structures. The caller
/// can expose the values through [`ExactSizeIterator`]s without first
/// materializing the complete vectors as contiguous buffers.
///
/// The iterators are consumed during diff generation.
///
/// Unlike [`diff_participation`], this function always returns
/// [`ParticipationDiff::Sparse`]. It does not perform the all-zero
/// specialization because the iterator is consumed while determining the
/// changed entries.
///
/// Values remaining in `target` after the common portion are treated as
/// newly appended participation flags and are stored in the delta's
/// `extension` field.
///
/// # Complexity
///
/// `O(n)` time and `O(m + e)` additional space, where `n` is the size of the
/// common portion, `m` is the number of modified entries, and `e` is the
/// number of appended target entries.
pub fn diff_participation_iter<I1, I2>(mut base: I1, mut target: I2) -> ParticipationDiff
where
    I1: ExactSizeIterator<Item = u8>,
    I2: ExactSizeIterator<Item = u8>,
{
    let common_len = base.len().min(target.len());

    let mut sparse_indices = Vec::with_capacity(50_000);
    let mut new_values = Vec::with_capacity(50_000);
    let mut last_idx = 0u64;

    for i in 0..common_len {
        let Some(v1) = base.next() else {
            break;
        };

        let Some(v2) = target.next() else {
            break;
        };

        if v1 != v2 {
            let idx = i as u64;

            let gap = idx
                .checked_sub(last_idx)
                .expect("changed indices are processed in strictly increasing iterator order");
            write_varint(gap, &mut sparse_indices);

            new_values.push(v2);
            last_idx = idx;
        }
    }

    let extension = target.collect();

    ParticipationDiff::Sparse {
        sparse_indices,
        new_values,
        extension,
    }
}

/// Applies a sparse participation delta to a mutable collection in place.
///
/// This API is intended for consensus clients whose participation flags are
/// stored in tree-backed or otherwise non-contiguous structures.
///
/// The destination collection is updated according to the sparse entries in
/// `delta`. Each encoded index gap identifies the next modified entry, whose
/// value is replaced with the corresponding entry from `new_values`. Values
/// in `extension` are then appended to the destination.
///
/// [`ParticipationDiff::AllZeros`] is not supported by this generic API
/// because [`crate::ListMutTarget`] does not provide an operation for clearing
/// or resizing an existing collection. Callers should use
/// [`apply_participation`] when they need to support that representation.
///
/// # Errors
///
/// Returns [`Error::InvalidDelta`] if the supplied delta is not a sparse
/// representation, if the number of encoded indices does not match the number
/// of replacement values, if an encoded index cannot be represented as a
/// `usize`, if an index falls outside the destination collection, or if the
/// destination collection cannot provide the required element.
///
/// Returns [`Error::MalformedDelta`] if the sparse index payload contains an
/// invalid or truncated varint.
///
/// # Complexity
///
/// `O(m + e)` time and `O(1)` additional working space, excluding allocations
/// performed by the destination collection when it grows.
///
/// Here `m` is the number of modified entries and `e` is the number of
/// appended entries.
pub fn apply_participation_iter<T: crate::ListMutTarget<u8>>(
    target: &mut T,
    delta: &ArchivedParticipationDiff,
) -> Result<(), Error> {
    let ArchivedParticipationDiff::Sparse {
        sparse_indices,
        new_values,
        extension,
    } = delta
    else {
        return Err(Error::InvalidDelta(
            "AllZeros participation delta cannot be applied through the generic iterator API"
                .into(),
        ));
    };

    let indices_raw = sparse_indices.as_slice();
    let mut cursor = 0usize;
    let mut current_idx = 0usize;

    for value in new_values.iter() {
        let gap = read_varint(indices_raw, &mut cursor)?;

        let gap = usize::try_from(gap).map_err(|_| {
            Error::MalformedDelta("participation index gap does not fit in usize".into())
        })?;

        current_idx = current_idx.checked_add(gap).ok_or_else(|| {
            Error::MalformedDelta(
                "participation index overflow while decoding sparse indices".into(),
            )
        })?;

        let Some(target_value) = target.get_mut(current_idx) else {
            return Err(Error::InvalidDelta(format!(
                "participation index {current_idx} is outside target collection of length {}",
                target.len()
            )));
        };

        *target_value = *value;
    }

    if cursor != indices_raw.len() {
        return Err(Error::InvalidDelta(
            "participation sparse index payload contains unused bytes".into(),
        ));
    }

    for byte in extension.iter() {
        target.push(*byte);
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::ArchivedParticipationDiff;
    use crate::ListMutTarget;

    struct MockTarget {
        inner: Vec<u8>,
    }

    impl ListMutTarget<u8> for MockTarget {
        fn len(&self) -> usize {
            self.inner.len()
        }

        fn get_mut(&mut self, index: usize) -> Option<&mut u8> {
            self.inner.get_mut(index)
        }

        fn push(&mut self, value: u8) {
            self.inner.push(value);
        }
    }

    /// Helper to assert a full roundtrip using the slice APIs, guaranteeing a sparse delta.
    fn assert_sparse_roundtrip(base: &[u8], target: &[u8]) {
        let delta = diff_participation(base, target);

        match &delta {
            ParticipationDiff::AllZeros(_) => panic!("test setup: expected sparse delta"),
            ParticipationDiff::Sparse { .. } => {}
        }

        let bytes = rkyv::to_bytes::<rkyv::rancor::Error>(&delta).expect("test setup: serialize");
        let archived = rkyv::access::<ArchivedParticipationDiff, rkyv::rancor::Error>(&bytes)
            .expect("test setup: failed to access archived delta");

        let mut reconstructed = base.to_vec();
        apply_participation(&mut reconstructed, archived).expect("test setup: apply");

        assert_eq!(reconstructed, target);
    }

    #[test]
    fn test_diff_slice_no_changes() {
        let base = vec![1, 2, 3];
        let target = vec![1, 2, 3];
        let delta = diff_participation(&base, &target);

        match delta {
            ParticipationDiff::Sparse {
                sparse_indices,
                new_values,
                extension,
            } => {
                assert!(sparse_indices.is_empty());
                assert!(new_values.is_empty());
                assert!(extension.is_empty());
            }
            _ => panic!("test setup: expected sparse"),
        }
    }

    #[test]
    fn test_diff_slice_all_zeros_fast_path() {
        let base = vec![1, 2, 3];
        let target = vec![0, 0, 0];
        let delta = diff_participation(&base, &target);

        match delta {
            ParticipationDiff::AllZeros(len) => assert_eq!(len, 3),
            _ => panic!("expected AllZeros fast path"),
        }
    }

    #[test]
    fn test_apply_all_zeros() {
        let mut base = vec![1, 2, 3];
        let delta = ParticipationDiff::AllZeros(5);

        let bytes = rkyv::to_bytes::<rkyv::rancor::Error>(&delta).expect("test setup: serialize");
        let archived = rkyv::access::<ArchivedParticipationDiff, rkyv::rancor::Error>(&bytes)
            .expect("test setup: failed to access archived delta");

        apply_participation(&mut base, archived).expect("test setup: apply");

        assert_eq!(base, vec![0, 0, 0, 0, 0]);
    }

    #[test]
    fn test_diff_iter_all_zeros_produces_sparse() {
        let base = vec![1, 2, 3];
        let target = vec![0, 0, 0];
        let delta = diff_participation_iter(base.into_iter(), target.into_iter());

        match delta {
            ParticipationDiff::Sparse { .. } => {} // Pass
            _ => panic!("iterator API must produce sparse, not AllZeros"),
        }
    }

    #[test]
    fn test_sparse_roundtrip_with_changes() {
        let base = vec![0, 0, 0, 0];
        let target = vec![1, 0, 2, 0];
        assert_sparse_roundtrip(&base, &target);
    }

    #[test]
    fn test_sparse_roundtrip_with_appended() {
        let base = vec![0];
        let target = vec![0, 5, 6];
        assert_sparse_roundtrip(&base, &target);
    }

    #[test]
    fn test_sparse_roundtrip_combined() {
        let base = vec![10, 20, 30];
        let target = vec![10, 99, 30, 40, 50]; // idx 1 changed, 3 & 4 appended
        assert_sparse_roundtrip(&base, &target);
    }

    #[test]
    fn test_apply_all_zeros_via_iter_api_errors() {
        let delta = ParticipationDiff::AllZeros(5);
        let bytes = rkyv::to_bytes::<rkyv::rancor::Error>(&delta).expect("test setup: serialize");
        let archived = rkyv::access::<ArchivedParticipationDiff, rkyv::rancor::Error>(&bytes)
            .expect("test setup: failed to access archived delta");

        let mut target = MockTarget { inner: vec![] };
        let result = apply_participation_iter(&mut target, archived);

        assert!(result.is_err());
        let err_str = format!("{}", result.expect_err("test setup"));
        assert!(err_str.contains("AllZeros participation delta cannot be applied"));
    }

    #[test]
    fn test_apply_sparse_mismatched_counts() {
        // 1 encoded index, but 0 replacement values
        let delta = ParticipationDiff::Sparse {
            sparse_indices: vec![0], // gap = 0
            new_values: vec![],
            extension: vec![],
        };
        let bytes = rkyv::to_bytes::<rkyv::rancor::Error>(&delta).expect("test setup: serialize");
        let archived = rkyv::access::<ArchivedParticipationDiff, rkyv::rancor::Error>(&bytes)
            .expect("test setup: failed to access archived delta");

        let mut target = MockTarget { inner: vec![0] };
        let result = apply_participation_iter(&mut target, archived);

        assert!(result.is_err());
        let err_str = format!("{}", result.expect_err("test setup"));
        assert!(err_str.contains("unused bytes"));
    }

    #[test]
    fn test_apply_sparse_index_out_of_bounds() {
        // gap = 5, but target collection has length 3
        let delta = ParticipationDiff::Sparse {
            sparse_indices: vec![5],
            new_values: vec![99],
            extension: vec![],
        };
        let bytes = rkyv::to_bytes::<rkyv::rancor::Error>(&delta).expect("test setup: serialize");
        let archived = rkyv::access::<ArchivedParticipationDiff, rkyv::rancor::Error>(&bytes)
            .expect("test setup: failed to access archived delta");

        let mut target = MockTarget { inner: vec![0; 3] };
        let result = apply_participation_iter(&mut target, archived);

        assert!(result.is_err());
        let err_str = format!("{}", result.expect_err("test setup"));
        assert!(err_str.contains("outside target collection"));
    }

    #[test]
    fn test_apply_sparse_truncated_varint() {
        let delta = ParticipationDiff::Sparse {
            sparse_indices: vec![0xFF], // Continuation bit set, but buffer ends
            new_values: vec![0],
            extension: vec![],
        };
        let bytes = rkyv::to_bytes::<rkyv::rancor::Error>(&delta).expect("test setup: serialize");
        let archived = rkyv::access::<ArchivedParticipationDiff, rkyv::rancor::Error>(&bytes)
            .expect("test setup: failed to access archived delta");

        let mut target = MockTarget { inner: vec![0] };
        let result = apply_participation_iter(&mut target, archived);

        assert!(result.is_err());
        let err_str = format!("{}", result.expect_err("test setup"));
        assert!(err_str.contains("truncated varint"));
    }

    #[test]
    fn test_apply_sparse_index_sum_overflow() {
        // Gap 1: 999 (valid index, fits in target)
        // Gap 2: u64::MAX (causes checked_add overflow to trigger on the 2nd loop)
        let sparse_indices: Vec<u8> = vec![
            0xE7, 0x07, // varint(999)
            0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x01, // varint(u64::MAX)
        ];

        let delta = ParticipationDiff::Sparse {
            sparse_indices,
            new_values: vec![0, 0],
            extension: vec![],
        };
        let bytes = rkyv::to_bytes::<rkyv::rancor::Error>(&delta).expect("test setup: serialize");
        let archived = rkyv::access::<ArchivedParticipationDiff, rkyv::rancor::Error>(&bytes)
            .expect("test setup: failed to access archived delta");

        // Target must be large enough to satisfy the first gap (999)
        let mut target = MockTarget {
            inner: vec![0; 1000],
        };
        let result = apply_participation_iter(&mut target, archived);

        assert!(result.is_err());
        let err_str = format!("{}", result.expect_err("test setup"));
        assert!(err_str.contains("index overflow"));
    }
}