companion: wait for relay sync progress; no 30s hard deadline

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),

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.
+ * <p>
+ * 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<Long, Point> 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<String> 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<Long, Point> 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))));
+        }
+    }
+}

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<Long, Point> 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