Merge branch 'unstable' into das-unstable-merge-0415

# Conflicts:
#	Cargo.lock
#	beacon_node/beacon_chain/src/data_availability_checker.rs
#	beacon_node/beacon_chain/src/data_availability_checker/availability_view.rs
#	beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs
#	beacon_node/beacon_chain/src/data_availability_checker/processing_cache.rs
#	beacon_node/lighthouse_network/src/rpc/methods.rs
#	beacon_node/network/src/network_beacon_processor/mod.rs
#	beacon_node/network/src/sync/block_lookups/tests.rs
#	crypto/kzg/Cargo.toml

Cargo.toml

@@ -1,7 +1,6 @@
 [workspace]
 members = [
     "account_manager",
-
     "beacon_node",
     "beacon_node/beacon_chain",
     "beacon_node/beacon_processor",
@@ -16,9 +15,7 @@ members = [
     "beacon_node/network",
     "beacon_node/store",
     "beacon_node/timer",
-
     "boot_node",
-
     "common/account_utils",
     "common/clap_utils",
     "common/compare_fields",
@@ -48,31 +45,24 @@ members = [
     "common/validator_dir",
     "common/warp_utils",
     "common/monitoring_api",
-
     "database_manager",
-
     "consensus/cached_tree_hash",
     "consensus/int_to_bytes",
     "consensus/fork_choice",
     "consensus/proto_array",
     "consensus/safe_arith",
     "consensus/state_processing",
     "consensus/swap_or_not_shuffle",
-
     "crypto/bls",
     "crypto/kzg",
     "crypto/eth2_key_derivation",
     "crypto/eth2_keystore",
     "crypto/eth2_wallet",
-
     "lcli",
-
     "lighthouse",
     "lighthouse/environment",
-
     "slasher",
     "slasher/service",
-
     "testing/ef_tests",
     "testing/eth1_test_rig",
     "testing/execution_engine_integration",
@@ -81,12 +71,9 @@ members = [
     "testing/test-test_logger",
     "testing/state_transition_vectors",
     "testing/web3signer_tests",
-
     "validator_client",
     "validator_client/slashing_protection",
-
     "validator_manager",
-
     "watch",
 ]
 resolver = "2"
@@ -102,6 +89,7 @@ bincode = "1"
 bitvec = "1"
 byteorder = "1"
 bytes = "1"
+c-kzg = { git = "https://github.com/ethereum/c-kzg-4844", branch = "das" }
 clap = "2"
 compare_fields_derive = { path = "common/compare_fields_derive" }
 criterion = "0.3"
@@ -181,7 +169,7 @@ zip = "0.6"
 account_utils = { path = "common/account_utils" }
 beacon_chain = { path = "beacon_node/beacon_chain" }
 beacon_node = { path = "beacon_node" }
-beacon_processor =  { path = "beacon_node/beacon_processor" }
+beacon_processor = { path = "beacon_node/beacon_processor" }
 bls = { path = "crypto/bls" }
 cached_tree_hash = { path = "consensus/cached_tree_hash" }
 clap_utils = { path = "common/clap_utils" }
@@ -218,7 +206,7 @@ network = { path = "beacon_node/network" }
 operation_pool = { path = "beacon_node/operation_pool" }
 pretty_reqwest_error = { path = "common/pretty_reqwest_error" }
 proto_array = { path = "consensus/proto_array" }
-safe_arith = {path = "consensus/safe_arith"}
+safe_arith = { path = "consensus/safe_arith" }
 sensitive_url = { path = "common/sensitive_url" }
 slasher = { path = "slasher" }
 slashing_protection = { path = "validator_client/slashing_protection" }

beacon_node/beacon_chain/src/beacon_block_streamer.rs

@@ -1,4 +1,4 @@
-use crate::{BeaconChain, BeaconChainError, BeaconChainTypes};
+use crate::{metrics, BeaconChain, BeaconChainError, BeaconChainTypes};
 use execution_layer::{ExecutionLayer, ExecutionPayloadBodyV1};
 use slog::{crit, debug, Logger};
 use std::collections::HashMap;
@@ -412,8 +412,13 @@ impl<T: BeaconChainTypes> BeaconBlockStreamer<T> {
     fn check_caches(&self, root: Hash256) -> Option<Arc<SignedBeaconBlock<T::EthSpec>>> {
         if self.check_caches == CheckCaches::Yes {
             self.beacon_chain
-                .data_availability_checker
-                .get_block(&root)
+                .reqresp_pre_import_cache
+                .read()
+                .get(&root)
+                .map(|block| {
+                    metrics::inc_counter(&metrics::BEACON_REQRESP_PRE_IMPORT_CACHE_HITS);
+                    block.clone()
+                })
                 .or(self.beacon_chain.early_attester_cache.get_block(root))
         } else {
             None

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -361,6 +361,9 @@ pub type BeaconStore<T> = Arc<
     >,
 >;
 
+/// Cache gossip verified blocks to serve over ReqResp before they are imported
+type ReqRespPreImportCache<E> = HashMap<Hash256, Arc<SignedBeaconBlock<E>>>;
+
 /// Represents the "Beacon Chain" component of Ethereum 2.0. Allows import of blocks and block
 /// operations and chooses a canonical head.
 pub struct BeaconChain<T: BeaconChainTypes> {
@@ -463,6 +466,8 @@ pub struct BeaconChain<T: BeaconChainTypes> {
     pub(crate) attester_cache: Arc<AttesterCache>,
     /// A cache used when producing attestations whilst the head block is still being imported.
     pub early_attester_cache: EarlyAttesterCache<T::EthSpec>,
+    /// Cache gossip verified blocks to serve over ReqResp before they are imported
+    pub reqresp_pre_import_cache: Arc<RwLock<ReqRespPreImportCache<T::EthSpec>>>,
     /// A cache used to keep track of various block timings.
     pub block_times_cache: Arc<RwLock<BlockTimesCache>>,
     /// A cache used to track pre-finalization block roots for quick rejection.
@@ -2936,8 +2941,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             }
         }
 
-        self.data_availability_checker
-            .notify_gossip_blob(block_root, &blob);
         let r = self.check_gossip_blob_availability_and_import(blob).await;
         self.remove_notified(&block_root, r)
     }
@@ -2994,8 +2997,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             }
         }
 
-        self.data_availability_checker
-            .notify_rpc_blobs(block_root, &blobs);
         let r = self
             .check_rpc_blob_availability_and_import(slot, block_root, blobs)
             .await;
@@ -3012,7 +3013,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         let has_missing_components =
             matches!(r, Ok(AvailabilityProcessingStatus::MissingComponents(_, _)));
         if !has_missing_components {
-            self.data_availability_checker.remove_notified(block_root);
+            self.reqresp_pre_import_cache.write().remove(block_root);
         }
         r
     }
@@ -3025,8 +3026,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         unverified_block: B,
         notify_execution_layer: NotifyExecutionLayer,
     ) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
-        self.data_availability_checker
-            .notify_block(block_root, unverified_block.block_cloned());
+        self.reqresp_pre_import_cache
+            .write()
+            .insert(block_root, unverified_block.block_cloned());
+
         let r = self
             .process_block(block_root, unverified_block, notify_execution_layer, || {
                 Ok(())

beacon_node/beacon_chain/src/blob_verification.rs

@@ -284,6 +284,9 @@ impl<E: EthSpec> KzgVerifiedBlob<E> {
     pub fn as_blob(&self) -> &BlobSidecar<E> {
         &self.blob
     }
+    pub fn get_commitment(&self) -> &KzgCommitment {
+        &self.blob.kzg_commitment
+    }
     /// This is cheap as we're calling clone on an Arc
     pub fn clone_blob(&self) -> Arc<BlobSidecar<E>> {
         self.blob.clone()

beacon_node/beacon_chain/src/builder.rs

@@ -25,7 +25,7 @@ use eth1::Config as Eth1Config;
 use execution_layer::ExecutionLayer;
 use fork_choice::{ForkChoice, ResetPayloadStatuses};
 use futures::channel::mpsc::Sender;
-use kzg::{Kzg, TrustedSetup};
+use kzg::Kzg;
 use operation_pool::{OperationPool, PersistedOperationPool};
 use parking_lot::{Mutex, RwLock};
 use proto_array::{DisallowedReOrgOffsets, ReOrgThreshold};
@@ -101,7 +101,7 @@ pub struct BeaconChainBuilder<T: BeaconChainTypes> {
     // Pending I/O batch that is constructed during building and should be executed atomically
     // alongside `PersistedBeaconChain` storage when `BeaconChainBuilder::build` is called.
     pending_io_batch: Vec<KeyValueStoreOp>,
-    trusted_setup: Option<TrustedSetup>,
+    kzg: Option<Arc<Kzg>>,
     task_executor: Option<TaskExecutor>,
     validator_monitor_config: Option<ValidatorMonitorConfig>,
 }
@@ -142,7 +142,7 @@ where
             graffiti: Graffiti::default(),
             slasher: None,
             pending_io_batch: vec![],
-            trusted_setup: None,
+            kzg: None,
             task_executor: None,
             validator_monitor_config: None,
         }
@@ -679,8 +679,8 @@ where
         self
     }
 
-    pub fn trusted_setup(mut self, trusted_setup: TrustedSetup) -> Self {
-        self.trusted_setup = Some(trusted_setup);
+    pub fn kzg(mut self, kzg: Option<Arc<Kzg>>) -> Self {
+        self.kzg = kzg;
         self
     }
 
@@ -728,15 +728,6 @@ where
             slot_clock.now().ok_or("Unable to read slot")?
         };
 
-        let kzg = if let Some(trusted_setup) = self.trusted_setup {
-            let kzg = Kzg::new_from_trusted_setup(trusted_setup)
-                .map_err(|e| format!("Failed to load trusted setup: {:?}", e))?;
-            let kzg_arc = Arc::new(kzg);
-            Some(kzg_arc)
-        } else {
-            None
-        };
-
         let initial_head_block_root = fork_choice
             .get_head(current_slot, &self.spec)
             .map_err(|e| format!("Unable to get fork choice head: {:?}", e))?;
@@ -965,6 +956,7 @@ where
             validator_pubkey_cache: TimeoutRwLock::new(validator_pubkey_cache),
             attester_cache: <_>::default(),
             early_attester_cache: <_>::default(),
+            reqresp_pre_import_cache: <_>::default(),
             light_client_server_cache: LightClientServerCache::new(),
             light_client_server_tx: self.light_client_server_tx,
             shutdown_sender: self
@@ -976,10 +968,10 @@ where
             validator_monitor: RwLock::new(validator_monitor),
             genesis_backfill_slot,
             data_availability_checker: Arc::new(
-                DataAvailabilityChecker::new(slot_clock, kzg.clone(), store, &log, self.spec)
+                DataAvailabilityChecker::new(slot_clock, self.kzg.clone(), store, &log, self.spec)
                     .map_err(|e| format!("Error initializing DataAvailabiltyChecker: {:?}", e))?,
             ),
-            kzg,
+            kzg: self.kzg.clone(),
             block_production_state: Arc::new(Mutex::new(None)),
         };