From d5139c99caab2581281abbe6d001baf18c1abac8 Mon Sep 17 00:00:00 2001 From: oskar Date: Tue, 14 Jul 2026 15:55:38 +0200 Subject: [PATCH] =?UTF-8?q?fix(observer):=20checkpoint=20by=20timestamp=20?= =?UTF-8?q?not=20lexical=20path=20=E2=80=94=20lexically-smaller-but-newer?= =?UTF-8?q?=20events=20were=20silently=20skipped=20forever=20(poisoned=20n?= =?UTF-8?q?ode)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per-node checkpoint now stores the last-processed event TIMESTAMP (int epoch) instead of a file path compared lexically. A file is "new" iff its timestamp (parsed from evt----.json, mtime fallback) exceeds the node's checkpoint; processing is ordered by timestamp, not path. Root cause (PIHA dead ~34d, 2026-07-12): a stray evt-unknown--… file landed in events/piha/, lexically greater than every evt-piha-… name. The lexical checkpoint pinned there, so every genuinely newer piha event sorted "before" it and was skipped forever. Event backlog grew to 7344 files, last_seen frozen, shadow-read logged false SHADOW_LIVENESS_MISMATCH event=dead prom=up. - _event_ts_from_path: filename epoch, mtime fallback; NEVER returns 0 for an existing file (0 == "older than checkpoint" == the poison). - _checkpoint_ts_from_value: graceful migration of pre-fix path-string checkpoints (and the older last_processed_file format) to int epochs; unparseable → 0 (reprocess all — safe, process_event is idempotent on last_seen/world_state; bias to reprocess, never to skip). - Preserved: quarantine of bad events, observer-source re-ingest guard. - Regression tests (test_incident_lifecycle.py section 9): lexically-smaller- but-newer processed, unparseable name falls back to mtime (not wedged), ts-not-path ordering, both checkpoint-format migrations, helper units. Separate bug filed in backlog (not fixed here): ha-diag-agent emits node= "unknown" events (config.py node_name default) into another node's dir when NODE_NAME reaches the compose volume path but not the app env — the source of the poison file. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/backlog.md | 60 ++++- scripts/observer/observer.py | 132 +++++++++-- .../tests/test_incident_lifecycle.py | 220 +++++++++++++++++- 3 files changed, 382 insertions(+), 30 deletions(-) diff --git a/docs/backlog.md b/docs/backlog.md index 6d1adf2..b86d9d4 100644 --- a/docs/backlog.md +++ b/docs/backlog.md @@ -579,11 +579,61 @@ więc KAŻDY nowy event był uznawany za starszy niż checkpoint i pomijany. observera → 7344 eventy przetworzone, `last_seen_age` spadł z 2 082 036 s (~24 dni) do 19 s, status=online/fresh, mismatch zniknął. -**Fix systemowy (DO ZROBIENIA).** Porównanie po ścieżce jest z gruntu kruche — -wystarczy JEDEN plik o "większej" nazwie (inny node w katalogu, inny prefiks) żeby -trwale zablokować węzeł. Checkpoint powinien opierać się na **timestampie** (jest -w nazwie: `evt---`) albo mtime pliku, nie na kolejności alfabetycznej. -Dodatkowo: zbadać, czemu eventy z node="unknown" (HA) trafiają do katalogu innego węzła. +**Fix systemowy (ZROBIONE 2026-07-14, `task/fix-observer-checkpoint`).** Checkpoint +per-węzeł trzyma teraz **TIMESTAMP** (int epoch), nie ścieżkę. „Nowy event" = +`ts_z_nazwy_pliku > checkpoint_ts_węzła`; kolejność przetwarzania sortowana po +timestampie, nie leksykalnie. Timestamp parsowany z nazwy `evt---…` +(regex `-(\d{9,11})-`, ten sam co `operator_ui._event_file_ts`); **fallback na +mtime** gdy nazwa nie pasuje — nieparsowalna nazwa NIGDY nie zwraca 0 (0 = leksykalne +„starszy niż checkpoint" = dokładnie ten poison). Migracja starych checkpointów +(ścieżka→ts) przy starcie; nieparsowalna wartość → 0 (reprocess wszystkiego — +bezpieczne, `process_event` jest idempotentne na `last_seen`/`world_state`; lepiej +przetworzyć duplikaty niż zgubić węzeł). Testy regresyjne w +`test_incident_lifecycle.py` (sekcja 9). Znany, akceptowalny warunek brzegowy: +strict `>` może pominąć event o `ts == checkpoint` dostarczony w PÓŹNIEJSZYM cyklu +niż inne eventy z tej samej sekundy — nierealne przy cadence shippingu (rsync co +60 s wysyła całą partię danej sekundy razem; kolejne partie są ~60 s od siebie). + +**Uwaga do idempotencji (zbadane).** Reprocess tego samego eventu NIE psuje +world_state (status/last_seen deterministyczne, resolve incydentu guardowany na +`status=="active"`), ALE `_handle_incident`/`deployment_*` inkrementują +`occurrence_count` i dopisują do `events[]` przy każdym przetworzeniu — reprocess +(np. jednorazowo po migracji) zawyża te liczniki. To kosmetyka, nie korupcja stanu. +Docelowo można dedupować po `event.id` w `events[]` — osobny, drobny task. + +## Bug: ha-diag-agent emituje eventy z node="unknown" do katalogu innego węzła (2026-07-14) + +**Kontekst.** To był plik-truciciel z buga checkpointu wyżej: +`evt-unknown-1781254800-ha_update_available-homeassistant-951.json` w +`events/piha/`. Node w evencie = `unknown`, ale plik wylądował w katalogu `piha/`. +Sufiks `-951` to `_seq` emittera → agent nachodził długo, wyemitował 951 eventów, +wszystkie jako `node="unknown"`. + +**Root cause (config-wiring).** Tożsamość agenta (`node_name`) i KATALOG eventów +pochodzą z DWÓCH niezależnych źródeł: +- `services/ha-diag-agent/src/ha_diag/config.py:20` → `node_name: str = "unknown"` + (domyślne, gdy env `NODE_NAME` nie dotrze do procesu w kontenerze). +- `services/ha-diag-agent/docker-compose.yml:12` → wolumen + `/opt/homelab/events/${NODE_NAME:-ha-diag}:/events` — `${NODE_NAME}` jest + interpolowane po stronie HOSTA (compose), a katalog jest dodatkowo twardo + przypięty do `piha` w `hosts/piha/runtime/ha-diag-agent/docker-compose.override.yml`. + +Jeśli `NODE_NAME` trafi do interpolacji wolumenu/override (→ `piha`), ale NIE do +`environment:` procesu (albo `Settings.load()` przez `os.environ.setdefault` go nie +nadpisze), aplikacja czyta `node_name="unknown"` i pisze eventy `node="unknown"` +do katalogu `events/piha/`. Rozjazd między nazwą w evencie a katalogiem docelowym. + +**Skutek.** Poza zatruciem checkpointu (już naprawione osobno): eventy `node="unknown"` +są bezużyteczne dla world_state (observer tworzy węzeł-widmo `unknown`, potem prune go +kasuje bo nie ma go w topologii) — realny sygnał z ha-diag na piha przepada. + +**Fix (DO ZROBIENIA — osobny task).** Wymusić spójność: albo (a) `node_name` NIGDY +nie może być `"unknown"` w produkcji — fail-fast/loud log gdy `NODE_NAME` nieustawione, +zamiast cichego defaultu; albo (b) wyprowadzać `node_name` z tej samej zmiennej co +ścieżka wolumenu i zweryfikować, że override piha faktycznie wstrzykuje `NODE_NAME=piha` +do `environment:` kontenera (jest w override — sprawdzić, czemu w praktyce nie zadziałało: +kolejność `env_file` vs `environment`, restart bez recreate, stary kontener). Docelowo +observer/emitter powinien odrzucać/kwarantannować event, którego `node` != katalog. ## Bug: deploy-local.sh control-plane pada na ghost-kontenerach i zostawia mózg rozłożony (2026-07-12) diff --git a/scripts/observer/observer.py b/scripts/observer/observer.py index 5eb240f..9847bac 100644 --- a/scripts/observer/observer.py +++ b/scripts/observer/observer.py @@ -1,4 +1,5 @@ import os +import re import sys import json import time @@ -55,6 +56,69 @@ def _parse_ts(ts) -> float: except Exception: return 0.0 + +# Event filenames follow evt----.json (node_agent.py, +# ha_diag/event_emitter.py, and the observer's own _emit_node_transition). The +# is the authoritative ordering key for checkpointing — matched the same +# way operator_ui.py::_event_file_ts does (a 9–11 digit run flanked by dashes; +# real epochs are 10 digits and stay so until year 2286). +_EVENT_TS_RE = re.compile(r"-(\d{9,11})-") + + +def _ts_from_event_name(name) -> "int | None": + """Parse the embedded from an event filename or full path. + + Returns the int epoch, or None when the name does not carry one (foreign + prefix, events.py naming, junk file) so the caller can fall back to mtime. + """ + m = _EVENT_TS_RE.search(Path(name).stem) + return int(m.group(1)) if m else None + + +def _event_ts_from_path(file_path) -> int: + """Ordering/checkpoint timestamp for an event file. + + Primary: the embedded in the filename. Fallback (name doesn't + parse): the file's mtime. NEVER returns 0 for an existing file — a value of + 0 would make the file compare as "older than the checkpoint" and be skipped + forever, which is exactly the lexical-path poisoning this fix removes. If + even stat() fails, fall back to now() so the file is treated as new and gets + a chance to be processed (and quarantined if truly unreadable). + """ + ts = _ts_from_event_name(file_path) + if ts is not None: + return ts + try: + return int(os.stat(file_path).st_mtime) + except OSError: + return int(time.time()) + + +def _checkpoint_ts_from_value(value) -> int: + """Coerce a stored checkpoint value into an int epoch (migration helper). + + Historical formats of observer_checkpoint.json values: + - int/float → already a timestamp (current format) — kept as-is + - path/name string → the pre-fix lexical checkpoint; parse the embedded + out of it so the node resumes near where it + left off instead of reprocessing everything + - anything else / unparseable string → 0 (reprocess all events for that + node). Reprocessing is safe: process_event is idempotent w.r.t. + last_seen/world_state, so re-ingesting duplicates cannot corrupt state — + whereas guessing too high a checkpoint could silently drop events (the + failure mode being fixed). Bias to reprocess, never to skip. + """ + if isinstance(value, bool): # bool is an int subclass — exclude explicitly + return 0 + if isinstance(value, (int, float)): + return int(value) + if isinstance(value, str) and value: + ts = _ts_from_event_name(value) + if ts is not None: + return ts + return 0 + + # Constants and Paths RUNTIME_PATH = os.getenv("RUNTIME_PATH", "/opt/homelab") EVENTS_DIR = Path(RUNTIME_PATH) / "events" @@ -83,9 +147,13 @@ logger = logging.getLogger("observer") class Observer: def __init__(self): - # Per-node-directory checkpoint: {"vps": "last/file/path", "piha": "last/file/path"} - # Replaces the old single last_processed_file which silently skipped event dirs - # that sort alphabetically before the checkpoint (e.g. piha/ < vps/). + # Per-node-directory checkpoint keyed on the last-processed event + # TIMESTAMP (int epoch): {"vps": 1784000000, "piha": 1784000123}. + # A file is "new" iff its event timestamp > the node's checkpoint. + # This replaces the earlier lexical-PATH comparison, which permanently + # poisoned a node the moment a single file with a lexically-larger name + # landed in its dir (e.g. evt-unknown-… > evt-piha-…): every genuinely + # newer event then sorted "before" the checkpoint and was skipped forever. self.node_checkpoints: dict = {} self.world_state = { "nodes": {}, @@ -168,15 +236,28 @@ class Observer: checkpoint = json.load(f) if "node_checkpoints" in checkpoint: - # New format: per-directory checkpoints. - self.node_checkpoints = checkpoint["node_checkpoints"] + # Per-directory checkpoints. Values may be int epochs (current + # format) OR pre-fix path strings — coerce every value to an + # int timestamp so a checkpoint file written by the old + # lexical-path observer migrates transparently on first start. + raw = checkpoint["node_checkpoints"] or {} + self.node_checkpoints = { + node: _checkpoint_ts_from_value(val) + for node, val in raw.items() + } + if any(not isinstance(v, (int, float)) for v in raw.values()): + logger.info( + "Migrated path-based node_checkpoints → timestamps: %s", + self.node_checkpoints, + ) elif "last_processed_file" in checkpoint: - # Migrate old single-file checkpoint: extract node dir from path. + # Migrate the very old single-file checkpoint: extract node dir + # from the path and the timestamp from the filename. old = checkpoint["last_processed_file"] if old: try: node_dir = Path(old).relative_to(EVENTS_DIR).parts[0] - self.node_checkpoints = {node_dir: old} + self.node_checkpoints = {node_dir: _checkpoint_ts_from_value(old)} logger.info(f"Migrated old checkpoint → node_checkpoints: {self.node_checkpoints}") except Exception: pass # Bad path — start fresh @@ -687,11 +768,12 @@ class Observer: except Exception as e: logger.error(f"Failed to touch heartbeat file: {e}") - # Collect all event files grouped by node directory. - # Per-node checkpoints are compared within each directory independently, - # so late-arriving events from remote nodes (sorted earlier in the path) - # are never skipped just because another node's checkpoint is further ahead. - all_files = sorted(glob.glob(str(EVENTS_DIR / "**" / "*.json"), recursive=True)) + # Collect all event files grouped by node directory. A file is "new" + # when its event TIMESTAMP (from the filename, mtime fallback) is greater + # than the node's checkpoint timestamp — never a lexical path compare, so + # a lexically-smaller-but-newer filename (evt-piha-… after a stray + # evt-unknown-…) can no longer poison a node into skipping every event. + all_files = glob.glob(str(EVENTS_DIR / "**" / "*.json"), recursive=True) new_files = [] for file_path in all_files: @@ -699,9 +781,15 @@ class Observer: node_dir = str(Path(file_path).relative_to(EVENTS_DIR).parts[0]) except (IndexError, ValueError): node_dir = "__unknown__" - last_for_node = self.node_checkpoints.get(node_dir, "") - if file_path > last_for_node: - new_files.append((node_dir, file_path)) + ev_ts = _event_ts_from_path(file_path) + last_for_node = self.node_checkpoints.get(node_dir, 0) + if ev_ts > last_for_node: + new_files.append((ev_ts, node_dir, file_path)) + + # Process oldest-first (tie-break on path for determinism) so the + # checkpoint advances monotonically in time and a mid-batch crash resumes + # from the right place. + new_files.sort(key=lambda t: (t[0], t[2])) if not new_files: # Even if no new events, prune stale entries and refresh summary freshness. @@ -710,8 +798,8 @@ class Observer: return logger.info(f"Processing {len(new_files)} new events across " - f"{len({n for n, _ in new_files})} node(s)") - for node_dir, file_path in new_files: + f"{len({n for _, n, _ in new_files})} node(s)") + for ev_ts, node_dir, file_path in new_files: try: with open(file_path, "r") as f: event = json.load(f) @@ -721,13 +809,13 @@ class Observer: # node we just declared dead. They are still consumed by the # supervisor (alerting) and the panel event feed. if event.get("source") == "observer": - if file_path > self.node_checkpoints.get(node_dir, ""): - self.node_checkpoints[node_dir] = file_path + if ev_ts > self.node_checkpoints.get(node_dir, 0): + self.node_checkpoints[node_dir] = ev_ts continue self.process_event(event) - # Advance per-node checkpoint (only forward — no regression). - if file_path > self.node_checkpoints.get(node_dir, ""): - self.node_checkpoints[node_dir] = file_path + # Advance per-node checkpoint by timestamp (only forward). + if ev_ts > self.node_checkpoints.get(node_dir, 0): + self.node_checkpoints[node_dir] = ev_ts except Exception as e: logger.error( "Error processing node_dir=%s file=%s (%s: %s)", diff --git a/services/control-plane/tests/test_incident_lifecycle.py b/services/control-plane/tests/test_incident_lifecycle.py index fb3f41d..0053d50 100644 --- a/services/control-plane/tests/test_incident_lifecycle.py +++ b/services/control-plane/tests/test_incident_lifecycle.py @@ -346,7 +346,11 @@ def test_run_once_quarantines_bad_event_and_processes_next_for_same_node(tmp_pat assert quarantined.exists() assert not bad_event.exists() assert obs.world_state["nodes"]["lustro"]["status"] == "online" - assert obs.node_checkpoints["lustro"] == str(good_event) + # Checkpoint is now a timestamp, not a path. The filenames here carry no + # 10-digit epoch (evt-lustro-2-good) so _event_ts_from_path falls back to + # mtime — a positive int, and the node is not wedged. + assert isinstance(obs.node_checkpoints["lustro"], int) + assert obs.node_checkpoints["lustro"] > 0 # --------------------------------------------------------------------------- @@ -538,5 +542,215 @@ def test_run_once_skips_observer_emitted_events(tmp_path): # Not ingested: last_seen stays old → vps stays dead/offline. assert obs.world_state["nodes"]["vps"]["status"] == "offline" - # Checkpoint still advanced over the skipped file. - assert obs.node_checkpoints.get("vps") == str(ev_path) + # Checkpoint still advanced past the skipped file — now by timestamp + # (the 9999999999 epoch embedded in the filename), not the path string. + assert obs.node_checkpoints.get("vps") == 9999999999 + + +# --------------------------------------------------------------------------- +# 9. Checkpoint by TIMESTAMP, not lexical path +# --------------------------------------------------------------------------- +# Regression suite for the "poisoned node" bug (2026-07-12): PIHA was dead to +# the observer for ~34 days because its checkpoint compared event PATHS +# lexically. A stray evt-unknown--… file (lexically > every real +# evt-piha-… name) pinned the checkpoint, so every genuinely newer piha event +# sorted "before" it and was skipped forever. The fix keys the checkpoint on +# the timestamp embedded in the filename (mtime fallback). + +from observer.observer import ( # noqa: E402 + _ts_from_event_name, + _event_ts_from_path, + _checkpoint_ts_from_value, +) + + +def _write_event(dir_path: Path, name: str, *, node: str, ts: int, + etype: str = "node_health", source: str | None = None) -> Path: + dir_path.mkdir(parents=True, exist_ok=True) + body = { + "id": name[:-5], + "timestamp": ts, + "date": "2026-07-14T00:00:00Z", + "type": etype, + "severity": "info", + "node": node, + "service": "", + "message": "test", + "payload": {"disk_pct": 1, "mem_pct": 2, "cpu_pct": 3}, + } + if source: + body["source"] = source + p = dir_path / name + p.write_text(json.dumps(body)) + return p + + +def _topology_with(obs, *nodes: str) -> None: + import observer.observer as obs_mod + lines = ["nodes:"] + for n in nodes: + lines.append(f" {n}:\n roles: [infra]\n connectivity: {{}}") + obs_mod.INVENTORY_TOPOLOGY.write_text("\n".join(lines) + "\n") + obs.inventory = obs._load_inventory() + + +def test_lexically_smaller_but_newer_event_is_processed(tmp_path): + """THE regression: a stray evt-unknown-… must not wedge piha forever. + + An evt-unknown--… file lands in the piha/ dir (lexically larger + than any evt-piha-… name). A genuinely NEWER evt-piha-… event, which is + lexically SMALLER, must still be processed. + """ + obs = _make_observer_simple(tmp_path) + import observer.observer as obs_mod + _topology_with(obs, "vps", "piha") + + piha_dir = obs_mod.EVENTS_DIR / "piha" + # The poison: HA event with node="unknown" that landed in piha/ (older ts). + _write_event( + piha_dir, + "evt-unknown-1781254800-ha_update_available-homeassistant-951.json", + node="unknown", ts=1781254800, etype="ha_update_available", + ) + obs.run_once() + # Checkpoint is the stray event's ts (unknown node itself is pruned). + assert obs.node_checkpoints["piha"] == 1781254800 + + # A newer real piha heartbeat — lexically "evt-piha-…" < "evt-unknown-…". + _write_event( + piha_dir, "evt-piha-1784000000-node_health-node.json", + node="piha", ts=1784000000, + ) + obs.run_once() + + # Under the old lexical logic this was skipped forever. Now it is processed. + assert "piha" in obs.world_state["nodes"] + assert obs.world_state["nodes"]["piha"]["last_seen"] == 1784000000 + assert obs.node_checkpoints["piha"] == 1784000000 + + +def test_unparseable_name_does_not_block_node(tmp_path): + """A file whose name carries no epoch falls back to mtime, never ts=0. + + ts=0 would compare as "older than the checkpoint" and be skipped forever — + the exact poisoning mechanism. The event must be ingested instead. + """ + obs = _make_observer_simple(tmp_path) + import observer.observer as obs_mod + _topology_with(obs, "vps", "solaria") + + solaria_dir = obs_mod.EVENTS_DIR / "solaria" + # No 10-digit epoch in the name → mtime fallback (≈ now, a positive int). + # Use a fresh event timestamp so read-time liveness keeps it online. + fresh = int(time.time()) + _write_event( + solaria_dir, "weird-legacy-name.json", + node="solaria", ts=fresh, + ) + obs.run_once() + + # Ingested (not skipped): the node exists with the event's last_seen, and + # the checkpoint advanced to a positive int via the mtime fallback. + assert "solaria" in obs.world_state["nodes"] + assert obs.world_state["nodes"]["solaria"]["last_seen"] == fresh + assert obs.world_state["nodes"]["solaria"]["status"] == "online" + assert isinstance(obs.node_checkpoints["solaria"], int) + assert obs.node_checkpoints["solaria"] > 0 + + +def test_checkpoint_governed_by_timestamp_not_path(tmp_path): + """A lexically-larger but temporally-OLDER file must be skipped. + + Proves the ordering key is the timestamp, not the path: an event whose name + sorts after the checkpoint yet whose ts predates it does not regress state. + """ + obs = _make_observer_simple(tmp_path) + import observer.observer as obs_mod + _topology_with(obs, "vps") + + obs.world_state["nodes"]["vps"] = { + "status": "online", "last_seen": 1784000000, "roles": ["infra"], + } + obs.node_checkpoints["vps"] = 1784000000 + + vps_dir = obs_mod.EVENTS_DIR / "vps" + # Lexically HUGE name ("evt-zzzz…" > "evt-vps-…") but an OLDER epoch. + _write_event( + vps_dir, "evt-zzzzzzzz-1000000000-node_health-node.json", + node="vps", ts=1000000000, + ) + obs.run_once() + # Skipped by ts (1e9 < checkpoint) despite the larger path → last_seen intact. + assert obs.world_state["nodes"]["vps"]["last_seen"] == 1784000000 + assert obs.node_checkpoints["vps"] == 1784000000 + + # A genuinely newer event advances the checkpoint by ts. + _write_event( + vps_dir, "evt-vps-1785000000-node_health-node.json", + node="vps", ts=1785000000, + ) + obs.run_once() + assert obs.world_state["nodes"]["vps"]["last_seen"] == 1785000000 + assert obs.node_checkpoints["vps"] == 1785000000 + + +def test_migration_path_checkpoint_to_timestamp(tmp_path): + """A pre-fix checkpoint file holding PATH strings migrates to int epochs.""" + obs = _make_observer_simple(tmp_path) # constructs to establish paths + import observer.observer as obs_mod + + old_path = str( + obs_mod.EVENTS_DIR / "piha" + / "evt-unknown-1781254800-ha_update_available-homeassistant-951.json" + ) + obs_mod.OBSERVER_STATE_FILE.write_text(json.dumps({ + "node_checkpoints": { + "piha": old_path, # old lexical path string + "vps": 1785000000, # already a timestamp int + "solaria": "junk-no-epoch-here.json", # unparseable → reprocess (0) + } + })) + + migrated = Observer() # re-loads the checkpoint we just wrote + assert migrated.node_checkpoints["piha"] == 1781254800 + assert migrated.node_checkpoints["vps"] == 1785000000 + assert migrated.node_checkpoints["solaria"] == 0 + + +def test_migration_old_single_file_checkpoint(tmp_path): + """The very old {"last_processed_file": } format also migrates.""" + obs = _make_observer_simple(tmp_path) + import observer.observer as obs_mod + + old = str(obs_mod.EVENTS_DIR / "piha" / "evt-piha-1784000000-node_health-node.json") + obs_mod.OBSERVER_STATE_FILE.write_text(json.dumps({"last_processed_file": old})) + + migrated = Observer() + assert migrated.node_checkpoints == {"piha": 1784000000} + + +def test_ts_helpers_units(): + """Unit-level coverage of the timestamp parsing helpers.""" + assert _ts_from_event_name("evt-piha-1784000000-node_health-node.json") == 1784000000 + assert _ts_from_event_name( + "evt-unknown-1781254800-ha_update_available-homeassistant-951.json" + ) == 1781254800 + assert _ts_from_event_name("no-epoch-here.json") is None + + assert _checkpoint_ts_from_value(1784000000) == 1784000000 + assert _checkpoint_ts_from_value(1784000000.9) == 1784000000 + assert _checkpoint_ts_from_value( + "/opt/homelab/events/piha/evt-piha-1784000000-node_health-node.json" + ) == 1784000000 + assert _checkpoint_ts_from_value("garbage") == 0 + assert _checkpoint_ts_from_value(None) == 0 + assert _checkpoint_ts_from_value(True) == 0 # bool is an int subclass — excluded + + +def test_event_ts_from_path_mtime_fallback(tmp_path): + """A real file with a no-epoch name returns its mtime, never 0.""" + p = tmp_path / "weird-name.json" + p.write_text("{}") + ts = _event_ts_from_path(str(p)) + assert isinstance(ts, int) + assert ts > 0