feat(scheduler): heartbeat-aware health check for scheduled sync

README.md

@@ -46,6 +46,15 @@ docker compose run --rm scheduler smt-scheduler --once
 
 Note: automatic sync runs in a separate scheduler service. The `web` and `scheduler` services share `./data` and `./logs` directories.
 
+### Scheduler heartbeat health
+
+The scheduler normally sleeps until the next configured daily run. Unlike the web service, it has **no HTTP listener**. Its Docker health check reads `./data/scheduler_heartbeat.json` inside the container and verifies the scheduler process, a recently updated heartbeat, and that the next scheduled run is not overdue. While a sync is running, heartbeats continue during the child process wait; a run exceeding four hours is considered stuck.
+
+The heartbeat file contains only timestamps, PID, status (`starting`, `sleeping`, `running`, or `finished`) and the last completed sync return code. It is atomically written with file mode `0600`; no credentials or consumption data are included. By default it updates every 30 seconds. The health check allows up to 180 seconds without a heartbeat and a 15-minute scheduling grace period. Routine daily sleep is healthy. A sync returning nonzero is **recorded** but does not, by itself, mark the scheduler unhealthy; investigate it separately in logs.
+
+Inspect the state from the Docker host with `cat data/scheduler_heartbeat.json`, and use `docker compose ps` or `docker inspect` for container health. Docker marks unhealthy containers but does **not** automatically restart them solely because the health check fails; configure external alerting if needed. A one-off `smt-scheduler --once` run will not overwrite the long-running scheduler's heartbeat.
+
+
 ## Configuration
 
 See [`.env.example`](.env.example). The utility account variables `SMT_USERNAME`, `SMT_PASSWORD`, and `SMT_ESIID` are needed for usage retrieval. Application-level Basic authentication additionally needs a username and password unless explicitly delegated to a trusted upstream access gateway.

docker-compose.yml

@@ -24,4 +24,8 @@ services:
       - ./logs:/app/logs
     restart: unless-stopped
     healthcheck:
-      disable: true
+      test: ["CMD", "python", "-m", "smart_meter_texas.scheduler_health"]
+      interval: 60s
+      timeout: 10s
+      start_period: 45s
+      retries: 3

src/smart_meter_texas/scheduler.py

@@ -9,6 +9,8 @@
 from datetime import datetime, timedelta
 from zoneinfo import ZoneInfo
 
+from smart_meter_texas.scheduler_health import HEARTBEAT_INTERVAL_SECONDS, SchedulerHeartbeat, timestamp
+
 
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(description="Run Smart Meter Texas interval refresh on a daily schedule.")
@@ -28,7 +30,11 @@ def next_run(now: datetime, hhmm: str) -> datetime:
     return candidate
 
 
-def run_refresh(overlap_days: int) -> int:
+def run_refresh(
+    overlap_days: int,
+    heartbeat: SchedulerHeartbeat | None = None,
+    scheduled_for: datetime | None = None,
+) -> int:
     command = [
         sys.executable,
         "-m",
@@ -42,36 +48,92 @@ def run_refresh(overlap_days: int) -> int:
         "--raw-payload-dir",
         "logs/raw-payloads",
     ]
+    if heartbeat is not None:
+        heartbeat.pulse(
+            status="running",
+            next_run_at=None,
+            current_scheduled_at=scheduled_for.isoformat() if scheduled_for is not None else None,
+            current_run_started_at=timestamp(),
+        )
+
     print(json.dumps({"event": "refresh_start", "command": command, "ts": datetime.now().isoformat()}), flush=True)
