rpc refactoring #1296
rpc refactoring #1296
Conversation
khaidarkairbek
changed the title
khaidarkairbek
changed the title
| ? await _eth_getBlockByNumber(args.requestQueue, { | ||
| blockTag: "latest", | ||
| }) | ||
| : await _eth_getBlockByHash(args.requestQueue, { |
There was a problem hiding this comment.
how come we still need to fetch the block? Does newHeads not return the full block object?
There was a problem hiding this comment.
newHeads returns the block header, which does not include transactions. So, we still need to fetch via _eth_getBlockByHash to access complete block data.
| // After a certain number of attempts, emit a fatal error. | ||
| if (++consecutiveErrors === ERROR_TIMEOUT.length) { | ||
| args.common.logger.error({ | ||
| service: "realtime", | ||
| msg: `Fatal warning: Failed to subscribe to the latest '${args.network.name}' block after ${ERROR_TIMEOUT.length} attempts.`, | ||
| error, | ||
| }); | ||
|
|
||
| args.onFatalError(error); |
There was a problem hiding this comment.
Is there any way we could reuse the same retry logic we have in enqueue?
| export type SubscribeParameters = Parameters< | ||
| ResolvedWebSocketTransport["value"]["subscribe"] | ||
| >[0] & { | ||
| method: "eth_subscribe"; | ||
| }; | ||
|
|
||
| export type SubscribeReturnType = Awaited< | ||
| ReturnType<ResolvedWebSocketTransport["value"]["subscribe"]> | ||
| >; |
There was a problem hiding this comment.
Are these types from viem?
There was a problem hiding this comment.
Yes, ResolvedWebSocketTransport is based on the WebSocketTransport type defined in viem.
type ResolvedWebSocketTransport = Omit<
ReturnType<WebSocketTransport>,
"value"
> & {
value: NonNullable<ReturnType<WebSocketTransport>["value"]>;
};| const wsTransport = resolveWebsocketTransport(network.transport); | ||
|
|
||
| if (wsTransport === undefined) { | ||
| throw new Error( |
There was a problem hiding this comment.
I think this should be NonRetryableError
There was a problem hiding this comment.
agree, it should always be defined by the way we make use of _eth_subscribe in sync-realtime.
| export function resolveWebsocketTransport( | ||
| transport: ReturnType<Transport>, | ||
| ): ResolvedWebSocketTransport | undefined { | ||
| if (transport.config.type === "http") { | ||
| return undefined; | ||
| } | ||
|
|
||
| if (transport.config.type === "fallback") { | ||
| const fallbackTransport: ReturnType<FallbackTransport> = | ||
| transport as ReturnType<FallbackTransport>; | ||
|
|
||
| const wsTransport = fallbackTransport.value!.transports.find( | ||
| (t: ReturnType<Transport>) => t.config.type === "webSocket", | ||
| ) as ResolvedWebSocketTransport | undefined; | ||
|
|
||
| return wsTransport; | ||
| } | ||
|
|
||
| if (transport.config.type === "webSocket") { | ||
| return transport as ResolvedWebSocketTransport; | ||
| } | ||
|
|
||
| return undefined; | ||
| } |
There was a problem hiding this comment.
What do you think about getting away from using the Transport type in this module?
In the future this would let us implement features that aren't available in the transports like request cancellation. I also feel confident that ponder.config.ts should use rpc urls, not transports.
There was a problem hiding this comment.
I think this makes sense, and we can make this change once we get to config and switch to rpcUrls instead of transports.
packages/core/src/rpc/index.ts
Outdated
| subscribe: (params: SubscribeParameters) => { | ||
| const stopClockLag = startClock(); | ||
|
|
||
| return requestQueue.add({ request: params, stopClockLag }); |
There was a problem hiding this comment.
Thinking about subscribe() not adding to the queue. I think it's safe to assume that we aren't going to get a 429 with the requests sent to it, because there should just be one per chain.
There was a problem hiding this comment.
Yeah that makes sense.
packages/core/src/rpc/index.ts
Outdated
| return await withRetry({ | ||
| fn: subscribe, | ||
| request: task.request, | ||
| }); |
There was a problem hiding this comment.
What are the difference between the errors throw in subscribe() vs the errors passed to onError() in the subscribe params
There was a problem hiding this comment.
I was confused a bit too, as far as I know, the onError handles the errors thrown during the active websocket connection, while the errors thrown in subscribe happen when establishing this connection. Feel free to correct me here.
There was a problem hiding this comment.
I'm not sure either but that makes sense
| interval = setInterval(enqueue, args.network.pollingInterval); | ||
| } else { | ||
| // Check on the connection every second | ||
| interval = setInterval(watchWebsocket, 1000); |
There was a problem hiding this comment.
Do we need to do this? Looking at the viem implementation of watchBlocks seems like they are handling it differently
https://github.com/wevm/viem/blob/main/src/actions/public/watchBlocks.ts
There was a problem hiding this comment.
My idea behind this implementation was to have persistency of websocket connection, so if error happens during active connection, we drop it and reconnect again. From the watchblocks implementation, they seem to leave this flexibility via onError callback.
Relates to #1291 and #1290