Reduce the number of RPC calls (#203)

This PR reduces the number of RPC calls to index non-finalized blocks by
optimistically fetching a block range.

Author: fracek

change/@apibara-evm-rpc-336c4f49-bb04-49d3-b041-9e3e4ee3079a.json

@@ -0,0 +1,7 @@
+{
+  "type": "prerelease",
+  "comment": "rpc: reduce the number of rpc calls",
+  "packageName": "@apibara/evm-rpc",
+  "email": "francesco@ceccon.me",
+  "dependentChangeType": "patch"
+}

change/@apibara-protocol-4b27cd4e-3d41-49a9-b1fd-4c8ef92a5fe5.json

@@ -0,0 +1,7 @@
+{
+  "type": "prerelease",
+  "comment": "rpc: reduce the number of rpc calls",
+  "packageName": "@apibara/protocol",
+  "email": "francesco@ceccon.me",
+  "dependentChangeType": "patch"
+}

change/@apibara-starknet-406308ed-54b4-4314-843d-15f3c9a194ec.json

@@ -0,0 +1,7 @@
+{
+  "type": "prerelease",
+  "comment": "rpc: reduce the number of rpc calls",
+  "packageName": "@apibara/starknet",
+  "email": "francesco@ceccon.me",
+  "dependentChangeType": "patch"
+}

examples/evm-client/src/main-rpc.ts

@@ -19,7 +19,8 @@ const command = defineCommand({
     },
     contract: {
       type: "string",
-      default: "0xf08A50178dfcDe18524640EA6618a1f965821715",
+      // default: "0xf08A50178dfcDe18524640EA6618a1f965821715",
+      default: "0xaf88d065e77c8cC2239327C5EDb3A432268e5831",
       description: "Contract address to monitor",
     },
     startBlock: {
@@ -29,7 +30,7 @@ const command = defineCommand({
     },
     finality: {
       type: "string",
-      default: "finalized",
+      default: "accepted",
       description: "Block finality (finalized, accepted)",
     },
   },
@@ -51,7 +52,11 @@ const command = defineCommand({
       new EvmRpcStream(viemClient, {
         // This parameter changes based on the rpc provider.
         // The stream automatically shrinks the batch size when the provider returns an error.
-        getLogsRangeSize: 1_000n,
+        // getLogsRangeSize: 1_000n,
+        getLogsRangeSize: 100n,
+        mergeGetLogsFilter: "always",
+        // alwaysSendAcceptedHeaders: true,
+        headRefreshIntervalMs: 500,
       }),
     );
 

packages/evm-rpc/src/stream-config.ts

@@ -7,6 +7,7 @@ import {
   type FetchBlockRangeResult,
   type FetchBlockResult,
   type FetchCursorArgs,
+  type FetchCursorRangeArgs,
   RpcStreamConfig,
   type ValidateFilterResult,
 } from "@apibara/protocol/rpc";
@@ -123,14 +124,33 @@ export class EvmRpcStream extends RpcStreamConfig<Filter, Block> {
     };
   }
 
+  async fetchCursorRange({
+    startBlockNumber,
+    endBlockNumber,
+  }: FetchCursorRangeArgs): Promise<BlockInfo[]> {
+    const requestCount = Number(endBlockNumber - startBlockNumber + 1n);
+    return await Promise.all(
+      Array.from({ length: requestCount }, async (_, i) => {
+        const blockNumber = startBlockNumber + BigInt(i);
+        const info = await this.fetchCursor({ blockNumber });
+        if (!info) {
+          throw new Error(
+            `RPC returned null block for block number ${blockNumber}`,
+          );
+        }
+        return info;
+      }),
+    );
+  }
+
   async fetchBlockRange({
     startBlock,
-    finalizedBlock,
+    maxBlock,
     force,
     filter,
   }: FetchBlockRangeArgs<Filter>): Promise<FetchBlockRangeResult<Block>> {
     const { start: fromBlock, end: toBlock } = this.blockRangeOracle.clampRange(
-      { start: startBlock, end: finalizedBlock },
+      { start: startBlock, end: maxBlock },
     );
 
     // console.log("Fetching block range", fromBlock, toBlock, filter);

packages/evm-rpc/tests/merge-get-logs.integration.test.ts

@@ -340,14 +340,14 @@ describe("EvmRpcStream", () => {
 
     const merged = await mergedStream.fetchBlockRange({
       startBlock,
-      finalizedBlock: endBlock,
+      maxBlock: endBlock,
       force: true,
       filter,
     });
 
     const standard = await standardStream.fetchBlockRange({
       startBlock,
-      finalizedBlock: endBlock,
+      maxBlock: endBlock,
       force: true,
       filter,
     });

packages/protocol/package.json

@@ -38,6 +38,7 @@
   "scripts": {
     "build": "pnpm build:proto && unbuild",
     "build:proto": "buf generate proto",
+    "buf": "buf",
     "typecheck": "tsc --noEmit",
     "test": "vitest",
     "test:ci": "vitest run",

packages/protocol/src/rpc/chain-tracker.ts

@@ -1,11 +1,12 @@
 import { fromHex } from "viem";
 import type { Bytes, Cursor } from "../common";
-import type { BlockInfo } from "./config";
+import type { BlockInfo, FetchCursorRangeArgs } from "./config";
 import { blockInfoToCursor } from "./helpers";
 
 type UpdateHeadArgs = {
   newHead: BlockInfo;
   fetchCursorByHash: (hash: Bytes) => Promise<BlockInfo | null>;
+  fetchCursorRange: (args: FetchCursorRangeArgs) => Promise<BlockInfo[]>;
 };
 
 type UpdateHeadResult =
@@ -24,10 +25,21 @@ export class ChainTracker {
   #finalized: BlockInfo;
   #head: BlockInfo;
   #canonical: Map<bigint, BlockInfo>;
+  #batchSize: bigint;
 
-  constructor({ head, finalized }: { finalized: BlockInfo; head: BlockInfo }) {
+  constructor({
+    head,
+    finalized,
+    batchSize,
+  }: {
+    finalized: BlockInfo;
+    head: BlockInfo;
+    batchSize: bigint;
+  }) {
     this.#finalized = finalized;
     this.#head = head;
+    this.#batchSize = batchSize;
+
     this.#canonical = new Map([
       [finalized.blockNumber, finalized],
       [head.blockNumber, head],
@@ -106,6 +118,7 @@ export class ChainTracker {
   async updateHead({
     newHead,
     fetchCursorByHash,
+    fetchCursorRange,
   }: UpdateHeadArgs): Promise<UpdateHeadResult> {
     // console.debug(
     //   `updateHead: new=${newHead.blockNumber} old=${this.#head.blockNumber}`,
@@ -192,10 +205,47 @@ export class ChainTracker {
     // The new chain is longer and we need the missing blocks.
     // This may result in reorgs.
 
-    let current = newHead;
-    let reorgDetected = false;
-    const blocksToApply = [newHead];
+    // console.log(
+    //   `Moving from ${this.#head.blockNumber} to ${newHead.blockNumber} (${newHead.blockNumber - this.#head.blockNumber} blocks)`,
+    // );
+
+    let currentBlockNumber = this.#head.blockNumber + 1n;
+
+    while (true) {
+      let endBlockNumber = currentBlockNumber + this.#batchSize;
+      if (endBlockNumber > newHead.blockNumber) {
+        endBlockNumber = newHead.blockNumber;
+      }
+
+      const missing = await fetchCursorRange({
+        startBlockNumber: currentBlockNumber,
+        endBlockNumber,
+      });
+
+      for (const block of missing) {
+        const canonicalParent = this.#canonical.get(block.blockNumber - 1n);
+        if (
+          !canonicalParent ||
+          canonicalParent.blockHash !== block.parentBlockHash
+        ) {
+          throw new Error(
+            "Chain reorganization detected. Recovery not implemented",
+          );
+        }
+
+        this.#canonical.set(block.blockNumber, block);
+
+        // console.log(`Applied block ${block.blockNumber}`);
+      }
+
+      if (endBlockNumber === newHead.blockNumber) {
+        break;
+      }
+
+      currentBlockNumber = endBlockNumber + 1n;
+    }
 
+    /*
     while (true) {
       const parent = await fetchCursorByHash(current.parentBlockHash);
 
@@ -226,24 +276,39 @@ export class ChainTracker {
       blocksToApply.push(parent);
       current = parent;
     }
+    */
+    // throw new Error("FUCK IT");
 
-    for (const block of blocksToApply.reverse()) {
-      this.#canonical.set(block.blockNumber, block);
-    }
+    // for (const block of blocksToApply.reverse()) {
+    //   this.#canonical.set(block.blockNumber, block);
+    // }
 
-    const previousHead = this.#head;
+    // const previousHead = this.#head;
     this.#head = newHead;
 
-    if (reorgDetected) {
-      return {
-        status: "reorg",
-        cursor: blockInfoToCursor(previousHead),
-      };
-    }
+    // if (reorgDetected) {
+    //   return {
+    //     status: "reorg",
+    //     cursor: blockInfoToCursor(previousHead),
+    //   };
+    // }
 
     return { status: "success" };
   }
 
+  isCanonical({ orderKey, uniqueKey }: Cursor) {
+    if (!uniqueKey) {
+      return true;
+    }
+
+    const block = this.#canonical.get(orderKey);
+    if (!block) {
+      return true;
+    }
+
+    return block.blockHash === uniqueKey;
+  }
+
   async initializeStartingCursor({
     cursor,
     fetchCursor,
@@ -319,9 +384,11 @@ export class ChainTracker {
 export function createChainTracker({
   head,
   finalized,
+  batchSize,
 }: {
   head: BlockInfo;
   finalized: BlockInfo;
+  batchSize: bigint;
 }): ChainTracker {
-  return new ChainTracker({ finalized, head });
+  return new ChainTracker({ finalized, head, batchSize });
 }

packages/protocol/src/rpc/config.ts

@@ -2,7 +2,7 @@ import type { Bytes, Cursor } from "../common";
 
 export type FetchBlockRangeArgs<TFilter> = {
   startBlock: bigint;
-  finalizedBlock: bigint;
+  maxBlock: bigint;
   force: boolean;
   filter: TFilter;
 };
@@ -43,6 +43,11 @@ export type FetchBlockByNumberResult<TBlock> =
       blockInfo: BlockInfo;
     };
 
+export type FetchCursorRangeArgs = {
+  startBlockNumber: bigint;
+  endBlockNumber: bigint;
+};
+
 export type FetchCursorArgs =
   | {
       blockTag: "latest" | "finalized";
@@ -74,6 +79,7 @@ export abstract class RpcStreamConfig<TFilter, TBlock> {
   abstract headRefreshIntervalMs(): number;
   abstract finalizedRefreshIntervalMs(): number;
 
+  abstract fetchCursorRange(args: FetchCursorRangeArgs): Promise<BlockInfo[]>;
   abstract fetchCursor(args: FetchCursorArgs): Promise<BlockInfo | null>;
 
   abstract validateFilter(filter: TFilter): ValidateFilterResult;