feat(node): add beacon RPC fallback for blob sidecar fetching

node/derivation/base_client_test.go

@@ -0,0 +1,169 @@
+package derivation
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+
+	"github.com/morph-l2/go-ethereum/common"
+	"github.com/stretchr/testify/require"
+)
+
+// zero-value but correctly-sized hex so the beacon JSON decodes into a
+// BlobSidecar. GetBlobSidecarsEnhanced does not verify blob contents (that
+// happens downstream), it only counts sidecars, so dummy bytes are fine here.
+var (
+	hex32 = "0x" + strings.Repeat("00", 32)
+	hex48 = "0x" + strings.Repeat("00", 48)
+)
+
+func sidecarJSON(index int) string {
+	return fmt.Sprintf(`{"block_root":%q,"slot":"1","blob":"0x00","index":"%d","kzg_commitment":%q,"kzg_proof":%q}`,
+		hex32, index, hex48, hex48)
+}
+
+// beaconBehavior controls what a stub beacon returns for blob_sidecars.
+type beaconBehavior int
+
+const (
+	beaconServesBlob  beaconBehavior = iota // 200 with one sidecar
+	beaconServesEmpty                       // 200 with an empty list (pruned / not indexed)
+	beaconServerError                       // 500
+)
+
+// newStubBeacon serves the genesis + spec endpoints (needed for slot math) and
+// answers blob_sidecars according to behavior. It returns the base URL and a
+// counter of blob_sidecars requests received.
+func newStubBeacon(t *testing.T, behavior beaconBehavior) (string, *int32) {
+	t.Helper()
+	var blobHits int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case strings.Contains(r.URL.Path, genesisMethod):
+			_, _ = w.Write([]byte(`{"data":{"genesis_time":"0"}}`))
+		case strings.Contains(r.URL.Path, specMethod):
+			_, _ = w.Write([]byte(`{"data":{"SECONDS_PER_SLOT":"12"}}`))
+		case strings.Contains(r.URL.Path, sidecarsMethodPrefix):
+			atomic.AddInt32(&blobHits, 1)
+			switch behavior {
+			case beaconServesBlob:
+				_, _ = w.Write([]byte(`{"data":[` + sidecarJSON(0) + `]}`))
+			case beaconServesEmpty:
+				_, _ = w.Write([]byte(`{"data":[]}`))
+			case beaconServerError:
+				w.WriteHeader(http.StatusInternalServerError)
+			}
+		default:
+			w.WriteHeader(http.StatusNotFound)
+		}
+	}))
+	t.Cleanup(srv.Close)
+	return srv.URL, &blobHits
+}
+
+func oneHash() []IndexedBlobHash {
+	return []IndexedBlobHash{{Index: 0, Hash: common.Hash{}}}
+}
+
+type canceledHTTP struct {
+	calls *int32
+}
+
+func (h canceledHTTP) Get(ctx context.Context, _ string, _ http.Header) (*http.Response, error) {
+	atomic.AddInt32(h.calls, 1)
+	return nil, ctx.Err()
+}
+
+func fetch(t *testing.T, c *FallbackBeaconClient) ([]*BlobSidecar, error) {
+	t.Helper()
+	return c.GetBlobSidecarsEnhanced(context.Background(), L1BlockRef{Time: 12}, oneHash())
+}
+
+// The primary serves the blob and the fallback is never queried.
+func TestFallbackBeacon_PrimaryServesBlob(t *testing.T) {
+	primary, primaryHits := newStubBeacon(t, beaconServesBlob)
+	fallback, fallbackHits := newStubBeacon(t, beaconServesBlob)
+
+	c := NewFallbackBeaconClient([]string{primary, fallback}, nil, nil)
+	sidecars, err := fetch(t, c)
+	require.NoError(t, err)
+	require.Len(t, sidecars, 1)
+	require.EqualValues(t, 1, atomic.LoadInt32(primaryHits))
+	require.EqualValues(t, 0, atomic.LoadInt32(fallbackHits), "fallback must not be queried while primary serves the blob")
+}
+
+// The key case: primary answers 200 but with NO sidecars (pruned / not indexed).
+// This must fall back to a beacon that actually has the blob.
+func TestFallbackBeacon_FallsBackOnEmptyResult(t *testing.T) {
+	primary, primaryHits := newStubBeacon(t, beaconServesEmpty)
+	fallback, fallbackHits := newStubBeacon(t, beaconServesBlob)
+
+	c := NewFallbackBeaconClient([]string{primary, fallback}, nil, nil)
+	sidecars, err := fetch(t, c)
+	require.NoError(t, err)
+	require.Len(t, sidecars, 1)
+	require.Positive(t, atomic.LoadInt32(primaryHits))
+	require.Positive(t, atomic.LoadInt32(fallbackHits))
+}
+
+// A 5xx on the primary also falls back.
+func TestFallbackBeacon_FallsBackOnServerError(t *testing.T) {
+	primary, _ := newStubBeacon(t, beaconServerError)
+	fallback, fallbackHits := newStubBeacon(t, beaconServesBlob)
+
+	c := NewFallbackBeaconClient([]string{primary, fallback}, nil, nil)
+	sidecars, err := fetch(t, c)
+	require.NoError(t, err)
+	require.Len(t, sidecars, 1)
+	require.Positive(t, atomic.LoadInt32(fallbackHits))
+}
+
+// An unreachable primary (transport error) falls back.
+func TestFallbackBeacon_FallsBackOnTransportError(t *testing.T) {
+	fallback, fallbackHits := newStubBeacon(t, beaconServesBlob)
+
+	c := NewFallbackBeaconClient([]string{"http://127.0.0.1:0", fallback}, nil, nil)
+	sidecars, err := fetch(t, c)
+	require.NoError(t, err)
+	require.Len(t, sidecars, 1)
+	require.Positive(t, atomic.LoadInt32(fallbackHits))
+}
+
+func TestFallbackBeacon_StopsOnContextCancellation(t *testing.T) {
+	var primaryCalls, fallbackCalls int32
+	c := &FallbackBeaconClient{
+		clients: []*L1BeaconClient{
+			NewL1BeaconClient(canceledHTTP{calls: &primaryCalls}),
+			NewL1BeaconClient(canceledHTTP{calls: &fallbackCalls}),
+		},
+		endpoints: []string{"primary", "fallback"},
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	sidecars, err := c.GetBlobSidecarsEnhanced(ctx, L1BlockRef{Time: 12}, oneHash())
+
+	require.ErrorIs(t, err, context.Canceled)
+	require.Nil(t, sidecars)
+	require.Positive(t, atomic.LoadInt32(&primaryCalls))
+	require.Zero(t, atomic.LoadInt32(&fallbackCalls))
+}
+
+// When every beacon fails to serve the blob, an error is returned and every
+// endpoint's failure is recorded in metrics (exercised via a real *Metrics).
+func TestFallbackBeacon_AllFailReturnsError(t *testing.T) {
+	primary, primaryHits := newStubBeacon(t, beaconServesEmpty)
+	fallback, fallbackHits := newStubBeacon(t, beaconServerError)
+
+	m := PrometheusMetrics("morphnode_test_" + t.Name())
+	c := NewFallbackBeaconClient([]string{primary, fallback}, nil, m)
+	sidecars, err := fetch(t, c)
+	require.Error(t, err)
+	require.Nil(t, sidecars)
+	require.Positive(t, atomic.LoadInt32(primaryHits))
+	require.Positive(t, atomic.LoadInt32(fallbackHits))
+}

node/derivation/beacon.go

@@ -16,6 +16,7 @@ import (
 	"github.com/morph-l2/go-ethereum/core/types"
 	"github.com/morph-l2/go-ethereum/crypto/kzg4844"
 	"github.com/morph-l2/go-ethereum/params"
+	tmlog "github.com/tendermint/tendermint/libs/log"
 )
 
 const (
@@ -208,6 +209,64 @@ func dataAndHashesFromTxs(txs types.Transactions, targetTx *types.Transaction) [
 	return hashes
 }
 
+// FallbackBeaconClient queries several beacon nodes in order for blob sidecars.
+// A beacon is skipped and the next one tried when it errors, is unreachable, or
+// answers with too few sidecars — a beacon that pruned the slot or has not
+// indexed it yet still replies 200 with an empty/partial list, which is exactly
+// the "temporarily failed to fetch blob" case seen in production and which a
+// transport-level fallback would miss. The failing endpoint is recorded in the
+// beacon_request_failure_total metric so a flaky node is visible on dashboards.
+// With a single endpoint it behaves like a bare L1BeaconClient.
+type FallbackBeaconClient struct {
+	clients   []*L1BeaconClient
+	endpoints []string // parallel to clients, used only for logs/metrics
+	log       tmlog.Logger
+	metrics   *Metrics
+}
+
+func NewFallbackBeaconClient(endpoints []string, log tmlog.Logger, metrics *Metrics) *FallbackBeaconClient {
+	clients := make([]*L1BeaconClient, 0, len(endpoints))
+	for _, endpoint := range endpoints {
+		clients = append(clients, NewL1BeaconClient(NewBasicHTTPClient(endpoint, log)))
+	}
+	return &FallbackBeaconClient{
+		clients:   clients,
+		endpoints: endpoints,
+		log:       log,
+		metrics:   metrics,
+	}
+}
+
+// GetBlobSidecarsEnhanced tries each configured beacon in order and returns the
+// first response that actually carries the requested blobs (at least len(hashes)
+// sidecars, or at least one when no explicit hashes are requested). A beacon
+// that errors or returns an incomplete set is recorded as a failure and skipped.
+// If every beacon fails, the last error is returned.
+func (c *FallbackBeaconClient) GetBlobSidecarsEnhanced(ctx context.Context, ref L1BlockRef, hashes []IndexedBlobHash) ([]*BlobSidecar, error) {
+	var lastErr error
+	for i, cl := range c.clients {
+		sidecars, err := cl.GetBlobSidecarsEnhanced(ctx, ref, hashes)
+		if err == nil && len(sidecars) > 0 && len(sidecars) >= len(hashes) {
+			return sidecars, nil
+		}
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+		if err == nil {
+			err = fmt.Errorf("beacon returned %d sidecars, want at least %d", len(sidecars), len(hashes))
+		}
+		if c.metrics != nil {
+			c.metrics.IncBeaconRequestFailure(c.endpoints[i])
+		}
+		if c.log != nil {
+			c.log.Error("beacon failed to serve blob sidecars, trying next endpoint",
+				"endpoint", c.endpoints[i], "err", err)
+		}
+		lastErr = err
+	}
+	return nil, lastErr
+}
+
 // Note: ForceGetAllBlobs is defined in derivation.go in the same package
 
 // GetBlobSidecarsEnhanced is an enhanced version of GetBlobSidecars method, combining two approaches to fetch blob data

node/derivation/config.go

@@ -84,6 +84,22 @@ type Config struct {
 	MetricsPort           uint64          `json:"metrics_port"`
 }
 
+// BeaconRpcList splits the configured beacon RPC endpoint(s) on commas and
+// returns them trimmed, in order (primary first). Operators can supply several
+// beacon nodes via a single comma-separated --l1.beaconrpc value; derivation
+// then falls back to the next endpoint when one temporarily fails to serve
+// blob sidecars. A plain single URL yields a one-element slice, so existing
+// configs keep working unchanged.
+func (c *Config) BeaconRpcList() []string {
+	var endpoints []string
+	for _, endpoint := range strings.Split(c.BeaconRpc, ",") {
+		if endpoint = strings.TrimSpace(endpoint); endpoint != "" {
+			endpoints = append(endpoints, endpoint)
+		}
+	}
+	return endpoints
+}
+
 // DefaultConfig returns the default derivation configuration.
 func DefaultConfig() *Config {
 	return &Config{
@@ -126,7 +142,7 @@ func (c *Config) SetCliContext(ctx *cli.Context) error {
 		c.RollupContractAddress = types.HoodiRollupContractAddress
 	}
 	c.BeaconRpc = ctx.GlobalString(flags.L1BeaconAddr.Name)
-	if c.BeaconRpc == "" {
+	if len(c.BeaconRpcList()) == 0 {
 		return errors.New("invalid L1BeaconAddr")
 	}
 

node/derivation/config_test.go

@@ -6,6 +6,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/stretchr/testify/require"
 	"github.com/urfave/cli"
 
 	"morph-l2/node/flags"
@@ -89,6 +90,26 @@ func TestVerifyMode_RejectsUnknown(t *testing.T) {
 	}
 }
 
+func TestBeaconRpcList(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		in   string
+		want []string
+	}{
+		{"single", "http://a", []string{"http://a"}},
+		{"multiple", "http://a,http://b,http://c", []string{"http://a", "http://b", "http://c"}},
+		{"trims spaces", " http://a , http://b ", []string{"http://a", "http://b"}},
+		{"drops empties", "http://a,,http://b,", []string{"http://a", "http://b"}},
+		{"empty", "", nil},
+		{"only separators", " , , ", nil},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			got := (&Config{BeaconRpc: tc.in}).BeaconRpcList()
+			require.Equal(t, tc.want, got)
+		})
+	}
+}
+
 func validateAndDefaultVerifyModeErr(t *testing.T, s string) error {
 	t.Helper()
 	_, err := validateAndDefaultVerifyMode(s)

node/derivation/derivation.go

@@ -47,7 +47,7 @@ type Derivation struct {
 	logger                tmlog.Logger
 	rollup                *bindings.Rollup
 	metrics               *Metrics
-	l1BeaconClient        *L1BeaconClient
+	l1BeaconClient        *FallbackBeaconClient
 	L2ToL1MessagePasser   *bindings.L2ToL1MessagePasser
 
 	rollupABI             *abi.ABI
@@ -124,8 +124,7 @@ func NewDerivationClient(ctx context.Context, cfg *Config, syncer *sync.Syncer,
 	// itself is started once at the top level (cmd/node/main.go) so every
 	// verify-mode and sequencer-mode produces exactly one /metrics URL.
 	metrics := PrometheusMetrics("morphnode")
-	baseHttp := NewBasicHTTPClient(cfg.BeaconRpc, logger)
-	l1BeaconClient := NewL1BeaconClient(baseHttp)
+	l1BeaconClient := NewFallbackBeaconClient(cfg.BeaconRpcList(), logger, metrics)
 
 	l2Client := types.NewRetryableClient(aClient, eClient, logger)
 	tagAdv := newTagAdvancer(l2Client, metrics, logger)
@@ -389,7 +388,7 @@ func (d *Derivation) derivationBlock(ctx context.Context) {
 				// or fails — without it, a deriveForce error would leave
 				// reactors stopped indefinitely (Stop is idempotent on
 				// retry, but Start is never reached).
-				err = d.withReactorsQuiesced(ctx, batchInfo.batchIndex, func() error {
+				err = d.withReactorsQuiesced(batchInfo.batchIndex, func() error {
 					var derErr error
 					lastHeader, derErr = d.deriveForce(batchInfoFull)
 					return derErr
@@ -433,7 +432,7 @@ func (d *Derivation) derivationBlock(ctx context.Context) {
 					// so the deferred Start runs whether deriveForce succeeds
 					// or fails — without it, a deriveForce error would leave
 					// reactors stopped indefinitely.
-					err = d.withReactorsQuiesced(ctx, batchInfo.batchIndex, func() error {
+					err = d.withReactorsQuiesced(batchInfo.batchIndex, func() error {
 						var derErr error
 						lastHeader, derErr = d.deriveForce(batchInfoFull)
 						return derErr
@@ -838,7 +837,7 @@ func (d *Derivation) derive(rollupData *BatchInfo) (*eth.Header, error) {
 // zero (which would tell blocksync to re-fetch from genesis). HA
 // sequencers and mock-mode skip (d.node == nil): sequencers don't
 // auto-reorg, mock has no reactors.
-func (d *Derivation) withReactorsQuiesced(ctx context.Context, batchIndex uint64, body func() error) error {
+func (d *Derivation) withReactorsQuiesced(batchIndex uint64, body func() error) error {
 	if d.node == nil {
 		return body()
 	}

node/derivation/metrics.go

@@ -36,6 +36,12 @@ type Metrics struct {
 	FinalizedL2BlockNumber     metrics.Gauge
 	L1ReorgResetTotal          metrics.Counter
 	TagInvariantViolationTotal metrics.Counter
+
+	// BeaconRequestFailure counts beacons that failed to serve blob sidecars
+	// (unreachable, non-200, or an incomplete/empty result), labeled by
+	// endpoint, before the FallbackBeaconClient moved on to the next configured
+	// beacon. A rising count for one endpoint pinpoints the flaky beacon node.
+	BeaconRequestFailure metrics.Counter
 }
 
 func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
@@ -122,6 +128,14 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
 			Name:      "tag_invariant_violation_total",
 			Help:      "Times the finalized <= safe <= unsafe invariant failed; SetBlockTags is skipped on each occurrence.",
 		}, labels).With(labelsAndValues...),
+		// The endpoint label is bound per-increment in IncBeaconRequestFailure,
+		// so it is declared here rather than pre-bound via With.
+		BeaconRequestFailure: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: metricsSubsystem,
+			Name:      "beacon_request_failure_total",
+			Help:      "Beacon requests that failed (transport error or non-200) per endpoint, before falling back to the next configured beacon.",
+		}, append(append([]string{}, labels...), "endpoint")),
 	}
 }
 
@@ -176,3 +190,7 @@ func (m *Metrics) IncL1ReorgReset() {
 func (m *Metrics) IncTagInvariantViolation() {
 	m.TagInvariantViolationTotal.Add(1)
 }
+
+func (m *Metrics) IncBeaconRequestFailure(endpoint string) {
+	m.BeaconRequestFailure.With("endpoint", endpoint).Add(1)
+}