Skip to content

Gloas Column and Blob variants and deprecating full blob support #8674

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
eserilev wants to merge 4 commits into from
+1,475 −2,753
Closed

Gloas Column and Blob variants and deprecating full blob support #8674

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1ef3ce7
superstruct columns and blobs and delete unused codepaths
eserilev Jan 17, 2026
c161333
I just yeeted a bunch of code
eserilev Jan 17, 2026
66dbc04
More lints
eserilev Jan 19, 2026
c1d2d0c
Revert blob changes
eserilev Jan 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
170 changes: 65 additions & 105 deletions beacon_node/beacon_chain/src/beacon_chain.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@ use crate::beacon_block_streamer::{BeaconBlockStreamer, CheckCaches};
use crate::beacon_proposer_cache::{
BeaconProposerCache, EpochBlockProposers, ensure_state_can_determine_proposers_for_epoch,
};
use crate::blob_verification::{GossipBlobError, GossipVerifiedBlob};
use crate::block_times_cache::BlockTimesCache;
use crate::block_verification::POS_PANDA_BANNER;
use crate::block_verification::{
Expand Down Expand Up @@ -412,8 +411,6 @@ pub struct BeaconChain<T: BeaconChainTypes> {
pub(crate) observed_sync_aggregators: RwLock<ObservedSyncAggregators<T::EthSpec>>,
/// Maintains a record of which validators have proposed blocks for each slot.
pub observed_block_producers: RwLock<ObservedBlockProducers<T::EthSpec>>,
/// Maintains a record of blob sidecars seen over the gossip network.
pub observed_blob_sidecars: RwLock<ObservedDataSidecars<BlobSidecar<T::EthSpec>>>,
/// Maintains a record of column sidecars seen over the gossip network.
pub observed_column_sidecars: RwLock<ObservedDataSidecars<DataColumnSidecar<T::EthSpec>>>,
/// Maintains a record of slashable message seen over the gossip network or RPC.
Expand Down Expand Up @@ -1131,13 +1128,18 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
.or_else(|| self.early_attester_cache.get_data_columns(block_root));

if let Some(mut all_cached_columns) = all_cached_columns_opt {
all_cached_columns.retain(|col| indices.contains(&col.index));
all_cached_columns.retain(|col| indices.contains(col.index()));
Ok(all_cached_columns)
} else {
} else if let Some(block) = self.get_blinded_block(&block_root)? {
indices
.iter()
.filter_map(|index| self.get_data_column(&block_root, index).transpose())
.filter_map(|index| {
self.get_data_column(&block_root, index, block.fork_name_unchecked())
.transpose()
})
.collect::<Result<_, _>>()
} else {
Ok(vec![])
}
}

Expand Down Expand Up @@ -1222,8 +1224,11 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
pub fn get_data_columns(
&self,
block_root: &Hash256,
fork_name: ForkName,
) -> Result<Option<DataColumnSidecarList<T::EthSpec>>, Error> {
self.store.get_data_columns(block_root).map_err(Error::from)
self.store
.get_data_columns(block_root, fork_name)
.map_err(Error::from)
}

/// Returns the blobs at the given root, if any.
Expand All @@ -1244,7 +1249,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
};

if self.spec.is_peer_das_enabled_for_epoch(block.epoch()) {
if let Some(columns) = self.store.get_data_columns(block_root)? {
let fork_name = self.spec.fork_name_at_epoch(block.epoch());
if let Some(columns) = self.store.get_data_columns(block_root, fork_name)? {
let num_required_columns = T::EthSpec::number_of_columns() / 2;
let reconstruction_possible = columns.len() >= num_required_columns;
if reconstruction_possible {
Expand All @@ -1260,7 +1266,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
Ok(None)
}
} else {
self.get_blobs(block_root).map(|b| b.blobs())
Ok(self.get_blobs(block_root)?.blobs())
}
}

Expand All @@ -1272,8 +1278,11 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
&self,
block_root: &Hash256,
column_index: &ColumnIndex,
fork_name: ForkName,
) -> Result<Option<Arc<DataColumnSidecar<T::EthSpec>>>, Error> {
Ok(self.store.get_data_column(block_root, column_index)?)
Ok(self
.store
.get_data_column(block_root, column_index, fork_name)?)
}

pub fn get_blinded_block(
Expand Down Expand Up @@ -2194,19 +2203,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
})
}

#[instrument(skip_all, level = "trace")]
pub fn verify_blob_sidecar_for_gossip(
self: &Arc<Self>,
blob_sidecar: Arc<BlobSidecar<T::EthSpec>>,
subnet_id: u64,
) -> Result<GossipVerifiedBlob<T>, GossipBlobError> {
metrics::inc_counter(&metrics::BLOBS_SIDECAR_PROCESSING_REQUESTS);
let _timer = metrics::start_timer(&metrics::BLOBS_SIDECAR_GOSSIP_VERIFICATION_TIMES);
GossipVerifiedBlob::new(blob_sidecar, subnet_id, self).inspect(|_| {
metrics::inc_counter(&metrics::BLOBS_SIDECAR_PROCESSING_SUCCESSES);
})
}

