Avoid refetching empty live RPC ranges (#220)

 ## Summary

Fix repeated live RPC range scans after emitting empty accepted blocks.

When live filtering reaches the current head with no matching data, the
stream emits an empty accepted block but
intentionally does not advance `state.cursor`. Subsequent live scans
were still starting from `cursor + 1`, so sparse
filters could repeatedly refetch previously scanned empty blocks as the
head advanced.

This changes live range fetching to resume from `max(cursor,
lastEmptyBlockNumber) + 1`, treats `lastEmptyBlockNumber ===
head` as being at head, and clears the empty-block marker on reorg
invalidation.

  ## Testing

- Added a protocol test covering empty accepted live blocks as the head
advances.
- Not run locally: dependencies are not installed in this checkout, so
`vitest` is unavailable.

Author: fracek

change/@apibara-protocol-cefba257-a3ad-451e-b3d6-13613ecf7903.json

@@ -0,0 +1,7 @@
+{
+  "type": "patch",
+  "comment": "Avoid refetching empty live RPC ranges",
+  "packageName": "@apibara/protocol",
+  "email": "francesco@ceccon.me",
+  "dependentChangeType": "patch"
+}

packages/protocol/src/rpc/data-stream.ts

@@ -285,12 +285,9 @@ async function* produceLiveBlocks<TFilter, TBlock>(
     if (result.status === "reorg") {
       const { cursor } = result;
       // Only handle reorgs if they involve blocks already processed.
-      if (
-        cursor.orderKey < state.cursor.orderKey ||
-        (state.lastEmptyBlockNumber !== undefined &&
-          cursor.orderKey < state.lastEmptyBlockNumber)
-      ) {
+      if (shouldInvalidateProcessedLiveData(state, cursor)) {
         state.cursor = cursor;
+        state.lastEmptyBlockNumber = undefined;
 
         yield {
           _tag: "invalidate",
@@ -412,8 +409,9 @@ async function* waitForHeadChange<TBlock>(
       case "reorg": {
         const { cursor } = result;
         // Only handle reorgs if they involve blocks already processed.
-        if (cursor.orderKey < state.cursor.orderKey) {
+        if (shouldInvalidateProcessedLiveData(state, cursor)) {
           state.cursor = cursor;
+          state.lastEmptyBlockNumber = undefined;
 
           yield {
             _tag: "invalidate",
@@ -464,7 +462,32 @@ function shouldRefreshHead(state: State<unknown, unknown>): boolean {
 
 function isAtHead(state: State<unknown, unknown>): boolean {
   const head = state.chainTracker.head();
-  return state.cursor.orderKey === head.orderKey;
+  return (
+    state.cursor.orderKey === head.orderKey ||
+    state.lastEmptyBlockNumber === head.orderKey
+  );
+}
+
+function shouldInvalidateProcessedLiveData(
+  state: State<unknown, unknown>,
+  cursor: Cursor,
+): boolean {
+  return (
+    cursor.orderKey < state.cursor.orderKey ||
+    (state.lastEmptyBlockNumber !== undefined &&
+      cursor.orderKey < state.lastEmptyBlockNumber)
+  );
+}
+
+function lastProcessedLiveBlock(state: State<unknown, unknown>): bigint {
+  if (
+    state.lastEmptyBlockNumber !== undefined &&
+    state.lastEmptyBlockNumber > state.cursor.orderKey
+  ) {
+    return state.lastEmptyBlockNumber;
+  }
+
+  return state.cursor.orderKey;
 }
 
 function sleep(duration: number): Promise<void> {

packages/protocol/tests/rpc-data-stream.test.ts

@@ -0,0 +1,188 @@
+import { describe, expect, it } from "vitest";
+import type { Bytes, Cursor } from "../src/common";
+import { RpcClient } from "../src/rpc/client";
+import {
+  type BlockInfo,
+  type FetchBlockByHashArgs,
+  type FetchBlockByHashResult,
+  type FetchBlockRangeArgs,
+  type FetchBlockRangeResult,
+  type FetchCursorArgs,
+  type FetchCursorRangeArgs,
+  RpcStreamConfig,
+} from "../src/rpc/config";
+
+type TestBlock = {
+  blockNumber: bigint;
+};
+
+class EmptyLiveHeadConfig extends RpcStreamConfig<string, TestBlock> {
+  fetchBlockRangeCalls: FetchBlockRangeArgs<string>[] = [];
+  headBlock = 2n;
+
+  headRefreshIntervalMs(): number {
+    return 10;
+  }
+
+  finalizedRefreshIntervalMs(): number {
+    return 10_000;
+  }
+
+  validateFilter() {
+    return { valid: true as const };
+  }
+
+  async fetchCursor(args: FetchCursorArgs): Promise<BlockInfo | null> {
+    if (args.blockTag === "latest") {
+      return blockInfo(this.headBlock);
+    }
+
+    if (args.blockTag === "finalized") {
+      return blockInfo(1n);
+    }
+
+    if (args.blockNumber !== undefined) {
+      return blockInfo(args.blockNumber);
+    }
+
+    return blockInfo(blockNumberFromHash(args.blockHash ?? "0x0"));
+  }
+
+  async fetchCursorRange({
+    startBlockNumber,
+    endBlockNumber,
+  }: FetchCursorRangeArgs): Promise<BlockInfo[]> {
+    const blocks: BlockInfo[] = [];
+    for (
+      let blockNumber = startBlockNumber;
+      blockNumber <= endBlockNumber;
+      blockNumber++
+    ) {
+      blocks.push(blockInfo(blockNumber));
+    }
+    return blocks;
+  }
+
+  async fetchBlockRange(
+    args: FetchBlockRangeArgs<string>,
+  ): Promise<FetchBlockRangeResult<TestBlock>> {
+    this.fetchBlockRangeCalls.push(args);
+    await sleep(1);
+
+    return {
+      startBlock: args.startBlock,
+      endBlock: args.maxBlock,
+      data: [],
+    };
+  }
+
+  async fetchHeaderByHash({
+    blockHash,
+  }: FetchBlockByHashArgs<string>): Promise<FetchBlockByHashResult<TestBlock>> {
+    const blockNumber = blockNumberFromHash(blockHash);
+    const info = blockInfo(blockNumber);
+
+    return {
+      blockInfo: info,
+      data: {
+        cursor: cursorForBlock(blockNumber - 1n),
+        endCursor: cursorForBlock(blockNumber)!,
+        block: { blockNumber },
+      },
+    };
+  }
+}
+
+describe("RpcDataStream", () => {
+  it("does not refetch empty accepted blocks as the live head advances", async () => {
+    const config = new EmptyLiveHeadConfig();
+    const client = new RpcClient(config);
+    const stream = client.streamData({
+      finality: "accepted",
+      filter: ["logs"],
+      startingCursor: { orderKey: 1n },
+    });
+    const iterator = stream[Symbol.asyncIterator]();
+
+    try {
+      const first = await iterator.next();
+
+      expect(first.done).toBe(false);
+      expect(first.value).toMatchObject({
+        _tag: "data",
+        data: {
+          endCursor: { orderKey: 2n },
+          finality: "accepted",
+          production: "live",
+        },
+      });
+      expect(config.fetchBlockRangeCalls).toHaveLength(1);
+      expect(config.fetchBlockRangeCalls[0]).toMatchObject({
+        startBlock: 2n,
+        maxBlock: 2n,
+      });
+
+      config.headBlock = 3n;
+      const second = await withTimeout(iterator.next(), 1_000);
+
+      expect(second.done).toBe(false);
+      expect(second.value).toMatchObject({
+        _tag: "data",
+        data: {
+          endCursor: { orderKey: 3n },
+          finality: "accepted",
+          production: "live",
+        },
+      });
+      expect(config.fetchBlockRangeCalls).toHaveLength(2);
+      expect(config.fetchBlockRangeCalls[1]).toMatchObject({
+        startBlock: 2n,
+        maxBlock: 3n,
+      });
+      await iterator.return?.();
+    } finally {
+      await iterator.return?.();
+    }
+  });
+});
+
+function blockInfo(blockNumber: bigint): BlockInfo {
+  return {
+    blockNumber,
+    blockHash: blockHash(blockNumber),
+    parentBlockHash: blockHash(blockNumber - 1n),
+  };
+}
+
+function cursorForBlock(blockNumber: bigint): Cursor | undefined {
+  if (blockNumber < 0n) {
+    return undefined;
+  }
+
+  return {
+    orderKey: blockNumber,
+    uniqueKey: blockHash(blockNumber),
+  };
+}
+
+function blockHash(blockNumber: bigint): Bytes {
+  const value = blockNumber < 0n ? 0n : blockNumber;
+  return `0x${value.toString(16).padStart(64, "0")}`;
+}
+
+function blockNumberFromHash(hash: Bytes): bigint {
+  return BigInt(hash);
+}
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+function withTimeout<T>(promise: Promise<T>, ms: number): Promise<T> {
+  return Promise.race([
+    promise,
+    sleep(ms).then(() => {
+      throw new Error(`Timed out after ${ms}ms`);
+    }),
+  ]);
+}