From a6aabaf810f43d6ab027c1dc4d08e98c86abb18d Mon Sep 17 00:00:00 2001 From: Moody Salem Date: Mon, 11 May 2026 14:37:07 -0400 Subject: [PATCH 1/3] Avoid refetching empty live RPC ranges --- packages/protocol/src/rpc/data-stream.ts | 40 +++- .../protocol/tests/rpc-data-stream.test.ts | 190 ++++++++++++++++++ 2 files changed, 222 insertions(+), 8 deletions(-) create mode 100644 packages/protocol/tests/rpc-data-stream.test.ts diff --git a/packages/protocol/src/rpc/data-stream.ts b/packages/protocol/src/rpc/data-stream.ts index 0f4d1c83..bfa25640 100644 --- a/packages/protocol/src/rpc/data-stream.ts +++ b/packages/protocol/src/rpc/data-stream.ts @@ -285,12 +285,9 @@ async function* produceLiveBlocks( 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", @@ -303,9 +300,10 @@ async function* produceLiveBlocks( } const head = chainTracker.head(); + const startBlock = lastProcessedLiveBlock(state) + 1n; const filterData = await config.fetchBlockRange({ - startBlock: cursor.orderKey + 1n, + startBlock, maxBlock: head.orderKey, force: false, clampAllowed: false, @@ -412,8 +410,9 @@ async function* waitForHeadChange( 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 +463,32 @@ function shouldRefreshHead(state: State): boolean { function isAtHead(state: State): 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, + cursor: Cursor, +): boolean { + return ( + cursor.orderKey < state.cursor.orderKey || + (state.lastEmptyBlockNumber !== undefined && + cursor.orderKey < state.lastEmptyBlockNumber) + ); +} + +function lastProcessedLiveBlock(state: State): bigint { + if ( + state.lastEmptyBlockNumber !== undefined && + state.lastEmptyBlockNumber > state.cursor.orderKey + ) { + return state.lastEmptyBlockNumber; + } + + return state.cursor.orderKey; } function sleep(duration: number): Promise { diff --git a/packages/protocol/tests/rpc-data-stream.test.ts b/packages/protocol/tests/rpc-data-stream.test.ts new file mode 100644 index 00000000..5d2a8ab2 --- /dev/null +++ b/packages/protocol/tests/rpc-data-stream.test.ts @@ -0,0 +1,190 @@ +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 { + fetchBlockRangeCalls: FetchBlockRangeArgs[] = []; + headBlock = 2n; + + headRefreshIntervalMs(): number { + return 10; + } + + finalizedRefreshIntervalMs(): number { + return 10_000; + } + + validateFilter() { + return { valid: true as const }; + } + + async fetchCursor(args: FetchCursorArgs): Promise { + 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)); + } + + async fetchCursorRange({ + startBlockNumber, + endBlockNumber, + }: FetchCursorRangeArgs): Promise { + const blocks: BlockInfo[] = []; + for ( + let blockNumber = startBlockNumber; + blockNumber <= endBlockNumber; + blockNumber++ + ) { + blocks.push(blockInfo(blockNumber)); + } + return blocks; + } + + async fetchBlockRange( + args: FetchBlockRangeArgs, + ): Promise> { + this.fetchBlockRangeCalls.push(args); + await sleep(1); + + return { + startBlock: args.startBlock, + endBlock: args.maxBlock, + data: [], + }; + } + + async fetchHeaderByHash({ + blockHash, + }: FetchBlockByHashArgs): Promise< + FetchBlockByHashResult + > { + 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: 3n, + 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 { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function withTimeout(promise: Promise, ms: number): Promise { + return Promise.race([ + promise, + sleep(ms).then(() => { + throw new Error(`Timed out after ${ms}ms`); + }), + ]); +} From 69f6e7010e5837d909a64f26752dab51271abd05 Mon Sep 17 00:00:00 2001 From: Francesco Ceccon Date: Mon, 11 May 2026 21:01:04 +0200 Subject: [PATCH 2/3] protocol: advance cursor only on data --- packages/protocol/src/rpc/data-stream.ts | 3 +-- packages/protocol/tests/rpc-data-stream.test.ts | 8 +++----- 2 files changed, 4 insertions(+), 7 deletions(-) diff --git a/packages/protocol/src/rpc/data-stream.ts b/packages/protocol/src/rpc/data-stream.ts index bfa25640..8774557f 100644 --- a/packages/protocol/src/rpc/data-stream.ts +++ b/packages/protocol/src/rpc/data-stream.ts @@ -300,10 +300,9 @@ async function* produceLiveBlocks( } const head = chainTracker.head(); - const startBlock = lastProcessedLiveBlock(state) + 1n; const filterData = await config.fetchBlockRange({ - startBlock, + startBlock: cursor.orderKey + 1n, maxBlock: head.orderKey, force: false, clampAllowed: false, diff --git a/packages/protocol/tests/rpc-data-stream.test.ts b/packages/protocol/tests/rpc-data-stream.test.ts index 5d2a8ab2..b52d9d42 100644 --- a/packages/protocol/tests/rpc-data-stream.test.ts +++ b/packages/protocol/tests/rpc-data-stream.test.ts @@ -45,7 +45,7 @@ class EmptyLiveHeadConfig extends RpcStreamConfig { return blockInfo(args.blockNumber); } - return blockInfo(blockNumberFromHash(args.blockHash)); + return blockInfo(blockNumberFromHash(args.blockHash ?? "0x0")); } async fetchCursorRange({ @@ -78,9 +78,7 @@ class EmptyLiveHeadConfig extends RpcStreamConfig { async fetchHeaderByHash({ blockHash, - }: FetchBlockByHashArgs): Promise< - FetchBlockByHashResult - > { + }: FetchBlockByHashArgs): Promise> { const blockNumber = blockNumberFromHash(blockHash); const info = blockInfo(blockNumber); @@ -138,7 +136,7 @@ describe("RpcDataStream", () => { }); expect(config.fetchBlockRangeCalls).toHaveLength(2); expect(config.fetchBlockRangeCalls[1]).toMatchObject({ - startBlock: 3n, + startBlock: 2n, maxBlock: 3n, }); await iterator.return?.(); From 3d285f90a74f2b91c425cec4324079a50f12dfdf Mon Sep 17 00:00:00 2001 From: Francesco Ceccon Date: Mon, 11 May 2026 21:01:18 +0200 Subject: [PATCH 3/3] Change files --- ...bara-protocol-cefba257-a3ad-451e-b3d6-13613ecf7903.json | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 change/@apibara-protocol-cefba257-a3ad-451e-b3d6-13613ecf7903.json diff --git a/change/@apibara-protocol-cefba257-a3ad-451e-b3d6-13613ecf7903.json b/change/@apibara-protocol-cefba257-a3ad-451e-b3d6-13613ecf7903.json new file mode 100644 index 00000000..157032f8 --- /dev/null +++ b/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" +}