From b2707758f8f8bf8ac27b1d12a37404cb2fb5fb79 Mon Sep 17 00:00:00 2001 From: "A. Shannon" <204582608+a-shannon@users.noreply.github.com> Date: Fri, 4 Sep 2026 21:00:24 +0200 Subject: [PATCH 1/7] fix: reject trailing transaction bytes --- .../http/api/TransactionsApiRoute.scala | 21 +++++++++-- .../http/routes/TransactionApiRouteSpec.scala | 37 ++++++++++++++++++- 2 files changed, 54 insertions(+), 4 deletions(-) diff --git a/src/main/scala/org/ergoplatform/http/api/TransactionsApiRoute.scala b/src/main/scala/org/ergoplatform/http/api/TransactionsApiRoute.scala index 95ec88a5b8..852534283b 100644 --- a/src/main/scala/org/ergoplatform/http/api/TransactionsApiRoute.scala +++ b/src/main/scala/org/ergoplatform/http/api/TransactionsApiRoute.scala @@ -18,12 +18,14 @@ import org.ergoplatform.settings.{Algos, Constants, ErgoSettings, RESTApiSetting import scorex.core.api.http.ApiResponse import scorex.crypto.authds.ADKey import scorex.util.encode.Base16 +import scorex.util.serialization.VLQByteBufferReader import sigma.VersionContext import sigma.ast.{EvaluatedValue, SType} import sigmastate.eval.Extensions.ArrayByteOps +import java.nio.ByteBuffer import scala.concurrent.Future -import scala.util.{Failure, Success} +import scala.util.{Failure, Success, Try} case class TransactionsApiRoute(readersHolder: ActorRef, nodeViewActorRef: ActorRef, @@ -170,6 +172,19 @@ case class TransactionsApiRoute(readersHolder: ActorRef, } } + private def parseTransactionBytes(txBytesStr: String): Try[ErgoTransaction] = { + Base16.decode(fromJsonOrPlain(txBytesStr)).flatMap { txBytes => + Try { + val reader = new VLQByteBufferReader(ByteBuffer.wrap(txBytes)) + val tx = ErgoTransactionSerializer.parse(reader) + if (reader.remaining != 0) { + throw new IllegalArgumentException("Transaction bytes contain trailing data") + } + tx + } + } + } + def sendTransactionR: Route = (pathEnd & post & entity(as[ErgoTransaction])) { tx => validateTransactionAndProcess(tx)(validTx => sendLocalTransactionRoute(nodeViewActorRef, validTx)) @@ -183,7 +198,7 @@ case class TransactionsApiRoute(readersHolder: ActorRef, // we check parsed with max version available val version = ergoSettings.chainSettings.protocolVersion VersionContext.withVersions(version, version) { - Base16.decode(fromJsonOrPlain(txBytesStr)).flatMap(ErgoTransactionSerializer.parseBytesTry) match { + parseTransactionBytes(txBytesStr) match { case Success(tx) => validateTransactionAndProcess(tx)(validTx => sendLocalTransactionRoute(nodeViewActorRef, validTx)) case Failure(e) => @@ -203,7 +218,7 @@ case class TransactionsApiRoute(readersHolder: ActorRef, // actual tree version is properly set in ErgoTreeSerializer inside val version = ergoSettings.chainSettings.protocolVersion VersionContext.withVersions(version, version) { - Base16.decode(fromJsonOrPlain(txBytesStr)).flatMap(ErgoTransactionSerializer.parseBytesTry) match { + parseTransactionBytes(txBytesStr) match { case Success(tx) => validateTransactionAndProcess(tx)(validTx => ApiResponse(validTx.transaction.id)) case Failure(e) => diff --git a/src/test/scala/org/ergoplatform/http/routes/TransactionApiRouteSpec.scala b/src/test/scala/org/ergoplatform/http/routes/TransactionApiRouteSpec.scala index a9a06fc040..9d718eac82 100644 --- a/src/test/scala/org/ergoplatform/http/routes/TransactionApiRouteSpec.scala +++ b/src/test/scala/org/ergoplatform/http/routes/TransactionApiRouteSpec.scala @@ -9,7 +9,7 @@ import io.circe.Json import io.circe.syntax._ import org.ergoplatform.ErgoBox.{AdditionalRegisters, NonMandatoryRegisterId, TokenId} import org.ergoplatform.http.api.{ApiCodecs, TransactionsApiRoute} -import org.ergoplatform.modifiers.mempool.{ErgoTransaction, UnconfirmedTransaction} +import org.ergoplatform.modifiers.mempool.{ErgoTransaction, ErgoTransactionSerializer, UnconfirmedTransaction} import org.ergoplatform.nodeView.ErgoReadersHolder.{GetDataFromHistory, GetReaders, Readers} import org.ergoplatform.settings.RESTApiSettings import org.ergoplatform.utils.Stubs @@ -17,6 +17,7 @@ import org.ergoplatform.{DataInput, ErgoBox, ErgoBoxCandidate, Input} import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers import scorex.util.encode.Base16 +import sigma.VersionContext import sigmastate.eval.Extensions._ import sigma.Extensions.ArrayOps import sigma.ast.{ByteArrayConstant, EvaluatedValue, SType} @@ -56,6 +57,12 @@ class TransactionApiRouteSpec extends AnyFlatSpec val output: ErgoBoxCandidate = new ErgoBoxCandidate(inputBox.value, TrueTree, creationHeight = 0, tokens.toArray.toColl, registers) val tx: ErgoTransaction = ErgoTransaction(IndexedSeq(input), IndexedSeq(dataInput), IndexedSeq(output)) + val canonicalTxBytes: String = { + val version = settings.chainSettings.protocolVersion + VersionContext.withVersions(version, version) { + Base16.encode(ErgoTransactionSerializer.toBytes(tx)) + } + } val chainedInput = Input(tx.outputs.head.id, emptyProverResult) val chainedTx: ErgoTransaction = ErgoTransaction(IndexedSeq(chainedInput), IndexedSeq(output)) @@ -126,6 +133,34 @@ class TransactionApiRouteSpec extends AnyFlatSpec } } + it should "accept canonical transaction bytes" in { + Seq("checkBytes", "bytes").foreach { endpoint => + withClue(endpoint) { + Post(prefix + s"/$endpoint", canonicalTxBytes) ~> route ~> check { + status shouldBe StatusCodes.OK + responseAs[String] shouldBe tx.id + } + } + } + } + + it should "reject transaction bytes with trailing data" in { + val txBytes = canonicalTxBytes + "00" + + Seq("checkBytes", "bytes").foreach { endpoint => + withClue(endpoint) { + Post(prefix + s"/$endpoint", txBytes) ~> route ~> check { + status shouldBe StatusCodes.BadRequest + val response = responseAs[Json].hcursor + response.get[Int]("error") shouldBe Right(StatusCodes.BadRequest.intValue) + response.get[String]("reason") shouldBe Right("bad.request") + response.get[String]("detail") shouldBe + Right("Can not parse transaction bytes: Transaction bytes contain trailing data") + } + } + } + } + it should "get unconfirmed txs from mempool" in { Get(prefix + "/unconfirmed") ~> route ~> check { status shouldBe StatusCodes.OK 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 2/7] 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 3/7] 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 4/7] 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 bcfcd61166151d8000ae2a05a62fa4f389aa7126 Mon Sep 17 00:00:00 2001 From: "A. Shannon" <217575710+a-shannon@users.noreply.github.com> Date: Sun, 13 Sep 2026 04:33:47 +0200 Subject: [PATCH 5/7] Share selected-header convergence observation --- .../it/UtxoStateNodesSyncSpec.scala | 55 +++++++++++++++---- 1 file changed, 45 insertions(+), 10 deletions(-) diff --git a/src/it/scala/org/ergoplatform/it/UtxoStateNodesSyncSpec.scala b/src/it/scala/org/ergoplatform/it/UtxoStateNodesSyncSpec.scala index 9e0038b8a6..2d8190359f 100644 --- a/src/it/scala/org/ergoplatform/it/UtxoStateNodesSyncSpec.scala +++ b/src/it/scala/org/ergoplatform/it/UtxoStateNodesSyncSpec.scala @@ -2,6 +2,7 @@ package org.ergoplatform.it import com.typesafe.config.Config import org.ergoplatform.it.container.{IntegrationSuite, Node} +import org.ergoplatform.it.util.ConvergenceObservations import org.scalatest.flatspec.AnyFlatSpec import scala.concurrent.duration._ @@ -27,21 +28,55 @@ class UtxoStateNodesSyncSpec extends AnyFlatSpec with IntegrationSuite { val nodes: List[Node] = docker.startDevNetNodes(nodeConfigs).get it should s"Utxo state nodes synchronisation ($blocksQty blocks)" in { + val deadline = 15.minutes.fromNow + val observations = new ConvergenceObservations + @volatile var recent = Vector.empty[String] val result = for { initHeight <- Future.traverse(nodes)(_.fullHeight).map(x => math.max(x.max, 1)) _ <- Future.traverse(nodes)(_.waitForHeight(initHeight + blocksQty)) - headers <- Future.traverse(nodes)( - _.headerIdsByHeight(initHeight + blocksQty - forkDepth) - ) + headers <- { + val height = initHeight + blocksQty - forkDepth + val probes = nodes.map { node => + observations.probe(node.singleGet(s"/blocks/at/$height", _.setRequestTimeout(5000)) + .map { r => + require(r.getStatusCode == 200, "Unexpected observation status") + node.ergoJsonAnswerAs[Seq[String]](r.getResponseBody) + }) + } + observations.until(deadline, 1.second, 5.seconds) { budget => + Future.traverse(probes.zipWithIndex) { case (probe, index) => + probe.sample(budget).map { result => + val selection = result.fold(error => s"errorClass=$error", ids => + ids.headOption.map(ConvergenceObservations.headerId).getOrElse("missing")) + synchronized { + recent = (recent :+ s"node$index height=$height selected=$selection sampledAt=${System.currentTimeMillis()}") + .takeRight(nodes.size * 3) + } + result.toOption.getOrElse(Seq.empty) + } + } + }(ConvergenceObservations.selectedHeadersAgree)( + s"Selected headers did not converge at height $height; recent observations: ${recent.mkString("; ")}") + } } yield { - log.info( - s"Headers at height ${initHeight + blocksQty - forkDepth}: ${headers.mkString(",")}" - ) - val headerIdsAtSameHeight = headers.flatten - val sample = headerIdsAtSameHeight.head - headerIdsAtSameHeight should contain only sample + log.info(s"Selected header convergence: ${recent.mkString("; ")}") + // `/blocks/at/{height}` returns *every* header id known at the given height, with the + // best-chain one first (see `HeadersProcessor.headerIdsAtHeight`). Nodes are in sync + // when their best-chain header at that height matches; a node may legitimately also + // know orphaned headers of a fork that was already resolved, so flattening all the + // returned ids makes the assertion fail on a perfectly synchronised network. + // Same convention as ForkResolutionSpec and DeepRollBackSpec, which compare `.head`. + headers.foreach(_ should not be empty) + val bestChainHeaderIds = headers.map(_.head) + val sample = bestChainHeaderIds.head + bestChainHeaderIds should contain only sample } - Await.result(result, 15.minutes) + try Await.result(result, deadline.timeLeft.max(Duration.Zero)) + catch { + case error: java.util.concurrent.TimeoutException => + log.error(s"UTXO synchronization timed out; recent observations: ${recent.mkString("; ")}") + throw error + } finally observations.close() } } 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 6/7] 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 7/7] 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") +}