Merge pull request #2156 from ergoplatform/i2155

Clear mempool transactions with inputs being spent immediately when a block arrives

Author: kushti

src/main/resources/application.conf

@@ -79,8 +79,8 @@ ergo {
     mempoolCapacity = 1000
 
     # Interval for mempool transaction re-check. We check transaction when it is entering the mempool, and then
-    # re-check it every interval value
-    mempoolCleanupDuration = 20m
+    # re-check it every interval value (but only on new block arrival)
+    mempoolCleanupDuration = 30s
 
     # Mempool transaction sorting scheme ("random", "bySize", or "byExecutionCost")
     mempoolSorting = "random"

src/main/scala/org/ergoplatform/local/CleanupWorker.scala

@@ -66,8 +66,8 @@ class CleanupWorker(nodeViewHolderRef: ActorRef,
 
     val now = System.currentTimeMillis()
 
-    val allPoolTxs = mempool.getAllPrioritized
     // Check transactions sorted by priority. Parent transaction comes before its children.
+    val allPoolTxs = mempool.getAllPrioritized
     val txsToValidate = allPoolTxs.filter { utx =>
       (now - utx.lastCheckedTime) > TimeLimit
     }.toList

src/main/scala/org/ergoplatform/nodeView/ErgoNodeViewHolder.scala

@@ -402,7 +402,7 @@ abstract class ErgoNodeViewHolder[State <: ErgoState[State]](settings: ErgoSetti
       .flatMap(extractTransactions)
       .filter(tx => !appliedTxs.exists(_.id == tx.id))
       .map(tx => UnconfirmedTransaction(tx, None))
-    memPool.remove(appliedTxs).put(rolledBackTxs)
+    memPool.removeWithDoubleSpends(appliedTxs).put(rolledBackTxs)
   }
 
   /**

src/main/scala/org/ergoplatform/nodeView/mempool/ErgoMemPool.scala

@@ -106,15 +106,37 @@ class ErgoMemPool private[mempool](private[mempool] val pool: OrderedTxPool,
   }
 
   /**
-    * Remove transaction from the pool
+    * Remove transaction from the pool along with its double-spends
     */
-  def remove(tx: ErgoTransaction): ErgoMemPool = {
-    log.debug(s"Removing transaction ${tx.id} from the mempool")
-    new ErgoMemPool(pool.remove(tx), updateStatsOnRemoval(tx), sortingOption)
+  def removeTxAndDoubleSpends(tx: ErgoTransaction): ErgoMemPool = {
+    def removeTx(mp: ErgoMemPool, tx: ErgoTransaction): ErgoMemPool = {
+      log.debug(s"Removing transaction ${tx.id} from the mempool")
+      new ErgoMemPool(mp.pool.remove(tx), mp.updateStatsOnRemoval(tx), sortingOption)
+    }
+
+    val poolWithoutTx = removeTx(this, tx)
+    val doubleSpentTransactionIds = tx.inputs.flatMap(i =>
+      poolWithoutTx.pool.inputs.get(i.boxId)
+    ).toSet
+    val doubleSpentTransactions = doubleSpentTransactionIds.flatMap { txId =>
+      poolWithoutTx.pool.orderedTransactions.get(txId)
+    }
+    doubleSpentTransactions.foldLeft(poolWithoutTx) { case (pool, tx) =>
+      removeTx(pool, tx.transaction)
+    }
   }
 
-  def remove(txs: TraversableOnce[ErgoTransaction]): ErgoMemPool = {
-    txs.foldLeft(this) { case (acc, tx) => acc.remove(tx) }
+  /**
+    * Remove provided transactions and their doublespends from the pool
+    */
+  def removeWithDoubleSpends(txs: TraversableOnce[ErgoTransaction]): ErgoMemPool = {
+    txs.foldLeft(this) { case (memPool, tx) =>
+      if (memPool.contains(tx.id)) { // tx could be removed earlier in this loop as double-spend of another tx
+        memPool.removeTxAndDoubleSpends(tx)
+      } else {
+        memPool
+      }
+    }
   }
 
   /**

src/test/scala/org/ergoplatform/nodeView/mempool/ErgoMemPoolSpec.scala

@@ -1,11 +1,11 @@
 package org.ergoplatform.nodeView.mempool
 
-import org.ergoplatform.modifiers.mempool.{ErgoTransaction, UnconfirmedTransaction}
+import org.ergoplatform.{ErgoBoxCandidate, Input}
 import org.ergoplatform.nodeView.mempool.ErgoMemPoolUtils.{ProcessingOutcome, SortingOption}
+import org.ergoplatform.modifiers.mempool.{ErgoTransaction, UnconfirmedTransaction}
 import org.ergoplatform.nodeView.state.wrapped.WrappedUtxoState
 import org.ergoplatform.settings.ErgoSettings
-import org.ergoplatform.utils.ErgoTestHelpers
-import org.ergoplatform.{ErgoBoxCandidate, Input}
+import org.ergoplatform.utils.{ErgoTestHelpers, RandomWrapper}
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatestplus.scalacheck.ScalaCheckPropertyChecks
 import sigma.ast.ErgoTree.ZeroHeader
@@ -299,11 +299,41 @@ class ErgoMemPoolSpec extends AnyFlatSpec
     }
     pool.size shouldBe (family_depth + 1) * txs.size
     allTxs.foreach { tx =>
-      pool = pool.remove(tx.transaction)
+      pool = pool.removeTxAndDoubleSpends(tx.transaction)
     }
     pool.size shouldBe 0
   }
 
+  it should "correctly remove doublespents of a transaction from pool" in {
+    val (us, bh) = createUtxoState(settings)
+    val genesis = validFullBlock(None, us, bh)
+    val wus = WrappedUtxoState(us, bh, settings, parameters).applyModifier(genesis)(_ => ()).get
+    val boxes = wus.takeBoxes(4)
+
+    val limit = 10000
+
+    val tx1 = validTransactionsFromBoxes(limit, boxes.take(1), new RandomWrapper)
+                ._1.map(tx => UnconfirmedTransaction(tx, None)).head
+
+    val tx2 = validTransactionsFromBoxes(limit, boxes.takeRight(2), new RandomWrapper)
+      ._1.map(tx => UnconfirmedTransaction(tx, None)).head
+
+    val tx3 = validTransactionsFromBoxes(limit, boxes.take(1), new RandomWrapper)
+      ._1.map(tx => UnconfirmedTransaction(tx, None)).head
+
+    tx1.transaction.inputs.head.boxId shouldBe tx3.transaction.inputs.head.boxId
+
+    var pool = ErgoMemPool.empty(settings)
+    Seq(tx2, tx3).foreach { tx =>
+      pool = pool.put(tx)
+    }
+
+    pool = pool.removeTxAndDoubleSpends(tx1.transaction)
+    pool.contains(tx1.transaction) shouldBe false
+    pool.contains(tx2.transaction) shouldBe true
+    pool.contains(tx3.transaction) shouldBe false
+  }
+
   it should "return results take / getAll / getAllPrioritized sorted by priority" in {
     val feeProp = settings.chainSettings.monetary.feeProposition
 
@@ -378,7 +408,7 @@ class ErgoMemPoolSpec extends AnyFlatSpec
     pool.stats.snapTakenTxns shouldBe MemPoolStatistics(System.currentTimeMillis(),0,System.currentTimeMillis()).snapTakenTxns
 
     allTxs.foreach { tx =>
-      pool = pool.remove(tx.transaction)
+      pool = pool.removeTxAndDoubleSpends(tx.transaction)
     }
     pool.size shouldBe 0
     pool.stats.takenTxns shouldBe (family_depth + 1) * txs.size

src/test/scala/scorex/testkit/properties/mempool/MempoolTransactionsTest.scala

@@ -84,15 +84,15 @@ trait MempoolTransactionsTest
   property("Size of mempool should decrease when removing a present transaction") {
     forAll(memPoolGenerator, unconfirmedTxSeqGenerator) { (mp: ErgoMemPool, unconfirmedTxs: Seq[UnconfirmedTransaction]) =>
       val m: ErgoMemPool = mp.put(unconfirmedTxs)
-      val m2: ErgoMemPool = m.remove(unconfirmedTxs.headOption.get.transaction)
+      val m2: ErgoMemPool = m.removeTxAndDoubleSpends(unconfirmedTxs.headOption.get.transaction)
       m2.size shouldBe unconfirmedTxs.size - 1
     }
   }
 
   property("Size of mempool should not decrease when removing a non-present transaction") {
     forAll(memPoolGenerator, unconfirmedTxSeqGenerator, unconfirmedTxGenerator) { (mp: ErgoMemPool, unconfirmedTxs: Seq[UnconfirmedTransaction], unconfirmedTx: UnconfirmedTransaction) =>
       val m: ErgoMemPool = mp.put(unconfirmedTxs)
-      val m2: ErgoMemPool = m.remove(unconfirmedTx.transaction)
+      val m2: ErgoMemPool = m.removeTxAndDoubleSpends(unconfirmedTx.transaction)
       m2.size shouldBe unconfirmedTxs.size
     }
   }