diff --git a/Dockerfile b/Dockerfile index b8d358d..d9b8089 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,5 +1,5 @@ # build stage -FROM golang:1.25-alpine AS builder +FROM golang:1.26-alpine AS builder WORKDIR /build diff --git a/README.md b/README.md index 51e9444..ada72ce 100644 --- a/README.md +++ b/README.md @@ -77,6 +77,7 @@ Metrics will be available at `http://localhost:2112/metrics` - `--balance.scrape-interval`: Balance check scrape interval in seconds (default: 30) - `--verifier.workers`: Number of concurrent workers for block verification (default: 50) - `--verbose`: Enable verbose (debug level) logging (default: false) +- `--log-format`: Log output format, `console` (human-readable) or `json` (one JSON object per line, for Loki/ELK ingestion) (default: `console`) ### Example with Custom Endpoints @@ -119,6 +120,8 @@ Monitor native token balances for one or more Celestia addresses: This will query account balances every 30 seconds with automatic failover between RPC endpoints. +`https://` consensus endpoints are dialed over TLS (port 443 if none is given); `http://` and bare `host:port` endpoints use plaintext gRPC. + ## Prometheus Metrics The following metrics are exposed: @@ -276,7 +279,7 @@ When `--balance.addresses` and `--balance.consensus-rpc-urls` are provided: There are two independent debug logging mechanisms: -- `--verbose` flag: sets the application log level to debug. +- `--verbose` flag: sets the application log level to debug. Per-block and per-poll success messages (block verification, drift polls, balance checks, JSON-RPC health checks) are logged at debug level, so use `--verbose` to see them. Warnings and errors are always logged. - `EVMETRICS_DEBUG=true` environment variable: logs DA submission details from the metrics package. **Use cases**: Troubleshoot submission failures, verify submission flow. diff --git a/cmd/monitor.go b/cmd/monitor.go index f45835f..7d6dbc2 100644 --- a/cmd/monitor.go +++ b/cmd/monitor.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "io" "os" "os/signal" "strings" @@ -33,6 +34,7 @@ const ( flagDataNS = "data-namespace" flagDuration = "duration" flagVerbose = "verbose" + flagLogFormat = "log-format" flagPort = "port" flagChain = "chain-id" flagReferenceNode = "reference-node" @@ -59,6 +61,7 @@ type flagValues struct { dataNS string duration int verbose bool + logFormat string port int chainID string referenceNode string @@ -92,6 +95,7 @@ func NewMonitorCmd() *cobra.Command { cmd.Flags().StringVar(&flags.dataNS, flagDataNS, "", "Data namespace (my_app_data_namespace)") cmd.Flags().IntVar(&flags.duration, flagDuration, 0, "Duration in seconds to stream (0 = infinite)") cmd.Flags().BoolVar(&flags.verbose, flagVerbose, false, "Enable verbose logging") + cmd.Flags().StringVar(&flags.logFormat, flagLogFormat, "console", "Log output format: console or json") cmd.Flags().IntVar(&flags.port, flagPort, 2112, "HTTP server port for metrics") cmd.Flags().StringVar(&flags.chainID, flagChain, "testnet", "chainID identifier for metrics labels") cmd.Flags().StringVar(&flags.referenceNode, flagReferenceNode, "", "Reference node RPC endpoint URL (sequencer) for drift monitoring") @@ -159,17 +163,27 @@ func newClients(ctx context.Context, flags flagValues, logger zerolog.Logger) (* return evnodeClient, evmClient, celestiaClient, nil } -func monitorAndExportMetrics(_ *cobra.Command, _ []string) error { - // Setup logger +func newLogger(format string, verbose bool, w io.Writer) (zerolog.Logger, error) { + switch format { + case "console": + w = zerolog.ConsoleWriter{Out: w, TimeFormat: time.RFC3339} + case "json": + default: + return zerolog.Logger{}, fmt.Errorf("invalid --%s %q: must be console or json", flagLogFormat, format) + } + logLevel := zerolog.InfoLevel - if flags.verbose { + if verbose { logLevel = zerolog.DebugLevel } - logger := zerolog.New(zerolog.ConsoleWriter{Out: os.Stdout, TimeFormat: time.RFC3339}). - Level(logLevel). - With(). - Timestamp(). - Logger() + return zerolog.New(w).Level(logLevel).With().Timestamp().Logger(), nil +} + +func monitorAndExportMetrics(_ *cobra.Command, _ []string) error { + logger, err := newLogger(flags.logFormat, flags.verbose, os.Stdout) + if err != nil { + return err + } // setup context with signal handling for graceful shutdown ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) diff --git a/cmd/monitor_test.go b/cmd/monitor_test.go new file mode 100644 index 0000000..9da9fda --- /dev/null +++ b/cmd/monitor_test.go @@ -0,0 +1,54 @@ +package cmd + +import ( + "bytes" + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestNewLogger(t *testing.T) { + t.Run("json", func(t *testing.T) { + var buf bytes.Buffer + logger, err := newLogger("json", false, &buf) + require.NoError(t, err) + + logger.Info().Msg("hello") + + var entry map[string]any + require.NoError(t, json.Unmarshal(buf.Bytes(), &entry)) + require.Equal(t, "info", entry["level"]) + require.Equal(t, "hello", entry["message"]) + require.Contains(t, entry, "time") + }) + + t.Run("console", func(t *testing.T) { + var buf bytes.Buffer + logger, err := newLogger("console", false, &buf) + require.NoError(t, err) + + logger.Info().Msg("hello") + + require.Contains(t, buf.String(), "hello") + require.False(t, json.Valid(buf.Bytes())) + }) + + t.Run("invalid format", func(t *testing.T) { + _, err := newLogger("xml", false, &bytes.Buffer{}) + require.Error(t, err) + }) + + t.Run("verbose enables debug", func(t *testing.T) { + var buf bytes.Buffer + logger, err := newLogger("json", false, &buf) + require.NoError(t, err) + logger.Debug().Msg("hidden") + require.Empty(t, buf.String()) + + logger, err = newLogger("json", true, &buf) + require.NoError(t, err) + logger.Debug().Msg("shown") + require.Contains(t, buf.String(), `"message":"shown"`) + }) +} diff --git a/go.mod b/go.mod index 1b371f2..db0cd1d 100644 --- a/go.mod +++ b/go.mod @@ -1,19 +1,19 @@ module github.com/evstack/ev-metrics -go 1.25.6 +go 1.26.0 require ( connectrpc.com/connect v1.19.1 github.com/celestiaorg/go-square/v3 v3.0.2 github.com/cosmos/cosmos-sdk v0.50.14 - github.com/ethereum/go-ethereum v1.16.5 + github.com/ethereum/go-ethereum v1.17.0 github.com/evstack/ev-node v1.0.0 github.com/prometheus/client_golang v1.23.2 github.com/rs/zerolog v1.34.0 github.com/spf13/cobra v1.10.2 github.com/stretchr/testify v1.11.1 - golang.org/x/sync v0.20.0 - google.golang.org/grpc v1.79.2 + golang.org/x/sync v0.23.0 + google.golang.org/grpc v1.83.1 google.golang.org/protobuf v1.36.11 ) @@ -33,6 +33,7 @@ require ( github.com/DataDog/datadog-go v3.2.0+incompatible // indirect github.com/DataDog/zstd v1.5.5 // indirect github.com/Microsoft/go-winio v0.6.2 // indirect + github.com/ProjectZKM/Ziren/crates/go-runtime/zkvm_runtime v0.0.0-20251001021608-1fe7b43fc4d6 // indirect github.com/StackExchange/wmi v1.2.1 // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/bgentry/speakeasy v0.1.1-0.20220910012023-760eaf8b6816 // indirect @@ -47,7 +48,7 @@ require ( github.com/cockroachdb/pebble v1.1.5 // indirect github.com/cockroachdb/redact v1.1.5 // indirect github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06 // indirect - github.com/cometbft/cometbft v0.38.19 // indirect + github.com/cometbft/cometbft v0.38.21 // indirect github.com/cometbft/cometbft-db v0.14.1 // indirect github.com/consensys/gnark-crypto v0.18.1 // indirect github.com/cosmos/btcutil v1.0.5 // indirect @@ -60,7 +61,6 @@ require ( github.com/cosmos/ics23/go v0.11.0 // indirect github.com/cosmos/ledger-cosmos-go v0.15.0 // indirect github.com/crate-crypto/go-eth-kzg v1.4.0 // indirect - github.com/crate-crypto/go-ipa v0.0.0-20240724233137-53bbb0ceb27a // indirect github.com/danieljoos/wincred v1.1.2 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/deckarep/golang-set/v2 v2.6.0 // indirect @@ -71,8 +71,7 @@ require ( github.com/dustin/go-humanize v1.0.1 // indirect github.com/dvsekhvalnov/jose2go v1.7.0 // indirect github.com/emicklei/dot v1.6.2 // indirect - github.com/ethereum/c-kzg-4844/v2 v2.1.3 // indirect - github.com/ethereum/go-verkle v0.2.2 // indirect + github.com/ethereum/c-kzg-4844/v2 v2.1.5 // indirect github.com/fatih/color v1.18.0 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect github.com/filecoin-project/go-jsonrpc v0.10.1 // indirect @@ -81,6 +80,8 @@ require ( github.com/go-kit/kit v0.13.0 // indirect github.com/go-kit/log v0.2.1 // indirect github.com/go-logfmt/logfmt v0.6.0 // indirect + github.com/go-logr/logr v1.4.3 // indirect + github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.3.0 // indirect github.com/go-viper/mapstructure/v2 v2.4.0 // indirect github.com/godbus/dbus v0.0.0-20190726142602-4481cbc300e2 // indirect @@ -153,20 +154,24 @@ require ( github.com/zondax/ledger-go v1.0.0 // indirect go.etcd.io/bbolt v1.4.0-alpha.0.0.20240404170359-43604f3112c5 // indirect go.opencensus.io v0.24.0 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/otel v1.44.0 // indirect + go.opentelemetry.io/otel/metric v1.44.0 // indirect + go.opentelemetry.io/otel/trace v1.44.0 // indirect go.uber.org/multierr v1.11.0 // indirect go.uber.org/zap v1.27.1 // indirect go.yaml.in/yaml/v2 v2.4.3 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect - golang.org/x/crypto v0.48.0 // indirect + golang.org/x/crypto v0.57.0 // indirect golang.org/x/exp v0.0.0-20260112195511-716be5621a96 // indirect - golang.org/x/net v0.51.0 // indirect - golang.org/x/sys v0.41.0 // indirect - golang.org/x/term v0.40.0 // indirect - golang.org/x/text v0.34.0 // indirect + golang.org/x/net v0.60.0 // indirect + golang.org/x/sys v0.48.0 // indirect + golang.org/x/term v0.46.0 // indirect + golang.org/x/text v0.42.0 // indirect golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect google.golang.org/genproto v0.0.0-20240227224415-6ceb2ff114de // indirect - google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect gopkg.in/yaml.v3 v3.0.1 // indirect gotest.tools/v3 v3.5.2 // indirect nhooyr.io/websocket v1.8.6 // indirect diff --git a/go.sum b/go.sum index 2e203a5..dde1828 100644 --- a/go.sum +++ b/go.sum @@ -36,6 +36,8 @@ github.com/DataDog/zstd v1.5.5/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwS github.com/Knetic/govaluate v3.0.1-0.20171022003610-9aa49832a739+incompatible/go.mod h1:r7JcOSlj0wfOMncg0iLm8Leh48TZaKVeNIfJntJ2wa0= github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= +github.com/ProjectZKM/Ziren/crates/go-runtime/zkvm_runtime v0.0.0-20251001021608-1fe7b43fc4d6 h1:1zYrtlhrZ6/b6SAjLSfKzWtdgqK0U+HtH/VcBWh1BaU= +github.com/ProjectZKM/Ziren/crates/go-runtime/zkvm_runtime v0.0.0-20251001021608-1fe7b43fc4d6/go.mod h1:ioLG6R+5bUSO1oeGSDxOV3FADARuMoytZCSX6MEMQkI= github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo= github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI= github.com/StackExchange/wmi v1.2.1 h1:VIkavFPXSjcnS+O8yTq7NI32k0R5Aj+v39y29VYDOSA= @@ -124,8 +126,8 @@ github.com/cockroachdb/redact v1.1.5/go.mod h1:BVNblN9mBWFyMyqK1k3AAiSxhvhfK2oOZ github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06 h1:zuQyyAKVxetITBuuhv3BI9cMrmStnpT18zmgmTxunpo= github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06/go.mod h1:7nc4anLGjupUW/PeY5qiNYsdNXj7zopG+eqsS7To5IQ= github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd/go.mod h1:sE/e/2PUdi/liOCUjSTXgM1o87ZssimdTWN964YiIeI= -github.com/cometbft/cometbft v0.38.19 h1:vNdtCkvhuwUlrcLPAyigV7lQpmmo+tAq8CsB8gZjEYw= -github.com/cometbft/cometbft v0.38.19/go.mod h1:UCu8dlHqvkAsmAFmWDRWNZJPlu6ya2fTWZlDrWsivwo= +github.com/cometbft/cometbft v0.38.21 h1:qcIJSH9LiwU5s6ZgKR5eRbsLNucbubfraDs5bzgjtOI= +github.com/cometbft/cometbft v0.38.21/go.mod h1:UCu8dlHqvkAsmAFmWDRWNZJPlu6ya2fTWZlDrWsivwo= github.com/cometbft/cometbft-db v0.14.1 h1:SxoamPghqICBAIcGpleHbmoPqy+crij/++eZz3DlerQ= github.com/cometbft/cometbft-db v0.14.1/go.mod h1:KHP1YghilyGV/xjD5DP3+2hyigWx0WTp9X+0Gnx0RxQ= github.com/consensys/gnark-crypto v0.18.1 h1:RyLV6UhPRoYYzaFnPQA4qK3DyuDgkTgskDdoGqFt3fI= @@ -160,8 +162,6 @@ github.com/cpuguy83/go-md2man/v2 v2.0.6 h1:XJtiaUW6dEEqVuZiMTn1ldk455QWwEIsMIJlo github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= github.com/crate-crypto/go-eth-kzg v1.4.0 h1:WzDGjHk4gFg6YzV0rJOAsTK4z3Qkz5jd4RE3DAvPFkg= github.com/crate-crypto/go-eth-kzg v1.4.0/go.mod h1:J9/u5sWfznSObptgfa92Jq8rTswn6ahQWEuiLHOjCUI= -github.com/crate-crypto/go-ipa v0.0.0-20240724233137-53bbb0ceb27a h1:W8mUrRp6NOVl3J+MYp5kPMoUZPp7aOYHtaua31lwRHg= -github.com/crate-crypto/go-ipa v0.0.0-20240724233137-53bbb0ceb27a/go.mod h1:sTwzHBvIzm2RfVCGNEBZgRyjwK40bVoun3ZnGOCafNM= github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7DoTY= github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/danieljoos/wincred v1.1.2 h1:QLdCxFs1/Yl4zduvBdcHB8goaYk9RARS2SgLLRuAyr0= @@ -206,14 +206,12 @@ github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.m github.com/envoyproxy/go-control-plane v0.9.10-0.20210907150352-cf90f659a021/go.mod h1:AFq3mo9L8Lqqiid3OhADV3RfLJnjiw63cSpi+fDTRC0= github.com/envoyproxy/go-control-plane v0.10.2-0.20220325020618-49ff273808a1/go.mod h1:KJwIaB5Mv44NWtYuAOFCVOjcI94vtpEz2JU/D2v6IjE= github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c= -github.com/ethereum/c-kzg-4844/v2 v2.1.3 h1:DQ21UU0VSsuGy8+pcMJHDS0CV1bKmJmxsJYK8l3MiLU= -github.com/ethereum/c-kzg-4844/v2 v2.1.3/go.mod h1:fyNcYI/yAuLWJxf4uzVtS8VDKeoAaRM8G/+ADz/pRdA= +github.com/ethereum/c-kzg-4844/v2 v2.1.5 h1:aVtoLK5xwJ6c5RiqO8g8ptJ5KU+2Hdquf6G3aXiHh5s= +github.com/ethereum/c-kzg-4844/v2 v2.1.5/go.mod h1:u59hRTTah4Co6i9fDWtiCjTrblJv0UwsqZKCc0GfgUs= github.com/ethereum/go-bigmodexpfix v0.0.0-20250911101455-f9e208c548ab h1:rvv6MJhy07IMfEKuARQ9TKojGqLVNxQajaXEp/BoqSk= github.com/ethereum/go-bigmodexpfix v0.0.0-20250911101455-f9e208c548ab/go.mod h1:IuLm4IsPipXKF7CW5Lzf68PIbZ5yl7FFd74l/E0o9A8= -github.com/ethereum/go-ethereum v1.16.5 h1:GZI995PZkzP7ySCxEFaOPzS8+bd8NldE//1qvQDQpe0= -github.com/ethereum/go-ethereum v1.16.5/go.mod h1:kId9vOtlYg3PZk9VwKbGlQmSACB5ESPTBGT+M9zjmok= -github.com/ethereum/go-verkle v0.2.2 h1:I2W0WjnrFUIzzVPwm8ykY+7pL2d4VhlsePn4j7cnFk8= -github.com/ethereum/go-verkle v0.2.2/go.mod h1:M3b90YRnzqKyyzBEWJGqj8Qff4IDeXnzFw0P9bFw3uk= +github.com/ethereum/go-ethereum v1.17.0 h1:2D+1Fe23CwZ5tQoAS5DfwKFNI1HGcTwi65/kRlAVxes= +github.com/ethereum/go-ethereum v1.17.0/go.mod h1:2W3msvdosS/MCWytpqTcqgFiRYbTH59FxDJzqah120o= github.com/evstack/ev-node v1.0.0 h1:m3e51fo4Dk9Z32XRV56GJKEeAiqvjiJ9n3SRjG7C5n8= github.com/evstack/ev-node v1.0.0/go.mod h1:85H7BPvvRoA+uPfCiIcyWMBN728Wv1uLNhOsjLifJgw= github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4= @@ -262,6 +260,7 @@ github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A= github.com/go-logfmt/logfmt v0.6.0 h1:wGYYu3uicYdqXVgoYbvnkrPVXkuLM1p1ifugDMEdRi4= github.com/go-logfmt/logfmt v0.6.0/go.mod h1:WYhtIu8zTZfxdn5+rREduYbwxfcBr/Vr6KEVveWlfTs= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= @@ -382,6 +381,10 @@ github.com/gorilla/websocket v0.0.0-20170926233335-4201258b820c/go.mod h1:E7qHFY github.com/gorilla/websocket v1.4.1/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/grafana/pyroscope-go v1.2.7 h1:VWBBlqxjyR0Cwk2W6UrE8CdcdD80GOFNutj0Kb1T8ac= +github.com/grafana/pyroscope-go v1.2.7/go.mod h1:o/bpSLiJYYP6HQtvcoVKiE9s5RiNgjYTj1DhiddP2Pc= +github.com/grafana/pyroscope-go/godeltaprof v0.1.9 h1:c1Us8i6eSmkW+Ez05d3co8kasnuOY813tbMN8i/a3Og= +github.com/grafana/pyroscope-go/godeltaprof v0.1.9/go.mod h1:2+l7K7twW49Ct4wFluZD3tZ6e0SjanjcUUBPVD/UuGU= github.com/grpc-ecosystem/go-grpc-middleware v1.0.1-0.20190118093823-f849b5445de4/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs= github.com/grpc-ecosystem/go-grpc-middleware v1.2.2/go.mod h1:EaizFBKfUKtMIF5iaDEhniwNedqGo9FuLFzppDr3uwI= github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 h1:UH//fgunKIs4JdUbpDl1VZCDaL56wXCB/5+wF6uHfaI= @@ -532,8 +535,6 @@ github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/mattn/go-runewidth v0.0.2/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU= -github.com/mattn/go-runewidth v0.0.13 h1:lTGmDsbAYt5DmK6OnoV7EuIF1wEIFAcxld6ypU4OSgU= -github.com/mattn/go-runewidth v0.0.13/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w= github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg= github.com/minio/highwayhash v1.0.3 h1:kbnuUMoHYyVl7szWjSxJnxw11k2U709jqFPPmIUyD6Q= @@ -586,8 +587,6 @@ github.com/oklog/run v1.0.0/go.mod h1:dlhp/R75TPv97u0XWUtDeV/lRKWPKSdTuV0TZvrmrQ github.com/oklog/run v1.1.0 h1:GEenZ1cK0+q0+wsJew9qUg/DyD8k3JzYsZAi5gYi2mA= github.com/oklog/run v1.1.0/go.mod h1:sVPdnTZT1zYwAJeCMu2Th4T21pA3FPOQRfWjQlk7DVU= github.com/olekukonko/tablewriter v0.0.0-20170122224234-a0225b3f23b5/go.mod h1:vsDQFd/mU46D+Z4whnwzcISnGGzXWMclvtLoiIKAKIo= -github.com/olekukonko/tablewriter v0.0.5 h1:P2Ga83D34wi1o9J6Wh1mRuqd4mF/x/lgBS7N7AbDhec= -github.com/olekukonko/tablewriter v0.0.5/go.mod h1:hPp6KlRPjbx+hW8ykQs1w3UBbZlj6HuIJcUGPhkA7kY= github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk= @@ -684,8 +683,6 @@ github.com/prometheus/procfs v0.17.0/go.mod h1:oPQLaDAMRbA+u8H5Pbfq+dl3VDAvHxMUO github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4= github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 h1:N/ElC8H3+5XpJzTSTfLsJV/mx9Q9g7kxmchpfZyxgzM= github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4= -github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY= -github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc= github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg= github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ= github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4= @@ -810,16 +807,16 @@ go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0= go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= -go.opentelemetry.io/otel v1.42.0 h1:lSQGzTgVR3+sgJDAU/7/ZMjN9Z+vUip7leaqBKy4sho= -go.opentelemetry.io/otel v1.42.0/go.mod h1:lJNsdRMxCUIWuMlVJWzecSMuNjE7dOYyWlqOXWkdqCc= -go.opentelemetry.io/otel/metric v1.42.0 h1:2jXG+3oZLNXEPfNmnpxKDeZsFI5o4J+nz6xUlaFdF/4= -go.opentelemetry.io/otel/metric v1.42.0/go.mod h1:RlUN/7vTU7Ao/diDkEpQpnz3/92J9ko05BIwxYa2SSI= -go.opentelemetry.io/otel/sdk v1.42.0 h1:LyC8+jqk6UJwdrI/8VydAq/hvkFKNHZVIWuslJXYsDo= -go.opentelemetry.io/otel/sdk v1.42.0/go.mod h1:rGHCAxd9DAph0joO4W6OPwxjNTYWghRWmkHuGbayMts= -go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2WKg+sEJTtB8= -go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew= -go.opentelemetry.io/otel/trace v1.42.0 h1:OUCgIPt+mzOnaUTpOQcBiM/PLQ/Op7oq6g4LenLmOYY= -go.opentelemetry.io/otel/trace v1.42.0/go.mod h1:f3K9S+IFqnumBkKhRJMeaZeNk9epyhnCmQh/EysQCdc= +go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= +go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= +go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= +go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= +go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= +go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= +go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= +go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= +go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= +go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI= go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= @@ -851,8 +848,8 @@ golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8U golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20200728195943-123391ffb6de/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= -golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= -golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20200331195152-e8c3332aa8e5/go.mod h1:4M0jN8W1tt0AVLNr8HDosyJCDCDuyL9N9+3m7wDWgKw= @@ -899,8 +896,8 @@ golang.org/x/net v0.0.0-20210428140749-89ef3d95e781/go.mod h1:OJAsFXCWl8Ukc7SiCT golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk= golang.org/x/net v0.0.0-20220607020251-c690dde0001d/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= -golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo= -golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y= +golang.org/x/net v0.60.0 h1:79p50tfZlm0J9YfoDsSi639qSXNGVwEzOPLCxM2FsYU= +golang.org/x/net v0.60.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= @@ -912,8 +909,8 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= -golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= +golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -971,20 +968,20 @@ golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k= -golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= -golang.org/x/term v0.40.0 h1:36e4zGLqU4yhjlmxEaagx2KuYbJq3EwY8K943ZsHcvg= -golang.org/x/term v0.40.0/go.mod h1:w2P8uVp06p2iyKKuvXIm7N/y0UCRt3UfJTfZ7oOpglM= +golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE= +golang.org/x/term v0.46.0/go.mod h1:+K02xbkittuwc0Am4abfA3Fc+XRGXkvBXNO88NCXPoc= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= -golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk= -golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= +golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= +golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE= @@ -1035,10 +1032,10 @@ google.golang.org/genproto v0.0.0-20210126160654-44e461bb6506/go.mod h1:FWY/as6D google.golang.org/genproto v0.0.0-20220314164441-57ef72a4c106/go.mod h1:hAL49I2IFola2sVEjAn7MEwsja0xp51I0tlGAf9hz4E= google.golang.org/genproto v0.0.0-20240227224415-6ceb2ff114de h1:F6qOa9AZTYJXOUEr4jDysRDLrm4PHePlge4v4TGAlxY= google.golang.org/genproto v0.0.0-20240227224415-6ceb2ff114de/go.mod h1:VUhTRKeHn9wwcdrk73nvdC9gF178Tzhmt/qyaFcPLSo= -google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 h1:JLQynH/LBHfCTSbDWl+py8C+Rg/k1OVH3xfcaiANuF0= -google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57/go.mod h1:kSJwQxqmFXeo79zOmbrALdflXQeAYcUbgS7PbpMknCY= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 h1:mWPCjDEyshlQYzBpMNHaEof6UX1PmHcaUODUywQ0uac= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57/go.mod h1:j9x/tPzZkyxcgEFkiKEEGxfvyumM01BEtsW8xzOahRQ= +google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:Kjn0N0tCrDgiAFW+lGO4JZ3ck44CehvJQMAwj9QF0G8= +google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:q4lMZS6kskjT5HvCPrnnypcDPVJqT/f4nfxmkE7gryY= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= google.golang.org/grpc v1.17.0/go.mod h1:6QZJwpn2B+Zp71q/5VxRsJ6NXXVCE5NRUHRo+f3cWCs= google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c= google.golang.org/grpc v1.20.0/go.mod h1:chYK+tFQF0nDUGJgXMSgLCQk3phJEuONr2DCgLDdAQM= @@ -1057,8 +1054,8 @@ google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv google.golang.org/grpc v1.36.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU= google.golang.org/grpc v1.45.0/go.mod h1:lN7owxKUQEqMfSyQikvvk5tf/6zMPsrK+ONuO11+0rQ= google.golang.org/grpc v1.49.0/go.mod h1:ZgQEeidpAuNRZ8iRrlBKXZQP1ghovWIVhdJRyCDK+GI= -google.golang.org/grpc v1.79.2 h1:fRMD94s2tITpyJGtBBn7MkMseNpOZU8ZxgC3MMBaXRU= -google.golang.org/grpc v1.79.2/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ= +google.golang.org/grpc v1.83.1 h1:HIO0+BEtBP6soyqvqC8sNUjZ7bTs+0hFQuFF+RAy++Y= +google.golang.org/grpc v1.83.1/go.mod h1:kDyl6SKsiHKt0uylY5gtn5cEjkrIOhQOGDgIc4JGwzQ= google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8= google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0= google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM= diff --git a/internal/clients/celestia/client.go b/internal/clients/celestia/client.go index 4338a34..d1e4459 100644 --- a/internal/clients/celestia/client.go +++ b/internal/clients/celestia/client.go @@ -8,21 +8,96 @@ import ( "io" "net/http" "strings" + "sync" "time" libshare "github.com/celestiaorg/go-square/v3/share" "github.com/evstack/ev-node/pkg/da/jsonrpc" evnode "github.com/evstack/ev-node/types/pb/evnode/v1" "github.com/rs/zerolog" + "golang.org/x/sync/singleflight" "google.golang.org/protobuf/proto" ) +const ( + // defaultCacheTTL keeps per-DA-height lookups long enough for every block in an + // ev-node batch to be verified against a single fetch. + defaultCacheTTL = 2 * time.Minute + // sharedFetchTimeout bounds a deduplicated fetch, which runs detached from the + // cancellation of whichever caller started it. + sharedFetchTimeout = 30 * time.Second +) + // Client is a small wrapper around the evnode celestia jsonrpc client. type Client struct { *jsonrpc.Client logger zerolog.Logger url string token string + + // getAll defaults to Blob.GetAll; overridable in tests. + getAll func(ctx context.Context, height uint64, namespaces []libshare.Namespace) ([]*jsonrpc.Blob, error) + + // simplify: plain maps swept on insert, no LRU. Memory is bounded by + // cacheTTL × DA heights in flight; switch to golang-lru/expirable if that grows. + cacheTTL time.Duration + mu sync.Mutex + timestamps map[uint64]cacheEntry[time.Time] + blobs map[blobKey]cacheEntry[[][]byte] + inflight singleflight.Group +} + +type blobKey struct { + height uint64 + namespace string +} + +type cacheEntry[V any] struct { + value V + expires time.Time +} + +func cacheGet[K comparable, V any](mu *sync.Mutex, m *map[K]cacheEntry[V], k K) (V, bool) { + mu.Lock() + defer mu.Unlock() + e, ok := (*m)[k] + if !ok || time.Now().After(e.expires) { + var zero V + return zero, false + } + return e.value, true +} + +func cachePut[K comparable, V any](mu *sync.Mutex, m *map[K]cacheEntry[V], k K, v V, ttl time.Duration) { + mu.Lock() + defer mu.Unlock() + now := time.Now() + if *m == nil { + *m = make(map[K]cacheEntry[V]) + } + for key, e := range *m { + if now.After(e.expires) { + delete(*m, key) + } + } + (*m)[k] = cacheEntry[V]{value: v, expires: now.Add(ttl)} +} + +// shared runs fetch once per key across concurrent callers. The fetch is detached +// from ctx cancellation (bounded by sharedFetchTimeout) so one caller giving up +// does not fail the others; each caller still returns as soon as its own ctx is done. +func (c *Client) shared(ctx context.Context, key string, fetch func(context.Context) (any, error)) (any, error) { + ch := c.inflight.DoChan(key, func() (any, error) { + fetchCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), sharedFetchTimeout) + defer cancel() + return fetch(fetchCtx) + }) + select { + case r := <-ch: + return r.Val, r.Err + case <-ctx.Done(): + return nil, ctx.Err() + } } func NewClient(ctx context.Context, url, token string, logger zerolog.Logger) (*Client, error) { @@ -33,21 +108,44 @@ func NewClient(ctx context.Context, url, token string, logger zerolog.Logger) (* } return &Client{ - Client: client, - logger: logger.With().Str("component", "celestia_client").Logger(), - url: url, - token: token, + Client: client, + logger: logger.With().Str("component", "celestia_client").Logger(), + url: url, + token: token, + getAll: client.Blob.GetAll, + cacheTTL: defaultCacheTTL, }, nil } -// GetBlobsAtHeight retrieves all blobs at a specific DA height and namespace +// GetBlobsAtHeight retrieves all blobs at a specific DA height and namespace. +// Non-empty results are memoized for cacheTTL; the returned slices are shared and +// must not be modified. func (c *Client) GetBlobsAtHeight(ctx context.Context, daHeight uint64, namespace []byte) ([][]byte, error) { ns, err := libshare.NewNamespaceFromBytes(namespace) if err != nil { return nil, fmt.Errorf("invalid namespace: %w", err) } - blobs, err := c.Blob.GetAll(ctx, daHeight, []libshare.Namespace{ns}) + key := blobKey{height: daHeight, namespace: string(namespace)} + if blobs, ok := cacheGet(&c.mu, &c.blobs, key); ok { + return blobs, nil + } + + v, err := c.shared(ctx, fmt.Sprintf("blobs/%d/%x", daHeight, namespace), func(ctx context.Context) (any, error) { + blobs, err := c.fetchBlobs(ctx, daHeight, ns) + if err == nil && len(blobs) > 0 { + cachePut(&c.mu, &c.blobs, key, blobs, c.cacheTTL) + } + return blobs, err + }) + if err != nil { + return nil, err + } + return v.([][]byte), nil +} + +func (c *Client) fetchBlobs(ctx context.Context, daHeight uint64, ns libshare.Namespace) ([][]byte, error) { + blobs, err := c.getAll(ctx, daHeight, []libshare.Namespace{ns}) if err != nil { if strings.Contains(err.Error(), "blob: not found") { return nil, nil @@ -98,7 +196,7 @@ func (c *Client) VerifyBlobAtHeight(ctx context.Context, blob []byte, daHeight u } if result != nil { - c.logger.Info(). + c.logger.Debug(). Str("commitment", fmt.Sprintf("%x", b.Commitment)). Msg("blob verified on Celestia") } @@ -165,7 +263,7 @@ func (c *Client) VerifyDataBlobAtHeight(ctx context.Context, unwrappedDataBlob [ Msg("comparing unwrapped Data") if equal { - c.logger.Info(). + c.logger.Debug(). Int("blob_index", i). Msg("found matching Data") return true, nil @@ -196,8 +294,27 @@ type RPCResponse struct { Error *RPCError `json:"error"` } -// GetBlockTimestamp retrieves the timestamp of a celestia da block at the given height +// GetBlockTimestamp retrieves the timestamp of a celestia da block at the given height. +// Successful lookups are memoized for cacheTTL. func (c *Client) GetBlockTimestamp(ctx context.Context, daHeight uint64) (time.Time, error) { + if ts, ok := cacheGet(&c.mu, &c.timestamps, daHeight); ok { + return ts, nil + } + + v, err := c.shared(ctx, fmt.Sprintf("timestamp/%d", daHeight), func(ctx context.Context) (any, error) { + ts, err := c.fetchBlockTimestamp(ctx, daHeight) + if err == nil { + cachePut(&c.mu, &c.timestamps, daHeight, ts, c.cacheTTL) + } + return ts, err + }) + if err != nil { + return time.Time{}, err + } + return v.(time.Time), nil +} + +func (c *Client) fetchBlockTimestamp(ctx context.Context, daHeight uint64) (time.Time, error) { payload := map[string]interface{}{ "jsonrpc": "2.0", "id": 1, diff --git a/internal/clients/celestia/client_test.go b/internal/clients/celestia/client_test.go index ae82180..bf52d82 100644 --- a/internal/clients/celestia/client_test.go +++ b/internal/clients/celestia/client_test.go @@ -2,11 +2,18 @@ package celestia import ( "context" + "errors" "net/http" "net/http/httptest" + "sync" + "sync/atomic" "testing" + "time" + libshare "github.com/celestiaorg/go-square/v3/share" + "github.com/evstack/ev-node/pkg/da/jsonrpc" "github.com/rs/zerolog" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -42,3 +49,168 @@ func TestGetBlockTimestamp_AuthorizationHeader(t *testing.T) { }) } } + +const okHeader = `{"result":{"header":{"time":"2024-01-02T03:04:05Z"}}}` + +func newTimestampClient(t *testing.T, handler func(w http.ResponseWriter, call int64)) (*Client, *atomic.Int64) { + t.Helper() + var hits atomic.Int64 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + handler(w, hits.Add(1)) + })) + t.Cleanup(srv.Close) + return &Client{logger: zerolog.Nop(), url: srv.URL, cacheTTL: time.Minute}, &hits +} + +func TestGetBlockTimestamp_Cache(t *testing.T) { + t.Run("same height fetched once", func(t *testing.T) { + c, hits := newTimestampClient(t, func(w http.ResponseWriter, _ int64) { + time.Sleep(20 * time.Millisecond) + _, _ = w.Write([]byte(okHeader)) + }) + + var wg sync.WaitGroup + for range 20 { + wg.Add(1) + go func() { + defer wg.Done() + _, err := c.GetBlockTimestamp(context.Background(), 7) + assert.NoError(t, err) + }() + } + wg.Wait() + for range 5 { + ts, err := c.GetBlockTimestamp(context.Background(), 7) + require.NoError(t, err) + require.Equal(t, time.Date(2024, 1, 2, 3, 4, 5, 0, time.UTC), ts) + } + require.EqualValues(t, 1, hits.Load()) + + _, err := c.GetBlockTimestamp(context.Background(), 8) + require.NoError(t, err) + require.EqualValues(t, 2, hits.Load()) + }) + + t.Run("http error not cached", func(t *testing.T) { + c, hits := newTimestampClient(t, func(w http.ResponseWriter, call int64) { + if call == 1 { + w.WriteHeader(http.StatusInternalServerError) + return + } + _, _ = w.Write([]byte(okHeader)) + }) + _, err := c.GetBlockTimestamp(context.Background(), 1) + require.Error(t, err) + _, err = c.GetBlockTimestamp(context.Background(), 1) + require.NoError(t, err) + require.EqualValues(t, 2, hits.Load()) + }) + + t.Run("rpc error not cached", func(t *testing.T) { + c, hits := newTimestampClient(t, func(w http.ResponseWriter, call int64) { + if call == 1 { + _, _ = w.Write([]byte(`{"error":{"code":1,"message":"header: not found"}}`)) + return + } + _, _ = w.Write([]byte(okHeader)) + }) + _, err := c.GetBlockTimestamp(context.Background(), 1) + require.Error(t, err) + _, err = c.GetBlockTimestamp(context.Background(), 1) + require.NoError(t, err) + require.EqualValues(t, 2, hits.Load()) + }) + + t.Run("expired entry refetched", func(t *testing.T) { + c, hits := newTimestampClient(t, func(w http.ResponseWriter, _ int64) { + _, _ = w.Write([]byte(okHeader)) + }) + c.cacheTTL = 10 * time.Millisecond + _, err := c.GetBlockTimestamp(context.Background(), 1) + require.NoError(t, err) + time.Sleep(20 * time.Millisecond) + _, err = c.GetBlockTimestamp(context.Background(), 1) + require.NoError(t, err) + require.EqualValues(t, 2, hits.Load()) + }) + + t.Run("first caller cancel does not fail others", func(t *testing.T) { + started, release := make(chan struct{}), make(chan struct{}) + c, hits := newTimestampClient(t, func(w http.ResponseWriter, _ int64) { + close(started) + <-release + _, _ = w.Write([]byte(okHeader)) + }) + + ctx, cancel := context.WithCancel(context.Background()) + firstErr := make(chan error, 1) + go func() { + _, err := c.GetBlockTimestamp(ctx, 1) + firstErr <- err + }() + <-started + + secondErr := make(chan error, 1) + go func() { + _, err := c.GetBlockTimestamp(context.Background(), 1) + secondErr <- err + }() + + cancel() + require.ErrorIs(t, <-firstErr, context.Canceled) + close(release) + require.NoError(t, <-secondErr) + require.EqualValues(t, 1, hits.Load()) + }) +} + +func TestGetBlobsAtHeight_Cache(t *testing.T) { + ns := libshare.MustNewV0Namespace([]byte("ev-metrics")) + blob, err := jsonrpc.NewBlobV0(ns, []byte("payload")) + require.NoError(t, err) + + newClient := func(results ...func() ([]*jsonrpc.Blob, error)) (*Client, *atomic.Int64) { + var calls atomic.Int64 + return &Client{ + logger: zerolog.Nop(), + cacheTTL: time.Minute, + getAll: func(context.Context, uint64, []libshare.Namespace) ([]*jsonrpc.Blob, error) { + n := calls.Add(1) + return results[min(int(n), len(results))-1]() + }, + }, &calls + } + found := func() ([]*jsonrpc.Blob, error) { return []*jsonrpc.Blob{blob}, nil } + + t.Run("repeat calls fetch once", func(t *testing.T) { + c, calls := newClient(found) + for range 10 { + blobs, err := c.GetBlobsAtHeight(context.Background(), 5, ns.Bytes()) + require.NoError(t, err) + require.Equal(t, [][]byte{[]byte("payload")}, blobs) + } + require.EqualValues(t, 1, calls.Load()) + + _, err := c.GetBlobsAtHeight(context.Background(), 6, ns.Bytes()) + require.NoError(t, err) + require.EqualValues(t, 2, calls.Load()) + }) + + tests := map[string]func() ([]*jsonrpc.Blob, error){ + "not found": func() ([]*jsonrpc.Blob, error) { return nil, errors.New("blob: not found") }, + "empty": func() ([]*jsonrpc.Blob, error) { return []*jsonrpc.Blob{}, nil }, + "other error": func() ([]*jsonrpc.Blob, error) { return nil, errors.New("connection refused") }, + } + for name, first := range tests { + t.Run(name+" not cached", func(t *testing.T) { + c, calls := newClient(first, found) + blobs, _ := c.GetBlobsAtHeight(context.Background(), 5, ns.Bytes()) + require.Empty(t, blobs) + + blobs, err := c.GetBlobsAtHeight(context.Background(), 5, ns.Bytes()) + require.NoError(t, err) + require.Len(t, blobs, 1) + require.EqualValues(t, 2, calls.Load()) + }) + } +} diff --git a/internal/clients/cosmos/client.go b/internal/clients/cosmos/client.go index b50707b..0c410be 100644 --- a/internal/clients/cosmos/client.go +++ b/internal/clients/cosmos/client.go @@ -2,13 +2,16 @@ package cosmos import ( "context" + "crypto/tls" "fmt" + "net" "strings" "github.com/cosmos/cosmos-sdk/types" banktypes "github.com/cosmos/cosmos-sdk/x/bank/types" "github.com/rs/zerolog" "google.golang.org/grpc" + "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/insecure" ) @@ -20,25 +23,40 @@ type Client struct { logger zerolog.Logger } -// stripScheme removes http:// or https:// prefix from endpoint -func stripScheme(endpoint string) string { - endpoint = strings.TrimPrefix(endpoint, "http://") - endpoint = strings.TrimPrefix(endpoint, "https://") - return endpoint +// parseEndpoint converts an endpoint URL into a gRPC "host:port" target and reports whether TLS +// should be used. https:// endpoints use TLS and default to port 443; http:// and bare host:port +// endpoints are plaintext. Any path is dropped. +func parseEndpoint(endpoint string) (target string, useTLS bool) { + useTLS = strings.HasPrefix(endpoint, "https://") + target = strings.TrimPrefix(strings.TrimPrefix(endpoint, "https://"), "http://") + if i := strings.IndexByte(target, '/'); i >= 0 { + target = target[:i] + } + if useTLS { + if _, _, err := net.SplitHostPort(target); err != nil { + target = net.JoinHostPort(strings.Trim(target, "[]"), "443") + } + } + return target, useTLS } // NewClient creates a new Cosmos SDK client connected to the given endpoint func NewClient(endpoint string, logger zerolog.Logger) (*Client, error) { - // grpc expects "host:port" not "http://host:port" - grpcEndpoint := stripScheme(endpoint) + grpcEndpoint, useTLS := parseEndpoint(endpoint) + + creds := insecure.NewCredentials() + if useTLS { + creds = credentials.NewTLS(&tls.Config{MinVersion: tls.VersionTLS12}) + } - conn, err := grpc.NewClient(grpcEndpoint, grpc.WithTransportCredentials(insecure.NewCredentials())) + conn, err := grpc.NewClient(grpcEndpoint, grpc.WithTransportCredentials(creds)) if err != nil { return nil, fmt.Errorf("failed to connect to %s: %w", grpcEndpoint, err) } logger.Debug(). Str("endpoint", grpcEndpoint). + Bool("tls", useTLS). Msg("established connection to cosmos endpoint") return &Client{ diff --git a/internal/clients/cosmos/client_test.go b/internal/clients/cosmos/client_test.go new file mode 100644 index 0000000..2838b5c --- /dev/null +++ b/internal/clients/cosmos/client_test.go @@ -0,0 +1,28 @@ +package cosmos + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestParseEndpoint(t *testing.T) { + tests := []struct { + endpoint string + target string + useTLS bool + }{ + {"http://h:9090", "h:9090", false}, + {"https://h", "h:443", true}, + {"https://h:443/path", "h:443", true}, + {"https://h:9443", "h:9443", true}, + {"h:9090", "h:9090", false}, + } + for _, tt := range tests { + t.Run(tt.endpoint, func(t *testing.T) { + target, useTLS := parseEndpoint(tt.endpoint) + require.Equal(t, tt.target, target) + require.Equal(t, tt.useTLS, useTLS) + }) + } +} diff --git a/pkg/exporters/balance/balance.go b/pkg/exporters/balance/balance.go index 9219097..b7b2198 100644 --- a/pkg/exporters/balance/balance.go +++ b/pkg/exporters/balance/balance.go @@ -13,6 +13,8 @@ import ( var _ metrics.Exporter = &exporter{} +const defaultRPCTimeout = 5 * time.Second + // NewMetricsExporter creates a new balance monitoring exporter func NewMetricsExporter( chainID string, @@ -26,6 +28,7 @@ func NewMetricsExporter( addresses: addresses, endpoints: endpoints, scrapeInterval: scrapeInterval, + rpcTimeout: defaultRPCTimeout, logger: logger.With().Str("component", "balance_monitor").Logger(), } } @@ -36,6 +39,7 @@ type exporter struct { addresses []string endpoints []string scrapeInterval int + rpcTimeout time.Duration logger zerolog.Logger } @@ -44,6 +48,16 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error ticker := time.NewTicker(time.Duration(e.scrapeInterval) * time.Second) defer ticker.Stop() + // one client per endpoint, created on first use and shared across scrapes and addresses + clients := make([]*cosmos.Client, len(e.endpoints)) + defer func() { + for _, c := range clients { + if c != nil { + _ = c.Close() + } + } + }() + e.logger.Info(). Strs("addresses", e.addresses). Strs("endpoints", e.endpoints). @@ -51,40 +65,41 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error Msg("starting balance monitoring") for { + for _, address := range e.addresses { + e.checkBalance(ctx, m, clients, address) + } + select { case <-ctx.Done(): e.logger.Info().Msg("stopping balance monitoring") return ctx.Err() case <-ticker.C: - for _, address := range e.addresses { - e.checkBalance(ctx, m, address) - } } } } // checkBalance queries balance with fallback across endpoints -func (e *exporter) checkBalance(ctx context.Context, m *metrics.Metrics, address string) { +func (e *exporter) checkBalance(ctx context.Context, m *metrics.Metrics, clients []*cosmos.Client, address string) { // try each endpoint until one succeeds - for _, endpoint := range e.endpoints { - // create client - client, err := cosmos.NewClient(endpoint, e.logger) - if err != nil { - e.logger.Warn(). - Err(err). - Str("endpoint", endpoint). - Str("address", address). - Msg("failed to create client, trying next endpoint") - m.RecordConsensusRpcEndpointAvailability(e.chainID, endpoint, false) - m.RecordConsensusRpcEndpointError(e.chainID, endpoint, utils.CategorizeError(err)) - continue + for i, endpoint := range e.endpoints { + if clients[i] == nil { + client, err := cosmos.NewClient(endpoint, e.logger) + if err != nil { + e.logger.Warn(). + Err(err). + Str("endpoint", endpoint). + Str("address", address). + Msg("failed to create client, trying next endpoint") + m.RecordConsensusRpcEndpointAvailability(e.chainID, endpoint, false) + m.RecordConsensusRpcEndpointError(e.chainID, endpoint, utils.CategorizeError(err)) + continue + } + clients[i] = client } - // query balance - balances, err := client.GetBalance(ctx, address) - - // close client - _ = client.Close() + callCtx, cancel := context.WithTimeout(ctx, e.rpcTimeout) + balances, err := clients[i].GetBalance(callCtx, address) + cancel() if err != nil { e.logger.Warn(). @@ -116,7 +131,7 @@ func (e *exporter) checkBalance(ctx context.Context, m *metrics.Metrics, address balanceFloat, _ := amount.Float64() m.RecordAccountBalance(e.chainID, address, coin.Denom, balanceFloat) - e.logger.Info(). + e.logger.Debug(). Str("address", address). Str("denom", coin.Denom). Float64("balance", balanceFloat). diff --git a/pkg/exporters/balance/balance_test.go b/pkg/exporters/balance/balance_test.go new file mode 100644 index 0000000..ab83255 --- /dev/null +++ b/pkg/exporters/balance/balance_test.go @@ -0,0 +1,99 @@ +package balance + +import ( + "context" + "net" + "sync/atomic" + "testing" + "time" + + sdk "github.com/cosmos/cosmos-sdk/types" + banktypes "github.com/cosmos/cosmos-sdk/x/bank/types" + "github.com/evstack/ev-metrics/pkg/metrics" + "github.com/prometheus/client_golang/prometheus" + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" +) + +type bankServer struct { + banktypes.UnimplementedQueryServer +} + +func (*bankServer) AllBalances(_ context.Context, req *banktypes.QueryAllBalancesRequest) (*banktypes.QueryAllBalancesResponse, error) { + amount := int64(100) + if req.Address == "addr2" { + amount = 200 + } + return &banktypes.QueryAllBalancesResponse{Balances: sdk.NewCoins(sdk.NewInt64Coin("utia", amount))}, nil +} + +type countingListener struct { + net.Listener + accepted atomic.Int32 +} + +func (l *countingListener) Accept() (net.Conn, error) { + c, err := l.Listener.Accept() + if err == nil { + l.accepted.Add(1) + } + return c, err +} + +func TestExportMetrics_FailoverAndClientReuse(t *testing.T) { + inner, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + lis := &countingListener{Listener: inner} + srv := grpc.NewServer() + banktypes.RegisterQueryServer(srv, &bankServer{}) + go func() { _ = srv.Serve(lis) }() + t.Cleanup(srv.Stop) + + dead, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + deadAddr := dead.Addr().String() + require.NoError(t, dead.Close()) + + reg := prometheus.NewRegistry() + m := metrics.NewWithRegistry("test", reg) + e := NewMetricsExporter("chain", []string{"addr1", "addr2"}, []string{deadAddr, "http://" + lis.Addr().String()}, 60, zerolog.Nop()).(*exporter) + e.rpcTimeout = time.Second + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + _ = e.ExportMetrics(ctx, m) + }() + t.Cleanup(func() { + cancel() + <-done + }) + + balances := func() map[string]float64 { + families, err := reg.Gather() + require.NoError(t, err) + out := map[string]float64{} + for _, mf := range families { + if mf.GetName() != "test_account_balance" { + continue + } + for _, metric := range mf.GetMetric() { + for _, l := range metric.GetLabel() { + if l.GetName() == "address" { + out[l.GetValue()] = metric.GetGauge().GetValue() + } + } + } + } + return out + } + + // the first scrape runs immediately rather than after the 60s interval + require.Eventually(t, func() bool { + return len(balances()) == 2 + }, 3*time.Second, 20*time.Millisecond) + require.Equal(t, map[string]float64{"addr1": 100, "addr2": 200}, balances()) + require.Equal(t, int32(1), lis.accepted.Load(), "both addresses must share one connection") +} diff --git a/pkg/exporters/drift/drift.go b/pkg/exporters/drift/drift.go index 02a595c..b28dd59 100644 --- a/pkg/exporters/drift/drift.go +++ b/pkg/exporters/drift/drift.go @@ -3,15 +3,19 @@ package drift import ( "context" "fmt" + "sync" + "time" + "github.com/ethereum/go-ethereum/ethclient" "github.com/evstack/ev-metrics/pkg/metrics" "github.com/evstack/ev-metrics/pkg/utils" "github.com/rs/zerolog" - "time" ) var _ metrics.Exporter = &exporter{} +const defaultRPCTimeout = 5 * time.Second + // NewMetricsExporter creates and returns a metrics exporter for monitoring block height drift across multiple nodes. func NewMetricsExporter(chainID, referenceNode string, fullNodes []string, pollingInterval int, logger zerolog.Logger) metrics.Exporter { return &exporter{ @@ -19,6 +23,7 @@ func NewMetricsExporter(chainID, referenceNode string, fullNodes []string, polli referenceNode: referenceNode, fullNodes: fullNodes, pollingInterval: pollingInterval, + rpcTimeout: defaultRPCTimeout, logger: logger.With().Str("component", "drift_monitor").Logger(), } } @@ -29,16 +34,43 @@ type exporter struct { referenceNode string fullNodes []string pollingInterval int + rpcTimeout time.Duration logger zerolog.Logger lastHeight uint64 } +// node holds a lazily dialed client for one endpoint. Polls never overlap, so a node is +// only ever used by one goroutine at a time. +type node struct { + url string + client *ethclient.Client +} + +func (n *node) close() { + if n.client != nil { + n.client.Close() + n.client = nil + } +} + // ExportMetrics continuously checks block heights of a reference node and multiple full nodes, // recording the heights and their drift into the provided metrics instance at specified intervals. func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error { ticker := time.NewTicker(time.Duration(e.pollingInterval) * time.Second) defer ticker.Stop() + ref := &node{url: e.referenceNode} + fullNodes := make([]*node, len(e.fullNodes)) + for i, url := range e.fullNodes { + fullNodes[i] = &node{url: url} + } + defer func() { + ref.close() + for _, n := range fullNodes { + n.close() + } + }() + e.logger.Info(). Str("reference_node", e.referenceNode). Strs("full_nodes", e.fullNodes). @@ -46,67 +78,87 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error Msg("starting node drift monitoring") for { + e.poll(ctx, m, ref, fullNodes) + select { case <-ctx.Done(): e.logger.Info().Msg("stopping node drift monitoring") return ctx.Err() case <-ticker.C: - // get reference node height - refHeight, err := getBlockHeight(ctx, e.referenceNode) - if err != nil { - e.logger.Error().Err(err).Str("endpoint", e.referenceNode).Msg("failed to get reference node block height") - m.RecordEndpointAvailability(e.chainID, e.referenceNode, false) - m.RecordEndpointError(e.chainID, e.referenceNode, utils.CategorizeError(err)) - continue - } - - m.RecordEndpointAvailability(e.chainID, e.referenceNode, true) - m.RecordReferenceBlockHeight(e.chainID, e.referenceNode, refHeight) - // Also update time since last block metric when we see new blocks via polling - // This ensures the metric stays accurate even if WebSocket subscription fails - if refHeight > e.lastHeight { - m.UpdateLastBlockTime(e.chainID, time.Now()) - e.lastHeight = refHeight - } - e.logger.Info().Uint64("height", refHeight).Str("endpoint", e.referenceNode).Msg("recorded reference node height") - - // get each full node height and calculate drift - for _, fullNode := range e.fullNodes { - currentHeight, err := getBlockHeight(ctx, fullNode) - if err != nil { - e.logger.Error().Err(err).Str("endpoint", fullNode).Msg("failed to get full node block height") - m.RecordEndpointAvailability(e.chainID, fullNode, false) - m.RecordEndpointError(e.chainID, fullNode, utils.CategorizeError(err)) - continue - } - - m.RecordEndpointAvailability(e.chainID, fullNode, true) - m.RecordCurrentBlockHeight(e.chainID, fullNode, currentHeight) - m.RecordBlockHeightDrift(e.chainID, fullNode, refHeight, currentHeight) - - drift := int64(refHeight) - int64(currentHeight) - e.logger.Info(). - Uint64("ref_height", refHeight). - Uint64("target_height", currentHeight). - Int64("drift", drift). - Str("endpoint", fullNode). - Msg("recorded full node height and drift") - } } } } -// getBlockHeight queries an EVM RPC endpoint for its current block height -func getBlockHeight(ctx context.Context, rpcURL string) (uint64, error) { - client, err := ethclient.DialContext(ctx, rpcURL) +// poll records the reference node height, then queries all full nodes concurrently and records their drift. +func (e *exporter) poll(ctx context.Context, m *metrics.Metrics, ref *node, fullNodes []*node) { + refHeight, err := e.blockHeight(ctx, ref) + if err != nil { + e.logger.Error().Err(err).Str("endpoint", ref.url).Msg("failed to get reference node block height") + m.RecordEndpointAvailability(e.chainID, ref.url, false) + m.RecordEndpointError(e.chainID, ref.url, utils.CategorizeError(err)) + return + } + + m.RecordEndpointAvailability(e.chainID, ref.url, true) + m.RecordReferenceBlockHeight(e.chainID, ref.url, refHeight) + // Also update time since last block metric when we see new blocks via polling + // This ensures the metric stays accurate even if WebSocket subscription fails + if refHeight > e.lastHeight { + m.UpdateLastBlockTime(e.chainID, time.Now()) + e.lastHeight = refHeight + } + e.logger.Debug().Uint64("height", refHeight).Str("endpoint", ref.url).Msg("recorded reference node height") + + var wg sync.WaitGroup + for _, n := range fullNodes { + wg.Add(1) + go func(n *node) { + defer wg.Done() + e.pollFullNode(ctx, m, n, refHeight) + }(n) + } + wg.Wait() +} + +func (e *exporter) pollFullNode(ctx context.Context, m *metrics.Metrics, n *node, refHeight uint64) { + currentHeight, err := e.blockHeight(ctx, n) if err != nil { - return 0, fmt.Errorf("failed to connect to %s: %w", rpcURL, err) + e.logger.Error().Err(err).Str("endpoint", n.url).Msg("failed to get full node block height") + m.RecordEndpointAvailability(e.chainID, n.url, false) + m.RecordEndpointError(e.chainID, n.url, utils.CategorizeError(err)) + return + } + + m.RecordEndpointAvailability(e.chainID, n.url, true) + m.RecordCurrentBlockHeight(e.chainID, n.url, currentHeight) + m.RecordBlockHeightDrift(e.chainID, n.url, refHeight, currentHeight) + + e.logger.Debug(). + Uint64("ref_height", refHeight). + Uint64("target_height", currentHeight). + Int64("drift", int64(refHeight)-int64(currentHeight)). + Str("endpoint", n.url). + Msg("recorded full node height and drift") +} + +// blockHeight queries an EVM RPC endpoint for its current block height, dialing on first use +// and dropping the client on failure so the next poll redials. +func (e *exporter) blockHeight(ctx context.Context, n *node) (uint64, error) { + ctx, cancel := context.WithTimeout(ctx, e.rpcTimeout) + defer cancel() + + if n.client == nil { + client, err := ethclient.DialContext(ctx, n.url) + if err != nil { + return 0, fmt.Errorf("failed to connect to %s: %w", n.url, err) + } + n.client = client } - defer client.Close() - height, err := client.BlockNumber(ctx) + height, err := n.client.BlockNumber(ctx) if err != nil { - return 0, fmt.Errorf("failed to get block number from %s: %w", rpcURL, err) + n.close() + return 0, fmt.Errorf("failed to get block number from %s: %w", n.url, err) } return height, nil diff --git a/pkg/exporters/drift/drift_test.go b/pkg/exporters/drift/drift_test.go index 15bc635..416c7a0 100644 --- a/pkg/exporters/drift/drift_test.go +++ b/pkg/exporters/drift/drift_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "io" "net/http" "net/http/httptest" "sync/atomic" @@ -66,17 +67,83 @@ func TestExportMetrics_TimeSinceLastBlockOnlyResetsWhenHeightAdvances(t *testing return 0 } - // polls at ~1s (height 10, first seen) and ~2s (height unchanged) + // polls at ~0s (height 10, first seen), ~1s and ~2s (height unchanged) time.Sleep(2500 * time.Millisecond) - require.GreaterOrEqual(t, timeSince(), 1.2, "constant height must not reset the gauge") + require.GreaterOrEqual(t, timeSince(), 2.2, "constant height must not reset the gauge") // poll at ~3s sees a lower height height.Store(5) time.Sleep(time.Second) - require.GreaterOrEqual(t, timeSince(), 2.2, "decreasing height must not reset the gauge") + require.GreaterOrEqual(t, timeSince(), 3.2, "decreasing height must not reset the gauge") // poll at ~4s sees a new height height.Store(11) time.Sleep(time.Second) require.Less(t, timeSince(), 1.0, "advancing height must reset the gauge") } + +func gaugeValue(t *testing.T, reg *prometheus.Registry, name, endpoint string) (float64, bool) { + t.Helper() + families, err := reg.Gather() + require.NoError(t, err) + for _, mf := range families { + if mf.GetName() != name { + continue + } + for _, metric := range mf.GetMetric() { + for _, l := range metric.GetLabel() { + if l.GetValue() == endpoint { + return metric.GetGauge().GetValue(), true + } + } + } + } + return 0, false +} + +func TestExportMetrics_HungFullNodeDoesNotBlockOthers(t *testing.T) { + var refHeight, goodHeight atomic.Uint64 + refHeight.Store(100) + goodHeight.Store(90) + ref := newBlockNumberServer(t, &refHeight) + good := newBlockNumberServer(t, &goodHeight) + hung := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // the body must be drained for r.Context() to observe the client disconnecting + _, _ = io.Copy(io.Discard, r.Body) + select { + case <-r.Context().Done(): + case <-time.After(5 * time.Second): + } + })) + t.Cleanup(hung.Close) + + reg := prometheus.NewRegistry() + m := metrics.NewWithRegistry("test", reg) + e := NewMetricsExporter("chain", ref.URL, []string{hung.URL, good.URL}, 60, zerolog.Nop()).(*exporter) + e.rpcTimeout = time.Second + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + _ = e.ExportMetrics(ctx, m) + }() + t.Cleanup(func() { + cancel() + <-done + }) + + // well under rpcTimeout: the good node must not wait for the hung one + require.Eventually(t, func() bool { + drift, ok := gaugeValue(t, reg, "test_block_height_drift", good.URL) + return ok && drift == 10 + }, 500*time.Millisecond, 10*time.Millisecond) + height, ok := gaugeValue(t, reg, "test_target_block_height", good.URL) + require.True(t, ok) + require.Equal(t, 90.0, height) + + require.Eventually(t, func() bool { + avail, ok := gaugeValue(t, reg, "test_endpoint_availability", hung.URL) + return ok && avail == 0 + }, 3*time.Second, 20*time.Millisecond) +} diff --git a/pkg/exporters/jsonrpc/json_rpc.go b/pkg/exporters/jsonrpc/json_rpc.go index da4d76a..f2ca26f 100644 --- a/pkg/exporters/jsonrpc/json_rpc.go +++ b/pkg/exporters/jsonrpc/json_rpc.go @@ -76,7 +76,7 @@ func performHealthCheck( m.RecordEndpointAvailability(chainID, "jsonrpc", true) m.RecordJsonRpcRequestDuration(chainID, duration) - logger.Info(). + logger.Debug(). Dur("duration", duration). Float64("duration_seconds", duration.Seconds()). Msg("JSON-RPC health check completed") diff --git a/pkg/exporters/verifier/verifier.go b/pkg/exporters/verifier/verifier.go index 2b65fed..71e2a6d 100644 --- a/pkg/exporters/verifier/verifier.go +++ b/pkg/exporters/verifier/verifier.go @@ -2,13 +2,12 @@ package verifier import ( "context" + "math/big" "sync" "time" ethereum "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/core/types" - "github.com/evstack/ev-metrics/internal/clients/celestia" - "github.com/evstack/ev-metrics/internal/clients/evm" "github.com/evstack/ev-metrics/internal/clients/evnode" "github.com/evstack/ev-metrics/pkg/metrics" "github.com/rs/zerolog" @@ -16,11 +15,52 @@ import ( var _ metrics.Exporter = &exporter{} +type blockSource interface { + GetBlockWithBlobs(ctx context.Context, height uint64) (*evnode.GetBlockResult, error) +} + +type daVerifier interface { + GetBlockTimestamp(ctx context.Context, daHeight uint64) (time.Time, error) + VerifyBlobAtHeight(ctx context.Context, blob []byte, daHeight uint64, namespace []byte) (bool, error) + VerifyDataBlobAtHeight(ctx context.Context, unwrappedDataBlob []byte, daHeight uint64, namespace []byte) (bool, error) +} + +type headSource interface { + SubscribeNewHead(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error) + HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error) +} + +// job is one pending verification attempt for a block. start is zero until the +// block is first processed by a worker. +type job struct { + header *types.Header + attempt int + start time.Time +} + +// defaultRetryIntervals is the delay before each verification attempt, matching +// observed DA submission timing. +var defaultRetryIntervals = []time.Duration{ + 0, // immediate first attempt + 20 * time.Second, + 40 * time.Second, + 60 * time.Second, + 90 * time.Second, + 120 * time.Second, +} + +const ( + // defaultCooldown is how long to wait before re-verifying a block whose retries were exhausted. + defaultCooldown = 5 * time.Minute + // maxBackfill caps how many heights are fetched after a gap in the header subscription. + maxBackfill = 1000 +) + // NewMetricsExporter creates a new exporter func NewMetricsExporter( - evnodeClient *evnode.Client, - celestiaClient *celestia.Client, - evmClient *evm.Client, + evnodeClient blockSource, + celestiaClient daVerifier, + evmClient headSource, headerNS, dataNS []byte, chainID string, workers int, @@ -34,19 +74,23 @@ func NewMetricsExporter( dataNS: dataNS, chainID: chainID, workers: workers, + retryIntervals: defaultRetryIntervals, + cooldown: defaultCooldown, logger: logger.With().Str("component", "verification_monitor").Logger(), } } // exporter handles verification of blocks against Celestia DA type exporter struct { - evnodeClient *evnode.Client - celestiaClient *celestia.Client - evmClient *evm.Client + evnodeClient blockSource + celestiaClient daVerifier + evmClient headSource headerNS []byte dataNS []byte chainID string workers int + retryIntervals []time.Duration + cooldown time.Duration logger zerolog.Logger } @@ -59,7 +103,7 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error } // create buffered channel for block queue - blockQueue := make(chan *types.Header, e.workers*2) + blockQueue := make(chan job, e.workers*2) // start work pool var workerGroup sync.WaitGroup @@ -89,6 +133,9 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error workerGroup.Wait() } + // lastHeight is the highest block seen, used to detect gaps after reconnects. + var lastHeight uint64 + // main subscription loop for { select { @@ -122,14 +169,24 @@ func (e *exporter) ExportMetrics(ctx context.Context, m *metrics.Metrics) error arrivalTime := time.Now() m.RecordBlockTime(e.chainID, arrivalTime) + height := header.Number.Uint64() e.logger.Debug(). - Uint64("block_height", header.Number.Uint64()). + Uint64("block_height", height). Time("arrival_time", arrivalTime). Msg("received block header from subscription") + if lastHeight != 0 && height > lastHeight+1 { + workerGroup.Add(1) + go func(from, to uint64) { + defer workerGroup.Done() + e.backfill(ctx, blockQueue, from, to) + }(lastHeight+1, height-1) + } + lastHeight = max(lastHeight, height) + // send block to work pool, blocking until space is available select { - case blockQueue <- header: + case blockQueue <- job{header: header}: // block queued successfully case <-ctx.Done(): shutdown() @@ -166,35 +223,56 @@ func (e *exporter) reconnectSubscription(ctx context.Context, headers chan *type } } +// backfill enqueues the blocks in [from, to] that were missed by the subscription, +// keeping only the most recent maxBackfill heights. +func (e *exporter) backfill(ctx context.Context, blockQueue chan<- job, from, to uint64) { + if n := to - from + 1; n > maxBackfill { + e.logger.Warn().Uint64("dropped", n-maxBackfill).Uint64("from", from).Uint64("to", to).Msg("backfill gap too large, skipping oldest blocks") + from = to - maxBackfill + 1 + } + e.logger.Info().Uint64("from", from).Uint64("to", to).Msg("backfilling blocks missed by subscription") + + for h := from; h <= to; h++ { + header, err := e.evmClient.HeaderByNumber(ctx, new(big.Int).SetUint64(h)) + if err != nil { + if ctx.Err() != nil { + return + } + e.logger.Warn().Err(err).Uint64("block_height", h).Msg("failed to fetch header for backfill, skipping") + continue + } + select { + case blockQueue <- job{header: header}: + case <-ctx.Done(): + return + } + } +} + +// enqueueAfter sends j to blockQueue after d without blocking the caller. +func enqueueAfter(ctx context.Context, blockQueue chan<- job, d time.Duration, j job) { + time.AfterFunc(d, func() { + select { + case blockQueue <- j: + case <-ctx.Done(): + } + }) +} + // processBlocks processes blocks from the queue -func (e *exporter) processBlocks(ctx context.Context, m *metrics.Metrics, workerID int, blockQueue chan *types.Header) { +func (e *exporter) processBlocks(ctx context.Context, m *metrics.Metrics, workerID int, blockQueue chan job) { logger := e.logger.With().Int("worker_id", workerID).Logger() logger.Debug().Msg("worker started") - // blockQueue is never closed: delayed re-queue goroutines may still send to it, + // blockQueue is never closed: delayed re-queue timers may still send to it, // so workers exit on ctx cancellation instead. for { select { case <-ctx.Done(): logger.Debug().Msg("worker stopped") return - case header := <-blockQueue: - reEnqueue := e.verifyBlock(ctx, m, header) - if reEnqueue { - // re-queue with delay in background - go func(h *types.Header) { - select { - case <-time.After(5 * time.Minute): - select { - case blockQueue <- h: - logger.Debug().Uint64("block", h.Number.Uint64()).Msg("re-queued block after cooldown") - case <-ctx.Done(): - // graceful shutdown, don't send - } - case <-ctx.Done(): - } - }(header) - } + case j := <-blockQueue: + e.verifyBlock(ctx, m, blockQueue, j) } } } @@ -217,8 +295,10 @@ func (e *exporter) onVerified(m *metrics.Metrics, namespace string, blockHeight, // freezing metrics. const verifyAttemptTimeout = 30 * time.Second -// verifyBlock attempts to verify a DA height for a given block status. -func (e *exporter) verifyBlock(ctx context.Context, m *metrics.Metrics, header *types.Header) bool { +// verifyBlock makes one verification attempt for j and schedules the next attempt +// (or a cooldown re-queue once retries are exhausted) without blocking the worker. +func (e *exporter) verifyBlock(ctx context.Context, m *metrics.Metrics, blockQueue chan<- job, j job) { + header := j.header blockHeight := header.Number.Uint64() // check if block has transactions @@ -232,46 +312,29 @@ func (e *exporter) verifyBlock(ctx context.Context, m *metrics.Metrics, header * blockTime := time.Unix(int64(header.Time), 0) logger := e.logger.With().Str("namespace", namespace).Uint64("block_height", blockHeight).Logger() - logger.Info(). - Str("hash", header.Hash().Hex()). - Time("time", blockTime). - Uint64("gas_used", header.GasUsed). - Bool("has_transactions", hasTransactions). - Msg("processing block") - - startTime := time.Now() - - // exponential backoff intervals matching observed DA submission timing - retryIntervals := []time.Duration{ - 0, // immediate first attempt - 20 * time.Second, - 40 * time.Second, - 60 * time.Second, - 90 * time.Second, - 120 * time.Second, + if j.start.IsZero() { + j.start = time.Now() + logger.Debug(). + Str("hash", header.Hash().Hex()). + Time("time", blockTime). + Uint64("gas_used", header.GasUsed). + Bool("has_transactions", hasTransactions). + Msg("processing block") } - for i, interval := range retryIntervals { - retries := i + 1 - - select { - case <-ctx.Done(): - // context cancelled during graceful shutdown, not an error - logger.Debug().Msg("block verification stopped due to shutdown") - return false - case <-time.After(interval): - // proceed with retry - } + if e.verifyAttempt(ctx, m, logger, j.attempt+1, blockHeight, namespace, blockTime, j.start) { + return + } - if e.verifyAttempt(ctx, m, logger, retries, blockHeight, namespace, blockTime, startTime) { - return false - } + if next := j.attempt + 1; next < len(e.retryIntervals) { + j.attempt = next + enqueueAfter(ctx, blockQueue, e.retryIntervals[next], j) + return } - // if loop completes without success, log final error logger.Error().Msg("max retries exhausted: failed to verify block") e.onVerified(m, namespace, blockHeight, 0, false, 0) - return true + enqueueAfter(ctx, blockQueue, e.cooldown, job{header: header}) } // verifyAttempt performs one bounded RPC attempt to verify a block against Celestia DA. @@ -283,15 +346,15 @@ func (e *exporter) verifyAttempt(ctx context.Context, m *metrics.Metrics, logger attemptCtx, cancel := context.WithTimeout(ctx, verifyAttemptTimeout) defer cancel() - blockResult, err := e.evnodeClient.GetBlock(attemptCtx, blockHeight) + blockResultWithBlobs, err := e.evnodeClient.GetBlockWithBlobs(attemptCtx, blockHeight) if err != nil { - logger.Warn().Err(err).Int("attempt", retries).Msg("failed to re-query block from ev-node") + logger.Warn().Err(err).Int("attempt", retries).Msg("failed to query block from ev-node") return false } - daHeight := blockResult.HeaderDaHeight + daHeight := blockResultWithBlobs.HeaderDaHeight if namespace == "data" { - daHeight = blockResult.DataDaHeight + daHeight = blockResultWithBlobs.DataDaHeight } if daHeight == 0 { @@ -299,12 +362,6 @@ func (e *exporter) verifyAttempt(ctx context.Context, m *metrics.Metrics, logger return false } - blockResultWithBlobs, err := e.evnodeClient.GetBlockWithBlobs(attemptCtx, blockHeight) - if err != nil { - logger.Warn().Err(err).Int("attempt", retries).Msg("failed to query block from ev-node") - return false - } - daBlockTime, err := e.celestiaClient.GetBlockTimestamp(attemptCtx, daHeight) if err != nil { logger.Warn().Err(err).Uint64("da_height", daHeight).Msg("failed to get da block timestamp") @@ -322,7 +379,7 @@ func (e *exporter) verifyAttempt(ctx context.Context, m *metrics.Metrics, logger return false } if verified { - logger.Info(). + logger.Debug(). Uint64("da_height", daHeight). Dur("duration", time.Since(startTime)). Msg("header blob verified on Celestia") @@ -333,7 +390,7 @@ func (e *exporter) verifyAttempt(ctx context.Context, m *metrics.Metrics, logger case "data": if len(blockResultWithBlobs.DataBlob) == 0 { - logger.Info(). + logger.Debug(). Dur("duration", time.Since(startTime)). Msg("empty data block - no verification needed") e.onVerified(m, namespace, blockHeight, daHeight, true, submissionDuration) @@ -347,7 +404,7 @@ func (e *exporter) verifyAttempt(ctx context.Context, m *metrics.Metrics, logger return false } if verified { - logger.Info(). + logger.Debug(). Uint64("da_height", daHeight). Dur("duration", time.Since(startTime)). Msg("data blob verified on Celestia") diff --git a/pkg/exporters/verifier/verifier_test.go b/pkg/exporters/verifier/verifier_test.go index ec59afe..cc965c3 100644 --- a/pkg/exporters/verifier/verifier_test.go +++ b/pkg/exporters/verifier/verifier_test.go @@ -2,16 +2,333 @@ package verifier import ( "context" + "math/big" + "slices" + "sync" "testing" "time" + ethereum "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/core/types" + "github.com/evstack/ev-metrics/internal/clients/evnode" + "github.com/evstack/ev-metrics/pkg/metrics" + "github.com/prometheus/client_golang/prometheus" "github.com/rs/zerolog" ) +const daOffset = 100 + +// fakeBlocks reports a block as not yet submitted for notYet[height] attempts +// (forever if negative), then as submitted at DA height height+daOffset. +type fakeBlocks struct { + mu sync.Mutex + notYet map[uint64]int + calls map[uint64]int +} + +func (f *fakeBlocks) GetBlockWithBlobs(_ context.Context, height uint64) (*evnode.GetBlockResult, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls[height]++ + if n := f.notYet[height]; n < 0 || f.calls[height] <= n { + return &evnode.GetBlockResult{}, nil + } + return &evnode.GetBlockResult{HeaderDaHeight: height + daOffset, HeaderBlob: []byte{1}}, nil +} + +func (f *fakeBlocks) callCount(height uint64) int { + f.mu.Lock() + defer f.mu.Unlock() + return f.calls[height] +} + +type fakeDA struct { + mu sync.Mutex + verified []uint64 +} + +func (f *fakeDA) GetBlockTimestamp(context.Context, uint64) (time.Time, error) { + return time.Now(), nil +} + +func (f *fakeDA) VerifyBlobAtHeight(_ context.Context, _ []byte, daHeight uint64, _ []byte) (bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.verified = append(f.verified, daHeight-daOffset) + return true, nil +} + +func (f *fakeDA) VerifyDataBlobAtHeight(context.Context, []byte, uint64, []byte) (bool, error) { + return true, nil +} + +func (f *fakeDA) verifiedHeights() []uint64 { + f.mu.Lock() + defer f.mu.Unlock() + return slices.Clone(f.verified) +} + +type fakeSub struct{ err chan error } + +func (s fakeSub) Unsubscribe() {} +func (s fakeSub) Err() <-chan error { return s.err } + +type fakeHeads struct { + mu sync.Mutex + ch chan<- *types.Header + fetched []uint64 +} + +func (f *fakeHeads) SubscribeNewHead(_ context.Context, ch chan<- *types.Header) (ethereum.Subscription, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.ch = ch + return fakeSub{err: make(chan error)}, nil +} + +func (f *fakeHeads) HeaderByNumber(_ context.Context, number *big.Int) (*types.Header, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.fetched = append(f.fetched, number.Uint64()) + return newHeader(number.Uint64()), nil +} + +func (f *fakeHeads) fetchedHeights() []uint64 { + f.mu.Lock() + defer f.mu.Unlock() + return slices.Clone(f.fetched) +} + +func newHeader(height uint64) *types.Header { + return &types.Header{Number: new(big.Int).SetUint64(height), TxHash: types.EmptyRootHash} +} + +type harness struct { + e *exporter + blocks *fakeBlocks + da *fakeDA + heads *fakeHeads + m *metrics.Metrics + reg *prometheus.Registry +} + +func newHarness(workers int, notYet map[uint64]int, retryIntervals ...time.Duration) *harness { + h := &harness{ + blocks: &fakeBlocks{notYet: notYet, calls: map[uint64]int{}}, + da: &fakeDA{}, + heads: &fakeHeads{}, + reg: prometheus.NewRegistry(), + } + h.m = metrics.NewWithRegistry("test", h.reg) + h.e = NewMetricsExporter(h.blocks, h.da, h.heads, []byte("h"), []byte("d"), "chain", workers, zerolog.Nop()).(*exporter) + h.e.retryIntervals = retryIntervals + h.e.cooldown = time.Hour + return h +} + +// startWorkers runs the worker pool on a fresh queue and returns the queue and +// a function that cancels the workers and waits for them to exit. +func (h *harness) startWorkers(t *testing.T) (chan job, func()) { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + queue := make(chan job, 10) + var wg sync.WaitGroup + for i := 0; i < h.e.workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + h.e.processBlocks(ctx, h.m, i, queue) + }() + } + stop := func() { + cancel() + wg.Wait() + } + t.Cleanup(stop) + return queue, stop +} + +func (h *harness) metric(t *testing.T, name string) float64 { + t.Helper() + families, err := h.reg.Gather() + if err != nil { + t.Fatal(err) + } + var sum float64 + for _, mf := range families { + if mf.GetName() != "test_"+name { + continue + } + for _, m := range mf.GetMetric() { + switch { + case m.Gauge != nil: + sum += m.GetGauge().GetValue() + case m.Counter != nil: + sum += m.GetCounter().GetValue() + case m.Histogram != nil: + sum += float64(m.GetHistogram().GetSampleCount()) + } + } + } + return sum +} + +func eventually(t *testing.T, msg string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for !cond() { + if time.Now().After(deadline) { + t.Fatal(msg) + } + time.Sleep(time.Millisecond) + } +} + +func TestVerifiedOnFirstAttempt(t *testing.T) { + h := newHarness(1, nil, 0, time.Millisecond) + queue, _ := h.startWorkers(t) + + queue <- job{header: newHeader(1)} + + eventually(t, "block not verified", func() bool { return h.metric(t, "submission_da_height") == 1+daOffset }) + if got := h.blocks.callCount(1); got != 1 { + t.Fatalf("expected 1 ev-node call, got %d", got) + } +} + +func TestVerifiedAfterRetriesDoesNotBlockWorker(t *testing.T) { + interval := 30 * time.Millisecond + h := newHarness(1, map[uint64]int{1: 3}, 0, interval, interval, interval, interval) + queue, _ := h.startWorkers(t) + + queue <- job{header: newHeader(1)} + queue <- job{header: newHeader(2)} + + eventually(t, "blocks not verified", func() bool { return len(h.da.verifiedHeights()) == 2 }) + if got := h.da.verifiedHeights(); !slices.Equal(got, []uint64{2, 1}) { + t.Fatalf("expected block 2 to be verified while block 1 waits for a retry, got order %v", got) + } + if got := h.blocks.callCount(1); got != 4 { + t.Fatalf("expected 4 attempts for block 1, got %d", got) + } + if got := h.metric(t, "unsubmitted_blocks_total"); got != 0 { + t.Fatalf("expected no unsubmitted blocks, got %v", got) + } +} + +func TestRetriesExhaustedRecordsMissingBlock(t *testing.T) { + h := newHarness(1, map[uint64]int{1: -1}, 0, time.Millisecond, time.Millisecond) + queue, stop := h.startWorkers(t) + + queue <- job{header: newHeader(1)} + + eventually(t, "missing block not recorded", func() bool { return h.metric(t, "unsubmitted_blocks_total") == 1 }) + stop() + if got := h.blocks.callCount(1); got != 3 { + t.Fatalf("expected 3 attempts before cooldown, got %d", got) + } +} + +func TestRetriesExhaustedRequeuesAfterCooldown(t *testing.T) { + h := newHarness(1, map[uint64]int{1: 2}, 0, time.Millisecond) + h.e.cooldown = 10 * time.Millisecond + queue, _ := h.startWorkers(t) + + queue <- job{header: newHeader(1)} + + eventually(t, "missing block not cleared after cooldown", func() bool { + return len(h.da.verifiedHeights()) == 1 && h.metric(t, "unsubmitted_blocks_total") == 0 + }) + if got := h.blocks.callCount(1); got != 3 { + t.Fatalf("expected 3 attempts, got %d", got) + } +} + +func TestExportMetricsBackfillsGap(t *testing.T) { + h := newHarness(2, nil, 0, time.Millisecond) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error) + go func() { done <- h.e.ExportMetrics(ctx, h.m) }() + + var headers chan<- *types.Header + eventually(t, "not subscribed", func() bool { + h.heads.mu.Lock() + defer h.heads.mu.Unlock() + headers = h.heads.ch + return headers != nil + }) + for _, n := range []uint64{1, 2, 6, 4} { + headers <- newHeader(n) + time.Sleep(time.Millisecond) + } + + eventually(t, "backfilled blocks not verified", func() bool { return len(h.da.verifiedHeights()) == 7 }) + got := h.da.verifiedHeights() + slices.Sort(got) + if !slices.Equal(got, []uint64{1, 2, 3, 4, 4, 5, 6}) { + t.Fatalf("unexpected verified heights %v", got) + } + if fetched := h.heads.fetchedHeights(); !slices.Equal(fetched, []uint64{3, 4, 5}) { + t.Fatalf("expected heights 3..5 to be backfilled once, got %v", fetched) + } + // 4 live headers yield 3 block-time observations; backfilled headers add none. + if got := h.metric(t, "block_time_seconds"); got != 3 { + t.Fatalf("expected 3 block time observations, got %v", got) + } + + cancel() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("ExportMetrics did not return after context cancellation") + } +} + +func TestBackfillCapsGap(t *testing.T) { + h := newHarness(1, nil) + queue := make(chan job, maxBackfill+10) + + h.e.backfill(context.Background(), queue, 1, maxBackfill+5) + + if got := len(queue); got != maxBackfill { + t.Fatalf("expected %d backfilled jobs, got %d", maxBackfill, got) + } + if first := (<-queue).header.Number.Uint64(); first != 6 { + t.Fatalf("expected backfill to keep the most recent heights starting at 6, got %d", first) + } +} + +func TestExportMetricsStopsOnContextCancel(t *testing.T) { + h := newHarness(2, map[uint64]int{1: -1}, 0, time.Millisecond) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error) + go func() { done <- h.e.ExportMetrics(ctx, h.m) }() + + eventually(t, "not subscribed", func() bool { + h.heads.mu.Lock() + defer h.heads.mu.Unlock() + return h.heads.ch != nil + }) + h.heads.ch <- newHeader(1) + eventually(t, "block not attempted", func() bool { return h.blocks.callCount(1) > 0 }) + + cancel() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("ExportMetrics did not return after context cancellation") + } +} + func TestProcessBlocksStopsOnContextCancel(t *testing.T) { e := &exporter{logger: zerolog.Nop()} - queue := make(chan *types.Header, 1) + queue := make(chan job, 1) ctx, cancel := context.WithCancel(context.Background()) done := make(chan struct{})