feat: migrate to beacon blobs endpoint with valid KZG test data

Implements support for the new /eth/v1/beacon/blobs/{block_id} endpoint
with automatic fallback to the legacy blob_sidecars endpoint.

Key changes:
- Archiver: Prefer blobs endpoint, compute KZG commitments/proofs from
  blob data, fall back to blob_sidecars on failure
- API: Add /eth/v1/beacon/blobs/{id} endpoint with versioned_hashes
  filtering support using kzg4844.CalcBlobHashV1
- Test data: Generate valid BLS12-381 field elements by reducing random
  data modulo the field modulus to ensure KZG operations succeed
- Test helpers: Consolidate blob generation into NewBlobSidecars with
  proper header support for consistent test comparisons

All tests now use the blobs endpoint (except one that explicitly tests
the fallback behavior). The archiver stores complete blob sidecars with
derived KZG commitments and proofs, maintaining backward compatibility
with existing storage format.

Technical notes:
- Each 32-byte chunk in blob data must be < BLS12-381 field modulus
- KZG commitments/proofs are computed using gokzg4844 library
- Versioned hashes use ethereum/go-ethereum's kzg4844.CalcBlobHashV1
- Test data includes valid headers to match reconstructed sidecars

Author: niran

api/service/api.go

@@ -2,6 +2,7 @@ package service
 
 import (
 	"context"
+	"crypto/sha256"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -13,13 +14,15 @@ import (
 
 	client "github.com/attestantio/go-eth2-client"
 	"github.com/attestantio/go-eth2-client/api"
+	v1 "github.com/attestantio/go-eth2-client/api/v1"
 	"github.com/attestantio/go-eth2-client/spec/deneb"
 	m "github.com/base/blob-archiver/api/metrics"
 	"github.com/base/blob-archiver/api/version"
 	"github.com/base/blob-archiver/common/storage"
 	opmetrics "github.com/ethereum-optimism/optimism/op-service/metrics"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/ethereum/go-ethereum/crypto/kzg4844"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/go-chi/chi/v5"
 	"github.com/go-chi/chi/v5/middleware"
@@ -107,6 +110,7 @@ func NewAPI(dataStoreClient storage.DataStoreReader, beaconClient client.BeaconB
 	})
 
 	r.Get("/eth/v1/beacon/blob_sidecars/{id}", result.blobSidecarHandler)
+	r.Get("/eth/v1/beacon/blobs/{id}", result.blobsHandler)
 	r.Get("/eth/v1/node/version", result.versionHandler)
 
 	return result
@@ -264,3 +268,112 @@ func filterBlobs(blobs []*deneb.BlobSidecar, _indices []string) ([]*deneb.BlobSi
 
 	return filteredBlobs, nil
 }
+
+// filterBlobsByVersionedHashes filters sidecars by versioned hashes query parameter.
+// Returns the filtered sidecars in the order they were requested, or all sidecars if no hashes provided.
+func filterBlobsByVersionedHashes(sidecars []*deneb.BlobSidecar, _versionedHashes []string) ([]*deneb.BlobSidecar, *httpError) {
+	var versionedHashes []string
+	if len(_versionedHashes) == 0 {
+		return sidecars, nil
+	} else if len(_versionedHashes) == 1 {
+		versionedHashes = strings.Split(_versionedHashes[0], ",")
+	} else {
+		versionedHashes = _versionedHashes
+	}
+
+	// Build map of commitment hash -> sidecar for quick lookup
+	// CalcBlobHashV1 requires a sha256 hasher instance
+	hasher := sha256.New()
+	hashToSidecar := make(map[[32]byte]*deneb.BlobSidecar)
+	for _, sidecar := range sidecars {
+		hasher.Reset()
+		commitment := kzg4844.Commitment(sidecar.KZGCommitment)
+		vh := kzg4844.CalcBlobHashV1(hasher, &commitment)
+		hashToSidecar[vh] = sidecar
+	}
+
+	// Return sidecars in the order of requested hashes
+	filteredBlobs := make([]*deneb.BlobSidecar, 0, len(versionedHashes))
+	for _, hashStr := range versionedHashes {
+		hash := common.HexToHash(hashStr)
+		var versionedHash [32]byte
+		copy(versionedHash[:], hash[:])
+
+		if sidecar, ok := hashToSidecar[versionedHash]; ok {
+			filteredBlobs = append(filteredBlobs, sidecar)
+		}
+	}
+
+	return filteredBlobs, nil
+}
+
+// sidecarsToBlobs converts blob sidecars to a Blobs response by extracting only the blob data
+func sidecarsToBlobs(sidecars []*deneb.BlobSidecar) v1.Blobs {
+	blobs := make(v1.Blobs, len(sidecars))
+	for i, sidecar := range sidecars {
+		blobs[i] = &sidecar.Blob
+	}
+	return blobs
+}
+
+// blobsHandler implements the /eth/v1/beacon/blobs/{id} endpoint, using the underlying DataStoreReader
+// to fetch blobs instead of the beacon node. This endpoint serves blobs without KZG proofs.
+// Filtering by versioned_hashes query parameter is supported (per EIP-4844).
+func (a *API) blobsHandler(w http.ResponseWriter, r *http.Request) {
+	param := chi.URLParam(r, "id")
+	beaconBlockHash, err := a.toBeaconBlockHash(param)
+	if err != nil {
+		err.write(w)
+		return
+	}
+
+	result, storageErr := a.dataStoreClient.ReadBlob(r.Context(), beaconBlockHash)
+	if storageErr != nil {
+		if errors.Is(storageErr, storage.ErrNotFound) {
+			errUnknownBlock.write(w)
+		} else {
+			a.logger.Info("unexpected error fetching blobs", "err", storageErr, "beaconBlockHash", beaconBlockHash.String(), "param", param)
+			errServerError.write(w)
+		}
+		return
+	}
+
+	sidecars := result.BlobSidecars.Data
+
+	// Filter by versioned_hashes query parameter (not indices)
+	filteredSidecars, err := filterBlobsByVersionedHashes(sidecars, r.URL.Query()["versioned_hashes"])
+	if err != nil {
+		err.write(w)
+		return
+	}
+
+	// Convert sidecars to blobs
+	blobs := sidecarsToBlobs(filteredSidecars)
+	responseType := r.Header.Get("Accept")
+
+	if responseType == sszAcceptType {
+		w.Header().Set("Content-Type", sszAcceptType)
+		res, err := blobs.MarshalSSZ()
+		if err != nil {
+			a.logger.Error("unable to marshal blobs to SSZ", "err", err)
+			errServerError.write(w)
+			return
+		}
+
+		_, err = w.Write(res)
+
+		if err != nil {
+			a.logger.Error("unable to write ssz response", "err", err)
+			errServerError.write(w)
+			return
+		}
+	} else {
+		w.Header().Set("Content-Type", jsonAcceptType)
+		err := json.NewEncoder(w).Encode(blobs)
+		if err != nil {
+			a.logger.Error("unable to encode blobs to JSON", "err", err)
+			errServerError.write(w)
+			return
+		}
+	}
+}

api/service/api_test.go

@@ -3,6 +3,7 @@ package service
 import (
 	"compress/gzip"
 	"context"
+	"crypto/sha256"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -21,6 +22,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 	"github.com/ethereum-optimism/optimism/op-service/testlog"
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto/kzg4844"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/stretchr/testify/require"
 )
@@ -80,7 +82,7 @@ func TestAPIService(t *testing.T) {
 			BeaconBlockHash: rootOne,
 		},
 		BlobSidecars: storage.BlobSidecars{
-			Data: blobtest.NewBlobSidecars(t, 2),
+			Data: blobtest.NewBlobSidecars(t, 2, nil),
 		},
 	}
 
