Merge pull request #2086 from mempool/nymkappa/bugfix/index-blocks-prices-often

Missing blocks prices

backend/src/api/blocks.ts

@@ -22,6 +22,8 @@ import poolsParser from './pools-parser';
 import BlocksSummariesRepository from '../repositories/BlocksSummariesRepository';
 import mining from './mining/mining';
 import DifficultyAdjustmentsRepository from '../repositories/DifficultyAdjustmentsRepository';
+import PricesRepository from '../repositories/PricesRepository';
+import priceUpdater from '../tasks/price-updater';
 
 class Blocks {
   private blocks: BlockExtended[] = [];
@@ -457,6 +459,19 @@ class Blocks {
           }
           await blocksRepository.$saveBlockInDatabase(blockExtended);
 
+          const lastestPriceId = await PricesRepository.$getLatestPriceId();
+          if (priceUpdater.historyInserted === true && lastestPriceId !== null) {
+            await blocksRepository.$saveBlockPrices([{
+              height: blockExtended.height,
+              priceId: lastestPriceId,
+            }]);
+          } else {
+            logger.info(`Cannot save block price for ${blockExtended.height} because the price updater hasnt completed yet. Trying again in 10 seconds.`)
+            setTimeout(() => {
+              indexer.runSingleTask('blocksPrices');
+            }, 10000);
+          }
+
           // Save blocks summary for visualization if it's enabled
           if (Common.blocksSummariesIndexingEnabled() === true) {
             await this.$getStrippedBlockTransactions(blockExtended.id, true);

backend/src/api/mining/mining.ts

@@ -473,7 +473,7 @@ class Mining {
 
       for (const block of blocksWithoutPrices) {
         // Quick optimisation, out mtgox feed only goes back to 2010-07-19 02:00:00, so skip the first 68951 blocks
-        if (block.height < 68951) {
+        if (['mainnet', 'testnet'].includes(config.MEMPOOL.NETWORK) && block.height < 68951) {
           blocksPrices.push({
             height: block.height,
             priceId: prices[0].id,
@@ -492,11 +492,11 @@ class Mining {
 
         if (blocksPrices.length >= 100000) {
           totalInserted += blocksPrices.length;
+          let logStr = `Linking ${blocksPrices.length} blocks to their closest price`;
           if (blocksWithoutPrices.length > 200000) {
-            logger.debug(`Linking ${blocksPrices.length} newly indexed blocks to their closest price | Progress ${Math.round(totalInserted / blocksWithoutPrices.length * 100)}%`);
+            logStr += ` | Progress ${Math.round(totalInserted / blocksWithoutPrices.length * 100)}%`;
-          } else {
-            logger.debug(`Linking ${blocksPrices.length} newly indexed blocks to their closest price`);
           }
+          logger.debug(logStr);
           await BlocksRepository.$saveBlockPrices(blocksPrices);
           blocksPrices.length = 0;
         }
@@ -504,11 +504,11 @@ class Mining {
 
       if (blocksPrices.length > 0) {
         totalInserted += blocksPrices.length;
+        let logStr = `Linking ${blocksPrices.length} blocks to their closest price`;
         if (blocksWithoutPrices.length > 200000) {
-          logger.debug(`Linking ${blocksPrices.length} newly indexed blocks to their closest price | Progress ${Math.round(totalInserted / blocksWithoutPrices.length * 100)}%`);
+          logStr += ` | Progress ${Math.round(totalInserted / blocksWithoutPrices.length * 100)}%`;
-        } else {
-          logger.debug(`Linking ${blocksPrices.length} newly indexed blocks to their closest price`);
         }
+        logger.debug(logStr);
         await BlocksRepository.$saveBlockPrices(blocksPrices);
       }
     } catch (e) {

backend/src/indexer.ts

@@ -6,13 +6,12 @@ import logger from './logger';
 import HashratesRepository from './repositories/HashratesRepository';
 import bitcoinClient from './api/bitcoin/bitcoin-client';
 import priceUpdater from './tasks/price-updater';
+import PricesRepository from './repositories/PricesRepository';
 
 class Indexer {
   runIndexer = true;
   indexerRunning = false;
-
+  tasksRunning: string[] = [];
-  constructor() {
-  }
 
   public reindex() {
     if (Common.indexingEnabled()) {
@@ -20,6 +19,28 @@ class Indexer {
     }
   }
 
+  public async runSingleTask(task: 'blocksPrices') {
+    if (!Common.indexingEnabled()) {
+      return;
+    }
+
+    if (task === 'blocksPrices' && !this.tasksRunning.includes(task)) {
+      this.tasksRunning.push(task);
+      const lastestPriceId = await PricesRepository.$getLatestPriceId();
+      if (priceUpdater.historyInserted === false || lastestPriceId === null) {
+        logger.debug(`Blocks prices indexer is waiting for the price updater to complete`)
+        setTimeout(() => {
+          this.tasksRunning = this.tasksRunning.filter(runningTask => runningTask != task)
+          this.runSingleTask('blocksPrices');
+        }, 10000);
+      } else {
+        logger.debug(`Blocks prices indexer will run now`)
+        await mining.$indexBlockPrices();
+        this.tasksRunning = this.tasksRunning.filter(runningTask => runningTask != task)
+      }
+    }
+  }
+
   public async $run() {
     if (!Common.indexingEnabled() || this.runIndexer === false ||
       this.indexerRunning === true || mempool.hasPriority()
@@ -50,7 +71,7 @@ class Indexer {
         return;
       }
 
-      await mining.$indexBlockPrices();
+      this.runSingleTask('blocksPrices');
       await mining.$indexDifficultyAdjustments();
       await this.$resetHashratesIndexingState(); // TODO - Remove this as it's not efficient
       await mining.$generateNetworkHashrateHistory();

backend/src/repositories/PricesRepository.ts

@@ -27,6 +27,11 @@ class PricesRepository {
     return oldestRow[0] ? oldestRow[0].time : 0;
   }
 
+  public async $getLatestPriceId(): Promise<number | null> {
+    const [oldestRow] = await DB.query(`SELECT id from prices WHERE USD != -1 ORDER BY time DESC LIMIT 1`);
+    return oldestRow[0] ? oldestRow[0].id : null;
+  }
+
   public async $getLatestPriceTime(): Promise<number> {
     const [oldestRow] = await DB.query(`SELECT UNIX_TIMESTAMP(time) as time from prices WHERE USD != -1 ORDER BY time DESC LIMIT 1`);
     return oldestRow[0] ? oldestRow[0].time : 0;

backend/src/tasks/price-updater.ts

@@ -1,4 +1,5 @@
 import * as fs from 'fs';
+import { Common } from '../api/common';
 import config from '../config';
 import logger from '../logger';
 import PricesRepository from '../repositories/PricesRepository';
@@ -34,10 +35,10 @@ export interface Prices {
 }
 
 class PriceUpdater {
-  historyInserted: boolean = false;
+  public historyInserted = false;
-  lastRun: number = 0;
+  lastRun = 0;
-  lastHistoricalRun: number = 0;
+  lastHistoricalRun = 0;
-  running: boolean = false;
+  running = false;
   feeds: PriceFeed[] = [];
   currencies: string[] = ['USD', 'EUR', 'GBP', 'CAD', 'CHF', 'AUD', 'JPY'];
   latestPrices: Prices;