Single blob lookups

.github/workflows/test-suite.yml

@@ -119,6 +119,20 @@ jobs:
         repo-token: ${{ secrets.GITHUB_TOKEN }}
     - name: Run operation_pool tests for all known forks
       run: make test-op-pool
+  network-minimal-tests:
+    name: network-minimal-tests
+    runs-on: ubuntu-latest
+    needs: cargo-fmt
+    steps:
+    - uses: actions/checkout@v3
+    - name: Get latest version of stable Rust
+      run: rustup update stable
+    - name: Install Protoc
+      uses: arduino/setup-protoc@e52d9eb8f7b63115df1ac544a1376fdbf5a39612
+      with:
+        repo-token: ${{ secrets.GITHUB_TOKEN }}
+    - name: Run network tests for all known forks using the minimal spec
+      run: make test-network-minimal
   slasher-tests:
     name: slasher-tests
     runs-on: ubuntu-latest

Makefile

@@ -106,7 +106,7 @@ build-release-tarballs:
 # Runs the full workspace tests in **release**, without downloading any additional
 # test vectors.
 test-release:
-	cargo test --workspace --release --exclude ef_tests --exclude beacon_chain --exclude slasher
+	cargo test --workspace --release --exclude ef_tests --exclude beacon_chain --exclude slasher 
 
 # Runs the full workspace tests in **debug**, without downloading any additional test
 # vectors.
@@ -143,6 +143,13 @@ test-op-pool-%:
 		--features 'beacon_chain/fork_from_env'\
 		-p operation_pool
 
+test-network-minimal: $(patsubst %,test-network-minimal-%,$(FORKS))
+
+test-network-minimal-%:
+	env FORK_NAME=$* cargo test --release \
+		--features 'fork_from_env,spec-minimal'\
+		-p network
+
 # Run the tests in the `slasher` crate for all supported database backends.
 test-slasher:
 	cargo test --release -p slasher --features lmdb

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -117,7 +117,7 @@ use tokio_stream::Stream;
 use tree_hash::TreeHash;
 use types::beacon_block_body::KzgCommitments;
 use types::beacon_state::CloneConfig;
-use types::blob_sidecar::{BlobIdentifier, BlobSidecarList, Blobs};
+use types::blob_sidecar::{BlobSidecarList, Blobs};
 use types::consts::deneb::MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS;
 use types::*;
 
