Extract mempool re-validation logic from CleanupWorker and test it

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

@@ -19,13 +19,13 @@ import scala.concurrent.ExecutionContext.Implicits.global
 /**
   * Performs mempool transactions re-validation. Called on a new block coming.
   * Validation results sent directly to `NodeViewHolder`.
+  *
+  * The actual re-validation logic lives in [[CleanupWorker.validatePool]], so that it can be
+  * exercised without an actor system; this actor only wires it to the node view holder.
   */
 class CleanupWorker(nodeViewHolderRef: ActorRef,
                     nodeSettings: NodeConfigurationSettings) extends Actor with ScorexLogging {
 
-  // Limit for total cost of transactions to be re-checked. Hard-coded for now.
-  private val CostLimit = 7000000
-
   // Transaction can be re-checked only after this delay
   private val TimeLimit = nodeSettings.mempoolCleanupDuration.toMillis
 
@@ -36,8 +36,15 @@ class CleanupWorker(nodeViewHolderRef: ActorRef,
   override def receive: Receive = {
     case RunCleanup(validator, mempool) =>
       val s = sender()
-      validatePool(validator, mempool)
-        .map { case (validated, toEliminate) =>
+      Future {
+        CleanupWorker.validatePool(
+          validator = validator,
+          mempool = mempool,
+          maxTransactionCost = nodeSettings.maxTransactionCost,
+          timeLimit = TimeLimit,
+          now = System.currentTimeMillis()
+        )
+      }.map { case CleanupWorker.CleanupResult(validated, toEliminate) =>
           log.debug(s"${validated.size} re-checked mempool transactions were ok, " +
             s"${toEliminate.size} transactions were invalidated")
 
@@ -56,22 +63,69 @@ class CleanupWorker(nodeViewHolderRef: ActorRef,
     case a: Any => log.warn(s"Strange input: $a")
   }
 
+}
+
+object CleanupWorker extends ScorexLogging {
+
   /**
-    * Validates transactions from mempool for some specified amount of time.
     *
-    * @return - updated valid transactions and invalidated transaction ids
+    * A command to run (partial) memory pool cleanup
+    *
+    * @param validator - a state implementation which provides transaction validation
+    * @param mempool - mempool reader instance
+    */
+  case class RunCleanup(validator: UtxoStateReader, mempool: ErgoMemPoolReader)
+
+  /**
+    * Outcome of a memory pool re-validation pass.
+    *
+    * @param validated   - transactions which are still valid, with their costs updated
+    * @param invalidated - ids of transactions which are not valid anymore
+    */
+  case class CleanupResult(validated: Seq[UnconfirmedTransaction], invalidated: Seq[ModifierId])
+
+  // Limit for total cost of transactions to be re-checked. Hard-coded for now.
+  val CostLimit: Long = 7000000
+
+  /**
+    * Selects mempool transactions which were not re-checked recently enough.
+    *
+    * @param mempool   - mempool reader instance
+    * @param timeLimit - a transaction can be re-checked only after this delay, in milliseconds
+    * @param now       - current time, in milliseconds
+    * @return transactions to be re-validated, sorted by priority (a parent comes before its children)
     */
-  private def validatePool(validator: UtxoStateReader,
-                           mempool: ErgoMemPoolReader): Future[(Seq[UnconfirmedTransaction], Seq[ModifierId])] = Future {
+  def transactionsToValidate(mempool: ErgoMemPoolReader,
+                             timeLimit: Long,
+                             now: Long): Seq[UnconfirmedTransaction] =
+    mempool.getAllPrioritized.filter { utx =>
+      (now - utx.lastCheckedTime) > timeLimit
+    }
 
-    val now = System.currentTimeMillis()
+  /**
+    * Validates transactions from the memory pool, until `costLimit` of accumulated cost is reached.
+    *
+    * This is a pure function of its arguments: it does not read the clock and does not touch the
+    * node view, so it can be called directly from tests.
+    *
+    * @param validator          - a state implementation which provides transaction validation
+    * @param mempool            - mempool reader instance
+    * @param maxTransactionCost - maximum cost of a single transaction
+    * @param timeLimit          - a transaction can be re-checked only after this delay, in milliseconds
+    * @param now                - current time, in milliseconds
+    * @param costLimit          - limit for total cost of transactions to be re-checked
+    * @return - updated valid transactions and invalidated transaction ids
+    */
+  def validatePool(validator: UtxoStateReader,
+                   mempool: ErgoMemPoolReader,
+                   maxTransactionCost: Int,
+                   timeLimit: Long,
+                   now: Long,
+                   costLimit: Long = CostLimit): CleanupResult = {
 
     // 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
-
+    val txsToValidate = transactionsToValidate(mempool, timeLimit, now).toList
 
     // Take into account other transactions from the pool.
     // This provides possibility to validate transactions which are spending off-chain outputs.
@@ -85,9 +139,9 @@ class CleanupWorker(nodeViewHolderRef: ActorRef,
                        costAcc: Long
                       ): (mutable.ArrayBuilder[UnconfirmedTransaction], mutable.ArrayBuilder[ModifierId]) = {
       txs match {
-        case head :: tail if costAcc < CostLimit =>
+        case head :: tail if costAcc < costLimit =>
           val validationContext = state.stateContext.simplifiedUpcoming()
-          state.validateWithCost(head.transaction, validationContext, nodeSettings.maxTransactionCost, None) match {
+          state.validateWithCost(head.transaction, validationContext, maxTransactionCost, None) match {
             case Success(txCost) =>
               val updTx = head.withCost(txCost)
               validationLoop(tail, validated += updTx, invalidated, txCost + costAcc)
@@ -102,20 +156,7 @@ class CleanupWorker(nodeViewHolderRef: ActorRef,
     }
 
     val res = validationLoop(txsToValidate, mutable.ArrayBuilder.make(), mutable.ArrayBuilder.make(), 0L)
-    wrapRefArray(res._1.result()) -> wrapRefArray(res._2.result())
+    CleanupResult(wrapRefArray(res._1.result()), wrapRefArray(res._2.result()))
   }
 
 }
-
-object CleanupWorker {
-
-  /**
-    *
-    * A command to run (partial) memory pool cleanup
-    *
-    * @param validator - a state implementation which provides transaction validation
-    * @param mempool - mempool reader instance
-    */
-  case class RunCleanup(validator: UtxoStateReader, mempool: ErgoMemPoolReader)
-
-}

src/test/scala/org/ergoplatform/local/CleanupWorkerSpec.scala

@@ -0,0 +1,119 @@
+package org.ergoplatform.local
+
+import org.ergoplatform.modifiers.mempool.{ErgoTransaction, UnconfirmedTransaction}
+import org.ergoplatform.nodeView.state.UtxoState
+import org.ergoplatform.settings.ErgoSettings
+import org.ergoplatform.utils.{ErgoTestHelpers, MempoolTestHelpers, NodeViewTestOps, RandomWrapper}
+import org.scalatest.flatspec.AnyFlatSpec
+
+/**
+  * Tests for the memory pool re-validation logic extracted from the `CleanupWorker` actor.
+  */
+class CleanupWorkerSpec extends AnyFlatSpec with NodeViewTestOps with ErgoTestHelpers with MempoolTestHelpers {
+
+  import org.ergoplatform.utils.ErgoNodeTestConstants._
+  import org.ergoplatform.utils.generators.ErgoNodeTransactionGenerators._
+  import org.ergoplatform.utils.generators.ValidBlocksGenerators._
+
+  private val settingsToTest: ErgoSettings = settings
+  private val maxTransactionCost: Int = settingsToTest.nodeSettings.maxTransactionCost
+
+  private val TimeLimit = 1000L
+  private val Now = 1000000L
+  // the production limit would truncate the pools used here, which is exercised by its own test below
+  private val NoCostLimit = Long.MaxValue
+
+  /** Transaction last checked `age` milliseconds ago */
+  private def unconfirmed(tx: ErgoTransaction, age: Long): UnconfirmedTransaction =
+    new UnconfirmedTransaction(tx, lastCost = None, createdTime = Now - age,
+      lastCheckedTime = Now - age, transactionBytes = Some(tx.bytes), source = None)
+
+  /** A state with one block applied, and transactions which are valid against it */
+  private def stateWithValidTxs: (UtxoState, Seq[ErgoTransaction]) = {
+    val (us0, bh0) = createUtxoState(settingsToTest)
+    val (genesisTxs, bh1) = validTransactionsFromBoxHolder(bh0)
+    val block = validFullBlock(None, us0, genesisTxs)
+    val us = us0.applyModifier(block, None)(_ => ()).get
+
+    val boxes = bh1.boxes.values.toList.filter(_.proposition != genesisEmissionBox.proposition)
+    (us, validTransactionsFromBoxes(200000, boxes, new RandomWrapper)._1)
+  }
+
+  /** Transactions spending inputs which do not exist in any state */
+  private def invalidTxs(n: Int): Seq[ErgoTransaction] =
+    (1 to n).map(_ => invalidErgoTransactionGen.sample.get)
+
+  it should "not re-check transactions checked recently enough" in {
+    val (us, validTxs) = stateWithValidTxs
+    val fresh = validTxs.map(tx => unconfirmed(tx, age = TimeLimit / 2))
+
+    CleanupWorker.transactionsToValidate(new FakeMempool(fresh), TimeLimit, Now) shouldBe empty
+
+    val result = CleanupWorker.validatePool(us, new FakeMempool(fresh), maxTransactionCost, TimeLimit, Now)
+    result.validated shouldBe empty
+    result.invalidated shouldBe empty
+  }
+
+  it should "re-check transactions which are stale enough" in {
+    val (_, validTxs) = stateWithValidTxs
+    val stale = validTxs.map(tx => unconfirmed(tx, age = TimeLimit * 2))
+    val fresh = validTxs.map(tx => unconfirmed(tx, age = 0))
+
+    CleanupWorker.transactionsToValidate(new FakeMempool(stale), TimeLimit, Now).map(_.id) shouldBe stale.map(_.id)
+    CleanupWorker.transactionsToValidate(new FakeMempool(fresh), TimeLimit, Now) shouldBe empty
+    // the boundary is exclusive: exactly `TimeLimit` old is not stale yet
+    CleanupWorker.transactionsToValidate(
+      new FakeMempool(validTxs.map(tx => unconfirmed(tx, age = TimeLimit))), TimeLimit, Now) shouldBe empty
+  }
+
+  it should "keep still-valid transactions and update their cost" in {
+    val (us, validTxs) = stateWithValidTxs
+    validTxs.nonEmpty shouldBe true
+    val pool = validTxs.map(tx => unconfirmed(tx, age = TimeLimit * 2))
+
+    val result = CleanupWorker.validatePool(us, new FakeMempool(pool), maxTransactionCost, TimeLimit, Now, NoCostLimit)
+
+    result.invalidated shouldBe empty
+    result.validated.map(_.id) should contain theSameElementsAs pool.map(_.id)
+    // cost is unknown before the check and is filled in by it
+    pool.forall(_.lastCost.isEmpty) shouldBe true
+    result.validated.forall(_.lastCost.exists(_ > 0)) shouldBe true
+  }
+
+  it should "invalidate transactions which are not valid anymore" in {
+    val (us, _) = stateWithValidTxs
+    val broken = invalidTxs(4)
+    val pool = broken.map(tx => unconfirmed(tx, age = TimeLimit * 2))
+
+    val result = CleanupWorker.validatePool(us, new FakeMempool(pool), maxTransactionCost, TimeLimit, Now, NoCostLimit)
+
+    result.validated shouldBe empty
+    result.invalidated should contain theSameElementsAs pool.map(_.id)
+  }
+
+  it should "report both validated and invalidated transactions of a mixed pool" in {
+    val (us, validTxs) = stateWithValidTxs
+    val broken = invalidTxs(3)
+    val pool = (validTxs ++ broken).map(tx => unconfirmed(tx, age = TimeLimit * 2))
+
+    val result = CleanupWorker.validatePool(us, new FakeMempool(pool), maxTransactionCost, TimeLimit, Now, NoCostLimit)
+
+    result.validated.map(_.id) should contain theSameElementsAs validTxs.map(_.id)
+    result.invalidated should contain theSameElementsAs broken.map(_.id)
+  }
+
+  it should "stop validating once the cost limit is reached" in {
+    val (us, validTxs) = stateWithValidTxs
+    // more than one transaction is needed for the limit to be observable
+    validTxs.size > 1 shouldBe true
+    val pool = validTxs.map(tx => unconfirmed(tx, age = TimeLimit * 2))
+
+    // costAcc starts at 0, so the first transaction is always checked and then the limit stops the loop
+    val limited = CleanupWorker.validatePool(us, new FakeMempool(pool), maxTransactionCost, TimeLimit, Now, costLimit = 1L)
+    limited.validated.size + limited.invalidated.size shouldBe 1
+
+    val unlimited = CleanupWorker.validatePool(us, new FakeMempool(pool), maxTransactionCost, TimeLimit, Now, NoCostLimit)
+    unlimited.validated.size + unlimited.invalidated.size shouldBe pool.size
+  }
+
+}