Use minfee node to limit gbt input size

backend/src/api/mempool-blocks.ts

@@ -1,5 +1,5 @@
 import logger from '../logger';
-import { MempoolBlock, TransactionExtended, ThreadTransaction, TransactionStripped, MempoolBlockWithTransactions, MempoolBlockDelta, Ancestor } from '../mempool.interfaces';
+import { MempoolBlock, TransactionExtended, ThreadTransaction, TransactionStripped, MempoolBlockWithTransactions, MempoolBlockDelta, Ancestor, GbtCandidates } from '../mempool.interfaces';
 import { Common } from './common';
 import config from '../config';
 import { Worker } from 'worker_threads';
@@ -147,19 +147,23 @@ class MempoolBlocks {
     return mempoolBlockDeltas;
   }
 
-  public async makeBlockTemplates(newMempool: { [txid: string]: TransactionExtended }, saveResults: boolean = false): Promise<MempoolBlockWithTransactions[]> {
+  public async makeBlockTemplates(newMempool: { [txid: string]: TransactionExtended }, saveResults: boolean = false, candidates?: GbtCandidates): Promise<MempoolBlockWithTransactions[]> {
     // 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
+    const txids = candidates ? Object.keys(candidates.txs) : Object.keys(newMempool);
     const strippedMempool: { [txid: string]: ThreadTransaction } = {};
-    Object.values(newMempool).forEach(entry => {
+    txids.forEach(txid => {
-      strippedMempool[entry.txid] = {
+      const entry = newMempool[txid];
-        txid: entry.txid,
+      if (entry) {
-        fee: entry.fee,
+        strippedMempool[entry.txid] = {
-        weight: entry.weight,
+          txid: entry.txid,
-        feePerVsize: entry.fee / (entry.weight / 4),
+          fee: entry.fee,
-        effectiveFeePerVsize: entry.fee / (entry.weight / 4),
+          weight: entry.weight,
-        vin: entry.vin.map(v => v.txid),
+          feePerVsize: entry.fee / (entry.weight / 4),
-      };
+          effectiveFeePerVsize: entry.fee / (entry.weight / 4),
+          vin: entry.vin.map(v => v.txid),
+        };
+      }
     });
 
     // (re)initialize tx selection worker thread
@@ -191,31 +195,49 @@ class MempoolBlocks {
       // clean up thread error listener
       this.txSelectionWorker?.removeListener('error', threadErrorListener);
 
-      return this.processBlockTemplates(newMempool, blocks, clusters, saveResults);
+      return this.processBlockTemplates(newMempool, blocks, clusters, saveResults, candidates);
     } catch (e) {
       logger.err('makeBlockTemplates failed. ' + (e instanceof Error ? e.message : e));
     }
     return this.mempoolBlocks;
   }
 
-  public async updateBlockTemplates(newMempool: { [txid: string]: TransactionExtended }, added: TransactionExtended[], removed: string[], saveResults: boolean = false): Promise<void> {
+  public async updateBlockTemplates(newMempool: { [txid: string]: TransactionExtended }, added: TransactionExtended[], removed: string[], saveResults: boolean = false, candidates?: GbtCandidates): Promise<void> {
     if (!this.txSelectionWorker) {
       // need to reset the worker
-      this.makeBlockTemplates(newMempool, saveResults);
+      this.makeBlockTemplates(newMempool, saveResults, candidates);
       return;
     }
     // 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
-    const addedStripped: ThreadTransaction[] = added.map(entry => {
+    let addedStripped: ThreadTransaction[] = [];
-      return {
+    let removedList;
-        txid: entry.txid,
+    if (candidates) {
-        fee: entry.fee,
+      addedStripped = candidates.added.filter(txid => newMempool[txid]).map(txid => {
-        weight: entry.weight,
+        const entry = newMempool[txid];
-        feePerVsize: entry.fee / (entry.weight / 4),
+        return {
-        effectiveFeePerVsize: entry.fee / (entry.weight / 4),
+          txid: entry.txid,
-        vin: entry.vin.map(v => v.txid),
+          fee: entry.fee,
-      };
+          weight: entry.weight,
-    });
+          feePerVsize: entry.fee / (entry.weight / 4),
+          effectiveFeePerVsize: entry.fee / (entry.weight / 4),
+          vin: entry.vin.map(v => v.txid),
+        };
+      });
+      removedList = candidates.removed;
+    } else {
+      addedStripped = added.map(entry => {
+        return {
+          txid: entry.txid,
+          fee: entry.fee,
+          weight: entry.weight,
+          feePerVsize: entry.fee / (entry.weight / 4),
+          effectiveFeePerVsize: entry.fee / (entry.weight / 4),
+          vin: entry.vin.map(v => v.txid),
+        };
+      });
+      removedList = removed;
+    }
 
     // run the block construction algorithm in a separate thread, and wait for a result
     let threadErrorListener;
@@ -227,19 +249,19 @@ class MempoolBlocks {
         });
         this.txSelectionWorker?.once('error', reject);
       });
-      this.txSelectionWorker.postMessage({ type: 'update', added: addedStripped, removed });
+      this.txSelectionWorker.postMessage({ type: 'update', added: addedStripped, removed: removedList });
       const { blocks, clusters } = await workerResultPromise;
 
       // clean up thread error listener
       this.txSelectionWorker?.removeListener('error', threadErrorListener);
 
-      this.processBlockTemplates(newMempool, blocks, clusters, saveResults);
+      this.processBlockTemplates(newMempool, blocks, clusters, saveResults, candidates);
     } catch (e) {
       logger.err('updateBlockTemplates failed. ' + (e instanceof Error ? e.message : e));
     }
   }
 
