Add prysmctl backfill verify command for blob and data column verification

Implements a new verification tool to check that backfill successfully
retrieved blobs and data columns from the beacon chain. The tool:

- Auto-detects network configuration by querying beacon node genesis info
- Calculates blob retention window based on MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUEST
- Verifies blobs via /eth/v1/beacon/blob_sidecars/{slot} API
- Verifies data columns (post-Fulu) via /eth/v1/debug/beacon/data_column_sidecars/{slot}
- Provides configurable verbosity with progress updates every 1000 slots
- Generates comprehensive statistics summary

Usage:
  prysmctl backfill verify --beacon-node-host=http://localhost:3500
  prysmctl backfill verify --beacon-node-host=http://localhost:3500 --verbose

Supports Hoodi testnet and mainnet with automatic config detection.

Fix backfill verify tool

.github/PULL_REQUEST_TEMPLATE.md

@@ -34,5 +34,4 @@ Fixes #
 
 - [ ] I have read [CONTRIBUTING.md](https://github.com/prysmaticlabs/prysm/blob/develop/CONTRIBUTING.md).
 - [ ] I have included a uniquely named [changelog fragment file](https://github.com/prysmaticlabs/prysm/blob/develop/CONTRIBUTING.md#maintaining-changelogmd).
-- [ ] I have added a description with sufficient context for reviewers to understand this PR.
-- [ ] I have tested that my changes work as expected and I added a testing plan to the PR description (if applicable). 
+- [ ] I have added a description to this PR with sufficient context for reviewers to understand this PR.

BUILD.bazel

@@ -193,7 +193,6 @@ nogo(
         "//tools/analyzers/featureconfig:go_default_library",
         "//tools/analyzers/gocognit:go_default_library",
         "//tools/analyzers/ineffassign:go_default_library",
-        "//tools/analyzers/httperror:go_default_library",
         "//tools/analyzers/interfacechecker:go_default_library",
         "//tools/analyzers/logcapitalization:go_default_library",
         "//tools/analyzers/logruswitherror:go_default_library",

CHANGELOG.md

@@ -4,91 +4,6 @@ All notable changes to this project will be documented in this file.
 
 The format is based on Keep a Changelog, and this project adheres to Semantic Versioning.
 
-## [v7.1.0](https://github.com/prysmaticlabs/prysm/compare/v7.0.0...v7.1.0) - 2025-12-10
-
-This release includes several key features/fixes. If you are running v7.0.0 then you should update to v7.0.1 or later and remove the flag `--disable-last-epoch-targets`. 
-
-Release highlights:
-
-- Backfill is now supported in Fulu. Backfill from checkpoint sync now supports data columns. Run with `--enable-backfill` when using checkpoint sync.
-- A new node configuration to custody enough data columns to reconstruct blobs. Use flag `--semi-supernode` to custody at least 50% of the data columns.
-- Critical fixes in attestation processing.
-
-A post mortem doc with full details on the mainnet attestation processing issue from December 4th is expected in the coming days. 
-
-### Added
-
-- add fulu support to light client processing. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15995)
-- Record data column gossip KZG batch verification latency in both the pooled worker and fallback paths so the `beacon_kzg_verification_data_column_batch_milliseconds` histogram reflects gossip traffic, annotated with `path` labels to distinguish the sources. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16018)
-- Implement Gloas state. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15611)
-- Add initial configs for the state-diff feature. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15903)
-- Add kv functions for the state-diff feature. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15903)
-- Add supported version for fork versions. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16030)
-- prometheus metric `gossip_attestation_verification_milliseconds` to track attestation gossip topic validation latency. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15785)
-- Integrate state-diff into `State()`. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16033)
-- Implement Gloas fork support in consensus-types/blocks with factory methods, getters, setters, and proto handling. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15618)
-- Integrate state-diff into `HasState()`. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16045)
-- Added `--semi-supernode` flag to custody half of a super node's datacolumn requirements but allowing for reconstruction for blob retrieval. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16029)
-- Data column backfill. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15580)
-- Backfill metrics for columns: backfill_data_column_sidecar_downloaded, backfill_data_column_sidecar_downloaded_bytes, backfill_batch_columns_download_ms, backfill_batch_columns_verify_ms. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15580)
-- prometheus summary `gossip_data_column_sidecar_arrival_milliseconds` to track data column sidecar arrival latency since slot start. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16099)
-
-### Changed
-
-- Improve readability in slashing import and remove duplicated code. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15957)
-- Use dependent root instead of target when possible. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15996)
-- Changed `--subscribe-all-data-subnets` flag to `--supernode` and aliased `--subscribe-all-data-subnets` for existing users. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16012)
-- Use explicit slot component timing configs. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15999)
-- Downgraded log level from INFO to DEBUG on PrepareBeaconProposer updated fee recipients. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15998)
-- Change the logging behaviour of Updated fee recipients to only log count of validators at Debug level and all validator indices at Trace level. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15998)
-- Stop emitting payload attribute events during late block handling when we are not proposing the next slot. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16026)
-- Initialize the `ExecutionRequests` field in gossip block map. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16047)
-- Avoid redundant WithHttpEndpoint when JWT is provided. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16032)
-- Removed dead slot parameter from blobCacheEntry.filter. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16021)
-- Added log prefix to the `genesis` package. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16075)
-- Added log prefix to the `params` package. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16075)
-- `WithGenesisValidatorsRoot`: Use camelCase for log field param. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16075)
-- Move `Origin checkpoint found in db` from WARN to INFO, since it is the expected behaviour. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16075)
-- backfill metrics that changed name and/or histogram buckets: backfill_batch_time_verify -> backfill_batch_verify_ms, backfill_batch_time_waiting -> backfill_batch_waiting_ms, backfill_batch_time_roundtrip -> backfill_batch_roundtrip_ms, backfill_blocks_bytes_downloaded -> backfill_blocks_downloaded_bytes, backfill_batch_time_verify -> backfill_batch_verify_ms, backfill_batch_blocks_time_download -> backfill_batch_blocks_download_ms, backfill_batch_blobs_time_download -> backfill_batch_blobs_download_ms, backfill_blobs_bytes_downloaded -> backfill_blocks_downloaded_bytes. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15580)
-- Move the "Not enough connected peers" (for a given subnet) from WARN to DEBUG. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16087)
-- `blobsDataFromStoredDataColumns`: Ask the use to use the `--supernode` flag and shorten the error mesage. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16097)
-- Introduced flag `--ignore-unviable-attestations` (replaces and deprecates `--disable-last-epoch-targets`) to drop attestations whose target state is not viable; default remains to process them unless explicitly enabled. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16094)
-
-### Removed
-
-- Remove validator cross-client from end-to-end tests. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16025)
-- `NUMBER_OF_COLUMNS` configuration (not in the specification any more, replaced by a preset). [[PR]](https://github.com/prysmaticlabs/prysm/pull/16073)
-- `MAX_CELLS_IN_EXTENDED_MATRIX` configuration (not in the specification any more). [[PR]](https://github.com/prysmaticlabs/prysm/pull/16073)
-
-### Fixed
-
-- Nil check for block if it doesn't exist in the DB in fetchOriginSidecars. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16006)
-- Fix proposals progress bar count [#16020](https://github.com/OffchainLabs/prysm/pull/16020). [[PR]](https://github.com/prysmaticlabs/prysm/pull/16020)
-- Move `BlockGossipReceived` event to the end of gossip validation. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16031)
-- Fix state diff repetitive anchor slot bug. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16037)
-- Check the JWT secret length is exactly 256 bits (32 bytes) as per Engine API specification. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15939)
-- http_error_count now matches the other cases by listing the endpoint name rather than the actual URL requested. This improves metrics cardinality. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16055)
-- Fix array out of bounds in static analyzer. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16058)
-- fixes E2E tests to be able to start from Electra genesis fork or future forks. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16048)
-- Use head state to validate attestations for old blocks if they are compatible. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16095)
-
-## [v7.0.1](https://github.com/prysmaticlabs/prysm/compare/v7.0.0...v7.0.1) - 2025-12-08
-
-This patch release contains 4 cherry-picked changes to address the mainnet attestation processing issue from 2025-12-04. Operators are encouraged to update to this release as soon as practical. As of this release, the feature flag `--disable-last-epoch-targets` has been deprecated and can be safely removed from your node configuration. 
-
-A post mortem doc with full details is expected to be published later this week.
-
-### Changed
-
-- Move the "Not enough connected peers" (for a given subnet) from WARN to DEBUG. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16087)
-- Use dependent root instead of target when possible. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15996)
-- Introduced flag `--ignore-unviable-attestations` (replaces and deprecates `--disable-last-epoch-targets`) to drop attestations whose target state is not viable; default remains to process them unless explicitly enabled. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16094)
-
-### Fixed
-
-- Use head state to validate attestations for old blocks if they are compatible. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16095)
-
-
 ## [v7.0.0](https://github.com/prysmaticlabs/prysm/compare/v6.1.4...v7.0.0) - 2025-11-10
 
 This is our initial mainnet release for the Ethereum mainnet Fulu fork on December 3rd, 2025. All operators MUST update to v7.0.0 or later release prior to the fulu fork epoch `411392`. See the [Ethereum Foundation blog post](https://blog.ethereum.org/2025/11/06/fusaka-mainnet-announcement) for more information on Fulu.

beacon-chain/blockchain/kzg/kzg.go

@@ -110,7 +110,7 @@ func VerifyCellKZGProofBatch(commitmentsBytes []Bytes48, cellIndices []uint64, c
 	ckzgCells := make([]ckzg4844.Cell, len(cells))
 
 	for i := range cells {
-		ckzgCells[i] = ckzg4844.Cell(cells[i])
+		copy(ckzgCells[i][:], cells[i][:])
 	}
 	return ckzg4844.VerifyCellKZGProofBatch(commitmentsBytes, cellIndices, ckzgCells, proofsBytes)
 }

beacon-chain/blockchain/process_attestation_helpers.go

@@ -22,7 +22,10 @@ import (
 // The caller of this function must have a lock on forkchoice.
 func (s *Service) getRecentPreState(ctx context.Context, c *ethpb.Checkpoint) state.ReadOnlyBeaconState {
 	headEpoch := slots.ToEpoch(s.HeadSlot())
-	if c.Epoch < headEpoch || c.Epoch == 0 {
+	if c.Epoch < headEpoch {
+		return nil
+	}
+	if !s.cfg.ForkChoiceStore.IsCanonical([32]byte(c.Root)) {
 		return nil
 	}
 	// Only use head state if the head state is compatible with the target checkpoint.
@@ -30,11 +33,11 @@ func (s *Service) getRecentPreState(ctx context.Context, c *ethpb.Checkpoint) st
 	if err != nil {
 		return nil
 	}
-	headDependent, err := s.cfg.ForkChoiceStore.DependentRootForEpoch([32]byte(headRoot), c.Epoch-1)
+	headDependent, err := s.cfg.ForkChoiceStore.DependentRootForEpoch([32]byte(headRoot), c.Epoch)
 	if err != nil {
 		return nil
 	}
-	targetDependent, err := s.cfg.ForkChoiceStore.DependentRootForEpoch([32]byte(c.Root), c.Epoch-1)
+	targetDependent, err := s.cfg.ForkChoiceStore.DependentRootForEpoch([32]byte(c.Root), c.Epoch)
 	if err != nil {
 		return nil
 	}
@@ -50,11 +53,7 @@ func (s *Service) getRecentPreState(ctx context.Context, c *ethpb.Checkpoint) st
 		}
 		return st
 	}
-	// At this point we can only have c.Epoch > headEpoch.
-	if !s.cfg.ForkChoiceStore.IsCanonical([32]byte(c.Root)) {
-		return nil
-	}
-	// Advance the head state to the start of the target epoch.
+	// Otherwise we need to advance the head state to the start of the target epoch.
 	// This point can only be reached if c.Root == headRoot and c.Epoch > headEpoch.
 	slot, err := slots.EpochStart(c.Epoch)
 	if err != nil {

beacon-chain/blockchain/process_attestation_test.go

@@ -181,123 +181,6 @@ func TestService_GetRecentPreState(t *testing.T) {
 	require.NotNil(t, service.getRecentPreState(ctx, &ethpb.Checkpoint{Epoch: 1, Root: ckRoot}))
 }
 
-func TestService_GetRecentPreState_Epoch_0(t *testing.T) {
-	service, _ := minimalTestService(t)
-	ctx := t.Context()
-	require.IsNil(t, service.getRecentPreState(ctx, &ethpb.Checkpoint{}))
-}
-
-func TestService_GetRecentPreState_Old_Checkpoint(t *testing.T) {
-	service, _ := minimalTestService(t)
-	ctx := t.Context()
-	s, err := util.NewBeaconState()
-	require.NoError(t, err)
-	ckRoot := bytesutil.PadTo([]byte{'A'}, fieldparams.RootLength)
-	cp0 := &ethpb.Checkpoint{Epoch: 0, Root: ckRoot}
-	err = s.SetFinalizedCheckpoint(cp0)
-	require.NoError(t, err)
-
-	st, root, err := prepareForkchoiceState(ctx, 33, [32]byte(ckRoot), [32]byte{}, [32]byte{'R'}, cp0, cp0)
-	require.NoError(t, err)
-	require.NoError(t, service.cfg.ForkChoiceStore.InsertNode(ctx, st, root))
-	service.head = &head{
-		root:  [32]byte(ckRoot),
-		state: s,
-		slot:  33,
-	}
-	require.IsNil(t, service.getRecentPreState(ctx, &ethpb.Checkpoint{}))
-}
-
-func TestService_GetRecentPreState_Same_DependentRoots(t *testing.T) {
-	service, _ := minimalTestService(t)
-	ctx := t.Context()
-	s, err := util.NewBeaconState()
-	require.NoError(t, err)
-	ckRoot := bytesutil.PadTo([]byte{'A'}, fieldparams.RootLength)
-	cp0 := &ethpb.Checkpoint{Epoch: 0, Root: ckRoot}
-
-	// Create a fork 31 <-- 32 <--- 64
-	//                 \---------33
-	// With the same dependent root at epoch 0 for a checkpoint at epoch 2
-	st, blk, err := prepareForkchoiceState(ctx, 31, [32]byte(ckRoot), [32]byte{}, [32]byte{}, cp0, cp0)
-	require.NoError(t, err)
-	require.NoError(t, service.cfg.ForkChoiceStore.InsertNode(ctx, st, blk))
-	st, blk, err = prepareForkchoiceState(ctx, 32, [32]byte{'S'}, blk.Root(), [32]byte{}, cp0, cp0)
-	require.NoError(t, err)
-	require.NoError(t, service.cfg.ForkChoiceStore.InsertNode(ctx, st, blk))
-	st, blk, err = prepareForkchoiceState(ctx, 64, [32]byte{'T'}, blk.Root(), [32]byte{}, cp0, cp0)
-	require.NoError(t, err)
-	require.NoError(t, service.cfg.ForkChoiceStore.InsertNode(ctx, st, blk))
-	st, blk, err = prepareForkchoiceState(ctx, 33, [32]byte{'U'}, [32]byte(ckRoot), [32]byte{}, cp0, cp0)
-	require.NoError(t, err)
-	require.NoError(t, service.cfg.ForkChoiceStore.InsertNode(ctx, st, blk))
-	cpRoot := blk.Root()
-
-	service.head = &head{
-		root:  [32]byte{'T'},
-		state: s,
-		slot:  64,
-	}
-	require.NotNil(t, service.getRecentPreState(ctx, &ethpb.Checkpoint{Epoch: 2, Root: cpRoot[:]}))
-}
-
-func TestService_GetRecentPreState_Different_DependentRoots(t *testing.T) {
-	service, _ := minimalTestService(t)
-	ctx := t.Context()
-	s, err := util.NewBeaconState()
-	require.NoError(t, err)
-	ckRoot := bytesutil.PadTo([]byte{'A'}, fieldparams.RootLength)
-	cp0 := &ethpb.Checkpoint{Epoch: 0, Root: ckRoot}
-
-	// Create a fork 30 <-- 31 <-- 32 <--- 64
-	//                 \---------33
-	// With the same dependent root at epoch 0 for a checkpoint at epoch 2
-	st, blk, err := prepareForkchoiceState(ctx, 30, [32]byte(ckRoot), [32]byte{}, [32]byte{}, cp0, cp0)
-	require.NoError(t, err)
-	require.NoError(t, service.cfg.ForkChoiceStore.InsertNode(ctx, st, blk))
-	st, blk, err = prepareForkchoiceState(ctx, 31, [32]byte{'S'}, blk.Root(), [32]byte{}, cp0, cp0)
-	require.NoError(t, err)
-	require.NoError(t, service.cfg.ForkChoiceStore.InsertNode(ctx, st, blk))
-	st, blk, err = prepareForkchoiceState(ctx, 32, [32]byte{'T'}, blk.Root(), [32]byte{}, cp0, cp0)
-	require.NoError(t, err)
-	require.NoError(t, service.cfg.ForkChoiceStore.InsertNode(ctx, st, blk))
-	st, blk, err = prepareForkchoiceState(ctx, 64, [32]byte{'U'}, blk.Root(), [32]byte{}, cp0, cp0)
-	require.NoError(t, err)
-	require.NoError(t, service.cfg.ForkChoiceStore.InsertNode(ctx, st, blk))
-	st, blk, err = prepareForkchoiceState(ctx, 33, [32]byte{'V'}, [32]byte(ckRoot), [32]byte{}, cp0, cp0)
-	require.NoError(t, err)
-	require.NoError(t, service.cfg.ForkChoiceStore.InsertNode(ctx, st, blk))
-	cpRoot := blk.Root()
-
-	service.head = &head{
-		root:  [32]byte{'T'},
-		state: s,
-		slot:  64,
-	}
-	require.IsNil(t, service.getRecentPreState(ctx, &ethpb.Checkpoint{Epoch: 2, Root: cpRoot[:]}))
-}
-
-func TestService_GetRecentPreState_Different(t *testing.T) {
-	service, _ := minimalTestService(t)
-	ctx := t.Context()
-	s, err := util.NewBeaconState()
-	require.NoError(t, err)
-	ckRoot := bytesutil.PadTo([]byte{'A'}, fieldparams.RootLength)
-	cp0 := &ethpb.Checkpoint{Epoch: 0, Root: ckRoot}
-	err = s.SetFinalizedCheckpoint(cp0)
-	require.NoError(t, err)
-
-	st, root, err := prepareForkchoiceState(ctx, 33, [32]byte(ckRoot), [32]byte{}, [32]byte{'R'}, cp0, cp0)
-	require.NoError(t, err)
-	require.NoError(t, service.cfg.ForkChoiceStore.InsertNode(ctx, st, root))
-	service.head = &head{
-		root:  [32]byte(ckRoot),
-		state: s,
-		slot:  33,
-	}
-	require.IsNil(t, service.getRecentPreState(ctx, &ethpb.Checkpoint{}))
-}
-
 func TestService_GetAttPreState_Concurrency(t *testing.T) {
 	service, _ := minimalTestService(t)
 	ctx := t.Context()

beacon-chain/blockchain/process_block.go

@@ -634,7 +634,9 @@ func missingDataColumnIndices(store *filesystem.DataColumnStorage, root [fieldpa
 		return nil, nil
 	}
 
-	if len(expected) > fieldparams.NumberOfColumns {
+	numberOfColumns := params.BeaconConfig().NumberOfColumns
+
+	if uint64(len(expected)) > numberOfColumns {
 		return nil, errMaxDataColumnsExceeded
 	}
 
@@ -816,9 +818,10 @@ func (s *Service) areDataColumnsAvailable(
 
 		case <-ctx.Done():
 			var missingIndices any = "all"
-			missingIndicesCount := len(missing)
+			numberOfColumns := params.BeaconConfig().NumberOfColumns
+			missingIndicesCount := uint64(len(missing))
 
-			if missingIndicesCount < fieldparams.NumberOfColumns {
+			if missingIndicesCount < numberOfColumns {
 				missingIndices = helpers.SortedPrettySliceFromMap(missing)
 			}
 

beacon-chain/blockchain/process_block_test.go

@@ -2495,8 +2495,7 @@ func TestMissingBlobIndices(t *testing.T) {
 }
 
 func TestMissingDataColumnIndices(t *testing.T) {
-	const countPlusOne = fieldparams.NumberOfColumns + 1
-
+	countPlusOne := params.BeaconConfig().NumberOfColumns + 1
 	tooManyColumns := make(map[uint64]bool, countPlusOne)
 	for i := range countPlusOne {
 		tooManyColumns[uint64(i)] = true

beacon-chain/blockchain/service_test.go

@@ -603,6 +603,7 @@ func TestUpdateCustodyInfoInDB(t *testing.T) {
 		custodyRequirement    = uint64(4)
 		earliestStoredSlot    = primitives.Slot(12)
 		numberOfCustodyGroups = uint64(64)
+		numberOfColumns       = uint64(128)
 	)
 
 	params.SetupTestConfigCleanup(t)
@@ -610,6 +611,7 @@ func TestUpdateCustodyInfoInDB(t *testing.T) {
 	cfg.FuluForkEpoch = fuluForkEpoch
 	cfg.CustodyRequirement = custodyRequirement
 	cfg.NumberOfCustodyGroups = numberOfCustodyGroups
+	cfg.NumberOfColumns = numberOfColumns
 	params.OverrideBeaconConfig(cfg)
 
 	ctx := t.Context()

beacon-chain/blockchain/setup_test.go

@@ -89,7 +89,7 @@ func (mb *mockBroadcaster) BroadcastLightClientFinalityUpdate(_ context.Context,
 	return nil
 }
 
-func (mb *mockBroadcaster) BroadcastDataColumnSidecars(_ context.Context, _ []blocks.VerifiedRODataColumn, _ []blocks.PartialDataColumn) error {
+func (mb *mockBroadcaster) BroadcastDataColumnSidecars(_ context.Context, _ []blocks.VerifiedRODataColumn) error {
 	mb.broadcastCalled = true
 	return nil
 }

beacon-chain/core/blocks/eth1_data.go

@@ -60,7 +60,7 @@ func Eth1DataHasEnoughSupport(beaconState state.ReadOnlyBeaconState, data *ethpb
 	voteCount := uint64(0)
 
 	for _, vote := range beaconState.Eth1DataVotes() {
-		if AreEth1DataEqual(vote, data) {
+		if AreEth1DataEqual(vote, data.Copy()) {
 			voteCount++
 		}
 	}

beacon-chain/core/helpers/validators.go

@@ -152,7 +152,7 @@ func ActiveValidatorIndices(ctx context.Context, s state.ReadOnlyBeaconState, ep
 	}
 
 	if err := UpdateCommitteeCache(ctx, s, epoch); err != nil {
-		log.WithError(err).Error("Could not update committee cache")
+		return nil, errors.Wrap(err, "could not update committee cache")
 	}
 
 	return indices, nil

beacon-chain/core/peerdas/BUILD.bazel

@@ -33,7 +33,6 @@ go_library(
         "@com_github_pkg_errors//:go_default_library",
         "@com_github_prometheus_client_golang//prometheus:go_default_library",
         "@com_github_prometheus_client_golang//prometheus/promauto:go_default_library",
-        "@com_github_prysmaticlabs_go_bitfield//:go_default_library",
         "@org_golang_x_sync//errgroup:go_default_library",
     ],
 )

beacon-chain/core/peerdas/das_core.go

@@ -5,7 +5,6 @@ import (
 	"math"
 	"slices"
 
-	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
 	"github.com/OffchainLabs/prysm/v7/config/params"
 	"github.com/OffchainLabs/prysm/v7/crypto/hash"
 	"github.com/OffchainLabs/prysm/v7/encoding/bytesutil"
@@ -97,7 +96,8 @@ func ComputeColumnsForCustodyGroup(custodyGroup uint64) ([]uint64, error) {
 		return nil, ErrCustodyGroupTooLarge
 	}
 
-	numberOfColumns := uint64(fieldparams.NumberOfColumns)
+	numberOfColumns := cfg.NumberOfColumns
+
 	columnsPerGroup := numberOfColumns / numberOfCustodyGroups
 
 	columns := make([]uint64, 0, columnsPerGroup)
@@ -112,9 +112,8 @@ func ComputeColumnsForCustodyGroup(custodyGroup uint64) ([]uint64, error) {
 // ComputeCustodyGroupForColumn computes the custody group for a given column.
 // It is the reciprocal function of ComputeColumnsForCustodyGroup.
 func ComputeCustodyGroupForColumn(columnIndex uint64) (uint64, error) {
-	const numberOfColumns = fieldparams.NumberOfColumns
-
 	cfg := params.BeaconConfig()
+	numberOfColumns := cfg.NumberOfColumns
 	numberOfCustodyGroups := cfg.NumberOfCustodyGroups
 
 	if columnIndex >= numberOfColumns {

beacon-chain/core/peerdas/das_core_test.go

@@ -30,6 +30,7 @@ func TestComputeColumnsForCustodyGroup(t *testing.T) {
 func TestComputeCustodyGroupForColumn(t *testing.T) {
 	params.SetupTestConfigCleanup(t)
 	config := params.BeaconConfig()
+	config.NumberOfColumns = 128
 	config.NumberOfCustodyGroups = 64
 	params.OverrideBeaconConfig(config)
 

beacon-chain/core/peerdas/metrics.go

@@ -5,20 +5,10 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 )
 
-var (
-	dataColumnComputationTime = promauto.NewHistogram(
-		prometheus.HistogramOpts{
-			Name:    "beacon_data_column_sidecar_computation_milliseconds",
-			Help:    "Captures the time taken to compute data column sidecars from blobs.",
-			Buckets: []float64{25, 50, 100, 250, 500, 750, 1000},
-		},
-	)
-
-	cellsAndProofsFromStructuredComputationTime = promauto.NewHistogram(
-		prometheus.HistogramOpts{
-			Name:    "cells_and_proofs_from_structured_computation_milliseconds",
-			Help:    "Captures the time taken to compute cells and proofs from structured computation.",
-			Buckets: []float64{10, 20, 30, 40, 50, 100, 200},
-		},
-	)
+var dataColumnComputationTime = promauto.NewHistogram(
+	prometheus.HistogramOpts{
+		Name:    "beacon_data_column_sidecar_computation_milliseconds",
+		Help:    "Captures the time taken to compute data column sidecars from blobs.",
+		Buckets: []float64{25, 50, 100, 250, 500, 750, 1000},
+	},
 )

beacon-chain/core/peerdas/p2p_interface.go

@@ -1,15 +1,11 @@
 package peerdas
 
 import (
-	stderrors "errors"
-	"iter"
-
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain/kzg"
 	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
 	"github.com/OffchainLabs/prysm/v7/config/params"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
 	"github.com/OffchainLabs/prysm/v7/container/trie"
-	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
 	"github.com/ethereum/go-ethereum/p2p/enr"
 	"github.com/pkg/errors"
 )
@@ -20,7 +16,6 @@ var (
 	ErrIndexTooLarge               = errors.New("column index is larger than the specified columns count")
 	ErrNoKzgCommitments            = errors.New("no KZG commitments found")
 	ErrMismatchLength              = errors.New("mismatch in the length of the column, commitments or proofs")
-	ErrEmptySegment                = errors.New("empty segment in batch")
 	ErrInvalidKZGProof             = errors.New("invalid KZG proof")
 	ErrBadRootLength               = errors.New("bad root length")
 	ErrInvalidInclusionProof       = errors.New("invalid inclusion proof")
@@ -38,7 +33,8 @@ func (Cgc) ENRKey() string { return params.BeaconNetworkConfig().CustodyGroupCou
 // https://github.com/ethereum/consensus-specs/blob/master/specs/fulu/p2p-interface.md#verify_data_column_sidecar
 func VerifyDataColumnSidecar(sidecar blocks.RODataColumn) error {
 	// The sidecar index must be within the valid range.
-	if sidecar.Index >= fieldparams.NumberOfColumns {
+	numberOfColumns := params.BeaconConfig().NumberOfColumns
+	if sidecar.Index >= numberOfColumns {
 		return ErrIndexTooLarge
 	}
 
@@ -62,136 +58,62 @@ func VerifyDataColumnSidecar(sidecar blocks.RODataColumn) error {
 	return nil
 }
 
-// CellProofBundleSegment is returned when a batch fails. The caller can call
-// the `.Verify` method to verify just this segment.
-type CellProofBundleSegment struct {
-	indices     []uint64      // column index i.e in [0-127] for each cell
-	commitments []kzg.Bytes48 // KZG commitment for the blob that contains the cell that needs to be verified
-	cells       []kzg.Cell    // actual cell data for each cell (2KB each)
-	proofs      []kzg.Bytes48 // Cell proof for each cell
-	// Note: All the above arrays are of the same length.
-}
-
-// Verify verifies this segment without batching.
-func (s CellProofBundleSegment) Verify() error {
-	if len(s.cells) == 0 {
-		return ErrEmptySegment
-	}
-	verified, err := kzg.VerifyCellKZGProofBatch(s.commitments, s.indices, s.cells, s.proofs)
-	if err != nil {
-		return stderrors.Join(err, ErrInvalidKZGProof)
-	}
-	if !verified {
-		return ErrInvalidKZGProof
-	}
-	return nil
-}
-
-func VerifyDataColumnsCellsKZGProofs(sizeHint int, cellProofsIter iter.Seq[blocks.CellProofBundle]) error {
-	// The BatchVerifyDataColumnsCellsKZGProofs call also returns the failed segments if beatch verification fails
-	// so we can verify each segment seperately to identify the failed cell ("BatchVerifyDataColumnsCellsKZGProofs" is all or nothing).
-	// But we ignore the failed segment list since we are just passing in one segment.
-	_, err := BatchVerifyDataColumnsCellsKZGProofs(sizeHint, []iter.Seq[blocks.CellProofBundle]{cellProofsIter})
-	return err
-}
-
-// BatchVerifyDataColumnsCellsKZGProofs verifies if the KZG proofs are correct.
+// VerifyDataColumnsSidecarKZGProofs verifies if the KZG proofs are correct.
 // Note: We are slightly deviating from the specification here:
 // The specification verifies the KZG proofs for each sidecar separately,
 // while we are verifying all the KZG proofs from multiple sidecars in a batch.
 // This is done to improve performance since the internal KZG library is way more
-// efficient when verifying in batch. If the batch fails, the failed segments
-// are returned to the caller so that they may try segment by segment without
-// batching. On success the failed segment list is empty.
-//
+// efficient when verifying in batch.
 // https://github.com/ethereum/consensus-specs/blob/master/specs/fulu/p2p-interface.md#verify_data_column_sidecar_kzg_proofs
-// sizeHint is the total number of cells to verify across all sidecars.
-func BatchVerifyDataColumnsCellsKZGProofs(sizeHint int, cellProofsIters []iter.Seq[blocks.CellProofBundle]) ( /* failed segment list */ []CellProofBundleSegment, error) {
-	commitments := make([]kzg.Bytes48, 0, sizeHint)
-	indices := make([]uint64, 0, sizeHint)
-	cells := make([]kzg.Cell, 0, sizeHint)
-	proofs := make([]kzg.Bytes48, 0, sizeHint)
+func VerifyDataColumnsSidecarKZGProofs(sidecars []blocks.RODataColumn) error {
+	// Compute the total count.
+	count := 0
+	for _, sidecar := range sidecars {
+		count += len(sidecar.Column)
+	}
 
-	var anySegmentEmpty bool
-	var segments []CellProofBundleSegment
-	for _, cellProofsIter := range cellProofsIters {
-		startIdx := len(cells) // starts at 0, and increments by the number of cells in the current segment
-		for bundle := range cellProofsIter {
+	commitments := make([]kzg.Bytes48, 0, count)
+	indices := make([]uint64, 0, count)
+	cells := make([]kzg.Cell, 0, count)
+	proofs := make([]kzg.Bytes48, 0, count)
+
+	for _, sidecar := range sidecars {
+		for i := range sidecar.Column {
 			var (
 				commitment kzg.Bytes48
 				cell       kzg.Cell
 				proof      kzg.Bytes48
 			)
 
-			// sanity check that each commitment is 48 bytes, each cell is 2KB and each proof is 48 bytes.
-			if len(bundle.Commitment) != len(commitment) ||
-				len(bundle.Cell) != len(cell) ||
-				len(bundle.Proof) != len(proof) {
-				return nil, ErrMismatchLength
+			commitmentBytes := sidecar.KzgCommitments[i]
+			cellBytes := sidecar.Column[i]
+			proofBytes := sidecar.KzgProofs[i]
+
+			if len(commitmentBytes) != len(commitment) ||
+				len(cellBytes) != len(cell) ||
+				len(proofBytes) != len(proof) {
+				return ErrMismatchLength
 			}
 
-			copy(commitment[:], bundle.Commitment)
-			copy(cell[:], bundle.Cell)
-			copy(proof[:], bundle.Proof)
+			copy(commitment[:], commitmentBytes)
+			copy(cell[:], cellBytes)
+			copy(proof[:], proofBytes)
 
 			commitments = append(commitments, commitment)
-			indices = append(indices, bundle.ColumnIndex)
+			indices = append(indices, sidecar.Index)
 			cells = append(cells, cell)
 			proofs = append(proofs, proof)
 		}
-		if len(cells[startIdx:]) == 0 {
-			// this basically means that the current segment i.e. the current  "bundle" we're
-			// iterating over is empty because "startIdx" did not increment after the last iteration.
-			anySegmentEmpty = true
-		}
-		segments = append(segments, CellProofBundleSegment{
-			indices:     indices[startIdx:],
-			commitments: commitments[startIdx:],
-			cells:       cells[startIdx:],
-			proofs:      proofs[startIdx:],
-		})
-	}
-
-	if anySegmentEmpty {
-		// if any segment is empty, we don't verify anything as it's a malformed batch and we return all
-		// the segments here so caller can verify each segment seperately.
-		return segments, ErrEmptySegment
 	}
 
 	// Batch verify that the cells match the corresponding commitments and proofs.
 	verified, err := kzg.VerifyCellKZGProofBatch(commitments, indices, cells, proofs)
 	if err != nil {
-		return segments, stderrors.Join(err, ErrInvalidKZGProof)
+		return errors.Wrap(err, "verify cell KZG proof batch")
 	}
 
 	if !verified {
-		return segments, ErrInvalidKZGProof
-	}
-
-	return nil, nil
-}
-
-// verifyKzgCommitmentsInclusionProof is the shared implementation for inclusion proof verification.
-func verifyKzgCommitmentsInclusionProof(bodyRoot []byte, kzgCommitments [][]byte, inclusionProof [][]byte) error {
-	if len(bodyRoot) != fieldparams.RootLength {
-		return ErrBadRootLength
-	}
-
-	leaves := blocks.LeavesFromCommitments(kzgCommitments)
-
-	sparse, err := trie.GenerateTrieFromItems(leaves, fieldparams.LogMaxBlobCommitments)
-	if err != nil {
-		return errors.Wrap(err, "generate trie from items")
-	}
-
-	hashTreeRoot, err := sparse.HashTreeRoot()
-	if err != nil {
-		return errors.Wrap(err, "hash tree root")
-	}
-
-	verified := trie.VerifyMerkleProof(bodyRoot, hashTreeRoot[:], kzgPosition, inclusionProof)
-	if !verified {
-		return ErrInvalidInclusionProof
+		return ErrInvalidKZGProof
 	}
 
 	return nil
@@ -203,23 +125,30 @@ func VerifyDataColumnSidecarInclusionProof(sidecar blocks.RODataColumn) error {
 	if sidecar.SignedBlockHeader == nil || sidecar.SignedBlockHeader.Header == nil {
 		return ErrNilBlockHeader
 	}
-	return verifyKzgCommitmentsInclusionProof(
-		sidecar.SignedBlockHeader.Header.BodyRoot,
-		sidecar.KzgCommitments,
-		sidecar.KzgCommitmentsInclusionProof,
-	)
-}
 
-// VerifyPartialDataColumnHeaderInclusionProof verifies if the KZG commitments are included in the beacon block.
-func VerifyPartialDataColumnHeaderInclusionProof(header *ethpb.PartialDataColumnHeader) error {
-	if header.SignedBlockHeader == nil || header.SignedBlockHeader.Header == nil {
-		return ErrNilBlockHeader
+	root := sidecar.SignedBlockHeader.Header.BodyRoot
+	if len(root) != fieldparams.RootLength {
+		return ErrBadRootLength
 	}
-	return verifyKzgCommitmentsInclusionProof(
-		header.SignedBlockHeader.Header.BodyRoot,
-		header.KzgCommitments,
-		header.KzgCommitmentsInclusionProof,
-	)
+
+	leaves := blocks.LeavesFromCommitments(sidecar.KzgCommitments)
+
+	sparse, err := trie.GenerateTrieFromItems(leaves, fieldparams.LogMaxBlobCommitments)
+	if err != nil {
+		return errors.Wrap(err, "generate trie from items")
+	}
+
+	hashTreeRoot, err := sparse.HashTreeRoot()
+	if err != nil {
+		return errors.Wrap(err, "hash tree root")
+	}
+
+	verified := trie.VerifyMerkleProof(root, hashTreeRoot[:], kzgPosition, sidecar.KzgCommitmentsInclusionProof)
+	if !verified {
+		return ErrInvalidInclusionProof
+	}
+
+	return nil
 }
 
 // ComputeSubnetForDataColumnSidecar computes the subnet for a data column sidecar.

beacon-chain/core/peerdas/p2p_interface_test.go

@@ -3,7 +3,6 @@ package peerdas_test
 import (
 	"crypto/rand"
 	"fmt"
-	"iter"
 	"testing"
 
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain/kzg"
@@ -73,7 +72,7 @@ func TestVerifyDataColumnSidecarKZGProofs(t *testing.T) {
 		sidecars := generateRandomSidecars(t, seed, blobCount)
 		sidecars[0].Column[0] = sidecars[0].Column[0][:len(sidecars[0].Column[0])-1] // Remove one byte to create size mismatch
 
-		err := peerdas.VerifyDataColumnsCellsKZGProofs(0, blocks.RODataColumnsToCellProofBundles(sidecars))
+		err := peerdas.VerifyDataColumnsSidecarKZGProofs(sidecars)
 		require.ErrorIs(t, err, peerdas.ErrMismatchLength)
 	})
 
@@ -81,15 +80,14 @@ func TestVerifyDataColumnSidecarKZGProofs(t *testing.T) {
 		sidecars := generateRandomSidecars(t, seed, blobCount)
 		sidecars[0].Column[0][0]++ // It is OK to overflow
 
-		err := peerdas.VerifyDataColumnsCellsKZGProofs(0, blocks.RODataColumnsToCellProofBundles(sidecars))
+		err := peerdas.VerifyDataColumnsSidecarKZGProofs(sidecars)
 		require.ErrorIs(t, err, peerdas.ErrInvalidKZGProof)
 	})
 
 	t.Run("nominal", func(t *testing.T) {
 		sidecars := generateRandomSidecars(t, seed, blobCount)
-		failedSegments, err := peerdas.BatchVerifyDataColumnsCellsKZGProofs(blobCount, []iter.Seq[blocks.CellProofBundle]{blocks.RODataColumnsToCellProofBundles(sidecars)})
+		err := peerdas.VerifyDataColumnsSidecarKZGProofs(sidecars)
 		require.NoError(t, err)
-		require.Equal(t, 0, len(failedSegments))
 	})
 }
 
@@ -275,7 +273,7 @@ func BenchmarkVerifyDataColumnSidecarKZGProofs_SameCommitments_NoBatch(b *testin
 		for _, sidecar := range sidecars {
 			sidecars := []blocks.RODataColumn{sidecar}
 			b.StartTimer()
-			err := peerdas.VerifyDataColumnsCellsKZGProofs(0, blocks.RODataColumnsToCellProofBundles(sidecars))
+			err := peerdas.VerifyDataColumnsSidecarKZGProofs(sidecars)
 			b.StopTimer()
 			require.NoError(b, err)
 		}
@@ -283,11 +281,8 @@ func BenchmarkVerifyDataColumnSidecarKZGProofs_SameCommitments_NoBatch(b *testin
 }
 
 func BenchmarkVerifyDataColumnSidecarKZGProofs_DiffCommitments_Batch(b *testing.B) {
-	const (
-		blobCount       = 12
-		numberOfColumns = fieldparams.NumberOfColumns
-	)
-
+	const blobCount = 12
+	numberOfColumns := int64(params.BeaconConfig().NumberOfColumns)
 	err := kzg.Start()
 	require.NoError(b, err)
 
@@ -310,7 +305,7 @@ func BenchmarkVerifyDataColumnSidecarKZGProofs_DiffCommitments_Batch(b *testing.
 				}
 
 				b.StartTimer()
-				err := peerdas.VerifyDataColumnsCellsKZGProofs(0, blocks.RODataColumnsToCellProofBundles(allSidecars))
+				err := peerdas.VerifyDataColumnsSidecarKZGProofs(allSidecars)
 				b.StopTimer()
 				require.NoError(b, err)
 			}
@@ -343,7 +338,7 @@ func BenchmarkVerifyDataColumnSidecarKZGProofs_DiffCommitments_Batch4(b *testing
 
 		for _, sidecars := range allSidecars {
 			b.StartTimer()
-			err := peerdas.VerifyDataColumnsCellsKZGProofs(len(allSidecars), blocks.RODataColumnsToCellProofBundles(sidecars))
+			err := peerdas.VerifyDataColumnsSidecarKZGProofs(sidecars)
 			b.StopTimer()
 			require.NoError(b, err)
 		}

beacon-chain/core/peerdas/reconstruction.go

@@ -3,9 +3,7 @@ package peerdas
 import (
 	"sort"
 	"sync"
-	"time"
 
-	"github.com/OffchainLabs/go-bitfield"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain/kzg"
 	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
 	"github.com/OffchainLabs/prysm/v7/config/params"
@@ -28,7 +26,7 @@ var (
 func MinimumColumnCountToReconstruct() uint64 {
 	// If the number of columns is odd, then we need total / 2 + 1 columns to reconstruct.
 	// If the number of columns is even, then we need total / 2 columns to reconstruct.
-	return (fieldparams.NumberOfColumns + 1) / 2
+	return (params.BeaconConfig().NumberOfColumns + 1) / 2
 }
 
 // MinimumCustodyGroupCountToReconstruct returns the minimum number of custody groups needed to
@@ -36,11 +34,10 @@ func MinimumColumnCountToReconstruct() uint64 {
 // custody groups and columns, making it future-proof if these values change.
 // Returns an error if the configuration values are invalid (zero or would cause division by zero).
 func MinimumCustodyGroupCountToReconstruct() (uint64, error) {
-	const numberOfColumns = fieldparams.NumberOfColumns
 	cfg := params.BeaconConfig()
 
 	// Validate configuration values
-	if numberOfColumns == 0 {
+	if cfg.NumberOfColumns == 0 {
 		return 0, errors.New("NumberOfColumns cannot be zero")
 	}
 	if cfg.NumberOfCustodyGroups == 0 {
@@ -50,13 +47,13 @@ func MinimumCustodyGroupCountToReconstruct() (uint64, error) {
 	minimumColumnCount := MinimumColumnCountToReconstruct()
 
 	// Calculate how many columns each custody group represents
-	columnsPerGroup := numberOfColumns / cfg.NumberOfCustodyGroups
+	columnsPerGroup := cfg.NumberOfColumns / cfg.NumberOfCustodyGroups
 
 	// If there are more groups than columns (columnsPerGroup = 0), this is an invalid configuration
 	// for reconstruction purposes as we cannot determine a meaningful custody group count
 	if columnsPerGroup == 0 {
 		return 0, errors.Errorf("invalid configuration: NumberOfCustodyGroups (%d) exceeds NumberOfColumns (%d)",
-			cfg.NumberOfCustodyGroups, numberOfColumns)
+			cfg.NumberOfCustodyGroups, cfg.NumberOfColumns)
 	}
 
 	// Use ceiling division to ensure we have enough groups to cover the minimum columns
@@ -288,8 +285,7 @@ func ReconstructBlobSidecars(block blocks.ROBlock, verifiedDataColumnSidecars []
 
 // ComputeCellsAndProofsFromFlat computes the cells and proofs from blobs and cell flat proofs.
 func ComputeCellsAndProofsFromFlat(blobs [][]byte, cellProofs [][]byte) ([][]kzg.Cell, [][]kzg.Proof, error) {
-	const numberOfColumns = fieldparams.NumberOfColumns
-
+	numberOfColumns := params.BeaconConfig().NumberOfColumns
 	blobCount := uint64(len(blobs))
 	cellProofsCount := uint64(len(cellProofs))
 
@@ -298,116 +294,78 @@ func ComputeCellsAndProofsFromFlat(blobs [][]byte, cellProofs [][]byte) ([][]kzg
 		return nil, nil, ErrBlobsCellsProofsMismatch
 	}
 
-	var wg errgroup.Group
-
-	cellsPerBlob := make([][]kzg.Cell, blobCount)
-	proofsPerBlob := make([][]kzg.Proof, blobCount)
-
+	cellsPerBlob := make([][]kzg.Cell, 0, blobCount)
+	proofsPerBlob := make([][]kzg.Proof, 0, blobCount)
 	for i, blob := range blobs {
-		wg.Go(func() error {
-			var kzgBlob kzg.Blob
-			if copy(kzgBlob[:], blob) != len(kzgBlob) {
-				return errors.New("wrong blob size - should never happen")
+		var kzgBlob kzg.Blob
+		if copy(kzgBlob[:], blob) != len(kzgBlob) {
+			return nil, nil, errors.New("wrong blob size - should never happen")
+		}
+
+		// Compute the extended cells from the (non-extended) blob.
+		cells, err := kzg.ComputeCells(&kzgBlob)
+		if err != nil {
+			return nil, nil, errors.Wrap(err, "compute cells")
+		}
+
+		var proofs []kzg.Proof
+		for idx := uint64(i) * numberOfColumns; idx < (uint64(i)+1)*numberOfColumns; idx++ {
+			var kzgProof kzg.Proof
+			if copy(kzgProof[:], cellProofs[idx]) != len(kzgProof) {
+				return nil, nil, errors.New("wrong KZG proof size - should never happen")
 			}
 
-			// Compute the extended cells from the (non-extended) blob.
-			cells, err := kzg.ComputeCells(&kzgBlob)
-			if err != nil {
-				return errors.Wrap(err, "compute cells")
-			}
+			proofs = append(proofs, kzgProof)
+		}
 
-			proofs := make([]kzg.Proof, 0, numberOfColumns)
-			for idx := uint64(i) * numberOfColumns; idx < (uint64(i)+1)*numberOfColumns; idx++ {
-				var kzgProof kzg.Proof
-				if copy(kzgProof[:], cellProofs[idx]) != len(kzgProof) {
-					return errors.New("wrong KZG proof size - should never happen")
-				}
-
-				proofs = append(proofs, kzgProof)
-			}
-
-			cellsPerBlob[i] = cells
-			proofsPerBlob[i] = proofs
-			return nil
-		})
-	}
-
-	if err := wg.Wait(); err != nil {
-		return nil, nil, err
+		cellsPerBlob = append(cellsPerBlob, cells)
+		proofsPerBlob = append(proofsPerBlob, proofs)
 	}
 
 	return cellsPerBlob, proofsPerBlob, nil
 }
 
 // ComputeCellsAndProofsFromStructured computes the cells and proofs from blobs and cell proofs.
-// commitmentCount (i.e. number of blobs) is required to return the correct sized bitlist even if we see a nil slice of blobsAndProofs.
-// Returns the bitlist of which blobs are present, the cells for each blob that is present, and the proofs for each cell in each blob that is present.
-// The returned cells and proofs are compacted and will not contain entries for missing blobs.
-func ComputeCellsAndProofsFromStructured(commitmentCount uint64, blobsAndProofs []*pb.BlobAndProofV2) (bitfield.Bitlist /* parts included */, [][]kzg.Cell, [][]kzg.Proof, error) {
-	start := time.Now()
-	defer func() {
-		cellsAndProofsFromStructuredComputationTime.Observe(float64(time.Since(start).Milliseconds()))
-	}()
+func ComputeCellsAndProofsFromStructured(blobsAndProofs []*pb.BlobAndProofV2) ([][]kzg.Cell, [][]kzg.Proof, error) {
+	numberOfColumns := params.BeaconConfig().NumberOfColumns
 
-	var wg errgroup.Group
-
-	var blobsPresent int
+	cellsPerBlob := make([][]kzg.Cell, 0, len(blobsAndProofs))
+	proofsPerBlob := make([][]kzg.Proof, 0, len(blobsAndProofs))
 	for _, blobAndProof := range blobsAndProofs {
-		if blobAndProof != nil {
-			blobsPresent++
-		}
-	}
-	cellsPerBlob := make([][]kzg.Cell, blobsPresent)   // array of cells for each blob
-	proofsPerBlob := make([][]kzg.Proof, blobsPresent) // array of proofs for each blob
-	included := bitfield.NewBitlist(commitmentCount)   // bitlist of which blobs are present
-
-	var j int
-	for i, blobAndProof := range blobsAndProofs {
 		if blobAndProof == nil {
-			continue
+			return nil, nil, ErrNilBlobAndProof
 		}
-		included.SetBitAt(uint64(i), true) // blob at index i is present
 
-		compactIndex := j // compact index is the index of the blob in the returned arrays after accounting for nil/missing blocks.
-		wg.Go(func() error {
-			var kzgBlob kzg.Blob
-			// the number of copied bytes should be equal to the expected size of a blob.
-			if copy(kzgBlob[:], blobAndProof.Blob) != len(kzgBlob) {
-				return errors.New("wrong blob size - should never happen")
+		var kzgBlob kzg.Blob
+		if copy(kzgBlob[:], blobAndProof.Blob) != len(kzgBlob) {
+			return nil, nil, errors.New("wrong blob size - should never happen")
+		}
+
+		// Compute the extended cells from the (non-extended) blob.
+		cells, err := kzg.ComputeCells(&kzgBlob)
+		if err != nil {
+			return nil, nil, errors.Wrap(err, "compute cells")
+		}
+
+		kzgProofs := make([]kzg.Proof, 0, numberOfColumns)
+		for _, kzgProofBytes := range blobAndProof.KzgProofs {
+			if len(kzgProofBytes) != kzg.BytesPerProof {
+				return nil, nil, errors.New("wrong KZG proof size - should never happen")
 			}
 
-			// Compute the extended cells from the (non-extended) blob.
-			cells, err := kzg.ComputeCells(&kzgBlob)
-			if err != nil {
-				return errors.Wrap(err, "compute cells")
+			var kzgProof kzg.Proof
+			if copy(kzgProof[:], kzgProofBytes) != len(kzgProof) {
+				return nil, nil, errors.New("wrong copied KZG proof size - should never happen")
 			}
 
-			kzgProofs := make([]kzg.Proof, 0, fieldparams.NumberOfColumns)
-			for _, kzgProofBytes := range blobAndProof.KzgProofs {
-				if len(kzgProofBytes) != kzg.BytesPerProof {
-					return errors.New("wrong KZG proof size - should never happen")
-				}
+			kzgProofs = append(kzgProofs, kzgProof)
+		}
 
-				var kzgProof kzg.Proof
-				if copy(kzgProof[:], kzgProofBytes) != len(kzgProof) {
-					return errors.New("wrong copied KZG proof size - should never happen")
-				}
-
-				kzgProofs = append(kzgProofs, kzgProof)
-			}
-
-			cellsPerBlob[compactIndex] = cells
-			proofsPerBlob[compactIndex] = kzgProofs
-			return nil
-		})
-		j++
+		cellsPerBlob = append(cellsPerBlob, cells)
+		proofsPerBlob = append(proofsPerBlob, kzgProofs)
 	}
 
-	if err := wg.Wait(); err != nil {
-		return nil, nil, nil, err
-	}
-
-	return included, cellsPerBlob, proofsPerBlob, nil
+	return cellsPerBlob, proofsPerBlob, nil
 }
 
 // ReconstructBlobs reconstructs blobs from data column sidecars without computing KZG proofs or creating sidecars.

beacon-chain/core/peerdas/reconstruction_test.go

@@ -17,9 +17,41 @@ import (
 )
 
 func TestMinimumColumnsCountToReconstruct(t *testing.T) {
-	const expected = uint64(64)
-	actual := peerdas.MinimumColumnCountToReconstruct()
-	require.Equal(t, expected, actual)
+	testCases := []struct {
+		name            string
+		numberOfColumns uint64
+		expected        uint64
+	}{
+		{
+			name:            "numberOfColumns=128",
+			numberOfColumns: 128,
+			expected:        64,
+		},
+		{
+			name:            "numberOfColumns=129",
+			numberOfColumns: 129,
+			expected:        65,
+		},
+		{
+			name:            "numberOfColumns=130",
+			numberOfColumns: 130,
+			expected:        65,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// Set the total number of columns.
+			params.SetupTestConfigCleanup(t)
+			cfg := params.BeaconConfig().Copy()
+			cfg.NumberOfColumns = tc.numberOfColumns
+			params.OverrideBeaconConfig(cfg)
+
+			// Compute the minimum number of columns needed to reconstruct.
+			actual := peerdas.MinimumColumnCountToReconstruct()
+			require.Equal(t, tc.expected, actual)
+		})
+	}
 }
 
 func TestReconstructDataColumnSidecars(t *testing.T) {
@@ -168,6 +200,7 @@ func TestReconstructBlobSidecars(t *testing.T) {
 
 	t.Run("nominal", func(t *testing.T) {
 		const blobCount = 3
+		numberOfColumns := params.BeaconConfig().NumberOfColumns
 
 		roBlock, roBlobSidecars := util.GenerateTestElectraBlockWithSidecar(t, [fieldparams.RootLength]byte{}, 42, blobCount)
 
@@ -203,7 +236,7 @@ func TestReconstructBlobSidecars(t *testing.T) {
 		require.NoError(t, err)
 
 		// Flatten proofs.
-		cellProofs := make([][]byte, 0, blobCount*fieldparams.NumberOfColumns)
+		cellProofs := make([][]byte, 0, blobCount*numberOfColumns)
 		for _, proofs := range inputProofsPerBlob {
 			for _, proof := range proofs {
 				cellProofs = append(cellProofs, proof[:])
@@ -395,12 +428,13 @@ func TestReconstructBlobs(t *testing.T) {
 }
 
 func TestComputeCellsAndProofsFromFlat(t *testing.T) {
-	const numberOfColumns = fieldparams.NumberOfColumns
 	// Start the trusted setup.
 	err := kzg.Start()
 	require.NoError(t, err)
 
 	t.Run("mismatched blob and proof counts", func(t *testing.T) {
+		numberOfColumns := params.BeaconConfig().NumberOfColumns
+
 		// Create one blob but proofs for two blobs
 		blobs := [][]byte{{}}
 
@@ -413,6 +447,7 @@ func TestComputeCellsAndProofsFromFlat(t *testing.T) {
 
 	t.Run("nominal", func(t *testing.T) {
 		const blobCount = 2
+		numberOfColumns := params.BeaconConfig().NumberOfColumns
 
 		// Generate test blobs
 		_, roBlobSidecars := util.GenerateTestElectraBlockWithSidecar(t, [fieldparams.RootLength]byte{}, 42, blobCount)
@@ -479,9 +514,8 @@ func TestComputeCellsAndProofsFromFlat(t *testing.T) {
 
 func TestComputeCellsAndProofsFromStructured(t *testing.T) {
 	t.Run("nil blob and proof", func(t *testing.T) {
-		included, _, _, err := peerdas.ComputeCellsAndProofsFromStructured(0, []*pb.BlobAndProofV2{nil})
-		require.NoError(t, err)
-		require.Equal(t, uint64(0), included.Count())
+		_, _, err := peerdas.ComputeCellsAndProofsFromStructured([]*pb.BlobAndProofV2{nil})
+		require.ErrorIs(t, err, peerdas.ErrNilBlobAndProof)
 	})
 
 	t.Run("nominal", func(t *testing.T) {
@@ -534,8 +568,7 @@ func TestComputeCellsAndProofsFromStructured(t *testing.T) {
 		require.NoError(t, err)
 
 		// Test ComputeCellsAndProofs
-		included, actualCellsPerBlob, actualProofsPerBlob, err := peerdas.ComputeCellsAndProofsFromStructured(uint64(len(blobsAndProofs)), blobsAndProofs)
-		require.Equal(t, included.Count(), uint64(len(actualCellsPerBlob)))
+		actualCellsPerBlob, actualProofsPerBlob, err := peerdas.ComputeCellsAndProofsFromStructured(blobsAndProofs)
 		require.NoError(t, err)
 		require.Equal(t, blobCount, len(actualCellsPerBlob))
 

beacon-chain/core/peerdas/semi_supernode_test.go

@@ -3,18 +3,16 @@ package peerdas
 import (
 	"testing"
 
-	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
 	"github.com/OffchainLabs/prysm/v7/config/params"
 	"github.com/OffchainLabs/prysm/v7/testing/require"
 	"github.com/ethereum/go-ethereum/p2p/enode"
 )
 
 func TestSemiSupernodeCustody(t *testing.T) {
-	const numberOfColumns = fieldparams.NumberOfColumns
-
 	params.SetupTestConfigCleanup(t)
 	cfg := params.BeaconConfig()
 	cfg.NumberOfCustodyGroups = 128
+	cfg.NumberOfColumns = 128
 	params.OverrideBeaconConfig(cfg)
 
 	// Create a test node ID
@@ -36,8 +34,8 @@ func TestSemiSupernodeCustody(t *testing.T) {
 
 		// Verify the columns are valid (within 0-127 range)
 		for columnIndex := range custodyColumns {
-			if columnIndex >= numberOfColumns {
-				t.Fatalf("Invalid column index %d, should be less than %d", columnIndex, numberOfColumns)
+			if columnIndex >= cfg.NumberOfColumns {
+				t.Fatalf("Invalid column index %d, should be less than %d", columnIndex, cfg.NumberOfColumns)
 			}
 		}
 	})
@@ -77,23 +75,33 @@ func TestSemiSupernodeCustody(t *testing.T) {
 func TestMinimumCustodyGroupCountToReconstruct(t *testing.T) {
 	tests := []struct {
 		name           string
-		numberOfGroups uint64
-		expectedResult uint64
+		numberOfColumns uint64
+		numberOfGroups  uint64
+		expectedResult  uint64
 	}{
 		{
 			name:           "Standard 1:1 ratio (128 columns, 128 groups)",
-			numberOfGroups: 128,
-			expectedResult: 64, // Need half of 128 groups
+			numberOfColumns: 128,
+			numberOfGroups:  128,
+			expectedResult:  64, // Need half of 128 groups
 		},
 		{
 			name:           "2 columns per group (128 columns, 64 groups)",
-			numberOfGroups: 64,
-			expectedResult: 32, // Need 64 columns, which is 32 groups (64/2)
+			numberOfColumns: 128,
+			numberOfGroups:  64,
+			expectedResult:  32, // Need 64 columns, which is 32 groups (64/2)
 		},
 		{
 			name:           "4 columns per group (128 columns, 32 groups)",
-			numberOfGroups: 32,
-			expectedResult: 16, // Need 64 columns, which is 16 groups (64/4)
+			numberOfColumns: 128,
+			numberOfGroups:  32,
+			expectedResult:  16, // Need 64 columns, which is 16 groups (64/4)
+		},
+		{
+			name:           "Odd number requiring ceiling division (100 columns, 30 groups)",
+			numberOfColumns: 100,
+			numberOfGroups:  30,
+			expectedResult:  17, // Need 50 columns, 3 columns per group (100/30), ceiling(50/3) = 17
 		},
 	}
 
@@ -101,6 +109,7 @@ func TestMinimumCustodyGroupCountToReconstruct(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			params.SetupTestConfigCleanup(t)
 			cfg := params.BeaconConfig()
+			cfg.NumberOfColumns = tt.numberOfColumns
 			cfg.NumberOfCustodyGroups = tt.numberOfGroups
 			params.OverrideBeaconConfig(cfg)
 
@@ -112,9 +121,22 @@ func TestMinimumCustodyGroupCountToReconstruct(t *testing.T) {
 }
 
 func TestMinimumCustodyGroupCountToReconstruct_ErrorCases(t *testing.T) {
+	t.Run("Returns error when NumberOfColumns is zero", func(t *testing.T) {
+		params.SetupTestConfigCleanup(t)
+		cfg := params.BeaconConfig()
+		cfg.NumberOfColumns = 0
+		cfg.NumberOfCustodyGroups = 128
+		params.OverrideBeaconConfig(cfg)
+
+		_, err := MinimumCustodyGroupCountToReconstruct()
+		require.NotNil(t, err)
+		require.Equal(t, true, err.Error() == "NumberOfColumns cannot be zero")
+	})
+
 	t.Run("Returns error when NumberOfCustodyGroups is zero", func(t *testing.T) {
 		params.SetupTestConfigCleanup(t)
 		cfg := params.BeaconConfig()
+		cfg.NumberOfColumns = 128
 		cfg.NumberOfCustodyGroups = 0
 		params.OverrideBeaconConfig(cfg)
 
@@ -126,6 +148,7 @@ func TestMinimumCustodyGroupCountToReconstruct_ErrorCases(t *testing.T) {
 	t.Run("Returns error when NumberOfCustodyGroups exceeds NumberOfColumns", func(t *testing.T) {
 		params.SetupTestConfigCleanup(t)
 		cfg := params.BeaconConfig()
+		cfg.NumberOfColumns = 128
 		cfg.NumberOfCustodyGroups = 256
 		params.OverrideBeaconConfig(cfg)
 

beacon-chain/core/peerdas/validator.go

@@ -3,7 +3,6 @@ package peerdas
 import (
 	"time"
 
-	"github.com/OffchainLabs/go-bitfield"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain/kzg"
 	beaconState "github.com/OffchainLabs/prysm/v7/beacon-chain/state"
 	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
@@ -103,13 +102,11 @@ func ValidatorsCustodyRequirement(state beaconState.ReadOnlyBeaconState, validat
 // https://github.com/ethereum/consensus-specs/blob/master/specs/fulu/validator.md#get_data_column_sidecars_from_block and
 // https://github.com/ethereum/consensus-specs/blob/master/specs/fulu/validator.md#get_data_column_sidecars_from_column_sidecar
 func DataColumnSidecars(cellsPerBlob [][]kzg.Cell, proofsPerBlob [][]kzg.Proof, src ConstructionPopulator) ([]blocks.RODataColumn, error) {
-	const numberOfColumns = uint64(fieldparams.NumberOfColumns)
-
 	if len(cellsPerBlob) == 0 {
 		return nil, nil
 	}
 	start := time.Now()
-	cells, proofs, err := rotateRowsToCols(cellsPerBlob, proofsPerBlob, numberOfColumns)
+	cells, proofs, err := rotateRowsToCols(cellsPerBlob, proofsPerBlob, params.BeaconConfig().NumberOfColumns)
 	if err != nil {
 		return nil, errors.Wrap(err, "rotate cells and proofs")
 	}
@@ -118,8 +115,9 @@ func DataColumnSidecars(cellsPerBlob [][]kzg.Cell, proofsPerBlob [][]kzg.Proof,
 		return nil, errors.Wrap(err, "extract block info")
 	}
 
-	roSidecars := make([]blocks.RODataColumn, 0, numberOfColumns)
-	for idx := range numberOfColumns {
+	maxIdx := params.BeaconConfig().NumberOfColumns
+	roSidecars := make([]blocks.RODataColumn, 0, maxIdx)
+	for idx := range maxIdx {
 		sidecar := &ethpb.DataColumnSidecar{
 			Index:                        idx,
 			Column:                       cells[idx],
@@ -144,51 +142,6 @@ func DataColumnSidecars(cellsPerBlob [][]kzg.Cell, proofsPerBlob [][]kzg.Proof,
 	return roSidecars, nil
 }
 
-// Note: cellsPerBlob
-// We should also assert that "included.Count() (i.e. number of bits set) == uint64(len(cellsPerBlob))" otherwise "cells[idx] = cells[idx][1:]" below will be out of bounds.
-func PartialColumns(included bitfield.Bitlist, cellsPerBlob [][]kzg.Cell, proofsPerBlob [][]kzg.Proof, src ConstructionPopulator) ([]blocks.PartialDataColumn, error) {
-	start := time.Now()
-	const numberOfColumns = uint64(fieldparams.NumberOfColumns)
-	// rotate the cells and proofs from being per blob(i.e. "row indexed") to being per column ("column indexed").
-	// The returned arrays are of size "numberOfColumns" in length and the "included" bitfield is used to filter out cells that are not present in each column.
-	cells, proofs, err := rotateRowsToCols(cellsPerBlob, proofsPerBlob, numberOfColumns)
-	if err != nil {
-		return nil, errors.Wrap(err, "rotate cells and proofs")
-	}
-	info, err := src.extract()
-	if err != nil {
-		return nil, errors.Wrap(err, "extract block info")
-	}
-
-	dataColumns := make([]blocks.PartialDataColumn, 0, numberOfColumns)
-	for idx := range numberOfColumns {
-		dc, err := blocks.NewPartialDataColumn(info.signedBlockHeader, idx, info.kzgCommitments, info.kzgInclusionProof)
-		if err != nil {
-			return nil, errors.Wrap(err, "new ro data column")
-		}
-
-		// info.kzgCommitments is the array of KZG commitments for each blob in the block so it's length is the number of blobs in the block.
-		// The included bitlist is the bitlist of which blobs are present i.e. which are the blobs we have the blob data for.
-		// we're now iterating over each row and extending the column one cell at a time for each blob that is present.
-		for i := range len(info.kzgCommitments) {
-			if !included.BitAt(uint64(i)) {
-				continue
-			}
-			// TODO: Check for "out of bounds" here. How do we know that cells always have upto "numberOfColumns" entries?
-			// Okay, this probably works because "rotateRowsToCols" above allocates "numberOfColumns" entries for each cell and proof.
-			// The "included" bitlist is used to filter out cells that are not present in each column so we only set the cells that are present in
-			// this call to "ExtendFromVerfifiedCell".
-			dc.ExtendFromVerfifiedCell(uint64(i), cells[idx][0], proofs[idx][0])
-			cells[idx] = cells[idx][1:]
-			proofs[idx] = proofs[idx][1:]
-		}
-		dataColumns = append(dataColumns, dc)
-	}
-
-	dataColumnComputationTime.Observe(float64(time.Since(start).Milliseconds()))
-	return dataColumns, nil
-}
-
 // Slot returns the slot of the source
 func (s *BlockReconstructionSource) Slot() primitives.Slot {
 	return s.Block().Slot()

beacon-chain/core/peerdas/validator_test.go

@@ -6,7 +6,7 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain/kzg"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/peerdas"
 	state_native "github.com/OffchainLabs/prysm/v7/beacon-chain/state/state-native"
-	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
+	"github.com/OffchainLabs/prysm/v7/config/params"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
 	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
@@ -59,8 +59,6 @@ func TestValidatorsCustodyRequirement(t *testing.T) {
 }
 
 func TestDataColumnSidecars(t *testing.T) {
-	const numberOfColumns = fieldparams.NumberOfColumns
-
 	t.Run("sizes mismatch", func(t *testing.T) {
 		// Create a protobuf signed beacon block.
 		signedBeaconBlockPb := util.NewBeaconBlockDeneb()
@@ -71,10 +69,10 @@ func TestDataColumnSidecars(t *testing.T) {
 
 		// Create cells and proofs.
 		cellsPerBlob := [][]kzg.Cell{
-			make([]kzg.Cell, numberOfColumns),
+			make([]kzg.Cell, params.BeaconConfig().NumberOfColumns),
 		}
 		proofsPerBlob := [][]kzg.Proof{
-			make([]kzg.Proof, numberOfColumns),
+			make([]kzg.Proof, params.BeaconConfig().NumberOfColumns),
 		}
 
 		rob, err := blocks.NewROBlock(signedBeaconBlock)
@@ -119,6 +117,7 @@ func TestDataColumnSidecars(t *testing.T) {
 		require.NoError(t, err)
 
 		// Create cells and proofs with sufficient cells but insufficient proofs.
+		numberOfColumns := params.BeaconConfig().NumberOfColumns
 		cellsPerBlob := [][]kzg.Cell{
 			make([]kzg.Cell, numberOfColumns),
 		}
@@ -150,6 +149,7 @@ func TestDataColumnSidecars(t *testing.T) {
 		require.NoError(t, err)
 
 		// Create cells and proofs with correct dimensions.
+		numberOfColumns := params.BeaconConfig().NumberOfColumns
 		cellsPerBlob := [][]kzg.Cell{
 			make([]kzg.Cell, numberOfColumns),
 			make([]kzg.Cell, numberOfColumns),
@@ -197,7 +197,6 @@ func TestDataColumnSidecars(t *testing.T) {
 }
 
 func TestReconstructionSource(t *testing.T) {
-	const numberOfColumns = fieldparams.NumberOfColumns
 	// Create a Fulu block with blob commitments.
 	signedBeaconBlockPb := util.NewBeaconBlockFulu()
 	commitment1 := make([]byte, 48)
@@ -213,6 +212,7 @@ func TestReconstructionSource(t *testing.T) {
 	require.NoError(t, err)
 
 	// Create cells and proofs with correct dimensions.
+	numberOfColumns := params.BeaconConfig().NumberOfColumns
 	cellsPerBlob := [][]kzg.Cell{
 		make([]kzg.Cell, numberOfColumns),
 		make([]kzg.Cell, numberOfColumns),

beacon-chain/das/BUILD.bazel

@@ -11,7 +11,6 @@ go_library(
         "iface.go",
         "log.go",
         "mock.go",
-        "needs.go",
     ],
     importpath = "github.com/OffchainLabs/prysm/v7/beacon-chain/das",
     visibility = ["//visibility:public"],
@@ -39,7 +38,6 @@ go_test(
         "availability_columns_test.go",
         "blob_cache_test.go",
         "data_column_cache_test.go",
-        "needs_test.go",
     ],
     embed = [":go_default_library"],
     deps = [

beacon-chain/das/availability_blobs.go

@@ -11,6 +11,7 @@ import (
 	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
 	"github.com/OffchainLabs/prysm/v7/runtime/logging"
 	"github.com/OffchainLabs/prysm/v7/runtime/version"
+	"github.com/OffchainLabs/prysm/v7/time/slots"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 )
@@ -23,10 +24,9 @@ var (
 // This implementation will hold any blobs passed to Persist until the IsDataAvailable is called for their
 // block, at which time they will undergo full verification and be saved to the disk.
 type LazilyPersistentStoreBlob struct {
-	store        *filesystem.BlobStorage
-	cache        *blobCache
-	verifier     BlobBatchVerifier
-	shouldRetain RetentionChecker
+	store    *filesystem.BlobStorage
+	cache    *blobCache
+	verifier BlobBatchVerifier
 }
 
 var _ AvailabilityChecker = &LazilyPersistentStoreBlob{}
@@ -42,12 +42,11 @@ type BlobBatchVerifier interface {
 
 // NewLazilyPersistentStore creates a new LazilyPersistentStore. This constructor should always be used
 // when creating a LazilyPersistentStore because it needs to initialize the cache under the hood.
-func NewLazilyPersistentStore(store *filesystem.BlobStorage, verifier BlobBatchVerifier, shouldRetain RetentionChecker) *LazilyPersistentStoreBlob {
+func NewLazilyPersistentStore(store *filesystem.BlobStorage, verifier BlobBatchVerifier) *LazilyPersistentStoreBlob {
 	return &LazilyPersistentStoreBlob{
-		store:        store,
-		cache:        newBlobCache(),
-		verifier:     verifier,
-		shouldRetain: shouldRetain,
+		store:    store,
+		cache:    newBlobCache(),
+		verifier: verifier,
 	}
 }
 
@@ -67,6 +66,9 @@ func (s *LazilyPersistentStoreBlob) Persist(current primitives.Slot, sidecars ..
 			}
 		}
 	}
+	if !params.WithinDAPeriod(slots.ToEpoch(sidecars[0].Slot()), slots.ToEpoch(current)) {
+		return nil
+	}
 	key := keyFromSidecar(sidecars[0])
 	entry := s.cache.ensure(key)
 	for _, blobSidecar := range sidecars {
@@ -89,7 +91,7 @@ func (s *LazilyPersistentStoreBlob) IsDataAvailable(ctx context.Context, current
 }
 
 func (s *LazilyPersistentStoreBlob) checkOne(ctx context.Context, current primitives.Slot, b blocks.ROBlock) error {
-	blockCommitments, err := commitmentsToCheck(b, s.shouldRetain)
+	blockCommitments, err := commitmentsToCheck(b, current)
 	if err != nil {
 		return errors.Wrapf(err, "could not check data availability for block %#x", b.Root())
 	}
@@ -138,12 +140,13 @@ func (s *LazilyPersistentStoreBlob) checkOne(ctx context.Context, current primit
 	return nil
 }
 
-func commitmentsToCheck(b blocks.ROBlock, shouldRetain RetentionChecker) ([][]byte, error) {
+func commitmentsToCheck(b blocks.ROBlock, current primitives.Slot) ([][]byte, error) {
 	if b.Version() < version.Deneb {
 		return nil, nil
 	}
 
-	if !shouldRetain(b.Block().Slot()) {
+	// We are only required to check within MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUEST
+	if !params.WithinDAPeriod(slots.ToEpoch(b.Block().Slot()), slots.ToEpoch(current)) {
 		return nil, nil
 	}
 

beacon-chain/das/availability_blobs_test.go

@@ -17,10 +17,6 @@ import (
 	errors "github.com/pkg/errors"
 )
 
-func testShouldRetainAlways(s primitives.Slot) bool {
-	return true
-}
-
 func Test_commitmentsToCheck(t *testing.T) {
 	params.SetupTestConfigCleanup(t)
 	params.BeaconConfig().FuluForkEpoch = params.BeaconConfig().ElectraForkEpoch + 4096*2
@@ -34,12 +30,11 @@ func Test_commitmentsToCheck(t *testing.T) {
 		commits[i] = bytesutil.PadTo([]byte{byte(i)}, 48)
 	}
 	cases := []struct {
-		name         string
-		commits      [][]byte
-		block        func(*testing.T) blocks.ROBlock
-		slot         primitives.Slot
-		err          error
-		shouldRetain RetentionChecker
+		name    string
+		commits [][]byte
+		block   func(*testing.T) blocks.ROBlock
+		slot    primitives.Slot
+		err     error
 	}{
 		{
 			name: "pre deneb",
@@ -65,7 +60,6 @@ func Test_commitmentsToCheck(t *testing.T) {
 				require.NoError(t, err)
 				return rb
 			},
-			shouldRetain: testShouldRetainAlways,
 			commits: func() [][]byte {
 				mb := params.GetNetworkScheduleEntry(slots.ToEpoch(fulu + 100)).MaxBlobsPerBlock
 				return commits[:mb]
@@ -85,8 +79,7 @@ func Test_commitmentsToCheck(t *testing.T) {
 				require.NoError(t, err)
 				return rb
 			},
-			shouldRetain: func(s primitives.Slot) bool { return false },
-			slot:         fulu + windowSlots + 1,
+			slot: fulu + windowSlots + 1,
 		},
 		{
 			name: "excessive commitments",
@@ -104,15 +97,14 @@ func Test_commitmentsToCheck(t *testing.T) {
 				require.Equal(t, true, len(c) > params.BeaconConfig().MaxBlobsPerBlock(sb.Block().Slot()))
 				return rb
 			},
-			shouldRetain: testShouldRetainAlways,
-			slot:         windowSlots + 1,
-			err:          errIndexOutOfBounds,
+			slot: windowSlots + 1,
+			err:  errIndexOutOfBounds,
 		},
 	}
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
 			b := c.block(t)
-			co, err := commitmentsToCheck(b, c.shouldRetain)
+			co, err := commitmentsToCheck(b, c.slot)
 			if c.err != nil {
 				require.ErrorIs(t, err, c.err)
 			} else {
@@ -134,7 +126,7 @@ func TestLazilyPersistent_Missing(t *testing.T) {
 	blk, blobSidecars := util.GenerateTestDenebBlockWithSidecar(t, [32]byte{}, ds, 3)
 
 	mbv := &mockBlobBatchVerifier{t: t, scs: blobSidecars}
-	as := NewLazilyPersistentStore(store, mbv, testShouldRetainAlways)
+	as := NewLazilyPersistentStore(store, mbv)
 
 	// Only one commitment persisted, should return error with other indices
 	require.NoError(t, as.Persist(ds, blobSidecars[2]))
@@ -161,7 +153,7 @@ func TestLazilyPersistent_Mismatch(t *testing.T) {
 
 	mbv := &mockBlobBatchVerifier{t: t, err: errors.New("kzg check should not run")}
 	blobSidecars[0].KzgCommitment = bytesutil.PadTo([]byte("nope"), 48)
-	as := NewLazilyPersistentStore(store, mbv, testShouldRetainAlways)
+	as := NewLazilyPersistentStore(store, mbv)
 
 	// Only one commitment persisted, should return error with other indices
 	require.NoError(t, as.Persist(ds, blobSidecars[0]))
@@ -174,7 +166,7 @@ func TestLazyPersistOnceCommitted(t *testing.T) {
 	ds := util.SlotAtEpoch(t, params.BeaconConfig().DenebForkEpoch)
 	_, blobSidecars := util.GenerateTestDenebBlockWithSidecar(t, [32]byte{}, ds, 6)
 
-	as := NewLazilyPersistentStore(filesystem.NewEphemeralBlobStorage(t), &mockBlobBatchVerifier{}, testShouldRetainAlways)
+	as := NewLazilyPersistentStore(filesystem.NewEphemeralBlobStorage(t), &mockBlobBatchVerifier{})
 	// stashes as expected
 	require.NoError(t, as.Persist(ds, blobSidecars...))
 	// ignores duplicates
@@ -191,7 +183,7 @@ func TestLazyPersistOnceCommitted(t *testing.T) {
 	require.NoError(t, as.Persist(slotOOB, moreBlobSidecars[0]))
 
 	// doesn't ignore new sidecars with a different block root
-	require.NoError(t, as.Persist(ds, moreBlobSidecars[1:]...))
+	require.NoError(t, as.Persist(ds, moreBlobSidecars...))
 }
 
 type mockBlobBatchVerifier struct {

beacon-chain/das/availability_columns.go

@@ -7,8 +7,10 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/peerdas"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/db/filesystem"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/verification"
+	"github.com/OffchainLabs/prysm/v7/config/params"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
+	"github.com/OffchainLabs/prysm/v7/time/slots"
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	errors "github.com/pkg/errors"
 )
@@ -22,7 +24,6 @@ type LazilyPersistentStoreColumn struct {
 	newDataColumnsVerifier verification.NewDataColumnsVerifier
 	custody                *custodyRequirement
 	bisector               Bisector
-	shouldRetain           RetentionChecker
 }
 
 var _ AvailabilityChecker = &LazilyPersistentStoreColumn{}
@@ -43,7 +44,6 @@ func NewLazilyPersistentStoreColumn(
 	nodeID enode.ID,
 	cgc uint64,
 	bisector Bisector,
-	shouldRetain RetentionChecker,
 ) *LazilyPersistentStoreColumn {
 	return &LazilyPersistentStoreColumn{
 		store:                  store,
@@ -51,15 +51,18 @@ func NewLazilyPersistentStoreColumn(
 		newDataColumnsVerifier: newDataColumnsVerifier,
 		custody:                &custodyRequirement{nodeID: nodeID, cgc: cgc},
 		bisector:               bisector,
-		shouldRetain:           shouldRetain,
 	}
 }
 
 // PersistColumns adds columns to the working column cache. Columns stored in this cache will be persisted
 // for at least as long as the node is running. Once IsDataAvailable succeeds, all columns referenced
 // by the given block are guaranteed to be persisted for the remainder of the retention period.
-func (s *LazilyPersistentStoreColumn) Persist(_ primitives.Slot, sidecars ...blocks.RODataColumn) error {
+func (s *LazilyPersistentStoreColumn) Persist(current primitives.Slot, sidecars ...blocks.RODataColumn) error {
+	currentEpoch := slots.ToEpoch(current)
 	for _, sidecar := range sidecars {
+		if !params.WithinDAPeriod(slots.ToEpoch(sidecar.Slot()), currentEpoch) {
+			continue
+		}
 		if err := s.cache.stash(sidecar); err != nil {
 			return errors.Wrap(err, "stash DataColumnSidecar")
 		}
@@ -70,12 +73,14 @@ func (s *LazilyPersistentStoreColumn) Persist(_ primitives.Slot, sidecars ...blo
 
 // IsDataAvailable returns nil if all the commitments in the given block are persisted to the db and have been verified.
 // DataColumnsSidecars already in the db are assumed to have been previously verified against the block.
-func (s *LazilyPersistentStoreColumn) IsDataAvailable(ctx context.Context, _ primitives.Slot, blks ...blocks.ROBlock) error {
+func (s *LazilyPersistentStoreColumn) IsDataAvailable(ctx context.Context, current primitives.Slot, blks ...blocks.ROBlock) error {
+	currentEpoch := slots.ToEpoch(current)
+
 	toVerify := make([]blocks.RODataColumn, 0)
 	for _, block := range blks {
-		indices, err := s.required(block)
+		indices, err := s.required(block, currentEpoch)
 		if err != nil {
-			return errors.Wrapf(err, "full commitments to check with block root `%#x`", block.Root())
+			return errors.Wrapf(err, "full commitments to check with block root `%#x` and current slot `%d`", block.Root(), current)
 		}
 		if indices.Count() == 0 {
 			continue
@@ -101,8 +106,10 @@ func (s *LazilyPersistentStoreColumn) IsDataAvailable(ctx context.Context, _ pri
 }
 
 // required returns the set of column indices to check for a given block.
-func (s *LazilyPersistentStoreColumn) required(block blocks.ROBlock) (peerdas.ColumnIndices, error) {
-	if !s.shouldRetain(block.Block().Slot()) {
+func (s *LazilyPersistentStoreColumn) required(block blocks.ROBlock, current primitives.Epoch) (peerdas.ColumnIndices, error) {
+	eBlk := slots.ToEpoch(block.Block().Slot())
+	eFulu := params.BeaconConfig().FuluForkEpoch
+	if current < eFulu || eBlk < eFulu || !params.WithinDAPeriod(eBlk, current) {
 		return peerdas.NewColumnIndices(), nil
 	}
 
@@ -117,14 +124,11 @@ func (s *LazilyPersistentStoreColumn) required(block blocks.ROBlock) (peerdas.Co
 		return peerdas.NewColumnIndices(), nil
 	}
 
-	return s.custody.required()
+	return s.custody.required(current)
 }
 
 // verifyAndSave calls Save on the column store if the columns pass verification.
 func (s *LazilyPersistentStoreColumn) verifyAndSave(columns []blocks.RODataColumn) error {
-	if len(columns) == 0 {
-		return nil
-	}
 	verified, err := s.verifyColumns(columns)
 	if err != nil {
 		return errors.Wrap(err, "verify columns")
@@ -137,9 +141,6 @@ func (s *LazilyPersistentStoreColumn) verifyAndSave(columns []blocks.RODataColum
 }
 
 func (s *LazilyPersistentStoreColumn) verifyColumns(columns []blocks.RODataColumn) ([]blocks.VerifiedRODataColumn, error) {
-	if len(columns) == 0 {
-		return nil, nil
-	}
 	verifier := s.newDataColumnsVerifier(columns, verification.ByRangeRequestDataColumnSidecarRequirements)
 	if err := verifier.ValidFields(); err != nil {
 		return nil, errors.Wrap(err, "valid fields")
@@ -235,7 +236,7 @@ type custodyRequirement struct {
 	indices peerdas.ColumnIndices
 }
 
-func (c *custodyRequirement) required() (peerdas.ColumnIndices, error) {
+func (c *custodyRequirement) required(current primitives.Epoch) (peerdas.ColumnIndices, error) {
 	peerInfo, _, err := peerdas.Info(c.nodeID, c.cgc)
 	if err != nil {
 		return peerdas.NewColumnIndices(), errors.Wrap(err, "peer info")

beacon-chain/das/availability_columns_test.go

@@ -20,12 +20,6 @@ import (
 	"github.com/pkg/errors"
 )
 
-func mockShouldRetain(current primitives.Epoch) RetentionChecker {
-	return func(slot primitives.Slot) bool {
-		return params.WithinDAPeriod(slots.ToEpoch(slot), current)
-	}
-}
-
 var commitments = [][]byte{
 	bytesutil.PadTo([]byte("a"), 48),
 	bytesutil.PadTo([]byte("b"), 48),
@@ -36,7 +30,7 @@ var commitments = [][]byte{
 func TestPersist(t *testing.T) {
 	t.Run("no sidecars", func(t *testing.T) {
 		dataColumnStorage := filesystem.NewEphemeralDataColumnStorage(t)
-		lazilyPersistentStoreColumns := NewLazilyPersistentStoreColumn(dataColumnStorage, nil, enode.ID{}, 0, nil, mockShouldRetain(0))
+		lazilyPersistentStoreColumns := NewLazilyPersistentStoreColumn(dataColumnStorage, nil, enode.ID{}, 0, nil)
 		err := lazilyPersistentStoreColumns.Persist(0)
 		require.NoError(t, err)
 		require.Equal(t, 0, len(lazilyPersistentStoreColumns.cache.entries))
@@ -49,14 +43,12 @@ func TestPersist(t *testing.T) {
 			{Slot: 1, Index: 1},
 		}
 
-		var current primitives.Slot = 1_000_000
-		sr := mockShouldRetain(slots.ToEpoch(current))
 		roSidecars, _ := util.CreateTestVerifiedRoDataColumnSidecars(t, dataColumnParamsByBlockRoot)
-		lazilyPersistentStoreColumns := NewLazilyPersistentStoreColumn(dataColumnStorage, nil, enode.ID{}, 0, nil, sr)
+		lazilyPersistentStoreColumns := NewLazilyPersistentStoreColumn(dataColumnStorage, nil, enode.ID{}, 0, nil)
 
-		err := lazilyPersistentStoreColumns.Persist(current, roSidecars...)
+		err := lazilyPersistentStoreColumns.Persist(1_000_000, roSidecars...)
 		require.NoError(t, err)
-		require.Equal(t, len(roSidecars), len(lazilyPersistentStoreColumns.cache.entries))
+		require.Equal(t, 0, len(lazilyPersistentStoreColumns.cache.entries))
 	})
 
 	t.Run("nominal", func(t *testing.T) {
@@ -69,7 +61,7 @@ func TestPersist(t *testing.T) {
 		}
 
 		roSidecars, roDataColumns := util.CreateTestVerifiedRoDataColumnSidecars(t, dataColumnParamsByBlockRoot)
-		avs := NewLazilyPersistentStoreColumn(store, nil, enode.ID{}, 0, nil, mockShouldRetain(slots.ToEpoch(slot)))
+		avs := NewLazilyPersistentStoreColumn(store, nil, enode.ID{}, 0, nil)
 
 		err := avs.Persist(slot, roSidecars...)
 		require.NoError(t, err)
@@ -114,9 +106,9 @@ func TestIsDataAvailable(t *testing.T) {
 		signedRoBlock := newSignedRoBlock(t, signedBeaconBlockFulu)
 
 		dataColumnStorage := filesystem.NewEphemeralDataColumnStorage(t)
-		lazilyPersistentStoreColumns := NewLazilyPersistentStoreColumn(dataColumnStorage, newDataColumnsVerifier, enode.ID{}, 0, nil, mockShouldRetain(0))
+		lazilyPersistentStoreColumns := NewLazilyPersistentStoreColumn(dataColumnStorage, newDataColumnsVerifier, enode.ID{}, 0, nil)
 
-		err := lazilyPersistentStoreColumns.IsDataAvailable(ctx, 0, signedRoBlock)
+		err := lazilyPersistentStoreColumns.IsDataAvailable(ctx, 0 /*current slot*/, signedRoBlock)
 		require.NoError(t, err)
 	})
 
@@ -138,7 +130,7 @@ func TestIsDataAvailable(t *testing.T) {
 		storage := filesystem.NewEphemeralDataColumnStorage(t)
 
 		indices := []uint64{1, 17, 19, 42, 75, 87, 102, 117}
-		avs := NewLazilyPersistentStoreColumn(storage, newDataColumnsVerifier, enode.ID{}, uint64(len(indices)), nil, mockShouldRetain(slots.ToEpoch(slot)))
+		avs := NewLazilyPersistentStoreColumn(storage, newDataColumnsVerifier, enode.ID{}, uint64(len(indices)), nil)
 		dcparams := make([]util.DataColumnParam, 0, len(indices))
 		for _, index := range indices {
 			dataColumnParams := util.DataColumnParam{
@@ -179,19 +171,15 @@ func TestIsDataAvailable(t *testing.T) {
 	})
 }
 
-func TestRetentionWindow(t *testing.T) {
+func TestFullCommitmentsToCheck(t *testing.T) {
 	windowSlots, err := slots.EpochEnd(params.BeaconConfig().MinEpochsForDataColumnSidecarsRequest)
 	require.NoError(t, err)
-	fuluSlot, err := slots.EpochStart(params.BeaconConfig().FuluForkEpoch)
-	require.NoError(t, err)
 
-	numberOfColumns := fieldparams.NumberOfColumns
 	testCases := []struct {
 		name        string
 		commitments [][]byte
 		block       func(*testing.T) blocks.ROBlock
 		slot        primitives.Slot
-		wantedCols  int
 	}{
 		{
 			name: "Pre-Fulu block",
@@ -209,33 +197,35 @@ func TestRetentionWindow(t *testing.T) {
 
 				return newSignedRoBlock(t, beaconBlockElectra)
 			},
-			slot: fuluSlot + windowSlots,
+			slot: windowSlots + 1,
 		},
 		{
 			name: "Commitments within data availability window",
 			block: func(t *testing.T) blocks.ROBlock {
 				signedBeaconBlockFulu := util.NewBeaconBlockFulu()
 				signedBeaconBlockFulu.Block.Body.BlobKzgCommitments = commitments
-				signedBeaconBlockFulu.Block.Slot = fuluSlot + windowSlots - 1
+				signedBeaconBlockFulu.Block.Slot = 100
 
 				return newSignedRoBlock(t, signedBeaconBlockFulu)
 			},
 			commitments: commitments,
-			slot:        fuluSlot + windowSlots,
-			wantedCols:  numberOfColumns,
+			slot:        100,
 		},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
+			numberOfColumns := params.BeaconConfig().NumberOfColumns
 
 			b := tc.block(t)
-			s := NewLazilyPersistentStoreColumn(nil, nil, enode.ID{}, uint64(numberOfColumns), nil, mockShouldRetain(slots.ToEpoch(tc.slot)))
+			s := NewLazilyPersistentStoreColumn(nil, nil, enode.ID{}, numberOfColumns, nil)
 
-			indices, err := s.required(b)
+			commitmentsArray, err := s.required(b, slots.ToEpoch(tc.slot))
 			require.NoError(t, err)
 
-			require.Equal(t, tc.wantedCols, len(indices))
+			for _, commitments := range commitmentsArray {
+				require.DeepEqual(t, tc.commitments, commitments)
+			}
 		})
 	}
 }

beacon-chain/das/blob_cache_test.go

@@ -34,8 +34,7 @@ type filterTestCaseSetupFunc func(t *testing.T) (*blobCacheEntry, [][]byte, []bl
 func filterTestCaseSetup(slot primitives.Slot, nBlobs int, onDisk []int, numExpected int) filterTestCaseSetupFunc {
 	return func(t *testing.T) (*blobCacheEntry, [][]byte, []blocks.ROBlob) {
 		blk, blobs := util.GenerateTestDenebBlockWithSidecar(t, [32]byte{}, slot, nBlobs)
-		shouldRetain := func(s primitives.Slot) bool { return true }
-		commits, err := commitmentsToCheck(blk, shouldRetain)
+		commits, err := commitmentsToCheck(blk, blk.Block().Slot())
 		require.NoError(t, err)
 		entry := &blobCacheEntry{}
 		if len(onDisk) > 0 {

beacon-chain/das/iface.go

@@ -14,6 +14,3 @@ import (
 type AvailabilityChecker interface {
 	IsDataAvailable(ctx context.Context, current primitives.Slot, b ...blocks.ROBlock) error
 }
-
-// RetentionChecker is a callback that determines whether blobs at the given slot are within the retention period.
-type RetentionChecker func(primitives.Slot) bool

beacon-chain/das/needs.go

@@ -1,135 +0,0 @@
-package das
-
-import (
-	"github.com/OffchainLabs/prysm/v7/config/params"
-	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
-	"github.com/OffchainLabs/prysm/v7/time/slots"
-	"github.com/pkg/errors"
-)
-
-// NeedSpan represents the need for a resource over a span of slots.
-type NeedSpan struct {
-	Begin primitives.Slot
-	End   primitives.Slot
-}
-
-// At returns whether blocks/blobs/columns are needed At the given slot.
-func (n NeedSpan) At(slot primitives.Slot) bool {
-	return slot >= n.Begin && slot < n.End
-}
-
-// CurrentNeeds fields can be used to check whether the given resource type is needed
-// at a given slot. The values are based on the current slot, so this value shouldn't
-// be retained / reused across slots.
-type CurrentNeeds struct {
-	Block NeedSpan
-	Blob  NeedSpan
-	Col   NeedSpan
-}
-
-// SyncNeeds holds configuration and state for determining what data is needed
-// at any given slot during backfill based on the current slot.
-type SyncNeeds struct {
-	current func() primitives.Slot
-	deneb   primitives.Slot
-	fulu    primitives.Slot
-
-	oldestSlotFlagPtr  *primitives.Slot
-	validOldestSlotPtr *primitives.Slot
-	blockRetention     primitives.Epoch
-
-	blobRetentionFlag primitives.Epoch
-	blobRetention     primitives.Epoch
-	colRetention      primitives.Epoch
-}
-
-type CurrentSlotter func() primitives.Slot
-
-func NewSyncNeeds(current CurrentSlotter, oldestSlotFlagPtr *primitives.Slot, blobRetentionFlag primitives.Epoch) (SyncNeeds, error) {
-	deneb, err := slots.EpochStart(params.BeaconConfig().DenebForkEpoch)
-	if err != nil {
-		return SyncNeeds{}, errors.Wrap(err, "deneb fork slot")
-	}
-	fuluBoundary := min(params.BeaconConfig().FuluForkEpoch, slots.MaxSafeEpoch())
-	fulu, err := slots.EpochStart(fuluBoundary)
-	if err != nil {
-		return SyncNeeds{}, errors.Wrap(err, "fulu fork slot")
-	}
-	sn := SyncNeeds{
-		current:           func() primitives.Slot { return current() },
-		deneb:             deneb,
-		fulu:              fulu,
-		blobRetentionFlag: blobRetentionFlag,
-	}
-	// We apply the --blob-retention-epochs flag to both blob and column retention.
-	sn.blobRetention = max(sn.blobRetentionFlag, params.BeaconConfig().MinEpochsForBlobsSidecarsRequest)
-	sn.colRetention = max(sn.blobRetentionFlag, params.BeaconConfig().MinEpochsForDataColumnSidecarsRequest)
-
-	// Override spec minimum block retention with user-provided flag only if it is lower than the spec minimum.
-	sn.blockRetention = primitives.Epoch(params.BeaconConfig().MinEpochsForBlockRequests)
-	if oldestSlotFlagPtr != nil {
-		oldestEpoch := slots.ToEpoch(*oldestSlotFlagPtr)
-		if oldestEpoch < sn.blockRetention {
-			sn.validOldestSlotPtr = oldestSlotFlagPtr
-		} else {
-			log.WithField("backfill-oldest-slot", *oldestSlotFlagPtr).
-				WithField("specMinSlot", syncEpochOffset(current(), sn.blockRetention)).
-				Warn("Ignoring user-specified slot > MIN_EPOCHS_FOR_BLOCK_REQUESTS.")
-		}
-	}
-
-	return sn, nil
-}
-
-// Currently is the main callback given to the different parts of backfill to determine
-// what resources are needed at a given slot. It assumes the current instance of SyncNeeds
-// is the result of calling initialize.
-func (n SyncNeeds) Currently() CurrentNeeds {
-	current := n.current()
-	c := CurrentNeeds{
-		Block: n.blockSpan(current),
-		Blob:  NeedSpan{Begin: syncEpochOffset(current, n.blobRetention), End: n.fulu},
-		Col:   NeedSpan{Begin: syncEpochOffset(current, n.colRetention), End: current},
-	}
-	// Adjust the minimums forward to the slots where the sidecar types were introduced
-	c.Blob.Begin = max(c.Blob.Begin, n.deneb)
-	c.Col.Begin = max(c.Col.Begin, n.fulu)
-
-	return c
-}
-
-func (n SyncNeeds) blockSpan(current primitives.Slot) NeedSpan {
-	if n.validOldestSlotPtr != nil { // assumes validation done in initialize()
-		return NeedSpan{Begin: *n.validOldestSlotPtr, End: current}
-	}
-	return NeedSpan{Begin: syncEpochOffset(current, n.blockRetention), End: current}
-}
-
-func (n SyncNeeds) BlobRetentionChecker() RetentionChecker {
-	return func(slot primitives.Slot) bool {
-		current := n.Currently()
-		return current.Blob.At(slot)
-	}
-}
-
-func (n SyncNeeds) DataColumnRetentionChecker() RetentionChecker {
-	return func(slot primitives.Slot) bool {
-		current := n.Currently()
-		return current.Col.At(slot)
-	}
-}
-
-// syncEpochOffset subtracts a number of epochs as slots from the current slot, with underflow checks.
-// It returns slot 1 if the result would be 0 or underflow. It doesn't return slot 0 because the
-// genesis block needs to be specially synced (it doesn't have a valid signature).
-func syncEpochOffset(current primitives.Slot, subtract primitives.Epoch) primitives.Slot {
-	minEpoch := min(subtract, slots.MaxSafeEpoch())
-	// compute slot offset - offset is a number of slots to go back from current (not an absolute slot).
-	offset := slots.UnsafeEpochStart(minEpoch)
-	// Undeflow protection: slot 0 is the genesis block, therefore the signature in it is invalid.
-	// To prevent us from rejecting a batch, we restrict the minimum backfill batch till only slot 1
-	if offset >= current {
-		return 1
-	}
-	return current - offset
-}

beacon-chain/db/filesystem/data_column.go

@@ -270,7 +270,7 @@ func (dcs *DataColumnStorage) Save(dataColumnSidecars []blocks.VerifiedRODataCol
 	// Check the number of columns is the one expected.
 	// While implementing this, we expect the number of columns won't change.
 	// If it does, we will need to create a new version of the data column sidecar file.
-	if fieldparams.NumberOfColumns != mandatoryNumberOfColumns {
+	if params.BeaconConfig().NumberOfColumns != mandatoryNumberOfColumns {
 		return errWrongNumberOfColumns
 	}
 
@@ -515,11 +515,6 @@ func (dcs *DataColumnStorage) Clear() error {
 
 // prune clean the cache, the filesystem and mutexes.
 func (dcs *DataColumnStorage) prune() {
-	startTime := time.Now()
-	defer func() {
-		dataColumnPruneLatency.Observe(float64(time.Since(startTime).Milliseconds()))
-	}()
-
 	highestStoredEpoch := dcs.cache.HighestEpoch()
 
 	// Check if we need to prune.
@@ -627,9 +622,6 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsExistingFile(filePath string
 	// Create the SSZ encoded data column sidecars.
 	var sszEncodedDataColumnSidecars []byte
 
-	// Initialize the count of the saved SSZ encoded data column sidecar.
-	storedCount := uint8(0)
-
 	for {
 		dataColumnSidecars := pullChan(inputDataColumnSidecars)
 		if len(dataColumnSidecars) == 0 {
@@ -676,9 +668,6 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsExistingFile(filePath string
 				return errors.Wrap(err, "set index")
 			}
 
-			// Increment the count of the saved SSZ encoded data column sidecar.
-			storedCount++
-
 			// Append the SSZ encoded data column sidecar to the SSZ encoded data column sidecars.
 			sszEncodedDataColumnSidecars = append(sszEncodedDataColumnSidecars, sszEncodedDataColumnSidecar...)
 		}
@@ -703,12 +692,9 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsExistingFile(filePath string
 		return errWrongBytesWritten
 	}
 
-	syncStart := time.Now()
 	if err := file.Sync(); err != nil {
 		return errors.Wrap(err, "sync")
 	}
-	dataColumnFileSyncLatency.Observe(float64(time.Since(syncStart).Milliseconds()))
-	dataColumnBatchStoreCount.Observe(float64(storedCount))
 
 	return nil
 }
@@ -822,14 +808,10 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsNewFile(filePath string, inp
 		return errWrongBytesWritten
 	}
 
-	syncStart := time.Now()
 	if err := file.Sync(); err != nil {
 		return errors.Wrap(err, "sync")
 	}
 
-	dataColumnFileSyncLatency.Observe(float64(time.Since(syncStart).Milliseconds()))
-	dataColumnBatchStoreCount.Observe(float64(storedCount))
-
 	return nil
 }
 
@@ -982,7 +964,8 @@ func (si *storageIndices) set(dataColumnIndex uint64, position uint8) error {
 
 // pullChan pulls data column sidecars from the input channel until it is empty.
 func pullChan(inputRoDataColumns chan []blocks.VerifiedRODataColumn) []blocks.VerifiedRODataColumn {
-	dataColumnSidecars := make([]blocks.VerifiedRODataColumn, 0, fieldparams.NumberOfColumns)
+	numberOfColumns := params.BeaconConfig().NumberOfColumns
+	dataColumnSidecars := make([]blocks.VerifiedRODataColumn, 0, numberOfColumns)
 
 	for {
 		select {

beacon-chain/db/filesystem/data_column_cache.go

@@ -117,6 +117,8 @@ func (sc *dataColumnStorageSummaryCache) HighestEpoch() primitives.Epoch {
 
 // set updates the cache.
 func (sc *dataColumnStorageSummaryCache) set(dataColumnsIdent DataColumnsIdent) error {
+	numberOfColumns := params.BeaconConfig().NumberOfColumns
+
 	sc.mu.Lock()
 	defer sc.mu.Unlock()
 
@@ -125,7 +127,7 @@ func (sc *dataColumnStorageSummaryCache) set(dataColumnsIdent DataColumnsIdent)
 
 	count := uint64(0)
 	for _, index := range dataColumnsIdent.Indices {
-		if index >= fieldparams.NumberOfColumns {
+		if index >= numberOfColumns {
 			return errDataColumnIndexOutOfBounds
 		}
 

beacon-chain/db/filesystem/data_column_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 
 	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
+	"github.com/OffchainLabs/prysm/v7/config/params"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
 	"github.com/OffchainLabs/prysm/v7/testing/require"
@@ -87,6 +88,22 @@ func TestWarmCache(t *testing.T) {
 }
 
 func TestSaveDataColumnsSidecars(t *testing.T) {
+	t.Run("wrong numbers of columns", func(t *testing.T) {
+		cfg := params.BeaconConfig().Copy()
+		cfg.NumberOfColumns = 0
+		params.OverrideBeaconConfig(cfg)
+		params.SetupTestConfigCleanup(t)
+
+		_, verifiedRoDataColumnSidecars := util.CreateTestVerifiedRoDataColumnSidecars(
+			t,
+			[]util.DataColumnParam{{Index: 12}, {Index: 1_000_000}, {Index: 48}},
+		)
+
+		_, dataColumnStorage := NewEphemeralDataColumnStorageAndFs(t)
+		err := dataColumnStorage.Save(verifiedRoDataColumnSidecars)
+		require.ErrorIs(t, err, errWrongNumberOfColumns)
+	})
+
 	t.Run("one of the column index is too large", func(t *testing.T) {
 		_, verifiedRoDataColumnSidecars := util.CreateTestVerifiedRoDataColumnSidecars(
 			t,

beacon-chain/db/filesystem/metrics.go

@@ -36,15 +36,16 @@ var (
 	})
 
 	// Data columns
+	dataColumnBuckets     = []float64{3, 5, 7, 9, 11, 13}
 	dataColumnSaveLatency = promauto.NewHistogram(prometheus.HistogramOpts{
 		Name:    "data_column_storage_save_latency",
 		Help:    "Latency of DataColumnSidecar storage save operations in milliseconds",
-		Buckets: []float64{10, 20, 30, 50, 100, 200, 500},
+		Buckets: dataColumnBuckets,
 	})
 	dataColumnFetchLatency = promauto.NewHistogram(prometheus.HistogramOpts{
 		Name:    "data_column_storage_get_latency",
 		Help:    "Latency of DataColumnSidecar storage get operations in milliseconds",
-		Buckets: []float64{3, 5, 7, 9, 11, 13},
+		Buckets: dataColumnBuckets,
 	})
 	dataColumnPrunedCounter = promauto.NewCounter(prometheus.CounterOpts{
 		Name: "data_column_pruned",
@@ -58,16 +59,4 @@ var (
 		Name: "data_column_disk_count",
 		Help: "Approximate number of data columns in storage",
 	})
-	dataColumnFileSyncLatency = promauto.NewSummary(prometheus.SummaryOpts{
-		Name: "data_column_file_sync_latency",
-		Help: "Latency of sync operations when saving data columns in milliseconds",
-	})
-	dataColumnBatchStoreCount = promauto.NewSummary(prometheus.SummaryOpts{
-		Name: "data_column_batch_store_count",
-		Help: "Number of data columns stored in a batch",
-	})
-	dataColumnPruneLatency = promauto.NewSummary(prometheus.SummaryOpts{
-		Name: "data_column_prune_latency",
-		Help: "Latency of data column prune operations in milliseconds",
-	})
 )

beacon-chain/execution/BUILD.bazel

@@ -72,7 +72,6 @@ go_library(
         "@com_github_pkg_errors//:go_default_library",
         "@com_github_prometheus_client_golang//prometheus:go_default_library",
         "@com_github_prometheus_client_golang//prometheus/promauto:go_default_library",
-        "@com_github_prysmaticlabs_go_bitfield//:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
         "@io_k8s_client_go//tools/cache:go_default_library",
         "@org_golang_google_protobuf//proto:go_default_library",

beacon-chain/execution/engine_client.go

@@ -7,7 +7,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/OffchainLabs/go-bitfield"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain/kzg"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/peerdas"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/execution/types"
@@ -58,7 +57,6 @@ var (
 	fuluEngineEndpoints = []string{
 		GetPayloadMethodV5,
 		GetBlobsV2,
-		GetBlobsV3,
 	}
 )
 
@@ -100,8 +98,6 @@ const (
 	GetBlobsV1 = "engine_getBlobsV1"
 	// GetBlobsV2 request string for JSON-RPC.
 	GetBlobsV2 = "engine_getBlobsV2"
-	// GetBlobsV3 request string for JSON-RPC.
-	GetBlobsV3 = "engine_getBlobsV3"
 	// Defines the seconds before timing out engine endpoints with non-block execution semantics.
 	defaultEngineTimeout = time.Second
 )
@@ -125,7 +121,7 @@ type Reconstructor interface {
 		ctx context.Context, blindedBlocks []interfaces.ReadOnlySignedBeaconBlock,
 	) ([]interfaces.SignedBeaconBlock, error)
 	ReconstructBlobSidecars(ctx context.Context, block interfaces.ReadOnlySignedBeaconBlock, blockRoot [fieldparams.RootLength]byte, hi func(uint64) bool) ([]blocks.VerifiedROBlob, error)
-	ConstructDataColumnSidecars(ctx context.Context, populator peerdas.ConstructionPopulator) ([]blocks.VerifiedRODataColumn, []blocks.PartialDataColumn, error)
+	ConstructDataColumnSidecars(ctx context.Context, populator peerdas.ConstructionPopulator) ([]blocks.VerifiedRODataColumn, error)
 }
 
 // EngineCaller defines a client that can interact with an Ethereum
@@ -536,35 +532,12 @@ func (s *Service) GetBlobsV2(ctx context.Context, versionedHashes []common.Hash)
 	ctx, span := trace.StartSpan(ctx, "powchain.engine-api-client.GetBlobsV2")
 	defer span.End()
 
-	start := time.Now()
-
 	if !s.capabilityCache.has(GetBlobsV2) {
 		return nil, errors.New(fmt.Sprintf("%s is not supported", GetBlobsV2))
 	}
 
 	result := make([]*pb.BlobAndProofV2, len(versionedHashes))
 	err := s.rpcClient.CallContext(ctx, &result, GetBlobsV2, versionedHashes)
-
-	if len(result) != 0 {
-		getBlobsV2Latency.Observe(float64(time.Since(start).Milliseconds()))
-	}
-
-	return result, handleRPCError(err)
-}
-
-func (s *Service) GetBlobsV3(ctx context.Context, versionedHashes []common.Hash) ([]*pb.BlobAndProofV2, error) {
-	ctx, span := trace.StartSpan(ctx, "powchain.engine-api-client.GetBlobsV3")
-	defer span.End()
-	start := time.Now()
-
-	if !s.capabilityCache.has(GetBlobsV3) {
-		return nil, errors.New(fmt.Sprintf("%s is not supported", GetBlobsV3))
-	}
-
-	getBlobsV3RequestsTotal.Inc()
-	result := make([]*pb.BlobAndProofV2, len(versionedHashes))
-	err := s.rpcClient.CallContext(ctx, &result, GetBlobsV3, versionedHashes)
-	getBlobsV3Latency.Observe(time.Since(start).Seconds())
 	return result, handleRPCError(err)
 }
 
@@ -678,51 +651,40 @@ func (s *Service) ReconstructBlobSidecars(ctx context.Context, block interfaces.
 	return verifiedBlobs, nil
 }
 
-func (s *Service) ConstructDataColumnSidecars(ctx context.Context, populator peerdas.ConstructionPopulator) ([]blocks.VerifiedRODataColumn, []blocks.PartialDataColumn, error) {
+func (s *Service) ConstructDataColumnSidecars(ctx context.Context, populator peerdas.ConstructionPopulator) ([]blocks.VerifiedRODataColumn, error) {
 	root := populator.Root()
 
 	// Fetch cells and proofs from the execution client using the KZG commitments from the sidecar.
 	commitments, err := populator.Commitments()
 	if err != nil {
-		return nil, nil, wrapWithBlockRoot(err, root, "commitments")
+		return nil, wrapWithBlockRoot(err, root, "commitments")
 	}
 
-	included, cellsPerBlob, proofsPerBlob, err := s.fetchCellsAndProofsFromExecution(ctx, commitments)
-	log.Info("Received cells and proofs from execution client", "included", included, "cells count", len(cellsPerBlob), "err", err)
+	cellsPerBlob, proofsPerBlob, err := s.fetchCellsAndProofsFromExecution(ctx, commitments)
 	if err != nil {
-		return nil, nil, wrapWithBlockRoot(err, root, "fetch cells and proofs from execution client")
+		return nil, wrapWithBlockRoot(err, root, "fetch cells and proofs from execution client")
 	}
 
-	partialColumns, err := peerdas.PartialColumns(included, cellsPerBlob, proofsPerBlob, populator)
-	haveAllBlobs := included.Count() == uint64(len(commitments))
-	log.Info("Constructed partial columns", "haveAllBlobs", haveAllBlobs)
-
-	// TODO: Check for err==nil before we go here.
-	if haveAllBlobs {
-		// Construct data column sidears from the signed block and cells and proofs.
-		roSidecars, err := peerdas.DataColumnSidecars(cellsPerBlob, proofsPerBlob, populator)
-		if err != nil {
-			return nil, nil, wrapWithBlockRoot(err, populator.Root(), "data column sidcars from column sidecar")
-		}
-
-		// Upgrade the sidecars to verified sidecars.
-		// We trust the execution layer we are connected to, so we can upgrade the sidecar into a verified one.
-		verifiedROSidecars := upgradeSidecarsToVerifiedSidecars(roSidecars)
-
-		return verifiedROSidecars, partialColumns, nil
+	// Return early if nothing is returned from the EL.
+	if len(cellsPerBlob) == 0 {
+		return nil, nil
 	}
 
+	// Construct data column sidears from the signed block and cells and proofs.
+	roSidecars, err := peerdas.DataColumnSidecars(cellsPerBlob, proofsPerBlob, populator)
 	if err != nil {
-		return nil, nil, wrapWithBlockRoot(err, populator.Root(), "partial columns from column sidecar")
+		return nil, wrapWithBlockRoot(err, populator.Root(), "data column sidcars from column sidecar")
 	}
-	return nil, partialColumns, nil
+
+	// Upgrade the sidecars to verified sidecars.
+	// We trust the execution layer we are connected to, so we can upgrade the sidecar into a verified one.
+	verifiedROSidecars := upgradeSidecarsToVerifiedSidecars(roSidecars)
+
+	return verifiedROSidecars, nil
 }
 
 // fetchCellsAndProofsFromExecution fetches cells and proofs from the execution client (using engine_getBlobsV2 execution API method)
-// The returned cells and proofs are compacted and will not contain entries for missing blobs.
-// The returned bitlist is the bitlist of which blobs are present i.e. which are the blobs we have the blob data for. This
-// will help index into the cells and proofs arrays.
-func (s *Service) fetchCellsAndProofsFromExecution(ctx context.Context, kzgCommitments [][]byte) (bitfield.Bitlist /* included parts */, [][]kzg.Cell, [][]kzg.Proof, error) {
+func (s *Service) fetchCellsAndProofsFromExecution(ctx context.Context, kzgCommitments [][]byte) ([][]kzg.Cell, [][]kzg.Proof, error) {
 	// Collect KZG hashes for all blobs.
 	versionedHashes := make([]common.Hash, 0, len(kzgCommitments))
 	for _, commitment := range kzgCommitments {
@@ -730,44 +692,24 @@ func (s *Service) fetchCellsAndProofsFromExecution(ctx context.Context, kzgCommi
 		versionedHashes = append(versionedHashes, versionedHash)
 	}
 
-	var blobAndProofs []*pb.BlobAndProofV2
-
 	// Fetch all blobsAndCellsProofs from the execution client.
-	var err error
-	useV3 := s.capabilityCache.has(GetBlobsV3)
-	if useV3 {
-		// v3 can return a partial response. V2 is all or nothing
-		blobAndProofs, err = s.GetBlobsV3(ctx, versionedHashes)
-	} else {
-		blobAndProofs, err = s.GetBlobsV2(ctx, versionedHashes)
-	}
-
+	blobAndProofV2s, err := s.GetBlobsV2(ctx, versionedHashes)
 	if err != nil {
-		return nil, nil, nil, errors.Wrapf(err, "get blobs V2/3")
+		return nil, nil, errors.Wrapf(err, "get blobs V2")
 	}
 
-	// Track complete vs partial responses for V3
-	if useV3 && len(blobAndProofs) > 0 {
-		nonNilCount := 0
-		for _, bp := range blobAndProofs {
-			if bp != nil {
-				nonNilCount++
-			}
-		}
-		if nonNilCount == len(versionedHashes) {
-			getBlobsV3CompleteResponsesTotal.Inc()
-		} else if nonNilCount > 0 {
-			getBlobsV3PartialResponsesTotal.Inc()
-		}
+	// Return early if nothing is returned from the EL.
+	if len(blobAndProofV2s) == 0 {
+		return nil, nil, nil
 	}
 
 	// Compute cells and proofs from the blobs and cell proofs.
-	included, cellsPerBlob, proofsPerBlob, err := peerdas.ComputeCellsAndProofsFromStructured(uint64(len(kzgCommitments)), blobAndProofs)
+	cellsPerBlob, proofsPerBlob, err := peerdas.ComputeCellsAndProofsFromStructured(blobAndProofV2s)
 	if err != nil {
-		return nil, nil, nil, errors.Wrap(err, "compute cells and proofs")
+		return nil, nil, errors.Wrap(err, "compute cells and proofs")
 	}
 
-	return included, cellsPerBlob, proofsPerBlob, nil
+	return cellsPerBlob, proofsPerBlob, nil
 }
 
 // upgradeSidecarsToVerifiedSidecars upgrades a list of data column sidecars into verified data column sidecars.

beacon-chain/execution/engine_client_test.go

@@ -2587,7 +2587,7 @@ func TestConstructDataColumnSidecars(t *testing.T) {
 	ctx := context.Background()
 
 	t.Run("GetBlobsV2 is not supported", func(t *testing.T) {
-		_, _, err := client.ConstructDataColumnSidecars(ctx, peerdas.PopulateFromBlock(roBlock))
+		_, err := client.ConstructDataColumnSidecars(ctx, peerdas.PopulateFromBlock(roBlock))
 		require.ErrorContains(t, "engine_getBlobsV2 is not supported", err)
 	})
 
@@ -2598,7 +2598,7 @@ func TestConstructDataColumnSidecars(t *testing.T) {
 		rpcClient, client := setupRpcClientV2(t, srv.URL, client)
 		defer rpcClient.Close()
 
-		dataColumns, _, err := client.ConstructDataColumnSidecars(ctx, peerdas.PopulateFromBlock(roBlock))
+		dataColumns, err := client.ConstructDataColumnSidecars(ctx, peerdas.PopulateFromBlock(roBlock))
 		require.NoError(t, err)
 		require.Equal(t, 0, len(dataColumns))
 	})
@@ -2611,7 +2611,7 @@ func TestConstructDataColumnSidecars(t *testing.T) {
 		rpcClient, client := setupRpcClientV2(t, srv.URL, client)
 		defer rpcClient.Close()
 
-		dataColumns, _, err := client.ConstructDataColumnSidecars(ctx, peerdas.PopulateFromBlock(roBlock))
+		dataColumns, err := client.ConstructDataColumnSidecars(ctx, peerdas.PopulateFromBlock(roBlock))
 		require.NoError(t, err)
 		require.Equal(t, 128, len(dataColumns))
 	})
@@ -2683,7 +2683,7 @@ func createBlobServerV2(t *testing.T, numBlobs int, blobMasks []bool) *httptest.
 				Blob:      []byte("0xblob"),
 				KzgProofs: []hexutil.Bytes{},
 			}
-			for range fieldparams.NumberOfColumns {
+			for j := 0; j < int(params.BeaconConfig().NumberOfColumns); j++ {
 				cellProof := make([]byte, 48)
 				blobAndCellProofs[i].KzgProofs = append(blobAndCellProofs[i].KzgProofs, cellProof)
 			}

beacon-chain/execution/metrics.go

@@ -27,32 +27,6 @@ var (
 			Buckets: []float64{25, 50, 100, 200, 500, 1000, 2000, 4000},
 		},
 	)
-	getBlobsV2Latency = promauto.NewHistogram(
-		prometheus.HistogramOpts{
-			Name:    "get_blobs_v2_latency_milliseconds",
-			Help:    "Captures RPC latency for getBlobsV2 in milliseconds",
-			Buckets: []float64{25, 50, 100, 200, 500, 1000, 2000, 4000},
-		},
-	)
-	getBlobsV3RequestsTotal = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "beacon_engine_getBlobsV3_requests_total",
-		Help: "Total number of engine_getBlobsV3 requests sent",
-	})
-	getBlobsV3CompleteResponsesTotal = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "beacon_engine_getBlobsV3_complete_responses_total",
-		Help: "Total number of complete engine_getBlobsV3 successful responses received",
-	})
-	getBlobsV3PartialResponsesTotal = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "beacon_engine_getBlobsV3_partial_responses_total",
-		Help: "Total number of engine_getBlobsV3 partial responses received",
-	})
-	getBlobsV3Latency = promauto.NewHistogram(
-		prometheus.HistogramOpts{
-			Name:    "beacon_engine_getBlobsV3_request_duration_seconds",
-			Help:    "Duration of engine_getBlobsV3 requests in seconds",
-			Buckets: []float64{0.025, 0.05, 0.1, 0.2, 0.5, 1, 2, 4},
-		},
-	)
 	errParseCount = promauto.NewCounter(prometheus.CounterOpts{
 		Name: "execution_parse_error_count",
 		Help: "The number of errors that occurred while parsing execution payload",

beacon-chain/execution/testing/mock_engine_client.go

@@ -118,8 +118,8 @@ func (e *EngineClient) ReconstructBlobSidecars(context.Context, interfaces.ReadO
 }
 
 // ConstructDataColumnSidecars is a mock implementation of the ConstructDataColumnSidecars method.
-func (e *EngineClient) ConstructDataColumnSidecars(context.Context, peerdas.ConstructionPopulator) ([]blocks.VerifiedRODataColumn, []blocks.PartialDataColumn, error) {
-	return e.DataColumnSidecars, nil, e.ErrorDataColumnSidecars
+func (e *EngineClient) ConstructDataColumnSidecars(context.Context, peerdas.ConstructionPopulator) ([]blocks.VerifiedRODataColumn, error) {
+	return e.DataColumnSidecars, e.ErrorDataColumnSidecars
 }
 
 // GetTerminalBlockHash --

beacon-chain/forkchoice/doubly-linked-tree/forkchoice.go

@@ -240,7 +240,7 @@ func (f *ForkChoice) IsViableForCheckpoint(cp *forkchoicetypes.Checkpoint) (bool
 	if node.slot == epochStart {
 		return true, nil
 	}
-	if !features.Get().IgnoreUnviableAttestations {
+	if !features.Get().DisableLastEpochTargets {
 		// Allow any node from the checkpoint epoch - 1 to be viable.
 		nodeEpoch := slots.ToEpoch(node.slot)
 		if nodeEpoch+1 == cp.Epoch {
@@ -642,12 +642,8 @@ func (f *ForkChoice) DependentRootForEpoch(root [32]byte, epoch primitives.Epoch
 	if !ok || node == nil {
 		return [32]byte{}, ErrNilNode
 	}
-	if slots.ToEpoch(node.slot) >= epoch {
-		if node.parent != nil {
-			node = node.parent
-		} else {
-			return f.store.finalizedDependentRoot, nil
-		}
+	if slots.ToEpoch(node.slot) >= epoch && node.parent != nil {
+		node = node.parent
 	}
 	return node.root, nil
 }

beacon-chain/forkchoice/doubly-linked-tree/store.go

@@ -212,9 +212,6 @@ func (s *Store) prune(ctx context.Context) error {
 		return nil
 	}
 
-	// Save the new finalized dependent root because it will be pruned
-	s.finalizedDependentRoot = finalizedNode.parent.root
-
 	// Prune nodeByRoot starting from root
 	if err := s.pruneFinalizedNodeByRootMap(ctx, s.treeRootNode, finalizedNode); err != nil {
 		return err

beacon-chain/forkchoice/doubly-linked-tree/store_test.go

@@ -465,7 +465,6 @@ func TestStore_TargetRootForEpoch(t *testing.T) {
 	ctx := t.Context()
 	f := setup(1, 1)
 
-	// Insert a block in slot 32
 	state, blk, err := prepareForkchoiceState(ctx, params.BeaconConfig().SlotsPerEpoch, [32]byte{'a'}, params.BeaconConfig().ZeroHash, params.BeaconConfig().ZeroHash, 1, 1)
 	require.NoError(t, err)
 	require.NoError(t, f.InsertNode(ctx, state, blk))
@@ -476,7 +475,6 @@ func TestStore_TargetRootForEpoch(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, dependent, [32]byte{})
 
-	// Insert a block in slot 33
 	state, blk1, err := prepareForkchoiceState(ctx, params.BeaconConfig().SlotsPerEpoch+1, [32]byte{'b'}, blk.Root(), params.BeaconConfig().ZeroHash, 1, 1)
 	require.NoError(t, err)
 	require.NoError(t, f.InsertNode(ctx, state, blk1))
@@ -490,7 +488,7 @@ func TestStore_TargetRootForEpoch(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, dependent, [32]byte{})
 
-	// Insert a block for the next epoch (missed slot 0), slot 65
+	// Insert a block for the next epoch (missed slot 0)
 
 	state, blk2, err := prepareForkchoiceState(ctx, 2*params.BeaconConfig().SlotsPerEpoch+1, [32]byte{'c'}, blk1.Root(), params.BeaconConfig().ZeroHash, 1, 1)
 	require.NoError(t, err)
@@ -511,7 +509,6 @@ func TestStore_TargetRootForEpoch(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, dependent, blk1.Root())
 
-	// Insert a block at slot 66
 	state, blk3, err := prepareForkchoiceState(ctx, 2*params.BeaconConfig().SlotsPerEpoch+2, [32]byte{'d'}, blk2.Root(), params.BeaconConfig().ZeroHash, 1, 1)
 	require.NoError(t, err)
 	require.NoError(t, f.InsertNode(ctx, state, blk3))
@@ -536,11 +533,8 @@ func TestStore_TargetRootForEpoch(t *testing.T) {
 	dependent, err = f.DependentRoot(1)
 	require.NoError(t, err)
 	require.Equal(t, [32]byte{}, dependent)
-	dependent, err = f.DependentRoot(2)
-	require.NoError(t, err)
-	require.Equal(t, blk1.Root(), dependent)
 
-	// Insert a block for the next epoch, slot 96 (descends from finalized at slot 33)
+	// Insert a block for next epoch (slot 0 present)
 	state, blk4, err := prepareForkchoiceState(ctx, 3*params.BeaconConfig().SlotsPerEpoch, [32]byte{'e'}, blk1.Root(), params.BeaconConfig().ZeroHash, 1, 1)
 	require.NoError(t, err)
 	require.NoError(t, f.InsertNode(ctx, state, blk4))
@@ -557,7 +551,6 @@ func TestStore_TargetRootForEpoch(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, dependent, blk1.Root())
 
-	// Insert a block at slot 97
 	state, blk5, err := prepareForkchoiceState(ctx, 3*params.BeaconConfig().SlotsPerEpoch+1, [32]byte{'f'}, blk4.Root(), params.BeaconConfig().ZeroHash, 1, 1)
 	require.NoError(t, err)
 	require.NoError(t, f.InsertNode(ctx, state, blk5))
@@ -607,16 +600,12 @@ func TestStore_TargetRootForEpoch(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, target, blk1.Root())
 
-	// Prune finalization, finalize the block at slot 96
+	// Prune finalization
 	s.finalizedCheckpoint.Root = blk4.Root()
 	require.NoError(t, s.prune(ctx))
 	target, err = f.TargetRootForEpoch(blk4.Root(), 3)
 	require.NoError(t, err)
 	require.Equal(t, blk4.Root(), target)
-	// Dependent root for the finalized block should be the root of the pruned block at slot 33
-	dependent, err = f.DependentRootForEpoch(blk4.Root(), 3)
-	require.NoError(t, err)
-	require.Equal(t, blk1.Root(), dependent)
 }
 
 func TestStore_DependentRootForEpoch(t *testing.T) {

beacon-chain/forkchoice/doubly-linked-tree/types.go

@@ -31,7 +31,6 @@ type Store struct {
 	proposerBoostRoot             [fieldparams.RootLength]byte           // latest block root that was boosted after being received in a timely manner.
 	previousProposerBoostRoot     [fieldparams.RootLength]byte           // previous block root that was boosted after being received in a timely manner.
 	previousProposerBoostScore    uint64                                 // previous proposer boosted root score.
-	finalizedDependentRoot        [fieldparams.RootLength]byte           // dependent root at finalized checkpoint.
 	committeeWeight               uint64                                 // tracks the total active validator balance divided by the number of slots per Epoch.
 	treeRootNode                  *Node                                  // the root node of the store tree.
 	headNode                      *Node                                  // last head Node

beacon-chain/graffiti/graffiti-proposal-brief.md

@@ -1,95 +0,0 @@
-# Graffiti Version Info Implementation
-
-## Summary
-Add automatic EL+CL version info to block graffiti following [ethereum/execution-apis#517](https://github.com/ethereum/execution-apis/pull/517). Uses the [flexible standard](https://hackmd.io/@wmoBhF17RAOH2NZ5bNXJVg/BJX2c9gja) to pack client info into leftover space after user graffiti.
-
-More details: https://github.com/ethereum/execution-apis/blob/main/src/engine/identification.md 
-
-## Implementation
-
-### Core Component: GraffitiInfo Struct
-Thread-safe struct holding version information:
-```go
-const clCode = "PR"
-
-type GraffitiInfo struct {
-    mu           sync.RWMutex
-    userGraffiti string  // From --graffiti flag (set once at startup)
-    clCommit     string  // From version.GetCommitPrefix() helper function
-    elCode       string  // From engine_getClientVersionV1
-    elCommit     string  // From engine_getClientVersionV1
-}
-```
-
-### Flow
-1. **Startup**: Parse flags, create GraffitiInfo with user graffiti and CL info.
-2. **Wiring**: Pass struct to both execution service and RPC validator server
-3. **Runtime**: Execution service goroutine periodically calls `engine_getClientVersionV1` and updates EL fields
-4. **Block Proposal**: RPC validator server calls `GenerateGraffiti()` to get formatted graffiti
-
-### Flexible Graffiti Format
-Packs as much client info as space allows (after user graffiti):
-
-| Available Space | Format | Example |
-|----------------|--------|---------|
-| ≥12 bytes | `EL(2)+commit(4)+CL(2)+commit(4)+user` | `GE168dPR63afBob` |
-| 8-11 bytes | `EL(2)+commit(2)+CL(2)+commit(2)+user` | `GE16PR63my node here` |
-| 4-7 bytes | `EL(2)+CL(2)+user` | `GEPRthis is my graffiti msg` |
-| 2-3 bytes | `EL(2)+user` | `GEalmost full graffiti message` |
-| <2 bytes | user only | `full 32 byte user graffiti here` |
-
-```go
-func (g *GraffitiInfo) GenerateGraffiti() [32]byte {
-    available := 32 - len(userGraffiti)
-
-    if elCode == "" {
-        elCommit2 = elCommit4 = ""
-    }
-
-    switch {
-    case available >= 12:
-        return elCode + elCommit4 + clCode + clCommit4 + userGraffiti
-    case available >= 8:
-        return elCode + elCommit2 + clCode + clCommit2 + userGraffiti
-    case available >= 4:
-        return elCode + clCode + userGraffiti
-    case available >= 2:
-        return elCode + userGraffiti
-    default:
-        return userGraffiti
-    }
-}
-```
-
-### Update Logic
-Single testable function in execution service:
-```go
-func (s *Service) updateGraffitiInfo() {
-    versions, err := s.GetClientVersion(ctx)
-    if err != nil {
-        return  // Keep last good value
-    }
-    if len(versions) == 1 {
-        s.graffitiInfo.UpdateFromEngine(versions[0].Code, versions[0].Commit)
-    }
-}
-```
-
-Goroutine calls this on `slot % 8 == 4` timing (4 times per epoch, avoids slot boundaries).
-
-### Files Changes Required
-
-**New:**
-- `beacon-chain/execution/graffiti_info.go` - The struct and methods
-- `beacon-chain/execution/graffiti_info_test.go` - Unit tests
-- `runtime/version/version.go` - Add `GetCommitPrefix()` helper that extracts first 4 hex chars from the git commit injected via Bazel ldflags at build time
-
-**Modified:**
-- `beacon-chain/execution/service.go` - Add goroutine + updateGraffitiInfo()
-- `beacon-chain/execution/engine_client.go` - Add GetClientVersion() method that does engine call
-- `beacon-chain/rpc/.../validator/proposer.go` - Call GenerateGraffiti()
-- `beacon-chain/node/node.go` - Wire GraffitiInfo to services
-
-### Testing Strategy
-- Unit test GraffitiInfo methods (priority logic, thread safety)
-- Unit test updateGraffitiInfo() with mocked engine client

beacon-chain/node/BUILD.bazel

@@ -23,7 +23,6 @@ go_library(
         "//beacon-chain/builder:go_default_library",
         "//beacon-chain/cache:go_default_library",
         "//beacon-chain/cache/depositsnapshot:go_default_library",
-        "//beacon-chain/das:go_default_library",
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/db/filesystem:go_default_library",
         "//beacon-chain/db/kv:go_default_library",

beacon-chain/node/node.go

@@ -26,7 +26,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/builder"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/cache"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/cache/depositsnapshot"
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/das"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/db"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/db/filesystem"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/db/kv"
@@ -117,7 +116,7 @@ type BeaconNode struct {
 	GenesisProviders         []genesis.Provider
 	CheckpointInitializer    checkpoint.Initializer
 	forkChoicer              forkchoice.ForkChoicer
-	ClockWaiter              startup.ClockWaiter
+	clockWaiter              startup.ClockWaiter
 	BackfillOpts             []backfill.ServiceOption
 	initialSyncComplete      chan struct{}
 	BlobStorage              *filesystem.BlobStorage
@@ -130,7 +129,6 @@ type BeaconNode struct {
 	slasherEnabled           bool
 	lcStore                  *lightclient.Store
 	ConfigOptions            []params.Option
-	SyncNeedsWaiter          func() (das.SyncNeeds, error)
 }
 
 // New creates a new node instance, sets up configuration options, and registers
@@ -195,7 +193,7 @@ func New(cliCtx *cli.Context, cancel context.CancelFunc, opts ...Option) (*Beaco
 	params.LogDigests(params.BeaconConfig())
 
 	synchronizer := startup.NewClockSynchronizer()
-	beacon.ClockWaiter = synchronizer
+	beacon.clockWaiter = synchronizer
 	beacon.forkChoicer = doublylinkedtree.New()
 
 	depositAddress, err := execution.DepositContractAddress()
@@ -235,13 +233,12 @@ func New(cliCtx *cli.Context, cancel context.CancelFunc, opts ...Option) (*Beaco
 
 	beacon.lhsp = &verification.LazyHeadStateProvider{}
 	beacon.verifyInitWaiter = verification.NewInitializerWaiter(
-		beacon.ClockWaiter, forkchoice.NewROForkChoice(beacon.forkChoicer), beacon.stateGen, beacon.lhsp)
+		beacon.clockWaiter, forkchoice.NewROForkChoice(beacon.forkChoicer), beacon.stateGen, beacon.lhsp)
 
 	beacon.BackfillOpts = append(
 		beacon.BackfillOpts,
 		backfill.WithVerifierWaiter(beacon.verifyInitWaiter),
 		backfill.WithInitSyncWaiter(initSyncWaiter(ctx, beacon.initialSyncComplete)),
-		backfill.WithSyncNeedsWaiter(beacon.SyncNeedsWaiter),
 	)
 
 	if err := registerServices(cliCtx, beacon, synchronizer, bfs); err != nil {
@@ -667,8 +664,7 @@ func (b *BeaconNode) registerP2P(cliCtx *cli.Context) error {
 		StateNotifier:         b,
 		DB:                    b.db,
 		StateGen:              b.stateGen,
-		ClockWaiter:           b.ClockWaiter,
-		PartialDataColumns:    b.cliCtx.Bool(flags.PartialDataColumns.Name),
+		ClockWaiter:           b.clockWaiter,
 	})
 	if err != nil {
 		return err
@@ -710,7 +706,7 @@ func (b *BeaconNode) registerSlashingPoolService() error {
 		return err
 	}
 
-	s := slashings.NewPoolService(b.ctx, b.slashingsPool, slashings.WithElectraTimer(b.ClockWaiter, chainService.CurrentSlot))
+	s := slashings.NewPoolService(b.ctx, b.slashingsPool, slashings.WithElectraTimer(b.clockWaiter, chainService.CurrentSlot))
 	return b.services.RegisterService(s)
 }
 
@@ -832,7 +828,7 @@ func (b *BeaconNode) registerSyncService(initialSyncComplete chan struct{}, bFil
 		regularsync.WithSlasherAttestationsFeed(b.slasherAttestationsFeed),
 		regularsync.WithSlasherBlockHeadersFeed(b.slasherBlockHeadersFeed),
 		regularsync.WithReconstructor(web3Service),
-		regularsync.WithClockWaiter(b.ClockWaiter),
+		regularsync.WithClockWaiter(b.clockWaiter),
 		regularsync.WithInitialSyncComplete(initialSyncComplete),
 		regularsync.WithStateNotifier(b),
 		regularsync.WithBlobStorage(b.BlobStorage),
@@ -863,8 +859,7 @@ func (b *BeaconNode) registerInitialSyncService(complete chan struct{}) error {
 		P2P:                 b.fetchP2P(),
 		StateNotifier:       b,
 		BlockNotifier:       b,
-		ClockWaiter:         b.ClockWaiter,
-		SyncNeedsWaiter:     b.SyncNeedsWaiter,
+		ClockWaiter:         b.clockWaiter,
 		InitialSyncComplete: complete,
 		BlobStorage:         b.BlobStorage,
 		DataColumnStorage:   b.DataColumnStorage,
@@ -895,7 +890,7 @@ func (b *BeaconNode) registerSlasherService() error {
 		SlashingPoolInserter:    b.slashingsPool,
 		SyncChecker:             syncService,
 		HeadStateFetcher:        chainService,
-		ClockWaiter:             b.ClockWaiter,
+		ClockWaiter:             b.clockWaiter,
 	})
 	if err != nil {
 		return err
@@ -988,7 +983,7 @@ func (b *BeaconNode) registerRPCService(router *http.ServeMux) error {
 		MaxMsgSize:                maxMsgSize,
 		BlockBuilder:              b.fetchBuilderService(),
 		Router:                    router,
-		ClockWaiter:               b.ClockWaiter,
+		ClockWaiter:               b.clockWaiter,
 		BlobStorage:               b.BlobStorage,
 		DataColumnStorage:         b.DataColumnStorage,
 		TrackedValidatorsCache:    b.trackedValidatorsCache,
@@ -1133,7 +1128,7 @@ func (b *BeaconNode) registerPrunerService(cliCtx *cli.Context) error {
 
 func (b *BeaconNode) RegisterBackfillService(cliCtx *cli.Context, bfs *backfill.Store) error {
 	pa := peers.NewAssigner(b.fetchP2P().Peers(), b.forkChoicer)
-	bf, err := backfill.NewService(cliCtx.Context, bfs, b.BlobStorage, b.DataColumnStorage, b.ClockWaiter, b.fetchP2P(), pa, b.BackfillOpts...)
+	bf, err := backfill.NewService(cliCtx.Context, bfs, b.BlobStorage, b.DataColumnStorage, b.clockWaiter, b.fetchP2P(), pa, b.BackfillOpts...)
 	if err != nil {
 		return errors.Wrap(err, "error initializing backfill service")
 	}

beacon-chain/p2p/BUILD.bazel

@@ -51,7 +51,6 @@ go_library(
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/db/kv:go_default_library",
         "//beacon-chain/p2p/encoder:go_default_library",
-        "//beacon-chain/p2p/partialdatacolumnbroadcaster:go_default_library",
         "//beacon-chain/p2p/peers:go_default_library",
         "//beacon-chain/p2p/peers/peerdata:go_default_library",
         "//beacon-chain/p2p/peers/scorers:go_default_library",

beacon-chain/p2p/broadcaster.go

@@ -342,7 +342,7 @@ func (s *Service) BroadcastLightClientFinalityUpdate(ctx context.Context, update
 // there is at least one peer in each needed subnet. If not, it will attempt to find one before broadcasting.
 // This function is non-blocking. It stops trying to broadcast a given sidecar when more than one slot has passed, or the context is
 // cancelled (whichever comes first).
-func (s *Service) BroadcastDataColumnSidecars(ctx context.Context, sidecars []blocks.VerifiedRODataColumn, partialColumns []blocks.PartialDataColumn) error {
+func (s *Service) BroadcastDataColumnSidecars(ctx context.Context, sidecars []blocks.VerifiedRODataColumn) error {
 	// Increase the number of broadcast attempts.
 	dataColumnSidecarBroadcastAttempts.Add(float64(len(sidecars)))
 
@@ -352,7 +352,7 @@ func (s *Service) BroadcastDataColumnSidecars(ctx context.Context, sidecars []bl
 		return errors.Wrap(err, "current fork digest")
 	}
 
-	go s.broadcastDataColumnSidecars(ctx, forkDigest, sidecars, partialColumns)
+	go s.broadcastDataColumnSidecars(ctx, forkDigest, sidecars)
 
 	return nil
 }
@@ -360,7 +360,7 @@ func (s *Service) BroadcastDataColumnSidecars(ctx context.Context, sidecars []bl
 // broadcastDataColumnSidecars broadcasts multiple data column sidecars to the p2p network, after ensuring
 // there is at least one peer in each needed subnet. If not, it will attempt to find one before broadcasting.
 // It returns when all broadcasts are complete, or the context is cancelled (whichever comes first).
-func (s *Service) broadcastDataColumnSidecars(ctx context.Context, forkDigest [fieldparams.VersionLength]byte, sidecars []blocks.VerifiedRODataColumn, partialColumns []blocks.PartialDataColumn) {
+func (s *Service) broadcastDataColumnSidecars(ctx context.Context, forkDigest [fieldparams.VersionLength]byte, sidecars []blocks.VerifiedRODataColumn) {
 	type rootAndIndex struct {
 		root  [fieldparams.RootLength]byte
 		index uint64
@@ -374,7 +374,7 @@ func (s *Service) broadcastDataColumnSidecars(ctx context.Context, forkDigest [f
 	logLevel := logrus.GetLevel()
 
 	slotPerRoot := make(map[[fieldparams.RootLength]byte]primitives.Slot, 1)
-	for i, sidecar := range sidecars {
+	for _, sidecar := range sidecars {
 		slotPerRoot[sidecar.BlockRoot()] = sidecar.Slot()
 
 		wg.Go(func() {
@@ -398,14 +398,6 @@ func (s *Service) broadcastDataColumnSidecars(ctx context.Context, forkDigest [f
 				return
 			}
 
-			if s.partialColumnBroadcaster != nil && i < len(partialColumns) {
-				fullTopicStr := topic + s.Encoding().ProtocolSuffix()
-				if err := s.partialColumnBroadcaster.Publish(fullTopicStr, partialColumns[i]); err != nil {
-					tracing.AnnotateError(span, err)
-					log.WithError(err).Error("Cannot partial broadcast data column sidecar")
-				}
-			}
-
 			// Broadcast the data column sidecar to the network.
 			if err := s.broadcastObject(ctx, sidecar, topic); err != nil {
 				tracing.AnnotateError(span, err)

beacon-chain/p2p/broadcaster_test.go

@@ -773,7 +773,7 @@ func TestService_BroadcastDataColumn(t *testing.T) {
 	time.Sleep(50 * time.Millisecond)
 
 	// Broadcast to peers and wait.
-	err = service.BroadcastDataColumnSidecars(ctx, []blocks.VerifiedRODataColumn{verifiedRoSidecar}, nil)
+	err = service.BroadcastDataColumnSidecars(ctx, []blocks.VerifiedRODataColumn{verifiedRoSidecar})
 	require.NoError(t, err)
 
 	// Receive the message.

beacon-chain/p2p/config.go

@@ -26,7 +26,6 @@ const (
 // Config for the p2p service. These parameters are set from application level flags
 // to initialize the p2p service.
 type Config struct {
-	PartialDataColumns    bool
 	NoDiscovery           bool
 	EnableUPnP            bool
 	StaticPeerID          bool

beacon-chain/p2p/interfaces.go

@@ -4,7 +4,6 @@ import (
 	"context"
 
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/encoder"
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/partialdatacolumnbroadcaster"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/peers"
 	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
@@ -29,7 +28,6 @@ type (
 		Broadcaster
 		SetStreamHandler
 		PubSubProvider
-		PartialColumnBroadcasterProvider
 		PubSubTopicUser
 		SenderEncoder
 		PeerManager
@@ -54,7 +52,7 @@ type (
 		BroadcastBlob(ctx context.Context, subnet uint64, blob *ethpb.BlobSidecar) error
 		BroadcastLightClientOptimisticUpdate(ctx context.Context, update interfaces.LightClientOptimisticUpdate) error
 		BroadcastLightClientFinalityUpdate(ctx context.Context, update interfaces.LightClientFinalityUpdate) error
-		BroadcastDataColumnSidecars(ctx context.Context, sidecars []blocks.VerifiedRODataColumn, partialColumns []blocks.PartialDataColumn) error
+		BroadcastDataColumnSidecars(ctx context.Context, sidecars []blocks.VerifiedRODataColumn) error
 	}
 
 	// SetStreamHandler configures p2p to handle streams of a certain topic ID.
@@ -94,11 +92,6 @@ type (
 		PubSub() *pubsub.PubSub
 	}
 
-	// PubSubProvider provides the p2p pubsub protocol.
-	PartialColumnBroadcasterProvider interface {
-		PartialColumnBroadcaster() *partialdatacolumnbroadcaster.PartialColumnBroadcaster
-	}
-
 	// PeerManager abstracts some peer management methods from libp2p.
 	PeerManager interface {
 		Disconnect(peer.ID) error

beacon-chain/p2p/monitoring.go

@@ -157,11 +157,6 @@ var (
 		Help: "The number of publish messages received via rpc for a particular topic",
 	},
 		[]string{"topic"})
-	pubsubRPCPubRecvSize = promauto.NewCounterVec(prometheus.CounterOpts{
-		Name: "p2p_pubsub_rpc_recv_pub_size_total",
-		Help: "The total size of publish messages received via rpc for a particular topic",
-	},
-		[]string{"topic", "is_partial"})
 	pubsubRPCDrop = promauto.NewCounterVec(prometheus.CounterOpts{
 		Name: "p2p_pubsub_rpc_drop_total",
 		Help: "The number of messages dropped via rpc for a particular control message",
@@ -176,11 +171,6 @@ var (
 		Help: "The number of publish messages dropped via rpc for a particular topic",
 	},
 		[]string{"topic"})
-	pubsubRPCPubDropSize = promauto.NewCounterVec(prometheus.CounterOpts{
-		Name: "p2p_pubsub_rpc_drop_pub_size_total",
-		Help: "The total size of publish messages dropped via rpc for a particular topic",
-	},
-		[]string{"topic", "is_partial"})
 	pubsubRPCSent = promauto.NewCounterVec(prometheus.CounterOpts{
 		Name: "p2p_pubsub_rpc_sent_total",
 		Help: "The number of messages sent via rpc for a particular control message",
@@ -195,11 +185,6 @@ var (
 		Help: "The number of publish messages sent via rpc for a particular topic",
 	},
 		[]string{"topic"})
-	pubsubRPCPubSentSize = promauto.NewCounterVec(prometheus.CounterOpts{
-		Name: "gossipsub_pubsub_rpc_sent_pub_size_total",
-		Help: "The total size of publish messages sent via rpc for a particular topic",
-	},
-		[]string{"topic", "is_partial"})
 )
 
 func (s *Service) updateMetrics() {

beacon-chain/p2p/partialdatacolumnbroadcaster/BUILD.bazel

@@ -1,27 +0,0 @@
-load("@prysm//tools/go:def.bzl", "go_library")
-
-go_library(
-    name = "go_default_library",
-    srcs = [
-        "metrics.go",
-        "partial.go",
-    ],
-    importpath = "github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/partialdatacolumnbroadcaster",
-    visibility = ["//visibility:public"],
-    deps = [
-        "//config/params:go_default_library",
-        "//consensus-types/blocks:go_default_library",
-        "//internal/logrusadapter:go_default_library",
-        "//proto/prysm/v1alpha1:go_default_library",
-        "@com_github_libp2p_go_libp2p//core/peer:go_default_library",
-        "@com_github_libp2p_go_libp2p_pubsub//:go_default_library",
-        "@com_github_libp2p_go_libp2p_pubsub//partialmessages:go_default_library",
-        "@com_github_libp2p_go_libp2p_pubsub//partialmessages/bitmap:go_default_library",
-        "@com_github_libp2p_go_libp2p_pubsub//pb:go_default_library",
-        "@com_github_pkg_errors//:go_default_library",
-        "@com_github_prometheus_client_golang//prometheus:go_default_library",
-        "@com_github_prometheus_client_golang//prometheus/promauto:go_default_library",
-        "@com_github_prysmaticlabs_go_bitfield//:go_default_library",
-        "@com_github_sirupsen_logrus//:go_default_library",
-    ],
-)

beacon-chain/p2p/partialdatacolumnbroadcaster/integrationtest/BUILD.bazel

@@ -1,25 +0,0 @@
-load("@prysm//tools/go:def.bzl", "go_test")
-
-go_test(
-    name = "go_default_test",
-    size = "medium",
-    srcs = ["two_node_test.go"],
-    deps = [
-        "//beacon-chain/core/peerdas:go_default_library",
-        "//beacon-chain/p2p:go_default_library",
-        "//beacon-chain/p2p/encoder:go_default_library",
-        "//beacon-chain/p2p/partialdatacolumnbroadcaster:go_default_library",
-        "//config/fieldparams:go_default_library",
-        "//config/params:go_default_library",
-        "//consensus-types/blocks:go_default_library",
-        "//proto/prysm/v1alpha1:go_default_library",
-        "//testing/assert:go_default_library",
-        "//testing/require:go_default_library",
-        "//testing/util:go_default_library",
-        "@com_github_libp2p_go_libp2p//core/peer:go_default_library",
-        "@com_github_libp2p_go_libp2p//x/simlibp2p:go_default_library",
-        "@com_github_libp2p_go_libp2p_pubsub//:go_default_library",
-        "@com_github_marcopolo_simnet//:go_default_library",
-        "@com_github_sirupsen_logrus//:go_default_library",
-    ],
-)

beacon-chain/p2p/partialdatacolumnbroadcaster/integrationtest/two_node_test.go

@@ -1,238 +0,0 @@
-package integrationtest
-
-import (
-	"context"
-	"crypto/rand"
-	"fmt"
-	"testing"
-	"testing/synctest"
-	"time"
-
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/peerdas"
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p"
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/encoder"
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/partialdatacolumnbroadcaster"
-	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
-	"github.com/OffchainLabs/prysm/v7/config/params"
-	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
-	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
-	"github.com/OffchainLabs/prysm/v7/testing/assert"
-	"github.com/OffchainLabs/prysm/v7/testing/require"
-	"github.com/OffchainLabs/prysm/v7/testing/util"
-	pubsub "github.com/libp2p/go-libp2p-pubsub"
-	"github.com/libp2p/go-libp2p/core/peer"
-	simlibp2p "github.com/libp2p/go-libp2p/x/simlibp2p"
-	"github.com/marcopolo/simnet"
-	"github.com/sirupsen/logrus"
-)
-
-// TestTwoNodePartialColumnExchange tests that two nodes can exchange partial columns
-// and reconstruct the complete column. Node 1 has cells 0-2, Node 2 has cells 3-5.
-// After exchange, both should have all cells.
-func TestTwoNodePartialColumnExchange(t *testing.T) {
-	synctest.Test(t, func(t *testing.T) {
-		// Create a simulated libp2p network
-		latency := time.Millisecond * 10
-		network, meta, err := simlibp2p.SimpleLibp2pNetwork([]simlibp2p.NodeLinkSettingsAndCount{
-			{LinkSettings: simnet.NodeBiDiLinkSettings{
-				Downlink: simnet.LinkSettings{BitsPerSecond: 20 * simlibp2p.OneMbps, Latency: latency / 2},
-				Uplink:   simnet.LinkSettings{BitsPerSecond: 20 * simlibp2p.OneMbps, Latency: latency / 2},
-			}, Count: 2},
-		}, simlibp2p.NetworkSettings{UseBlankHost: true})
-		require.NoError(t, err)
-		require.NoError(t, network.Start())
-		defer func() {
-			require.NoError(t, network.Close())
-		}()
-		defer func() {
-			for _, node := range meta.Nodes {
-				err := node.Close()
-				if err != nil {
-					panic(err)
-				}
-			}
-		}()
-
-		h1 := meta.Nodes[0]
-		h2 := meta.Nodes[1]
-
-		logger := logrus.New()
-		logger.SetLevel(logrus.DebugLevel)
-		broadcaster1 := partialdatacolumnbroadcaster.NewBroadcaster(logger)
-		broadcaster2 := partialdatacolumnbroadcaster.NewBroadcaster(logger)
-
-		opts1 := broadcaster1.AppendPubSubOpts([]pubsub.Option{
-			pubsub.WithMessageSigning(false),
-			pubsub.WithStrictSignatureVerification(false),
-		})
-		opts2 := broadcaster2.AppendPubSubOpts([]pubsub.Option{
-			pubsub.WithMessageSigning(false),
-			pubsub.WithStrictSignatureVerification(false),
-		})
-
-		ctx, cancel := context.WithCancel(context.Background())
-		defer cancel()
-		ps1, err := pubsub.NewGossipSub(ctx, h1, opts1...)
-		require.NoError(t, err)
-		ps2, err := pubsub.NewGossipSub(ctx, h2, opts2...)
-		require.NoError(t, err)
-
-		go broadcaster1.Start()
-		go broadcaster2.Start()
-		defer func() {
-			broadcaster1.Stop()
-			broadcaster2.Stop()
-		}()
-
-		// Generate Test Data
-		var blockRoot [fieldparams.RootLength]byte
-		copy(blockRoot[:], []byte("test-block-root"))
-
-		numCells := 6
-		commitments := make([][]byte, numCells)
-		cells := make([][]byte, numCells)
-		proofs := make([][]byte, numCells)
-
-		for i := range numCells {
-			commitments[i] = make([]byte, 48)
-
-			cells[i] = make([]byte, 2048)
-			rand.Read(cells[i])
-			proofs[i] = make([]byte, 48)
-			fmt.Appendf(proofs[i][:0], "proof %d", i)
-		}
-
-		roDC, _ := util.CreateTestVerifiedRoDataColumnSidecars(t, []util.DataColumnParam{
-			{
-				BodyRoot:       blockRoot[:],
-				KzgCommitments: commitments,
-				Column:         cells,
-				KzgProofs:      proofs,
-			},
-		})
-
-		pc1, err := blocks.NewPartialDataColumn(roDC[0].DataColumnSidecar.SignedBlockHeader, roDC[0].Index, roDC[0].KzgCommitments, roDC[0].KzgCommitmentsInclusionProof)
-		require.NoError(t, err)
-		pc2, err := blocks.NewPartialDataColumn(roDC[0].DataColumnSidecar.SignedBlockHeader, roDC[0].Index, roDC[0].KzgCommitments, roDC[0].KzgCommitmentsInclusionProof)
-		require.NoError(t, err)
-
-		// Split data
-		for i := range numCells {
-			if i%2 == 0 {
-				pc1.ExtendFromVerfifiedCell(uint64(i), roDC[0].Column[i], roDC[0].KzgProofs[i])
-			} else {
-				pc2.ExtendFromVerfifiedCell(uint64(i), roDC[0].Column[i], roDC[0].KzgProofs[i])
-			}
-		}
-
-		// Setup Topic and Subscriptions
-		digest := params.ForkDigest(0)
-		columnIndex := uint64(12)
-		subnet := peerdas.ComputeSubnetForDataColumnSidecar(columnIndex)
-		topicStr := fmt.Sprintf(p2p.DataColumnSubnetTopicFormat, digest, subnet) +
-			encoder.SszNetworkEncoder{}.ProtocolSuffix()
-
-		time.Sleep(100 * time.Millisecond)
-
-		topic1, err := ps1.Join(topicStr, pubsub.RequestPartialMessages())
-		require.NoError(t, err)
-		topic2, err := ps2.Join(topicStr, pubsub.RequestPartialMessages())
-		require.NoError(t, err)
-
-		// Header validator that verifies the inclusion proof
-		headerValidator := func(header *ethpb.PartialDataColumnHeader) (reject bool, err error) {
-			if header == nil {
-				return false, fmt.Errorf("nil header")
-			}
-			if header.SignedBlockHeader == nil || header.SignedBlockHeader.Header == nil {
-				return true, fmt.Errorf("nil signed block header")
-			}
-			if len(header.KzgCommitments) == 0 {
-				return true, fmt.Errorf("empty kzg commitments")
-			}
-			// Verify inclusion proof
-			if err := peerdas.VerifyPartialDataColumnHeaderInclusionProof(header); err != nil {
-				return true, fmt.Errorf("invalid inclusion proof: %w", err)
-			}
-			t.Log("Header validation passed")
-			return false, nil
-		}
-
-		cellValidator := func(_ []blocks.CellProofBundle) error {
-			return nil
-		}
-
-		node1Complete := make(chan blocks.VerifiedRODataColumn, 1)
-		node2Complete := make(chan blocks.VerifiedRODataColumn, 1)
-
-		handler1 := func(topic string, col blocks.VerifiedRODataColumn) {
-			t.Logf("Node 1: Completed! Column has %d cells", len(col.Column))
-			node1Complete <- col
-		}
-
-		handler2 := func(topic string, col blocks.VerifiedRODataColumn) {
-			t.Logf("Node 2: Completed! Column has %d cells", len(col.Column))
-			node2Complete <- col
-		}
-
-		// Connect hosts
-		err = h1.Connect(context.Background(), peer.AddrInfo{
-			ID:    h2.ID(),
-			Addrs: h2.Addrs(),
-		})
-		require.NoError(t, err)
-		time.Sleep(300 * time.Millisecond)
-
-		// Subscribe to regular GossipSub (critical for partial message RPC exchange!)
-		sub1, err := topic1.Subscribe()
-		require.NoError(t, err)
-		defer sub1.Cancel()
-
-		sub2, err := topic2.Subscribe()
-		require.NoError(t, err)
-		defer sub2.Cancel()
-
-		err = broadcaster1.Subscribe(topic1, headerValidator, cellValidator, handler1)
-		require.NoError(t, err)
-		err = broadcaster2.Subscribe(topic2, headerValidator, cellValidator, handler2)
-		require.NoError(t, err)
-
-		// Wait for mesh to form
-		time.Sleep(2 * time.Second)
-
-		// Publish
-		t.Log("Publishing from Node 1")
-		err = broadcaster1.Publish(topicStr, pc1)
-		require.NoError(t, err)
-
-		time.Sleep(200 * time.Millisecond)
-
-		t.Log("Publishing from Node 2")
-		err = broadcaster2.Publish(topicStr, pc2)
-		require.NoError(t, err)
-
-		//  Wait for Completion
-		timeout := time.After(10 * time.Second)
-		var col1, col2 blocks.VerifiedRODataColumn
-		receivedCount := 0
-
-		for receivedCount < 2 {
-			select {
-			case col1 = <-node1Complete:
-				t.Log("Node 1 completed reconstruction")
-				receivedCount++
-			case col2 = <-node2Complete:
-				t.Log("Node 2 completed reconstruction")
-				receivedCount++
-			case <-timeout:
-				t.Fatalf("Timeout: Only %d/2 nodes completed", receivedCount)
-			}
-		}
-
-		// Verify both columns have all cells
-		assert.Equal(t, numCells, len(col1.Column), "Node 1 should have all cells")
-		assert.Equal(t, numCells, len(col2.Column), "Node 2 should have all cells")
-		assert.DeepSSZEqual(t, cells, col1.Column, "Node 1 cell mismatch")
-		assert.DeepSSZEqual(t, cells, col2.Column, "Node 2 cell mismatch")
-	})
-}

beacon-chain/p2p/partialdatacolumnbroadcaster/metrics.go

@@ -1,18 +0,0 @@
-package partialdatacolumnbroadcaster
-
-import (
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
-)
-
-var (
-	partialMessageUsefulCellsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
-		Name: "beacon_partial_message_useful_cells_total",
-		Help: "Number of useful cells received via a partial message",
-	}, []string{"column_index"})
-
-	partialMessageCellsReceivedTotal = promauto.NewCounterVec(prometheus.CounterOpts{
-		Name: "beacon_partial_message_cells_received_total",
-		Help: "Number of total cells received via a partial message",
-	}, []string{"column_index"})
-)

beacon-chain/p2p/partialdatacolumnbroadcaster/partial.go

@@ -1,544 +0,0 @@
-package partialdatacolumnbroadcaster
-
-import (
-	"bytes"
-	"log/slog"
-	"regexp"
-	"strconv"
-	"time"
-
-	"github.com/OffchainLabs/go-bitfield"
-	"github.com/OffchainLabs/prysm/v7/config/params"
-	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
-	"github.com/OffchainLabs/prysm/v7/internal/logrusadapter"
-	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
-	pubsub "github.com/libp2p/go-libp2p-pubsub"
-	"github.com/libp2p/go-libp2p-pubsub/partialmessages"
-	"github.com/libp2p/go-libp2p-pubsub/partialmessages/bitmap"
-	pubsub_pb "github.com/libp2p/go-libp2p-pubsub/pb"
-	"github.com/libp2p/go-libp2p/core/peer"
-	"github.com/pkg/errors"
-	"github.com/sirupsen/logrus"
-)
-
-// TODOs:
-// different eager push strategies:
-// - no eager push
-// - full column eager push
-//   - With debouncing - some factor of RTT
-// - eager push missing cells
-
-const TTLInSlots = 3
-const maxConcurrentValidators = 128
-
-var dataColumnTopicRegex = regexp.MustCompile(`data_column_sidecar_(\d+)`)
-
-func extractColumnIndexFromTopic(topic string) (uint64, error) {
-	matches := dataColumnTopicRegex.FindStringSubmatch(topic)
-	if len(matches) < 2 {
-		return 0, errors.New("could not extract column index from topic")
-	}
-	return strconv.ParseUint(matches[1], 10, 64)
-}
-
-// HeaderValidator validates a PartialDataColumnHeader.
-// Returns (reject, err) where:
-//   - reject=true, err!=nil: REJECT - peer should be penalized
-//   - reject=false, err!=nil: IGNORE - don't penalize, just ignore
-//   - reject=false, err=nil: valid header
-type HeaderValidator func(header *ethpb.PartialDataColumnHeader) (reject bool, err error)
-type ColumnValidator func(cells []blocks.CellProofBundle) error
-
-type PartialColumnBroadcaster struct {
-	logger *logrus.Logger
-
-	ps   *pubsub.PubSub
-	stop chan struct{}
-
-	// map topic -> headerValidators
-	headerValidators map[string]HeaderValidator
-	// map topic -> Validator
-	validators map[string]ColumnValidator
-
-	// map topic -> handler
-	handlers map[string]SubHandler
-
-	// map topic -> *pubsub.Topic
-	topics map[string]*pubsub.Topic
-
-	concurrentValidatorSemaphore chan struct{}
-
-	// map topic -> map[groupID]PartialColumn
-	partialMsgStore map[string]map[string]*blocks.PartialDataColumn
-
-	groupTTL map[string]int8
-
-	// validHeaderCache caches validated headers by group ID (works across topics)
-	validHeaderCache map[string]*ethpb.PartialDataColumnHeader
-
-	incomingReq chan request
-}
-
-type requestKind uint8
-
-const (
-	requestKindPublish requestKind = iota
-	requestKindSubscribe
-	requestKindUnsubscribe
-	requestKindHandleIncomingRPC
-	requestKindCellsValidated
-)
-
-type request struct {
-	kind           requestKind
-	response       chan error
-	sub            subscribe
-	unsub          unsubscribe
-	publish        publish
-	incomingRPC    rpcWithFrom
-	cellsValidated *cellsValidated
-}
-
-type publish struct {
-	topic string
-	c     blocks.PartialDataColumn
-}
-
-type subscribe struct {
-	t               *pubsub.Topic
-	headerValidator HeaderValidator
-	validator       ColumnValidator
-	handler         SubHandler
-}
-
-type unsubscribe struct {
-	topic string
-}
-
-type rpcWithFrom struct {
-	*pubsub_pb.PartialMessagesExtension
-	from peer.ID
-}
-
-type cellsValidated struct {
-	validationTook time.Duration
-	topic          string
-	group          []byte
-	cellIndices    []uint64
-	cells          []blocks.CellProofBundle
-}
-
-func NewBroadcaster(logger *logrus.Logger) *PartialColumnBroadcaster {
-	return &PartialColumnBroadcaster{
-		validators:       make(map[string]ColumnValidator),
-		headerValidators: make(map[string]HeaderValidator),
-		handlers:         make(map[string]SubHandler),
-		topics:           make(map[string]*pubsub.Topic),
-		partialMsgStore:  make(map[string]map[string]*blocks.PartialDataColumn),
-		groupTTL:         make(map[string]int8),
-		validHeaderCache: make(map[string]*ethpb.PartialDataColumnHeader),
-		// GossipSub sends the messages to this channel. The buffer should be
-		// big enough to avoid dropping messages. We don't want to block the gossipsub event loop for this.
-		incomingReq: make(chan request, 128*16),
-		logger:      logger,
-
-		concurrentValidatorSemaphore: make(chan struct{}, maxConcurrentValidators),
-	}
-}
-
-// AppendPubSubOpts adds the necessary pubsub options to enable partial messages.
-func (p *PartialColumnBroadcaster) AppendPubSubOpts(opts []pubsub.Option) []pubsub.Option {
-	slogger := slog.New(logrusadapter.Handler{Logger: p.logger})
-	opts = append(opts,
-		pubsub.WithPartialMessagesExtension(&partialmessages.PartialMessagesExtension{
-			Logger: slogger,
-			MergePartsMetadata: func(topic string, left, right partialmessages.PartsMetadata) partialmessages.PartsMetadata {
-				if len(left) == 0 {
-					return right
-				}
-				merged, err := bitfield.Bitlist(left).Or(bitfield.Bitlist(right))
-				if err != nil {
-					p.logger.Warn("Failed to merge bitfields", "err", err, "left", left, "right", right)
-					return left
-				}
-				return partialmessages.PartsMetadata(merged)
-			},
-			ValidateRPC: func(from peer.ID, rpc *pubsub_pb.PartialMessagesExtension) error {
-				// TODO. Add some basic and fast sanity checks
-				return nil
-			},
-			OnIncomingRPC: func(from peer.ID, rpc *pubsub_pb.PartialMessagesExtension) error {
-				select {
-				case p.incomingReq <- request{
-					kind:        requestKindHandleIncomingRPC,
-					incomingRPC: rpcWithFrom{rpc, from},
-				}:
-				default:
-					p.logger.Warn("Dropping incoming partial RPC", "rpc", rpc)
-				}
-				return nil
-			},
-		}),
-		func(ps *pubsub.PubSub) error {
-			p.ps = ps
-			return nil
-		},
-	)
-	return opts
-}
-
-// Start starts the event loop of the PartialColumnBroadcaster. Should be called
-// within a goroutine (go p.Start())
-func (p *PartialColumnBroadcaster) Start() {
-	if p.stop != nil {
-		return
-	}
-	p.stop = make(chan struct{})
-	p.loop()
-}
-
-func (p *PartialColumnBroadcaster) loop() {
-	cleanup := time.NewTicker(time.Second * time.Duration(params.BeaconConfig().SecondsPerSlot))
-	defer cleanup.Stop()
-	for {
-		select {
-		case <-p.stop:
-			return
-		case <-cleanup.C:
-			for groupID, ttl := range p.groupTTL {
-				if ttl > 0 {
-					p.groupTTL[groupID] = ttl - 1
-					continue
-				}
-
-				delete(p.groupTTL, groupID)
-				delete(p.validHeaderCache, groupID)
-				for topic, msgStore := range p.partialMsgStore {
-					delete(msgStore, groupID)
-					if len(msgStore) == 0 {
-						delete(p.partialMsgStore, topic)
-					}
-				}
-			}
-		case req := <-p.incomingReq:
-			switch req.kind {
-			case requestKindPublish:
-				req.response <- p.publish(req.publish.topic, req.publish.c)
-			case requestKindSubscribe:
-				req.response <- p.subscribe(req.sub.t, req.sub.headerValidator, req.sub.validator, req.sub.handler)
-			case requestKindUnsubscribe:
-				req.response <- p.unsubscribe(req.unsub.topic)
-			case requestKindHandleIncomingRPC:
-				err := p.handleIncomingRPC(req.incomingRPC)
-				if err != nil {
-					p.logger.Error("Failed to handle incoming partial RPC", "err", err)
-				}
-			case requestKindCellsValidated:
-				err := p.handleCellsValidated(req.cellsValidated)
-				if err != nil {
-					p.logger.Error("Failed to handle cells validated", "err", err)
-				}
-			default:
-				p.logger.Error("Unknown request kind", "kind", req.kind)
-			}
-		}
-	}
-}
-
-func (p *PartialColumnBroadcaster) getDataColumn(topic string, group []byte) *blocks.PartialDataColumn {
-	topicStore, ok := p.partialMsgStore[topic]
-	if !ok {
-		return nil
-	}
-	msg, ok := topicStore[string(group)]
-	if !ok {
-		return nil
-	}
-	return msg
-}
-
-func (p *PartialColumnBroadcaster) handleIncomingRPC(rpcWithFrom rpcWithFrom) error {
-	if p.ps == nil {
-		return errors.New("pubsub not initialized")
-	}
-
-	hasMessage := len(rpcWithFrom.PartialMessage) > 0
-
-	var message ethpb.PartialDataColumnSidecar
-	if hasMessage {
-		err := message.UnmarshalSSZ(rpcWithFrom.PartialMessage)
-		if err != nil {
-			return errors.Wrap(err, "failed to unmarshal partial message data")
-		}
-	}
-
-	topicID := rpcWithFrom.GetTopicID()
-	groupID := rpcWithFrom.GroupID
-	ourDataColumn := p.getDataColumn(topicID, groupID)
-	var shouldRepublish bool
-
-	if ourDataColumn == nil && hasMessage {
-		var header *ethpb.PartialDataColumnHeader
-		// Check cache first for this group
-		if cachedHeader, ok := p.validHeaderCache[string(groupID)]; ok {
-			header = cachedHeader
-		} else {
-			// We haven't seen this group before. Check if we have a valid header.
-			if len(message.Header) == 0 {
-				p.logger.Debug("No partial column found and no header in message, ignoring")
-				return nil
-			}
-
-			header = message.Header[0]
-			headerValidator, ok := p.headerValidators[topicID]
-			if !ok || headerValidator == nil {
-				p.logger.Debug("No header validator registered for topic")
-				return nil
-			}
-
-			reject, err := headerValidator(header)
-			if err != nil {
-				p.logger.Debug("Header validation failed", "err", err, "reject", reject)
-				if reject {
-					// REJECT case: penalize the peer
-					_ = p.ps.PeerFeedback(topicID, rpcWithFrom.from, pubsub.PeerFeedbackInvalidMessage)
-				}
-				// Both REJECT and IGNORE: don't process further
-				return nil
-			}
-			// Cache the valid header
-			p.validHeaderCache[string(groupID)] = header
-
-			// TODO: We now have the information we need to call GetBlobsV3, we should do that to see what we have locally.
-		}
-
-		columnIndex, err := extractColumnIndexFromTopic(topicID)
-		if err != nil {
-			return err
-		}
-
-		newColumn, err := blocks.NewPartialDataColumn(
-			header.SignedBlockHeader,
-			columnIndex,
-			header.KzgCommitments,
-			header.KzgCommitmentsInclusionProof,
-		)
-		if err != nil {
-			p.logger.WithError(err).WithFields(logrus.Fields{
-				"topic":          topicID,
-				"columnIndex":    columnIndex,
-				"numCommitments": len(header.KzgCommitments),
-			}).Error("Failed to create partial data column from header")
-			return err
-		}
-
-		// Save to store
-		topicStore, ok := p.partialMsgStore[topicID]
-		if !ok {
-			topicStore = make(map[string]*blocks.PartialDataColumn)
-			p.partialMsgStore[topicID] = topicStore
-		}
-		topicStore[string(newColumn.GroupID())] = &newColumn
-		p.groupTTL[string(newColumn.GroupID())] = TTLInSlots
-
-		ourDataColumn = &newColumn
-		shouldRepublish = true
-	}
-
-	if ourDataColumn == nil {
-		// We don't have a partial column for this. Can happen if we got cells
-		// without a header.
-		return nil
-	}
-
-	logger := p.logger.WithFields(logrus.Fields{
-		"from":  rpcWithFrom.from,
-		"topic": topicID,
-		"group": groupID,
-	})
-
-	validator, validatorOK := p.validators[topicID]
-	if len(rpcWithFrom.PartialMessage) > 0 && validatorOK {
-		// TODO: is there any penalty we want to consider for giving us data we didn't request?
-		// Note that we need to be careful around race conditions and eager data.
-		// Also note that protobufs by design allow extra data that we don't parse.
-		// Marco's thoughts. No, we don't need to do anything else here.
-		cellIndices, cellsToVerify, err := ourDataColumn.CellsToVerifyFromPartialMessage(&message)
-		if err != nil {
-			return err
-		}
-		// Track cells received via partial message
-		if len(cellIndices) > 0 {
-			columnIndexStr := strconv.FormatUint(ourDataColumn.Index, 10)
-			partialMessageCellsReceivedTotal.WithLabelValues(columnIndexStr).Add(float64(len(cellIndices)))
-		}
-		if len(cellsToVerify) > 0 {
-			p.concurrentValidatorSemaphore <- struct{}{}
-			go func() {
-				defer func() {
-					<-p.concurrentValidatorSemaphore
-				}()
-				start := time.Now()
-				err := validator(cellsToVerify)
-				if err != nil {
-					logger.Error("failed to validate cells", "err", err)
-					_ = p.ps.PeerFeedback(topicID, rpcWithFrom.from, pubsub.PeerFeedbackInvalidMessage)
-					return
-				}
-				_ = p.ps.PeerFeedback(topicID, rpcWithFrom.from, pubsub.PeerFeedbackUsefulMessage)
-				p.incomingReq <- request{
-					kind: requestKindCellsValidated,
-					cellsValidated: &cellsValidated{
-						validationTook: time.Since(start),
-						topic:          topicID,
-						group:          groupID,
-						cells:          cellsToVerify,
-						cellIndices:    cellIndices,
-					},
-				}
-			}()
-		}
-	}
-
-	peerHas := bitmap.Bitmap(rpcWithFrom.PartsMetadata)
-	iHave := bitmap.Bitmap(ourDataColumn.PartsMetadata())
-	if !shouldRepublish && len(peerHas) > 0 && !bytes.Equal(peerHas, iHave) {
-		// Either we have something they don't or vice versa
-		shouldRepublish = true
-		logger.Debug("republishing due to parts metadata difference")
-	}
-
-	if shouldRepublish {
-		err := p.ps.PublishPartialMessage(topicID, ourDataColumn, partialmessages.PublishOptions{})
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func (p *PartialColumnBroadcaster) handleCellsValidated(cells *cellsValidated) error {
-	ourDataColumn := p.getDataColumn(cells.topic, cells.group)
-	if ourDataColumn == nil {
-		return errors.New("data column not found for verified cells")
-	}
-	extended := ourDataColumn.ExtendFromVerfifiedCells(cells.cellIndices, cells.cells)
-	p.logger.Debug("Extended partial message", "duration", cells.validationTook, "extended", extended)
-
-	columnIndexStr := strconv.FormatUint(ourDataColumn.Index, 10)
-	if extended {
-		// Track useful cells (cells that extended our data)
-		partialMessageUsefulCellsTotal.WithLabelValues(columnIndexStr).Add(float64(len(cells.cells)))
-
-		// TODO: we could use the heuristic here that if this data was
-		// useful to us, it's likely useful to our peers and we should
-		// republish eagerly
-
-		if col, ok := ourDataColumn.Complete(p.logger); ok {
-			p.logger.Info("Completed partial column", "topic", cells.topic, "group", cells.group)
-			handler, handlerOK := p.handlers[cells.topic]
-
-			if handlerOK {
-				go handler(cells.topic, col)
-			}
-		} else {
-			p.logger.Info("Extended partial column", "topic", cells.topic, "group", cells.group)
-		}
-
-		err := p.ps.PublishPartialMessage(cells.topic, ourDataColumn, partialmessages.PublishOptions{})
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func (p *PartialColumnBroadcaster) Stop() {
-	if p.stop != nil {
-		close(p.stop)
-		p.stop = nil
-	}
-}
-
-// Publish publishes the partial column.
-func (p *PartialColumnBroadcaster) Publish(topic string, c blocks.PartialDataColumn) error {
-	if p.ps == nil {
-		return errors.New("pubsub not initialized")
-	}
-	respCh := make(chan error)
-	p.incomingReq <- request{
-		kind:     requestKindPublish,
-		response: respCh,
-		publish: publish{
-			topic: topic,
-			c:     c,
-		},
-	}
-	return <-respCh
-}
-
-func (p *PartialColumnBroadcaster) publish(topic string, c blocks.PartialDataColumn) error {
-	topicStore, ok := p.partialMsgStore[topic]
-	if !ok {
-		topicStore = make(map[string]*blocks.PartialDataColumn)
-		p.partialMsgStore[topic] = topicStore
-	}
-	topicStore[string(c.GroupID())] = &c
-	p.groupTTL[string(c.GroupID())] = TTLInSlots
-
-	return p.ps.PublishPartialMessage(topic, &c, partialmessages.PublishOptions{})
-}
-
-type SubHandler func(topic string, col blocks.VerifiedRODataColumn)
-
-func (p *PartialColumnBroadcaster) Subscribe(t *pubsub.Topic, headerValidator HeaderValidator, validator ColumnValidator, handler SubHandler) error {
-	respCh := make(chan error)
-	p.incomingReq <- request{
-		kind: requestKindSubscribe,
-		sub: subscribe{
-			t:               t,
-			headerValidator: headerValidator,
-			validator:       validator,
-			handler:         handler,
-		},
-		response: respCh,
-	}
-	return <-respCh
-}
-func (p *PartialColumnBroadcaster) subscribe(t *pubsub.Topic, headerValidator HeaderValidator, validator ColumnValidator, handler SubHandler) error {
-	topic := t.String()
-	if _, ok := p.topics[topic]; ok {
-		return errors.New("already subscribed")
-	}
-
-	p.topics[topic] = t
-	p.headerValidators[topic] = headerValidator
-	p.validators[topic] = validator
-	p.handlers[topic] = handler
-	return nil
-}
-
-func (p *PartialColumnBroadcaster) Unsubscribe(topic string) error {
-	respCh := make(chan error)
-	p.incomingReq <- request{
-		kind: requestKindUnsubscribe,
-		unsub: unsubscribe{
-			topic: topic,
-		},
-		response: respCh,
-	}
-	return <-respCh
-}
-func (p *PartialColumnBroadcaster) unsubscribe(topic string) error {
-	t, ok := p.topics[topic]
-	if !ok {
-		return errors.New("topic not found")
-	}
-	delete(p.topics, topic)
-	delete(p.partialMsgStore, topic)
-	delete(p.headerValidators, topic)
-	delete(p.validators, topic)
-	delete(p.handlers, topic)
-
-	return t.Close()
-}

beacon-chain/p2p/peers/status_test.go

@@ -58,7 +58,7 @@ func TestPeerExplicitAdd(t *testing.T) {
 
 	resAddress, err := p.Address(id)
 	require.NoError(t, err)
-	assert.Equal(t, address.Equal(resAddress), true, "Unexpected address")
+	assert.Equal(t, address, resAddress, "Unexpected address")
 
 	resDirection, err := p.Direction(id)
 	require.NoError(t, err)
@@ -72,7 +72,7 @@ func TestPeerExplicitAdd(t *testing.T) {
 
 	resAddress2, err := p.Address(id)
 	require.NoError(t, err)
-	assert.Equal(t, address2.Equal(resAddress2), true, "Unexpected address")
+	assert.Equal(t, address2, resAddress2, "Unexpected address")
 
 	resDirection2, err := p.Direction(id)
 	require.NoError(t, err)

beacon-chain/p2p/pubsub.go

@@ -160,9 +160,6 @@ func (s *Service) pubsubOptions() []pubsub.Option {
 		}
 		psOpts = append(psOpts, pubsub.WithDirectPeers(directPeersAddrInfos))
 	}
-	if s.partialColumnBroadcaster != nil {
-		psOpts = s.partialColumnBroadcaster.AppendPubSubOpts(psOpts)
-	}
 
 	return psOpts
 }

beacon-chain/p2p/pubsub_tracer.go

@@ -88,20 +88,20 @@ func (g gossipTracer) ThrottlePeer(p peer.ID) {
 
 // RecvRPC .
 func (g gossipTracer) RecvRPC(rpc *pubsub.RPC) {
-	g.setMetricFromRPC(recv, pubsubRPCSubRecv, pubsubRPCPubRecv, pubsubRPCPubRecvSize, pubsubRPCRecv, rpc)
+	g.setMetricFromRPC(recv, pubsubRPCSubRecv, pubsubRPCPubRecv, pubsubRPCRecv, rpc)
 }
 
 // SendRPC .
 func (g gossipTracer) SendRPC(rpc *pubsub.RPC, p peer.ID) {
-	g.setMetricFromRPC(send, pubsubRPCSubSent, pubsubRPCPubSent, pubsubRPCPubSentSize, pubsubRPCSent, rpc)
+	g.setMetricFromRPC(send, pubsubRPCSubSent, pubsubRPCPubSent, pubsubRPCSent, rpc)
 }
 
 // DropRPC .
 func (g gossipTracer) DropRPC(rpc *pubsub.RPC, p peer.ID) {
-	g.setMetricFromRPC(drop, pubsubRPCSubDrop, pubsubRPCPubDrop, pubsubRPCPubDropSize, pubsubRPCDrop, rpc)
+	g.setMetricFromRPC(drop, pubsubRPCSubDrop, pubsubRPCPubDrop, pubsubRPCDrop, rpc)
 }
 
-func (g gossipTracer) setMetricFromRPC(act action, subCtr prometheus.Counter, pubCtr, pubSizeCtr, ctrlCtr *prometheus.CounterVec, rpc *pubsub.RPC) {
+func (g gossipTracer) setMetricFromRPC(act action, subCtr prometheus.Counter, pubCtr, ctrlCtr *prometheus.CounterVec, rpc *pubsub.RPC) {
 	subCtr.Add(float64(len(rpc.Subscriptions)))
 	if rpc.Control != nil {
 		ctrlCtr.WithLabelValues("graft").Add(float64(len(rpc.Control.Graft)))
@@ -110,17 +110,12 @@ func (g gossipTracer) setMetricFromRPC(act action, subCtr prometheus.Counter, pu
 		ctrlCtr.WithLabelValues("iwant").Add(float64(len(rpc.Control.Iwant)))
 		ctrlCtr.WithLabelValues("idontwant").Add(float64(len(rpc.Control.Idontwant)))
 	}
-	// For incoming messages from pubsub, we do not record metrics for them as these values
-	// could be junk.
-	if act == recv {
-		return
-	}
 	for _, msg := range rpc.Publish {
-		pubCtr.WithLabelValues(msg.GetTopic()).Inc()
-		pubSizeCtr.WithLabelValues(msg.GetTopic(), "false").Add(float64(msg.Size()))
-	}
-	if rpc.Partial != nil {
-		pubCtr.WithLabelValues(rpc.Partial.GetTopicID()).Inc()
-		pubSizeCtr.WithLabelValues(rpc.Partial.GetTopicID(), "true").Add(float64(rpc.Partial.Size()))
+		// For incoming messages from pubsub, we do not record metrics for them as these values
+		// could be junk.
+		if act == recv {
+			continue
+		}
+		pubCtr.WithLabelValues(*msg.Topic).Inc()
 	}
 }

beacon-chain/p2p/service.go

@@ -11,7 +11,6 @@ import (
 
 	"github.com/OffchainLabs/prysm/v7/async"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/encoder"
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/partialdatacolumnbroadcaster"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/peers"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/peers/scorers"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/types"
@@ -78,7 +77,6 @@ type Service struct {
 	privKey                  *ecdsa.PrivateKey
 	metaData                 metadata.Metadata
 	pubsub                   *pubsub.PubSub
-	partialColumnBroadcaster *partialdatacolumnbroadcaster.PartialColumnBroadcaster
 	joinedTopics             map[string]*pubsub.Topic
 	joinedTopicsLock         sync.RWMutex
 	subnetsLock              map[uint64]*sync.RWMutex
@@ -149,10 +147,6 @@ func NewService(ctx context.Context, cfg *Config) (*Service, error) {
 		custodyInfoSet:        make(chan struct{}),
 	}
 
-	if cfg.PartialDataColumns {
-		s.partialColumnBroadcaster = partialdatacolumnbroadcaster.NewBroadcaster(log.Logger)
-	}
-
 	ipAddr := prysmnetwork.IPAddr()
 
 	opts, err := s.buildOptions(ipAddr, s.privKey)
@@ -311,10 +305,6 @@ func (s *Service) Start() {
 		logExternalDNSAddr(s.host.ID(), p2pHostDNS, p2pTCPPort)
 	}
 	go s.forkWatcher()
-
-	if s.partialColumnBroadcaster != nil {
-		go s.partialColumnBroadcaster.Start()
-	}
 }
 
 // Stop the p2p service and terminate all peer connections.
@@ -324,10 +314,6 @@ func (s *Service) Stop() error {
 	if s.dv5Listener != nil {
 		s.dv5Listener.Close()
 	}
-
-	if s.partialColumnBroadcaster != nil {
-		s.partialColumnBroadcaster.Stop()
-	}
 	return nil
 }
 
@@ -364,10 +350,6 @@ func (s *Service) PubSub() *pubsub.PubSub {
 	return s.pubsub
 }
 
-func (s *Service) PartialColumnBroadcaster() *partialdatacolumnbroadcaster.PartialColumnBroadcaster {
-	return s.partialColumnBroadcaster
-}
-
 // Host returns the currently running libp2p
 // host of the service.
 func (s *Service) Host() host.Host {

beacon-chain/p2p/testing/BUILD.bazel

@@ -21,7 +21,6 @@ go_library(
     deps = [
         "//beacon-chain/core/peerdas:go_default_library",
         "//beacon-chain/p2p/encoder:go_default_library",
-        "//beacon-chain/p2p/partialdatacolumnbroadcaster:go_default_library",
         "//beacon-chain/p2p/peers:go_default_library",
         "//beacon-chain/p2p/peers/scorers:go_default_library",
         "//config/fieldparams:go_default_library",

beacon-chain/p2p/testing/fuzz_p2p.go

@@ -4,7 +4,6 @@ import (
 	"context"
 
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/encoder"
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/partialdatacolumnbroadcaster"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/peers"
 	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
@@ -109,10 +108,6 @@ func (*FakeP2P) PubSub() *pubsub.PubSub {
 	return nil
 }
 
-func (*FakeP2P) PartialColumnBroadcaster() *partialdatacolumnbroadcaster.PartialColumnBroadcaster {
-	return nil
-}
-
 // MetadataSeq -- fake.
 func (*FakeP2P) MetadataSeq() uint64 {
 	return 0
@@ -174,7 +169,7 @@ func (*FakeP2P) BroadcastLightClientFinalityUpdate(_ context.Context, _ interfac
 }
 
 // BroadcastDataColumnSidecar -- fake.
-func (*FakeP2P) BroadcastDataColumnSidecars(_ context.Context, _ []blocks.VerifiedRODataColumn, _ []blocks.PartialDataColumn) error {
+func (*FakeP2P) BroadcastDataColumnSidecars(_ context.Context, _ []blocks.VerifiedRODataColumn) error {
 	return nil
 }
 

beacon-chain/p2p/testing/mock_broadcaster.go

@@ -63,7 +63,7 @@ func (m *MockBroadcaster) BroadcastLightClientFinalityUpdate(_ context.Context,
 }
 
 // BroadcastDataColumnSidecar broadcasts a data column for mock.
-func (m *MockBroadcaster) BroadcastDataColumnSidecars(context.Context, []blocks.VerifiedRODataColumn, []blocks.PartialDataColumn) error {
+func (m *MockBroadcaster) BroadcastDataColumnSidecars(context.Context, []blocks.VerifiedRODataColumn) error {
 	m.BroadcastCalled.Store(true)
 	return nil
 }

beacon-chain/p2p/testing/p2p.go

@@ -12,7 +12,6 @@ import (
 	"time"
 
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/peerdas"
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/partialdatacolumnbroadcaster"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/encoder"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/peers"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/peers/scorers"
@@ -234,7 +233,7 @@ func (p *TestP2P) BroadcastLightClientFinalityUpdate(_ context.Context, _ interf
 }
 
 // BroadcastDataColumnSidecar broadcasts a data column for mock.
-func (p *TestP2P) BroadcastDataColumnSidecars(context.Context, []blocks.VerifiedRODataColumn, []blocks.PartialDataColumn) error {
+func (p *TestP2P) BroadcastDataColumnSidecars(context.Context, []blocks.VerifiedRODataColumn) error {
 	p.BroadcastCalled.Store(true)
 	return nil
 }
@@ -300,10 +299,6 @@ func (p *TestP2P) PubSub() *pubsub.PubSub {
 	return p.pubsub
 }
 
-func (p *TestP2P) PartialColumnBroadcaster() *partialdatacolumnbroadcaster.PartialColumnBroadcaster {
-	return nil
-}
-
 // Disconnect from a peer.
 func (p *TestP2P) Disconnect(pid peer.ID) error {
 	return p.BHost.Network().ClosePeer(pid)

beacon-chain/p2p/types/object_mapping.go

@@ -204,9 +204,6 @@ func InitializeDataMaps() {
 		bytesutil.ToBytes4(params.BeaconConfig().ElectraForkVersion): func() (interfaces.LightClientOptimisticUpdate, error) {
 			return lightclientConsensusTypes.NewEmptyOptimisticUpdateDeneb(), nil
 		},
-		bytesutil.ToBytes4(params.BeaconConfig().FuluForkVersion): func() (interfaces.LightClientOptimisticUpdate, error) {
-			return lightclientConsensusTypes.NewEmptyOptimisticUpdateDeneb(), nil
-		},
 	}
 
 	// Reset our light client finality update map.
@@ -226,8 +223,5 @@ func InitializeDataMaps() {
 		bytesutil.ToBytes4(params.BeaconConfig().ElectraForkVersion): func() (interfaces.LightClientFinalityUpdate, error) {
 			return lightclientConsensusTypes.NewEmptyFinalityUpdateElectra(), nil
 		},
-		bytesutil.ToBytes4(params.BeaconConfig().FuluForkVersion): func() (interfaces.LightClientFinalityUpdate, error) {
-			return lightclientConsensusTypes.NewEmptyFinalityUpdateElectra(), nil
-		},
 	}
 }

beacon-chain/rpc/eth/beacon/handlers.go

@@ -993,11 +993,10 @@ func (s *Server) validateEquivocation(blk interfaces.ReadOnlyBeaconBlock) error
 }
 
 func (s *Server) validateBlobs(blk interfaces.SignedBeaconBlock, blobs [][]byte, proofs [][]byte) error {
-	const numberOfColumns = fieldparams.NumberOfColumns
-
 	if blk.Version() < version.Deneb {
 		return nil
 	}
+	numberOfColumns := params.BeaconConfig().NumberOfColumns
 	commitments, err := blk.Block().Body().BlobKzgCommitments()
 	if err != nil {
 		return errors.Wrap(err, "could not get blob kzg commitments")

beacon-chain/rpc/eth/beacon/handlers_pool.go

@@ -711,7 +711,6 @@ func (s *Server) SubmitAttesterSlashingsV2(w http.ResponseWriter, r *http.Reques
 	versionHeader := r.Header.Get(api.VersionHeader)
 	if versionHeader == "" {
 		httputil.HandleError(w, api.VersionHeader+" header is required", http.StatusBadRequest)
-		return
 	}
 	v, err := version.FromString(versionHeader)
 	if err != nil {

beacon-chain/rpc/eth/beacon/handlers_pool_test.go

@@ -2112,33 +2112,6 @@ func TestSubmitAttesterSlashingsV2(t *testing.T) {
 		assert.Equal(t, http.StatusBadRequest, e.Code)
 		assert.StringContains(t, "Invalid attester slashing", e.Message)
 	})
-
-	t.Run("missing-version-header", func(t *testing.T) {
-		bs, err := util.NewBeaconStateElectra()
-		require.NoError(t, err)
-
-		broadcaster := &p2pMock.MockBroadcaster{}
-		s := &Server{
-			ChainInfoFetcher: &blockchainmock.ChainService{State: bs},
-			SlashingsPool:    &slashingsmock.PoolMock{},
-			Broadcaster:      broadcaster,
-		}
-
-		var body bytes.Buffer
-		_, err = body.WriteString(invalidAttesterSlashing)
-		require.NoError(t, err)
-		request := httptest.NewRequest(http.MethodPost, "http://example.com/beacon/pool/attester_slashings", &body)
-		// Intentionally do not set api.VersionHeader to verify missing header handling.
-		writer := httptest.NewRecorder()
-		writer.Body = &bytes.Buffer{}
-
-		s.SubmitAttesterSlashingsV2(writer, request)
-		require.Equal(t, http.StatusBadRequest, writer.Code)
-		e := &httputil.DefaultJsonError{}
-		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
-		assert.Equal(t, http.StatusBadRequest, e.Code)
-		assert.StringContains(t, api.VersionHeader+" header is required", e.Message)
-	})
 }
 
 func TestSubmitProposerSlashing_InvalidSlashing(t *testing.T) {

beacon-chain/rpc/eth/beacon/handlers_state_test.go

@@ -654,10 +654,6 @@ func (m *futureSyncMockFetcher) StateBySlot(context.Context, primitives.Slot) (s
 	return m.BeaconState, nil
 }
 
-func (m *futureSyncMockFetcher) StateByEpoch(context.Context, primitives.Epoch) (state.BeaconState, error) {
-	return m.BeaconState, nil
-}
-
 func TestGetSyncCommittees_Future(t *testing.T) {
 	st, _ := util.DeterministicGenesisStateAltair(t, params.BeaconConfig().SyncCommitteeSize)
 	syncCommittee := make([][]byte, params.BeaconConfig().SyncCommitteeSize)

beacon-chain/rpc/eth/beacon/handlers_test.go

@@ -27,7 +27,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/testutil"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/state"
 	mockSync "github.com/OffchainLabs/prysm/v7/beacon-chain/sync/initial-sync/testing"
-	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
 	"github.com/OffchainLabs/prysm/v7/config/params"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/interfaces"
@@ -3757,7 +3756,6 @@ func Test_validateBlobs(t *testing.T) {
 	})
 
 	t.Run("Fulu block with valid cell proofs", func(t *testing.T) {
-		const numberOfColumns = fieldparams.NumberOfColumns
 		blk := util.NewBeaconBlockFulu()
 		blk.Block.Slot = fs
 
@@ -3785,13 +3783,14 @@ func Test_validateBlobs(t *testing.T) {
 		require.NoError(t, err)
 
 		// Generate cell proofs for the blobs (flattened format like execution client)
+		numberOfColumns := params.BeaconConfig().NumberOfColumns
 		cellProofs := make([][]byte, uint64(blobCount)*numberOfColumns)
 		for blobIdx := range blobCount {
 			_, proofs, err := kzg.ComputeCellsAndKZGProofs(&kzgBlobs[blobIdx])
 			require.NoError(t, err)
 
 			for colIdx := range numberOfColumns {
-				cellProofIdx := blobIdx*numberOfColumns + colIdx
+				cellProofIdx := uint64(blobIdx)*numberOfColumns + colIdx
 				cellProofs[cellProofIdx] = proofs[colIdx][:]
 			}
 		}

beacon-chain/rpc/eth/config/handlers_test.go

@@ -155,19 +155,20 @@ func TestGetSpec(t *testing.T) {
 	config.MaxAttesterSlashingsElectra = 88
 	config.MaxAttestationsElectra = 89
 	config.MaxWithdrawalRequestsPerPayload = 90
-	config.UnsetDepositRequestsStartIndex = 91
-	config.MaxDepositRequestsPerPayload = 92
-	config.MaxPendingDepositsPerEpoch = 93
-	config.MaxBlobCommitmentsPerBlock = 94
-	config.MaxBytesPerTransaction = 95
-	config.MaxExtraDataBytes = 96
-	config.BytesPerLogsBloom = 97
-	config.MaxTransactionsPerPayload = 98
-	config.FieldElementsPerBlob = 99
-	config.KzgCommitmentInclusionProofDepth = 100
-	config.BlobsidecarSubnetCount = 101
-	config.BlobsidecarSubnetCountElectra = 102
-	config.SyncMessageDueBPS = 103
+	config.MaxCellsInExtendedMatrix = 91
+	config.UnsetDepositRequestsStartIndex = 92
+	config.MaxDepositRequestsPerPayload = 93
+	config.MaxPendingDepositsPerEpoch = 94
+	config.MaxBlobCommitmentsPerBlock = 95
+	config.MaxBytesPerTransaction = 96
+	config.MaxExtraDataBytes = 97
+	config.BytesPerLogsBloom = 98
+	config.MaxTransactionsPerPayload = 99
+	config.FieldElementsPerBlob = 100
+	config.KzgCommitmentInclusionProofDepth = 101
+	config.BlobsidecarSubnetCount = 102
+	config.BlobsidecarSubnetCountElectra = 103
+	config.SyncMessageDueBPS = 104
 
 	var dbp [4]byte
 	copy(dbp[:], []byte{'0', '0', '0', '1'})
@@ -205,7 +206,7 @@ func TestGetSpec(t *testing.T) {
 	require.NoError(t, json.Unmarshal(writer.Body.Bytes(), &resp))
 	data, ok := resp.Data.(map[string]any)
 	require.Equal(t, true, ok)
-	assert.Equal(t, 175, len(data))
+	assert.Equal(t, 176, len(data))
 	for k, v := range data {
 		t.Run(k, func(t *testing.T) {
 			switch k {
@@ -499,6 +500,8 @@ func TestGetSpec(t *testing.T) {
 				assert.Equal(t, "1024", v)
 			case "MAX_REQUEST_BLOCKS_DENEB":
 				assert.Equal(t, "128", v)
+			case "NUMBER_OF_COLUMNS":
+				assert.Equal(t, "128", v)
 			case "MIN_PER_EPOCH_CHURN_LIMIT_ELECTRA":
 				assert.Equal(t, "128000000000", v)
 			case "MAX_PER_EPOCH_ACTIVATION_EXIT_CHURN_LIMIT":
@@ -535,12 +538,14 @@ func TestGetSpec(t *testing.T) {
 				assert.Equal(t, "89", v)
 			case "MAX_WITHDRAWAL_REQUESTS_PER_PAYLOAD":
 				assert.Equal(t, "90", v)
-			case "UNSET_DEPOSIT_REQUESTS_START_INDEX":
+			case "MAX_CELLS_IN_EXTENDED_MATRIX":
 				assert.Equal(t, "91", v)
-			case "MAX_DEPOSIT_REQUESTS_PER_PAYLOAD":
+			case "UNSET_DEPOSIT_REQUESTS_START_INDEX":
 				assert.Equal(t, "92", v)
-			case "MAX_PENDING_DEPOSITS_PER_EPOCH":
+			case "MAX_DEPOSIT_REQUESTS_PER_PAYLOAD":
 				assert.Equal(t, "93", v)
+			case "MAX_PENDING_DEPOSITS_PER_EPOCH":
+				assert.Equal(t, "94", v)
 			case "MAX_BLOBS_PER_BLOCK_ELECTRA":
 				assert.Equal(t, "9", v)
 			case "MAX_REQUEST_BLOB_SIDECARS_ELECTRA":
@@ -558,25 +563,25 @@ func TestGetSpec(t *testing.T) {
 			case "MIN_EPOCHS_FOR_DATA_COLUMN_SIDECARS_REQUESTS":
 				assert.Equal(t, "4096", v)
 			case "MAX_BLOB_COMMITMENTS_PER_BLOCK":
-				assert.Equal(t, "94", v)
-			case "MAX_BYTES_PER_TRANSACTION":
 				assert.Equal(t, "95", v)
-			case "MAX_EXTRA_DATA_BYTES":
+			case "MAX_BYTES_PER_TRANSACTION":
 				assert.Equal(t, "96", v)
-			case "BYTES_PER_LOGS_BLOOM":
+			case "MAX_EXTRA_DATA_BYTES":
 				assert.Equal(t, "97", v)
-			case "MAX_TRANSACTIONS_PER_PAYLOAD":
+			case "BYTES_PER_LOGS_BLOOM":
 				assert.Equal(t, "98", v)
-			case "FIELD_ELEMENTS_PER_BLOB":
+			case "MAX_TRANSACTIONS_PER_PAYLOAD":
 				assert.Equal(t, "99", v)
-			case "KZG_COMMITMENT_INCLUSION_PROOF_DEPTH":
+			case "FIELD_ELEMENTS_PER_BLOB":
 				assert.Equal(t, "100", v)
-			case "BLOB_SIDECAR_SUBNET_COUNT":
+			case "KZG_COMMITMENT_INCLUSION_PROOF_DEPTH":
 				assert.Equal(t, "101", v)
-			case "BLOB_SIDECAR_SUBNET_COUNT_ELECTRA":
+			case "BLOB_SIDECAR_SUBNET_COUNT":
 				assert.Equal(t, "102", v)
-			case "SYNC_MESSAGE_DUE_BPS":
+			case "BLOB_SIDECAR_SUBNET_COUNT_ELECTRA":
 				assert.Equal(t, "103", v)
+			case "SYNC_MESSAGE_DUE_BPS":
+				assert.Equal(t, "104", v)
 			case "BLOB_SCHEDULE":
 				blobSchedule, ok := v.([]any)
 				assert.Equal(t, true, ok)

beacon-chain/rpc/eth/debug/BUILD.bazel

@@ -17,7 +17,6 @@ go_library(
         "//beacon-chain/rpc/eth/helpers:go_default_library",
         "//beacon-chain/rpc/eth/shared:go_default_library",
         "//beacon-chain/rpc/lookup:go_default_library",
-        "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",
         "//consensus-types/blocks:go_default_library",
         "//consensus-types/primitives:go_default_library",

beacon-chain/rpc/eth/debug/handlers.go

@@ -15,7 +15,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/core"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/eth/helpers"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/eth/shared"
-	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
 	"github.com/OffchainLabs/prysm/v7/config/params"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
@@ -309,7 +308,7 @@ func (s *Server) DataColumnSidecars(w http.ResponseWriter, r *http.Request) {
 
 // parseDataColumnIndices filters out invalid and duplicate data column indices
 func parseDataColumnIndices(url *url.URL) ([]int, error) {
-	const numberOfColumns = fieldparams.NumberOfColumns
+	numberOfColumns := params.BeaconConfig().NumberOfColumns
 	rawIndices := url.Query()["indices"]
 	indices := make([]int, 0, numberOfColumns)
 	invalidIndices := make([]string, 0)

beacon-chain/rpc/eth/debug/handlers_test.go

@@ -709,6 +709,15 @@ func TestDataColumnSidecars(t *testing.T) {
 }
 
 func TestParseDataColumnIndices(t *testing.T) {
+	// Save the original config
+	originalConfig := params.BeaconConfig()
+	defer func() { params.OverrideBeaconConfig(originalConfig) }()
+
+	// Set NumberOfColumns to 128 for testing
+	config := params.BeaconConfig().Copy()
+	config.NumberOfColumns = 128
+	params.OverrideBeaconConfig(config)
+
 	tests := []struct {
 		name        string
 		queryParams map[string][]string

beacon-chain/rpc/eth/light-client/handlers.go

@@ -116,7 +116,6 @@ func (s *Server) GetLightClientUpdatesByRange(w http.ResponseWriter, req *http.R
 		for _, update := range updates {
 			if ctx.Err() != nil {
 				httputil.HandleError(w, "Context error: "+ctx.Err().Error(), http.StatusInternalServerError)
-				return
 			}
 
 			updateSlot := update.AttestedHeader().Beacon().Slot
@@ -132,15 +131,12 @@ func (s *Server) GetLightClientUpdatesByRange(w http.ResponseWriter, req *http.R
 			chunkLength = ssz.MarshalUint64(chunkLength, uint64(len(updateSSZ)+4))
 			if _, err := w.Write(chunkLength); err != nil {
 				httputil.HandleError(w, "Could not write chunk length: "+err.Error(), http.StatusInternalServerError)
-				return
 			}
 			if _, err := w.Write(updateEntry.ForkDigest[:]); err != nil {
 				httputil.HandleError(w, "Could not write fork digest: "+err.Error(), http.StatusInternalServerError)
-				return
 			}
 			if _, err := w.Write(updateSSZ); err != nil {
 				httputil.HandleError(w, "Could not write update SSZ: "+err.Error(), http.StatusInternalServerError)
-				return
 			}
 		}
 	} else {
@@ -149,7 +145,6 @@ func (s *Server) GetLightClientUpdatesByRange(w http.ResponseWriter, req *http.R
 		for _, update := range updates {
 			if ctx.Err() != nil {
 				httputil.HandleError(w, "Context error: "+ctx.Err().Error(), http.StatusInternalServerError)
-				return
 			}
 
 			updateJson, err := structs.LightClientUpdateFromConsensus(update)

beacon-chain/rpc/eth/node/handlers.go

@@ -132,7 +132,6 @@ func (s *Server) GetHealth(w http.ResponseWriter, r *http.Request) {
 	optimistic, err := s.OptimisticModeFetcher.IsOptimistic(ctx)
 	if err != nil {
 		httputil.HandleError(w, "Could not check optimistic status: "+err.Error(), http.StatusInternalServerError)
-		return
 	}
 	if s.SyncChecker.Synced() && !optimistic {
 		return

beacon-chain/rpc/eth/rewards/handlers.go

@@ -228,7 +228,7 @@ func (s *Server) attRewardsState(w http.ResponseWriter, r *http.Request) (state.
 	}
 	st, err := s.Stater.StateBySlot(r.Context(), nextEpochEnd)
 	if err != nil {
-		shared.WriteStateFetchError(w, err)
+		httputil.HandleError(w, "Could not get state for epoch's starting slot: "+err.Error(), http.StatusInternalServerError)
 		return nil, false
 	}
 	return st, true

beacon-chain/rpc/eth/validator/BUILD.bazel

@@ -19,6 +19,7 @@ go_library(
         "//beacon-chain/cache:go_default_library",
         "//beacon-chain/core/feed/operation:go_default_library",
         "//beacon-chain/core/helpers:go_default_library",
+        "//beacon-chain/core/transition:go_default_library",
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/operations/attestations:go_default_library",
         "//beacon-chain/operations/synccommittee:go_default_library",
@@ -77,7 +78,6 @@ go_test(
         "//beacon-chain/rpc/core:go_default_library",
         "//beacon-chain/rpc/eth/rewards/testing:go_default_library",
         "//beacon-chain/rpc/eth/shared/testing:go_default_library",
-        "//beacon-chain/rpc/lookup:go_default_library",
         "//beacon-chain/rpc/testutil:go_default_library",
         "//beacon-chain/state:go_default_library",
         "//beacon-chain/state/stategen:go_default_library",

beacon-chain/rpc/eth/validator/handlers.go

@@ -19,6 +19,7 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/builder"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/cache"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/helpers"
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/transition"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/core"
 	rpchelpers "github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/eth/helpers"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/eth/shared"
@@ -897,15 +898,20 @@ func (s *Server) GetAttesterDuties(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// For next epoch requests, we use the current epoch's state since committee
-	// assignments for next epoch can be computed from current epoch's state.
-	epochForState := requestedEpoch
+	var startSlot primitives.Slot
 	if requestedEpoch == nextEpoch {
-		epochForState = currentEpoch
+		startSlot, err = slots.EpochStart(currentEpoch)
+	} else {
+		startSlot, err = slots.EpochStart(requestedEpoch)
 	}
-	st, err := s.Stater.StateByEpoch(ctx, epochForState)
 	if err != nil {
-		shared.WriteStateFetchError(w, err)
+		httputil.HandleError(w, fmt.Sprintf("Could not get start slot from epoch %d: %v", requestedEpoch, err), http.StatusInternalServerError)
+		return
+	}
+
+	st, err := s.Stater.StateBySlot(ctx, startSlot)
+	if err != nil {
+		httputil.HandleError(w, "Could not get state: "+err.Error(), http.StatusInternalServerError)
 		return
 	}
 
@@ -1014,11 +1020,39 @@ func (s *Server) GetProposerDuties(w http.ResponseWriter, r *http.Request) {
 		nextEpochLookahead = true
 	}
 
-	st, err := s.Stater.StateByEpoch(ctx, requestedEpoch)
+	epochStartSlot, err := slots.EpochStart(requestedEpoch)
 	if err != nil {
-		shared.WriteStateFetchError(w, err)
+		httputil.HandleError(w, fmt.Sprintf("Could not get start slot of epoch %d: %v", requestedEpoch, err), http.StatusInternalServerError)
 		return
 	}
+	var st state.BeaconState
+	// if the requested epoch is new, use the head state and the next slot cache
+	if requestedEpoch < currentEpoch {
+		st, err = s.Stater.StateBySlot(ctx, epochStartSlot)
+		if err != nil {
+			httputil.HandleError(w, fmt.Sprintf("Could not get state for slot %d: %v ", epochStartSlot, err), http.StatusInternalServerError)
+			return
+		}
+	} else {
+		st, err = s.HeadFetcher.HeadState(ctx)
+		if err != nil {
+			httputil.HandleError(w, fmt.Sprintf("Could not get head state: %v ", err), http.StatusInternalServerError)
+			return
+		}
+		// Notice that even for Fulu requests for the next epoch, we are only advancing the state to the start of the current epoch.
+		if st.Slot() < epochStartSlot {
+			headRoot, err := s.HeadFetcher.HeadRoot(ctx)
+			if err != nil {
+				httputil.HandleError(w, fmt.Sprintf("Could not get head root: %v ", err), http.StatusInternalServerError)
+				return
+			}
+			st, err = transition.ProcessSlotsUsingNextSlotCache(ctx, st, headRoot, epochStartSlot)
+			if err != nil {
+				httputil.HandleError(w, fmt.Sprintf("Could not process slots up to %d: %v ", epochStartSlot, err), http.StatusInternalServerError)
+				return
+			}
+		}
+	}
 
 	var assignments map[primitives.ValidatorIndex][]primitives.Slot
 	if nextEpochLookahead {
@@ -1069,8 +1103,7 @@ func (s *Server) GetProposerDuties(w http.ResponseWriter, r *http.Request) {
 		httputil.HandleError(w, "Could not check optimistic status: "+err.Error(), http.StatusInternalServerError)
 		return
 	}
-	if err = sortProposerDuties(duties); err != nil {
-		httputil.HandleError(w, "Could not sort proposer duties: "+err.Error(), http.StatusInternalServerError)
+	if !sortProposerDuties(w, duties) {
 		return
 	}
 
@@ -1141,10 +1174,14 @@ func (s *Server) GetSyncCommitteeDuties(w http.ResponseWriter, r *http.Request)
 	}
 
 	startingEpoch := min(requestedEpoch, currentEpoch)
-
-	st, err := s.Stater.StateByEpoch(ctx, startingEpoch)
+	slot, err := slots.EpochStart(startingEpoch)
 	if err != nil {
-		shared.WriteStateFetchError(w, err)
+		httputil.HandleError(w, "Could not get sync committee slot: "+err.Error(), http.StatusInternalServerError)
+		return
+	}
+	st, err := s.Stater.State(ctx, []byte(strconv.FormatUint(uint64(slot), 10)))
+	if err != nil {
+		httputil.HandleError(w, "Could not get sync committee state: "+err.Error(), http.StatusInternalServerError)
 		return
 	}
 
@@ -1290,7 +1327,7 @@ func (s *Server) GetLiveness(w http.ResponseWriter, r *http.Request) {
 		}
 		st, err = s.Stater.StateBySlot(ctx, epochEnd)
 		if err != nil {
-			shared.WriteStateFetchError(w, err)
+			httputil.HandleError(w, "Could not get slot for requested epoch: "+err.Error(), http.StatusInternalServerError)
 			return
 		}
 		participation, err = st.CurrentEpochParticipation()
@@ -1410,20 +1447,22 @@ func syncCommitteeDutiesAndVals(
 	return duties, vals, nil
 }
 
-func sortProposerDuties(duties []*structs.ProposerDuty) error {
-	var err error
+func sortProposerDuties(w http.ResponseWriter, duties []*structs.ProposerDuty) bool {
+	ok := true
 	sort.Slice(duties, func(i, j int) bool {
-		si, parseErr := strconv.ParseUint(duties[i].Slot, 10, 64)
-		if parseErr != nil {
-			err = errors.Wrap(parseErr, "could not parse slot")
+		si, err := strconv.ParseUint(duties[i].Slot, 10, 64)
+		if err != nil {
+			httputil.HandleError(w, "Could not parse slot: "+err.Error(), http.StatusInternalServerError)
+			ok = false
 			return false
 		}
-		sj, parseErr := strconv.ParseUint(duties[j].Slot, 10, 64)
-		if parseErr != nil {
-			err = errors.Wrap(parseErr, "could not parse slot")
+		sj, err := strconv.ParseUint(duties[j].Slot, 10, 64)
+		if err != nil {
+			httputil.HandleError(w, "Could not parse slot: "+err.Error(), http.StatusInternalServerError)
+			ok = false
 			return false
 		}
 		return si < sj
 	})
-	return err
+	return ok
 }

beacon-chain/rpc/eth/validator/handlers_test.go

@@ -25,7 +25,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/operations/synccommittee"
 	p2pmock "github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/testing"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/core"
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/lookup"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/testutil"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/state"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/state/stategen"
@@ -2007,7 +2006,6 @@ func TestGetAttesterDuties(t *testing.T) {
 		TimeFetcher:           chain,
 		SyncChecker:           &mockSync.Sync{IsSyncing: false},
 		OptimisticModeFetcher: chain,
-		HeadFetcher:           chain,
 		BeaconDB:              db,
 	}
 
@@ -2186,7 +2184,6 @@ func TestGetAttesterDuties(t *testing.T) {
 			Stater:                &testutil.MockStater{StatesBySlot: map[primitives.Slot]state.BeaconState{0: bs}},
 			TimeFetcher:           chain,
 			OptimisticModeFetcher: chain,
-			HeadFetcher:           chain,
 			SyncChecker:           &mockSync.Sync{IsSyncing: false},
 			BeaconDB:              db,
 		}
@@ -2227,62 +2224,6 @@ func TestGetAttesterDuties(t *testing.T) {
 		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
 		assert.Equal(t, http.StatusServiceUnavailable, e.Code)
 	})
-	t.Run("state not found returns 404", func(t *testing.T) {
-		chainSlot := primitives.Slot(0)
-		chain := &mockChain.ChainService{
-			State: bs, Root: genesisRoot[:], Slot: &chainSlot,
-		}
-		stateNotFoundErr := lookup.NewStateNotFoundError(8192, []byte("test"))
-		s := &Server{
-			Stater:                &testutil.MockStater{CustomError: &stateNotFoundErr},
-			TimeFetcher:           chain,
-			SyncChecker:           &mockSync.Sync{IsSyncing: false},
-			OptimisticModeFetcher: chain,
-			HeadFetcher:           chain,
-		}
-
-		var body bytes.Buffer
-		_, err = body.WriteString("[\"0\"]")
-		require.NoError(t, err)
-		request := httptest.NewRequest(http.MethodGet, "http://www.example.com/eth/v1/validator/duties/attester/{epoch}", &body)
-		request.SetPathValue("epoch", "0")
-		writer := httptest.NewRecorder()
-		writer.Body = &bytes.Buffer{}
-
-		s.GetAttesterDuties(writer, request)
-		assert.Equal(t, http.StatusNotFound, writer.Code)
-		e := &httputil.DefaultJsonError{}
-		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
-		assert.Equal(t, http.StatusNotFound, e.Code)
-		assert.StringContains(t, "State not found", e.Message)
-	})
-	t.Run("state fetch error returns 500", func(t *testing.T) {
-		chainSlot := primitives.Slot(0)
-		chain := &mockChain.ChainService{
-			State: bs, Root: genesisRoot[:], Slot: &chainSlot,
-		}
-		s := &Server{
-			Stater:                &testutil.MockStater{CustomError: errors.New("internal error")},
-			TimeFetcher:           chain,
-			SyncChecker:           &mockSync.Sync{IsSyncing: false},
-			OptimisticModeFetcher: chain,
-			HeadFetcher:           chain,
-		}
-
-		var body bytes.Buffer
-		_, err = body.WriteString("[\"0\"]")
-		require.NoError(t, err)
-		request := httptest.NewRequest(http.MethodGet, "http://www.example.com/eth/v1/validator/duties/attester/{epoch}", &body)
-		request.SetPathValue("epoch", "0")
-		writer := httptest.NewRecorder()
-		writer.Body = &bytes.Buffer{}
-
-		s.GetAttesterDuties(writer, request)
-		assert.Equal(t, http.StatusInternalServerError, writer.Code)
-		e := &httputil.DefaultJsonError{}
-		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
-		assert.Equal(t, http.StatusInternalServerError, e.Code)
-	})
 }
 
 func TestGetProposerDuties(t *testing.T) {
@@ -2486,60 +2427,6 @@ func TestGetProposerDuties(t *testing.T) {
 		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
 		assert.Equal(t, http.StatusServiceUnavailable, e.Code)
 	})
-	t.Run("state not found returns 404", func(t *testing.T) {
-		bs, err := transition.GenesisBeaconState(t.Context(), deposits, 0, eth1Data)
-		require.NoError(t, err)
-		chainSlot := primitives.Slot(0)
-		chain := &mockChain.ChainService{
-			State: bs, Root: genesisRoot[:], Slot: &chainSlot,
-		}
-		stateNotFoundErr := lookup.NewStateNotFoundError(8192, []byte("test"))
-		s := &Server{
-			Stater:                &testutil.MockStater{CustomError: &stateNotFoundErr},
-			TimeFetcher:           chain,
-			SyncChecker:           &mockSync.Sync{IsSyncing: false},
-			OptimisticModeFetcher: chain,
-			HeadFetcher:           chain,
-		}
-
-		request := httptest.NewRequest(http.MethodGet, "http://www.example.com/eth/v1/validator/duties/proposer/{epoch}", nil)
-		request.SetPathValue("epoch", "0")
-		writer := httptest.NewRecorder()
-		writer.Body = &bytes.Buffer{}
-
-		s.GetProposerDuties(writer, request)
-		assert.Equal(t, http.StatusNotFound, writer.Code)
-		e := &httputil.DefaultJsonError{}
-		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
-		assert.Equal(t, http.StatusNotFound, e.Code)
-		assert.StringContains(t, "State not found", e.Message)
-	})
-	t.Run("state fetch error returns 500", func(t *testing.T) {
-		bs, err := transition.GenesisBeaconState(t.Context(), deposits, 0, eth1Data)
-		require.NoError(t, err)
-		chainSlot := primitives.Slot(0)
-		chain := &mockChain.ChainService{
-			State: bs, Root: genesisRoot[:], Slot: &chainSlot,
-		}
-		s := &Server{
-			Stater:                &testutil.MockStater{CustomError: errors.New("internal error")},
-			TimeFetcher:           chain,
-			SyncChecker:           &mockSync.Sync{IsSyncing: false},
-			OptimisticModeFetcher: chain,
-			HeadFetcher:           chain,
-		}
-
-		request := httptest.NewRequest(http.MethodGet, "http://www.example.com/eth/v1/validator/duties/proposer/{epoch}", nil)
-		request.SetPathValue("epoch", "0")
-		writer := httptest.NewRecorder()
-		writer.Body = &bytes.Buffer{}
-
-		s.GetProposerDuties(writer, request)
-		assert.Equal(t, http.StatusInternalServerError, writer.Code)
-		e := &httputil.DefaultJsonError{}
-		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
-		assert.Equal(t, http.StatusInternalServerError, e.Code)
-	})
 }
 
 func TestGetSyncCommitteeDuties(t *testing.T) {
@@ -2570,7 +2457,7 @@ func TestGetSyncCommitteeDuties(t *testing.T) {
 	}
 	require.NoError(t, st.SetNextSyncCommittee(nextCommittee))
 
-	mockChainService := &mockChain.ChainService{Genesis: genesisTime, State: st}
+	mockChainService := &mockChain.ChainService{Genesis: genesisTime}
 	s := &Server{
 		Stater:                &testutil.MockStater{BeaconState: st},
 		SyncChecker:           &mockSync.Sync{IsSyncing: false},
@@ -2761,7 +2648,7 @@ func TestGetSyncCommitteeDuties(t *testing.T) {
 				return newSyncPeriodSt
 			}
 		}
-		mockChainService := &mockChain.ChainService{Genesis: genesisTime, Slot: &newSyncPeriodStartSlot, State: newSyncPeriodSt}
+		mockChainService := &mockChain.ChainService{Genesis: genesisTime, Slot: &newSyncPeriodStartSlot}
 		s := &Server{
 			Stater:                &testutil.MockStater{BeaconState: stateFetchFn(newSyncPeriodStartSlot)},
 			SyncChecker:           &mockSync.Sync{IsSyncing: false},
@@ -2842,7 +2729,8 @@ func TestGetSyncCommitteeDuties(t *testing.T) {
 		slot, err := slots.EpochStart(1)
 		require.NoError(t, err)
 
-		st2 := st.Copy()
+		st2, err := util.NewBeaconStateBellatrix()
+		require.NoError(t, err)
 		require.NoError(t, st2.SetSlot(slot))
 
 		mockChainService := &mockChain.ChainService{
@@ -2856,7 +2744,7 @@ func TestGetSyncCommitteeDuties(t *testing.T) {
 			State: st2,
 		}
 		s := &Server{
-			Stater:                &testutil.MockStater{BeaconState: st2},
+			Stater:                &testutil.MockStater{BeaconState: st},
 			SyncChecker:           &mockSync.Sync{IsSyncing: false},
 			TimeFetcher:           mockChainService,
 			HeadFetcher:           mockChainService,
@@ -2901,62 +2789,6 @@ func TestGetSyncCommitteeDuties(t *testing.T) {
 		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
 		assert.Equal(t, http.StatusServiceUnavailable, e.Code)
 	})
-	t.Run("state not found returns 404", func(t *testing.T) {
-		slot := 2 * params.BeaconConfig().SlotsPerEpoch
-		chainService := &mockChain.ChainService{
-			Slot: &slot,
-		}
-		stateNotFoundErr := lookup.NewStateNotFoundError(8192, []byte("test"))
-		s := &Server{
-			Stater:                &testutil.MockStater{CustomError: &stateNotFoundErr},
-			TimeFetcher:           chainService,
-			SyncChecker:           &mockSync.Sync{IsSyncing: false},
-			OptimisticModeFetcher: chainService,
-			HeadFetcher:           chainService,
-		}
-
-		var body bytes.Buffer
-		_, err := body.WriteString("[\"1\"]")
-		require.NoError(t, err)
-		request := httptest.NewRequest(http.MethodGet, "http://www.example.com/eth/v1/validator/duties/sync/{epoch}", &body)
-		request.SetPathValue("epoch", "1")
-		writer := httptest.NewRecorder()
-		writer.Body = &bytes.Buffer{}
-
-		s.GetSyncCommitteeDuties(writer, request)
-		assert.Equal(t, http.StatusNotFound, writer.Code)
-		e := &httputil.DefaultJsonError{}
-		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
-		assert.Equal(t, http.StatusNotFound, e.Code)
-		assert.StringContains(t, "State not found", e.Message)
-	})
-	t.Run("state fetch error returns 500", func(t *testing.T) {
-		slot := 2 * params.BeaconConfig().SlotsPerEpoch
-		chainService := &mockChain.ChainService{
-			Slot: &slot,
-		}
-		s := &Server{
-			Stater:                &testutil.MockStater{CustomError: errors.New("internal error")},
-			TimeFetcher:           chainService,
-			SyncChecker:           &mockSync.Sync{IsSyncing: false},
-			OptimisticModeFetcher: chainService,
-			HeadFetcher:           chainService,
-		}
-
-		var body bytes.Buffer
-		_, err := body.WriteString("[\"1\"]")
-		require.NoError(t, err)
-		request := httptest.NewRequest(http.MethodGet, "http://www.example.com/eth/v1/validator/duties/sync/{epoch}", &body)
-		request.SetPathValue("epoch", "1")
-		writer := httptest.NewRecorder()
-		writer.Body = &bytes.Buffer{}
-
-		s.GetSyncCommitteeDuties(writer, request)
-		assert.Equal(t, http.StatusInternalServerError, writer.Code)
-		e := &httputil.DefaultJsonError{}
-		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
-		assert.Equal(t, http.StatusInternalServerError, e.Code)
-	})
 }
 
 func TestPrepareBeaconProposer(t *testing.T) {

beacon-chain/rpc/lookup/BUILD.bazel

@@ -11,7 +11,6 @@ go_library(
     deps = [
         "//beacon-chain/blockchain:go_default_library",
         "//beacon-chain/core/peerdas:go_default_library",
-        "//beacon-chain/core/transition:go_default_library",
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/db/filesystem:go_default_library",
         "//beacon-chain/rpc/core:go_default_library",

beacon-chain/rpc/lookup/blocker.go

@@ -450,7 +450,7 @@ func (p *BeaconDbBlocker) blobsDataFromStoredDataColumns(root [fieldparams.RootL
 	if count < peerdas.MinimumColumnCountToReconstruct() {
 		// There is no way to reconstruct the data columns.
 		return nil, &core.RpcError{
-			Err:    errors.Errorf("the node does not custody enough data columns to reconstruct blobs - please start the beacon node with the `--%s` flag to ensure this call to succeed", flags.SemiSupernode.Name),
+			Err:    errors.Errorf("the node does not custody enough data columns to reconstruct blobs - please start the beacon node with the `--%s` flag to ensure this call to succeed, or retry later if it is already the case", flags.Supernode.Name),
 			Reason: core.NotFound,
 		}
 	}
@@ -628,8 +628,6 @@ func (p *BeaconDbBlocker) neededDataColumnSidecars(root [fieldparams.RootLength]
 //   - no block, 404
 //   - block exists, before Fulu fork, 400 (data columns are not supported before Fulu fork)
 func (p *BeaconDbBlocker) DataColumns(ctx context.Context, id string, indices []int) ([]blocks.VerifiedRODataColumn, *core.RpcError) {
-	const numberOfColumns = fieldparams.NumberOfColumns
-
 	// Check for genesis block first (not supported for data columns)
 	if id == "genesis" {
 		return nil, &core.RpcError{Err: errors.New("data columns are not supported for Phase 0 fork"), Reason: core.BadRequest}
@@ -683,6 +681,7 @@ func (p *BeaconDbBlocker) DataColumns(ctx context.Context, id string, indices []
 		}
 	} else {
 		// Validate and convert indices
+		numberOfColumns := params.BeaconConfig().NumberOfColumns
 		for _, index := range indices {
 			if index < 0 || uint64(index) >= numberOfColumns {
 				return nil, &core.RpcError{

beacon-chain/rpc/lookup/stater.go

@@ -8,7 +8,6 @@ import (
 	"strings"
 
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain"
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/transition"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/db"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/state"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/state/stategen"
@@ -83,8 +82,8 @@ type StateRootNotFoundError struct {
 }
 
 // NewStateRootNotFoundError creates a new error instance.
-func NewStateRootNotFoundError(stateRootsSize int) StateRootNotFoundError {
-	return StateRootNotFoundError{
+func NewStateRootNotFoundError(stateRootsSize int) StateNotFoundError {
+	return StateNotFoundError{
 		message: fmt.Sprintf("state root not found in the last %d state roots", stateRootsSize),
 	}
 }
@@ -99,7 +98,6 @@ type Stater interface {
 	State(ctx context.Context, id []byte) (state.BeaconState, error)
 	StateRoot(ctx context.Context, id []byte) ([]byte, error)
 	StateBySlot(ctx context.Context, slot primitives.Slot) (state.BeaconState, error)
-	StateByEpoch(ctx context.Context, epoch primitives.Epoch) (state.BeaconState, error)
 }
 
 // BeaconDbStater is an implementation of Stater. It retrieves states from the beacon chain database.
@@ -269,46 +267,6 @@ func (p *BeaconDbStater) StateBySlot(ctx context.Context, target primitives.Slot
 	return st, nil
 }
 
-// StateByEpoch returns the state for the start of the requested epoch.
-// For current or next epoch, it uses the head state and next slot cache for efficiency.
-// For past epochs, it replays blocks from the most recent canonical state.
-func (p *BeaconDbStater) StateByEpoch(ctx context.Context, epoch primitives.Epoch) (state.BeaconState, error) {
-	ctx, span := trace.StartSpan(ctx, "statefetcher.StateByEpoch")
-	defer span.End()
-
-	targetSlot, err := slots.EpochStart(epoch)
-	if err != nil {
-		return nil, errors.Wrap(err, "could not get epoch start slot")
-	}
-
-	currentSlot := p.GenesisTimeFetcher.CurrentSlot()
-	currentEpoch := slots.ToEpoch(currentSlot)
-
-	// For past epochs, use the replay mechanism
-	if epoch < currentEpoch {
-		return p.StateBySlot(ctx, targetSlot)
-	}
-
-	// For current or next epoch, use head state + next slot cache (much faster)
-	headState, err := p.ChainInfoFetcher.HeadState(ctx)
-	if err != nil {
-		return nil, errors.Wrap(err, "could not get head state")
-	}
-
-	// If head state is already at or past the target slot, return it
-	if headState.Slot() >= targetSlot {
-		return headState, nil
-	}
-
-	// Process slots using the next slot cache
-	headRoot := p.ChainInfoFetcher.CachedHeadRoot()
-	st, err := transition.ProcessSlotsUsingNextSlotCache(ctx, headState, headRoot[:], targetSlot)
-	if err != nil {
-		return nil, errors.Wrapf(err, "could not process slots up to %d", targetSlot)
-	}
-	return st, nil
-}
-
 func (p *BeaconDbStater) headStateRoot(ctx context.Context) ([]byte, error) {
 	b, err := p.ChainInfoFetcher.HeadBlock(ctx)
 	if err != nil {

beacon-chain/rpc/lookup/stater_test.go

@@ -444,111 +444,3 @@ func TestStateBySlot_AfterHeadSlot(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, primitives.Slot(101), st.Slot())
 }
-
-func TestStateByEpoch(t *testing.T) {
-	ctx := t.Context()
-	slotsPerEpoch := params.BeaconConfig().SlotsPerEpoch
-
-	t.Run("current epoch uses head state", func(t *testing.T) {
-		// Head is at slot 5 (epoch 0), requesting epoch 0
-		headSlot := primitives.Slot(5)
-		headSt, err := statenative.InitializeFromProtoPhase0(&ethpb.BeaconState{Slot: headSlot})
-		require.NoError(t, err)
-
-		currentSlot := headSlot
-		mock := &chainMock.ChainService{State: headSt, Slot: &currentSlot}
-		p := BeaconDbStater{ChainInfoFetcher: mock, GenesisTimeFetcher: mock}
-
-		st, err := p.StateByEpoch(ctx, 0)
-		require.NoError(t, err)
-		// Should return head state since it's already past epoch start
-		assert.Equal(t, headSlot, st.Slot())
-	})
-
-	t.Run("current epoch processes slots to epoch start", func(t *testing.T) {
-		// Head is at slot 5 (epoch 0), requesting epoch 1
-		// Current slot is 32 (epoch 1), so epoch 1 is current epoch
-		headSlot := primitives.Slot(5)
-		headSt, err := statenative.InitializeFromProtoPhase0(&ethpb.BeaconState{Slot: headSlot})
-		require.NoError(t, err)
-
-		currentSlot := slotsPerEpoch // slot 32, epoch 1
-		mock := &chainMock.ChainService{State: headSt, Slot: &currentSlot}
-		p := BeaconDbStater{ChainInfoFetcher: mock, GenesisTimeFetcher: mock}
-
-		// Note: This will fail since ProcessSlotsUsingNextSlotCache requires proper setup
-		// In real usage, the transition package handles this properly
-		_, err = p.StateByEpoch(ctx, 1)
-		// The error is expected since we don't have a fully initialized beacon state
-		// that can process slots (missing committees, etc.)
-		assert.NotNil(t, err)
-	})
-
-	t.Run("past epoch uses replay", func(t *testing.T) {
-		// Head is at epoch 2, requesting epoch 0 (past)
-		headSlot := slotsPerEpoch * 2 // slot 64, epoch 2
-		headSt, err := statenative.InitializeFromProtoPhase0(&ethpb.BeaconState{Slot: headSlot})
-		require.NoError(t, err)
-
-		pastEpochSt, err := statenative.InitializeFromProtoPhase0(&ethpb.BeaconState{Slot: 0})
-		require.NoError(t, err)
-
-		currentSlot := headSlot
-		mock := &chainMock.ChainService{State: headSt, Slot: &currentSlot}
-		mockReplayer := mockstategen.NewReplayerBuilder()
-		mockReplayer.SetMockStateForSlot(pastEpochSt, 0)
-		p := BeaconDbStater{ChainInfoFetcher: mock, GenesisTimeFetcher: mock, ReplayerBuilder: mockReplayer}
-
-		st, err := p.StateByEpoch(ctx, 0)
-		require.NoError(t, err)
-		assert.Equal(t, primitives.Slot(0), st.Slot())
-	})
-
-	t.Run("next epoch uses head state path", func(t *testing.T) {
-		// Head is at slot 30 (epoch 0), requesting epoch 1 (next)
-		// Current slot is 30 (epoch 0), so epoch 1 is next epoch
-		headSlot := primitives.Slot(30)
-		headSt, err := statenative.InitializeFromProtoPhase0(&ethpb.BeaconState{Slot: headSlot})
-		require.NoError(t, err)
-
-		currentSlot := headSlot
-		mock := &chainMock.ChainService{State: headSt, Slot: &currentSlot}
-		p := BeaconDbStater{ChainInfoFetcher: mock, GenesisTimeFetcher: mock}
-
-		// Note: This will fail since ProcessSlotsUsingNextSlotCache requires proper setup
-		_, err = p.StateByEpoch(ctx, 1)
-		// The error is expected since we don't have a fully initialized beacon state
-		assert.NotNil(t, err)
-	})
-
-	t.Run("head state already at target slot returns immediately", func(t *testing.T) {
-		// Head is at slot 32 (epoch 1 start), requesting epoch 1
-		headSlot := slotsPerEpoch // slot 32
-		headSt, err := statenative.InitializeFromProtoPhase0(&ethpb.BeaconState{Slot: headSlot})
-		require.NoError(t, err)
-
-		currentSlot := headSlot
-		mock := &chainMock.ChainService{State: headSt, Slot: &currentSlot}
-		p := BeaconDbStater{ChainInfoFetcher: mock, GenesisTimeFetcher: mock}
-
-		st, err := p.StateByEpoch(ctx, 1)
-		require.NoError(t, err)
-		assert.Equal(t, headSlot, st.Slot())
-	})
-
-	t.Run("head state past target slot returns head state", func(t *testing.T) {
-		// Head is at slot 40, requesting epoch 1 (starts at slot 32)
-		headSlot := primitives.Slot(40)
-		headSt, err := statenative.InitializeFromProtoPhase0(&ethpb.BeaconState{Slot: headSlot})
-		require.NoError(t, err)
-
-		currentSlot := headSlot
-		mock := &chainMock.ChainService{State: headSt, Slot: &currentSlot}
-		p := BeaconDbStater{ChainInfoFetcher: mock, GenesisTimeFetcher: mock}
-
-		st, err := p.StateByEpoch(ctx, 1)
-		require.NoError(t, err)
-		// Returns head state since it's already >= epoch start
-		assert.Equal(t, headSlot, st.Slot())
-	})
-}

beacon-chain/rpc/prysm/v1alpha1/validator/proposer.go

@@ -7,7 +7,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/OffchainLabs/go-bitfield"
 	builderapi "github.com/OffchainLabs/prysm/v7/api/client/builder"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/builder"
@@ -309,7 +308,6 @@ func (vs *Server) ProposeBeaconBlock(ctx context.Context, req *ethpb.GenericSign
 	}
 
 	rob, err := blocks.NewROBlockWithRoot(block, root)
-	var partialColumns []blocks.PartialDataColumn
 	if block.IsBlinded() {
 		block, blobSidecars, err = vs.handleBlindedBlock(ctx, block)
 		if errors.Is(err, builderapi.ErrBadGateway) {
@@ -317,7 +315,7 @@ func (vs *Server) ProposeBeaconBlock(ctx context.Context, req *ethpb.GenericSign
 			return &ethpb.ProposeResponse{BlockRoot: root[:]}, nil
 		}
 	} else if block.Version() >= version.Deneb {
-		blobSidecars, dataColumnSidecars, partialColumns, err = vs.handleUnblindedBlock(rob, req)
+		blobSidecars, dataColumnSidecars, err = vs.handleUnblindedBlock(rob, req)
 	}
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "%s: %v", "handle block failed", err)
@@ -337,7 +335,7 @@ func (vs *Server) ProposeBeaconBlock(ctx context.Context, req *ethpb.GenericSign
 
 	wg.Wait()
 
-	if err := vs.broadcastAndReceiveSidecars(ctx, block, root, blobSidecars, dataColumnSidecars, partialColumns); err != nil {
+	if err := vs.broadcastAndReceiveSidecars(ctx, block, root, blobSidecars, dataColumnSidecars); err != nil {
 		return nil, status.Errorf(codes.Internal, "Could not broadcast/receive sidecars: %v", err)
 	}
 	if err := <-errChan; err != nil {
@@ -354,10 +352,9 @@ func (vs *Server) broadcastAndReceiveSidecars(
 	root [fieldparams.RootLength]byte,
 	blobSidecars []*ethpb.BlobSidecar,
 	dataColumnSidecars []blocks.RODataColumn,
-	partialColumns []blocks.PartialDataColumn,
 ) error {
 	if block.Version() >= version.Fulu {
-		if err := vs.broadcastAndReceiveDataColumns(ctx, dataColumnSidecars, partialColumns); err != nil {
+		if err := vs.broadcastAndReceiveDataColumns(ctx, dataColumnSidecars); err != nil {
 			return errors.Wrap(err, "broadcast and receive data columns")
 		}
 		return nil
@@ -406,41 +403,34 @@ func (vs *Server) handleBlindedBlock(ctx context.Context, block interfaces.Signe
 func (vs *Server) handleUnblindedBlock(
 	block blocks.ROBlock,
 	req *ethpb.GenericSignedBeaconBlock,
-) ([]*ethpb.BlobSidecar, []blocks.RODataColumn, []blocks.PartialDataColumn, error) {
+) ([]*ethpb.BlobSidecar, []blocks.RODataColumn, error) {
 	rawBlobs, proofs, err := blobsAndProofs(req)
 	if err != nil {
-		return nil, nil, nil, err
+		return nil, nil, err
 	}
 
 	if block.Version() >= version.Fulu {
 		// Compute cells and proofs from the blobs and cell proofs.
 		cellsPerBlob, proofsPerBlob, err := peerdas.ComputeCellsAndProofsFromFlat(rawBlobs, proofs)
 		if err != nil {
-			return nil, nil, nil, errors.Wrap(err, "compute cells and proofs")
+			return nil, nil, errors.Wrap(err, "compute cells and proofs")
 		}
 
 		// Construct data column sidecars from the signed block and cells and proofs.
 		roDataColumnSidecars, err := peerdas.DataColumnSidecars(cellsPerBlob, proofsPerBlob, peerdas.PopulateFromBlock(block))
 		if err != nil {
-			return nil, nil, nil, errors.Wrap(err, "data column sidcars")
+			return nil, nil, errors.Wrap(err, "data column sidcars")
 		}
 
-		included := bitfield.NewBitlist(uint64(len(cellsPerBlob)))
-		included = included.Not() // all bits set to 1
-		partialColumns, err := peerdas.PartialColumns(included, cellsPerBlob, proofsPerBlob, peerdas.PopulateFromBlock(block))
-		if err != nil {
-			return nil, nil, nil, errors.Wrap(err, "data column sidcars")
-		}
-
-		return nil, roDataColumnSidecars, partialColumns, nil
+		return nil, roDataColumnSidecars, nil
 	}
 
 	blobSidecars, err := BuildBlobSidecars(block, rawBlobs, proofs)
 	if err != nil {
-		return nil, nil, nil, errors.Wrap(err, "build blob sidecars")
+		return nil, nil, errors.Wrap(err, "build blob sidecars")
 	}
 
-	return blobSidecars, nil, nil, nil
+	return blobSidecars, nil, nil
 }
 
 // broadcastReceiveBlock broadcasts a block and handles its reception.
@@ -507,7 +497,7 @@ func (vs *Server) broadcastAndReceiveBlobs(ctx context.Context, sidecars []*ethp
 }
 
 // broadcastAndReceiveDataColumns handles the broadcasting and reception of data columns sidecars.
-func (vs *Server) broadcastAndReceiveDataColumns(ctx context.Context, roSidecars []blocks.RODataColumn, partialColumns []blocks.PartialDataColumn) error {
+func (vs *Server) broadcastAndReceiveDataColumns(ctx context.Context, roSidecars []blocks.RODataColumn) error {
 	// We built this block ourselves, so we can upgrade the read only data column sidecar into a verified one.
 	verifiedSidecars := make([]blocks.VerifiedRODataColumn, 0, len(roSidecars))
 	for _, sidecar := range roSidecars {
@@ -516,7 +506,7 @@ func (vs *Server) broadcastAndReceiveDataColumns(ctx context.Context, roSidecars
 	}
 
 	// Broadcast sidecars (non blocking).
-	if err := vs.P2P.BroadcastDataColumnSidecars(ctx, verifiedSidecars, partialColumns); err != nil {
+	if err := vs.P2P.BroadcastDataColumnSidecars(ctx, verifiedSidecars); err != nil {
 		return errors.Wrap(err, "broadcast data column sidecars")
 	}
 

beacon-chain/rpc/testutil/BUILD.bazel

@@ -26,6 +26,5 @@ go_library(
         "//proto/prysm/v1alpha1:go_default_library",
         "//testing/require:go_default_library",
         "//testing/util:go_default_library",
-        "//time/slots:go_default_library",
     ],
 )

beacon-chain/rpc/testutil/mock_stater.go

@@ -6,7 +6,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/state"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
 	"github.com/OffchainLabs/prysm/v7/encoding/bytesutil"
-	"github.com/OffchainLabs/prysm/v7/time/slots"
 )
 
 // MockStater is a fake implementation of lookup.Stater.
@@ -15,7 +14,6 @@ type MockStater struct {
 	StateProviderFunc func(ctx context.Context, stateId []byte) (state.BeaconState, error)
 	BeaconStateRoot   []byte
 	StatesBySlot      map[primitives.Slot]state.BeaconState
-	StatesByEpoch     map[primitives.Epoch]state.BeaconState
 	StatesByRoot      map[[32]byte]state.BeaconState
 	CustomError       error
 }
@@ -45,22 +43,3 @@ func (m *MockStater) StateRoot(context.Context, []byte) ([]byte, error) {
 func (m *MockStater) StateBySlot(_ context.Context, s primitives.Slot) (state.BeaconState, error) {
 	return m.StatesBySlot[s], nil
 }
-
-// StateByEpoch --
-func (m *MockStater) StateByEpoch(_ context.Context, e primitives.Epoch) (state.BeaconState, error) {
-	if m.CustomError != nil {
-		return nil, m.CustomError
-	}
-	if m.StatesByEpoch != nil {
-		return m.StatesByEpoch[e], nil
-	}
-	// Fall back to StatesBySlot if StatesByEpoch is not set
-	slot, err := slots.EpochStart(e)
-	if err != nil {
-		return nil, err
-	}
-	if m.StatesBySlot != nil {
-		return m.StatesBySlot[slot], nil
-	}
-	return m.BeaconState, nil
-}

beacon-chain/state/stategen/migrate.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/state"
-	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
 	"github.com/OffchainLabs/prysm/v7/encoding/bytesutil"
 	"github.com/OffchainLabs/prysm/v7/monitoring/tracing/trace"
 	"github.com/sirupsen/logrus"
@@ -38,84 +37,76 @@ func (s *State) MigrateToCold(ctx context.Context, fRoot [32]byte) error {
 		return nil
 	}
 
-	// Calculate the first archived point slot >= oldFSlot (but > 0).
-	// This avoids iterating through every slot and only visits archived points directly.
-	var startSlot primitives.Slot
-	if oldFSlot == 0 {
-		startSlot = s.slotsPerArchivedPoint
-	} else {
-		// Round up to the next archived point
-		startSlot = (oldFSlot + s.slotsPerArchivedPoint - 1) / s.slotsPerArchivedPoint * s.slotsPerArchivedPoint
-	}
-
-	// Start at the first archived point after old finalized slot, stop before current finalized slot.
-	// Jump directly between archived points.
-	for slot := startSlot; slot < fSlot; slot += s.slotsPerArchivedPoint {
+	// Start at previous finalized slot, stop at current finalized slot (it will be handled in the next migration).
+	// If the slot is on archived point, save the state of that slot to the DB.
+	for slot := oldFSlot; slot < fSlot; slot++ {
 		if ctx.Err() != nil {
 			return ctx.Err()
 		}
 
-		cached, exists, err := s.epochBoundaryStateCache.getBySlot(slot)
-		if err != nil {
-			return fmt.Errorf("could not get epoch boundary state for slot %d", slot)
-		}
-
-		var aRoot [32]byte
-		var aState state.BeaconState
-
-		// When the epoch boundary state is not in cache due to skip slot scenario,
-		// we have to regenerate the state which will represent epoch boundary.
-		// By finding the highest available block below epoch boundary slot, we
-		// generate the state for that block root.
-		if exists {
-			aRoot = cached.root
-			aState = cached.state
-		} else {
-			_, roots, err := s.beaconDB.HighestRootsBelowSlot(ctx, slot)
+		if slot%s.slotsPerArchivedPoint == 0 && slot != 0 {
+			cached, exists, err := s.epochBoundaryStateCache.getBySlot(slot)
 			if err != nil {
-				return err
+				return fmt.Errorf("could not get epoch boundary state for slot %d", slot)
 			}
-			// Given the block has been finalized, the db should not have more than one block in a given slot.
-			// We should error out when this happens.
-			if len(roots) != 1 {
-				return errUnknownBlock
-			}
-			aRoot = roots[0]
-			// There's no need to generate the state if the state already exists in the DB.
-			// We can skip saving the state.
-			if !s.beaconDB.HasState(ctx, aRoot) {
-				aState, err = s.StateByRoot(ctx, aRoot)
+
+			var aRoot [32]byte
+			var aState state.BeaconState
+
+			// When the epoch boundary state is not in cache due to skip slot scenario,
+			// we have to regenerate the state which will represent epoch boundary.
+			// By finding the highest available block below epoch boundary slot, we
+			// generate the state for that block root.
+			if exists {
+				aRoot = cached.root
+				aState = cached.state
+			} else {
+				_, roots, err := s.beaconDB.HighestRootsBelowSlot(ctx, slot)
 				if err != nil {
 					return err
 				}
-			}
-		}
-
-		if s.beaconDB.HasState(ctx, aRoot) {
-			// If you are migrating a state and its already part of the hot state cache saved to the db,
-			// you can just remove it from the hot state cache as it becomes redundant.
-			s.saveHotStateDB.lock.Lock()
-			roots := s.saveHotStateDB.blockRootsOfSavedStates
-			for i := range roots {
-				if aRoot == roots[i] {
-					s.saveHotStateDB.blockRootsOfSavedStates = append(roots[:i], roots[i+1:]...)
-					// There shouldn't be duplicated roots in `blockRootsOfSavedStates`.
-					// Break here is ok.
-					break
+				// Given the block has been finalized, the db should not have more than one block in a given slot.
+				// We should error out when this happens.
+				if len(roots) != 1 {
+					return errUnknownBlock
+				}
+				aRoot = roots[0]
+				// There's no need to generate the state if the state already exists in the DB.
+				// We can skip saving the state.
+				if !s.beaconDB.HasState(ctx, aRoot) {
+					aState, err = s.StateByRoot(ctx, aRoot)
+					if err != nil {
+						return err
+					}
 				}
 			}
-			s.saveHotStateDB.lock.Unlock()
-			continue
-		}
 
-		if err := s.beaconDB.SaveState(ctx, aState, aRoot); err != nil {
-			return err
+			if s.beaconDB.HasState(ctx, aRoot) {
+				// If you are migrating a state and its already part of the hot state cache saved to the db,
+				// you can just remove it from the hot state cache as it becomes redundant.
+				s.saveHotStateDB.lock.Lock()
+				roots := s.saveHotStateDB.blockRootsOfSavedStates
+				for i := range roots {
+					if aRoot == roots[i] {
+						s.saveHotStateDB.blockRootsOfSavedStates = append(roots[:i], roots[i+1:]...)
+						// There shouldn't be duplicated roots in `blockRootsOfSavedStates`.
+						// Break here is ok.
+						break
+					}
+				}
+				s.saveHotStateDB.lock.Unlock()
+				continue
+			}
+
+			if err := s.beaconDB.SaveState(ctx, aState, aRoot); err != nil {
+				return err
+			}
+			log.WithFields(
+				logrus.Fields{
+					"slot": aState.Slot(),
+					"root": hex.EncodeToString(bytesutil.Trunc(aRoot[:])),
+				}).Info("Saved state in DB")
 		}
-		log.WithFields(
-			logrus.Fields{
-				"slot": aState.Slot(),
-				"root": hex.EncodeToString(bytesutil.Trunc(aRoot[:])),
-			}).Info("Saved state in DB")
 	}
 
 	// Update finalized info in memory.

beacon-chain/sync/BUILD.bazel

@@ -58,7 +58,6 @@ go_library(
         "validate_bls_to_execution_change.go",
         "validate_data_column.go",
         "validate_light_client.go",
-        "validate_partial_header.go",
         "validate_proposer_slashing.go",
         "validate_sync_committee_message.go",
         "validate_sync_contribution_proof.go",
@@ -99,7 +98,6 @@ go_library(
         "//beacon-chain/operations/voluntaryexits:go_default_library",
         "//beacon-chain/p2p:go_default_library",
         "//beacon-chain/p2p/encoder:go_default_library",
-        "//beacon-chain/p2p/partialdatacolumnbroadcaster:go_default_library",
         "//beacon-chain/p2p/peers:go_default_library",
         "//beacon-chain/p2p/types:go_default_library",
         "//beacon-chain/slasher/types:go_default_library",

beacon-chain/sync/backfill/BUILD.bazel

@@ -11,6 +11,7 @@ go_library(
         "fulu_transition.go",
         "log.go",
         "metrics.go",
+        "needs.go",
         "pool.go",
         "service.go",
         "status.go",
@@ -61,6 +62,7 @@ go_test(
         "columns_test.go",
         "fulu_transition_test.go",
         "log_test.go",
+        "needs_test.go",
         "pool_test.go",
         "service_test.go",
         "status_test.go",
@@ -69,7 +71,6 @@ go_test(
     ],
     embed = [":go_default_library"],
     deps = [
-        "//beacon-chain/blockchain/kzg:go_default_library",
         "//beacon-chain/core/peerdas:go_default_library",
         "//beacon-chain/core/signing:go_default_library",
         "//beacon-chain/das:go_default_library",

beacon-chain/sync/backfill/batch.go

@@ -6,7 +6,6 @@ import (
 	"sort"
 	"time"
 
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/das"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/sync"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
 	eth "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
@@ -183,6 +182,7 @@ func (b batch) withState(s batchState) batch {
 	}
 	if s == batchImportComplete {
 		backfillBatchTimeRoundtrip.Observe(float64(time.Since(b.firstScheduled).Milliseconds()))
+		log.WithFields(b.logFields()).Debug("Backfill batch imported")
 	}
 	b.state = s
 	b.seq += 1
@@ -238,7 +238,7 @@ func (b batch) validatingColumnRequest(cb *columnBisector) (*validatingColumnReq
 // to the toDownload structure for any blocks where those columns failed, and resets the bisector state.
 // Note that this method will also prune any columns that have expired, meaning we no longer need them
 // per spec and/or our backfill & retention settings.
-func resetToRetryColumns(b batch, needs das.CurrentNeeds) batch {
+func resetToRetryColumns(b batch, needs currentNeeds) batch {
 	// return the given batch as-is if it isn't in a state that this func should handle.
 	if b.columns == nil || b.columns.bisector == nil || len(b.columns.bisector.errs) == 0 {
 		return b.transitionToNext()
@@ -290,9 +290,9 @@ func (b batch) workComplete() bool {
 	return b.state == batchImportable
 }
 
-func (b batch) expired(needs das.CurrentNeeds) bool {
-	if !needs.Block.At(b.end - 1) {
-		log.WithFields(b.logFields()).WithField("retentionStartSlot", needs.Block.Begin).Debug("Batch outside retention window")
+func (b batch) expired(needs currentNeeds) bool {
+	if !needs.block.at(b.end - 1) {
+		log.WithFields(b.logFields()).WithField("retentionStartSlot", needs.block.begin).Debug("Batch outside retention window")
 		return true
 	}
 	return false

beacon-chain/sync/backfill/batcher.go

@@ -1,7 +1,6 @@
 package backfill
 
 import (
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/das"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
 	"github.com/pkg/errors"
 )
@@ -13,7 +12,7 @@ var errCannotDecreaseMinimum = errors.New("the minimum backfill slot can only be
 type batchSequencer struct {
 	batcher      batcher
 	seq          []batch
-	currentNeeds func() das.CurrentNeeds
+	currentNeeds func() currentNeeds
 }
 
 // sequence() is meant as a verb "arrange in a particular order".
@@ -162,24 +161,24 @@ func (c *batchSequencer) numTodo() int {
 	return todo
 }
 
-func newBatchSequencer(seqLen int, max, size primitives.Slot, needsCb func() das.CurrentNeeds) *batchSequencer {
+func newBatchSequencer(seqLen int, max, size primitives.Slot, needsCb func() currentNeeds) *batchSequencer {
 	b := batcher{currentNeeds: needsCb, max: max, size: size}
 	seq := make([]batch, seqLen)
 	return &batchSequencer{batcher: b, seq: seq, currentNeeds: needsCb}
 }
 
 type batcher struct {
-	currentNeeds func() das.CurrentNeeds
+	currentNeeds func() currentNeeds
 	max          primitives.Slot
 	size         primitives.Slot
 }
 
 func (r batcher) remaining(upTo primitives.Slot) int {
 	needs := r.currentNeeds()
-	if !needs.Block.At(upTo) {
+	if !needs.block.at(upTo) {
 		return 0
 	}
-	delta := upTo - needs.Block.Begin
+	delta := upTo - needs.block.begin
 	if delta%r.size != 0 {
 		return int(delta/r.size) + 1
 	}
@@ -197,12 +196,12 @@ func (r batcher) before(upTo primitives.Slot) batch {
 	// The upper bound is exclusive, so we shouldn't return in this case where the previous
 	// batch beginning sits at the exact slot of the start of the retention window. In that case
 	// we've actually hit the end of the sync sequence.
-	if !needs.Block.At(upTo) || needs.Block.Begin == upTo {
+	if !needs.block.at(upTo) || needs.block.begin == upTo {
 		return batch{begin: upTo, end: upTo, state: batchEndSequence}
 	}
 
-	begin := needs.Block.Begin
-	if upTo > r.size+needs.Block.Begin {
+	begin := needs.block.begin
+	if upTo > r.size+needs.block.begin {
 		begin = upTo - r.size
 	}
 

beacon-chain/sync/backfill/batcher_expiration_test.go

@@ -3,7 +3,6 @@ package backfill
 import (
 	"testing"
 
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/das"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
 	"github.com/OffchainLabs/prysm/v7/testing/require"
 )
@@ -30,11 +29,11 @@ func newDynamicNeeds(blockBegin, blockEnd primitives.Slot) *dynamicNeeds {
 	}
 }
 
-func (d *dynamicNeeds) get() das.CurrentNeeds {
-	return das.CurrentNeeds{
-		Block: das.NeedSpan{Begin: d.blockBegin, End: d.blockEnd},
-		Blob:  das.NeedSpan{Begin: d.blobBegin, End: d.blobEnd},
-		Col:   das.NeedSpan{Begin: d.colBegin, End: d.colEnd},
+func (d *dynamicNeeds) get() currentNeeds {
+	return currentNeeds{
+		block: needSpan{begin: d.blockBegin, end: d.blockEnd},
+		blob:  needSpan{begin: d.blobBegin, end: d.blobEnd},
+		col:   needSpan{begin: d.colBegin, end: d.colEnd},
 	}
 }
 

beacon-chain/sync/backfill/batcher_test.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/das"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
 	"github.com/OffchainLabs/prysm/v7/testing/require"
 )
@@ -213,8 +212,8 @@ func TestBatchSequencer(t *testing.T) {
 	// set the min for the batcher close to the lowest slot. This will force the next batch to be partial and the batch
 	// after that to be the final batch.
 	newMin := seq.seq[len(seq.seq)-1].begin - 30
-	seq.currentNeeds = func() das.CurrentNeeds {
-		return das.CurrentNeeds{Block: das.NeedSpan{Begin: newMin, End: seq.batcher.max}}
+	seq.currentNeeds = func() currentNeeds {
+		return currentNeeds{block: needSpan{begin: newMin, end: seq.batcher.max}}
 	}
 	seq.batcher.currentNeeds = seq.currentNeeds
 	first = seq.seq[0]
@@ -878,8 +877,8 @@ func TestBatcherRemaining(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			needs := func() das.CurrentNeeds {
-				return das.CurrentNeeds{Block: das.NeedSpan{Begin: tc.min, End: tc.upTo + 1}}
+			needs := func() currentNeeds {
+				return currentNeeds{block: needSpan{begin: tc.min, end: tc.upTo + 1}}
 			}
 			b := batcher{size: tc.size, currentNeeds: needs}
 			result := b.remaining(tc.upTo)

beacon-chain/sync/backfill/blobs.go

@@ -33,7 +33,7 @@ type blobSummary struct {
 type blobSyncConfig struct {
 	nbv          verification.NewBlobVerifier
 	store        *filesystem.BlobStorage
-	currentNeeds func() das.CurrentNeeds
+	currentNeeds func() currentNeeds
 }
 
 func newBlobSync(current primitives.Slot, vbs verifiedROBlocks, cfg *blobSyncConfig) (*blobSync, error) {
@@ -42,12 +42,7 @@ func newBlobSync(current primitives.Slot, vbs verifiedROBlocks, cfg *blobSyncCon
 		return nil, err
 	}
 	bbv := newBlobBatchVerifier(cfg.nbv)
-	shouldRetain := func(slot primitives.Slot) bool {
-		needs := cfg.currentNeeds()
-		return needs.Blob.At(slot)
-	}
-	as := das.NewLazilyPersistentStore(cfg.store, bbv, shouldRetain)
-
+	as := das.NewLazilyPersistentStore(cfg.store, bbv)
 	return &blobSync{current: current, expected: expected, bbv: bbv, store: as}, nil
 }
 

beacon-chain/sync/backfill/blobs_test.go

@@ -3,7 +3,6 @@ package backfill
 import (
 	"testing"
 
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/das"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/db/filesystem"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/verification"
 	"github.com/OffchainLabs/prysm/v7/config/params"
@@ -28,11 +27,12 @@ func testBlobGen(t *testing.T, start primitives.Slot, n int) ([]blocks.ROBlock,
 	return blks, blobs
 }
 
-func setupCurrentNeeds(t *testing.T, current primitives.Slot) das.SyncNeeds {
-	cs := func() primitives.Slot { return current }
-	sn, err := das.NewSyncNeeds(cs, nil, 0)
-	require.NoError(t, err)
-	return sn
+func setupCurrentNeeds(t *testing.T, current primitives.Slot) syncNeeds {
+	return syncNeeds{
+		current: func() primitives.Slot { return current },
+		deneb:   slots.UnsafeEpochStart(params.BeaconConfig().DenebForkEpoch),
+		fulu:    slots.UnsafeEpochStart(params.BeaconConfig().FuluForkEpoch),
+	}
 }
 
 func TestValidateNext_happy(t *testing.T) {
@@ -64,32 +64,28 @@ func TestValidateNext_happy(t *testing.T) {
 }
 
 func TestValidateNext_cheapErrors(t *testing.T) {
-	denebSlot, err := slots.EpochStart(params.BeaconConfig().DenebForkEpoch)
-	require.NoError(t, err)
 	current := primitives.Slot(128)
 	syncNeeds := setupCurrentNeeds(t, current)
 	cfg := &blobSyncConfig{
 		nbv:          testNewBlobVerifier(),
 		store:        filesystem.NewEphemeralBlobStorage(t),
-		currentNeeds: syncNeeds.Currently,
+		currentNeeds: syncNeeds.currently,
 	}
-	blks, blobs := testBlobGen(t, denebSlot, 2)
+	blks, blobs := testBlobGen(t, syncNeeds.deneb, 2)
 	bsync, err := newBlobSync(current, blks, cfg)
 	require.NoError(t, err)
 	require.ErrorIs(t, bsync.validateNext(blobs[len(blobs)-1][0]), errUnexpectedResponseContent)
 }
 
 func TestValidateNext_sigMatch(t *testing.T) {
-	denebSlot, err := slots.EpochStart(params.BeaconConfig().DenebForkEpoch)
-	require.NoError(t, err)
 	current := primitives.Slot(128)
 	syncNeeds := setupCurrentNeeds(t, current)
 	cfg := &blobSyncConfig{
 		nbv:          testNewBlobVerifier(),
 		store:        filesystem.NewEphemeralBlobStorage(t),
-		currentNeeds: syncNeeds.Currently,
+		currentNeeds: syncNeeds.currently,
 	}
-	blks, blobs := testBlobGen(t, denebSlot, 1)
+	blks, blobs := testBlobGen(t, syncNeeds.deneb, 1)
 	bsync, err := newBlobSync(current, blks, cfg)
 	require.NoError(t, err)
 	blobs[0][0].SignedBlockHeader.Signature = bytesutil.PadTo([]byte("derp"), 48)

beacon-chain/sync/backfill/columns.go

@@ -65,9 +65,9 @@ func (cs *columnBatch) needed() peerdas.ColumnIndices {
 
 // pruneExpired removes any columns from the batch that are no longer needed.
 // If `pruned` is non-nil, it is populated with the roots that were removed.
-func (cs *columnBatch) pruneExpired(needs das.CurrentNeeds, pruned map[[32]byte]struct{}) {
+func (cs *columnBatch) pruneExpired(needs currentNeeds, pruned map[[32]byte]struct{}) {
 	for root, td := range cs.toDownload {
-		if !needs.Col.At(td.slot) {
+		if !needs.col.at(td.slot) {
 			delete(cs.toDownload, root)
 			if pruned != nil {
 				pruned[root] = struct{}{}
@@ -118,16 +118,12 @@ func newColumnSync(ctx context.Context, b batch, blks verifiedROBlocks, current
 	if cb == nil {
 		return &columnSync{}, nil
 	}
-	shouldRetain := func(sl primitives.Slot) bool {
-		needs := cfg.currentNeeds()
-		return needs.Col.At(sl)
-	}
 
 	bisector := newColumnBisector(cfg.downscore)
 	return &columnSync{
 		columnBatch: cb,
 		current:     current,
-		store:       das.NewLazilyPersistentStoreColumn(cfg.colStore, cfg.newVC, p.NodeID(), cgc, bisector, shouldRetain),
+		store:       das.NewLazilyPersistentStoreColumn(cfg.colStore, cfg.newVC, p.NodeID(), cgc, bisector),
 		bisector:    bisector,
 	}, nil
 }
@@ -235,12 +231,12 @@ func currentCustodiedColumns(ctx context.Context, p p2p.P2P) (peerdas.ColumnIndi
 	return peerdas.NewColumnIndicesFromMap(peerInfo.CustodyColumns), nil
 }
 
-func buildColumnBatch(ctx context.Context, b batch, blks verifiedROBlocks, p p2p.P2P, store *filesystem.DataColumnStorage, needs das.CurrentNeeds) (*columnBatch, error) {
+func buildColumnBatch(ctx context.Context, b batch, blks verifiedROBlocks, p p2p.P2P, store *filesystem.DataColumnStorage, needs currentNeeds) (*columnBatch, error) {
 	if len(blks) == 0 {
 		return nil, nil
 	}
 
-	if !needs.Col.At(b.begin) && !needs.Col.At(b.end-1) {
+	if !needs.col.at(b.begin) && !needs.col.at(b.end-1) {
 		return nil, nil
 	}
 
@@ -254,7 +250,7 @@ func buildColumnBatch(ctx context.Context, b batch, blks verifiedROBlocks, p p2p
 	}
 	for _, b := range blks {
 		slot := b.Block().Slot()
-		if !needs.Col.At(slot) {
+		if !needs.col.at(slot) {
 			continue
 		}
 		cmts, err := b.Block().Body().BlobKzgCommitments()

beacon-chain/sync/backfill/columns_test.go

@@ -5,7 +5,6 @@ import (
 	"slices"
 	"testing"
 
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain/kzg"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/peerdas"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/das"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/db/filesystem"
@@ -787,9 +786,6 @@ func TestValidatingColumnRequest_CountedValidation(t *testing.T) {
 		remaining := peerdas.NewColumnIndicesFromSlice([]uint64{0})
 		bisector := newColumnBisector(func(peer.ID, string, error) {})
 
-		sr := func(slot primitives.Slot) bool {
-			return true
-		}
 		vcr := &validatingColumnRequest{
 			columnSync: &columnSync{
 				columnBatch: &columnBatch{
@@ -800,7 +796,7 @@ func TestValidatingColumnRequest_CountedValidation(t *testing.T) {
 						},
 					},
 				},
-				store:   das.NewLazilyPersistentStoreColumn(colStore, testNewDataColumnsVerifier(), p2p.NodeID(), 1, bisector, sr),
+				store:   das.NewLazilyPersistentStoreColumn(colStore, testNewDataColumnsVerifier(), p2p.NodeID(), 1, bisector),
 				current: currentSlot,
 				peer:    mockPeer,
 			},
@@ -875,7 +871,7 @@ func TestNewColumnSync(t *testing.T) {
 		cfg := &workerCfg{
 			colStore:     colStore,
 			downscore:    func(peer.ID, string, error) {},
-			currentNeeds: func() das.CurrentNeeds { return mockCurrentSpecNeeds() },
+			currentNeeds: func() currentNeeds { return mockCurrentSpecNeeds() },
 		}
 
 		cs, err := newColumnSync(ctx, b, blks, current, p2p, cfg)
@@ -1629,140 +1625,3 @@ func testNewDataColumnsVerifier() verification.NewDataColumnsVerifier {
 		return &verification.MockDataColumnsVerifier{}
 	}
 }
-
-// Helper to create a verifier that marks all columns as verified
-func markAllVerified(m *verification.MockDataColumnsVerifier) verification.NewDataColumnsVerifier {
-	return func(cols []blocks.RODataColumn, reqs []verification.Requirement) verification.DataColumnsVerifier {
-		m.AppendRODataColumns(cols...)
-		return m
-	}
-}
-
-func TestRetentionCheckWithOverride(t *testing.T) {
-	require.NoError(t, kzg.Start())
-	fuluEpoch := params.BeaconConfig().FuluForkEpoch
-	fuluSlot, err := slots.EpochStart(fuluEpoch)
-	require.NoError(t, err)
-	colRetention := params.BeaconConfig().MinEpochsForDataColumnSidecarsRequest
-	savedSlotRange := func(start, end primitives.Slot) map[primitives.Slot]struct{} {
-		m := make(map[primitives.Slot]struct{})
-		for s := start; s < end; s++ {
-			m[s] = struct{}{}
-		}
-		return m
-	}
-	cases := []struct {
-		name          string
-		span          das.NeedSpan
-		savedSlots    map[primitives.Slot]struct{}
-		currentEpoch  primitives.Epoch
-		retentionFlag primitives.Epoch
-		cgc           uint64
-		oldestSlot    *primitives.Slot
-	}{
-		{
-			name:         "Before retention period, none saved",
-			span:         das.NeedSpan{Begin: fuluSlot, End: fuluSlot + 5},
-			currentEpoch: fuluEpoch + colRetention + 1,
-			cgc:          4,
-		},
-		{
-			name:         "At retention period boundary, all saved",
-			span:         das.NeedSpan{Begin: fuluSlot, End: fuluSlot + 5},
-			currentEpoch: fuluEpoch + colRetention,
-			savedSlots:   savedSlotRange(fuluSlot, fuluSlot+5),
-			cgc:          4,
-		},
-		{
-			name:         "Across boundary, saved after",
-			span:         das.NeedSpan{Begin: fuluSlot + 30, End: fuluSlot + 35},
-			currentEpoch: fuluEpoch + colRetention + 1,
-			savedSlots:   savedSlotRange(fuluSlot+32, fuluSlot+35),
-			cgc:          4,
-		},
-		{
-			name:          "Before retention period with override, all saved",
-			span:          das.NeedSpan{Begin: fuluSlot, End: fuluSlot + 5},
-			currentEpoch:  fuluEpoch + colRetention*2, // well past retention without override
-			savedSlots:    savedSlotRange(fuluSlot, fuluSlot+5),
-			retentionFlag: fuluEpoch + colRetention*2, // flag covers all slots
-		},
-		{
-			name:          "Before retention period and just before override, none saved",
-			span:          das.NeedSpan{Begin: fuluSlot, End: fuluSlot + 5},
-			currentEpoch:  1 + fuluEpoch + colRetention*2, // current slot is beyond base retention
-			retentionFlag: fuluEpoch + colRetention*2,     // stops 1 short flag coverage
-		},
-	}
-	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
-			colStore := filesystem.NewEphemeralDataColumnStorage(t)
-			p2p := p2ptest.NewTestP2P(t)
-			bisector := newColumnBisector(func(peer.ID, string, error) {})
-			current, err := slots.EpochStart(tc.currentEpoch)
-			require.NoError(t, err)
-			cs := func() primitives.Slot { return current }
-			sn, err := das.NewSyncNeeds(cs, tc.oldestSlot, tc.retentionFlag)
-			require.NoError(t, err)
-
-			sr := func(slot primitives.Slot) bool {
-				current := sn.Currently()
-				return current.Col.At(slot)
-			}
-
-			nodeId := p2p.NodeID()
-			peerInfo, _, err := peerdas.Info(nodeId, tc.cgc)
-			require.NoError(t, err)
-			indices := peerdas.NewColumnIndicesFromMap(peerInfo.CustodyColumns)
-			// note that markAllVerified considers all columns that get through the rest of the da check as verified,
-			// not all columns in the test.
-			verifier := markAllVerified(&verification.MockDataColumnsVerifier{})
-			avs := das.NewLazilyPersistentStoreColumn(colStore, verifier, p2p.NodeID(), tc.cgc, bisector, sr)
-			fixtures := testBlockWithColumnSpan(t, tc.span.Begin, tc.span, 3)
-			blocks := make([]blocks.ROBlock, 0, len(fixtures))
-			for _, fix := range fixtures {
-				for _, col := range fix.cols {
-					if indices.Has(col.Index) {
-						require.NoError(t, avs.Persist(current, col))
-					}
-				}
-				blocks = append(blocks, fix.block)
-			}
-			require.NoError(t, avs.IsDataAvailable(t.Context(), current, blocks...))
-			for _, block := range blocks {
-				slot := block.Block().Slot()
-				sum := colStore.Summary(block.Root())
-				stored := sum.Stored()
-				// If we don't expect storage for the slot, verify none are stored
-				if _, exists := tc.savedSlots[slot]; !exists {
-					require.Equal(t, 0, len(stored), "expected no columns stored for slot %d", slot)
-					continue
-				}
-				// If we expect storage, verify all stored columns are expected, and that all expected columns are stored.
-				missing := indices.Copy()
-				for idx := range stored {
-					require.Equal(t, true, missing.Has(idx), "unexpected stored column %d for slot %d", idx, slot)
-					missing.Unset(idx)
-				}
-				require.Equal(t, 0, missing.Count(), "expected all columns to be stored for slot %d, missing %v", slot, missing.ToSlice())
-			}
-		})
-	}
-}
-
-type blockFixture struct {
-	block blocks.ROBlock
-	cols  []blocks.RODataColumn
-	vcols []blocks.VerifiedRODataColumn
-}
-
-func testBlockWithColumnSpan(t *testing.T, slot primitives.Slot, colSpan das.NeedSpan, numBlobs int) []blockFixture {
-	res := make([]blockFixture, 0, colSpan.End-colSpan.Begin)
-	parent := [32]byte{0x00}
-	for bs := colSpan.Begin; bs < colSpan.End; bs++ {
-		blk, c, vc := util.GenerateTestFuluBlockWithSidecars(t, numBlobs, util.WithSlot(bs), util.WithParentRoot(parent))
-		res = append(res, blockFixture{block: blk, cols: c, vcols: vc})
-		parent = blk.Root()
-	}
-	return res
-}