op-node,sources: Add Beacon source option to fetch all sidecars (#9151)

* op-node,soruces: Add Beacon source option to fetch all sidecars

* Update op-service/sources/l1_beacon_client.go

Break out of inner loop after idx found

Co-authored-by: protolambda <proto@protolambda.com>

* sources: Remove odering from blobsFromSidecars

* sources: Fix godoc of GetBlobSidecars

---------

Co-authored-by: protolambda <proto@protolambda.com>

op-e2e/l1_beacon_client_test.go

@@ -21,7 +21,8 @@ func TestGetVersion(t *testing.T) {
 	})
 	require.NoError(t, beaconApi.Start("127.0.0.1:0"))
 
-	cl := sources.NewL1BeaconClient(client.NewBasicHTTPClient(beaconApi.BeaconAddr(), l))
+	beaconCfg := sources.L1BeaconClientConfig{FetchAllSidecars: false}
+	cl := sources.NewL1BeaconClient(client.NewBasicHTTPClient(beaconApi.BeaconAddr(), l), beaconCfg)
 
 	version, err := cl.GetVersion(context.Background())
 	require.NoError(t, err)

op-node/flags/flags.go

@@ -56,6 +56,13 @@ var (
 		Value:    false,
 		EnvVars:  prefixEnvVars("L1_BEACON_IGNORE"),
 	}
