Organize blobs on disk by epoch (#14023)

* organize blob directories by period and epoch

* changelog

* remove Indices and replace with Summary

* old PR feedback

* log to advise about the speed of blob migration

* rename level->layer (hoping term is more clear)

* assert path in tests for increased legibility

* lint

* lint

* remove test covering a newly impossible error

* improve feedback from flag validation failure

* Try to clean dangling dirs epoch->flat migration

* lint

* Preston feedback

* try all layouts and short-circuit if base not found

---------

Co-authored-by: Kasey Kirkham <kasey@users.noreply.github.com>

beacon-chain/execution/BUILD.bazel

@@ -103,6 +103,7 @@ go_test(
         "//beacon-chain/core/helpers:go_default_library",
         "//beacon-chain/core/signing:go_default_library",
         "//beacon-chain/db:go_default_library",
+        "//beacon-chain/db/filesystem:go_default_library",
         "//beacon-chain/db/testing:go_default_library",
         "//beacon-chain/execution/testing:go_default_library",
         "//beacon-chain/execution/types:go_default_library",

beacon-chain/execution/engine_client.go

@@ -105,7 +105,7 @@ type Reconstructor interface {
 	ReconstructFullBellatrixBlockBatch(
 		ctx context.Context, blindedBlocks []interfaces.ReadOnlySignedBeaconBlock,
 	) ([]interfaces.SignedBeaconBlock, error)
-	ReconstructBlobSidecars(ctx context.Context, block interfaces.ReadOnlySignedBeaconBlock, blockRoot [32]byte, indices []bool) ([]blocks.VerifiedROBlob, error)
+	ReconstructBlobSidecars(ctx context.Context, block interfaces.ReadOnlySignedBeaconBlock, blockRoot [32]byte, hi func(uint64) bool) ([]blocks.VerifiedROBlob, error)
 }
 
 // EngineCaller defines a client that can interact with an Ethereum
@@ -531,31 +531,20 @@ func (s *Service) ReconstructFullBellatrixBlockBatch(
 // It retrieves the KZG commitments from the block body, fetches the associated blobs and proofs,
 // and constructs the corresponding verified read-only blob sidecars.
 //
-// The 'exists' argument is a boolean list (must be the same length as body.BlobKzgCommitments), where each element corresponds to whether a
-// particular blob sidecar already exists. If exists[i] is true, the blob for the i-th KZG commitment
-// has already been retrieved and does not need to be fetched again from the execution layer (EL).
-//
-// For example:
-//   - len(block.Body().BlobKzgCommitments()) == 6
-//   - If exists = [true, false, true, false, true, false], the function will fetch the blobs
-//     associated with indices 1, 3, and 5 (since those are marked as non-existent).
-//   - If exists = [false ... x 6], the function will attempt to fetch all blobs.
-//
-// Only the blobs that do not already exist (where exists[i] is false) are fetched using the KZG commitments from block body.
-func (s *Service) ReconstructBlobSidecars(ctx context.Context, block interfaces.ReadOnlySignedBeaconBlock, blockRoot [32]byte, exists []bool) ([]blocks.VerifiedROBlob, error) {
+// The 'hasIndex' argument is a function returns true if the given uint64 blob index already exists on disc.
+// Only the blobs that do not already exist (where hasIndex(i) is false)
+// will be fetched from the execution engine using the KZG commitments from block body.
+func (s *Service) ReconstructBlobSidecars(ctx context.Context, block interfaces.ReadOnlySignedBeaconBlock, blockRoot [32]byte, hasIndex func(uint64) bool) ([]blocks.VerifiedROBlob, error) {
 	blockBody := block.Block().Body()
 	kzgCommitments, err := blockBody.BlobKzgCommitments()
 	if err != nil {
 		return nil, errors.Wrap(err, "could not get blob KZG commitments")
 	}
-	if len(kzgCommitments) > len(exists) {
-		return nil, fmt.Errorf("length of KZG commitments (%d) is greater than length of exists (%d)", len(kzgCommitments), len(exists))
-	}
 
 	// Collect KZG hashes for non-existing blobs
 	var kzgHashes []common.Hash
 	for i, commitment := range kzgCommitments {
-		if !exists[i] {
+		if !hasIndex(uint64(i)) {
 			kzgHashes = append(kzgHashes, primitives.ConvertKzgCommitmentToVersionedHash(commitment))
 		}
 	}
@@ -580,7 +569,7 @@ func (s *Service) ReconstructBlobSidecars(ctx context.Context, block interfaces.
 	// Reconstruct verified blob sidecars
 	var verifiedBlobs []blocks.VerifiedROBlob
 	for i, blobIndex := 0, 0; i < len(kzgCommitments); i++ {
-		if exists[i] {
+		if hasIndex(uint64(i)) {
 			continue
 		}
 

beacon-chain/execution/engine_client_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/holiman/uint256"
 	"github.com/pkg/errors"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/db/filesystem"
 	mocks "github.com/prysmaticlabs/prysm/v5/beacon-chain/execution/testing"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/verification"
 	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
@@ -2395,6 +2396,12 @@ func Test_ExchangeCapabilities(t *testing.T) {
 	})
 }
 
+func mockSummary(t *testing.T, exists []bool) func(uint64) bool {
+	hi, err := filesystem.NewBlobStorageSummary(params.BeaconConfig().DenebForkEpoch, exists)
+	require.NoError(t, err)
+	return hi.HasIndex
+}
+
 func TestReconstructBlobSidecars(t *testing.T) {
 	client := &Service{capabilityCache: &capabilityCache{}}
 	b := util.NewBeaconBlockDeneb()
@@ -2408,15 +2415,15 @@ func TestReconstructBlobSidecars(t *testing.T) {
 
 	ctx := context.Background()
 	t.Run("all seen", func(t *testing.T) {
-		exists := []bool{true, true, true, true, true, true}
-		verifiedBlobs, err := client.ReconstructBlobSidecars(ctx, sb, r, exists)
+		hi := mockSummary(t, []bool{true, true, true, true, true, true})
+		verifiedBlobs, err := client.ReconstructBlobSidecars(ctx, sb, r, hi)
 		require.NoError(t, err)
 		require.Equal(t, 0, len(verifiedBlobs))
 	})
 
 	t.Run("get-blobs end point is not supported", func(t *testing.T) {
-		exists := []bool{true, true, true, true, true, false}
-		verifiedBlobs, err := client.ReconstructBlobSidecars(ctx, sb, r, exists)
+		hi := mockSummary(t, []bool{true, true, true, true, true, false})
+		verifiedBlobs, err := client.ReconstructBlobSidecars(ctx, sb, r, hi)
 		require.NoError(t, err)
 		require.Equal(t, 0, len(verifiedBlobs))
 	})
@@ -2430,8 +2437,8 @@ func TestReconstructBlobSidecars(t *testing.T) {
 		rpcClient, client := setupRpcClient(t, srv.URL, client)
 		defer rpcClient.Close()
 
-		exists := [6]bool{}
-		verifiedBlobs, err := client.ReconstructBlobSidecars(ctx, sb, r, exists[:])
+		hi := mockSummary(t, make([]bool, 6))
+		verifiedBlobs, err := client.ReconstructBlobSidecars(ctx, sb, r, hi)
 		require.NoError(t, err)
 		require.Equal(t, 6, len(verifiedBlobs))
 	})
@@ -2443,23 +2450,11 @@ func TestReconstructBlobSidecars(t *testing.T) {
 		rpcClient, client := setupRpcClient(t, srv.URL, client)
 		defer rpcClient.Close()
 
-		exists := []bool{true, false, true, false, true, false}
-		verifiedBlobs, err := client.ReconstructBlobSidecars(ctx, sb, r, exists)
+		hi := mockSummary(t, []bool{true, false, true, false, true, false})
+		verifiedBlobs, err := client.ReconstructBlobSidecars(ctx, sb, r, hi)
 		require.NoError(t, err)
 		require.Equal(t, 3, len(verifiedBlobs))
 	})
-
-	t.Run("kzg is longer than exist", func(t *testing.T) {
-		srv := createBlobServer(t, 3)
-		defer srv.Close()
-
-		rpcClient, client := setupRpcClient(t, srv.URL, client)
-		defer rpcClient.Close()
-
-		exists := []bool{true, false, true, false, true}
-		_, err := client.ReconstructBlobSidecars(ctx, sb, r, exists)
-		require.ErrorContains(t, "length of KZG commitments (6) is greater than length of exists (5)", err)
-	})
 }
 
 func createRandomKzgCommitments(t *testing.T, num int) [][]byte {

beacon-chain/execution/testing/mock_engine_client.go

@@ -109,7 +109,7 @@ func (e *EngineClient) ReconstructFullBellatrixBlockBatch(
 }
 
 // ReconstructBlobSidecars is a mock implementation of the ReconstructBlobSidecars method.
-func (e *EngineClient) ReconstructBlobSidecars(context.Context, interfaces.ReadOnlySignedBeaconBlock, [32]byte, []bool) ([]blocks.VerifiedROBlob, error) {
+func (e *EngineClient) ReconstructBlobSidecars(context.Context, interfaces.ReadOnlySignedBeaconBlock, [32]byte, func(uint64) bool) ([]blocks.VerifiedROBlob, error) {
 	return e.BlobSidecars, e.ErrorBlobSidecars
 }