cache cleanup

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -6216,9 +6216,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             return Err(BlobError::TransactionCommitmentMismatch);
         }
 
-        self.blob_cache
+        // Validate that the kzg proof is valid against the commitments and blobs
-
-        // Validatate that the kzg proof is valid against the commitments and blobs
         let kzg = self
             .kzg
             .as_ref()

beacon_node/beacon_chain/src/gossip_blob_cache.rs

@@ -1,45 +1,42 @@
 use std::collections::{BTreeMap, HashMap, HashSet};
+use std::future::Future;
 use std::sync::Arc;
 use parking_lot::{Mutex, RwLock};
-use kzg::KzgCommitment;
+use eth2::reqwest::header::Entry;
+use kzg::{Kzg, KzgCommitment};
 use ssz_types::VariableList;
 use types::blob_sidecar::{BlobIdentifier, BlobSidecar};
 use types::{EthSpec, Hash256};
 use crate::blob_verification::{AvailabilityPendingBlock, verify_data_availability};
-use crate::block_verification::IntoExecutionPendingBlock;
+use crate::block_verification::{ExecutedBlock, IntoExecutionPendingBlock};
+use crate::{BeaconChainError, BlockError};
+use crate::kzg_utils::validate_blob;
 
+pub enum BlobCacheError {
+    DuplicateBlob(Hash256)
+}
 /// This cache contains
 ///  - blobs that have been gossip verified
 ///  - commitments for blocks that have been gossip verified, but the commitments themselves
 ///    have not been verified against blobs
 ///  - blocks that have been fully verified and only require a data availability check
 pub struct GossipBlobCache<T: EthSpec> {
-    blob_cache: Mutex<GossipBlobCacheInner<T>>
+    rpc_blob_cache: RwLock<HashMap<BlobIdentifier, Arc<BlobSidecar<T>>>>,
+    gossip_blob_cache: Mutex<HashMap<Hash256, GossipBlobCacheInner<T>>>,
+    kzg: Kzg,
 }
 
 struct GossipBlobCacheInner<T: EthSpec> {
-    // used when all blobs are not yet present and when the block is not yet present
+    verified_blobs: Vec<Arc<BlobSidecar<T>>>,
-
+    executed_block: Option<ExecutedBlock<T>>,
-    //TODO(sean) do we want two versions of this cache, one meant to serve RPC?
-    unverified_blobs: BTreeMap<BlobIdentifier, Arc<BlobSidecar<T>>>,
-    // used when the block was fully processed before we received all blobs
-    availability_pending_blocks: HashMap<Hash256, AvailabilityPendingBlock<T>>,
-    // used to cache kzg commitments from gossip verified blocks in case we receive all blobs during block processing
-    unverified_commitments: HashMap<Hash256, VariableList<KzgCommitment, T::MaxBlobsPerBlock>>,
-    // used when block + blob kzg verification completes prior before block processing
-    verified_commitments: HashSet<Hash256>,
 }
 
 impl <T: EthSpec> GossipBlobCache<T> {
-    pub fn new() -> Self {
+    pub fn new(kzg: Kzg) -> Self {
-
         Self {
-            blob_cache: Mutex::new(GossipBlobCacheInner {
+            rpc_blob_cache: RwLock::new(HashMap::new()),
-                unverified_blobs:  BTreeMap::new(),
+            gossip_blob_cache: Mutex::new(HashMap::new()),
-                availability_pending_blocks:  HashMap::new(),
+            kzg,
-                unverified_commitments: HashMap::new(),
-                verified_commitments: HashSet::new(),
-            })
         }
 
     }
@@ -49,59 +46,52 @@ impl <T: EthSpec> GossipBlobCache<T> {
     /// cached, verify the block and import it.
     ///
     /// This should only accept gossip verified blobs, so we should not have to worry about dupes.
-    pub fn put_blob(&self, blob: Arc<BlobSidecar<T>>) {
+    pub fn put_blob(&self, blob: Arc<BlobSidecar<T>>) -> Result<(),BlobCacheError> {
-        let blob_id = blob.id();
-        let blob_cache = self.blob_cache.lock();
 
-        if let Some(dup) = blob_cache.unverified_blobs.insert(blob_id, blob) {
+        // TODO(remove clones)
-            // return error relating to gossip validation failure
+        let verified =  validate_blob(&self.kzg, blob.blob.clone(), blob.kzg_commitment.clone(), blob.kzg_proof)?;
+
+        if verified {
+            let mut blob_cache = self.gossip_blob_cache.lock();
+
+            // Gossip cache.
+            blob_cache.entry(blob.block_root)
+                .and_modify(|mut inner| {
+                    // All blobs reaching this cache should be gossip verified and gossip verification
+                    // should filter duplicates, as well as validate indices.
+                    inner.verified_blobs.insert(blob.index as usize, blob.clone());
+
+                    if let Some (executed_block) = inner.executed_block.as_ref() {
+                        // trigger reprocessing ?
+                    }
+                })
+                .or_insert(GossipBlobCacheInner {
+               verified_blobs: vec![blob.clone()],
+                    executed_block: None
+            });
+
+            drop(blob_cache);
+
+            // RPC cache.
+            self.rpc_blob_cache.write().insert(blob.id(), blob.clone());
         }
 
-        if let Some(availability_pending_block) = blob_cache.availability_pending_blocks.get(&blob.block_root) {
+        Ok(())
-            let num_blobs = availability_pending_block.kzg_commitments().len();
-            let mut blobs : Vec<BlobIdentifier, BlobSidecar<T>> = blob_cache.unverified_blobs.range(BlobIdentifier::new(blob.block_root, 0)
-                ..BlobIdentifier::new(blob.block_root, num_blobs as u64)).collect();
-
-            if blobs.len() == num_blobs {
-               // verify
-                // import
-            }
-        } else if let Some(commitments) = blob_cache.unverified_commitments.get(&blob.block_root) {
-            let num_blobs = commitments.len();
-            let mut blobs : Vec<BlobIdentifier, BlobSidecar<T>> = blob_cache.unverified_blobs.range(BlobIdentifier::new(blob.block_root, 0)
-                ..BlobIdentifier::new(blob.block_root, num_blobs as u64)).collect();
-
-            if blobs.len() == num_blobs {
-                // verify
-                // cache
-            }
-        }
     }
 
-
+    pub fn put_block(&self, block: ExecutedBlock<T>) -> () {
-    pub fn put_commitments(&self, block_root: Hash256, kzg_commitments: VariableList<KzgCommitment, T::MaxBlobsPerBlock>) {
+        let mut guard =         self.gossip_blob_cache.lock();
-        let blob_cache = self.blob_cache.lock();
+        guard.entry(block.block_root).and_modify(|cache|{
-        if let Some(dup) = blob_cache.unverified_commitments.insert(block_root, kzg_commitments) {
+            if cache.verified_blobs == block.block.message_eip4844().blob_kzg_commitments() {
-            // return error relating to gossip validation failure
+                // send to reprocessing queue ?
-        }
+            } else if let Some(dup) = cache.executed_block.insert(block) {
-
+                    // return error
-        let num_blobs = commitments.len();
-        let mut blobs : Vec<BlobIdentifier, BlobSidecar<T>> = blob_cache.unverified_blobs.range(BlobIdentifier::new(blob.block_root, 0)
-            ..BlobIdentifier::new(blob.block_root, num_blobs as u64)).collect();
-
-        if blobs.len() == num_blobs {
-            // verify
-            // cache
-        }
-    }
-
-    pub fn check_availability_and_import(&self, block_root: Hash256, block: AvailabilityPendingBlock<T>) -> bool {
-        let blob_cache = self.blob_cache.lock();
-       if blob_cache.verified_commitments.contains(&block_root) {
-          true
                 } else {
-           // cache the block
+                    // log that we cached it
-            false
+                }
-       }
+        }).or_insert(GossipBlobCacheInner {
+            verified_blobs: vec![],
+            executed_block: Some(block)
+        });
     }
 }

beacon_node/execution_layer/src/lib.rs

@@ -44,12 +44,11 @@ use tree_hash::TreeHash;
 use types::consts::eip4844::BLOB_TX_TYPE;
 use types::transaction::{AccessTuple, BlobTransaction, EcdsaSignature, SignedBlobTransaction};
 use types::Withdrawals;
+use types::{AbstractExecPayload, BeaconStateError, ExecPayload, VersionedHash};
 use types::{
-    blobs_sidecar::{Blobs, KzgCommitments},
+    BlindedPayload, Blobs, BlockType, ChainSpec, Epoch, ExecutionBlockHash, ExecutionPayload,
-    BlindedPayload, BlockType, ChainSpec, Epoch, ExecutionBlockHash, ExecutionPayload,
     ExecutionPayloadCapella, ExecutionPayloadEip4844, ExecutionPayloadMerge, ForkName,
 };
-use types::{AbstractExecPayload, BeaconStateError, ExecPayload, VersionedHash};
 use types::{
     ProposerPreparationData, PublicKeyBytes, Signature, SignedBeaconBlock, Slot, Transaction,
     Uint256,

consensus/types/src/beacon_block_body.rs

@@ -1,5 +1,5 @@
+use crate::test_utils::TestRandom;
 use crate::*;
-use crate::{blobs_sidecar::KzgCommitments, test_utils::TestRandom};
 use derivative::Derivative;
 use serde_derive::{Deserialize, Serialize};
 use ssz_derive::{Decode, Encode};

consensus/types/src/blob_sidecar.rs

@@ -1,5 +1,6 @@
 use crate::test_utils::TestRandom;
 use crate::{Blob, EthSpec, Hash256, SignedRoot, Slot};
+use bls::Signature;
 use derivative::Derivative;
 use kzg::{KzgCommitment, KzgProof};
 use serde_derive::{Deserialize, Serialize};
@@ -34,7 +35,6 @@ pub struct BlobIdentifier {
 #[derivative(PartialEq, Hash(bound = "T: EthSpec"))]
 pub struct BlobSidecar<T: EthSpec> {
     pub block_root: Hash256,
-    // TODO: fix the type, should fit in u8 as well
     #[serde(with = "eth2_serde_utils::quoted_u64")]
     pub index: u64,
     pub slot: Slot,
@@ -52,6 +52,13 @@ pub type BlobSidecarList<T> = VariableList<BlobSidecar<T>, <T as EthSpec>::MaxBl
 impl<T: EthSpec> SignedRoot for BlobSidecar<T> {}
 
 impl<T: EthSpec> BlobSidecar<T> {
+    pub fn id(&self) -> BlobIdentifier {
+        BlobIdentifier {
+            block_root: self.block_root,
+            index: self.index,
+        }
+    }
+
     pub fn empty() -> Self {
         Self::default()
     }