diff --git a/sync_tests/scripts/heartbeat.sh b/sync_tests/scripts/heartbeat.sh index af091a88..c4787be8 100755 --- a/sync_tests/scripts/heartbeat.sh +++ b/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 diff --git a/sync_tests/utils/db_sync/__init__.py b/sync_tests/utils/db_sync/__init__.py index 72bf0775..93b50114 100755 --- a/sync_tests/utils/db_sync/__init__.py +++ b/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 diff --git a/sync_tests/utils/helpers.py b/sync_tests/utils/helpers.py index 2e60e7d2..7cf38fa4 100644 --- a/sync_tests/utils/helpers.py +++ b/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: diff --git a/sync_tests/utils/node/__init__.py b/sync_tests/utils/node/__init__.py index 7e0d011f..0d7f2503 100644 --- a/sync_tests/utils/node/__init__.py +++ b/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) diff --git a/sync_tests/utils/sync_entries.py b/sync_tests/utils/sync_entries.py index afbbd5ec..d3a20aab 100644 --- a/sync_tests/utils/sync_entries.py +++ b/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)