fix: reverted orphan blocks algorithm behavior

Author: mschead

indexer/src/services/define-canonical.ts

@@ -5,42 +5,37 @@ import { getRequiredEnvString } from '@/utils/helpers';
 import { processPayload, saveBlock } from './streaming';
 import { sequelize } from '@/config/database';
 import { Transaction } from 'sequelize';
+import { BlockOutput } from '@/kadena-server/repository/application/block-repository';
 
 const blockRepository = new BlockDbRepository();
 const SYNC_BASE_URL = getRequiredEnvString('SYNC_BASE_URL');
 const SYNC_NETWORK = getRequiredEnvString('SYNC_NETWORK');
 
-export async function defineCanonicalBaseline(
-  blockHash: string,
-  parentHash: string,
-  height: number,
-  chainId: number,
-) {
+export async function defineCanonicalBaseline(blockHash: string) {
+  let tipBlock: BlockOutput | null = null;
   try {
-    const parentBlock = await blockRepository.getBlockByHash(parentHash);
-    if (!parentBlock) {
-      console.log(`[INFO][SYNC][DEFINE_CANONICAL] parentBlock not found. filling gaps ...`);
-      console.log(`[chainId: ${chainId}, parentHash: ${parentHash}, height: ${height - 1}]`);
-      await fillChainGapAndConfirmBlockPath(parentHash, height - 1, chainId);
-    }
+    tipBlock = await blockRepository.getBlockByHash(blockHash);
   } catch (error) {
-    console.error(`[ERROR][SYNC][DEFINE_CANONICAL] Error filling gaps:`, error);
-    return;
+    console.error(`[FATAL][DB][BASELINE] There was an error with the database:`, error);
+    process.exit(1);
   }
 
-  const tipBlock = await blockRepository.getBlockByHash(blockHash);
   if (!tipBlock) {
     // this scenario should not happen, but if it does, terminate the app.
-    console.error(`[ERROR][SYNC][DEFINE_CANONICAL] Block ${blockHash} not found in database`);
+    console.error(`[FATAL][DB][BASELINE] Block ${blockHash} not found in database`);
     process.exit(1);
   }
 
   let tx: Transaction;
   try {
     tx = await sequelize.transaction();
   } catch (error) {
-    console.error(`[ERROR][SYNC][DEFINE_CANONICAL] Failed to start transaction:`, error);
-    throw error;
+    console.error(
+      `[ERROR][DB][BASELINE] Failed to start transaction for canonical algorithm:`,
+      blockHash,
+      error,
+    );
+    return;
   }
 
   try {

indexer/src/services/missing.ts

@@ -1,6 +1,7 @@
 import { rootPgPool, sequelize } from '@/config/database';
 import { getRequiredEnvString } from '@/utils/helpers';
 import { processPayload, saveBlock } from './streaming';
+import { Transaction } from 'sequelize';
 
 const SYNC_BASE_URL = getRequiredEnvString('SYNC_BASE_URL');
 const NETWORK_ID = getRequiredEnvString('SYNC_NETWORK');
@@ -148,3 +149,77 @@ async function fillChainGaps(
     console.info('[INFO][SYNC][MISSING] Processed:', chainIdDiff);
   }
 }
+
+export async function fillChainGapsBeforeDefiningCanonicalBaseline({
+  chainId,
+  lastHeight,
+  tx,
+}: {
+  chainId: number;
+  lastHeight: number;
+  tx: Transaction;
+}): Promise<void> {
+  try {
+    console.info('[INFO][SYNC][MISSING] Filling initial chain gaps:', chainId);
+    const cutUrl = `${SYNC_BASE_URL}/${NETWORK_ID}/cut`;
+    const cutRes = await fetch(cutUrl, {
+      method: 'GET',
+      headers: {
+        'Content-Type': 'application/json',
+      },
+    });
+
+    const cutData = await cutRes.json();
+
+    const chainsAndHashes = Object.keys(cutData.hashes).map(chainId => ({
+      chainId,
+      hash: cutData.hashes[chainId].hash,
+    }));
+
+    const dbQuery = `
+      SELECT MAX(height) as height
+      FROM "Blocks"
+      WHERE "chainId" = $1
+    `;
+
+    const { rows } = await rootPgPool.query(dbQuery, [chainId]);
+
+    const fromHeight = rows[0].height + 1;
+    const toHeight = lastHeight - 1;
+
+    if (fromHeight > toHeight) {
+      console.info(`[INFO][SYNC][MISSING] No gaps to fill for chain ${chainId}`);
+      return;
+    }
+
+    const url = `${SYNC_BASE_URL}/${NETWORK_ID}/chain/${chainId}/block/branch?minheight=${fromHeight}&maxheight=${toHeight}`;
+
+    const res = await fetch(url, {
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/json',
+        Accept: 'application/json',
+      },
+      body: JSON.stringify({
+        upper: [chainsAndHashes[chainId].hash],
+      }),
+    });
+
+    const data = await res.json();
+
+    const promises = data.items.map(async (item: any) => {
+      const payload = processPayload(item.payloadWithOutputs);
+      return saveBlock({ header: item.header, payload, canonical: true }, tx);
+    });
+
+    await Promise.all(promises);
+
+    console.info(`[INFO][SYNC][MISSING] Initial chain gaps filled:`, chainId, fromHeight, toHeight);
+  } catch (error) {
+    console.error(
+      `[FATAL][SYNC][MISSING] Error filling chain ${chainId} gaps before defining canonical baseline:`,
+      error,
+    );
+    process.exit(1);
+  }
+}

indexer/src/services/streaming.ts

@@ -23,7 +23,10 @@ import { backfillGuards } from './guards';
 import { Transaction } from 'sequelize';
 import { PriceUpdaterService } from './price/price-updater.service';
 import { defineCanonicalBaseline } from '@/services/define-canonical';
-import { startMissingBlocksBeforeStreamingProcess } from '@/services/missing';
+import {
+  fillChainGapsBeforeDefiningCanonicalBaseline,
+  startMissingBlocksBeforeStreamingProcess,
+} from '@/services/missing';
 import { EventAttributes } from '@/models/event';
 import { startPairCreation } from '@/services/start-pair-creation';
 
@@ -53,10 +56,7 @@ export async function startStreaming() {
   await startMissingBlocksBeforeStreamingProcess();
 
   const nextBlocksToProcess: any[] = [];
-  const blocksRecentlyProcessed = new Set<string>();
-
-  // Initialize price updater
-  PriceUpdaterService.getInstance();
+  const initialChainGapsAlreadyFilled = new Set<number>();
 
   // Initialize EventSource connection to the blockchain node
   const eventSource = new EventSource(`${SYNC_BASE_URL}/${SYNC_NETWORK}/block/updates`);
@@ -77,35 +77,62 @@ export async function startStreaming() {
   };
 
   const processBlock = async (block: any) => {
-    const blockIdentifier = block.header.hash;
-
-    if (blocksRecentlyProcessed.has(blockIdentifier)) {
-      await defineCanonicalBaseline(
-        block.header.hash,
-        block.header.parent,
-        block.header.height,
-        block.header.chainId,
+    const blockHash = block.header.hash;
+
+    let blockInDatabase: Block | null = null;
+    try {
+      blockInDatabase = await Block.findOne({ where: { hash: blockHash } });
+    } catch (error) {
+      console.error(
+        '[FATAL][DB][STREAMING] There was an error with the database:',
+        blockHash,
+        error,
+      );
+      process.exit(1);
+    }
+
+    if (blockInDatabase) {
+      await defineCanonicalBaseline(block.header.hash);
+      return;
+    }
+
+    let tx: Transaction;
+    try {
+      tx = await sequelize.transaction();
+    } catch (error) {
+      console.error(
+        '[ERROR][DB][STREAMING] Failed to start transaction for new block:',
+        blockHash,
+        error,
       );
       return;
     }
 
-    const tx = await sequelize.transaction();
     try {
       const payload = processPayload(block.payloadWithOutputs);
       await saveBlock({ header: block.header, payload, canonical: null }, tx);
+
+      if (!initialChainGapsAlreadyFilled.has(block.header.chainId)) {
+        initialChainGapsAlreadyFilled.add(block.header.chainId);
+        await fillChainGapsBeforeDefiningCanonicalBaseline({
+          chainId: block.header.chainId,
+          lastHeight: block.header.height,
+          tx,
+        });
+      }
+
       await tx.commit();
-      blocksRecentlyProcessed.add(blockIdentifier);
     } catch (error) {
       await tx.rollback();
+      console.error(
+        '[ERROR][DB][STREAMING] Failed to start transaction for new block:',
+        blockHash,
+        error,
+      );
       return;
     }
 
-    await defineCanonicalBaseline(
-      block.header.hash,
-      block.header.parent,
-      block.header.height,
-      block.header.chainId,
-    );
+    await defineCanonicalBaseline(block.header.hash);
   };
 
   const processBlocks = async () => {
@@ -134,13 +161,6 @@ export async function startStreaming() {
     setTimeout(processBlocks, 1000);
   };
 
-  setInterval(
-    () => {
-      blocksRecentlyProcessed.clear();
-    },
-    1000 * 60 * 60 * 1, // 1 hour
-  );
-
   processBlocks();
   backfillGuards();