/// Accepts some 'LightClientOptimisticUpdate' from the network and attempts to verify it
pub fn verify_optimistic_update_for_gossip(
self: &Arc<Self>,
Expand Down Expand Up @@ -3002,35 +2998,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
.map_err(BeaconChainError::TokioJoin)?
}

/// Cache the blob in the processing cache, process it, then evict it from the cache if it was
/// imported or errors.
#[instrument(skip_all, level = "debug")]
pub async fn process_gossip_blob(
self: &Arc<Self>,
blob: GossipVerifiedBlob<T>,
) -> Result<AvailabilityProcessingStatus, BlockError> {
let block_root = blob.block_root();

// If this block has already been imported to forkchoice it must have been available, so
// we don't need to process its blobs again.
if self
.canonical_head
.fork_choice_read_lock()
.contains_block(&block_root)
{
return Err(BlockError::DuplicateFullyImported(blob.block_root()));
}

// No need to process and import blobs beyond the PeerDAS epoch.
if self.spec.is_peer_das_enabled_for_epoch(blob.epoch()) {
return Err(BlockError::BlobNotRequired(blob.slot()));
}

self.emit_sse_blob_sidecar_events(&block_root, std::iter::once(blob.as_blob()));

self.check_gossip_blob_availability_and_import(blob).await
}

/// Cache the data columns in the processing cache, process it, then evict it from the cache if it was
/// imported or errors.
#[instrument(skip_all, level = "debug")]
Expand Down Expand Up @@ -3093,19 +3060,21 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
return Err(BlockError::DuplicateFullyImported(block_root));
}

// Reject RPC blobs referencing unknown parents. Otherwise we allow potentially invalid data
// into the da_checker, where invalid = descendant of invalid blocks.
// Note: blobs should have at least one item and all items have the same parent root.
if let Some(parent_root) = blobs
.iter()
.filter_map(|b| b.as_ref().map(|b| b.block_parent_root()))
.next()
&& !self
.canonical_head
.fork_choice_read_lock()
.contains_block(&parent_root)
{
return Err(BlockError::ParentUnknown { parent_root });
for blob in &blobs {
if let Some(blob) = blob.as_ref() {
// Reject RPC blobs referencing unknown parents. Otherwise we allow potentially invalid data
// into the da_checker, where invalid = descendant of invalid blocks.
// Note: blobs should have at least one item and all items have the same parent root.
if !self
.canonical_head
.fork_choice_read_lock()
.contains_block(&blob.block_parent_root())
{
return Err(BlockError::ParentUnknown {
parent_root: blob.block_parent_root(),
});
}
}
}

self.emit_sse_blob_sidecar_events(&block_root, blobs.iter().flatten().map(Arc::as_ref));
Expand All @@ -3132,9 +3101,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
}

