feat(ci): show live sync progress in heartbeat

sync_tests/scripts/heartbeat.sh

@@ -17,6 +17,44 @@ INTERVAL="${SYNC_TESTS_HEARTBEAT_INTERVAL_SEC:-600}"
 TAIL_LINES="${SYNC_TESTS_HEARTBEAT_TAIL_LINES:-30}"
 HALF_TAIL=$(( TAIL_LINES / 2 ))
 
+_newest_file() {
+  local newest=""
+  local f
+  for f in "$@"; do
+    if [ -z "$newest" ] || [ "$f" -nt "$newest" ]; then
+      newest="$f"
+    fi
+  done
+  echo "$newest"
+}
+
+_progress_line() {
+  local label="$1"
+  local file="$2"
+  if [ ! -s "$file" ] || ! command -v jq >/dev/null 2>&1; then
+    return
+  fi
+  # One jq call per label: era, epoch, slot, updated_at, sync_progress as TSV.
+  # Empty output means the label is absent (or the file is unparsable).
+  local row
+  row="$(jq -r --arg k "$label" '
+    def dflt: if (. // "") == "" then "?" else . end;
+    if has($k) then
+      [ (.[$k].era | dflt), (.[$k].epoch | dflt), (.[$k].slot | dflt),
+        (.[$k].updated_at | dflt), (.[$k].sync_progress // "") ] | @tsv
+    else empty end' "$file" 2>/dev/null)"
+  [ -z "$row" ] && return
+  local era epoch slot updated pct
+  IFS=$'\t' read -r era epoch slot updated pct <<< "$row"
+  if [ -n "$pct" ]; then
+    printf 'progress[%s]: %s%% synced - era=%s epoch=%s slot=%s (as of %s)\n' \
+      "$label" "$pct" "$era" "$epoch" "$slot" "$updated"
+  else
+    printf 'progress[%s]: syncProgress unavailable - era=%s epoch=%s slot=%s (as of %s)\n' \
+      "$label" "$era" "$epoch" "$slot" "$updated"
+  fi
+}
+
 _tail_group() {
   local label="$1"
   local file="$2"
@@ -68,6 +106,7 @@ _heartbeat_tick() {
 
   shopt -s nullglob
   local markers_files=( "$WORKDIR"/sync_markers_*.json )
+  local progress_files=( "$WORKDIR"/sync_progress_*.json )
   shopt -u nullglob
   if [ "${#markers_files[@]}" -gt 0 ]; then
     grep -q SYNC_MARKER_NODE_DONE   "${markers_files[@]}" 2>/dev/null && node_done=1
@@ -112,6 +151,12 @@ _heartbeat_tick() {
 
   echo "::group::Heartbeat [${phase}] $(date -u +%FT%TZ)"
   echo "status: ${status}"
+  if [ "${#progress_files[@]}" -gt 0 ]; then
+    local progress_file
+    progress_file="$(_newest_file "${progress_files[@]}")"
+    _progress_line "node" "$progress_file"
+    _progress_line "dbsync" "$progress_file"
+  fi
   df -h "${GITHUB_WORKSPACE:-.}" 2>/dev/null || df -h . 2>/dev/null || echo "[df unavailable]"
 
   if [ "$MODE" = "node-only" ]; then

sync_tests/utils/db_sync/__init__.py

@@ -291,6 +291,7 @@ def _log_sync_progress(config: DbSyncConfig, env: str, start_sync: float) -> flo
             tip.slot,
             tip.era,
         )
+        node.write_progress_file(workdir=config.workdir, env=env, tip=tip)
     try:
         db_sync_tip = postgres.get_db_sync_tip(config)
     except Exception:
@@ -314,6 +315,18 @@ def _log_sync_progress(config: DbSyncConfig, env: str, start_sync: float) -> flo
         db_sync_tip.block_no,
         db_sync_tip.slot_no,
     )
+    helpers.write_sync_progress(
+        workdir=config.workdir,
+        env=env,
+        key="dbsync",
+        payload={
+            "epoch": db_sync_tip.epoch_no,
+            "block": db_sync_tip.block_no,
+            "slot": db_sync_tip.slot_no,
+            "sync_progress": db_sync_progress,
+            "sync_time_h_m_s": sync_time_h_m_s,
+        },
+    )
     helpers.print_last_n_lines(config.db_sync_log_file, 5)
     return db_sync_progress
 

sync_tests/utils/helpers.py

@@ -4,6 +4,7 @@
 
 import argparse
 import collections
+import datetime
 import hashlib
 import json
 import logging
@@ -14,6 +15,7 @@
 import shutil
 import stat
 import subprocess
+import tempfile
 import time
 import typing as tp
 import zipfile
@@ -257,15 +259,67 @@ def execute_command(
         raise
 
 
-def update_json_file(file_path: pl.Path, updates: dict) -> None:
-    """Read a JSON file, updates it with the provided dictionary, and writes it back."""
-    with open(file_path) as json_file:
-        data = json.load(json_file)
+def update_json_file(file_path: str | pl.Path, updates: dict) -> None:
+    """Read a JSON file, merge it with ``updates``, and write it back.
 
+    A missing, unreadable, or invalid existing file is treated as empty
+    rather than raised, so status files must never abort a caller just
+    because a previous write was interrupted. The write itself goes through
+    ``write_json_to_file``, so it is atomic and safe against a concurrent
+    *reader* (e.g. the CI heartbeat script). The read-modify-write as a whole
+    is not atomic, so two *writers* racing on the same file can still lose an
+    update; all current callers write from a single process.
+    """
+    file_path = pl.Path(file_path)
+    data: dict = {}
+    if file_path.exists():
+        try:
+            with open(file_path) as json_file:
+                loaded = json.load(json_file)
+            if isinstance(loaded, dict):
+                data = loaded
+        # ValueError covers both JSONDecodeError and UnicodeDecodeError, i.e. a
+        # truncated or binary-garbage file, which must not abort the caller.
+        except (ValueError, OSError):
+            LOGGER.warning(
+                "Ignoring unreadable status file %s, starting fresh", file_path, exc_info=True
+            )
     data.update(updates)
+    write_json_to_file(file_path, data)
 
-    with open(file_path, "w") as json_file:
-        json.dump(data, json_file, indent=2)
+
+def write_sync_progress(workdir: pl.Path | None, env: str, key: str, payload: dict) -> None:
+    """Upsert one top-level key of the shared ``sync_progress_{env}.json`` status file.
+
+    The file is read by the CI heartbeat script and is independent of pytest's
+    own logging (level, capture), so CI verbosity settings can never silently
+    hide sync progress from the heartbeat. The node side writes the ``"node"``
+    key, the db-sync side the ``"dbsync"`` key.
+
+    An ``updated_at`` UTC timestamp is added unless the payload already has one,
+    so a stale entry is always recognizable as stale.
+
+    Never raises: this is observability only, and must not abort a multi-hour
+    sync run over a disk-full, permission, or serialization error.
+
+    Args:
+        workdir: Directory holding the progress file. When ``None``, writing is
+            skipped.
+        env: Environment name (preview, preprod, mainnet).
+        key: Top-level key to upsert ("node" or "dbsync").
+        payload: Values to store under ``key``.
+    """
+    if workdir is None:
+        return
+    entry = dict(payload)
+    entry.setdefault(
+        "updated_at",
+        datetime.datetime.now(tz=datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
+    )
+    try:
+        update_json_file(pl.Path(workdir) / f"sync_progress_{env}.json", {key: entry})
+    except Exception:
+        LOGGER.warning("Failed to write %s sync progress file for CI heartbeat", key, exc_info=True)
 
 
 def remove_json_keys(file_path: pl.Path, keys: list[str]) -> None:
@@ -352,14 +406,33 @@ def make_tarfile(output_filename: str, source_dir: str) -> None:
 def write_json_to_file(file_path: str | pl.Path, data: dict | list) -> None:
     """Write data to a file in JSON format.
 
+    Writes to a uniquely named temp file in the same directory first, flushes
+    it to disk, then renames it into place. A concurrent reader (e.g. the CI
+    heartbeat script) or a process killed mid-write can then never observe a
+    truncated or empty file, and two writers of the same path cannot interleave
+    into a shared temp file.
+
     Args:
         file_path: Path to the output JSON file.
         data: Dictionary or list to serialize as JSON.
     """
     file_path = pl.Path(file_path)
     file_path.parent.mkdir(parents=True, exist_ok=True)
-    with open(file_path, "w") as f:
-        json.dump(data, f, indent=2)
+    fd, tmp_name = tempfile.mkstemp(
+        dir=file_path.parent, prefix=f"{file_path.name}.", suffix=".tmp"
+    )
+    tmp_path = pl.Path(tmp_name)
+    try:
+        with os.fdopen(fd, "w") as f:
+            json.dump(data, f, indent=2)
+            f.flush()
+            os.fsync(f.fileno())
+        # mkstemp creates 0600; keep the readable perms a plain open() would give.
+        tmp_path.chmod(0o644)
+        os.replace(tmp_path, file_path)
+    except BaseException:
+        tmp_path.unlink(missing_ok=True)
+        raise
 
 
 def manage_directory(dir_name: str, action: str, root: str = ".") -> str | None:

sync_tests/utils/node/__init__.py

@@ -441,6 +441,32 @@ def get_testnet_args(env: str) -> tp.Iterable[str]:
         raise exceptions.SyncError(msg) from e
 
 
+def write_progress_file(workdir: pl.Path | None, env: str, tip: "Tip") -> None:
+    """Record the node's current sync position for CI heartbeats.
+
+    Thin ``Tip`` adapter over ``helpers.write_sync_progress``; see there for the
+    file format and the never-raises guarantee.
+
+    Args:
+        workdir: Directory to write the CI heartbeat progress file to. When
+            ``None``, progress-file writing is skipped.
+        env: Environment name (preview, preprod, mainnet).
+        tip: Current node tip, as returned by ``get_current_tip``.
+    """
+    helpers.write_sync_progress(
+        workdir=workdir,
+        env=env,
+        key="node",
+        payload={
+            "era": tip.era,
+            "epoch": tip.epoch,
+            "block": tip.block,
+            "slot": tip.slot,
+            "sync_progress": tip.sync_progress,
+        },
+    )
+
+
 def get_current_tip(env: str) -> Tip:
     """Retrieve the current tip of the Cardano node."""
     cardano_cli_path = os.environ.get("CARDANO_CLI_PATH") or "cardano-cli"
@@ -766,6 +792,7 @@ def wait_for_shelley_era(
     timeout_minutes: int = 60,
     min_era: str = "shelley",
     logfile_path: pl.Path | None = None,
+    workdir: pl.Path | None = None,
 ) -> None:
     """Wait for the node to reach at least a target era before starting db-sync.
 
@@ -780,6 +807,8 @@ def wait_for_shelley_era(
         logfile_path: Node stdout/stderr log file (same as ``start_node``). When
             ``None``, defaults to ``base_dir / NODE_LOG_FILE_NAME`` for backward
             compatibility.
+        workdir: Directory to write the CI heartbeat progress file to. When
+            ``None``, progress-file writing is skipped.
 
     Raises:
         exceptions.SyncError: If the target era is not reached within timeout.
@@ -818,6 +847,7 @@ def wait_for_shelley_era(
                 f"elapsed: {elapsed_minutes} minutes, "
                 f"node logfile: {logfile_size} bytes"
             )
+            write_progress_file(workdir=workdir, env=env, tip=tip)
 
         # Check if we've reached the target era or later
         current_idx = era_order.get(str(tip.era).lower())
@@ -826,6 +856,7 @@ def wait_for_shelley_era(
                 f"Node reached {tip.era} era at epoch {tip.epoch}, block {tip.block}. "
                 f"Proceeding to start db-sync (min_era={min_era})."
             )
+            write_progress_file(workdir=workdir, env=env, tip=tip)
             return
 
         # Check timeout
@@ -840,7 +871,7 @@ def wait_for_shelley_era(
         count += 1
 
 
-def wait_for_node_to_sync(env: str, base_dir: pl.Path) -> tuple:
+def wait_for_node_to_sync(env: str, base_dir: pl.Path, workdir: pl.Path | None = None) -> tuple:
     """Wait for the Cardano node to start."""
     LOGGER.info("Waiting for the node to sync")
     era_details_dict = {}
@@ -863,6 +894,7 @@ def wait_for_node_to_sync(env: str, base_dir: pl.Path) -> tuple:
                 f" - actual_slot : {tip.slot} "
                 f" - syncProgress: {tip.sync_progress}",
             )
+            write_progress_file(workdir=workdir, env=env, tip=tip)
 
         # Use the same current time for both era and epoch updates.
         current_time_str = datetime.datetime.now(tz=datetime.timezone.utc).strftime(
@@ -890,10 +922,12 @@ def wait_for_node_to_sync(env: str, base_dir: pl.Path) -> tuple:
         # Check termination condition:
         # For nodes reporting sync progress, we wait until progress reaches 100.
         if tip.sync_progress is not None and tip.sync_progress >= 100:
+            write_progress_file(workdir=workdir, env=env, tip=tip)
             break
         # Otherwise (for nodes without sync progress) wait until the slot number passes
         # the calculated value.
         if tip.sync_progress is None and tip.slot > last_slot_no:
+            write_progress_file(workdir=workdir, env=env, tip=tip)
             break
 
         time.sleep(5)

sync_tests/utils/sync_entries.py

@@ -146,7 +146,7 @@ def run_node_sync(
             latest_chunk_no,
             era_details,
             epoch_details,
-        ) = node.wait_for_node_to_sync(env=env, base_dir=base_dir)
+        ) = node.wait_for_node_to_sync(env=env, base_dir=base_dir, workdir=node_logfile_path.parent)
         LOGGER.info(
             "--- Full sync complete: sync_time_sec=%s last_slot_no=%s latest_chunk_no=%s eras=%s",
             sync_time_sec,
@@ -168,6 +168,7 @@ def run_node_sync(
             timeout_minutes=shelley_timeout_minutes,
             min_era=start_era,
             logfile_path=node_logfile_path,
+            workdir=node_logfile_path.parent,
         )
         phase_end = time.perf_counter()
         sync_time_sec = int(phase_end - phase_start)