From 7c753910bf28b7c69f69228e120e101de2d83bb7 Mon Sep 17 00:00:00 2001 From: "A. Shannon" <217575710+a-shannon@users.noreply.github.com> Date: Tue, 8 Sep 2026 12:28:47 +0200 Subject: [PATCH 1/9] Extract shared test fixtures and convergence observations Keep synchronization scenarios with their correction in #2511. Reuse the spendable sorting fixture already present in #2480. --- .../it/util/ConvergenceObservations.scala | 100 ++++++++++++++++++ .../it/util/ConvergenceObservationsSpec.scala | 96 +++++++++++++++++ .../mining/CandidateGeneratorSpec.scala | 88 +++++++-------- .../mempool/ErgoNodeTransactionSpec.scala | 43 ++++---- .../nodeView/mempool/ErgoMemPoolSpec.scala | 19 ++-- .../viewholder/ErgoNodeViewHolderSpec.scala | 4 +- .../ErgoNodeTransactionGenerators.scala | 8 +- .../FundedBoxHolderGeneratorsSpec.scala | 45 ++++++++ 8 files changed, 325 insertions(+), 78 deletions(-) create mode 100644 src/it/scala/org/ergoplatform/it/util/ConvergenceObservations.scala create mode 100644 src/it/scala/org/ergoplatform/it/util/ConvergenceObservationsSpec.scala create mode 100644 src/test/scala/org/ergoplatform/utils/generators/FundedBoxHolderGeneratorsSpec.scala diff --git a/src/it/scala/org/ergoplatform/it/util/ConvergenceObservations.scala b/src/it/scala/org/ergoplatform/it/util/ConvergenceObservations.scala new file mode 100644 index 0000000000..09ce30420b --- /dev/null +++ b/src/it/scala/org/ergoplatform/it/util/ConvergenceObservations.scala @@ -0,0 +1,100 @@ +package org.ergoplatform.it.util + +import java.util.concurrent.{ScheduledThreadPoolExecutor, ThreadFactory, TimeUnit, TimeoutException} +import org.ergoplatform.it.api.NodeApi.NodeInfo + +import scala.concurrent.{ExecutionContext, Future, Promise} +import scala.concurrent.duration._ +import scala.util.{Failure, Success, Try} + +/** Bounded, single-flight observations for integration assertions. */ +final class ConvergenceObservations(implicit ec: ExecutionContext) extends AutoCloseable { + private val timer = new ScheduledThreadPoolExecutor(1, new ThreadFactory { + override def newThread(runnable: Runnable): Thread = { + val thread = new Thread(runnable, "convergence-observations") + thread.setDaemon(true) + thread + } + }) + timer.setRemoveOnCancelPolicy(true) + + private def bounded[A](future: Future[A], budget: FiniteDuration): Future[A] = { + val result = Promise[A]() + val timeout = timer.schedule(new Runnable { + override def run(): Unit = result.tryFailure(new TimeoutException("Observation deadline")) + }, math.max(0L, budget.toNanos), TimeUnit.NANOSECONDS) + future.onComplete { value => + result.tryComplete(value) + timeout.cancel(false) + } + result.future + } + + final class Probe[A](request: () => Future[A]) { + private var pending: Option[Future[A]] = None + + def sample(budget: FiniteDuration): Future[Either[String, A]] = { + val response = synchronized { + val current = pending.filterNot(_.isCompleted).getOrElse { + Try(request()) match { + case Success(value) => value + case Failure(error) => Future.failed(error) + } + } + pending = Some(current) + current + } + bounded(response, budget).map(value => Right(value): Either[String, A]).recover { + case scala.util.control.NonFatal(error) => Left(error.getClass.getSimpleName) + } + } + } + + def probe[A](request: => Future[A]): Probe[A] = new Probe(() => request) + + def until[A](deadline: Deadline, interval: FiniteDuration, sampleBudget: FiniteDuration) + (observe: FiniteDuration => Future[A])(accept: A => Boolean) + (failure: => String): Future[A] = { + def expired: Future[A] = Future.failed(new TimeoutException(failure)) + + def loop(): Future[A] = { + if (deadline.isOverdue()) expired + else { + val remaining = deadline.timeLeft + bounded(observe(sampleBudget.min(remaining)), remaining).flatMap { value => + if (deadline.isOverdue()) expired + else if (accept(value)) Future.successful(value) + else { + val next = Promise[Unit]() + timer.schedule(new Runnable { + override def run(): Unit = next.trySuccess(()) + }, interval.min(deadline.timeLeft).max(Duration.Zero).toNanos, TimeUnit.NANOSECONDS) + next.future.flatMap(_ => loop()) + } + }.recoverWith { + case _: TimeoutException => expired + } + } + } + + loop() + } + + override def close(): Unit = timer.shutdownNow() +} + +object ConvergenceObservations { + def sameBestBlock(infoA: NodeInfo, infoB: NodeInfo, minHeight: Int): Boolean = { + val sameHeight = infoA.bestBlockHeightOpt.nonEmpty && infoA.bestBlockHeightOpt == infoB.bestBlockHeightOpt + val sameBlock = infoA.bestBlockIdOpt.nonEmpty && infoA.bestBlockIdOpt == infoB.bestBlockIdOpt + val highEnough = infoA.bestBlockHeightOpt.exists(_ >= minHeight) + sameHeight && sameBlock && highEnough + } + + def selectedHeadersAgree(headers: Seq[Seq[String]]): Boolean = + headers.nonEmpty && headers.forall(_.headOption.exists(_.nonEmpty)) && + headers.map(_.head).distinct.size == 1 + + def headerId(value: String): String = + if (value.matches("[0-9a-fA-F]{64}")) value else "invalid-header-id" +} diff --git a/src/it/scala/org/ergoplatform/it/util/ConvergenceObservationsSpec.scala b/src/it/scala/org/ergoplatform/it/util/ConvergenceObservationsSpec.scala new file mode 100644 index 0000000000..bdbcfcc3ed --- /dev/null +++ b/src/it/scala/org/ergoplatform/it/util/ConvergenceObservationsSpec.scala @@ -0,0 +1,96 @@ +package org.ergoplatform.it.util + +import java.util.concurrent.TimeoutException +import java.util.concurrent.atomic.AtomicInteger + +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import org.ergoplatform.it.api.NodeApi.NodeInfo + +import scala.concurrent.{Await, ExecutionContext, Future, Promise} +import scala.concurrent.duration._ + +class ConvergenceObservationsSpec extends AnyFlatSpec with Matchers { + implicit private val ec: ExecutionContext = ExecutionContext.global + + private def withObserver(test: ConvergenceObservations => Unit): Unit = { + val observer = new ConvergenceObservations + try test(observer) + finally observer.close() + } + + "Selected header agreement" should "accept retained alternatives only when every first ID agrees" in { + ConvergenceObservations.selectedHeadersAgree(Seq(Seq("a", "b"), Seq("a", "c"))) shouldBe true + ConvergenceObservations.selectedHeadersAgree(Seq(Seq("a", "b"), Seq("b", "a"))) shouldBe false + ConvergenceObservations.selectedHeadersAgree(Seq(Seq("a"), Seq.empty)) shouldBe false + ConvergenceObservations.selectedHeadersAgree(Seq(Seq(""), Seq(""))) shouldBe false + ConvergenceObservations.selectedHeadersAgree(Seq.empty) shouldBe false + } + + "Full block agreement" should "require both heights and IDs and the original minimum height" in { + val info = NodeInfo(Some("header"), Some("block"), Some(60), Some(50), None, None) + ConvergenceObservations.sameBestBlock(info, info, 50) shouldBe true + ConvergenceObservations.sameBestBlock(info, info, 51) shouldBe false + ConvergenceObservations.sameBestBlock(info, info.copy(bestBlockHeightOpt = Some(51)), 50) shouldBe false + ConvergenceObservations.sameBestBlock(info, info.copy(bestBlockIdOpt = Some("other")), 50) shouldBe false + ConvergenceObservations.sameBestBlock(info.copy(bestBlockHeightOpt = None), info, 50) shouldBe false + ConvergenceObservations.sameBestBlock(info, info.copy(bestBlockHeightOpt = None), 50) shouldBe false + ConvergenceObservations.sameBestBlock(info.copy(bestBlockIdOpt = None), info, 50) shouldBe false + ConvergenceObservations.sameBestBlock(info, info.copy(bestBlockIdOpt = None), 50) shouldBe false + } + + it should "resample the entire group until its current selections agree" in withObserver { observer => + val samples = new AtomicInteger() + val result = observer.until(2.seconds.fromNow, 1.millis, 100.millis) { _ => + val headers = if (samples.incrementAndGet() == 1) Seq(Seq("a", "b"), Seq("b", "a")) + else Seq(Seq("b", "a"), Seq("b")) + Future.successful(headers) + }(ConvergenceObservations.selectedHeadersAgree)("selected headers disagree") + Await.result(result, 3.seconds).map(_.head) shouldBe Seq("b", "b") + samples.get() shouldBe 2 + } + + it should "fail persistent disagreement within the original deadline with recent evidence" in withObserver { observer => + var recent = "none" + val result = observer.until(100.millis.fromNow, 1.millis, 20.millis) { _ => + recent = "node0=a node1=b" + Future.successful(Seq(Seq("a"), Seq("b"))) + }(ConvergenceObservations.selectedHeadersAgree)(s"last: $recent") + intercept[TimeoutException](Await.result(result, 2.seconds)).getMessage should include("node0=a node1=b") + } + + "Observation probes" should "bound a stalled endpoint and not start overlapping requests" in withObserver { observer => + val calls = new AtomicInteger() + val never = Promise[Int]() + val probe = observer.probe { calls.incrementAndGet(); never.future } + Await.result(probe.sample(20.millis), 2.seconds) shouldBe Left("TimeoutException") + Await.result(probe.sample(20.millis), 2.seconds) shouldBe Left("TimeoutException") + calls.get() shouldBe 1 + never.success(3) + Await.result(never.future, 2.seconds) shouldBe 3 + Await.result(probe.sample(100.millis), 2.seconds) shouldBe Right(3) + calls.get() shouldBe 2 + } + + it should "keep successful status available while the peer sample times out" in withObserver { observer => + val status = observer.probe(Future.successful(42)).sample(100.millis) + val peers = observer.probe(Promise[Int]().future).sample(30.millis) + Await.result(status, 2.seconds) shouldBe Right(42) + Await.result(status.zip(peers), 2.seconds) shouldBe (Right(42) -> Left("TimeoutException")) + } + + it should "retain only an error class for endpoint failures" in withObserver { observer => + val result = observer.probe(Future.failed[Int](new IllegalArgumentException("private diagnostic payload"))) + Await.result(result.sample(100.millis), 2.seconds) shouldBe Left("IllegalArgumentException") + } + + it should "enforce the convergence deadline even when observation never returns" in withObserver { observer => + val result = observer.until(30.millis.fromNow, 1.millis, 10.millis)(_ => Promise[Boolean]().future)(identity)("recent status") + intercept[TimeoutException](Await.result(result, 2.seconds)).getMessage shouldBe "recent status" + } + + "Observation identifiers" should "exclude arbitrary response text" in { + ConvergenceObservations.headerId("ab" * 32) shouldBe "ab" * 32 + ConvergenceObservations.headerId("unexpected response text") shouldBe "invalid-header-id" + } +} diff --git a/src/test/scala/org/ergoplatform/mining/CandidateGeneratorSpec.scala b/src/test/scala/org/ergoplatform/mining/CandidateGeneratorSpec.scala index c049a80872..9a5d26e4d3 100644 --- a/src/test/scala/org/ergoplatform/mining/CandidateGeneratorSpec.scala +++ b/src/test/scala/org/ergoplatform/mining/CandidateGeneratorSpec.scala @@ -892,74 +892,68 @@ class CandidateGeneratorSpec extends AnyFlatSpec with Matchers with ErgoTestHelp it should "ignore cached candidate when forced = true" in new TestKit(ActorSystem()) { val testProbe = new TestProbe(system) - system.eventStream.subscribe(testProbe.ref, newBlockSignal) + val viewHolderProbe = new TestProbe(system) val testDir = s"${defaultSettings.directory}-ignore-cache-${System.currentTimeMillis()}" - val settingsWithShortRegeneration: ErgoSettings = + val settingsForExplicitForcing: ErgoSettings = ErgoSettingsReader.read() .copy( nodeSettings = defaultSettings.nodeSettings - .copy(blockCandidateGenerationInterval = 1.millis), + // Keep automatic refresh outside this test of explicit forcing. + .copy(blockCandidateGenerationInterval = 1.hour), chainSettings = ErgoSettingsReader.read().chainSettings.copy(blockInterval = 1.seconds), directory = testDir ) - val viewHolderRef: ActorRef = ErgoNodeViewRef(settingsWithShortRegeneration) - val readersHolderRef: ActorRef = ErgoReadersHolderRef(viewHolderRef) + // Keep readers fixed: a later ChangedMempool event may legitimately regenerate the cache. + val (initialState, boxes) = createUtxoState(settingsForExplicitForcing) + val block = validFullBlock(None, initialState, boxes) + val state = initialState.applyModifier(block, None)(_ => ()).get + // Initialization computes mining-time averages from pairs of headers. + val history = historyWithBestFullBlock(Seq(block)) + val readers = Readers(history, state, ErgoMemPool.empty(settingsForExplicitForcing), walletStub) + val readersHolderRef = system.actorOf(Props(new FixedReadersHolder(readers))) val candidateGenerator: ActorRef = CandidateGenerator( defaultMinerSecret.publicImage, readersHolderRef, - viewHolderRef, - settingsWithShortRegeneration + viewHolderProbe.ref, + settingsForExplicitForcing ) - val powScheme = settingsWithShortRegeneration.chainSettings.powScheme - - // First mine a block to establish chain (needed for avg mining time calculation) - candidateGenerator.tell(GenerateCandidate(Seq.empty, reply = true, forced = false), testProbe.ref) - val initCandidate = testProbe.expectMsgPF(candidateGenDelay) { - case StatusReply.Success(c: Candidate) => c - } - val initBlock = powScheme - .proveCandidate(initCandidate.candidateBlock, defaultMinerSecret.w, 0, 1000) - .get - candidateGenerator.tell(initBlock.header.powSolution, testProbe.ref) - testProbe.fishForMessage(blockValidationDelay) { - case StatusReply.Success(()) => true - case FullBlockApplied(header) if header.id != initBlock.header.parentId => true - case _ => false - } + try { + // Get first candidate from the coherent, already applied block snapshot. + candidateGenerator.tell(GenerateCandidate(Seq.empty, reply = true, forced = false), testProbe.ref) + val candidate1 = testProbe.expectMsgPF(candidateGenDelay) { + case StatusReply.Success(c: Candidate) => c + } + candidate1.candidateBlock.parentOpt.map(_.id) shouldBe Some(block.header.id) - // Get first candidate after chain is established - candidateGenerator.tell(GenerateCandidate(Seq.empty, reply = true, forced = false), testProbe.ref) - val candidate1 = testProbe.expectMsgPF(candidateGenDelay) { - case StatusReply.Success(c: Candidate) => c - } + // Request with forced = false should return cached candidate immediately. + candidateGenerator.tell(GenerateCandidate(Seq.empty, reply = true, forced = false), testProbe.ref) + val candidate2 = testProbe.expectMsgPF(100.millis) { + case StatusReply.Success(c: Candidate) => c + } + candidate2 should be theSameInstanceAs candidate1 + candidate2.candidateBlock.timestamp shouldBe candidate1.candidateBlock.timestamp - // Request with forced = false should return cached candidate immediately - candidateGenerator.tell(GenerateCandidate(Seq.empty, reply = true, forced = false), testProbe.ref) - val candidate2 = testProbe.expectMsgPF(100.millis) { - case StatusReply.Success(c: Candidate) => c - } - // Should be the exact same cached candidate - candidate2.candidateBlock.timestamp shouldBe candidate1.candidateBlock.timestamp + // Request with forced = true should bypass cache and regenerate. + candidateGenerator.tell(GenerateCandidate(Seq.empty, reply = true, forced = true), testProbe.ref) + val candidate3 = testProbe.expectMsgPF(candidateGenDelay) { + case StatusReply.Success(c: Candidate) => c + } - // Request with forced = true should bypass cache and regenerate - candidateGenerator.tell(GenerateCandidate(Seq.empty, reply = true, forced = true), testProbe.ref) - val candidate3 = testProbe.fishForMessage(candidateGenDelay) { - case StatusReply.Success(_: Candidate) => true - case _: FullBlockApplied => false - } match { - case StatusReply.Success(c: Candidate) => c + // Identity proves regeneration even when the clock has not advanced. + candidate3 should not be theSameInstanceAs(candidate1) + candidate3.candidateBlock.parentOpt.map(_.id) shouldBe Some(block.header.id) + candidate3.candidateBlock.timestamp should be >= candidate1.candidateBlock.timestamp + } finally { + TestKit.shutdownActorSystem(system) + history.closeStorage() + state.closeStorage() } - - // candidate3 should have timestamp >= candidate1 (regenerated, possibly same or newer) - candidate3.candidateBlock.timestamp should be >= candidate1.candidateBlock.timestamp - - system.terminate() } it should "preserve previous candidate when forced regeneration occurs" in new TestKit(ActorSystem()) { diff --git a/src/test/scala/org/ergoplatform/modifiers/mempool/ErgoNodeTransactionSpec.scala b/src/test/scala/org/ergoplatform/modifiers/mempool/ErgoNodeTransactionSpec.scala index 7768778054..e1f8de0ab8 100644 --- a/src/test/scala/org/ergoplatform/modifiers/mempool/ErgoNodeTransactionSpec.scala +++ b/src/test/scala/org/ergoplatform/modifiers/mempool/ErgoNodeTransactionSpec.scala @@ -16,8 +16,6 @@ import org.ergoplatform.wallet.boxes.{ErgoBoxAssetExtractor, ErgoBoxSerializer} import org.ergoplatform.wallet.interpreter.TransactionHintsBag import org.ergoplatform.wallet.protocol.context.InputContext import org.scalacheck.Gen -import sigma.util.BenchmarkUtil -import scorex.crypto.hash.Blake2b256 import scorex.util.encode.Base16 import sigma.{Colls, VersionContext} import sigma.ast.ErgoTree.DefaultHeader @@ -326,34 +324,29 @@ class ErgoNodeTransactionSpec extends ErgoCorePropertyTest with ErgoCompilerHelp property("transaction with too many inputs should be rejected") { - //we assume that verifier must finish verification of any script in less time than 250K hash calculations - // (for the Blake2b256 hash function over a single block input) - val Timeout: Long = { - val hf = Blake2b256 - - //just in case to heat up JVM - (1 to 5000000).foreach(i => hf(s"$i-$i")) - - val t0 = System.currentTimeMillis() - (1 to 250000).foreach(i => hf(s"$i")) - val t = System.currentTimeMillis() - t - t0 - } + // This test used to calibrate a wall-clock budget by timing 250K Blake2b256 hashes and then + // assert that validation fits in it. On a loaded CI machine the calibration window and the + // measurement window get different shares of the CPU, so the assertions flipped at random + // (see issue #2095). What the node is actually protected by is the block cost limit, not the + // clock, so the assertions below are on cost, which is deterministic. val gen = validErgoTransactionGenTemplate(0, 0, 2000, trueLeafGen) val (from, tx) = gen.sample.get tx.statelessValidity().isSuccess shouldBe true - //check that spam transaction is being rejected quickly implicit val verifier: ErgoInterpreter = ErgoInterpreter(parameters) - val (validity, time0) = BenchmarkUtil.measureTime(tx.statefulValidity(from, IndexedSeq(), emptyStateContext)) + + // with the block cost limit in force, the spam transaction is rejected, and validation is + // aborted as soon as the accumulated cost passes the limit + val validity = tx.statefulValidity(from, IndexedSeq(), emptyStateContext) validity.isSuccess shouldBe false - assert(time0 <= Timeout) val cause = validity.failed.get.getMessage cause should startWith(ValidationRules.errorMessage(bsBlockTransactionsCost, "", emptyModifierId, ErgoTransaction.modifierTypeId).take(30)) - //check that spam transaction validation with no cost limit is indeed taking too much time + // with a cost limit high enough to let it through, the same transaction validates, and the cost + // it accumulates is far above the block limit - that gap is what makes it spam, and it does not + // depend on how fast the machine running the test is import Parameters._ val maxCost = (Int.MaxValue - 10) / 10 // cannot use Int.MaxValue directly due to overflow when it is converted to block cost val ps = Parameters(0, DefaultParameters.updated(MaxBlockCostIncrease, maxCost), emptyVSUpdate) @@ -365,11 +358,15 @@ class ErgoNodeTransactionSpec extends ErgoCorePropertyTest with ErgoCompilerHelp Array.fill(3)(0.toByte), ErgoValidationSettingsUpdate.empty, 0.toByte) - val (_, time) = BenchmarkUtil.measureTime( - tx.statefulValidity(from, IndexedSeq(), sc)(verifier) - ) - assert(time > Timeout) + val blockCostLimit = emptyStateContext.currentParameters.maxBlockCost + val fullCost = tx.statefulValidity(from, IndexedSeq(), sc)(verifier).get + // the generator produces exactly `maxInputs` inputs, so this ratio is stable (~4.4x as of today) + fullCost.toLong should be > (blockCostLimit.toLong * 2) + + // and it is rejected by any limit below its cost, one unit below included + val justBelow = stateContextWith(Parameters(0, DefaultParameters.updated(MaxBlockCostIncrease, fullCost - 1), emptyVSUpdate)) + tx.statefulValidity(from, IndexedSeq(), justBelow)(ErgoInterpreter(justBelow.currentParameters)) shouldBe 'failure } property("transaction cost") { diff --git a/src/test/scala/org/ergoplatform/nodeView/mempool/ErgoMemPoolSpec.scala b/src/test/scala/org/ergoplatform/nodeView/mempool/ErgoMemPoolSpec.scala index d35e22703d..86d25c1448 100644 --- a/src/test/scala/org/ergoplatform/nodeView/mempool/ErgoMemPoolSpec.scala +++ b/src/test/scala/org/ergoplatform/nodeView/mempool/ErgoMemPoolSpec.scala @@ -64,8 +64,13 @@ class ErgoMemPoolSpec extends AnyFlatSpec val (us, bh) = createUtxoState(settings) val genesis = validFullBlock(None, us, bh) val wus = WrappedUtxoState(us, bh, settings).applyModifier(genesis)(_ => ()).get - val inputBox = wus.takeBoxes(1).head - val feeOut = new ErgoBoxCandidate(inputBox.value, feeProp, creationHeight = 0) + val inputBox = wus.takeBoxes(100).find(_.ergoTree == TrueTree).get + val feeOut = new ErgoBoxCandidate( + inputBox.value, + feeProp, + creationHeight = 0, + additionalTokens = inputBox.additionalTokens + ) val tx = ErgoTransaction( IndexedSeq(new Input(inputBox.id, ProverResult.empty)), IndexedSeq(feeOut) @@ -79,8 +84,9 @@ class ErgoMemPoolSpec extends AnyFlatSpec mempoolSorting = SortingOption.FeePerByte, )) - var poolSize = ErgoMemPool.empty(sortBySizeSettings) - poolSize = poolSize.process(UnconfirmedTransaction(tx, None), wus)._1 + val (poolSize, sizeOutcome) = + ErgoMemPool.empty(sortBySizeSettings).process(UnconfirmedTransaction(tx, None), wus) + sizeOutcome.isInstanceOf[ProcessingOutcome.Accepted] shouldBe true val size = tx.size poolSize.pool.orderedTransactions.firstKey.weight shouldBe OrderedTxPool.weighted(tx, size).weight @@ -89,8 +95,9 @@ class ErgoMemPoolSpec extends AnyFlatSpec mempoolSorting = SortingOption.FeePerCycle, )) - var poolCost = ErgoMemPool.empty(sortByCostSettings) - poolCost = poolCost.process(UnconfirmedTransaction(tx, None), wus)._1 + val (poolCost, costOutcome) = + ErgoMemPool.empty(sortByCostSettings).process(UnconfirmedTransaction(tx, None), wus) + costOutcome.isInstanceOf[ProcessingOutcome.Accepted] shouldBe true val validationContext = wus.stateContext.simplifiedUpcoming() val cost = wus.validateWithCost(tx, validationContext, Int.MaxValue, None).get poolCost.pool.orderedTransactions.firstKey.weight shouldBe OrderedTxPool.weighted(tx, cost).weight diff --git a/src/test/scala/org/ergoplatform/nodeView/viewholder/ErgoNodeViewHolderSpec.scala b/src/test/scala/org/ergoplatform/nodeView/viewholder/ErgoNodeViewHolderSpec.scala index 2d8e8f4359..4172bb3a95 100644 --- a/src/test/scala/org/ergoplatform/nodeView/viewholder/ErgoNodeViewHolderSpec.scala +++ b/src/test/scala/org/ergoplatform/nodeView/viewholder/ErgoNodeViewHolderSpec.scala @@ -593,7 +593,9 @@ class ErgoNodeViewHolderSpec extends ErgoCorePropertyTest with NodeViewTestOps w val wusAfterGenesis = wus.applyModifier(genesis)(_ => ()).get // Create a valid tx that pays to a FalseTree output. - val box = wusAfterGenesis.takeBoxes(1).head + val box = wusAfterGenesis.takeBoxes(wusAfterGenesis.size) + .find(_.ergoTree == TrueTree) + .getOrElse(fail("Expected an unspent TrueTree box after genesis")) val validTx = validTransactionFromBoxes(IndexedSeq(box), outputsProposition = FalseTree) val validBlock = validFullBlock(Some(genesis), wusAfterGenesis, Seq(validTx)) diff --git a/src/test/scala/org/ergoplatform/utils/generators/ErgoNodeTransactionGenerators.scala b/src/test/scala/org/ergoplatform/utils/generators/ErgoNodeTransactionGenerators.scala index 5766cc1ad9..855a848f01 100644 --- a/src/test/scala/org/ergoplatform/utils/generators/ErgoNodeTransactionGenerators.scala +++ b/src/test/scala/org/ergoplatform/utils/generators/ErgoNodeTransactionGenerators.scala @@ -57,7 +57,13 @@ object ErgoNodeTransactionGenerators extends ScorexLogging { def ergoBoxGenForTokens(tokens: Seq[(TokenId, Long)], propositionGen: Gen[ErgoTree]): Gen[ErgoBox] = { - ergoBoxGen(propGen = propositionGen, tokensGen = Gen.oneOf(tokens, tokens), heightGen = EmptyHistoryHeight) + val minValue = BoxUtils.sufficientAmount(extendedParameters) + ergoBoxGen( + propGen = propositionGen, + tokensGen = Gen.oneOf(tokens, tokens), + valueGenOpt = Some(validValueGen.map(value => Math.max(value, minValue))), + heightGen = EmptyHistoryHeight + ) } def unspendableErgoBoxGen(minValue: Long = parameters.minValuePerByte * 200, diff --git a/src/test/scala/org/ergoplatform/utils/generators/FundedBoxHolderGeneratorsSpec.scala b/src/test/scala/org/ergoplatform/utils/generators/FundedBoxHolderGeneratorsSpec.scala new file mode 100644 index 0000000000..cf50e1d882 --- /dev/null +++ b/src/test/scala/org/ergoplatform/utils/generators/FundedBoxHolderGeneratorsSpec.scala @@ -0,0 +1,45 @@ +package org.ergoplatform.utils.generators + +import org.ergoplatform.utils.BoxUtils +import org.ergoplatform.utils.generators.ErgoCoreGenerators.trueLeafGen +import org.ergoplatform.utils.ErgoNodeTestConstants.extendedParameters +import org.ergoplatform.utils.generators.ErgoNodeTransactionGenerators._ +import org.scalacheck.Gen +import org.scalacheck.rng.Seed +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import sigmastate.eval.Extensions._ + +class FundedBoxHolderGeneratorsSpec extends AnyFlatSpec with Matchers { + "Generated box holders" should "fund small input groups and conserve their value" in { + val minimum = BoxUtils.sufficientAmount(extendedParameters) + // This seed includes a value below the transaction generator's conservative floor. + val holder = boxesHolderGenOfSize(5).pureApply(Gen.Parameters.default, Seed(803L)) + holder.size shouldBe 5 + val boxes = holder.boxes.values.toIndexedSeq + Seq(1, 2).foreach { inputCount => + boxes.grouped(inputCount).foreach { inputs => + val tx = validUnsignedTransactionFromBoxes(inputs, issueNew = false) + tx.inputs.map(_.boxId.toSeq) shouldBe inputs.map(_.id.toSeq) + tx.outputCandidates.map(_.value).sum shouldBe inputs.map(_.value).sum + tx.outputCandidates.foreach { output => + output.value should be >= minimum + output.additionalTokens.length shouldBe 0 + } + } + } + boxes.foreach(_.value should be >= minimum) + } + + it should "preserve supplied tokens when the node generator funds a box" in { + val token = Array.fill[Byte](32)(1).toTokenId + val box = ergoBoxGenForTokens(Seq(token -> 7L), trueLeafGen) + .pureApply(Gen.Parameters.default, Seed(803L)) + val tx = validUnsignedTransactionFromBoxes(IndexedSeq(box), issueNew = false) + val tokens = tx.outputCandidates.flatMap(_.additionalTokens.toArray) + box.value should be >= BoxUtils.sufficientAmount(extendedParameters) + tx.outputCandidates.map(_.value).sum shouldBe box.value + tokens.map(_._1.toArray.toSeq).distinct shouldBe Seq(token.toArray.toSeq) + tokens.map(_._2).sum shouldBe 7L + } +} From bf06cc62edcfc4480df184184baac5e58638792c Mon Sep 17 00:00:00 2001 From: "A. Shannon" <217575710+a-shannon@users.noreply.github.com> Date: Sun, 6 Sep 2026 22:52:13 +0200 Subject: [PATCH 2/9] Keep mining reward maturity in the block wallet checkpoint --- .../nodeView/wallet/WalletScanLogic.scala | 10 +- .../wallet/persistence/WalletRegistry.scala | 20 ++- .../nodeView/wallet/WalletScanLogicSpec.scala | 122 ++++++++++++++++++ 3 files changed, 143 insertions(+), 9 deletions(-) diff --git a/src/main/scala/org/ergoplatform/nodeView/wallet/WalletScanLogic.scala b/src/main/scala/org/ergoplatform/nodeView/wallet/WalletScanLogic.scala index 68eb204d9c..efc071a6bd 100644 --- a/src/main/scala/org/ergoplatform/nodeView/wallet/WalletScanLogic.scala +++ b/src/main/scala/org/ergoplatform/nodeView/wallet/WalletScanLogic.scala @@ -92,8 +92,7 @@ object WalletScanLogic extends ScorexLogging { val maxMiningHeight = height - walletVars.settings.miningRewardDelay val miningBoxes = registry.unspentBoxes(MiningScanId).filter(_.inclusionHeightOpt.getOrElse(0) <= maxMiningHeight) val resolvedBoxes = miningBoxes.map { tb => - registry.removeScan(tb.box.id, MiningScanId) - tb.copy(scans = Set(PaymentsScanId)) + tb.copy(scans = (tb.scans - MiningScanId) + PaymentsScanId) } val initialScanResults = ScanResults(resolvedBoxes, ArraySeq.empty, ArraySeq.empty) @@ -138,8 +137,9 @@ object WalletScanLogic extends ScorexLogging { val inpId = inp.boxId unspentBoxes.get(bytesToId(inpId)).flatMap { _ => - registry.getBox(inpId) - .orElse(scanResults.outputs.find(tb => tb.box.id.sameElements(inpId))) + // Prefer this block's association changes to the preceding registry state. + scanResults.outputs.find(tb => tb.box.id.sameElements(inpId)) + .orElse(registry.getBox(inpId)) } } } else { @@ -165,7 +165,7 @@ object WalletScanLogic extends ScorexLogging { } // function effects: updating registry and offchainRegistry datasets - registry.updateOnBlock(scanRes, blockId, height) + registry.updateOnBlock(scanRes, blockId, height, miningBoxes) .map { _ => //data needed to update the offchain-registry val walletUnspent = registry.walletUnspentBoxes() diff --git a/src/main/scala/org/ergoplatform/nodeView/wallet/persistence/WalletRegistry.scala b/src/main/scala/org/ergoplatform/nodeView/wallet/persistence/WalletRegistry.scala index f3e7deba48..572ed54899 100644 --- a/src/main/scala/org/ergoplatform/nodeView/wallet/persistence/WalletRegistry.scala +++ b/src/main/scala/org/ergoplatform/nodeView/wallet/persistence/WalletRegistry.scala @@ -255,18 +255,30 @@ class WalletRegistry(private val store: LDBVersionedStore)(ws: WalletSettings) e * @param scanResults - block scan data (outputs created and spent along with corresponding transactions) * @param blockId - block identifier * @param blockHeight - block height + * @param maturedBoxes - preceding versions of mining rewards resolved in scanResults.outputs */ - def updateOnBlock(scanResults: ScanResults, blockId: ModifierId, blockHeight: Int): Try[Unit] = { + def updateOnBlock(scanResults: ScanResults, + blockId: ModifierId, + blockHeight: Int, + maturedBoxes: Seq[TrackedBox] = Seq.empty): Try[Unit] = { + // Resolve mining associations in the same checkpoint as this block's outputs and spends. + val bag0 = removeBoxes(KeyValuePairsBag.empty, maturedBoxes) // first, put newly created outputs and related transactions into key-value bag cache ++= scanResults.outputs.map(b => b.boxId -> b) - val bag1 = putBoxes(KeyValuePairsBag.empty, scanResults.outputs) + val bag1 = putBoxes(bag0, scanResults.outputs) val bag2 = putTxs(bag1, scanResults.relatedTransactions) // process spent boxes val spentBoxesWithTx = scanResults.inputsSpent.map(t => t.inputTxId -> t.trackedBox) val bag3 = processSpentBoxes(bag2, spentBoxesWithTx, blockHeight) + // A shared reward can already belong to payments; its assets are already in the digest. + val previouslyWalletBoxIds = maturedBoxes.filter(_.scans.contains(PaymentsScanId)).map(_.boxId).toSet + val receivedWalletBoxes = scanResults.outputs.filter { tb => + tb.scans.contains(PaymentsScanId) && !previouslyWalletBoxIds.contains(tb.boxId) + } + // and update wallet digest updateDigest(bag3) { case WalletDigest(height, wBalance, wTokensSeq) => if (height + 1 != blockHeight) { @@ -279,7 +291,7 @@ class WalletRegistry(private val store: LDBVersionedStore)(ws: WalletSettings) e .foldLeft(Map.empty[EncodedTokenId, Long]) { case (acc, (id, amt)) => acc.updated(encodedTokenId(id), acc.getOrElse(encodedTokenId(id), 0L) + amt) } - val receivedTokensAmt = scanResults.outputs.filter(_.scans.contains(PaymentsScanId)) + val receivedTokensAmt = receivedWalletBoxes .flatMap(_.box.additionalTokens.toArray) .foldLeft(Map.empty[EncodedTokenId, Long]) { case (acc, (id, amt)) => acc.updated(encodedTokenId(id), acc.getOrElse(encodedTokenId(id), 0L) + amt) @@ -301,7 +313,7 @@ class WalletRegistry(private val store: LDBVersionedStore)(ws: WalletSettings) e } } - val receivedAmt = scanResults.outputs.filter(_.scans.contains(PaymentsScanId)).map(_.box.value).sum + val receivedAmt = receivedWalletBoxes.map(_.box.value).sum val newBalance = wBalance + receivedAmt - spentAmt if ((newBalance >= 0 && newTokensBalance.forall(_._2 >= 0)) || ws.testMnemonic.isDefined) Success(WalletDigest(blockHeight, newBalance, newTokensBalance.toSeq)) diff --git a/src/test/scala/org/ergoplatform/nodeView/wallet/WalletScanLogicSpec.scala b/src/test/scala/org/ergoplatform/nodeView/wallet/WalletScanLogicSpec.scala index 11bc1fe7f5..ea1cfc05f5 100644 --- a/src/test/scala/org/ergoplatform/nodeView/wallet/WalletScanLogicSpec.scala +++ b/src/test/scala/org/ergoplatform/nodeView/wallet/WalletScanLogicSpec.scala @@ -8,10 +8,15 @@ import org.ergoplatform.nodeView.wallet.persistence.{OffChainRegistry, WalletReg import org.ergoplatform.nodeView.wallet.scanning.{EqualsScanningPredicate, ScanRequest, ScanWalletInteraction} import org.ergoplatform.wallet.Constants import org.ergoplatform.wallet.Constants.ScanId +import org.ergoplatform.wallet.boxes.TrackedBox +import org.ergoplatform.core.VersionTag import org.ergoplatform.{ErgoBox, ErgoBoxCandidate, Input} import org.scalacheck.Gen import sigma.ast.{ByteArrayConstant, ErgoTree} import org.ergoplatform.settings.Constants.{FalseTree, TrueTree} +import scorex.util.bytesToId +import sigma.Colls +import sigmastate.eval.Extensions.ArrayByteOps import scala.util.Random @@ -121,6 +126,123 @@ class WalletScanLogicSpec extends ErgoCorePropertyTest with DBSpec with WalletTe } } + private def withMiningReward(scans: Set[ScanId], keepSpent: Boolean = true) + (check: (WalletRegistry, WalletVars, TrackedBox, VersionTag) => Unit): Unit = { + withVersionedStore(10) { store => + val registry = new WalletRegistry(store)(settings.walletSettings.copy(keepSpentBoxes = keepSpent)) + val walletVars = WalletVars(None, Seq.empty, Some(WalletCache(pubkeys, s)))(s) + val tokens = Colls.fromItems(Array.fill[Byte](32)(1).toTokenId -> 7L) + val output = new ErgoBoxCandidate(1000L, miningScripts.head, creationHeight = 1, additionalTokens = tokens) + val tx = new ErgoTransaction(fakeInputs, IndexedSeq.empty, IndexedSeq(output)) + val reward = TrackedBox(tx.id, 0, Some(1), None, None, tx.outputs.head, scans) + val previousId = bytesToId(versionId("before maturity")) + registry.updateOnBlock(WalletScanLogic.ScanResults(Seq(reward), Seq.empty, Seq.empty), previousId, 720).get + check(registry, walletVars, reward, VersionTag @@ previousId.toString) + } + } + + private def scanMaturityBlock(registry: WalletRegistry, walletVars: WalletVars, + transactions: Seq[ErgoTransaction] = Seq.empty): Unit = { + scanBlockTransactions(registry, OffChainRegistry.empty, walletVars, 721, + bytesToId(versionId("maturity block")), transactions, None, None, WalletProfile.User).get + } + + property("mining maturity leaves an immature reward unchanged") { + withMiningReward(Set(Constants.MiningScanId)) { (registry, walletVars, reward, _) => + scanBlockTransactions(registry, OffChainRegistry.empty, walletVars, 720, + bytesToId(versionId("immature block")), Seq.empty, None, None, WalletProfile.User).get + registry.getBox(reward.box.id) shouldBe Some(reward) + registry.walletUnspentBoxes() shouldBe empty + registry.fetchDigest().walletBalance shouldBe 0L + registry.fetchDigest().walletAssetBalances shouldBe empty + } + } + + Seq(false, true).foreach { shared => + val originalScans = if (shared) Set(Constants.MiningScanId, scanId) else Set(Constants.MiningScanId) + val resolvedScans = originalScans - Constants.MiningScanId + Constants.PaymentsScanId + + property(s"mining maturity preserves unspent reward associations and assets (shared=$shared)") { + withMiningReward(originalScans) { (registry, walletVars, reward, _) => + scanMaturityBlock(registry, walletVars) + registry.getBox(reward.box.id) shouldBe Some(reward.copy(scans = resolvedScans)) + registry.getBox(reward.box.id).get.scans shouldBe resolvedScans + registry.getBox(reward.box.id).get.inclusionHeightOpt shouldBe reward.inclusionHeightOpt + registry.unspentBoxes(Constants.MiningScanId) shouldBe empty + registry.boxesByInclusionHeight(Constants.MiningScanId, 0, 721) shouldBe empty + registry.walletUnspentBoxes().map(_.boxId) shouldBe Seq(reward.boxId) + if (shared) registry.unspentBoxes(scanId).map(_.boxId) shouldBe Seq(reward.boxId) + registry.fetchDigest().walletBalance shouldBe reward.value + registry.fetchDigest().walletAssetBalances.toMap shouldBe + Map(IdUtils.encodedTokenId(reward.box.additionalTokens(0)._1) -> 7L) + } + } + + Seq(false, true).foreach { keepSpent => + property(s"mining maturity recognizes current block spending (shared=$shared, history=$keepSpent)") { + withMiningReward(originalScans, keepSpent) { (registry, walletVars, reward, _) => + val spendingTx = new ErgoTransaction(IndexedSeq(Input(reward.box.id, emptyProverResult)), + IndexedSeq.empty, IndexedSeq(new ErgoBoxCandidate(reward.value, FalseTree, 721, + reward.box.additionalTokens))) + scanMaturityBlock(registry, walletVars, Seq(spendingTx)) + registry.walletUnspentBoxes() shouldBe empty + registry.unspentBoxes(Constants.MiningScanId) shouldBe empty + registry.unspentBoxes(scanId) shouldBe empty + registry.fetchDigest().walletBalance shouldBe 0L + registry.fetchDigest().walletAssetBalances shouldBe empty + registry.getTx(spendingTx.id).get.scanIds.toSet shouldBe resolvedScans + val expected = if (keepSpent) Some(reward.copy(scans = resolvedScans, + spendingHeightOpt = Some(721), spendingTxIdOpt = Some(spendingTx.id))) else None + registry.getBox(reward.box.id) shouldBe expected + expected.foreach { tb => + val stored = registry.getBox(reward.box.id).get + stored.scans shouldBe tb.scans + stored.inclusionHeightOpt shouldBe tb.inclusionHeightOpt + stored.spendingHeightOpt shouldBe tb.spendingHeightOpt + stored.spendingTxIdOpt shouldBe tb.spendingTxIdOpt + } + registry.walletSpentBoxes().map(_.boxId) shouldBe expected.toSeq.map(_.boxId) + if (shared) registry.spentBoxes(scanId).map(_.boxId) shouldBe expected.toSeq.map(_.boxId) + } + } + } + + Seq(false, true).foreach { spent => + property(s"mining maturity rollback restores preceding state (shared=$shared, spent=$spent)") { + withMiningReward(originalScans) { (registry, walletVars, reward, previousVersion) => + val previousDigest = registry.fetchDigest() + val transactions = if (spent) Seq(new ErgoTransaction( + IndexedSeq(Input(reward.box.id, emptyProverResult)), IndexedSeq.empty, + IndexedSeq(new ErgoBoxCandidate(reward.value, FalseTree, 721, reward.box.additionalTokens)))) + else Seq.empty + scanMaturityBlock(registry, walletVars, transactions) + registry.rollback(previousVersion).get + registry.getBox(reward.box.id) shouldBe Some(reward) + registry.getBox(reward.box.id).get.scans shouldBe originalScans + registry.getBox(reward.box.id).get.inclusionHeightOpt shouldBe reward.inclusionHeightOpt + registry.getBox(reward.box.id).get.spendingHeightOpt shouldBe None + registry.getBox(reward.box.id).get.spendingTxIdOpt shouldBe None + registry.unspentBoxes(Constants.MiningScanId) shouldBe Seq(reward) + registry.walletUnspentBoxes() shouldBe empty + registry.walletSpentBoxes() shouldBe empty + if (shared) registry.unspentBoxes(scanId) shouldBe Seq(reward) + registry.fetchDigest() shouldBe previousDigest + transactions.foreach(tx => registry.getTx(tx.id) shouldBe None) + } + } + } + } + + property("mining maturity does not credit an existing payment association twice") { + withMiningReward(Set(Constants.MiningScanId, Constants.PaymentsScanId, scanId)) { + (registry, walletVars, reward, _) => + val previousDigest = registry.fetchDigest() + scanMaturityBlock(registry, walletVars) + registry.fetchDigest() shouldBe previousDigest.copy(height = 721) + registry.getBox(reward.box.id).get.scans shouldBe Set(Constants.PaymentsScanId, scanId) + } + } + property("scanBlockTransactions") { withVersionedStore(10) { store => val walletVars = walletVarsGen.sample.get From 459f6bbd5d80f31dd55a3eb240377339a6c3622f Mon Sep 17 00:00:00 2001 From: "A. Shannon" <217575710+a-shannon@users.noreply.github.com> Date: Tue, 8 Sep 2026 11:59:01 +0200 Subject: [PATCH 3/9] Compose wallet scan metadata and write results with reward maturity --- .../nodeView/wallet/ErgoWalletActor.scala | 3 +- .../wallet/persistence/WalletRegistry.scala | 73 ++--- .../persistence/WalletRegistrySpec.scala | 149 +++++++++- .../WalletRegistryWriteResultSpec.scala | 257 ++++++++++++++++++ 4 files changed, 448 insertions(+), 34 deletions(-) create mode 100644 src/test/scala/org/ergoplatform/nodeView/wallet/persistence/WalletRegistryWriteResultSpec.scala diff --git a/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletActor.scala b/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletActor.scala index 78d7621a25..4191f61367 100644 --- a/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletActor.scala +++ b/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletActor.scala @@ -437,8 +437,7 @@ class ErgoWalletActor(settings: ErgoSettings, } case AddBox(box: ErgoBox, scanIds: Set[ScanId]) => - state.registry.updateScans(scanIds, box) - sender() ! AddBoxResponse(Success(())) // todo: what is the reasoning behind returning always success? + sender() ! AddBoxResponse(state.registry.updateScans(scanIds, box)) case StopTracking(scanId: ScanId, boxId: BoxId) => sender() ! StopTrackingResponse(state.registry.removeScan(boxId, scanId)) diff --git a/src/main/scala/org/ergoplatform/nodeView/wallet/persistence/WalletRegistry.scala b/src/main/scala/org/ergoplatform/nodeView/wallet/persistence/WalletRegistry.scala index 572ed54899..76f434228c 100644 --- a/src/main/scala/org/ergoplatform/nodeView/wallet/persistence/WalletRegistry.scala +++ b/src/main/scala/org/ergoplatform/nodeView/wallet/persistence/WalletRegistry.scala @@ -263,9 +263,12 @@ class WalletRegistry(private val store: LDBVersionedStore)(ws: WalletSettings) e maturedBoxes: Seq[TrackedBox] = Seq.empty): Try[Unit] = { // Resolve mining associations in the same checkpoint as this block's outputs and spends. - val bag0 = removeBoxes(KeyValuePairsBag.empty, maturedBoxes) + val maturityBag = removeBoxes(KeyValuePairsBag.empty, maturedBoxes) // first, put newly created outputs and related transactions into key-value bag - cache ++= scanResults.outputs.map(b => b.boxId -> b) + // An output may have been registered before its creating block was scanned. + val bag0 = scanResults.outputs.foldLeft(maturityBag) { (bag, output) => + getBox(output.box.id).map(previous => removeBox(bag, previous)).getOrElse(bag) + } val bag1 = putBoxes(bag0, scanResults.outputs) val bag2 = putTxs(bag1, scanResults.relatedTransactions) @@ -280,7 +283,7 @@ class WalletRegistry(private val store: LDBVersionedStore)(ws: WalletSettings) e } // and update wallet digest - updateDigest(bag3) { case WalletDigest(height, wBalance, wTokensSeq) => + val result = updateDigest(bag3) { case WalletDigest(height, wBalance, wTokensSeq) => if (height + 1 != blockHeight) { log.error(s"Blocks were skipped during wallet scanning, from $height until $blockHeight") } @@ -322,6 +325,10 @@ class WalletRegistry(private val store: LDBVersionedStore)(ws: WalletSettings) e }.flatMap { bag4 => bag4.transact(store, idToBytes(blockId)) } + // Reload affected boxes from storage after either outcome; never expose a prepared batch. + cache --= scanResults.outputs.map(_.boxId) + cache --= scanResults.inputsSpent.map(_.trackedBox.boxId) + result } def rollback(version: VersionTag): Try[Unit] = { @@ -335,21 +342,24 @@ class WalletRegistry(private val store: LDBVersionedStore)(ws: WalletSettings) e private[persistence] def processSpentBoxes(bag: KeyValuePairsBag, spentBoxes: Seq[(ModifierId, TrackedBox)], spendingHeight: Int): KeyValuePairsBag = { - if (keepHistory) { - val outSpent: Seq[TrackedBox] = spentBoxes.flatMap { case (_, tb) => - getBox(tb.box.id).orElse { - bag.toInsert.find(_._1.sameElements(boxKey(tb))).flatMap { case (_, tbBytes) => - TrackedBoxSerializer.parseBytesTry(tbBytes).toOption - } match { - case s@Some(_) => s - case None => - log.warn(s"Output spent hasn't found in the wallet: ${Algos.encode(tb.box.id)}, " + - s"could be okay if it was created before wallet init") - None - } - }: Option[TrackedBox] + val outSpent: Seq[TrackedBox] = spentBoxes.flatMap { case (_, tb) => + // The current block's output metadata takes precedence over an earlier registration. + bag.toInsert.find(_._1.sameElements(boxKey(tb))).flatMap { case (_, tbBytes) => + TrackedBoxSerializer.parseBytesTry(tbBytes).toOption + }.orElse(getBox(tb.box.id)) match { + case s@Some(_) => s + case None => + log.warn(s"Output spent hasn't found in the wallet: ${Algos.encode(tb.box.id)}, " + + s"could be okay if it was created before wallet init") + None } + } + val removalBoxes = spentBoxes.map { case (_, tb) => + outSpent.find(_.boxId == tb.boxId).getOrElse(tb) + } + val bagBeforePut = removeBoxes(bag, removalBoxes) + if (keepHistory) { val updatedBoxes = outSpent.map { tb => val spendingTxIdOpt = spentBoxes .find { case (_, x) => x.box.id.sameElements(tb.box.id) } @@ -357,13 +367,9 @@ class WalletRegistry(private val store: LDBVersionedStore)(ws: WalletSettings) e tb.copy(spendingHeightOpt = Some(spendingHeight), spendingTxIdOpt = spendingTxIdOpt) } - cache --= spentBoxes.map(_._2.boxId) - val bagBeforePut = removeBoxes(bag, spentBoxes.map(_._2)) - cache ++= updatedBoxes.map(b => b.boxId -> b) putBoxes(bagBeforePut, updatedBoxes) } else { - cache --= spentBoxes.map(_._2.boxId) - removeBoxes(bag, spentBoxes.map(_._2)) + bagBeforePut } } @@ -376,36 +382,41 @@ class WalletRegistry(private val store: LDBVersionedStore)(ws: WalletSettings) e * @param box - box to be updated (new version) * @return */ - def updateScans(newScans: Set[ScanId], box: ErgoBox): Try[Unit] = Try { + def updateScans(newScans: Set[ScanId], box: ErgoBox): Try[Unit] = { + val result = prepareScanUpdate(newScans, box).flatMap { bag => + bag.transact(store, store.lastVersionID.getOrElse(scorex.util.Random.randomBytes(32))) + } + cache.remove(bytesToId(box.id)) + result + } + + private def prepareScanUpdate(newScans: Set[ScanId], box: ErgoBox): Try[KeyValuePairsBag] = Try { val bag0 = KeyValuePairsBag.empty val oldBox = getBox(box.id) // read old version from the database val oldScans = oldBox.map(_.scans).getOrElse(Set.empty) - val newBox = TrackedBox(box, box.creationHeight, newScans) + val newBox = oldBox.map(_.copy(scans = newScans)) + .getOrElse(TrackedBox(box, box.creationHeight, newScans)) val bag1 = (oldScans.isEmpty, newScans.isEmpty) match { case (false, false) => // replace scans of the box by removing it along with indexes related to old scans, // and then adding the box with indexes related to the new scans - cache.update(oldBox.get.boxId, newBox) putBox(removeBox(bag0, oldBox.get), newBox) case (false, true) => // if new scans are empty, remove the box along with indexes - cache.remove(oldBox.get.boxId) removeBox(bag0, oldBox.get) case (true, false) => // if old scans are empty, add the box along with indexes - cache.put(newBox.boxId, newBox) putBox(bag0, newBox) case (true, true) => //old and new scans are empty, can't do anything useful throw new Exception("Can't remove a box which does not exist") } - // Flag showing that box has been added to the payments app (p2pk-wallet) or removed from it - // If true, we need to update wallet digest - val digestChanged = (oldScans.contains(Constants.PaymentsScanId) || newScans.contains(Constants.PaymentsScanId)) && - !(oldScans.contains(Constants.PaymentsScanId) && newScans.contains(Constants.PaymentsScanId)) + // Only unspent boxes contribute assets when their payment association changes. + val digestChanged = !newBox.isSpent && + (oldScans.contains(Constants.PaymentsScanId) != newScans.contains(Constants.PaymentsScanId)) val bag2 = if (digestChanged) { val digest = fetchDigest() @@ -434,7 +445,7 @@ class WalletRegistry(private val store: LDBVersionedStore)(ws: WalletSettings) e bag1 } - bag2.transact(store, store.lastVersionID.getOrElse(scorex.util.Random.randomBytes(32))) + bag2 } /** diff --git a/src/test/scala/org/ergoplatform/nodeView/wallet/persistence/WalletRegistrySpec.scala b/src/test/scala/org/ergoplatform/nodeView/wallet/persistence/WalletRegistrySpec.scala index 03ec0aa24b..62b577d642 100644 --- a/src/test/scala/org/ergoplatform/nodeView/wallet/persistence/WalletRegistrySpec.scala +++ b/src/test/scala/org/ergoplatform/nodeView/wallet/persistence/WalletRegistrySpec.scala @@ -4,6 +4,10 @@ import com.google.common.primitives.{Ints, Shorts} import org.ergoplatform.wallet.Constants.{PaymentsScanId, ScanId} import org.ergoplatform.db.DBSpec import org.ergoplatform.nodeView.wallet.WalletScanLogic.{ScanResults, SpentInputData} +import org.ergoplatform.nodeView.wallet.IdUtils +import org.ergoplatform.modifiers.mempool.ErgoTransaction +import org.ergoplatform.{ErgoBoxCandidate, Input} +import org.ergoplatform.settings.Constants.TrueTree import org.ergoplatform.wallet.boxes.TrackedBox import org.ergoplatform.core.VersionTag import org.scalacheck.Gen @@ -11,6 +15,9 @@ import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers import org.scalatestplus.scalacheck.ScalaCheckPropertyChecks import scorex.util.encode.Base16 +import scorex.crypto.authds.ADKey +import sigma.Colls +import sigmastate.eval.Extensions.ArrayByteOps import scala.collection.compat.immutable.ArraySeq import scala.util.Success @@ -21,6 +28,7 @@ class WalletRegistrySpec with DBSpec with ScalaCheckPropertyChecks { import org.ergoplatform.utils.ErgoNodeTestConstants._ + import org.ergoplatform.utils.ErgoCoreTestConstants.emptyProverResult import org.ergoplatform.utils.generators.ErgoNodeWalletGenerators._ import org.ergoplatform.utils.generators.CoreObjectGenerators._ import org.ergoplatform.utils.generators.ErgoNodeTransactionGenerators._ @@ -224,7 +232,103 @@ class WalletRegistrySpec } } - it should "update scans correctly" in { + it should "preserve inclusion metadata when updating scans of an existing unspent box" in { + val appId1: ScanId = ScanId @@ 21.toShort + val appId2: ScanId = ScanId @@ 22.toShort + + forAll(trackedBoxGen) { tb0 => + withVersionedStore(10) { store => + val inclusionHeight = if (tb0.box.creationHeight == 5) 6 else 5 + val existingBox = tb0.copy( + inclusionHeightOpt = Some(inclusionHeight), + spendingHeightOpt = None, + spendingTxIdOpt = None, + scans = Set(appId1)) + + WalletRegistry.putBox(emptyBag, existingBox).transact(store).get + val reg = new WalletRegistry(store)(ws) + + reg.updateScans(Set(appId2), existingBox.box).get + + val updatedBox = reg.getBox(existingBox.box.id).get + updatedBox.inclusionHeightOpt shouldBe existingBox.inclusionHeightOpt + updatedBox.spendingHeightOpt shouldBe existingBox.spendingHeightOpt + updatedBox.spendingTxIdOpt shouldBe existingBox.spendingTxIdOpt + updatedBox.scans shouldBe Set(appId2) + reg.unspentBoxes(appId1) shouldBe empty + reg.unspentBoxesByInclusionHeight(appId2, inclusionHeight, inclusionHeight) should have length 1 + } + } + } + + it should "preserve inclusion and spending metadata when updating scans of an existing spent box" in { + val appId1: ScanId = ScanId @@ 21.toShort + val appId2: ScanId = ScanId @@ 22.toShort + + forAll(trackedBoxGen, modifierIdGen) { case (tb0, spendingTxId) => + withVersionedStore(10) { store => + val inclusionHeight = if (tb0.box.creationHeight == 5) 6 else 5 + val existingBox = tb0.copy( + inclusionHeightOpt = Some(inclusionHeight), + spendingHeightOpt = Some(10), + spendingTxIdOpt = Some(spendingTxId), + scans = Set(appId1)) + + WalletRegistry.putBox(emptyBag, existingBox).transact(store).get + val reg = new WalletRegistry(store)(ws) + + reg.updateScans(Set(appId2), existingBox.box).get + + val updatedBox = reg.getBox(existingBox.box.id).get + updatedBox.inclusionHeightOpt shouldBe existingBox.inclusionHeightOpt + updatedBox.spendingHeightOpt shouldBe existingBox.spendingHeightOpt + updatedBox.spendingTxIdOpt shouldBe existingBox.spendingTxIdOpt + updatedBox.scans shouldBe Set(appId2) + reg.spentBoxes(appId1) shouldBe empty + reg.spentBoxes(appId2) should have length 1 + reg.spentBoxesByInclusionHeight(appId2, inclusionHeight, inclusionHeight) should have length 1 + } + } + } + + for (spent <- Seq(false, true); adding <- Seq(false, true)) { + it should s"account for payment scan membership according to spending state (spent=$spent, adding=$adding)" in { + withVersionedStore(10) { store => + val appId = ScanId @@ 21.toShort + val tokenId = Array.fill[Byte](32)(1).toTokenId + val tokens = Colls.fromItems(tokenId -> 7L) + val output = new ErgoBoxCandidate(1000L, TrueTree, 1, tokens) + val inputs = IndexedSeq(Input(ADKey @@ Array.fill(32)(0: Byte), emptyProverResult)) + val transaction = new ErgoTransaction(inputs, IndexedSeq.empty, IndexedSeq(output)) + val oldScans = if (adding) Set(appId) else Set(appId, PaymentsScanId) + val newScans = if (adding) Set(appId, PaymentsScanId) else Set(appId) + val tracked = TrackedBox(transaction.id, 0, Some(2), + if (spent) Some(transaction.id) else None, + if (spent) Some(3) else None, transaction.outputs.head, oldScans) + val encodedToken = IdUtils.encodedTokenId(tokenId) + val digest = WalletDigest(10, 5000L, Seq(encodedToken -> 20L)) + WalletRegistry.putDigest(WalletRegistry.putBox(emptyBag, tracked), digest).transact(store).get + val registry = new WalletRegistry(store)(ws) + + registry.updateScans(newScans, tracked.box).get + + val change = if (spent) 0 else if (adding) 1 else -1 + registry.fetchDigest().walletBalance shouldBe 5000L + change * 1000L + registry.fetchDigest().walletAssetBalances.toMap shouldBe Map(encodedToken -> (20L + change * 7L)) + registry.fetchDigest().height shouldBe 10 + val updated = registry.getBox(tracked.box.id).get + updated.scans shouldBe newScans + updated.inclusionHeightOpt shouldBe tracked.inclusionHeightOpt + updated.spendingHeightOpt shouldBe tracked.spendingHeightOpt + updated.spendingTxIdOpt shouldBe tracked.spendingTxIdOpt + registry.spentBoxes(appId).size shouldBe (if (spent) 1 else 0) + registry.unspentBoxes(appId).size shouldBe (if (spent) 0 else 1) + registry.walletUnspentBoxes().size shouldBe (if (!spent && adding) 1 else 0) + } + } + } + + it should "update non-payment scan indexes correctly" in { val appId1: ScanId = ScanId @@ 21.toShort val appId2: ScanId = ScanId @@ 22.toShort @@ -248,6 +352,49 @@ class WalletRegistrySpec } } + it should "construct an initially tracked record when updating scans for a new box" in { + val appId: ScanId = ScanId @@ 21.toShort + + forAll(trackedBoxGen) { tb => + withVersionedStore(10) { store => + val reg = new WalletRegistry(store)(ws) + + reg.updateScans(Set(appId), tb.box).get + + val insertedBox = reg.getBox(tb.box.id).get + insertedBox.creationTxId shouldBe tb.box.transactionId + insertedBox.creationOutIndex shouldBe tb.box.index + insertedBox.inclusionHeightOpt shouldBe Some(tb.box.creationHeight) + insertedBox.spendingHeightOpt shouldBe None + insertedBox.spendingTxIdOpt shouldBe None + insertedBox.scans shouldBe Set(appId) + reg.unspentBoxes(appId) should have length 1 + } + } + } + + it should "delete a tracked box when its final scan association is removed" in { + val appId: ScanId = ScanId @@ 21.toShort + + forAll(trackedBoxGen) { tb0 => + withVersionedStore(10) { store => + val existingBox = tb0.copy( + inclusionHeightOpt = Some(5), + spendingHeightOpt = None, + spendingTxIdOpt = None, + scans = Set(appId)) + WalletRegistry.putBox(emptyBag, existingBox).transact(store).get + val reg = new WalletRegistry(store)(ws) + + reg.updateScans(Set.empty, existingBox.box).get + + reg.getBox(existingBox.box.id) shouldBe None + reg.unspentBoxes(appId) shouldBe empty + reg.boxesByInclusionHeight(appId, 5, 5) shouldBe empty + } + } + } + it should "get unspent boxes by height from/to inclusive" in { val appId1: ScanId = ScanId @@ 21.toShort val appId2: ScanId = ScanId @@ 22.toShort diff --git a/src/test/scala/org/ergoplatform/nodeView/wallet/persistence/WalletRegistryWriteResultSpec.scala b/src/test/scala/org/ergoplatform/nodeView/wallet/persistence/WalletRegistryWriteResultSpec.scala new file mode 100644 index 0000000000..0e39742502 --- /dev/null +++ b/src/test/scala/org/ergoplatform/nodeView/wallet/persistence/WalletRegistryWriteResultSpec.scala @@ -0,0 +1,257 @@ +package org.ergoplatform.nodeView.wallet.persistence + +import akka.testkit.{TestActorRef, TestKit, TestProbe} +import org.ergoplatform.ErgoBoxCandidate +import org.ergoplatform.db.DBSpec +import org.ergoplatform.nodeView.wallet.ErgoWalletActorMessages.{AddBox, AddBoxResponse, ReadWallet} +import org.ergoplatform.nodeView.wallet.WalletScanLogic.{ScanResults, SpentInputData} +import org.ergoplatform.nodeView.wallet.{ErgoWalletActor, ErgoWalletServiceImpl, ErgoWalletState, WalletVars} +import org.ergoplatform.sdk.SecretString +import org.ergoplatform.settings.Constants.TrueTree +import org.ergoplatform.utils.ErgoNodeTestConstants.{extendedParameters, settings} +import org.ergoplatform.wallet.Constants.{MiningScanId, PaymentsScanId, ScanId} +import org.ergoplatform.wallet.boxes.TrackedBox +import org.ergoplatform.wallet.settings.SecretStorageSettings +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import scorex.db.LDBVersionedStore +import scorex.testkit.utils.AkkaFixture +import scorex.util.bytesToId + +import scala.util.{Failure, Success, Try} + +class WalletRegistryWriteResultSpec extends AnyFlatSpec with Matchers with DBSpec { + info(s"Registry persistence backend: ${scorex.db.LDBFactory.factory.asInstanceOf[scorex.db.StoreRegistry].factory.getClass.getName}") + private val appScan = ScanId @@ 11.toShort + private val nextScan = ScanId @@ 12.toShort + private val box = new ErgoBoxCandidate(1000000L, TrueTree, 1) + .toBox(bytesToId(Array.fill(32)(1.toByte)), 0) + private val tracked = TrackedBox(box, 1, Set(appScan)) + private val failure = new IllegalStateException("registry write sentinel") + + private class ControlledStore extends LDBVersionedStore(createTempDir, 10) { + var rejectWrites = false + override def update(versionID: Array[Byte], + toRemove: TraversableOnce[Array[Byte]], + toUpdate: TraversableOnce[(Array[Byte], Array[Byte])]): Try[Unit] = + if (rejectWrites) Failure(failure) else super.update(versionID, toRemove, toUpdate) + } + + private def withRegistry(keepSpent: Boolean = true)(body: (ControlledStore, WalletRegistry) => Unit): Unit = { + val store = new ControlledStore + val registry = new WalletRegistry(store)(settings.walletSettings.copy(keepSpentBoxes = keepSpent, testMnemonic = None)) + try body(store, registry) finally registry.close() + } + + for ((label, original, updated) <- Seq( + ("insert", Set.empty[ScanId], Set(appScan)), + ("replace", Set(appScan), Set(nextScan)), + ("remove", Set(appScan), Set.empty[ScanId]) + )) { + def prepare(store: ControlledStore, registry: WalletRegistry): Unit = { + if (original.nonEmpty) registry.updateScans(original, box).get + registry.getBox(box.id) + store.rejectWrites = true + } + + it should s"return the persistence failure for scan $label" in withRegistry() { (store, registry) => + prepare(store, registry) + registry.updateScans(updated, box) shouldBe Failure(failure) + } + + it should s"keep cached and persisted metadata aligned after failed scan $label" in withRegistry() { (store, registry) => + prepare(store, registry) + registry.updateScans(updated, box) + val persisted = new WalletRegistry(store)(settings.walletSettings).getBox(box.id) + registry.getBox(box.id).map(_.scans) shouldBe persisted.map(_.scans) + persisted.map(_.scans) shouldBe (if (original.isEmpty) None else Some(original)) + registry.unspentBoxes(appScan).map(_.scans) shouldBe persisted.toSeq.map(_.scans) + } + } + + it should "leave new outputs invisible when a block write fails" in withRegistry() { (store, registry) => + store.rejectWrites = true + registry.updateOnBlock(ScanResults(Seq(tracked), Seq.empty, Seq.empty), bytesToId(versionId("block")), 1) shouldBe Failure(failure) + registry.getBox(box.id) shouldBe None + registry.allUnspentBoxes() shouldBe empty + registry.fetchDigest() shouldBe WalletDigest.empty + } + + for (keepSpent <- Seq(true, false)) { + it should s"keep committed spending metadata after a failed block write with history $keepSpent" in withRegistry(keepSpent) { (store, registry) => + registry.updateScans(Set(appScan), box).get + registry.getBox(box.id).get.spendingHeightOpt shouldBe None + store.rejectWrites = true + val spent = SpentInputData(bytesToId(versionId("spend")), tracked) + registry.updateOnBlock(ScanResults(Seq.empty, Seq(spent), Seq.empty), bytesToId(versionId("block")), 2) shouldBe Failure(failure) + val cached = registry.getBox(box.id).get + cached.spendingHeightOpt shouldBe None + cached.spendingTxIdOpt shouldBe None + registry.unspentBoxes(appScan).map(_.spendingHeightOpt) shouldBe Seq(None) + } + } + + it should "leave outputs invisible when block digest validation fails" in withRegistry() { (_, registry) => + val spent = SpentInputData(bytesToId(versionId("spend")), tracked.copy(scans = Set(PaymentsScanId))) + registry.updateOnBlock(ScanResults(Seq(tracked), Seq(spent), Seq.empty), bytesToId(versionId("block")), 1).isFailure shouldBe true + registry.getBox(box.id) shouldBe None + registry.fetchDigest() shouldBe WalletDigest.empty + } + + it should "persist payment assets only when a failed scan write is retried successfully" in withRegistry() { (store, registry) => + registry.updateScans(Set(appScan), box).get + store.rejectWrites = true + registry.updateScans(Set(PaymentsScanId), box) shouldBe Failure(failure) + registry.fetchDigest() shouldBe WalletDigest.empty + registry.getBox(box.id).get.scans shouldBe Set(appScan) + store.rejectWrites = false + registry.updateScans(Set(PaymentsScanId), box) shouldBe Success(()) + registry.fetchDigest().walletBalance shouldBe box.value + registry.getBox(box.id).get.scans shouldBe Set(PaymentsScanId) + registry.unspentBoxes(appScan) shouldBe empty + registry.walletUnspentBoxes().map(_.boxId) shouldBe Seq(tracked.boxId) + } + + for (keepSpent <- Seq(true, false); createdInBlock <- Seq(true, false)) { + it should s"reload committed block metadata with history $keepSpent and new output $createdInBlock" in withRegistry(keepSpent) { (_, registry) => + if (!createdInBlock) { + registry.updateScans(Set(appScan), box).get + registry.getBox(box.id).get.spendingHeightOpt shouldBe None + } + val spendingTx = bytesToId(versionId("spend")) + val spent = SpentInputData(spendingTx, tracked) + val outputs = if (createdInBlock) Seq(tracked) else Seq.empty + registry.updateOnBlock(ScanResults(outputs, Seq(spent), Seq.empty), bytesToId(versionId("block")), 2) shouldBe Success(()) + if (keepSpent) { + val updated = registry.getBox(box.id).get + updated.scans shouldBe Set(appScan) + updated.inclusionHeightOpt shouldBe Some(1) + updated.spendingHeightOpt shouldBe Some(2) + updated.spendingTxIdOpt shouldBe Some(spendingTx) + registry.spentBoxes(appScan).map(_.spendingHeightOpt) shouldBe Seq(Some(2)) + } else { + registry.getBox(box.id) shouldBe None + } + registry.unspentBoxes(appScan) shouldBe empty + } + } + + for (keepSpent <- Seq(true, false); changedField <- Seq("height", "scans")) { + it should s"use current output $changedField for an overlapping stored spent box with history $keepSpent" in withRegistry(keepSpent) { (_, registry) => + registry.updateScans(Set(appScan), box).get + val previous = registry.getBox(box.id).get + val current = if (changedField == "height") previous.copy(inclusionHeightOpt = Some(2)) + else previous.copy(scans = Set(nextScan)) + val spendingTx = bytesToId(versionId("overlap-spend")) + val spent = SpentInputData(spendingTx, previous) + registry.updateOnBlock(ScanResults(Seq(current), Seq(spent), Seq.empty), bytesToId(versionId("overlap-block")), 2) shouldBe Success(()) + registry.unspentBoxes(appScan) shouldBe empty + registry.unspentBoxes(nextScan) shouldBe empty + if (keepSpent) { + val updated = registry.getBox(box.id).get + updated.inclusionHeightOpt shouldBe current.inclusionHeightOpt + updated.scans shouldBe current.scans + updated.spendingHeightOpt shouldBe Some(2) + updated.spendingTxIdOpt shouldBe Some(spendingTx) + val currentScan = current.scans.head + registry.spentBoxes(currentScan).map(_.boxId) shouldBe Seq(previous.boxId) + registry.boxesByInclusionHeight(currentScan, current.inclusionHeightOpt.get, current.inclusionHeightOpt.get) + .map(_.inclusionHeightOpt) shouldBe Seq(current.inclusionHeightOpt) + if (changedField == "height") registry.boxesByInclusionHeight(appScan, 1, 1) shouldBe empty + else registry.confirmedBoxes(appScan) shouldBe empty + } else { + registry.getBox(box.id) shouldBe None + registry.confirmedBoxes(appScan) shouldBe empty + registry.confirmedBoxes(nextScan) shouldBe empty + registry.boxesByInclusionHeight(appScan, 1, 2) shouldBe empty + registry.boxesByInclusionHeight(nextScan, 1, 2) shouldBe empty + } + } + } + + for (keepSpent <- Seq(true, false)) { + it should s"commit maturity and output cleanup after a failed write with history $keepSpent" in withRegistry(keepSpent) { (store, registry) => + registry.updateScans(Set(appScan), box).get + val previousOutput = registry.getBox(box.id).get + val rewardBox = new ErgoBoxCandidate(2000000L, TrueTree, 1) + .toBox(bytesToId(versionId("reward")), 0) + registry.updateScans(Set(MiningScanId, PaymentsScanId), rewardBox).get + val previousReward = registry.getBox(rewardBox.id).get + val previousDigest = registry.fetchDigest() + val maturedReward = previousReward.copy(scans = Set(PaymentsScanId)) + val currentOutput = previousOutput.copy(scans = Set(nextScan), inclusionHeightOpt = Some(2)) + val spendingTx = bytesToId(versionId("composition-spend")) + val scan = ScanResults(Seq(maturedReward, currentOutput), + Seq(SpentInputData(spendingTx, previousOutput)), Seq.empty) + val blockId = bytesToId(versionId("composition-block")) + + store.rejectWrites = true + registry.updateOnBlock(scan, blockId, 2, Seq(previousReward)) shouldBe Failure(failure) + registry.getBox(rewardBox.id).get.scans shouldBe previousReward.scans + registry.getBox(box.id).get.scans shouldBe previousOutput.scans + registry.getBox(box.id).get.inclusionHeightOpt shouldBe previousOutput.inclusionHeightOpt + registry.getBox(box.id).get.spendingHeightOpt shouldBe None + registry.unspentBoxes(MiningScanId).map(_.boxId) shouldBe Seq(previousReward.boxId) + registry.unspentBoxes(appScan).map(_.boxId) shouldBe Seq(previousOutput.boxId) + registry.fetchDigest() shouldBe previousDigest + + store.rejectWrites = false + registry.updateOnBlock(scan, blockId, 2, Seq(previousReward)) shouldBe Success(()) + registry.getBox(rewardBox.id).get.scans shouldBe Set(PaymentsScanId) + registry.unspentBoxes(MiningScanId) shouldBe empty + registry.boxesByInclusionHeight(MiningScanId, 1, 2) shouldBe empty + registry.confirmedBoxes(appScan) shouldBe empty + registry.boxesByInclusionHeight(appScan, 1, 2) shouldBe empty + registry.unspentBoxes(nextScan) shouldBe empty + registry.walletUnspentBoxes().map(_.boxId) shouldBe Seq(previousReward.boxId) + registry.fetchDigest() shouldBe previousDigest.copy(height = 2) + if (keepSpent) { + val persisted = registry.getBox(box.id).get + persisted.scans shouldBe currentOutput.scans + persisted.inclusionHeightOpt shouldBe currentOutput.inclusionHeightOpt + persisted.spendingHeightOpt shouldBe Some(2) + persisted.spendingTxIdOpt shouldBe Some(spendingTx) + registry.spentBoxes(nextScan).map(_.boxId) shouldBe Seq(previousOutput.boxId) + } else { + registry.getBox(box.id) shouldBe None + registry.confirmedBoxes(nextScan) shouldBe empty + } + } + } + + for (outcome <- Seq("success", "write failure", "validation failure")) { + it should s"reply to AddBox with the registry $outcome" in withRegistry() { (store, registry) => + val fixture = new AkkaFixture + implicit val system = fixture.system + val state = ErgoWalletState(null, None, registry, OffChainRegistry.empty, None, + WalletVars(None, Seq.empty), None, None, None, extendedParameters, 10, rescanInProgress = false) + val service = new ErgoWalletServiceImpl(settings) { + override def readWallet(state: ErgoWalletState, mnemonic: Option[SecretString], + keys: Option[Int], storage: SecretStorageSettings): ErgoWalletState = state + } + val actor = TestActorRef(new ErgoWalletActor(settings, extendedParameters, service, null, null) { + override def preStart(): Unit = () + }) + try { + val probe = TestProbe() + actor ! ReadWallet(state) + store.rejectWrites = outcome == "write failure" + val scans = if (outcome == "validation failure") Set.empty[ScanId] else Set(appScan) + actor.tell(AddBox(box, scans), probe.ref) + val reply = probe.expectMsgType[AddBoxResponse].status + outcome match { + case "success" => + reply shouldBe Success(()) + registry.getBox(box.id).get.scans shouldBe scans + case "write failure" => reply shouldBe Failure(failure) + case _ => + reply.isFailure shouldBe true + reply.failed.get.getMessage shouldBe "Can't remove a box which does not exist" + } + } finally { + system.stop(actor) + TestKit.shutdownActorSystem(system) + } + } + } +} From a772a5cbf666584db3ed981838dbc11223d502ed Mon Sep 17 00:00:00 2001 From: "A. Shannon" <217575710+a-shannon@users.noreply.github.com> Date: Sun, 13 Sep 2026 02:56:51 +0200 Subject: [PATCH 4/9] Retry cached block sections after remote headers --- .../nodeView/ErgoNodeViewHolder.scala | 3 + .../HeaderBodyCacheWakeupSpec.scala | 68 +++++++++++++++++++ 2 files changed, 71 insertions(+) create mode 100644 src/test/scala/org/ergoplatform/nodeView/viewholder/HeaderBodyCacheWakeupSpec.scala diff --git a/src/main/scala/org/ergoplatform/nodeView/ErgoNodeViewHolder.scala b/src/main/scala/org/ergoplatform/nodeView/ErgoNodeViewHolder.scala index 0363d7bd26..157ca08f38 100644 --- a/src/main/scala/org/ergoplatform/nodeView/ErgoNodeViewHolder.scala +++ b/src/main/scala/org/ergoplatform/nodeView/ErgoNodeViewHolder.scala @@ -369,6 +369,9 @@ abstract class ErgoNodeViewHolder[State <: ErgoState[State]](settings: ErgoSetti applyFromCacheLoop(headersCache) + // Newly accepted headers may unblock sections received before their headers. + applyFromCacheLoop(modifiersCache) + val cleared = headersCache.cleanOverfull() val upd = BlockSectionsProcessingCacheUpdate( headersCache.size, diff --git a/src/test/scala/org/ergoplatform/nodeView/viewholder/HeaderBodyCacheWakeupSpec.scala b/src/test/scala/org/ergoplatform/nodeView/viewholder/HeaderBodyCacheWakeupSpec.scala new file mode 100644 index 0000000000..12f9546a72 --- /dev/null +++ b/src/test/scala/org/ergoplatform/nodeView/viewholder/HeaderBodyCacheWakeupSpec.scala @@ -0,0 +1,68 @@ +package org.ergoplatform.nodeView.viewholder + +import akka.testkit.TestProbe +import org.ergoplatform.core.idToVersion +import org.ergoplatform.modifiers.BlockSection +import org.ergoplatform.network.ErgoNodeViewSynchronizerMessages.BlockSectionsProcessingCacheUpdate +import org.ergoplatform.nodeView.ErgoNodeViewHolder.ReceivableMessages.ModifiersFromRemote +import org.ergoplatform.nodeView.state.StateType +import org.ergoplatform.nodeView.state.wrapped.WrappedUtxoState +import org.ergoplatform.utils.{ErgoCorePropertyTest, NodeViewTestConfig, NodeViewTestOps} +import org.ergoplatform.utils.fixtures.NodeViewFixture +import org.ergoplatform.utils.generators.ValidBlocksGenerators._ + +import scala.concurrent.Await +import scala.concurrent.duration._ + +class HeaderBodyCacheWakeupSpec extends ErgoCorePropertyTest with NodeViewTestOps { + import org.ergoplatform.utils.ErgoCoreTestConstants.parameters + + Seq(StateType.Utxo, StateType.Digest).foreach { stateType => + property(s"remote header wakes already cached block sections in $stateType state") { + val fixture = new NodeViewFixture( + NodeViewTestConfig(stateType, verifyTransactions = true, popowBootstrap = false).toSettings, + parameters) + import fixture._ + val (generationState, boxes) = createUtxoState(fixture.settings) + try { + val prefix = validFullBlock(None, generationState, boxes) + val afterPrefix = WrappedUtxoState(generationState, boxes, fixture.settings) + .applyModifier(prefix)(_ => ()).get + val next = validFullBlock(Some(prefix), afterPrefix) + applyBlock(prefix).isSuccess shouldBe true + getCurrentState.version shouldBe idToVersion(prefix.id) + + val cacheProbe = TestProbe()(actorSystem) + actorSystem.eventStream.subscribe(cacheProbe.ref, classOf[BlockSectionsProcessingCacheUpdate]) + val sections: Seq[BlockSection] = Seq(next.blockTransactions, next.extension, next.adProofs.get) + sections.map(_.modifierTypeId).distinct.size shouldBe 3 + sections.zipWithIndex.foreach { case (section, index) => + nodeViewHolderRef ! ModifiersFromRemote(Seq(section)) + val cached = cacheProbe.expectMsgType[BlockSectionsProcessingCacheUpdate](5.seconds) + cached.blockSectionsCacheSize shouldBe index + 1 + cached.cleared._2 shouldBe empty + } + val beforeHeader = getCurrentView + beforeHeader.history.contains(next.header.id) shouldBe false + sections.foreach(section => beforeHeader.history.contains(section.id) shouldBe false) + beforeHeader.state.version shouldBe idToVersion(prefix.id) + + // The cache-update event is an actor-processing barrier, not another drain trigger. + // No body section is resent after its prerequisite header arrives. + nodeViewHolderRef ! ModifiersFromRemote(Seq(next.header)) + val afterHeader = cacheProbe.expectMsgType[BlockSectionsProcessingCacheUpdate](5.seconds) + afterHeader.headersCacheSize shouldBe 0 + val current = getCurrentView + current.history.getFullBlock(next.header).map(_.id) shouldBe Some(next.id) + sections.foreach(section => current.history.contains(section.id) shouldBe true) + current.history.bestFullBlockIdOpt shouldBe Some(next.id) + current.state.version shouldBe idToVersion(next.id) + current.state.rootDigest.toSeq shouldBe next.header.stateRoot.toSeq + afterHeader.blockSectionsCacheSize shouldBe 0 + } finally { + generationState.closeStorage() + Await.result(actorSystem.terminate(), 15.seconds) + } + } + } +} From 76e5ec4bfc377334fdced4bd60a61df06e6b4cb6 Mon Sep 17 00:00:00 2001 From: "A. Shannon" <217575710+a-shannon@users.noreply.github.com> Date: Sun, 13 Sep 2026 03:24:33 +0200 Subject: [PATCH 5/9] Share synchronizer fixture reader ownership and isolation --- ...rgoNodeViewSynchronizerSpecification.scala | 46 +++++++++++++++---- 1 file changed, 37 insertions(+), 9 deletions(-) diff --git a/src/test/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerSpecification.scala b/src/test/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerSpecification.scala index 3e82649f70..a226c9bde3 100644 --- a/src/test/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerSpecification.scala +++ b/src/test/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerSpecification.scala @@ -6,8 +6,9 @@ import org.ergoplatform.modifiers.history.header.{Header, HeaderSerializer} import org.ergoplatform.modifiers.{BlockSection, ErgoFullBlock} import org.ergoplatform.network.ErgoNodeViewSynchronizerMessages._ import org.ergoplatform.nodeView.ErgoNodeViewHolder +import org.ergoplatform.nodeView.ErgoNodeViewHolder.ReceivableMessages.GetNodeViewChanges import org.ergoplatform.nodeView.history.{ErgoHistory, ErgoHistoryReader, ErgoSyncInfoMessageSpec, ErgoSyncInfoV2} -import org.ergoplatform.nodeView.mempool.ErgoMemPool +import org.ergoplatform.nodeView.mempool.{ErgoMemPool, ErgoMemPoolReader} import org.ergoplatform.nodeView.state.wrapped.WrappedUtxoState import org.ergoplatform.nodeView.state.{StateType, UtxoState} import org.ergoplatform.sanity.ErgoSanity._ @@ -67,7 +68,27 @@ class ErgoNodeViewSynchronizerSpecification extends AnyPropSpec } } - class NodeViewHolderMock extends ErgoNodeViewHolder[UtxoState](settings) + private def isolatedNodeSettings(prototype: ErgoSettings): ErgoSettings = { + val directory = createTempDir + prototype.copy(directory = directory.getAbsolutePath, + walletSettings = prototype.walletSettings.copy(secretStorage = + prototype.walletSettings.secretStorage.copy( + secretDir = new java.io.File(directory, "keystore").getAbsolutePath))) + } + + class NodeViewHolderMock(nodeSettings: ErgoSettings) extends ErgoNodeViewHolder[UtxoState](nodeSettings) + + class InjectedReadersNodeViewHolder(nodeSettings: ErgoSettings, + injectedHistory: ErgoHistoryReader, + injectedMempool: ErgoMemPoolReader) extends NodeViewHolderMock(nodeSettings) { + // This fixture owns its history and mempool; startup replies must use those same readers. + override protected def getNodeViewChanges: Receive = { + case request: GetNodeViewChanges => + if (request.history) sender() ! ChangedHistory(injectedHistory) + super.getNodeViewChanges(request.copy(history = false, mempool = false)) + if (request.mempool) sender() ! ChangedMempool(injectedMempool) + } + } class SynchronizerMock(networkControllerRef: ActorRef, viewHolderRef: ActorRef, @@ -129,7 +150,7 @@ class ErgoNodeViewSynchronizerSpecification extends AnyPropSpec val h = localHistoryGen.sample.get @SuppressWarnings(Array("org.wartremover.warts.OptionPartial")) val s = localStateGen.sample.get - val settings = ErgoSettingsReader.read() + val settings = isolatedNodeSettings(ErgoSettingsReader.read()) val pool = ErgoMemPool.empty(settings) implicit val ec: ExecutionContextExecutor = system.dispatcher val ncProbe = TestProbe("NetworkControllerProbe") @@ -138,9 +159,7 @@ class ErgoNodeViewSynchronizerSpecification extends AnyPropSpec val syncTracker = ErgoSyncTracker(settings.scorexSettings.network) val deliveryTracker: DeliveryTracker = DeliveryTracker.empty(settings) - // each test should always start with empty history - deleteRecursive(ErgoHistory.historyDir(settings)) - val nodeViewHolderMockRef = system.actorOf(Props(new NodeViewHolderMock)) + val nodeViewHolderMockRef = system.actorOf(Props(new InjectedReadersNodeViewHolder(settings, h, pool))) val synchronizerMockRef = system.actorOf(Props( new SynchronizerMock( @@ -174,15 +193,14 @@ class ErgoNodeViewSynchronizerSpecification extends AnyPropSpec } class Synchronizer2Fixture extends AkkaFixture { + val settings: ErgoSettings = isolatedNodeSettings(org.ergoplatform.utils.ErgoNodeTestConstants.settings) implicit val ec: ExecutionContextExecutor = system.dispatcher val ncProbe = TestProbe("NetworkControllerProbe") val pchProbe = TestProbe("PeerHandlerProbe") val syncTracker = ErgoSyncTracker(settings.scorexSettings.network) val deliveryTracker: DeliveryTracker = DeliveryTracker.empty(settings) - // each test should always start with empty history - deleteRecursive(ErgoHistory.historyDir(settings)) - val nodeViewHolderMockRef = system.actorOf(Props(new NodeViewHolderMock)) + val nodeViewHolderMockRef = system.actorOf(Props(new NodeViewHolderMock(settings))) val synchronizerMockRef = system.actorOf(Props( new SynchronizerMock( @@ -510,6 +528,16 @@ class ErgoNodeViewSynchronizerSpecification extends AnyPropSpec withFixture { ctx => import ctx._ + // Replay a late startup request before observing the peer response. The fixture's + // injected history and mempool must not be replaced by the holder's empty readers. + val startupReaders = TestProbe("StartupReaders") + startupReaders.send(nodeViewHolder, + GetNodeViewChanges(history = true, state = false, vault = false, mempool = true)) + val changedHistory = startupReaders.expectMsgType[ChangedHistory](3.seconds) + val changedMempool = startupReaders.expectMsgType[ChangedMempool](3.seconds) + synchronizer ! changedHistory + synchronizer ! changedMempool + val sync = ErgoSyncInfoV2(Seq(altchain.last)) // Neighbour is sending From 0f9f75c1fc1df3ff3309482b83c0475a05cdd7a2 Mon Sep 17 00:00:00 2001 From: "A. Shannon" <217575710+a-shannon@users.noreply.github.com> Date: Sun, 13 Sep 2026 04:20:10 +0200 Subject: [PATCH 6/9] Share convergence predicates independently of wallet fixtures --- .../it/util/ConvergenceObservations.scala | 110 +++++++++++++ .../it/util/ConvergenceObservationsSpec.scala | 145 ++++++++++++++++++ 2 files changed, 255 insertions(+) create mode 100644 src/it/scala/org/ergoplatform/it/util/ConvergenceObservations.scala create mode 100644 src/it/scala/org/ergoplatform/it/util/ConvergenceObservationsSpec.scala diff --git a/src/it/scala/org/ergoplatform/it/util/ConvergenceObservations.scala b/src/it/scala/org/ergoplatform/it/util/ConvergenceObservations.scala new file mode 100644 index 0000000000..ca0f61b238 --- /dev/null +++ b/src/it/scala/org/ergoplatform/it/util/ConvergenceObservations.scala @@ -0,0 +1,110 @@ +package org.ergoplatform.it.util + +import java.util.concurrent.{ScheduledThreadPoolExecutor, ThreadFactory, TimeUnit, TimeoutException} +import org.ergoplatform.it.api.NodeApi.NodeInfo + +import scala.concurrent.{ExecutionContext, Future, Promise} +import scala.concurrent.duration._ +import scala.util.{Failure, Success, Try} + +/** Bounded, single-flight observations for integration assertions. */ +final class ConvergenceObservations(implicit ec: ExecutionContext) extends AutoCloseable { + private val timer = new ScheduledThreadPoolExecutor(1, new ThreadFactory { + override def newThread(runnable: Runnable): Thread = { + val thread = new Thread(runnable, "convergence-observations") + thread.setDaemon(true) + thread + } + }) + timer.setRemoveOnCancelPolicy(true) + + private def bounded[A](future: Future[A], budget: FiniteDuration): Future[A] = { + val result = Promise[A]() + val timeout = timer.schedule(new Runnable { + override def run(): Unit = result.tryFailure(new TimeoutException("Observation deadline")) + }, math.max(0L, budget.toNanos), TimeUnit.NANOSECONDS) + future.onComplete { value => + result.tryComplete(value) + timeout.cancel(false) + } + result.future + } + + final class Probe[A](request: () => Future[A]) { + private var pending: Option[Future[A]] = None + + def sample(budget: FiniteDuration): Future[Either[String, A]] = { + val response = synchronized { + val current = pending.filterNot(_.isCompleted).getOrElse { + Try(request()) match { + case Success(value) => value + case Failure(error) => Future.failed(error) + } + } + pending = Some(current) + current + } + bounded(response, budget).map(value => Right(value): Either[String, A]).recover { + case scala.util.control.NonFatal(error) => Left(error.getClass.getSimpleName) + } + } + } + + def probe[A](request: => Future[A]): Probe[A] = new Probe(() => request) + + def until[A](deadline: Deadline, interval: FiniteDuration, sampleBudget: FiniteDuration) + (observe: FiniteDuration => Future[A])(accept: A => Boolean) + (failure: => String): Future[A] = { + def expired: Future[A] = Future.failed(new TimeoutException(failure)) + + def loop(): Future[A] = { + if (deadline.isOverdue()) expired + else { + val remaining = deadline.timeLeft + bounded(observe(sampleBudget.min(remaining)), remaining).flatMap { value => + if (deadline.isOverdue()) expired + else if (accept(value)) Future.successful(value) + else { + val next = Promise[Unit]() + timer.schedule(new Runnable { + override def run(): Unit = next.trySuccess(()) + }, interval.min(deadline.timeLeft).max(Duration.Zero).toNanos, TimeUnit.NANOSECONDS) + next.future.flatMap(_ => loop()) + } + }.recoverWith { + case _: TimeoutException => expired + } + } + } + + loop() + } + + override def close(): Unit = timer.shutdownNow() +} + +object ConvergenceObservations { + def sameBestBlock(infoA: NodeInfo, infoB: NodeInfo, minHeight: Int): Boolean = { + val sameHeight = infoA.bestBlockHeightOpt.nonEmpty && infoA.bestBlockHeightOpt == infoB.bestBlockHeightOpt + val sameBlock = infoA.bestBlockIdOpt.nonEmpty && infoA.bestBlockIdOpt == infoB.bestBlockIdOpt + val highEnough = infoA.bestBlockHeightOpt.exists(_ >= minHeight) + sameHeight && sameBlock && highEnough + } + + def selectedHeadersAgree(headers: Seq[Seq[String]]): Boolean = + headers.nonEmpty && headers.forall(_.headOption.exists(_.nonEmpty)) && + headers.map(_.head).distinct.size == 1 + + /** Both nodes report mining disabled and share a completely applied selected header tip. */ + def sameFullyAppliedNonMiningBlock(infoA: NodeInfo, infoB: NodeInfo, minHeight: Int): Boolean = + infoA.isMining.contains(false) && infoB.isMining.contains(false) && + sameBestBlock(infoA, infoB, minHeight) && + infoA.bestBlockIdOpt.exists(_.nonEmpty) && + infoA.bestHeaderHeightOpt == infoA.bestBlockHeightOpt && + infoB.bestHeaderHeightOpt == infoB.bestBlockHeightOpt && + infoA.bestHeaderIdOpt == infoA.bestBlockIdOpt && + infoB.bestHeaderIdOpt == infoB.bestBlockIdOpt + + def headerId(value: String): String = + if (value.matches("[0-9a-fA-F]{64}")) value else "invalid-header-id" +} diff --git a/src/it/scala/org/ergoplatform/it/util/ConvergenceObservationsSpec.scala b/src/it/scala/org/ergoplatform/it/util/ConvergenceObservationsSpec.scala new file mode 100644 index 0000000000..b4ddf9b47c --- /dev/null +++ b/src/it/scala/org/ergoplatform/it/util/ConvergenceObservationsSpec.scala @@ -0,0 +1,145 @@ +package org.ergoplatform.it.util + +import java.util.concurrent.TimeoutException +import java.util.concurrent.atomic.AtomicInteger + +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import org.ergoplatform.it.api.NodeApi.NodeInfo + +import scala.concurrent.{Await, ExecutionContext, Future, Promise} +import scala.concurrent.duration._ + +class ConvergenceObservationsSpec extends AnyFlatSpec with Matchers { + implicit private val ec: ExecutionContext = ExecutionContext.global + + private def withObserver(test: ConvergenceObservations => Unit): Unit = { + val observer = new ConvergenceObservations + try test(observer) + finally observer.close() + } + + "Selected header agreement" should "accept retained alternatives only when every first ID agrees" in { + ConvergenceObservations.selectedHeadersAgree(Seq(Seq("a", "b"), Seq("a", "c"))) shouldBe true + ConvergenceObservations.selectedHeadersAgree(Seq(Seq("a", "b"), Seq("b", "a"))) shouldBe false + ConvergenceObservations.selectedHeadersAgree(Seq(Seq("a"), Seq.empty)) shouldBe false + ConvergenceObservations.selectedHeadersAgree(Seq(Seq(""), Seq(""))) shouldBe false + ConvergenceObservations.selectedHeadersAgree(Seq.empty) shouldBe false + } + + "Full block agreement" should "require both heights and IDs and the original minimum height" in { + val info = NodeInfo(Some("header"), Some("block"), Some(60), Some(50), None, None) + ConvergenceObservations.sameBestBlock(info, info, 50) shouldBe true + ConvergenceObservations.sameBestBlock(info, info, 51) shouldBe false + ConvergenceObservations.sameBestBlock(info, info.copy(bestBlockHeightOpt = Some(51)), 50) shouldBe false + ConvergenceObservations.sameBestBlock(info, info.copy(bestBlockIdOpt = Some("other")), 50) shouldBe false + ConvergenceObservations.sameBestBlock(info.copy(bestBlockHeightOpt = None), info, 50) shouldBe false + ConvergenceObservations.sameBestBlock(info, info.copy(bestBlockHeightOpt = None), 50) shouldBe false + ConvergenceObservations.sameBestBlock(info.copy(bestBlockIdOpt = None), info, 50) shouldBe false + ConvergenceObservations.sameBestBlock(info, info.copy(bestBlockIdOpt = None), 50) shouldBe false + } + + "Fully applied block agreement" should "accept a complete shared tip at the minimum height" in { + val info = NodeInfo(Some("tip"), Some("tip"), Some(12), Some(12), None, Some(false)) + ConvergenceObservations.sameFullyAppliedNonMiningBlock(info, info, 12) shouldBe true + ConvergenceObservations.sameFullyAppliedNonMiningBlock(info, info, 13) shouldBe false + } + + it should "reject a shared 21-header 12-full-block seed despite full-block agreement" in { + val lagging = NodeInfo(Some("header21"), Some("block12"), Some(21), Some(12), None, Some(false)) + ConvergenceObservations.sameBestBlock(lagging, lagging, 1) shouldBe true + ConvergenceObservations.sameFullyAppliedNonMiningBlock(lagging, lagging, 1) shouldBe false + } + + private val completeSeed = NodeInfo(Some("tip"), Some("tip"), Some(12), Some(12), None, Some(false)) + + Seq[(String, NodeInfo => NodeInfo)]( + "missing header height" -> ((info: NodeInfo) => info.copy(bestHeaderHeightOpt = None)), + "missing full-block height" -> ((info: NodeInfo) => info.copy(bestBlockHeightOpt = None)), + "missing header ID" -> ((info: NodeInfo) => info.copy(bestHeaderIdOpt = None)), + "missing full-block ID" -> ((info: NodeInfo) => info.copy(bestBlockIdOpt = None)), + "different header ID" -> ((info: NodeInfo) => info.copy(bestHeaderIdOpt = Some("other"))), + "different full-block ID" -> ((info: NodeInfo) => info.copy(bestBlockIdOpt = Some("other"))), + "different header height" -> ((info: NodeInfo) => info.copy(bestHeaderHeightOpt = Some(21))), + "unknown mining status" -> ((info: NodeInfo) => info.copy(isMining = None)), + "active mining" -> ((info: NodeInfo) => info.copy(isMining = Some(true))) + ).foreach { case (fault, change) => + Seq("A", "B").foreach { side => + it should s"reject $fault on node $side independently" in { + val (a, b) = if (side == "A") (change(completeSeed), completeSeed) + else (completeSeed, change(completeSeed)) + ConvergenceObservations.sameFullyAppliedNonMiningBlock(a, b, 1) shouldBe false + } + } + } + + it should "reject two empty tip identifiers" in { + val empty = completeSeed.copy(bestHeaderIdOpt = Some(""), bestBlockIdOpt = Some("")) + ConvergenceObservations.sameFullyAppliedNonMiningBlock(empty, empty, 1) shouldBe false + } + + it should "reject different complete tips at the same height" in { + val other = completeSeed.copy(bestHeaderIdOpt = Some("other"), bestBlockIdOpt = Some("other")) + ConvergenceObservations.sameFullyAppliedNonMiningBlock(completeSeed, other, 1) shouldBe false + } + + it should "reject different complete heights independently of matching IDs" in { + val other = completeSeed.copy(bestHeaderHeightOpt = Some(13), bestBlockHeightOpt = Some(13)) + ConvergenceObservations.sameFullyAppliedNonMiningBlock(completeSeed, other, 1) shouldBe false + } + + "Full block agreement" should "resample the entire group until its current selections agree" in withObserver { observer => + val samples = new AtomicInteger() + val result = observer.until(2.seconds.fromNow, 1.millis, 100.millis) { _ => + val headers = if (samples.incrementAndGet() == 1) Seq(Seq("a", "b"), Seq("b", "a")) + else Seq(Seq("b", "a"), Seq("b")) + Future.successful(headers) + }(ConvergenceObservations.selectedHeadersAgree)("selected headers disagree") + Await.result(result, 3.seconds).map(_.head) shouldBe Seq("b", "b") + samples.get() shouldBe 2 + } + + it should "fail persistent disagreement within the original deadline with recent evidence" in withObserver { observer => + var recent = "none" + val result = observer.until(100.millis.fromNow, 1.millis, 20.millis) { _ => + recent = "node0=a node1=b" + Future.successful(Seq(Seq("a"), Seq("b"))) + }(ConvergenceObservations.selectedHeadersAgree)(s"last: $recent") + intercept[TimeoutException](Await.result(result, 2.seconds)).getMessage should include("node0=a node1=b") + } + + "Observation probes" should "bound a stalled endpoint and not start overlapping requests" in withObserver { observer => + val calls = new AtomicInteger() + val never = Promise[Int]() + val probe = observer.probe { calls.incrementAndGet(); never.future } + Await.result(probe.sample(20.millis), 2.seconds) shouldBe Left("TimeoutException") + Await.result(probe.sample(20.millis), 2.seconds) shouldBe Left("TimeoutException") + calls.get() shouldBe 1 + never.success(3) + Await.result(never.future, 2.seconds) shouldBe 3 + Await.result(probe.sample(100.millis), 2.seconds) shouldBe Right(3) + calls.get() shouldBe 2 + } + + it should "keep successful status available while the peer sample times out" in withObserver { observer => + val status = observer.probe(Future.successful(42)).sample(100.millis) + val peers = observer.probe(Promise[Int]().future).sample(30.millis) + Await.result(status, 2.seconds) shouldBe Right(42) + Await.result(status.zip(peers), 2.seconds) shouldBe (Right(42) -> Left("TimeoutException")) + } + + it should "retain only an error class for endpoint failures" in withObserver { observer => + val result = observer.probe(Future.failed[Int](new IllegalArgumentException("private diagnostic payload"))) + Await.result(result.sample(100.millis), 2.seconds) shouldBe Left("IllegalArgumentException") + } + + it should "enforce the convergence deadline even when observation never returns" in withObserver { observer => + val result = observer.until(30.millis.fromNow, 1.millis, 10.millis)(_ => Promise[Boolean]().future)(identity)("recent status") + intercept[TimeoutException](Await.result(result, 2.seconds)).getMessage shouldBe "recent status" + } + + "Observation identifiers" should "exclude arbitrary response text" in { + ConvergenceObservations.headerId("ab" * 32) shouldBe "ab" * 32 + ConvergenceObservations.headerId("unexpected response text") shouldBe "invalid-header-id" + } +} From d62c945a4d6314c54a7b57f18a94aee8fd258aa7 Mon Sep 17 00:00:00 2001 From: "A. Shannon" <217575710+a-shannon@users.noreply.github.com> Date: Sun, 13 Sep 2026 04:43:20 +0200 Subject: [PATCH 7/9] Extract shared synchronization continuation prerequisite --- .../network/ErgoNodeViewSynchronizer.scala | 27 ++- .../nodeView/history/ErgoHistoryReader.scala | 8 +- .../ToDownloadProcessor.scala | 19 +- ...rgoNodeViewSynchronizerSpecification.scala | 12 +- .../network/SyncInfoV2CacheSpec.scala | 62 ++++++ ...BlockDownloadSchedulingSpecification.scala | 158 ++++++++++++++ .../SyncInfoV2HistorySpecification.scala | 193 ++++++++++++++++++ 7 files changed, 462 insertions(+), 17 deletions(-) create mode 100644 src/test/scala/org/ergoplatform/network/SyncInfoV2CacheSpec.scala create mode 100644 src/test/scala/org/ergoplatform/nodeView/history/BlockDownloadSchedulingSpecification.scala create mode 100644 src/test/scala/org/ergoplatform/nodeView/history/SyncInfoV2HistorySpecification.scala diff --git a/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala b/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala index ced63d1e8a..aa7f8cca8d 100644 --- a/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala +++ b/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala @@ -75,7 +75,7 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, private var syncInfoV1CacheByHeadersHeight: Option[(Int, ErgoSyncInfoV1)] = Option.empty - private var syncInfoV2CacheByHeadersHeight: Option[(Int, ErgoSyncInfoV2)] = Option.empty + private val syncInfoV2Cache = new SyncInfoV2Cache private val networkSettings: NetworkSettings = settings.scorexSettings.network @@ -315,14 +315,7 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, /** Get V2 sync info from cache or load it from history and add to cache */ private def getV2SyncInfo(history: ErgoHistory, full: Boolean): ErgoSyncInfoV2 = { - val headersHeight = history.headersHeight - syncInfoV2CacheByHeadersHeight - .collect { case (height, syncInfo) if height == headersHeight => syncInfo } - .getOrElse { - val v2SyncInfo = history.syncInfoV2(full) - syncInfoV2CacheByHeadersHeight = Some(headersHeight -> v2SyncInfo) - v2SyncInfo - } + syncInfoV2Cache.getOrElseUpdate(history.bestHeaderIdOpt, full)(history.syncInfoV2(full)) } /** @@ -1659,6 +1652,22 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, object ErgoNodeViewSynchronizer { + /** Single-entry cache owned by the synchronizer actor. */ + private[network] final class SyncInfoV2Cache { + private var cached: Option[(Option[ModifierId], Boolean, ErgoSyncInfoV2)] = None + + def getOrElseUpdate(bestHeaderId: Option[ModifierId], full: Boolean) + (build: => ErgoSyncInfoV2): ErgoSyncInfoV2 = { + cached.collect { + case (tip, mode, info) if tip == bestHeaderId && mode == full => info + }.getOrElse { + val info = build + cached = Some((bestHeaderId, full, info)) + info + } + } + } + private def props(networkControllerRef: ActorRef, viewHolderRef: ActorRef, syncInfoSpec: ErgoSyncInfoMessageSpec.type, diff --git a/src/main/scala/org/ergoplatform/nodeView/history/ErgoHistoryReader.scala b/src/main/scala/org/ergoplatform/nodeView/history/ErgoHistoryReader.scala index 9f7a45f0fe..82798e354c 100644 --- a/src/main/scala/org/ergoplatform/nodeView/history/ErgoHistoryReader.scala +++ b/src/main/scala/org/ergoplatform/nodeView/history/ErgoHistoryReader.scala @@ -403,8 +403,14 @@ trait ErgoHistoryReader } val headers = offsets.flatMap(offset => bestHeaderAtHeight(h - offset)) + // Keep a shared starting point in full summaries when it is available locally. + val genesisAnchor = if (full) { + bestHeaderAtHeight(GenesisHeight).filterNot(genesis => headers.exists(_.id == genesis.id)).toSeq + } else { + Seq.empty + } - ErgoSyncInfoV2(headers) + ErgoSyncInfoV2(headers.toSeq ++ genesisAnchor) } } diff --git a/src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/ToDownloadProcessor.scala b/src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/ToDownloadProcessor.scala index 090610c3c2..6aba3571f7 100644 --- a/src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/ToDownloadProcessor.scala +++ b/src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/ToDownloadProcessor.scala @@ -71,7 +71,7 @@ trait ToDownloadProcessor val toDownload = headersAtThisHeight.flatMap(requiredModifiersForHeader).filter { case (mtid, mid) => condition(mtid, mid) } // add new modifiers to download to accumulator val newAcc = toDownload.foldLeft(acc) { case (newAcc, (mType, mId)) => newAcc.adjust(mType)(_.fold(Vector(mId))(_ :+ mId)) } - continuation(height + 1, newAcc, maxHeight) + if (height == maxHeight) newAcc else continuation(height + 1, newAcc, maxHeight) } else { acc } @@ -84,8 +84,21 @@ trait ToDownloadProcessor // do not download full blocks if no headers-chain synced yet or SPV mode Map.empty case Some(fb) if farAwayFromBeingSynced(fb) => - // when far away from blockchain tip - continuation(fb.height + 1, Map.empty, fb.height + FullBlocksToDownloadAhead) + // A different best headers-chain may need block sections below the old full-chain tip. + val commonAncestor = if (isInBestChain(fb.id)) { + None + } else { + val parentSteps = Math.max(0L, nodeSettings.keepVersions.toLong) + val limit = Math.min(fb.height.toLong, parentSteps + 1L).toInt + headerChainBack(limit, fb.header, h => isInBestChain(h.id)).headOption + .filter(h => isInBestChain(h.id)) + } + val fromHeight = commonAncestor + .map(h => Math.max(minimalFullBlockHeight, h.height + 1)) + .getOrElse(fb.height + 1) + // Extending the scan backward must preserve forward progress beyond the existing full-chain tip. + val maxHeight = Math.min(fb.height.toLong + FullBlocksToDownloadAhead, Int.MaxValue.toLong).toInt + continuation(fromHeight, Map.empty, maxHeight) case Some(fb) => // when blockchain is about to be synced, // download children blocks of last 100 full blocks applied to the best chain, to get block sections from forks diff --git a/src/test/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerSpecification.scala b/src/test/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerSpecification.scala index a226c9bde3..ae9d1e0f67 100644 --- a/src/test/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerSpecification.scala +++ b/src/test/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerSpecification.scala @@ -510,14 +510,16 @@ class ErgoNodeViewSynchronizerSpecification extends AnyPropSpec // we check that in case of neighbour with older history (it has more blocks), // sync message will be sent by our node (to get invs from the neighbour), - // sync message will consist of 4 headers + // sync message will contain the sampled headers followed by genesis synchronizer ! Message(ErgoSyncInfoMessageSpec, Left(msgBytes), Some(peer)) ncProbe.fishForMessage(3 seconds) { case m => m match { case stn: SendToNetwork => val msg = stn.message val headers = msg.data.get.asInstanceOf[ErgoSyncInfoV2].lastHeaders - msg.spec.messageCode == ErgoSyncInfoMessageSpec.messageCode && headers.length == 4 + msg.spec.messageCode == ErgoSyncInfoMessageSpec.messageCode && + headers.map(_.id) == (Seq(0, 16, 128, 512) + .map(offset => localChain(localChain.size - offset - 1).id) :+ localChain.head.id) case _ => false } } @@ -545,14 +547,16 @@ class ErgoNodeViewSynchronizerSpecification extends AnyPropSpec // we check that in case of neighbour with older history (it has more blocks), // sync message will be sent by our node (to get invs from the neighbour), - // sync message will consist of 4 headers + // sync message will contain the sampled headers followed by genesis synchronizer ! Message(ErgoSyncInfoMessageSpec, Left(msgBytes), Some(peer)) ncProbe.fishForMessage(3 seconds) { case m => m match { case stn: SendToNetwork => val msg = stn.message val headers = msg.data.get.asInstanceOf[ErgoSyncInfoV2].lastHeaders - msg.spec.messageCode == ErgoSyncInfoMessageSpec.messageCode && headers.length == 4 + msg.spec.messageCode == ErgoSyncInfoMessageSpec.messageCode && + headers.map(_.id) == (Seq(0, 16, 128, 512) + .map(offset => localChain(localChain.size - offset - 1).id) :+ localChain.head.id) case _ => false } } diff --git a/src/test/scala/org/ergoplatform/network/SyncInfoV2CacheSpec.scala b/src/test/scala/org/ergoplatform/network/SyncInfoV2CacheSpec.scala new file mode 100644 index 0000000000..6fe5949fe7 --- /dev/null +++ b/src/test/scala/org/ergoplatform/network/SyncInfoV2CacheSpec.scala @@ -0,0 +1,62 @@ +package org.ergoplatform.network + +import org.ergoplatform.network.ErgoNodeViewSynchronizer.SyncInfoV2Cache +import org.ergoplatform.nodeView.history.ErgoSyncInfoV2 +import org.ergoplatform.utils.generators.ChainGenerator.genHeaderChain +import org.scalatest.matchers.should.Matchers +import org.scalatest.propspec.AnyPropSpec + +class SyncInfoV2CacheSpec extends AnyPropSpec with Matchers { + private val headers = genHeaderChain(3, diffBitsOpt = None, useRealTs = false).headers + private val tip = headers.last + private val fullSummary = ErgoSyncInfoV2(Seq(tip, headers.head)) + private val reducedSummary = ErgoSyncInfoV2(Seq(tip)) + + property("reuse a summary only for the same tip and requested mode") { + Seq(false, true).foreach { full => + val cache = new SyncInfoV2Cache + val expected = if (full) fullSummary else reducedSummary + cache.getOrElseUpdate(Some(tip.id), full)(expected) shouldBe expected + cache.getOrElseUpdate(Some(tip.id), full) { + fail("An unchanged tip and mode should reuse the cached summary") + } shouldBe expected + } + } + + property("alternating reduced and full requests preserves each requested summary") { + val cache = new SyncInfoV2Cache + Seq(false, true, false, true).foreach { full => + val expected = if (full) fullSummary else reducedSummary + cache.getOrElseUpdate(Some(tip.id), full)(expected) shouldBe expected + } + } + + property("a different best header at the same height replaces the cached summary") { + val otherTip = genHeaderChain(1, prefixOpt = Some(headers(1)), + diffBitsOpt = None, useRealTs = false).last + otherTip.height shouldBe tip.height + otherTip.id should not be tip.id + + Seq(false, true).foreach { full => + val cache = new SyncInfoV2Cache + val original = if (full) fullSummary else reducedSummary + val replacement = ErgoSyncInfoV2(if (full) Seq(otherTip, headers.head) else Seq(otherTip)) + cache.getOrElseUpdate(Some(tip.id), full)(original) shouldBe original + cache.getOrElseUpdate(Some(otherTip.id), full)(replacement) shouldBe replacement + cache.getOrElseUpdate(Some(otherTip.id), full) { + fail("The replacement tip should now be cached") + } shouldBe replacement + } + } + + property("empty history and populated history do not share cached summaries") { + val cache = new SyncInfoV2Cache + val empty = ErgoSyncInfoV2(Nil) + cache.getOrElseUpdate(None, full = true)(empty) shouldBe empty + cache.getOrElseUpdate(None, full = true) { + fail("An unchanged empty history should reuse its summary") + } shouldBe empty + cache.getOrElseUpdate(Some(tip.id), full = true)(fullSummary) shouldBe fullSummary + cache.getOrElseUpdate(None, full = true)(empty) shouldBe empty + } +} diff --git a/src/test/scala/org/ergoplatform/nodeView/history/BlockDownloadSchedulingSpecification.scala b/src/test/scala/org/ergoplatform/nodeView/history/BlockDownloadSchedulingSpecification.scala new file mode 100644 index 0000000000..b45da75214 --- /dev/null +++ b/src/test/scala/org/ergoplatform/nodeView/history/BlockDownloadSchedulingSpecification.scala @@ -0,0 +1,158 @@ +package org.ergoplatform.nodeView.history + +import org.ergoplatform.consensus.ProgressInfo +import org.ergoplatform.mining.AutolykosPowScheme +import org.ergoplatform.modifiers.{BlockSection, ErgoFullBlock, NetworkObjectTypeId, NonHeaderBlockSection} +import org.ergoplatform.modifiers.history.{BlockTransactions, HeaderChain} +import org.ergoplatform.modifiers.history.extension.Extension +import org.ergoplatform.modifiers.history.header.Header +import org.ergoplatform.nodeView.history.storage.HistoryStorage +import org.ergoplatform.nodeView.state.StateType +import org.ergoplatform.settings.ErgoSettings +import org.ergoplatform.utils.ErgoCorePropertyTest +import org.ergoplatform.utils.ErgoNodeTestConstants.initSettings +import org.ergoplatform.utils.generators.ErgoCoreGenerators.defaultHeaderGen +import org.ergoplatform.utils.generators.ErgoCoreTransactionGenerators.invalidErgoTransactionGen +import scorex.util.ModifierId + +import scala.reflect.ClassTag +import scala.util.Try + +class BlockDownloadSchedulingSpecification extends ErgoCorePropertyTest { + + // These in-memory records exercise scheduling, not block or chain validation. + private val template = defaultHeaderGen.sample.get + private val transaction = invalidErgoTransactionGen.sample.get + private val maximumHeightHeader = template.copy(height = Int.MaxValue) + private val bestChain = (1 to 300).foldLeft(Vector.empty[Header]) { (chain, height) => + chain :+ template.copy(height = height, parentId = chain.lastOption.map(_.id).getOrElse(Header.GenesisParentId)) + } + private val oldChain = (21 to 60).foldLeft(bestChain.take(20)) { (chain, height) => + chain :+ template.copy(height = height, parentId = chain.last.id, timestamp = template.timestamp + 1) + } + + private class SchedulingReader(fullHeader: Header, + retainedVersions: Int = 50, + minimumHeight: Int = 1, + synced: Boolean = true, + verify: Boolean = true, + tipHeight: Int = 300, + missingHeader: Option[ModifierId] = None) extends ErgoHistoryReader { + override protected[history] val historyStorage: HistoryStorage = null + override protected val settings: ErgoSettings = initSettings.copy(nodeSettings = initSettings.nodeSettings.copy( + keepVersions = retainedVersions, verifyTransactions = verify, stateType = StateType.Utxo)) + override val powScheme: AutolykosPowScheme = null + override protected def requireProofs: Boolean = false + override protected def process(m: NonHeaderBlockSection): Try[ProgressInfo[BlockSection]] = + fail("Scheduling snapshots do not process sections") + override protected def validate(m: NonHeaderBlockSection): Try[Unit] = + fail("Scheduling snapshots do not validate sections") + + private val headersById = (bestChain ++ oldChain :+ maximumHeightHeader).map(h => h.id -> h).toMap + var traversals: Vector[(Int, ModifierId)] = Vector.empty + var heightLookups: Vector[Int] = Vector.empty + override def bestFullBlockOpt: Option[ErgoFullBlock] = Some(ErgoFullBlock(fullHeader, + BlockTransactions(fullHeader.id, fullHeader.version, Seq(transaction)), Extension(fullHeader.id, Seq.empty), None)) + override def bestHeaderIdOpt: Option[ModifierId] = Some(bestChain(tipHeight - 1).id) + override def estimatedTip(): Option[Int] = Some(tipHeight) + override def minimalFullBlockHeight: Int = minimumHeight + override def isHeadersChainSynced: Boolean = synced + override def isInBestChain(id: ModifierId): Boolean = bestChain.exists(_.id == id) + override def headerIdsAtHeight(height: Int): Seq[ModifierId] = { + heightLookups :+= height + (bestChain :+ maximumHeightHeader).find(_.height == height).map(_.id).toSeq + } + override def typedModifierById[T <: BlockSection : ClassTag](id: ModifierId): Option[T] = + headersById.get(id).filterNot(h => missingHeader.contains(h.id)).collect { case section: T => section } + override def headerChainBack(limit: Int, startHeader: Header, until: Header => Boolean): HeaderChain = { + traversals :+= limit -> startHeader.id + super.headerChainBack(limit, startHeader, until) + } + } + + private def expected(reader: SchedulingReader, heights: Range): Map[NetworkObjectTypeId.Value, Seq[ModifierId]] = + heights.flatMap(h => reader.requiredModifiersForHeader(bestChain(h - 1))) + .groupBy(_._1).map { case (kind, sections) => kind -> sections.map(_._2) } + + private val acceptAll: (NetworkObjectTypeId.Value, ModifierId) => Boolean = (_, _) => true + + property("linear far-behind downloads retain the forward window without walking history") { + val reader = new SchedulingReader(bestChain(59)) + reader.nextModifiersToDownload(1000, acceptAll) shouldBe expected(reader, 61 to 252) + reader.traversals shouldBe empty + } + + property("ancestor scheduling extends backward without shortening the forward window") { + val reader = new SchedulingReader(oldChain.last, retainedVersions = 40) + reader.nextModifiersToDownload(1000, acceptAll) shouldBe expected(reader, 21 to 252) + reader.traversals shouldBe Vector(41 -> oldChain.last.id) + } + + property("a truncated traversal does not infer a common ancestor") { + val reader = new SchedulingReader(oldChain.last, retainedVersions = 39) + reader.nextModifiersToDownload(1000, acceptAll) shouldBe expected(reader, 61 to 252) + reader.traversals shouldBe Vector(40 -> oldChain.last.id) + } + + property("non-positive retention does not walk to a parent") { + Seq(0, -1, Int.MinValue).foreach { retained => + val reader = new SchedulingReader(oldChain.last, retainedVersions = retained) + reader.nextModifiersToDownload(1000, acceptAll) shouldBe expected(reader, 61 to 252) + reader.traversals shouldBe Vector(1 -> oldChain.last.id) + } + } + + property("a missing parent does not turn a partial traversal into a common ancestor") { + val reader = new SchedulingReader(oldChain.last, missingHeader = Some(oldChain(39).id)) + reader.nextModifiersToDownload(1000, acceptAll) shouldBe expected(reader, 61 to 252) + reader.traversals shouldBe Vector(51 -> oldChain.last.id) + } + + property("ancestor scheduling respects the minimum retained full-block height") { + Seq(30, 253).foreach { minimumHeight => + val reader = new SchedulingReader(oldChain.last, minimumHeight = minimumHeight) + reader.nextModifiersToDownload(1000, acceptAll) shouldBe expected(reader, minimumHeight to 252) + } + } + + property("maximum retention does not overflow the traversal bound") { + val reader = new SchedulingReader(oldChain.last, retainedVersions = Int.MaxValue) + reader.nextModifiersToDownload(1000, acceptAll) shouldBe expected(reader, 21 to 252) + reader.traversals shouldBe Vector(60 -> oldChain.last.id) + } + + property("a download window ending at the maximum height terminates without wrapping") { + val fullHeader = oldChain.last.copy(height = Int.MaxValue - 191) + val reader = new SchedulingReader(fullHeader, minimumHeight = Int.MaxValue) { + override def estimatedTip(): Option[Int] = Some(Int.MaxValue) + } + val sections = reader.requiredModifiersForHeader(maximumHeightHeader) + reader.nextModifiersToDownload(1000, acceptAll) shouldBe + sections.groupBy(_._1).map { case (kind, entries) => kind -> entries.map(_._2) } + reader.heightLookups shouldBe Vector(Int.MaxValue) + } + + property("ancestor scheduling preserves per-type caps and the section filter") { + val reader = new SchedulingReader(oldChain.last) + val firstSections = reader.requiredModifiersForHeader(bestChain(20)) + val excluded = firstSections.head._2 + val result = reader.nextModifiersToDownload(2, (_, id) => id != excluded) + val all = expected(reader, 21 to 22) + result shouldBe all.map { case (kind, ids) => kind -> ids.filterNot(_ == excluded) } + result.values.foreach(_.size should be <= 2) + } + + property("unsynced and non-verifying readers do not schedule sections or walk history") { + Seq(new SchedulingReader(oldChain.last, synced = false), new SchedulingReader(oldChain.last, verify = false)) + .foreach { reader => + reader.nextModifiersToDownload(1000, acceptAll) shouldBe empty + reader.traversals shouldBe empty + } + } + + property("near-tip scheduling retains its existing lookback without walking history") { + val reader = new SchedulingReader(oldChain.last, tipHeight = 100) + reader.nextModifiersToDownload(1000, acceptAll) shouldBe expected(reader, 1 to 300) + reader.traversals shouldBe empty + } +} diff --git a/src/test/scala/org/ergoplatform/nodeView/history/SyncInfoV2HistorySpecification.scala b/src/test/scala/org/ergoplatform/nodeView/history/SyncInfoV2HistorySpecification.scala new file mode 100644 index 0000000000..5e3e8311ce --- /dev/null +++ b/src/test/scala/org/ergoplatform/nodeView/history/SyncInfoV2HistorySpecification.scala @@ -0,0 +1,193 @@ +package org.ergoplatform.nodeView.history + +import org.ergoplatform.consensus.ProgressInfo +import org.ergoplatform.mining.AutolykosPowScheme +import org.ergoplatform.modifiers.{BlockSection, NonHeaderBlockSection} +import org.ergoplatform.modifiers.history.HeaderChain +import org.ergoplatform.modifiers.history.header.Header +import org.ergoplatform.nodeView.history.storage.HistoryStorage +import org.ergoplatform.nodeView.state.StateType +import org.ergoplatform.settings.ErgoSettings +import org.ergoplatform.utils.ErgoCorePropertyTest +import org.ergoplatform.utils.HistoryTestHelpers.generateHistory +import org.ergoplatform.utils.generators.ChainGenerator.{applyHeaderChain, genHeaderChain} +import org.ergoplatform.utils.generators.ErgoCoreGenerators.defaultHeaderGen +import scorex.crypto.hash.Digest32 + +import scala.util.Try + +class SyncInfoV2HistorySpecification extends ErgoCorePropertyTest { + + private def newHistory(): ErgoHistory = + generateHistory( + verifyTransactions = false, + stateType = StateType.Digest, + PoPoWBootstrap = false, + blocksToKeep = 0, + epochLength = 1000 + ) + + private def checkRoundtrip(summary: ErgoSyncInfoV2): Unit = { + val bytes = ErgoSyncInfoSerializer.toBytes(summary) + val decoded = ErgoSyncInfoSerializer.parseBytes(bytes).asInstanceOf[ErgoSyncInfoV2] + decoded.lastHeaders.map(_.id) shouldBe summary.lastHeaders.map(_.id) + decoded.height shouldBe summary.height + ErgoSyncInfoSerializer.toBytes(decoded).sameElements(bytes) shouldBe true + } + + // These snapshots exercise summary selection, not header or chain validity. + private def snapshotReader(headers: Seq[Header]): ErgoHistoryReader = new ErgoHistoryReader { + override protected[history] val historyStorage: HistoryStorage = null + override protected val settings: ErgoSettings = null + override val powScheme: AutolykosPowScheme = null + override protected def requireProofs: Boolean = false + override protected def process(m: NonHeaderBlockSection): Try[ProgressInfo[BlockSection]] = + fail("Summary snapshots do not process block sections") + override protected def validate(m: NonHeaderBlockSection): Try[Unit] = + fail("Summary snapshots do not validate block sections") + override def bestHeaderOpt: Option[Header] = headers.sortBy(_.height).lastOption + override def headersHeight: Int = bestHeaderOpt.map(_.height).getOrElse(0) + override def isEmpty: Boolean = headers.isEmpty + override def bestHeaderAtHeight(height: Int): Option[Header] = headers.find(_.height == height) + } + + property("full and reduced sync summaries remain empty for empty history") { + val history = newHistory() + try { + Seq(true, false).foreach { full => + val summary = history.syncInfoV2(full) + summary.lastHeaders shouldBe empty + summary.nonEmpty shouldBe false + summary.height shouldBe None + checkRoundtrip(summary) + } + } finally { + history.closeStorage() + } + } + + property("linear history summaries retain recent samples and exactly one genesis anchor") { + var history = newHistory() + try { + val chain = genHeaderChain(17, history, diffBitsOpt = None, useRealTs = false) + val expectedHeights = Seq( + 1 -> Seq(1), + 2 -> Seq(2, 1), + 10 -> Seq(10, 1), + 17 -> Seq(17, 1) + ) + var appliedHeight = 0 + expectedHeights.foreach { case (height, heights) => + history = applyHeaderChain(history, HeaderChain(chain.headers.slice(appliedHeight, height))) + appliedHeight = height + history.headersHeight shouldBe height + + val full = history.syncInfoV2(full = true) + val expectedIds = heights.map(h => chain.headers(h - 1).id) + full.lastHeaders.map(_.id) shouldBe expectedIds + full.lastHeaders.map(_.height) shouldBe heights + full.lastHeaders.head.id shouldBe history.bestHeaderOpt.get.id + full.lastHeaders.last.id shouldBe chain.head.id + full.lastHeaders.count(_.id == chain.head.id) shouldBe 1 + full.lastHeaders.map(_.id).distinct.size shouldBe full.lastHeaders.size + full.lastHeaders.size should be <= 5 + full.lastHeaders.size should be <= ErgoSyncInfoSerializer.MaxHeadersAllowed + full.height shouldBe Some(height) + checkRoundtrip(full) + + val reduced = history.syncInfoV2(full = false) + reduced.lastHeaders.map(_.id) shouldBe Seq(chain.headers(height - 1).id) + reduced.height shouldBe Some(height) + checkRoundtrip(reduced) + } + } finally { + history.closeStorage() + } + } + + property("full summary snapshots preserve sparse samples and deduplicate the genesis anchor") { + val template = defaultHeaderGen.sample.get + val cases = Seq( + Seq(129, 113, 1), + Seq(513, 497, 385, 1), + Seq(600, 584, 472, 88, 1) + ) + cases.foreach { heights => + val headers = heights.map(height => template.copy(height = height)) + val reader = snapshotReader(headers) + val full = reader.syncInfoV2(full = true) + full.lastHeaders.map(_.id) shouldBe headers.map(_.id) + full.lastHeaders.map(_.height) shouldBe heights + full.lastHeaders.count(_.height == 1) shouldBe 1 + full.lastHeaders.map(_.id).distinct.size shouldBe full.lastHeaders.size + full.lastHeaders.size should be <= 5 + full.lastHeaders.size should be <= ErgoSyncInfoSerializer.MaxHeadersAllowed + full.height shouldBe Some(heights.head) + checkRoundtrip(full) + + val reduced = reader.syncInfoV2(full = false) + reduced.lastHeaders.map(_.id) shouldBe Seq(headers.head.id) + checkRoundtrip(reduced) + } + } + + property("full summary preserves available samples when genesis is unavailable") { + val template = defaultHeaderGen.sample.get + val headers = Seq(600, 584, 472, 88).map(height => template.copy(height = height)) + val reader = snapshotReader(headers) + reader.bestHeaderAtHeight(1) shouldBe None + val full = reader.syncInfoV2(full = true) + full.lastHeaders.map(_.id) shouldBe headers.map(_.id) + full.height shouldBe Some(600) + checkRoundtrip(full) + val reduced = reader.syncInfoV2(full = false) + reduced.lastHeaders.map(_.id) shouldBe Seq(headers.head.id) + checkRoundtrip(reduced) + } + + Seq((3, 9, 13), (18, 249, 70)).foreach { case (sharedHeight, heightA, heightB) => + property(s"full summaries recover both fork continuations after shared height $sharedHeight at tips $heightA/$heightB") { + var historyA = newHistory() + var historyB = newHistory() + try { + // Exercise stored header histories and the actual summary/continuation methods. + // Full block application and network delivery are separate integration contracts. + val shared = genHeaderChain(sharedHeight, historyA, diffBitsOpt = None, useRealTs = false) + val branchA = genHeaderChain(heightA - sharedHeight, prefixOpt = Some(shared.last), + control = historyA.difficultyCalculator, extensionHash = Digest32 @@ Array.fill[Byte](32)(1), + diffBitsOpt = None, useRealTs = false).headers.tail + val branchB = genHeaderChain(heightB - sharedHeight, prefixOpt = Some(shared.last), + control = historyB.difficultyCalculator, extensionHash = Digest32 @@ Array.fill[Byte](32)(2), + diffBitsOpt = None, useRealTs = false).headers.tail + val chainA = shared.headers ++ branchA + val chainB = shared.headers ++ branchB + historyA = applyHeaderChain(historyA, HeaderChain(chainA)) + historyB = applyHeaderChain(historyB, HeaderChain(chainB)) + historyA.headersHeight shouldBe heightA + historyB.headersHeight shouldBe heightB + branchA.head.id should not be branchB.head.id + historyA.contains(branchB.head.id) shouldBe false + historyB.contains(branchA.head.id) shouldBe false + + Seq((historyA, historyB, chainA), (historyB, historyA, chainB)).foreach { + case (local, peer, localChain) => + val sparseHeaders = ErgoHistoryReader.FullV2SyncOffsets.toSeq + .flatMap(offset => peer.bestHeaderAtHeight(peer.headersHeight - offset)) + sparseHeaders should not be empty + sparseHeaders.exists(header => local.contains(header.id)) shouldBe false + local.continuationIdsV2(ErgoSyncInfoV2(sparseHeaders), size = 400) shouldBe empty + + val full = peer.syncInfoV2(full = true) + checkRoundtrip(full) + val continuation = local.continuationIdsV2(full, size = 400) + continuation.map(_._2) shouldBe localChain.tail.map(_.id) + continuation.map(_._1).distinct shouldBe Seq(Header.modifierTypeId) + continuation.map(_._2) should contain(localChain(sharedHeight).id) + full.lastHeaders.last.id shouldBe shared.head.id + } + } finally { + try historyA.closeStorage() finally historyB.closeStorage() + } + } + } +} From 71f33c6e45a9394557cf3af1bfb422cbc53d26cf Mon Sep 17 00:00:00 2001 From: "A. Shannon" <217575710+a-shannon@users.noreply.github.com> Date: Sun, 13 Sep 2026 04:46:14 +0200 Subject: [PATCH 8/9] Extract shared deep rollback fixture prerequisite --- .../it/DeepRollBackIsolationSpec.scala | 58 ++++++ .../ergoplatform/it/DeepRollBackSpec.scala | 185 ++++++++++++++---- 2 files changed, 201 insertions(+), 42 deletions(-) create mode 100644 src/it/scala/org/ergoplatform/it/DeepRollBackIsolationSpec.scala diff --git a/src/it/scala/org/ergoplatform/it/DeepRollBackIsolationSpec.scala b/src/it/scala/org/ergoplatform/it/DeepRollBackIsolationSpec.scala new file mode 100644 index 0000000000..f7432730ea --- /dev/null +++ b/src/it/scala/org/ergoplatform/it/DeepRollBackIsolationSpec.scala @@ -0,0 +1,58 @@ +package org.ergoplatform.it + +import akka.actor.{ActorRef, ActorSystem} +import akka.io.Tcp +import akka.testkit.{ExplicitlyTriggeredScheduler, TestActorRef, TestProbe} +import com.typesafe.config.ConfigFactory +import org.ergoplatform.network.message.MessageConstants.MessageCode +import org.ergoplatform.network.peer.PeerManager.ReceivableMessages.RandomPeerExcluding +import org.ergoplatform.settings.ScorexSettings +import org.ergoplatform.utils.ErgoNodeTestConstants.settings +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import scorex.core.app.ScorexContext +import scorex.core.network.NetworkController + +import java.net.InetSocketAddress +import scala.concurrent.{Await, ExecutionContext} +import scala.concurrent.duration._ + +class DeepRollBackIsolationSpec extends AnyFlatSpec with Matchers { + "Isolated rollback mining" should "disable automatic peer selection and reject incoming connections" in { + val ordinaryConfig = ConfigFactory.load() + val isolatedNetwork = ScorexSettings.fromConfig( + DeepRollBackSpec.isolatedMiningConfig.withFallback(ordinaryConfig).resolve()).network + ScorexSettings.fromConfig(ordinaryConfig).network.maxConnections should be > 0 + + implicit val system: ActorSystem = ActorSystem("RollbackIsolation", ConfigFactory.parseString( + "akka.scheduler.implementation = akka.testkit.ExplicitlyTriggeredScheduler")) + implicit val ec: ExecutionContext = system.dispatcher + try { + def controller(maxConnections: Int): (TestActorRef[NetworkController], TestProbe) = { + val peers = TestProbe() + val tcp = TestProbe() + val controllerSettings = settings.copy(scorexSettings = settings.scorexSettings.copy( + network = settings.scorexSettings.network.copy(maxConnections = maxConnections))) + val ref = TestActorRef(new NetworkController(controllerSettings, peers.ref, + ScorexContext(Seq.empty, None, None), tcp.ref, _ => Map.empty[MessageCode, ActorRef])) + tcp.expectMsgType[Tcp.Bind] + ref ! Tcp.Bound(controllerSettings.scorexSettings.network.bindAddress) + (ref, peers) + } + + val (isolated, isolatedPeers) = controller(isolatedNetwork.maxConnections) + val (_, ordinaryPeers) = controller(maxConnections = 1) + system.scheduler.asInstanceOf[ExplicitlyTriggeredScheduler].timePasses(5.seconds) + ordinaryPeers.expectMsgType[RandomPeerExcluding](3.seconds) + isolatedPeers.expectNoMessage(200.millis) + + val incoming = TestProbe() + incoming.send(isolated, Tcp.Connected(new InetSocketAddress("127.0.0.2", 9000), + settings.scorexSettings.network.bindAddress)) + incoming.expectMsg(Tcp.Close) + isolatedPeers.expectNoMessage(200.millis) + } finally { + Await.result(system.terminate(), 10.seconds) + } + } +} diff --git a/src/it/scala/org/ergoplatform/it/DeepRollBackSpec.scala b/src/it/scala/org/ergoplatform/it/DeepRollBackSpec.scala index ea16faf015..63546b5c7a 100644 --- a/src/it/scala/org/ergoplatform/it/DeepRollBackSpec.scala +++ b/src/it/scala/org/ergoplatform/it/DeepRollBackSpec.scala @@ -2,13 +2,15 @@ package org.ergoplatform.it import java.io.File import java.util.concurrent.TimeoutException -import com.typesafe.config.Config -import org.ergoplatform.it.api.NodeApi.NodeInfo +import com.typesafe.config.{Config, ConfigFactory} +import io.circe.Json +import org.ergoplatform.it.api.NodeApi.{NodeInfo, nodeInfoDecoder} import org.ergoplatform.it.container.{IntegrationSuite, Node} import org.ergoplatform.nodeView.history.ErgoHistoryUtils +import org.ergoplatform.it.util.ConvergenceObservations import org.scalatest.freespec.AnyFreeSpec import scala.async.Async -import scala.concurrent.{Await, Future, blocking} +import scala.concurrent.{Await, Future} import scala.concurrent.duration._ class DeepRollBackSpec extends AnyFreeSpec with IntegrationSuite { @@ -45,41 +47,114 @@ class DeepRollBackSpec extends AnyFreeSpec with IntegrationSuite { .withFallback(nonGeneratingPeerConfig) .withFallback(allowLocalConfig) - private def waitForSameBestBlock( + private val observations = new ConvergenceObservations + private val seedObservations = new ConvergenceObservations + @volatile private var lastSeedObservation = "Initial seed has not been sampled" + + private def waitForSettledSeed( nodeA: Node, nodeB: Node, - minHeight: Int, timeout: FiniteDuration ): Future[(NodeInfo, NodeInfo)] = { - def sameBestBlock(infoA: NodeInfo, infoB: NodeInfo): Boolean = { - val sameHeight = - infoA.bestBlockHeightOpt.nonEmpty && - infoA.bestBlockHeightOpt == infoB.bestBlockHeightOpt - val sameBlock = - infoA.bestBlockIdOpt.nonEmpty && infoA.bestBlockIdOpt == infoB.bestBlockIdOpt - val highEnough = infoA.bestBlockHeightOpt.exists(_ >= minHeight) - sameHeight && sameBlock && highEnough + def infoProbe(node: Node): seedObservations.Probe[NodeInfo] = + seedObservations.probe(node.singleGet("/info", _.setRequestTimeout(5000)).map { response => + require(response.getStatusCode == 200, "Unexpected seed observation status") + node.ergoJsonAnswerAs[NodeInfo](response.getResponseBody) + }) + + val probeA = infoProbe(nodeA) + val probeB = infoProbe(nodeB) + def describe(result: Either[String, NodeInfo]): String = result.fold( + error => s"errorClass=$error", + info => s"headersHeight=${info.bestHeaderHeightOpt}; fullHeight=${info.bestBlockHeightOpt}; " + + s"headerId=${info.bestHeaderIdOpt.map(ConvergenceObservations.headerId)}; " + + s"fullId=${info.bestBlockIdOpt.map(ConvergenceObservations.headerId)}; mining=${info.isMining}") + seedObservations.until(timeout.fromNow, 1.second, 5.seconds) { budget => + probeA.sample(budget).zip(probeB.sample(budget)).map { pair => + lastSeedObservation = s"A=${describe(pair._1)}; B=${describe(pair._2)}" + log.info(s"Initial shared-chain readiness: $lastSeedObservation") + pair + } + } { + case (Right(a), Right(b)) => + ConvergenceObservations.sameFullyAppliedNonMiningBlock(a, b, ErgoHistoryUtils.GenesisHeight) + case _ => false + }( + s"Initial chain did not settle with mining disabled and matching full/header tips; $lastSeedObservation" + ).map { case (a, b) => (a.toOption.get, b.toOption.get) } + } + + private case class NodeSnapshot(info: Option[NodeInfo]) + private val probes = scala.collection.mutable.Map.empty[ + Node, (observations.Probe[NodeInfo], observations.Probe[Int])] + + @volatile private var lastObservation = "No node pair observed" + private var recentObservations = Vector.empty[String] + + private def remember(phase: String, label: String, endpoint: String, summary: String): Unit = synchronized { + val observation = s"$phase $label $endpoint sampledAt=${System.currentTimeMillis()} $summary" + recentObservations = (recentObservations :+ observation).takeRight(12) + lastObservation = recentObservations.mkString("; ") + log.info(observation) + } + + private def snapshot(phase: String, label: String, node: Node, budget: FiniteDuration): Future[NodeSnapshot] = { + val (statusProbe, peerProbe) = probes.getOrElseUpdate(node, ( + observations.probe(node.singleGet("/info", _.setRequestTimeout(5000)) + .map { r => + require(r.getStatusCode == 200, "Unexpected observation status") + node.ergoJsonAnswerAs[NodeInfo](r.getResponseBody) + }), + observations.probe(node.singleGet("/peers/connected", _.setRequestTimeout(5000)).map { r => + require(r.getStatusCode == 200, "Unexpected observation status") + node.ergoJsonAnswerAs[Json](r.getResponseBody).asArray.getOrElse( + throw new IllegalArgumentException("Expected peer array")).size + }) + )) + val status = statusProbe.sample(budget).map { + case Right(info) => + val summary = Json.obj( + "headersHeight" -> info.bestHeaderHeightOpt.map(Json.fromInt).getOrElse(Json.Null), + "fullHeight" -> info.bestBlockHeightOpt.map(Json.fromInt).getOrElse(Json.Null), + "bestHeaderId" -> info.bestHeaderIdOpt.map(ConvergenceObservations.headerId).map(Json.fromString).getOrElse(Json.Null), + "bestFullHeaderId" -> info.bestBlockIdOpt.map(ConvergenceObservations.headerId).map(Json.fromString).getOrElse(Json.Null) + ) + remember(phase, label, "status", summary.noSpaces) + Some(info) + case Left(error) => + remember(phase, label, "status", s"errorClass=$error") + None } + val peers = peerProbe.sample(budget).map { result => + val summary = result.fold(error => s"errorClass=$error", count => s"connectedPeerCount=$count") + remember(phase, label, "peers", summary) + } + status.zip(peers).map { case (info, _) => NodeSnapshot(info) } + } - def retryAfterDelay(deadline: Deadline): Future[(NodeInfo, NodeInfo)] = - Future { - blocking(Thread.sleep(1000)) - }.flatMap(_ => loop(deadline)) - - def loop(deadline: Deadline): Future[(NodeInfo, NodeInfo)] = - nodeA.info.zip(nodeB.info).flatMap { case (infoA, infoB) => - if (sameBestBlock(infoA, infoB)) { - Future.successful((infoA, infoB)) - } else if (deadline.isOverdue()) { - Future.failed(new TimeoutException( - s"Nodes did not converge to the same best full block at height >= $minHeight" - )) - } else { - retryAfterDelay(deadline) - } - } + private def observeNodes( + phase: String, + nodeA: Node, + nodeB: Node, + budget: FiniteDuration = 5.seconds + ): Future[(NodeSnapshot, NodeSnapshot)] = { + snapshot(phase, "A", nodeA, budget).zip(snapshot(phase, "B", nodeB, budget)) + } - loop(timeout.fromNow) + private def waitForSameBestBlock( + nodeA: Node, + nodeB: Node, + minHeight: Int, + timeout: FiniteDuration + ): Future[(NodeInfo, NodeInfo)] = { + observations.until(timeout.fromNow, 1.second, 5.seconds)( + budget => observeNodes("convergence", nodeA, nodeB, budget) + ) { case (a, b) => + a.info.exists(infoA => b.info.exists(infoB => ConvergenceObservations.sameBestBlock(infoA, infoB, minHeight))) + }( + s"Nodes did not converge to the same best full block at height >= $minHeight; " + + s"recent observations: $lastObservation" + ).map { case (a, b) => (a.info.get, b.info.get) } } "Deep rollback handling" in { @@ -100,34 +175,44 @@ class DeepRollBackSpec extends AnyFreeSpec with IntegrationSuite { val genesisAGen = Async.await(minerAGen.headerIdsByHeight(ErgoHistoryUtils.GenesisHeight)).head val genesisBGen = Async.await(minerBGen.headerIdsByHeight(ErgoHistoryUtils.GenesisHeight)).head - val minerAGenBestHeight = Async.await(minerAGen.fullHeight) - val minerBGenBestHeight = Async.await(minerBGen.fullHeight) - - log.info("heightA: " + minerAGenBestHeight) - log.info("heightB: " + minerBGenBestHeight) - genesisAGen shouldBe genesisBGen + Async.await(observeNodes("initial shared chain", minerAGen, minerBGen)) - // 2. Stop all nodes + // Freeze the producer while B can still retrieve every header's full block. docker.stopNode(minerAGen.containerId) + val minerASeed: Node = docker.startDevNetNode(minerAConfigNonGen, + specialVolumeOpt = Some((localVolumeA, remoteVolumeA))).get + val (seedA, seedB) = Async.await(waitForSettledSeed(minerASeed, minerBGen, 2.minutes)) + val seedHeight = seedA.bestBlockHeightOpt.get + require(seedHeight < chainLength, + s"Initial shared chain already reached $seedHeight; isolated node B must mine to $chainLength") + log.info(s"Settled shared chain: heightA=$seedHeight, heightB=${seedB.bestBlockHeightOpt.get}") + + // 2. Stop the restarted A and B only after both have the complete shared seed. + docker.stopNode(minerASeed.containerId) docker.stopNode(minerBGen.containerId) - val minerAIsolated: Node = docker.startDevNetNode(minerAConfig, isolatedPeersConfig, + val minerAIsolated: Node = docker.startDevNetNode(DeepRollBackSpec.isolatedMiningConfig.withFallback(minerAConfig), isolatedPeersConfig, specialVolumeOpt = Some((localVolumeA, remoteVolumeA))).get // 1. Let nodeA mine `chainLength + delta` blocks in isolation Async.await(minerAIsolated.waitForHeight(chainLength + delta)) - val minerBIsolated: Node = docker.startDevNetNode(minerBConfig, isolatedPeersConfig, + val minerBIsolated: Node = docker.startDevNetNode(DeepRollBackSpec.isolatedMiningConfig.withFallback(minerBConfig), isolatedPeersConfig, specialVolumeOpt = Some((localVolumeB, remoteVolumeB))).get + Async.await(observeNodes("isolated miners started", minerAIsolated, minerBIsolated)) // 2. Let nodeB mine `chainLength` blocks in isolation Async.await(minerBIsolated.waitForHeight(chainLength, 100.millis)) + Async.await(minerAIsolated.connectedPeers) shouldBe empty + Async.await(minerBIsolated.connectedPeers) shouldBe empty + log.info("Mining phase done") val minerABestHeight = Async.await(minerAIsolated.fullHeight) val minerBBestHeight = Async.await(minerBIsolated.fullHeight) + Async.await(observeNodes("isolated mining complete", minerAIsolated, minerBIsolated)) docker.stopNode(minerAIsolated.containerId) docker.stopNode(minerBIsolated.containerId) @@ -143,7 +228,7 @@ class DeepRollBackSpec extends AnyFreeSpec with IntegrationSuite { val minerB: Node = docker.startDevNetNode(minerBConfigNonGen, specialVolumeOpt = Some((localVolumeB, remoteVolumeB))).get - + Async.await(observeNodes("restarted without mining", minerA, minerB)) val isMiningAOpt = Async.await(minerA.info).isMining log.info("isminingA: " + isMiningAOpt) @@ -162,7 +247,23 @@ class DeepRollBackSpec extends AnyFreeSpec with IntegrationSuite { minerBInfo.bestBlockIdOpt shouldEqual minerAInfo.bestBlockIdOpt } - Await.result(result, 20.minutes) + try { + Await.result(result, 20.minutes) + } catch { + case error: TimeoutException => + log.error(s"Deep rollback timed out; last initial-seed observation: $lastSeedObservation") + log.error(s"Deep rollback timed out; last observation: $lastObservation") + throw error + } finally { + seedObservations.close() + observations.close() + } } } + +object DeepRollBackSpec { + // The retained peer database survives restarts. Disable automatic outgoing connections + // and incoming admission during mining; final restarts use the ordinary node configs. + private[it] val isolatedMiningConfig: Config = ConfigFactory.parseString("scorex.network.maxConnections = 0") +} From a1c1bd26e89ec2f24d5c52acc1fdd35346d65371 Mon Sep 17 00:00:00 2001 From: "A. Shannon" <217575710+a-shannon@users.noreply.github.com> Date: Sun, 13 Sep 2026 05:28:01 +0200 Subject: [PATCH 9/9] Extract shared fork-resolution observation diagnostics --- .../it/ForkResolutionDiagnosticsSpec.scala | 57 +++++ .../ergoplatform/it/ForkResolutionSpec.scala | 215 ++++++++++++++---- 2 files changed, 231 insertions(+), 41 deletions(-) create mode 100644 src/it/scala/org/ergoplatform/it/ForkResolutionDiagnosticsSpec.scala diff --git a/src/it/scala/org/ergoplatform/it/ForkResolutionDiagnosticsSpec.scala b/src/it/scala/org/ergoplatform/it/ForkResolutionDiagnosticsSpec.scala new file mode 100644 index 0000000000..e4522d911e --- /dev/null +++ b/src/it/scala/org/ergoplatform/it/ForkResolutionDiagnosticsSpec.scala @@ -0,0 +1,57 @@ +package org.ergoplatform.it + +import org.ergoplatform.it.api.NodeApi.NodeInfo +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +class ForkResolutionDiagnosticsSpec extends AnyFlatSpec with Matchers { + import ForkResolutionDiagnostics._ + + "Fork observations" should "retain independently accepted results while later selections change" in { + val first = latch(Vector.fill[Option[String]](4)(None), Vector(Some("a"), None, Some("a"), None)) + val later = latch(first, Vector(None, Some("a"), None, Some("a"))) + later shouldBe Vector.fill(4)(Some("a")) + latch(later, Vector.fill(4)(Some("b"))) shouldBe later + } + + it should "retain missing results until their own predicate succeeds" in { + latch(Vector(Some("a"), None), Vector(None, None)) shouldBe Vector(Some("a"), None) + } + + it should "report the fixed target separately from the current selected header" in { + val target = "ab" * 32 + val current = "cd" * 32 + val snapshot = Snapshot(Right(NodeInfo(None, None, Some(25), Some(20), None, Some(false))), + Right(3), Right(Seq(current, target)), Right(20), reachable = true) + val text = describe("match-anchor", 2, Some(10), Some(target), snapshot) + text should include(s"fixed=Some($target)") + text should include(s"selected=$current") + text should include("headerHeight=Some(25) fullHeight=Some(20) mining=Some(false)") + text should include("peerCount=3") + } + + it should "exclude response text, addresses and paths from observed fields" in { + val payload = "http://example.invalid/private/location" + val snapshot = Snapshot(Right(NodeInfo(Some(payload), Some(payload), Some(1), Some(1), Some(payload), None)), + Left(payload), Right(Seq(payload)), Right(1), reachable = true) + val text = describe("match-anchor", 0, Some(1), Some(payload), snapshot) + text should not include payload + text should include("peerErrorClass=ObservationError") + text should include("selected=invalid-header-id") + text should include("fixed=Some(invalid-header-id)") + describe("initial-height", 0, None, None, + Snapshot(Left("TimeoutException"), Left("IOException"), Left("ParsingFailure"), + Left("TimeoutException"), reachable = false)) should + include("statusErrorClass=TimeoutException peerErrorClass=IOException headerErrorClass=ParsingFailure") + } + + it should "keep supplementary failures out of the startup and height acceptance gates" in { + val snapshot = Snapshot(Left("ParsingFailure"), Left("TimeoutException"), Left("IOException"), + Right(20), reachable = true) + startupReached(snapshot) shouldBe Some(()) + heightReached(20)(snapshot) shouldBe Some(20) + heightReached(21)(snapshot) shouldBe None + heightReached(20)(snapshot.copy(fullHeight = Left("ParsingFailure"))) shouldBe None + startupReached(snapshot.copy(reachable = false)) shouldBe None + } +} diff --git a/src/it/scala/org/ergoplatform/it/ForkResolutionSpec.scala b/src/it/scala/org/ergoplatform/it/ForkResolutionSpec.scala index fbe8273e6b..39ca9ab163 100644 --- a/src/it/scala/org/ergoplatform/it/ForkResolutionSpec.scala +++ b/src/it/scala/org/ergoplatform/it/ForkResolutionSpec.scala @@ -1,20 +1,24 @@ package org.ergoplatform.it import java.io.File -import cats.implicits._ +import java.util.concurrent.atomic.AtomicBoolean import com.typesafe.config.Config +import io.circe.Json +import org.ergoplatform.it.api.NodeApi.NodeInfo import org.ergoplatform.it.container.Docker.{ExtraConfig, noExtraConfig} import org.ergoplatform.it.container.{IntegrationSuite, Node} -import org.scalatest.concurrent.Eventually +import org.ergoplatform.it.util.ConvergenceObservations import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers -import scala.async.Async import scala.concurrent.duration._ import scala.concurrent.{Await, Future} import scala.util.Try +import scala.util.control.NonFatal -class ForkResolutionSpec extends AnyFlatSpec with Matchers with IntegrationSuite with Eventually { +class ForkResolutionSpec extends AnyFlatSpec with Matchers with IntegrationSuite { + + import ForkResolutionDiagnostics._ val nodesQty: Int = 4 @@ -43,24 +47,107 @@ class ForkResolutionSpec extends AnyFlatSpec with Matchers with IntegrationSuite def localVolume(n: Int): String = s"$localDataDir/fork-resolution-spec/node-$n/data" + private var phase = "initial-start" + private var recent = Vector.empty[String] + private val active = new AtomicBoolean(true) + + private def enter(next: String): Unit = synchronized { + requireActive() + phase = next + log.info(s"Fork resolution phase=$phase") + } + + private def remember(entry: String): Unit = synchronized { + recent = (recent :+ entry).takeRight(nodesQty * 3) + } + + private def evidence: String = synchronized { s"phase=$phase; ${recent.mkString("; ")}" } + def clearPeerDatabases(): Unit = { - volumesMapping.foreach { case (localVolume, remoteVolume) => + volumesMapping.zipWithIndex.foreach { case ((localVolume, remoteVolume), index) => + requireActive() + remember(s"phase=$phase node=$index operation=clear-peers") docker.removeFromMountedVolume(localVolume, remoteVolume, "peers") } } - def startNodesWithBinds(nodeConfigs: List[Config], + private def startNodesWithBinds(nodeConfigs: List[Config], observations: ConvergenceObservations, + deadline: Option[Deadline] = None, configEnrich: ExtraConfig = noExtraConfig): List[Node] = { - log.trace(s"Starting ${nodeConfigs.size} containers") - val nodes: Try[List[Node]] = nodeConfigs + val nodes = nodeConfigs .map(_.withFallback(specialDataDirConfig(remoteVolume))) .zip(volumesMapping) - .map { case (cfg, vol) => docker.startDevNetNode(cfg, configEnrich, Some(vol)) } - .sequence - implicit val patienceConfig: PatienceConfig = PatienceConfig((nodeConfigs.size * 2).seconds, 3.second) - eventually { - Await.result(Future.traverse(nodes.get)(_.waitForStartup), 180.seconds) + .zipWithIndex.map { case ((cfg, vol), index) => + requireActive() + deadline.foreach(d => requireTime(d)) + remember(s"phase=$phase node=$index operation=start") + docker.startDevNetNode(cfg, configEnrich, Some(vol)).get + } + val startupDeadline = deadline.map(d => d.timeLeft.min(180.seconds).fromNow).getOrElse(180.seconds.fromNow) + waitFor(nodes, observations, startupDeadline, None, None)(startupReached) + nodes + } + + private def requireTime(deadline: Deadline): Unit = { + requireActive() + if (deadline.isOverdue()) throw new java.util.concurrent.TimeoutException("Fork resolution deadline") + } + + private def requireActive(): Unit = { + if (!active.get()) throw new java.util.concurrent.CancellationException("Fork resolution finished") + } + + private def waitFor[A](nodes: List[Node], observations: ConvergenceObservations, deadline: Deadline, + headerHeight: Option[Int], fixedSample: Option[String]) + (accept: Snapshot => Option[A]): Vector[A] = { + val currentPhase = phase + val probes = nodes.map { node => + def json(path: String): Future[Json] = { + requireTime(deadline) + node.singleGet(path, _.setRequestTimeout(5000)).map { response => + require(response.getStatusCode == 200, "Unexpected observation status") + node.ergoJsonAnswerAs[Json](response.getResponseBody) + } + } + val status = observations.probe { + requireTime(deadline) + node.singleGet("/info", _.setRequestTimeout(5000)).map { response => + require(response.getStatusCode == 200, "Unexpected observation status") + val parsed = Try(node.ergoJsonAnswerAs[Json](response.getResponseBody)) + val info = parsed.flatMap(j => Try(j.as[NodeInfo].fold(throw _, identity))) + .toEither.left.map(_.getClass.getSimpleName) + // Height gates retain NodeApi.fullHeight's exact field/default semantics. + val height = parsed.flatMap(j => Try(j.hcursor.downField("fullHeight").as[Option[Int]] + .fold(throw _, identity).getOrElse(0))).toEither.left.map(_.getClass.getSimpleName) + (info, height) + } + } + val peers = observations.probe(json("/peers/connected").map(_.asArray.get.size)) + val headers = headerHeight.map { height => + observations.probe(json(s"/blocks/at/$height").map(_.as[Seq[String]].fold(throw _, identity))) + } + (status, peers, headers) } + var accepted = Vector.fill[Option[A]](nodes.size)(None) + val result = observations.until(deadline, 100.millis, 5.seconds) { budget => + requireTime(deadline) + Future.traverse(probes.zipWithIndex) { case ((status, peers, headers), index) => + val infoResult = status.sample(budget) + val peerResult = peers.sample(budget) + val headerResult = headers.map(_.sample(budget)).getOrElse(Future.successful(Right(Seq.empty[String]))) + infoResult.zip(peerResult).zip(headerResult).map { case ((status, peerCount), ids) => + val snapshot = Snapshot(status.flatMap(_._1), peerCount, ids, + status.flatMap(_._2), status.isRight) + remember(describe(currentPhase, index, headerHeight, fixedSample, snapshot) + + s" sampledAt=${System.currentTimeMillis()}") + accept(snapshot) + } + }.map { observed => + accepted = latch(accepted, observed.toVector) + accepted + } + }(_.forall(_.isDefined))(s"Fork resolution did not complete; $evidence") + Await.result(result, deadline.timeLeft.max(Duration.Zero)).map(_.get) } // Testing scenario: @@ -71,38 +158,84 @@ class ForkResolutionSpec extends AnyFlatSpec with Matchers with IntegrationSuite // 5. Check that nodes reached consensus on created forks; it should "Fork resolution after isolated mining" in { - log.info(minerConfig.toString) - onlineSyncNodesConfig.foreach(x => log.info(x.toString)) - - val nodes: List[Node] = startNodesWithBinds(minerConfig +: onlineSyncNodesConfig) - - val result = Async.async { - val initMaxHeight = Async.await(Future.traverse(nodes)(_.fullHeight).map(_.max)) - Async.await(Future.traverse(nodes)(_.waitForHeight(initMaxHeight + commonChainLength, 100.millis))) - val isolatedNodes = Async.await { - nodes.foreach(node => docker.stopNode(node.containerId)) + val observations = new ConvergenceObservations + try { + enter("initial-start") + val nodes = startNodesWithBinds(minerConfig +: onlineSyncNodesConfig, observations) + // Match the original budget: initial startup precedes the 15-minute scenario deadline. + val deadline = 15.minutes.fromNow + val result = Future { + enter("initial-height") + val initMaxHeight = waitFor(nodes, observations, deadline, None, None)(_.fullHeight.toOption).max + val forkHeight = initMaxHeight + commonChainLength + forkLength + enter(s"common-height target=${initMaxHeight + commonChainLength}") + waitFor(nodes, observations, deadline, Some(initMaxHeight + commonChainLength), None)( + heightReached(initMaxHeight + commonChainLength)) + def stop(current: List[Node]): Unit = current.zipWithIndex.foreach { case (node, index) => + requireTime(deadline) + remember(s"phase=$phase node=$index operation=stop") + docker.stopNode(node.containerId) + } + enter("isolate-stop") + stop(nodes) + enter("isolate-clear-peers") clearPeerDatabases() - Future.successful(startNodesWithBinds(minerConfig +: offlineMiningNodesConfig, isolatedPeersConfig)) - } - val forkHeight = initMaxHeight + commonChainLength + forkLength - Async.await(Future.traverse(isolatedNodes)(_.waitForHeight(forkHeight, 100.millis))) - val regularNodes = Async.await { - isolatedNodes.foreach(node => docker.stopNode(node.containerId)) + enter("isolate-start") + val isolatedNodes = startNodesWithBinds(minerConfig +: offlineMiningNodesConfig, + observations, Some(deadline), isolatedPeersConfig) + enter(s"isolated-height target=$forkHeight") + waitFor(isolatedNodes, observations, deadline, Some(forkHeight), None)(heightReached(forkHeight)) + enter("reconnect-stop") + stop(isolatedNodes) + enter("reconnect-clear-peers") clearPeerDatabases() - Future.successful(startNodesWithBinds(minerConfig +: onlineSyncNodesConfig)) + enter("reconnect-start") + val regularNodes = startNodesWithBinds(minerConfig +: onlineSyncNodesConfig, observations, Some(deadline)) + enter(s"reconnected-height target=${forkHeight + syncLength}") + waitFor(regularNodes, observations, deadline, Some(forkHeight), None)(heightReached(forkHeight + syncLength)) + enter("select-anchor") + val sample = waitFor(regularNodes.take(1), observations, deadline, Some(forkHeight), None)( + _.headers.toOption).head.headOption.value + enter("match-anchor") + val headers = waitFor(regularNodes, observations, deadline, Some(forkHeight), Some(sample))( + _.headers.toOption.filter(_.headOption.contains(sample))) + val headerIdsAtSameHeight = headers.map(_.headOption.value) + headerIdsAtSameHeight should contain only sample + log.info(s"Fork resolution completed; $evidence") } - Async.await(Future.traverse(regularNodes)(_.waitForHeight(forkHeight + syncLength, 100.millis))) - val sample = Async.await(regularNodes.head.headerIdsByHeight(forkHeight)).headOption.value - val headers = Async.await(Future.traverse(regularNodes) { node => - node.waitFor[Seq[String]](_.headerIdsByHeight(forkHeight), _.headOption.contains(sample), 100.millis) - }) - - log.debug(s"Headers at height $forkHeight: ${headers.mkString(",")}") - val headerIdsAtSameHeight = headers.map(_.headOption.value) - headerIdsAtSameHeight should contain only sample + Await.result(result, deadline.timeLeft.max(Duration.Zero)) + } catch { + case NonFatal(error) => + log.error(s"Fork resolution failed errorClass=${error.getClass.getSimpleName}; $evidence") + throw error + } finally { + active.set(false) + observations.close() } - - Await.result(result, 15.minutes) } } + +private[it] object ForkResolutionDiagnostics { + final case class Snapshot(info: Either[String, NodeInfo], peers: Either[String, Int], + headers: Either[String, Seq[String]], fullHeight: Either[String, Int], + reachable: Boolean) + + def startupReached(snapshot: Snapshot): Option[Unit] = if (snapshot.reachable) Some(()) else None + + def heightReached(target: Int)(snapshot: Snapshot): Option[Int] = + snapshot.fullHeight.toOption.filter(_ >= target) + + def latch[A](previous: Vector[Option[A]], observed: Vector[Option[A]]): Vector[Option[A]] = + previous.zip(observed).map { case (accepted, current) => accepted.orElse(current) } + + def describe(phase: String, index: Int, height: Option[Int], fixedSample: Option[String], snapshot: Snapshot): String = { + def error(value: String): String = if (value.matches("[A-Za-z0-9_$]+")) value else "ObservationError" + val info = snapshot.info.fold(e => s"statusErrorClass=${error(e)}", value => + s"headerHeight=${value.bestHeaderHeightOpt} fullHeight=${value.bestBlockHeightOpt} mining=${value.isMining}") + val peers = snapshot.peers.fold(e => s"peerErrorClass=${error(e)}", count => s"peerCount=$count") + val selected = snapshot.headers.fold(e => s"headerErrorClass=${error(e)}", ids => + s"selected=${ids.headOption.map(ConvergenceObservations.headerId).getOrElse("missing")}") + s"phase=$phase node=$index height=$height fixed=${fixedSample.map(ConvergenceObservations.headerId)} $info $peers $selected" + } +}