diff --git a/config-template.ini b/config-template.ini index 45799c9..29b5415 100644 --- a/config-template.ini +++ b/config-template.ini @@ -72,3 +72,14 @@ SFM_STATE_FILE = # `--seed-state` workflow that skips the historical backfill entirely. SFM_MAX_FORWARDS_PER_PASS = 500 +# --- Mirror (dual-send) --- +# Optional best-effort second destination: post each heartbeat and event +# to a mirror server (e.g. the office NAS) IN ADDITION to the primary +# above. Default off (blank URLs). The mirror can never delay or fail +# the primary; the event mirror keeps its own state file so nothing is +# lost while it's down. See docs/mirror-dual-send-design.md. +MIRROR_API_URL = ; base URL e.g. http://10.0.0.x:8001 (blank = off) +MIRROR_SFM_URL = ; base URL e.g. http://10.0.0.x:8200 (blank = off) +# Blank → defaults to /sfm_forwarded_mirror.json +MIRROR_SFM_STATE_FILE = + diff --git a/event_forwarder.py b/event_forwarder.py index cbc6d2b..a01a025 100644 --- a/event_forwarder.py +++ b/event_forwarder.py @@ -684,6 +684,74 @@ def forward_pending( return counts +# ── Mirror (dual-send) ──────────────────────────────────────────────────────── + + +def mirror_reachable(base_url: str, timeout: float = 3.0) -> bool: + """Quick liveness probe of a mirror SFM base URL. + + Run before a mirror forward pass so an unreachable mirror can't stall + the watcher loop on a per-event HTTP timeout for every pending file. + Probes ``/health`` (the SFM server exposes it). ANY HTTP + response — even 404/500 — means the server is up, so proceed; only a + connection-level failure (refused / timed out / DNS) counts as down. + Best-effort: never raises. + """ + if not base_url: + return False + url = base_url.rstrip("/") + "/health" + try: + with urllib.request.urlopen( + urllib.request.Request(url, method="GET"), timeout=timeout + ): + return True + except urllib.error.HTTPError: + return True # got a status back → server is alive + except Exception: + return False # refused / timeout / DNS / socket → down + + +def mirror_forward_pass( + watch_dir: str, + mirror_url: str, + mirror_state: ForwardState, + *, + reachable_fn=mirror_reachable, + reachable_timeout: float = 3.0, + **forward_kwargs: Any, +) -> Optional[Dict[str, int]]: + """One best-effort forwarding pass against a *mirror* SFM server. + + The dual-send entry point. Wraps :func:`forward_pending` with two + properties that keep the mirror from ever harming the primary path: + + 1. **Fast-fail reachability guard** — probe ``mirror_url`` first; if + it's down, return ``None`` immediately rather than let + :func:`forward_pending` block on a per-event timeout for every + pending file. + 2. **Total exception isolation** — any error (probe, forward, state + I/O) is swallowed and reported as ``None``. This function NEVER + raises, so the caller's already-completed primary forward is + untouched. + + The mirror keeps its OWN ``ForwardState`` file, so its idempotency + and retry are independent of the primary's. Nothing is lost while + the mirror is down — skipped events stay pending and are delivered + on a later pass once it returns. + + Returns the :func:`forward_pending` counts dict on a completed pass, + or ``None`` if the mirror was unreachable or the pass raised. + """ + try: + if not mirror_url: + return None + if not reachable_fn(mirror_url, reachable_timeout): + return None + return forward_pending(watch_dir, mirror_url, mirror_state, **forward_kwargs) + except Exception: + return None + + # ── Seed-state mode (skip historical backfill on first deploy) ──────────────── diff --git a/series3_watcher.py b/series3_watcher.py index 9d7eb98..bb4e420 100644 --- a/series3_watcher.py +++ b/series3_watcher.py @@ -110,6 +110,12 @@ def load_config(path: str) -> Dict[str, Any]: # first deploy in a folder that's been accumulating for years. # See README "First-time deployment" section. "SFM_MAX_FORWARDS_PER_PASS": get_int("SFM_MAX_FORWARDS_PER_PASS", 500), + + # Mirror (dual-send) — best-effort second destination, default OFF. + # See docs/mirror-dual-send-design.md. + "MIRROR_API_URL": get_str("MIRROR_API_URL", ""), + "MIRROR_SFM_URL": get_str("MIRROR_SFM_URL", ""), + "MIRROR_SFM_STATE_FILE": get_str("MIRROR_SFM_STATE_FILE", ""), } @@ -427,6 +433,33 @@ def run_watcher(state: Dict[str, Any], stop_event: threading.Event) -> None: else: print("[CFG] SFM_FORWARD_ENABLED=false (event forwarding disabled)") + # ---- Mirror (dual-send) setup ---- + # Best-effort second destination. The event mirror gets its OWN state + # file so its idempotency is independent of the primary's. + MIRROR_API_URL = cfg.get("MIRROR_API_URL", "") + MIRROR_SFM_URL = cfg.get("MIRROR_SFM_URL", "") + mirror_state = None + if sfm_state is not None and MIRROR_SFM_URL: + try: + from event_forwarder import ForwardState + mirror_state_file = cfg.get("MIRROR_SFM_STATE_FILE") or os.path.join( + os.path.dirname(LOG_FILE) or here, "sfm_forwarded_mirror.json" + ) + mirror_state = ForwardState(mirror_state_file) + print("[CFG] mirror event forwarding → {} state={} ({} known)".format( + MIRROR_SFM_URL, mirror_state_file, mirror_state.count())) + log_message(LOG_FILE, ENABLE_LOGGING, + "[mirror] event mirror enabled url={} state={} known={}".format( + MIRROR_SFM_URL, mirror_state_file, mirror_state.count())) + except Exception as e: + print("[WARN] mirror forwarder init failed: {}".format(e)) + log_message(LOG_FILE, ENABLE_LOGGING, + "[warn] mirror forwarder init failed: {}".format(e)) + mirror_state = None + if MIRROR_API_URL: + log_message(LOG_FILE, ENABLE_LOGGING, + "[mirror] heartbeat mirror enabled url={}".format(MIRROR_API_URL)) + while not stop_event.is_set(): try: now_local = datetime.now().isoformat() @@ -497,6 +530,13 @@ def run_watcher(state: Dict[str, Any], stop_event: threading.Event) -> None: hb_payload["log_tail"] = _read_log_tail(cfg.get("LOG_FILE", ""), 25) response = send_api_payload(hb_payload, cfg.get("API_URL", "")) last_api_ts = now_ts + # Best-effort heartbeat mirror — never affects primary. + if MIRROR_API_URL: + try: + send_api_payload(hb_payload, MIRROR_API_URL) + except Exception as e: + log_message(LOG_FILE, ENABLE_LOGGING, + "[mirror] heartbeat post failed: {}".format(e)) if response is not None: state["api_status"] = "ok" state["last_api"] = datetime.now() @@ -550,6 +590,26 @@ def run_watcher(state: Dict[str, Any], stop_event: threading.Event) -> None: print(err) log_message(LOG_FILE, ENABLE_LOGGING, err) state["sfm_status"] = "fail" + + # Best-effort event mirror — own state, fast-fail guard, + # fully isolated. Runs after the primary each tick. + if mirror_state is not None: + from event_forwarder import mirror_forward_pass + m_counts = mirror_forward_pass( + WATCH_PATH, MIRROR_SFM_URL, mirror_state, + max_age_days=MAX_EVENT_AGE_DAYS, + quiescence_seconds=int(cfg.get("SFM_QUIESCENCE_SECONDS", 5)), + missing_report_grace_seconds=int(cfg.get("SFM_MISSING_REPORT_GRACE_SECONDS", 60)), + timeout=int(cfg.get("SFM_HTTP_TIMEOUT", 60)), + max_per_pass=int(cfg.get("SFM_MAX_FORWARDS_PER_PASS", 500)), + logger=lambda m: log_message(LOG_FILE, ENABLE_LOGGING, "[mirror] " + m), + ) + if m_counts is None: + log_message(LOG_FILE, ENABLE_LOGGING, + "[mirror] forward skipped (unreachable/error): {}".format(MIRROR_SFM_URL)) + else: + log_message(LOG_FILE, ENABLE_LOGGING, + "[mirror] pass forwarded={forwarded} errors={errors}".format(**m_counts)) else: state["sfm_status"] = "disabled" diff --git a/settings_dialog.py b/settings_dialog.py index 0b22531..7255f04 100644 --- a/settings_dialog.py +++ b/settings_dialog.py @@ -53,6 +53,11 @@ DEFAULTS = { "SFM_HTTP_TIMEOUT": "60", "SFM_STATE_FILE": "", "SFM_MAX_FORWARDS_PER_PASS": "500", + + # Mirror (dual-send) — best-effort second destination, default OFF. + "MIRROR_API_URL": "", + "MIRROR_SFM_URL": "", + "MIRROR_SFM_STATE_FILE": "", } @@ -260,6 +265,14 @@ class SettingsDialog: self.var_sfm_state_file = tk.StringVar(value=v["SFM_STATE_FILE"]) self.var_sfm_max_per_pass = tk.StringVar(value=v["SFM_MAX_FORWARDS_PER_PASS"]) + # Mirror (dual-send) — best-effort second destination + _raw_mirror = v["MIRROR_API_URL"] + if _raw_mirror.endswith(_suffix): + _raw_mirror = _raw_mirror[:-len(_suffix)] + self.var_mirror_api_url = tk.StringVar(value=_raw_mirror) + self.var_mirror_sfm_url = tk.StringVar(value=v["MIRROR_SFM_URL"]) + self.var_mirror_sfm_state_file = tk.StringVar(value=v["MIRROR_SFM_STATE_FILE"]) + # --- UI construction --- def _build_ui(self): @@ -288,6 +301,7 @@ class SettingsDialog: self._build_tab_logging(nb) self._build_tab_updates(nb) self._build_tab_sfm(nb) + self._build_tab_mirror(nb) # Buttons btn_frame = tk.Frame(outer) @@ -566,6 +580,58 @@ class SettingsDialog: row=8, column=0, columnspan=2, sticky="w", padx=(8, 8), pady=(8, 4) ) + def _build_tab_mirror(self, nb): + """Optional dual-send mirror: a second heartbeat + event destination.""" + f = self._tab_frame(nb, "Mirror") + + intro = ( + "Optional dual-send: post each heartbeat and event to a SECOND\n" + "(\"mirror\") server in addition to the primary. Best-effort — the\n" + "mirror can never delay or fail the primary. Leave blank to disable." + ) + tk.Label(f, text=intro, justify="left", fg="#1a5276", wraplength=420).grid( + row=0, column=0, columnspan=2, sticky="w", padx=(8, 8), pady=(6, 8) + ) + + _add_label_entry(f, 1, "Mirror Terra-View URL", self.var_mirror_api_url) + _add_label_entry(f, 2, "Mirror SFM URL", self.var_mirror_sfm_url) + + tk.Label(f, text="Mirror State File", anchor="w").grid( + row=3, column=0, sticky="w", padx=(8, 4), pady=4 + ) + state_frame = tk.Frame(f) + state_frame.grid(row=3, column=1, sticky="ew", padx=(0, 8), pady=4) + state_frame.columnconfigure(0, weight=1) + ttk.Entry(state_frame, textvariable=self.var_mirror_sfm_state_file, width=32).grid( + row=0, column=0, sticky="ew" + ) + + def _browse_mirror_state(): + path = filedialog.asksaveasfilename( + title="Mirror forward-state file", + defaultextension=".json", + filetypes=[("JSON", "*.json"), ("All Files", "*.*")], + initialfile="sfm_forwarded_mirror.json", + ) + if path: + self.var_mirror_sfm_state_file.set(path) + + ttk.Button(state_frame, text="Browse...", width=10, + command=_browse_mirror_state).grid(row=0, column=1, padx=(4, 0)) + + hint_text = ( + "Mirror Terra-View URL → base like http://10.0.0.x:8001 (heartbeats).\n" + "Mirror SFM URL → base like http://10.0.0.x:8200 (events).\n\n" + "The event mirror keeps its OWN state file, so nothing is lost while\n" + "the mirror is down — pending events are delivered once it returns.\n" + "A quick reachability check skips the mirror cleanly when unreachable,\n" + "so the primary is never slowed.\n" + "State File blank → defaults to /sfm_forwarded_mirror.json." + ) + tk.Label(f, text=hint_text, justify="left", fg="#555555", wraplength=420).grid( + row=4, column=0, columnspan=2, sticky="w", padx=(8, 8), pady=(8, 4) + ) + def _test_sfm_connection(self): """GET /health and show the result.""" import urllib.request @@ -673,6 +739,12 @@ class SettingsDialog: else: api_url = api_url.rstrip("/") + "/api/series3/heartbeat" + # Mirror (dual-send) — best-effort second destination. + mirror_api_url = self.var_mirror_api_url.get().strip() + if mirror_api_url: + mirror_api_url = mirror_api_url.rstrip("/") + "/api/series3/heartbeat" + mirror_sfm_url = self.var_mirror_sfm_url.get().strip().rstrip("/") + values = { "API_ENABLED": "true" if self.var_api_enabled.get() else "false", "API_URL": api_url, @@ -698,6 +770,10 @@ class SettingsDialog: "SFM_HTTP_TIMEOUT": str(int_values["SFM HTTP Timeout"]), "SFM_STATE_FILE": self.var_sfm_state_file.get().strip(), "SFM_MAX_FORWARDS_PER_PASS": str(int_values["SFM Max Events Per Pass"]), + + "MIRROR_API_URL": mirror_api_url, + "MIRROR_SFM_URL": mirror_sfm_url, + "MIRROR_SFM_STATE_FILE": self.var_mirror_sfm_state_file.get().strip(), } try: diff --git a/test_event_forwarder.py b/test_event_forwarder.py index e425fe6..2387351 100644 --- a/test_event_forwarder.py +++ b/test_event_forwarder.py @@ -714,5 +714,127 @@ class TestForwardEventPair(unittest.TestCase): self.assertIn("serial=BE11529", req["path"]) +# ── Mirror (dual-send) ─────────────────────────────────────────────────────── + + +def _mk_mirror_event(watch_dir, name="M529LK44.AB0", age_seconds=200, content=b"binary"): + """Create a Blastware-shaped event binary with a controlled mtime.""" + p = Path(watch_dir) / name + p.write_bytes(content) + target = time.time() - age_seconds + os.utime(str(p), (target, target)) + return p + + +class TestMirrorReachable(unittest.TestCase): + + def test_empty_url_is_unreachable(self): + self.assertFalse(ef.mirror_reachable("", timeout=1.0)) + + def test_dead_port_is_unreachable_and_fast(self): + t0 = time.time() + self.assertFalse(ef.mirror_reachable("http://127.0.0.1:1", timeout=2.0)) + self.assertLess(time.time() - t0, 2.5) + + def test_any_http_response_counts_as_reachable(self): + server, base = _start_fake_server() + try: + self.assertTrue(ef.mirror_reachable(base, timeout=2.0)) + finally: + server.shutdown() + server.server_close() + + +class TestMirrorForwardPass(unittest.TestCase): + """The dual-send entry point: best-effort, isolated, own state.""" + + def setUp(self): + _FakeImportHandler.received = [] + self.server, self.base_url = _start_fake_server() + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + + def test_empty_mirror_url_is_noop(self): + with tempfile.TemporaryDirectory() as tmp: + _mk_mirror_event(tmp) + mstate = ef.ForwardState(os.path.join(tmp, "mirror.json")) + self.assertIsNone(ef.mirror_forward_pass(tmp, "", mstate, max_age_days=30)) + self.assertEqual(len(_FakeImportHandler.received), 0) + + def test_skips_when_unreachable_without_posting(self): + with tempfile.TemporaryDirectory() as tmp: + _mk_mirror_event(tmp) + mstate = ef.ForwardState(os.path.join(tmp, "mirror.json")) + result = ef.mirror_forward_pass( + tmp, self.base_url, mstate, + reachable_fn=lambda url, timeout=3.0: False, + max_age_days=30, quiescence_seconds=5, + missing_report_grace_seconds=60, timeout=5.0, + ) + self.assertIsNone(result) + self.assertEqual(mstate.count(), 0) + self.assertEqual(len(_FakeImportHandler.received), 0) + + def test_forwards_when_reachable_using_its_own_state(self): + with tempfile.TemporaryDirectory() as tmp: + _mk_mirror_event(tmp) + mstate = ef.ForwardState(os.path.join(tmp, "mirror.json")) + counts = ef.mirror_forward_pass( + tmp, self.base_url, mstate, + reachable_fn=lambda url, timeout=3.0: True, + max_age_days=30, quiescence_seconds=5, + missing_report_grace_seconds=60, timeout=5.0, + ) + self.assertIsNotNone(counts) + self.assertEqual(counts["forwarded"], 1) + self.assertEqual(mstate.count(), 1) + self.assertEqual(len(_FakeImportHandler.received), 1) + + def test_never_raises_when_forward_blows_up(self): + def _boom(*a, **k): + raise RuntimeError("boom") + with tempfile.TemporaryDirectory() as tmp: + _mk_mirror_event(tmp) + mstate = ef.ForwardState(os.path.join(tmp, "mirror.json")) + orig = ef.forward_pending + ef.forward_pending = _boom + try: + result = ef.mirror_forward_pass( + tmp, self.base_url, mstate, + reachable_fn=lambda url, timeout=3.0: True, + max_age_days=30, + ) + finally: + ef.forward_pending = orig + self.assertIsNone(result) + + def test_down_mirror_leaves_primary_state_untouched(self): + with tempfile.TemporaryDirectory() as tmp: + _mk_mirror_event(tmp) + + primary_state = ef.ForwardState(os.path.join(tmp, "primary.json")) + pcounts = ef.forward_pending( + tmp, self.base_url, primary_state, + max_age_days=30, quiescence_seconds=5, + missing_report_grace_seconds=60, timeout=5.0, + ) + self.assertEqual(pcounts["forwarded"], 1) + primary_snapshot = primary_state.count() + + mstate = ef.ForwardState(os.path.join(tmp, "mirror.json")) + t0 = time.time() + result = ef.mirror_forward_pass( + tmp, "http://127.0.0.1:1", mstate, # nothing listens + max_age_days=30, quiescence_seconds=5, + missing_report_grace_seconds=60, timeout=5.0, + ) + self.assertIsNone(result) + self.assertLess(time.time() - t0, 3.5) + self.assertEqual(mstate.count(), 0) + self.assertEqual(primary_state.count(), primary_snapshot) + + if __name__ == "__main__": unittest.main()