-    result = subprocess.run(command)
-    print(json.dumps({"event": "refresh_end", "returncode": result.returncode, "ts": datetime.now().isoformat()}), flush=True)
-    if result.returncode == 0:
+    try:
+        if heartbeat is None:
+            # One-off runs deliberately do not overwrite the live scheduler's
+            # heartbeat when a second container shares the same data volume.
+            code = subprocess.run(command).returncode
+        else:
+            process = subprocess.Popen(command)
+            while True:
+                try:
+                    code = process.wait(timeout=HEARTBEAT_INTERVAL_SECONDS)
+                    break
+                except subprocess.TimeoutExpired:
+                    heartbeat.pulse(status="running")
+    except BaseException:
+        if heartbeat is not None:
+            heartbeat.pulse(
+                status="finished",
+                last_run_started_at=heartbeat.state["current_run_started_at"],
+                last_scheduled_at=heartbeat.state["current_scheduled_at"],
+                current_run_started_at=None,
+                current_scheduled_at=None,
+                last_run_finished_at=timestamp(),
+                last_run_returncode=127,
+            )
+        raise
+
+    print(json.dumps({"event": "refresh_end", "returncode": code, "ts": datetime.now().isoformat()}), flush=True)
+    if code == 0:
         try:
             from smart_meter_web.notification_settings import maybe_notify_latest_anomaly
             notification = maybe_notify_latest_anomaly()
             print(json.dumps({"event": "anomaly_alert", "status": notification["status"]}), flush=True)
         except Exception:
             # Preserve sync success; avoid logging secret webhook URLs or usage details.
             print(json.dumps({"event": "anomaly_alert", "status": "error"}), flush=True)
-    return result.returncode
+
+    if heartbeat is not None:
+        heartbeat.pulse(
+            status="finished",
+            last_run_started_at=heartbeat.state["current_run_started_at"],
+            last_scheduled_at=heartbeat.state["current_scheduled_at"],
+            current_run_started_at=None,
+            current_scheduled_at=None,
+            last_run_finished_at=timestamp(),
+            last_run_returncode=code,
+        )
+    return code
+
+
+def sleep_until(target: datetime, heartbeat: SchedulerHeartbeat) -> None:
+    heartbeat.pulse(status="sleeping", next_run_at=target.isoformat())
+    while True:
+        remaining = (target - datetime.now(target.tzinfo)).total_seconds()
+        if remaining <= 0:
+            return
+        time.sleep(min(remaining, HEARTBEAT_INTERVAL_SECONDS))
+        heartbeat.pulse()
 
 
 def main() -> None:
     args = parse_args()
     timezone = ZoneInfo(args.timezone)
 
-    if args.once or args.run_on_start:
-        code = run_refresh(args.overlap_days)
-        if args.once:
-            raise SystemExit(code)
+    if args.once:
+        raise SystemExit(run_refresh(args.overlap_days))
+
+    heartbeat = SchedulerHeartbeat()
+    if args.run_on_start:
+        run_refresh(args.overlap_days, heartbeat=heartbeat)
 
     while True:
         now = datetime.now(timezone)
         target = next_run(now, args.time)
         sleep_seconds = max(1, int((target - now).total_seconds()))
         print(json.dumps({"event": "sleep", "until": target.isoformat(), "seconds": sleep_seconds}), flush=True)
-        time.sleep(sleep_seconds)
-        run_refresh(args.overlap_days)
+        sleep_until(target, heartbeat)
+        run_refresh(args.overlap_days, heartbeat=heartbeat, scheduled_for=target)
 
 
 if __name__ == "__main__":

src/smart_meter_texas/scheduler_health.py