@@ -89,7 +91,7 @@ func TestAPIService(t *testing.T) {
 			BeaconBlockHash: rootTwo,
 		},
 		BlobSidecars: storage.BlobSidecars{
-			Data: blobtest.NewBlobSidecars(t, 2),
+			Data: blobtest.NewBlobSidecars(t, 2, nil),
 		},
 	}
 
@@ -99,7 +101,7 @@ func TestAPIService(t *testing.T) {
 			BeaconBlockHash: rootThree,
 		},
 		BlobSidecars: storage.BlobSidecars{
-			Data: blobtest.NewBlobSidecars(t, 8), // More than 6 blobs
+			Data: blobtest.NewBlobSidecars(t, 8, nil), // More than 6 blobs
 		},
 	}
 
@@ -364,3 +366,130 @@ func TestHealthHandler(t *testing.T) {
 
 	require.Equal(t, 200, response.Code)
 }
+
+func TestBlobsHandlerJSON(t *testing.T) {
+	a, fs, _, cleanup := setup(t)
+	defer cleanup()
+
+	// Pre-populate storage with blob sidecars
+	testBlobs := blobtest.NewBlobSidecars(t, 3, nil)
+	data := storage.BlobData{
+		Header: storage.Header{
+			BeaconBlockHash: blobtest.Five,
+		},
+		BlobSidecars: storage.BlobSidecars{
+			Data: testBlobs,
+		},
+	}
+	err := fs.WriteBlob(context.Background(), data)
+	require.NoError(t, err)
+
+	// Request blobs endpoint with JSON encoding
+	request := httptest.NewRequest("GET", "/eth/v1/beacon/blobs/"+blobtest.Five.String(), nil)
+	request.Header.Set("Accept", "application/json")
+	response := httptest.NewRecorder()
+
+	a.router.ServeHTTP(response, request)
+
+	require.Equal(t, 200, response.Code)
+	require.Equal(t, "application/json", response.Header().Get("Content-Type"))
+
+	var blobs v1.Blobs
+	err = json.Unmarshal(response.Body.Bytes(), &blobs)
+	require.NoError(t, err)
+	require.Equal(t, len(testBlobs), len(blobs))
+
+	// Verify blob data matches
+	for i, blob := range blobs {
+		require.Equal(t, testBlobs[i].Blob, *blob)
+	}
+}
+
+func TestBlobsHandlerSSZ(t *testing.T) {
+	a, fs, _, cleanup := setup(t)
+	defer cleanup()
+
+	// Pre-populate storage with blob sidecars
+	testBlobs := blobtest.NewBlobSidecars(t, 2, nil)
+	data := storage.BlobData{
+		Header: storage.Header{
+			BeaconBlockHash: blobtest.One,
+		},
+		BlobSidecars: storage.BlobSidecars{
+			Data: testBlobs,
+		},
+	}
+	err := fs.WriteBlob(context.Background(), data)
+	require.NoError(t, err)
+
+	// Request blobs endpoint with SSZ encoding
+	request := httptest.NewRequest("GET", "/eth/v1/beacon/blobs/"+blobtest.One.String(), nil)
+	request.Header.Set("Accept", "application/octet-stream")
+	response := httptest.NewRecorder()
+
+	a.router.ServeHTTP(response, request)
+
+	require.Equal(t, 200, response.Code)
+	require.Equal(t, "application/octet-stream", response.Header().Get("Content-Type"))
+
+	// Unmarshal SSZ response
+	blobs := v1.Blobs{}
+	err = blobs.UnmarshalSSZ(response.Body.Bytes())
+	require.NoError(t, err)
+	require.Equal(t, len(testBlobs), len(blobs))
+}
+
+func TestBlobsHandlerWithVersionedHashes(t *testing.T) {
+	a, fs, _, cleanup := setup(t)
+	defer cleanup()
+
+	// Pre-populate storage with blob sidecars
+	testBlobs := blobtest.NewBlobSidecars(t, 5, nil)
+	data := storage.BlobData{
+		Header: storage.Header{
+			BeaconBlockHash: blobtest.Four,
+		},
+		BlobSidecars: storage.BlobSidecars{
+			Data: testBlobs,
+		},
+	}
+	err := fs.WriteBlob(context.Background(), data)
+	require.NoError(t, err)
+
+	// Compute versioned hashes from commitments
+	hasher := sha256.New()
+	commitment0 := kzg4844.Commitment(testBlobs[0].KZGCommitment)
+	vh0 := kzg4844.CalcBlobHashV1(hasher, &commitment0)
+
+	hasher.Reset()
+	commitment2 := kzg4844.Commitment(testBlobs[2].KZGCommitment)
+	vh2 := kzg4844.CalcBlobHashV1(hasher, &commitment2)
+
+	// Request blobs endpoint with versioned_hashes filter
+	request := httptest.NewRequest("GET", fmt.Sprintf("/eth/v1/beacon/blobs/%s?versioned_hashes=%s&versioned_hashes=%s", blobtest.Four.String(), common.Hash(vh0).Hex(), common.Hash(vh2).Hex()), nil)
+	request.Header.Set("Accept", "application/json")
+	response := httptest.NewRecorder()
+
+	a.router.ServeHTTP(response, request)
+
+	require.Equal(t, 200, response.Code)
+
+	var blobs v1.Blobs
+	err = json.Unmarshal(response.Body.Bytes(), &blobs)
+	require.NoError(t, err)
+	require.Equal(t, 2, len(blobs))
+}
+
+func TestBlobsHandlerNotFound(t *testing.T) {
+	a, _, _, cleanup := setup(t)
+	defer cleanup()
+
+	// Request blobs endpoint for non-existent block
+	request := httptest.NewRequest("GET", "/eth/v1/beacon/blobs/"+blobtest.Seven.String(), nil)
+	request.Header.Set("Accept", "application/json")
+	response := httptest.NewRecorder()
+
+	a.router.ServeHTTP(response, request)
+
+	require.Equal(t, 404, response.Code)
+}