From 03a642193c2083cbd80422754a1c60cb82744cf2 Mon Sep 17 00:00:00 2001 From: rouzwelt Date: Fri, 4 Sep 2026 22:47:51 +0000 Subject: [PATCH 1/2] init --- config.env.yaml | 2 + config.example.yaml | 8 ++ src/cli/commands/sweep.ts | 1 + src/cli/index.test.ts | 4 + src/cli/index.ts | 5 + src/common/abis/orderbook.ts | 2 + src/config/validators.test.ts | 29 +++++ src/config/validators.ts | 13 ++ src/config/yaml.test.ts | 6 + src/config/yaml.ts | 17 +++ src/core/process/order.test.ts | 5 +- src/core/process/order.ts | 17 +-- src/core/process/round.test.ts | 71 +++++++++- src/core/process/round.ts | 19 ++- src/rpc/helpers.ts | 5 + src/rpc/hooks.test.ts | 162 ----------------------- src/rpc/hooks.ts | 73 ----------- src/rpc/index.ts | 1 - src/rpc/rpc.test.ts | 130 ++++++++++++++++++- src/rpc/rpc.ts | 76 +++++++++-- src/rpc/transport.test.ts | 214 +++++++++++++++++++++++++++++- src/rpc/transport.ts | 75 +++++++---- src/state/index.test.ts | 231 ++++++++++++++++++++++++++++++++- src/state/index.ts | 121 ++++++++++++++++- test/e2e/e2e.test.js | 10 ++ 25 files changed, 989 insertions(+), 308 deletions(-) delete mode 100644 src/rpc/hooks.test.ts delete mode 100644 src/rpc/hooks.ts diff --git a/config.env.yaml b/config.env.yaml index 17c7486e..0ae075a3 100644 --- a/config.env.yaml +++ b/config.env.yaml @@ -4,6 +4,7 @@ walletCount: $WALLET_COUNT topupAmount: $TOPUP_AMOUNT rpc: $RPC_URL writeRpc: $WRITE_RPC +wsRpc: $WS_RPC subgraph: $SUBGRAPH contracts: v4: @@ -36,6 +37,7 @@ quoteGas: $QUOTE_GAS botMinBalance: $BOT_MIN_BALANCE gasPriceMultiplier: $GAS_PRICE_MULTIPLIER txTimeThreshold: $TX_TIME_THRESHOLD +blockTime: $BLOCK_TIME gasBoostProfitThreshold: $GAS_BOOST_PROFIT_THRESHOLD gasBoostMultiplier: $GAS_BOOST_MULTIPLIER gasBoostUsdThreshold: $GAS_BOOST_USD_THRESHOLD diff --git a/config.example.yaml b/config.example.yaml index b8ce9519..ce73dead 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -30,6 +30,11 @@ writeRpc: - url: https://rpc-url2.com weight: 0.5 +# Optional websocket rpc url, used explicitly for the block number watcher +# new heads subscription for the earliest possible block updates, when unset +# the block number is polled over http at blockTime intervals instead +wsRpc: wss://rpc-url1.com + # List of subgraph urls, required # for specifying more than 1 subgraph URL in the env, separate them by a comma # NOTE: for v6 orderbook subgraph, prepend the URL with "v6=" because the v5 and v6 @@ -117,6 +122,9 @@ gasPriceMultiplier: 107 # Time threshold (in ms) for a transaction mine time before it counts as a trigger to increase gas price multiplier for future transactions, default is 15000ms (15 seconds) txTimeThreshold: 15000 +# The average block time (in ms) of the operating chain, used as the polling interval of the block number watcher, default is 5000ms (5 seconds) +blockTime: 2000 + # Time (in minutes) to to check the operating wallet balances, 0 means dont ever check wallet balance, default is 15 mins checkWalletBalanceTime: 15 diff --git a/src/cli/commands/sweep.ts b/src/cli/commands/sweep.ts index 1295229e..00637397 100644 --- a/src/cli/commands/sweep.ts +++ b/src/cli/commands/sweep.ts @@ -106,6 +106,7 @@ export async function sweepFunds(opts: SweepOptions) { gasLimitMultiplier: 100, gasPriceMultiplier: 107, txTimeThreshold: 2_500, + blockTime: 5_000, timeout: 15_000, // unused fields but need to be defined diff --git a/src/cli/index.test.ts b/src/cli/index.test.ts index a0b55868..a620ff23 100644 --- a/src/cli/index.test.ts +++ b/src/cli/index.test.ts @@ -103,6 +103,7 @@ describe("Test RainSolverCli", () => { sleep: 1000, poolUpdateInterval: 60, gasCoveragePercentage: "100", + blockTime: 2000, } as any; mockState = { @@ -173,6 +174,7 @@ describe("Test RainSolverCli", () => { }, liquidityProviders: ["uniswap"], client: {}, + watchBlockNumber: vi.fn(), avgGasCost: 1000000000000000000n, gasCosts: [500000000000000000n, 1500000000000000000n], contracts, @@ -273,6 +275,7 @@ describe("Test RainSolverCli", () => { expect(AppOptions.tryFromYamlPath).toHaveBeenCalledWith("config.yaml"); expect(SharedStateConfig.tryFromAppOptions).toHaveBeenCalledWith(mockAppOptions); expect(SharedState).toHaveBeenCalledWith(mockStateConfig); + expect(mockState.watchBlockNumber).toHaveBeenCalledWith(mockAppOptions.blockTime); expect(SubgraphConfig.tryFromAppOptions).toHaveBeenCalledWith(mockAppOptions); expect(SubgraphManager).toHaveBeenCalledWith(mockSgManagerConfig); expect(mockSubgraphManager.statusCheck).toHaveBeenCalledTimes(1); @@ -465,6 +468,7 @@ describe("Test RainSolverCli", () => { sleep: 1000, poolUpdateInterval: 60, gasCoveragePercentage: "100", + blockTime: 2000, key: "N/A", mnemonic: "N/A", }), diff --git a/src/cli/index.ts b/src/cli/index.ts index 044776c6..20487b52 100644 --- a/src/cli/index.ts +++ b/src/cli/index.ts @@ -136,6 +136,11 @@ export class RainSolverCli { } const state = new SharedState(stateConfig.value); + // watch block number during runtime, so the current block number + // is always available in the state without repeated rpc calls, + // polled at the chain's block time cadence + state.watchBlockNumber(appOptions.blockTime); + report.setStatus({ code: SpanStatusCode.OK }); report.end(); logger.exportPreAssembledSpan(report); diff --git a/src/common/abis/orderbook.ts b/src/common/abis/orderbook.ts index 10460f08..8418dd18 100644 --- a/src/common/abis/orderbook.ts +++ b/src/common/abis/orderbook.ts @@ -39,6 +39,8 @@ namespace _v4 { `event TakeOrderV2(address sender, ${TakeOrderConfigV3} config, uint256 input, uint256 output)`, `function quote(${Quote} calldata quoteConfig) external view returns (bool, uint256, uint256)`, `event ClearV2(address sender, ${OrderV3} alice, ${OrderV3} bob, ${ClearConfig} clearConfig)`, + `event Deposit(address sender, address token, uint256 vaultId, uint256 amount)`, + `event Withdraw(address sender, address token, uint256 vaultId, uint256 targetAmount, uint256 amount)`, ] as const; export const Arb = [ `function arb2(${TakeOrdersConfigV3} calldata takeOrders, uint256 minimumSenderOutput, ${EvaluableV3} calldata evaluable) external payable`, diff --git a/src/config/validators.test.ts b/src/config/validators.test.ts index 68685aa0..6adc67a7 100644 --- a/src/config/validators.test.ts +++ b/src/config/validators.test.ts @@ -721,6 +721,35 @@ describe("Test yaml Validator methods", async function () { ); }); + it("test Validator resolveWsRpc", async function () { + // happy: wss and ws urls + assert.equal( + Validator.resolveWsRpc("wss://ws-rpc.example.com"), + "wss://ws-rpc.example.com", + ); + assert.equal(Validator.resolveWsRpc("ws://ws-rpc.example.com"), "ws://ws-rpc.example.com"); + + // happy: unset returns undefined + assert.equal(Validator.resolveWsRpc(undefined), undefined); + + // happy: env variable + process.env.TEST_WS_RPC = "wss://ws-rpc.example.com"; + assert.equal(Validator.resolveWsRpc("$TEST_WS_RPC"), "wss://ws-rpc.example.com"); + delete process.env.TEST_WS_RPC; + + // unhappy: non websocket url + assert.throws( + () => Validator.resolveWsRpc("https://rpc.example.com"), + /invalid wsRpc value, expected a websocket url starting with ws:\/\/ or wss:\/\//, + ); + + // unhappy: non string value + assert.throws( + () => Validator.resolveWsRpc(123), + /invalid wsRpc value, expected a websocket url starting with ws:\/\/ or wss:\/\//, + ); + }); + it("test Validator resolveAddressSet", async function () { // happy let input: any = [`0x${"1".repeat(40)}`, `0x${"2".repeat(40)}`]; diff --git a/src/config/validators.ts b/src/config/validators.ts index 1cdf45c2..a648f6d2 100644 --- a/src/config/validators.ts +++ b/src/config/validators.ts @@ -78,6 +78,19 @@ export namespace Validator { return Array.from(new Set(urls.value)) as any; } + /** Resolves config's optional websocket rpc url */ + export function resolveWsRpc(input: any): string | undefined { + const url = readValue(input); + if (url.value === undefined) return undefined; + assert( + typeof url.value === "string" && /^wss?:\/\//.test(url.value), + validationError( + "invalid wsRpc value, expected a websocket url starting with ws:// or wss://", + ), + ); + return url.value; + } + /** Resolves config's list of liquidity providers */ export function resolveLiquidityProviders(input: any) { const lps = readValue(input); diff --git a/src/config/yaml.test.ts b/src/config/yaml.test.ts index 5b7ef7d4..b312a841 100644 --- a/src/config/yaml.test.ts +++ b/src/config/yaml.test.ts @@ -17,6 +17,7 @@ walletCount: 10 topupAmount: 0.5 writeRpc: - url: http://write-rpc.example.com +wsRpc: wss://ws-rpc.example.com subgraph: ["http://subgraph.example.com"] contracts: v4: @@ -37,6 +38,7 @@ quoteGas: 2000000 botMinBalance: 50.5 gasPriceMultiplier: 150 txTimeThreshold: 4000 +blockTime: 3000 checkWalletBalanceTime: 30 gasBoostProfitThreshold: 7 gasBoostMultiplier: 3.5 @@ -96,6 +98,7 @@ orderbookTradeTypes: walletCount: 10, topupAmount: "0.5", writeRpc: [{ url: "http://write-rpc.example.com" }], + wsRpc: "wss://ws-rpc.example.com", subgraph: ["http://subgraph.example.com"], contracts: { v4: { @@ -116,6 +119,7 @@ orderbookTradeTypes: botMinBalance: "50.5", gasPriceMultiplier: 150, txTimeThreshold: 4000, + blockTime: 3000, gasLimitMultiplier: 90, timeout: 20000, maxRatio: true, @@ -294,6 +298,7 @@ orderbookTradeTypes: assert.deepEqual(result.botMinBalance, "50.5"); assert.deepEqual(result.gasPriceMultiplier, 150); assert.deepEqual(result.txTimeThreshold, 4000); + assert.deepEqual(result.blockTime, 5000); // should be default 5000 assert.deepEqual(result.gasLimitMultiplier, 90); assert.deepEqual(result.timeout, 20000); assert.equal(result.maxRatio, true); @@ -354,6 +359,7 @@ orderbookTradeTypes: assert.equal(result.convertToGasTime, 2); assert.equal(result.rotateMultiWallet, true); assert.equal(result.checkWalletBalanceTime, 15); // should be default 15 + assert.equal(result.wsRpc, undefined); // no ws rpc when unset assert.equal(result.gasBoostProfitThreshold, undefined); // no boost when unset assert.equal(result.gasBoostMultiplier, undefined); // no boost when unset assert.equal(result.gasBoostUsdThreshold, undefined); // no boost when unset diff --git a/src/config/yaml.ts b/src/config/yaml.ts index 490aceb2..80e033cb 100644 --- a/src/config/yaml.ts +++ b/src/config/yaml.ts @@ -69,6 +69,8 @@ export type AppOptions = { rpc: RpcConfig[]; /** List of write rpc configs used explicitly for write transactions */ writeRpc?: RpcConfig[]; + /** Optional websocket rpc url used explicitly for the block number watcher new heads subscription */ + wsRpc?: string; /** List of subgraph urls */ subgraph: string[]; /** Option to maximize maxIORatio, default is true */ @@ -119,6 +121,8 @@ export type AppOptions = { rotateMultiWallet: boolean; /** Time threshold (in ms) for a transaction mine time before it counts as a trigger to increase gas price multiplier for future transactions, default is 15 seconds */ txTimeThreshold: number; + /** The average block time (in ms) of the operating chain, used as the polling interval of the block number watcher, default is 5000 ms */ + blockTime: number; /** Time (in minutes) to to check the operating wallet balances, 0 means dont ever check wallet balance, default is 15 mins */ checkWalletBalanceTime: number; /** Optional threshold as the min expected bounty multiple that the estimated profit must exceed to boost the tx gas price, no boost applies if unset */ @@ -184,6 +188,7 @@ export namespace AppOptions { contracts: Validator.resolveContracts(input), rpc: Validator.resolveRpc(input.rpc), writeRpc: Validator.resolveRpc(input.writeRpc, true), + wsRpc: Validator.resolveWsRpc(input.wsRpc), subgraph: Validator.resolveUrls( input.subgraph, "expected array of subgraph urls with at least 1 url", @@ -362,6 +367,18 @@ export namespace AppOptions { "invalid txTimeThreshold value, must be an integer greater than 0", ), ), + blockTime: Validator.resolveNumericValue( + input.blockTime, + INT_PATTERN, + "invalid blockTime value, must be an integer greater than 0", + "5000", + undefined, + (blockTime) => + assert( + blockTime > 0, + "invalid blockTime value, must be an integer greater than 0", + ), + ), checkWalletBalanceTime: Validator.resolveNumericValue( input.checkWalletBalanceTime, INT_PATTERN, diff --git a/src/core/process/order.test.ts b/src/core/process/order.test.ts index 96b38ce3..86b41772 100644 --- a/src/core/process/order.test.ts +++ b/src/core/process/order.test.ts @@ -46,9 +46,8 @@ describe("Test processOrder", () => { id: 1, nativeWrappedToken: "0xWETH", }, - client: { - getBlockNumber: vi.fn().mockResolvedValue(123), - }, + blockNumber: 123n, + client: {}, router: { sushi: { update: vi.fn().mockResolvedValue(undefined), diff --git a/src/core/process/order.ts b/src/core/process/order.ts index 9dfab6d5..c768937f 100644 --- a/src/core/process/order.ts +++ b/src/core/process/order.ts @@ -3,7 +3,6 @@ import { Pair } from "../../order"; import { Result } from "../../common"; import { toUsdValue, toNumber } from "../../math"; import { Token } from "sushi/currency"; -import { errorSnapshot } from "../../error"; import { SpanWithContext } from "../../logger"; import { formatUnits, parseUnits } from "viem"; import { Attributes } from "@opentelemetry/api"; @@ -252,19 +251,9 @@ export async function processOrder( spanEvents["findBestTrade"] = { startTime: findBestTradeTime, duration: findBestTradeDuration }; // get block number - let blockNumber: number; - try { - blockNumber = Number(await this.state.client.getBlockNumber()); - spanAttributes["details.blockNumber"] = blockNumber; - spanAttributes["details.blockNumberDiff"] = blockNumber - oppBlockNumber; - } catch (e) { - // dont reject if getting block number fails but just record it, - // since an opp is found and can ultimately be cleared - spanAttributes["details.blockNumberError"] = await errorSnapshot( - "failed to get block number", - e, - ); - } + const blockNumber = Number(this.state.blockNumber); + spanAttributes["details.blockNumber"] = blockNumber; + spanAttributes["details.blockNumberDiff"] = blockNumber - oppBlockNumber; // process the found transaction opportunity return processTransaction.call(this, { diff --git a/src/core/process/round.test.ts b/src/core/process/round.test.ts index 6e4dac8f..4eb23610 100644 --- a/src/core/process/round.test.ts +++ b/src/core/process/round.test.ts @@ -65,10 +65,10 @@ describe("Test initializeRound", () => { // mock state mockState = { chainConfig: { id: 1 }, + blockNumber: 123n, updateGasTokenUsdPrice: vi.fn().mockResolvedValue(undefined), client: { name: "viem-client", - getBlockNumber: vi.fn().mockResolvedValue(123n), }, contracts: { getAddressesForTrade: vi.fn().mockReturnValue({ @@ -142,6 +142,41 @@ describe("Test initializeRound", () => { expect(checkpointReport.endTime).toBeTypeOf("number"); }); + it("should refresh block number from state at batch boundaries", async () => { + (mockSolver.appOptions as any).maxConcurrency = 1; + const makeOrder = (id: string) => ({ + orderbook: "0x3333333333333333333333333333333333333333", + buyTokenSymbol: "ETH", + sellTokenSymbol: "USDC", + sellToken: "0xsellToken", + buyToken: "0xbuyToken", + takeOrder: { id, struct: { order: { owner: "0xOwner123" } } }, + }); + (mockOrderManager.getNextRoundOrders as Mock).mockReturnValue({ + nonZeroOutput: [makeOrder("0xOrder1"), makeOrder("0xOrder2")], + zeroOutput: [], + }); + (mockWalletManager.getRandomSigner as Mock).mockResolvedValue(mockSigner); + const seenBlockNumbers: bigint[] = []; + (mockSolver.processOrder as Mock).mockImplementation(async (args: any) => { + seenBlockNumbers.push(args.blockNumber); + // simulate the watcher observing a new block during the batch + (mockState as any).blockNumber = args.blockNumber + 10n; + return vi.fn(); + }); + + const result: initializeRoundType = await initializeRound.call( + mockSolver, + undefined, + false, // no shuffle, keep the order sequence deterministic + ); + + expect(result.settlements).toHaveLength(2); + // the first batch uses the initial block number and the second + // batch uses the refreshed state value from the batch boundary + expect(seenBlockNumbers).toEqual([123n, 133n]); + }); + it("should handle multiple orders from multiple orderbooks", async () => { const mockOrders = [ { @@ -229,6 +264,38 @@ describe("Test initializeRound", () => { }); }); + describe("block number not available handling", () => { + it("should skip the round and report error when watched block number is not available yet", async () => { + (mockOrderManager.getNextRoundOrders as Mock).mockReturnValue({ + nonZeroOutput: [], + zeroOutput: [], + }); + mockState.blockNumber = 0n; + (mockSolver as any).logger = { + exportPreAssembledSpan: vi.fn(), + } as any; + + const result: initializeRoundType = await initializeRound.call(mockSolver); + + expect(result.settlements).toHaveLength(0); + expect(result.checkpointReports).toHaveLength(0); + expect(mockWalletManager.getRandomSigner).not.toHaveBeenCalled(); + expect(mockSolver.processOrder).not.toHaveBeenCalled(); + expect(mockSolver.logger?.exportPreAssembledSpan).toHaveBeenCalledTimes(1); + expect(mockSolver.logger?.exportPreAssembledSpan).toHaveBeenCalledWith( + expect.objectContaining({ + name: "order_batch_preprocess", + status: expect.objectContaining({ + message: "block number is not available yet for orders batch process", + }), + }), + undefined, + ); + + (mockSolver as any).logger = undefined; // reset logger + }); + }); + describe("method call verification", () => { it("should call getNextRoundOrders with correct parameter", async () => { (mockOrderManager.getNextRoundOrders as Mock).mockReturnValue({ @@ -1702,9 +1769,9 @@ describe("Test processOrderInit", () => { // mock state mockState = { + blockNumber: 123n, client: { name: "viem-client", - getBlockNumber: vi.fn().mockResolvedValue(123n), }, contracts: { getAddressesForTrade: vi.fn().mockReturnValue({ diff --git a/src/core/process/round.ts b/src/core/process/round.ts index e36bac4b..3601a3d6 100644 --- a/src/core/process/round.ts +++ b/src/core/process/round.ts @@ -36,18 +36,16 @@ export async function initializeRound( const settlements: Settlement[] = []; const checkpointReports: PreAssembledSpan[] = []; - let blockNumber: bigint; let concurrencyProcessBatch = []; let maxConcurrencyCounter = this.appOptions.maxConcurrency; - try { - blockNumber = await this.state.client.getBlockNumber(); - } catch (error) { - const message = await errorSnapshot( - "failed to get block number for orders batch process", - error, - ); + let blockNumber = this.state.blockNumber; + if (blockNumber <= 0n) { + // the block number watcher has not observed any block yet const report = new PreAssembledSpan(`order_batch_preprocess`); - report.setStatus({ code: SpanStatusCode.ERROR, message }); + report.setStatus({ + code: SpanStatusCode.ERROR, + message: "block number is not available yet for orders batch process", + }); this.logger?.exportPreAssembledSpan(report, roundSpanCtx?.context); return { settlements, @@ -86,8 +84,7 @@ export async function initializeRound( // reset counter and batch vector concurrencyProcessBatch = []; maxConcurrencyCounter = this.appOptions.maxConcurrency; - const temp = await this.state.client.getBlockNumber().catch(() => undefined); - if (typeof temp === "bigint") blockNumber = temp; + blockNumber = this.state.blockNumber; } } diff --git a/src/rpc/helpers.ts b/src/rpc/helpers.ts index 4a950d2d..c16310b0 100644 --- a/src/rpc/helpers.ts +++ b/src/rpc/helpers.ts @@ -97,6 +97,11 @@ export function normalizeUrl(url: string): string { return url.endsWith("/") ? url : `${url}/`; } +/** Checks if the given url is a websocket url */ +export function isWebSocketUrl(url: string): boolean { + return /^wss?:\/\//.test(url); +} + /** * Probably picks an item from the given array of success rates as probablity ranges * which are in 2 fixed point decimalss diff --git a/src/rpc/hooks.test.ts b/src/rpc/hooks.test.ts deleted file mode 100644 index 6a5fd38e..00000000 --- a/src/rpc/hooks.test.ts +++ /dev/null @@ -1,162 +0,0 @@ -import { RpcMetrics, RpcState } from "./rpc"; -import { normalizeUrl, shouldThrow } from "./helpers"; -import { onFetchRequest, onFetchResponse } from "./hooks"; -import { describe, it, expect, vi, beforeEach, Mock } from "vitest"; - -vi.mock("./helpers", () => ({ - normalizeUrl: vi.fn(), - shouldThrow: vi.fn(), -})); - -vi.mock("./rpc", () => ({ - RpcState: vi.fn(), - RpcMetrics: vi.fn(), -})); - -describe("Test RPC hooks", () => { - let mockRpcState: RpcState; - let mockRpcMetrics: any; - - beforeEach(() => { - vi.clearAllMocks(); - mockRpcMetrics = { - recordRequest: vi.fn(), - recordSuccess: vi.fn(), - recordFailure: vi.fn(), - }; - - (RpcMetrics as Mock).mockImplementation(() => mockRpcMetrics); - - mockRpcState = { - metrics: {}, - } as RpcState; - }); - - describe("Test onFetchRequest", () => { - it("should record request for new URL", () => { - const mockRequest = new Request("https://api.example.com/rpc"); - (normalizeUrl as Mock).mockReturnValue("https://api.example.com/rpc"); - - onFetchRequest.call(mockRpcState, mockRequest); - - expect(normalizeUrl).toHaveBeenCalledWith(mockRequest.url); - expect(RpcMetrics).toHaveBeenCalledOnce(); - expect(mockRpcMetrics.recordRequest).toHaveBeenCalledOnce(); - expect(mockRpcState.metrics["https://api.example.com/rpc"]).toBe(mockRpcMetrics); - }); - - it("should record request for existing URL", () => { - const mockRequest = new Request("https://api.example.com/rpc"); - const existingMetrics = { - recordRequest: vi.fn(), - recordSuccess: vi.fn(), - recordFailure: vi.fn(), - } as any; - - (normalizeUrl as Mock).mockReturnValue("https://api.example.com/rpc"); - mockRpcState.metrics["https://api.example.com/rpc"] = existingMetrics; - - onFetchRequest.call(mockRpcState, mockRequest); - - expect(existingMetrics.recordRequest).toHaveBeenCalledOnce(); - expect(RpcMetrics).not.toHaveBeenCalled(); - }); - }); - - describe("Test onFetchResponse", () => { - let mockResponse: Response; - - beforeEach(() => { - mockResponse = { - clone: vi.fn().mockReturnThis(), - url: "https://api.example.com/rpc", - ok: true, - headers: { - get: vi.fn(), - }, - json: vi.fn(), - text: vi.fn(), - } as any; - - (normalizeUrl as Mock).mockReturnValue("https://api.example.com/rpc"); - mockRpcState.metrics["https://api.example.com/rpc"] = mockRpcMetrics; - }); - - it("should record failure for non-ok response", async () => { - (mockResponse as any).ok = false; - - await onFetchResponse.call(mockRpcState, mockResponse); - - expect(mockRpcMetrics.recordFailure).toHaveBeenCalledOnce(); - }); - - it("should record success for JSON response with result", async () => { - (mockResponse.headers.get as Mock) = vi.fn().mockReturnValue("application/json"); - (mockResponse.json as Mock) = vi.fn().mockResolvedValue({ result: "success" }); - - await onFetchResponse.call(mockRpcState, mockResponse); - - expect(mockRpcMetrics.recordSuccess).toHaveBeenCalledOnce(); - }); - - it("should record success for JSON response with throwable error", async () => { - (mockResponse.headers.get as Mock) = vi.fn().mockReturnValue("application/json"); - (mockResponse.json as Mock) = vi.fn().mockResolvedValue({ error: { code: -32000 } }); - (shouldThrow as Mock).mockReturnValue(true); - - await onFetchResponse.call(mockRpcState, mockResponse); - - expect(shouldThrow).toHaveBeenCalledWith({ code: -32000 }); - expect(mockRpcMetrics.recordSuccess).toHaveBeenCalledOnce(); - }); - - it("should record failure for JSON response with non-throwable error", async () => { - (mockResponse.headers.get as Mock) = vi.fn().mockReturnValue("application/json"); - (mockResponse.json as Mock) = vi.fn().mockResolvedValue({ error: { code: -32001 } }); - (shouldThrow as Mock).mockReturnValue(false); - - await onFetchResponse.call(mockRpcState, mockResponse); - - expect(mockRpcMetrics.recordFailure).toHaveBeenCalledOnce(); - }); - - it("should record failure for invalid JSON response", async () => { - (mockResponse.headers.get as Mock) = vi.fn().mockReturnValue("application/json"); - (mockResponse.json as Mock) = vi.fn().mockRejectedValue(new Error("Invalid JSON")); - - await onFetchResponse.call(mockRpcState, mockResponse); - - expect(mockRpcMetrics.recordFailure).toHaveBeenCalledOnce(); - }); - - it("should handle text response with valid JSON", async () => { - (mockResponse.headers.get as Mock) = vi.fn().mockReturnValue("text/plain"); - (mockResponse.text as Mock) = vi.fn().mockResolvedValue('{"result": "success"}'); - - await onFetchResponse.call(mockRpcState, mockResponse); - - expect(mockRpcMetrics.recordSuccess).toHaveBeenCalledOnce(); - }); - - it("should record failure for text response with invalid JSON", async () => { - (mockResponse.headers.get as Mock) = vi.fn().mockReturnValue("text/plain"); - (mockResponse.text as Mock) = vi.fn().mockResolvedValue("invalid json"); - - await onFetchResponse.call(mockRpcState, mockResponse); - - expect(mockRpcMetrics.recordFailure).toHaveBeenCalledOnce(); - }); - - it("should create new metrics if not found", async () => { - delete mockRpcState.metrics["https://api.example.com/rpc"]; - (mockResponse.headers.get as Mock) = vi.fn().mockReturnValue("application/json"); - (mockResponse.json as Mock) = vi.fn().mockResolvedValue({ result: "success" }); - - await onFetchResponse.call(mockRpcState, mockResponse); - - expect(RpcMetrics).toHaveBeenCalledOnce(); - expect(mockRpcMetrics.recordRequest).toHaveBeenCalledOnce(); - expect(mockRpcMetrics.recordSuccess).toHaveBeenCalledOnce(); - }); - }); -}); diff --git a/src/rpc/hooks.ts b/src/rpc/hooks.ts deleted file mode 100644 index 7647e25b..00000000 --- a/src/rpc/hooks.ts +++ /dev/null @@ -1,73 +0,0 @@ -import { RpcMetrics, RpcState } from "./rpc"; -import { normalizeUrl, shouldThrow } from "./helpers"; - -/** - * A fetch request hook for the viem http client that is used to keeps track of rpc metrics - * @param request - The fetch request object - */ -export function onFetchRequest(this: RpcState, request: Request) { - const url = normalizeUrl(request.url); - let record = this.metrics[url]; - if (!record) { - record = this.metrics[url] = new RpcMetrics(); - } - record.recordRequest(); -} - -/** - * A fetch response hook for the viem http client that is used to keeps track of rpc metrics - * @param response - The fetch response object - */ -export async function onFetchResponse(this: RpcState, response: Response) { - const _response = response.clone(); - const url = normalizeUrl(_response.url); - let record = this.metrics[url]; - if (!record) { - // this cannot really happen, but just to be sure, - // initialize this rpc record if its not already - record = this.metrics[url] = new RpcMetrics(); - record.recordRequest(); - } - - if (!_response.ok) { - record.recordFailure(); - return; - } - - const handleResponse = (res: any) => { - if ("result" in res) { - record.recordSuccess(); - return; - } else if ("error" in res) { - if (shouldThrow(res.error)) { - record.recordSuccess(); - return; - } - } - record.recordFailure(); - }; - if (_response.headers.get("Content-Type")?.startsWith("application/json")) { - await _response - .json() - .then((res: any) => { - handleResponse(res); - }) - .catch(() => { - record.recordFailure(); - }); - } else { - await _response - .text() - .then((text) => { - try { - const res = JSON.parse(text || "{}"); - handleResponse(res); - } catch (err) { - record.recordFailure(); - } - }) - .catch(() => { - record.recordFailure(); - }); - } -} diff --git a/src/rpc/index.ts b/src/rpc/index.ts index 0464d173..af46f237 100644 --- a/src/rpc/index.ts +++ b/src/rpc/index.ts @@ -1,4 +1,3 @@ export * from "./rpc"; -export * from "./hooks"; export * from "./helpers"; export * from "./transport"; diff --git a/src/rpc/rpc.test.ts b/src/rpc/rpc.test.ts index 8f048963..3a0e5d66 100644 --- a/src/rpc/rpc.test.ts +++ b/src/rpc/rpc.test.ts @@ -1,7 +1,8 @@ import { sleep } from "../common"; import { RainSolverTransportTimeoutError } from "./transport"; +import { Transport, TimeoutError, ExecutionRevertedError } from "viem"; import { vi, describe, it, assert, Mock, beforeEach, expect } from "vitest"; -import { RpcState, RpcConfig, RpcMetrics, RpcProgress, RpcBufferType } from "./rpc"; +import { RpcState, RpcConfig, RpcMetrics, RpcProgress, RpcBufferType, withRpcMetrics } from "./rpc"; vi.mock("../common", async (importOriginal) => ({ ...(await importOriginal()), @@ -122,9 +123,132 @@ describe("Test RpcState", async function () { new RainSolverTransportTimeoutError(0), ); }); + + it("should return the picked transport atomically paired with its url", async function () { + const urls = configs.map((v) => v.url); + const state = new RpcState(configs); + + for (let i = 0; i < 10; i++) { + const { transport, url } = await state.nextRpc({ + pollingInterval: 0, + timeout: 600000, + }); + expect(urls).toContain(url); + expect(transport).toBe(state.transports[url]); + expect(url).toBe(state.lastUsedUrl); + } + }); + + it("should create the underlying transport based on the url scheme", async function () { + const state = new RpcState([ + { url: "wss://ws-example.com" }, + { url: "https://http-example.com" }, + ]); + + expect((state.transports["wss://ws-example.com/"]({}) as any).config.type).toBe( + "webSocket", + ); + expect((state.transports["https://http-example.com/"]({}) as any).config.type).toBe("http"); + }); +}); + +describe("Test withRpcMetrics", async function () { + let record: RpcMetrics; + const makeMockTransport = (request: Mock) => + (() => ({ + config: { type: "mock" }, + request, + value: undefined, + })) as any as Transport; + + beforeEach(() => { + record = new RpcMetrics(); + }); + + it("should pass args and options through to the underlying transport request", async function () { + const request = vi.fn().mockResolvedValue("0x1234"); + const wrapped = withRpcMetrics(makeMockTransport(request), record)({} as any); + + await wrapped.request({ method: "eth_call", params: ["0xdata"] }, { dedupe: true } as any); + + expect(request).toHaveBeenCalledWith( + { method: "eth_call", params: ["0xdata"] }, + { dedupe: true }, + ); + }); + + it("should record success for resolved request", async function () { + const request = vi.fn().mockResolvedValue("0x1234"); + const wrapped = withRpcMetrics(makeMockTransport(request), record)({} as any); + + const result = await wrapped.request({ method: "eth_blockNumber" }); + + expect(result).toBe("0x1234"); + expect(request).toHaveBeenCalledTimes(1); + assert.equal(record.req, 1); + assert.equal(record.success, 1); + assert.equal(record.failure, 0); + assert.equal(record.timeout, 0); + }); + + it("should record success for node level error rejection", async function () { + // execution reverted is a node level error, meaning the + // rpc itself responded fine to the request + const error = new ExecutionRevertedError({ + cause: new Error("execution reverted") as any, + }); + const request = vi.fn().mockRejectedValue(error); + const wrapped = withRpcMetrics(makeMockTransport(request), record)({} as any); + + await expect(wrapped.request({ method: "eth_call" })).rejects.toThrow(error); + + assert.equal(record.req, 1); + assert.equal(record.success, 1); + assert.equal(record.failure, 0); + assert.equal(record.timeout, 0); + }); + + it("should record neither success nor failure for timeout rejection", async function () { + const error = new TimeoutError({ + body: { method: "eth_blockNumber" }, + url: "https://example.com", + }); + const request = vi.fn().mockRejectedValue(error); + const wrapped = withRpcMetrics(makeMockTransport(request), record)({} as any); + + await expect(wrapped.request({ method: "eth_blockNumber" })).rejects.toThrow(error); + + assert.equal(record.req, 1); + assert.equal(record.success, 0); + assert.equal(record.failure, 0); + assert.equal(record.timeout, 1); // derived from req - (success + failure) + }); + + it("should record failure for any other rejection", async function () { + const error = new Error("connection refused"); + const request = vi.fn().mockRejectedValue(error); + const wrapped = withRpcMetrics(makeMockTransport(request), record)({} as any); + + await expect(wrapped.request({ method: "eth_blockNumber" })).rejects.toThrow(error); + + assert.equal(record.req, 1); + assert.equal(record.success, 0); + assert.equal(record.failure, 1); + assert.equal(record.timeout, 0); + }); }); describe("Test RpcMetrics", async function () { + beforeEach(() => { + // interval tests below depend on sleep having real delays + (sleep as Mock).mockImplementation( + (ms: number) => + new Promise((resolve) => { + setTimeout(() => resolve(""), ms); + }), + ); + }); + it("should init RpcMetrics", async function () { const result = new RpcMetrics(); assert.equal(result.req, 0); @@ -150,7 +274,9 @@ describe("Test RpcMetrics", async function () { assert.deepEqual(result.requestIntervals, []); assert.ok(result.lastRequestTimestamp > 0); assert.equal(result.timeout, 1); - assert.equal(result.avgRequestIntervals, 0); + // with no recorded intervals this equals Date.now() minus the last request + // timestamp, so allow a small delta for the time it takes to get here + assert.closeTo(result.avgRequestIntervals, 0, 50); // wait 2 seconds and then record another request await sleep(2000); diff --git a/src/rpc/rpc.ts b/src/rpc/rpc.ts index 350ef6f7..b0be0d76 100644 --- a/src/rpc/rpc.ts +++ b/src/rpc/rpc.ts @@ -1,18 +1,18 @@ +import { isTimeout } from "../error"; import { promiseTimeout, sleep } from "../common"; -import { onFetchRequest, onFetchResponse } from "./hooks"; -import { http, Transport, HttpTransportConfig } from "viem"; -import { normalizeUrl, probablyPicksFrom } from "./helpers"; import { RainSolverTransportTimeoutError } from "./transport"; +import { http, webSocket, Transport, HttpTransportConfig } from "viem"; +import { normalizeUrl, probablyPicksFrom, isWebSocketUrl, shouldThrow } from "./helpers"; /** The rpc configurations */ export type RpcConfig = { - /** The rpc url */ + /** The rpc url, either http(s) or ws(s) */ url: string; /** The number of latest requests to keep track of, default is 100 */ trackSize?: number; /** The selection weight for this rpc, default is 1 */ selectionWeight?: number; - /** Viem transport configuration */ + /** Viem transport configuration, batch and fetchOptions only apply to http rpcs */ transportConfig?: Pick; }; @@ -45,12 +45,20 @@ export class RpcState { this.metrics = {}; this.transports = {}; configs.forEach((conf, i) => { - this.metrics[this.urls[i]] = new RpcMetrics(conf); - this.transports[this.urls[i]] = http(conf.url, { - ...conf.transportConfig, - onFetchRequest: onFetchRequest.bind(this), - onFetchResponse: onFetchResponse.bind(this), - }); + const url = this.urls[i]; + const record = (this.metrics[url] = new RpcMetrics(conf)); + // pick the underlying viem transport by the url scheme and wrap + // it with the metrics recorder, so the rpc consumption details + // are tracked the same way regardless of the transport kind + const transport = isWebSocketUrl(conf.url) + ? webSocket(conf.url, { + key: conf.transportConfig?.key, + name: conf.transportConfig?.name, + keepAlive: true, + reconnect: true, + }) + : http(conf.url, conf.transportConfig); + this.transports[url] = withRpcMetrics(transport, record); }); } @@ -68,7 +76,9 @@ export class RpcState { } /** - * Get next rpc to use which is picked based on past performance + * Get next rpc to use which is picked based on past performance, returns + * the picked transport atomically paired with its url, since reading the + * shared lastUsedUrl after the call can race with concurrent callers */ async nextRpc({ timeout = 10_000, @@ -76,7 +86,7 @@ export class RpcState { }: { timeout?: number; pollingInterval?: number; - }): Promise { + }): Promise<{ transport: Transport; url: string }> { // rpcs selection rate, each rate determines the probability of selecting that // rpc which is just a percentage of that rpc's latest success rate in 2 fixed // point decimals relative to other rpcs sucess rates, so the bigger the rate, @@ -92,7 +102,8 @@ export class RpcState { await sleep(pollingInterval); } else { this.lastUsedRpcIndex = index; - return this.transports[this.urls[index]]; + const url = this.urls[index]; + return { transport: this.transports[url], url }; } } })(), @@ -102,6 +113,43 @@ export class RpcState { } } +/** + * Wraps the given viem transport with the given rpc metrics recorder, the + * request outcome determines the records the same way regardless of the + * underlying transport kind (http or websocket): + * - a resolved request records a success + * - a rejection with a node level error (shouldThrow) records a success, + * since the rpc itself responded fine and the call semantically failed + * - a rejection with a timeout error records neither, it is derived as a + * timeout from the count of requests without a success/failure record + * - any other rejection records a failure + * @param transport - The transport to wrap + * @param record - The rpc metrics recorder of the wrapped transport's rpc + */ +export function withRpcMetrics(transport: Transport, record: RpcMetrics): Transport { + return ((opts) => { + const instance = transport(opts); + return { + ...instance, + request: async (args: any, options?: any) => { + record.recordRequest(); + try { + const result = await instance.request(args, options); + record.recordSuccess(); + return result; + } catch (error: any) { + if (shouldThrow(error)) { + record.recordSuccess(); + } else if (!isTimeout(error)) { + record.recordFailure(); + } + throw error; + } + }, + }; + }) as Transport; +} + /** * Metrics of a rpc consumption details */ diff --git a/src/rpc/transport.test.ts b/src/rpc/transport.test.ts index 8c62cc95..0bbf7f54 100644 --- a/src/rpc/transport.test.ts +++ b/src/rpc/transport.test.ts @@ -1,7 +1,7 @@ import { randomInt } from "crypto"; import { getLocal } from "mockttp"; import { polygon } from "viem/chains"; -import { describe, it, assert, expect } from "vitest"; +import { describe, it, assert, expect, vi } from "vitest"; import { normalizeUrl, RpcConfig, RpcBufferType, RpcState } from "."; import { rainSolverTransport, @@ -78,6 +78,218 @@ describe("Test transport", async function () { await mockServer2.stop(); }); + it("should dedupe identical concurrent requests into a single wire request", async function () { + const mockServer = getLocal(); + await mockServer.start(9494); + + const state = new RpcState([{ url: mockServer.url }]); + // pin the rpc at 100% success rate so nextRpc always picks it on the + // first try without ever yielding to the macrotask queue, otherwise a + // caller could enter after the first response already settled and got + // knocked out of viem's dedupe cache, firing a second wire request + const dedupeRecord = state.metrics[normalizeUrl(mockServer.url)]; + dedupeRecord.progress.buffer = Array(100).fill(RpcBufferType.Success); + dedupeRecord.progress.success = 100; + const transport = rainSolverTransport(state, { + retryCount: 0, + timeout: 60_000, + pollingInterval: 0, + })({ chain: polygon }); + + const endpoint = await mockServer.forPost().thenSendJsonRpcResult(1234); + + // fire 5 identical requests concurrently + const results = await Promise.all( + Array.from({ length: 5 }, () => transport.request({ method: "eth_blockNumber" })), + ); + + // all callers get the result, but only one request hits the wire + expect(results).toEqual([1234, 1234, 1234, 1234, 1234]); + const seenRequests = await endpoint.getSeenRequests(); + expect(seenRequests.length).toBe(1); + + // metrics record all 5 logical requests + const record = state.metrics[normalizeUrl(mockServer.url)]; + expect(record.req).toBe(5); + expect(record.success).toBe(5); + + await mockServer.stop(); + }); + + it("should retry the same rpc when it has a good success rate", async function () { + const mockServer = getLocal(); + await mockServer.start(9595); + + const state = new RpcState([{ url: mockServer.url }]); + // set a healthy success rate (90%) so same rpc retries kick in + const record = state.metrics[normalizeUrl(mockServer.url)]; + record.progress.buffer = Array(100).fill(RpcBufferType.Success); + record.progress.success = 90; + + // fail twice, then succeed + await mockServer.forPost().times(2).thenSendJsonRpcError({ + code: -32000, + message: "ratelimit exceeded", + }); + const successEndpoint = await mockServer.forPost().thenSendJsonRpcResult(1234); + + const transport = rainSolverTransport(state, { + retryCount: 2, + retryCountNext: 1, + retryDelay: 10, + timeout: 60_000, + pollingInterval: 0, + })({ chain: polygon }); + + // succeeds via 2 same-rpc retries without consuming the next-rpc retry, + // with the next-rpc path alone (retryCountNext 1) only 2 attempts would + // be made and the request would have failed + const result = await transport.request({ method: "eth_blockNumber" }); + expect(result).toBe(1234); + expect((await successEndpoint.getSeenRequests()).length).toBe(1); + + await mockServer.stop(); + }); + + it("should not retry the same rpc when it has a poor success rate", async function () { + const mockServer = getLocal(); + await mockServer.start(9696); + + const state = new RpcState([{ url: mockServer.url }]); + // set a poor success rate (10%) so same rpc retries are skipped + const record = state.metrics[normalizeUrl(mockServer.url)]; + record.progress.buffer = Array(100).fill(RpcBufferType.Failure); + record.progress.success = 10; + + // always fail + await mockServer.forPost().thenSendJsonRpcError({ + code: -32000, + message: "ratelimit exceeded", + }); + + const transport = rainSolverTransport(state, { + retryCount: 2, + retryCountNext: 1, + retryDelay: 10, + timeout: 60_000, + pollingInterval: 0, + })({ chain: polygon }); + + await expect(transport.request({ method: "eth_blockNumber" })).rejects.toThrow(); + + // only the initial attempt: no same-rpc retries despite retryCount + // being 2 (poor success rate), and the next-rpc rotation bails out + // since it lands on the same rpc that just failed (only rpc there is) + expect(state.metrics[normalizeUrl(mockServer.url)].req).toBe(1); + + await mockServer.stop(); + }); + + it("should rotate to the next rpc when the picked one fails", async function () { + const mockServer1 = getLocal(); + const mockServer2 = getLocal(); + await mockServer1.start(9797); + await mockServer2.start(9898); + + const state = new RpcState([{ url: mockServer1.url }, { url: mockServer2.url }]); + // both rpcs start at 50% success rate, so each occupies the first half of + // its 10000 wide slot, pin the picks so rpc1 is picked first and rpc2 second + const randomSpy = vi + .spyOn(Math, "random") + .mockReturnValueOnce(0) // pick 1 -> rpc1 slot + .mockReturnValueOnce(0.55); // pick 11001 -> rpc2 slot + + await mockServer1.forPost().thenSendJsonRpcError({ + code: -32000, + message: "ratelimit exceeded", + }); + await mockServer2.forPost().thenSendJsonRpcResult(1234); + + const transport = rainSolverTransport(state, { + retryCount: 0, // no same-rpc retries, isolate the rotation path + retryCountNext: 1, + timeout: 60_000, + pollingInterval: 0, + })({ chain: polygon }); + + const result = await transport.request({ method: "eth_blockNumber" }); + expect(result).toBe(1234); + expect(state.metrics[normalizeUrl(mockServer1.url)].req).toBe(1); + expect(state.metrics[normalizeUrl(mockServer2.url)].req).toBe(1); + + randomSpy.mockRestore(); + await mockServer1.stop(); + await mockServer2.stop(); + }); + + it("should bail out of rotation when it lands on the rpc that just failed", async function () { + const mockServer = getLocal(); + await mockServer.start(9999); + + const state = new RpcState([{ url: mockServer.url }]); + // healthy rpc, so the same-rpc retry does kick in before rotation + const record = state.metrics[normalizeUrl(mockServer.url)]; + record.progress.buffer = Array(100).fill(RpcBufferType.Success); + record.progress.success = 100; + + // always fail + await mockServer.forPost().thenSendJsonRpcError({ + code: -32000, + message: "ratelimit exceeded", + }); + + const transport = rainSolverTransport(state, { + retryCount: 1, + retryCountNext: 1, + retryDelay: 10, + timeout: 60_000, + pollingInterval: 0, + })({ chain: polygon }); + + await expect(transport.request({ method: "eth_blockNumber" })).rejects.toThrow(); + + // initial attempt plus one same-rpc retry, then the rotation lands on + // the same rpc (only rpc there is) and bails out with the error + expect(record.req).toBe(2); + + await mockServer.stop(); + }); + + it("should not retry at all when the error is a node level error", async function () { + const mockServer = getLocal(); + await mockServer.start(10101); + + const state = new RpcState([{ url: mockServer.url }]); + // healthy rpc, retries would kick in if the error classification allowed + const record = state.metrics[normalizeUrl(mockServer.url)]; + record.progress.buffer = Array(100).fill(RpcBufferType.Success); + record.progress.success = 100; + + // node level error, the rpc itself responded fine + await mockServer.forPost().thenSendJsonRpcError({ + code: 3, + message: "execution reverted", + }); + + const transport = rainSolverTransport(state, { + retryCount: 2, + retryCountNext: 1, + retryDelay: 10, + timeout: 60_000, + pollingInterval: 0, + })({ chain: polygon }); + + await expect(transport.request({ method: "eth_call" })).rejects.toThrow(); + + // single attempt, no same-rpc retries and no rotation since node level + // errors are deterministic, and it counts as a success for the metrics + expect(record.req).toBe(1); + expect(record.success).toBe(1); + expect(record.failure).toBe(0); + + await mockServer.stop(); + }); + it("test RainSolver transport unhappy", async function () { // setup 2 rpc mock servers const mockServer1 = getLocal(); diff --git a/src/rpc/transport.ts b/src/rpc/transport.ts index d2cc6ee5..b0bd92b0 100644 --- a/src/rpc/transport.ts +++ b/src/rpc/transport.ts @@ -1,4 +1,5 @@ import { RpcState } from "./rpc"; +import { sleep } from "../common"; import { shouldThrow } from "./helpers"; import { BaseError, createTransport, Transport, TransportConfig } from "viem"; @@ -9,9 +10,9 @@ export namespace RainSolverTransportDefaults { export const DEDUPE = true as const; export const RETRY_COUNT = 1 as const; export const TIMEOUT = 10_000 as const; - export const RETRY_DELAY = 150 as const; + export const RETRY_DELAY = 50 as const; export const RETRY_COUNT_NEXT = 1 as const; - export const POLLING_INTERVAL = 100 as const; + export const POLLING_INTERVAL = 50 as const; export const POLLING_TIMEOUT = 10_000 as const; export const KEY = "RainSolverTransport" as const; export const NAME = "Rain Solver Transport" as const; @@ -31,7 +32,7 @@ export type RainSolverTransportConfig = { retryDelay?: TransportConfig["retryDelay"]; /** The polling timeout in milliseconds when no rpc becomes available, default: 10_000ms */ pollingTimeout?: number; - /** The polling interval (in ms) to check for next available rpc, default: 250ms */ + /** The polling interval (in ms) to check for next available rpc, default: 100ms */ pollingInterval?: number; /** The max number of times to retry with next rpc, default: 1 */ retryCountNext?: number; @@ -86,6 +87,13 @@ export function rainSolverTransport( retryCountNext = RainSolverTransportDefaults.RETRY_COUNT_NEXT, } = config; return (({ chain }) => { + // cached transport instance of each rpc url, an instance is created once + // on first use and reused for all future requests to that url, a stable + // instance keeps viem's dedupe id stable, so identical concurrent requests + // to the same rpc get deduplicated into a single wire request by viem + const instances: Record> = {}; + const getInstance = (transport: Transport, url: string) => + (instances[url] ??= transport({ chain, retryCount: 0 })); return createTransport({ key, name, @@ -94,28 +102,51 @@ export function rainSolverTransport( retryCount: 0, type: "RainSolverTransport", async request(args, options) { - const req = async (tryNextCount: number): Promise => { + const req = async ( + tryNextCount: number, + prevUrl?: string, + prevError?: any, + ): Promise => { + // transport comes atomically paired with its url, reading + // state.lastUsedUrl here instead could race with concurrent + // requests that already moved it by their own nextRpc calls + const { transport, url } = await state.nextRpc({ + timeout: pollingTimeout, + pollingInterval, + }); + // when the rotation lands on the same rpc that just failed, + // dont waste another attempt on it and bail out with its + // error, its recorded failure has already lowered its chance + // of selection, so this balances out over future requests + if (url === prevUrl) throw prevError; + const instance = getInstance(transport, url); + const attempt = async (retrySameCount: number): Promise => { + try { + return await instance.request(args, { + ...options, + dedupe, + }); + } catch (error: any) { + if (shouldThrow(error)) throw error; + // retry the same rpc as long as it keeps a success rate + // above 20% threshold, this replaces the viem transport + // inner retries which were cancelled by the same criteria + if ( + retrySameCount > 0 && + tryNextCount > 0 && + state.metrics[url].progress.successRate > 2000 + ) { + await sleep(retryDelay); + return attempt(retrySameCount - 1); + } + throw error; + } + }; try { - const transport = await state.nextRpc({ - timeout: pollingTimeout, - pollingInterval, - }); - // cancel inner transport retry when success rate is below 20% threshold - const resolvedRetryCount = - tryNextCount && - state.metrics[state.lastUsedUrl].progress.successRate > 2000 - ? retryCount - : 0; - return await transport({ - chain, - retryCount: resolvedRetryCount, - }).request(args, { - ...options, - dedupe, - }); + return await attempt(retryCount); } catch (error: any) { if (shouldThrow(error)) throw error; - if (tryNextCount) return req(tryNextCount - 1); + if (tryNextCount) return req(tryNextCount - 1, url, error); throw error; } }; diff --git a/src/state/index.test.ts b/src/state/index.test.ts index b4154ead..45e7b986 100644 --- a/src/state/index.test.ts +++ b/src/state/index.test.ts @@ -7,8 +7,8 @@ import { LiquidityProviders } from "sushi"; import { SolverContracts } from "./contracts"; import { RainSolverRouter } from "../router/router"; import { Result, TokenDetails } from "../common"; -import { describe, it, expect, vi, beforeEach, Mock, assert } from "vitest"; -import { SharedState, SharedStateConfig, SharedStateErrorType } from "."; +import { describe, it, expect, vi, beforeEach, afterEach, Mock, assert } from "vitest"; +import { SharedState, SharedStateConfig, SharedStateErrorType, WS_RESUBSCRIBE_DELAY } from "."; vi.mock("../gas", () => ({ GasManager: { @@ -315,6 +315,233 @@ describe("Test SharedState", () => { }); }); + describe("Test block number watcher", () => { + beforeEach(() => { + vi.useFakeTimers(); + config.client.getBlockNumber = vi.fn().mockResolvedValue(100n); + sharedState = new SharedState(config); + }); + + afterEach(() => { + sharedState.unwatchBlockNumber(); + vi.useRealTimers(); + }); + + it("should update block number from rpc", async () => { + expect(sharedState.blockNumber).toBe(0n); + await sharedState.updateBlockNumber(); + expect(sharedState.blockNumber).toBe(100n); + }); + + it("should keep previous block number when the call fails", async () => { + await sharedState.updateBlockNumber(); + expect(sharedState.blockNumber).toBe(100n); + + (config.client.getBlockNumber as Mock).mockRejectedValue(new Error("rpc failed")); + await sharedState.updateBlockNumber(); + expect(sharedState.blockNumber).toBe(100n); + }); + + it("should not move block number backwards", async () => { + await sharedState.updateBlockNumber(); + expect(sharedState.blockNumber).toBe(100n); + + (config.client.getBlockNumber as Mock).mockResolvedValue(90n); + await sharedState.updateBlockNumber(); + expect(sharedState.blockNumber).toBe(100n); + + (config.client.getBlockNumber as Mock).mockResolvedValue(110n); + await sharedState.updateBlockNumber(); + expect(sharedState.blockNumber).toBe(110n); + }); + + it("should watch block number with an immediate update and periodic updates", async () => { + expect(sharedState.isWatchingBlockNumber).toBe(false); + sharedState.watchBlockNumber(5000); + expect(sharedState.isWatchingBlockNumber).toBe(true); + + // immediate update on start + await vi.advanceTimersByTimeAsync(0); + expect(config.client.getBlockNumber).toHaveBeenCalledTimes(1); + expect(sharedState.blockNumber).toBe(100n); + + // periodic updates + (config.client.getBlockNumber as Mock).mockResolvedValue(101n); + await vi.advanceTimersByTimeAsync(5000); + expect(config.client.getBlockNumber).toHaveBeenCalledTimes(2); + expect(sharedState.blockNumber).toBe(101n); + + // should not start a second watcher + sharedState.watchBlockNumber(5000); + await vi.advanceTimersByTimeAsync(5000); + expect(config.client.getBlockNumber).toHaveBeenCalledTimes(3); + }); + + it("should unwatch block number", async () => { + sharedState.watchBlockNumber(5000); + await vi.advanceTimersByTimeAsync(0); + expect(sharedState.isWatchingBlockNumber).toBe(true); + + sharedState.unwatchBlockNumber(); + expect(sharedState.isWatchingBlockNumber).toBe(false); + + // no more updates after unwatch + await vi.advanceTimersByTimeAsync(15000); + expect(config.client.getBlockNumber).toHaveBeenCalledTimes(1); + }); + + describe("websocket new heads subscription", () => { + let onBlockNumber: (blockNumber: bigint) => void; + let onError: (error: Error) => void; + let wsUnwatch: Mock; + let watchBlockNumberSpy: Mock; + + beforeEach(() => { + config.appOptions.wsRpc = "wss://ws-rpc.example.com"; + wsUnwatch = vi.fn(); + watchBlockNumberSpy = vi.fn().mockImplementation((args: any) => { + onBlockNumber = args.onBlockNumber; + onError = args.onError; + return wsUnwatch; + }); + (createPublicClient as Mock).mockReturnValue({ + watchBlockNumber: watchBlockNumberSpy, + }); + sharedState = new SharedState(config); + }); + + it("should subscribe to new heads and update block number", async () => { + sharedState.watchBlockNumber(5000); + expect(sharedState.isWatchingBlockNumber).toBe(true); + expect(watchBlockNumberSpy).toHaveBeenCalledTimes(1); + + // immediate update over http on start + await vi.advanceTimersByTimeAsync(0); + expect(config.client.getBlockNumber).toHaveBeenCalledTimes(1); + expect(sharedState.blockNumber).toBe(100n); + + // new heads push updates the block number + onBlockNumber(105n); + expect(sharedState.blockNumber).toBe(105n); + + // should not move backwards + onBlockNumber(101n); + expect(sharedState.blockNumber).toBe(105n); + + // no polling should be active + await vi.advanceTimersByTimeAsync(15000); + expect(config.client.getBlockNumber).toHaveBeenCalledTimes(1); + }); + + it("should fall back to polling on subscription error and stop it on recovery", async () => { + sharedState.watchBlockNumber(5000); + await vi.advanceTimersByTimeAsync(0); + expect(config.client.getBlockNumber).toHaveBeenCalledTimes(1); + + // subscription error starts the polling fallback + onError(new Error("ws failed")); + (config.client.getBlockNumber as Mock).mockResolvedValue(101n); + await vi.advanceTimersByTimeAsync(5000); + expect(config.client.getBlockNumber).toHaveBeenCalledTimes(2); + expect(sharedState.blockNumber).toBe(101n); + + // subscription recovery stops the polling fallback + onBlockNumber(102n); + expect(sharedState.blockNumber).toBe(102n); + await vi.advanceTimersByTimeAsync(15000); + expect(config.client.getBlockNumber).toHaveBeenCalledTimes(2); + }); + + it("should not start a second subscription when already watching", async () => { + sharedState.watchBlockNumber(5000); + expect(watchBlockNumberSpy).toHaveBeenCalledTimes(1); + + sharedState.watchBlockNumber(5000); + expect(watchBlockNumberSpy).toHaveBeenCalledTimes(1); + }); + + it("should not start a second polling fallback on repeated subscription errors", async () => { + sharedState.watchBlockNumber(5000); + await vi.advanceTimersByTimeAsync(0); + expect(config.client.getBlockNumber).toHaveBeenCalledTimes(1); + + // repeated errors should keep a single poller + onError(new Error("ws failed")); + onError(new Error("ws failed again")); + await vi.advanceTimersByTimeAsync(5000); + expect(config.client.getBlockNumber).toHaveBeenCalledTimes(2); + }); + + it("should re-establish the subscription after the resubscribe delay", async () => { + sharedState.watchBlockNumber(5000); + await vi.advanceTimersByTimeAsync(0); + expect(watchBlockNumberSpy).toHaveBeenCalledTimes(1); + + // subscription error schedules a fresh subscription + onError(new Error("ws failed")); + await vi.advanceTimersByTimeAsync(WS_RESUBSCRIBE_DELAY); + expect(wsUnwatch).toHaveBeenCalledTimes(1); // old subscription is unwatched + expect(watchBlockNumberSpy).toHaveBeenCalledTimes(2); // fresh subscription + + // the fresh subscription pushes and stops the polling fallback + onBlockNumber(200n); + expect(sharedState.blockNumber).toBe(200n); + const callCount = (config.client.getBlockNumber as Mock).mock.calls.length; + await vi.advanceTimersByTimeAsync(15000); + expect(config.client.getBlockNumber).toHaveBeenCalledTimes(callCount); + }); + + it("should schedule a single resubscribe for repeated subscription errors", async () => { + sharedState.watchBlockNumber(5000); + await vi.advanceTimersByTimeAsync(0); + + onError(new Error("ws failed")); + onError(new Error("ws failed again")); + await vi.advanceTimersByTimeAsync(WS_RESUBSCRIBE_DELAY); + expect(watchBlockNumberSpy).toHaveBeenCalledTimes(2); + }); + + it("should cancel the pending resubscribe when the subscription recovers", async () => { + sharedState.watchBlockNumber(5000); + await vi.advanceTimersByTimeAsync(0); + + onError(new Error("ws failed")); + onBlockNumber(101n); // recovery before the resubscribe delay passes + + await vi.advanceTimersByTimeAsync(WS_RESUBSCRIBE_DELAY); + expect(wsUnwatch).not.toHaveBeenCalled(); + expect(watchBlockNumberSpy).toHaveBeenCalledTimes(1); + }); + + it("should cancel the pending resubscribe on unwatch", async () => { + sharedState.watchBlockNumber(5000); + await vi.advanceTimersByTimeAsync(0); + + onError(new Error("ws failed")); + sharedState.unwatchBlockNumber(); + + await vi.advanceTimersByTimeAsync(WS_RESUBSCRIBE_DELAY); + expect(watchBlockNumberSpy).toHaveBeenCalledTimes(1); + }); + + it("should unwatch the subscription and the polling fallback", async () => { + sharedState.watchBlockNumber(5000); + onError(new Error("ws failed")); // start polling fallback too + expect(sharedState.isWatchingBlockNumber).toBe(true); + + sharedState.unwatchBlockNumber(); + expect(wsUnwatch).toHaveBeenCalledTimes(1); + expect(sharedState.isWatchingBlockNumber).toBe(false); + + // no more polling after unwatch + await vi.advanceTimersByTimeAsync(0); + expect(config.client.getBlockNumber).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(15000); + expect(config.client.getBlockNumber).toHaveBeenCalledTimes(1); + }); + }); + }); + describe("Test avgGasCost", () => { it("should return 0 when gasCosts array is empty", () => { const state = new SharedState(config); diff --git a/src/state/index.ts b/src/state/index.ts index 8c048932..f539b76e 100644 --- a/src/state/index.ts +++ b/src/state/index.ts @@ -17,7 +17,16 @@ import { OrderManagerConfig } from "../order/config"; import { RainSolverRouterError } from "../router/error"; import { ChainConfig, ChainConfigError, getChainConfig } from "./chain"; import { RpcState, rainSolverTransport, RainSolverTransportConfig } from "../rpc"; -import { createPublicClient, parseUnits, PublicClient, ReadContractErrorType } from "viem"; +import { + webSocket, + parseUnits, + PublicClient, + createPublicClient, + ReadContractErrorType, +} from "viem"; + +/** Delay (in ms) before an errored ws new heads subscription is re-established */ +export const WS_RESUBSCRIBE_DELAY = 15_000; /** Enumerates the possible error types that can occur within the chain config */ export enum SharedStateErrorType { @@ -230,6 +239,12 @@ export class SharedState { oracleHealth: OracleHealthMap = new Map(); /** The current native gas token to USD price (18 decimals fixed point number as decimal string), updated once per round */ gasTokenUsdPrice?: string; + /** The latest observed block number of the operating chain, kept up-to-date by the block number watcher */ + blockNumber = 0n; + + private blockNumberWatcher: ReturnType | undefined; + private wsBlockNumberUnwatcher: (() => void) | undefined; + private wsResubscribeTimer: ReturnType | undefined; constructor(config: SharedStateConfig) { this.appOptions = config.appOptions; @@ -301,6 +316,110 @@ export class SharedState { this.gasManager.unwatchGasPrice(); } + /** Whether the block number watcher is active */ + get isWatchingBlockNumber(): boolean { + return this.blockNumberWatcher !== undefined || this.wsBlockNumberUnwatcher !== undefined; + } + + /** + * Updates the block number by reading it from the rpc once, + * keeps the previous value if the call fails + */ + async updateBlockNumber() { + const blockNumber = await this.client.getBlockNumber().catch(() => undefined); + if (typeof blockNumber === "bigint" && blockNumber > this.blockNumber) { + this.blockNumber = blockNumber; + } + } + + /** + * Watches the chain's block number during runtime, an immediate update is + * fired on start so the value becomes available asap, if a websocket rpc is + * configured, the block number is kept up-to-date by a new heads subscription + * for the earliest possible updates, with polling acting as a fallback while + * the subscription errors, otherwise it is polled periodically over http + * @param interval - Interval to poll block number in milliseconds, default is 5 seconds + */ + watchBlockNumber(interval = 5_000) { + if (this.isWatchingBlockNumber) return; + this.updateBlockNumber(); + if (this.appOptions.wsRpc) { + const wsClient = createPublicClient({ + chain: this.chainConfig, + transport: webSocket(this.appOptions.wsRpc, { + keepAlive: true, + reconnect: true, + }), + }); + this.subscribeToBlockNumber(wsClient, interval); + } else { + this.startPollingBlockNumber(interval); + } + } + + /** Unwatches block number if the watcher has been already active */ + unwatchBlockNumber() { + this.stopPollingBlockNumber(); + this.clearWsResubscribeTimer(); + this.wsBlockNumberUnwatcher?.(); + this.wsBlockNumberUnwatcher = undefined; + } + + /** + * Establishes the ws new heads subscription for block number updates, + * an errored subscription degrades to polling and schedules a fresh + * subscription after a delay, since viem reconnects the dropped socket + * but does not replay its subscriptions, when the ws endpoint is still + * down, the fresh subscription errors again and re-enters this path, + * which forms a retry loop paced by the delay, with polling covering + * the block number updates for the whole outage + */ + private subscribeToBlockNumber(wsClient: PublicClient, interval: number) { + this.wsBlockNumberUnwatcher = wsClient.watchBlockNumber({ + onBlockNumber: (blockNumber) => { + if (blockNumber > this.blockNumber) { + this.blockNumber = blockNumber; + } + // subscription is healthy, so stop the polling fallback + // and cancel any pending resubscribe if active + this.stopPollingBlockNumber(); + this.clearWsResubscribeTimer(); + }, + onError: () => { + this.startPollingBlockNumber(interval); + if (this.wsResubscribeTimer === undefined) { + this.wsResubscribeTimer = setTimeout(() => { + this.wsResubscribeTimer = undefined; + this.wsBlockNumberUnwatcher?.(); + this.subscribeToBlockNumber(wsClient, interval); + }, WS_RESUBSCRIBE_DELAY); + } + }, + }); + } + + /** Cancels the pending ws resubscribe if there is one scheduled */ + private clearWsResubscribeTimer() { + if (this.wsResubscribeTimer !== undefined) { + clearTimeout(this.wsResubscribeTimer); + this.wsResubscribeTimer = undefined; + } + } + + /** Starts polling block number periodically, no-op if already polling */ + private startPollingBlockNumber(interval: number) { + if (this.blockNumberWatcher !== undefined) return; + this.blockNumberWatcher = setInterval(() => this.updateBlockNumber(), interval); + } + + /** Stops polling block number if the poller is active */ + private stopPollingBlockNumber() { + if (this.blockNumberWatcher !== undefined) { + clearInterval(this.blockNumberWatcher); + this.blockNumberWatcher = undefined; + } + } + /** Watches the given token by putting on the watchedToken map */ watchToken(tokenDetails: TokenDetails) { if (!this.watchedTokens.has(tokenDetails.address.toLowerCase())) { diff --git a/test/e2e/e2e.test.js b/test/e2e/e2e.test.js index 6ef15d57..23b50434 100644 --- a/test/e2e/e2e.test.js +++ b/test/e2e/e2e.test.js @@ -171,6 +171,7 @@ for (let i = 0; i < testData.length; i++) { config.rpc = [rpc]; const viemClient = await viem.getPublicClient(); state.client = viemClient; + await state.updateBlockNumber(); const sushiRouterResult = await sushiRouterPromise; assert(sushiRouterResult.isOk()); state.router = new RainSolverRouter( @@ -466,6 +467,7 @@ for (let i = 0; i < testData.length; i++) { config.rpc = [rpc]; const viemClient = await viem.getPublicClient(); state.client = viemClient; + await state.updateBlockNumber(); const sushiRouterResult = await sushiRouterPromise; assert(sushiRouterResult.isOk()); state.router = new RainSolverRouter( @@ -855,6 +857,7 @@ for (let i = 0; i < testData.length; i++) { config.rpc = [rpc]; const viemClient = await viem.getPublicClient(); state.client = viemClient; + await state.updateBlockNumber(); const sushiRouterResult = await sushiRouterPromise; assert(sushiRouterResult.isOk()); state.router = new RainSolverRouter( @@ -1661,6 +1664,7 @@ for (let i = 0; i < testData.length; i++) { config.rpc = [rpc]; const viemClient = await viem.getPublicClient(); state.client = viemClient; + await state.updateBlockNumber(); state.client.simulateContract = client.simulateContract; const sushiRouterResult = await sushiRouterPromise; assert(sushiRouterResult.isOk()); @@ -1995,6 +1999,7 @@ for (let i = 0; i < testData.length; i++) { config.rpc = [rpc]; const viemClient = await viem.getPublicClient(); state.client = viemClient; + await state.updateBlockNumber(); const sushiRouterResult = await sushiRouterPromise; assert(sushiRouterResult.isOk()); state.router = new RainSolverRouter( @@ -2312,6 +2317,7 @@ for (let i = 0; i < testData.length; i++) { config.rpc = [rpc]; const viemClient = await viem.getPublicClient(); state.client = viemClient; + await state.updateBlockNumber(); const sushiRouterResult = await sushiRouterPromise; assert(sushiRouterResult.isOk()); state.router = new RainSolverRouter( @@ -2715,6 +2721,7 @@ for (let i = 0; i < testData.length; i++) { config.rpc = [rpc]; const viemClient = await viem.getPublicClient(); state.client = viemClient; + await state.updateBlockNumber(); const sushiRouterResult = await sushiRouterPromise; assert(sushiRouterResult.isOk()); state.router = new RainSolverRouter( @@ -3124,6 +3131,7 @@ for (let i = 0; i < testData.length; i++) { config.rpc = [rpc]; const viemClient = await viem.getPublicClient(); state.client = viemClient; + await state.updateBlockNumber(); const sushiRouterResult = await sushiRouterPromise; assert(sushiRouterResult.isOk()); state.router = new RainSolverRouter( @@ -3548,6 +3556,7 @@ for (let i = 0; i < testData.length; i++) { config.rpc = [rpc]; const viemClient = await viem.getPublicClient(); state.client = viemClient; + await state.updateBlockNumber(); state.client.simulateContract = client.simulateContract; const sushiRouterResult = await sushiRouterPromise; assert(sushiRouterResult.isOk()); @@ -3902,6 +3911,7 @@ for (let i = 0; i < testData.length; i++) { config.rpc = [rpc]; const viemClient = await viem.getPublicClient(); state.client = viemClient; + await state.updateBlockNumber(); state.client.simulateContract = client.simulateContract; const sushiRouterResult = await sushiRouterPromise; assert(sushiRouterResult.isOk()); From 18b377d7c301c8304f8a4e93e304e54174f1c69c Mon Sep 17 00:00:00 2001 From: rouzwelt Date: Fri, 4 Sep 2026 23:08:38 +0000 Subject: [PATCH 2/2] Update data.js --- test/e2e/data.js | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/test/e2e/data.js b/test/e2e/data.js index 53808d9f..732432a8 100644 --- a/test/e2e/data.js +++ b/test/e2e/data.js @@ -117,12 +117,12 @@ module.exports = [ // decimals: 18, // symbol: "WLTH", // }), - new Token({ - chainId: ChainId.BASE, - address: "0x71DDE9436305D2085331AF4737ec6f1fe876Cf9f", - decimals: 18, - symbol: "PAID", - }), + // new Token({ + // chainId: ChainId.BASE, + // address: "0x71DDE9436305D2085331AF4737ec6f1fe876Cf9f", + // decimals: 18, + // symbol: "PAID", + // }), new Token({ chainId: ChainId.BASE, address: "0x3982E57fF1b193Ca8eb03D16Db268Bd4B40818f8", @@ -134,7 +134,7 @@ module.exports = [ [ "0xb2cc224c1c9feE385f8ad6a55b4d94E92359DC59", // "0xe3715B2a3bB826cd9EC5429eE85B651f95879D34", - "0x4617C0F3e55930fdD72ec6EA92e79D384987C464", + // "0x4617C0F3e55930fdD72ec6EA92e79D384987C464", "0x7731D522011b4ACE5D812C15539321F373d0E964", "0x8da91A6298eA5d1A8Bc985e99798fd0A0f05701a", ], @@ -144,7 +144,7 @@ module.exports = [ LiquidityProviders.BaseSwap, LiquidityProviders.AerodromeSlipstream, ], - ["1", "10000", "10000", "100"], + ["1", "10000", "100"], ], // [ // // unique test for aerodrome slipstream