websocker transport

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

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
 

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

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",
                 }),

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);

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`,

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)}`];

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);

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

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,

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),

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, {

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({