diff --git a/src/main/scala/org/ergoplatform/local/CleanupWorker.scala b/src/main/scala/org/ergoplatform/local/CleanupWorker.scala index ecc9c16e00..501e7d94da 100644 --- a/src/main/scala/org/ergoplatform/local/CleanupWorker.scala +++ b/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) - -} diff --git a/src/test/scala/org/ergoplatform/local/CleanupWorkerSpec.scala b/src/test/scala/org/ergoplatform/local/CleanupWorkerSpec.scala new file mode 100644 index 0000000000..566cadd8ac --- /dev/null +++ b/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 + } + +}