#5684 review comments

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -3081,15 +3081,15 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         self.remove_notified(&block_root, r)
     }
 
-    /// Cache the blobs in the processing cache, process it, then evict it from the cache if it was
+    /// Cache the columns in the processing cache, process it, then evict it from the cache if it was
     /// imported or errors.
     pub async fn process_rpc_custody_columns(
         self: &Arc<Self>,
         block_root: Hash256,
         custody_columns: Vec<CustodyDataColumn<T::EthSpec>>,
     ) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
         // If this block has already been imported to forkchoice it must have been available, so
-        // we don't need to process its blobs again.
+        // we don't need to process its columns again.
         if self
             .canonical_head
             .fork_choice_read_lock()
@@ -3099,9 +3099,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         }
 
         // TODO(das): custody column SSE event
-        // TODO(das): Why is the slot necessary here?
-        let slot = Slot::new(0);
-
+        let slot = custody_columns
+            .first()
+            .ok_or(BeaconChainError::EmptyRpcCustodyColumns)?
+            .as_data_column()
+            .signed_block_header
+            .message
+            .slot;
         let r = self
             .check_rpc_custody_columns_availability_and_import(slot, block_root, custody_columns)
             .await;
@@ -3494,6 +3498,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             );
         }
 
+        // TODO(das) record custody column available timestamp
+
         // import
         let chain = self.clone();
         let block_root = self

beacon_node/beacon_chain/src/block_verification_types.rs

@@ -96,10 +96,7 @@ enum RpcBlockInner<E: EthSpec> {
     BlockAndBlobs(Arc<SignedBeaconBlock<E>>, BlobSidecarList<E>),
     /// This variant is used with parent lookups and by-range responses. It should have all
     /// requested data columns, all block roots matching for this block.
-    BlockAndCustodyColumns(
-        Arc<SignedBeaconBlock<E>>,
-        VariableList<CustodyDataColumn<E>, <E as EthSpec>::DataColumnCount>,
-    ),
+    BlockAndCustodyColumns(Arc<SignedBeaconBlock<E>>, CustodyDataColumnList<E>),
 }
 
 impl<E: EthSpec> RpcBlock<E> {
@@ -168,13 +165,9 @@ impl<E: EthSpec> RpcBlock<E> {
             // The number of required custody columns is out of scope here.
             return Err(AvailabilityCheckError::MissingCustodyColumns);
         }
-        // Treat empty blob lists as if they are missing.
-        let inner = if custody_columns.is_empty() {
-            RpcBlockInner::BlockAndCustodyColumns(
-                block,
-                VariableList::new(custody_columns)
-                    .expect("TODO(das): custody vec should never exceed len"),
-            )
+        // Treat empty data column lists as if they are missing.
+        let inner = if !custody_columns.is_empty() {
+            RpcBlockInner::BlockAndCustodyColumns(block, VariableList::new(custody_columns)?)
         } else {
             RpcBlockInner::Block(block)
         };

beacon_node/beacon_chain/src/data_availability_checker.rs

@@ -196,9 +196,10 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         };
 
         // TODO(das): report which column is invalid for proper peer scoring
+        // TODO(das): batch KZG verification here
         let verified_custody_columns = custody_columns
             .into_iter()
-            .map(|c| KzgVerifiedCustodyDataColumn::new(c, kzg).map_err(AvailabilityCheckError::Kzg))
+            .map(|c| KzgVerifiedCustodyDataColumn::new(c, kzg))
             .collect::<Result<Vec<_>, _>>()?;
 
         self.availability_cache

beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs

@@ -81,7 +81,7 @@ impl<E: EthSpec> PendingComponents<E> {
         &self.verified_blobs
     }
 
-    /// Returns a mutable reference to the cached data column.
+    /// Returns an immutable reference to the cached data column.
     pub fn get_cached_data_column(
         &self,
         data_column_index: u64,
@@ -340,7 +340,6 @@ impl<E: EthSpec> PendingComponents<E> {
             block,
             blobs,
             blobs_available_timestamp,
-            // TODO(das) Do we need a check here for number of expected custody columns?
             // TODO(das): Update store types to prevent this conversion
             data_columns: Some(
                 VariableList::new(

beacon_node/beacon_chain/src/data_column_verification.rs

@@ -228,8 +228,8 @@ impl<E: EthSpec> KzgVerifiedCustodyDataColumn<E> {
     }
 
     /// Verify a column already marked as custody column
-    pub fn new(data_column: CustodyDataColumn<E>, _kzg: &Kzg) -> Result<Self, KzgError> {
-        // TODO(das): verify kzg proof
+    pub fn new(data_column: CustodyDataColumn<E>, kzg: &Kzg) -> Result<Self, KzgError> {
+        verify_kzg_for_data_column(data_column.clone_arc(), kzg)?;
         Ok(Self {
             data: data_column.data,
         })

beacon_node/beacon_chain/src/errors.rs

@@ -83,8 +83,6 @@ pub enum BeaconChainError {
         max_task_runtime: Duration,
     },
     MissingFinalizedStateRoot(Slot),
-    /// Returned when an internal check fails, indicating corrupt data.
-    InvariantViolated(String),
     SszTypesError(SszTypesError),
     NoProposerForSlot(Slot),
     CanonicalHeadLockTimeout,
@@ -227,6 +225,7 @@ pub enum BeaconChainError {
     LightClientError(LightClientError),
     UnsupportedFork,
     MilhouseError(MilhouseError),
+    EmptyRpcCustodyColumns,
 }
 
 easy_from_to!(SlotProcessingError, BeaconChainError);

beacon_node/network/src/network_beacon_processor/mod.rs

@@ -504,7 +504,8 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         })
     }
 
-    /// Create a new `Work` event for some data_columns from ReqResp
+    /// Create a new `Work` event for some sampling columns, and reports the verification result
+    /// back to sync.
     pub fn send_rpc_validate_data_columns(
         self: &Arc<Self>,
         block_root: Hash256,

beacon_node/network/src/sync/block_lookups/tests.rs

@@ -897,7 +897,7 @@ impl TestRig {
                 None
             }
         })
-        .unwrap_or_else(|e| panic!("Expected sample verify work: {e}"))
+        .unwrap_or_else(|e| panic!("Expected RPC custody column work: {e}"))
     }
 
     fn expect_rpc_sample_verify_work_event(&mut self) {
@@ -1629,8 +1629,8 @@ fn custody_lookup_happy_path() {
     // Should not request blobs
     let id = r.expect_block_lookup_request(block.canonical_root());
     r.complete_valid_block_request(id, block.into(), true);
-    // TODO(das): do not hardcode 4
-    let custody_ids = r.expect_only_data_columns_by_root_requests(block_root, 4);
+    let custody_column_count = E::min_custody_requirement() * E::data_columns_per_subnet();
+    let custody_ids = r.expect_only_data_columns_by_root_requests(block_root, custody_column_count);
     r.complete_valid_custody_request(custody_ids, data_columns, false);
     r.expect_no_active_lookups();
 }