peerDAS: Implement dataColumnSidecarByRootRPCHandler. (#15405)

* `CreateTestVerifiedRoDataColumnSidecars`: Use consistent block root.

* peerDAS: Implement `dataColumnSidecarByRootRPCHandler`.

* Fix James' comment.

* Fix James' comment.

cmd/beacon-chain/flags/base.go

@@ -212,6 +212,11 @@ var (
 		Usage: "The factor by which blob batch limit may increase on burst.",
 		Value: 3,
 	}
+	DataColumnBatchLimit = &cli.IntFlag{
+		Name:  "data-column-batch-limit",
+		Usage: "The amount of data columns the local peer is bounded to request and respond to in a batch.",
+		Value: 4096,
+	}
 	// DisableDebugRPCEndpoints disables the debug Beacon API namespace.
 	DisableDebugRPCEndpoints = &cli.BoolFlag{
 		Name:  "disable-debug-rpc-endpoints",

cmd/beacon-chain/flags/config.go

@@ -16,6 +16,7 @@ type GlobalFlags struct {
 	BlockBatchLimit            int
 	BlockBatchLimitBurstFactor int
 	BlobBatchLimit             int
+	DataColumnBatchLimit       int
 	BlobBatchLimitBurstFactor  int
 }
 
@@ -53,6 +54,7 @@ func ConfigureGlobalFlags(ctx *cli.Context) {
 	cfg.BlockBatchLimitBurstFactor = ctx.Int(BlockBatchLimitBurstFactor.Name)
 	cfg.BlobBatchLimit = ctx.Int(BlobBatchLimit.Name)
 	cfg.BlobBatchLimitBurstFactor = ctx.Int(BlobBatchLimitBurstFactor.Name)
+	cfg.DataColumnBatchLimit = ctx.Int(DataColumnBatchLimit.Name)
 	cfg.MinimumPeersPerSubnet = ctx.Int(MinPeersPerSubnet.Name)
 	cfg.MaxConcurrentDials = ctx.Int(MaxConcurrentDials.Name)
 	configureMinimumPeers(ctx, cfg)

cmd/beacon-chain/storage/options.go

@@ -14,7 +14,6 @@ import (
 )
 
 var (
-	// BlobStoragePathFlag defines a flag to start the beacon chain from a give genesis state file.
 	BlobStoragePathFlag = &cli.PathFlag{
 		Name:  "blob-path",
 		Usage: "Location for blob storage. Default location will be a 'blobs' directory next to the beacon db.",
@@ -30,6 +29,10 @@ var (
 		Usage: layoutFlagUsage(),
 		Value: filesystem.LayoutNameFlat,
 	}
+	DataColumnStoragePathFlag = &cli.PathFlag{
+		Name:  "data-column-path",
+		Usage: "Location for data column storage. Default location will be a 'data-columns' directory next to the beacon db.",
+	}
 )
 
 func layoutOptions() string {
@@ -54,15 +57,23 @@ func validateLayoutFlag(_ *cli.Context, v string) error {
 // create a cancellable context. If we switch to using App.RunContext, we can set up this cancellation in the cmd
 // package instead, and allow the functional options to tap into context cancellation.
 func BeaconNodeOptions(c *cli.Context) ([]node.Option, error) {
-	e, err := blobRetentionEpoch(c)
+	blobRetentionEpoch, err := blobRetentionEpoch(c)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrap(err, "blob retention epoch")
 	}
-	opts := []node.Option{node.WithBlobStorageOptions(
-		filesystem.WithBlobRetentionEpochs(e),
+
+	blobStorageOptions := node.WithBlobStorageOptions(
+		filesystem.WithBlobRetentionEpochs(blobRetentionEpoch),
 		filesystem.WithBasePath(blobStoragePath(c)),
 		filesystem.WithLayout(c.String(BlobStorageLayout.Name)), // This is validated in the Action func for BlobStorageLayout.
-	)}
+	)
+
+	dataColumnStorageOption := node.WithDataColumnStorageOptions(
+		filesystem.WithDataColumnRetentionEpochs(blobRetentionEpoch),
+		filesystem.WithDataColumnBasePath(dataColumnStoragePath(c)),
+	)
+
+	opts := []node.Option{blobStorageOptions, dataColumnStorageOption}
 	return opts, nil
 }
 
@@ -75,6 +86,16 @@ func blobStoragePath(c *cli.Context) string {
 	return blobsPath
 }
 
+func dataColumnStoragePath(c *cli.Context) string {
+	dataColumnsPath := c.Path(DataColumnStoragePathFlag.Name)
+	if dataColumnsPath == "" {
+		// append a "data-columns" subdir to the end of the data dir path
+		dataColumnsPath = path.Join(c.String(cmd.DataDirFlag.Name), "data-columns")
+	}
+
+	return dataColumnsPath
+}
+
 var errInvalidBlobRetentionEpochs = errors.New("value is smaller than spec minimum")
 
 // blobRetentionEpoch returns the spec default MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUEST