Propagate history repair failures before publishing recovered state

src/it/scala/org/ergoplatform/it/util/ConvergenceObservations.scala

@@ -0,0 +1,100 @@
+package org.ergoplatform.it.util
+
+import java.util.concurrent.{ScheduledThreadPoolExecutor, ThreadFactory, TimeUnit, TimeoutException}
+import org.ergoplatform.it.api.NodeApi.NodeInfo
+
+import scala.concurrent.{ExecutionContext, Future, Promise}
+import scala.concurrent.duration._
+import scala.util.{Failure, Success, Try}
+
+/** Bounded, single-flight observations for integration assertions. */
+final class ConvergenceObservations(implicit ec: ExecutionContext) extends AutoCloseable {
+  private val timer = new ScheduledThreadPoolExecutor(1, new ThreadFactory {
+    override def newThread(runnable: Runnable): Thread = {
+      val thread = new Thread(runnable, "convergence-observations")
+      thread.setDaemon(true)
+      thread
+    }
+  })
+  timer.setRemoveOnCancelPolicy(true)
+
+  private def bounded[A](future: Future[A], budget: FiniteDuration): Future[A] = {
+    val result = Promise[A]()
+    val timeout = timer.schedule(new Runnable {
+      override def run(): Unit = result.tryFailure(new TimeoutException("Observation deadline"))
+    }, math.max(0L, budget.toNanos), TimeUnit.NANOSECONDS)
+    future.onComplete { value =>
+      result.tryComplete(value)
+      timeout.cancel(false)
+    }
+    result.future
+  }
+
+  final class Probe[A](request: () => Future[A]) {
+    private var pending: Option[Future[A]] = None
+
+    def sample(budget: FiniteDuration): Future[Either[String, A]] = {
+      val response = synchronized {
+        val current = pending.filterNot(_.isCompleted).getOrElse {
+          Try(request()) match {
+            case Success(value) => value
+            case Failure(error) => Future.failed(error)
+          }
+        }
+        pending = Some(current)
+        current
+      }
+      bounded(response, budget).map(value => Right(value): Either[String, A]).recover {
+        case scala.util.control.NonFatal(error) => Left(error.getClass.getSimpleName)
+      }
+    }
+  }
+
+  def probe[A](request: => Future[A]): Probe[A] = new Probe(() => request)
+
+  def until[A](deadline: Deadline, interval: FiniteDuration, sampleBudget: FiniteDuration)
+              (observe: FiniteDuration => Future[A])(accept: A => Boolean)
+              (failure: => String): Future[A] = {
+    def expired: Future[A] = Future.failed(new TimeoutException(failure))
+
+    def loop(): Future[A] = {
+      if (deadline.isOverdue()) expired
+      else {
+        val remaining = deadline.timeLeft
+        bounded(observe(sampleBudget.min(remaining)), remaining).flatMap { value =>
+          if (deadline.isOverdue()) expired
+          else if (accept(value)) Future.successful(value)
+          else {
+            val next = Promise[Unit]()
+            timer.schedule(new Runnable {
+              override def run(): Unit = next.trySuccess(())
+            }, interval.min(deadline.timeLeft).max(Duration.Zero).toNanos, TimeUnit.NANOSECONDS)
+            next.future.flatMap(_ => loop())
+          }
+        }.recoverWith {
+          case _: TimeoutException => expired
+        }
+      }
+    }
+
+    loop()
+  }
+
+  override def close(): Unit = timer.shutdownNow()
+}
+
+object ConvergenceObservations {
+  def sameBestBlock(infoA: NodeInfo, infoB: NodeInfo, minHeight: Int): Boolean = {
+    val sameHeight = infoA.bestBlockHeightOpt.nonEmpty && infoA.bestBlockHeightOpt == infoB.bestBlockHeightOpt
+    val sameBlock = infoA.bestBlockIdOpt.nonEmpty && infoA.bestBlockIdOpt == infoB.bestBlockIdOpt
+    val highEnough = infoA.bestBlockHeightOpt.exists(_ >= minHeight)
+    sameHeight && sameBlock && highEnough
+  }
+
+  def selectedHeadersAgree(headers: Seq[Seq[String]]): Boolean =
+    headers.nonEmpty && headers.forall(_.headOption.exists(_.nonEmpty)) &&
+      headers.map(_.head).distinct.size == 1
+
+  def headerId(value: String): String =
+    if (value.matches("[0-9a-fA-F]{64}")) value else "invalid-header-id"
+}

