Apply wallet confirmation filters within valid height ranges

src/it/scala/org/ergoplatform/it/DeepRollBackSpec.scala

@@ -3,8 +3,9 @@ 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 org.ergoplatform.it.api.NodeApi.{NodeInfo, nodeInfoDecoder}
 import org.ergoplatform.it.container.{IntegrationSuite, Node}
+import org.ergoplatform.it.util.ConvergenceObservations
 import org.ergoplatform.nodeView.history.ErgoHistoryUtils
 import org.scalatest.freespec.AnyFreeSpec
 import scala.async.Async
@@ -45,6 +46,42 @@ class DeepRollBackSpec extends AnyFreeSpec with IntegrationSuite {
     .withFallback(nonGeneratingPeerConfig)
     .withFallback(allowLocalConfig)
 
+  private val seedObservations = new ConvergenceObservations
+  @volatile private var lastSeedObservation = "Initial seed has not been sampled"
+
+  private def waitForSettledSeed(
+    nodeA: Node,
+    nodeB: Node,
+    timeout: FiniteDuration
+  ): Future[(NodeInfo, NodeInfo)] = {
+    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 def waitForSameBestBlock(
     nodeA: Node,
     nodeB: Node,
@@ -100,16 +137,20 @@ 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
 
-      // 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,
@@ -162,7 +203,15 @@ 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")
+        throw error
+    } finally {
+      seedObservations.close()
+    }
   }
 
 }

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

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