diff --git a/packages/agent/src/dkg-agent-base.ts b/packages/agent/src/dkg-agent-base.ts index e911d644f2..e0f0565add 100644 --- a/packages/agent/src/dkg-agent-base.ts +++ b/packages/agent/src/dkg-agent-base.ts @@ -96,7 +96,7 @@ import { pickNetworkTunables, } from '@origintrail-official/dkg-core'; import { GraphManager, PrivateContentStore, createTripleStore, isExternalBackend, type TripleStore, type TripleStoreConfig, type Quad, type LargeLiteralStorageConfig } from '@origintrail-official/dkg-storage'; -import { EVMChainAdapter, NoChainAdapter, enrichEvmError, buildKnowledgeAssetUal, type EVMAdapterConfig, type ChainAdapter, type CreateContextGraphParams, type CreateOnChainContextGraphParams, type CreateOnChainContextGraphResult, type TxResult, type V10PublishingConvictionAccountInfo } from '@origintrail-official/dkg-chain'; +import { emptyRpcUsageWindow, EVMChainAdapter, NoChainAdapter, enrichEvmError, buildKnowledgeAssetUal, type EVMAdapterConfig, type ChainAdapter, type CreateContextGraphParams, type CreateOnChainContextGraphParams, type CreateOnChainContextGraphResult, type TxResult, type V10PublishingConvictionAccountInfo, type RpcUsageWindow } from '@origintrail-official/dkg-chain'; import { DKGPublisher, PublishHandler, SharedMemoryHandler, UpdateHandler, ChainEventPoller, AccessHandler, AccessClient, PublishJournal, StaleWriteError, @@ -1405,4 +1405,15 @@ export class DKGAgentBase { this.publisher.setWorkspaceSenderKeyEncryptor((input) => (this as unknown as DKGAgent).encryptWorkspacePayloadWithSenderKey(input)); this.syncCheckpoints = config.syncCheckpointStore ?? this.syncCheckpoints; } + + /** + * RpcUsageDrainable: drain the chain adapter's raw JSON-RPC usage window + * (delta since the previous drain — the provider-billing unit). The + * adapter capability is optional; an adapter without it reports an empty + * window here, so consumers never see the optionality. Keeps `chain` + * protected instead of the daemon reaching through a cast. + */ + drainRpcUsage(): RpcUsageWindow { + return this.chain.drainRpcUsage?.() ?? emptyRpcUsageWindow(); + } } diff --git a/packages/agent/test/rpc-usage-boundary.test.ts b/packages/agent/test/rpc-usage-boundary.test.ts new file mode 100644 index 0000000000..d0328f2727 --- /dev/null +++ b/packages/agent/test/rpc-usage-boundary.test.ts @@ -0,0 +1,44 @@ +// SPDX-License-Identifier: Apache-2.0 +/** + * Review coverage gap (PR #1409): the agent's `drainRpcUsage()` is THE + * boundary the daemon telemetry consumes (`startRpcUsageTelemetry({ source: + * agent })`), and a delegation regression here would silence every rpc_usage + * log while the chain-side and cli-side tests stayed green. These drive the + * REAL method (prototype-invoked with a chain-bearing `this`, the established + * pattern for agent unit tests) against the real MockChainAdapter, a + * counts-bearing adapter, and an adapter lacking the capability. + */ +import { describe, it, expect } from 'vitest'; +import { DKGAgentBase } from '../src/dkg-agent-base.js'; +import { DKGAgent } from '../src/dkg-agent.js'; +import { MockChainAdapter } from '@origintrail-official/dkg-chain'; + +const drain = DKGAgentBase.prototype.drainRpcUsage; + +describe('DKGAgent.drainRpcUsage — the adapter→agent telemetry boundary', () => { + it('delegates to the adapter and passes the window through VERBATIM', () => { + const window = { byMethod: { eth_call: 42, eth_getLogs: 7 }, lifetimeTotal: 49 }; + const out = drain.call({ chain: { drainRpcUsage: () => window } } as never); + expect(out).toBe(window); // same object — no reshaping/copying in the boundary + }); + + it('returns the real MockChainAdapter empty window (capability present, no transport)', () => { + const chain = new MockChainAdapter(); + const out = drain.call({ chain } as never); + expect(out).toEqual({ byMethod: {}, lifetimeTotal: 0 }); + }); + + it('collapses a missing adapter capability to a concrete EMPTY window (consumers never see undefined)', () => { + const out = drain.call({ chain: {} } as never); + expect(out).toEqual({ byMethod: {}, lifetimeTotal: 0 }); + }); + + it('is inherited by the composed DKGAgent class (the daemon consumes agent, not the base)', () => { + // The daemon passes the DKGAgent instance as the telemetry source — the + // method must exist on the composed class's prototype chain. + expect(typeof DKGAgent.prototype.drainRpcUsage).toBe('function'); + const window = { byMethod: { eth_sendRawTransaction: 3 }, lifetimeTotal: 3 }; + const out = DKGAgent.prototype.drainRpcUsage.call({ chain: { drainRpcUsage: () => window } } as never); + expect(out).toBe(window); + }); +}); diff --git a/packages/chain/src/chain-adapter.ts b/packages/chain/src/chain-adapter.ts index 0473e8e5fc..d0c8444769 100644 --- a/packages/chain/src/chain-adapter.ts +++ b/packages/chain/src/chain-adapter.ts @@ -1,4 +1,5 @@ import type { ethers } from 'ethers'; +import type { RpcUsageWindow } from './rpc-usage.js'; /** * The Publishing-Conviction-Account read methods the funded-wallet selector @@ -797,6 +798,16 @@ export interface ChainAdapter { */ deploymentId: string; + /** + * OPTIONAL RPC-usage capability: drain the raw JSON-RPC request counts + * accumulated since the previous drain (a DELTA window — summing drains over + * time yields exact request totals, the provider-billing unit). The EVM + * adapter counts at its HTTP transport; the mock adapter returns an + * always-empty window (it has no RPC transport). Consumed by the daemon's + * minutely `rpc_usage` telemetry log line. + */ + drainRpcUsage?(): RpcUsageWindow; + // Identity registerIdentity(proof: IdentityProof): Promise; getIdentityId(): Promise; diff --git a/packages/chain/src/evm-adapter-base.ts b/packages/chain/src/evm-adapter-base.ts index 76831d97da..ced2a4dae8 100644 --- a/packages/chain/src/evm-adapter-base.ts +++ b/packages/chain/src/evm-adapter-base.ts @@ -25,6 +25,7 @@ import { resolveRpcUrls, boundedRetryFetchRequest, withTimeout, isRetryableRpcEr import { rpcHost } from './rpc-failover-log.js'; import { ChainRpcTransportError, createRpcTimeoutError } from './chain-rpc-transport-error.js'; import { RpcFailoverClient, type ReadOpts } from './rpc-failover-client.js'; +import { RpcUsageTracker, createCountingJsonRpcProvider, type RpcUsageWindow } from './rpc-usage.js'; import { computeApprovalAction, effectivePublishAllowance, V10_PUBLISH_ONCHAIN_MIN_ALLOWANCE } from './evm-adapter-allowance.js'; import { formatProviderContext } from './evm-adapter-types.js'; import type { ContractCache, EVMAdapterConfig } from './evm-adapter-types.js'; @@ -343,6 +344,8 @@ export class EVMChainAdapterBase { * the adapter and never owns tx-safety state. */ protected readonly rpcFailover: RpcFailoverClient; + /** Raw JSON-RPC request accounting (provider-billing unit). See rpc-usage.ts. */ + protected readonly rpcUsage: RpcUsageTracker; protected readonly filterErrorSilencer: FilterErrorSilencer; @@ -625,8 +628,16 @@ export class EVMChainAdapterBase { // single-RPC node keeps the bounded `RPC_REQUEST_MAX_RETRIES` retry (its // only resilience; #894) via the default. See `boundedRetryFetchRequest`. const perEndpointRetries = this.rpcUrls.length > 1 ? 0 : undefined; + // CountingJsonRpcProvider reports every raw JSON-RPC request to the usage + // tracker (the PROVIDER-BILLING unit — see rpc-usage.ts). With + // `batchMaxCount: 1` below, one send() == one HTTP request, so the count is + // exact. `this.chainId` is assigned later in this constructor → live thunk. + this.rpcUsage = new RpcUsageTracker(() => this.chainId); + // One transport factory wires BOTH billing-exact accounting hooks (first + // attempt at `_send` + every ethers-internal retry attempt) to the tracker — + // see createCountingJsonRpcProvider for the invariant. this.providers = this.rpcUrls.map( - (url) => new JsonRpcProvider(boundedRetryFetchRequest(url, perEndpointRetries), undefined, { + (url) => createCountingJsonRpcProvider(url, perEndpointRetries, this.rpcUsage, { cacheTimeout: -1, polling: true, batchMaxCount: 1, @@ -2734,6 +2745,17 @@ export class EVMChainAdapterBase { return [...this.rpcUrls]; } + /** + * Drain the raw JSON-RPC request counts accumulated since the previous drain + * (delta window) — consumed by the daemon's minutely `rpc_usage` telemetry + * log line. Part of the optional `ChainAdapter.drainRpcUsage` capability + * (the mock adapter implements it as an always-empty window — it has no RPC + * transport). + */ + drainRpcUsage(): RpcUsageWindow { + return this.rpcUsage.drainWindow(); + } + async getContract(name: string): Promise { await this.init(); return this.resolveContract(name); diff --git a/packages/chain/src/evm-adapter-rpc.ts b/packages/chain/src/evm-adapter-rpc.ts index 3132f56b5e..896143a852 100644 --- a/packages/chain/src/evm-adapter-rpc.ts +++ b/packages/chain/src/evm-adapter-rpc.ts @@ -95,7 +95,7 @@ export function boundedRetryFetchRequest( maxRetries: number = RPC_REQUEST_MAX_RETRIES, ): FetchRequest { const req = new FetchRequest(url); - req.retryFunc = async (_req, _response, attempt) => { + req.retryFunc = async (_attemptReq, _response, attempt) => { if (attempt >= maxRetries) return false; await sleep(Math.min(500 * (attempt + 1), RPC_REQUEST_RETRY_BACKOFF_CAP_MS)); return true; diff --git a/packages/chain/src/index.ts b/packages/chain/src/index.ts index eae7ebef00..52b0320034 100644 --- a/packages/chain/src/index.ts +++ b/packages/chain/src/index.ts @@ -1,4 +1,10 @@ export * from './chain-adapter.js'; +// RPC-usage accounting: ONLY the typed window contract is public API (consumed +// by the ChainAdapter.drainRpcUsage capability, the agent boundary, and the +// daemon's rpc_usage log emission). The parsing/label-bounding helpers stay +// package-internal (imported via ./rpc-usage.js) so the transport accounting +// implementation can change without a public-API break. +export { emptyRpcUsageWindow, mergeRpcUsageWindows, rpcUsageWindowTotal, type RpcUsageDrainable, type RpcUsageWindow } from './rpc-usage.js'; export { MockChainAdapter, MOCK_DEFAULT_SIGNER } from './mock-adapter.js'; export { EVMChainAdapter, diff --git a/packages/chain/src/mock-adapter.ts b/packages/chain/src/mock-adapter.ts index b6460c3813..75a92f79be 100644 --- a/packages/chain/src/mock-adapter.ts +++ b/packages/chain/src/mock-adapter.ts @@ -26,6 +26,7 @@ import type { NodePublishingConvictionAccount, VerifyACKIdentityResult, } from './chain-adapter.js'; +import type { RpcUsageWindow } from './rpc-usage.js'; import { NoEligibleContextGraphError, NoEligibleKnowledgeCollectionError, @@ -125,6 +126,11 @@ export class MockChainAdapter implements ChainAdapter { return []; } + /** RPC-usage capability: the mock has no RPC transport → always-empty window. */ + drainRpcUsage(): RpcUsageWindow { + return { byMethod: {}, lifetimeTotal: 0 }; + } + async ensureProfile(_options?: { nodeName?: string; stakeAmount?: bigint; lockTier?: number }): Promise { const existing = await this.getIdentityId(); if (existing > 0n) return existing; diff --git a/packages/chain/src/rpc-usage.ts b/packages/chain/src/rpc-usage.ts new file mode 100644 index 0000000000..b6e9e98f49 --- /dev/null +++ b/packages/chain/src/rpc-usage.ts @@ -0,0 +1,280 @@ +// SPDX-License-Identifier: Apache-2.0 +/** + * RAW JSON-RPC request accounting — the PROVIDER-BILLING view of chain usage. + * + * Why this exists (incident-driven): a node burned ~$200 of RPC credits in a + * day and nothing measured it. The existing `dkg.chain.rpc.total` counts + * LOGICAL chain operations (one contract read, one broadcast, …), but providers + * bill per raw JSON-RPC request, and one logical op fans out into several raw + * requests (a tx populate alone issues eth_estimateGas + eth_getTransactionCount + * + eth_chainId + fee reads). So credit burn must be counted at the TRANSPORT + * choke-point: `JsonRpcProvider.send(method, …)`. The adapter constructs its + * providers with `batchMaxCount: 1` (see evm-adapter-base), so one `send()` == + * one HTTP JSON-RPC request == one billable unit — the count is exact. + * + * Two consumers, one tracker: + * - OTel counter `dkg.chain.rpc.requests.total{rpc_method, chain_id}` — for + * the metrics backend once one is provisioned. + * - `drainWindow()` — per-window DELTA counts the daemon logs as structured + * `rpc_usage` lines every minute, which ride the already-deployed + * OTLP-logs → Alloy → Loki path so Grafana can chart RPC usage per node / + * per method TODAY, with exact sums (deltas, not cumulative gauges). + */ + +import { JsonRpcProvider } from 'ethers'; +import type { + FetchRequest, + Networkish, + JsonRpcApiProviderOptions, + JsonRpcPayload, + JsonRpcResult, +} from 'ethers'; +import { getMetrics } from '@origintrail-official/dkg-core'; +import { boundedRetryFetchRequest } from './evm-adapter-rpc.js'; + +/** + * The JSON-RPC methods our own code (via ethers v6) can issue. Used to BOUND + * the metric label — anything outside maps to 'other' so the label set can + * never grow unbounded. (Methods are self-generated, not peer input, so this is + * defensive; the raw method still appears verbatim in the rpc_usage log lines, + * where cardinality is not a concern.) + */ +export const KNOWN_RPC_METHODS: ReadonlySet = new Set([ + 'eth_chainId', + 'eth_blockNumber', + 'eth_call', + 'eth_estimateGas', + 'eth_gasPrice', + 'eth_maxPriorityFeePerGas', + 'eth_feeHistory', + 'eth_getBalance', + 'eth_getTransactionCount', + 'eth_getCode', + 'eth_getStorageAt', + 'eth_getLogs', + 'eth_getBlockByNumber', + 'eth_getBlockByHash', + 'eth_getTransactionByHash', + 'eth_getTransactionReceipt', + 'eth_sendRawTransaction', + 'eth_newFilter', + 'eth_getFilterChanges', + 'eth_uninstallFilter', + 'net_version', + 'web3_clientVersion', +]); + +/** Bound a method name for use as a metric label (unknown → 'other'). */ +export function boundedRpcMethodLabel(method: string): string { + return KNOWN_RPC_METHODS.has(method) ? method : 'other'; +} + +/** + * Extract the JSON-RPC method name(s) from an encoded request body (single + * payload or batch). Usage-accounting policy owned HERE, not by the transport + * retry code — the retry hook just forwards the body. Best-effort: anything + * unparseable yields ['other'] so a retry attempt is never lost from the count. + */ +export function jsonRpcMethodsFromBody(body: Uint8Array | null | undefined): string[] { + try { + if (!body || body.length === 0) return ['other']; + const parsed = JSON.parse(new TextDecoder().decode(body)); + const entries = Array.isArray(parsed) ? parsed : [parsed]; + const methods = entries.map((e: { method?: unknown }) => String(e?.method ?? 'other')); + return methods.length > 0 ? methods : ['other']; + } catch { + return ['other']; + } +} + +/** A fresh all-zero window — the identity element for merging and the value of "nothing to report". */ +export function emptyRpcUsageWindow(): RpcUsageWindow { + return { byMethod: {}, lifetimeTotal: 0 }; +} + +/** + * THE drain contract — one name for one concept, wherever a usage window can + * be drained from: an agent (delegates to its adapter), a publisher runtime + * (merges its per-wallet adapters), or the daemon's composite source. Deltas + * since the previous drain. Always a concrete window: capability-optionality + * exists only at the ChainAdapter API edge and is collapsed to + * emptyRpcUsageWindow() the moment it crosses into a drainable, so consumers + * never juggle "missing capability" vs "empty window". + */ +export interface RpcUsageDrainable { + drainRpcUsage(): RpcUsageWindow; +} + +/** + * Merge usage windows from multiple trackers into one (pure model operation: + * per-method sums, summed lifetimes). A process can own several + * chain adapters with independent trackers — one per configured RPC consumer + * (e.g. the agent's adapter plus one per publisher wallet); billing-exact + * accounting is the SUM across all of them. undefined inputs (an absent + * optional capability, a not-yet-started runtime) are skipped; the result is + * always a concrete window — empty when there is nothing to merge. + */ +export function mergeRpcUsageWindows( + ...windows: Array +): RpcUsageWindow { + const defined = windows.filter((w): w is RpcUsageWindow => w !== undefined); + if (defined.length === 0) return emptyRpcUsageWindow(); + const byMethod: Record = {}; + let lifetimeTotal = 0; + for (const w of defined) { + for (const [m, c] of Object.entries(w.byMethod)) byMethod[m] = (byMethod[m] ?? 0) + c; + lifetimeTotal += w.lifetimeTotal; + } + return { byMethod, lifetimeTotal }; +} + +export interface RpcUsageWindow { + /** + * Raw requests since the previous drain, keyed by the RAW JSON-RPC method + * name (NOT sanitized — full diagnostic fidelity is the point). Keys are + * bounded in COUNT, not value: at most {@link RpcUsageTracker.MAX_WINDOW_METHODS} + * distinct names per window, overflow aggregated under 'other'. Consumers + * must sanitize keys for their own sink (the cli logfmt formatter does). + */ + byMethod: Record; + /** Raw requests since process start (monotonic; NOT reset by drain). */ + lifetimeTotal: number; +} + +/** + * Total raw requests in a window — DERIVED from byMethod on demand. The + * window model deliberately stores no separate total, so an inconsistent + * {byMethod, total} pair is unrepresentable. + */ +export function rpcUsageWindowTotal(window: RpcUsageWindow): number { + let total = 0; + for (const count of Object.values(window.byMethod)) total += count; + return total; +} + +/** + * In-process accumulator for raw JSON-RPC request counts. `record()` is on the + * hot path of every RPC — it does one map increment and one counter add, never + * throws, and never touches the network. + */ +export class RpcUsageTracker { + private window = new Map(); + private lifetime = 0; + + constructor( + // Live thunk (matches RpcFailoverClient): the adapter assigns `chainId` + // after construction, so resolve it at record time. + private readonly chainId: () => string, + ) {} + + /** + * Count one raw JSON-RPC request. Called from the provider's `_send` and the + * FetchRequest retry hook. Window keys keep the RAW method name BY DESIGN: + * the log path's `method` is parsed at Loki query time (logfmt), so raw + * names cost no index cardinality and preserve full diagnostic fidelity + * (seeing the real `debug_traceTransaction` is the point). Only the METRIC + * label is bounded to the known set — Prometheus label cardinality is a + * real storage cost. As a spam guard, a window holds at most + * MAX_WINDOW_METHODS distinct raw keys; pathological method-name churn + * (buggy caller, hostile input) overflows into 'other' instead of emitting + * one rpc_usage log line per fabricated name every minute. + */ + static readonly MAX_WINDOW_METHODS = 64; + + record(method: string): void { + // Authoritative window/lifetime state first, OUTSIDE any try — pure map + // arithmetic that cannot realistically throw, and it must never be + // skipped because an OPTIONAL sink misbehaved. + const raw = typeof method === 'string' && method.length > 0 && method.length <= 128 ? method : 'other'; + const key = this.window.has(raw) || this.window.size < RpcUsageTracker.MAX_WINDOW_METHODS ? raw : 'other'; + this.window.set(key, (this.window.get(key) ?? 0) + 1); + this.lifetime += 1; + // Best-effort applies ONLY to the OTel side effect (and the chainId + // thunk it evaluates) — a throwing metrics backend must not break the + // RPC call, and the window above is already committed either way. + try { + getMetrics().chainRpcRequestsTotal.add(1, { + rpc_method: boundedRpcMethodLabel(method), + chain_id: this.chainId(), + }); + } catch { + /* metrics emission must never break an RPC call */ + } + } + + /** + * Return the DELTA since the previous drain and reset the window. Deltas (not + * cumulative totals) are what the daemon logs, so `sum_over_time` in Grafana + * yields exact request counts over any range. + */ + drainWindow(): RpcUsageWindow { + const byMethod: Record = {}; + for (const [method, count] of this.window) byMethod[method] = count; + this.window.clear(); + return { byMethod, lifetimeTotal: this.lifetime }; + } +} + +/** + * `JsonRpcProvider` that reports every outgoing JSON-RPC request to the + * tracker before delegating. The hook is `_send(payload)` — the ACTUAL wire + * dispatch in ethers v6 — not the higher-level `send()`, because ethers' + * network detection (`_detectNetwork` → eth_chainId) deliberately bypasses + * `send()` and would otherwise go uncounted. `_send` sees exactly what leaves + * the process: every payload entry is one billable JSON-RPC request (and with + * the adapter's `batchMaxCount: 1` a batch is a single-entry array anyway). + */ +export class CountingJsonRpcProvider extends JsonRpcProvider { + constructor( + url: string | FetchRequest, + network: Networkish | undefined, + options: JsonRpcApiProviderOptions | undefined, + private readonly onRpcRequest: (method: string) => void, + ) { + super(url, network, options); + } + + override _send(payload: JsonRpcPayload | Array): Promise> { + try { + const entries = Array.isArray(payload) ? payload : [payload]; + for (const entry of entries) this.onRpcRequest(String(entry?.method ?? 'unknown')); + } catch { + /* accounting must never break an RPC call */ + } + return super._send(payload); + } +} + +/** + * The ONE transport factory for a usage-counted provider: wires BOTH accounting + * hooks (the `_send` first-attempt count and the FetchRequest retry-attempt + * count) to the same tracker, so a construction site can never pair them + * inconsistently — the pairing IS the billing-exactness invariant (first + * attempt via `_send` + every ethers-internal retry attempt via the retry + * hook). Keeps the adapter constructor free of accounting mechanics. + */ +export function createCountingJsonRpcProvider( + url: string, + maxRetries: number | undefined, + tracker: RpcUsageTracker, + options: JsonRpcApiProviderOptions, +): CountingJsonRpcProvider { + // boundedRetryFetchRequest stays PURE retry policy; the accounting + // composition lives HERE, with the rest of the accounting. Ethers' throttle + // retries happen BELOW JsonRpcProvider._send (one dispatch can issue 1 + N + // HTTP attempts under 429/5xx) and every attempt bills at the provider, so + // the retryFunc is decorated to record each RE-attempt's methods; the first + // attempt is counted at _send by CountingJsonRpcProvider. + const fetchRequest = boundedRetryFetchRequest(url, maxRetries); + const pureRetry = fetchRequest.retryFunc!; + fetchRequest.retryFunc = async (attemptReq, response, attempt) => { + const retry = await pureRetry(attemptReq, response, attempt); + if (retry) { + try { + for (const method of jsonRpcMethodsFromBody(attemptReq?.body)) tracker.record(method); + } catch { /* accounting must never break the retry path */ } + } + return retry; + }; + return new CountingJsonRpcProvider(fetchRequest, undefined, options, (method) => tracker.record(method)); +} diff --git a/packages/chain/test/rpc-usage.unit.test.ts b/packages/chain/test/rpc-usage.unit.test.ts new file mode 100644 index 0000000000..ec28d1c8b8 --- /dev/null +++ b/packages/chain/test/rpc-usage.unit.test.ts @@ -0,0 +1,198 @@ +// SPDX-License-Identifier: Apache-2.0 +/** + * RPC usage accounting (the provider-billing view) — proven through the REAL + * transport path with the loopback JSON-RPC server as the SOURCE OF TRUTH: + * the tracker's counts must EQUAL the raw HTTP JSON-RPC requests the server + * actually received, total and per method — including ethers' internal + * 429-retry attempts, which happen below JsonRpcProvider._send and each bill + * at the provider. Also asserts the OTel counter's bounded {rpc_method, + * chain_id} labels, drain-resets-window semantics, and label bounding. + */ +import { describe, it, expect, afterEach } from 'vitest'; +import { metrics } from '@opentelemetry/api'; +import { + MeterProvider, + PeriodicExportingMetricReader, + InMemoryMetricExporter, + AggregationTemporality, +} from '@opentelemetry/sdk-metrics'; +import { rebuildMetrics } from '@origintrail-official/dkg-core'; +import { EVMChainAdapter, type EVMAdapterConfig } from '../src/evm-adapter.js'; +import { boundedRpcMethodLabel, mergeRpcUsageWindows, rpcUsageWindowTotal, RpcUsageTracker } from '../src/rpc-usage.js'; +import { startLoopbackRpc, type LoopbackRpc } from './loopback-rpc-harness.js'; + +const DEPLOYER_PK = '0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80'; +const HUB = '0x0000000000000000000000000000000000000001'; + +function minimalConfig(overrides: Partial = {}): EVMAdapterConfig { + return { + rpcUrl: 'http://127.0.0.1:1', + privateKey: DEPLOYER_PK, + hubAddress: HUB, + chainId: 'evm:31337', + allowNoAdminSigner: true, + ...overrides, + }; +} + +describe('RPC usage accounting — raw request counts EQUAL the server-received requests', () => { + let mp: MeterProvider | null = null; + let exporter: InMemoryMetricExporter; + const adapters: EVMChainAdapter[] = []; + const servers: LoopbackRpc[] = []; + + function installMeter(): void { + exporter = new InMemoryMetricExporter(AggregationTemporality.CUMULATIVE); + mp = new MeterProvider({ readers: [new PeriodicExportingMetricReader({ exporter, exportIntervalMillis: 60_000 })] }); + metrics.setGlobalMeterProvider(mp); + rebuildMetrics(); + } + + async function requestPoints(): Promise>> { + await mp!.forceFlush(); + const out: Array> = []; + for (const rm of exporter.getMetrics()) + for (const sm of rm.scopeMetrics) + for (const m of sm.metrics) + if (m.descriptor.name === 'dkg.chain.rpc.requests.total') + for (const dp of m.dataPoints) out.push(dp.attributes as Record); + return out; + } + + afterEach(async () => { + for (const a of adapters.splice(0)) { try { a.destroy(); } catch { /* idempotent */ } } + for (const s of servers.splice(0)) await s.close(); + if (mp) { await mp.forceFlush().catch(() => {}); await mp.shutdown().catch(() => {}); mp = null; } + metrics.disable(); + rebuildMetrics(); + }); + + it('EXACT: tracker total + per-method counts equal what the server received; drain resets', async () => { + installMeter(); + const rpc = await startLoopbackRpc(); + servers.push(rpc); + const a: any = new EVMChainAdapter(minimalConfig({ rpcUrl: rpc.url })); + adapters.push(a); + + await expect(a.getEvmChainId()).resolves.toBe(31337n); + await expect(a.getEvmChainId()).resolves.toBe(31337n); + + // Source of truth: the server's own per-method request log. The tracker's + // window must EQUAL it — no over- or under-counting. + const usage = a.drainRpcUsage(); + expect(rpcUsageWindowTotal(usage)).toBe(rpc.totalHits()); + expect(rpcUsageWindowTotal(usage)).toBeGreaterThanOrEqual(1); // guard against a 0==0 vacuous pass + for (const [method, count] of Object.entries(usage.byMethod)) { + expect(rpc.hits(method), `method ${method}`).toBe(count); + } + // ...and per-method the other way: every server-observed method was counted. + // (The harness has no method-list accessor, so probe the ones this read issues.) + expect(usage.byMethod['eth_chainId'] ?? 0).toBe(rpc.hits('eth_chainId')); + + // OTel counter: bounded labels only. + const pts = await requestPoints(); + expect(pts.length).toBeGreaterThanOrEqual(1); + const keys = new Set(pts.flatMap((p) => Object.keys(p))); + expect([...keys].sort()).toEqual(['chain_id', 'rpc_method']); + expect(pts.some((p) => p.rpc_method === 'eth_chainId' && p.chain_id === 'evm:31337')).toBe(true); + + // Drain semantics: deltas, not cumulative; lifetime monotonic. + const drained = a.drainRpcUsage(); + expect(rpcUsageWindowTotal(drained)).toBe(0); + expect(drained.lifetimeTotal).toBe(usage.lifetimeTotal); + }); + + it('RETRIES BILL: ethers 429-retry attempts below _send are counted (tracker == server hits)', async () => { + installMeter(); + // Single-RPC adapter → boundedRetryFetchRequest keeps the default retry + // budget (5), and a perpetually-throttled method makes ethers issue + // 1 + 5 HTTP attempts inside ONE _send dispatch. Every attempt is a + // billable provider request and the server sees each one — the tracker + // must match exactly (this is the undercount the review flagged). + const rpc = await startLoopbackRpc({ throttle: ['eth_chainId'] }); + servers.push(rpc); + const a: any = new EVMChainAdapter(minimalConfig({ rpcUrl: rpc.url })); + adapters.push(a); + + await expect(a.getEvmChainId()).rejects.toBeTruthy(); // perpetual 429 → bounded failure + + const usage = a.drainRpcUsage(); + expect(rpc.hits('eth_chainId')).toBeGreaterThanOrEqual(2); // initial + ≥1 retry actually happened + expect(usage.byMethod['eth_chainId'] ?? 0).toBe(rpc.hits('eth_chainId')); + expect(rpcUsageWindowTotal(usage)).toBe(rpc.totalHits()); + }, 30_000); + + it('bounds unknown methods to "other" for the metric label', () => { + expect(boundedRpcMethodLabel('eth_getLogs')).toBe('eth_getLogs'); + expect(boundedRpcMethodLabel('eth_sendRawTransaction')).toBe('eth_sendRawTransaction'); + expect(boundedRpcMethodLabel('debug_traceTransaction')).toBe('other'); + expect(boundedRpcMethodLabel('weird method !!')).toBe('other'); + }); + + it('MULTI-ENDPOINT FAILOVER: failed attempts on endpoint A and the fallback success on B both bill (tracker == A hits + B hits)', async () => { + installMeter(); + // Endpoint A rate-limits eth_chainId (HTTP 429 — the request REACHED the + // provider, so it bills); endpoint B is healthy. A multi-RPC adapter uses + // perEndpointRetries = 0, so the failover client moves to B after A's + // billed refusal. The drained window must equal the SUM of what both + // servers actually received — the failed attempt and the fallback success + // are each one billable raw request. + const rpcA = await startLoopbackRpc({ throttle: ['eth_chainId'] }); + const rpcB = await startLoopbackRpc(); + servers.push(rpcA, rpcB); + const a: any = new EVMChainAdapter(minimalConfig({ rpcUrl: rpcA.url, rpcUrls: [rpcA.url, rpcB.url] })); + adapters.push(a); + + await expect(a.getEvmChainId()).resolves.toBe(31337n); // failover succeeds via B + + const usage = a.drainRpcUsage(); + expect(rpcA.hits('eth_chainId')).toBeGreaterThanOrEqual(1); // billed refusal happened + expect(rpcB.hits('eth_chainId')).toBeGreaterThanOrEqual(1); // fallback success happened + expect(usage.byMethod['eth_chainId'] ?? 0).toBe(rpcA.hits('eth_chainId') + rpcB.hits('eth_chainId')); + expect(rpcUsageWindowTotal(usage)).toBe(rpcA.totalHits() + rpcB.totalHits()); + }, 30_000); + + it('caps distinct window keys at MAX_WINDOW_METHODS; overflow aggregates into "other", existing keys keep counting raw', () => { + const t = new RpcUsageTracker(() => 'evm:31337'); + const max = RpcUsageTracker.MAX_WINDOW_METHODS; + for (let i = 0; i < max + 6; i++) t.record(`fabricated_${i}`); + t.record('fabricated_0'); // existing key increments raw even after the cap + const w = t.drainWindow(); + expect(Object.keys(w.byMethod).length).toBeLessThanOrEqual(max + 1); // raw keys + 'other' + expect(w.byMethod['fabricated_0']).toBe(2); + expect(w.byMethod['other']).toBe(6); // the 6 overflow names + expect(rpcUsageWindowTotal(w)).toBe(max + 7); + }); + + it('mergeRpcUsageWindows sums per-method and lifetime across trackers', () => { + const merged = mergeRpcUsageWindows( + { byMethod: { eth_call: 5, eth_estimateGas: 2 }, lifetimeTotal: 100 }, + { byMethod: { eth_call: 3, eth_sendRawTransaction: 4 }, lifetimeTotal: 50 }, + ); + expect(merged).toEqual({ + byMethod: { eth_call: 8, eth_estimateGas: 2, eth_sendRawTransaction: 4 }, + lifetimeTotal: 150, + }); + }); + + it('mergeRpcUsageWindows skips undefined inputs; nothing to merge yields a concrete EMPTY window', () => { + const w = { byMethod: { eth_call: 1 }, lifetimeTotal: 1 }; + const empty = { byMethod: {}, lifetimeTotal: 0 }; + expect(mergeRpcUsageWindows(undefined, w, undefined)).toEqual(w); + expect(mergeRpcUsageWindows(undefined, undefined)).toEqual(empty); + expect(mergeRpcUsageWindows()).toEqual(empty); + }); + + it('tracker.record never throws; window keys stay RAW (log token-safety is the formatter concern)', () => { + const t = new RpcUsageTracker(() => 'evm:31337'); + expect(() => t.record('eth_call')).not.toThrow(); + expect(() => t.record('debug_traceTransaction')).not.toThrow(); // raw key preserved in window + expect(() => t.record('')).not.toThrow(); // degenerate → 'other' + expect(() => t.record('x'.repeat(500))).not.toThrow(); // oversized → 'other' + const w = t.drainWindow(); + expect(rpcUsageWindowTotal(w)).toBe(4); + expect(w.byMethod['eth_call']).toBe(1); + expect(w.byMethod['debug_traceTransaction']).toBe(1); // NOT sanitized to 'other' + expect(w.byMethod['other']).toBe(2); + }); +}); diff --git a/packages/cli/src/daemon/lifecycle.ts b/packages/cli/src/daemon/lifecycle.ts index b2d28156e0..26be5588ea 100644 --- a/packages/cli/src/daemon/lifecycle.ts +++ b/packages/cli/src/daemon/lifecycle.ts @@ -65,6 +65,7 @@ const execAsync = promisify(exec); const execFileAsync = promisify(execFile); import { MockChainAdapter, + mergeRpcUsageWindows, type ApprovalPolicy, } from '@origintrail-official/dkg-chain'; import { DKGAgent, loadOpWallets, KaNumberAllocator } from '@origintrail-official/dkg-agent'; @@ -137,6 +138,7 @@ import { } from '../config.js'; import { resolveOtelSignals, resolveLogExporterMode, isUnknownLogExporter } from '../telemetry-config.js'; import { createDaemonLogSink } from './log-sink.js'; +import { startRpcUsageTelemetry } from './rpc-usage-log.js'; import { createPublicSnapshotStore, createPublisherControlFromStore, startPublisherRuntimeIfEnabled, type PublisherRuntime } from '../publisher-runner.js'; import { createCatchupRunner, type CatchupJobResult, type CatchupRunner } from '../catchup-runner.js'; import { loadTokens, httpAuthGuard } from '../auth.js'; @@ -2430,6 +2432,27 @@ export async function runDaemonInner( }, PRUNE_INTERVAL_MS); pruneTimer.unref(); + // RPC usage telemetry — the "RPC credit burn" signal (incident: a node spent + // ~$200 of RPC credits in a day with nothing measuring it). The whole + // Wiring only (scheduling/format live in rpc-usage-log.ts, unit-tested). + // The composite source merges EVERY drainable this process owns: the agent + // (its chain adapter) plus the async-publisher runtime's per-wallet + // adapters — `publisherRuntime` is late-bound, so read the live variable at + // each drain. chainId is the EFFECTIVE one (chainBase resolves field-level + // inheritance; config.chain?.chainId alone is undefined for configs that + // override only rpcUrl). + const rpcUsageLogger = new Logger("chain-rpc"); + const rpcUsageTelemetry = startRpcUsageTelemetry({ + source: { + drainRpcUsage: () => mergeRpcUsageWindows( + agent.drainRpcUsage(), + publisherRuntime?.drainRpcUsage(), + ), + }, + emit: (line) => rpcUsageLogger.info(createOperationContext("system"), line), + chainId: chainBase?.chainId ?? config.chain?.chainId, + }); + const tracker = new OperationTracker(dashDb); // Track peer connections @@ -3257,6 +3280,10 @@ export async function runDaemonInner( clearInterval(chainScanTimer); clearInterval(pingTimer); clearInterval(pruneTimer); + // Clears the timer AND performs the final best-effort drain (BEFORE + // telemetry stops), so a partial window still reaches Loki — keeps + // log-derived request totals exact across process lifecycles. + rpcUsageTelemetry.stop(); rateLimiter.destroy(); metricsCollector.stop(); // Stops log exporters AND flushes + shuts down the OTel SDK. diff --git a/packages/cli/src/daemon/rpc-usage-log.ts b/packages/cli/src/daemon/rpc-usage-log.ts new file mode 100644 index 0000000000..7fcd218bb7 --- /dev/null +++ b/packages/cli/src/daemon/rpc-usage-log.ts @@ -0,0 +1,100 @@ +/** + * Formats + schedules the minutely `rpc_usage` telemetry log lines. Invariants + * here: one logfmt line per method, DELTA counts (so `sum_over_time` over any + * range is the true request count), logfmt-token safety, idle windows emit + * nothing. WHY the accounting exists and how it counts lives in ONE place: + * packages/chain/src/rpc-usage.ts (module overview). + * + * Line shape (parsed in LogQL with `| logfmt` on the log body): + * rpc_usage method=eth_call count=42 window_s=60 chain=base:8453 + */ + +import { rpcUsageWindowTotal, type RpcUsageDrainable, type RpcUsageWindow } from '@origintrail-official/dkg-chain'; + +/** logfmt-token safety: methods/chain ids are self-generated, but never emit a token that could break parsing. */ +function safeToken(value: string, fallback: string): string { + return /^[A-Za-z0-9_.:-]{1,64}$/.test(value) ? value : fallback; +} + +/** + * One logfmt line per method with a nonzero count in the window. Returns [] when + * the window is empty (the caller skips logging entirely — an idle node emits + * nothing rather than a stream of zeros). + */ +export function formatRpcUsageLines( + usage: RpcUsageWindow, + windowSeconds: number, + chainId?: string, +): string[] { + if (!usage || rpcUsageWindowTotal(usage) <= 0) return []; + const chain = chainId ? ` chain=${safeToken(chainId, 'unknown')}` : ''; + const lines: string[] = []; + for (const [method, count] of Object.entries(usage.byMethod)) { + if (!Number.isFinite(count) || count <= 0) continue; + lines.push(`rpc_usage method=${safeToken(method, 'other')} count=${Math.floor(count)} window_s=${windowSeconds}${chain}`); + } + return lines; +} + +/** What the daemon drains: anything (partially) implementing the shared contract. */ +export type RpcUsageSource = Partial; + +/** + * Drain the source's RPC-usage window and emit one `rpc_usage` line per method + * through `emit`. The COMPLETE daemon emission step (drain → format → emit), + * extracted so the lifecycle timer AND the shutdown final-drain share one + * unit-tested implementation. Returns the number of lines emitted (0 for a + * missing capability or an empty window). Never throws. + */ +export function emitRpcUsage( + source: RpcUsageSource | undefined, + emit: (line: string) => void, + windowSeconds: number, + chainId?: string, +): number { + try { + const usage = source?.drainRpcUsage?.(); + if (!usage || rpcUsageWindowTotal(usage) <= 0) return 0; + const lines = formatRpcUsageLines(usage, windowSeconds, chainId); + for (const line of lines) emit(line); + return lines.length; + } catch { + return 0; // usage accounting must never break the node + } +} + +/** Handle returned by {@link startRpcUsageTelemetry}. */ +export interface RpcUsageTelemetryHandle { + /** + * Clear the timer and perform ONE final best-effort drain+emit, so a partial + * window (e.g. a publish burst right before a restart) still reaches the log + * pipeline instead of dying in memory. Idempotent-safe to call once at + * daemon teardown, BEFORE telemetry shuts down. + */ + stop(): void; +} + +/** + * The COMPLETE RPC-usage telemetry lifecycle: schedules the minutely + * drain→format→emit tick (unref'd — never keeps the process alive) and owns + * the shutdown final-drain. `runDaemonInner` just wires source/emit and calls + * `stop()` at teardown — no feature scheduling embedded in the daemon monolith. + */ +export function startRpcUsageTelemetry(opts: { + source: RpcUsageSource; + emit: (line: string) => void; + chainId?: string; + /** Window length in seconds (default 60). */ + windowSeconds?: number; +}): RpcUsageTelemetryHandle { + const windowSeconds = opts.windowSeconds ?? 60; + const tick = () => emitRpcUsage(opts.source, opts.emit, windowSeconds, opts.chainId); + const timer = setInterval(tick, windowSeconds * 1000); + timer.unref?.(); + return { + stop() { + clearInterval(timer); + tick(); // final best-effort drain — emitRpcUsage never throws + }, + }; +} diff --git a/packages/cli/src/publisher-runner.ts b/packages/cli/src/publisher-runner.ts index c9dc8fd129..a2748f3efe 100644 --- a/packages/cli/src/publisher-runner.ts +++ b/packages/cli/src/publisher-runner.ts @@ -1,6 +1,6 @@ import { join } from 'node:path'; import { DKGAgentWallet } from '@origintrail-official/dkg-agent'; -import { EVMChainAdapter, NoChainAdapter } from '@origintrail-official/dkg-chain'; +import { EVMChainAdapter, NoChainAdapter, mergeRpcUsageWindows, type ChainAdapter, type RpcUsageWindow } from '@origintrail-official/dkg-chain'; import { TypedEventBus, type Ed25519Keypair } from '@origintrail-official/dkg-core'; import { ACKCollector, AsyncLiftRunner, DKGPublisher, FileWorkspacePublicSnapshotStore, TripleStoreAsyncLiftPublisher, wrapAsRpcPreconditionIfApplicable, type AsyncLiftPublishExecutionInput, type AsyncLiftPublisher, type AsyncLiftPublisherConfig, type AsyncLiftPublisherRecoveryResult, type LiftJobBroadcast, type LiftJobIncluded, type PublishOptions, type WorkspacePublicSnapshotStore } from '@origintrail-official/dkg-publisher'; import { createTripleStore, type TripleStore } from '@origintrail-official/dkg-storage'; @@ -13,6 +13,8 @@ export interface PublisherRuntime { readonly walletIds: string[]; readonly wallets: readonly PublisherRuntimeWallet[]; readonly stop: () => Promise; + /** RpcUsageDrainable: merged window across every per-wallet chain adapter. */ + readonly drainRpcUsage: () => RpcUsageWindow; } export interface PublisherRuntimeWallet { @@ -40,6 +42,8 @@ type PublishEncryptionFactory = (publishOptions: PublishOptions) => interface ConfiguredPublisherWallet extends PublisherRuntimeWallet { readonly publisher: DKGPublisher; + /** The wallet's own chain adapter — also the wallet's RpcUsageDrainable source. */ + readonly chain: ChainAdapter; } export async function startPublisherRuntimeIfEnabled(args: { @@ -249,6 +253,7 @@ async function createPublisherRuntimeFromBase(args: PublisherRuntimeBaseArgs): P wallets.push({ address: wallet.address, identityId, + chain, publisher: new DKGPublisher({ store: args.store, chain, @@ -346,6 +351,7 @@ async function createPublisherRuntimeFromBase(args: PublisherRuntimeBaseArgs): P publisher: asyncPublisher, walletIds: validWalletIds, wallets: wallets.map(({ address, identityId }) => ({ address, identityId })), + drainRpcUsage: () => mergeRpcUsageWindows(...wallets.map((w) => w.chain.drainRpcUsage?.())), stop: async () => { await runner.stop(); if (args.closeStoreOnStop) { diff --git a/packages/cli/test/publisher-runner-rpc-usage.test.ts b/packages/cli/test/publisher-runner-rpc-usage.test.ts new file mode 100644 index 0000000000..f657ee7887 --- /dev/null +++ b/packages/cli/test/publisher-runner-rpc-usage.test.ts @@ -0,0 +1,118 @@ +// SPDX-License-Identifier: Apache-2.0 +/** + * The REAL publisher-runtime RPC-usage boundary (#1409 review): a runtime + * constructed through the production entry point (createPublisherRuntimeFromAgent + * → per-wallet EVMChainAdapter instances) must expose the raw JSON-RPC requests + * its adapters actually made via drainRpcUsage(). Construction itself performs + * real RPC (getIdentityId per wallet) against a loopback server whose own + * per-method hit log is the source of truth — so this dies if the runtime + * stops retaining its adapters (`chainAdapters.push`) or drainRpcUsage stops + * mapping them, the exact mutations the review called out as invisible to the + * fake-runtime composite test. + */ +import { createServer, type Server } from 'node:http'; +import { mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { describe, it, expect, afterEach } from 'vitest'; +import { generateEd25519Keypair } from '@origintrail-official/dkg-core'; +import { createTripleStore, type TripleStore } from '@origintrail-official/dkg-storage'; +import { rpcUsageWindowTotal } from '@origintrail-official/dkg-chain'; +import { createPublisherRuntimeFromAgent, type PublisherRuntime } from '../src/publisher-runner.js'; + +// Hardhat dev keys #0 and #1 — loopback only, never touch a real network. +// TWO wallets so the merge across ALL per-wallet adapters is what's proven: +// a regression to draining only chainAdapters[0] undercounts and fails the +// server-hit equality below. +const WALLETS = [ + { address: '0xf39Fd6e51aad88F6F4ce6aB8827279cffFb92266', privateKey: '0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80' }, + { address: '0x70997970C51812dc3A010C7d01b50e0d17dc79C8', privateKey: '0x59c6995e998f97a5a0044966f0945389dc9e86dae88c7a8412f4603b6b78690d' }, +]; +const HUB = '0x0000000000000000000000000000000000000001'; +// eth_call answer: 32 bytes that decode BOTH as a non-zero address (Hub +// contract resolution during adapter init must not see ZeroAddress) and as a +// non-zero uint (identityId lookup) — 12 zero bytes + 20 bytes of 0x22. +const CALL32 = '0x' + '00'.repeat(12) + '22'.repeat(20); + +/** Minimal loopback JSON-RPC server with a per-method hit log (source of truth). */ +async function startLoopback(): Promise<{ url: string; hits: (m: string) => number; totalHits: () => number; close: () => Promise }> { + const hits = new Map(); + const server: Server = createServer((req, res) => { + let raw = ''; + req.on('data', (c) => { raw += c; }); + req.on('end', () => { + const parsed = JSON.parse(raw); + const entries = Array.isArray(parsed) ? parsed : [parsed]; + const answer = (e: { id: number; method: string }) => { + hits.set(e.method, (hits.get(e.method) ?? 0) + 1); + const result = e.method === 'eth_chainId' ? '0x7a69' + : e.method === 'net_version' ? '31337' + : e.method === 'eth_blockNumber' ? '0x10' + : e.method === 'eth_getCode' ? '0x1234' + : CALL32; + return { jsonrpc: '2.0', id: e.id, result }; + }; + const body = Array.isArray(parsed) ? entries.map(answer) : answer(entries[0]); + res.writeHead(200, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify(body)); + }); + }); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + const addr = server.address() as { port: number }; + return { + url: `http://127.0.0.1:${addr.port}`, + hits: (m) => hits.get(m) ?? 0, + totalHits: () => [...hits.values()].reduce((a, b) => a + b, 0), + close: () => new Promise((resolve) => server.close(() => resolve())), + }; +} + +describe('publisher runtime drainRpcUsage — REAL runtime, real adapters, loopback as source of truth', () => { + let runtime: PublisherRuntime | null = null; + let store: TripleStore | null = null; + let loopback: Awaited> | null = null; + let dataDir: string | null = null; + + afterEach(async () => { + await runtime?.stop().catch(() => {}); + await store?.close().catch(() => {}); + await loopback?.close().catch(() => {}); + if (dataDir) await rm(dataDir, { recursive: true, force: true }); + runtime = null; store = null; loopback = null; dataDir = null; + }); + + it('merges BOTH per-wallet adapters’ raw request counts (== loopback hits); drain resets', async () => { + loopback = await startLoopback(); + dataDir = await mkdtemp(join(tmpdir(), 'pub-rpc-usage-')); + await writeFile( + join(dataDir, 'publisher-wallets.json'), + JSON.stringify({ wallets: WALLETS }), + ); + store = await createTripleStore({ backend: 'oxigraph' }); + + runtime = await createPublisherRuntimeFromAgent({ + dataDir, + store, + keypair: await generateEd25519Keypair(), + chainBase: { rpcUrl: loopback.url, hubAddress: HUB, chainId: 'evm:31337' }, + }); + + // Constructing the runtime performed real RPC through BOTH per-wallet + // adapters (identity lookup et al). The drained window must EQUAL what + // the loopback actually received — total and per method — which only + // holds if EVERY adapter's tracker is merged (each wallet's identity + // lookups bill separately). + expect(runtime.walletIds).toHaveLength(WALLETS.length); + const usage = runtime.drainRpcUsage(); + expect(usage).toBeDefined(); + expect(rpcUsageWindowTotal(usage!)).toBeGreaterThanOrEqual(2); // non-vacuous: both adapters made calls + expect(rpcUsageWindowTotal(usage!)).toBe(loopback.totalHits()); + for (const [method, count] of Object.entries(usage!.byMethod)) { + expect(loopback.hits(method), `method ${method}`).toBe(count); + } + + // Delta semantics survive the runtime boundary: second drain is empty. + const drained = runtime.drainRpcUsage(); + expect(rpcUsageWindowTotal(drained)).toBe(0); + }, 30_000); +}); diff --git a/packages/cli/test/rpc-usage-log.test.ts b/packages/cli/test/rpc-usage-log.test.ts new file mode 100644 index 0000000000..ea1989af4e --- /dev/null +++ b/packages/cli/test/rpc-usage-log.test.ts @@ -0,0 +1,164 @@ +import { describe, it, expect, vi, afterEach } from 'vitest'; +import { mergeRpcUsageWindows } from '@origintrail-official/dkg-chain'; +import { formatRpcUsageLines, emitRpcUsage, startRpcUsageTelemetry } from '../src/daemon/rpc-usage-log.js'; + +/** + * The rpc_usage line format is a CONTRACT with the Grafana dashboards (parsed + * in LogQL via `| json | line_format "{{.body}}" | logfmt | unwrap count`), so + * pin it: one logfmt line per method, delta counts, token-safe values. + */ +describe('formatRpcUsageLines — the Grafana-facing rpc_usage contract', () => { + it('emits one logfmt line per method with count/window/chain', () => { + const lines = formatRpcUsageLines( + { byMethod: { eth_call: 42, eth_getLogs: 7 }, lifetimeTotal: 49 }, + 60, + 'base:8453', + ); + expect(lines).toHaveLength(2); + expect(lines).toContain('rpc_usage method=eth_call count=42 window_s=60 chain=base:8453'); + expect(lines).toContain('rpc_usage method=eth_getLogs count=7 window_s=60 chain=base:8453'); + // Every line is logfmt-parseable: key=value tokens, no quotes needed. + for (const l of lines) expect(l).toMatch(/^rpc_usage( [a-z_]+=[A-Za-z0-9_.:-]+)+$/); + }); + + it('returns [] for an empty window (idle node logs nothing, not zeros)', () => { + expect(formatRpcUsageLines({ byMethod: {}, lifetimeTotal: 123 }, 60, 'base:8453')).toEqual([]); + }); + + it('omits chain when unset and skips zero/negative/NaN counts', () => { + const lines = formatRpcUsageLines( + { byMethod: { eth_call: 3, eth_getLogs: 0, eth_gasPrice: NaN as unknown as number }, lifetimeTotal: 3 }, + 60, + ); + expect(lines).toEqual(['rpc_usage method=eth_call count=3 window_s=60']); + }); + + it('sanitizes non-token-safe method/chain values (defensive logfmt safety)', () => { + const lines = formatRpcUsageLines( + { byMethod: { 'bad method "x"': 5 }, lifetimeTotal: 5 }, + 60, + 'weird chain id', + ); + expect(lines).toEqual(['rpc_usage method=other count=5 window_s=60 chain=unknown']); + }); +}); + +describe('composite daemon source — agent + publisher-runtime windows merged at drain time', () => { + it('COMPOSITE SOURCE end-to-end: publisher-runtime traffic reaches the emitted rpc_usage lines', () => { + // The exact daemon wiring shape: the source merges the agent window with + // the (lazily started) publisher runtime's window at every drain. The + // publish-transaction methods MUST appear in the emitted lines — this is + // the undercount the review flagged (per-wallet publisher adapters were + // never drained). + let runtime: { drainRpcUsage: () => { byMethod: Record; total: number; lifetimeTotal: number } | undefined } | null = null; + const agentLike = { drainRpcUsage: () => ({ byMethod: { eth_call: 2 }, lifetimeTotal: 2 }) }; + const source = { drainRpcUsage: () => mergeRpcUsageWindows(agentLike.drainRpcUsage(), runtime?.drainRpcUsage()) }; + + const before: string[] = []; + emitRpcUsage(source, (l) => before.push(l), 60, 'base:8453'); + expect(before).toEqual(['rpc_usage method=eth_call count=2 window_s=60 chain=base:8453']); + + // runtime boots later (the daemon assigns the live variable) — its + // per-wallet adapter traffic must now be merged in. + runtime = { drainRpcUsage: () => ({ byMethod: { eth_sendRawTransaction: 3, eth_call: 1 }, lifetimeTotal: 4 }) }; + const after: string[] = []; + emitRpcUsage(source, (l) => after.push(l), 60, 'base:8453'); + expect(after).toContain('rpc_usage method=eth_call count=3 window_s=60 chain=base:8453'); + expect(after).toContain('rpc_usage method=eth_sendRawTransaction count=3 window_s=60 chain=base:8453'); + }); +}); + +describe('emitRpcUsage — the complete daemon emission step (drain → format → emit)', () => { + it('drains the agent source and emits one line per method', () => { + const emitted: string[] = []; + const agentLike = { + drainRpcUsage: () => ({ byMethod: { eth_call: 12, eth_getLogs: 3 }, lifetimeTotal: 15 }), + }; + const n = emitRpcUsage(agentLike, (l) => emitted.push(l), 60, 'base:8453'); + expect(n).toBe(2); + expect(emitted).toContain('rpc_usage method=eth_call count=12 window_s=60 chain=base:8453'); + expect(emitted).toContain('rpc_usage method=eth_getLogs count=3 window_s=60 chain=base:8453'); + }); + + it('emits nothing for an empty window, a missing capability, or no source', () => { + const emitted: string[] = []; + expect(emitRpcUsage({ drainRpcUsage: () => ({ byMethod: {}, lifetimeTotal: 9 }) }, (l) => emitted.push(l), 60)).toBe(0); + expect(emitRpcUsage({ drainRpcUsage: () => undefined } as never, (l) => emitted.push(l), 60)).toBe(0); // rogue undefined still tolerated at runtime + expect(emitRpcUsage({}, (l) => emitted.push(l), 60)).toBe(0); // adapter without the capability + expect(emitRpcUsage(undefined, (l) => emitted.push(l), 60)).toBe(0); + expect(emitted).toEqual([]); + }); + + it('never throws — a throwing drain or emitter is swallowed (accounting must not break the node)', () => { + expect(emitRpcUsage({ drainRpcUsage: () => { throw new Error('boom'); } }, () => {}, 60)).toBe(0); + expect( + emitRpcUsage( + { drainRpcUsage: () => ({ byMethod: { eth_call: 1 }, lifetimeTotal: 1 }) }, + () => { throw new Error('sink down'); }, + 60, + ), + ).toBe(0); + }); +}); + +describe('emitRpcUsage ← DKGAgent.drainRpcUsage — the REAL daemon boundary end-to-end', () => { + it('adapter window → real agent delegation → rpc_usage lines', async () => { + // The exact wiring the daemon uses: the source is a DKGAgent whose + // drainRpcUsage (REAL inherited method, prototype-invoked with a + // chain-bearing `this`) delegates to the adapter capability. A broken + // delegation would emit zero lines here and fail this test — closing the + // gap where chain-side and formatter tests both pass but the daemon signal + // is silently dead. + const { DKGAgent } = await import('@origintrail-official/dkg-agent'); + const chain = { drainRpcUsage: () => ({ byMethod: { eth_call: 11 }, lifetimeTotal: 11 }) }; + const agentLike = { drainRpcUsage: () => DKGAgent.prototype.drainRpcUsage.call({ chain } as never) }; + const emitted: string[] = []; + const n = emitRpcUsage(agentLike, (l) => emitted.push(l), 60, 'base:8453'); + expect(n).toBe(1); + expect(emitted).toEqual(['rpc_usage method=eth_call count=11 window_s=60 chain=base:8453']); + }); +}); + +describe('startRpcUsageTelemetry — the full scheduling lifecycle (timer + shutdown drain)', () => { + afterEach(() => vi.useRealTimers()); + + it('ticks every window, emits the drained lines, and stop() drains the partial window then halts', () => { + vi.useFakeTimers(); + const emitted: string[] = []; + let next = { byMethod: { eth_call: 5 }, lifetimeTotal: 5 }; + const handle = startRpcUsageTelemetry({ + source: { drainRpcUsage: () => next }, + emit: (l) => emitted.push(l), + chainId: 'base:8453', + windowSeconds: 60, + }); + + expect(emitted).toEqual([]); // nothing before the first tick + vi.advanceTimersByTime(60_000); + expect(emitted).toEqual(['rpc_usage method=eth_call count=5 window_s=60 chain=base:8453']); + + // Partial window accumulated after the last tick → stop() must drain it. + next = { byMethod: { eth_getLogs: 2 }, lifetimeTotal: 7 }; + handle.stop(); + expect(emitted).toContain('rpc_usage method=eth_getLogs count=2 window_s=60 chain=base:8453'); + + // ...and the timer is really gone: no further emissions after stop(). + const after = emitted.length; + next = { byMethod: { eth_call: 9 }, lifetimeTotal: 16 }; + vi.advanceTimersByTime(300_000); + expect(emitted.length).toBe(after); + }); + + it('idle windows emit nothing on tick or stop', () => { + vi.useFakeTimers(); + const emitted: string[] = []; + const handle = startRpcUsageTelemetry({ + source: { drainRpcUsage: () => ({ byMethod: {}, lifetimeTotal: 3 }) }, + emit: (l) => emitted.push(l), + windowSeconds: 60, + }); + vi.advanceTimersByTime(180_000); + handle.stop(); + expect(emitted).toEqual([]); + }); +}); diff --git a/packages/core/src/telemetry-api.ts b/packages/core/src/telemetry-api.ts index 20ae94f1d9..c4cf81d152 100644 --- a/packages/core/src/telemetry-api.ts +++ b/packages/core/src/telemetry-api.ts @@ -137,6 +137,12 @@ export interface DkgMetrics { ackHandlerTotal: Counter; /** rpc_method, outcome={ok|error|timeout}, retryable, chain_id */ chainRpcTotal: Counter; + /** rpc_method (bounded known JSON-RPC set, else 'other'), chain_id — RAW + * transport-level JSON-RPC client requests, i.e. the PROVIDER-BILLING unit. + * Distinct from chainRpcTotal (logical chain operations): one logical op can + * issue several raw requests (populate = estimateGas + getTransactionCount + * + chainId + …), and raw request volume is what burns RPC credits. */ + chainRpcRequestsTotal: Counter; /** ms; rpc_method, chain_id */ chainRpcDuration: Histogram; /** rpc_method, chain_id, reason={exhausted|recovered} */ @@ -164,6 +170,9 @@ function buildMetrics(): DkgMetrics { ackVerifyTotal: meter.createCounter('dkg.ack.verify.total', { description: 'ACK identity verification results' }), ackHandlerTotal: meter.createCounter('dkg.ack.handler.total', { description: 'Inbound storage-ACK handler outcomes' }), chainRpcTotal: meter.createCounter('dkg.chain.rpc.total', { description: 'Chain RPC calls by method/outcome' }), + chainRpcRequestsTotal: meter.createCounter('dkg.chain.rpc.requests.total', { + description: 'Raw JSON-RPC client requests by method (provider billing unit)', + }), chainRpcDuration: meter.createHistogram('dkg.chain.rpc.duration', { unit: 'ms', description: 'Chain RPC wall-time', advice: { explicitBucketBoundaries: RPC_DURATION_BUCKETS }, }), diff --git a/packages/evm-module/deployments/localhost_contracts.json b/packages/evm-module/deployments/localhost_contracts.json index 28ee21a1dc..ad64491f2e 100644 --- a/packages/evm-module/deployments/localhost_contracts.json +++ b/packages/evm-module/deployments/localhost_contracts.json @@ -3,271 +3,289 @@ "Hub": { "evmAddress": "0x5FbDB2315678afecb367f032d93F642f64180aa3", "version": "1.0.0", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 1, - "deploymentTimestamp": 1781650667800, + "deploymentTimestamp": 1783072376041, "deployed": true }, "Token": { "evmAddress": "0xe7f1725E7734CE288F8367e1Bb143E90bb3F0512", "version": null, - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 2, - "deploymentTimestamp": 1781650667874, + "deploymentTimestamp": 1783072376270, "deployed": true }, "ParametersStorage": { "evmAddress": "0xe70f935c32dA4dB13e7876795f1e175465e6458e", - "version": "10.0.3", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "version": "10.0.4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 205, - "deploymentTimestamp": 1781650668281, + "deploymentTimestamp": 1783072376913, "deployed": true }, "WhitelistStorage": { "evmAddress": "0x2625760C4A8e8101801D3a48eE64B2bEA42f1E96", "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 211, - "deploymentTimestamp": 1781650668508, + "deploymentTimestamp": 1783072377385, "deployed": true }, "IdentityStorage": { "evmAddress": "0xD6b040736e948621c5b6E0a494473c47a6113eA8", "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 213, - "deploymentTimestamp": 1781650668618, + "deploymentTimestamp": 1783072377699, "deployed": true }, "ShardingTableStorage": { "evmAddress": "0xAdE429ba898c34722e722415D722A70a297cE3a2", "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 215, - "deploymentTimestamp": 1781650668728, + "deploymentTimestamp": 1783072377952, "deployed": true }, "StakingStorage": { "evmAddress": "0xcE0066b1008237625dDDBE4a751827de037E53D2", "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 217, - "deploymentTimestamp": 1781650668855, + "deploymentTimestamp": 1783072378244, "deployed": true }, "ProfileStorage": { "evmAddress": "0x51C65cd0Cdb1A8A8b79dfc2eE965B1bA0bb8fc89", - "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "version": "10.0.4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 220, - "deploymentTimestamp": 1781650668977, + "deploymentTimestamp": 1783072378513, "deployed": true }, "Chronos": { "evmAddress": "0xC7143d5bA86553C06f5730c8dC9f8187a621A8D4", "version": null, - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 222, - "deploymentTimestamp": 1781650669065, + "deploymentTimestamp": 1783072378738, "deployed": true }, "EpochStorageV8": { "evmAddress": "0xc9952Fc93Fa9bE383ccB39008c786b9f94eAc95d", - "version": "10.0.3", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "version": "10.0.4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 224, - "deploymentTimestamp": 1781650669176, + "deploymentTimestamp": 1783072379010, "deployed": true }, "DKGKnowledgeAssets": { "evmAddress": "0x70eE76691Bdd9696552AF8d4fd634b3cF79DD529", "version": "10.1.0", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 227, - "deploymentTimestamp": 1781650669295, + "deploymentTimestamp": 1783072379305, "deployed": true }, "AskStorage": { "evmAddress": "0x162700d1613DfEC978032A909DE02643bC55df1A", "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 230, - "deploymentTimestamp": 1781650669408, + "deploymentTimestamp": 1783072379555, "deployed": true }, "Identity": { "evmAddress": "0xcD0048A5628B37B8f743cC2FeA18817A29e97270", "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 233, - "deploymentTimestamp": 1781650669519, + "deploymentTimestamp": 1783072379822, "deployed": true }, "ConvictionStakingStorage": { "evmAddress": "0x942ED2fa862887Dc698682cc6a86355324F0f01e", - "version": "10.0.3", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "version": "10.0.6", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 236, - "deploymentTimestamp": 1781650669635, + "deploymentTimestamp": 1783072380083, "deployed": true }, "ShardingTable": { "evmAddress": "0xa722bdA6968F50778B973Ae2701e90200C564B49", "version": "10.0.3", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 239, - "deploymentTimestamp": 1781650669747, + "deploymentTimestamp": 1783072380347, "deployed": true }, "Ask": { "evmAddress": "0xe1708FA6bb2844D5384613ef0846F9Bc1e8eC55E", "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 242, - "deploymentTimestamp": 1781650669860, + "deploymentTimestamp": 1783072380632, "deployed": true }, "RandomSamplingStorage": { "evmAddress": "0x871ACbEabBaf8Bed65c22ba7132beCFaBf8c27B5", - "version": "10.1.0", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "version": "10.2.0", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 245, - "deploymentTimestamp": 1781650669975, + "deploymentTimestamp": 1783072380897, "deployed": true }, "StakingKPI": { "evmAddress": "0x683d9CDD3239E0e01E8dC6315fA50AD92aB71D2d", "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 248, - "deploymentTimestamp": 1781650670094, + "deploymentTimestamp": 1783072381149, "deployed": true }, "Profile": { "evmAddress": "0x71a0b8A2245A9770A4D887cE1E4eCc6C1d4FF28c", "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 251, - "deploymentTimestamp": 1781650670210, + "deploymentTimestamp": 1783072381424, "deployed": true }, "ContextGraphStorage": { "evmAddress": "0x193521C8934bCF3473453AF4321911E7A89E0E12", - "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "version": "10.0.6", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 254, - "deploymentTimestamp": 1781650670325, + "deploymentTimestamp": 1783072381710, "deployed": true }, "ContextGraphValueStorage": { "evmAddress": "0x3C1Cb427D20F15563aDa8C249E71db76d7183B6c", "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 257, - "deploymentTimestamp": 1781650670442, + "deploymentTimestamp": 1783072381964, "deployed": true }, - "RandomSampling": { + "CGWeightTreeStorage": { "evmAddress": "0x547382C0D1b23f707918D3c83A77317B71Aa8470", - "version": "10.0.4", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "version": "1.0.0", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 260, - "deploymentTimestamp": 1781650670557, + "deploymentTimestamp": 1783072382225, "deployed": true }, - "ContextGraphs": { + "RandomSampling": { "evmAddress": "0x5e6CB7E728E1C320855587E1D9C6F7972ebdD6D5", - "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "version": "10.6.0", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 263, - "deploymentTimestamp": 1781650670672, + "deploymentTimestamp": 1783072382522, "deployed": true }, - "PublishingConvictionStorage": { + "ContextGraphWaiverStorage": { "evmAddress": "0xeAd789bd8Ce8b9E94F5D0FCa99F8787c7e758817", - "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "version": "1.0.0", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 266, - "deploymentTimestamp": 1781650670783, + "deploymentTimestamp": 1783072382763, "deployed": true }, - "PublishingConviction": { - "evmAddress": "0xd3FFD73C53F139cEBB80b6A524bE280955b3f4db", - "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", - "deploymentBlock": 269, - "deploymentTimestamp": 1781650670893, + "ContextGraphs": { + "evmAddress": "0xd9fEc8238711935D6c8d79Bef2B9546ef23FC046", + "version": "10.0.4", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", + "deploymentBlock": 268, + "deploymentTimestamp": 1783072383012, "deployed": true }, - "DKGPublishingConvictionNFT": { - "evmAddress": "0xCBBe2A5c3A22BE749D5DDF24e9534f98951983e2", - "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", - "deploymentBlock": 272, - "deploymentTimestamp": 1781650671004, + "PublishingConvictionStorage": { + "evmAddress": "0x9fD16eA9E31233279975D99D5e8Fc91dd214c7Da", + "version": "10.0.3", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", + "deploymentBlock": 271, + "deploymentTimestamp": 1783072383282, "deployed": true }, - "KnowledgeAssetsLifecycle": { - "evmAddress": "0xE8F7d98bE6722d42F29b50500B0E318EF2be4fc8", - "version": "10.1.0", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", - "deploymentBlock": 275, - "deploymentTimestamp": 1781650671123, + "PublishingConviction": { + "evmAddress": "0xb932C8342106776E73E39D695F3FFC3A9624eCE0", + "version": "10.0.7", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", + "deploymentBlock": 274, + "deploymentTimestamp": 1783072383541, "deployed": true }, - "V8MigrationEligibility": { - "evmAddress": "0x7580708993de7CA120E957A62f26A5dDD4b3D8aC", - "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", - "deploymentBlock": 278, - "deploymentTimestamp": 1781650671232, + "DKGPublishingConvictionNFT": { + "evmAddress": "0x2c8ED11fd7A058096F2e5828799c68BE88744E2F", + "version": "10.0.3", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", + "deploymentBlock": 277, + "deploymentTimestamp": 1783072383796, "deployed": true }, - "StakingV10": { + "KnowledgeAssetsLifecycle": { "evmAddress": "0x572316aC11CB4bc5daf6BDae68f43EA3CCE3aE0e", - "version": "10.0.3", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "version": "10.1.6", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 280, - "deploymentTimestamp": 1781650671344, + "deploymentTimestamp": 1783072384075, "deployed": true }, - "DKGStakingConvictionNFT": { + "StakingV10": { "evmAddress": "0xCd7c00Ac6dc51e8dCc773971Ac9221cC582F3b1b", - "version": "10.0.2", - "gitBranch": "integration/rfc49-full", - "gitCommitHash": "fe1104a40e0c61a35fdde6433df5938c876d31a4", + "version": "10.0.5", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", "deploymentBlock": 283, - "deploymentTimestamp": 1781650671457, + "deploymentTimestamp": 1783072384345, + "deployed": true + }, + "DKGStakingConvictionNFT": { + "evmAddress": "0xCa1D199b6F53Af7387ac543Af8e8a34455BBe5E0", + "version": "10.0.3", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", + "deploymentBlock": 286, + "deploymentTimestamp": 1783072384604, + "deployed": true + }, + "MigrationCreditRecovery": { + "evmAddress": "0xFD2Cf3b56a73c75A7535fFe44EBABe7723c64719", + "version": "1.0.0", + "gitBranch": "feat/rpc-usage-metrics", + "gitCommitHash": "42a8781d8857abb0c3fecd0d9be9e6db09bb176f", + "deploymentBlock": 289, + "deploymentTimestamp": 1783072384860, "deployed": true } } diff --git a/tools/log-collection-poc/OPERATOR-GUIDE.md b/tools/log-collection-poc/OPERATOR-GUIDE.md index 6e762cbf30..ce41b57691 100644 --- a/tools/log-collection-poc/OPERATOR-GUIDE.md +++ b/tools/log-collection-poc/OPERATOR-GUIDE.md @@ -41,6 +41,12 @@ queued unboundedly. - **Resource labels:** `service.name=dkg-node`, `service.instance.id=`, `deployment.environment=`, `dkg.node.role`, `dkg.chain` (matches the traces/metrics resource). Loki sanitizes dots to underscores, so these appear as `service_name`, `service_instance_id`, `deployment_environment`, `dkg_node_role`, `dkg_chain`. - **Per-record attributes:** `dkg.operation_id`, `dkg.operation_name`, `dkg.source_operation_id`, `dkg.module`, severity, plus `trace_id`/`span_id` when emitted inside a span. - **Body:** the log message, with secrets already redacted. +- **RPC usage lines:** once per minute (only when RPC activity happened) the node + emits one `rpc_usage method= count= window_s=60 chain=` + line per method — the RAW JSON-RPC request count (the unit your RPC provider + bills), counted at the transport level. Each line carries the DELTA for its + window, so summing them over any range gives the exact request count. This is + what powers the Grafana "RPC requests per node / by method" panels. ## Traces & metrics (optional) Logs go through a hand-rolled OTLP/HTTP exporter; **traces and metrics use the