Merge pull request #6540 from mempool/mononaut/cm-misc-fixes

misc cluster mempool fixes

backend/src/__e2e__/cluster-mempool/harness/run-harness.ts

@@ -244,8 +244,10 @@ class Harness {
 
       const added = await this.fetchTransactions(addedTxids);
 
-      // Apply diff before deleting from cache — processRemovals needs the tx data
-      const diff: MempoolDiff = { added, removed: removedTxids, accelerations: {} };
+      const removed = removedTxids
+        .map(txid => this.mempool[txid])
+        .filter((tx): tx is MempoolTransactionExtended => !!tx);
+      const diff: MempoolDiff = { added, removed, accelerations: {} };
       const t0 = Date.now();
       this.clusterMempool.applyMempoolChange(diff);
       const dt = Date.now() - t0;

backend/src/__tests__/cluster-mempool/cluster-mempool.test.ts

@@ -95,18 +95,36 @@ describe('ClusterMempool', () => {
       expect(cm.getTxCount()).toBe(2);
     });
 
+    it('should skip duplicate tx additions', () => {
+      const mempool = buildMempool([]);
+      const cm = new ClusterMempool(mempool);
+      const tx = makeTx(txid('a1'), 100, 100);
+      mempool[tx.txid] = tx;
+
+      cm.applyMempoolChange({
+        added: [tx, tx],
+        removed: [],
+        accelerations: {},
+      });
+
+      expect(cm.getClusterCount()).toBe(1);
+      expect(cm.getTxCount()).toBe(1);
+      expect(cm.getBlocks(1)[0].txids).toEqual([tx.txid]);
+    });
+
     it('should handle removing a tx', () => {
       const parentId = txid('a1');
       const childId = txid('a2');
+      const childTx = makeTx(childId, 5000, 100, [parentId]);
       const mempool = buildMempool([
         makeTx(parentId, 100, 100),
-        makeTx(childId, 5000, 100, [parentId]),
+        childTx,
       ]);
       const cm = new ClusterMempool(mempool);
 
       cm.applyMempoolChange({
         added: [],
-        removed: [childId],
+        removed: [childTx],
         accelerations: {},
       });
 
@@ -115,13 +133,61 @@ describe('ClusterMempool', () => {
       expect(cm.getClusterInfo(parentId)).not.toBeNull();
     });
 
+    it('should clean spentBy when removed tx is already missing from mempool', () => {
+      const parentId = txid('a1');
+      const childId = txid('a2');
+      const childTx = makeTx(childId, 5000, 100, [parentId]);
+      const mempool = buildMempool([
+        makeTx(parentId, 100, 100),
+        childTx,
+      ]);
+      const cm = new ClusterMempool(mempool);
+      const spentBy = (cm as unknown as { spentBy: Map<string, string> }).spentBy;
+
+      expect(spentBy.get(`${parentId}:0`)).toBe(childId);
+
+      delete mempool[childId];
+      cm.applyMempoolChange({
+        added: [],
+        removed: [childTx],
+        accelerations: {},
+      });
+
+      expect(spentBy.has(`${parentId}:0`)).toBe(false);
+    });
+
+    it('should not delete spentBy for a replacement transaction', () => {
+      const parentId = txid('a1');
+      const replacedId = txid('a2');
+      const replacementId = txid('a3');
+      const replacedTx = makeTx(replacedId, 5000, 100, [parentId]);
+      const replacementTx = makeTx(replacementId, 6000, 100, [parentId]);
+      const mempool = buildMempool([
+        makeTx(parentId, 100, 100),
+        replacementTx,
+      ]);
+      const cm = new ClusterMempool(mempool);
+      const spentBy = (cm as unknown as { spentBy: Map<string, string> }).spentBy;
+
+      expect(spentBy.get(`${parentId}:0`)).toBe(replacementId);
+
+      cm.applyMempoolChange({
+        added: [],
+        removed: [replacedTx],
+        accelerations: {},
+      });
+
+      expect(spentBy.get(`${parentId}:0`)).toBe(replacementId);
+    });
+
     it('should split cluster when middle tx is removed', () => {
       const a = txid('a1');
       const b = txid('b1');
       const c = txid('c1');
+      const bTx = makeTx(b, 200, 100, [a]);
       const mempool = buildMempool([
         makeTx(a, 100, 100),
-        makeTx(b, 200, 100, [a]),
+        bTx,
         makeTx(c, 300, 100, [b]),
       ]);
       const cm = new ClusterMempool(mempool);
@@ -129,7 +195,7 @@ describe('ClusterMempool', () => {
 
       cm.applyMempoolChange({
         added: [],
-        removed: [b],
+        removed: [bTx],
         accelerations: {},
       });
 
@@ -260,7 +326,7 @@ describe('ClusterMempool', () => {
       const cm = new ClusterMempool(mempool);
       expect(cm.getClusterCount()).toBe(1);
 
-      cm.applyMempoolChange({ added: [], removed: [center], accelerations: {} });
+      cm.applyMempoolChange({ added: [], removed: [centerTx], accelerations: {} });
 
       expect(cm.getTxCount()).toBe(5);
       expect(cm.getClusterCount()).toBe(5);
@@ -280,7 +346,7 @@ describe('ClusterMempool', () => {
       const cm = new ClusterMempool(mempool);
       expect(cm.getClusterCount()).toBe(1);
 
-      cm.applyMempoolChange({ added: [], removed: [c], accelerations: {} });
+      cm.applyMempoolChange({ added: [], removed: [cTx], accelerations: {} });
 
       expect(cm.getClusterCount()).toBe(1);
       expect(cm.getTxCount()).toBe(2);
@@ -289,13 +355,14 @@ describe('ClusterMempool', () => {
     it('should produce singleton when tx is removed from 2-tx cluster', () => {
       const a = txid('a1');
       const b = txid('b1');
+      const bTx = makeTx(b, 200, 100, [a]);
       const mempool = buildMempool([
         makeTx(a, 100, 100),
-        makeTx(b, 200, 100, [a]),
+        bTx,
       ]);
       const cm = new ClusterMempool(mempool);
 
-      cm.applyMempoolChange({ added: [], removed: [b], accelerations: {} });
+      cm.applyMempoolChange({ added: [], removed: [bTx], accelerations: {} });
 
       expect(cm.getClusterCount()).toBe(1);
       expect(cm.getTxCount()).toBe(1);
@@ -322,7 +389,7 @@ describe('ClusterMempool', () => {
       const cm = new ClusterMempool(mempool);
       expect(cm.getClusterCount()).toBe(1);
 
-      cm.applyMempoolChange({ added: [], removed: [hub], accelerations: {} });
+      cm.applyMempoolChange({ added: [], removed: [hubTx], accelerations: {} });
 
       expect(cm.getClusterCount()).toBe(3);
       expect(cm.getTxCount()).toBe(3);
@@ -467,7 +534,7 @@ describe('ClusterMempool', () => {
 
       cm.applyMempoolChange({
         added: [],
-        removed: [txid('nonexistent')],
+        removed: [makeTx(txid('nonexistent'), 100, 100)],
         accelerations: {},
       });
 

backend/src/api/blocks.ts

@@ -511,7 +511,7 @@ class Blocks {
     if (config.MEMPOOL.CLUSTER_MEMPOOL) {
       memPool.clusterMempool?.applyMempoolChange({
         added: [],
-        removed: txIds,
+        removed: transactions,
         accelerations: mempool.getAccelerations(),
       });
     }

backend/src/api/mempool.ts

@@ -399,7 +399,7 @@ class Mempool {
     if (config.MEMPOOL.CLUSTER_MEMPOOL && (newTransactions.length || deletedTransactions.length || accelerationDelta.length)) {
       this.clusterMempool?.applyMempoolChange({
         added: newTransactions,
-        removed: deletedTransactions.map(tx => tx.txid),
+        removed: deletedTransactions,
         accelerations: this.getAccelerations(),
       });
     }

backend/src/cluster-mempool/cluster-mempool.ts

@@ -6,7 +6,7 @@ import logger from '../logger';
 
 export interface MempoolDiff {
   added: MempoolTransactionExtended[];
-  removed: string[];
+  removed: MempoolTransactionExtended[];
   accelerations: { [txid: string]: { feeDelta: number } };
 }
 
@@ -42,7 +42,6 @@ const DEFAULT_COST_BUDGET = 75000;
 export class ClusterMempool {
   private clusters = new Map<number, Cluster>();
   private txToCluster = new Map<string, number>();
-  private parentMap = new Map<string, Set<string>>();
   private spentBy = new Map<string, string>();
   private mempool: Readonly<{ [txid: string]: MempoolTransactionExtended }>;
   private accelerations: { [txid: string]: { feeDelta: number } } = {};
@@ -147,15 +146,15 @@ export class ClusterMempool {
   }
 
   private buildFromMempool(): void {
-    this.buildRelativeMaps();
-    const components = this.findMempoolComponents();
+    const parentMap = this.buildRelativeMaps();
+    const components = this.findMempoolComponents(parentMap);
     for (const component of components) {
-      this.createClusterFromTxids(component);
+      this.createClusterFromTxids(component, parentMap);
     }
   }
 
-  private buildRelativeMaps(): void {
-    this.parentMap.clear();
+  private buildRelativeMaps(): Map<string, Set<string>> {
+    const parentMap = new Map<string, Set<string>>();
     this.spentBy.clear();
     for (const txid in this.mempool) {
       const tx = this.mempool[txid];
@@ -167,18 +166,19 @@ export class ClusterMempool {
         }
       }
       if (txParents.size > 0) {
-        this.parentMap.set(txid, txParents);
+        parentMap.set(txid, txParents);
       }
     }
+    return parentMap;
   }
 
-  private findMempoolComponents(): Set<string>[] {
+  private findMempoolComponents(parentMap: Map<string, Set<string>>): Set<string>[] {
     const visited = new Set<string>();
     const components: Set<string>[] = [];
 
     for (const txid in this.mempool) {
       if (!visited.has(txid)) {
-        const component = this.dfsComponent(txid, visited);
+        const component = this.dfsComponent(txid, visited, parentMap);
         components.push(component);
       }
     }
@@ -187,7 +187,8 @@ export class ClusterMempool {
 
   private dfsComponent(
     startTxid: string,
-    visited: Set<string>
+    visited: Set<string>,
+    parentMap: Map<string, Set<string>>
   ): Set<string> {
     const component = new Set<string>();
     const stack = [startTxid];
@@ -197,7 +198,7 @@ export class ClusterMempool {
         visited.add(current);
         component.add(current);
 
-        const txParents = this.parentMap.get(current);
+        const txParents = parentMap.get(current);
         if (txParents) {
           for (const p of txParents) {
             if (!visited.has(p)) {
@@ -229,7 +230,8 @@ export class ClusterMempool {
   }
 
   private createClusterFromTxids(
-    txids: Set<string>
+    txids: Set<string>,
+    parentMap: Map<string, Set<string>>
   ): Cluster | null {
     const clusterId = this.nextClusterId++;
     const depgraph = new DepGraph();
@@ -246,7 +248,7 @@ export class ClusterMempool {
     }
 
     for (const txid of txids) {
-      const txParents = this.parentMap.get(txid);
+      const txParents = parentMap.get(txid);
       if (txParents) {
         for (const parentTxid of txParents) {
           if (txids.has(parentTxid)) {
@@ -349,27 +351,25 @@ export class ClusterMempool {
     return relatives;
   }
 
-  private processRemovals(removed: string[]): void {
-    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}`);
+  private processRemovals(removed: MempoolTransactionExtended[]): void {
+    for (const tx of removed) {
+      for (const vin of tx.vin) {
+        if (!vin.is_coinbase) {
+          const spentOutpoint = `${vin.txid}:${vin.vout}`;
+          if (this.spentBy.get(spentOutpoint) === tx.txid) {
+            this.spentBy.delete(spentOutpoint);
           }
         }
-      } else if (this.txToCluster.has(txid)) {
-        logger.warn(`ClusterMempool.processRemovals: ${txid} missing from mempool, spentBy cleanup skipped`);
       }
     }
 
-    for (const txid of removed) {
-      const match = this.getClusterForTx(txid);
+    for (const tx of removed) {
+      const match = this.getClusterForTx(tx.txid);
       if (match) {
         match.cluster.depgraph.removeTransactions(new Set([match.clusterTx]));
-        match.cluster.txs.delete(txid);
+        match.cluster.txs.delete(tx.txid);
         match.cluster.linearization = match.cluster.linearization.filter(t => t !== match.clusterTx);
-        this.txToCluster.delete(txid);
+        this.txToCluster.delete(tx.txid);
         match.cluster.dirty = true;
       }
     }
@@ -431,6 +431,12 @@ export class ClusterMempool {
     for (const tx of added) {
       const txid = tx.txid;
 
+      // sanity check for duplicate transactions
+      if (this.getClusterForTx(txid) !== null) {
+        logger.warn(`ClusterMempool.processAdditions: ${txid} was already added, skipping`);
+        continue;
+      }
+
       for (const vin of tx.vin) {
         if (!vin.is_coinbase) {
           this.spentBy.set(`${vin.txid}:${vin.vout}`, txid);