From 23a7dc561fc65b22ee2f697aabc7b2a1ef3670de Mon Sep 17 00:00:00 2001 From: Eitan Seri- Levi Date: Tue, 10 Feb 2026 21:13:40 -0800 Subject: [PATCH] Fix --- .../src/data_availability_checker.rs | 230 +++++++++--------- 1 file changed, 110 insertions(+), 120 deletions(-) diff --git a/beacon_node/beacon_chain/src/data_availability_checker.rs b/beacon_node/beacon_chain/src/data_availability_checker.rs index 75b0760b4e..f6daf386a9 100644 --- a/beacon_node/beacon_chain/src/data_availability_checker.rs +++ b/beacon_node/beacon_chain/src/data_availability_checker.rs @@ -136,6 +136,10 @@ impl DataAvailabilityChecker { }) } + pub fn custody_context(&self) -> &Arc> { + &self.custody_context + } + /// Checks if the block root is currently in the availability cache awaiting import because /// of missing components. /// @@ -159,6 +163,30 @@ impl DataAvailabilityChecker { }) } + /// Return the set of cached custody column indexes for `block_root`. Returns None if there is + /// no block component for `block_root`. + pub fn cached_data_column_indexes(&self, block_root: &Hash256) -> Option> { + self.availability_cache + .peek_pending_components(block_root, |components| { + components.map(|components| components.get_cached_data_columns_indices()) + }) + } + + /// Check if the exact data column is in the availability cache. + pub fn is_data_column_cached( + &self, + block_root: &Hash256, + data_column: &DataColumnSidecar, + ) -> bool { + self.availability_cache + .peek_pending_components(block_root, |components| { + components.is_some_and(|components| { + let cached_column_opt = components.get_cached_data_column(*data_column.index()); + cached_column_opt.is_some_and(|cached| *cached == *data_column) + }) + }) + } + /// Get a blob from the availability cache. pub fn get_blob( &self, @@ -167,6 +195,14 @@ impl DataAvailabilityChecker { self.availability_cache.peek_blob(blob_id) } + /// Get data columns for a block from the availability cache. + pub fn get_data_columns( + &self, + block_root: Hash256, + ) -> Option> { + self.availability_cache.peek_data_columns(block_root) + } + /// Put a list of blobs received via RPC into the availability cache. This performs KZG /// verification on the blobs in the list. #[instrument(skip_all, level = "trace")] @@ -194,6 +230,39 @@ impl DataAvailabilityChecker { .put_kzg_verified_blobs(block_root, verified_blobs) } + /// Put a list of custody columns received via RPC into the availability cache. This performs KZG + /// verification on the blobs in the list. + #[allow(clippy::type_complexity)] + #[instrument(skip_all, level = "trace")] + pub fn put_rpc_custody_columns( + &self, + block_root: Hash256, + slot: Slot, + custody_columns: DataColumnSidecarList, + ) -> Result, AvailabilityCheckError> { + // Attributes fault to the specific peer that sent an invalid column + let kzg_verified_columns = + KzgVerifiedDataColumn::from_batch_with_scoring(custody_columns, &self.kzg) + .map_err(AvailabilityCheckError::InvalidColumn)?; + + // Filter out columns that aren't required for custody for this slot + // This is required because `data_columns_by_root` requests the **latest** CGC that _may_ + // not be yet effective for data availability check, as CGC changes are only effecive from + // a new epoch. + let epoch = slot.epoch(T::EthSpec::slots_per_epoch()); + let sampling_columns = self + .custody_context + .sampling_columns_for_epoch(epoch, &self.spec); + let verified_custody_columns = kzg_verified_columns + .into_iter() + .filter(|col| sampling_columns.contains(&col.index())) + .map(KzgVerifiedCustodyDataColumn::from_asserted_custody) + .collect::>(); + + self.availability_cache + .put_kzg_verified_data_columns(block_root, verified_custody_columns) + } + /// Check if we've cached other blobs for this block. If it completes a set and we also /// have a block cached, return the `Availability` variant triggering block import. /// Otherwise cache the blob sidecar. @@ -222,6 +291,47 @@ impl DataAvailabilityChecker { .put_kzg_verified_blobs(block_root, blobs) } + /// Check if we've cached other data columns for this block. If it satisfies the custody requirement and we also + /// have a block cached, return the `Availability` variant triggering block import. + /// Otherwise cache the data column sidecar. + /// + /// This should only accept gossip verified data columns, so we should not have to worry about dupes. + #[instrument(skip_all, level = "trace")] + pub fn put_gossip_verified_data_columns< + O: ObservationStrategy, + I: IntoIterator>, + >( + &self, + block_root: Hash256, + slot: Slot, + data_columns: I, + ) -> Result, AvailabilityCheckError> { + let epoch = slot.epoch(T::EthSpec::slots_per_epoch()); + let sampling_columns = self + .custody_context + .sampling_columns_for_epoch(epoch, &self.spec); + let custody_columns = data_columns + .into_iter() + .filter(|col| sampling_columns.contains(&col.index())) + .map(|c| KzgVerifiedCustodyDataColumn::from_asserted_custody(c.into_inner())) + .collect::>(); + + self.availability_cache + .put_kzg_verified_data_columns(block_root, custody_columns) + } + + #[instrument(skip_all, level = "trace")] + pub fn put_kzg_verified_custody_data_columns< + I: IntoIterator>, + >( + &self, + block_root: Hash256, + custody_columns: I, + ) -> Result, AvailabilityCheckError> { + self.availability_cache + .put_kzg_verified_data_columns(block_root, custody_columns) + } + /// Check if we have all the blobs for a block. Returns `Availability` which has information /// about whether all components have been received or more are required. pub fn put_executed_block( @@ -356,113 +466,6 @@ impl DataAvailabilityChecker { block_cache_size: self.availability_cache.block_cache_size(), } } -} - -impl DataAvailabilityChecker { - pub fn custody_context(&self) -> &Arc> { - &self.custody_context - } - - /// Get data columns for a block from the availability cache. - pub fn get_data_columns( - &self, - block_root: Hash256, - ) -> Option> { - self.availability_cache.peek_data_columns(block_root) - } - - /// Return the set of cached custody column indices for `block_root`. Returns None if there is - /// no block component for `block_root`. - pub fn cached_data_column_indexes(&self, block_root: &Hash256) -> Option> { - self.availability_cache - .peek_pending_components(block_root, |components| { - components.map(|components| components.get_cached_data_columns_indices()) - }) - } - - /// Check if the exact data column is in the availability cache. - pub fn is_data_column_cached( - &self, - block_root: &Hash256, - data_column: &DataColumnSidecar, - ) -> bool { - self.availability_cache - .peek_pending_components(block_root, |components| { - components.is_some_and(|components| { - let cached_column_opt = components.get_cached_data_column(*data_column.index()); - cached_column_opt.is_some_and(|cached| *cached == *data_column) - }) - }) - } - - /// Put a list of custody columns received via RPC into the availability cache. This performs KZG - /// verification on the blobs in the list. - #[allow(clippy::type_complexity)] - #[instrument(skip_all, level = "trace")] - pub fn put_rpc_custody_columns( - &self, - block_root: Hash256, - slot: Slot, - custody_columns: DataColumnSidecarList, - ) -> Result, AvailabilityCheckError> { - // Attributes fault to the specific peer that sent an invalid column - let kzg_verified_columns = - KzgVerifiedDataColumn::from_batch_with_scoring(custody_columns, &self.kzg) - .map_err(AvailabilityCheckError::InvalidColumn)?; - - // Filter out columns that aren't required for custody for this slot - // This is required because `data_columns_by_root` requests the **latest** CGC that _may_ - // not be yet effective for data availability check, as CGC changes are only effecive from - // a new epoch. - let epoch = slot.epoch(T::EthSpec::slots_per_epoch()); - let sampling_columns = self - .custody_context - .sampling_columns_for_epoch(epoch, &self.spec); - let verified_custody_columns = kzg_verified_columns - .into_iter() - .filter(|col| sampling_columns.contains(&col.index())) - .map(KzgVerifiedCustodyDataColumn::from_asserted_custody) - .collect::>(); - - self.availability_cache - .put_kzg_verified_data_columns(block_root, verified_custody_columns) - } - - /// Check if we've cached other data columns for this block. If it satisfies the custody requirement and we also - /// have a block cached, return the `Availability` variant triggering block import. - /// Otherwise cache the data column sidecar. - /// - /// This should only accept gossip verified data columns, so we should not have to worry about dupes. - #[instrument(skip_all, level = "trace")] - pub fn put_gossip_verified_data_columns( - &self, - block_root: Hash256, - slot: Slot, - data_columns: Vec>, - ) -> Result, AvailabilityCheckError> { - let epoch = slot.epoch(T::EthSpec::slots_per_epoch()); - let sampling_columns = self - .custody_context - .sampling_columns_for_epoch(epoch, &self.spec); - let custody_columns = data_columns - .into_iter() - .filter(|col| sampling_columns.contains(&col.index())) - .map(|c| KzgVerifiedCustodyDataColumn::from_asserted_custody(c.into_inner())) - .collect::>(); - - self.availability_cache - .put_kzg_verified_data_columns(block_root, custody_columns) - } - - #[instrument(skip_all, level = "trace")] - pub fn put_kzg_verified_custody_data_columns( - &self, - block_root: Hash256, - custody_columns: Vec>, - ) -> Result, AvailabilityCheckError> { - self.availability_cache - .put_kzg_verified_data_columns(block_root, custody_columns) - } #[instrument(skip_all, level = "debug")] pub fn reconstruct_data_columns( @@ -551,18 +554,6 @@ impl DataAvailabilityChecker { )) }) } - - /// Verifies KZG commitments for data columns. - pub fn verify_kzg_for_data_columns( - &self, - data_columns: &DataColumnSidecarList, - ) -> Result<(), AvailabilityCheckError> { - if !data_columns.is_empty() { - verify_kzg_for_data_column_list(data_columns.iter(), &self.kzg) - .map_err(AvailabilityCheckError::InvalidColumn)?; - } - Ok(()) - } } /// Helper struct to group data availability checker metrics. @@ -590,7 +581,6 @@ pub fn start_availability_cache_maintenance_service( } } -// TODO(gloas) we can shut down this service once we reach the gloas fork epoch async fn availability_cache_maintenance_service( chain: Arc>, overflow_cache: Arc>,