mirror of
https://github.com/mempool/mempool.git
synced 2026-08-13 12:33:11 +02:00
Integrate cluster mempool into block audits & cpfp indexing
This commit is contained in:
Mononaut
mononaut
parent
1f2e09c569
commit
f4ce7d9c3c
9 changed files with 545 additions and 46 deletions
|
|
@ -2,7 +2,7 @@ import config from '../config';
|
|||
import bitcoinApi, { bitcoinCoreApi } from './bitcoin/bitcoin-api-factory';
|
||||
import logger from '../logger';
|
||||
import memPool from './mempool';
|
||||
import { BlockExtended, BlockExtension, BlockSummary, PoolTag, TransactionExtended, TransactionMinerInfo, CpfpSummary, MempoolTransactionExtended, TransactionClassified, BlockAudit, TransactionAudit } from '../mempool.interfaces';
|
||||
import { BlockExtended, BlockExtension, BlockSummary, PoolTag, TransactionExtended, TransactionMinerInfo, CpfpSummary, MempoolTransactionExtended, TransactionClassified, BlockAudit, TransactionAudit, TemplateAlgorithm } from '../mempool.interfaces';
|
||||
import { Common } from './common';
|
||||
import diskCache from './disk-cache';
|
||||
import transactionUtils from './transaction-utils';
|
||||
|
|
@ -30,7 +30,7 @@ import redisCache from './redis-cache';
|
|||
import rbfCache from './rbf-cache';
|
||||
import { calcBitsDifference } from './difficulty-adjustment';
|
||||
import AccelerationRepository from '../repositories/AccelerationRepository';
|
||||
import { calculateFastBlockCpfp, calculateGoodBlockCpfp } from './cpfp';
|
||||
import { calculateFastBlockCpfp, calculateGoodBlockCpfp, calculateClusterMempoolBlockCpfp } from './cpfp';
|
||||
import mempool from './mempool';
|
||||
import CpfpRepository from '../repositories/CpfpRepository';
|
||||
import { parseDATUMTemplateCreator } from '../utils/bitcoin-script';
|
||||
|
|
@ -1059,7 +1059,11 @@ class Blocks {
|
|||
const pool = await this.$findBlockMiner(transactionUtils.stripCoinbaseTransaction(transactions[0]));
|
||||
accelerations = accelerations.filter(a => a.pools.includes(pool.uniqueId));
|
||||
}
|
||||
const cpfpSummary: CpfpSummary = calculateGoodBlockCpfp(block.height, transactions, accelerations.map(a => ({ txid: a.txid, max_bid: a.feeDelta })));
|
||||
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);
|
||||
this.updateTimerProgress(timer, `got block data for ${this.currentBlockHeight}`);
|
||||
|
|
@ -1096,7 +1100,7 @@ class Blocks {
|
|||
await this.$getStrippedBlockTransactions(blockExtended.id, true, false, cpfpSummary, blockExtended.height);
|
||||
this.updateTimerProgress(timer, `saved block summary for ${this.currentBlockHeight}`);
|
||||
}
|
||||
if (config.MEMPOOL.CPFP_INDEXING) {
|
||||
if (config.MEMPOOL.CPFP_INDEXING && !config.MEMPOOL.AUDIT) {
|
||||
void this.$saveCpfp(blockExtended.id, this.currentBlockHeight, cpfpSummary);
|
||||
this.updateTimerProgress(timer, `saved cpfp for ${this.currentBlockHeight}`);
|
||||
}
|
||||
|
|
@ -1648,7 +1652,14 @@ class Blocks {
|
|||
}
|
||||
|
||||
if (transactions?.length != null) {
|
||||
const summary = calculateFastBlockCpfp(height, transactions);
|
||||
const algo = await this.detectTemplateAlgorithm(hash, height, transactions);
|
||||
|
||||
let summary: CpfpSummary;
|
||||
if (algo === TemplateAlgorithm.clusterMempool) {
|
||||
summary = calculateClusterMempoolBlockCpfp(height, transactions, []);
|
||||
} else {
|
||||
summary = calculateFastBlockCpfp(height, transactions);
|
||||
}
|
||||
|
||||
if (!stale) {
|
||||
await this.$saveCpfp(hash, height, summary);
|
||||
|
|
@ -1664,6 +1675,30 @@ class Blocks {
|
|||
}
|
||||
}
|
||||
|
||||
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<TemplateAlgorithm> {
|
||||
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<void> {
|
||||
try {
|
||||
|
|
@ -1731,4 +1766,116 @@ 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<string, Set<string>>();
|
||||
const childMap = new Map<string, Set<string>>();
|
||||
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<string>();
|
||||
let cmLikely = false;
|
||||
|
||||
for (const tx of transactions) {
|
||||
if (visited.has(tx.txid)) {
|
||||
continue;
|
||||
}
|
||||
const component = new Set<string>();
|
||||
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<string>([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();
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ import cpfpRepository from '../repositories/CpfpRepository';
|
|||
import { RowDataPacket } from 'mysql2';
|
||||
|
||||
class DatabaseMigration {
|
||||
private static currentVersion = 109;
|
||||
private static currentVersion = 111;
|
||||
private queryTimeout = 3600_000;
|
||||
private statisticsAddedIndexed = false;
|
||||
private uniqueLogs: string[] = [];
|
||||
|
|
@ -1246,6 +1246,15 @@ class DatabaseMigration {
|
|||
`);
|
||||
await this.updateToSchemaVersion(109);
|
||||
}
|
||||
if (databaseSchemaVersion < 110 && isBitcoin === true) {
|
||||
await this.$executeQuery('ALTER TABLE `blocks_audits` ADD template_algo TINYINT UNSIGNED NOT NULL DEFAULT 0');
|
||||
await this.updateToSchemaVersion(110);
|
||||
}
|
||||
|
||||
if (databaseSchemaVersion < 111 && isBitcoin === true) {
|
||||
await this.$executeQuery('ALTER TABLE `compact_cpfp_clusters` ADD template_algo TINYINT UNSIGNED NOT NULL DEFAULT 0');
|
||||
await this.updateToSchemaVersion(111);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -1093,16 +1093,40 @@ class WebsocketHandler {
|
|||
memPool.removeFromSpendMap(transactions);
|
||||
|
||||
if (config.MEMPOOL.AUDIT && memPool.isInSync()) {
|
||||
let projectedBlocks;
|
||||
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);
|
||||
|
||||
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);
|
||||
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 {
|
||||
projectedBlocks = await mempoolBlocks.$makeBlockTemplates(transactionIds, auditMempool, candidates, false, isAccelerated, block.extras.pool.id);
|
||||
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()) {
|
||||
|
|
@ -1124,11 +1148,12 @@ class WebsocketHandler {
|
|||
id: block.id,
|
||||
transactions: stripped,
|
||||
},
|
||||
version: 1,
|
||||
version: auditVersion,
|
||||
});
|
||||
|
||||
void BlocksAuditsRepository.$saveAudit({
|
||||
version: 1,
|
||||
version: auditVersion,
|
||||
templateAlgorithm,
|
||||
time: block.timestamp,
|
||||
height: block.height,
|
||||
hash: block.id,
|
||||
|
|
@ -1151,6 +1176,22 @@ class WebsocketHandler {
|
|||
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();
|
||||
|
|
@ -1161,6 +1202,14 @@ class WebsocketHandler {
|
|||
|
||||
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];
|
||||
|
|
@ -1179,7 +1228,10 @@ class WebsocketHandler {
|
|||
}
|
||||
|
||||
|
||||
if (config.MEMPOOL.RUST_GBT) {
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -276,6 +276,7 @@ export class ClusterMempool {
|
|||
): void {
|
||||
const txid = clusterTx.txid;
|
||||
if (!this.mempool[txid]) {
|
||||
logger.warn(`ClusterMempool.writeBackTx: ${txid} missing from mempool (cluster ${cluster.id})`);
|
||||
return;
|
||||
}
|
||||
const tx = this.mempool[txid];
|
||||
|
|
@ -308,6 +309,8 @@ export class ClusterMempool {
|
|||
const mempoolTx = this.mempool[rel.txid];
|
||||
if (mempoolTx) {
|
||||
relatives.push({ txid: rel.txid, fee: mempoolTx.fee, weight: mempoolTx.weight });
|
||||
} else {
|
||||
logger.warn(`ClusterMempool.getChunkRelatives: ${rel.txid} missing from mempool`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -315,11 +318,16 @@ export class ClusterMempool {
|
|||
}
|
||||
|
||||
private processRemovals(removed: string[]): void {
|
||||
for (const tx of removed.map(txid => this.mempool[txid]).filter(tx => !!tx)) {
|
||||
for (const vin of tx.vin) {
|
||||
if (!vin.is_coinbase) {
|
||||
this.spentBy.delete(`${vin.txid}:${vin.vout}`);
|
||||
for (const txid of removed) {
|
||||
const tx = this.mempool[txid];
|
||||
if (tx) {
|
||||
for (const vin of tx.vin) {
|
||||
if (!vin.is_coinbase) {
|
||||
this.spentBy.delete(`${vin.txid}:${vin.vout}`);
|
||||
}
|
||||
}
|
||||
} else if (this.txToCluster.has(txid)) {
|
||||
logger.warn(`ClusterMempool.processRemovals: ${txid} missing from mempool, spentBy cleanup skipped`);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -626,6 +634,8 @@ export class ClusterMempool {
|
|||
}
|
||||
const mempoolTx = this.mempool[clusterTx.txid];
|
||||
txs.push({ txid: clusterTx.txid, fee: mempoolTx.fee, weight: mempoolTx.weight, parents });
|
||||
} else {
|
||||
logger.warn(`ClusterMempool.buildClusterData: ${clusterTx.txid} missing from mempool (cluster ${cluster.id})`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -50,6 +50,7 @@ export interface BlockAudit {
|
|||
expectedFees?: number,
|
||||
expectedWeight?: number,
|
||||
template?: any[];
|
||||
templateAlgorithm?: TemplateAlgorithm,
|
||||
}
|
||||
|
||||
export interface TransactionAudit {
|
||||
|
|
@ -396,6 +397,25 @@ export interface CpfpCluster {
|
|||
height: number,
|
||||
txs: Ancestor[],
|
||||
effectiveFeePerVsize: number,
|
||||
templateAlgorithm?: TemplateAlgorithm,
|
||||
clusterData?: CpfpClusterData,
|
||||
}
|
||||
|
||||
export interface CpfpClusterTx {
|
||||
txid: string;
|
||||
fee: number;
|
||||
weight: number;
|
||||
parents: number[];
|
||||
}
|
||||
|
||||
export interface CpfpClusterChunk {
|
||||
txs: number[];
|
||||
feerate: number;
|
||||
}
|
||||
|
||||
export interface CpfpClusterData {
|
||||
txs: CpfpClusterTx[];
|
||||
chunks: CpfpClusterChunk[];
|
||||
}
|
||||
|
||||
export interface CpfpSummary {
|
||||
|
|
|
|||
|
|
@ -114,6 +114,7 @@ class AuditReplication {
|
|||
});
|
||||
await blocksAuditsRepository.$saveAudit({
|
||||
version: auditSummary.version || 0,
|
||||
templateAlgorithm: auditSummary.templateAlgorithm ?? 0,
|
||||
hash: blockHash,
|
||||
height: auditSummary.height,
|
||||
time: auditSummary.timestamp || auditSummary.time,
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import DB from '../database';
|
||||
import logger from '../logger';
|
||||
import bitcoinApi from '../api/bitcoin/bitcoin-api-factory';
|
||||
import { BlockAudit, AuditScore, TransactionAudit, TransactionStripped } from '../mempool.interfaces';
|
||||
import { BlockAudit, AuditScore, TransactionAudit, TransactionStripped, TemplateAlgorithm } from '../mempool.interfaces';
|
||||
|
||||
interface MigrationAudit {
|
||||
version: number,
|
||||
|
|
@ -18,8 +18,8 @@ class BlocksAuditRepositories {
|
|||
/** @asyncSafe */
|
||||
public async $saveAudit(audit: BlockAudit): Promise<void> {
|
||||
try {
|
||||
await DB.query(`INSERT INTO blocks_audits(version, time, height, hash, unseen_txs, missing_txs, added_txs, prioritized_txs, fresh_txs, sigop_txs, fullrbf_txs, accelerated_txs, match_rate, expected_fees, expected_weight)
|
||||
VALUE (?, FROM_UNIXTIME(?), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, [audit.version, audit.time, audit.height, audit.hash, JSON.stringify(audit.unseenTxs), JSON.stringify(audit.missingTxs),
|
||||
await DB.query(`INSERT INTO blocks_audits(version, template_algo, time, height, hash, unseen_txs, missing_txs, added_txs, prioritized_txs, fresh_txs, sigop_txs, fullrbf_txs, accelerated_txs, match_rate, expected_fees, expected_weight)
|
||||
VALUE (?, ?, FROM_UNIXTIME(?), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, [audit.version, audit.templateAlgorithm ?? 0, audit.time, audit.height, audit.hash, JSON.stringify(audit.unseenTxs), JSON.stringify(audit.missingTxs),
|
||||
JSON.stringify(audit.addedTxs), JSON.stringify(audit.prioritizedTxs), JSON.stringify(audit.freshTxs), JSON.stringify(audit.sigopTxs), JSON.stringify(audit.fullrbfTxs), JSON.stringify(audit.acceleratedTxs), audit.matchRate, audit.expectedFees, audit.expectedWeight]);
|
||||
} catch (e: any) {
|
||||
if (e.errno === 1062) { // ER_DUP_ENTRY - This scenario is possible upon node backend restart
|
||||
|
|
@ -80,6 +80,7 @@ class BlocksAuditRepositories {
|
|||
const [rows]: any[] = await DB.query(
|
||||
`SELECT
|
||||
blocks_audits.version,
|
||||
blocks_audits.template_algo as templateAlgorithm,
|
||||
blocks_audits.height,
|
||||
blocks_audits.hash as id,
|
||||
UNIX_TIMESTAMP(blocks_audits.time) as timestamp,
|
||||
|
|
@ -120,6 +121,23 @@ class BlocksAuditRepositories {
|
|||
}
|
||||
}
|
||||
|
||||
/** @asyncSafe */
|
||||
public async $getBlockTemplateAlgo(hash: string): Promise<TemplateAlgorithm | null> {
|
||||
try {
|
||||
const [rows]: any[] = await DB.query(
|
||||
`SELECT template_algo FROM blocks_audits WHERE hash = ?`,
|
||||
[hash]
|
||||
);
|
||||
if (rows.length) {
|
||||
return rows[0].template_algo as TemplateAlgorithm;
|
||||
}
|
||||
return null;
|
||||
} catch (e: any) {
|
||||
logger.err(`Cannot fetch block template algo from db. Reason: ` + (e instanceof Error ? e.message : e));
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/** @asyncSafe */
|
||||
public async $getBlockTxAudit(hash: string, txid: string): Promise<TransactionAudit | null> {
|
||||
try {
|
||||
|
|
|
|||
|
|
@ -1,31 +1,47 @@
|
|||
import { RowDataPacket } from 'mysql2';
|
||||
import DB from '../database';
|
||||
import logger from '../logger';
|
||||
import { Ancestor, CpfpCluster } from '../mempool.interfaces';
|
||||
import { Ancestor, CpfpCluster, CpfpClusterData, CpfpClusterTx, TemplateAlgorithm } from '../mempool.interfaces';
|
||||
import transactionRepository from '../repositories/TransactionRepository';
|
||||
|
||||
class CpfpRepository {
|
||||
public async $batchSaveClusters(clusters: { root: string, height: number, txs: Ancestor[], effectiveFeePerVsize: number }[]): Promise<boolean> {
|
||||
public async $batchSaveClusters(clusters: CpfpCluster[]): Promise<boolean> {
|
||||
try {
|
||||
const clusterValues: [string, number, Buffer, number][] = [];
|
||||
const clusterValues: [string, number, Buffer, number, number][] = [];
|
||||
const txs: { txid: string, cluster: string }[] = [];
|
||||
|
||||
for (const cluster of clusters) {
|
||||
if (cluster.txs?.length) {
|
||||
const roundedEffectiveFee = Math.round(cluster.effectiveFeePerVsize * 100) / 100;
|
||||
const equalFee = cluster.txs.length > 1 && cluster.txs.reduce((acc, tx) => {
|
||||
return (acc && Math.round(((tx.fee || 0) / (tx.weight / 4)) * 100) / 100 === roundedEffectiveFee);
|
||||
}, true);
|
||||
if (!equalFee) {
|
||||
const isCM = cluster.templateAlgorithm === TemplateAlgorithm.clusterMempool;
|
||||
|
||||
if (isCM && cluster.clusterData) {
|
||||
clusterValues.push([
|
||||
cluster.root,
|
||||
cluster.height,
|
||||
Buffer.from(this.pack(cluster.txs)),
|
||||
cluster.effectiveFeePerVsize
|
||||
Buffer.from(this.packCM(cluster.clusterData)),
|
||||
0,
|
||||
TemplateAlgorithm.clusterMempool,
|
||||
]);
|
||||
for (const tx of cluster.txs) {
|
||||
for (const tx of cluster.clusterData.txs) {
|
||||
txs.push({ txid: tx.txid, cluster: cluster.root });
|
||||
}
|
||||
} else {
|
||||
const roundedEffectiveFee = Math.round(cluster.effectiveFeePerVsize * 100) / 100;
|
||||
const equalFee = cluster.txs.length > 1 && cluster.txs.reduce((acc, tx) => {
|
||||
return (acc && Math.round(((tx.fee || 0) / (tx.weight / 4)) * 100) / 100 === roundedEffectiveFee);
|
||||
}, true);
|
||||
if (!equalFee) {
|
||||
clusterValues.push([
|
||||
cluster.root,
|
||||
cluster.height,
|
||||
Buffer.from(this.pack(cluster.txs)),
|
||||
cluster.effectiveFeePerVsize,
|
||||
TemplateAlgorithm.legacy,
|
||||
]);
|
||||
for (const tx of cluster.txs) {
|
||||
txs.push({ txid: tx.txid, cluster: cluster.root });
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -42,11 +58,11 @@ class CpfpRepository {
|
|||
while (chunkIndex < clusterValues.length) {
|
||||
const chunk = clusterValues.slice(chunkIndex, chunkIndex + maxChunk);
|
||||
let query = `
|
||||
INSERT IGNORE INTO compact_cpfp_clusters(root, height, txs, fee_rate)
|
||||
INSERT IGNORE INTO compact_cpfp_clusters(root, height, txs, fee_rate, template_algo)
|
||||
VALUES
|
||||
`;
|
||||
query += chunk.map(chunk => {
|
||||
return (' (UNHEX(?), ?, ?, ?)');
|
||||
return (' (UNHEX(?), ?, ?, ?, ?)');
|
||||
}) + ';';
|
||||
const values = chunk.flat();
|
||||
queries.push({
|
||||
|
|
@ -86,8 +102,16 @@ class CpfpRepository {
|
|||
);
|
||||
const cluster = clusterRows[0];
|
||||
if (cluster?.txs) {
|
||||
cluster.effectiveFeePerVsize = cluster.fee_rate;
|
||||
cluster.txs = this.unpack(cluster.txs);
|
||||
if (cluster.template_algo === TemplateAlgorithm.clusterMempool) {
|
||||
cluster.templateAlgorithm = TemplateAlgorithm.clusterMempool;
|
||||
cluster.clusterData = this.unpackCM(cluster.txs);
|
||||
cluster.txs = cluster.clusterData.txs.map(tx => ({ txid: tx.txid, weight: tx.weight, fee: tx.fee }));
|
||||
cluster.effectiveFeePerVsize = 0;
|
||||
} else {
|
||||
cluster.templateAlgorithm = TemplateAlgorithm.legacy;
|
||||
cluster.effectiveFeePerVsize = cluster.fee_rate;
|
||||
cluster.txs = this.unpack(cluster.txs);
|
||||
}
|
||||
return cluster;
|
||||
}
|
||||
return;
|
||||
|
|
@ -105,8 +129,16 @@ class CpfpRepository {
|
|||
);
|
||||
return clusterRows.map(cluster => {
|
||||
if (cluster?.txs) {
|
||||
cluster.effectiveFeePerVsize = cluster.fee_rate;
|
||||
cluster.txs = this.unpack(cluster.txs);
|
||||
if (cluster.template_algo === TemplateAlgorithm.clusterMempool) {
|
||||
cluster.templateAlgorithm = TemplateAlgorithm.clusterMempool;
|
||||
cluster.clusterData = this.unpackCM(cluster.txs);
|
||||
cluster.txs = cluster.clusterData.txs.map(tx => ({ txid: tx.txid, weight: tx.weight, fee: tx.fee }));
|
||||
cluster.effectiveFeePerVsize = 0;
|
||||
} else {
|
||||
cluster.templateAlgorithm = TemplateAlgorithm.legacy;
|
||||
cluster.effectiveFeePerVsize = cluster.fee_rate;
|
||||
cluster.txs = this.unpack(cluster.txs);
|
||||
}
|
||||
return cluster;
|
||||
} else {
|
||||
return null;
|
||||
|
|
@ -119,16 +151,16 @@ class CpfpRepository {
|
|||
try {
|
||||
const [rows] = await DB.query(
|
||||
`
|
||||
SELECT txs, height, root from compact_cpfp_clusters
|
||||
SELECT txs, height, root, template_algo 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);
|
||||
const txids = this.extractTxids(clusterToDelete);
|
||||
for (const txid of txids) {
|
||||
await transactionRepository.$removeTransaction(txid);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -150,16 +182,16 @@ class CpfpRepository {
|
|||
try {
|
||||
const [rows] = await DB.query(
|
||||
`
|
||||
SELECT txs, height, root from compact_cpfp_clusters
|
||||
SELECT txs, height, root, template_algo 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);
|
||||
const txids = this.extractTxids(clusterToDelete);
|
||||
for (const txid of txids) {
|
||||
await transactionRepository.$removeTransaction(txid);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -176,6 +208,13 @@ class CpfpRepository {
|
|||
}
|
||||
}
|
||||
|
||||
private extractTxids(row: any): string[] {
|
||||
if (row.template_algo === TemplateAlgorithm.clusterMempool) {
|
||||
return this.unpackCM(row.txs).txs.map(tx => tx.txid);
|
||||
}
|
||||
return this.unpack(row.txs).map(tx => tx.txid);
|
||||
}
|
||||
|
||||
// insert a dummy row to mark that we've indexed as far as this block
|
||||
public async $insertProgressMarker(height: number): Promise<void> {
|
||||
try {
|
||||
|
|
@ -245,6 +284,117 @@ class CpfpRepository {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Pack cluster mempool data into binary format:
|
||||
* [num_chunks: uint16]
|
||||
* Per chunk: [num_txs: uint16]
|
||||
* Per tx (in linearization order, grouped by chunk):
|
||||
* [txid: 32 bytes LE] [weight: uint32] [fee: uint64] [num_parents: uint8] [parent_indices: uint8 each]
|
||||
*/
|
||||
public packCM(clusterData: CpfpClusterData): ArrayBuffer {
|
||||
const headerSize = 2;
|
||||
const chunkHeadersSize = clusterData.chunks.length * 2;
|
||||
let txDataSize = 0;
|
||||
for (const tx of clusterData.txs) {
|
||||
txDataSize += 32 + 4 + 8 + 1 + tx.parents.length;
|
||||
}
|
||||
const totalSize = headerSize + chunkHeadersSize + txDataSize;
|
||||
|
||||
const buf = new ArrayBuffer(totalSize);
|
||||
const view = new DataView(buf);
|
||||
let offset = 0;
|
||||
|
||||
view.setUint16(offset, clusterData.chunks.length);
|
||||
offset += 2;
|
||||
|
||||
for (const chunk of clusterData.chunks) {
|
||||
view.setUint16(offset, chunk.txs.length);
|
||||
offset += 2;
|
||||
}
|
||||
|
||||
for (const tx of clusterData.txs) {
|
||||
for (let x = 0; x < 32; x++) {
|
||||
view.setUint8(offset + (31 - x), parseInt(tx.txid.slice(x * 2, (x * 2) + 2), 16));
|
||||
}
|
||||
offset += 32;
|
||||
view.setUint32(offset, tx.weight);
|
||||
offset += 4;
|
||||
view.setBigUint64(offset, BigInt(Math.round(tx.fee)));
|
||||
offset += 8;
|
||||
view.setUint8(offset, tx.parents.length);
|
||||
offset += 1;
|
||||
for (const parentIdx of tx.parents) {
|
||||
view.setUint8(offset, parentIdx);
|
||||
offset += 1;
|
||||
}
|
||||
}
|
||||
|
||||
return buf;
|
||||
}
|
||||
|
||||
public unpackCM(buf: Buffer): CpfpClusterData {
|
||||
if (!buf) {
|
||||
return { txs: [], chunks: [] };
|
||||
}
|
||||
|
||||
try {
|
||||
const arrayBuffer = buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength);
|
||||
const view = new DataView(arrayBuffer);
|
||||
let offset = 0;
|
||||
|
||||
const numChunks = view.getUint16(offset);
|
||||
offset += 2;
|
||||
|
||||
const chunkSizes: number[] = [];
|
||||
for (let i = 0; i < numChunks; i++) {
|
||||
chunkSizes.push(view.getUint16(offset));
|
||||
offset += 2;
|
||||
}
|
||||
|
||||
const txs: CpfpClusterTx[] = [];
|
||||
const chunks: { txs: number[], feerate: number }[] = [];
|
||||
let txIndex = 0;
|
||||
|
||||
for (let chunkIdx = 0; chunkIdx < numChunks; chunkIdx++) {
|
||||
const chunkTxIndices: number[] = [];
|
||||
let chunkFee = 0;
|
||||
let chunkWeight = 0;
|
||||
|
||||
for (let t = 0; t < chunkSizes[chunkIdx]; t++) {
|
||||
const txid = Array.from(new Uint8Array(arrayBuffer, offset, 32)).reverse().map(b => b.toString(16).padStart(2, '0')).join('');
|
||||
offset += 32;
|
||||
const weight = view.getUint32(offset);
|
||||
offset += 4;
|
||||
const fee = Number(view.getBigUint64(offset));
|
||||
offset += 8;
|
||||
const numParents = view.getUint8(offset);
|
||||
offset += 1;
|
||||
const parents: number[] = [];
|
||||
for (let p = 0; p < numParents; p++) {
|
||||
parents.push(view.getUint8(offset));
|
||||
offset += 1;
|
||||
}
|
||||
|
||||
txs.push({ txid, fee, weight, parents });
|
||||
chunkTxIndices.push(txIndex);
|
||||
chunkFee += fee;
|
||||
chunkWeight += weight;
|
||||
txIndex++;
|
||||
}
|
||||
|
||||
chunks.push({
|
||||
txs: chunkTxIndices,
|
||||
feerate: chunkWeight > 0 ? (chunkFee * 4) / chunkWeight : 0,
|
||||
});
|
||||
}
|
||||
|
||||
return { txs, chunks };
|
||||
} catch (e) {
|
||||
logger.warn(`Failed to unpack CM CPFP cluster. Reason: ` + (e instanceof Error ? e.message : e));
|
||||
return { txs: [], chunks: [] };
|
||||
}
|
||||
}
|
||||
|
||||
// returns `true` if two sets of CPFP clusters are deeply identical
|
||||
public compareClusters(clustersA: CpfpCluster[], clustersB: CpfpCluster[]): boolean {
|
||||
if (clustersA.length !== clustersB.length) {
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
import DB from '../database';
|
||||
import logger from '../logger';
|
||||
import { Ancestor, CpfpInfo } from '../mempool.interfaces';
|
||||
import { Ancestor, CpfpCluster, CpfpInfo, TemplateAlgorithm } from '../mempool.interfaces';
|
||||
import cpfpRepository from './CpfpRepository';
|
||||
|
||||
class TransactionRepository {
|
||||
|
|
@ -72,6 +72,9 @@ class TransactionRepository {
|
|||
const clusterId = txRows[0].root.toLowerCase();
|
||||
const cluster = await cpfpRepository.$getCluster(clusterId);
|
||||
if (cluster) {
|
||||
if (cluster.templateAlgorithm === TemplateAlgorithm.clusterMempool && cluster.clusterData) {
|
||||
return this.convertCpfpCM(txid, cluster);
|
||||
}
|
||||
return this.convertCpfp(txid, cluster);
|
||||
}
|
||||
}
|
||||
|
|
@ -116,6 +119,95 @@ class TransactionRepository {
|
|||
effectiveFeePerVsize: cluster.effectiveFeePerVsize,
|
||||
};
|
||||
}
|
||||
|
||||
private convertCpfpCM(txid: string, cluster: CpfpCluster): CpfpInfo {
|
||||
const clusterData = cluster.clusterData;
|
||||
if (!clusterData) {
|
||||
return { ancestors: [], descendants: [], effectiveFeePerVsize: 0 };
|
||||
}
|
||||
|
||||
// Find which chunk this tx belongs to
|
||||
let txFlatIdx = -1;
|
||||
let txChunkIndex = -1;
|
||||
for (let i = 0; i < clusterData.txs.length; i++) {
|
||||
if (clusterData.txs[i].txid === txid) {
|
||||
txFlatIdx = i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// Find the chunk containing this tx
|
||||
for (let chunkIdx = 0; chunkIdx < clusterData.chunks.length; chunkIdx++) {
|
||||
if (clusterData.chunks[chunkIdx].txs.includes(txFlatIdx)) {
|
||||
txChunkIndex = chunkIdx;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// Derive ancestors/descendants from in-chunk depgraph parents
|
||||
// For CM, ancestors are the tx's depgraph parents within the cluster,
|
||||
// descendants are txs that depend on this tx
|
||||
const ancestors: Ancestor[] = [];
|
||||
const descendants: Ancestor[] = [];
|
||||
|
||||
if (txFlatIdx >= 0) {
|
||||
// Build child map
|
||||
const childMap = new Map<number, number[]>();
|
||||
for (let i = 0; i < clusterData.txs.length; i++) {
|
||||
for (const parentIdx of clusterData.txs[i].parents) {
|
||||
let children = childMap.get(parentIdx);
|
||||
if (!children) {
|
||||
children = [];
|
||||
childMap.set(parentIdx, children);
|
||||
}
|
||||
children.push(i);
|
||||
}
|
||||
}
|
||||
|
||||
const ancestorSet = new Set<number>();
|
||||
const stack = [...clusterData.txs[txFlatIdx].parents];
|
||||
while (stack.length) {
|
||||
const idx = stack.pop();
|
||||
if (idx === undefined || ancestorSet.has(idx)) {
|
||||
continue;
|
||||
}
|
||||
ancestorSet.add(idx);
|
||||
stack.push(...clusterData.txs[idx].parents);
|
||||
}
|
||||
|
||||
const descendantSet = new Set<number>();
|
||||
const dStack = [...(childMap.get(txFlatIdx) || [])];
|
||||
while (dStack.length) {
|
||||
const idx = dStack.pop();
|
||||
if (idx === undefined || descendantSet.has(idx)) {
|
||||
continue;
|
||||
}
|
||||
descendantSet.add(idx);
|
||||
dStack.push(...(childMap.get(idx) || []));
|
||||
}
|
||||
|
||||
for (const idx of ancestorSet) {
|
||||
const tx = clusterData.txs[idx];
|
||||
ancestors.push({ txid: tx.txid, weight: tx.weight, fee: tx.fee });
|
||||
}
|
||||
for (const idx of descendantSet) {
|
||||
const tx = clusterData.txs[idx];
|
||||
descendants.push({ txid: tx.txid, weight: tx.weight, fee: tx.fee });
|
||||
}
|
||||
}
|
||||
|
||||
const effectiveFeePerVsize = txChunkIndex >= 0 ? clusterData.chunks[txChunkIndex].feerate : 0;
|
||||
|
||||
return {
|
||||
ancestors,
|
||||
descendants,
|
||||
effectiveFeePerVsize,
|
||||
cluster: {
|
||||
...clusterData,
|
||||
chunkIndex: txChunkIndex,
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
export default new TransactionRepository();
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue