From 33b8fd123845442acde88245f141fb60b29d9a06 Mon Sep 17 00:00:00 2001 From: pragmaxim Date: Thu, 2 Jul 2026 12:46:21 +0200 Subject: [PATCH 1/8] perf(e2e): parallelize cross-coin execution, preload samples, pool WS connections, reduce timeouts - Use Promise.all for cross-coin execution: coins run concurrently (they connect to independent backends). Each coin buffers its own output and flushes contiguously after completion. - Add preloadSamples() to TestContext: eagerly resolves block, tx, address, and fiat samples once at coin startup so downstream tests hit the cache instead of triggering redundant probe chains. - Replace per-call WebSocket connections with a wsPool: maintains 3 persistent connections, eliminating the 5s handshake overhead for every WS test. Pool acquires/releases per request with automatic dead-connection replacement. - Reduce timeouts: HTTP 30s->15s, WS dial 5s->3s, WS message 15s->10s. --- tests/openapi/src/client.ts | 2 +- tests/openapi/src/constants.ts | 4 +- tests/openapi/src/context.ts | 265 +++++++++++++++++++++++++++------ tests/openapi/src/runner.ts | 54 ++++--- 4 files changed, 260 insertions(+), 65 deletions(-) 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..15569422b1 100644 --- a/tests/openapi/src/context.ts +++ b/tests/openapi/src/context.ts @@ -31,6 +31,181 @@ import { import type { Capability, AddressResponse, BlockHashResponse, BlockResponse, BlockSummary, FiatTickerResponse, StatusResponse, TxResponse, UtxoResponse, WsEnvelope, WsInfoResponse, WsMethod, WsResponse } from "./types.js"; +// --------------------------------------------------------------------------- +// WebSocket connection pool +// --------------------------------------------------------------------------- + +// 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) => { + this.ws = new WebSocket(this.url, { + handshakeTimeout: wsDialTimeoutMs, + rejectUnauthorized: process.env.OPENAPI_INSECURE_TLS === "0", + agent: wsProxyAgent(), + }); + this.ws.on("open", () => resolve()); + this.ws.on("error", (err) => { + this.closed = true; + reject(err); + }); + this.ws.on("close", () => { + 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) { + this.ws.send(JSON.stringify(request)); + } 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; + } +} + +// wsPool maintains a small set of persistent WebSocket connections so that +// WS tests don't pay a handshake timeout per call. acquire/release follow +// a simple lease pattern; callers must release after the request completes. +class wsPool { + private conns: wsConnection[] = []; + private busy = new Set(); + + constructor( + private readonly url: string, + private readonly contract: OpenApiContract, + private readonly size: number, + ) {} + + get aliveCount(): number { + return this.conns.length; + } + + async init(): Promise { + const errors: Error[] = []; + for (let i = 0; i < this.size; i++) { + try { + const c = new wsConnection(this.url, this.contract); + await c.connect(); + this.conns.push(c); + } catch (error) { + errors.push(error instanceof Error ? error : new Error(String(error))); + } + } + // If none connected, propagate the first error so the caller can attempt + // a protocol upgrade (ws→wss). A partial pool is still usable. + if (this.conns.length === 0 && errors.length > 0) { + throw errors[0]; + } + } + + async acquire(): Promise { + // eslint-disable-next-line no-constant-condition + while (true) { + for (const c of this.conns) { + if (!this.busy.has(c) && !c.isClosed) { + this.busy.add(c); + return c; + } + } + // All connections are busy or dead. If any died, replace them lazily. + for (let i = 0; i < this.conns.length; i++) { + if (this.conns[i].isClosed) { + try { + const c = new wsConnection(this.url, this.contract); + await c.connect(); + this.conns[i] = c; + this.busy.add(c); + return c; + } catch { + // Replacement failed, try the next slot. + } + } + } + await new Promise((resolve) => setTimeout(resolve, 20)); + } + } + + release(conn: wsConnection): void { + this.busy.delete(conn); + } + + closeAll(): void { + for (const c of this.conns) { + c.close(); + } + this.conns = []; + this.busy.clear(); + } +} + // 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 +226,7 @@ function wsProxyAgent(): HttpsProxyAgent | undefined { export class TestContext { readonly client: OpenApiFetchClient; + private wsPool_: wsPool | undefined; private status?: NonNullable; private nextWSReq = 0; @@ -139,6 +315,26 @@ export class TestContext { return found; } + // preloadSamples eagerly resolves every sample type so that all downstream + // tests hit the cache rather than triggering redundant probe chains. Safe to + // call multiple times; subsequent calls are no-ops because every resolver is + // guarded by a resolved flag. A SkipTest from a fiat ticker that is + // temporarily unavailable is swallowed — tests that need fiat data check the + // availability flag and skip themselves. + async preloadSamples() { + await this.getSampleIndexedBlock(); + await this.getSampleTxID(); + await this.getSampleAddress(); + await this.getSampleAddressTx(); + 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,17 +694,32 @@ export class TestContext { } async wsCallWithID(id: string, method: WsMethod, params: unknown, dataSchemaRef?: string) { + // Lazily initialise the WS connection pool. The first call tries the + // configured ws:// URL; if that fails the pool tries wss:// and updates + // this.wsURL so subsequent calls skip the fallback. + if (!this.wsPool_) { + this.wsPool_ = new wsPool(this.wsURL, this.contract, 3); + try { + await this.wsPool_.init(); + } catch (firstError) { + const upgraded = upgradeWSBaseToWSS(this.wsURL); + if (!upgraded) { + this.wsPool_ = undefined; + throw firstError; + } + const pool2 = new wsPool(upgraded, this.contract, 3); + await pool2.init(); + this.wsPool_ = pool2; + this.wsURL = upgraded; + } + } + const request: WsEnvelope = { id, method, params }; + const conn = await this.wsPool_.acquire(); 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 await conn.request(request, dataSchemaRef); + } finally { + this.wsPool_.release(conn); } } @@ -660,41 +871,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..1021f1086a 100644 --- a/tests/openapi/src/runner.ts +++ b/tests/openapi/src/runner.ts @@ -34,17 +34,29 @@ export async function runOpenApiE2E() { return; } - const summaries: CoinSummary[] = []; - for (const coin of selectedCoins) { - summaries.push(await runCoin(coin, contract)); + // 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 after completion so per-coin lines are + // contiguous rather than interleaved. + type CoinResult = { summary: CoinSummary; output: string[] }; + const results: CoinResult[] = await Promise.all( + selectedCoins.map(async (coin) => runCoin(coin, contract)), + ); + + // Flush buffered per-coin output in the original order. + for (const { output } of results) { + for (const line of output) { + console.log(line); + } } // 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. console.log("\nOpenAPI e2e summary:"); - for (const summary of summaries) { + for (const { summary } of results) { console.log(` ${summary.coin}: ${summary.ok} ok, ${summary.skip} skip, ${summary.fail} fail`); } + const summaries = results.map((r) => r.summary); const totals = summaries.reduce( (acc, s) => ({ ok: acc.ok + s.ok, skip: acc.skip + s.skip, fail: acc.fail + s.fail }), { ok: 0, skip: 0, fail: 0 }, @@ -78,17 +90,21 @@ 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`); + 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 +113,20 @@ 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 }; } From 5ae78af71fd9c6e0106fc673b40dd27861d4a1ac Mon Sep 17 00:00:00 2001 From: pragmaxim Date: Thu, 2 Jul 2026 13:04:45 +0200 Subject: [PATCH 2/8] fix(e2e): add connect timeout to WS pool, skip heavy preload, flush output per-coin MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add a mandatory timeout to wsConnection.connect() using a 'settled' guard that rejects after wsDialTimeoutMs even when DNS or TCP stalls (the ws library's handshakeTimeout only applies after TCP connect). - Make wsPool.init() parallel (Promise.allSettled) so all connections are attempted concurrently; a partial pool is still usable. - Remove getSampleAddressTx from preloadSamples() — it probes up to 40 transactions with paginated lookups and is only needed by a few address-listing tests; keep it lazy. - Flush each coin's output immediately upon completion instead of waiting for Promise.all, so results from finished coins are visible even if another coin hangs. - Use Promise.allSettled instead of Promise.all so a rejection in one coin does not discard results from other coins. --- tests/openapi/src/context.ts | 63 ++++++++++++++++++++++++------------ tests/openapi/src/runner.ts | 26 +++++++-------- 2 files changed, 55 insertions(+), 34 deletions(-) diff --git a/tests/openapi/src/context.ts b/tests/openapi/src/context.ts index 15569422b1..63ebcc2123 100644 --- a/tests/openapi/src/context.ts +++ b/tests/openapi/src/context.ts @@ -55,18 +55,40 @@ class wsConnection { } connect(): Promise { - return new Promise((resolve, reject) => { + 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", () => resolve()); + 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); @@ -149,20 +171,23 @@ class wsPool { } async init(): Promise { - const errors: Error[] = []; - for (let i = 0; i < this.size; i++) { - try { + const results = await Promise.allSettled( + Array.from({ length: this.size }, () => { const c = new wsConnection(this.url, this.contract); - await c.connect(); - this.conns.push(c); - } catch (error) { - errors.push(error instanceof Error ? error : new Error(String(error))); + return c.connect().then(() => c); + }), + ); + for (const r of results) { + if (r.status === "fulfilled") { + this.conns.push(r.value); } } - // If none connected, propagate the first error so the caller can attempt - // a protocol upgrade (ws→wss). A partial pool is still usable. - if (this.conns.length === 0 && errors.length > 0) { - throw errors[0]; + // A partial pool (fewer than this.size connections) is still usable — + // acquire() replaces dead slots lazily. If nothing connected, propagate + // the first rejection so the caller can attempt a protocol upgrade. + if (this.conns.length === 0) { + const first = results.find((r) => r.status === "rejected"); + throw first ? (first as PromiseRejectedResult).reason : new Error("wsPool init: no connections and no errors"); } } @@ -315,17 +340,15 @@ export class TestContext { return found; } - // preloadSamples eagerly resolves every sample type so that all downstream - // tests hit the cache rather than triggering redundant probe chains. Safe to - // call multiple times; subsequent calls are no-ops because every resolver is - // guarded by a resolved flag. A SkipTest from a fiat ticker that is - // temporarily unavailable is swallowed — tests that need fiat data check the - // availability flag and skip themselves. + // 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(); - await this.getSampleAddressTx(); try { await this.sampleFiatTickerOrSkip(); } catch (error) { diff --git a/tests/openapi/src/runner.ts b/tests/openapi/src/runner.ts index 1021f1086a..675312fa99 100644 --- a/tests/openapi/src/runner.ts +++ b/tests/openapi/src/runner.ts @@ -36,27 +36,25 @@ export async function runOpenApiE2E() { // 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 after completion so per-coin lines are - // contiguous rather than interleaved. - type CoinResult = { summary: CoinSummary; output: string[] }; - const results: CoinResult[] = await Promise.all( - selectedCoins.map(async (coin) => runCoin(coin, contract)), + // console output and flushes it immediately upon completion so that even + // if another coin hangs the completed results are still visible. + const summaries: CoinSummary[] = []; + await Promise.allSettled( + selectedCoins.map(async (coin) => { + const { summary, output } = await runCoin(coin, contract); + for (const line of output) { + console.log(line); + } + summaries.push(summary); + }), ); - // Flush buffered per-coin output in the original order. - for (const { output } of results) { - for (const line of output) { - console.log(line); - } - } - // 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. console.log("\nOpenAPI e2e summary:"); - for (const { summary } of results) { + for (const summary of summaries) { console.log(` ${summary.coin}: ${summary.ok} ok, ${summary.skip} skip, ${summary.fail} fail`); } - const summaries = results.map((r) => r.summary); const totals = summaries.reduce( (acc, s) => ({ ok: acc.ok + s.ok, skip: acc.skip + s.skip, fail: acc.fail + s.fail }), { ok: 0, skip: 0, fail: 0 }, From eae15ac0fc3927a63079eb7b5030d919c2b4e7bb Mon Sep 17 00:00:00 2001 From: pragmaxim Date: Thu, 2 Jul 2026 19:02:36 +0000 Subject: [PATCH 3/8] fix(e2e): reject pending WS request when ws.send() throws, guard pool init race E2e-only remainder of the original review-findings commit; its tests/integration.go part (configurable init retry) was dropped along with the Go-side test parallelization. --- tests/openapi/src/context.ts | 54 ++++++++++++++++++++++++------------ tests/openapi/src/runner.ts | 8 ++++-- 2 files changed, 43 insertions(+), 19 deletions(-) diff --git a/tests/openapi/src/context.ts b/tests/openapi/src/context.ts index 63ebcc2123..01e93bc360 100644 --- a/tests/openapi/src/context.ts +++ b/tests/openapi/src/context.ts @@ -137,7 +137,13 @@ class wsConnection { timer, }); if (this.ws) { - this.ws.send(JSON.stringify(request)); + 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); @@ -252,6 +258,7 @@ function wsProxyAgent(): HttpsProxyAgent | undefined { export class TestContext { readonly client: OpenApiFetchClient; private wsPool_: wsPool | undefined; + private wsPoolInit_: Promise | undefined; private status?: NonNullable; private nextWSReq = 0; @@ -720,29 +727,36 @@ export class TestContext { // Lazily initialise the WS connection pool. The first call tries the // configured ws:// URL; if that fails the pool tries wss:// and updates // this.wsURL so subsequent calls skip the fallback. - if (!this.wsPool_) { - this.wsPool_ = new wsPool(this.wsURL, this.contract, 3); - try { - await this.wsPool_.init(); - } catch (firstError) { - const upgraded = upgradeWSBaseToWSS(this.wsURL); - if (!upgraded) { - this.wsPool_ = undefined; - throw firstError; + // wsPoolInit_ guards against concurrent calls creating two pools. + if (!this.wsPoolInit_) { + this.wsPoolInit_ = (async () => { + const pool = new wsPool(this.wsURL, this.contract, 3); + try { + await pool.init(); + } catch (firstError) { + const upgraded = upgradeWSBaseToWSS(this.wsURL); + if (!upgraded) { + throw firstError; + } + const pool2 = new wsPool(upgraded, this.contract, 3); + await pool2.init(); + this.wsURL = upgraded; + this.wsPool_ = pool2; + return; } - const pool2 = new wsPool(upgraded, this.contract, 3); - await pool2.init(); - this.wsPool_ = pool2; - this.wsURL = upgraded; - } + this.wsPool_ = pool; + })(); } + await this.wsPoolInit_; + // wsPoolInit_ resolved, so wsPool_ is guaranteed to be set. + const pool = this.wsPool_!; const request: WsEnvelope = { id, method, params }; - const conn = await this.wsPool_.acquire(); + const conn = await pool.acquire(); try { return await conn.request(request, dataSchemaRef); } finally { - this.wsPool_.release(conn); + pool.release(conn); } } @@ -755,6 +769,12 @@ export class TestContext { return isEVMAddressValue(address) || (this.coin === "tron" && isTronAddress(address)); } + close() { + this.wsPool_?.closeAll(); + this.wsPool_ = undefined; + this.wsPoolInit_ = undefined; + } + private async probeCapabilities() { return { utxo: await this.probeUTXOSupport(), diff --git a/tests/openapi/src/runner.ts b/tests/openapi/src/runner.ts index 675312fa99..6b96f2f046 100644 --- a/tests/openapi/src/runner.ts +++ b/tests/openapi/src/runner.ts @@ -102,7 +102,8 @@ async function runCoin(coin: string, contract: OpenApiContract): Promise<{ summa } const ctx = await TestContext.create(coin, contract); - emit(`\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 @@ -162,5 +163,8 @@ async function runCoin(coin: string, contract: OpenApiContract): Promise<{ summa return { summary: summarize(coin, results), output }; } - return { summary, output }; + return { summary, output }; + } finally { + ctx.close(); + } } From f9b187ec79db2f4df12a64c9ecb9ba081525ed40 Mon Sep 17 00:00:00 2001 From: pragmaxim Date: Thu, 2 Jul 2026 14:39:14 +0200 Subject: [PATCH 4/8] feat: add rpc_sync_latency histogram for sync-internal RPC calls --- bchain/coins/eth/ethrpc.go | 27 ++++++++++++-- common/metrics.go | 1 + configs/grafana/panels.yaml | 10 ++++++ configs/grafana/template.json | 66 +++++++++++++++++++++++++++++++++++ configs/metrics.yaml | 6 ++++ 5 files changed, 108 insertions(+), 2 deletions(-) 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..de3b680d88 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 for 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. + queries: + p95: + promql: histogram_quantile(0.95, sum(rate({{name:rpc_sync_latency}}_bucket{job="blockbook", coin="$coin"}[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"}[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..ec6311bcb4 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 + 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 From adc6b3b2f9134d13ac4d926c48ab400d97e88936 Mon Sep 17 00:00:00 2001 From: pragmaxim Date: Thu, 2 Jul 2026 18:37:43 +0000 Subject: [PATCH 5/8] fix(e2e): record coin aborts as failures instead of swallowing rejections Promise.allSettled discarded runCoin rejections: a coin whose context creation or preload threw vanished from the summaries with its buffered output unflushed, and the run printed 'passed for N coin(s)' with exit code 0. Catch the rejection per coin and push a failed CoinRun summary so the aggregate failure path fires. Co-Authored-By: Claude Fable 5 --- tests/openapi/src/runner.ts | 21 ++++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/tests/openapi/src/runner.ts b/tests/openapi/src/runner.ts index 6b96f2f046..142376ac94 100644 --- a/tests/openapi/src/runner.ts +++ b/tests/openapi/src/runner.ts @@ -39,13 +39,24 @@ export async function runOpenApiE2E() { // console output and flushes it immediately upon completion so that even // if another coin hangs the completed results are still visible. const summaries: CoinSummary[] = []; - await Promise.allSettled( + await Promise.all( selectedCoins.map(async (coin) => { - const { summary, output } = await runCoin(coin, contract); - for (const line of output) { - console.log(line); + 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 }]), + ); } - summaries.push(summary); }), ); From a9904e98c59e5267a0be8619ddfbb13257bec70b Mon Sep 17 00:00:00 2001 From: pragmaxim Date: Thu, 2 Jul 2026 18:38:02 +0000 Subject: [PATCH 6/8] fix(e2e): fail the coin visibly when sample preload throws preloadSamples() ran bare inside runCoin, outside the per-test try/catch, so one transient HTTP or schema-validation error during the probe chain rejected the whole coin. Handle it like the status preflight: emit a single SamplePreload failure with the buffered output preserved and stop the coin there. Co-Authored-By: Claude Fable 5 --- tests/openapi/src/runner.ts | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/tests/openapi/src/runner.ts b/tests/openapi/src/runner.ts index 142376ac94..c3cb663843 100644 --- a/tests/openapi/src/runner.ts +++ b/tests/openapi/src/runner.ts @@ -129,8 +129,17 @@ async function runCoin(coin: string, contract: OpenApiContract): Promise<{ summa } // Eagerly resolve all samples once so downstream tests hit the cache - // instead of each triggering its own probe chain. - await ctx.preloadSamples(); + // instead of each triggering its own probe chain. A probe error here is a + // coin-level failure like the status preflight — every sample-derived test + // would fail downstream for the same root cause. + try { + await ctx.preloadSamples(); + } catch (error) { + const message = errorMessage(error); + emit(` fail (sample preload): ${message}`); + results.push({ coin, name: "SamplePreload", group: "preflight", status: "fail", durationMs: 0, message }); + return { summary: summarize(coin, results), output }; + } for (const testName of apiTests) { const def = testRegistry[testName]; From 962dbf49b3708cd463b66560ce87106f0afd6464 Mon Sep 17 00:00:00 2001 From: pragmaxim Date: Thu, 2 Jul 2026 19:04:52 +0000 Subject: [PATCH 7/8] refactor(e2e): replace WS pool with a single persistent connection wsConnection already multiplexes concurrent requests by ID and tests within a coin run sequentially, so the 3-connection lease pool bought nothing over one persistent socket while adding an unbounded acquire retry loop and a permanently-cached failed init. One shared connection with redial-on-close keeps the handshake-per-coin win; every wait is bounded by the dial and message timeouts, and a failed dial is not cached so the next call retries. Co-Authored-By: Claude Fable 5 --- tests/openapi/src/context.ts | 152 ++++++++++------------------------- 1 file changed, 43 insertions(+), 109 deletions(-) diff --git a/tests/openapi/src/context.ts b/tests/openapi/src/context.ts index 01e93bc360..996df20a89 100644 --- a/tests/openapi/src/context.ts +++ b/tests/openapi/src/context.ts @@ -32,7 +32,7 @@ import { import type { Capability, AddressResponse, BlockHashResponse, BlockResponse, BlockSummary, FiatTickerResponse, StatusResponse, TxResponse, UtxoResponse, WsEnvelope, WsInfoResponse, WsMethod, WsResponse } from "./types.js"; // --------------------------------------------------------------------------- -// WebSocket connection pool +// WebSocket connection // --------------------------------------------------------------------------- // wsConnection wraps a single persistent WebSocket and multiplexes requests @@ -159,84 +159,6 @@ class wsConnection { } } -// wsPool maintains a small set of persistent WebSocket connections so that -// WS tests don't pay a handshake timeout per call. acquire/release follow -// a simple lease pattern; callers must release after the request completes. -class wsPool { - private conns: wsConnection[] = []; - private busy = new Set(); - - constructor( - private readonly url: string, - private readonly contract: OpenApiContract, - private readonly size: number, - ) {} - - get aliveCount(): number { - return this.conns.length; - } - - async init(): Promise { - const results = await Promise.allSettled( - Array.from({ length: this.size }, () => { - const c = new wsConnection(this.url, this.contract); - return c.connect().then(() => c); - }), - ); - for (const r of results) { - if (r.status === "fulfilled") { - this.conns.push(r.value); - } - } - // A partial pool (fewer than this.size connections) is still usable — - // acquire() replaces dead slots lazily. If nothing connected, propagate - // the first rejection so the caller can attempt a protocol upgrade. - if (this.conns.length === 0) { - const first = results.find((r) => r.status === "rejected"); - throw first ? (first as PromiseRejectedResult).reason : new Error("wsPool init: no connections and no errors"); - } - } - - async acquire(): Promise { - // eslint-disable-next-line no-constant-condition - while (true) { - for (const c of this.conns) { - if (!this.busy.has(c) && !c.isClosed) { - this.busy.add(c); - return c; - } - } - // All connections are busy or dead. If any died, replace them lazily. - for (let i = 0; i < this.conns.length; i++) { - if (this.conns[i].isClosed) { - try { - const c = new wsConnection(this.url, this.contract); - await c.connect(); - this.conns[i] = c; - this.busy.add(c); - return c; - } catch { - // Replacement failed, try the next slot. - } - } - } - await new Promise((resolve) => setTimeout(resolve, 20)); - } - } - - release(conn: wsConnection): void { - this.busy.delete(conn); - } - - closeAll(): void { - for (const c of this.conns) { - c.close(); - } - this.conns = []; - this.busy.clear(); - } -} - // 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. @@ -257,8 +179,8 @@ function wsProxyAgent(): HttpsProxyAgent | undefined { export class TestContext { readonly client: OpenApiFetchClient; - private wsPool_: wsPool | undefined; - private wsPoolInit_: Promise | undefined; + private wsConn_: wsConnection | undefined; + private wsConnInit_: Promise | undefined; private status?: NonNullable; private nextWSReq = 0; @@ -724,40 +646,52 @@ export class TestContext { } async wsCallWithID(id: string, method: WsMethod, params: unknown, dataSchemaRef?: string) { - // Lazily initialise the WS connection pool. The first call tries the - // configured ws:// URL; if that fails the pool tries wss:// and updates - // this.wsURL so subsequent calls skip the fallback. - // wsPoolInit_ guards against concurrent calls creating two pools. - if (!this.wsPoolInit_) { - this.wsPoolInit_ = (async () => { - const pool = new wsPool(this.wsURL, this.contract, 3); + const conn = await this.ensureWSConnection(); + const request: WsEnvelope = { id, method, params }; + 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 pool.init(); + await conn.connect(); + return conn; } catch (firstError) { const upgraded = upgradeWSBaseToWSS(this.wsURL); if (!upgraded) { throw firstError; } - const pool2 = new wsPool(upgraded, this.contract, 3); - await pool2.init(); + const conn2 = new wsConnection(upgraded, this.contract); + await conn2.connect(); this.wsURL = upgraded; - this.wsPool_ = pool2; - return; + return conn2; } - this.wsPool_ = pool; - })(); - } - await this.wsPoolInit_; - // wsPoolInit_ resolved, so wsPool_ is guaranteed to be set. - const pool = this.wsPool_!; - - const request: WsEnvelope = { id, method, params }; - const conn = await pool.acquire(); - try { - return await conn.request(request, dataSchemaRef); - } finally { - pool.release(conn); + })().then( + (conn) => { + this.wsConn_ = conn; + this.wsConnInit_ = undefined; + return conn; + }, + (error) => { + this.wsConnInit_ = undefined; + throw error; + }, + ); } + return this.wsConnInit_; } isEVMTxID(txid: string) { @@ -770,9 +704,9 @@ export class TestContext { } close() { - this.wsPool_?.closeAll(); - this.wsPool_ = undefined; - this.wsPoolInit_ = undefined; + this.wsConn_?.close(); + this.wsConn_ = undefined; + this.wsConnInit_ = undefined; } private async probeCapabilities() { From e01beed6a2ac2d1a821ced2a1b4e56257a305982 Mon Sep 17 00:00:00 2001 From: pragmaxim Date: Thu, 2 Jul 2026 19:16:38 +0000 Subject: [PATCH 8/8] fix(metrics): reflect real sync RPC latency distribution in help text and panel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit debug_traceBlockByHash is typically sub-100ms, reaches seconds only when the node is overloaded, and ~10-15s observations are trace_timeout / rpc_timeout expiries — not routine latency. Say so in the histogram help text, and exclude failed calls from the Grafana p50/p95 queries so timeout expiries do not skew the quantiles (failures are already charted in the sync RPC errors panel). Co-Authored-By: Claude Fable 5 --- configs/grafana/panels.yaml | 6 +++--- configs/metrics.yaml | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/configs/grafana/panels.yaml b/configs/grafana/panels.yaml index de3b680d88..c694cb8e99 100644 --- a/configs/grafana/panels.yaml +++ b/configs/grafana/panels.yaml @@ -288,13 +288,13 @@ panels: legend: '{{subscription}} - {{event}}' sync.rpc_sync_latency: title: Sync RPC latency (p50/p95) - description: Round-trip time for 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. + 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"}[1m])) by (coin, method, le)) + 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"}[1m])) by (coin, method, le)) + 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 diff --git a/configs/metrics.yaml b/configs/metrics.yaml index ec6311bcb4..55d767fd83 100644 --- a/configs/metrics.yaml +++ b/configs/metrics.yaml @@ -403,7 +403,7 @@ metrics: 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 + 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: