Shrink persisted fork choice data (#7805)

Closes:

- https://github.com/sigp/lighthouse/issues/7760


  - [x] Remove `balances_cache` from `PersistedForkChoiceStore` (~65 MB saving on mainnet)
- [x] Remove `justified_balances` from `PersistedForkChoiceStore` (~16 MB saving on mainnet)
- [x] Remove `balances` from `ProtoArray`/`SszContainer`.
- [x] Implement zstd compression for votes
- [x] Fix bug in justified state usage
- [x] Bump schema version to V28 and implement migration.

beacon_node/store/src/hdiff.rs

@@ -6,14 +6,12 @@ use serde::{Deserialize, Serialize};
 use ssz::{Decode, Encode};
 use ssz_derive::{Decode, Encode};
 use std::cmp::Ordering;
-use std::io::{Read, Write};
 use std::ops::RangeInclusive;
 use std::str::FromStr;
 use std::sync::LazyLock;
 use superstruct::superstruct;
 use types::historical_summary::HistoricalSummary;
 use types::{BeaconState, ChainSpec, Epoch, EthSpec, Hash256, List, Slot, Validator};
-use zstd::{Decoder, Encoder};
 
 static EMPTY_PUBKEY: LazyLock<PublicKeyBytes> = LazyLock::new(PublicKeyBytes::empty);
 
@@ -395,13 +393,17 @@ impl CompressedU64Diff {
             .collect();
 
         Ok(CompressedU64Diff {
-            bytes: compress_bytes(&uncompressed_bytes, config)?,
+            bytes: config
+                .compress_bytes(&uncompressed_bytes)
+                .map_err(Error::Compression)?,
         })
     }
 
     pub fn apply(&self, xs: &mut Vec<u64>, config: &StoreConfig) -> Result<(), Error> {
         // Decompress balances diff.
-        let balances_diff_bytes = uncompress_bytes(&self.bytes, config)?;
+        let balances_diff_bytes = config
+            .decompress_bytes(&self.bytes)
+            .map_err(Error::Compression)?;
 
         for (i, diff_bytes) in balances_diff_bytes
             .chunks(u64::BITS as usize / 8)
@@ -428,22 +430,6 @@ impl CompressedU64Diff {
     }
 }
 
-fn compress_bytes(input: &[u8], config: &StoreConfig) -> Result<Vec<u8>, Error> {
-    let compression_level = config.compression_level;
-    let mut out = Vec::with_capacity(config.estimate_compressed_size(input.len()));
-    let mut encoder = Encoder::new(&mut out, compression_level).map_err(Error::Compression)?;
-    encoder.write_all(input).map_err(Error::Compression)?;
-    encoder.finish().map_err(Error::Compression)?;
-    Ok(out)
-}
-
-fn uncompress_bytes(input: &[u8], config: &StoreConfig) -> Result<Vec<u8>, Error> {
-    let mut out = Vec::with_capacity(config.estimate_decompressed_size(input.len()));
-    let mut decoder = Decoder::new(input).map_err(Error::Compression)?;
-    decoder.read_to_end(&mut out).map_err(Error::Compression)?;
-    Ok(out)
-}
-
 impl ValidatorsDiff {
     pub fn compute(
         xs: &[Validator],
@@ -534,12 +520,16 @@ impl ValidatorsDiff {
             .collect::<Vec<u8>>();
 
         Ok(Self {
-            bytes: compress_bytes(&uncompressed_bytes, config)?,
+            bytes: config
+                .compress_bytes(&uncompressed_bytes)
+                .map_err(Error::Compression)?,
         })
     }
 
     pub fn apply(&self, xs: &mut Vec<Validator>, config: &StoreConfig) -> Result<(), Error> {
-        let validator_diff_bytes = uncompress_bytes(&self.bytes, config)?;
+        let validator_diff_bytes = config
+            .decompress_bytes(&self.bytes)
+            .map_err(Error::Compression)?;
 
         for diff_bytes in
             validator_diff_bytes.chunks(<ValidatorDiffEntry as Decode>::ssz_fixed_len())