Merge pull request #1369 from nymkappa/feature/detect-re-org

[Indexing] - Support 10 blocks depth reorgs

backend/src/api/blocks.ts

@@ -23,6 +23,7 @@ class Blocks {
   private newBlockCallbacks: ((block: BlockExtended, txIds: string[], transactions: TransactionExtended[]) => void)[] = [];
   private blockIndexingStarted = false;
   public blockIndexingCompleted = false;
+  public reindexFlag = true; // Always re-index the latest indexed data in case the node went offline with an invalid block tip (reorg)
 
   constructor() { }
 
@@ -189,16 +190,19 @@ class Blocks {
    * [INDEXING] Index all blocks metadata for the mining dashboard
    */
   public async $generateBlockDatabase() {
-    if (this.blockIndexingStarted) {
+    if (this.blockIndexingStarted && !this.reindexFlag) {
       return;
     }
 
+    this.reindexFlag = false;
+
     const blockchainInfo = await bitcoinClient.getBlockchainInfo();
     if (blockchainInfo.blocks !== blockchainInfo.headers) { // Wait for node to sync
       return;
     }
 
     this.blockIndexingStarted = true;
+    this.blockIndexingCompleted = false;
 
     try {
       let currentBlockHeight = blockchainInfo.blocks;
@@ -316,6 +320,12 @@ class Blocks {
 
       if (Common.indexingEnabled()) {
         await blocksRepository.$saveBlockInDatabase(blockExtended);
+
+        // If the last 10 blocks chain is not valid, re-index them (reorg)
+        const chainValid = await blocksRepository.$validateRecentBlocks();
+        if (!chainValid) {
+          this.reindexFlag = true;
+        }
       }
 
       if (block.height % 2016 === 0) {

backend/src/index.ts

@@ -27,6 +27,7 @@ import icons from './api/liquid/icons';
 import { Common } from './api/common';
 import mining from './api/mining';
 import HashratesRepository from './repositories/HashratesRepository';
+import BlocksRepository from './repositories/BlocksRepository';
 import poolsUpdater from './tasks/pools-updater';
 
 class Server {
@@ -179,6 +180,10 @@ class Server {
 
     try {
       await poolsUpdater.updatePoolsJson();
+      if (blocks.reindexFlag) {
+        await BlocksRepository.$deleteBlocks(10);
+        await HashratesRepository.$deleteLastEntries();
+      }
       blocks.$generateBlockDatabase();
       await mining.$generateNetworkHashrateHistory();
       await mining.$generatePoolHashrateHistory();

backend/src/repositories/BlocksRepository.ts

@@ -10,9 +10,11 @@ class BlocksRepository {
    * Save indexed block data in the database
    */
   public async $saveBlockInDatabase(block: BlockExtended) {
-    const connection = await DB.getConnection();
+    let connection;
 
     try {
+      connection = await DB.getConnection();
+
       const query = `INSERT INTO blocks(
         height,           hash,                blockTimestamp, size,
         weight,           tx_count,            coinbase_raw,   difficulty,
@@ -72,8 +74,9 @@ class BlocksRepository {
       return [];
     }
 
-    const connection = await DB.getConnection();
+    let connection;
     try {
+      connection = await DB.getConnection();
       const [rows]: any[] = await connection.query(`
         SELECT height
         FROM blocks
@@ -118,8 +121,9 @@ class BlocksRepository {
 
     query += ` GROUP by pools.id`;
 
-    const connection = await DB.getConnection();
+    let connection;
     try {
+      connection = await DB.getConnection();
       const [rows] = await connection.query(query, params);
       connection.release();
 
@@ -155,8 +159,9 @@ class BlocksRepository {
       query += ` blockTimestamp BETWEEN DATE_SUB(NOW(), INTERVAL ${interval}) AND NOW()`;
     }
 
-    const connection = await DB.getConnection();
+    let connection;
     try {
+      connection = await DB.getConnection();
       const [rows] = await connection.query(query, params);
       connection.release();
 
@@ -194,8 +199,9 @@ class BlocksRepository {
     }
     query += ` blockTimestamp BETWEEN FROM_UNIXTIME('${from}') AND FROM_UNIXTIME('${to}')`;
 
-    const connection = await DB.getConnection();
+    let connection;
     try {
+      connection = await DB.getConnection();
       const [rows] = await connection.query(query, params);
       connection.release();
 
@@ -216,8 +222,9 @@ class BlocksRepository {
       ORDER BY height
       LIMIT 1;`;
 
-    const connection = await DB.getConnection();
+    let connection;
     try {
+      connection = await DB.getConnection();
       const [rows]: any[] = await connection.query(query);
       connection.release();
 
@@ -257,8 +264,9 @@ class BlocksRepository {
     query += ` ORDER BY height DESC
       LIMIT 10`;
 
-    const connection = await DB.getConnection();
+    let connection;
     try {
+      connection = await DB.getConnection();
       const [rows] = await connection.query(query, params);
       connection.release();
 
@@ -279,8 +287,9 @@ class BlocksRepository {
    * Get one block by height
    */
   public async $getBlockByHeight(height: number): Promise<object | null> {
-    const connection = await DB.getConnection();
+    let connection;
     try {
+      connection = await DB.getConnection();
       const [rows]: any[] = await connection.query(`
         SELECT *, UNIX_TIMESTAMP(blocks.blockTimestamp) as blockTimestamp,
         pools.id as pool_id, pools.name as pool_name, pools.link as pool_link, pools.slug as pool_slug,
@@ -310,8 +319,6 @@ class BlocksRepository {
   public async $getBlocksDifficulty(interval: string | null): Promise<object[]> {
     interval = Common.getSqlInterval(interval);
 
-    const connection = await DB.getConnection();
-
     // :D ... Yeah don't ask me about this one https://stackoverflow.com/a/40303162
     // Basically, using temporary user defined fields, we are able to extract all
     // difficulty adjustments from the blocks tables.
@@ -344,14 +351,17 @@ class BlocksRepository {
       ORDER BY t.height
     `;
 
+    let connection;
     try {
+      connection = await DB.getConnection();
       const [rows]: any[] = await connection.query(query);
       connection.release();
 
-      for (let row of rows) {
+      for (const row of rows) {
         delete row['rn'];
       }
 
+      connection.release();
       return rows;
     } catch (e) {
       connection.release();
@@ -386,6 +396,49 @@ class BlocksRepository {
       throw e;
     }
   }
+
+  /*
+   * Check if the last 10 blocks chain is valid
+   */
+  public async $validateRecentBlocks(): Promise<boolean> {
+    let connection;
+
+    try {
+      connection = await DB.getConnection();
+      const [lastBlocks] = await connection.query(`SELECT height, hash, previous_block_hash FROM blocks ORDER BY height DESC LIMIT 10`);
+      connection.release();
+
+      for (let i = 0; i < lastBlocks.length - 1; ++i) {
+        if (lastBlocks[i].previous_block_hash !== lastBlocks[i + 1].hash) {
+          logger.notice(`Chain divergence detected at block ${lastBlocks[i].height}, re-indexing most recent data`);
+          return false;
+        }
+      }
+
+      return true;
+    } catch (e) {
+      connection.release();
+
+      return true; // Don't do anything if there is a db error
+    }
+  }
+
+  /**
+   * Delete $count blocks from the database
+   */
+  public async $deleteBlocks(count: number) {
+    let connection;
+
+    try {
+      connection = await DB.getConnection();
+      logger.debug(`Delete ${count} most recent indexed blocks from the database`);
+      await connection.query(`DELETE FROM blocks ORDER BY height DESC LIMIT ${count};`);
+    } catch (e) {
+      logger.err('$deleteBlocks() error' + (e instanceof Error ? e.message : e));
+    }
+
+    connection.release();
+  }
 }
 
 export default new BlocksRepository();

backend/src/repositories/HashratesRepository.ts

@@ -169,6 +169,9 @@ class HashratesRepository {
     }
   }
 
+  /**
+   * Set latest run timestamp
+   */
   public async $setLatestRunTimestamp(key: string, val: any = null) {
     const connection = await DB.getConnection();
     const query = `UPDATE state SET number = ? WHERE name = ?`;
@@ -181,6 +184,9 @@ class HashratesRepository {
     }
   }
 
+  /**
+   * Get latest run timestamp
+   */
   public async $getLatestRunTimestamp(key: string): Promise<number> {
     const connection = await DB.getConnection();
     const query = `SELECT number FROM state WHERE name = ?`;
@@ -199,6 +205,29 @@ class HashratesRepository {
       throw e;
     }
   }
+
+  /**
+   * Delete most recent data points for re-indexing
+   */
+  public async $deleteLastEntries() {
+    logger.debug(`Delete latest hashrates data points from the database`);
+
+    let connection;
+    try {
+      connection = await DB.getConnection();
+      const [rows] = await connection.query(`SELECT MAX(hashrate_timestamp) as timestamp FROM hashrates GROUP BY type`);
+      for (const row of rows) {
+        await connection.query(`DELETE FROM hashrates WHERE hashrate_timestamp = ?`, [row.timestamp]);
+      }
+      // Re-run the hashrate indexing to fill up missing data
+      await this.$setLatestRunTimestamp('last_hashrates_indexing', 0);
+      await this.$setLatestRunTimestamp('last_weekly_hashrates_indexing', 0);
+    } catch (e) {
+      logger.err('$deleteLastEntries() error' + (e instanceof Error ? e.message : e));
+    }
+
+    connection.release();
+  }
 }
 
 export default new HashratesRepository();