rpc-capabilities: cherry-pick integrity + consensus code to main (unblocks bench 083)

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 {

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 {

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

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