Use functional options for --blob-retention-epochs (#13283)

* blob retention period functional opts * missed unstaged change * missed other init after cleardb * fix ineffassign * fix dup import * config failsafe for tests --------- Co-authored-by: Kasey Kirkham <kasey@users.noreply.github.com>
2026-01-08 21:08:10 -05:00 · 2023-12-06 14:20:34 -06:00
parent 604c82626f
commit 737e0e0d3a
26 changed files with 190 additions and 172 deletions
--- a/beacon-chain/db/db.go
+++ b/beacon-chain/db/db.go
@@ -3,16 +3,9 @@
 package db

 import (
-	"context"
-
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/db/kv"
 )

-// NewDB initializes a new DB.
-func NewDB(ctx context.Context, dirPath string) (Database, error) {
-	return kv.NewKVStore(ctx, dirPath)
-}
-
 // NewFileName uses the KVStoreDatafilePath so that if this layer of
 // indirection between db.NewDB->kv.NewKVStore ever changes, it will be easy to remember
 // to also change this filename indirection at the same time.
--- a/beacon-chain/db/filesystem/BUILD.bazel
+++ b/beacon-chain/db/filesystem/BUILD.bazel
@@ -5,13 +5,11 @@ go_library(
    srcs = [
        "blob.go",
        "ephemeral.go",
-        "flags.go",
    ],
    importpath = "github.com/prysmaticlabs/prysm/v4/beacon-chain/db/filesystem",
    visibility = ["//visibility:public"],
    deps = [
        "//beacon-chain/verification:go_default_library",
-        "//cmd/beacon-chain/flags:go_default_library",
        "//config/fieldparams:go_default_library",
        "//config/params:go_default_library",
        "//consensus-types/blocks:go_default_library",
@@ -24,22 +22,16 @@ go_library(
        "@com_github_pkg_errors//:go_default_library",
        "@com_github_sirupsen_logrus//:go_default_library",
        "@com_github_spf13_afero//:go_default_library",
-        "@com_github_urfave_cli_v2//:go_default_library",
    ],
 )

 go_test(
    name = "go_default_test",
-    srcs = [
-        "blob_test.go",
-        "flags_test.go",
-    ],
+    srcs = ["blob_test.go"],
    embed = [":go_default_library"],
    deps = [
        "//beacon-chain/verification:go_default_library",
-        "//cmd/beacon-chain/flags:go_default_library",
        "//config/fieldparams:go_default_library",
-        "//config/params:go_default_library",
        "//consensus-types/primitives:go_default_library",
        "//encoding/bytesutil:go_default_library",
        "//proto/prysm/v1alpha1:go_default_library",
@@ -48,6 +40,5 @@ go_test(
        "@com_github_ethereum_go_ethereum//common/hexutil:go_default_library",
        "@com_github_prysmaticlabs_fastssz//:go_default_library",
        "@com_github_spf13_afero//:go_default_library",
-        "@com_github_urfave_cli_v2//:go_default_library",
    ],
 )
--- a/beacon-chain/db/filesystem/blob.go
+++ b/beacon-chain/db/filesystem/blob.go
@@ -13,6 +13,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/verification"
 	fieldparams "github.com/prysmaticlabs/prysm/v4/config/fieldparams"
+	"github.com/prysmaticlabs/prysm/v4/config/params"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/blocks"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
 	"github.com/prysmaticlabs/prysm/v4/io/file"
@@ -35,16 +36,30 @@ const (
 	directoryPermissions = 0700
 )

+// BlobStorageOption is a functional option for configuring a BlobStorage.
+type BlobStorageOption func(*BlobStorage)
+
+// WithBlobRetentionEpochs is an option that changes the number of epochs blobs will be persisted.
+func WithBlobRetentionEpochs(e primitives.Epoch) BlobStorageOption {
+	return func(b *BlobStorage) {
+		b.retentionEpochs = e
+	}
+}
+
 // NewBlobStorage creates a new instance of the BlobStorage object. Note that the implementation of BlobStorage may
 // attempt to hold a file lock to guarantee exclusive control of the blob storage directory, so this should only be
 // initialized once per beacon node.
-func NewBlobStorage(base string) (*BlobStorage, error) {
+func NewBlobStorage(base string, opts ...BlobStorageOption) (*BlobStorage, error) {
 	base = path.Clean(base)
 	if err := file.MkdirAll(base); err != nil {
 		return nil, fmt.Errorf("failed to create blob storage at %s: %w", base, err)
 	}
 	fs := afero.NewBasePathFs(afero.NewOsFs(), base)
-	return &BlobStorage{fs: fs, retentionEpochs: MaxEpochsToPersistBlobs}, nil
+	b := &BlobStorage{fs: fs, retentionEpochs: params.BeaconNetworkConfig().MinEpochsForBlobsSidecarsRequest}
+	for _, o := range opts {
+		o(b)
+	}
+	return b, nil
 }

 // BlobStorage is the concrete implementation of the filesystem backend for saving and retrieving BlobSidecars.
--- a/beacon-chain/db/filesystem/flags.go
+++ b/beacon-chain/db/filesystem/flags.go
@@ -1,33 +0,0 @@
-package filesystem
-
-import (
-	"fmt"
-
-	"github.com/prysmaticlabs/prysm/v4/cmd/beacon-chain/flags"
-	"github.com/prysmaticlabs/prysm/v4/config/params"
-	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
-	"github.com/urfave/cli/v2"
-)
-
-var MaxEpochsToPersistBlobs = params.BeaconNetworkConfig().MinEpochsForBlobsSidecarsRequest
-
-// ConfigureBlobRetentionEpoch sets the epoch for blob retention based on command-line context. It sets the local config `MaxEpochsToPersistBlobs`.
-// If the flag is not set, the spec default `MinEpochsForBlobsSidecarsRequest` is used.
-// An error if the input epoch is smaller than the spec default value.
-func ConfigureBlobRetentionEpoch(cliCtx *cli.Context) error {
-	// Check if the blob retention epoch flag is set.
-	if cliCtx.IsSet(flags.BlobRetentionEpoch.Name) {
-		// Retrieve and cast the epoch value.
-		epochValue := cliCtx.Uint64(flags.BlobRetentionEpoch.Name)
-		e := primitives.Epoch(epochValue)
-
-		// Validate the epoch value against the spec default.
-		if e < params.BeaconNetworkConfig().MinEpochsForBlobsSidecarsRequest {
-			return fmt.Errorf("%s smaller than spec default, %d < %d", flags.BlobRetentionEpoch.Name, e, params.BeaconNetworkConfig().MinEpochsForBlobsSidecarsRequest)
-		}
-
-		MaxEpochsToPersistBlobs = e
-	}
-
-	return nil
-}
--- a/beacon-chain/db/filesystem/flags_test.go
+++ b/beacon-chain/db/filesystem/flags_test.go
@@ -1,39 +0,0 @@
-package filesystem
-
-import (
-	"flag"
-	"strconv"
-	"testing"
-
-	"github.com/prysmaticlabs/prysm/v4/cmd/beacon-chain/flags"
-	"github.com/prysmaticlabs/prysm/v4/config/params"
-	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
-	"github.com/prysmaticlabs/prysm/v4/testing/require"
-	"github.com/urfave/cli/v2"
-)
-
-func TestConfigureBlobRetentionEpoch(t *testing.T) {
-	MaxEpochsToPersistBlobs = params.BeaconNetworkConfig().MinEpochsForBlobsSidecarsRequest
-	params.SetupTestConfigCleanup(t)
-	app := cli.App{}
-	set := flag.NewFlagSet("test", 0)
-
-	// Test case: Spec default.
-	require.NoError(t, ConfigureBlobRetentionEpoch(cli.NewContext(&app, set, nil)))
-	require.Equal(t, params.BeaconNetworkConfig().MinEpochsForBlobsSidecarsRequest, MaxEpochsToPersistBlobs)
-
-	set.Uint64(flags.BlobRetentionEpoch.Name, 0, "")
-	minEpochsForSidecarRequest := uint64(params.BeaconNetworkConfig().MinEpochsForBlobsSidecarsRequest)
-	require.NoError(t, set.Set(flags.BlobRetentionEpoch.Name, strconv.FormatUint(2*minEpochsForSidecarRequest, 10)))
-	cliCtx := cli.NewContext(&app, set, nil)
-
-	// Test case: Input epoch is greater than or equal to spec value.
-	require.NoError(t, ConfigureBlobRetentionEpoch(cliCtx))
-	require.Equal(t, primitives.Epoch(2*minEpochsForSidecarRequest), MaxEpochsToPersistBlobs)
-
-	// Test case: Input epoch is less than spec value.
-	require.NoError(t, set.Set(flags.BlobRetentionEpoch.Name, strconv.FormatUint(minEpochsForSidecarRequest-1, 10)))
-	cliCtx = cli.NewContext(&app, set, nil)
-	err := ConfigureBlobRetentionEpoch(cliCtx)
-	require.ErrorContains(t, "blob-retention-epochs smaller than spec default", err)
-}
--- a/beacon-chain/db/kv/BUILD.bazel
+++ b/beacon-chain/db/kv/BUILD.bazel
@@ -33,7 +33,6 @@ go_library(
    visibility = ["//visibility:public"],
    deps = [
        "//beacon-chain/core/blocks:go_default_library",
-        "//beacon-chain/db/filesystem:go_default_library",
        "//beacon-chain/db/filters:go_default_library",
        "//beacon-chain/db/iface:go_default_library",
        "//beacon-chain/state:go_default_library",
@@ -99,13 +98,11 @@ go_test(
    data = glob(["testdata/**"]),
    embed = [":go_default_library"],
    deps = [
-        "//beacon-chain/db/filesystem:go_default_library",
        "//beacon-chain/db/filters:go_default_library",
        "//beacon-chain/db/iface:go_default_library",
        "//beacon-chain/state:go_default_library",
        "//beacon-chain/state/genesis:go_default_library",
        "//beacon-chain/state/state-native:go_default_library",
-        "//cmd/beacon-chain/flags:go_default_library",
        "//config/features:go_default_library",
        "//config/fieldparams:go_default_library",
        "//config/params:go_default_library",
@@ -123,7 +120,6 @@ go_test(
        "@com_github_golang_snappy//:go_default_library",
        "@com_github_pkg_errors//:go_default_library",
        "@com_github_sirupsen_logrus//:go_default_library",
-        "@com_github_urfave_cli_v2//:go_default_library",
        "@io_bazel_rules_go//go/tools/bazel:go_default_library",
        "@io_etcd_go_bbolt//:go_default_library",
        "@org_golang_google_protobuf//proto:go_default_library",
--- a/beacon-chain/db/kv/blob.go
+++ b/beacon-chain/db/kv/blob.go
@@ -3,11 +3,9 @@ package kv
 import (
 	"bytes"
 	"context"
-	"fmt"
 	"sort"

 	"github.com/pkg/errors"
-	"github.com/prysmaticlabs/prysm/v4/beacon-chain/db/filesystem"
 	fieldparams "github.com/prysmaticlabs/prysm/v4/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v4/config/params"
 	types "github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
@@ -64,7 +62,7 @@ func (s *Store) SaveBlobSidecar(ctx context.Context, scs []*ethpb.DeprecatedBlob
 	defer span.End()

 	first := scs[0]
-	newKey := blobSidecarKey(first)
+	newKey := s.blobSidecarKey(first)
 	prefix := newKey.BufferPrefix()
 	var prune []blobRotatingKey
 	return s.db.Update(func(tx *bolt.Tx) error {
@@ -236,7 +234,7 @@ func (s *Store) BlobSidecarsBySlot(ctx context.Context, slot types.Slot, indices
 	defer span.End()

 	var enc []byte
-	sk := slotKey(slot)
+	sk := s.slotKey(slot)
 	if err := s.db.View(func(tx *bolt.Tx) error {
 		c := tx.Bucket(blobsBucket).Cursor()
 		// Bucket size is bounded and bolt cursors are fast. Moreover, a thin caching layer can be added.
@@ -282,32 +280,37 @@ func (s *Store) DeleteBlobSidecars(ctx context.Context, beaconBlockRoot [32]byte

 // We define a blob sidecar key as: bytes(slot_to_rotating_buffer(blob.slot)) ++ bytes(blob.slot) ++ blob.block_root
 // where slot_to_rotating_buffer(slot) = slot % MAX_SLOTS_TO_PERSIST_BLOBS.
-func blobSidecarKey(blob *ethpb.DeprecatedBlobSidecar) blobRotatingKey {
-	key := slotKey(blob.Slot)
+func (s *Store) blobSidecarKey(blob *ethpb.DeprecatedBlobSidecar) blobRotatingKey {
+	key := s.slotKey(blob.Slot)
 	key = append(key, bytesutil.SlotToBytesBigEndian(blob.Slot)...)
 	key = append(key, blob.BlockRoot...)
 	return key
 }

-func slotKey(slot types.Slot) []byte {
-	slotsPerEpoch := params.BeaconConfig().SlotsPerEpoch
-	maxSlotsToPersistBlobs := types.Slot(filesystem.MaxEpochsToPersistBlobs.Mul(uint64(slotsPerEpoch)))
-	return bytesutil.SlotToBytesBigEndian(slot.ModSlot(maxSlotsToPersistBlobs))
+func (s *Store) slotKey(slot types.Slot) []byte {
+	return bytesutil.SlotToBytesBigEndian(slot.ModSlot(s.blobRetentionSlots()))
 }

-func checkEpochsForBlobSidecarsRequestBucket(db *bolt.DB) error {
+func (s *Store) blobRetentionSlots() types.Slot {
+	return types.Slot(s.blobRetentionEpochs.Mul(uint64(params.BeaconConfig().SlotsPerEpoch)))
+}
+
+var errBlobRetentionEpochMismatch = errors.New("epochs for blobs request value in DB does not match runtime config")
+
+func (s *Store) checkEpochsForBlobSidecarsRequestBucket(db *bolt.DB) error {
+	uRetentionEpochs := uint64(s.blobRetentionEpochs)
 	if err := db.Update(func(tx *bolt.Tx) error {
 		b := tx.Bucket(chainMetadataBucket)
 		v := b.Get(blobRetentionEpochsKey)
 		if v == nil {
-			if err := b.Put(blobRetentionEpochsKey, bytesutil.Uint64ToBytesBigEndian(uint64(filesystem.MaxEpochsToPersistBlobs))); err != nil {
+			if err := b.Put(blobRetentionEpochsKey, bytesutil.Uint64ToBytesBigEndian(uRetentionEpochs)); err != nil {
 				return err
 			}
 			return nil
 		}
 		e := bytesutil.BytesToUint64BigEndian(v)
-		if e != uint64(filesystem.MaxEpochsToPersistBlobs) {
-			return fmt.Errorf("epochs for blobs request value in DB %d does not match config value %d", e, filesystem.MaxEpochsToPersistBlobs)
+		if e != uRetentionEpochs {
+			return errors.Wrapf(errBlobRetentionEpochMismatch, "db=%d, config=%d", e, uRetentionEpochs)
 		}
 		return nil
 	}); err != nil {
--- a/beacon-chain/db/kv/blob_test.go
+++ b/beacon-chain/db/kv/blob_test.go
@@ -3,14 +3,10 @@ package kv
 import (
 	"context"
 	"crypto/rand"
-	"flag"
 	"fmt"
-	"strconv"
 	"testing"

 	"github.com/pkg/errors"
-	"github.com/prysmaticlabs/prysm/v4/beacon-chain/db/filesystem"
-	"github.com/prysmaticlabs/prysm/v4/cmd/beacon-chain/flags"
 	fieldparams "github.com/prysmaticlabs/prysm/v4/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v4/config/params"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
@@ -19,7 +15,6 @@ import (
 	ethpb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v4/testing/assertions"
 	"github.com/prysmaticlabs/prysm/v4/testing/require"
-	"github.com/urfave/cli/v2"
 	bolt "go.etcd.io/bbolt"
 )

@@ -494,7 +489,7 @@ func BenchmarkStore_BlobSidecarsByRoot(b *testing.B) {
 			scs := []*ethpb.DeprecatedBlobSidecar{
 				{BlockRoot: r, Slot: primitives.Slot(i)},
 			}
-			k := blobSidecarKey(scs[0])
+			k := s.blobSidecarKey(scs[0])
 			encodedBlobSidecar, err := encode(ctx, &ethpb.BlobSidecars{Sidecars: scs})
 			require.NoError(b, err)
 			require.NoError(b, bkt.Put(k, encodedBlobSidecar))
@@ -515,27 +510,23 @@ func BenchmarkStore_BlobSidecarsByRoot(b *testing.B) {
 }

 func Test_checkEpochsForBlobSidecarsRequestBucket(t *testing.T) {
-	dbStore := setupDB(t)
+	s := setupDB(t)

-	require.NoError(t, checkEpochsForBlobSidecarsRequestBucket(dbStore.db)) // First write
-	require.NoError(t, checkEpochsForBlobSidecarsRequestBucket(dbStore.db)) // First check
+	require.NoError(t, s.checkEpochsForBlobSidecarsRequestBucket(s.db)) // First write
+	require.NoError(t, s.checkEpochsForBlobSidecarsRequestBucket(s.db)) // First check

-	params.SetupTestConfigCleanup(t)
-	set := flag.NewFlagSet("test", 0)
-	set.Uint64(flags.BlobRetentionEpoch.Name, 0, "")
-	require.NoError(t, set.Set(flags.BlobRetentionEpoch.Name, strconv.FormatUint(42069, 10)))
-	cliCtx := cli.NewContext(&cli.App{}, set, nil)
-	require.NoError(t, filesystem.ConfigureBlobRetentionEpoch(cliCtx))
-	require.ErrorContains(t, "epochs for blobs request value in DB 4096 does not match config value 42069", checkEpochsForBlobSidecarsRequestBucket(dbStore.db))
+	s.blobRetentionEpochs += 1
+	require.ErrorIs(t, s.checkEpochsForBlobSidecarsRequestBucket(s.db), errBlobRetentionEpochMismatch)
 }

 func TestBlobRotatingKey(t *testing.T) {
-	k := blobSidecarKey(&ethpb.DeprecatedBlobSidecar{
+	s := setupDB(t)
+	k := s.blobSidecarKey(&ethpb.DeprecatedBlobSidecar{
 		Slot:      1,
 		BlockRoot: []byte{2},
 	})

 	require.Equal(t, types.Slot(1), k.Slot())
 	require.DeepEqual(t, []byte{2}, k.BlockRoot())
-	require.DeepEqual(t, slotKey(types.Slot(1)), k.BufferPrefix())
+	require.DeepEqual(t, s.slotKey(types.Slot(1)), k.BufferPrefix())
 }
--- a/beacon-chain/db/kv/kv.go
+++ b/beacon-chain/db/kv/kv.go
@@ -18,6 +18,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v4/config/features"
 	"github.com/prysmaticlabs/prysm/v4/config/params"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/blocks"
+	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
 	"github.com/prysmaticlabs/prysm/v4/io/file"
 	bolt "go.etcd.io/bbolt"
 )
@@ -90,6 +91,7 @@ type Store struct {
 	validatorEntryCache *ristretto.Cache
 	stateSummaryCache   *stateSummaryCache
 	ctx                 context.Context
+	blobRetentionEpochs primitives.Epoch
 }

 // StoreDatafilePath is the canonical construction of a full
@@ -133,10 +135,20 @@ var Buckets = [][]byte{
 	blobsBucket,
 }

+// KVStoreOption is a functional option that modifies a kv.Store.
+type KVStoreOption func(*Store)
+
+// WithBlobRetentionEpochs sets the variable configuring the blob retention window.
+func WithBlobRetentionEpochs(e primitives.Epoch) KVStoreOption {
+	return func(s *Store) {
+		s.blobRetentionEpochs = e
+	}
+}
+
 // NewKVStore initializes a new boltDB key-value store at the directory
 // path specified, creates the kv-buckets based on the schema, and stores
 // an open connection db object as a property of the Store struct.
-func NewKVStore(ctx context.Context, dirPath string) (*Store, error) {
+func NewKVStore(ctx context.Context, dirPath string, opts ...KVStoreOption) (*Store, error) {
 	hasDir, err := file.HasDir(dirPath)
 	if err != nil {
 		return nil, err
@@ -189,6 +201,9 @@ func NewKVStore(ctx context.Context, dirPath string) (*Store, error) {
 		stateSummaryCache:   newStateSummaryCache(),
 		ctx:                 ctx,
 	}
+	for _, o := range opts {
+		o(kv)
+	}
 	if err := kv.db.Update(func(tx *bolt.Tx) error {
 		return createBuckets(tx, Buckets...)
 	}); err != nil {
@@ -202,10 +217,14 @@ func NewKVStore(ctx context.Context, dirPath string) (*Store, error) {
 		return nil, err
 	}

-	if err := checkEpochsForBlobSidecarsRequestBucket(boltDB); err != nil {
+	if err := kv.checkEpochsForBlobSidecarsRequestBucket(boltDB); err != nil {
 		return nil, errors.Wrap(err, "failed to check epochs for blob sidecars request bucket")
 	}

+	// set a default so that tests don't break
+	if kv.blobRetentionEpochs == 0 {
+		kv.blobRetentionEpochs = params.BeaconNetworkConfig().MinEpochsForBlobsSidecarsRequest
+	}
 	return kv, nil
 }

--- a/beacon-chain/db/kv/kv_test.go
+++ b/beacon-chain/db/kv/kv_test.go
@@ -6,6 +6,7 @@ import (
 	"testing"

 	"github.com/prysmaticlabs/prysm/v4/config/features"
+	"github.com/prysmaticlabs/prysm/v4/config/params"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/blocks"
 	ethpb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v4/testing/require"
@@ -15,7 +16,8 @@ import (

 // setupDB instantiates and returns a Store instance.
 func setupDB(t testing.TB) *Store {
-	db, err := NewKVStore(context.Background(), t.TempDir())
+	opt := WithBlobRetentionEpochs(params.BeaconNetworkConfig().MinEpochsForBlobsSidecarsRequest)
+	db, err := NewKVStore(context.Background(), t.TempDir(), opt)
 	require.NoError(t, err, "Failed to instantiate DB")
 	t.Cleanup(func() {
 		require.NoError(t, db.Close(), "Failed to close database")
--- a/beacon-chain/node/node.go
+++ b/beacon-chain/node/node.go
@@ -113,6 +113,7 @@ type BeaconNode struct {
 	clockWaiter             startup.ClockWaiter
 	initialSyncComplete     chan struct{}
 	BlobStorage             *filesystem.BlobStorage
+	blobRetentionEpochs     primitives.Epoch
 }

 // New creates a new node instance, sets up configuration options, and registers
@@ -155,9 +156,6 @@ func New(cliCtx *cli.Context, cancel context.CancelFunc, opts ...Option) (*Beaco
 	if err := configureExecutionSetting(cliCtx); err != nil {
 		return nil, err
 	}
-	if err := filesystem.ConfigureBlobRetentionEpoch(cliCtx); err != nil {
-		return nil, err
-	}
 	configureFastSSZHashingAlgorithm()

 	// Initializes any forks here.
@@ -380,7 +378,7 @@ func (b *BeaconNode) startDB(cliCtx *cli.Context, depositAddress string) error {

 	log.WithField("database-path", dbPath).Info("Checking DB")

-	d, err := db.NewDB(b.ctx, dbPath)
+	d, err := kv.NewKVStore(b.ctx, dbPath, kv.WithBlobRetentionEpochs(b.blobRetentionEpochs))
 	if err != nil {
 		return err
 	}
@@ -402,7 +400,8 @@ func (b *BeaconNode) startDB(cliCtx *cli.Context, depositAddress string) error {
 		if err := d.ClearDB(); err != nil {
 			return errors.Wrap(err, "could not clear database")
 		}
-		d, err = db.NewDB(b.ctx, dbPath)
+
+		d, err = kv.NewKVStore(b.ctx, dbPath, kv.WithBlobRetentionEpochs(b.blobRetentionEpochs))
 		if err != nil {
 			return errors.Wrap(err, "could not create new database")
 		}
--- a/beacon-chain/node/options.go
+++ b/beacon-chain/node/options.go
@@ -5,6 +5,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/builder"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/db/filesystem"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/execution"
+	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
 )

 // Option for beacon node configuration.
@@ -41,3 +42,11 @@ func WithBlobStorage(bs *filesystem.BlobStorage) Option {
 		return nil
 	}
 }
+
+// WithBlobRetentionEpochs sets the blobRetentionEpochs value, used in kv store initialization.
+func WithBlobRetentionEpochs(e primitives.Epoch) Option {
+	return func(bn *BeaconNode) error {
+		bn.blobRetentionEpochs = e
+		return nil
+	}
+}
--- a/cmd/beacon-chain/flags/base.go
+++ b/cmd/beacon-chain/flags/base.go
@@ -265,10 +265,4 @@ var (
 		Usage: "Directory for the slasher database",
 		Value: cmd.DefaultDataDir(),
 	}
-	BlobRetentionEpoch = &cli.Uint64Flag{
-		Name:    "blob-retention-epochs",
-		Usage:   "Override the default blob retention period (measured in epochs). The node will exit with an error at startup if the value is less than the default of 4096 epochs.",
-		Value:   uint64(params.BeaconNetworkConfig().MinEpochsForBlobsSidecarsRequest),
-		Aliases: []string{"extend-blob-retention-epoch"},
-	}
 )
--- a/cmd/beacon-chain/main.go
+++ b/cmd/beacon-chain/main.go
@@ -78,7 +78,6 @@ var appFlags = []cli.Flag{
 	flags.MaxBuilderConsecutiveMissedSlots,
 	flags.EngineEndpointTimeoutSeconds,
 	flags.LocalBlockValueBoost,
-	flags.BlobRetentionEpoch,
 	cmd.BackupWebhookOutputDir,
 	cmd.MinimalConfigFlag,
 	cmd.E2EConfigFlag,
@@ -137,6 +136,8 @@ var appFlags = []cli.Flag{
 	genesis.BeaconAPIURL,
 	flags.SlasherDirFlag,
 	flags.JwtId,
+	storage.BlobStoragePathFlag,
+	storage.BlobRetentionEpochFlag,
 }

 func init() {
@@ -282,7 +283,7 @@ func startNode(ctx *cli.Context, cancel context.CancelFunc) error {
 		node.WithBuilderFlagOptions(builderFlagOpts),
 	}

-	optFuncs := []func(*cli.Context) (node.Option, error){
+	optFuncs := []func(*cli.Context) ([]node.Option, error){
 		genesis.BeaconNodeOptions,
 		checkpoint.BeaconNodeOptions,
 		storage.BeaconNodeOptions,
@@ -293,7 +294,7 @@ func startNode(ctx *cli.Context, cancel context.CancelFunc) error {
 			return err
 		}
 		if ofo != nil {
-			opts = append(opts, ofo)
+			opts = append(opts, ofo...)
 		}
 	}

--- a/cmd/beacon-chain/storage/BUILD.bazel
+++ b/cmd/beacon-chain/storage/BUILD.bazel
@@ -9,6 +9,9 @@ go_library(
        "//beacon-chain/db/filesystem:go_default_library",
        "//beacon-chain/node:go_default_library",
        "//cmd:go_default_library",
+        "//config/params:go_default_library",
+        "//consensus-types/primitives:go_default_library",
+        "@com_github_pkg_errors//:go_default_library",
        "@com_github_urfave_cli_v2//:go_default_library",
    ],
 )
@@ -19,7 +22,10 @@ go_test(
    embed = [":go_default_library"],
    deps = [
        "//cmd:go_default_library",
+        "//config/params:go_default_library",
+        "//consensus-types/primitives:go_default_library",
        "//testing/assert:go_default_library",
+        "//testing/require:go_default_library",
        "@com_github_urfave_cli_v2//:go_default_library",
    ],
 )
--- a/cmd/beacon-chain/storage/options.go
+++ b/cmd/beacon-chain/storage/options.go
@@ -3,38 +3,70 @@ package storage
 import (
 	"path"

+	"github.com/pkg/errors"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/db/filesystem"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/node"
 	"github.com/prysmaticlabs/prysm/v4/cmd"
+	"github.com/prysmaticlabs/prysm/v4/config/params"
+	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
 	"github.com/urfave/cli/v2"
 )

 var (
-	// BlobStoragePath defines a flag to start the beacon chain from a give genesis state file.
-	BlobStoragePath = &cli.PathFlag{
+	// 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.",
 	}
+	BlobRetentionEpochFlag = &cli.Uint64Flag{
+		Name:    "blob-retention-epochs",
+		Usage:   "Override the default blob retention period (measured in epochs). The node will exit with an error at startup if the value is less than the default of 4096 epochs.",
+		Value:   uint64(params.BeaconNetworkConfig().MinEpochsForBlobsSidecarsRequest),
+		Aliases: []string{"extend-blob-retention-epoch"},
+	}
 )

 // BeaconNodeOptions sets configuration values on the node.BeaconNode value at node startup.
 // Note: we can't get the right context from cli.Context, because the beacon node setup code uses this context to
 // 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) {
-	blobsPath := blobStoragePath(c)
-	bs, err := filesystem.NewBlobStorage(blobsPath)
+func BeaconNodeOptions(c *cli.Context) ([]node.Option, error) {
+	e, err := blobRetentionEpoch(c)
 	if err != nil {
 		return nil, err
 	}
-	return node.WithBlobStorage(bs), nil
+	bs, err := filesystem.NewBlobStorage(blobStoragePath(c), filesystem.WithBlobRetentionEpochs(e))
+	if err != nil {
+		return nil, err
+	}
+	return []node.Option{node.WithBlobStorage(bs), node.WithBlobRetentionEpochs(e)}, nil
 }

 func blobStoragePath(c *cli.Context) string {
-	blobsPath := c.Path(BlobStoragePath.Name)
+	blobsPath := c.Path(BlobStoragePathFlag.Name)
 	if blobsPath == "" {
 		// append a "blobs" subdir to the end of the data dir path
 		blobsPath = path.Join(c.String(cmd.DataDirFlag.Name), "blobs")
 	}
 	return blobsPath
 }
+
+var errInvalidBlobRetentionEpochs = errors.New("value is smaller than spec minimum")
+
+// blobRetentionEpoch returns the spec deffault MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUEST
+// or a user-specified flag overriding this value. If a user-specified override is
+// smaller than the spec default, an error will be returned.
+func blobRetentionEpoch(cliCtx *cli.Context) (primitives.Epoch, error) {
+	spec := params.BeaconNetworkConfig().MinEpochsForBlobsSidecarsRequest
+	if !cliCtx.IsSet(BlobRetentionEpochFlag.Name) {
+		return spec, nil
+	}
+
+	re := primitives.Epoch(cliCtx.Uint64(BlobRetentionEpochFlag.Name))
+	// Validate the epoch value against the spec default.
+	if re < params.BeaconNetworkConfig().MinEpochsForBlobsSidecarsRequest {
+		return spec, errors.Wrapf(errInvalidBlobRetentionEpochs, "%s=%d, spec=%d", BlobRetentionEpochFlag.Name, re, spec)
+	}
+
+	return re, nil
+}
--- a/cmd/beacon-chain/storage/options_test.go
+++ b/cmd/beacon-chain/storage/options_test.go
@@ -2,10 +2,14 @@ package storage

 import (
 	"flag"
+	"fmt"
 	"testing"

 	"github.com/prysmaticlabs/prysm/v4/cmd"
+	"github.com/prysmaticlabs/prysm/v4/config/params"
+	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
 	"github.com/prysmaticlabs/prysm/v4/testing/assert"
+	"github.com/prysmaticlabs/prysm/v4/testing/require"
 	"github.com/urfave/cli/v2"
 )

@@ -22,9 +26,38 @@ func TestBlobStoragePath_NoFlagSpecified(t *testing.T) {
 func TestBlobStoragePath_FlagSpecified(t *testing.T) {
 	app := cli.App{}
 	set := flag.NewFlagSet("test", 0)
-	set.String(BlobStoragePath.Name, "/blah/blah", BlobStoragePath.Usage)
+	set.String(BlobStoragePathFlag.Name, "/blah/blah", BlobStoragePathFlag.Usage)
 	cliCtx := cli.NewContext(&app, set, nil)
 	storagePath := blobStoragePath(cliCtx)

 	assert.Equal(t, "/blah/blah", storagePath)
 }
+
+func TestConfigureBlobRetentionEpoch(t *testing.T) {
+	params.SetupTestConfigCleanup(t)
+	specMinEpochs := params.BeaconNetworkConfig().MinEpochsForBlobsSidecarsRequest
+	app := cli.App{}
+	set := flag.NewFlagSet("test", 0)
+	cliCtx := cli.NewContext(&app, set, nil)
+
+	// Test case: Spec default.
+	epochs, err := blobRetentionEpoch(cliCtx)
+	require.NoError(t, err)
+	require.Equal(t, specMinEpochs, epochs)
+
+	// manually define the flag in the set, so the following code can use set.Set
+	set.Uint64(BlobRetentionEpochFlag.Name, 0, "")
+
+	// Test case: Input epoch is greater than or equal to spec value.
+	expectedChange := specMinEpochs + 1
+	require.NoError(t, set.Set(BlobRetentionEpochFlag.Name, fmt.Sprintf("%d", expectedChange)))
+	epochs, err = blobRetentionEpoch(cliCtx)
+	require.NoError(t, err)
+	require.Equal(t, primitives.Epoch(expectedChange), epochs)
+
+	// Test case: Input epoch is less than spec value.
+	expectedChange = specMinEpochs - 1
+	require.NoError(t, set.Set(BlobRetentionEpochFlag.Name, fmt.Sprintf("%d", expectedChange)))
+	_, err = blobRetentionEpoch(cliCtx)
+	require.ErrorIs(t, err, errInvalidBlobRetentionEpochs)
+}
--- a/cmd/beacon-chain/sync/checkpoint/options.go
+++ b/cmd/beacon-chain/sync/checkpoint/options.go
@@ -33,19 +33,20 @@ var (
 // BeaconNodeOptions is responsible for determining if the checkpoint sync options have been used, and if so,
 // reading the block and state ssz-serialized values from the filesystem locations specified and preparing a
 // checkpoint.Initializer, which uses the provided io.ReadClosers to initialize the beacon node database.
-func BeaconNodeOptions(c *cli.Context) (node.Option, error) {
+func BeaconNodeOptions(c *cli.Context) ([]node.Option, error) {
 	blockPath := c.Path(BlockPath.Name)
 	statePath := c.Path(StatePath.Name)
 	remoteURL := c.String(RemoteURL.Name)
 	if remoteURL != "" {
-		return func(node *node.BeaconNode) error {
+		opt := func(node *node.BeaconNode) error {
 			var err error
 			node.CheckpointInitializer, err = checkpoint.NewAPIInitializer(remoteURL)
 			if err != nil {
 				return errors.Wrap(err, "error while constructing beacon node api client for checkpoint sync")
 			}
 			return nil
-		}, nil
+		}
+		return []node.Option{opt}, nil
 	}

 	if blockPath == "" && statePath == "" {
@@ -58,11 +59,12 @@ func BeaconNodeOptions(c *cli.Context) (node.Option, error) {
 		return nil, fmt.Errorf("--checkpoint-state specified, but not --checkpoint-block. both are required")
 	}

-	return func(node *node.BeaconNode) (err error) {
+	opt := func(node *node.BeaconNode) (err error) {
 		node.CheckpointInitializer, err = checkpoint.NewFileInitializer(blockPath, statePath)
 		if err != nil {
 			return errors.Wrap(err, "error preparing to initialize checkpoint from local ssz files")
 		}
 		return nil
-	}, nil
+	}
+	return []node.Option{opt}, nil
 }
--- a/cmd/beacon-chain/sync/genesis/options.go
+++ b/cmd/beacon-chain/sync/genesis/options.go
@@ -27,7 +27,7 @@ var (
 // BeaconNodeOptions is responsible for determining if the checkpoint sync options have been used, and if so,
 // reading the block and state ssz-serialized values from the filesystem locations specified and preparing a
 // checkpoint.Initializer, which uses the provided io.ReadClosers to initialize the beacon node database.
-func BeaconNodeOptions(c *cli.Context) (node.Option, error) {
+func BeaconNodeOptions(c *cli.Context) ([]node.Option, error) {
 	statePath := c.Path(StatePath.Name)
 	remoteURL := c.String(BeaconAPIURL.Name)
 	if remoteURL == "" && c.String(checkpoint.RemoteURL.Name) != "" {
@@ -35,25 +35,27 @@ func BeaconNodeOptions(c *cli.Context) (node.Option, error) {
 		remoteURL = c.String(checkpoint.RemoteURL.Name)
 	}
 	if remoteURL != "" {
-		return func(node *node.BeaconNode) error {
+		opt := func(node *node.BeaconNode) error {
 			var err error
 			node.GenesisInitializer, err = genesis.NewAPIInitializer(remoteURL)
 			if err != nil {
 				return errors.Wrap(err, "error constructing beacon node api client for genesis state init")
 			}
 			return nil
-		}, nil
+		}
+		return []node.Option{opt}, nil
 	}

 	if statePath == "" {
 		return nil, nil
 	}

-	return func(node *node.BeaconNode) (err error) {
+	opt := func(node *node.BeaconNode) (err error) {
 		node.GenesisInitializer, err = genesis.NewFileInitializer(statePath)
 		if err != nil {
 			return errors.Wrap(err, "error preparing to initialize genesis db state from local ssz files")
 		}
 		return nil
-	}, nil
+	}
+	return []node.Option{opt}, nil
 }
--- a/cmd/beacon-chain/usage.go
+++ b/cmd/beacon-chain/usage.go
@@ -7,6 +7,7 @@ import (

 	"github.com/prysmaticlabs/prysm/v4/cmd"
 	"github.com/prysmaticlabs/prysm/v4/cmd/beacon-chain/flags"
+	"github.com/prysmaticlabs/prysm/v4/cmd/beacon-chain/storage"
 	"github.com/prysmaticlabs/prysm/v4/cmd/beacon-chain/sync/checkpoint"
 	"github.com/prysmaticlabs/prysm/v4/cmd/beacon-chain/sync/genesis"
 	"github.com/prysmaticlabs/prysm/v4/config/features"
@@ -128,13 +129,14 @@ var appHelpFlagGroups = []flagGroup{
 			flags.EngineEndpointTimeoutSeconds,
 			flags.SlasherDirFlag,
 			flags.LocalBlockValueBoost,
-			flags.BlobRetentionEpoch,
 			flags.JwtId,
 			checkpoint.BlockPath,
 			checkpoint.StatePath,
 			checkpoint.RemoteURL,
 			genesis.StatePath,
 			genesis.BeaconAPIURL,
+			storage.BlobStoragePathFlag,
+			storage.BlobRetentionEpochFlag,
 		},
 	},
 	{
--- a/tools/blocktree/BUILD.bazel
+++ b/tools/blocktree/BUILD.bazel
@@ -7,8 +7,8 @@ go_library(
    importpath = "github.com/prysmaticlabs/prysm/v4/tools/blocktree",
    visibility = ["//visibility:private"],
    deps = [
-        "//beacon-chain/db:go_default_library",
        "//beacon-chain/db/filters:go_default_library",
+        "//beacon-chain/db/kv:go_default_library",
        "//consensus-types/primitives:go_default_library",
        "@com_github_emicklei_dot//:go_default_library",
    ],
--- a/tools/blocktree/main.go
+++ b/tools/blocktree/main.go
@@ -16,8 +16,8 @@ import (
 	"strconv"

 	"github.com/emicklei/dot"
-	"github.com/prysmaticlabs/prysm/v4/beacon-chain/db"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/db/filters"
+	"github.com/prysmaticlabs/prysm/v4/beacon-chain/db/kv"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
 )

@@ -37,7 +37,7 @@ type node struct {

 func main() {
 	flag.Parse()
-	database, err := db.NewDB(context.Background(), *datadir)
+	database, err := kv.NewKVStore(context.Background(), *datadir)
 	if err != nil {
 		panic(err)
 	}
--- a/tools/extractor/BUILD.bazel
+++ b/tools/extractor/BUILD.bazel
@@ -8,7 +8,7 @@ go_library(
    visibility = ["//visibility:public"],
    deps = [
        "//beacon-chain/core/transition/interop:go_default_library",
-        "//beacon-chain/db:go_default_library",
+        "//beacon-chain/db/kv:go_default_library",
        "//config/features:go_default_library",
        "//consensus-types/primitives:go_default_library",
    ],
--- a/tools/extractor/main.go
+++ b/tools/extractor/main.go
@@ -6,7 +6,7 @@ import (
 	"fmt"

 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/core/transition/interop"
-	"github.com/prysmaticlabs/prysm/v4/beacon-chain/db"
+	"github.com/prysmaticlabs/prysm/v4/beacon-chain/db/kv"
 	"github.com/prysmaticlabs/prysm/v4/config/features"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
 )
@@ -23,7 +23,7 @@ func main() {
 	defer resetCfg()
 	flag.Parse()
 	fmt.Println("Starting process...")
-	d, err := db.NewDB(context.Background(), *datadir)
+	d, err := kv.NewKVStore(context.Background(), *datadir)
 	if err != nil {
 		panic(err)
 	}
--- a/tools/interop/export-genesis/BUILD.bazel
+++ b/tools/interop/export-genesis/BUILD.bazel
@@ -7,7 +7,7 @@ go_library(
    importpath = "github.com/prysmaticlabs/prysm/v4/tools/interop/export-genesis",
    visibility = ["//visibility:private"],
    deps = [
-        "//beacon-chain/db:go_default_library",
+        "//beacon-chain/db/kv:go_default_library",
        "//io/file:go_default_library",
    ],
 )
--- a/tools/interop/export-genesis/main.go
+++ b/tools/interop/export-genesis/main.go
@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"os"

-	"github.com/prysmaticlabs/prysm/v4/beacon-chain/db"
+	"github.com/prysmaticlabs/prysm/v4/beacon-chain/db/kv"
 	"github.com/prysmaticlabs/prysm/v4/io/file"
 )

@@ -21,7 +21,7 @@ func main() {

 	fmt.Printf("Reading db at %s and writing ssz output to %s.\n", os.Args[1], os.Args[2])

-	d, err := db.NewDB(context.Background(), os.Args[1])
+	d, err := kv.NewKVStore(context.Background(), os.Args[1])
 	if err != nil {
 		panic(err)
 	}