machankura/mempool
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use minfee node to limit gbt input size

Browse Source
This commit is contained in:
Mononaut 2023-02-17 17:54:29 -06:00
parent 5c430e57a4
commit e2d3bb4cc5
No known key found for this signature in database
GPG Key ID: A3F058E41374C04E
6 changed files with 176 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
backend/src/api/mempool-blocks.ts
View File

@ -1,6 +1,6 @@
import { GbtGenerator, GbtResult, ThreadTransaction as RustThreadTransaction, ThreadAcceleration as RustThreadAcceleration } from 'rust-gbt'; import { GbtGenerator, GbtResult, ThreadTransaction as RustThreadTransaction, ThreadAcceleration as RustThreadAcceleration } from 'rust-gbt';
import logger from '../logger'; import logger from '../logger';
import { MempoolBlock, MempoolTransactionExtended, MempoolBlockWithTransactions, MempoolBlockDelta, Ancestor, CompactThreadTransaction, EffectiveFeeStats, PoolTag, TransactionClassified, TransactionCompressed, MempoolDeltaChange } from '../mempool.interfaces'; import { MempoolBlock, MempoolTransactionExtended, MempoolBlockWithTransactions, MempoolBlockDelta, Ancestor, CompactThreadTransaction, EffectiveFeeStats, TransactionClassified, TransactionCompressed, MempoolDeltaChange, GbtCandidates } from '../mempool.interfaces';
import { Common, OnlineFeeStatsCalculator } from './common'; import { Common, OnlineFeeStatsCalculator } from './common';
import config from '../config'; import config from '../config';
import { Worker } from 'worker_threads'; import { Worker } from 'worker_threads';
@ -40,12 +40,9 @@ class MempoolBlocks {
return this.mempoolBlockDeltas; return this.mempoolBlockDeltas;
} }
public updateMempoolBlocks(memPool: { [txid: string]: MempoolTransactionExtended }, saveResults: boolean = false): MempoolBlockWithTransactions[] { public updateMempoolBlocks(transactions: string[], memPool: { [txid: string]: MempoolTransactionExtended }, candidates: GbtCandidates | undefined, saveResults: boolean = false): MempoolBlockWithTransactions[] {
const latestMempool = memPool; const latestMempool = memPool;
const memPoolArray: MempoolTransactionExtended[] = []; const memPoolArray: MempoolTransactionExtended[] = transactions.map(txid => memPool[txid]);
for (const i in latestMempool) {
memPoolArray.push(latestMempool[i]);
}
const start = new Date().getTime(); const start = new Date().getTime();
// Clear bestDescendants & ancestors // Clear bestDescendants & ancestors
@ -207,7 +204,7 @@ class MempoolBlocks {
return mempoolBlockDeltas; return mempoolBlockDeltas;
} }
public async $makeBlockTemplates(newMempool: { [txid: string]: MempoolTransactionExtended }, saveResults: boolean = false, useAccelerations: boolean = false, accelerationPool?: number): Promise<MempoolBlockWithTransactions[]> { public async $makeBlockTemplates(transactions: string[], newMempool: { [txid: string]: MempoolTransactionExtended }, candidates: GbtCandidates | undefined, saveResults: boolean = false, useAccelerations: boolean = false, accelerationPool?: number): Promise<MempoolBlockWithTransactions[]> {
const start = Date.now(); const start = Date.now();
// reset mempool short ids // reset mempool short ids
@ -215,8 +212,9 @@ class MempoolBlocks {
this.resetUids(); this.resetUids();
} }
// set missing short ids // set missing short ids
for (const tx of Object.values(newMempool)) { for (const txid of transactions) {
this.setUid(tx, !saveResults); const tx = newMempool[txid];
this.setUid(tx, false);
} }
const accelerations = useAccelerations ? mempool.getAccelerations() : {}; const accelerations = useAccelerations ? mempool.getAccelerations() : {};
@ -224,7 +222,8 @@ class MempoolBlocks {
// prepare a stripped down version of the mempool with only the minimum necessary data // prepare a stripped down version of the mempool with only the minimum necessary data
// to reduce the overhead of passing this data to the worker thread // to reduce the overhead of passing this data to the worker thread
const strippedMempool: Map<number, CompactThreadTransaction> = new Map(); const strippedMempool: Map<number, CompactThreadTransaction> = new Map();
Object.values(newMempool).forEach(entry => { for (const txid of transactions) {
const entry = newMempool[txid];
if (entry.uid !== null && entry.uid !== undefined) { if (entry.uid !== null && entry.uid !== undefined) {
const stripped = { const stripped = {
uid: entry.uid, uid: entry.uid,
@ -237,7 +236,7 @@ class MempoolBlocks {
}; };
strippedMempool.set(entry.uid, stripped); strippedMempool.set(entry.uid, stripped);
} }
}); }
// (re)initialize tx selection worker thread // (re)initialize tx selection worker thread
if (!this.txSelectionWorker) { if (!this.txSelectionWorker) {
@ -268,7 +267,7 @@ class MempoolBlocks {
// clean up thread error listener // clean up thread error listener
this.txSelectionWorker?.removeListener('error', threadErrorListener); this.txSelectionWorker?.removeListener('error', threadErrorListener);
const processed = this.processBlockTemplates(newMempool, blocks, null, Object.entries(rates), Object.values(clusters), accelerations, accelerationPool, saveResults); const processed = this.processBlockTemplates(newMempool, blocks, null, Object.entries(rates), Object.values(clusters), candidates, accelerations, accelerationPool, saveResults);
logger.debug(`makeBlockTemplates completed in ${(Date.now() - start)/1000} seconds`); logger.debug(`makeBlockTemplates completed in ${(Date.now() - start)/1000} seconds`);
@ -279,10 +278,10 @@ class MempoolBlocks {
return this.mempoolBlocks; return this.mempoolBlocks;
} }
public async $updateBlockTemplates(newMempool: { [txid: string]: MempoolTransactionExtended }, added: MempoolTransactionExtended[], removed: MempoolTransactionExtended[], accelerationDelta: string[] = [], saveResults: boolean = false, useAccelerations: boolean = false): Promise<void> { public async $updateBlockTemplates(transactions: string[], newMempool: { [txid: string]: MempoolTransactionExtended }, added: MempoolTransactionExtended[], removed: MempoolTransactionExtended[], candidates: GbtCandidates | undefined, accelerationDelta: string[] = [], saveResults: boolean = false, useAccelerations: boolean = false): Promise<void> {
if (!this.txSelectionWorker) { if (!this.txSelectionWorker) {
// need to reset the worker // need to reset the worker
await this.$makeBlockTemplates(newMempool, saveResults, useAccelerations); await this.$makeBlockTemplates(transactions, newMempool, candidates, saveResults, useAccelerations);
return; return;
} }
@ -292,7 +291,7 @@ class MempoolBlocks {
const addedAndChanged: MempoolTransactionExtended[] = useAccelerations ? accelerationDelta.map(txid => newMempool[txid]).filter(tx => tx != null).concat(added) : added; const addedAndChanged: MempoolTransactionExtended[] = useAccelerations ? accelerationDelta.map(txid => newMempool[txid]).filter(tx => tx != null).concat(added) : added;
for (const tx of addedAndChanged) { for (const tx of addedAndChanged) {
this.setUid(tx, true); this.setUid(tx, false);
} }
const removedUids = removed.map(tx => this.getUid(tx)).filter(uid => uid != null) as number[]; const removedUids = removed.map(tx => this.getUid(tx)).filter(uid => uid != null) as number[];
@ -328,7 +327,7 @@ class MempoolBlocks {
// clean up thread error listener // clean up thread error listener
this.txSelectionWorker?.removeListener('error', threadErrorListener); this.txSelectionWorker?.removeListener('error', threadErrorListener);
this.processBlockTemplates(newMempool, blocks, null, Object.entries(rates), Object.values(clusters), accelerations, null, saveResults); this.processBlockTemplates(newMempool, blocks, null, Object.entries(rates), Object.values(clusters), candidates, accelerations, null, saveResults);
logger.debug(`updateBlockTemplates completed in ${(Date.now() - start) / 1000} seconds`); logger.debug(`updateBlockTemplates completed in ${(Date.now() - start) / 1000} seconds`);
} catch (e) { } catch (e) {
logger.err('updateBlockTemplates failed. ' + (e instanceof Error ? e.message : e)); logger.err('updateBlockTemplates failed. ' + (e instanceof Error ? e.message : e));
@ -336,23 +335,26 @@ class MempoolBlocks {
} }
private resetRustGbt(): void { private resetRustGbt(): void {
console.log('reset rust gbt');
this.rustInitialized = false; this.rustInitialized = false;
this.rustGbtGenerator = new GbtGenerator(); this.rustGbtGenerator = new GbtGenerator();
} }
public async $rustMakeBlockTemplates(newMempool: { [txid: string]: MempoolTransactionExtended }, saveResults: boolean = false, useAccelerations: boolean = false, accelerationPool?: number): Promise<MempoolBlockWithTransactions[]> { public async $rustMakeBlockTemplates(txids: string[], newMempool: { [txid: string]: MempoolTransactionExtended }, candidates: GbtCandidates | undefined, saveResults: boolean = false, useAccelerations: boolean = false, accelerationPool?: number): Promise<MempoolBlockWithTransactions[]> {
const start = Date.now(); const start = Date.now();
// reset mempool short ids // reset mempool short ids
if (saveResults) { if (saveResults) {
this.resetUids(); this.resetUids();
} }
const transactions = txids.map(txid => newMempool[txid]).filter(tx => tx != null);
// set missing short ids // set missing short ids
for (const tx of Object.values(newMempool)) { for (const tx of transactions) {
this.setUid(tx, !saveResults); this.setUid(tx, false);
} }
// set short ids for transaction inputs // set short ids for transaction inputs
for (const tx of Object.values(newMempool)) { for (const tx of transactions) {
tx.inputs = tx.vin.map(v => this.getUid(newMempool[v.txid])).filter(uid => (uid !== null && uid !== undefined)) as number[]; tx.inputs = tx.vin.map(v => this.getUid(newMempool[v.txid])).filter(uid => (uid !== null && uid !== undefined)) as number[];
} }
@ -369,15 +371,15 @@ class MempoolBlocks {
const rustGbt = saveResults ? this.rustGbtGenerator : new GbtGenerator(); const rustGbt = saveResults ? this.rustGbtGenerator : new GbtGenerator();
try { try {
const { blocks, blockWeights, rates, clusters, overflow } = this.convertNapiResultTxids( const { blocks, blockWeights, rates, clusters, overflow } = this.convertNapiResultTxids(
await rustGbt.make(Object.values(newMempool) as RustThreadTransaction[], convertedAccelerations as RustThreadAcceleration[], this.nextUid), await rustGbt.make(transactions as RustThreadTransaction[], convertedAccelerations as RustThreadAcceleration[], this.nextUid),
); );
if (saveResults) { if (saveResults) {
this.rustInitialized = true; this.rustInitialized = true;
} }
const mempoolSize = Object.keys(newMempool).length; const expectedSize = transactions.length;
const resultMempoolSize = blocks.reduce((total, block) => total + block.length, 0) + overflow.length; const resultMempoolSize = blocks.reduce((total, block) => total + block.length, 0) + overflow.length;
logger.debug(`RUST updateBlockTemplates returned ${resultMempoolSize} txs out of ${mempoolSize} in the mempool, ${overflow.length} were unmineable`); logger.debug(`RUST updateBlockTemplates returned ${resultMempoolSize} txs out of ${expectedSize} in the mempool, ${overflow.length} were unmineable`);
const processed = this.processBlockTemplates(newMempool, blocks, blockWeights, rates, clusters, accelerations, accelerationPool, saveResults); const processed = this.processBlockTemplates(newMempool, blocks, blockWeights, rates, clusters, candidates, accelerations, accelerationPool, saveResults);
logger.debug(`RUST makeBlockTemplates completed in ${(Date.now() - start)/1000} seconds`); logger.debug(`RUST makeBlockTemplates completed in ${(Date.now() - start)/1000} seconds`);
return processed; return processed;
} catch (e) { } catch (e) {
@ -389,26 +391,26 @@ class MempoolBlocks {
return this.mempoolBlocks; return this.mempoolBlocks;
} }
public async $oneOffRustBlockTemplates(newMempool: { [txid: string]: MempoolTransactionExtended }, useAccelerations: boolean, accelerationPool?: number): Promise<MempoolBlockWithTransactions[]> { public async $oneOffRustBlockTemplates(transactions: string[], newMempool: { [txid: string]: MempoolTransactionExtended }, candidates: GbtCandidates | undefined, useAccelerations: boolean, accelerationPool?: number): Promise<MempoolBlockWithTransactions[]> {
return this.$rustMakeBlockTemplates(newMempool, false, useAccelerations, accelerationPool); return this.$rustMakeBlockTemplates(transactions, newMempool, candidates, false, useAccelerations, accelerationPool);
} }
public async $rustUpdateBlockTemplates(newMempool: { [txid: string]: MempoolTransactionExtended }, mempoolSize: number, added: MempoolTransactionExtended[], removed: MempoolTransactionExtended[], useAccelerations: boolean, accelerationPool?: number): Promise<MempoolBlockWithTransactions[]> { public async $rustUpdateBlockTemplates(transactions: string[], newMempool: { [txid: string]: MempoolTransactionExtended }, added: MempoolTransactionExtended[], removed: MempoolTransactionExtended[], candidates: GbtCandidates | undefined, useAccelerations: boolean, accelerationPool?: number): Promise<MempoolBlockWithTransactions[]> {
// GBT optimization requires that uids never get too sparse // GBT optimization requires that uids never get too sparse
// as a sanity check, we should also explicitly prevent uint32 uid overflow // as a sanity check, we should also explicitly prevent uint32 uid overflow
if (this.nextUid + added.length >= Math.min(Math.max(262144, 2 * mempoolSize), MAX_UINT32)) { if (this.nextUid + added.length >= Math.min(Math.max(262144, 2 * transactions.length), MAX_UINT32)) {
this.resetRustGbt(); this.resetRustGbt();
} }
if (!this.rustInitialized) { if (!this.rustInitialized) {
// need to reset the worker // need to reset the worker
return this.$rustMakeBlockTemplates(newMempool, true, useAccelerations, accelerationPool); return this.$rustMakeBlockTemplates(transactions, newMempool, candidates, true, useAccelerations, accelerationPool);
} }
const start = Date.now(); const start = Date.now();
// set missing short ids // set missing short ids
for (const tx of added) { for (const tx of added) {
this.setUid(tx, true); this.setUid(tx, false);
} }
// set short ids for transaction inputs // set short ids for transaction inputs
for (const tx of added) { for (const tx of added) {
@ -416,6 +418,8 @@ class MempoolBlocks {
} }
const removedUids = removed.map(tx => this.getUid(tx)).filter(uid => (uid !== null && uid !== undefined)) as number[]; const removedUids = removed.map(tx => this.getUid(tx)).filter(uid => (uid !== null && uid !== undefined)) as number[];
console.log(`removing ${removed.length} (${removedUids.length} with uids)`);
const accelerations = useAccelerations ? mempool.getAccelerations() : {}; const accelerations = useAccelerations ? mempool.getAccelerations() : {};
const acceleratedList = accelerationPool ? Object.values(accelerations).filter(acc => newMempool[acc.txid] && acc.pools.includes(accelerationPool)) : Object.values(accelerations).filter(acc => newMempool[acc.txid]); const acceleratedList = accelerationPool ? Object.values(accelerations).filter(acc => newMempool[acc.txid] && acc.pools.includes(accelerationPool)) : Object.values(accelerations).filter(acc => newMempool[acc.txid]);
const convertedAccelerations = acceleratedList.map(acc => { const convertedAccelerations = acceleratedList.map(acc => {
@ -425,6 +429,8 @@ class MempoolBlocks {
}; };
}); });
console.log(`${acceleratedList.length} accelerations`);
// run the block construction algorithm in a separate thread, and wait for a result // run the block construction algorithm in a separate thread, and wait for a result
try { try {
const { blocks, blockWeights, rates, clusters, overflow } = this.convertNapiResultTxids( const { blocks, blockWeights, rates, clusters, overflow } = this.convertNapiResultTxids(
@ -436,11 +442,11 @@ class MempoolBlocks {
), ),
); );
const resultMempoolSize = blocks.reduce((total, block) => total + block.length, 0) + overflow.length; const resultMempoolSize = blocks.reduce((total, block) => total + block.length, 0) + overflow.length;
logger.debug(`RUST updateBlockTemplates returned ${resultMempoolSize} txs out of ${mempoolSize} in the mempool, ${overflow.length} were unmineable`); logger.debug(`RUST updateBlockTemplates returned ${resultMempoolSize} txs out of ${transactions.length} candidates, ${overflow.length} were unmineable`);
if (mempoolSize !== resultMempoolSize) { if (transactions.length !== resultMempoolSize) {
throw new Error('GBT returned wrong number of transactions , cache is probably out of sync'); throw new Error(`GBT returned wrong number of transactions ${transactions.length} vs ${resultMempoolSize}, cache is probably out of sync`);
} else { } else {
const processed = this.processBlockTemplates(newMempool, blocks, blockWeights, rates, clusters, accelerations, accelerationPool, true); const processed = this.processBlockTemplates(newMempool, blocks, blockWeights, rates, clusters, candidates, accelerations, accelerationPool, true);
this.removeUids(removedUids); this.removeUids(removedUids);
logger.debug(`RUST updateBlockTemplates completed in ${(Date.now() - start)/1000} seconds`); logger.debug(`RUST updateBlockTemplates completed in ${(Date.now() - start)/1000} seconds`);
return processed; return processed;
@ -452,7 +458,7 @@ class MempoolBlocks {
} }
} }
private processBlockTemplates(mempool: { [txid: string]: MempoolTransactionExtended }, blocks: string[][], blockWeights: number[] | null, rates: [string, number][], clusters: string[][], accelerations, accelerationPool, saveResults): MempoolBlockWithTransactions[] { private processBlockTemplates(mempool: { [txid: string]: MempoolTransactionExtended }, blocks: string[][], blockWeights: number[] | null, rates: [string, number][], clusters: string[][], candidates: GbtCandidates | undefined, accelerations, accelerationPool, saveResults): MempoolBlockWithTransactions[] {
for (const [txid, rate] of rates) { for (const [txid, rate] of rates) {
if (txid in mempool) { if (txid in mempool) {
mempool[txid].cpfpDirty = (rate !== mempool[txid].effectiveFeePerVsize); mempool[txid].cpfpDirty = (rate !== mempool[txid].effectiveFeePerVsize);
@ -486,6 +492,9 @@ class MempoolBlocks {
if (txid === memberTxid) { if (txid === memberTxid) {
matched = true; matched = true;
} else { } else {
if (!mempool[txid]) {
console.log('txid missing from mempool! ', txid, candidates?.txs[txid]);
}
const relative = { const relative = {
txid: txid, txid: txid,
fee: mempool[txid].fee, fee: mempool[txid].fee,
@ -518,6 +527,16 @@ class MempoolBlocks {
let totalWeight = 0; let totalWeight = 0;
let totalFees = 0; let totalFees = 0;
const transactions: MempoolTransactionExtended[] = []; const transactions: MempoolTransactionExtended[] = [];
// backfill purged transactions
if (candidates?.txs && blockIndex === blocks.length - 1) {
for (const txid of Object.keys(mempool)) {
if (!candidates.txs[txid]) {
block.push(txid);
}
}
}
for (const txid of block) { for (const txid of block) {
if (txid) { if (txid) {
mempoolTx = mempool[txid]; mempoolTx = mempool[txid];

70
backend/src/api/mempool.ts
View File

@ -1,6 +1,6 @@
import config from '../config'; import config from '../config';
import bitcoinApi from './bitcoin/bitcoin-api-factory'; import bitcoinApi from './bitcoin/bitcoin-api-factory';
import { MempoolTransactionExtended, TransactionExtended, VbytesPerSecond } from '../mempool.interfaces'; import { MempoolTransactionExtended, TransactionExtended, VbytesPerSecond, GbtCandidates } from '../mempool.interfaces';
import logger from '../logger'; import logger from '../logger';
import { Common } from './common'; import { Common } from './common';
import transactionUtils from './transaction-utils'; import transactionUtils from './transaction-utils';
@ -11,18 +11,21 @@ import bitcoinSecondClient from './bitcoin/bitcoin-second-client';
import rbfCache from './rbf-cache'; import rbfCache from './rbf-cache';
import { Acceleration } from './services/acceleration'; import { Acceleration } from './services/acceleration';
import redisCache from './redis-cache'; import redisCache from './redis-cache';
import blocks from './blocks';
class Mempool { class Mempool {
private inSync: boolean = false; private inSync: boolean = false;
private mempoolCacheDelta: number = -1; private mempoolCacheDelta: number = -1;
private mempoolCache: { [txId: string]: MempoolTransactionExtended } = {}; private mempoolCache: { [txId: string]: MempoolTransactionExtended } = {};
private mempoolCandidates: { [txid: string ]: boolean } = {};
private minFeeMempool: { [txId: string]: boolean } = {};
private spendMap = new Map<string, MempoolTransactionExtended>(); private spendMap = new Map<string, MempoolTransactionExtended>();
private mempoolInfo: IBitcoinApi.MempoolInfo = { loaded: false, size: 0, bytes: 0, usage: 0, total_fee: 0, private mempoolInfo: IBitcoinApi.MempoolInfo = { loaded: false, size: 0, bytes: 0, usage: 0, total_fee: 0,
maxmempool: 300000000, mempoolminfee: Common.isLiquid() ? 0.00000100 : 0.00001000, minrelaytxfee: Common.isLiquid() ? 0.00000100 : 0.00001000 }; maxmempool: 300000000, mempoolminfee: Common.isLiquid() ? 0.00000100 : 0.00001000, minrelaytxfee: Common.isLiquid() ? 0.00000100 : 0.00001000 };
private mempoolChangedCallback: ((newMempool: {[txId: string]: MempoolTransactionExtended; }, newTransactions: MempoolTransactionExtended[], private mempoolChangedCallback: ((newMempool: {[txId: string]: MempoolTransactionExtended; }, newTransactions: MempoolTransactionExtended[],
deletedTransactions: MempoolTransactionExtended[], accelerationDelta: string[]) => void) | undefined; deletedTransactions: MempoolTransactionExtended[], accelerationDelta: string[]) => void) | undefined;
private $asyncMempoolChangedCallback: ((newMempool: {[txId: string]: MempoolTransactionExtended; }, mempoolSize: number, newTransactions: MempoolTransactionExtended[], private $asyncMempoolChangedCallback: ((newMempool: {[txId: string]: MempoolTransactionExtended; }, mempoolSize: number, newTransactions: MempoolTransactionExtended[],
deletedTransactions: MempoolTransactionExtended[], accelerationDelta: string[]) => Promise<void>) | undefined; deletedTransactions: MempoolTransactionExtended[], accelerationDelta: string[], candidates?: GbtCandidates) => Promise<void>) | undefined;
private accelerations: { [txId: string]: Acceleration } = {}; private accelerations: { [txId: string]: Acceleration } = {};
@ -40,6 +43,8 @@ class Mempool {
private missingTxCount = 0; private missingTxCount = 0;
private mainLoopTimeout: number = 120000; private mainLoopTimeout: number = 120000;
public limitGBT = config.MEMPOOL.USE_SECOND_NODE_FOR_MINFEE && config.MEMPOOL.LIMIT_GBT;
constructor() { constructor() {
setInterval(this.updateTxPerSecond.bind(this), 1000); setInterval(this.updateTxPerSecond.bind(this), 1000);
} }
@ -74,7 +79,8 @@ class Mempool {
} }
public setAsyncMempoolChangedCallback(fn: (newMempool: { [txId: string]: MempoolTransactionExtended; }, mempoolSize: number, public setAsyncMempoolChangedCallback(fn: (newMempool: { [txId: string]: MempoolTransactionExtended; }, mempoolSize: number,
newTransactions: MempoolTransactionExtended[], deletedTransactions: MempoolTransactionExtended[], accelerationDelta: string[]) => Promise<void>): void { newTransactions: MempoolTransactionExtended[], deletedTransactions: MempoolTransactionExtended[], accelerationDelta: string[],
candidates?: GbtCandidates) => Promise<void>): void {
this.$asyncMempoolChangedCallback = fn; this.$asyncMempoolChangedCallback = fn;
} }
@ -117,7 +123,7 @@ class Mempool {
this.mempoolChangedCallback(this.mempoolCache, [], [], []); this.mempoolChangedCallback(this.mempoolCache, [], [], []);
} }
if (this.$asyncMempoolChangedCallback) { if (this.$asyncMempoolChangedCallback) {
await this.$asyncMempoolChangedCallback(this.mempoolCache, count, [], [], []); await this.$asyncMempoolChangedCallback(this.mempoolCache, count, [], [], [], this.limitGBT ? { txs: {}, added: [], removed: [] } : undefined);
} }
this.addToSpendMap(Object.values(this.mempoolCache)); this.addToSpendMap(Object.values(this.mempoolCache));
} }
@ -160,6 +166,10 @@ class Mempool {
return newTransactions; return newTransactions;
} }
public getMempoolCandidates(): { [txid: string]: boolean } {
return this.mempoolCandidates;
}
public async $updateMemPoolInfo() { public async $updateMemPoolInfo() {
this.mempoolInfo = await this.$getMempoolInfo(); this.mempoolInfo = await this.$getMempoolInfo();
} }
@ -189,7 +199,7 @@ class Mempool {
return txTimes; return txTimes;
} }
public async $updateMempool(transactions: string[], accelerations: Acceleration[] | null, pollRate: number): Promise<void> { public async $updateMempool(transactions: string[], accelerations: Acceleration[] | null, minFeeMempool: string[], minFeeTip: number, pollRate: number): Promise<void> {
logger.debug(`Updating mempool...`); logger.debug(`Updating mempool...`);
// warn if this run stalls the main loop for more than 2 minutes // warn if this run stalls the main loop for more than 2 minutes
@ -311,6 +321,8 @@ class Mempool {
}, 1000 * 60 * config.MEMPOOL.CLEAR_PROTECTION_MINUTES); }, 1000 * 60 * config.MEMPOOL.CLEAR_PROTECTION_MINUTES);
} }
const candidates = await this.getNextCandidates(minFeeMempool, minFeeTip);
const deletedTransactions: MempoolTransactionExtended[] = []; const deletedTransactions: MempoolTransactionExtended[] = [];
if (this.mempoolProtection !== 1) { if (this.mempoolProtection !== 1) {
@ -341,12 +353,14 @@ class Mempool {
this.mempoolCacheDelta = Math.abs(transactions.length - newMempoolSize); this.mempoolCacheDelta = Math.abs(transactions.length - newMempoolSize);
const candidatesChanged = candidates?.added?.length || candidates?.removed?.length;
if (this.mempoolChangedCallback && (hasChange || deletedTransactions.length)) { if (this.mempoolChangedCallback && (hasChange || deletedTransactions.length)) {
this.mempoolChangedCallback(this.mempoolCache, newTransactions, deletedTransactions, accelerationDelta); this.mempoolChangedCallback(this.mempoolCache, newTransactions, deletedTransactions, accelerationDelta);
} }
if (this.$asyncMempoolChangedCallback && (hasChange || deletedTransactions.length)) { if (this.$asyncMempoolChangedCallback && (hasChange || deletedTransactions.length || candidatesChanged)) {
this.updateTimerProgress(timer, 'running async mempool callback'); this.updateTimerProgress(timer, 'running async mempool callback');
await this.$asyncMempoolChangedCallback(this.mempoolCache, newMempoolSize, newTransactions, deletedTransactions, accelerationDelta); await this.$asyncMempoolChangedCallback(this.mempoolCache, newMempoolSize, newTransactions, deletedTransactions, accelerationDelta, candidates);
this.updateTimerProgress(timer, 'completed async mempool callback'); this.updateTimerProgress(timer, 'completed async mempool callback');
} }
@ -432,6 +446,48 @@ class Mempool {
} }
} }
public async getNextCandidates(minFeeTransactions: string[], blockHeight: number): Promise<GbtCandidates | undefined> {
if (this.limitGBT) {
const newCandidateTxMap = {};
this.minFeeMempool = {};
for (const txid of minFeeTransactions) {
if (this.mempoolCache[txid]) {
newCandidateTxMap[txid] = true;
}
this.minFeeMempool[txid] = true;
}
const removed: MempoolTransactionExtended[] = [];
const added: MempoolTransactionExtended[] = [];
// don't prematurely remove txs included in a new block
if (blockHeight > blocks.getCurrentBlockHeight()) {
for (const txid of Object.keys(this.mempoolCandidates)) {
newCandidateTxMap[txid] = true;
}
} else {
for (const txid of Object.keys(this.mempoolCandidates)) {
if (!newCandidateTxMap[txid]) {
const tx = this.mempoolCache[txid];
removed.push(tx);
}
}
}
for (const txid of Object.keys(newCandidateTxMap)) {
if (!this.mempoolCandidates[txid]) {
const tx = this.mempoolCache[txid];
added.push(tx);
}
}
this.mempoolCandidates = newCandidateTxMap;
return {
txs: this.mempoolCandidates,
added,
removed
};
}
}
private startTimer() { private startTimer() {
const state: any = { const state: any = {
start: Date.now(), start: Date.now(),

60
backend/src/api/websocket-handler.ts
View File

@ -2,7 +2,7 @@ import logger from '../logger';
import * as WebSocket from 'ws'; import * as WebSocket from 'ws';
import { import {
BlockExtended, TransactionExtended, MempoolTransactionExtended, WebsocketResponse, BlockExtended, TransactionExtended, MempoolTransactionExtended, WebsocketResponse,
OptimizedStatistic, ILoadingIndicators OptimizedStatistic, ILoadingIndicators, GbtCandidates,
} from '../mempool.interfaces'; } from '../mempool.interfaces';
import blocks from './blocks'; import blocks from './blocks';
import memPool from './mempool'; import memPool from './mempool';
@ -32,6 +32,7 @@ interface AddressTransactions {
confirmed: MempoolTransactionExtended[], confirmed: MempoolTransactionExtended[],
removed: MempoolTransactionExtended[], removed: MempoolTransactionExtended[],
} }
import bitcoinSecondClient from './bitcoin/bitcoin-second-client';
// valid 'want' subscriptions // valid 'want' subscriptions
const wantable = [ const wantable = [
@ -436,21 +437,33 @@ class WebsocketHandler {
} }
async $handleMempoolChange(newMempool: { [txid: string]: MempoolTransactionExtended }, mempoolSize: number, async $handleMempoolChange(newMempool: { [txid: string]: MempoolTransactionExtended }, mempoolSize: number,
newTransactions: MempoolTransactionExtended[], deletedTransactions: MempoolTransactionExtended[], accelerationDelta: string[]): Promise<void> { newTransactions: MempoolTransactionExtended[], deletedTransactions: MempoolTransactionExtended[], accelerationDelta: string[],
candidates?: GbtCandidates): Promise<void> {
if (!this.wss) { if (!this.wss) {
throw new Error('WebSocket.Server is not set'); throw new Error('WebSocket.Server is not set');
} }
this.printLogs(); this.printLogs();
const transactionIds = (memPool.limitGBT && candidates) ? Object.keys(candidates?.txs || {}) : Object.keys(newMempool);
let added = newTransactions;
let removed = deletedTransactions;
console.log(`handleMempoolChange: ${newTransactions.length} new txs, ${deletedTransactions.length} removed, ${candidates?.added.length} candidate added, ${candidates?.removed.length} candidate removed`);
console.log(`mempool size ${Object.keys(newMempool).length}, candidates: ${transactionIds.length}`);
if (memPool.limitGBT) {
console.log('GBT on limited mempool...');
added = candidates?.added || [];
removed = candidates?.removed || [];
}
if (config.MEMPOOL.ADVANCED_GBT_MEMPOOL) { if (config.MEMPOOL.ADVANCED_GBT_MEMPOOL) {
if (config.MEMPOOL.RUST_GBT) { if (config.MEMPOOL.RUST_GBT) {
await mempoolBlocks.$rustUpdateBlockTemplates(newMempool, mempoolSize, newTransactions, deletedTransactions, config.MEMPOOL_SERVICES.ACCELERATIONS); await mempoolBlocks.$rustUpdateBlockTemplates(transactionIds, newMempool, added, removed, candidates, config.MEMPOOL_SERVICES.ACCELERATIONS);
} else { } else {
await mempoolBlocks.$updateBlockTemplates(newMempool, newTransactions, deletedTransactions, accelerationDelta, true, config.MEMPOOL_SERVICES.ACCELERATIONS); await mempoolBlocks.$updateBlockTemplates(transactionIds, newMempool, added, removed, candidates, accelerationDelta, true, config.MEMPOOL_SERVICES.ACCELERATIONS);
} }
} else { } else {
mempoolBlocks.updateMempoolBlocks(newMempool, true); mempoolBlocks.updateMempoolBlocks(transactionIds, newMempool, candidates, true);
} }
const mBlocks = mempoolBlocks.getMempoolBlocks(); const mBlocks = mempoolBlocks.getMempoolBlocks();
@ -739,6 +752,9 @@ class WebsocketHandler {
await statistics.runStatistics(); await statistics.runStatistics();
const _memPool = memPool.getMempool(); const _memPool = memPool.getMempool();
const candidateTxs = await memPool.getMempoolCandidates();
let candidates: GbtCandidates | undefined = (memPool.limitGBT && candidateTxs) ? { txs: candidateTxs, added: [], removed: [] } : undefined;
let transactionIds: string[] = (memPool.limitGBT && candidates) ? Object.keys(_memPool) : Object.keys(candidates?.txs || {});
const isAccelerated = config.MEMPOOL_SERVICES.ACCELERATIONS && accelerationApi.isAcceleratedBlock(block, Object.values(mempool.getAccelerations())); const isAccelerated = config.MEMPOOL_SERVICES.ACCELERATIONS && accelerationApi.isAcceleratedBlock(block, Object.values(mempool.getAccelerations()));
@ -759,19 +775,23 @@ class WebsocketHandler {
auditMempool = deepClone(_memPool); auditMempool = deepClone(_memPool);
if (config.MEMPOOL.ADVANCED_GBT_AUDIT) { if (config.MEMPOOL.ADVANCED_GBT_AUDIT) {
if (config.MEMPOOL.RUST_GBT) { if (config.MEMPOOL.RUST_GBT) {
projectedBlocks = await mempoolBlocks.$oneOffRustBlockTemplates(auditMempool, isAccelerated, block.extras.pool.id); projectedBlocks = await mempoolBlocks.$oneOffRustBlockTemplates(transactionIds, auditMempool, candidates, isAccelerated, block.extras.pool.id);
} else { } else {
projectedBlocks = await mempoolBlocks.$makeBlockTemplates(auditMempool, false, isAccelerated, block.extras.pool.id); projectedBlocks = await mempoolBlocks.$makeBlockTemplates(transactionIds, auditMempool, candidates, false, isAccelerated, block.extras.pool.id);
} }
} else { } else {
projectedBlocks = mempoolBlocks.updateMempoolBlocks(auditMempool, false); projectedBlocks = mempoolBlocks.updateMempoolBlocks(transactionIds, auditMempool, candidates, false);
} }
} else { } else {
if ((config.MEMPOOL_SERVICES.ACCELERATIONS)) { if ((config.MEMPOOL_SERVICES.ACCELERATIONS)) {
if (config.MEMPOOL.RUST_GBT) { if (config.MEMPOOL.RUST_GBT) {
projectedBlocks = await mempoolBlocks.$rustUpdateBlockTemplates(auditMempool, Object.keys(auditMempool).length, [], [], isAccelerated, block.extras.pool.id); console.log(`handleNewBlock: ${transactions.length} mined txs, ${candidates?.added.length} candidate added, ${candidates?.removed.length} candidate removed`);
console.log(`mempool size ${Object.keys(auditMempool).length}, candidates: ${transactionIds.length}`);
let added = memPool.limitGBT ? (candidates?.added || []) : [];
let removed = memPool.limitGBT ? (candidates?.removed || []) : [];
projectedBlocks = await mempoolBlocks.$rustUpdateBlockTemplates(transactionIds, auditMempool, added, removed, candidates, isAccelerated, block.extras.pool.id);
} else { } else {
projectedBlocks = await mempoolBlocks.$makeBlockTemplates(auditMempool, false, isAccelerated, block.extras.pool.id); projectedBlocks = await mempoolBlocks.$makeBlockTemplates(transactionIds, auditMempool, candidates, false, isAccelerated, block.extras.pool.id);
} }
} else { } else {
projectedBlocks = mempoolBlocks.getMempoolBlocksWithTransactions(); projectedBlocks = mempoolBlocks.getMempoolBlocksWithTransactions();
@ -838,14 +858,28 @@ class WebsocketHandler {
confirmedTxids[txId] = true; confirmedTxids[txId] = true;
} }
if (memPool.limitGBT) {
const minFeeMempool = memPool.limitGBT ? await bitcoinSecondClient.getRawMemPool() : null;
const minFeeTip = memPool.limitGBT ? await bitcoinSecondClient.getBlockCount() : -1;
candidates = await memPool.getNextCandidates(minFeeMempool, minFeeTip);
transactionIds = Object.keys(candidates?.txs || {});
} else {
candidates = undefined;
transactionIds = Object.keys(memPool.getMempool());
}
if (config.MEMPOOL.ADVANCED_GBT_MEMPOOL) { if (config.MEMPOOL.ADVANCED_GBT_MEMPOOL) {
if (config.MEMPOOL.RUST_GBT) { if (config.MEMPOOL.RUST_GBT) {
await mempoolBlocks.$rustUpdateBlockTemplates(_memPool, Object.keys(_memPool).length, [], transactions, true); console.log(`handleNewBlock (after): ${transactions.length} mined txs, ${candidates?.added.length} candidate added, ${candidates?.removed.length} candidate removed`);
console.log(`mempool size ${Object.keys(_memPool).length}, candidates: ${transactionIds.length}`);
let added = memPool.limitGBT ? (candidates?.added || []) : [];
let removed = memPool.limitGBT ? (candidates?.removed || []) : transactions;
await mempoolBlocks.$rustUpdateBlockTemplates(transactionIds, _memPool, added, removed, candidates, true);
} else { } else {
await mempoolBlocks.$makeBlockTemplates(_memPool, true, config.MEMPOOL_SERVICES.ACCELERATIONS); await mempoolBlocks.$makeBlockTemplates(transactionIds, _memPool, candidates, true, config.MEMPOOL_SERVICES.ACCELERATIONS);
} }
} else { } else {
mempoolBlocks.updateMempoolBlocks(_memPool, true); mempoolBlocks.updateMempoolBlocks(transactionIds, _memPool, candidates, true);
} }
const mBlocks = mempoolBlocks.getMempoolBlocks(); const mBlocks = mempoolBlocks.getMempoolBlocks();
const mBlockDeltas = mempoolBlocks.getMempoolBlockDeltas(); const mBlockDeltas = mempoolBlocks.getMempoolBlockDeltas();

2
backend/src/config.ts
View File

@ -35,6 +35,7 @@ interface IConfig {
ADVANCED_GBT_AUDIT: boolean; ADVANCED_GBT_AUDIT: boolean;
ADVANCED_GBT_MEMPOOL: boolean; ADVANCED_GBT_MEMPOOL: boolean;
RUST_GBT: boolean; RUST_GBT: boolean;
LIMIT_GBT: boolean;
CPFP_INDEXING: boolean; CPFP_INDEXING: boolean;
MAX_BLOCKS_BULK_QUERY: number; MAX_BLOCKS_BULK_QUERY: number;
DISK_CACHE_BLOCK_INTERVAL: number; DISK_CACHE_BLOCK_INTERVAL: number;
@ -197,6 +198,7 @@ const defaults: IConfig = {
'ADVANCED_GBT_AUDIT': false, 'ADVANCED_GBT_AUDIT': false,
'ADVANCED_GBT_MEMPOOL': false, 'ADVANCED_GBT_MEMPOOL': false,
'RUST_GBT': false, 'RUST_GBT': false,
'LIMIT_GBT': false,
'CPFP_INDEXING': false, 'CPFP_INDEXING': false,
'MAX_BLOCKS_BULK_QUERY': 0, 'MAX_BLOCKS_BULK_QUERY': 0,
'DISK_CACHE_BLOCK_INTERVAL': 6, 'DISK_CACHE_BLOCK_INTERVAL': 6,

5
backend/src/index.ts
View File

@ -45,6 +45,7 @@ import { formatBytes, getBytesUnit } from './utils/format';
import redisCache from './api/redis-cache'; import redisCache from './api/redis-cache';
import accelerationApi from './api/services/acceleration'; import accelerationApi from './api/services/acceleration';
import bitcoinCoreRoutes from './api/bitcoin/bitcoin-core.routes'; import bitcoinCoreRoutes from './api/bitcoin/bitcoin-core.routes';
import bitcoinSecondClient from './api/bitcoin/bitcoin-second-client';
class Server { class Server {
private wss: WebSocket.Server | undefined; private wss: WebSocket.Server | undefined;
@ -215,11 +216,13 @@ class Server {
} }
} }
const newMempool = await bitcoinApi.$getRawMempool(); const newMempool = await bitcoinApi.$getRawMempool();
const minFeeMempool = memPool.limitGBT ? await bitcoinSecondClient.getRawMemPool() : null;
const minFeeTip = memPool.limitGBT ? await bitcoinSecondClient.getBlockCount() : -1;
const newAccelerations = await accelerationApi.$fetchAccelerations(); const newAccelerations = await accelerationApi.$fetchAccelerations();
const numHandledBlocks = await blocks.$updateBlocks(); const numHandledBlocks = await blocks.$updateBlocks();
const pollRate = config.MEMPOOL.POLL_RATE_MS * (indexer.indexerIsRunning() ? 10 : 1); const pollRate = config.MEMPOOL.POLL_RATE_MS * (indexer.indexerIsRunning() ? 10 : 1);
if (numHandledBlocks === 0) { if (numHandledBlocks === 0) {
await memPool.$updateMempool(newMempool, newAccelerations, pollRate); await memPool.$updateMempool(newMempool, newAccelerations, minFeeMempool, minFeeTip, pollRate);
} }
indexer.$run(); indexer.$run();
if (config.FIAT_PRICE.ENABLED) { if (config.FIAT_PRICE.ENABLED) {

6
backend/src/mempool.interfaces.ts
View File

@ -143,6 +143,12 @@ export interface CompactThreadTransaction {
dirty?: boolean; dirty?: boolean;
} }
export interface GbtCandidates {
txs: { [txid: string ]: boolean },
added: MempoolTransactionExtended[];
removed: MempoolTransactionExtended[];
}
export interface ThreadTransaction { export interface ThreadTransaction {
txid: string; txid: string;
fee: number; fee: number;