Improve support for sparse data (#196)

Author: fracek

change/@apibara-evm-rpc-4574c99a-543b-46c1-ad62-77f9d7332378.json

@@ -0,0 +1,7 @@
+{
+  "type": "prerelease",
+  "comment": "rpc: periodically send empty blocks",
+  "packageName": "@apibara/evm-rpc",
+  "email": "francesco@ceccon.me",
+  "dependentChangeType": "patch"
+}

change/@apibara-protocol-0bf11335-0d8e-496e-8e6c-f4f955860d97.json

@@ -0,0 +1,7 @@
+{
+  "type": "prerelease",
+  "comment": "rpc: periodically send empty blocks",
+  "packageName": "@apibara/protocol",
+  "email": "francesco@ceccon.me",
+  "dependentChangeType": "patch"
+}

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

@@ -121,6 +121,7 @@ export class EvmRpcStream extends RpcStreamConfig<Filter, Block> {
   async fetchBlockRange({
     startBlock,
     finalizedBlock,
+    force,
     filter,
   }: FetchBlockRangeArgs<Filter>): Promise<FetchBlockRangeResult<Block>> {
     const { start: fromBlock, end: toBlock } = this.blockRangeOracle.clampRange(
@@ -151,6 +152,12 @@ export class EvmRpcStream extends RpcStreamConfig<Filter, Block> {
           }),
         );
       }
+    } else if (force && blockNumbers.length === 0) {
+      blockNumberResponses.push(
+        this.fetchBlockHeaderByNumberWithRetry({
+          blockNumber: toBlock,
+        }),
+      );
     } else {
       for (const blockNumber of blockNumbers) {
         blockNumberResponses.push(
@@ -180,6 +187,7 @@ export class EvmRpcStream extends RpcStreamConfig<Filter, Block> {
   async fetchBlockByNumber({
     blockNumber,
     expectedParentBlockHash,
+    isAtHead,
     filter,
   }: FetchBlockByNumberArgs<Filter>): Promise<FetchBlockByNumberResult<Block>> {
     // Fetch block header and check it matches the expected parent block hash.
@@ -226,8 +234,12 @@ export class EvmRpcStream extends RpcStreamConfig<Filter, Block> {
 
     let block = null;
 
-    // TODO: handle header on new block.
-    if (filter.header === "always" || logs.length > 0) {
+    const shouldSendBlock =
+      filter.header === "always" ||
+      logs.length > 0 ||
+      (filter.header === "on_data_or_on_new_block" && isAtHead);
+
+    if (shouldSendBlock) {
       block = {
         header,
         logs,

packages/protocol/src/rpc/config.ts

@@ -3,6 +3,7 @@ import type { Bytes, Cursor } from "../common";
 export type FetchBlockRangeArgs<TFilter> = {
   startBlock: bigint;
   finalizedBlock: bigint;
+  force: boolean;
   filter: TFilter;
 };
 
@@ -27,6 +28,7 @@ export type BlockInfo = {
 export type FetchBlockByNumberArgs<TFilter> = {
   blockNumber: bigint;
   expectedParentBlockHash: Bytes;
+  isAtHead: boolean;
   filter: TFilter;
 };
 

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

@@ -16,6 +16,8 @@ type State<TFilter, TBlock> = {
   lastFinalizedRefresh: number;
   // When the last heartbeat was sent.
   lastHeartbeat: number;
+  // When the last backfill message was sent.
+  lastBackfillMessage: number;
   // Track the chain's state.
   chainTracker: ChainTracker;
   // Heartbeat interval in milliseconds.
@@ -95,6 +97,7 @@ export class RpcDataStream<TFilter, TBlock> {
       cursor,
       lastHeartbeat: Date.now(),
       lastFinalizedRefresh: Date.now(),
+      lastBackfillMessage: Date.now(),
       chainTracker,
       config: this.config,
       heartbeatIntervalMs: this.heartbeatIntervalMs,
@@ -165,9 +168,14 @@ async function* backfillFinalizedBlocks<TFilter, TBlock>(
   const { cursor, chainTracker, config, filter } = state;
   const finalized = chainTracker.finalized();
 
+  // While backfilling we want to regularly send some blocks (even if empty) so
+  // that the client can store the cursor.
+  const force = shouldForceBackfill(state);
+
   const filterData = await config.fetchBlockRange({
     startBlock: cursor.orderKey + 1n,
     finalizedBlock: finalized.orderKey,
+    force,
     filter,
   });
 
@@ -179,6 +187,7 @@ async function* backfillFinalizedBlocks<TFilter, TBlock>(
 
   for (const data of filterData.data) {
     state.lastHeartbeat = Date.now();
+    state.lastBackfillMessage = Date.now();
     yield {
       _tag: "data",
       data: {
@@ -213,6 +222,7 @@ async function* produceNextBlock<TFilter, TBlock>(
 
   const result = await state.config.fetchBlockByNumber({
     blockNumber: state.cursor.orderKey + 1n,
+    isAtHead: isAtHead(state),
     expectedParentBlockHash: currentBlockHash,
     filter: state.filter,
   });
@@ -325,6 +335,12 @@ function shouldSendHeartbeat(state: State<unknown, unknown>): boolean {
   return now - lastHeartbeat >= heartbeatIntervalMs;
 }
 
+function shouldForceBackfill(state: State<unknown, unknown>): boolean {
+  const { lastBackfillMessage, heartbeatIntervalMs } = state;
+  const now = Date.now();
+  return now - lastBackfillMessage >= heartbeatIntervalMs;
+}
+
 function shouldContinue(state: State<unknown, unknown>): boolean {
   const { endingCursor } = state.options || {};
   if (endingCursor === undefined) return true;