Preserve effective wallet keystore directory configuration

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/settings/ErgoSettingsReader.scala

@@ -4,7 +4,7 @@ import java.io.{File, FileOutputStream}
 import java.nio.channels.Channels
 import ch.qos.logback.classic.{Level, LoggerContext}
 import org.slf4j.{Logger, LoggerFactory}
-import com.typesafe.config.{Config, ConfigFactory, ConfigValueFactory}
+import com.typesafe.config.{Config, ConfigFactory}
 import net.ceedubs.ficus.Ficus._
 import net.ceedubs.ficus.readers.ArbitraryTypeReader._
 import org.ergoplatform.nodeView.state.StateType.Digest
@@ -82,24 +82,15 @@ object ErgoSettingsReader extends ScorexLogging
       require(new File(s"$secretDirName").canRead, s"Folder $secretDirName does not exist or not readable")
     }
 
-    val fullConfig = ConfigFactory
+    // Resolve path substitutions only after all configuration layers have been merged.
+    ConfigFactory
       .defaultOverrides()
       .withFallback(cfg)
       .withFallback(firstFallBack)
       .withFallback(ConfigFactory.defaultApplication())
       .withFallback(ConfigFactory.defaultReference())
       .resolve()
 
-    // If user provided only ergo.directory but not ergo.wallet.secretStorage.secretDir in his config,
-    // set ergo.wallet.secretStorage.secretDir like in reference.conf (so ergo.directory + "/wallet/keystore")
-    // Otherwise, a user may have an issue, especially with Powershell it seems from reports.
-    userDirOpt.map { userDir =>
-      if(walletKeystoreDirOpt.isEmpty) {
-        fullConfig.withValue(keystorePath, ConfigValueFactory.fromAnyRef(userDir + "/wallet/keystore"))
-      } else {
-        fullConfig
-      }
-    }.getOrElse(fullConfig)
   }
 
   private def readConfig(args: Args): Config = {

src/test/scala/org/ergoplatform/mining/CandidateGeneratorSpec.scala

@@ -892,74 +892,68 @@ class CandidateGeneratorSpec extends AnyFlatSpec with Matchers with ErgoTestHelp
 
   it should "ignore cached candidate when forced = true" in new TestKit(ActorSystem()) {
     val testProbe = new TestProbe(system)
-    system.eventStream.subscribe(testProbe.ref, newBlockSignal)
+    val viewHolderProbe = new TestProbe(system)
 
     val testDir = s"${defaultSettings.directory}-ignore-cache-${System.currentTimeMillis()}"
-    val settingsWithShortRegeneration: ErgoSettings =
+    val settingsForExplicitForcing: ErgoSettings =
       ErgoSettingsReader.read()
         .copy(
           nodeSettings = defaultSettings.nodeSettings
-            .copy(blockCandidateGenerationInterval = 1.millis),
+            // Keep automatic refresh outside this test of explicit forcing.
+            .copy(blockCandidateGenerationInterval = 1.hour),
           chainSettings =
             ErgoSettingsReader.read().chainSettings.copy(blockInterval = 1.seconds),
           directory = testDir
         )
 
-    val viewHolderRef: ActorRef = ErgoNodeViewRef(settingsWithShortRegeneration)
-    val readersHolderRef: ActorRef = ErgoReadersHolderRef(viewHolderRef)
+    // Keep readers fixed: a later ChangedMempool event may legitimately regenerate the cache.
+    val (initialState, boxes) = createUtxoState(settingsForExplicitForcing)
+    val block = validFullBlock(None, initialState, boxes)
+    val state = initialState.applyModifier(block, None)(_ => ()).get
+    // Initialization computes mining-time averages from pairs of headers.
+    val history = historyWithBestFullBlock(Seq(block))
+    val readers = Readers(history, state, ErgoMemPool.empty(settingsForExplicitForcing), walletStub)
+    val readersHolderRef = system.actorOf(Props(new FixedReadersHolder(readers)))
 
     val candidateGenerator: ActorRef =
       CandidateGenerator(
         defaultMinerSecret.publicImage,
         readersHolderRef,
-        viewHolderRef,
-        settingsWithShortRegeneration
+        viewHolderProbe.ref,
+        settingsForExplicitForcing
       )
 
-    val powScheme = settingsWithShortRegeneration.chainSettings.powScheme
-
-    // First mine a block to establish chain (needed for avg mining time calculation)
-    candidateGenerator.tell(GenerateCandidate(Seq.empty, reply = true, forced = false), testProbe.ref)
-    val initCandidate = testProbe.expectMsgPF(candidateGenDelay) {
-      case StatusReply.Success(c: Candidate) => c
-    }
-    val initBlock = powScheme
-      .proveCandidate(initCandidate.candidateBlock, defaultMinerSecret.w, 0, 1000)
-      .get
-    candidateGenerator.tell(initBlock.header.powSolution, testProbe.ref)
-    testProbe.fishForMessage(blockValidationDelay) {
-      case StatusReply.Success(()) => true
-      case FullBlockApplied(header) if header.id != initBlock.header.parentId => true
-      case _ => false
-    }
+    try {
+      // Get first candidate from the coherent, already applied block snapshot.
+      candidateGenerator.tell(GenerateCandidate(Seq.empty, reply = true, forced = false), testProbe.ref)
+      val candidate1 = testProbe.expectMsgPF(candidateGenDelay) {
+        case StatusReply.Success(c: Candidate) => c
+      }
+      candidate1.candidateBlock.parentOpt.map(_.id) shouldBe Some(block.header.id)
 
-    // Get first candidate after chain is established
-    candidateGenerator.tell(GenerateCandidate(Seq.empty, reply = true, forced = false), testProbe.ref)
-    val candidate1 = testProbe.expectMsgPF(candidateGenDelay) {
-      case StatusReply.Success(c: Candidate) => c
-    }
+      // Request with forced = false should return cached candidate immediately.
+      candidateGenerator.tell(GenerateCandidate(Seq.empty, reply = true, forced = false), testProbe.ref)
+      val candidate2 = testProbe.expectMsgPF(100.millis) {
+        case StatusReply.Success(c: Candidate) => c
+      }
+      candidate2 should be theSameInstanceAs candidate1
+      candidate2.candidateBlock.timestamp shouldBe candidate1.candidateBlock.timestamp
 
-    // Request with forced = false should return cached candidate immediately
-    candidateGenerator.tell(GenerateCandidate(Seq.empty, reply = true, forced = false), testProbe.ref)
-    val candidate2 = testProbe.expectMsgPF(100.millis) {
-      case StatusReply.Success(c: Candidate) => c
-    }
-    // Should be the exact same cached candidate
-    candidate2.candidateBlock.timestamp shouldBe candidate1.candidateBlock.timestamp
+      // Request with forced = true should bypass cache and regenerate.
+      candidateGenerator.tell(GenerateCandidate(Seq.empty, reply = true, forced = true), testProbe.ref)
+      val candidate3 = testProbe.expectMsgPF(candidateGenDelay) {
+        case StatusReply.Success(c: Candidate) => c
+      }
 
-    // Request with forced = true should bypass cache and regenerate
-    candidateGenerator.tell(GenerateCandidate(Seq.empty, reply = true, forced = true), testProbe.ref)
-    val candidate3 = testProbe.fishForMessage(candidateGenDelay) {
-      case StatusReply.Success(_: Candidate) => true
-      case _: FullBlockApplied => false
-    } match {
-      case StatusReply.Success(c: Candidate) => c
+      // Identity proves regeneration even when the clock has not advanced.
+      candidate3 should not be theSameInstanceAs(candidate1)
+      candidate3.candidateBlock.parentOpt.map(_.id) shouldBe Some(block.header.id)
+      candidate3.candidateBlock.timestamp should be >= candidate1.candidateBlock.timestamp
+    } finally {
+      TestKit.shutdownActorSystem(system)
+      history.closeStorage()
+      state.closeStorage()
     }
-
-    // candidate3 should have timestamp >= candidate1 (regenerated, possibly same or newer)
-    candidate3.candidateBlock.timestamp should be >= candidate1.candidateBlock.timestamp
-
-    system.terminate()
   }
 
   it should "preserve previous candidate when forced regeneration occurs" in new TestKit(ActorSystem()) {