Compare commits
13 Commits
997b5f053c
...
dev
| Author | SHA1 | Date | |
|---|---|---|---|
| 56d2027d08 | |||
| a92afd64cd | |||
| 302934ced6 | |||
| 3bef58b8dd | |||
| 3dadac4bcb | |||
| 7d22442bad | |||
| dae67afeb9 | |||
| 05429bdda3 | |||
| 5340332f32 | |||
| 4870f3ee87 | |||
| 76387ad17a | |||
| 4312efc15c | |||
| 0e2fe7c1e3 |
@@ -5,6 +5,38 @@ All notable changes to this project will be documented in this file.
|
||||
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
|
||||
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
||||
|
||||
## [0.4.1] - 2026-06-25
|
||||
|
||||
### Fixed
|
||||
- **Auto-updater no longer dies when settings are saved.** Saving the Settings dialog restarts the watcher thread via `_restart_watcher`, which sets and then replaces the shared `stop_event`. Because the updater loop keyed off that *same* event, saving settings would terminate the auto-updater thread (a race — it could survive one save and die on the next). The watcher kept running and reporting heartbeats normally, but the agent silently stopped checking for / applying updates until the tray app was relaunched. The updater now keys off its own app-lifetime event (`_app_stop`) that watcher restarts never touch; it stops only on Exit or when applying an update.
|
||||
- Added `test_thor_tray.py` covering the tray thread lifecycle (updater survives a watcher restart; stops on app exit; `_exit`/`_do_update` signal the lifetime event).
|
||||
|
||||
## [0.4.0] - 2026-06-25
|
||||
|
||||
### Added — Mirror (dual-send)
|
||||
- **Optional best-effort mirror server.** Each heartbeat and each forwarded event can now be sent to a *second* ("mirror") server in addition to the primary, so a standby box (e.g. the office NAS) stays a continuous replica during a migration — making the eventual cutover a non-event. **Default off** — blank `mirror_api_url`/`mirror_sfm_url` means existing installs are unchanged after auto-update.
|
||||
- **Heartbeat mirror (best-effort).** After the primary heartbeat POST, the same payload is fired at `mirror_api_url` when set. Bounded by the existing API timeout and fully wrapped — it can never delay or fail the primary heartbeat; its result is ignored except for a debug log line.
|
||||
- **Event mirror (reliable, isolated).** Events are re-forwarded to `mirror_sfm_url` using a *separate* sha256 state file (`mirror_sfm_state_file`, blank → `<log_dir>/thor_forwarded_mirror.json`), so the mirror tracks its own delivery independently of the primary. A quick reachability probe (~3 s) skips the mirror pass when the mirror server is down, so a dead mirror never stalls the loop on per-event timeouts; skipped events stay pending in the mirror state and deliver when the mirror returns — **no data loss**.
|
||||
- **Isolation invariant (the one rule).** Nothing on the mirror path can delay or fail the primary: its own state file, its own try/except, the reachability guard + bounded timeouts, all exceptions swallowed-and-logged. The primary path stays exactly as reliable as before.
|
||||
- **Mirror tab in the Settings dialog** with `Mirror API URL` + `Mirror SFM URL` fields (blank = off). The mirror rides along with the primary — there is no separate enable flag (heartbeat-mirror active iff the primary heartbeat is on *and* `mirror_api_url` is set; event-mirror active iff primary forwarding is on *and* `mirror_sfm_url` is set).
|
||||
- New unit tests in `test_event_forwarder.py` covering mirror reachability, the down-mirror skip, separate-state idempotency, and the isolation invariant (a mirror failure leaves the primary's result + state untouched).
|
||||
|
||||
### Configuration
|
||||
|
||||
New `config.json` keys (all default-empty = off; existing 0.3.x deployments don't change behaviour on auto-update):
|
||||
|
||||
| Key | Default | Notes |
|
||||
|---|---|---|
|
||||
| `mirror_api_url` | `""` | Second heartbeat destination — same URL form as `api_url` (blank = off) |
|
||||
| `mirror_sfm_url` | `""` | Second SFM base URL, e.g. `http://10.0.0.x:8200` (blank = off) |
|
||||
| `mirror_sfm_state_file` | `""` | Override the mirror's forwarded-sha256 state file. Blank → `<log_dir>\thor_forwarded_mirror.json` |
|
||||
|
||||
### Changed
|
||||
- Bumped `VERSION` to `0.4.0`
|
||||
|
||||
> **Operational note (migration seeding).** To make the mirror re-deliver only the gap since a one-time replica snapshot, copy the primary state file to the mirror state file and drop entries whose `forwarded_at` is after the snapshot. SFM-side dedup covers any overlap. See `docs/mirror-dual-send-design.md`.
|
||||
|
||||
|
||||
## [0.3.1] - 2026-05-20
|
||||
|
||||
### Hot fix
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
# Thor Watcher
|
||||
|
||||
**Version:** 0.3.1
|
||||
**Version:** 0.4.1
|
||||
|
||||
Micromate (Series 4) watcher agent for Terra-View fleet management. Runs as a Windows system tray application, scans THORDATA for Micromate unit activity, sends heartbeat data to Terra-View, and (optionally) forwards `.IDFH`/`.IDFW` event files to a seismo-relay SFM server.
|
||||
Micromate (Series 4) watcher agent for Terra-View fleet management. Runs as a Windows system tray application, scans THORDATA for Micromate unit activity, sends heartbeat data to Terra-View, and (optionally) forwards `.IDFH`/`.IDFW` event files to a seismo-relay SFM server. Heartbeats and events can optionally be **mirrored** to a second server (e.g. a standby/NAS) in addition to the primary.
|
||||
|
||||
---
|
||||
|
||||
@@ -29,7 +29,7 @@ build.bat
|
||||
```
|
||||
|
||||
Produces:
|
||||
- `dist\thor-watcher-0.3.1.exe` — upload to Gitea release
|
||||
- `dist\thor-watcher-0.4.1.exe` — upload to Gitea release
|
||||
- `dist\thor-watcher.exe` — use with Inno Setup
|
||||
|
||||
Then run Inno Setup Compiler on `installer.iss` to produce `thor-watcher-setup.exe`.
|
||||
@@ -71,6 +71,9 @@ Managed through the Settings dialog (right-click tray icon → Settings). A `con
|
||||
| `sfm_state_file` | string | `""` | Path to the sha256-keyed state file. Blank → `<log_dir>\thor_forwarded.json` |
|
||||
| `sfm_max_forwards_per_pass` | integer | `500` | Cap per pass to drip-feed large backfills |
|
||||
| `sfm_max_event_age_days` | integer | `365` | Skip event files older than this many days |
|
||||
| `mirror_api_url` | string | `""` | Second heartbeat destination (same URL form as `api_url`). Blank → heartbeat mirror off |
|
||||
| `mirror_sfm_url` | string | `""` | Second SFM base URL (e.g. `http://10.0.0.x:8200`). Blank → event mirror off |
|
||||
| `mirror_sfm_state_file` | string | `""` | Path to the mirror's sha256 state file. Blank → `<log_dir>\thor_forwarded_mirror.json` |
|
||||
|
||||
---
|
||||
|
||||
@@ -86,6 +89,21 @@ When `sfm_forward_enabled` is true and `sfm_url` is set, Thor Watcher walks the
|
||||
|
||||
---
|
||||
|
||||
## Mirror (Dual-Send)
|
||||
|
||||
Thor Watcher can post every heartbeat and event to a **second ("mirror") server** in addition to the primary — useful for keeping a standby box (e.g. the office NAS) as a continuous replica during a server migration, so the eventual cutover is a non-event. **Default off.**
|
||||
|
||||
The mirror rides along with the primary — there is no separate enable flag:
|
||||
|
||||
- **Heartbeat mirror** is active when the API heartbeat is enabled **and** `mirror_api_url` is set. After the primary heartbeat POST, the same payload is fired at `mirror_api_url`.
|
||||
- **Event mirror** is active when SFM forwarding is enabled **and** `mirror_sfm_url` is set. Events are re-forwarded to `mirror_sfm_url` using a **separate** sha256 state file (`mirror_sfm_state_file`, blank → `<log_dir>\thor_forwarded_mirror.json`).
|
||||
|
||||
**The mirror can never delay or fail the primary.** The heartbeat mirror is fully wrapped and bounded by the API timeout. The event mirror runs a quick reachability probe (~3 s) first and skips the pass if the mirror server is down — skipped events stay pending in the mirror state and deliver when it returns, so nothing is lost. All mirror exceptions are swallowed-and-logged; the primary path stays exactly as reliable as before.
|
||||
|
||||
- **Seeding for a migration.** To re-deliver only the gap since a one-time replica snapshot, copy the primary state file to the mirror state file and drop entries whose `forwarded_at` is after the snapshot. SFM-side dedup covers any overlap. See [`docs/mirror-dual-send-design.md`](docs/mirror-dual-send-design.md).
|
||||
|
||||
---
|
||||
|
||||
## Tray Icon Colors
|
||||
|
||||
| Color | Meaning |
|
||||
@@ -121,7 +139,7 @@ Posted to `api_url` on each API interval:
|
||||
{
|
||||
"source_id": "THOR-PC",
|
||||
"source_type": "series4_watcher",
|
||||
"version": "0.3.1",
|
||||
"version": "0.4.1",
|
||||
"generated_at": "2026-03-20T14:30:00Z",
|
||||
"log_tail": ["...last 25 log lines..."],
|
||||
"units": [
|
||||
|
||||
+5
-1
@@ -24,5 +24,9 @@
|
||||
"sfm_http_timeout": 60,
|
||||
"sfm_state_file": "",
|
||||
"sfm_max_forwards_per_pass": 500,
|
||||
"sfm_max_event_age_days": 365
|
||||
"sfm_max_event_age_days": 365,
|
||||
|
||||
"mirror_api_url": "",
|
||||
"mirror_sfm_url": "",
|
||||
"mirror_sfm_state_file": ""
|
||||
}
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
# Watcher Mirror (Dual-Send) — Design
|
||||
|
||||
**Date:** 2026-06-24 · **Branch:** `feat/mirror-dual-send` (both `thor-watcher` + `series3-watcher`) · **Status:** approved, implementing
|
||||
|
||||
## Goal
|
||||
Let each watcher post every heartbeat and event to a **second ("mirror") server** in addition to the primary, so the office NAS stays a continuous replica of prod during the migration. The eventual cutover becomes a non-event. **Default off** → existing installs unchanged.
|
||||
|
||||
## Pattern
|
||||
Classic dual-write migration. The primary path stays **exactly as reliable as today**; the mirror is pure best-effort bonus that can **never delay or fail** the primary.
|
||||
|
||||
## The two watchers (parallel changes)
|
||||
| | thor-watcher (series4) | series3-watcher (s3) |
|
||||
|---|---|---|
|
||||
| Config | `config.json` (JSON) | `config.ini` (`[agent]`, INI) |
|
||||
| Heartbeat fn | `send_api_payload(payload, api_url, timeout)` | `send_api_payload(payload, api_url)` |
|
||||
| Event fn | `event_forwarder.forward_pending(root, sfm_url, state, …)` | same (`event_forwarder` is a sibling port) |
|
||||
| Loop | `series4_ingest.run_watcher` | `series3_watcher.run_watcher` |
|
||||
| Settings GUI | `thor_settings_dialog.py` | `settings_dialog.py` |
|
||||
|
||||
## Design (Approach A — reliable event mirror, best-effort heartbeat)
|
||||
|
||||
### Config additions (default empty = off)
|
||||
- thor (`config.json`): `mirror_api_url`, `mirror_sfm_url`, `mirror_sfm_state_file` (blank → `<log_dir>/thor_forwarded_mirror.json`).
|
||||
- s3 (`config.ini`): `MIRROR_API_URL`, `MIRROR_SFM_URL`, `MIRROR_SFM_STATE_FILE` (blank → `sfm_forwarded_mirror.json`).
|
||||
|
||||
The mirror **rides along** with the primary: heartbeat-mirror active iff primary heartbeat on **and** `mirror_api_url` set; event-mirror active iff primary forwarding on **and** `mirror_sfm_url` set. No separate enable flag.
|
||||
|
||||
### Heartbeat mirror (best-effort)
|
||||
After the primary `send_api_payload`, if `mirror_api_url` is set, fire one more `send_api_payload(payload, mirror_api_url, …)`. Wrapped so it can never raise into the loop; result ignored except a debug log. Bounded by the existing short API timeout (~5 s).
|
||||
|
||||
### Event mirror (reliable, isolated)
|
||||
- Init a **second** `ForwardState(mirror_state_path)` alongside the primary state.
|
||||
- Each forward tick, **after** the primary `forward_pending`, run a second `forward_pending(root, mirror_sfm_url, mirror_state, …)` in its **own** try/except.
|
||||
- **Reachability guard:** before the mirror forward, a quick TCP/HTTP probe of `mirror_sfm_url` (~3 s). If unreachable → skip this pass, log, retry next tick. Prevents a down NAS from stalling the loop on per-event timeouts. The mirror state file means skipped events stay pending → delivered when the NAS returns. **No data loss.**
|
||||
- Reliable + idempotent via its own sha256 state, fully independent of the primary's state.
|
||||
|
||||
### Isolation invariant (the one rule)
|
||||
Nothing on the mirror path can delay or fail the primary: separate state file, its own try/except, the reachability guard + bounded timeouts, all exceptions swallowed-and-logged.
|
||||
|
||||
### Settings dialog
|
||||
Add `Mirror API URL` + `Mirror SFM URL` fields (blank = off) to each watcher's settings dialog.
|
||||
|
||||
## Seeding (operational, for this migration)
|
||||
The mirror state file should reflect what the NAS already has (everything ≤ the migration snapshot, ~16:23 on 2026-06-24). Recipe: copy the primary state file → mirror state file, then drop entries whose `forwarded_at` is after the snapshot, so the mirror re-delivers exactly the gap. SFM-side dedup covers any overlap. (If a watcher had no events in the gap, a straight copy suffices.)
|
||||
|
||||
## Testing
|
||||
- thor `test_event_forwarder.py`: mirror uses a separate state file (idempotent, independent); a mirror failure leaves the primary's result + state untouched; reachability guard skips cleanly when the mirror is down.
|
||||
- s3: add a minimal test for the same isolation/idempotency.
|
||||
|
||||
## Out of scope / handoff
|
||||
Rebuilding the Windows installers + redeploying (s3 on the Win7 box, thor on the Win10 box) is the operator's job — this change is **code + tests only**.
|
||||
@@ -692,6 +692,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) ────────────────
|
||||
|
||||
|
||||
|
||||
+1
-1
@@ -3,7 +3,7 @@
|
||||
|
||||
[Setup]
|
||||
AppName=Thor Watcher
|
||||
AppVersion=0.3.1
|
||||
AppVersion=0.4.1
|
||||
AppPublisher=Terra-Mechanics Inc.
|
||||
DefaultDirName={pf}\ThorWatcher
|
||||
DefaultGroupName=Thor Watcher
|
||||
|
||||
+61
-2
@@ -1,5 +1,5 @@
|
||||
"""
|
||||
Thor Watcher — Series 4 Ingest Agent v0.3.1
|
||||
Thor Watcher — Series 4 Ingest Agent v0.4.1
|
||||
|
||||
Micromate (Series 4) ingest agent for Terra-View.
|
||||
|
||||
@@ -29,7 +29,7 @@ import event_forwarder
|
||||
|
||||
# ── Version ───────────────────────────────────────────────────────────────────
|
||||
|
||||
VERSION = "0.3.1"
|
||||
VERSION = "0.4.1"
|
||||
|
||||
|
||||
# ── Config ────────────────────────────────────────────────────────────────────
|
||||
@@ -69,6 +69,15 @@ def load_config(config_path: str) -> Dict[str, Any]:
|
||||
"sfm_state_file": "", # blank → <log_dir>/thor_forwarded.json
|
||||
"sfm_max_forwards_per_pass": 500,
|
||||
"sfm_max_event_age_days": 365,
|
||||
|
||||
# Mirror (dual-send) — best-effort second destination, default OFF.
|
||||
# See docs/mirror-dual-send-design.md. Heartbeat-mirror fires iff
|
||||
# api_url AND mirror_api_url are set; event-mirror runs iff SFM
|
||||
# forwarding is on AND mirror_sfm_url is set. Own state file →
|
||||
# idempotency independent of the primary.
|
||||
"mirror_api_url": "",
|
||||
"mirror_sfm_url": "",
|
||||
"mirror_sfm_state_file": "", # blank → <log_dir>/thor_forwarded_mirror.json
|
||||
}
|
||||
|
||||
with open(config_path, "r", encoding="utf-8") as f:
|
||||
@@ -329,6 +338,12 @@ def run_watcher(state: Dict[str, Any], stop_event: threading.Event) -> None:
|
||||
sfm_state_path = str(cfg.get("sfm_state_file", "")).strip() or \
|
||||
os.path.join(state["log_dir"], "thor_forwarded.json")
|
||||
|
||||
# Mirror (dual-send) config
|
||||
MIRROR_API_URL = str(cfg.get("mirror_api_url", "")).strip()
|
||||
MIRROR_SFM_URL = str(cfg.get("mirror_sfm_url", "")).strip()
|
||||
mirror_state_path = str(cfg.get("mirror_sfm_state_file", "")).strip() or \
|
||||
os.path.join(state["log_dir"], "thor_forwarded_mirror.json")
|
||||
|
||||
log_message(log_file, ENABLE_LOGGING,
|
||||
"[cfg] THORDATA_PATH={} SCAN_INTERVAL={}s API_INTERVAL={}s API={} SFM={}".format(
|
||||
THORDATA_PATH, SCAN_INTERVAL, API_INTERVAL, bool(API_URL),
|
||||
@@ -362,6 +377,24 @@ def run_watcher(state: Dict[str, Any], stop_event: threading.Event) -> None:
|
||||
else:
|
||||
state["sfm_status"] = "disabled"
|
||||
|
||||
# Mirror event-forward state — independent of the primary's. Created
|
||||
# only when the primary forwarder is on AND a mirror URL is set.
|
||||
mirror_state_obj: Optional[event_forwarder.ForwardState] = None
|
||||
if sfm_state_obj is not None and MIRROR_SFM_URL:
|
||||
try:
|
||||
mirror_state_obj = event_forwarder.ForwardState(mirror_state_path)
|
||||
log_message(log_file, ENABLE_LOGGING,
|
||||
"[mirror] event mirror ready url={} state={} known={}".format(
|
||||
MIRROR_SFM_URL, mirror_state_path, mirror_state_obj.count()))
|
||||
print("[MIRROR] event mirror ready url={}".format(MIRROR_SFM_URL))
|
||||
except Exception as exc:
|
||||
log_message(log_file, ENABLE_LOGGING,
|
||||
"[mirror] event mirror init failed: {}".format(exc))
|
||||
mirror_state_obj = None
|
||||
if MIRROR_API_URL:
|
||||
log_message(log_file, ENABLE_LOGGING,
|
||||
"[mirror] heartbeat mirror enabled url={}".format(MIRROR_API_URL))
|
||||
|
||||
state["last_forward"] = None
|
||||
state["last_forward_counts"] = None
|
||||
|
||||
@@ -419,6 +452,13 @@ def run_watcher(state: Dict[str, Any], stop_event: threading.Event) -> None:
|
||||
payload["log_tail"] = _read_log_tail(log_file, 25)
|
||||
response = send_api_payload(payload, API_URL, API_TIMEOUT)
|
||||
last_api_ts = now_ts
|
||||
# Best-effort heartbeat mirror — never affects primary.
|
||||
if MIRROR_API_URL:
|
||||
try:
|
||||
send_api_payload(payload, MIRROR_API_URL, API_TIMEOUT)
|
||||
except Exception as exc:
|
||||
log_message(log_file, ENABLE_LOGGING,
|
||||
"[mirror] heartbeat post failed: {}".format(exc))
|
||||
if response is not None:
|
||||
state["api_status"] = "ok"
|
||||
if response.get("update_available"):
|
||||
@@ -459,6 +499,25 @@ def run_watcher(state: Dict[str, Any], stop_event: threading.Event) -> None:
|
||||
print(msg)
|
||||
log_message(log_file, ENABLE_LOGGING, msg)
|
||||
|
||||
# Best-effort event mirror — own state, fast-fail guard,
|
||||
# fully isolated. Runs after the primary each tick.
|
||||
if mirror_state_obj is not None:
|
||||
m_counts = event_forwarder.mirror_forward_pass(
|
||||
THORDATA_PATH, MIRROR_SFM_URL, mirror_state_obj,
|
||||
max_age_days=SFM_MAX_AGE_DAYS,
|
||||
quiescence_seconds=SFM_QUIESCENCE,
|
||||
missing_report_grace_seconds=SFM_GRACE,
|
||||
timeout=SFM_HTTP_TIMEOUT,
|
||||
max_per_pass=SFM_MAX_PER_PASS,
|
||||
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))
|
||||
|
||||
except Exception as e:
|
||||
err = "[loop-error] {}".format(e)
|
||||
print(err)
|
||||
|
||||
@@ -712,5 +712,134 @@ class TestForwardPending(unittest.TestCase):
|
||||
self.assertEqual(len(_FakeImportHandler.received), 2)
|
||||
|
||||
|
||||
# ── Mirror (dual-send) ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
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):
|
||||
# Nothing listens on 127.0.0.1:1 → connection refused → False,
|
||||
# and it must NOT hang (this is the whole point of the guard).
|
||||
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):
|
||||
# The fake server 501s on GET /health, but it's UP — so reachable.
|
||||
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 _one_event(self, root: Path) -> None:
|
||||
unit = _make_thordata(root, "Project A", "UM11719")
|
||||
_make_event(unit, "UM11719_20231219163444.IDFW", age_seconds=200, content=b"binary")
|
||||
_make_txt(unit, "UM11719_20231219163444.IDFW", age_seconds=100, content=b"report")
|
||||
|
||||
def test_empty_mirror_url_is_noop(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
self._one_event(root)
|
||||
mstate = ef.ForwardState(str(root / "mirror.json"))
|
||||
self.assertIsNone(ef.mirror_forward_pass(str(root), "", mstate, max_age_days=30))
|
||||
self.assertEqual(len(_FakeImportHandler.received), 0)
|
||||
|
||||
def test_skips_when_unreachable_without_posting(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
self._one_event(root)
|
||||
mstate = ef.ForwardState(str(root / "mirror.json"))
|
||||
result = ef.mirror_forward_pass(
|
||||
str(root), self.base_url, mstate,
|
||||
reachable_fn=lambda url, timeout=3.0: False, # force "down"
|
||||
max_age_days=30, quiescence_seconds=5,
|
||||
missing_report_grace_seconds=60, timeout=5.0,
|
||||
)
|
||||
self.assertIsNone(result)
|
||||
self.assertEqual(mstate.count(), 0) # nothing forwarded
|
||||
self.assertEqual(len(_FakeImportHandler.received), 0) # never POSTed
|
||||
|
||||
def test_forwards_when_reachable_using_its_own_state(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
self._one_event(root)
|
||||
mstate = ef.ForwardState(str(root / "mirror.json"))
|
||||
counts = ef.mirror_forward_pass(
|
||||
str(root), 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):
|
||||
# Even if forward_pending raises, the mirror swallows it → None.
|
||||
def _boom(*a, **k):
|
||||
raise RuntimeError("boom")
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
self._one_event(root)
|
||||
mstate = ef.ForwardState(str(root / "mirror.json"))
|
||||
orig = ef.forward_pending
|
||||
ef.forward_pending = _boom
|
||||
try:
|
||||
result = ef.mirror_forward_pass(
|
||||
str(root), 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):
|
||||
# The invariant: a real down mirror (via the real reachability
|
||||
# probe) is a fast no-op and the primary forward is unaffected.
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
self._one_event(root)
|
||||
|
||||
primary_state = ef.ForwardState(str(root / "primary.json"))
|
||||
pcounts = ef.forward_pending(
|
||||
str(root), 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(str(root / "mirror.json"))
|
||||
t0 = time.time()
|
||||
result = ef.mirror_forward_pass(
|
||||
str(root), "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) # fast-failed, didn't hang
|
||||
self.assertEqual(mstate.count(), 0)
|
||||
self.assertEqual(primary_state.count(), primary_snapshot) # untouched
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
"""
|
||||
Tests for thor_tray's thread lifecycle.
|
||||
|
||||
Regression guard for the bug where saving settings — which restarts the
|
||||
watcher thread via ``_restart_watcher`` — also killed the auto-updater
|
||||
thread, because both shared a single ``stop_event``. The updater must key
|
||||
off a separate lifetime event (``_app_stop``) that a watcher restart never
|
||||
disturbs; it should only stop when the app actually exits.
|
||||
"""
|
||||
import sys
|
||||
import time
|
||||
import threading
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
# Thor's tray pulls in Windows-only GUI deps at import time; stub them so the
|
||||
# module imports on a headless dev/CI box.
|
||||
for _name in ("pystray", "PIL", "PIL.Image", "PIL.ImageDraw"):
|
||||
sys.modules.setdefault(_name, mock.MagicMock())
|
||||
|
||||
import thor_tray # noqa: E402
|
||||
|
||||
|
||||
class TrayThreadLifecycle(unittest.TestCase):
|
||||
def setUp(self):
|
||||
# Keep _start_watcher cheap: a no-op watcher loop so restarting the
|
||||
# watcher doesn't spin up the real THORDATA scanner / network heartbeat.
|
||||
patcher = mock.patch.object(
|
||||
thor_tray.watcher, "run_watcher", lambda state, stop_event: None
|
||||
)
|
||||
patcher.start()
|
||||
self.addCleanup(patcher.stop)
|
||||
|
||||
def test_app_stop_is_separate_lifetime_event(self):
|
||||
app = thor_tray.WatcherTray()
|
||||
self.assertIsInstance(app._app_stop, threading.Event)
|
||||
self.assertIsNot(app._app_stop, app.stop_event)
|
||||
self.assertFalse(app._app_stop.is_set())
|
||||
|
||||
def test_exit_signals_app_stop(self):
|
||||
app = thor_tray.WatcherTray()
|
||||
icon = mock.MagicMock()
|
||||
app._exit(icon, None)
|
||||
self.assertTrue(app._app_stop.is_set())
|
||||
icon.stop.assert_called_once()
|
||||
|
||||
def test_restart_watcher_leaves_watcher_stop_unset(self):
|
||||
app = thor_tray.WatcherTray()
|
||||
app._restart_watcher()
|
||||
self.assertFalse(app.stop_event.is_set())
|
||||
|
||||
def test_updater_survives_watcher_restart_then_stops_on_exit(self):
|
||||
app = thor_tray.WatcherTray()
|
||||
app._icon = None
|
||||
with mock.patch.object(thor_tray, "check_for_update", return_value=(None, None)), \
|
||||
mock.patch.object(thor_tray, "_update_log"), \
|
||||
mock.patch.object(app, "_tray_status", return_value="ok"):
|
||||
t = threading.Thread(
|
||||
target=app._icon_updater, daemon=True, name="test-updater"
|
||||
)
|
||||
t.start()
|
||||
time.sleep(0.1)
|
||||
self.assertTrue(t.is_alive(), "updater failed to start")
|
||||
|
||||
# A settings save restarts the watcher — it must NOT kill the updater.
|
||||
app._restart_watcher()
|
||||
time.sleep(0.2)
|
||||
self.assertTrue(
|
||||
t.is_alive(), "updater thread died on watcher restart (the bug)"
|
||||
)
|
||||
|
||||
# Real app exit DOES stop the updater.
|
||||
app._app_stop.set()
|
||||
t.join(timeout=2)
|
||||
self.assertFalse(
|
||||
t.is_alive(), "updater thread did not stop on app exit"
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+68
-1
@@ -1,5 +1,5 @@
|
||||
"""
|
||||
Thor Watcher — Settings Dialog v0.3.1
|
||||
Thor Watcher — Settings Dialog v0.4.1
|
||||
|
||||
Provides a Tkinter settings dialog that doubles as a first-run wizard.
|
||||
|
||||
@@ -47,6 +47,11 @@ DEFAULTS = {
|
||||
"sfm_state_file": "",
|
||||
"sfm_max_forwards_per_pass": 500,
|
||||
"sfm_max_event_age_days": 365,
|
||||
|
||||
# Mirror (dual-send) — best-effort second destination, default OFF.
|
||||
"mirror_api_url": "",
|
||||
"mirror_sfm_url": "",
|
||||
"mirror_sfm_state_file": "",
|
||||
}
|
||||
|
||||
|
||||
@@ -220,6 +225,14 @@ class SettingsDialog:
|
||||
self.var_sfm_max_age_days = tk.StringVar(value=str(v.get("sfm_max_event_age_days", 365)))
|
||||
self.var_sfm_state_file = tk.StringVar(value=str(v.get("sfm_state_file", "")))
|
||||
|
||||
# Mirror (dual-send) — best-effort second destination
|
||||
raw_mirror = str(v.get("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=str(v.get("mirror_sfm_url", "")))
|
||||
self.var_mirror_sfm_state_file = tk.StringVar(value=str(v.get("mirror_sfm_state_file", "")))
|
||||
|
||||
# ── UI construction ───────────────────────────────────────────────────────
|
||||
|
||||
def _build_ui(self):
|
||||
@@ -245,6 +258,7 @@ class SettingsDialog:
|
||||
self._build_tab_scanning(nb)
|
||||
self._build_tab_logging(nb)
|
||||
self._build_tab_forwarding(nb)
|
||||
self._build_tab_mirror(nb)
|
||||
self._build_tab_updates(nb)
|
||||
|
||||
btn_frame = tk.Frame(outer)
|
||||
@@ -484,6 +498,49 @@ class SettingsDialog:
|
||||
finally:
|
||||
self._sfm_test_btn.config(state="normal")
|
||||
|
||||
def _build_tab_mirror(self, nb):
|
||||
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)
|
||||
|
||||
def browse_mirror_state():
|
||||
p = filedialog.asksaveasfilename(
|
||||
title="Select Mirror State File",
|
||||
defaultextension=".json",
|
||||
filetypes=[("JSON files", "*.json"), ("All files", "*.*")],
|
||||
initialfile=os.path.basename(
|
||||
self.var_mirror_sfm_state_file.get() or "thor_forwarded_mirror.json"),
|
||||
initialdir=os.path.dirname(self.var_mirror_sfm_state_file.get() or "C:\\"),
|
||||
)
|
||||
if p:
|
||||
self.var_mirror_sfm_state_file.set(p.replace("/", "\\"))
|
||||
|
||||
_add_label_browse_entry(f, 3, "Mirror State File", self.var_mirror_sfm_state_file,
|
||||
browse_mirror_state)
|
||||
|
||||
help_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"
|
||||
"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>\\thor_forwarded_mirror.json."
|
||||
)
|
||||
tk.Label(f, text=help_text, justify="left", fg="#555555", wraplength=420).grid(
|
||||
row=4, column=0, columnspan=2, sticky="w", padx=(8, 8), pady=(8, 4)
|
||||
)
|
||||
|
||||
def _build_tab_updates(self, nb):
|
||||
f = self._tab_frame(nb, "Updates")
|
||||
|
||||
@@ -599,6 +656,12 @@ class SettingsDialog:
|
||||
sfm_url = ""
|
||||
sfm_url = sfm_url.rstrip("/") # event_forwarder adds the endpoint path
|
||||
|
||||
# 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/series4/heartbeat"
|
||||
mirror_sfm_url = self.var_mirror_sfm_url.get().strip().rstrip("/")
|
||||
|
||||
values = {
|
||||
"thordata_path": self.var_thordata_path.get().strip(),
|
||||
"scan_interval": int_values["Scan Interval"],
|
||||
@@ -623,6 +686,10 @@ class SettingsDialog:
|
||||
"sfm_max_forwards_per_pass": int_values["Max Forwards Per Pass"],
|
||||
"sfm_max_event_age_days": int_values["Max Event Age (days)"],
|
||||
"sfm_state_file": self.var_sfm_state_file.get().strip(),
|
||||
|
||||
"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:
|
||||
|
||||
+7
-4
@@ -1,5 +1,5 @@
|
||||
"""
|
||||
Thor Watcher — System Tray Launcher v0.3.1
|
||||
Thor Watcher — System Tray Launcher v0.4.1
|
||||
Requires: pystray, Pillow, tkinter (stdlib)
|
||||
|
||||
Run with: pythonw thor_tray.py (no console window)
|
||||
@@ -334,7 +334,8 @@ def _show_cancel_message():
|
||||
class WatcherTray:
|
||||
def __init__(self):
|
||||
self.state = {}
|
||||
self.stop_event = threading.Event()
|
||||
self.stop_event = threading.Event() # watcher lifecycle; replaced on restart
|
||||
self._app_stop = threading.Event() # app lifetime; only set on exit/update
|
||||
self._watcher_thread = None
|
||||
self._icon = None
|
||||
self._menu_lock = threading.Lock()
|
||||
@@ -390,6 +391,7 @@ class WatcherTray:
|
||||
subprocess.Popen(["explorer", HERE])
|
||||
|
||||
def _exit(self, icon, item):
|
||||
self._app_stop.set()
|
||||
self.stop_event.set()
|
||||
icon.stop()
|
||||
|
||||
@@ -478,7 +480,7 @@ class WatcherTray:
|
||||
# line in the log soon after startup; subsequent checks every ~5 min.
|
||||
update_check_counter = 27
|
||||
|
||||
while not self.stop_event.is_set():
|
||||
while not self._app_stop.is_set():
|
||||
icon_status = self._tray_status()
|
||||
|
||||
if self._icon is not None:
|
||||
@@ -504,7 +506,7 @@ class WatcherTray:
|
||||
self._do_update(url)
|
||||
return
|
||||
|
||||
self.stop_event.wait(timeout=10)
|
||||
self._app_stop.wait(timeout=10)
|
||||
|
||||
def _do_update(self, download_url=None):
|
||||
"""Notify tray then apply update. If url is None, fetch it first."""
|
||||
@@ -519,6 +521,7 @@ class WatcherTray:
|
||||
|
||||
success = apply_update(download_url)
|
||||
if success:
|
||||
self._app_stop.set()
|
||||
self.stop_event.set()
|
||||
if self._icon is not None:
|
||||
self._icon.stop()
|
||||
|
||||
Reference in New Issue
Block a user