diff --git a/bchain/coins/eth/ethrpc.go b/bchain/coins/eth/ethrpc.go index a2cc4081e7..12922ad9da 100644 --- a/bchain/coins/eth/ethrpc.go +++ b/bchain/coins/eth/ethrpc.go @@ -395,6 +395,17 @@ func (b *EthereumRPC) observeEthSyncRpcError(method string, err error) { b.metrics.EthSyncRpcErrors.With(common.Labels{"method": method, "status": ethSyncRpcErrStatus(err)}).Inc() } +func (b *EthereumRPC) observeSyncRPCLatency(method string, start time.Time, err error) { + if b.metrics == nil { + return + } + errorLabel := "" + if err != nil { + errorLabel = "failure" + } + b.metrics.RPCSyncLatency.With(common.Labels{"method": method, "error": errorLabel}).Observe(float64(time.Since(start)) / 1e6) +} + // EnsureSameRPCHost validates both RPC URLs and logs a warning if hosts differ. func EnsureSameRPCHost(httpURL, wsURL string) error { if httpURL == "" || wsURL == "" { @@ -960,12 +971,17 @@ func (b *EthereumRPC) GetChainInfo() (*bchain.ChainInfo, error) { } ctx, cancel := context.WithTimeout(context.Background(), b.Timeout) defer cancel() + netStart := time.Now() id, err := b.Client.NetworkID(ctx) + b.observeSyncRPCLatency("net_version", netStart, err) if err != nil { return nil, err } var ver string - if err := b.RPC.CallContext(ctx, &ver, "web3_clientVersion"); err != nil { + web3Start := time.Now() + err = b.RPC.CallContext(ctx, &ver, "web3_clientVersion") + b.observeSyncRPCLatency("web3_clientVersion", web3Start, err) + if err != nil { return nil, err } rv := &bchain.ChainInfo{ @@ -996,7 +1012,9 @@ func (b *EthereumRPC) getBestHeader() (bchain.EVMHeader, error) { var err error ctx, cancel := context.WithTimeout(context.Background(), b.Timeout) defer cancel() + headerStart := time.Now() b.bestHeader, err = b.Client.HeaderByNumber(ctx, nil) + b.observeSyncRPCLatency("eth_getBlockByNumber", headerStart, err) if err != nil { b.bestHeader = nil return nil, err @@ -1363,6 +1381,7 @@ func (b *EthereumRPC) getBlockRaw(hash string, height uint32, fullTxs bool) (jso var raw json.RawMessage var err error var method string + defer func(s time.Time) { b.observeSyncRPCLatency(method, s, err) }(time.Now()) if hash != "" { if hash == "pending" { method = "eth_getBlockByNumber" @@ -1395,7 +1414,9 @@ func (b *EthereumRPC) processEventsForBlock(blockNumber string) (map[string][]*b var logs []rpcLogWithTxHash var ensRecords []bchain.AddressAliasRecord var method = "eth_getLogs" - err := b.RPC.CallContext(ctx, &logs, method, map[string]interface{}{ + var err error + defer func(s time.Time) { b.observeSyncRPCLatency(method, s, err) }(time.Now()) + err = b.RPC.CallContext(ctx, &logs, method, map[string]interface{}{ "fromBlock": blockNumber, "toBlock": blockNumber, }) @@ -1499,7 +1520,9 @@ func (b *EthereumRPC) getInternalDataForBlock(ctx context.Context, blockHash str if b.ChainConfig.TraceTimeout != "" { traceConfig["timeout"] = b.ChainConfig.TraceTimeout } + traceStart := time.Now() err := b.RPC.CallContext(ctx, &trace, "debug_traceBlockByHash", blockHash, traceConfig) // Use caller-provided ctx for timeout/cancel. + b.observeSyncRPCLatency("debug_traceBlockByHash", traceStart, err) b.observeEthSyncRpcError("debug_traceBlockByHash", err) if err != nil { glog.Error("debug_traceBlockByHash block ", blockHash, ", error ", err) diff --git a/common/metrics.go b/common/metrics.go index d83bd33746..0e11783038 100644 --- a/common/metrics.go +++ b/common/metrics.go @@ -102,6 +102,7 @@ type Metrics struct { EthEip1559FeeSource *prometheus.CounterVec `metric:"eth_eip1559_fee_source_total"` EthBlockGasUsedRatio prometheus.Gauge `metric:"eth_block_gas_used_ratio"` EthBlockBaseFee prometheus.Gauge `metric:"eth_block_base_fee"` + RPCSyncLatency *prometheus.HistogramVec `metric:"rpc_sync_latency"` EthSyncRpcErrors *prometheus.CounterVec `metric:"eth_sync_rpc_errors"` } diff --git a/configs/grafana/panels.yaml b/configs/grafana/panels.yaml index a3aae3f2b4..c694cb8e99 100644 --- a/configs/grafana/panels.yaml +++ b/configs/grafana/panels.yaml @@ -286,6 +286,16 @@ panels: events: promql: sum by (subscription, event) (rate({{name:backend_subscription_events}}{job="blockbook", coin="$coin"}[5m])) legend: '{{subscription}} - {{event}}' + sync.rpc_sync_latency: + title: Sync RPC latency (p50/p95) + description: Round-trip time of successful sync-internal RPC calls (eth_getBlockByHash, eth_getLogs, debug_traceBlockByHash, etc.) that are not tracked by the general RPC latency panel; these directly affect block indexing speed. Typically sub-100ms; a sustained multi-second p95 means the backend is overloaded. Failed calls are excluded so timeout expiries (~10-15s) do not skew the quantiles; they show up in the Ethereum sync RPC errors panel instead. + queries: + p95: + promql: histogram_quantile(0.95, sum(rate({{name:rpc_sync_latency}}_bucket{job="blockbook", coin="$coin", error=""}[1m])) by (coin, method, le)) + legend: '{{coin}} - {{method}}' + p50: + promql: histogram_quantile(0.50, sum(rate({{name:rpc_sync_latency}}_bucket{job="blockbook", coin="$coin", error=""}[1m])) by (coin, method, le)) + legend: '{{coin}} - {{method}}' row.websocket: title: Websocket websocket.clients: diff --git a/configs/grafana/template.json b/configs/grafana/template.json index 8c9f907c6d..bdc937d509 100644 --- a/configs/grafana/template.json +++ b/configs/grafana/template.json @@ -2436,6 +2436,72 @@ "type": "timeseries", "x-panel-key": "sync.tip_feed_events" }, + { + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "lineWidth": 1, + "pointSize": 5, + "showPoints": "never", + "stacking": { + "group": "A", + "mode": "none" + } + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + }, + "unit": "ms" + }, + "overrides": [] + }, + "id": 348, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max", + "mean" + ], + "displayMode": "table", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "pluginVersion": "12.3.0", + "targets": [ + { + "editorMode": "code", + "range": true, + "refId": "A", + "x-query-key": "p95" + }, + { + "editorMode": "code", + "range": true, + "refId": "B", + "x-query-key": "p50" + } + ], + "type": "timeseries", + "x-panel-key": "sync.rpc_sync_latency" + }, { "collapsed": false, "id": 315, diff --git a/configs/metrics.yaml b/configs/metrics.yaml index 5018502f5e..55d767fd83 100644 --- a/configs/metrics.yaml +++ b/configs/metrics.yaml @@ -400,6 +400,12 @@ metrics: name: blockbook_eth_block_base_fee type: gauge help: Base fee per gas of the most recently connected EVM block, in wei - divide by 1e9 for Gwei. The realized base fee from the block header (push path); compare with eth_eip1559_base_fee (the pull path's next-block projection) to see whether the projection tracks the chain. Updated once per connected block + rpc_sync_latency: + name: blockbook_rpc_sync_latency + type: histogram_vec + help: Blockbook-to-backend round-trip time for sync-internal RPC calls (eth_getBlockByHash, eth_getLogs, debug_traceBlockByHash, etc.), in milliseconds, labeled by method and whether the call failed. Typically sub-100ms; multi-second values mean an overloaded backend, and observations near 10-15s are trace_timeout/rpc_timeout expiries rather than real round trips + labels: [method, error] + buckets: [1, 5, 10, 25, 50, 100, 250, 500, 1000, 2000, 5000, 10000, 20000] eth_sync_rpc_errors: name: blockbook_eth_sync_rpc_errors type: counter_vec diff --git a/tests/openapi/src/client.ts b/tests/openapi/src/client.ts index 28a405bd97..07c10c0e2a 100644 --- a/tests/openapi/src/client.ts +++ b/tests/openapi/src/client.ts @@ -61,7 +61,7 @@ export class OpenApiFetchClient { try { const response = await fetch(url, { method: "GET", - signal: AbortSignal.timeout(30_000), + signal: AbortSignal.timeout(15_000), }); const body = await response.text(); if (attempt < 2 && isRetryableHTTPStatus(response.status)) { diff --git a/tests/openapi/src/constants.ts b/tests/openapi/src/constants.ts index 0fc1fe54d7..3bc5ee9175 100644 --- a/tests/openapi/src/constants.ts +++ b/tests/openapi/src/constants.ts @@ -1,5 +1,5 @@ -export const wsDialTimeoutMs = 5_000; -export const wsMessageTimeoutMs = 15_000; +export const wsDialTimeoutMs = 3_000; +export const wsMessageTimeoutMs = 10_000; export const txSearchWindow = 12; export const blockPageSize = 1; export const sampleBlockPageSize = 3; diff --git a/tests/openapi/src/context.ts b/tests/openapi/src/context.ts index 9165bc03f8..996df20a89 100644 --- a/tests/openapi/src/context.ts +++ b/tests/openapi/src/context.ts @@ -31,6 +31,134 @@ import { import type { Capability, AddressResponse, BlockHashResponse, BlockResponse, BlockSummary, FiatTickerResponse, StatusResponse, TxResponse, UtxoResponse, WsEnvelope, WsInfoResponse, WsMethod, WsResponse } from "./types.js"; +// --------------------------------------------------------------------------- +// WebSocket connection +// --------------------------------------------------------------------------- + +// wsConnection wraps a single persistent WebSocket and multiplexes requests +// over it by matching responses to the originating request ID. +class wsConnection { + private ws: WebSocket | null = null; + private pendings = new Map< + string, + { resolve: (value: unknown) => void; reject: (reason: unknown) => void; timer: NodeJS.Timeout } + >(); + private closed = false; + + constructor( + private readonly url: string, + private readonly contract: OpenApiContract, + ) {} + + get isClosed(): boolean { + return this.closed; + } + + connect(): Promise { + return new Promise((resolve, reject) => { + // Guard against hangs below the ws handshake layer (DNS / TCP). + let settled = false; + const guard = setTimeout(() => { + if (settled) { return; } + settled = true; + this.ws?.terminate(); + reject(new Error(`websocket connect timed out for ${this.url}`)); + }, wsDialTimeoutMs); + + this.ws = new WebSocket(this.url, { + handshakeTimeout: wsDialTimeoutMs, + rejectUnauthorized: process.env.OPENAPI_INSECURE_TLS === "0", + agent: wsProxyAgent(), + }); + this.ws.on("open", () => { + if (settled) { return; } + settled = true; + clearTimeout(guard); + resolve(); + }); + this.ws.on("error", (err) => { + if (settled) { return; } + settled = true; + clearTimeout(guard); + this.closed = true; + reject(err); + }); + this.ws.on("close", () => { + if (!settled) { + settled = true; + clearTimeout(guard); + reject(new Error("websocket closed before handshake completed")); + } + this.closed = true; + for (const [, p] of this.pendings) { + clearTimeout(p.timer); + p.reject(new Error("websocket connection closed")); + } + this.pendings.clear(); + }); + this.ws.on("message", (data) => { + let response: WsResponse; + try { + response = JSON.parse(data.toString()) as WsResponse; + } catch { + return; + } + const pending = this.pendings.get(response.id); + if (!pending) { + return; + } + clearTimeout(pending.timer); + this.pendings.delete(response.id); + if (isWsError(response.data)) { + pending.reject( + new Error(`websocket ${response.id} returned error: ${response.data.error.message}`), + ); + return; + } + pending.resolve(response.data); + }); + }); + } + + async request(request: WsEnvelope, dataSchemaRef?: string): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + this.pendings.delete(request.id); + reject(new Error(`websocket ${request.method} timed out for ${this.url}`)); + }, wsMessageTimeoutMs); + this.pendings.set(request.id, { + resolve: (value) => { + if (dataSchemaRef) { + this.contract.validateSchemaRef(dataSchemaRef, `WS ${request.method} response data`, value); + } + resolve(value as T); + }, + reject, + timer, + }); + if (this.ws) { + try { + this.ws.send(JSON.stringify(request)); + } catch (err) { + clearTimeout(timer); + this.pendings.delete(request.id); + reject(err); + } + } else { + clearTimeout(timer); + this.pendings.delete(request.id); + reject(new Error("websocket not connected")); + } + }); + } + + close(): void { + this.closed = true; + this.ws?.close(); + this.ws = null; + } +} + // proxyFromEnv returns the configured egress proxy URL (sandboxed/corporate networks that only allow // outbound traffic via HTTP(S)_PROXY), or "" when none is set. Shared by the fetch dispatcher setup // in runner.ts and the ws client below. @@ -51,6 +179,8 @@ function wsProxyAgent(): HttpsProxyAgent | undefined { export class TestContext { readonly client: OpenApiFetchClient; + private wsConn_: wsConnection | undefined; + private wsConnInit_: Promise | undefined; private status?: NonNullable; private nextWSReq = 0; @@ -139,6 +269,24 @@ export class TestContext { return found; } + // preloadSamples eagerly resolves the most commonly-needed sample types so + // that downstream tests hit the cache rather than triggering redundant probe + // chains. getSampleAddressTx is deliberately excluded here — it can probe + // 40+ transactions with paginated address lookups and is only needed by a + // handful of tests; it stays lazy. + async preloadSamples() { + await this.getSampleIndexedBlock(); + await this.getSampleTxID(); + await this.getSampleAddress(); + try { + await this.sampleFiatTickerOrSkip(); + } catch (error) { + if (!(error instanceof SkipTest)) { + throw error; + } + } + } + async getSampleIndexedHeight() { if (this.sampleIndexResolved) { return this.sampleIndexHash ? { height: this.sampleIndexHeight, hash: this.sampleIndexHash } : undefined; @@ -498,18 +646,52 @@ export class TestContext { } async wsCallWithID(id: string, method: WsMethod, params: unknown, dataSchemaRef?: string) { + const conn = await this.ensureWSConnection(); const request: WsEnvelope = { id, method, params }; - try { - return await this.wsCallOnce(this.wsURL, request, dataSchemaRef); - } catch (error) { - const upgraded = upgradeWSBaseToWSS(this.wsURL); - if (!upgraded) { - throw error; - } - const result = await this.wsCallOnce(upgraded, request, dataSchemaRef); - this.wsURL = upgraded; - return result; + return conn.request(request, dataSchemaRef); + } + + // ensureWSConnection lazily dials the shared WebSocket connection (so WS + // tests pay a single handshake per coin, not one per call) and redials when + // the previous connection died. The connection multiplexes concurrent + // requests by ID, so callers share it freely. The first dial tries the + // configured ws:// URL; if that fails it tries wss:// and updates this.wsURL + // so later dials skip the fallback. wsConnInit_ guards against concurrent + // calls dialing twice; a failed dial is not cached, so a transient outage + // does not poison every later WS test of the coin. + private ensureWSConnection(): Promise { + if (this.wsConn_ && !this.wsConn_.isClosed) { + return Promise.resolve(this.wsConn_); + } + if (!this.wsConnInit_) { + this.wsConnInit_ = (async () => { + const conn = new wsConnection(this.wsURL, this.contract); + try { + await conn.connect(); + return conn; + } catch (firstError) { + const upgraded = upgradeWSBaseToWSS(this.wsURL); + if (!upgraded) { + throw firstError; + } + const conn2 = new wsConnection(upgraded, this.contract); + await conn2.connect(); + this.wsURL = upgraded; + return conn2; + } + })().then( + (conn) => { + this.wsConn_ = conn; + this.wsConnInit_ = undefined; + return conn; + }, + (error) => { + this.wsConnInit_ = undefined; + throw error; + }, + ); } + return this.wsConnInit_; } isEVMTxID(txid: string) { @@ -521,6 +703,12 @@ export class TestContext { return isEVMAddressValue(address) || (this.coin === "tron" && isTronAddress(address)); } + close() { + this.wsConn_?.close(); + this.wsConn_ = undefined; + this.wsConnInit_ = undefined; + } + private async probeCapabilities() { return { utxo: await this.probeUTXOSupport(), @@ -660,41 +848,5 @@ export class TestContext { return summary; } - private wsCallOnce(wsURL: string, request: WsEnvelope, dataSchemaRef?: string) { - return new Promise((resolve, reject) => { - const ws = new WebSocket(wsURL, { - handshakeTimeout: wsDialTimeoutMs, - rejectUnauthorized: process.env.OPENAPI_INSECURE_TLS === "0", - agent: wsProxyAgent(), - }); - const timeout = setTimeout(() => { - ws.terminate(); - reject(new Error(`websocket ${request.method} timed out for ${wsURL}`)); - }, wsMessageTimeoutMs); - ws.on("open", () => { - ws.send(JSON.stringify(request)); - }); - ws.on("message", (data) => { - const response = JSON.parse(data.toString()) as WsResponse; - if (response.id !== request.id) { - return; - } - clearTimeout(timeout); - ws.close(); - if (isWsError(response.data)) { - reject(new Error(`websocket ${request.method} returned error: ${response.data.error.message}`)); - return; - } - if (dataSchemaRef) { - this.contract.validateSchemaRef(dataSchemaRef, `WS ${request.method} response data`, response.data); - } - resolve(response.data as T); - }); - ws.on("error", (error) => { - clearTimeout(timeout); - reject(error); - }); - }); - } } diff --git a/tests/openapi/src/runner.ts b/tests/openapi/src/runner.ts index b774ee69dd..c3cb663843 100644 --- a/tests/openapi/src/runner.ts +++ b/tests/openapi/src/runner.ts @@ -34,10 +34,31 @@ export async function runOpenApiE2E() { return; } + // Run coins in parallel — they connect to independent Blockbook instances, + // so there is no shared state to contend over. Each coin buffers its own + // console output and flushes it immediately upon completion so that even + // if another coin hangs the completed results are still visible. const summaries: CoinSummary[] = []; - for (const coin of selectedCoins) { - summaries.push(await runCoin(coin, contract)); - } + await Promise.all( + selectedCoins.map(async (coin) => { + try { + const { summary, output } = await runCoin(coin, contract); + for (const line of output) { + console.log(line); + } + summaries.push(summary); + } catch (error) { + // A rejection here means the coin aborted outside runCoin's own error + // handling. Record it as a coin-level failure so the run cannot go + // green while a selected coin was never tested. + const message = errorMessage(error); + console.error(`\nOpenAPI e2e ${coin}: aborted: ${message}`); + summaries.push( + summarize(coin, [{ coin, name: "CoinRun", group: "run", status: "fail", durationMs: 0, message }]), + ); + } + }), + ); // Per-coin and aggregate summary so a run that is green-but-empty (everything skipped) is // visible at a glance, not hidden behind a zero-failure exit. @@ -78,17 +99,22 @@ export async function runOpenApiE2E() { console.log(`\nOpenAPI e2e passed for ${summaries.length} coin(s): ${selectedCoins.join(", ")}`); } -async function runCoin(coin: string, contract: OpenApiContract): Promise { +async function runCoin(coin: string, contract: OpenApiContract): Promise<{ summary: CoinSummary; output: string[] }> { const testsConfig = loadTestsConfig(); const apiTests = testsConfig[coin]?.api ?? []; const results: TestResult[] = []; + const output: string[] = []; + + const emit = (msg: string) => output.push(msg); + if (apiTests.length === 0) { - console.log(`OpenAPI e2e ${coin}: no api tests configured, skipping.`); - return summarize(coin, results); + emit(`OpenAPI e2e ${coin}: no api tests configured, skipping.`); + return { summary: summarize(coin, results), output }; } const ctx = await TestContext.create(coin, contract); - console.log(`\nOpenAPI e2e ${coin}: ${apiTests.length} tests`); + try { + emit(`\nOpenAPI e2e ${coin}: ${apiTests.length} tests`); // Status preflight: if the node is unreachable or not in sync, every sample-derived test would // fail or skip downstream for the same root cause. Surface it once as a single coin-level failure @@ -97,16 +123,29 @@ async function runCoin(coin: string, contract: OpenApiContract): Promise 0) { const message = `coin ${coin} had 0 passing tests (${summary.skip} skipped) — likely not in sync or the recent-block window yielded no usable sample data`; - console.error(` fail (coin health): ${message}`); + emit(` fail (coin health): ${message}`); results.push({ coin, name: "CoinHealth", group: "health", status: "fail", durationMs: 0, message }); - return summarize(coin, results); + return { summary: summarize(coin, results), output }; } - return summary; + return { summary, output }; + } finally { + ctx.close(); + } }