Skip malformed live chain keys

crates/storage/src/store.rs

@@ -468,11 +468,23 @@ fn encode_live_chain_key(slot: u64, root: &H256) -> Vec<u8> {
     result
 }
 
+#[derive(Debug, Error)]
+enum DecodeLiveChainKeyError {
+    #[error("invalid LiveChain key length: expected 40 bytes, got {actual}")]
+    InvalidLength { actual: usize },
+}
+
 /// Decode a LiveChain key from bytes.
-fn decode_live_chain_key(bytes: &[u8]) -> (u64, H256) {
+fn decode_live_chain_key(bytes: &[u8]) -> Result<(u64, H256), DecodeLiveChainKeyError> {
+    if bytes.len() != 40 {
+        return Err(DecodeLiveChainKeyError::InvalidLength {
+            actual: bytes.len(),
+        });
+    }
+
     let slot = u64::from_be_bytes(bytes[..8].try_into().expect("valid slot bytes"));
     let root = H256::from_slice(&bytes[8..]);
-    (slot, root)
+    Ok((slot, root))
 }
 
 /// Fork choice store backed by a pluggable storage backend.
@@ -819,10 +831,15 @@ impl Store {
         view.prefix_iterator(Table::LiveChain, &[])
             .expect("iterator")
             .filter_map(|res| res.ok())
-            .map(|(k, v)| {
-                let (slot, root) = decode_live_chain_key(&k);
-                let parent_root = H256::from_ssz_bytes(&v).expect("valid parent_root");
-                (root, (slot, parent_root))
+            .filter_map(|(k, v)| match decode_live_chain_key(&k) {
+                Ok((slot, root)) => {
+                    let parent_root = H256::from_ssz_bytes(&v).expect("valid parent_root");
+                    Some((root, (slot, parent_root)))
+                }
+                Err(error) => {
+                    warn!(%error, "Skipping malformed LiveChain key");
+                    None
+                }
             })
             .collect()
     }
@@ -833,7 +850,13 @@ impl Store {
         view.prefix_iterator(Table::LiveChain, &[])
             .expect("iterator")
             .filter_map(Result::ok)
-            .map(|(key, _)| decode_live_chain_key(&key).0)
+            .filter_map(|(key, _)| match decode_live_chain_key(&key) {
+                Ok((slot, _)) => Some(slot),
+                Err(error) => {
+                    warn!(%error, "Skipping malformed LiveChain key");
+                    None
+                }
+            })
             .max()
     }
 
@@ -845,9 +868,12 @@ impl Store {
         view.prefix_iterator(Table::LiveChain, &[])
             .expect("iterator")
             .filter_map(|res| res.ok())
-            .map(|(k, _)| {
-                let (_, root) = decode_live_chain_key(&k);
-                root
+            .filter_map(|(k, _)| match decode_live_chain_key(&k) {
+                Ok((_, root)) => Some(root),
+                Err(error) => {
+                    warn!(%error, "Skipping malformed LiveChain key");
+                    None
+                }
             })
             .collect()
     }
@@ -867,9 +893,12 @@ impl Store {
             .prefix_iterator(Table::LiveChain, &[])
             .expect("iterator")
             .filter_map(|res| res.ok())
-            .take_while(|(k, _)| {
-                let (slot, _) = decode_live_chain_key(k);
-                slot < finalized_slot
+            .take_while(|(k, _)| match decode_live_chain_key(k) {
+                Ok((slot, _)) => slot < finalized_slot,
+                Err(error) => {
+                    warn!(%error, "Pruning malformed LiveChain key");
+                    true
+                }
             })
             .map(|(k, _)| k.to_vec())
             .collect();
@@ -1420,6 +1449,33 @@ mod tests {
     use super::*;
     use crate::backend::InMemoryBackend;
 
+    #[test]
+    fn live_chain_queries_skip_malformed_keys() {
+        let backend = Arc::new(InMemoryBackend::new());
+        let store = Store::test_store_with_backend(backend.clone());
+        let root = root(1);
+        let parent = H256::ZERO;
+
+        let mut batch = backend.begin_write().expect("write batch");
+        batch
+            .put_batch(
+                Table::LiveChain,
+                vec![
+                    (vec![1, 2, 3], parent.to_ssz()),
+                    (encode_live_chain_key(42, &root), parent.to_ssz()),
+                ],
+            )
+            .expect("put live chain");
+        batch.commit().expect("commit");
+
+        assert_eq!(store.max_live_chain_slot(), Some(42));
+        assert_eq!(store.get_block_roots(), HashSet::from([root]));
+
+        let live_chain = store.get_live_chain();
+        assert_eq!(live_chain.len(), 1);
+        assert_eq!(live_chain.get(&root), Some(&(42, parent)));
+    }
+
     /// Insert a block header (and dummy body + signature) for a given root and slot.
     fn insert_header(backend: &dyn StorageBackend, root: H256, slot: u64) {
         let header = BlockHeader {