diff --git a/harnesses/rpc-capabilities/cmd/script/archive.go b/harnesses/rpc-capabilities/cmd/script/archive.go index d7889264..1c9e74e6 100644 --- a/harnesses/rpc-capabilities/cmd/script/archive.go +++ b/harnesses/rpc-capabilities/cmd/script/archive.go @@ -45,8 +45,10 @@ func StartArchiveLoop(ctx context.Context) { c := c if c.Kind == "solana" || c.Kind == "polkadot" { // eth_getBalance at historical heights has no Solana or - // Substrate equivalent on public endpoints; skip the - // archive loop. + // Substrate equivalent on public endpoints (Polkadot state + // is accessed via state_getStorage keyed by a Blake2 + // hashed storage key, no chain-agnostic depth analog); + // skip the archive loop. continue } for _, p := range c.Providers { diff --git a/harnesses/rpc-capabilities/cmd/script/config.go b/harnesses/rpc-capabilities/cmd/script/config.go index 740d3537..d9c769ca 100644 --- a/harnesses/rpc-capabilities/cmd/script/config.go +++ b/harnesses/rpc-capabilities/cmd/script/config.go @@ -36,7 +36,10 @@ type Chain struct { // no archive-depth loop. // "polkadot": chain_getHeader against Substrate JSON-RPC, // block-based staleness (Polkadot relay produces one block every - // ~6 s, override via polkadotStaleBlockGap), no archive-depth loop. + // ~6 s so staleBlockGap needs a Polkadot-specific override, see + // polkadotStaleBlockGap), no archive-depth loop (Polkadot's + // state model does not map onto the eth_getBalance-by-depth + // probe cleanly). Kind string } @@ -62,8 +65,12 @@ type Chain struct { // (drpc caches eth_blockNumber → only 2 clean providers), opBNB // (1rpc 429s at probe cadence, only 3 solid providers), Mode // (3 providers), Zora / Abstract / HyperEVM (≤2 keyless providers). +// +// Local-run filter: OCB_CHAINS=ethereum,base restricts the matrix to +// the listed slugs (unset or unmatched = full matrix). Used for local +// smoke runs; never set in production. func chains() []Chain { - return []Chain{ + all := []Chain{ // ─── Polkadot relay chain — first non-EVM, non-Solana chain // added to the cohort. Substrate JSON-RPC via chain_getHeader // (returns hex block number, staleness by relay-block gap). @@ -391,6 +398,25 @@ func chains() []Chain { }, }, } + + filter := strings.TrimSpace(os.Getenv("OCB_CHAINS")) + if filter == "" { + return all + } + keep := make(map[string]bool) + for _, s := range strings.Split(filter, ",") { + keep[strings.TrimSpace(s)] = true + } + var out []Chain + for _, c := range all { + if keep[c.Slug] { + out = append(out, c) + } + } + if len(out) == 0 { + return all + } + return out } func envDefault(key, def string) string { diff --git a/harnesses/rpc-capabilities/cmd/script/consensus.go b/harnesses/rpc-capabilities/cmd/script/consensus.go new file mode 100644 index 00000000..695567e5 --- /dev/null +++ b/harnesses/rpc-capabilities/cmd/script/consensus.go @@ -0,0 +1,154 @@ +package main + +import "sync" + +// consensus.go — bench 083 rpc-reliability, cross-provider head +// agreement. The latency probe already fetches the full latest header +// (anti-cache design) and used to throw the hash away; here every +// successful probe feeds a per-chain height→hash quorum map so we can +// publish two things the latency bench cannot see: +// +// rpc_consensus_lag_blocks — how far this provider's head sits +// behind the highest head any probed provider reported for the +// same chain (the chainTips rolling max). Zero for a provider at +// the shared tip; 1-2 for a gateway one block behind (observed +// live on 1rpc/tenderly); tens for something wedged. +// +// rpc_hash_mismatch_total — the provider's hash at height H +// disagrees with the hash a >=2-provider strict plurality agreed +// on at H. Same-height hash divergence is the serious signal +// (serving a non-canonical or fabricated block), so it is a +// counter, incremented at most once per (provider, height). +// +// Reorg honesty: during a real reorg two providers can legitimately +// sit on different hashes at the same height. quorumHash requires a +// strict plurality (>=2 votes AND strictly more than any competing +// hash), so a 2-2 split counts nobody. Only a provider outvoted by an +// established majority is flagged. + +const ( + // quorumMinProviders: minimum providers agreeing on one hash + // before that hash is treated as canonical at its height. + quorumMinProviders = 2 + // consensusPruneDepth: heights this far below the chain tip are + // dropped from the vote map so memory stays bounded (~2 blocks/s + // chains would otherwise grow forever). + consensusPruneDepth uint64 = 128 +) + +type heightVotes struct { + hash map[string]string // provider -> reported hash + flagged map[string]bool // providers already counted at this height +} + +type chainConsensus struct { + heights map[uint64]*heightVotes +} + +type consensusTracker struct { + mu sync.Mutex + chains map[string]*chainConsensus +} + +func newConsensusTracker() *consensusTracker { + return &consensusTracker{chains: make(map[string]*chainConsensus)} +} + +var consensus = newConsensusTracker() + +// observe records one valid head probe. Called from probeOne for +// results classified ok or stale (a stale block is still a valid +// (height, hash) observation — its lag is exactly the point). hash is +// "" on the Solana path (getSlot carries no hash): lag is emitted, +// quorum voting is skipped. +func (t *consensusTracker) observe(chain, provider string, height uint64, hash string) { + // Lag against the cross-provider rolling max the staleness check + // already maintains. The caller updates tips before observing, so + // a provider that IS the tip reads 0. Noise note: tips is updated + // asynchronously by per-provider goroutines staggered across the + // probe interval, so ±1 block of jitter on fast chains is expected + // and averages out in the 24h quantiles. + tip := tips.get(chain) + lag := 0.0 + if tip > height { + lag = float64(tip - height) + } + rpcConsensusLag.WithLabelValues(provider, chain, currentRegion).Set(lag) + + if hash == "" { + return + } + + t.mu.Lock() + defer t.mu.Unlock() + + cc := t.chains[chain] + if cc == nil { + cc = &chainConsensus{heights: make(map[uint64]*heightVotes)} + t.chains[chain] = cc + } + hv := cc.heights[height] + if hv == nil { + hv = &heightVotes{hash: make(map[string]string), flagged: make(map[string]bool)} + cc.heights[height] = hv + } + hv.hash[provider] = hash + + if q := quorumHash(hv.hash); q != "" { + for p, h := range hv.hash { + if h != q && !hv.flagged[p] { + hv.flagged[p] = true + rpcHashMismatch.WithLabelValues(p, chain, currentRegion).Inc() + } + } + } + + for h := range cc.heights { + if h+consensusPruneDepth < tip { + delete(cc.heights, h) + } + } +} + +// quorumHash returns the hash backed by >= quorumMinProviders votes +// AND strictly more votes than any competing hash. Ties (the 2-2 reorg +// split) return "" so no side is punished without a real majority. +func quorumHash(votes map[string]string) string { + counts := make(map[string]int, len(votes)) + for _, h := range votes { + counts[h]++ + } + best, bestN, secondN := "", 0, 0 + for h, n := range counts { + if n > bestN { + best, secondN, bestN = h, bestN, n + } else if n > secondN { + secondN = n + } + } + if bestN >= quorumMinProviders && bestN > secondN { + return best + } + return "" +} + +// initConsensusMetrics zero-initializes the mismatch counters for the +// full (provider × chain) matrix so `increase()` in the bench's +// incident queries resolves to 0 instead of an absent series for +// providers that never misbehave (which is, hopefully, most of them). +func initConsensusMetrics() { + for _, c := range chains() { + if c.Kind == "solana" || c.Kind == "polkadot" { + // solana: getSlot returns a number only, no hash to vote on. + // polkadot: chain_getHeader returns parentHash (N-1) but no + // current block hash; deriving it needs Blake2 over the + // SCALE-encoded header, which is not worth wiring for a + // bench 083 cross-provider quorum check that already has + // value on 20+ EVM chains. + continue + } + for _, p := range c.Providers { + rpcHashMismatch.WithLabelValues(p.Slug, c.Slug, currentRegion).Add(0) + } + } +} diff --git a/harnesses/rpc-capabilities/cmd/script/consensus_test.go b/harnesses/rpc-capabilities/cmd/script/consensus_test.go new file mode 100644 index 00000000..27e36d65 --- /dev/null +++ b/harnesses/rpc-capabilities/cmd/script/consensus_test.go @@ -0,0 +1,144 @@ +package main + +import ( + "testing" + + "github.com/prometheus/client_golang/prometheus/testutil" +) + +func TestQuorumHash(t *testing.T) { + cases := []struct { + name string + votes map[string]string + want string + }{ + {"empty", map[string]string{}, ""}, + {"single vote no quorum", map[string]string{"a": "0xaa"}, ""}, + {"two agree", map[string]string{"a": "0xaa", "b": "0xaa"}, "0xaa"}, + {"majority beats minority", map[string]string{"a": "0xaa", "b": "0xaa", "c": "0xbb"}, "0xaa"}, + {"reorg 2-2 split counts nobody", map[string]string{"a": "0xaa", "b": "0xaa", "c": "0xbb", "d": "0xbb"}, ""}, + {"3-2 split resolves", map[string]string{"a": "0xaa", "b": "0xaa", "c": "0xaa", "d": "0xbb", "e": "0xbb"}, "0xaa"}, + {"all distinct", map[string]string{"a": "0xaa", "b": "0xbb", "c": "0xcc"}, ""}, + } + for _, tc := range cases { + if got := quorumHash(tc.votes); got != tc.want { + t.Errorf("%s: quorumHash = %q, want %q", tc.name, got, tc.want) + } + } +} + +func TestObserveFlagsMinorityOnce(t *testing.T) { + tr := newConsensusTracker() + chain := "testchain-mismatch" + tips.update(chain, 100) + + mismatches := func(p string) float64 { + return testutil.ToFloat64(rpcHashMismatch.WithLabelValues(p, chain, currentRegion)) + } + + // First vote: no quorum yet, nobody flagged. + tr.observe(chain, "alpha", 100, "0xaa") + if got := mismatches("alpha"); got != 0 { + t.Fatalf("alpha flagged before any quorum existed: %v", got) + } + + // Quorum forms on 0xaa; gamma disagrees and is flagged exactly once. + tr.observe(chain, "beta", 100, "0xaa") + tr.observe(chain, "gamma", 100, "0xbb") + if got := mismatches("gamma"); got != 1 { + t.Fatalf("gamma mismatch count = %v, want 1", got) + } + if mismatches("alpha") != 0 || mismatches("beta") != 0 { + t.Fatalf("quorum members were flagged") + } + + // Re-reporting the same bad hash at the same height must not + // double-count: 60s probes revisit heights on slow chains. + tr.observe(chain, "gamma", 100, "0xbb") + if got := mismatches("gamma"); got != 1 { + t.Fatalf("gamma double-counted at same height: %v", got) + } + + // A different height is a new incident. + tr.observe(chain, "alpha", 101, "0xcc") + tr.observe(chain, "beta", 101, "0xcc") + tr.observe(chain, "gamma", 101, "0xdd") + if got := mismatches("gamma"); got != 2 { + t.Fatalf("gamma mismatch count after 2nd height = %v, want 2", got) + } +} + +func TestObserveReorgSplitCountsNobody(t *testing.T) { + tr := newConsensusTracker() + chain := "testchain-reorg" + tips.update(chain, 50) + + tr.observe(chain, "a", 50, "0xaa") + tr.observe(chain, "b", 50, "0xaa") + tr.observe(chain, "c", 50, "0xbb") + tr.observe(chain, "d", 50, "0xbb") + // c was flagged while 0xaa held a 2-1 plurality; d's vote made it + // 2-2, and from that point no NEW flags may appear. + before := testutil.ToFloat64(rpcHashMismatch.WithLabelValues("d", chain, currentRegion)) + if before != 0 { + t.Fatalf("d flagged on a 2-2 split: %v", before) + } +} + +func TestObserveConsensusLagGauge(t *testing.T) { + tr := newConsensusTracker() + chain := "testchain-lag" + tips.update(chain, 200) + + tr.observe(chain, "laggy", 197, "0xaa") + got := testutil.ToFloat64(rpcConsensusLag.WithLabelValues("laggy", chain, currentRegion)) + if got != 3 { + t.Fatalf("lag gauge = %v, want 3", got) + } + + // Provider at (or ahead of) the recorded tip reads 0. + tr.observe(chain, "fresh", 200, "0xaa") + if got := testutil.ToFloat64(rpcConsensusLag.WithLabelValues("fresh", chain, currentRegion)); got != 0 { + t.Fatalf("fresh lag gauge = %v, want 0", got) + } + + // Solana path: empty hash emits lag but never votes. + tr.observe(chain, "solananode", 195, "") + if got := testutil.ToFloat64(rpcConsensusLag.WithLabelValues("solananode", chain, currentRegion)); got != 5 { + t.Fatalf("solana lag gauge = %v, want 5", got) + } + if got := testutil.ToFloat64(rpcHashMismatch.WithLabelValues("solananode", chain, currentRegion)); got != 0 { + t.Fatalf("hashless observation voted: %v", got) + } +} + +func TestConsensusPruning(t *testing.T) { + tr := newConsensusTracker() + chain := "testchain-prune" + tips.update(chain, 10) + + tr.observe(chain, "a", 10, "0xaa") + tips.update(chain, 10+consensusPruneDepth+5) + tr.observe(chain, "a", 10+consensusPruneDepth+5, "0xbb") + + tr.mu.Lock() + defer tr.mu.Unlock() + if _, ok := tr.chains[chain].heights[10]; ok { + t.Fatalf("height 10 not pruned at tip %d", 10+consensusPruneDepth+5) + } +} + +func TestMajority(t *testing.T) { + if v, ok := majority(map[int]int{1371: 3, 1370: 1}); !ok || v != 1371 { + t.Fatalf("majority(3-1) = %v %v", v, ok) + } + if _, ok := majority(map[int]int{1371: 2, 1370: 2}); ok { + t.Fatalf("tie must not resolve") + } + if _, ok := majority(map[int]int{1371: 1}); ok { + t.Fatalf("single answer must not resolve") + } + if v, ok := majority(map[string]int{"0x1": 2}); !ok || v != "0x1" { + t.Fatalf("majority(2-0) = %v %v", v, ok) + } +} diff --git a/harnesses/rpc-capabilities/cmd/script/integrity.go b/harnesses/rpc-capabilities/cmd/script/integrity.go new file mode 100644 index 00000000..e8663ed4 --- /dev/null +++ b/harnesses/rpc-capabilities/cmd/script/integrity.go @@ -0,0 +1,304 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "time" +) + +// integrity.go — bench 083 rpc-reliability, fixed-vector correctness +// checks. Every 5 minutes ONE chain (rotating through the set below) +// gets the same two questions asked of every provider serving it: +// +// logs — `eth_getLogs` for the chain's canonical USDC contract +// over a 10-block window ending at tip-N. Completeness: +// a provider silently dropping logs (or blocking the +// method, or gating the depth behind a paid tier) shows +// up against the cross-provider majority count. +// balance — `eth_getBalance` of a fixed well-known address at the +// same tip-N block. Consistency: the hex answer must be +// byte-identical across providers; the comparison works +// whether or not the balance is non-zero. +// +// Anti-gaming: N rotates daily over {20, 30, 40, 50, 60} blocks so a +// provider cannot special-case a fixed range, and every request id +// rotates (same edge-cache defeat as the latency probe). All depths +// stay far inside non-archive territory so pruned-but-honest nodes +// are never penalized; a provider that gates even 60-blocks-deep data +// behind a key (observed live: publicnode -32602 "Archive requests +// require a personal token") is emitting exactly the signal this +// bench exists to record. +// +// Errors ARE signal: result="error" on rpc_integrity_check_total is +// counted as an incident by the spec, because "method blocked" and +// "depth gated" are reliability failures from the caller's seat. + +const ( + integrityInterval = 5 * time.Minute + integrityTimeout = 15 * time.Second + // integrityCallSpacing: sequential per-provider spacing; meowrpc + // 429s on rapid bursts (same lesson as archive.go). + integrityCallSpacing = 1500 * time.Millisecond + // integrityLogsSpan: width of the getLogs window, in blocks. + integrityLogsSpan uint64 = 10 + // integrityBalanceAddr: same well-known address the archive loop + // uses (Vitalik). The check compares answers across providers + // byte-for-byte, so it is divergence we measure, not the value. + integrityBalanceAddr = archiveTestAddr +) + +// integrityVectors maps chain slug -> canonical USDC contract used as +// the fixed eth_getLogs vector. Only chains with a heavily traded +// canonical USDC participate (a quiet contract would return 0 logs +// everywhere and the completeness check would be vacuous). Native +// Circle deployments except BNB (Binance-peg, still the busiest +// USDC-family contract there). +var integrityVectors = map[string]string{ + "ethereum": "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48", + "arbitrum": "0xaf88d065e77c8cC2239327C5EDb3A432268e5831", + "optimism": "0x0b2C639c533813f4Aa9D7837CAf62653d097Ff85", + "base": "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", + "polygon": "0x3c499c542cEF5E3811e1192ce70d8cC03d5c3359", + "bnb": "0x8AC76a51cc950d9822D68b83fE1Ad97B32Cd580d", + "avalanche": "0xB97EF9Ef8734C71904D8002F8b6Bc66Dd9c48a6E", +} + +// integrityOffset returns today's tip-N base depth: 20 + 10*(day%5), +// i.e. {20, 30, 40, 50, 60}, rotating at UTC midnight. +func integrityOffset() uint64 { + day := uint64(time.Now().UTC().Unix() / 86400) + return 20 + 10*(day%5) +} + +// StartIntegrityLoop rotates through the vector-equipped chains, one +// chain per 5-minute tick. Non-blocking; spawns its own goroutine. +func StartIntegrityLoop(ctx context.Context) { + var targets []Chain + for _, c := range chains() { + if c.Kind == "" && integrityVectors[c.Slug] != "" { + targets = append(targets, c) + } + } + if len(targets) == 0 { + return + } + initIntegrityMetrics(targets) + + go func() { + // Let the latency loop populate chainTips first: the vector + // block range is derived from the cross-provider tip. + select { + case <-ctx.Done(): + return + case <-time.After(90 * time.Second): + } + i := 0 + integrityTick(ctx, targets[i%len(targets)]) + i++ + t := time.NewTicker(integrityInterval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + integrityTick(ctx, targets[i%len(targets)]) + i++ + } + } + }() +} + +type integrityObs struct { + p Provider + logsN int + logsErr error + balHex string + balErr error +} + +func integrityTick(ctx context.Context, c Chain) { + tip := tips.get(c.Slug) + off := integrityOffset() + if tip <= off+integrityLogsSpan { + fmt.Printf("[integrity/%s] no tip yet (tip=%d), skipping round\n", c.Slug, tip) + return + } + to := tip - off + from := to - (integrityLogsSpan - 1) + + obs := make([]integrityObs, 0, len(c.Providers)) + for _, p := range c.Providers { + o := integrityObs{p: p} + o.logsN, o.logsErr = fetchLogsCount(ctx, p.URL, integrityVectors[c.Slug], from, to) + if !integritySleep(ctx) { + return + } + o.balHex, o.balErr = fetchBalanceHex(ctx, p.URL, integrityBalanceAddr, to) + obs = append(obs, o) + if !integritySleep(ctx) { + return + } + } + + // Logs completeness: strict majority of the successful counts. + logsCounts := make(map[int]int) + for _, o := range obs { + if o.logsErr == nil { + logsCounts[o.logsN]++ + } + } + majLogs, logsQuorum := majority(logsCounts) + for _, o := range obs { + switch { + case o.logsErr != nil: + rpcLogsCount.DeleteLabelValues(o.p.Slug, c.Slug, currentRegion) + rpcIntegrityCheck.WithLabelValues(o.p.Slug, c.Slug, currentRegion, "logs", "error").Inc() + fmt.Printf("[integrity/%s/%s] logs range=%d-%d err=%v\n", c.Slug, o.p.Slug, from, to, o.logsErr) + case logsQuorum && o.logsN != majLogs: + rpcLogsCount.WithLabelValues(o.p.Slug, c.Slug, currentRegion).Set(float64(o.logsN)) + rpcLogsDisagreement.WithLabelValues(o.p.Slug, c.Slug, currentRegion).Inc() + rpcIntegrityCheck.WithLabelValues(o.p.Slug, c.Slug, currentRegion, "logs", "disagree").Inc() + fmt.Printf("[integrity/%s/%s] logs DISAGREE got=%d majority=%d range=%d-%d\n", c.Slug, o.p.Slug, o.logsN, majLogs, from, to) + default: + // No >=2 quorum this round (too few successes) also lands + // here: a lone answer is unverifiable, not wrong. + rpcLogsCount.WithLabelValues(o.p.Slug, c.Slug, currentRegion).Set(float64(o.logsN)) + rpcIntegrityCheck.WithLabelValues(o.p.Slug, c.Slug, currentRegion, "logs", "ok").Inc() + fmt.Printf("[integrity/%s/%s] logs ok count=%d range=%d-%d\n", c.Slug, o.p.Slug, o.logsN, from, to) + } + } + + // State consistency: strict majority of the raw hex answers. + balCounts := make(map[string]int) + for _, o := range obs { + if o.balErr == nil { + balCounts[o.balHex]++ + } + } + majBal, balQuorum := majority(balCounts) + for _, o := range obs { + switch { + case o.balErr != nil: + rpcIntegrityCheck.WithLabelValues(o.p.Slug, c.Slug, currentRegion, "balance", "error").Inc() + fmt.Printf("[integrity/%s/%s] balance block=%d err=%v\n", c.Slug, o.p.Slug, to, o.balErr) + case balQuorum && o.balHex != majBal: + rpcStateDisagreement.WithLabelValues(o.p.Slug, c.Slug, currentRegion).Inc() + rpcIntegrityCheck.WithLabelValues(o.p.Slug, c.Slug, currentRegion, "balance", "disagree").Inc() + fmt.Printf("[integrity/%s/%s] balance DISAGREE got=%s majority=%s block=%d\n", c.Slug, o.p.Slug, o.balHex, majBal, to) + default: + rpcIntegrityCheck.WithLabelValues(o.p.Slug, c.Slug, currentRegion, "balance", "ok").Inc() + fmt.Printf("[integrity/%s/%s] balance ok block=%d\n", c.Slug, o.p.Slug, to) + } + } +} + +func integritySleep(ctx context.Context) bool { + select { + case <-ctx.Done(): + return false + case <-time.After(integrityCallSpacing): + return true + } +} + +// majority returns the value backed by >=2 votes and strictly more +// than any competing value; ok=false when no such strict majority +// exists (all-distinct answers, or a tie). +func majority[T comparable](counts map[T]int) (T, bool) { + var best T + bestN, secondN := 0, 0 + for v, n := range counts { + if n > bestN { + best, secondN, bestN = v, bestN, n + } else if n > secondN { + secondN = n + } + } + return best, bestN >= 2 && bestN > secondN +} + +func fetchLogsCount(ctx context.Context, url, addr string, from, to uint64) (int, error) { + body := fmt.Sprintf( + `{"jsonrpc":"2.0","method":"eth_getLogs","params":[{"fromBlock":"0x%x","toBlock":"0x%x","address":"%s"}],"id":%d}`, + from, to, addr, time.Now().UnixNano(), + ) + raw, err := integrityPost(ctx, url, body) + if err != nil { + return 0, err + } + var r rpcBlockEnvelope + if err := json.Unmarshal(raw, &r); err != nil { + return 0, err + } + if r.Error != nil { + return 0, fmt.Errorf("rpc %d: %s", r.Error.Code, r.Error.Message) + } + var logs []json.RawMessage + if err := json.Unmarshal(r.Result, &logs); err != nil { + return 0, fmt.Errorf("non-array result") + } + return len(logs), nil +} + +func fetchBalanceHex(ctx context.Context, url, addr string, block uint64) (string, error) { + body := fmt.Sprintf( + `{"jsonrpc":"2.0","method":"eth_getBalance","params":["%s","0x%x"],"id":%d}`, + addr, block, time.Now().UnixNano(), + ) + raw, err := integrityPost(ctx, url, body) + if err != nil { + return "", err + } + var r rpcEnvelope + if err := json.Unmarshal(raw, &r); err != nil { + return "", err + } + if r.Error != nil { + return "", fmt.Errorf("rpc %d: %s", r.Error.Code, r.Error.Message) + } + if r.Result == "" { + return "", fmt.Errorf("empty result") + } + return r.Result, nil +} + +func integrityPost(ctx context.Context, url, body string) ([]byte, error) { + c, cancel := context.WithTimeout(ctx, integrityTimeout) + defer cancel() + req, _ := http.NewRequestWithContext(c, "POST", url, bytes.NewReader([]byte(body))) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("User-Agent", "OpenChainBench/1.0 (+https://openchainbench.com)") + client := &http.Client{Timeout: integrityTimeout} + resp, err := client.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode != 200 { + _, _ = io.Copy(io.Discard, resp.Body) + return nil, fmt.Errorf("http %d", resp.StatusCode) + } + return io.ReadAll(resp.Body) +} + +// initIntegrityMetrics zero-initializes every counter in the vector +// matrix so the bench's `increase()` queries read 0, not absent, for +// clean providers. +func initIntegrityMetrics(targets []Chain) { + for _, c := range targets { + for _, p := range c.Providers { + for _, check := range []string{"logs", "balance"} { + for _, result := range []string{"ok", "error", "disagree"} { + rpcIntegrityCheck.WithLabelValues(p.Slug, c.Slug, currentRegion, check, result).Add(0) + } + } + rpcLogsDisagreement.WithLabelValues(p.Slug, c.Slug, currentRegion).Add(0) + rpcStateDisagreement.WithLabelValues(p.Slug, c.Slug, currentRegion).Add(0) + } + } +} diff --git a/harnesses/rpc-capabilities/cmd/script/main.go b/harnesses/rpc-capabilities/cmd/script/main.go index 9bea8647..9b18b764 100644 --- a/harnesses/rpc-capabilities/cmd/script/main.go +++ b/harnesses/rpc-capabilities/cmd/script/main.go @@ -73,11 +73,16 @@ func main() { } } fmt.Println() - fmt.Println("Metrics server: :2112/metrics") + + // OCB_METRICS_ADDR: explicit local-run override only (e.g. a dev + // box where :2112 is taken). We still ignore Railway's injected + // $PORT on purpose; Prometheus scrapes :2112 in prod. + addr := envDefault("OCB_METRICS_ADDR", ":2112") + fmt.Printf("Metrics server: %s/metrics\n", addr) fmt.Println() go func() { - if err := StartMetricsServer(":2112"); err != nil { + if err := StartMetricsServer(addr); err != nil { fmt.Printf("[fatal] metrics server: %v\n", err) os.Exit(1) } @@ -88,6 +93,7 @@ func main() { StartProbeLoop(ctx) StartArchiveLoop(ctx) + StartIntegrityLoop(ctx) sig := make(chan os.Signal, 1) signal.Notify(sig, os.Interrupt, syscall.SIGTERM) diff --git a/harnesses/rpc-capabilities/cmd/script/metrics.go b/harnesses/rpc-capabilities/cmd/script/metrics.go index df15637f..304c860e 100644 --- a/harnesses/rpc-capabilities/cmd/script/metrics.go +++ b/harnesses/rpc-capabilities/cmd/script/metrics.go @@ -57,6 +57,57 @@ var ( }, []string{"provider", "chain", "region", "depth"}, ) + + // ─── Bench 083 rpc-reliability: correctness / integrity metrics ──── + // Emitted from the same probe matrix; consensus.go + integrity.go. + + rpcConsensusLag = promauto.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "rpc_consensus_lag_blocks", + Help: "Blocks (slots on Solana) between this provider's reported head and the highest head any probed provider reported for the same chain. Set on every valid head probe (ok or stale); deleted on failure so a dead endpoint ages out instead of freezing at its last lag.", + }, + []string{"provider", "chain", "region"}, + ) + + rpcHashMismatch = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "rpc_hash_mismatch_total", + Help: "Times a provider reported a block hash at height H that disagrees with the hash at least 2 other-or-same providers agreed on at H (strict plurality). Incremented at most once per (provider, height); a 2-2 reorg split yields no quorum and nobody is counted.", + }, + []string{"provider", "chain", "region"}, + ) + + rpcLogsCount = promauto.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "rpc_logs_count", + Help: "Number of logs the provider returned for the fixed-vector `eth_getLogs` check (canonical USDC contract, 10-block range at a daily-rotating depth behind tip). Deleted when the call errors so a blocked method doesn't freeze a stale count.", + }, + []string{"provider", "chain", "region"}, + ) + + rpcLogsDisagreement = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "rpc_logs_disagreement_total", + Help: "Times a provider's `eth_getLogs` count for the fixed vector deviated from the strict cross-provider majority count in the same round.", + }, + []string{"provider", "chain", "region"}, + ) + + rpcStateDisagreement = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "rpc_state_disagreement_total", + Help: "Times a provider's `eth_getBalance` hex at the fixed recent block was not byte-identical to the strict cross-provider majority answer in the same round.", + }, + []string{"provider", "chain", "region"}, + ) + + rpcIntegrityCheck = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "rpc_integrity_check_total", + Help: "Fixed-vector integrity check outcomes. check is `logs` (eth_getLogs USDC 10-block window) or `balance` (eth_getBalance at a fixed recent block). result is ok (matches majority), error (method blocked, archive gated, transport failure: errors are signal), or disagree (diverged from the >=2-provider majority).", + }, + []string{"provider", "chain", "region", "check", "result"}, + ) ) // StartMetricsServer binds /metrics + /health on addr. Blocking call — diff --git a/harnesses/rpc-capabilities/cmd/script/probe.go b/harnesses/rpc-capabilities/cmd/script/probe.go index 96859460..a8b92ddb 100644 --- a/harnesses/rpc-capabilities/cmd/script/probe.go +++ b/harnesses/rpc-capabilities/cmd/script/probe.go @@ -88,6 +88,7 @@ type rpcBlockEnvelope struct { type blockHeader struct { Number string `json:"number"` + Hash string `json:"hash"` } // callLatestBlock issues `eth_getBlockByNumber("latest", false)` and @@ -100,8 +101,9 @@ type blockHeader struct { // node, which let cache-fronted gateways top the latency leaderboard // on cache hits rather than real RPC work. Fetching the full latest // header with a rotating request id defeats body-keyed edge caches; -// the header's `number` field keeps the staleness check intact. -func callLatestBlock(ctx context.Context, url string) (block uint64, result string, latencyMs float64, err error) { +// the header's `number` field keeps the staleness check intact and its +// `hash` feeds the bench-083 cross-provider quorum map (consensus.go). +func callLatestBlock(ctx context.Context, url string) (block uint64, hash string, result string, latencyMs float64, err error) { body := []byte(fmt.Sprintf( `{"jsonrpc":"2.0","method":"eth_getBlockByNumber","params":["latest",false],"id":%d}`, time.Now().UnixNano(), @@ -117,48 +119,49 @@ func callLatestBlock(ctx context.Context, url string) (block uint64, result stri if err != nil { if ctx.Err() != nil || strings.Contains(err.Error(), "deadline exceeded") || strings.Contains(err.Error(), "Timeout") { - return 0, "timeout", latencyMs, err + return 0, "", "timeout", latencyMs, err } - return 0, "http_err", latencyMs, err + return 0, "", "http_err", latencyMs, err } defer resp.Body.Close() if resp.StatusCode != 200 { _, _ = io.Copy(io.Discard, resp.Body) - return 0, "http_err", latencyMs, fmt.Errorf("status %d", resp.StatusCode) + return 0, "", "http_err", latencyMs, fmt.Errorf("status %d", resp.StatusCode) } raw, err := io.ReadAll(resp.Body) if err != nil { - return 0, "http_err", latencyMs, err + return 0, "", "http_err", latencyMs, err } var r rpcBlockEnvelope if err := json.Unmarshal(raw, &r); err != nil { - return 0, "http_err", latencyMs, err + return 0, "", "http_err", latencyMs, err } if r.Error != nil { - return 0, "jsonrpc_err", latencyMs, fmt.Errorf("rpc -%d: %s", r.Error.Code, r.Error.Message) + return 0, "", "jsonrpc_err", latencyMs, fmt.Errorf("rpc -%d: %s", r.Error.Code, r.Error.Message) } if len(r.Result) == 0 || string(r.Result) == "null" { - return 0, "jsonrpc_err", latencyMs, fmt.Errorf("empty result") + return 0, "", "jsonrpc_err", latencyMs, fmt.Errorf("empty result") } var hdr blockHeader if err := json.Unmarshal(r.Result, &hdr); err != nil { - return 0, "jsonrpc_err", latencyMs, err + return 0, "", "jsonrpc_err", latencyMs, err } if hdr.Number == "" { - return 0, "jsonrpc_err", latencyMs, fmt.Errorf("header missing number") + return 0, "", "jsonrpc_err", latencyMs, fmt.Errorf("header missing number") } n, err := strconv.ParseUint(strings.TrimPrefix(hdr.Number, "0x"), 16, 64) if err != nil { - return 0, "jsonrpc_err", latencyMs, err + return 0, "", "jsonrpc_err", latencyMs, err } - return n, "ok", latencyMs, nil + return n, hdr.Hash, "ok", latencyMs, nil } // StartProbeLoop spawns one goroutine per (chain × provider). Each // goroutine runs forever, ticking every probeInterval. func StartProbeLoop(ctx context.Context) { + initConsensusMetrics() for _, c := range chains() { c := c for _, p := range c.Providers { @@ -185,6 +188,7 @@ func probeOne(ctx context.Context, c Chain, p Provider) { probeCtx, cancel := context.WithTimeout(ctx, probeTimeout) defer cancel() var block uint64 + var hash string var result string var latency float64 var err error @@ -192,9 +196,9 @@ func probeOne(ctx context.Context, c Chain, p Provider) { case "solana": block, result, latency, err = callLatestSlot(probeCtx, p.URL) case "polkadot": - block, result, latency, err = callSubstrateHeader(probeCtx, p.URL) + block, hash, result, latency, err = callSubstrateHeader(probeCtx, p.URL) default: - block, result, latency, err = callLatestBlock(probeCtx, p.URL) + block, hash, result, latency, err = callLatestBlock(probeCtx, p.URL) } if result == "ok" { @@ -211,6 +215,23 @@ func probeOne(ctx context.Context, c Chain, p Provider) { result = "stale" } } + // Bench 083: valid observations (fresh or stale, both carry a + // real height + hash) feed the consensus lag gauge and the + // height→hash quorum map; anything else deletes the lag series + // so a dead endpoint ages out instead of freezing. Skipped on + // chains whose probe cannot return the current-block hash + // (Solana: getSlot returns a number only; Polkadot: header + // carries parentHash, not the current block hash). + switch c.Kind { + case "solana", "polkadot": + // no consensus participation + default: + if result == "ok" || result == "stale" { + consensus.observe(c.Slug, p.Slug, block, hash) + } else { + rpcConsensusLag.DeleteLabelValues(p.Slug, c.Slug, currentRegion) + } + } rpcCallTotal.WithLabelValues(p.Slug, c.Slug, currentRegion, result).Inc() if result == "ok" { // Latency is recorded ONLY for fresh, valid responses. Error @@ -309,19 +330,24 @@ func callLatestSlot(ctx context.Context, url string) (slot uint64, result string } // substrateHeader is the shape of the `chain_getHeader` result on -// Polkadot and every Substrate-based relay chain. The block number is -// hex-encoded (`0x` prefix) matching the EVM header convention, so the -// parse path reuses the same strconv rule. +// Polkadot / Kusama and every Substrate-based relay chain. The block +// number is hex-encoded (`0x` prefix) matching the EVM header convention, +// so the parse path reuses the same strconv rule. `parentHash` is here +// so the consensus.go quorum map can key on it identically to EVM +// (bench 083 cross-provider height-hash agreement). type substrateHeader struct { - Number string `json:"number"` + Number string `json:"number"` + ParentHash string `json:"parentHash"` } // callSubstrateHeader is the Polkadot probe path: chain_getHeader with a // rotating request id (same anti-cache rule as the EVM header fetch). -// Returns the relay-chain block height so the caller can plug into the -// same tips machinery the EVM path uses; staleness classification in -// probeOne uses polkadotStaleBlockGap. -func callSubstrateHeader(ctx context.Context, url string) (block uint64, result string, latencyMs float64, err error) { +// Returns the block number and parent hash so the caller can plug the +// probe result into the same tips / consensus machinery the EVM chain +// path uses. The returned "block" is the relay-chain block height, not +// a chain-agnostic slot: staleness classification uses +// polkadotStaleBlockGap in probeOne. +func callSubstrateHeader(ctx context.Context, url string) (block uint64, hash string, result string, latencyMs float64, err error) { body := []byte(fmt.Sprintf( `{"jsonrpc":"2.0","method":"chain_getHeader","params":[],"id":%d}`, time.Now().UnixNano(), @@ -337,41 +363,41 @@ func callSubstrateHeader(ctx context.Context, url string) (block uint64, result if err != nil { if ctx.Err() != nil || strings.Contains(err.Error(), "deadline exceeded") || strings.Contains(err.Error(), "Timeout") { - return 0, "timeout", latencyMs, err + return 0, "", "timeout", latencyMs, err } - return 0, "http_err", latencyMs, err + return 0, "", "http_err", latencyMs, err } defer resp.Body.Close() if resp.StatusCode != 200 { _, _ = io.Copy(io.Discard, resp.Body) - return 0, "http_err", latencyMs, fmt.Errorf("status %d", resp.StatusCode) + return 0, "", "http_err", latencyMs, fmt.Errorf("status %d", resp.StatusCode) } raw, err := io.ReadAll(resp.Body) if err != nil { - return 0, "http_err", latencyMs, err + return 0, "", "http_err", latencyMs, err } var r rpcBlockEnvelope if err := json.Unmarshal(raw, &r); err != nil { - return 0, "http_err", latencyMs, err + return 0, "", "http_err", latencyMs, err } if r.Error != nil { - return 0, "jsonrpc_err", latencyMs, fmt.Errorf("rpc -%d: %s", r.Error.Code, r.Error.Message) + return 0, "", "jsonrpc_err", latencyMs, fmt.Errorf("rpc -%d: %s", r.Error.Code, r.Error.Message) } if len(r.Result) == 0 || string(r.Result) == "null" { - return 0, "jsonrpc_err", latencyMs, fmt.Errorf("empty result") + return 0, "", "jsonrpc_err", latencyMs, fmt.Errorf("empty result") } var hdr substrateHeader if err := json.Unmarshal(r.Result, &hdr); err != nil { - return 0, "jsonrpc_err", latencyMs, err + return 0, "", "jsonrpc_err", latencyMs, err } if hdr.Number == "" { - return 0, "jsonrpc_err", latencyMs, fmt.Errorf("substrate header missing number") + return 0, "", "jsonrpc_err", latencyMs, fmt.Errorf("substrate header missing number") } n, err := strconv.ParseUint(strings.TrimPrefix(hdr.Number, "0x"), 16, 64) if err != nil { - return 0, "jsonrpc_err", latencyMs, err + return 0, "", "jsonrpc_err", latencyMs, err } - return n, "ok", latencyMs, nil + return n, hdr.ParentHash, "ok", latencyMs, nil } diff --git a/harnesses/rpc-capabilities/go.mod b/harnesses/rpc-capabilities/go.mod index 74e2e1fb..3c5d672c 100644 --- a/harnesses/rpc-capabilities/go.mod +++ b/harnesses/rpc-capabilities/go.mod @@ -8,6 +8,7 @@ require ( github.com/beorn7/perks v1.0.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/kr/text v0.2.0 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.66.1 // indirect