@@ -185,12 +185,10 @@ pub enum WhenSlotSkipped {
 
 #[derive(Debug, PartialEq)]
 pub enum AvailabilityProcessingStatus {
-    PendingBlobs(Vec<BlobIdentifier>),
-    PendingBlock(Hash256),
+    MissingComponents(Slot, Hash256),
     Imported(Hash256),
 }
 
-//TODO(sean) using this in tests for now
 impl TryInto<SignedBeaconBlockHash> for AvailabilityProcessingStatus {
     type Error = ();
 
@@ -468,7 +466,7 @@ pub struct BeaconChain<T: BeaconChainTypes> {
     /// The slot at which blocks are downloaded back to.
     pub genesis_backfill_slot: Slot,
     pub proposal_blob_cache: BlobCache<T::EthSpec>,
-    pub data_availability_checker: DataAvailabilityChecker<T>,
+    pub data_availability_checker: Arc<DataAvailabilityChecker<T>>,
     pub kzg: Option<Arc<Kzg>>,
 }
 
@@ -1985,8 +1983,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         self: &Arc<Self>,
         blob_sidecar: SignedBlobSidecar<T::EthSpec>,
         subnet_id: u64,
-    ) -> Result<GossipVerifiedBlob<T::EthSpec>, BlobError> // TODO(pawan): make a GossipVerifedBlob type
-    {
+    ) -> Result<GossipVerifiedBlob<T::EthSpec>, BlobError<T::EthSpec>> {
         blob_verification::validate_blob_sidecar_for_gossip(blob_sidecar, subnet_id, self)
     }
 
@@ -2674,7 +2671,24 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                     )
                     .await
                 {
-                    Ok(_) => imported_blocks += 1,
+                    Ok(status) => {
+                        match status {
+                            AvailabilityProcessingStatus::Imported(_) => {
+                                // The block was imported successfully.
+                                imported_blocks += 1;
+                            }
+                            AvailabilityProcessingStatus::MissingComponents(slot, block_root) => {
+                                warn!(self.log, "Blobs missing in response to range request";
+                                    "block_root" => ?block_root, "slot" => slot);
+                                return ChainSegmentResult::Failed {
+                                    imported_blocks,
+                                    error: BlockError::AvailabilityCheck(
+                                        AvailabilityCheckError::MissingBlobs,
+                                    ),
+                                };
+                            }
+                        }
+                    }
                     Err(error) => {
                         return ChainSegmentResult::Failed {
                             imported_blocks,
@@ -2748,6 +2762,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         count_unrealized: CountUnrealized,
     ) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
         self.check_availability_and_maybe_import(
+            blob.slot(),
             |chain| chain.data_availability_checker.put_gossip_blob(blob),
             count_unrealized,
         )
@@ -2804,6 +2819,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             }
             ExecutedBlock::AvailabilityPending(block) => {
                 self.check_availability_and_maybe_import(
+                    block.block.slot(),
                     |chain| {
                         chain
                             .data_availability_checker
@@ -2907,6 +2923,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     /// (i.e., this function is not atomic).
     pub async fn check_availability_and_maybe_import(
         self: &Arc<Self>,
+        slot: Slot,
         cache_fn: impl FnOnce(Arc<Self>) -> Result<Availability<T::EthSpec>, AvailabilityCheckError>,
         count_unrealized: CountUnrealized,
     ) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
@@ -2915,12 +2932,9 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             Availability::Available(block) => {
                 self.import_available_block(block, count_unrealized).await
             }
-            Availability::PendingBlock(block_root) => {
-                Ok(AvailabilityProcessingStatus::PendingBlock(block_root))
-            }
-            Availability::PendingBlobs(blob_ids) => {
-                Ok(AvailabilityProcessingStatus::PendingBlobs(blob_ids))
-            }
+            Availability::MissingComponents(block_root) => Ok(
+                AvailabilityProcessingStatus::MissingComponents(slot, block_root),
+            ),
         }
     }
 

beacon_node/beacon_chain/src/blob_verification.rs

@@ -16,15 +16,17 @@ use eth2::types::BlockContentsTuple;
 use kzg::Kzg;
 use slog::{debug, warn};
 use ssz_derive::{Decode, Encode};
+use ssz_types::FixedVector;
 use std::borrow::Cow;
+use types::blob_sidecar::{BlobIdentifier, FixedBlobSidecarList};
 use types::{
-    BeaconBlockRef, BeaconState, BeaconStateError, BlobSidecar, BlobSidecarList, ChainSpec,
-    CloneConfig, Epoch, EthSpec, FullPayload, Hash256, KzgCommitment, RelativeEpoch,
-    SignedBeaconBlock, SignedBeaconBlockHeader, SignedBlobSidecar, Slot,
+    BeaconBlockRef, BeaconState, BeaconStateError, BlobSidecar, ChainSpec, CloneConfig, Epoch,
+    EthSpec, FullPayload, Hash256, KzgCommitment, RelativeEpoch, SignedBeaconBlock,
+    SignedBeaconBlockHeader, SignedBlobSidecar, Slot,
 };
 
 #[derive(Debug)]
-pub enum BlobError {
+pub enum BlobError<T: EthSpec> {
     /// The blob sidecar is from a slot that is later than the current slot (with respect to the
     /// gossip clock disparity).
     ///
@@ -96,10 +98,7 @@ pub enum BlobError {
     /// ## Peer scoring
     ///
     /// We cannot process the blob without validating its parent, the peer isn't necessarily faulty.
-    BlobParentUnknown {
-        blob_root: Hash256,
-        blob_parent_root: Hash256,
-    },
+    BlobParentUnknown(Arc<BlobSidecar<T>>),
 
     /// A blob has already been seen for the given `(sidecar.block_root, sidecar.index)` tuple
     /// over gossip or no gossip sources.
@@ -114,41 +113,50 @@ pub enum BlobError {
     },
 }
 
-impl From<BeaconChainError> for BlobError {
+impl<T: EthSpec> From<BeaconChainError> for BlobError<T> {
     fn from(e: BeaconChainError) -> Self {
         BlobError::BeaconChainError(e)
     }
 }
 
-impl From<BeaconStateError> for BlobError {
+impl<T: EthSpec> From<BeaconStateError> for BlobError<T> {
     fn from(e: BeaconStateError) -> Self {
         BlobError::BeaconChainError(BeaconChainError::BeaconStateError(e))
     }
 }
 
 /// A wrapper around a `BlobSidecar` that indicates it has been approved for re-gossiping on
 /// the p2p network.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct GossipVerifiedBlob<T: EthSpec> {
     blob: Arc<BlobSidecar<T>>,
 }
 
 impl<T: EthSpec> GossipVerifiedBlob<T> {
+    pub fn id(&self) -> BlobIdentifier {
+        self.blob.id()
+    }
     pub fn block_root(&self) -> Hash256 {
         self.blob.block_root
     }
+    pub fn to_blob(self) -> Arc<BlobSidecar<T>> {
+        self.blob
+    }
+    pub fn slot(&self) -> Slot {
+        self.blob.slot
+    }
 }
 
 pub fn validate_blob_sidecar_for_gossip<T: BeaconChainTypes>(
     signed_blob_sidecar: SignedBlobSidecar<T::EthSpec>,
     subnet: u64,
     chain: &BeaconChain<T>,
-) -> Result<GossipVerifiedBlob<T::EthSpec>, BlobError> {
+) -> Result<GossipVerifiedBlob<T::EthSpec>, BlobError<T::EthSpec>> {
     let blob_slot = signed_blob_sidecar.message.slot;
     let blob_index = signed_blob_sidecar.message.index;
-    let block_root = signed_blob_sidecar.message.block_root;
     let block_parent_root = signed_blob_sidecar.message.block_parent_root;
     let blob_proposer_index = signed_blob_sidecar.message.proposer_index;
+    let block_root = signed_blob_sidecar.message.block_root;
 
     // Verify that the blob_sidecar was received on the correct subnet.
     if blob_index != subnet {
@@ -211,10 +219,7 @@ pub fn validate_blob_sidecar_for_gossip<T: BeaconChainTypes>(
             });
         }
     } else {
-        return Err(BlobError::BlobParentUnknown {
-            blob_root: block_root,
-            blob_parent_root: block_parent_root,
-        });
+        return Err(BlobError::BlobParentUnknown(signed_blob_sidecar.message));
     }
 
     // Note: We check that the proposer_index matches against the shuffling first to avoid
@@ -366,7 +371,7 @@ fn cheap_state_advance_to_obtain_committees<'a, E: EthSpec>(
     state_root_opt: Option<Hash256>,
     blob_slot: Slot,
     spec: &ChainSpec,
-) -> Result<Cow<'a, BeaconState<E>>, BlobError> {
+) -> Result<Cow<'a, BeaconState<E>>, BlobError<E>> {
     let block_epoch = blob_slot.epoch(E::slots_per_epoch());
 
     if state.current_epoch() == block_epoch {
@@ -443,19 +448,14 @@ impl<T: EthSpec> KzgVerifiedBlob<T> {
 ///
 /// Returns an error if the kzg verification check fails.
 pub fn verify_kzg_for_blob<T: EthSpec>(
-    blob: GossipVerifiedBlob<T>,
+    blob: Arc<BlobSidecar<T>>,
     kzg: &Kzg,
 ) -> Result<KzgVerifiedBlob<T>, AvailabilityCheckError> {
     //TODO(sean) remove clone
-    if validate_blob::<T>(
-        kzg,
-        blob.blob.blob.clone(),
-        blob.blob.kzg_commitment,
-        blob.blob.kzg_proof,
-    )
-    .map_err(AvailabilityCheckError::Kzg)?
+    if validate_blob::<T>(kzg, blob.blob.clone(), blob.kzg_commitment, blob.kzg_proof)
+        .map_err(AvailabilityCheckError::Kzg)?
     {
-        Ok(KzgVerifiedBlob { blob: blob.blob })
+        Ok(KzgVerifiedBlob { blob })
     } else {
         Err(AvailabilityCheckError::KzgVerificationFailed)
     }
@@ -467,7 +467,7 @@ pub fn verify_kzg_for_blob<T: EthSpec>(
 /// Note: This function should be preferred over calling `verify_kzg_for_blob`
 /// in a loop since this function kzg verifies a list of blobs more efficiently.
 pub fn verify_kzg_for_blob_list<T: EthSpec>(
-    blob_list: BlobSidecarList<T>,
+    blob_list: Vec<Arc<BlobSidecar<T>>>,
     kzg: &Kzg,
 ) -> Result<KzgVerifiedBlobList<T>, AvailabilityCheckError> {
     let (blobs, (commitments, proofs)): (Vec<_>, (Vec<_>, Vec<_>)) = blob_list
@@ -608,7 +608,16 @@ impl<E: EthSpec> AsBlock<E> for &MaybeAvailableBlock<E> {
 #[derivative(Hash(bound = "E: EthSpec"))]
 pub enum BlockWrapper<E: EthSpec> {
     Block(Arc<SignedBeaconBlock<E>>),
-    BlockAndBlobs(Arc<SignedBeaconBlock<E>>, Vec<Arc<BlobSidecar<E>>>),
+    BlockAndBlobs(Arc<SignedBeaconBlock<E>>, FixedBlobSidecarList<E>),
+}
+
+impl<E: EthSpec> BlockWrapper<E> {
+    pub fn deconstruct(self) -> (Arc<SignedBeaconBlock<E>>, Option<FixedBlobSidecarList<E>>) {
+        match self {
+            BlockWrapper::Block(block) => (block, None),
+            BlockWrapper::BlockAndBlobs(block, blobs) => (block, Some(blobs)),
+        }
+    }
 }
 
 impl<E: EthSpec> AsBlock<E> for BlockWrapper<E> {
@@ -675,13 +684,15 @@ impl<E: EthSpec> From<SignedBeaconBlock<E>> for BlockWrapper<E> {
 impl<E: EthSpec> From<BlockContentsTuple<E, FullPayload<E>>> for BlockWrapper<E> {
     fn from(value: BlockContentsTuple<E, FullPayload<E>>) -> Self {
         match value.1 {
-            Some(variable_list) => Self::BlockAndBlobs(
-                Arc::new(value.0),
-                Vec::from(variable_list)
-                    .into_iter()
-                    .map(|signed_blob| signed_blob.message)
-                    .collect::<Vec<_>>(),
-            ),
+            Some(variable_list) => {
+                let mut blobs = Vec::with_capacity(E::max_blobs_per_block());
+                for blob in variable_list {
+                    if blob.message.index < E::max_blobs_per_block() as u64 {
+                        blobs.insert(blob.message.index as usize, Some(blob.message));
+                    }
+                }
+                Self::BlockAndBlobs(Arc::new(value.0), FixedVector::from(blobs))
+            }
             None => Self::Block(Arc::new(value.0)),
         }
     }