src/it/scala/org/ergoplatform/it/util/ConvergenceObservationsSpec.scala

@@ -0,0 +1,96 @@
+package org.ergoplatform.it.util
+
+import java.util.concurrent.TimeoutException
+import java.util.concurrent.atomic.AtomicInteger
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.ergoplatform.it.api.NodeApi.NodeInfo
+
+import scala.concurrent.{Await, ExecutionContext, Future, Promise}
+import scala.concurrent.duration._
+
+class ConvergenceObservationsSpec extends AnyFlatSpec with Matchers {
+  implicit private val ec: ExecutionContext = ExecutionContext.global
+
+  private def withObserver(test: ConvergenceObservations => Unit): Unit = {
+    val observer = new ConvergenceObservations
+    try test(observer)
+    finally observer.close()
+  }
+
+  "Selected header agreement" should "accept retained alternatives only when every first ID agrees" in {
+    ConvergenceObservations.selectedHeadersAgree(Seq(Seq("a", "b"), Seq("a", "c"))) shouldBe true
+    ConvergenceObservations.selectedHeadersAgree(Seq(Seq("a", "b"), Seq("b", "a"))) shouldBe false
+    ConvergenceObservations.selectedHeadersAgree(Seq(Seq("a"), Seq.empty)) shouldBe false
+    ConvergenceObservations.selectedHeadersAgree(Seq(Seq(""), Seq(""))) shouldBe false
+    ConvergenceObservations.selectedHeadersAgree(Seq.empty) shouldBe false
+  }
+
+  "Full block agreement" should "require both heights and IDs and the original minimum height" in {
+    val info = NodeInfo(Some("header"), Some("block"), Some(60), Some(50), None, None)
+    ConvergenceObservations.sameBestBlock(info, info, 50) shouldBe true
+    ConvergenceObservations.sameBestBlock(info, info, 51) shouldBe false
+    ConvergenceObservations.sameBestBlock(info, info.copy(bestBlockHeightOpt = Some(51)), 50) shouldBe false
+    ConvergenceObservations.sameBestBlock(info, info.copy(bestBlockIdOpt = Some("other")), 50) shouldBe false
+    ConvergenceObservations.sameBestBlock(info.copy(bestBlockHeightOpt = None), info, 50) shouldBe false
+    ConvergenceObservations.sameBestBlock(info, info.copy(bestBlockHeightOpt = None), 50) shouldBe false
+    ConvergenceObservations.sameBestBlock(info.copy(bestBlockIdOpt = None), info, 50) shouldBe false
+    ConvergenceObservations.sameBestBlock(info, info.copy(bestBlockIdOpt = None), 50) shouldBe false
+  }
+
+  it should "resample the entire group until its current selections agree" in withObserver { observer =>
+    val samples = new AtomicInteger()
+    val result = observer.until(2.seconds.fromNow, 1.millis, 100.millis) { _ =>
+      val headers = if (samples.incrementAndGet() == 1) Seq(Seq("a", "b"), Seq("b", "a"))
+                    else Seq(Seq("b", "a"), Seq("b"))
+      Future.successful(headers)
+    }(ConvergenceObservations.selectedHeadersAgree)("selected headers disagree")
+    Await.result(result, 3.seconds).map(_.head) shouldBe Seq("b", "b")
+    samples.get() shouldBe 2
+  }
+
+  it should "fail persistent disagreement within the original deadline with recent evidence" in withObserver { observer =>
+    var recent = "none"
+    val result = observer.until(100.millis.fromNow, 1.millis, 20.millis) { _ =>
+      recent = "node0=a node1=b"
+      Future.successful(Seq(Seq("a"), Seq("b")))
+    }(ConvergenceObservations.selectedHeadersAgree)(s"last: $recent")
+    intercept[TimeoutException](Await.result(result, 2.seconds)).getMessage should include("node0=a node1=b")
+  }
+
+  "Observation probes" should "bound a stalled endpoint and not start overlapping requests" in withObserver { observer =>
+    val calls = new AtomicInteger()
+    val never = Promise[Int]()
+    val probe = observer.probe { calls.incrementAndGet(); never.future }
+    Await.result(probe.sample(20.millis), 2.seconds) shouldBe Left("TimeoutException")
+    Await.result(probe.sample(20.millis), 2.seconds) shouldBe Left("TimeoutException")
+    calls.get() shouldBe 1
+    never.success(3)
+    Await.result(never.future, 2.seconds) shouldBe 3
+    Await.result(probe.sample(100.millis), 2.seconds) shouldBe Right(3)
+    calls.get() shouldBe 2
+  }
+
+  it should "keep successful status available while the peer sample times out" in withObserver { observer =>
+    val status = observer.probe(Future.successful(42)).sample(100.millis)
+    val peers = observer.probe(Promise[Int]().future).sample(30.millis)
+    Await.result(status, 2.seconds) shouldBe Right(42)
+    Await.result(status.zip(peers), 2.seconds) shouldBe (Right(42) -> Left("TimeoutException"))
+  }
+
+  it should "retain only an error class for endpoint failures" in withObserver { observer =>
+    val result = observer.probe(Future.failed[Int](new IllegalArgumentException("private diagnostic payload")))
+    Await.result(result.sample(100.millis), 2.seconds) shouldBe Left("IllegalArgumentException")
+  }
+
+  it should "enforce the convergence deadline even when observation never returns" in withObserver { observer =>
+    val result = observer.until(30.millis.fromNow, 1.millis, 10.millis)(_ => Promise[Boolean]().future)(identity)("recent status")
+    intercept[TimeoutException](Await.result(result, 2.seconds)).getMessage shouldBe "recent status"
+  }
+
+  "Observation identifiers" should "exclude arbitrary response text" in {
+    ConvergenceObservations.headerId("ab" * 32) shouldBe "ab" * 32
+    ConvergenceObservations.headerId("unexpected response text") shouldBe "invalid-header-id"
+  }
+}

