add: backend indexing + reorg handling

backend/src/api/blocks.ts

@@ -40,6 +40,7 @@ import CpfpRepository from '../repositories/CpfpRepository';
 import { parseDATUMTemplateCreator, parseDMNDTemplateCreator } from '../utils/bitcoin-script';
 import database from '../database';
 import { getBlockFirstSeenFromLogs, getOldestLogTimestampFromLogs, scanLogsForBlocksFirstSeen } from '../utils/file-read';
+import FlagValueRepository, { INDEXING_PRESETS } from '../repositories/FlagValueRepository';
 
 class Blocks {
   private blocks: BlockExtended[] = [];
@@ -54,6 +55,8 @@ class Blocks {
   private oldestCoreLogTimestamp: number | undefined | null = undefined;
 
   private mainLoopTimeout: number = 120000;
+  private indexingFlagValues: boolean = false;
+  private flagValuesDeleteQueue: number[]= [];
 
   constructor() { }
 
@@ -689,6 +692,155 @@ class Blocks {
     }
   }
 
+  /**
+   * [INDEXING] Index all blocks flag values for the goggles graph rendering
+   *
+   *  @asyncSafe
+   */
+  public async $generateFlagValuesDatabase(): Promise<void> {
+    const MAX_BLOCKS_PERQUERY = 144;
+    if (this.indexingFlagValues) {
+      return;
+    }
+
+    if (Common.blocksSummariesIndexingEnabled() === false || Common.isLiquid()) {
+      return;
+    }
+
+    this.indexingFlagValues = true;
+
+    const tipOfSummaries = await BlocksSummariesRepository.$getTipIndexed();
+    if (!tipOfSummaries) {
+      this.indexingFlagValues = false;
+      return;
+    }
+
+    let newlyIndexedBuckets = 0;
+
+    while (this.flagValuesDeleteQueue.length > 0) { // Deletion of in-queue heights due to reorg
+      const deletionHeight = this.flagValuesDeleteQueue.shift();
+      if (deletionHeight === undefined) {
+        continue;
+      }
+      await FlagValueRepository.$deleteFlagValuesFromHeight(deletionHeight);
+    }
+
+    for (const preset of INDEXING_PRESETS) {
+      let seedHeight = preset.retentionSpan > -1 ? tipOfSummaries - preset.retentionSpan : 0;
+      if (config.MEMPOOL.INDEXING_BLOCKS_AMOUNT > 0) {
+        seedHeight = Math.max(seedHeight, tipOfSummaries - config.MEMPOOL.INDEXING_BLOCKS_AMOUNT + 1);
+      }
+      const firstBucket = Math.floor((tipOfSummaries + 1) / preset.bucketSize) * preset.bucketSize - preset.bucketSize;
+      const lastBucket = Math.max(0, Math.floor(seedHeight / preset.bucketSize) * preset.bucketSize);
+
+      // Deletion of flag values out of retention span
+      const tipAndTailOfFlagValues = await FlagValueRepository.$getTipAndTailIndexedByBucketSize(preset.bucketSize);
+      if (tipAndTailOfFlagValues && lastBucket > tipAndTailOfFlagValues.tail) { // Drop buckets that fell out of block span
+        logger.debug(`Deleting all the flag values ${preset.name} below height #${lastBucket}`, logger.tags.goggles);
+        await FlagValueRepository.$deleteFlagValuesBelowHeight(lastBucket, preset.bucketSize);
+      }
+
+      if (firstBucket < lastBucket) {
+        continue; // no complete bucket in range
+      }
+
+      const indexedBuckets = await FlagValueRepository.$getIndexedStartHeights(preset.bucketSize, firstBucket, lastBucket);
+      const isBucketIndexed = {};
+      // We map the buckets that are already indexed to skip them
+      for (const startHeight of indexedBuckets) {
+        isBucketIndexed[startHeight] = true;
+      }
+
+      logger.debug(`Processing and indexing flag values from #${firstBucket} to #${lastBucket} ${preset.name}`, logger.tags.goggles);
+
+      let timer = Date.now() / 1000;
+      const startedAt = Date.now() / 1000;
+      let blocksComputedInTotal = 0;
+      let blocksComputedThisRun = 0;
+      const blocksToCompute = firstBucket + preset.bucketSize - lastBucket - (indexedBuckets.length * preset.bucketSize);
+      for (let bucketStart = firstBucket; bucketStart >= lastBucket; bucketStart -= preset.bucketSize) {
+        if (isBucketIndexed[bucketStart]) {
+          continue; // already indexed
+        }
+        try {
+          const bucketFirstHeight = bucketStart + preset.bucketSize - 1;
+          const bucketLastHeight = bucketStart - 1;
+
+          let step = bucketFirstHeight;
+
+          const dataPerFlag: Record<string, Record<string, number>> = {};
+          let sumTimestamps = 0;
+          let nBlocks = 0;
+          let incomplete = false;
+
+          // Incrementalized logic capped by max blocks per query, not bucket size
+          while (step > bucketLastHeight) {
+            const blocksPerQuery = Math.min(step - bucketLastHeight, MAX_BLOCKS_PERQUERY);
+            const cappedLastHeight = step - blocksPerQuery;
+
+            const blocks = await BlocksSummariesRepository.$getSummariesBetweenHeights(step, cappedLastHeight);
+            await Common.sleep$(250); // Don't query/index flag values too fast
+
+            if (!blocks || blocks.length < blocksPerQuery) {
+              incomplete = true;
+              break; // Incomplete bucket
+            }
+
+            // Flag values processing
+            for (const block of blocks) {
+              const txData = JSON.parse(block.transactions).map((tx) => ({flags: tx.flags, vsize: tx.vsize}));
+              for (const data of txData) {
+                if (dataPerFlag[data.flags] === undefined || Object.keys(dataPerFlag[data.flags]).length === 0) {
+                  dataPerFlag[data.flags] = {
+                    txCount: 0,
+                    vSizeTotal: 0
+                  };
+                }
+                dataPerFlag[data.flags].txCount = dataPerFlag[data.flags].txCount + 1;
+                dataPerFlag[data.flags].vSizeTotal = dataPerFlag[data.flags].vSizeTotal + data.vsize;
+              }
+              sumTimestamps += block.timestamp;
+              blocksComputedInTotal++;
+              blocksComputedThisRun++;
+              nBlocks++;
+            }
+
+            // Logging
+            const elapsedSeconds = (Date.now() / 1000) - timer;
+            if (elapsedSeconds > 5) {
+              const runningFor = (Date.now() / 1000) - startedAt;
+              const blocksPerSecond = blocksComputedThisRun / elapsedSeconds;
+              const completion = (blocksComputedInTotal / blocksToCompute) * 100;
+              logger.debug(`Indexing flag values ${preset.name} | ${blocksComputedInTotal}/${blocksToCompute} (${completion.toFixed(2)}%) | ~${blocksPerSecond.toFixed(2)} blocks/sec | elapsed: ${runningFor.toFixed(2)} seconds`,logger.tags.goggles);
+              timer = Date.now() / 1000;
+              blocksComputedThisRun = 0;
+            }
+
+            step -= blocksPerQuery;
+          }
+
+          if (incomplete) {
+            continue;
+          }
+
+          const avgTimestamp = sumTimestamps / nBlocks;
+          await FlagValueRepository.$saveBatchFlagValues(preset.bucketSize, bucketStart, dataPerFlag, avgTimestamp);
+          nBlocks = 0;
+          newlyIndexedBuckets++;
+        } catch (e) {
+          logger.err(`Failed to index flag values between #${bucketStart} and #${bucketStart + preset.bucketSize - 1}. Reason: ${(e instanceof Error ? e.message : e)}`, logger.tags.goggles);
+        }
+      }
+      logger.debug(`Successfully indexed #${blocksComputedInTotal} blocks ${preset.name} in ${((Date.now() / 1000) - startedAt).toFixed(2)} seconds`, logger.tags.goggles);
+    }
+    if (newlyIndexedBuckets > 0) {
+      logger.notice(`Flag values indexing completed: indexed ${newlyIndexedBuckets} buckets`, logger.tags.goggles);
+    } else {
+      logger.debug(`Flag values indexing completed: indexed ${newlyIndexedBuckets} buckets`, logger.tags.goggles);
+    }
+    this.indexingFlagValues = false;
+  }
+
   /** @asyncUnsafe */
   public async $indexBlockSummary(hash: string, height: number, stale?: boolean): Promise<void> {
     if (config.MEMPOOL.BACKEND === 'esplora') {
@@ -1409,6 +1561,7 @@ class Blocks {
       await DifficultyAdjustmentsRepository.$deleteAdjustementsFromHeight(forkTail.height);
       await cpfpRepository.$deleteClustersFrom(forkTail.height);
       await AccelerationRepository.$deleteAccelerationsFrom(forkTail.height);
+      this.flagValuesDeleteQueue.push(forkTail.height);
       chainTips.clearOrphanCacheAboveHeight(forkTail.height);
       this.updateTimerProgress(timer, `deleted stale block data`);
 
@@ -1769,7 +1922,7 @@ class Blocks {
     if (transactions?.length != null) {
       const { cpfpSummary } = await detectTemplateAlgorithm(height, transactions, [], true);
 
-      if (!stale) {
+      if (!stale && Common.cpfpIndexingEnabled() === true) {
         await this.$saveCpfp(hash, height, cpfpSummary);
       }
 

backend/src/indexer.ts

@@ -226,6 +226,8 @@ class Indexer {
       await AccelerationRepository.$indexPastAccelerations();
       await BlocksAuditsRepository.$migrateAuditsV0toV1();
       await BlocksRepository.$migrateBlocks();
+
+      void blocks.$generateFlagValuesDatabase();
       // do not wait for classify blocks to finish
       void blocks.$classifyBlocks();
       runSuccessful = true;

backend/src/repositories/BlocksSummariesRepository.ts

@@ -216,6 +216,34 @@ class BlocksSummariesRepository {
     }
     return false;
   }
+
+  /** @asyncSafe */
+  public async $getTipIndexed(): Promise<number | null> {
+    if (!Common.blocksSummariesIndexingEnabled()) {
+      return null;
+    }
+    try {
+      const [row]: any[] = await DB.query('SELECT MAX(height) as tip FROM blocks_summaries WHERE version >= 1');
+
+      if (row !== null && row.length > 0) {
+        return row[0].tip;
+      }
+    } catch (e) {
+      logger.err(`Cannot get latest block summary. Reason: ` + (e instanceof Error ? e.message : e));
+    }
+    return null;
+  }
+
+  public async $getSummariesBetweenHeights(startHeight: number, latestHeight: number): Promise<{height: number, transactions: string, timestamp: number}[]> {
+    try {
+      const [rows]: any[] = await DB.query(`SELECT bs.height, bs.transactions, UNIX_TIMESTAMP(b.blockTimestamp) as timestamp FROM blocks_summaries bs JOIN blocks b ON bs.id = b.hash WHERE bs.height <= ? AND bs.height > ? AND b.stale = 0 AND bs.version >= 1 ORDER BY height DESC`, [startHeight, latestHeight]);
+
+      return rows;
+    } catch (e) {
+      logger.err(`Cannot get blocks between ${startHeight} and ${latestHeight}. Reason: ` + (e instanceof Error ? e.message : e));
+      throw e;
+    }
+  }
 }
 
 export default new BlocksSummariesRepository();