+	BeaconFetchAllSidecars = &cli.BoolFlag{
+		Name:     "l1.beacon.fetch-all-sidecars",
+		Usage:    "If true, all sidecars are fetched and filtered locally. Workaround for buggy Beacon nodes.",
+		Required: false,
+		Value:    false,
+		EnvVars:  prefixEnvVars("L1_BEACON_FETCH_ALL_SIDECARS"),
+	}
 	SyncModeFlag = &cli.GenericFlag{
 		Name:    "syncmode",
 		Usage:   fmt.Sprintf("IN DEVELOPMENT: Options are: %s", openum.EnumString(sync.ModeStrings)),
@@ -285,6 +292,7 @@ var requiredFlags = []cli.Flag{
 var optionalFlags = []cli.Flag{
 	BeaconAddr,
 	BeaconCheckIgnore,
+	BeaconFetchAllSidecars,
 	SyncModeFlag,
 	RPCListenAddr,
 	RPCListenPort,

op-node/node/client.go

@@ -33,6 +33,7 @@ type L1BeaconEndpointSetup interface {
 	Setup(ctx context.Context, log log.Logger) (cl client.HTTP, err error)
 	// ShouldIgnoreBeaconCheck returns true if the Beacon-node version check should not halt startup.
 	ShouldIgnoreBeaconCheck() bool
+	ShouldFetchAllSidecars() bool
 	Check() error
 }
 
@@ -175,8 +176,9 @@ func (cfg *PreparedL1Endpoint) Check() error {
 }
 
 type L1BeaconEndpointConfig struct {
-	BeaconAddr        string // Address of L1 User Beacon-API endpoint to use (beacon namespace required)
-	BeaconCheckIgnore bool   // When false, halt startup if the beacon version endpoint fails
+	BeaconAddr             string // Address of L1 User Beacon-API endpoint to use (beacon namespace required)
+	BeaconCheckIgnore      bool   // When false, halt startup if the beacon version endpoint fails
+	BeaconFetchAllSidecars bool   // Whether to fetch all blob sidecars and filter locally
 }
 
 var _ L1BeaconEndpointSetup = (*L1BeaconEndpointConfig)(nil)
@@ -195,3 +197,7 @@ func (cfg *L1BeaconEndpointConfig) Check() error {
 func (cfg *L1BeaconEndpointConfig) ShouldIgnoreBeaconCheck() bool {
 	return cfg.BeaconCheckIgnore
 }
+
+func (cfg *L1BeaconEndpointConfig) ShouldFetchAllSidecars() bool {
+	return cfg.BeaconFetchAllSidecars
+}

op-node/node/node.go

@@ -30,9 +30,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-service/sources"
 )
 
-var (
-	ErrAlreadyClosed = errors.New("node is already closed")
-)
+var ErrAlreadyClosed = errors.New("node is already closed")
 
 type OpNode struct {
 	log        log.Logger
@@ -308,7 +306,10 @@ func (n *OpNode) initL1BeaconAPI(ctx context.Context, cfg *Config) error {
 	if err != nil {
 		return fmt.Errorf("failed to setup L1 Beacon API client: %w", err)
 	}
-	n.beacon = sources.NewL1BeaconClient(httpClient)
+	beaconCfg := sources.L1BeaconClientConfig{
+		FetchAllSidecars: cfg.Beacon.ShouldFetchAllSidecars(),
+	}
+	n.beacon = sources.NewL1BeaconClient(httpClient, beaconCfg)
 
 	// Retry retrieval of the Beacon API version, to be more robust on startup against Beacon API connection issues.
 	beaconVersion, missingEndpoint, err := retry.Do2[string, bool](ctx, 5, retry.Exponential(), func() (string, bool, error) {

op-node/service.go

@@ -126,8 +126,9 @@ func NewConfig(ctx *cli.Context, log log.Logger) (*node.Config, error) {
 
 func NewBeaconEndpointConfig(ctx *cli.Context) node.L1BeaconEndpointSetup {
 	return &node.L1BeaconEndpointConfig{
-		BeaconAddr:        ctx.String(flags.BeaconAddr.Name),
-		BeaconCheckIgnore: ctx.Bool(flags.BeaconCheckIgnore.Name),
+		BeaconAddr:             ctx.String(flags.BeaconAddr.Name),
+		BeaconCheckIgnore:      ctx.Bool(flags.BeaconCheckIgnore.Name),
+		BeaconFetchAllSidecars: ctx.Bool(flags.BeaconFetchAllSidecars.Name),
 	}
 }
 

op-service/sources/l1_beacon_client.go

@@ -8,7 +8,6 @@ import (
 	"net/http"
 	"net/url"
 	"path"
-	"slices"
 	"strconv"
 	"sync"
 
@@ -25,24 +24,29 @@ const (
 	sidecarsMethodPrefix = "eth/v1/beacon/blob_sidecars/"
 )
 
+type L1BeaconClientConfig struct {
+	FetchAllSidecars bool
+}
+
 type L1BeaconClient struct {
-	cl client.HTTP
+	cl  client.HTTP
+	cfg L1BeaconClientConfig
 
 	initLock     sync.Mutex
 	timeToSlotFn TimeToSlotFn
 }
 
 // NewL1BeaconClient returns a client for making requests to an L1 consensus layer node.
-func NewL1BeaconClient(cl client.HTTP) *L1BeaconClient {
-	return &L1BeaconClient{cl: cl}
+func NewL1BeaconClient(cl client.HTTP, cfg L1BeaconClientConfig) *L1BeaconClient {
+	return &L1BeaconClient{cl: cl, cfg: cfg}
 }
 
 func (cl *L1BeaconClient) apiReq(ctx context.Context, dest any, reqPath string, reqQuery url.Values) error {
 	headers := http.Header{}
 	headers.Add("Accept", "application/json")
 	resp, err := cl.cl.Get(ctx, reqPath, reqQuery, headers)
 	if err != nil {
-		return fmt.Errorf("%w: http Get failed", err)
+		return fmt.Errorf("http Get failed: %w", err)
 	}
 	if resp.StatusCode != http.StatusOK {
 		errMsg, _ := io.ReadAll(resp.Body)
@@ -54,7 +58,7 @@ func (cl *L1BeaconClient) apiReq(ctx context.Context, dest any, reqPath string,
 		return err
 	}
 	if err := resp.Body.Close(); err != nil {
-		return fmt.Errorf("%w: failed to close response body", err)
+		return fmt.Errorf("failed to close response body: %w", err)
 	}
 	return nil
 }
@@ -93,37 +97,54 @@ func (cl *L1BeaconClient) GetTimeToSlotFn(ctx context.Context) (TimeToSlotFn, er
 	return cl.timeToSlotFn, nil
 }
 
-// GetBlobSidecars fetches blob sidecars that were confirmed in the specified L1 block with the
-// given indexed hashes. Order of the returned sidecars is not guaranteed, and blob data is not
-// checked for validity.
+// GetBlobSidecars fetches blob sidecars that were confirmed in the specified
+// L1 block with the given indexed hashes.
+// Order of the returned sidecars is guaranteed to be that of the hashes.
+// Blob data is not checked for validity.
 func (cl *L1BeaconClient) GetBlobSidecars(ctx context.Context, ref eth.L1BlockRef, hashes []eth.IndexedBlobHash) ([]*eth.BlobSidecar, error) {
 	if len(hashes) == 0 {
 		return []*eth.BlobSidecar{}, nil
 	}
 	slotFn, err := cl.GetTimeToSlotFn(ctx)
 	if err != nil {
-		return nil, fmt.Errorf("%w: failed to get time to slot function", err)
+		return nil, fmt.Errorf("failed to get time to slot function: %w", err)
 	}
 	slot, err := slotFn(ref.Time)
 	if err != nil {
-		return nil, fmt.Errorf("%w: error in converting ref.Time to slot", err)
+		return nil, fmt.Errorf("error in converting ref.Time to slot: %w", err)
 	}
 
 	reqPath := path.Join(sidecarsMethodPrefix, strconv.FormatUint(slot, 10))
-	reqQuery := url.Values{}
-	for i := range hashes {
-		reqQuery.Add("indices", strconv.FormatUint(hashes[i].Index, 10))
+	var reqQuery url.Values
+	if !cl.cfg.FetchAllSidecars {
+		reqQuery = url.Values{}
+		for i := range hashes {
+			reqQuery.Add("indices", strconv.FormatUint(hashes[i].Index, 10))
+		}
 	}
+
 	var resp eth.APIGetBlobSidecarsResponse
 	if err := cl.apiReq(ctx, &resp, reqPath, reqQuery); err != nil {
-		return nil, fmt.Errorf("%w: failed to fetch blob sidecars for slot %v block %v", err, slot, ref)
+		return nil, fmt.Errorf("failed to fetch blob sidecars for slot %v block %v: %w", slot, ref, err)
+	}
+
+	apiscs := make([]*eth.APIBlobSidecar, 0, len(hashes))
+	// filter and order by hashes
+	for _, h := range hashes {
+		for _, apisc := range resp.Data {
+			if h.Index == uint64(apisc.Index) {
+				apiscs = append(apiscs, apisc)
+				break
+			}
+		}
 	}
-	if len(hashes) != len(resp.Data) {
+
+	if len(hashes) != len(apiscs) {
 		return nil, fmt.Errorf("expected %v sidecars but got %v", len(hashes), len(resp.Data))
 	}
 
 	bscs := make([]*eth.BlobSidecar, 0, len(hashes))
-	for _, apisc := range resp.Data {
+	for _, apisc := range apiscs {
 		bscs = append(bscs, apisc.BlobSidecar())
 	}
 
@@ -137,24 +158,22 @@ func (cl *L1BeaconClient) GetBlobSidecars(ctx context.Context, ref eth.L1BlockRe
 func (cl *L1BeaconClient) GetBlobs(ctx context.Context, ref eth.L1BlockRef, hashes []eth.IndexedBlobHash) ([]*eth.Blob, error) {
 	blobSidecars, err := cl.GetBlobSidecars(ctx, ref, hashes)
 	if err != nil {
-		return nil, fmt.Errorf("%w: failed to get blob sidecars for L1BlockRef %s", err, ref)
+		return nil, fmt.Errorf("failed to get blob sidecars for L1BlockRef %s: %w", ref, err)
 	}
 	return blobsFromSidecars(blobSidecars, hashes)
 }
 
 func blobsFromSidecars(blobSidecars []*eth.BlobSidecar, hashes []eth.IndexedBlobHash) ([]*eth.Blob, error) {
+	if len(blobSidecars) != len(hashes) {
+		return nil, fmt.Errorf("number of hashes and blobSidecars mismatch, %d != %d", len(hashes), len(blobSidecars))
+	}
+
 	out := make([]*eth.Blob, len(hashes))
 	for i, ih := range hashes {
-		// The beacon node api makes no guarantees on order of the returned blob sidecars, so
-		// search for the sidecar that matches the current indexed hash to ensure blobs are
-		// returned in the same order.
-		scIndex := slices.IndexFunc(
-			blobSidecars,
-			func(sc *eth.BlobSidecar) bool { return uint64(sc.Index) == ih.Index })
-		if scIndex == -1 {
-			return nil, fmt.Errorf("no blob in response matches desired index: %v", ih.Index)
+		sidecar := blobSidecars[i]
+		if sidx := uint64(sidecar.Index); sidx != ih.Index {
+			return nil, fmt.Errorf("expected sidecars to be ordered by hashes, but got %d != %d", sidx, ih.Index)
 		}
-		sidecar := blobSidecars[scIndex]
 
 		// make sure the blob's kzg commitment hashes to the expected value
 		hash := eth.KZGToVersionedHash(kzg4844.Commitment(sidecar.KZGCommitment))
@@ -164,7 +183,7 @@ func blobsFromSidecars(blobSidecars []*eth.BlobSidecar, hashes []eth.IndexedBlob
 
 		// confirm blob data is valid by verifying its proof against the commitment
 		if err := eth.VerifyBlobProof(&sidecar.Blob, kzg4844.Commitment(sidecar.KZGCommitment), kzg4844.Proof(sidecar.KZGProof)); err != nil {
-			return nil, fmt.Errorf("%w: blob at index %d failed verification", err, i)
+			return nil, fmt.Errorf("blob at index %d failed verification: %w", i, err)
 		}
 		out[i] = &sidecar.Blob
 	}

op-service/sources/l1_beacon_client_test.go

@@ -41,9 +41,18 @@ func TestBlobsFromSidecars(t *testing.T) {
 
 	hashes := []eth.IndexedBlobHash{index0, index1, index2}
 
-	// put the sidecars in scrambled order of expectation to confirm function appropriately
-	// reorders the output to match that of the blob hashes
+	// put the sidecars in scrambled order to confirm error
 	sidecars := []*eth.BlobSidecar{sidecar2, sidecar0, sidecar1}
+	_, err := blobsFromSidecars(sidecars, hashes)
+	require.Error(t, err)
+
+	// too few sidecars should error
+	sidecars = []*eth.BlobSidecar{sidecar0, sidecar1}
+	_, err = blobsFromSidecars(sidecars, hashes)
+	require.Error(t, err)
+
+	// correct order should work
+	sidecars = []*eth.BlobSidecar{sidecar0, sidecar1, sidecar2}
 	blobs, err := blobsFromSidecars(sidecars, hashes)
 	require.NoError(t, err)
 	// confirm order by checking first blob byte against expected index