New Data Column Sidecar Storage Design, Data Columns as a First-Class Citizen & Unit Testing (#15061)

* DB Filesystem: Move all data column related code to `data_columns.go`

Only code move.

* Implement data columns storage

* Kasey comment: Fix typo

* Kasey comment: Fix clutter

* Kasey comment: `IsDataAvailable`: Remove `nodeID`.

* Kasey comment: indice ==> index

* Kasey comment: Move `CreateTestVerifiedRoDataColumnSidecars` in `beacon-chain/verification/fake`.

* `Store` ==> `Save`.

* Kasey comment: AAAA!

* Kasey comment: Fix typo.

* Kasey comment: Add comment.

* Kasey commnet: Stop exporting errors for nothing.

* Kasey comment: Read all metadata at once.

* Kasey comment: Compute file size instead of reading it from stats.

* Kasey comment: Lock mutexes before checking if the file exists.

* Kasey comment: `limit` ==> `nonZeroOffset`.

* Kasey comment: `DataColumnStorage.Get`: Set verified into the `verification package`.

* Kasey comment: `prune` - Flatten the `==` case.

* Kasey comment: Implement and use `storageIndices`.

* `DataColumnsAlignWithBlock`: Move into its own file.

* `DataColumnSidecar`: Rename variables to stick with
https://github.com/ethereum/consensus-specs/blob/dev/specs/fulu/das-core.md#datacolumnsidecar

* Kasey comment: Add `file.Sync`.

* `DataColumnStorage.Get`: Remove useless cast.

* (Internal) Kasey comment: Set automatically the count of saved data columns.

Author: nalepae

beacon-chain/blockchain/BUILD.bazel

@@ -144,6 +144,7 @@ go_test(
         "//beacon-chain/core/feed/state:go_default_library",
         "//beacon-chain/core/helpers:go_default_library",
         "//beacon-chain/core/light-client:go_default_library",
+        "//beacon-chain/core/peerdas:go_default_library",
         "//beacon-chain/core/signing:go_default_library",
         "//beacon-chain/core/transition:go_default_library",
         "//beacon-chain/das:go_default_library",
@@ -165,6 +166,7 @@ go_test(
         "//beacon-chain/state:go_default_library",
         "//beacon-chain/state/state-native:go_default_library",
         "//beacon-chain/state/stategen:go_default_library",
+        "//beacon-chain/verification:go_default_library",
         "//config/features:go_default_library",
         "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",

beacon-chain/blockchain/execution_engine.go

@@ -444,6 +444,9 @@ func (s *Service) removeInvalidBlockAndState(ctx context.Context, blkRoots [][32
 			// Blobs may not exist for some blocks, leading to deletion failures. Log such errors at debug level.
 			log.WithError(err).Debug("Could not remove blob from blob storage")
 		}
+		if err := s.dataColumnStorage.Remove(root); err != nil {
+			log.WithError(err).Debug("Could not remove data columns from data column storage")
+		}
 	}
 	return nil
 }

beacon-chain/blockchain/kzg/kzg.go

@@ -99,7 +99,7 @@ func RecoverCellsAndKZGProofs(cellIndices []uint64, partialCells []Cell) (CellsA
 
 	ckzgCells, ckzgProofs, err := ckzg4844.RecoverCellsAndKZGProofs(cellIndices, ckzgPartialCells)
 	if err != nil {
-		return CellsAndProofs{}, err
+		return CellsAndProofs{}, errors.Wrap(err, "recover cells and KZG proofs")
 	}
 
 	return makeCellsAndProofs(ckzgCells[:], ckzgProofs[:])

beacon-chain/blockchain/options.go

@@ -1,6 +1,8 @@
 package blockchain
 
 import (
+	"time"
+
 	"github.com/prysmaticlabs/prysm/v5/async/event"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/cache"
 	statefeed "github.com/prysmaticlabs/prysm/v5/beacon-chain/core/feed/state"
@@ -208,6 +210,14 @@ func WithBlobStorage(b *filesystem.BlobStorage) Option {
 	}
 }
 
+// WithDataColumnStorage sets the data column storage backend for the blockchain service.
+func WithDataColumnStorage(b *filesystem.DataColumnStorage) Option {
+	return func(s *Service) error {
+		s.dataColumnStorage = b
+		return nil
+	}
+}
+
 func WithSyncChecker(checker Checker) Option {
 	return func(s *Service) error {
 		s.cfg.SyncChecker = checker
@@ -221,3 +231,10 @@ func WithCustodyInfo(custodyInfo *peerdas.CustodyInfo) Option {
 		return nil
 	}
 }
+
+func WithGenesisTime(genesisTime time.Time) Option {
+	return func(s *Service) error {
+		s.genesisTime = genesisTime
+		return nil
+	}
+}