Fix RPC chain tracker (#198)

This PR makes sure the starting cursor is in the chain tracker if it's a
non-reorged block.

Author: fracek

change/@apibara-protocol-01feb2dd-0a0a-468c-bca6-6cfaa832402a.json

@@ -0,0 +1,7 @@
+{
+  "type": "prerelease",
+  "comment": "protocol: insert starting cursor in chain tracker",
+  "packageName": "@apibara/protocol",
+  "email": "francesco@ceccon.me",
+  "dependentChangeType": "patch"
+}

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

@@ -43,6 +43,10 @@ export class ChainTracker {
   }
 
   updateFinalized(newFinalized: BlockInfo) {
+    // console.debug(
+    //   `updateFinalized: new=${newFinalized.blockNumber} old=${this.#finalized.blockNumber}`,
+    // );
+
     if (newFinalized.blockNumber < this.#finalized.blockNumber) {
       throw new Error("Finalized cursor moved backwards");
     }
@@ -70,7 +74,9 @@ export class ChainTracker {
     return true;
   }
 
-  addToCanonicalChain(blockInfo: BlockInfo) {
+  addToCanonicalChain({ blockInfo }: { blockInfo: BlockInfo }) {
+    // console.debug(`addToCanonicalChain: block=${blockInfo.blockNumber}`);
+
     const existing = this.#canonical.get(blockInfo.blockNumber);
 
     if (existing) {
@@ -83,7 +89,7 @@ export class ChainTracker {
 
     const parent = this.#canonical.get(blockInfo.blockNumber - 1n);
     if (!parent) {
-      throw new Error("Parent block not found");
+      throw new Error("Parent block not in canonical chain");
     }
 
     if (parent.blockHash !== blockInfo.parentBlockHash) {
@@ -101,6 +107,10 @@ export class ChainTracker {
     newHead,
     fetchCursorByHash,
   }: UpdateHeadArgs): Promise<UpdateHeadResult> {
+    // console.debug(
+    //   `updateHead: new=${newHead.blockNumber} old=${this.#head.blockNumber}`,
+    // );
+
     // No changes to the chain.
     if (
       newHead.blockNumber === this.#head.blockNumber &&
@@ -245,21 +255,24 @@ export class ChainTracker {
     | { canonical: false; reason: string; fullCursor?: undefined }
   > {
     const head = this.head();
+    const finalized = this.finalized();
+
     if (cursor.orderKey > head.orderKey) {
       return { canonical: false, reason: "cursor is ahead of head" };
     }
 
+    const expectedInfo = await fetchCursor(cursor.orderKey);
     if (!cursor.uniqueKey) {
-      const fullInfo = await fetchCursor(cursor.orderKey);
-
-      if (fullInfo === null) {
+      if (expectedInfo === null) {
         throw new Error("Failed to initialize canonical cursor");
       }
 
-      return { canonical: true, fullCursor: blockInfoToCursor(fullInfo) };
-    }
+      if (expectedInfo.blockNumber > finalized.orderKey) {
+        this.#canonical.set(expectedInfo.blockNumber, expectedInfo);
+      }
 
-    const expectedInfo = await fetchCursor(cursor.orderKey);
+      return { canonical: true, fullCursor: blockInfoToCursor(expectedInfo) };
+    }
 
     if (expectedInfo === null) {
       return {
@@ -295,6 +308,10 @@ export class ChainTracker {
       };
     }
 
+    if (expectedInfo.blockNumber > finalized.orderKey) {
+      this.#canonical.set(expectedInfo.blockNumber, expectedInfo);
+    }
+
     return { canonical: true, fullCursor: expectedCursor };
   }
 }

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

@@ -142,8 +142,8 @@ async function* dataStreamLoop<TFilter, TBlock>(
 
     const finalized = chainTracker.finalized();
 
-    // console.log(
-    //   `Loop: c=${cursor.orderKey} f=${finalized.orderKey} h=${chainTracker.head().orderKey}`,
+    // console.debug(
+    //   `RpcLoop: c=${cursor.orderKey} f=${finalized.orderKey} h=${chainTracker.head().orderKey}`,
     // );
 
     if (cursor.orderKey < finalized.orderKey) {
@@ -200,6 +200,7 @@ async function* backfillFinalizedBlocks<TFilter, TBlock>(
     };
   }
 
+  // Notice that we check that filteredData.endBlock <= finalized.orderKey above.
   if (filterData.endBlock === finalized.orderKey) {
     // Prepare for transition to non-finalized data.
     state.cursor = finalized;
@@ -240,8 +241,9 @@ async function* produceNextBlock<TFilter, TBlock>(
     uniqueKey: blockInfo.blockHash,
   };
 
-  const { status: headUpdateStatus } =
-    state.chainTracker.addToCanonicalChain(blockInfo);
+  const { status: headUpdateStatus } = state.chainTracker.addToCanonicalChain({
+    blockInfo,
+  });
 
   if (headUpdateStatus !== "success") {
     throw new Error("Failed to update head. Would cause reorg.");