Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
140 changes: 140 additions & 0 deletions node/derivation/base_client_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,140 @@
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{}}}
}

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))
}

// 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))
}
56 changes: 56 additions & 0 deletions node/derivation/beacon.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -208,6 +209,61 @@ 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 == nil {
err = fmt.Errorf("beacon returned %d sidecars, want at least %d", len(sidecars), len(hashes))
}
Comment on lines +246 to +254

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

Fail fast on context cancellation.

If ctx is canceled (e.g., during node shutdown or a request timeout), the current loop treats the resulting context.Canceled as a beacon failure, increments the metric, logs an error, and proceeds to the next endpoint (which will also immediately fail).

Abort early to avoid spamming logs and artificially skewing the failure metric.

💡 Proposed fix
 	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 ctx.Err() != nil {
+			return nil, ctx.Err()
+		}
 		if err == nil {
 			err = fmt.Errorf("beacon returned %d sidecars, want at least %d", len(sidecars), len(hashes))
 		}
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
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 == nil {
err = fmt.Errorf("beacon returned %d sidecars, want at least %d", len(sidecars), len(hashes))
}
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 ctx.Err() != nil {
return nil, ctx.Err()
}
if err == nil {
err = fmt.Errorf("beacon returned %d sidecars, want at least %d", len(sidecars), len(hashes))
}
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@node/derivation/beacon.go` around lines 246 - 254, Update the client loop
around GetBlobSidecarsEnhanced to detect context cancellation or deadline expiry
immediately after the call and return the context error without recording a
beacon failure, logging it, or trying additional clients. Preserve the existing
sidecar validation and retry behavior for non-context errors.

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
Expand Down
18 changes: 17 additions & 1 deletion node/derivation/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,22 @@ type Config struct {
ReorgCheckDepth uint64 `json:"reorg_check_depth"`
}

// 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
}

func DefaultConfig() *Config {
return &Config{
L1: &types.L1Config{
Expand Down Expand Up @@ -122,7 +138,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")
}

Expand Down
21 changes: 21 additions & 0 deletions node/derivation/config_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import (
"strings"
"testing"

"github.com/stretchr/testify/require"
"github.com/urfave/cli"

"morph-l2/node/flags"
Expand Down Expand Up @@ -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)
Expand Down
11 changes: 5 additions & 6 deletions node/derivation/derivation.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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()
}
Expand Down
18 changes: 18 additions & 0 deletions node/derivation/metrics.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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")),
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}
}

Expand Down Expand Up @@ -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)
}
2 changes: 1 addition & 1 deletion node/flags/flags.go
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ var (

L1BeaconAddr = cli.StringFlag{
Name: "l1.beaconrpc",
Usage: "Address of L1 Beacon JSON-RPC endpoint to use (eth namespace required)",
Usage: "Address of L1 Beacon JSON-RPC endpoint(s) to use (eth namespace required). Supports a comma-separated list; endpoints are tried in order and derivation falls back to the next one when a beacon temporarily fails to serve blob sidecars",
EnvVar: prefixEnvVar("L1_ETH_BEACON_RPC"),
}

Expand Down
Loading