feat: paginated logs + batching client

Author: doomsower

src/config/config.ts

@@ -5,10 +5,20 @@ import { createClassFromType, detectNetwork } from "../utils/index.js";
 import { envConfig } from "./env.js";
 import { ConfigSchema } from "./schema.js";
 
+// These limits work for DRPC and Alchemy
+const PAGE_SIZE: Record<NetworkType, bigint> = {
+  Mainnet: 100_000n,
+  Optimism: 500_000n,
+  Arbitrum: 500_000n,
+  Base: 500_000n,
+  Sonic: 500_000n,
+};
+
 interface DynamicConfig {
   readonly network: NetworkType;
   readonly chainId: number;
   readonly startBlock: bigint;
+  readonly logsPageSize: bigint;
 }
 
 const ConfigClass = createClassFromType<ConfigSchema & DynamicConfig>();
@@ -32,6 +42,7 @@ export class Config extends ConfigClass {
       startBlock,
       chainId: Number(chainId),
       network,
+      logsPageSize: PAGE_SIZE[network],
     });
   }
 

src/services/AddressProviderService.ts

@@ -7,7 +7,7 @@ import type { Config } from "../config/index.js";
 import { DI } from "../di.js";
 import { type ILogger, Logger } from "../log/index.js";
 import { TxParser } from "../utils/ethers-6-temp/txparser/index.js";