@@ -0,0 +1,138 @@
+"""File-backed scheduler heartbeat and Docker liveness/deadline check.
+
+The scheduler does not serve HTTP. Its heartbeat is written atomically and
+contains scheduling metadata only (never meter data or credentials).
+"""
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any
+
+HEARTBEAT_INTERVAL_SECONDS = 30
+MAX_HEARTBEAT_AGE_SECONDS = 180
+SCHEDULE_GRACE_SECONDS = 15 * 60
+MAX_RUN_SECONDS = 4 * 60 * 60
+
+
+def heartbeat_path() -> Path:
+    return Path(os.environ.get("SMT_SCHEDULER_HEARTBEAT_PATH", "data/scheduler_heartbeat.json"))
+
+
+def utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def timestamp() -> str:
+    return utc_now().isoformat()
+
+
+def save_heartbeat(path: Path, payload: dict[str, Any]) -> None:
+    """Replace the state file atomically with POSIX-restricted permissions."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            if hasattr(os, "fchmod"):
+                os.fchmod(handle.fileno(), 0o600)
+            json.dump(payload, handle, sort_keys=True)
+            handle.write("\n")
+        os.replace(temp_name, path)
+    finally:
+        if os.path.exists(temp_name):
+            os.unlink(temp_name)
+
+
+class SchedulerHeartbeat:
+    def __init__(self, path: Path | None = None) -> None:
+        self.path = path if path is not None else heartbeat_path()
+        started = timestamp()
+        self.state: dict[str, Any] = {
+            "schema_version": 1,
+            "pid": os.getpid(),
+            "started_at": started,
+            "status": "starting",
+            "next_run_at": None,
+            "last_scheduled_at": None,
+            "current_scheduled_at": None,
+            "current_run_started_at": None,
+            "last_run_started_at": None,
+            "last_run_finished_at": None,
+            "last_run_returncode": None,
+        }
+        self.pulse()
+
+    def pulse(self, **fields: Any) -> None:
+        self.state.update(fields)
+        self.state["heartbeat_at"] = timestamp()
+        save_heartbeat(self.path, self.state)
+
+
+def _read_time(value: Any) -> datetime:
+    if not isinstance(value, str):
+        raise ValueError("missing timestamp")
+    parsed = datetime.fromisoformat(value)
+    if parsed.tzinfo is None or parsed.utcoffset() is None:
+        raise ValueError("timestamp lacks timezone")
+    return parsed.astimezone(timezone.utc)
+
+
+def check_heartbeat(
+    path: Path | None = None,
+    *,
+    now: datetime | None = None,
+    max_age_seconds: int = MAX_HEARTBEAT_AGE_SECONDS,
+    schedule_grace_seconds: int = SCHEDULE_GRACE_SECONDS,
+    max_run_seconds: int = MAX_RUN_SECONDS,
+) -> tuple[bool, str]:
+    """Check process liveness, heartbeat freshness, and scheduled-run deadline."""
+    source = path if path is not None else heartbeat_path()
+    try:
+        state = json.loads(source.read_text(encoding="utf-8"))
+        if not isinstance(state, dict) or state.get("schema_version") != 1:
+            return False, "invalid heartbeat schema"
+        pid = state.get("pid")
+        if type(pid) is not int or pid < 1:
+            return False, "invalid scheduler pid"
+        try:
+            os.kill(pid, 0)
+        except ProcessLookupError:
+            return False, "scheduler process missing"
+        except PermissionError:
+            pass  # A running process may not be signalable by this user.
+
+        current = (now if now is not None else utc_now()).astimezone(timezone.utc)
+        heartbeat = _read_time(state.get("heartbeat_at"))
+        age = (current - heartbeat).total_seconds()
+        if age > max_age_seconds or age < -120:
+            return False, "scheduler heartbeat stale or clock skewed"
+
+        status = state.get("status")
+        if status == "sleeping":
+            next_run = _read_time(state.get("next_run_at"))
+            if current > next_run + timedelta(seconds=schedule_grace_seconds):
+                return False, "scheduled run overdue"
+        elif status == "running":
+            started = _read_time(state.get("current_run_started_at"))
+            if current > started + timedelta(seconds=max_run_seconds):
+                return False, "scheduled run exceeded maximum duration"
+        elif status not in {"starting", "finished"}:
+            return False, "invalid scheduler status"
+        # A failed sync does not mean the scheduler is stuck; report the result in
+        # the state file, rather than triggering unhelpful container restarts.
+        return True, f"scheduler {status}"
+    except (OSError, ValueError, TypeError, json.JSONDecodeError) as exc:
+        return False, f"unreadable scheduler heartbeat: {type(exc).__name__}"
+
+
+def main() -> None:
+    ok, message = check_heartbeat()
+    print(message)
+    raise SystemExit(0 if ok else 1)
+
+
+if __name__ == "__main__":
+    main()