Keep wallet scan metadata, caches and write acknowledgements consistent

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/ForkResolutionDiagnosticsSpec.scala

@@ -0,0 +1,57 @@
+package org.ergoplatform.it
+
+import org.ergoplatform.it.api.NodeApi.NodeInfo
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class ForkResolutionDiagnosticsSpec extends AnyFlatSpec with Matchers {
+  import ForkResolutionDiagnostics._
+
+  "Fork observations" should "retain independently accepted results while later selections change" in {
+    val first = latch(Vector.fill[Option[String]](4)(None), Vector(Some("a"), None, Some("a"), None))
+    val later = latch(first, Vector(None, Some("a"), None, Some("a")))
+    later shouldBe Vector.fill(4)(Some("a"))
+    latch(later, Vector.fill(4)(Some("b"))) shouldBe later
+  }
+
+  it should "retain missing results until their own predicate succeeds" in {
+    latch(Vector(Some("a"), None), Vector(None, None)) shouldBe Vector(Some("a"), None)
+  }
+
+  it should "report the fixed target separately from the current selected header" in {
+    val target = "ab" * 32
+    val current = "cd" * 32
+    val snapshot = Snapshot(Right(NodeInfo(None, None, Some(25), Some(20), None, Some(false))),
+      Right(3), Right(Seq(current, target)), Right(20), reachable = true)
+    val text = describe("match-anchor", 2, Some(10), Some(target), snapshot)
+    text should include(s"fixed=Some($target)")
+    text should include(s"selected=$current")
+    text should include("headerHeight=Some(25) fullHeight=Some(20) mining=Some(false)")
+    text should include("peerCount=3")
+  }
+
+  it should "exclude response text, addresses and paths from observed fields" in {
+    val payload = "http://example.invalid/private/location"
+    val snapshot = Snapshot(Right(NodeInfo(Some(payload), Some(payload), Some(1), Some(1), Some(payload), None)),
+      Left(payload), Right(Seq(payload)), Right(1), reachable = true)
+    val text = describe("match-anchor", 0, Some(1), Some(payload), snapshot)
+    text should not include payload
+    text should include("peerErrorClass=ObservationError")
+    text should include("selected=invalid-header-id")
+    text should include("fixed=Some(invalid-header-id)")
+    describe("initial-height", 0, None, None,
+      Snapshot(Left("TimeoutException"), Left("IOException"), Left("ParsingFailure"),
+        Left("TimeoutException"), reachable = false)) should
+      include("statusErrorClass=TimeoutException peerErrorClass=IOException headerErrorClass=ParsingFailure")
+  }
+
+  it should "keep supplementary failures out of the startup and height acceptance gates" in {
+    val snapshot = Snapshot(Left("ParsingFailure"), Left("TimeoutException"), Left("IOException"),
+      Right(20), reachable = true)
+    startupReached(snapshot) shouldBe Some(())
+    heightReached(20)(snapshot) shouldBe Some(20)
+    heightReached(21)(snapshot) shouldBe None
+    heightReached(20)(snapshot.copy(fullHeight = Left("ParsingFailure"))) shouldBe None
+    startupReached(snapshot.copy(reachable = false)) shouldBe None
+  }
+}