diff --git a/applications/cli/config/application.properties b/applications/cli/config/application.properties index 83cd5caf..5081b08e 100644 --- a/applications/cli/config/application.properties +++ b/applications/cli/config/application.properties @@ -34,6 +34,13 @@ bp.create.enabled=true ## A Haskell relay rejects a dense backfill at activeSlotsCoeff below 1 with VRFLeaderValueTooBig. #yano.past.time.travel.slot.leader.mode=auto +## Companion mode: how long the handover waits for the Haskell relay to copy Yano's bootstrap chain. +## The wait continues while the relay tip keeps advancing. It gives up when the tip has not moved for +## the stall timeout, or when the optional overall ceiling is reached (0 = no ceiling). +## A long epoch means a long backfill (roughly 550 blocks/s), so a fixed deadline would cut it off. +#yano.relay.sync.stall.timeout.seconds=30 +#yano.relay.sync.max.wait.seconds=0 + ###################################################### #To configure an external database for Yaci Store (Indexer), diff --git a/applications/cli/src/main/java/com/bloxbean/cardano/yacicli/localcluster/yano/RelaySyncWaiter.java b/applications/cli/src/main/java/com/bloxbean/cardano/yacicli/localcluster/yano/RelaySyncWaiter.java new file mode 100644 index 00000000..7fefd49c --- /dev/null +++ b/applications/cli/src/main/java/com/bloxbean/cardano/yacicli/localcluster/yano/RelaySyncWaiter.java @@ -0,0 +1,153 @@ +package com.bloxbean.cardano.yacicli.localcluster.yano; + +import com.bloxbean.cardano.yaci.core.protocol.chainsync.messages.Point; +import com.bloxbean.cardano.yacicli.common.Tuple; + +import java.util.function.Consumer; +import java.util.function.LongSupplier; + +/** + * Waits for the Haskell relay to copy Yano's bootstrap chain during the companion handover. + *