-  private processBlockTemplates(mempool, blocks, clusters, saveResults): MempoolBlockWithTransactions[] {
+  private processBlockTemplates(mempool: { [txid: string ]: TransactionExtended }, blocks, clusters, saveResults, candidates?: GbtCandidates): MempoolBlockWithTransactions[] {
     // update this thread's mempool with the results
     blocks.forEach(block => {
       block.forEach(tx => {
@@ -277,6 +299,32 @@ class MempoolBlocks {
       });
     });
 
+    // Add purged transactions at the end, if required
+    if (candidates) {
+      const purged: string[] = [];
+      Object.values(mempool).forEach(tx => {
+        if (!candidates.txs[tx.txid]) {
+          purged.push(tx.txid);
+        }
+      });
+      if (!blocks.length) {
+        blocks = [[]];
+      }
+      let blockIndex = blocks.length - 1;
+      let weight = blocks[blockIndex].reduce((acc, tx) => acc + tx.weight, 0);
+      purged.sort((a,b) => { return mempool[b].effectiveFeePerVsize - mempool[a].effectiveFeePerVsize});
+      purged.forEach(txid => {
+        const tx = mempool[txid];
+        if ((weight + tx.weight) >= (config.MEMPOOL.BLOCK_WEIGHT_UNITS - 4000) && blockIndex < 7) {
+          blocks.push([]);
+          blockIndex++;
+          weight = 0;
+        }
+        blocks[blockIndex].push(tx);
+        weight += tx.weight;
+      });
+    }
+
     // unpack the condensed blocks into proper mempool blocks
     const mempoolBlocks = blocks.map((transactions, blockIndex) => {
       return this.dataToMempoolBlocks(transactions.map(tx => {

backend/src/api/mempool.ts

@@ -1,6 +1,6 @@
 import config from '../config';
 import bitcoinApi from './bitcoin/bitcoin-api-factory';
-import { TransactionExtended, VbytesPerSecond } from '../mempool.interfaces';
+import { GbtCandidates, TransactionExtended, VbytesPerSecond } from '../mempool.interfaces';
 import logger from '../logger';
 import { Common } from './common';
 import transactionUtils from './transaction-utils';
@@ -9,6 +9,7 @@ import loadingIndicators from './loading-indicators';
 import bitcoinClient from './bitcoin/bitcoin-client';
 import bitcoinSecondClient from './bitcoin/bitcoin-second-client';
 import rbfCache from './rbf-cache';
+import blocks from './blocks';
 
 class Mempool {
   private static WEBSOCKET_REFRESH_RATE_MS = 10000;
@@ -16,12 +17,16 @@ class Mempool {
   private inSync: boolean = false;
   private mempoolCacheDelta: number = -1;
   private mempoolCache: { [txId: string]: TransactionExtended } = {};
+  private mempoolCandidates: { [txid: string ]: boolean } = {};
+  private minFeeMempool: { [txId: string]: boolean } = {};
   private mempoolInfo: IBitcoinApi.MempoolInfo = { loaded: false, size: 0, bytes: 0, usage: 0, total_fee: 0,
                                                     maxmempool: 300000000, mempoolminfee: 0.00001000, minrelaytxfee: 0.00001000 };
+  private secondMempoolInfo: IBitcoinApi.MempoolInfo = { loaded: false, size: 0, bytes: 0, usage: 0, total_fee: 0,
+                                                    maxmempool: 300000000, mempoolminfee: 0.00001000, minrelaytxfee: 0.00001000 };
   private mempoolChangedCallback: ((newMempool: {[txId: string]: TransactionExtended; }, newTransactions: TransactionExtended[],
     deletedTransactions: TransactionExtended[]) => void) | undefined;
   private asyncMempoolChangedCallback: ((newMempool: {[txId: string]: TransactionExtended; }, newTransactions: TransactionExtended[],
-    deletedTransactions: TransactionExtended[]) => Promise<void>) | undefined;
+    deletedTransactions: TransactionExtended[], candidates?: GbtCandidates) => Promise<void>) | undefined;
 
   private txPerSecondArray: number[] = [];
   private txPerSecond: number = 0;
@@ -71,7 +76,7 @@ class Mempool {
   }
 
   public setAsyncMempoolChangedCallback(fn: (newMempool: { [txId: string]: TransactionExtended; },
-    newTransactions: TransactionExtended[], deletedTransactions: TransactionExtended[]) => Promise<void>) {
+    newTransactions: TransactionExtended[], deletedTransactions: TransactionExtended[], candidates?: GbtCandidates) => Promise<void>) {
     this.asyncMempoolChangedCallback = fn;
   }
 
@@ -85,10 +90,18 @@ class Mempool {
       this.mempoolChangedCallback(this.mempoolCache, [], []);
     }
     if (this.asyncMempoolChangedCallback) {
-      this.asyncMempoolChangedCallback(this.mempoolCache, [], []);
+      if (config.MEMPOOL.USE_SECOND_NODE_FOR_MINFEE) {
+        this.asyncMempoolChangedCallback(this.mempoolCache, [], [], { txs: {}, added: [], removed: [] });
+      } else {
+        this.asyncMempoolChangedCallback(this.mempoolCache, [], [], );
+      }
     }
   }
 
+  public getMempoolCandidates(): { [txid: string]: boolean } {
+    return this.mempoolCandidates;
+  }
+
   public async $updateMemPoolInfo() {
     this.mempoolInfo = await this.$getMempoolInfo();
   }
@@ -124,6 +137,7 @@ class Mempool {
     let hasChange: boolean = false;
     const currentMempoolSize = Object.keys(this.mempoolCache).length;
     const transactions = await bitcoinApi.$getRawMempool();
+    const candidates = await this.getNextCandidates();
     const diff = transactions.length - currentMempoolSize;
     const newTransactions: TransactionExtended[] = [];
 
@@ -223,8 +237,8 @@ class Mempool {
     if (this.mempoolChangedCallback && (hasChange || deletedTransactions.length)) {
       this.mempoolChangedCallback(this.mempoolCache, newTransactions, deletedTransactions);
     }
-    if (this.asyncMempoolChangedCallback && (hasChange || deletedTransactions.length)) {
+    if (this.asyncMempoolChangedCallback && (hasChange || deletedTransactions.length || candidates?.added.length || candidates?.removed.length)) {
-      await this.asyncMempoolChangedCallback(this.mempoolCache, newTransactions, deletedTransactions);
+      await this.asyncMempoolChangedCallback(this.mempoolCache, newTransactions, deletedTransactions, candidates);
     }
 
     const end = new Date().getTime();
@@ -232,6 +246,55 @@ class Mempool {
     logger.debug(`Mempool updated in ${time / 1000} seconds. New size: ${Object.keys(this.mempoolCache).length} (${diff > 0 ? '+' + diff : diff})`);
   }
 
+  public isTxPurged(txid: string): boolean {
+    return !this.minFeeMempool[txid];
+  }
+
+  public async getNextCandidates(): Promise<GbtCandidates | undefined> {
+    if (config.MEMPOOL.USE_SECOND_NODE_FOR_MINFEE) {
+      const minFeeTransactions = await bitcoinSecondClient.getRawMemPool();
+      const blockHeight = await bitcoinSecondClient.getChainTips()
+        .then((result: IBitcoinApi.ChainTips[]) => {
+          return result.find(tip => tip.status === 'active')!.height;
+        });
+      const newCandidateTxMap = {};
+      this.minFeeMempool = {};
+      for (const txid of minFeeTransactions) {
+        if (this.mempoolCache[txid]) {
+          newCandidateTxMap[txid] = true;
+        }
+        this.minFeeMempool[txid] = true;
+      }
+      const removed: string[] = [];
+      const added: string[] = [];
+      // 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]) {
+            removed.push(txid);
+          }
+        }
+      }
+
+      for (const txid of Object.keys(newCandidateTxMap)) {
+        if (!this.mempoolCandidates[txid]) {
+          added.push(txid);
+        }
+      }
+
+      this.mempoolCandidates = newCandidateTxMap;
+      return {
+        txs: this.mempoolCandidates,
+        added,
+        removed
+      };
+    }
+  }
+
   public handleRbfTransactions(rbfTransactions: { [txid: string]: TransactionExtended; }) {
     for (const rbfTransaction in rbfTransactions) {
       if (this.mempoolCache[rbfTransaction]) {

backend/src/api/websocket-handler.ts

@@ -2,7 +2,7 @@ import logger from '../logger';
 import * as WebSocket from 'ws';
 import {
   BlockExtended, TransactionExtended, WebsocketResponse,
-  OptimizedStatistic, ILoadingIndicators
+  OptimizedStatistic, ILoadingIndicators, GbtCandidates
 } from '../mempool.interfaces';
 import blocks from './blocks';
 import memPool from './mempool';
@@ -92,6 +92,9 @@ class WebsocketHandler {
                   }
                 }
               }
+              if (config.MEMPOOL.USE_SECOND_NODE_FOR_MINFEE && memPool.getMempool()[client['track-tx']]) {
+                response['txPurged'] = memPool.isTxPurged(client['track-tx']);
+              }
             } else {
               client['track-tx'] = null;
             }
@@ -247,13 +250,14 @@ class WebsocketHandler {
   }
 
   async handleMempoolChange(newMempool: { [txid: string]: TransactionExtended },
-    newTransactions: TransactionExtended[], deletedTransactions: TransactionExtended[]): Promise<void> {
+    newTransactions: TransactionExtended[], deletedTransactions: TransactionExtended[],
+    candidates?: GbtCandidates): Promise<void> {
     if (!this.wss) {
       throw new Error('WebSocket.Server is not set');
     }
 
     if (config.MEMPOOL.ADVANCED_GBT_MEMPOOL) {
-      await mempoolBlocks.updateBlockTemplates(newMempool, newTransactions, deletedTransactions.map(tx => tx.txid), true);
+      await mempoolBlocks.updateBlockTemplates(newMempool, newTransactions, deletedTransactions.map(tx => tx.txid), true, candidates);
     } else {
       mempoolBlocks.updateMempoolBlocks(newMempool, true);
     }
@@ -395,6 +399,11 @@ class WebsocketHandler {
             }
           }
         }
+
+        // update purge status of unconfirmed tracked txs
+        if (config.MEMPOOL.USE_SECOND_NODE_FOR_MINFEE && newMempool[client['track-tx']]) {
+          response['txPurged'] = memPool.isTxPurged(client['track-tx']);
+        }
       }
 
       if (client['track-mempool-block'] >= 0) {
@@ -419,6 +428,7 @@ class WebsocketHandler {
     }
 
     const _memPool = memPool.getMempool();
+    const candidateTxs = await memPool.getMempoolCandidates();
 
     if (config.MEMPOOL.AUDIT) {
       let projectedBlocks;
@@ -426,7 +436,15 @@ class WebsocketHandler {
       // a cloned copy of the mempool if we're running a different algorithm for mempool updates
       const auditMempool = (config.MEMPOOL.ADVANCED_GBT_AUDIT === config.MEMPOOL.ADVANCED_GBT_MEMPOOL) ? _memPool : deepClone(_memPool);
       if (config.MEMPOOL.ADVANCED_GBT_AUDIT) {
-        projectedBlocks = await mempoolBlocks.makeBlockTemplates(auditMempool, false);
+        let candidates;
+        if (config.MEMPOOL.USE_SECOND_NODE_FOR_MINFEE) {
+          candidates = {
+            txs: candidateTxs,
+            added: [],
+            removed: [],
+          };
+        }
+        projectedBlocks = await mempoolBlocks.makeBlockTemplates(auditMempool, false, candidates);
       } else {
         projectedBlocks = mempoolBlocks.updateMempoolBlocks(auditMempool, false);
       }
@@ -478,12 +496,21 @@ class WebsocketHandler {
     // Update mempool to remove transactions included in the new block
     for (const txId of txIds) {
       delete _memPool[txId];
+      delete candidateTxs[txId];
       removed.push(txId);
       rbfCache.evict(txId);
     }
 
     if (config.MEMPOOL.ADVANCED_GBT_MEMPOOL) {
-      await mempoolBlocks.updateBlockTemplates(_memPool, [], removed, true);
+      let candidates;
+      if (config.MEMPOOL.USE_SECOND_NODE_FOR_MINFEE) {
+        candidates = {
+          txs: candidateTxs,
+          added: [],
+          removed: removed,
+        };
+      }
+      await mempoolBlocks.updateBlockTemplates(_memPool, [], removed, true, candidates);
     } else {
       mempoolBlocks.updateMempoolBlocks(_memPool, true);
     }

backend/src/mempool.interfaces.ts

@@ -100,6 +100,12 @@ export interface AuditTransaction {
   modifiedNode: HeapNode<AuditTransaction>;
 }
 
+export interface GbtCandidates {
+  txs: { [txid: string ]: boolean },
+  added: string[];
+  removed: string[];
+}
+
 export interface ThreadTransaction {
   txid: string;
   fee: number;

frontend/src/app/components/transaction/transaction.component.html

@@ -1,6 +1,10 @@
 <div class="container-xl">
 
   <div class="title-block">
+    <div *ngIf="isPurged" class="alert alert-mempool" role="alert">
+      <span i18n="transaction.purged|Purged from mempool">This transaction has been purged from our default sized 300MB mempool</span>
+    </div>
+
     <div *ngIf="rbfTransaction" class="alert alert-mempool" role="alert">
       <span i18n="transaction.rbf.replacement|RBF replacement">This transaction has been replaced by:</span>
       <app-truncate [text]="rbfTransaction.txid" [lastChars]="12" [link]="['/tx/' | relativeUrl, rbfTransaction.txid]"></app-truncate>

frontend/src/app/components/transaction/transaction.component.ts

@@ -46,6 +46,7 @@ export class TransactionComponent implements OnInit, AfterViewInit, OnDestroy {
   fetchRbfSubscription: Subscription;
   fetchCachedTxSubscription: Subscription;
   txReplacedSubscription: Subscription;
+  txPurgedSubscription: Subscription;
   blocksSubscription: Subscription;
   queryParamsSubscription: Subscription;
   urlFragmentSubscription: Subscription;
@@ -59,6 +60,7 @@ export class TransactionComponent implements OnInit, AfterViewInit, OnDestroy {
   fetchRbfHistory$ = new Subject<string>();
   fetchCachedTx$ = new Subject<string>();
   isCached: boolean = false;
+  isPurged: boolean = false;
   now = new Date().getTime();
   timeAvg$: Observable<number>;
   liquidUnblinding = new LiquidUnblinding();
@@ -380,6 +382,10 @@ export class TransactionComponent implements OnInit, AfterViewInit, OnDestroy {
       }
     });
 
+    this.txPurgedSubscription = this.stateService.txPurged$.subscribe((isPurged) => {
+      this.isPurged = isPurged;
+    });
+
     this.queryParamsSubscription = this.route.queryParams.subscribe((params) => {
       if (params.showFlow === 'false') {
         this.overrideFlowPreference = false;
@@ -532,6 +538,7 @@ export class TransactionComponent implements OnInit, AfterViewInit, OnDestroy {
     this.fetchRbfSubscription.unsubscribe();
     this.fetchCachedTxSubscription.unsubscribe();
     this.txReplacedSubscription.unsubscribe();
+    this.txPurgedSubscription.unsubscribe();
     this.blocksSubscription.unsubscribe();
     this.queryParamsSubscription.unsubscribe();
     this.flowPrefSubscription.unsubscribe();

frontend/src/app/interfaces/websocket.interface.ts

@@ -16,6 +16,7 @@ export interface WebsocketResponse {
   tx?: Transaction;
   rbfTransaction?: ReplacedTransaction;
   txReplaced?: ReplacedTransaction;
+  txPurged?: boolean;
   utxoSpent?: object;
   transactions?: TransactionStripped[];
   loadingIndicators?: ILoadingIndicators;

frontend/src/app/services/state.service.ts

@@ -98,6 +98,7 @@ export class StateService {
   mempoolBlockTransactions$ = new Subject<TransactionStripped[]>();
   mempoolBlockDelta$ = new Subject<MempoolBlockDelta>();
   txReplaced$ = new Subject<ReplacedTransaction>();
+  txPurged$ = new Subject<boolean>();
   utxoSpent$ = new Subject<object>();
   difficultyAdjustment$ = new ReplaySubject<DifficultyAdjustment>(1);
   mempoolTransactions$ = new Subject<Transaction>();

frontend/src/app/services/websocket.service.ts

@@ -261,6 +261,10 @@ export class WebsocketService {
       this.stateService.txReplaced$.next(response.txReplaced);
     }
 
+    if (response.txPurged != null) {
+      this.stateService.txPurged$.next(response.txPurged);
+    }
+
     if (response['mempool-blocks']) {
       this.stateService.mempoolBlocks$.next(response['mempool-blocks']);
     }