diff --git a/backend/src/api/audit.ts b/backend/src/api/audit.ts index 7b90c516e..cc1201919 100644 --- a/backend/src/api/audit.ts +++ b/backend/src/api/audit.ts @@ -6,11 +6,28 @@ import transactionUtils from './transaction-utils'; const PROPAGATION_MARGIN = 180; // in seconds, time since a transaction is first seen after which it is assumed to have propagated to all miners +export interface AuditResult { + unseen: string[]; + censored: string[]; + added: string[]; + prioritized: string[]; + fresh: string[]; + sigop: string[]; + fullrbf: string[]; + accelerated: string[]; + matchRate: number; + similarity: number; +} + class Audit { - auditBlock(height: number, transactions: MempoolTransactionExtended[], projectedBlocks: MempoolBlockWithTransactions[], mempool: { [txId: string]: MempoolTransactionExtended }) - : { unseen: string[], censored: string[], added: string[], prioritized: string[], fresh: string[], sigop: string[], fullrbf: string[], accelerated: string[], score: number, similarity: number } { + auditBlock( + height: number, + transactions: MempoolTransactionExtended[], + projectedBlocks: MempoolBlockWithTransactions[], + mempool: { [txId: string]: MempoolTransactionExtended } + ): AuditResult { if (!projectedBlocks?.[0]?.transactionIds || !mempool) { - return { unseen: [], censored: [], added: [], prioritized: [], fresh: [], sigop: [], fullrbf: [], accelerated: [], score: 1, similarity: 1 }; + return { unseen: [], censored: [], added: [], prioritized: [], fresh: [], sigop: [], fullrbf: [], accelerated: [], matchRate: 100, similarity: 1 }; } const matches: string[] = []; // present in both mined block and template @@ -176,6 +193,8 @@ class Audit { } const similarity = projectedWeight ? matchedWeight / projectedWeight : 1; + const matchRate = Math.round(score * 100 * 100) / 100; + return { unseen, censored: Object.keys(isCensored), @@ -185,7 +204,7 @@ class Audit { sigop: [], fullrbf: rbf, accelerated, - score, + matchRate, similarity, }; } diff --git a/backend/src/api/block-processor.ts b/backend/src/api/block-processor.ts new file mode 100644 index 000000000..7d8ce4f68 --- /dev/null +++ b/backend/src/api/block-processor.ts @@ -0,0 +1,242 @@ +import config from '../config'; +import logger from '../logger'; +import { + BlockExtended, + BlockSummary, + PoolTag, + MempoolTransactionExtended, + CpfpSummary, + TemplateAlgorithm, + MempoolBlockWithTransactions, +} from '../mempool.interfaces'; +import { IEsploraApi } from './bitcoin/esplora-api.interface'; +import { Acceleration } from './services/acceleration'; +import { calculateGoodBlockCpfp, calculateClusterMempoolBlockCpfp, calculateFastBlockCpfp } from './cpfp'; +import mempoolBlocks from './mempool-blocks'; +import memPool from './mempool'; +import Audit, { AuditResult } from './audit'; +import blocks from './blocks'; +import transactionUtils from './transaction-utils'; +import { ClusterMempool } from '../cluster-mempool/cluster-mempool'; +import { Common } from './common'; +import accelerationApi from './services/acceleration'; + +interface ProcessedAudit extends AuditResult { + expectedFees: number; + expectedWeight: number; + projectedBlocks: MempoolBlockWithTransactions[]; +} + +export interface BlockProcessingResult { + templateAlgorithm: TemplateAlgorithm; + cpfpSummary: CpfpSummary; + blockExtended: BlockExtended; + blockSummary: BlockSummary; + auditResult?: ProcessedAudit; +} + +const CM_ACTIVATION_HEIGHT: { [network: string]: number } = { + 'mainnet': 940000, + 'testnet': 4860000, + 'testnet4': 125000, + 'signet': 294000, + 'regtest': 0, +}; + +class BlockProcessor { + + /** @asyncUnsafe */ + public async $processNewBlock( + block: IEsploraApi.Block, + transactions: MempoolTransactionExtended[], + pool: PoolTag, + accelerations: Record + ): Promise { + const poolAccelerations = Object.values(accelerations) + .filter(a => a.pools.includes(pool.uniqueId)) + .map(a => ({ txid: a.txid, max_bid: a.feeDelta })); + + const { templateAlgorithm, cpfpSummary } = detectTemplateAlgorithm( + block.height, + transactions, + poolAccelerations + ); + + logger.debug(`Block #${block.height} detected template algorithm: ${templateAlgorithm === TemplateAlgorithm.clusterMempool ? 'cluster mempool' : 'legacy GBT'}`); + + const blockExtended = await blocks.$getBlockExtended(block, cpfpSummary.transactions, pool); + const blockSummary = blocks.summarizeBlockTransactions(block.id, block.height, cpfpSummary.transactions); + + let auditResult: ProcessedAudit | undefined; + if (config.MEMPOOL.AUDIT && memPool.isInSync()) { + auditResult = await this.$runAudit( + blockExtended, + transactions, + templateAlgorithm, + pool, + accelerations + ); + + if (blockExtended.extras) { + blockExtended.extras.matchRate = auditResult.matchRate; + blockExtended.extras.expectedFees = auditResult.expectedFees; + blockExtended.extras.expectedWeight = auditResult.expectedWeight; + blockExtended.extras.similarity = auditResult.similarity; + } + } else if (blockExtended.extras) { + const mBlocks = mempoolBlocks.getMempoolBlocksWithTransactions(); + if (mBlocks?.length && mBlocks[0].transactions) { + blockExtended.extras.similarity = Common.getSimilarity(mBlocks[0], transactions); + } + } + + return { + templateAlgorithm, + cpfpSummary, + blockExtended, + blockSummary, + auditResult, + }; + } + + private async $runAudit( + block: BlockExtended, + transactions: MempoolTransactionExtended[], + templateAlgorithm: TemplateAlgorithm, + pool: PoolTag, + accelerations: Record + ): Promise { + const auditMempool = memPool.getMempool(); + const isAccelerated = accelerationApi.isAcceleratedBlock(block, Object.values(accelerations)); + + const candidateTxs = memPool.getMempoolCandidates(); + const candidates = (memPool.limitGBT && candidateTxs) + ? { txs: candidateTxs, added: [], removed: [] } + : undefined; + const transactionIds = (memPool.limitGBT) + ? Object.keys(candidates?.txs || {}) + : Object.keys(auditMempool); + + let projectedBlocks: MempoolBlockWithTransactions[]; + + if (templateAlgorithm === TemplateAlgorithm.clusterMempool) { + const clusterMempool = memPool.clusterMempool ?? new ClusterMempool(auditMempool, accelerations); + const cmBlocks = clusterMempool.getBlocks(config.MEMPOOL.MEMPOOL_BLOCKS_AMOUNT) ?? []; + projectedBlocks = mempoolBlocks.processClusterMempoolBlocks( + cmBlocks, + auditMempool, + accelerations, + false, + pool.uniqueId + ); + } else if (config.MEMPOOL.RUST_GBT) { + const added = memPool.limitGBT ? (candidates?.added || []) : []; + const removed = memPool.limitGBT ? (candidates?.removed || []) : []; + projectedBlocks = await mempoolBlocks.$rustUpdateBlockTemplates( + transactionIds, + auditMempool, + added, + removed, + candidates, + isAccelerated, + pool.uniqueId, + true + ); + } else { + projectedBlocks = await mempoolBlocks.$makeBlockTemplates( + transactionIds, + auditMempool, + candidates, + false, + isAccelerated, + pool.uniqueId + ); + } + + const auditResult = Audit.auditBlock(block.height, structuredClone(transactions), projectedBlocks, auditMempool); + + const stripped = projectedBlocks[0]?.transactions ? projectedBlocks[0].transactions : []; + + let totalFees = 0; + let totalWeight = 0; + for (const tx of stripped) { + totalFees += tx.fee; + totalWeight += (tx.vsize * 4); + } + + return { + ...auditResult, + expectedFees: totalFees, + expectedWeight: totalWeight, + projectedBlocks, + }; + } +} + +export function detectTemplateAlgorithm( + height: number, + blockTransactions: MempoolTransactionExtended[], + poolAccelerations: { txid: string; max_bid: number }[], + fast: boolean = false +): { templateAlgorithm: TemplateAlgorithm; cpfpSummary: CpfpSummary } { + + const network = config.MEMPOOL.NETWORK || 'mainnet'; + const activationHeight = CM_ACTIVATION_HEIGHT[network] ?? Infinity; + + // always need the legacy CPFP summary + const legacyCpfpSummary = fast ? calculateFastBlockCpfp( + height, + structuredClone(blockTransactions), + ) : calculateGoodBlockCpfp( + height, + structuredClone(blockTransactions), + poolAccelerations + ); + + // assume legacy below the activation height + if (height < activationHeight) { + return { + templateAlgorithm: TemplateAlgorithm.legacy, + cpfpSummary: legacyCpfpSummary, + }; + } + + // calculate single-block CPFP rates for each algorithm + const clusterCpfpSummary = calculateClusterMempoolBlockCpfp( + height, + structuredClone(blockTransactions), + poolAccelerations + ); + const clusterRates = new Map(); + const legacyRates = new Map(); + for (const tx of clusterCpfpSummary.transactions) { + clusterRates.set(tx.txid, tx.effectiveFeePerVsize); + } + for (const tx of legacyCpfpSummary.transactions) { + legacyRates.set(tx.txid, tx.effectiveFeePerVsize); + } + const clusterTxs = blockTransactions.map(tx => ({...tx, effectiveFeePerVsize: clusterRates.get(tx.txid) || tx.effectiveFeePerVsize})); + const legacyTxs = blockTransactions.map(tx => ({...tx, effectiveFeePerVsize: legacyRates.get(tx.txid) || tx.effectiveFeePerVsize})); + + // identify apparent prioritizations using each algorithm's rates + const clusterPrioritization = transactionUtils.identifyPrioritizedTransactions(clusterTxs, 'effectiveFeePerVsize'); + const legacyPrioritization = transactionUtils.identifyPrioritizedTransactions(legacyTxs, 'effectiveFeePerVsize'); + + // choose the best fitting algorithm (or legacy if tied) + const clusterCount = clusterPrioritization.prioritized.length + clusterPrioritization.deprioritized.length; + const legacyCount = legacyPrioritization.prioritized.length + legacyPrioritization.deprioritized.length; + logger.debug(`Prioritization counts - cluster: ${clusterCount}, legacy: ${legacyCount}`); + if (clusterCount < legacyCount) { + return { + templateAlgorithm: TemplateAlgorithm.clusterMempool, + cpfpSummary: clusterCpfpSummary, + }; + } else { + return { + templateAlgorithm: TemplateAlgorithm.legacy, + cpfpSummary: legacyCpfpSummary, + }; + } +} + +export default new BlockProcessor(); diff --git a/backend/src/api/blocks.ts b/backend/src/api/blocks.ts index 974cfe638..faded1e08 100644 --- a/backend/src/api/blocks.ts +++ b/backend/src/api/blocks.ts @@ -28,9 +28,14 @@ import chainTips from './chain-tips'; import websocketHandler from './websocket-handler'; import redisCache from './redis-cache'; import rbfCache from './rbf-cache'; +import bitcoinSecondClient from './bitcoin/bitcoin-second-client'; +import mempoolBlocks from './mempool-blocks'; +import statistics from './statistics/statistics'; +import { Acceleration } from './services/acceleration'; import { calcBitsDifference } from './difficulty-adjustment'; import AccelerationRepository from '../repositories/AccelerationRepository'; import { calculateFastBlockCpfp, calculateGoodBlockCpfp, calculateClusterMempoolBlockCpfp } from './cpfp'; +import blockProcessor, { BlockProcessingResult, detectTemplateAlgorithm } from './block-processor'; import mempool from './mempool'; import CpfpRepository from '../repositories/CpfpRepository'; import { parseDATUMTemplateCreator } from '../utils/bitcoin-script'; @@ -46,7 +51,6 @@ class Blocks { private previousDifficultyRetarget = 0; private quarterEpochBlockTime: number | null = null; private newBlockCallbacks: ((block: BlockExtended, txIds: string[], transactions: TransactionExtended[]) => void)[] = []; - private newAsyncBlockCallbacks: ((block: BlockExtended, txIds: string[], transactions: MempoolTransactionExtended[]) => Promise)[] = []; private classifyingBlocks: boolean = false; private oldestCoreLogTimestamp: number | undefined | null = undefined; @@ -74,10 +78,6 @@ class Blocks { this.newBlockCallbacks.push(fn); } - public setNewAsyncBlockCallback(fn: (block: BlockExtended, txIds: string[], transactions: MempoolTransactionExtended[]) => Promise) { - this.newAsyncBlockCallbacks.push(fn); - } - /** * Return the list of transaction for a block * @param blockHash @@ -252,7 +252,7 @@ class Blocks { * * @asyncUnsafe */ - private async $getBlockExtended(block: IEsploraApi.Block, transactions: TransactionExtended[]): Promise { + public async $getBlockExtended(block: IEsploraApi.Block, transactions: TransactionExtended[], providedPool?: PoolTag): Promise { const coinbaseTx = transactionUtils.stripCoinbaseTransaction(transactions[0]); const blk: Partial = Object.assign({}, block); @@ -335,7 +335,9 @@ class Blocks { if (['mainnet', 'testnet', 'signet', 'testnet4', 'regtest'].includes(config.MEMPOOL.NETWORK)) { let pool: PoolTag; - if (coinbaseTx !== undefined) { + if (providedPool) { + pool = providedPool; + } else if (coinbaseTx !== undefined) { pool = await this.$findBlockMiner(coinbaseTx); } else { if (config.DATABASE.ENABLED === true) { @@ -386,7 +388,7 @@ class Blocks { return blk; } - private async $getBlockStats(block: IEsploraApi.Block, transactions: TransactionExtended[]): Promise { + public async $getBlockStats(block: IEsploraApi.Block, transactions: TransactionExtended[]): Promise { if (!block.stale) { return bitcoinClient.getBlockStats(block.id); } @@ -496,6 +498,128 @@ class Blocks { } } + /** @asyncUnsafe */ + private async $applyBlockTransactionsToMempool( + txIds: string[], + transactions: MempoolTransactionExtended[] + ): Promise<{ rbfTransactions: { [txid: string]: { replaced: MempoolTransactionExtended[], replacedBy: TransactionExtended }}}> { + const _memPool = memPool.getMempool(); + + const rbfTransactions = Common.findMinedRbfTransactions(transactions, memPool.getSpendMap()); + memPool.handleRbfTransactions(rbfTransactions); + memPool.removeFromSpendMap(transactions); + + if (config.MEMPOOL.CLUSTER_MEMPOOL) { + memPool.clusterMempool?.applyMempoolChange({ + added: [], + removed: txIds, + accelerations: mempool.getAccelerations(), + }); + } + + for (const txId of txIds) { + delete _memPool[txId]; + rbfCache.mined(txId); + } + + let candidates; + let transactionIds: string[]; + + if (memPool.limitGBT) { + const minFeeMempool = await bitcoinSecondClient.getRawMemPool(); + const minFeeTip = await bitcoinSecondClient.getBlockCount(); + candidates = memPool.getNextCandidates(minFeeMempool, minFeeTip, transactions); + transactionIds = Object.keys(candidates?.txs || {}); + } else { + candidates = undefined; + transactionIds = Object.keys(memPool.getMempool()); + } + + if (config.MEMPOOL.CLUSTER_MEMPOOL) { + const cmBlocks = mempool.clusterMempool?.getBlocks(config.MEMPOOL.MEMPOOL_BLOCKS_AMOUNT) ?? []; + mempoolBlocks.processClusterMempoolBlocks(cmBlocks, _memPool, mempool.getAccelerations()); + } else if (config.MEMPOOL.RUST_GBT) { + const added = memPool.limitGBT ? (candidates?.added || []) : []; + const removed = memPool.limitGBT ? (candidates?.removed || []) : transactions; + await mempoolBlocks.$rustUpdateBlockTemplates(transactionIds, _memPool, added, removed, candidates, true); + } else { + await mempoolBlocks.$makeBlockTemplates(transactionIds, _memPool, candidates, true, true); + } + + return { rbfTransactions }; + } + + /** @asyncUnsafe */ + private async $saveBlockData( + processingResult: BlockProcessingResult, + timer: number + ): Promise { + const blockExtended = processingResult.blockExtended; + const cpfpSummary = processingResult.cpfpSummary; + + let latestPriceId; + try { + latestPriceId = await PricesRepository.$getLatestPriceId(); + this.updateTimerProgress(timer, `got latest price id ${this.currentBlockHeight}`); + } catch (e) { + logger.debug('failed to fetch latest price id from db: ' + (e instanceof Error ? e.message : e)); + } + if (priceUpdater.historyInserted === true && latestPriceId !== null) { + await blocksRepository.$saveBlockPrices([{ + height: blockExtended.height, + priceId: latestPriceId, + }]); + this.updateTimerProgress(timer, `saved prices for ${this.currentBlockHeight}`); + } else { + logger.debug(`Cannot save block price for ${blockExtended.height} because the price updater hasnt completed yet. Trying again in 10 seconds.`, logger.tags.mining); + indexer.scheduleSingleTask('blocksPrices', 10000); + } + + if (Common.blocksSummariesIndexingEnabled() === true) { + // indexes the summary as a side effect + await this.$getStrippedBlockTransactions(blockExtended.id, true, false, cpfpSummary, blockExtended.height); + this.updateTimerProgress(timer, `saved block summary for ${this.currentBlockHeight}`); + } + + if (config.MEMPOOL.CPFP_INDEXING) { + // can be slow, and isn't critical, so don't await + void this.$saveCpfp(blockExtended.id, this.currentBlockHeight, cpfpSummary); + this.updateTimerProgress(timer, `saved cpfp for ${this.currentBlockHeight}`); + } + + if (processingResult.auditResult) { + void BlocksSummariesRepository.$saveTemplate({ + height: blockExtended.height, + template: { + id: blockExtended.id, + transactions: processingResult.auditResult.projectedBlocks[0].transactions, + }, + version: 1, + }); + this.updateTimerProgress(timer, `saved audit template for ${this.currentBlockHeight}`); + + void BlocksAuditsRepository.$saveAudit({ + version: 1, + templateAlgorithm: processingResult.templateAlgorithm, + time: blockExtended.timestamp, + height: blockExtended.height, + hash: blockExtended.id, + unseenTxs: processingResult.auditResult.unseen, + addedTxs: processingResult.auditResult.added, + prioritizedTxs: processingResult.auditResult.prioritized, + missingTxs: processingResult.auditResult.censored, + freshTxs: processingResult.auditResult.fresh, + sigopTxs: processingResult.auditResult.sigop, + fullrbfTxs: processingResult.auditResult.fullrbf, + acceleratedTxs: processingResult.auditResult.accelerated, + matchRate: processingResult.auditResult.matchRate, + expectedFees: processingResult.auditResult.expectedFees, + expectedWeight: processingResult.auditResult.expectedWeight, + }); + this.updateTimerProgress(timer, `saved audit results for ${this.currentBlockHeight}`); + } + } + /** * [INDEXING] Index all blocks summaries for the block txs visualization */ @@ -1054,63 +1178,55 @@ class Blocks { } } - let accelerations = Object.values(mempool.getAccelerations()); - if (accelerations?.length > 0) { - const pool = await this.$findBlockMiner(transactionUtils.stripCoinbaseTransaction(transactions[0])); - accelerations = accelerations.filter(a => a.pools.includes(pool.uniqueId)); - } - const cpfpTransactions = structuredClone(transactions); - const accelForCpfp = accelerations.map(a => ({ txid: a.txid, max_bid: a.feeDelta })); - const cpfpSummary: CpfpSummary = config.MEMPOOL.CLUSTER_MEMPOOL - ? calculateClusterMempoolBlockCpfp(block.height, cpfpTransactions, accelForCpfp) - : calculateGoodBlockCpfp(block.height, cpfpTransactions, accelForCpfp); - const blockExtended: BlockExtended = await this.$getBlockExtended(block, cpfpSummary.transactions); - const blockSummary: BlockSummary = this.summarizeBlockTransactions(block.id, block.height, cpfpSummary.transactions); + const pool = await this.$findBlockMiner(transactionUtils.stripCoinbaseTransaction(transactions[0])); + const accelerations = mempool.getAccelerations(); + + const processingResult = await blockProcessor.$processNewBlock( + block, + transactions, + pool, + accelerations + ); + + const blockExtended = processingResult.blockExtended; + const blockSummary = processingResult.blockSummary; + const cpfpSummary = processingResult.cpfpSummary; this.updateTimerProgress(timer, `got block data for ${this.currentBlockHeight}`); - if (Common.indexingEnabled()) { - if (!fastForwarded) { - await this.$handleReorgs(blockExtended, timer); - } + if (config.STATISTICS.ENABLED && config.DATABASE.ENABLED) { + await statistics.runStatistics(); + } + const { rbfTransactions } = await this.$applyBlockTransactionsToMempool(txIds, cpfpSummary.transactions); + this.updateTimerProgress(timer, `applied mempool changes for ${this.currentBlockHeight}`); + + if (config.STATISTICS.ENABLED && config.DATABASE.ENABLED) { + await statistics.runStatistics(); + } + + if (Common.indexingEnabled() && !fastForwarded) { + await this.$handleReorgs(blockExtended, timer); + } + + await websocketHandler.handleNewBlock(blockExtended, txIds, cpfpSummary.transactions, rbfTransactions); + this.updateTimerProgress(timer, `sent websocket updates for ${this.currentBlockHeight}`); + + if (Common.indexingEnabled()) { await blocksRepository.$saveBlockInDatabase(blockExtended); this.updateTimerProgress(timer, `saved ${this.currentBlockHeight} to database`); - if (!fastForwarded) { - let lastestPriceId; - try { - lastestPriceId = await PricesRepository.$getLatestPriceId(); - this.updateTimerProgress(timer, `got latest price id ${this.currentBlockHeight}`); - } catch (e) { - logger.debug('failed to fetch latest price id from db: ' + (e instanceof Error ? e.message : e)); - } - if (priceUpdater.historyInserted === true && lastestPriceId !== null) { - await blocksRepository.$saveBlockPrices([{ - height: blockExtended.height, - priceId: lastestPriceId, - }]); - this.updateTimerProgress(timer, `saved prices for ${this.currentBlockHeight}`); - } else { - logger.debug(`Cannot save block price for ${blockExtended.height} because the price updater hasnt completed yet. Trying again in 10 seconds.`, logger.tags.mining); - indexer.scheduleSingleTask('blocksPrices', 10000); - } + await AccelerationRepository.$indexAccelerationsForBlock( + blockExtended, + Object.values(accelerations), + structuredClone(cpfpSummary.transactions) + ); + this.updateTimerProgress(timer, `indexed accelerations for ${this.currentBlockHeight}`); - // Save blocks summary for visualization if it's enabled - if (Common.blocksSummariesIndexingEnabled() === true) { - await this.$getStrippedBlockTransactions(blockExtended.id, true, false, cpfpSummary, blockExtended.height); - this.updateTimerProgress(timer, `saved block summary for ${this.currentBlockHeight}`); - } - if (config.MEMPOOL.CPFP_INDEXING && !config.MEMPOOL.AUDIT) { - void this.$saveCpfp(blockExtended.id, this.currentBlockHeight, cpfpSummary); - this.updateTimerProgress(timer, `saved cpfp for ${this.currentBlockHeight}`); - } + if (!fastForwarded) { + await this.$saveBlockData(processingResult, timer); } } - // start async callbacks - this.updateTimerProgress(timer, `starting async callbacks for ${this.currentBlockHeight}`); - const callbackPromises = this.newAsyncBlockCallbacks.map((cb) => cb(blockExtended, txIds, cpfpSummary.transactions)); - if (block.height % 2016 === 0) { if (Common.indexingEnabled()) { let adjustment; @@ -1148,11 +1264,6 @@ class Blocks { await chainTips.updateOrphanedBlocks(); } - // wait for pending async callbacks to finish - this.updateTimerProgress(timer, `waiting for async callbacks to complete for ${this.currentBlockHeight}`); - await Promise.all(callbackPromises); - this.updateTimerProgress(timer, `async callbacks completed for ${this.currentBlockHeight}`); - this.blocks.push(blockExtended); if (this.blocks.length > config.MEMPOOL.INITIAL_BLOCKS_AMOUNT * 4) { this.blocks = this.blocks.slice(-config.MEMPOOL.INITIAL_BLOCKS_AMOUNT * 4); @@ -1652,53 +1763,22 @@ class Blocks { } if (transactions?.length != null) { - const algo = await this.detectTemplateAlgorithm(hash, height, transactions); - - let summary: CpfpSummary; - if (algo === TemplateAlgorithm.clusterMempool) { - summary = calculateClusterMempoolBlockCpfp(height, transactions, []); - } else { - summary = calculateFastBlockCpfp(height, transactions); - } + const { cpfpSummary } = await detectTemplateAlgorithm(height, transactions, []); if (!stale) { - await this.$saveCpfp(hash, height, summary); + await this.$saveCpfp(hash, height, cpfpSummary); } - const effectiveFeeStats = Common.calcEffectiveFeeStatistics(summary.transactions); + const effectiveFeeStats = Common.calcEffectiveFeeStatistics(cpfpSummary.transactions); await blocksRepository.$saveEffectiveFeeStats(hash, effectiveFeeStats); - return summary; + return cpfpSummary; } else { logger.err(`Cannot index CPFP for block ${height} - missing transaction data`); return null; } } - private static CM_ACTIVATION_HEIGHT: { [network: string]: number } = { - 'mainnet': 945000, - 'testnet': 4860000, - 'testnet4': 125000, - 'signet': 294000, - 'regtest': 0, - }; - - private async detectTemplateAlgorithm(hash: string, height: number, transactions: MempoolTransactionExtended[]): Promise { - const network = config.MEMPOOL.NETWORK || 'mainnet'; - const activationHeight = Blocks.CM_ACTIVATION_HEIGHT[network] ?? Infinity; - - if (height < activationHeight) { - return TemplateAlgorithm.legacy; - } - - const auditAlgo = await BlocksAuditsRepository.$getBlockTemplateAlgo(hash); - if (auditAlgo !== null) { - return auditAlgo; - } - - return detectAlgorithmHeuristic(transactions); - } - /** @asyncSafe */ public async $saveCpfp(hash: string, height: number, cpfpSummary: CpfpSummary): Promise { try { @@ -1766,116 +1846,4 @@ class Blocks { } } -export function detectAlgorithmHeuristic(transactions: MempoolTransactionExtended[]): TemplateAlgorithm { - const txMap: { [txid: string]: MempoolTransactionExtended } = {}; - for (const tx of transactions) { - txMap[tx.txid] = tx; - } - - const parentMap = new Map>(); - const childMap = new Map>(); - for (const tx of transactions) { - for (const vin of tx.vin) { - if (txMap[vin.txid]) { - let parents = parentMap.get(tx.txid); - if (!parents) { - parents = new Set(); - parentMap.set(tx.txid, parents); - } - parents.add(vin.txid); - let children = childMap.get(vin.txid); - if (!children) { - children = new Set(); - childMap.set(vin.txid, children); - } - children.add(tx.txid); - } - } - } - - const visited = new Set(); - let cmLikely = false; - - for (const tx of transactions) { - if (visited.has(tx.txid)) { - continue; - } - const component = new Set(); - const stack = [tx.txid]; - while (stack.length > 0) { - const current = stack.pop(); - if (current === undefined || visited.has(current)) { - continue; - } - visited.add(current); - component.add(current); - for (const p of (parentMap.get(current) || [])) { - if (!visited.has(p)) { - stack.push(p); - } - } - for (const c of (childMap.get(current) || [])) { - if (!visited.has(c)) { - stack.push(c); - } - } - } - - if (component.size < 3) { - continue; - } - - const leaves: string[] = []; - for (const txid of component) { - const children = childMap.get(txid); - if (!children || children.size === 0) { - leaves.push(txid); - } - } - - if (leaves.length >= 2) { - let totalFee = 0; - let totalWeight = 0; - for (const txid of component) { - totalFee += txMap[txid].fee || 0; - totalWeight += txMap[txid].weight; - } - const aggregateRate = totalWeight > 0 ? totalFee / (totalWeight / 4) : 0; - - let bestSingleRate = 0; - for (const leaf of leaves) { - let pkgFee = txMap[leaf].fee || 0; - let pkgWeight = txMap[leaf].weight; - const ancestorStack = [...(parentMap.get(leaf) || [])]; - const ancestorVisited = new Set([leaf]); - while (ancestorStack.length) { - const anc = ancestorStack.pop(); - if (anc === undefined || ancestorVisited.has(anc)) { - continue; - } - ancestorVisited.add(anc); - pkgFee += txMap[anc].fee || 0; - pkgWeight += txMap[anc].weight; - for (const p of (parentMap.get(anc) || [])) { - if (!ancestorVisited.has(p)) { - ancestorStack.push(p); - } - } - } - const rate = pkgWeight > 0 ? pkgFee / (pkgWeight / 4) : 0; - if (rate > bestSingleRate) { - bestSingleRate = rate; - } - } - - if (aggregateRate > bestSingleRate * 1.01) { - cmLikely = true; - break; - } - } - } - - return cmLikely ? TemplateAlgorithm.clusterMempool : TemplateAlgorithm.legacy; -} - export default new Blocks(); diff --git a/backend/src/api/websocket-handler.ts b/backend/src/api/websocket-handler.ts index 26ea015ce..4fbe3aa09 100644 --- a/backend/src/api/websocket-handler.ts +++ b/backend/src/api/websocket-handler.ts @@ -3,7 +3,7 @@ import * as WebSocket from 'ws'; import { BlockExtended, TransactionExtended, MempoolTransactionExtended, WebsocketResponse, OptimizedStatistic, ILoadingIndicators, GbtCandidates, TxTrackingInfo, - MempoolDelta, MempoolDeltaTxids, TemplateAlgorithm, CpfpInfo + MempoolDelta, MempoolDeltaTxids, CpfpInfo } from '../mempool.interfaces'; import blocks from './blocks'; import memPool from './mempool'; @@ -16,16 +16,12 @@ import transactionUtils from './transaction-utils'; import rbfCache, { ReplacementInfo } from './rbf-cache'; import difficultyAdjustment from './difficulty-adjustment'; import feeApi from './fee-api'; -import BlocksAuditsRepository from '../repositories/BlocksAuditsRepository'; -import BlocksSummariesRepository from '../repositories/BlocksSummariesRepository'; -import Audit from './audit'; import priceUpdater from '../tasks/price-updater'; import { ApiPrice } from '../repositories/PricesRepository'; import { Acceleration } from './services/acceleration'; import accelerationApi from './services/acceleration'; import mempool from './mempool'; import statistics from './statistics/statistics'; -import accelerationRepository from '../repositories/AccelerationRepository'; import bitcoinApi from './bitcoin/bitcoin-api-factory'; import walletApi from './services/wallets'; @@ -35,7 +31,7 @@ interface AddressTransactions { removed: MempoolTransactionExtended[], } import bitcoinSecondClient from './bitcoin/bitcoin-second-client'; -import { calculateMempoolTxCpfp, calculateGoodBlockCpfp, calculateClusterMempoolBlockCpfp } from './cpfp'; +import { calculateMempoolTxCpfp } from './cpfp'; import stratumApi, { StratumJob } from './services/stratum'; // valid 'want' subscriptions @@ -1065,179 +1061,25 @@ class WebsocketHandler { } } - /** @asyncUnsafe */ - async handleNewBlock(block: BlockExtended, txIds: string[], transactions: MempoolTransactionExtended[]): Promise { + /** @asyncSafe */ + async handleNewBlock( + block: BlockExtended, + txIds: string[], + transactions: MempoolTransactionExtended[], + rbfTransactions: { [txid: string]: { replaced: MempoolTransactionExtended[], replacedBy: TransactionExtended }} + ): Promise { if (!this.webSocketServers.length) { throw new Error('No WebSocket.Server have been set'); } - const blockTransactions = structuredClone(transactions); - this.printLogs(); - if (config.STATISTICS.ENABLED && config.DATABASE.ENABLED) { - await statistics.runStatistics(); - } const _memPool = memPool.getMempool(); - const candidateTxs = memPool.getMempoolCandidates(); - let candidates: GbtCandidates | undefined = (memPool.limitGBT && candidateTxs) ? { txs: candidateTxs, added: [], removed: [] } : undefined; - let transactionIds: string[] = (memPool.limitGBT) ? Object.keys(candidates?.txs || {}) : Object.keys(_memPool); - - if (config.DATABASE.ENABLED) { - const accelerations = Object.values(mempool.getAccelerations()); - await accelerationRepository.$indexAccelerationsForBlock(block, accelerations, structuredClone(transactions)); - } - - const rbfTransactions = Common.findMinedRbfTransactions(transactions, memPool.getSpendMap()); - memPool.handleRbfTransactions(rbfTransactions); - memPool.removeFromSpendMap(transactions); - - if (config.MEMPOOL.AUDIT && memPool.isInSync()) { - const auditMempool = _memPool; - const isAccelerated = accelerationApi.isAcceleratedBlock(block, Object.values(mempool.getAccelerations())); - const auditAccelerations = mempool.getAccelerations(); - - let projectedBlocks; - const auditVersion = 1; - let templateAlgorithm = TemplateAlgorithm.legacy; - - if (config.MEMPOOL.CLUSTER_MEMPOOL) { - const cmBlocks = mempoolBlocks.processClusterMempoolBlocks(mempool.clusterMempool?.getBlocks(config.MEMPOOL.MEMPOOL_BLOCKS_AMOUNT) ?? [], auditMempool, auditAccelerations, false, block.extras.pool.id); - const cmAudit = Audit.auditBlock(block.height, blockTransactions, cmBlocks, auditMempool); - - const added = memPool.limitGBT ? (candidates?.added || []) : []; - const removed = memPool.limitGBT ? (candidates?.removed || []) : []; - const legacyBlocks = config.MEMPOOL.RUST_GBT - ? await mempoolBlocks.$rustUpdateBlockTemplates(transactionIds, auditMempool, added, removed, candidates, isAccelerated, block.extras.pool.id, true) - : await mempoolBlocks.$makeBlockTemplates(transactionIds, auditMempool, candidates, false, isAccelerated, block.extras.pool.id); - const legacyAudit = Audit.auditBlock(block.height, blockTransactions, legacyBlocks, auditMempool); - - const SCORE_MARGIN = 0.001; - if (cmAudit.score > legacyAudit.score + SCORE_MARGIN) { - projectedBlocks = cmBlocks; - templateAlgorithm = TemplateAlgorithm.clusterMempool; - } else { - projectedBlocks = legacyBlocks; - } - } else { - if (config.MEMPOOL.RUST_GBT) { - const added = memPool.limitGBT ? (candidates?.added || []) : []; - const removed = memPool.limitGBT ? (candidates?.removed || []) : []; - projectedBlocks = await mempoolBlocks.$rustUpdateBlockTemplates(transactionIds, auditMempool, added, removed, candidates, isAccelerated, block.extras.pool.id); - } else { - projectedBlocks = await mempoolBlocks.$makeBlockTemplates(transactionIds, auditMempool, candidates, false, isAccelerated, block.extras.pool.id); - } - } - - if (Common.indexingEnabled()) { - const { unseen, censored, added, prioritized, fresh, sigop, fullrbf, accelerated, score, similarity } = Audit.auditBlock(block.height, blockTransactions, projectedBlocks, auditMempool); - const matchRate = Math.round(score * 100 * 100) / 100; - - const stripped = projectedBlocks[0]?.transactions ? projectedBlocks[0].transactions : []; - - let totalFees = 0; - let totalWeight = 0; - for (const tx of stripped) { - totalFees += tx.fee; - totalWeight += (tx.vsize * 4); - } - - void BlocksSummariesRepository.$saveTemplate({ - height: block.height, - template: { - id: block.id, - transactions: stripped, - }, - version: auditVersion, - }); - - void BlocksAuditsRepository.$saveAudit({ - version: auditVersion, - templateAlgorithm, - time: block.timestamp, - height: block.height, - hash: block.id, - unseenTxs: unseen, - addedTxs: added, - prioritizedTxs: prioritized, - missingTxs: censored, - freshTxs: fresh, - sigopTxs: sigop, - fullrbfTxs: fullrbf, - acceleratedTxs: accelerated, - matchRate: matchRate, - expectedFees: totalFees, - expectedWeight: totalWeight, - }); - - if (block.extras) { - block.extras.matchRate = matchRate; - block.extras.expectedFees = totalFees; - block.extras.expectedWeight = totalWeight; - block.extras.similarity = similarity; - } - - // Save CPFP data using the algorithm that best matched the mined block - // blockTransactions is already a structuredClone (line above), safe to mutate - if (config.MEMPOOL.CPFP_INDEXING) { - const blockAccelerations = Object.values(mempool.getAccelerations()) - .filter(a => a.pools.includes(block.extras.pool.id)) - .map(a => ({ txid: a.txid, max_bid: a.feeDelta })); - - let cpfpSummary; - if (templateAlgorithm === TemplateAlgorithm.clusterMempool) { - cpfpSummary = calculateClusterMempoolBlockCpfp(block.height, blockTransactions as MempoolTransactionExtended[], blockAccelerations); - } else { - cpfpSummary = calculateGoodBlockCpfp(block.height, blockTransactions as MempoolTransactionExtended[], blockAccelerations); - } - void blocks.$saveCpfp(block.id, block.height, cpfpSummary); - } - } - } else if (block.extras) { - const mBlocks = mempoolBlocks.getMempoolBlocksWithTransactions(); - if (mBlocks?.length && mBlocks[0].transactions) { - block.extras.similarity = Common.getSimilarity(mBlocks[0], transactions); - } - } - const confirmedTxids: { [txid: string]: boolean } = {}; - - if (config.MEMPOOL.CLUSTER_MEMPOOL) { - memPool.clusterMempool?.applyMempoolChange({ - added: [], - removed: txIds, - accelerations: mempool.getAccelerations(), - }); - } - - // Update mempool to remove transactions included in the new block for (const txId of txIds) { - delete _memPool[txId]; - rbfCache.mined(txId); confirmedTxids[txId] = true; } - if (memPool.limitGBT) { - const minFeeMempool = memPool.limitGBT ? await bitcoinSecondClient.getRawMemPool() : null; - const minFeeTip = memPool.limitGBT ? await bitcoinSecondClient.getBlockCount() : -1; - candidates = memPool.getNextCandidates(minFeeMempool, minFeeTip, transactions); - transactionIds = Object.keys(candidates?.txs || {}); - } else { - candidates = undefined; - transactionIds = Object.keys(memPool.getMempool()); - } - - - if (config.MEMPOOL.CLUSTER_MEMPOOL) { - const cmBlocks = mempool.clusterMempool?.getBlocks(config.MEMPOOL.MEMPOOL_BLOCKS_AMOUNT) ?? []; - mempoolBlocks.processClusterMempoolBlocks(cmBlocks, _memPool, mempool.getAccelerations()); - } else if (config.MEMPOOL.RUST_GBT) { - const added = memPool.limitGBT ? (candidates?.added || []) : []; - const removed = memPool.limitGBT ? (candidates?.removed || []) : transactions; - await mempoolBlocks.$rustUpdateBlockTemplates(transactionIds, _memPool, added, removed, candidates, true); - } else { - await mempoolBlocks.$makeBlockTemplates(transactionIds, _memPool, candidates, true, true); - } const mBlocks = mempoolBlocks.getMempoolBlocks(); const mBlockDeltas = mempoolBlocks.getMempoolBlockDeltas(); @@ -1502,10 +1344,6 @@ class WebsocketHandler { } }); } - - if (config.STATISTICS.ENABLED && config.DATABASE.ENABLED) { - await statistics.runStatistics(); - } } public handleNewStratumJob(job: StratumJob): void { diff --git a/backend/src/cluster-mempool/cluster-mempool.ts b/backend/src/cluster-mempool/cluster-mempool.ts index fc3859f36..f01abce45 100644 --- a/backend/src/cluster-mempool/cluster-mempool.ts +++ b/backend/src/cluster-mempool/cluster-mempool.ts @@ -667,5 +667,4 @@ export class ClusterMempool { } return null; } - } diff --git a/backend/src/index.ts b/backend/src/index.ts index c2a3d998c..2e7e9a13d 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -334,7 +334,6 @@ class Server { if (config.MEMPOOL.ENABLED) { statistics.setNewStatisticsEntryCallback(websocketHandler.handleNewStatistic.bind(websocketHandler)); memPool.setAsyncMempoolChangedCallback(websocketHandler.$handleMempoolChange.bind(websocketHandler)); - blocks.setNewAsyncBlockCallback(websocketHandler.handleNewBlock.bind(websocketHandler)); } if (config.FIAT_PRICE.ENABLED) { priceUpdater.setRatesChangedCallback(websocketHandler.handleNewConversionRates.bind(websocketHandler));