serversdown/series3-watcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

feat: best-effort mirror (dual-send) of heartbeats + events

Browse Source 
Optional second destination so each heartbeat and event is posted to a
mirror server (the office NAS) alongside the primary — dual-write to
de-risk the migration. Default off (blank mirror URLs); existing installs
unchanged.

- event_forwarder: mirror_reachable() fast-fail probe + mirror_forward_pass()
  (reliable event mirror with its OWN state file, total exception isolation,
  never raises into the primary path).
- series3_watcher: MIRROR_API_URL/MIRROR_SFM_URL/MIRROR_SFM_STATE_FILE config;
  best-effort heartbeat mirror; isolated event-mirror pass after the primary.
- settings dialog + config-template: new 'Mirror' tab / keys.
- 8 new tests incl. the isolation invariant. 44 passing.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
serversdown
2026-06-24 23:16:47 +00:00
parent 730f215e23
commit 4adaa09d75
5 changed files with 337 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config-template.ini
+11
View File
@@ -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 <log dir>/sfm_forwarded_mirror.json
MIRROR_SFM_STATE_FILE =
event_forwarder.py
+68
View File
@@ -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 ``<base>/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) ────────────────
series3_watcher.py
+60
View File
@@ -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"
settings_dialog.py
+76
View File
@@ -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 <log dir>/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 <sfm_url>/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:
test_event_forwarder.py
+122
View File
@@ -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()