+ * The wait is progress-based. It keeps polling while the relay's tip is still advancing and gives up + * only when the tip has not moved for {@code stallTimeoutMs}, or when the optional overall ceiling + * {@code maxWaitMs} is reached ({@code 0} = no ceiling). A fixed deadline cut the relay off mid-chain + * on long epochs: Yano was stopped, and the producer restarted with a tip already outside its own + * forecast horizon, so the chain was frozen from its first minute. + */ +final class RelaySyncWaiter { + static final long POLL_INTERVAL_MS = 1_000L; + static final long PROGRESS_REPORT_INTERVAL_MS = 5_000L; + static final long DEFAULT_STALL_TIMEOUT_SECONDS = 30; + // Largest value in seconds that still converts to milliseconds without overflowing + private static final long MAX_SECONDS = Long.MAX_VALUE / 1000; + + /** Stall timeout to use for a configured value: a positive value as given, anything else the default. */ + static long stallTimeoutSeconds(long configured) { + return configured > 0 ? Math.min(configured, MAX_SECONDS) : DEFAULT_STALL_TIMEOUT_SECONDS; + } + + /** Overall ceiling to use for a configured value: zero (no ceiling) or a positive value as given, negative is none. */ + static long maxWaitSeconds(long configured) { + return Math.min(Math.max(0, configured), MAX_SECONDS); + } + + /** Reads the relay's tip: block height and point. Returns {@code null} when the relay is not reachable yet. */ + interface TipReader { + Tuple readTip(); + } + + /** Sleeps for the given number of milliseconds. */ + interface Sleeper { + void sleep(long millis) throws InterruptedException; + } + + /** + * Result of one wait. + * + * @param synced true when the relay's tip reached the target epoch + * @param epoch epoch of the last tip seen, or -1 when no tip was ever read + * @param slot slot of the last tip seen, or -1 + * @param height block height of the last tip seen, or -1 + * @param reason why the wait ended when {@code synced} is false; {@code null} otherwise + */ + record Outcome(boolean synced, long epoch, long slot, long height, String reason) { + } + + private final long stallTimeoutMs; + private final long maxWaitMs; + private final LongSupplier clock; + private final Sleeper sleeper; + + RelaySyncWaiter(long stallTimeoutMs, long maxWaitMs) { + this(stallTimeoutMs, maxWaitMs, System::currentTimeMillis, Thread::sleep); + } + + RelaySyncWaiter(long stallTimeoutMs, long maxWaitMs, LongSupplier clock, Sleeper sleeper) { + if (stallTimeoutMs <= 0) { + throw new IllegalArgumentException("stallTimeoutMs must be positive, got " + stallTimeoutMs); + } + if (maxWaitMs < 0) { + throw new IllegalArgumentException("maxWaitMs must be zero (no ceiling) or positive, got " + maxWaitMs); + } + this.stallTimeoutMs = stallTimeoutMs; + this.maxWaitMs = maxWaitMs; + this.clock = clock; + this.sleeper = sleeper; + } + + /** + * Poll the relay's tip until it reaches {@code targetEpoch}, the tip stalls, or the ceiling is hit. + * + * @param targetEpoch epoch the relay must reach (tip slot / epochLength >= targetEpoch) + * @param epochLength slots per epoch; must be positive + * @param tipReader reads the relay's current tip + * @param progress receives a progress line about every {@link #PROGRESS_REPORT_INTERVAL_MS} + */ + Outcome await(long targetEpoch, long epochLength, TipReader tipReader, Consumer progress) + throws InterruptedException { + if (epochLength <= 0) { + return new Outcome(false, -1, -1, -1, "epoch length is not known"); + } + + final long startedAt = clock.getAsLong(); + long lastProgressAt = startedAt; + long lastReportAt = startedAt; + long lastReportHeight = -1; + long slot = -1; + long height = -1; + long epoch = -1; + + while (true) { + Tuple tip = tipReader.readTip(); + long now = clock.getAsLong(); + + if (tip != null && tip._2 != null) { + long tipSlot = tip._2.getSlot(); + long tipHeight = tip._1 != null ? tip._1 : -1; + if (tipSlot > slot || tipHeight > height) { + lastProgressAt = now; + } + slot = tipSlot; + height = tipHeight; + epoch = slot / epochLength; + + if (lastReportHeight < 0 && height >= 0) { + // First tip read: the baseline the first reported rate is measured from + lastReportHeight = height; + lastReportAt = now; + } + + if (epoch >= targetEpoch) { + return new Outcome(true, epoch, slot, height, null); + } + + if (now - lastReportAt >= PROGRESS_REPORT_INTERVAL_MS) { + long elapsedMs = Math.max(1, now - lastReportAt); + long blocksSinceReport = lastReportHeight >= 0 && height >= 0 ? height - lastReportHeight : 0; + long blocksPerSecond = blocksSinceReport * 1000 / elapsedMs; + progress.accept(String.format( + "Relay sync in progress: epoch %d of %d, slot %d, height %d (%d blocks/s)", + epoch, targetEpoch, slot, height, blocksPerSecond)); + lastReportAt = now; + lastReportHeight = height; + } + } + + long sinceProgress = now - lastProgressAt; + if (sinceProgress >= stallTimeoutMs) { + String where = slot >= 0 ? " at slot " + slot + " (epoch " + epoch + ")" : " before any tip was read"; + return new Outcome(false, epoch, slot, height, + "relay tip did not advance for " + (sinceProgress / 1000) + "s" + where); + } + if (maxWaitMs > 0 && now - startedAt >= maxWaitMs) { + return new Outcome(false, epoch, slot, height, + "overall wait ceiling of " + (maxWaitMs / 1000) + "s reached"); + } + + long untilStall = stallTimeoutMs - sinceProgress; + long untilCeiling = maxWaitMs > 0 ? maxWaitMs - (now - startedAt) : Long.MAX_VALUE; + sleeper.sleep(Math.max(1, Math.min(POLL_INTERVAL_MS, Math.min(untilStall, untilCeiling)))); + } + } +} diff --git a/applications/cli/src/main/java/com/bloxbean/cardano/yacicli/localcluster/yano/YanoCompanionService.java b/applications/cli/src/main/java/com/bloxbean/cardano/yacicli/localcluster/yano/YanoCompanionService.java index db5fd92a..4dc5e92f 100644 --- a/applications/cli/src/main/java/com/bloxbean/cardano/yacicli/localcluster/yano/YanoCompanionService.java +++ b/applications/cli/src/main/java/com/bloxbean/cardano/yacicli/localcluster/yano/YanoCompanionService.java @@ -16,6 +16,7 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.ObjectProvider; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import com.bloxbean.cardano.client.crypto.Blake2bUtil; @@ -58,6 +59,14 @@ public class YanoCompanionService { private final ClusterInfoService clusterInfoService; private final ObjectMapper objectMapper = new ObjectMapper(); + // Relay sync wait during handover. Stall timeout: give up when the relay tip stops advancing for this + // long. Max wait: optional overall ceiling, 0 = none. Both in seconds. + @Value("${yano.relay.sync.stall.timeout.seconds:30}") + private long relaySyncStallTimeoutSeconds = 30; + + @Value("${yano.relay.sync.max.wait.seconds:0}") + private long relaySyncMaxWaitSeconds = 0; + /** * Run the full companion bootstrap sequence before the Haskell node starts. * Returns true if bootstrap succeeded and Haskell node should proceed. @@ -253,43 +262,45 @@ public void performHandover(ClusterInfo clusterInfo, Path clusterFolder, Consume Thread.sleep(1000); } - // Wait for Haskell relay to sync past Yano's shifted bootstrap region (~3 epochs). - // Poll the relay's tip every ~1s and exit early once epoch >= BOOTSTRAP_EPOCH_SHIFT. - // Soft 30s deadline — getTip itself can block up to ~10s on a stuck socket, so - // wall-clock worst case is ~40s; on healthy runs this exits in ~10-20s. + // Wait for the Haskell relay to sync past Yano's shifted bootstrap region (~3 epochs). + // The wait is progress-based: it continues while the relay tip keeps advancing and gives up + // only when the tip stalls for yano.relay.sync.stall.timeout.seconds (default 30s) or the + // optional ceiling yano.relay.sync.max.wait.seconds is reached. A long epoch means a long + // backfill (about 550 blocks/s measured), so a fixed deadline would cut the relay off mid-chain. long epochLength = clusterInfo.getEpochLength(); - writer.accept(info("Waiting up to 30s for relay to sync past epoch %d...", BOOTSTRAP_EPOCH_SHIFT)); - final long deadlineMs = System.currentTimeMillis() + 30_000L; + // An unusable value must not throw here: the handover would then skip stopping Yano and restoring the + // topology, and the Haskell producer would restart next to a Yano that is still producing. + long stallTimeoutSeconds = RelaySyncWaiter.stallTimeoutSeconds(relaySyncStallTimeoutSeconds); + if (stallTimeoutSeconds != relaySyncStallTimeoutSeconds) + writer.accept(warn("yano.relay.sync.stall.timeout.seconds must be positive, got %d. Using %d.", + relaySyncStallTimeoutSeconds, stallTimeoutSeconds)); + long maxWaitSeconds = RelaySyncWaiter.maxWaitSeconds(relaySyncMaxWaitSeconds); + if (maxWaitSeconds != relaySyncMaxWaitSeconds) + writer.accept(warn("yano.relay.sync.max.wait.seconds must be 0 (no ceiling) or positive, got %d. Using %d.", + relaySyncMaxWaitSeconds, maxWaitSeconds)); + + String ceiling = maxWaitSeconds > 0 ? ", at most " + maxWaitSeconds + "s in total" : ""; + writer.accept(info("Waiting for relay to sync past epoch %d (giving up if the tip stalls for %ds%s)...", + BOOTSTRAP_EPOCH_SHIFT, stallTimeoutSeconds, ceiling)); + final ClusterUtilService clusterUtilService = clusterUtilServiceProvider.getObject(); - boolean synced = false; - - while (true) { - long remaining = deadlineMs - System.currentTimeMillis(); - if (remaining <= 0) break; - - // ClusterUtilService.getTip prints "Find tip error ..." directly via static - // ConsoleWriter on exception (bypassing the consumer); transient noise during - // the first one or two polls after the socket appears is expected. - Tuple tip = clusterUtilService.getTip(msg -> {}); - if (tip != null && tip._2 != null && epochLength > 0) { - long slot = tip._2.getSlot(); - long epoch = slot / epochLength; - if (epoch >= BOOTSTRAP_EPOCH_SHIFT) { - writer.accept(success("Relay synced to epoch " + epoch - + " (slot " + slot + ", height " + tip._1 + ")")); - synced = true; - break; - } - } - - remaining = deadlineMs - System.currentTimeMillis(); - if (remaining <= 0) break; - Thread.sleep(Math.min(1000L, remaining)); - } - if (!synced) { - writer.accept(warn("Relay sync did not reach epoch " - + BOOTSTRAP_EPOCH_SHIFT + " within 30s. Proceeding anyway.")); + // ClusterUtilService.getTip prints "Find tip error ..." directly via static + // ConsoleWriter on exception (bypassing the consumer); transient noise during + // the first one or two polls after the socket appears is expected. + RelaySyncWaiter waiter = new RelaySyncWaiter(stallTimeoutSeconds * 1000L, maxWaitSeconds * 1000L); + RelaySyncWaiter.Outcome outcome = waiter.await(BOOTSTRAP_EPOCH_SHIFT, epochLength, + () -> clusterUtilService.getTip(msg -> {}), + msg -> writer.accept(info(msg))); + + if (outcome.synced()) { + writer.accept(success("Relay synced to epoch " + outcome.epoch() + + " (slot " + outcome.slot() + ", height " + outcome.height() + ")")); + } else { + writer.accept(warn("Relay sync did not reach epoch " + BOOTSTRAP_EPOCH_SHIFT + + ": " + outcome.reason() + ". Proceeding anyway.")); + writer.accept(warn("If the Haskell node forges no blocks after the restart, its tip is outside " + + "its forecast horizon; recreate the devnet, or raise yano.relay.sync.stall.timeout.seconds.")); } // Stop Yano — relay has synced the chain to its db diff --git a/applications/cli/src/test/java/com/bloxbean/cardano/yacicli/localcluster/yano/RelaySyncWaiterTest.java b/applications/cli/src/test/java/com/bloxbean/cardano/yacicli/localcluster/yano/RelaySyncWaiterTest.java new file mode 100644 index 00000000..71ec9122 --- /dev/null +++ b/applications/cli/src/test/java/com/bloxbean/cardano/yacicli/localcluster/yano/RelaySyncWaiterTest.java @@ -0,0 +1,185 @@ +package com.bloxbean.cardano.yacicli.localcluster.yano; + +import com.bloxbean.cardano.yaci.core.protocol.chainsync.messages.Point; +import com.bloxbean.cardano.yacicli.common.Tuple; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +class RelaySyncWaiterTest { + private static final long EPOCH_LENGTH = 600; + private static final long TARGET_EPOCH = 3; + + /** Fake clock: sleeping advances time, and each tip read costs a little time too. */ + private static final class FakeTime { + long now = 1_000_000L; + final List sleeps = new ArrayList<>(); + + long read() { + return now; + } + + void sleep(long millis) { + sleeps.add(millis); + now += millis; + } + } + + private static Tuple tip(long height, long slot) { + return new Tuple<>(height, new Point(slot, "00")); + } + + /** Replays scripted tips in order; the last one repeats forever. */ + private static RelaySyncWaiter.TipReader script(List> tips) { + Iterator> it = tips.iterator(); + Tuple[] last = new Tuple[1]; + return () -> { + if (it.hasNext()) { + last[0] = it.next(); + } + return last[0]; + }; + } + + @Test + void returnsSyncedAsSoonAsTipReachesTargetEpoch() throws Exception { + FakeTime time = new FakeTime(); + RelaySyncWaiter waiter = new RelaySyncWaiter(30_000, 0, time::read, time::sleep); + List progress = new ArrayList<>(); + + RelaySyncWaiter.Outcome outcome = waiter.await(TARGET_EPOCH, EPOCH_LENGTH, + script(List.of(tip(10, 10), tip(700, 700), tip(1900, 1900))), progress::add); + + assertThat(outcome.synced()).isTrue(); + assertThat(outcome.epoch()).isEqualTo(3); + assertThat(outcome.slot()).isEqualTo(1900); + assertThat(outcome.height()).isEqualTo(1900); + assertThat(outcome.reason()).isNull(); + assertThat(time.sleeps).hasSize(2); + } + + @Test + void keepsWaitingBeyondStallTimeoutWhileTipAdvances() throws Exception { + FakeTime time = new FakeTime(); + // 30s stall timeout, but the relay needs 120 polls (2 minutes) to reach the target. + RelaySyncWaiter waiter = new RelaySyncWaiter(30_000, 0, time::read, time::sleep); + List> tips = new ArrayList<>(); + for (long i = 1; i <= 120; i++) { + tips.add(tip(i * 15, i * 15)); + } + List progress = new ArrayList<>(); + + RelaySyncWaiter.Outcome outcome = waiter.await(TARGET_EPOCH, EPOCH_LENGTH, script(tips), progress::add); + + assertThat(outcome.synced()).isTrue(); + assertThat(outcome.slot()).isEqualTo(1800); + assertThat(time.now - 1_000_000L).isGreaterThan(100_000L); + assertThat(progress).isNotEmpty(); + assertThat(progress.get(0)).startsWith("Relay sync in progress: epoch 0 of 3"); + assertThat(progress.get(0)).contains("blocks/s"); + } + + @Test + void givesUpWhenTipStopsAdvancing() throws Exception { + FakeTime time = new FakeTime(); + RelaySyncWaiter waiter = new RelaySyncWaiter(30_000, 0, time::read, time::sleep); + + RelaySyncWaiter.Outcome outcome = waiter.await(TARGET_EPOCH, EPOCH_LENGTH, + script(List.of(tip(100, 100), tip(200, 200))), msg -> {}); + + assertThat(outcome.synced()).isFalse(); + assertThat(outcome.slot()).isEqualTo(200); + assertThat(outcome.reason()).contains("did not advance for 30s").contains("slot 200"); + // one poll with progress, then 30s of stalled polls + assertThat(time.now - 1_000_000L).isEqualTo(31_000L); + } + + @Test + void givesUpWhenNoTipIsEverRead() throws Exception { + FakeTime time = new FakeTime(); + RelaySyncWaiter waiter = new RelaySyncWaiter(5_000, 0, time::read, time::sleep); + + RelaySyncWaiter.Outcome outcome = waiter.await(TARGET_EPOCH, EPOCH_LENGTH, () -> null, msg -> {}); + + assertThat(outcome.synced()).isFalse(); + assertThat(outcome.slot()).isEqualTo(-1); + assertThat(outcome.reason()).contains("before any tip was read"); + } + + @Test + void ceilingEndsTheWaitEvenWhileTipAdvances() throws Exception { + FakeTime time = new FakeTime(); + RelaySyncWaiter waiter = new RelaySyncWaiter(30_000, 10_000, time::read, time::sleep); + List> tips = new ArrayList<>(); + for (long i = 1; i <= 100; i++) { + tips.add(tip(i, i)); + } + + RelaySyncWaiter.Outcome outcome = waiter.await(TARGET_EPOCH, EPOCH_LENGTH, script(tips), msg -> {}); + + assertThat(outcome.synced()).isFalse(); + assertThat(outcome.reason()).contains("ceiling of 10s"); + assertThat(time.now - 1_000_000L).isEqualTo(10_000L); + } + + @Test + void unknownEpochLengthCannotBeWaitedFor() throws Exception { + FakeTime time = new FakeTime(); + RelaySyncWaiter waiter = new RelaySyncWaiter(30_000, 0, time::read, time::sleep); + + RelaySyncWaiter.Outcome outcome = waiter.await(TARGET_EPOCH, 0, () -> tip(1, 1), msg -> {}); + + assertThat(outcome.synced()).isFalse(); + assertThat(outcome.reason()).contains("epoch length"); + assertThat(time.sleeps).isEmpty(); + } + + @Test + void firstProgressReportMeasuresTheRateFromTheFirstTip() throws Exception { + FakeTime time = new FakeTime(); + RelaySyncWaiter waiter = new RelaySyncWaiter(30_000, 0, time::read, time::sleep); + // One poll per second, 100 blocks further each time + List> tips = new ArrayList<>(); + for (long i = 1; i <= 20; i++) { + tips.add(tip(i * 100, i * 100)); + } + List progress = new ArrayList<>(); + + waiter.await(TARGET_EPOCH, EPOCH_LENGTH, script(tips), progress::add); + + assertThat(progress).isNotEmpty(); + assertThat(progress).allSatisfy(line -> assertThat(line).endsWith("(100 blocks/s)")); + } + + @Test + void unusableStallTimeoutFallsBackToTheDefault() { + assertThat(RelaySyncWaiter.stallTimeoutSeconds(45)).isEqualTo(45); + assertThat(RelaySyncWaiter.stallTimeoutSeconds(0)).isEqualTo(RelaySyncWaiter.DEFAULT_STALL_TIMEOUT_SECONDS); + assertThat(RelaySyncWaiter.stallTimeoutSeconds(-5)).isEqualTo(RelaySyncWaiter.DEFAULT_STALL_TIMEOUT_SECONDS); + // Converts to milliseconds without overflowing into a negative timeout + assertThat(RelaySyncWaiter.stallTimeoutSeconds(Long.MAX_VALUE) * 1000L).isPositive(); + } + + @Test + void negativeMaxWaitMeansNoCeiling() { + assertThat(RelaySyncWaiter.maxWaitSeconds(600)).isEqualTo(600); + assertThat(RelaySyncWaiter.maxWaitSeconds(0)).isZero(); + assertThat(RelaySyncWaiter.maxWaitSeconds(-1)).isZero(); + assertThat(RelaySyncWaiter.maxWaitSeconds(Long.MAX_VALUE) * 1000L).isPositive(); + } + + @Test + void sanitizedSettingsAlwaysBuildAWaiter() { + // The handover builds the waiter from sanitized values, so a bad setting can no longer throw there + for (long stall : new long[]{-1, 0, 1, 30, Long.MAX_VALUE}) { + for (long max : new long[]{-1, 0, 60, Long.MAX_VALUE}) { + new RelaySyncWaiter(RelaySyncWaiter.stallTimeoutSeconds(stall) * 1000L, + RelaySyncWaiter.maxWaitSeconds(max) * 1000L); + } + } + } +} diff --git a/docs/pages/yano-node-modes.mdx b/docs/pages/yano-node-modes.mdx index 84407631..84863488 100644 --- a/docs/pages/yano-node-modes.mdx +++ b/docs/pages/yano-node-modes.mdx @@ -56,6 +56,25 @@ yano.past.time.travel.slot.leader.mode=auto The node configuration sets `TraceChainSyncClient` to `true` by default, so a rejected bootstrap chain shows up in the node log instead of failing silently. Set `traceChainSyncClient=false` in `node.properties` to turn it off. +### Relay sync wait during handover + +After Yano catches up to wall-clock time, the Haskell node starts as a relay and copies Yano's chain before it +takes over as block producer. The handover waits while the relay's tip keeps advancing. It gives up when the tip +has not moved for the stall timeout, or when the optional overall ceiling is reached. The wait is not a fixed +deadline: a long epoch means a long bootstrap chain (about 550 blocks per second are copied), and the producer +must own the whole chain before Yano stops. + +The two settings live in `config/application.properties` (the CLI's own properties, not `node.properties`): + +```properties +# Give up when the relay tip stops advancing for this long. Default 30. +yano.relay.sync.stall.timeout.seconds=30 +# Optional overall ceiling for the wait. Default 0 = no ceiling. +yano.relay.sync.max.wait.seconds=0 +``` + +Progress is printed about every five seconds while the relay is syncing. + ## Yano-only Mode In `yano-only` mode, Yano runs as the devnet node. There is no Haskell node handover.