-import { getLogsSafe } from "../utils/getLogsSafe.js";
+import { getLogsPaginated } from "../utils/getLogsPaginated.js";
 import type Client from "./Client.js";
 
 const AP_SERVICES = [
@@ -52,13 +52,14 @@ export class AddressProviderService {
 
     const toBlock = await this.client.pub.getBlockNumber();
 
-    const logs = await getLogsSafe(this.client.pub, {
+    const logs = await getLogsPaginated(this.client.logs, {
       address,
       event: getAbiItem({ abi: iAddressProviderV3Abi, name: "SetAddress" }),
       args: { key: AP_SERVICES.map(s => stringToHex(s, { size: 32 })) },
       fromBlock: AP_BLOCK_BY_NETWORK[this.config.network],
       toBlock,
       strict: true,
+      pageSize: this.config.logsPageSize,
     });
 
     for (const { args } of logs) {

src/services/Client.ts

@@ -100,6 +100,7 @@ export default class Client {
   #anvilInfo: AnvilNodeInfo | null = null;
 
   #publicClient?: PublicClient;
+  #logsClient?: PublicClient;
 
   #walletClient?: WalletClient<Transport, Chain, PrivateKeyAccount, undefined>;
 
@@ -113,15 +114,7 @@ export default class Client {
   >;
 
   public async launch(): Promise<void> {
-    const { ethProviderRpcs, chainId, network, optimistic, privateKey } =
-      this.config;
-    const rpcs = ethProviderRpcs.map(url =>
-      http(url, {
-        timeout: optimistic ? 240_000 : 10_000,
-        retryCount: optimistic ? 3 : undefined,
-      }),
-    );
-    const transport = rpcs.length > 1 && !optimistic ? fallback(rpcs) : rpcs[0];
+    const { chainId, network, optimistic, privateKey } = this.config;
     const chain = defineChain({
       ...CHAINS[network],
       id: chainId,
@@ -130,13 +123,19 @@ export default class Client {
     this.#publicClient = createPublicClient({
       cacheTime: 0,
       chain,
-      transport,
+      transport: this.#createTransport(),
+      pollingInterval: optimistic ? 25 : undefined,
+    });
+    this.#logsClient = createPublicClient({
+      cacheTime: 0,
+      chain,
+      transport: this.#createTransport(true),
       pollingInterval: optimistic ? 25 : undefined,
     });
     this.#walletClient = createWalletClient({
       account: privateKeyToAccount(privateKey),
       chain,
-      transport,
+      transport: this.#createTransport(),
       pollingInterval: optimistic ? 25 : undefined,
     });
     try {
@@ -148,7 +147,7 @@ export default class Client {
         AnvilRPCSchema
       >({
         mode: "anvil",
-        transport,
+        transport: this.#createTransport(),
         chain,
         pollingInterval: 25,
       });
@@ -310,6 +309,13 @@ export default class Client {
     return this.#publicClient;
   }
 
+  public get logs(): PublicClient {
+    if (!this.#logsClient) {
+      throw new Error("logs client not initialized");
+    }
+    return this.#logsClient;
+  }
+
   public get wallet(): WalletClient<
     Transport,
     Chain,
@@ -354,4 +360,16 @@ export default class Client {
     }
     return BigInt(n);
   }
+
+  #createTransport(batch = false): Transport {
+    const { ethProviderRpcs, optimistic } = this.config;
+    const rpcs = ethProviderRpcs.map(url =>
+      http(url, {
+        timeout: optimistic ? 240_000 : 10_000,
+        retryCount: optimistic ? 3 : undefined,
+        batch,
+      }),
+    );
+    return rpcs.length > 1 && !optimistic ? fallback(rpcs) : rpcs[0];
+  }
 }

src/services/OracleServiceV3.ts

@@ -14,7 +14,7 @@ import { DI } from "../di.js";
 import type { ILogger } from "../log/index.js";
 import { Logger } from "../log/index.js";
 import { TxParser } from "../utils/ethers-6-temp/txparser/index.js";
-import { getLogsSafe } from "../utils/getLogsSafe.js";
+import { getLogsPaginated } from "../utils/getLogsPaginated.js";
 import type { AddressProviderService } from "./AddressProviderService.js";
 import type Client from "./Client.js";
 
@@ -54,7 +54,7 @@ interface OracleEntry {
 
 const ORACLE_START_BLOCK: Record<NetworkType, bigint> = {
   Mainnet: 18797638n,
-  Optimism: 116864678n, // not deployed yet, arbitrary block here
+  Optimism: 118413958n,
   Arbitrum: 184650373n,
   Base: 12299805n, // not deployed yet, arbitrary block here
   Sonic: 8897028n, // not deployed yet, arbitrary block here
@@ -192,12 +192,13 @@ export default class OracleServiceV3 {
       return;
     }
     this.log.debug(`updating price feeds in [${this.#lastBlock}, ${toBlock}]`);
-    let logs = await getLogsSafe(this.client.pub, {
+    let logs = await getLogsPaginated(this.client.logs, {
       address: this.oracle,
       events: iPriceOracleV3EventsAbi,
       fromBlock: BigInt(this.#lastBlock),
       toBlock: BigInt(toBlock),
       strict: true,
+      pageSize: this.config.logsPageSize,
     });
     // sort logs by blockNumber ASC, logIndex ASC
     // on sonic sometimes events are not in order

src/utils/getLogsPaginated.ts

@@ -0,0 +1,85 @@
+import type { AbiEvent } from "abitype";
+import type {
+  BlockNumber,
+  Chain,
+  Client,
+  GetLogsParameters,
+  GetLogsReturnType,
+  MaybeAbiEventName,
+  Transport,
+} from "viem";
+import { getLogs } from "viem/actions";
+
+export type GetLogsPaginatedParameters<
+  abiEvent extends AbiEvent | undefined = undefined,
+  abiEvents extends
+    | readonly AbiEvent[]
+    | readonly unknown[]
+    | undefined = abiEvent extends AbiEvent ? [abiEvent] : undefined,
+  strict extends boolean | undefined = undefined,
+  //
+  _eventName extends string | undefined = MaybeAbiEventName<abiEvent>,
+> = GetLogsParameters<
+  abiEvent,
+  abiEvents,
+  strict,
+  BlockNumber,
+  BlockNumber,
+  _eventName
+> & {
+  pageSize: bigint;
+};
+
+export async function getLogsPaginated<
+  chain extends Chain | undefined,
+  const abiEvent extends AbiEvent | undefined = undefined,
+  const abiEvents extends
+    | readonly AbiEvent[]
+    | readonly unknown[]
+    | undefined = abiEvent extends AbiEvent ? [abiEvent] : undefined,
+  strict extends boolean | undefined = undefined,
+>(
+  client: Client<Transport, chain>,
+  params: GetLogsPaginatedParameters<abiEvent, abiEvents, strict>,
+): Promise<
+  GetLogsReturnType<abiEvent, abiEvents, strict, BlockNumber, BlockNumber>
+> {
+  const from_ = params.fromBlock as bigint;
+  const to_ = params.toBlock as bigint;
+  const pageSize = params.pageSize;
+  const requests: GetLogsParameters<
+    abiEvent,
+    abiEvents,
+    strict,
+    BlockNumber,
+    BlockNumber
+  >[] = [];
+  for (let fromBlock = from_; fromBlock < to_; fromBlock += pageSize) {
+    let toBlock = fromBlock + pageSize - 1n;
+    if (toBlock > to_) {
+      toBlock = to_;
+    }
+    requests.push({
+      ...params,
+      fromBlock,
+      toBlock,
+    } as GetLogsParameters<
+      abiEvent,
+      abiEvents,
+      strict,
+      BlockNumber,
+      BlockNumber
+    >);
+  }
+  const responses = await Promise.all(
+    requests.map(r =>
+      getLogs<chain, abiEvent, abiEvents, strict, BlockNumber, BlockNumber>(
+        client,
+        r,
+      ),
+    ),
+  );
+  const events = responses.flat();
+
+  return events;
+}

src/utils/getLogsSafe.ts

@@ -3,7 +3,7 @@ export * from "./bigint-utils.js";
 export * from "./detect-network.js";
 export * from "./etherscan.js";
 export * from "./formatters.js";
-export * from "./getLogsSafe.js";
+export * from "./getLogsPaginated.js";
 export * from "./retry.js";
 export * from "./simulateMulticall.js";
 export * from "./types.js";