src/main/scala/org/ergoplatform/nodeView/ErgoNodeViewHolder.scala

@@ -19,7 +19,7 @@ import org.ergoplatform.nodeView.mempool.ErgoMemPoolUtils.ProcessingOutcome
 import org.ergoplatform.nodeView.state._
 import org.ergoplatform.nodeView.wallet.ErgoWallet
 import org.ergoplatform.settings.{Algos, Constants, ErgoSettings, NetworkType, ScorexSettings}
-import org.ergoplatform.utils.ScorexEncoding
+import org.ergoplatform.utils.{LoggingUtil, ScorexEncoding}
 import org.ergoplatform.validation.{MalformedModifierError, RecoverableModifierError}
 import org.ergoplatform.wallet.utils.FileUtils
 import scorex.util.{ModifierId, ScorexLogging}
@@ -809,9 +809,13 @@ object ErgoNodeViewHolder {
         history.bestFullBlockOpt
           .filter(_.id != lastMod.id)
           .fold("")(fb => s"\n best full block: $fb")
-      val repairNeeded = ErgoHistory.repairIfNeeded(history)
+      val repairStatus = ErgoHistory.repairIfNeeded(history) match {
+        case Success(false) => "repair not needed"
+        case Success(true) => "repair completed"
+        case Failure(error) => s"repair failed: ${LoggingUtil.getReasonMsg(error)}"
+      }
       ChainIsStuck(s"Chain not modified for $chainUpdateDelay ms, headers-height: $headersHeight, " +
-        s"block-height $blockHeight, chain synced: $chainSynced, repair needed: $repairNeeded, " +
+        s"block-height $blockHeight, chain synced: $chainSynced, $repairStatus, " +
         s"last modifier applied: $lastMod, " +
         s"possible best full block $bestFullBlockOpt")
     } else {

src/main/scala/org/ergoplatform/nodeView/history/ErgoHistory.scala

@@ -203,24 +203,23 @@ trait ErgoHistory
   /**
     * Remove header, corresponding block parts, and corresponding indexes from storage and caches
     * @param headerId - header id
-    * @return
+    * @return Success after every removal, or the first removal failure
     */
-  def forgetHeader(headerId: ModifierId): Try[Unit] = Try {
-    val hOpt = typedModifierById[Header](headerId)
-      val hRes = historyStorage.remove(
+  def forgetHeader(headerId: ModifierId): Try[Unit] = Try(typedModifierById[Header](headerId)).flatMap { hOpt =>
+    // Keep the header until its sections are removed so a retry can still discover their identifiers.
+    hOpt.toSeq.flatMap(requiredModifiersForHeader).foldLeft[Try[Unit]](Success(())) {
+      case (result, (_, mId)) => result.flatMap { _ =>
+        val removal = historyStorage.remove(Array(validityKey(mId)), Array(mId))
+        log.info(s"Result of removing modifier $mId: " + removal)
+        removal
+      }
+    }.flatMap { _ =>
+      val removal = historyStorage.remove(
         indicesToRemove = Array(validityKey(headerId), headerHeightKey(headerId), headerScoreKey(headerId)),
         idsToRemove = Array(headerId)
       )
-    log.info(s"Result of removing header $headerId: " + hRes)
-
-    hOpt.foreach { h =>
-      requiredModifiersForHeader(h).foreach { case (_, mId) =>
-        val mRes = historyStorage.remove(
-          indicesToRemove = Array(validityKey(mId)),
-          idsToRemove = Array(mId)
-        )
-        log.info(s"Result of removing modifier $mId: " + mRes)
-      }
+      log.info(s"Result of removing header $headerId: " + removal)
+      removal
     }
   }
 
@@ -239,30 +238,34 @@ object ErgoHistory extends ScorexLogging {
     dir
   }
 
-  // check if there is possible database corruption when there is header after
-  // recognized blockchain tip marked as invalid
-  protected[nodeView] def repairIfNeeded(history: ErgoHistory): Boolean = history.historyStorage.synchronized {
+  // Success(false) means no repair was needed; Success(true) means all removals completed.
+  // Failure preserves the first error; the height index is removed only after all headers are forgotten.
+  protected[nodeView] def repairIfNeeded(history: ErgoHistory): Try[Boolean] = Try(history.historyStorage.synchronized {
     val bestHeaderHeight = history.headersHeight
     val bestFullBlockHeight = history.bestFullBlockOpt.map(_.height).getOrElse(-1)
     val afterHeaders = history.headerIdsAtHeight(bestHeaderHeight + 1)
 
     if (bestHeaderHeight == bestFullBlockHeight && afterHeaders.nonEmpty) {
       log.warn("Found suspicious continuation, clearing it...")
-      afterHeaders.map { hId =>
-        history.forgetHeader(hId)
-      }
-      history.historyStorage.remove(Array(history.heightIdsKey(bestHeaderHeight + 1)), Array.empty[ModifierId])
-      true
+      afterHeaders.foldLeft[Try[Unit]](Success(())) { (result, hId) =>
+        result.flatMap(_ => history.forgetHeader(hId))
+      }.flatMap { _ =>
+        history.historyStorage.remove(Array(history.heightIdsKey(bestHeaderHeight + 1)), Array.empty[ModifierId])
+      }.map(_ => true)
     } else {
-      false
+      Success(false)
     }
-  }
+  }).flatten
 
   /**
     * @return ErgoHistory instance with new database or database read from existing folder
     */
-  def readOrGenerate(ergoSettings: ErgoSettings)(implicit context: ActorContext): ErgoHistory = {
-    var db = HistoryStorage(ergoSettings)
+  def readOrGenerate(ergoSettings: ErgoSettings)(implicit context: ActorContext): ErgoHistory =
+    readOrGenerate(ergoSettings, HistoryStorage(ergoSettings))
+
+  private[history] def readOrGenerate(ergoSettings: ErgoSettings,
+                                    storage: HistoryStorage)(implicit context: ActorContext): ErgoHistory = {
+    var db = storage
 
     // ExtraIndexer db check
     if(ergoSettings.nodeSettings.extraIndex) { // check db schema
@@ -292,7 +295,14 @@ object ErgoHistory extends ScorexLogging {
         }
     }
 
-    repairIfNeeded(history)
+    repairIfNeeded(history) match {
+      case Failure(error) =>
+        Try(history.closeStorage()).failed.foreach { closeError =>
+          if (closeError ne error) error.addSuppressed(closeError)
+        }
+        throw error
+      case Success(_) =>
+    }
 
     log.info("History database read")
     if(ergoSettings.nodeSettings.extraIndex) // start extra indexer, if enabled

src/main/scala/org/ergoplatform/nodeView/history/storage/HistoryStorage.scala

@@ -185,7 +185,7 @@ class HistoryStorage(indexStore: LDBKVStore, objectsStore: LDBKVStore, extraStor
   def remove(indicesToRemove: Array[ByteArrayWrapper],
              idsToRemove: Array[ModifierId]): Try[Unit] = {
 
-      objectsStore.remove(idsToRemove.map(idToBytes)).map { _ =>
+      objectsStore.remove(idsToRemove.map(idToBytes)).flatMap { _ =>
         cfor(0)(_ < idsToRemove.length, _ + 1) { i => removeModifier(idsToRemove(i))}
         indexStore.remove(indicesToRemove.map(_.data)).map { _ =>
           cfor(0)(_ < indicesToRemove.length, _ + 1) { i => indexCache.invalidate(indicesToRemove(i))}