#!/usr/bin/env python3 """Error escalation watchdog with a capped, graceful auto-reboot. Polls the mission-critical services' state files (camera + sensors -- deliberately NOT lumen/epaper, matching led_status.py's reasoning: a display or light glitch isn't mission failure and doesn't warrant a disruptive reboot). A per-service systemd restart (Restart=on-failure) is given first crack at self-healing; only once a service has remained continuously in error for `watchdog.escalation_after_seconds` does this watchdog escalate to a full Pi reboot. Reboots are capped at `watchdog.max_reboot_attempts` (persisted on /data, survives the reboot it's counting) so a permanently broken device can't send the Pi into a reboot-crash-loop for the rest of the mission, burning battery and SD wear for no benefit. Once the cap is hit, this watchdog gives up rebooting and reports its own state as "error" -- which is what makes led_status.py leave the LED in its blinking-BLUE fault indication for the remainder of the mission, since watchdog is itself one of led_status.py's monitored mission-critical services. If this process itself dies, its state file goes stale, which state_lib.has_error() also treats as an error -- so a watchdog crash surfaces the same way a watchdog-detected fault would, rather than silently going unsupervised. """ from __future__ import annotations import os import subprocess import sys import time from pathlib import Path from typing import Dict, List, Optional sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from bubblecam import bounded_io, config, events, reboot_guard, state_lib # noqa: E402 # The data partition is probed on a slower cadence than the state files: it # costs a real (if tiny) write to the SD card, and unlike a tmpfs read that # is both wear and I/O contention with capture. STORAGE_PROBE_INTERVAL_SECONDS = 60 # How long to wait for a graceful `systemctl reboot` before resetting the # hard way. A clean shutdown tries to unmount /data, which never returns if # the card is what wedged -- the exact case this reboot exists to escape. FORCED_REBOOT_AFTER_SECONDS = 60 # Non-critical services get a systemctl restart (never a reboot) when # their heartbeat goes stale while systemd still says "active" -- a # process can be alive but wedged (observed: the LED service froze solid # in a clobbered render; systemd was none the wiser). Over an unattended # multi-day mission, "no lights until a human notices" is worth healing. NON_CRITICAL_UNITS = { "led": "bubblecam-led.service", "epaper": "bubblecam-epaper.service", "lumen": "bubblecam-lumen.service", } NON_CRITICAL_STALE_AFTER_SECONDS = 120 # Floor between restarts of the same service, so a permanently broken one # neither restart-storms nor floods the events log. NON_CRITICAL_RESTART_COOLDOWN_SECONDS = 300 def stale_noncritical_units(stale_after: float = NON_CRITICAL_STALE_AFTER_SECONDS): """(name, unit) pairs for non-critical services whose heartbeat exists but has gone stale. Services with no state file at all are skipped -- never-started is a choice (e.g. lumen deliberately disabled), not a fault this remediation should fight.""" stale = [] for name, unit in NON_CRITICAL_UNITS.items(): state = state_lib.read_state(name) if state is not None and not state_lib.is_fresh(state, stale_after): stale.append((name, unit)) return stale def _simulate() -> bool: # Read fresh on every call, not a module-level constant -- this module # is exec'd once by tests/conftest.py's load_bin_module() at collection # time, before any per-test monkeypatch of BUBBLECAM_SIM has run. return os.environ.get("BUBBLECAM_SIM") == "1" def camera_service_names(cfg: dict) -> List[str]: mode = cfg["camera"]["mode"] if mode.startswith("dual_node"): return ["camera_left", "camera_right"] return ["camera_primary"] def camera_systemd_units(cfg: dict) -> List[str]: mode = cfg["camera"]["mode"] instances = ["left", "right"] if mode.startswith("dual_node") else ["primary"] return [f"bubblecam-camera@{instance}.service" for instance in instances] def _run_abandonable(cmd: list, timeout: float) -> None: """Run a command, abandoning it (not just killing it) on timeout. On a machine whose storage has died, merely exec'ing systemctl can block forever in uninterruptible sleep -- the binary has to be paged in from the dead device. Python's subprocess timeout can't recover from that (it SIGKILLs then wait()s, and a D-state child never reaps), so the whole call runs on a throwaway thread instead. Observed for real: the watchdog froze inside a systemctl call mid-escalation and the Pi hung 12.5 hours with the fix one line away. """ bounded_io.run_with_timeout(lambda: subprocess.run(cmd, check=False), timeout=timeout) def stop_camera_services(cfg: dict) -> None: """Ask ffmpeg (via each capture_supervisor) to stop gracefully so the current video segment gets finalized before we reboot out from under it. The record-switch daemon is stopped first: otherwise its reconcile loop would see stopped-but-switch-says-record units and restart them inside the pre-reboot grace window, un-finalizing the very segment this stop exists to protect. It comes back with everything else after the reboot.""" for unit in ["bubblecam-record-switch.service"] + camera_systemd_units(cfg): if _simulate(): print(f"[SIM WATCHDOG] systemctl stop {unit}") continue _run_abandonable(["systemctl", "stop", unit], timeout=120) def storage_healthy(video_dir: Path) -> bool: """Time-bounded write probe of the data partition. A dead/wedged SD card doesn't announce itself: writes just never return, and every service that touches the card silently stops making progress while the kernel stays healthy enough that the hardware watchdog never fires. Observed in a real soak -- the Pi hung for 8.5 hours with no log line explaining it, because writing that log line needed the same dead card. This probe is what turns that silent hang into a reboot. """ if _simulate(): return True def _probe() -> bool: probe_path = video_dir / ".bubblecam-storage-probe" with open(probe_path, "w") as f: f.write("ok") f.flush() os.fsync(f.fileno()) probe_path.unlink() return True return bool(bounded_io.run_with_timeout(_probe, default=False)) def trigger_reboot() -> None: if _simulate(): print("[SIM WATCHDOG] systemctl reboot") return _run_abandonable(["systemctl", "reboot"], timeout=30) def force_reboot() -> None: """Reset via the kernel directly, bypassing systemd's shutdown sequence. Only reached when a graceful reboot didn't happen within FORCED_REBOOT_AFTER_SECONDS -- which on this hardware means systemd is blocked unmounting a filesystem whose device stopped responding. Data loss is not a concern at that point: the card already isn't accepting writes, and every completed segment is crash-safe by design. """ if _simulate(): print("[SIM WATCHDOG] forced reboot via sysrq") return try: with open("/proc/sysrq-trigger", "w") as f: f.write("b") except OSError: subprocess.run(["reboot", "-f"], check=False) def main() -> int: cfg = config.load() watchdog_cfg = cfg["watchdog"] stale_after = watchdog_cfg["stale_after_seconds"] escalation_after = watchdog_cfg["escalation_after_seconds"] max_attempts = watchdog_cfg["max_reboot_attempts"] healthy_reset_after = watchdog_cfg["healthy_reset_after_seconds"] poll_seconds = watchdog_cfg["poll_seconds"] graceful_stop_grace = watchdog_cfg["graceful_stop_grace_seconds"] camera_names = camera_service_names(cfg) video_dir = Path(cfg["paths"]["video_dir"]) error_since: Dict[str, float] = {} noncritical_restarted_at: Dict[str, float] = {} healthy_since: Optional[float] = None storage_ok = True storage_checked_at = 0.0 state_lib.write_state("watchdog", "running") boot_attempts = reboot_guard.read_count() events.log("watchdog", "system up" + ( f" -- after recovery reboot ({boot_attempts} attempt(s) on record)" if boot_attempts else "" )) while True: now = time.monotonic() if now - storage_checked_at >= STORAGE_PROBE_INTERVAL_SECONDS: was_ok = storage_ok storage_ok = storage_healthy(video_dir) storage_checked_at = now if was_ok and not storage_ok: # Best effort -- the card this line is written to may be # the thing that just died (events.log is time-bounded). events.log("watchdog", "STORAGE PROBE FAILED -- /data not accepting writes") elif storage_ok and not was_ok: events.log("watchdog", "storage recovered") # Heal wedged non-critical services on the same slow cadence. for name, unit in stale_noncritical_units(): last = noncritical_restarted_at.get(name) if last is not None and now - last < NON_CRITICAL_RESTART_COOLDOWN_SECONDS: continue noncritical_restarted_at[name] = now events.log("watchdog", f"{name} heartbeat stale -- restarting {unit}") if _simulate(): print(f"[SIM WATCHDOG] systemctl restart {unit}") else: _run_abandonable(["systemctl", "restart", unit], timeout=60) # A camera stopped on purpose by the physical record switch is not # a fault -- only monitor the camera services while a live switch # reading requests recording (missing/stale switch data fails # toward monitoring, see state_lib.recording_requested). recording_requested = state_lib.recording_requested(stale_after) critical_services = (camera_names if recording_requested else []) + ["sensors"] currently_erroring = [] for name in camera_names + ["sensors"]: if name in critical_services and state_lib.has_error(name, stale_after): error_since.setdefault(name, now) currently_erroring.append(name) else: error_since.pop(name, None) # Storage is mission-critical regardless of the record switch: a # wedged card stops sensor logging too, and is unrecoverable without # a reboot. if not storage_ok: error_since.setdefault("storage", now) currently_erroring.append("storage") else: error_since.pop("storage", None) escalate_candidates = [ name for name in currently_erroring if now - error_since[name] >= escalation_after ] if escalate_candidates: detail = "; ".join(f"{name} errored {now - error_since[name]:.0f}s" for name in escalate_candidates) if reboot_guard.exhausted(max_attempts): state_lib.write_state( "watchdog", "error", detail=f"reboot attempts exhausted ({max_attempts}); giving up: {detail}", ) events.log("watchdog", f"reboot attempts exhausted ({max_attempts}) -- giving up: {detail}") else: storage_failed = "storage" in escalate_candidates state_lib.write_state("watchdog", "error", detail=f"escalating reboot: {detail}") if not storage_failed: # Graceful path: let ffmpeg finalize the open segment # before the reboot pulls the rug out. stop_camera_services(cfg) time.sleep(graceful_stop_grace) attempt = reboot_guard.increment() state_lib.write_state( "watchdog", "error", detail=f"rebooting now (attempt {attempt}/{max_attempts}): {detail}", ) events.log( "watchdog", f"REBOOTING ({'forced' if storage_failed else 'graceful'}, " f"attempt {attempt}/{max_attempts}): {detail}", ) if storage_failed: # Dead storage: the graceful path is pointless (the # card stopped taking writes minutes ago; every # completed segment is crash-safe by design) and # actively dangerous -- systemctl must be exec'd FROM # the dead device, which blocks forever, and a # graceful shutdown wedges unmounting it. Straight to # the exec-free kernel reset (a pure /proc syscall). if not _simulate(): force_reboot() time.sleep(30) else: force_reboot() else: trigger_reboot() if not _simulate(): # Still alive after this long means systemd's # shutdown is itself blocked, so escalate to the # kernel-level reset rather than looping forever. time.sleep(FORCED_REBOOT_AFTER_SECONDS) force_reboot() time.sleep(30) healthy_since = None elif currently_erroring: # In error, but not yet past the escalation grace period -- # let a per-service systemd restart have its chance first. state_lib.write_state( "watchdog", "running", detail=f"monitoring, not yet escalating: {', '.join(currently_erroring)}", ) healthy_since = None else: state_lib.write_state("watchdog", "running") if healthy_since is None: healthy_since = now elif now - healthy_since >= healthy_reset_after and reboot_guard.read_count() > 0: reboot_guard.reset() time.sleep(poll_seconds) if __name__ == "__main__": sys.exit(main())