Several changes

* Fix state cache pruning of finalized state from block map
* Update to latest `milhouse`
* Check beacon state diffs in EF tests

beacon_node/store/src/chunked_vector.rs

@@ -543,7 +543,7 @@ pub fn load_variable_list_from_db<F: VariableLengthField<E>, E: EthSpec, S: KeyV
 
     let chunks: Vec<Chunk<F::Value>> = range_query(store, F::column(), start_cindex, end_cindex)?;
 
-    let mut result = VList::empty()?;
+    let mut result = VList::empty();
 
     for (chunk_index, chunk) in chunks.into_iter().enumerate() {
         for (i, value) in chunk.values.into_iter().enumerate() {

beacon_node/store/src/errors.rs

@@ -47,8 +47,7 @@ pub enum Error {
     BlockReplayError(BlockReplayError),
     #[cfg(feature = "milhouse")]
     MilhouseError(milhouse::Error),
-    Bincode(Box<bincode::ErrorKind>),
-    FlateCompression(std::io::Error),
+    Compression(std::io::Error),
 }
 
 pub trait HandleUnavailable<T> {
@@ -114,12 +113,6 @@ impl From<BlockReplayError> for Error {
     }
 }
 
-impl From<Box<bincode::ErrorKind>> for Error {
-    fn from(e: Box<bincode::ErrorKind>) -> Self {
-        Self::Bincode(e)
-    }
-}
-
 #[derive(Debug)]
 pub struct DBError {
     pub message: String,

beacon_node/store/src/state_cache.rs

@@ -95,8 +95,7 @@ impl<E: EthSpec> StateCache<E> {
                 finalized_state.state_root == state_root
             })
         {
-            // FIXME(sproul): this should technically be true
-            return Ok(false);
+            return Ok(true);
         }
         if self.states.peek(&state_root).is_some() {
             return Ok(true);
@@ -167,7 +166,7 @@ impl BlockMap {
 
         self.blocks.retain(|_, slot_map| {
             slot_map.slots.retain(|slot, state_root| {
-                let keep = *slot > finalized_slot;
+                let keep = *slot >= finalized_slot;
                 if !keep {
                     pruned_states.insert(*state_root);
                 }

beacon_node/store/src/state_diff.rs

@@ -1,8 +1,18 @@
 use crate::{metrics, DBColumn, Error, StoreItem};
-use flate2::bufread::{ZlibDecoder, ZlibEncoder};
 use ssz::{Decode, Encode};
-use std::io::Read;
+use std::io::{Read, Write};
 use types::{beacon_state::BeaconStateDiff, EthSpec};
+use zstd::{Decoder, Encoder};
+
+const EST_COMPRESSION_FACTOR: usize = 2;
+
+fn estimate_compressed_size(len: usize, compression_level: i32) -> usize {
+    if compression_level == 0 {
+        len
+    } else {
+        len / EST_COMPRESSION_FACTOR
+    }
+}
 
 impl<E: EthSpec> StoreItem for BeaconStateDiff<E> {
     fn db_column() -> DBColumn {
@@ -14,13 +24,13 @@ impl<E: EthSpec> StoreItem for BeaconStateDiff<E> {
         let value = self.as_ssz_bytes();
         drop(encode_timer);
 
-        // FIXME(sproul): try vec with capacity
         let compression_timer = metrics::start_timer(&metrics::BEACON_STATE_DIFF_COMPRESSION_TIME);
-        let mut encoder = ZlibEncoder::new(&value[..], flate2::Compression::fast());
-        let mut compressed_value = vec![];
-        encoder
-            .read_to_end(&mut compressed_value)
-            .map_err(Error::FlateCompression)?;
+
+        let level = 1;
+        let mut compressed_value = Vec::with_capacity(estimate_compressed_size(value.len(), level));
+        let mut encoder = Encoder::new(&mut compressed_value, level).map_err(Error::Compression)?;
+        encoder.write_all(&value).map_err(Error::Compression)?;
+        encoder.finish().map_err(Error::Compression)?;
         drop(compression_timer);
 
         let compression_ratio = value.len() as f64 / compressed_value.len() as f64;
@@ -39,11 +49,11 @@ impl<E: EthSpec> StoreItem for BeaconStateDiff<E> {
     }
 
     fn from_store_bytes(bytes: &[u8]) -> Result<Self, Error> {
-        let mut ssz_bytes = vec![];
-        let mut decoder = ZlibDecoder::new(bytes);
+        let mut ssz_bytes = Vec::with_capacity(EST_COMPRESSION_FACTOR * bytes.len());
+        let mut decoder = Decoder::new(bytes).map_err(Error::Compression)?;
         decoder
             .read_to_end(&mut ssz_bytes)
-            .map_err(Error::FlateCompression)?;
+            .map_err(Error::Compression)?;
         Ok(Self::from_ssz_bytes(&ssz_bytes)?)
     }
 }