Implement CPFP reindexing using mini-miner method (not activated)

backend/src/api/blocks.ts

@@ -32,6 +32,7 @@ import { calcBitsDifference } from './difficulty-adjustment';
 import AccelerationRepository from '../repositories/AccelerationRepository';
 import { calculateFastBlockCpfp, calculateGoodBlockCpfp } from './cpfp';
 import mempool from './mempool';
+import CpfpRepository from '../repositories/CpfpRepository';
 
 class Blocks {
   private blocks: BlockExtended[] = [];
@@ -569,8 +570,11 @@ class Blocks {
     const blockchainInfo = await bitcoinClient.getBlockchainInfo();
     const currentBlockHeight = blockchainInfo.blocks;
 
-    const unclassifiedBlocksList = await BlocksSummariesRepository.$getSummariesWithVersion(0);
+    const targetSummaryVersion: number = 1;
-    const unclassifiedTemplatesList = await BlocksSummariesRepository.$getTemplatesWithVersion(0);
+    const targetTemplateVersion: number = 1;
+
+    const unclassifiedBlocksList = await BlocksSummariesRepository.$getSummariesBelowVersion(targetSummaryVersion);
+    const unclassifiedTemplatesList = await BlocksSummariesRepository.$getTemplatesBelowVersion(targetTemplateVersion);
 
     // nothing to do
     if (!unclassifiedBlocksList?.length && !unclassifiedTemplatesList?.length) {
@@ -612,7 +616,15 @@ class Blocks {
           const cpfpSummary = calculateGoodBlockCpfp(height, txs, []);
           // classify
           const { transactions: classifiedTxs } = this.summarizeBlockTransactions(blockHash, cpfpSummary.transactions);
-          await BlocksSummariesRepository.$saveTransactions(height, blockHash, classifiedTxs, 1);
+          await BlocksSummariesRepository.$saveTransactions(height, blockHash, classifiedTxs, 2);
+          if (unclassifiedBlocks[height].version < 2 && targetSummaryVersion === 2) {
+            const cpfpClusters = await CpfpRepository.$getClustersAt(height);
+            if (!cpfpRepository.compareClusters(cpfpClusters, cpfpSummary.clusters)) {
+              // CPFP clusters changed - update the compact_cpfp tables
+              await CpfpRepository.$deleteClustersAt(height);
+              await this.$saveCpfp(blockHash, height, cpfpSummary);
+            }
+          }
           await Common.sleep$(250);
         }
         if (unclassifiedTemplates[height]) {

backend/src/repositories/BlocksSummariesRepository.ts

@@ -114,6 +114,43 @@ class BlocksSummariesRepository {
     return [];
   }
 
+  public async $getSummariesBelowVersion(version: number): Promise<{ height: number, id: string, version: number }[]> {
+    try {
+      const [rows]: any[] = await DB.query(`
+        SELECT
+          height,
+          id,
+          version
+        FROM blocks_summaries
+        WHERE version < ?
+        ORDER BY height DESC;`, [version]);
+      return rows;
+    } catch (e) {
+      logger.err(`Cannot get block summaries below version. Reason: ` + (e instanceof Error ? e.message : e));
+    }
+
+    return [];
+  }
+
+  public async $getTemplatesBelowVersion(version: number): Promise<{ height: number, id: string, version: number }[]> {
+    try {
+      const [rows]: any[] = await DB.query(`
+        SELECT
+          blocks_summaries.height as height,
+          blocks_templates.id as id,
+          blocks_templates.version as version
+        FROM blocks_templates
+        JOIN blocks_summaries ON blocks_templates.id = blocks_summaries.id
+        WHERE blocks_templates.version < ?
+        ORDER BY height DESC;`, [version]);
+      return rows;
+    } catch (e) {
+      logger.err(`Cannot get block summaries below version. Reason: ` + (e instanceof Error ? e.message : e));
+    }
+
+    return [];
+  }
+
   /**
    * Get the fee percentiles if the block has already been indexed, [] otherwise
    * 

backend/src/repositories/CpfpRepository.ts

@@ -91,6 +91,26 @@ class CpfpRepository {
     return;
   }
 
+  public async $getClustersAt(height: number): Promise<CpfpCluster[]> {
+    const [clusterRows]: any = await DB.query(
+      `
+        SELECT *
+        FROM compact_cpfp_clusters
+        WHERE height = ?
+      `,
+      [height]
+    );
+    return clusterRows.map(cluster => {
+      if (cluster?.txs) {
+        cluster.effectiveFeePerVsize = cluster.fee_rate;
+        cluster.txs = this.unpack(cluster.txs);
+        return cluster;
+      } else {
+        return null;
+      }
+    }).filter(cluster => cluster !== null);
+  }
+
   public async $deleteClustersFrom(height: number): Promise<void> {
     logger.info(`Delete newer cpfp clusters from height ${height} from the database`);
     try {
@@ -122,6 +142,37 @@ class CpfpRepository {
     }
   }
 
+  public async $deleteClustersAt(height: number): Promise<void> {
+    logger.info(`Delete cpfp clusters at height ${height} from the database`);
+    try {
+      const [rows] = await DB.query(
+        `
+          SELECT txs, height, root from compact_cpfp_clusters
+          WHERE height = ?
+        `,
+        [height]
+      ) as RowDataPacket[][];
+      if (rows?.length) {
+        for (const clusterToDelete of rows) {
+          const txs = this.unpack(clusterToDelete?.txs);
+          for (const tx of txs) {
+            await transactionRepository.$removeTransaction(tx.txid);
+          }
+        }
+      }
+      await DB.query(
+        `
+          DELETE from compact_cpfp_clusters
+          WHERE height = ?
+        `,
+        [height]
+      );
+    } catch (e: any) {
+      logger.err(`Cannot delete cpfp clusters from db. Reason: ` + (e instanceof Error ? e.message : e));
+      throw e;
+    }
+  }
+
   // insert a dummy row to mark that we've indexed as far as this block
   public async $insertProgressMarker(height: number): Promise<void> {
     try {
@@ -190,6 +241,32 @@ class CpfpRepository {
       return [];
     }
   }
+
+  // returns `true` if two sets of CPFP clusters are deeply identical
+  public compareClusters(clustersA: CpfpCluster[], clustersB: CpfpCluster[]): boolean {
+    if (clustersA.length !== clustersB.length) {
+      return false;
+    }
+
+    clustersA = clustersA.sort((a,b) => a.root.localeCompare(b.root));
+    clustersB = clustersB.sort((a,b) => a.root.localeCompare(b.root));
+
+    for (let i = 0; i < clustersA.length; i++) {
+      if (clustersA[i].root !== clustersB[i].root) {
+        return false;
+      }
+      if (clustersA[i].txs.length !== clustersB[i].txs.length) {
+        return false;
+      }
+      for (let j = 0; j < clustersA[i].txs.length; j++) {
+        if (clustersA[i].txs[j].txid !== clustersB[i].txs[j].txid) {
+          return false;
+        }
+      }
+    }
+
+    return true;
+  }
 }
 
 export default new CpfpRepository();