Skip to content

Commit 267d915

Browse files
committed
feat(rpc): export retried RPC errors and ingestion restarts as metrics
Errors that used to stop evm-dump or restart real-time ingestion are now retried, so a dump stuck on a persistently failing endpoint no longer shows up as restarts. Counting retried errors by kind, per endpoint, keeps that condition alertable. In the data service the RPC clients live in worker threads, so their counts are read on each scrape and the final counts of closed workers are kept. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
1 parent 9e0110b commit 267d915

20 files changed

Lines changed: 500 additions & 10 deletions
Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
{
2+
"changes": [
3+
{
4+
"packageName": "@subsquid/evm-data-service",
5+
"comment": "Report retried RPC errors from the worker threads",
6+
"type": "patch"
7+
}
8+
],
9+
"packageName": "@subsquid/evm-data-service"
10+
}
Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
{
2+
"changes": [
3+
{
4+
"packageName": "@subsquid/evm-rpc",
5+
"comment": "Classify retried errors as rate_limit, transient, internal or no_result for metrics",
6+
"type": "patch"
7+
}
8+
],
9+
"packageName": "@subsquid/evm-rpc"
10+
}
Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
{
2+
"changes": [
3+
{
4+
"packageName": "@subsquid/rpc-client",
5+
"comment": "Count retried errors by kind: `getRetryKind()` and `retriedErrors` in `getMetrics()`",
6+
"type": "minor"
7+
}
8+
],
9+
"packageName": "@subsquid/rpc-client"
10+
}
Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
{
2+
"changes": [
3+
{
4+
"packageName": "@subsquid/util-internal-data-service",
5+
"comment": "Export `sqd_hotblocks_ingestion_restarts_total` and `sqd_chain_rpc_retried_errors_total`, with `RpcMetricsCollector` to read RPC clients in worker threads",
6+
"type": "patch"
7+
}
8+
],
9+
"packageName": "@subsquid/util-internal-data-service"
10+
}
Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
{
2+
"changes": [
3+
{
4+
"packageName": "@subsquid/util-internal-dump-cli",
5+
"comment": "Export `sqd_chain_rpc_retried_errors_total` by kind",
6+
"type": "patch"
7+
}
8+
],
9+
"packageName": "@subsquid/util-internal-dump-cli"
10+
}

‎evm/evm-data-service/src/data-source/setup.ts‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
import {type CallFrameValidationMode, Rpc, EvmRpcDataSource, EvmRpcClient} from '@subsquid/evm-rpc'
22
import type {Block, DataSource} from '@subsquid/util-internal-data-service'
33
import {createLogger} from '@subsquid/logger'
4+
import type {RpcMetrics} from '@subsquid/rpc-client'
45
import {Mapping} from './mapping'
56

67

@@ -36,7 +37,13 @@ export interface DataSourceOptions {
3637
}
3738

3839

