fix: harden sync-progress file writes

A second independent review found two more real gaps and one design
issue in the sync-progress heartbeat work.

heartbeat.sh dropped the whole progress line whenever sync_progress
was null, even though era/epoch/slot were known and useful on their
own. cardano-cli can omit syncProgress; the Python side already has
a fallback for this case. The heartbeat now prints era/epoch/slot
with "syncProgress unavailable" instead of nothing at all.

jq's // operator only substitutes for null or false, not an empty
string. node.py defaults a missing era to "", not null, so the "?"
fallback for era never actually fired. Fixed by checking for an
empty string explicitly.

upsert_json_key duplicated an existing, unused helper,
update_json_file, which already did the same read-merge-write. Since
update_json_file had zero callers in the codebase, extended it with
the same missing/corrupt-file tolerance and atomic write instead of
keeping a second near-identical function, and pointed both progress
call sites at it.

Also adds a missing Args section to write_progress_file's docstring,
to match its sibling wait_for_shelley_era.

Author: OlufemiAdeOlusile

sync_tests/scripts/heartbeat.sh

@@ -34,15 +34,22 @@ _progress_line() {
   if [ ! -s "$file" ] || ! command -v jq >/dev/null 2>&1; then
     return
   fi
+  local present
+  present="$(jq -r --arg k "$label" 'has($k)' "$file" 2>/dev/null)"
+  [ "$present" != "true" ] && return
   local era epoch slot pct updated
-  pct="$(jq -r --arg k "$label" '.[$k].sync_progress // empty' "$file" 2>/dev/null)"
-  [ -z "$pct" ] && return
-  era="$(jq -r --arg k "$label" '.[$k].era // "?"' "$file" 2>/dev/null)"
+  era="$(jq -r --arg k "$label" '(.[$k].era // "") | if . == "" then "?" else . end' "$file" 2>/dev/null)"
   epoch="$(jq -r --arg k "$label" '.[$k].epoch // "?"' "$file" 2>/dev/null)"
   slot="$(jq -r --arg k "$label" '.[$k].slot // "?"' "$file" 2>/dev/null)"
   updated="$(jq -r --arg k "$label" '.[$k].updated_at // "?"' "$file" 2>/dev/null)"
-  printf 'progress[%s]: %s%% synced - era=%s epoch=%s slot=%s (as of %s)\n' \
-    "$label" "$pct" "$era" "$epoch" "$slot" "$updated"
+  pct="$(jq -r --arg k "$label" '.[$k].sync_progress // empty' "$file" 2>/dev/null)"
+  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() {

sync_tests/utils/db_sync/__init__.py

@@ -317,18 +317,19 @@ def _log_sync_progress(config: DbSyncConfig, env: str, start_sync: float) -> flo
         db_sync_tip.slot_no,
     )
     try:
-        helpers.upsert_json_key(
+        helpers.update_json_file(
             config.workdir / f"sync_progress_{env}.json",
-            "dbsync",
             {
-                "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,
-                "updated_at": datetime.datetime.now(tz=datetime.timezone.utc).strftime(
-                    "%Y-%m-%dT%H:%M:%SZ"
-                ),
+                "dbsync": {
+                    "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,
+                    "updated_at": datetime.datetime.now(tz=datetime.timezone.utc).strftime(
+                        "%Y-%m-%dT%H:%M:%SZ"
+                    ),
+                }
             },
         )
     except OSError:

sync_tests/utils/helpers.py

@@ -257,15 +257,29 @@ 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 this is safe to use for status files with more
+    than one writer, that 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 (temp file + rename).
+    """
+    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
+        except (json.JSONDecodeError, OSError):
+            LOGGER.warning(
+                "Ignoring unreadable status file %s, starting fresh", file_path, exc_info=True
+            )
     data.update(updates)
-
-    with open(file_path, "w") as json_file:
-        json.dump(data, json_file, indent=2)
+    write_json_to_file(file_path, data)
 
 
 def remove_json_keys(file_path: pl.Path, keys: list[str]) -> None:
@@ -368,39 +382,6 @@ def write_json_to_file(file_path: str | pl.Path, data: dict | list) -> None:
     os.replace(tmp_path, file_path)
 
 
-def upsert_json_key(file_path: str | pl.Path, key: str, value: dict) -> None:
-    """Update a single top-level key in a shared JSON status file.
-
-    Reads the existing file (if any), sets ``data[key] = value``, and writes
-    the whole file back, so unrelated keys written by other callers are
-    preserved. Same read-merge-write pattern as ``conftest.py``'s
-    ``_write_marker_to_status``, for status files with more than one writer.
-
-    An existing file that is missing, unreadable, or not valid JSON is
-    treated as empty rather than raised, since this is used for
-    observability status files that must never abort the caller.
-
-    Args:
-        file_path: Path to the shared JSON status file.
-        key: Top-level key to upsert.
-        value: Value to store under ``key``.
-    """
-    file_path = pl.Path(file_path)
-    data: dict = {}
-    if file_path.exists():
-        try:
-            with open(file_path) as fh:
-                loaded = json.load(fh)
-            if isinstance(loaded, dict):
-                data = loaded
-        except (json.JSONDecodeError, OSError):
-            LOGGER.warning(
-                "Ignoring unreadable status file %s, starting fresh", file_path, exc_info=True
-            )
-    data[key] = value
-    write_json_to_file(file_path, data)
-
-
 def manage_directory(dir_name: str, action: str, root: str = ".") -> str | None:
     """Manage a directory by creating or removing it based on the action specified.
 

sync_tests/utils/node/__init__.py

@@ -451,22 +451,29 @@ def write_progress_file(workdir: pl.Path | None, env: str, tip: "Tip") -> None:
 
     Never raises: this is observability only, and must not abort a
     multi-hour sync run over a disk-full or permission error.
+
+    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``.
     """
     if workdir is None:
         return
     try:
-        helpers.upsert_json_key(
+        helpers.update_json_file(
             workdir / f"sync_progress_{env}.json",
-            "node",
             {
-                "era": tip.era,
-                "epoch": tip.epoch,
-                "block": tip.block,
-                "slot": tip.slot,
-                "sync_progress": tip.sync_progress,
-                "updated_at": datetime.datetime.now(tz=datetime.timezone.utc).strftime(
-                    "%Y-%m-%dT%H:%M:%SZ"
-                ),
+                "node": {
+                    "era": tip.era,
+                    "epoch": tip.epoch,
+                    "block": tip.block,
+                    "slot": tip.slot,
+                    "sync_progress": tip.sync_progress,
+                    "updated_at": datetime.datetime.now(tz=datetime.timezone.utc).strftime(
+                        "%Y-%m-%dT%H:%M:%SZ"
+                    ),
+                }
             },
         )
     except OSError: