Add a Collector, owned by the Aggregator (process lifetime), that aggregates
KV-read metrics across every read path. Producers fill their own *DomainMetrics
lock-free and hand ownership over a buffered channel tagged with a Source
(exec/commitment/warmup/rpc/engine); a single collector goroutine folds them
into map[Source]*DomainMetrics with no lock or atomics on the aggregate. The
goroutine also self-publishes source-labelled Prometheus gauges
(kv_read_count / kv_read_duration_ns, labels {source,domain,op}) on a ticker —
process-level and independent of whether a block is executing.

Wiring:
- Aggregator owns the Collector: Start in newAggregator (on a.wg), Stop+drain
  in Close before wg.Wait. Exposed to SharedDomains via the duck-typed
  kvmetrics.MetricsCollectorProvider on *AggregatorRoTx (same pattern as
  BranchCacheProvider), so the leaf kvmetrics package stays cycle-free.
- SharedDomains.MergeMetrics(source, wm) now hands a finished worker's metrics
  to BOTH sinks: the per-batch sd.metrics (under one lock, for the existing log
  line) and the collector (lock-free, for Prometheus). Ownership of wm transfers
  to the collector, so producers allocate a fresh instance afterwards.
- Producers tagged: exec workers (SourceExec, per task), the commitment fold
  (SourceCommitment), trie warmup and concurrent mount (SourceWarmup).
- AsGetterCollected(tx, source) gives concurrent short-lived callers (RPC,
  engine) a per-getter instance + flush closure; gated on KVReadLevelledMetrics.

The new gauges are additive — the existing exec-scoped mxExec* gauges and the
per-batch log line are unchanged. The memBatch put-path (CachePut*) is left on
the existing shared aggregate deliberately: those counters are load-bearing for
SizeEstimate's flush accounting, so moving them belongs in a separate change.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Add a request-scoped read accumulator to SharedDomains
(StartRequestMetrics(source) / flush at Close) so a single-goroutine RPC handler
that reads through the plain AsGetter is metered without a shared accumulator or
a per-getter flush. getLatestMetered folds nil-metrics reads into it; this
short-circuits for exec workers (which pass their own per-worker instance), so
there is no cross-goroutine access to the request accumulator.

Wire eth_simulation's SimulateV1 to tag its reads as SourceRPC. Engine block
execution is already metered as SourceExec (it runs through the exec workers);
SourceEngine and the per-read-getter paths (exec_module CacheView, vm/runtime,
which build a getter per read) are left for a follow-up that needs a view-level
accumulator.

Also make Collector.Snapshot drain the buffered samples first so a snapshot
reflects everything sent so far, and add a -race collector test covering
concurrent Send + Snapshot + Close-drain.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

…roducers

The exec hot path must never block on metrics and must never lose counts. A
buffered channel alone satisfies neither at the boundary: a full buffer blocks a
plain send, and a non-blocking send drops. Resolve both with retain-on-full.

- Collector.TrySend is non-blocking and returns whether the sample was queued.
  Exec workers keep a retained accumulator (collectorAcc): each task folds its
  reads in and TrySends; on a full buffer the send is skipped and the worker
  keeps adding to the same accumulator, retrying next task. A single blocking
  flush at worker exit drains the remainder (off the hot path, lossless). So
  execution never waits on metrics and no count is dropped.
- Collector.Send is the blocking variant for low-frequency boundary producers
  (commitment fold, warmup teardown, an RPC request closing) where a rare brief
  wait is fine and losing the sample is not.
- SharedDomains.LogMergeMetrics folds a task into the per-batch sd.metrics log
  aggregate only; the exec path uses it each task and feeds the collector
  separately via the retained accumulator, so the two sinks (which reset on
  different conditions) never double-count.
- Dropped the drop counter and its gauge — nothing is dropped now.

Validated: build, make lint (0 issues), go test -race ./db/state/kvmetrics
(incl. a test proving TrySend never blocks on a full buffer), and hive cancun
with KV_READ_METRICS=true at the pinned ref = 226/0.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>