39-
export function createDataSource(options: DataSourceOptions): DataSource<Block> {
40+
export interface DataSourceSetup {
41+
source: DataSource<Block>
42+
getRpcMetrics(): RpcMetrics
43+
}
44+
45+
46+
export function createDataSource(options: DataSourceOptions): DataSourceSetup {
4047
let httpRpcClient = new EvmRpcClient({
4148
url: options.httpRpc,
4249
maxBatchCallSize: options.httpRpcMaxBatchCallSize,
@@ -78,8 +85,12 @@ export function createDataSource(options: DataSourceOptions): DataSource<Block>
7885
strideSize: options.httpRpcStrideSize,
7986
strideConcurrency: options.httpRpcStrideConcurrency
8087
})
81-
return new Mapping(rpcSource, {
88+
let source = new Mapping(rpcSource, {
8289
withTraces: options.withTraces,
8390
withStateDiffs: options.withStatediffs,
8491
})
92+
return {
93+
source,
94+
getRpcMetrics: () => httpRpcClient.getMetrics()
95+
}
8596
}

‎evm/evm-data-service/src/data-source/worker-client.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
import type {RpcMetrics} from '@subsquid/rpc-client'
12
import {Block, BlockRef, BlockStream, DataSource, StreamRequest} from '@subsquid/util-internal-data-service'
23
import {Client, createWorker} from '@subsquid/util-internal-worker-thread'
34
import {DataSourceOptions} from './setup'
@@ -33,6 +34,10 @@ export class WorkerClient implements DataSource<Block> {
3334
return this.stream('getStream', req)
3435
}
3536

37+
getRpcMetrics(): Promise<RpcMetrics> {
38+
return this.worker.call('getRpcMetrics', [])
39+
}
40+
3641
private async *stream(method: string, req: StreamRequest): BlockStream<Block> {
3742
let stream: BlockStream<Block> = await this.worker.call(method, [req])
3843
yield* stream

‎evm/evm-data-service/src/data-source/worker.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ import {getServer, getServerArguments} from '@subsquid/util-internal-worker-thre
33
import {createDataSource} from './setup'
44

55

6-
const source = createDataSource(getServerArguments())
6+
const {source, getRpcMetrics} = createDataSource(getServerArguments())
77

88

99
getServer()
@@ -15,4 +15,5 @@ getServer()
1515
.def('getStream', (req: StreamRequest) => {
1616
return source.getStream(req)
1717
})
18+
.def('getRpcMetrics', () => getRpcMetrics())
1819
.start()

‎evm/evm-data-service/src/main.ts‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import {
66
type Block,
77
type BlockStream,
88
type DataSource,
9+
RpcMetricsCollector,
910
runDataService,
1011
type StreamRequest
1112
} from '@subsquid/util-internal-data-service'
@@ -122,7 +123,9 @@ runProgram(async () => {
122123
useGasUsedForReceiptsRoot: args.useGasUsedForReceiptsRoot
123124
}
124125

126+
let rpcMetrics = new RpcMetricsCollector()
125127
let mainWorker = new WorkerClient(dataSourceOptions)
128+
rpcMetrics.add(mainWorker)
126129
let service: Awaited<ReturnType<typeof runDataService>> | undefined
127130
let dataSource: DataSource<Block> = {
128131
getHead() {
@@ -133,11 +136,13 @@ runProgram(async () => {
133136
},
134137
async *getFinalizedStream(req: StreamRequest): BlockStream<Block> {
135138
let worker = new WorkerClient(dataSourceOptions)
139+
rpcMetrics.add(worker)
136140
service?.metrics.incActiveWorkers()
137141
try {
138142
yield* worker.getFinalizedStream(req)
139143
} finally {
140144
service?.metrics.decActiveWorkers()
145+
await rpcMetrics.remove(worker)
141146
worker.close()
142147
}
143148
},
@@ -152,6 +157,7 @@ runProgram(async () => {
152157
port: args.port,
153158
autoAdjustFinalizedHead: args.autoAdjustFinalizedHead
154159
})
160+
service.metrics.setRpcMetricsSource(() => rpcMetrics.collect())
155161

156162
log.info(`listening on port ${service.port}`)
157163
return waitForInterruption(service)

‎evm/evm-rpc/src/rpc-client.ts‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { RpcClient, RpcError, RpcClientOptions } from '@subsquid/rpc-client'
1+
import { RetryError, RpcClient, RpcError, RpcClientOptions } from '@subsquid/rpc-client'
22
import { HttpError } from '@subsquid/http-client'
33

44
export interface EvmRpcClientOptions extends RpcClientOptions {
@@ -48,6 +48,23 @@ export class EvmRpcClient extends RpcClient {
4848
return false
4949
}
5050

51+
getRetryKind(err: Error): string {
52+
if (err instanceof RpcError) {
53+
if (this.isRpcRateLimitError(err)) return 'rate_limit'
54+
if (this.isRpcTransientError(err)) return 'transient'
55+
let kind = super.getRetryKind(err)
56+
if (kind != 'other') return kind
57+
if (this.isRpcInternalError(err)) return 'internal'
58+
}
59+
if (err instanceof HttpError && err.response.status === 500) {
60+
return this.retryInternalServerErrors ? 'internal' : 'transient'
61+
}
62+
// Thrown here when the endpoint had no usable answer yet: a null or an
63+
// error text in place of a result, or a block height it does not serve yet.
64+
if (err instanceof RetryError) return 'no_result'
65+
return super.getRetryKind(err)
66+
}
67+
5168
isRpcInternalError(err: RpcError): boolean {
5269
return (
5370
err.code === -32000 || // generic "catch-all" code

0 commit comments

Comments
 (0)