Reject trailing transaction bytes

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)
+    }
+  }
+}

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")
+}

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()
   }
 
 }