diff --git a/node/derivation/base_client_test.go b/node/derivation/base_client_test.go new file mode 100644 index 000000000..65bd146ec --- /dev/null +++ b/node/derivation/base_client_test.go @@ -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)) +} diff --git a/node/derivation/beacon.go b/node/derivation/beacon.go index ac663241f..1de364ade 100644 --- a/node/derivation/beacon.go +++ b/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,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)) + } + 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 diff --git a/node/derivation/config.go b/node/derivation/config.go index 29b6f9a39..41af11572 100644 --- a/node/derivation/config.go +++ b/node/derivation/config.go @@ -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{ @@ -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") } diff --git a/node/derivation/config_test.go b/node/derivation/config_test.go index f18819784..8701196a5 100644 --- a/node/derivation/config_test.go +++ b/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) diff --git a/node/derivation/derivation.go b/node/derivation/derivation.go index 4a5ce004a..b49ed1264 100644 --- a/node/derivation/derivation.go +++ b/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() } diff --git a/node/derivation/metrics.go b/node/derivation/metrics.go index dce79ca13..57e180518 100644 --- a/node/derivation/metrics.go +++ b/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) +} diff --git a/node/flags/flags.go b/node/flags/flags.go index 7ee931d48..cb433cb59 100644 --- a/node/flags/flags.go +++ b/node/flags/flags.go @@ -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"), }