Audit past acceleration data

backend/src/api/acceleration.ts

@@ -238,7 +238,7 @@ class AccelerationCosts {
   private convertToGraphTx(tx: MempoolTransactionExtended): GraphTx {
     return {
       txid: tx.txid,
-      vsize: tx.vsize,
+      vsize: Math.ceil(tx.weight / 4),
       weight: tx.weight,
       fees: {
         base: 0, // dummy
@@ -256,7 +256,7 @@ class AccelerationCosts {
         ancestor: tx.fees.base,
       },
       ancestorcount: 1,
-      ancestorsize: tx.vsize,
+      ancestorsize: Math.ceil(tx.weight / 4),
       ancestors: new Map<string, MempoolTx>(),
       ancestorRate: 0,
       individualRate: 0,
@@ -493,7 +493,7 @@ interface MinerTransaction extends TemplateTransaction {
 * Build a block using an approximation of the transaction selection algorithm from Bitcoin Core
 * (see BlockAssembler in https://github.com/bitcoin/bitcoin/blob/master/src/node/miner.cpp)
 */
-function makeBlockTemplate(candidates: IEsploraApi.Transaction[], accelerations: Acceleration[], maxBlocks: number = 8, weightLimit: number = BLOCK_WEIGHT_UNITS, sigopLimit: number = BLOCK_SIGOPS): TemplateTransaction[] {
+export function makeBlockTemplate(candidates: IEsploraApi.Transaction[], accelerations: Acceleration[], maxBlocks: number = 8, weightLimit: number = BLOCK_WEIGHT_UNITS, sigopLimit: number = BLOCK_SIGOPS): TemplateTransaction[] {
   const auditPool: Map<string, MinerTransaction> = new Map();
   const mempoolArray: MinerTransaction[] = [];
   

backend/src/api/database-migration.ts

@@ -7,7 +7,7 @@ import cpfpRepository from '../repositories/CpfpRepository';
 import { RowDataPacket } from 'mysql2';
 
 class DatabaseMigration {
-  private static currentVersion = 73;
+  private static currentVersion = 74;
   private queryTimeout = 3600_000;
   private statisticsAddedIndexed = false;
   private uniqueLogs: string[] = [];
@@ -619,6 +619,11 @@ class DatabaseMigration {
       this.uniqueLog(logger.notice, `'accelerations' table has been truncated`);
       await this.updateToSchemaVersion(73);
     }
+
+    if (databaseSchemaVersion < 74 && config.MEMPOOL.NETWORK === 'mainnet') {
+      await this.$executeQuery(`INSERT INTO state(name, number) VALUE ('last_acceleration_block', 0);`);
+      await this.updateToSchemaVersion(74);
+    }
   }
 
   /**

backend/src/api/services/acceleration.ts

@@ -7,7 +7,26 @@ export interface Acceleration {
   txid: string,
   feeDelta: number,
   pools: number[],
-}
+};
+
+export interface AccelerationHistory {
+  txid: string,
+  status: string,
+  feePaid: number,
+  added: number,
+  lastUpdated: number,
+  baseFee: number,
+  vsizeFee: number,
+  effectiveFee: number,
+  effectiveVsize: number,
+  feeDelta: number,
+  blockHash: string,
+  blockHeight: number,
+  pools: {
+    pool_unique_id: number,
+    username: string,
+  }[],
+};
 
 class AccelerationApi {
   public async $fetchAccelerations(): Promise<Acceleration[] | null> {
@@ -24,6 +43,27 @@ class AccelerationApi {
     }
   }
 
+  public async $fetchAccelerationHistory(page?: number, status?: string): Promise<AccelerationHistory[] | null> {
+    if (config.MEMPOOL_SERVICES.ACCELERATIONS) {
+      try {
+        const response = await axios.get(`${config.MEMPOOL_SERVICES.API}/accelerator/accelerations/history`, {
+          responseType: 'json',
+          timeout: 10000,
+          params: {
+            page,
+            status,
+          }
+        });
+        return response.data as AccelerationHistory[];
+      } catch (e) {
+        logger.warn('Failed to fetch acceleration history from the mempool services backend: ' + (e instanceof Error ? e.message : e));
+        return null;
+      }
+    } else {
+      return [];
+    }
+  }
+
   public isAcceleratedBlock(block: BlockExtended, accelerations: Acceleration[]): boolean {
     let anyAccelerated = false;
     for (let i = 0; i < accelerations.length && !anyAccelerated; i++) {

backend/src/api/websocket-handler.ts

@@ -24,7 +24,6 @@ import { ApiPrice } from '../repositories/PricesRepository';
 import accelerationApi from './services/acceleration';
 import mempool from './mempool';
 import statistics from './statistics/statistics';
-import accelerationCosts from './acceleration';
 import accelerationRepository from '../repositories/AccelerationRepository';
 import bitcoinApi from './bitcoin/bitcoin-api-factory';
 
@@ -742,25 +741,8 @@ class WebsocketHandler {
 
     const isAccelerated = config.MEMPOOL_SERVICES.ACCELERATIONS && accelerationApi.isAcceleratedBlock(block, Object.values(mempool.getAccelerations()));
 
-
-    if (isAccelerated) {
-      const blockTxs: { [txid: string]: MempoolTransactionExtended } = {};
-      for (const tx of transactions) {
-        blockTxs[tx.txid] = tx;
-      }
     const accelerations = Object.values(mempool.getAccelerations());
-      const boostRate = accelerationCosts.calculateBoostRate(
+    await accelerationRepository.$indexAccelerationsForBlock(block, accelerations, transactions);
-        accelerations.map(acc => ({ txid: acc.txid, max_bid: acc.feeDelta })),
-        transactions
-      );
-      for (const acc of accelerations) {
-        if (blockTxs[acc.txid]) {
-          const tx = blockTxs[acc.txid];
-          const accelerationInfo = accelerationCosts.getAccelerationInfo(tx, boostRate, transactions);
-          accelerationRepository.$saveAcceleration(accelerationInfo, block, block.extras.pool.id);
-        }
-      }
-    }
 
     const rbfTransactions = Common.findMinedRbfTransactions(transactions, memPool.getSpendMap());
     memPool.handleMinedRbfTransactions(rbfTransactions);

backend/src/indexer.ts

@@ -8,6 +8,7 @@ import priceUpdater from './tasks/price-updater';
 import PricesRepository from './repositories/PricesRepository';
 import config from './config';
 import auditReplicator from './replication/AuditReplication';
+import AccelerationRepository from './repositories/AccelerationRepository';
 
 export interface CoreIndex {
   name: string;
@@ -185,6 +186,7 @@ class Indexer {
       await blocks.$generateCPFPDatabase();
       await blocks.$generateAuditStats();
       await auditReplicator.$sync();
+      await AccelerationRepository.$indexPastAccelerations();
       // do not wait for classify blocks to finish
       blocks.$classifyBlocks();
     } catch (e) {

backend/src/repositories/AccelerationRepository.ts

@@ -1,10 +1,16 @@
-import { AccelerationInfo } from '../api/acceleration';
+import { AccelerationInfo, makeBlockTemplate } from '../api/acceleration';
-import { ResultSetHeader, RowDataPacket } from 'mysql2';
+import { RowDataPacket } from 'mysql2';
 import DB from '../database';
 import logger from '../logger';
 import { IEsploraApi } from '../api/bitcoin/esplora-api.interface';
 import { Common } from '../api/common';
 import config from '../config';
+import blocks from '../api/blocks';
+import accelerationApi, { Acceleration } from '../api/services/acceleration';
+import accelerationCosts from '../api/acceleration';
+import bitcoinApi from '../api/bitcoin/bitcoin-api-factory';
+import transactionUtils from '../api/transaction-utils';
+import { BlockExtended, MempoolTransactionExtended } from '../mempool.interfaces';
 
 export interface PublicAcceleration {
   txid: string,
@@ -21,19 +27,15 @@ export interface PublicAcceleration {
 }
 
 class AccelerationRepository {
+  private bidBoostV2Activated = 831580;
+
   public async $saveAcceleration(acceleration: AccelerationInfo, block: IEsploraApi.Block, pool_id: number): Promise<void> {
     try {
       await DB.query(`
         INSERT INTO accelerations(txid, added, height, pool, effective_vsize, effective_fee, boost_rate, boost_cost)
         VALUE (?, FROM_UNIXTIME(?), ?, ?, ?, ?, ?, ?)
         ON DUPLICATE KEY UPDATE
-          added = FROM_UNIXTIME(?),
+          height = ?
-          height = ?,
-          pool = ?,
-          effective_vsize = ?,
-          effective_fee = ?,
-          boost_rate = ?,
-          boost_cost = ?
       `, [
         acceleration.txSummary.txid,
         block.timestamp,
@@ -41,13 +43,9 @@ class AccelerationRepository {
         pool_id,
         acceleration.txSummary.effectiveVsize,
         acceleration.txSummary.effectiveFee,
-        acceleration.targetFeeRate, acceleration.cost,
+        acceleration.targetFeeRate,
-        block.timestamp,
+        acceleration.cost,
         block.height,
-        pool_id,
-        acceleration.txSummary.effectiveVsize,
-        acceleration.txSummary.effectiveFee,
-        acceleration.targetFeeRate, acceleration.cost,
       ]);
     } catch (e: any) {
       logger.err(`Cannot save acceleration (${acceleration.txSummary.txid}) into db. Reason: ` + (e instanceof Error ? e.message : e));
@@ -119,6 +117,167 @@ class AccelerationRepository {
       throw e;
     }
   }
+
+  public async $getLastSyncedHeight(): Promise<number> {
+    try {
+      const [rows] = await DB.query(`
+        SELECT * FROM state
+        WHERE name = 'last_acceleration_block'
+      `);
+      if (rows?.['length']) {
+        return rows[0].number;
+      }
+    } catch (e: any) {
+      logger.err(`Cannot find last acceleration sync height. Reason: ` + (e instanceof Error ? e.message : e));
+    }
+    return 0;
+  }
+
+  private async $setLastSyncedHeight(height: number): Promise<void> {
+    try {
+      await DB.query(`
+        UPDATE state
+        SET number = ?
+        WHERE name = 'last_acceleration_block'
+      `, [height]);
+    } catch (e: any) {
+      logger.err(`Cannot update last acceleration sync height. Reason: ` + (e instanceof Error ? e.message : e));
+    }
+  }
+
+  public async $indexAccelerationsForBlock(block: BlockExtended, accelerations: Acceleration[], transactions: MempoolTransactionExtended[]): Promise<void> {
+    const blockTxs: { [txid: string]: MempoolTransactionExtended } = {};
+    for (const tx of transactions) {
+      blockTxs[tx.txid] = tx;
+    }
+    const successfulAccelerations = accelerations.filter(acc => acc.pools.includes(block.extras.pool.id));
+    let boostRate: number | null = null;
+    for (const acc of successfulAccelerations) {
+      if (boostRate === null) {
+        boostRate = accelerationCosts.calculateBoostRate(
+          accelerations.map(acc => ({ txid: acc.txid, max_bid: acc.feeDelta })),
+          transactions
+        );
+      }
+      if (blockTxs[acc.txid]) {
+        const tx = blockTxs[acc.txid];
+        const accelerationInfo = accelerationCosts.getAccelerationInfo(tx, boostRate, transactions);
+        accelerationInfo.cost = Math.max(0, Math.min(acc.feeDelta, accelerationInfo.cost));
+        this.$saveAcceleration(accelerationInfo, block, block.extras.pool.id);
+      }
+    }
+    const lastSyncedHeight = await this.$getLastSyncedHeight();
+    // if we've missed any blocks, let the indexer catch up from the last synced height on the next run
+    if (block.height === lastSyncedHeight + 1) {
+      await this.$setLastSyncedHeight(block.height);
+    }
+  }
+
+  /**
+   * [INDEXING] Backfill missing acceleration data
+   */
+  async $indexPastAccelerations(): Promise<void> {
+    if (config.MEMPOOL.NETWORK !== 'mainnet' || !config.MEMPOOL_SERVICES.ACCELERATIONS) {
+      // acceleration history disabled
+      return;
+    }
+    const lastSyncedHeight = await this.$getLastSyncedHeight();
+    const currentHeight = blocks.getCurrentBlockHeight();
+    if (currentHeight <= lastSyncedHeight) {
+      // already in sync
+      return;
+    }
+
+    logger.debug(`Fetching accelerations between block ${lastSyncedHeight} and ${currentHeight}`);
+
+    // Fetch accelerations from mempool.space since the last synced block;
+    const accelerationsByBlock = {};
+    const blockHashes = {};
+    let done = false;
+    let page = 1;
+    let count = 0;
+    try {
+      while (!done) {
+        const accelerations = await accelerationApi.$fetchAccelerationHistory(page);
+        page++;
+        if (!accelerations?.length) {
+          done = true;
+          break;
+        }
+        for (const acc of accelerations) {
+          if (acc.status !== 'mined' && acc.status !== 'completed') {
+            continue;
+          }
+          if (!lastSyncedHeight || acc.blockHeight > lastSyncedHeight) {
+            if (!accelerationsByBlock[acc.blockHeight]) {
+              accelerationsByBlock[acc.blockHeight] = [];
+              blockHashes[acc.blockHeight] = acc.blockHash;
+            }
+            accelerationsByBlock[acc.blockHeight].push(acc);
+            count++;
+          } else {
+            done = true;
+          }
+        }
+      }
+    } catch (e) {
+      logger.err(`Failed to fetch full acceleration history. Reason: ` + (e instanceof Error ? e.message : e));
+    }
+
+    logger.debug(`Indexing ${count} accelerations between block ${lastSyncedHeight} and ${currentHeight}`);
+
+    // process accelerated blocks in order
+    const heights = Object.keys(accelerationsByBlock).map(key => parseInt(key)).sort((a,b) => a - b);
+    for (const height of heights) {
+      const accelerations = accelerationsByBlock[height];
+      try {
+        const block = await blocks.$getBlock(blockHashes[height]) as BlockExtended;
+        const transactions = (await bitcoinApi.$getTxsForBlock(blockHashes[height])).map(tx => transactionUtils.extendMempoolTransaction(tx));
+
+        const blockTxs = {};
+        for (const tx of transactions) {
+          blockTxs[tx.txid] = tx;
+        }
+
+        let boostRate = 0;
+        // use Bid Boost V2 if active
+        if (height > this.bidBoostV2Activated) {
+          boostRate = accelerationCosts.calculateBoostRate(
+            accelerations.map(acc => ({ txid: acc.txid, max_bid: acc.feeDelta })),
+            transactions
+          );
+        } else {
+          // default to Bid Boost V1 (median block fee rate)
+          const template = makeBlockTemplate(
+            transactions,
+            accelerations.map(acc => ({ txid: acc.txid, max_bid: acc.feeDelta })),
+            1,
+            Infinity,
+            Infinity
+          );
+          const feeStats = Common.calcEffectiveFeeStatistics(template);
+          boostRate = feeStats.medianFee;
+        }
+        for (const acc of accelerations) {
+          if (blockTxs[acc.txid]) {
+            const tx = blockTxs[acc.txid];
+            const accelerationInfo = accelerationCosts.getAccelerationInfo(tx, boostRate, transactions);
+            accelerationInfo.cost = Math.max(0, Math.min(acc.feeDelta, accelerationInfo.cost));
+            await this.$saveAcceleration(accelerationInfo, block, block.extras.pool.id);
+          }
+        }
+        await this.$setLastSyncedHeight(height);
+      } catch (e) {
+        logger.err(`Failed to process accelerations for block ${height}. Reason: ` + (e instanceof Error ? e.message : e));
+        return;
+      }
+      logger.debug(`Indexed ${accelerations.length} accelerations in block  ${height}`);
+    }
+
+    await this.$setLastSyncedHeight(currentHeight);
+
+    logger.debug(`Indexing accelerations completed`);
+  }
 }
 
 export default new AccelerationRepository();