match &engine_get_blobs_output {
EngineGetBlobsOutput::Blobs(blobs) => {
self.emit_sse_blob_sidecar_events(&block_root, blobs.iter().map(|b| b.as_blob()));
}
EngineGetBlobsOutput::CustodyColumns(columns) => {
self.emit_sse_data_column_sidecar_events(
&block_root,
Expand Down Expand Up @@ -3183,7 +3149,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
.cached_data_column_indexes(block_root)
.unwrap_or_default();
let new_data_columns =
data_columns_iter.filter(|b| !imported_data_columns.contains(&b.index));
data_columns_iter.filter(|b| !imported_data_columns.contains(b.index()));

for data_column in new_data_columns {
event_handler.register(EventKind::DataColumnSidecar(
Expand Down Expand Up @@ -3223,7 +3189,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
// Reject RPC columns referencing unknown parents. Otherwise we allow potentially invalid data
// into the da_checker, where invalid = descendant of invalid blocks.
// Note: custody_columns should have at least one item and all items have the same parent root.
if let Some(parent_root) = custody_columns.iter().map(|c| c.block_parent_root()).next()
if let Some(parent_root) = custody_columns
.iter()
.filter_map(|c| match c.as_ref() {
DataColumnSidecar::Fulu(column) => Some(column.block_parent_root()),
_ => None,
})
.next()
&& !self
.canonical_head
.fork_choice_read_lock()
Expand Down Expand Up @@ -3515,24 +3487,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
.await
}

/// Checks if the provided blob can make any cached blocks available, and imports immediately
/// if so, otherwise caches the blob in the data availability checker.
async fn check_gossip_blob_availability_and_import(
self: &Arc<Self>,
blob: GossipVerifiedBlob<T>,
) -> Result<AvailabilityProcessingStatus, BlockError> {
let slot = blob.slot();
if let Some(slasher) = self.slasher.as_ref() {
slasher.accept_block_header(blob.signed_block_header());
}
let availability = self
.data_availability_checker
.put_gossip_verified_blobs(blob.block_root(), std::iter::once(blob))?;

self.process_availability(slot, availability, || Ok(()))
.await
}

/// Checks if the provided data column can make any cached blocks available, and imports immediately
/// if so, otherwise caches the data column in the data availability checker.
async fn check_gossip_data_columns_availability_and_import(
Expand All @@ -3543,8 +3497,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
publish_fn: impl FnOnce() -> Result<(), BlockError>,
) -> Result<AvailabilityProcessingStatus, BlockError> {
if let Some(slasher) = self.slasher.as_ref() {
for data_colum in &data_columns {
slasher.accept_block_header(data_colum.signed_block_header());
for data_column in &data_columns {
if let DataColumnSidecar::Fulu(c) = data_column.as_data_column() {
slasher.accept_block_header(c.signed_block_header.clone());
}
}
}

Expand Down Expand Up @@ -3596,7 +3552,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
) -> Result<AvailabilityProcessingStatus, BlockError> {
self.check_blob_header_signature_and_slashability(
block_root,
blobs.iter().flatten().map(Arc::as_ref),
blobs.iter().flatten().map(|b| b.as_ref()),
)?;
let availability = self
.data_availability_checker
Expand All @@ -3613,18 +3569,15 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
engine_get_blobs_output: EngineGetBlobsOutput<T>,
) -> Result<AvailabilityProcessingStatus, BlockError> {
let availability = match engine_get_blobs_output {
EngineGetBlobsOutput::Blobs(blobs) => {
self.check_blob_header_signature_and_slashability(
block_root,
blobs.iter().map(|b| b.as_blob()),
)?;
self.data_availability_checker
.put_kzg_verified_blobs(block_root, blobs)?
}
EngineGetBlobsOutput::CustodyColumns(data_columns) => {
self.check_data_column_sidecar_header_signature_and_slashability(
block_root,
data_columns.iter().map(|c| c.as_data_column()),
data_columns
.iter()
.filter_map(|c| match c.as_data_column() {
DataColumnSidecar::Fulu(column) => Some(column),
_ => None,
}),
)?;
self.data_availability_checker
.put_kzg_verified_custody_data_columns(block_root, data_columns)?
Expand All @@ -3645,7 +3598,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
) -> Result<AvailabilityProcessingStatus, BlockError> {
self.check_data_column_sidecar_header_signature_and_slashability(
block_root,
custody_columns.iter().map(|c| c.as_ref()),
custody_columns.iter().filter_map(|c| match c.as_ref() {
DataColumnSidecar::Fulu(fulu) => Some(fulu),
_ => None,
}),
)?;

// This slot value is purely informative for the consumers of
Expand All @@ -3663,7 +3619,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
fn check_data_column_sidecar_header_signature_and_slashability<'a>(
self: &Arc<Self>,
block_root: Hash256,
custody_columns: impl IntoIterator<Item = &'a DataColumnSidecar<T::EthSpec>>,
custody_columns: impl IntoIterator<Item = &'a DataColumnSidecarFulu<T::EthSpec>>,
) -> Result<(), BlockError> {
let mut slashable_cache = self.observed_slashable.write();
// Process all unique block headers - previous logic assumed all headers were identical and
Expand Down Expand Up @@ -7366,7 +7322,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
// Supernodes need to persist all sampled custody columns
if columns_to_custody.len() != self.spec.number_of_custody_groups as usize {
data_columns
.retain(|data_column| columns_to_custody.contains(&data_column.index));
.retain(|data_column| columns_to_custody.contains(data_column.index()));
}
debug!(
%block_root,
Expand All @@ -7379,7 +7335,11 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
}

/// Retrieves block roots (in ascending slot order) within some slot range from fork choice.
pub fn block_roots_from_fork_choice(&self, start_slot: u64, count: u64) -> Vec<Hash256> {
pub fn block_roots_from_fork_choice(
&self,
start_slot: u64,
count: u64,
) -> Vec<(Hash256, Slot)> {
let head_block_root = self.canonical_head.cached_head().head_block_root();
let fork_choice_read_lock = self.canonical_head.fork_choice_read_lock();
let block_roots_iter = fork_choice_read_lock
Expand All @@ -7390,7 +7350,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {

for (root, slot) in block_roots_iter {
if slot < end_slot && slot >= start_slot {
roots.push(root);
roots.push((root, slot));
}
if slot < start_slot {
break;
Expand Down
Loading
Loading