diff --git a/Dockerfile.test b/Dockerfile.test index 1966699..e4b6682 100644 --- a/Dockerfile.test +++ b/Dockerfile.test @@ -25,7 +25,7 @@ COPY tests ./tests # docker-compose.yml.j2 is here because the compose-generation test renders the # REAL template — a stub would assert nothing about what actually ships. COPY requirements-api.txt requirements-mcpunifier.txt docker-compose.yml.example docker-compose.yml.j2 run.sh ./ -COPY scripts/config_helper.py scripts/start.bat scripts/check_health.py scripts/healthcheck.sh scripts/verify_binaries.py ./scripts/ +COPY scripts/config_helper.py scripts/start.bat scripts/check_health.py scripts/healthcheck.sh scripts/verify_binaries.py scripts/wickworks-healthcheck.py ./scripts/ COPY assets/binaries.lock.json ./assets/ ENV PYTHONPATH=/app diff --git a/docker-compose.yml.example b/docker-compose.yml.example index 2bfaf58..069a6d9 100644 --- a/docker-compose.yml.example +++ b/docker-compose.yml.example @@ -36,6 +36,13 @@ services: # ONLY from the mt5 container (and from the Windows VM via the dockurr # gateway 20.20.20.1:8000). No ports published; nothing else on the # docker network can talk to it. + # + # netns-shared sidecars are NOT re-joined when the owning VM container is + # recreated: Docker leaves wickworks running in the old, now-orphaned + # network namespace, where its own loopback healthcheck still passes while + # the VM can no longer reach it (every TA call then 502s). This healthcheck + # detects the orphan and kills wickworks so `restart: unless-stopped` + # recreates it into the current mt5 netns. See scripts/wickworks-healthcheck.py. wickworks: image: psyb0t/wickworks:v0.3.1 restart: unless-stopped @@ -44,6 +51,17 @@ services: LOG_LEVEL: INFO MAX_BARS: "5000" MIN_BARS: "50" + volumes: + - ./scripts/wickworks-healthcheck.py:/wickworks-healthcheck.py:ro + healthcheck: + test: ["CMD", "python", "/wickworks-healthcheck.py"] + interval: 15s + # Must exceed the script's worst case (self-health probe + concurrent + # gateway sweep, each up to 2s), or Docker aborts the check before the + # orphan path can kill the process and the self-heal never fires. + timeout: 12s + start_period: 30s + retries: 3 depends_on: - mt5 diff --git a/docker-compose.yml.j2 b/docker-compose.yml.j2 index 64b31d2..e8f8428 100644 --- a/docker-compose.yml.j2 +++ b/docker-compose.yml.j2 @@ -72,6 +72,23 @@ services: LOG_LEVEL: INFO MAX_BARS: "5000" MIN_BARS: "50" + volumes: + - ./scripts/wickworks-healthcheck.py:/wickworks-healthcheck.py:ro + # netns-shared sidecars are NOT re-joined when the owning VM container is + # recreated: Docker leaves wickworks running in the old, now-orphaned + # network namespace, where its own loopback healthcheck still passes while + # the VM can no longer reach it (every TA call then 502s). This healthcheck + # detects the orphan and kills wickworks so `restart: unless-stopped` + # recreates it into the current mt5 netns. See scripts/wickworks-healthcheck.py. + healthcheck: + test: ["CMD", "python", "/wickworks-healthcheck.py"] + interval: 15s + # Must exceed the script's worst case (self-health probe + concurrent + # gateway sweep, each up to 2s), or Docker aborts the check before the + # orphan path can kill the process and the self-heal never fires. + timeout: 12s + start_period: 30s + retries: 3 depends_on: - {{ vm.service }} diff --git a/mt5api/backtest/handler.py b/mt5api/backtest/handler.py index e74d9e5..895ed5b 100644 --- a/mt5api/backtest/handler.py +++ b/mt5api/backtest/handler.py @@ -32,6 +32,7 @@ from mt5api.config import ( ACCOUNT, BROKER, + INSTANCE, LOG_DIR, TERMINAL_DIR, TERMINAL_PATH, @@ -254,21 +255,38 @@ def _tail(text, limit=DIAGNOSTIC_TAIL_CHARS): def _tail_terminal_log(lines=20): + """Tail of the terminal's most recently written run log. + + Picked by modification time, not by name. The logs are named `.log`, + but the directory also holds `metaeditor.log`, which sorts after every one + of them ("m" > "2") and never changes — so an alphabetical pick attached a + months-old compile tail to every failure message and hid the actual reason + the run died. + """ log_dir = os.path.join(TERMINAL_DIR, "logs") if not os.path.isdir(log_dir): return "" try: - candidates = sorted( - file_name for file_name in os.listdir(log_dir) if file_name.endswith(".log") - ) + candidates = [ + entry + for entry in os.scandir(log_dir) + if entry.is_file() + and entry.name.endswith(".log") + and entry.name.lower() != "metaeditor.log" + ] except OSError: return "" if not candidates: return "" - latest_path = os.path.join(log_dir, candidates[-1]) + try: + newest = max(candidates, key=lambda entry: entry.stat().st_mtime) + except OSError: + return "" + + latest_path = newest.path try: with open(latest_path, "r", encoding="utf-16-le", errors="replace") as handle: content = handle.read() @@ -281,22 +299,77 @@ def _tail_terminal_log(lines=20): return "\n".join(tail_lines[-lines:]) -def _terminal_process_alive(): - """True while a terminal64.exe belonging to THIS terminal directory runs. +#: The tester runs as terminal64.exe plus one metatester64.exe per agent. The +#: agents are what hold the localhost ports a later run needs, so a cleanup that +#: only accounts for terminal64.exe leaves the terminal unusable. +TESTER_PROCESS_NAMES = frozenset({"terminal64.exe", "metatester64.exe"}) + + +def _in_terminal_dir(exe): + """True when ``exe`` lives inside THIS terminal directory. + + Compared by normalized path components, not by substring: a sibling at + ``...\\a2\\terminal64.exe`` must never match a terminal at ``...\\a``. + """ + if not exe: + return False + base = TERMINAL_DIR.replace("\\", "/").lower().rstrip("/") + path = exe.replace("\\", "/").lower().rstrip("/") + return path == base or path.startswith(base + "/") + - Matched by directory rather than by the PID we spawned on purpose: the - point of this check is to see the process MT5 started to *replace* the one - we launched, which we never get a handle on. +def _terminal_processes(names=TESTER_PROCESS_NAMES): + """Processes of the given names running from THIS terminal directory. + + Matched by directory rather than by the PID we spawned on purpose: MT5 may + replace the process we launched (see _await_self_relaunch), and the agents + are never ours to begin with. Sibling terminals live in sibling directories, + so this never reaches across to another instance's processes. """ for proc in psutil.process_iter(["name", "exe"]): try: - if (proc.info.get("name") or "").lower() != "terminal64.exe": + if (proc.info.get("name") or "").lower() not in names: continue exe = proc.info.get("exe") or "" - if exe and TERMINAL_DIR.lower() in exe.lower(): - return True + if _in_terminal_dir(exe): + yield proc + except (psutil.NoSuchProcess, psutil.AccessDenied): + continue + + +def kill_terminal_processes(grace_seconds=10): + """Stop this terminal's tester processes. Returns how many were signalled. + + Terminate first, then kill whatever is still standing, so MT5 gets the + chance to release its files cleanly before being shot. + """ + victims = list(_terminal_processes()) + if not victims: + return 0 + for proc in victims: + try: + proc.terminate() + except (psutil.NoSuchProcess, psutil.AccessDenied): + continue + _, alive = psutil.wait_procs(victims, timeout=grace_seconds) + for proc in alive: + try: + proc.kill() except (psutil.NoSuchProcess, psutil.AccessDenied): continue + if alive: + psutil.wait_procs(alive, timeout=grace_seconds) + return len(victims) + + +def _terminal_process_alive(): + """True while a terminal64.exe belonging to THIS terminal directory runs. + + Only terminal64.exe: this answers "is a run still going", and the agents + come and go within one. + """ + for _ in _terminal_processes({"terminal64.exe"}): + return True return False @@ -461,6 +534,9 @@ def run_backtest(): "status": "queued", "broker": BROKER, "account": ACCOUNT, + # Sibling terminals share broker+account and differ only by instance, so + # without this the startup sweep cannot tell whose job it is looking at. + "instance": INSTANCE, "submittedAt": jobs.now_iso(), "startedAt": None, "finishedAt": None, @@ -560,10 +636,24 @@ def _execute_job(job_id): ) except subprocess.TimeoutExpired: duration = round(time.time() - start_time, 3) + # subprocess.run kills the process it started, but not the + # metatester64.exe agents it spawned, and not a terminal MT5 + # relaunched in place of ours. Those keep the terminal's + # localhost agent ports bound, so every later run on this + # terminal dies with "bind error [10048]" until the host is + # rebooted. Clear the whole directory's tester processes. + killed = kill_terminal_processes() + error = f"Backtest timed out after {job['timeoutSeconds']}s" + if killed: + error = f"{error} (killed {killed} leftover tester process(es))" + log.warning( + "backtest timed out broker=%s account=%s job=%s killed=%d", + BROKER, ACCOUNT, job_id, killed, + ) jobs.update_job( job_id, status="failed", - error=f"Backtest timed out after {job['timeoutSeconds']}s", + error=error, durationSeconds=duration, finishedAt=jobs.now_iso(), ) diff --git a/mt5api/backtest/jobs.py b/mt5api/backtest/jobs.py index 78d773e..1d758c3 100644 --- a/mt5api/backtest/jobs.py +++ b/mt5api/backtest/jobs.py @@ -19,9 +19,14 @@ from datetime import datetime, timezone from mt5api.config import ( + ACCOUNT, BACKTEST_JOB_DIR, BACKTEST_JOB_RETENTION_SECONDS, BACKTEST_SWEEP_LOOKBACK_SECONDS, + BROKER, + INSTANCE, + load_yaml_config, + normalize_instance, ) from mt5api.logger import log @@ -157,8 +162,47 @@ def public_payload(job: dict) -> dict: return payload +def _configured_terminals(broker: str, account: str) -> int: + """How many terminals in config.yaml target this broker/account. + + Reads the real terminal list so the ownership decision reflects how the + install is actually configured, not just this API process's own identity. + """ + terms = load_yaml_config().get("terminals") or [] + return sum( + 1 + for t in terms + if t.get("broker") == broker and t.get("account", "") == account + ) + + +def owns_job(job: dict) -> bool: + """Is this job ours, rather than a sibling terminal's? + + Every backtest API on a host shares one job directory, so the sweep sees + every terminal's jobs. A job that names a terminal is ours only if that + terminal is this one. A job that does not name a terminal is ambiguous and + is claimed ONLY when this broker/account has a single configured terminal — + that is the shape a single-terminal install writes, and what every job + written before the instance field existed looks like. In a multi-clone + install, an un-instanced job belongs to no clone in particular, and claiming + it here would reproduce the cross-terminal failure this sweep exists to + prevent. + """ + broker = job.get("broker") + if broker is not None and broker != BROKER: + return False + account = job.get("account") + if account is not None and account != ACCOUNT: + return False + instance = job.get("instance") + if instance is not None: + return normalize_instance(instance) == normalize_instance(INSTANCE) + return _configured_terminals(BROKER, ACCOUNT) <= 1 + + def sweep_orphans(lookback_seconds: int | None = None) -> int: - """Mark any queued/running jobs on disk as failed. + """Mark this terminal's queued/running jobs on disk as failed. Called at API startup. Only state files touched within the last ``lookback_seconds`` are considered: a live job rewrites its file on every @@ -167,6 +211,11 @@ def sweep_orphans(lookback_seconds: int | None = None) -> int: parsed tens of thousands of files on every boot — and because every backtest API on a VM shares the same job directory, that full scan repeated once per process. Returns the number of jobs swept. + + Only jobs belonging to THIS terminal are swept. The directory is shared, so + sweeping everything meant restarting one terminal's API failed every other + terminal's in-flight backtest with "API restarted before completion" — + silently, and while those runs went on to finish perfectly well. """ if lookback_seconds is None: lookback_seconds = SWEEP_LOOKBACK_SECONDS @@ -198,6 +247,8 @@ def sweep_orphans(lookback_seconds: int | None = None) -> int: continue if job.get("status") not in ACTIVE_STATUSES: continue + if not owns_job(job): + continue job["status"] = "failed" job["error"] = "API restarted before completion" job["finishedAt"] = now_iso() diff --git a/mt5api/main.py b/mt5api/main.py index 91372ac..ac71e0d 100644 --- a/mt5api/main.py +++ b/mt5api/main.py @@ -140,6 +140,29 @@ def _run_backtest_startup_cleanup(): swept = backtest_jobs.sweep_orphans() if swept: log.warning("Backtest sweep marked %d orphaned job(s) as failed.", swept) + + # Any tester process still running for this terminal is a leftover: we + # are the only thing that launches one here, and we have only just + # started. It holds this terminal's localhost agent ports, so leaving it + # makes every subsequent run fail with "bind error [10048]" until the + # host is rebooted. + # + # Deliberately not conditional on the sweep having found anything. A run + # whose state file predates the sweep lookback — an API killed while a + # long test was live — is never swept, and gating the kill on `swept` + # would leave exactly that process holding the ports. + # + # Backtest mode only: this cleanup thread runs in every mode, and in + # live mode the terminal is meant to stay up. + if MODE == "backtest": + from mt5api.backtest.handler import kill_terminal_processes + + killed = kill_terminal_processes() + if killed: + log.warning( + "Backtest startup killed %d leftover tester process(es).", killed + ) + pruned = backtest_jobs.prune_old_jobs() if pruned: log.info("Backtest retention retired %d old job(s).", pruned) diff --git a/scripts/wickworks-healthcheck.py b/scripts/wickworks-healthcheck.py new file mode 100644 index 0000000..915cb3c --- /dev/null +++ b/scripts/wickworks-healthcheck.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +"""Healthcheck for the wickworks TA sidecar. + +The sidecar shares the mt5 VM container's network namespace via compose's +``network_mode: service:mt5``. Docker resolves that namespace once, at +container start. When the mt5 container is later restarted or recreated (a +VM reboot, a manual ``docker restart mt5``, a compose recreate), Docker gives +the mt5 container a FRESH network namespace and leaves this sidecar running +in the old, now-orphaned one. The sidecar's own health endpoint on loopback +still answers, so the image's built-in healthcheck stays green while the +Windows VM can no longer reach wickworks at all — every ``/rates/ta`` call +then fails with ``connection refused`` and the API surfaces a 502. + +This healthcheck therefore does two things beyond the image's loopback probe: + +1. It confirms the sidecar is still attached to a LIVE mt5 netns by probing + the dockurr gateway services (20.20.20.1:445/139/5900/5700) that only + exist while sharing the current mt5 netns. When the mt5 container is + recreated, those services move to the new netns and become unreachable + here, which is the earliest detectable sign of orphaning. The probes run + concurrently so the all-unreachable (orphan) path is bounded by one probe + timeout, keeping the whole check inside Docker's healthcheck timeout. + +2. When orphaning is detected it kills the container's main uvicorn process + (PID 1 is ``sh``; ``kill 1`` from an exec'd healthcheck is not delivered, + but killing the uvicorn child makes ``sh`` exit cleanly), so the compose + ``restart: unless-stopped`` policy recreates the container and it rejoins + the current mt5 netns. + +Exit codes: 0 = healthy, 1 = unhealthy. When unhealthy due to orphaning the +process also terminates itself so the restart policy can actually fire. +""" + +import os +import signal +import socket +import sys +from concurrent.futures import ThreadPoolExecutor + +# dockurr gateway services that only exist while sharing the LIVE mt5 netns. +# The SMB/VNC/dockurr ports are bound by the mt5 container's own processes, +# so they are present when the namespaces are shared and gone when orphaned. +GATEWAY_HOST = os.environ.get("WICKWORKS_GATEWAY_HOST", "20.20.20.1") +GATEWAY_PORTS = [445, 139, 5900, 5700] +PROBE_TIMEOUT = 2 + +# Loopback health endpoint of the wickworks service itself. +SELF_HEALTH_URL = "http://127.0.0.1:8000/health" + + +def _self_healthy(): + """True when wickworks answers its own health endpoint on loopback.""" + import urllib.request + + try: + with urllib.request.urlopen(SELF_HEALTH_URL, timeout=PROBE_TIMEOUT) as resp: + return resp.status == 200 + except Exception: + return False + + +def _probe(port): + """One gateway probe. Returns True when the port answers.""" + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(PROBE_TIMEOUT) + try: + return sock.connect_ex((GATEWAY_HOST, port)) == 0 + finally: + sock.close() + + +def _shares_live_mt5_netns(): + """True when any dockurr gateway service answers through the shared netns. + + These services are owned by the mt5 container's processes. When the mt5 + container is recreated, they live in the new netns, so a refused/unrouted + connection here means this sidecar has been orphaned. + + Probes run concurrently: a serial sweep would take up to + ``len(GATEWAY_PORTS) * PROBE_TIMEOUT`` seconds when every port is down, + and Docker kills this healthcheck after its compose ``timeout`` — so the + orphan path (where every probe fails) must complete well within that + window or the self-heal never fires. Parallel probing bounds the sweep at + one ``PROBE_TIMEOUT`` regardless of how many ports there are. + """ + with ThreadPoolExecutor(max_workers=len(GATEWAY_PORTS)) as pool: + return any(pool.map(_probe, GATEWAY_PORTS)) + + +def _kill_main_process(): + """Kill the uvicorn process so PID 1 (sh) exits and the container stops. + + ``kill 1`` from a Docker exec'd healthcheck is not delivered to the + container's init in this runtime, so targeting the uvicorn child (the + process whose death makes ``sh -c uvicorn ...`` return) is what actually + terminates the container. The ``sh -c`` wrapper is deliberately skipped: + its cmdline also contains ``uvicorn``, and signalling it is the one thing + that does not work. + """ + for entry in os.listdir("/proc"): + if not entry.isdigit() or entry == "1": + continue + try: + with open(f"/proc/{entry}/cmdline", "rb") as fh: + cmdline = fh.read().decode("utf-8", errors="replace") + except OSError: + continue + if "uvicorn" in cmdline and "--multiprocessing-fork" not in cmdline: + try: + os.kill(int(entry), signal.SIGKILL) + except (ProcessLookupError, PermissionError): + pass + return + # Fallback: no uvicorn found — kill whatever is PID 1 via the signal that + # does get delivered from an exec'd process. + try: + os.kill(1, signal.SIGTERM) + except (ProcessLookupError, PermissionError): + pass + + +def main(): + if not _self_healthy(): + # The service itself is down — unhealthy without trying to restart; + # a dead uvicorn already causes the container to stop on its own. + return 1 + if _shares_live_mt5_netns(): + # Still inside the live mt5 netns — normal healthy state. + return 0 + # Orphaned: the mt5 container was recreated under a new netns. Kill the + # main process so the restart policy recreates us into the current netns. + print( + "wickworks orphaned from the live mt5 netns; " + "killing main process so the restart policy rejoins it", + file=sys.stderr, + ) + _kill_main_process() + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_backtest_jobs.py b/tests/test_backtest_jobs.py index 131472f..12444ce 100644 --- a/tests/test_backtest_jobs.py +++ b/tests/test_backtest_jobs.py @@ -13,6 +13,9 @@ @pytest.fixture def tmp_jobs_dir(monkeypatch, tmp_path): monkeypatch.setattr(jobs, "BACKTEST_JOB_DIR", str(tmp_path)) + # Default to a single-terminal install so legacy (un-instanced) jobs are + # claimed. Ownership-scoping tests override this via _clone_count. + monkeypatch.setattr(jobs, "_configured_terminals", lambda b, a: 1) # Reset in-memory cache between tests. jobs.BACKTEST_JOBS.clear() return tmp_path @@ -235,3 +238,75 @@ def test_store_and_load_job_roundtrip(tmp_jobs_dir): def test_load_job_missing_returns_none(tmp_jobs_dir): assert jobs.load_job("nope") is None + + +# ── Ownership scoping ──────────────────────────────────────────────────────── +# +# Every backtest API on a host shares one job directory, and sibling terminals +# share broker+account (they differ only by instance). Before scoping, any +# terminal's restart failed every other terminal's in-flight run. + + +def _identity(monkeypatch, broker="darwinex", account="live", instance="a"): + monkeypatch.setattr(jobs, "BROKER", broker) + monkeypatch.setattr(jobs, "ACCOUNT", account) + monkeypatch.setattr(jobs, "INSTANCE", instance) + + +def _clone_count(monkeypatch, count): + monkeypatch.setattr(jobs, "_configured_terminals", lambda b, a: count) + + +def test_sweep_leaves_a_sibling_terminals_job_alone(tmp_jobs_dir, monkeypatch): + _identity(monkeypatch, instance="a") + _clone_count(monkeypatch, 3) + _write(tmp_jobs_dir, "mine", "running", broker="darwinex", account="live", instance="a") + _write(tmp_jobs_dir, "theirs", "running", broker="darwinex", account="live", instance="b") + + assert jobs.sweep_orphans() == 1 + + mine = json.loads((tmp_jobs_dir / "mine.json").read_text()) + theirs = json.loads((tmp_jobs_dir / "theirs.json").read_text()) + assert mine["status"] == "failed" + assert theirs["status"] == "running", "a sibling terminal's live run must survive our restart" + + +def test_sweep_does_not_claim_a_legacy_job_on_a_multi_clone_install(tmp_jobs_dir, monkeypatch): + # The real pre-PR shape: broker+account present, instance absent. On a + # multi-clone install every clone shares broker+account, so the job cannot + # be attributed — claiming it would fail a sibling's live run. It must be + # left alone (retention retires it eventually). + _identity(monkeypatch, instance="a") + _clone_count(monkeypatch, 3) + _write(tmp_jobs_dir, "legacy", "running", broker="darwinex", account="live") + + assert jobs.sweep_orphans() == 0 + assert json.loads((tmp_jobs_dir / "legacy.json").read_text())["status"] == "running" + + +def test_sweep_still_claims_a_legacy_job_on_a_single_terminal_install(tmp_jobs_dir, monkeypatch): + # What a single-terminal install writes: broker+account, no instance, and no + # sibling clone to be confused with. Behaviour must not change there. + _identity(monkeypatch) + _clone_count(monkeypatch, 1) + _write(tmp_jobs_dir, "legacy", "running", broker="darwinex", account="live") + + assert jobs.sweep_orphans() == 1 + assert json.loads((tmp_jobs_dir / "legacy.json").read_text())["status"] == "failed" + + +def test_sweep_skips_another_broker_or_account(tmp_jobs_dir, monkeypatch): + _identity(monkeypatch, broker="darwinex", account="live", instance="a") + _write(tmp_jobs_dir, "other-broker", "running", broker="icmarkets", account="live", instance="a") + _write(tmp_jobs_dir, "other-account", "running", broker="darwinex", account="demo", instance="a") + assert jobs.sweep_orphans() == 0 + + +def test_owns_job_normalizes_the_instance(monkeypatch): + # "" and None mean the default instance; they must not read as a stranger. + _identity(monkeypatch, instance="default") + _clone_count(monkeypatch, 1) + assert jobs.owns_job({"broker": "darwinex", "account": "live", "instance": ""}) + assert jobs.owns_job({"broker": "darwinex", "account": "live", "instance": None}) + assert jobs.owns_job({"broker": "darwinex", "account": "live", "instance": "default"}) + assert not jobs.owns_job({"broker": "darwinex", "account": "live", "instance": "a"}) diff --git a/tests/test_backtest_log_tail.py b/tests/test_backtest_log_tail.py new file mode 100644 index 0000000..d6f3957 --- /dev/null +++ b/tests/test_backtest_log_tail.py @@ -0,0 +1,83 @@ +"""_tail_terminal_log picks the log that actually describes the run. + +The terminal's logs/ directory holds `.log` files plus a `metaeditor.log` +that is written once at install and never again. `metaeditor.log` sorts after +every dated log ("m" > "2"), so selecting by name attached a stale compile tail +to every backtest failure message — which is what hid an agent bind error +behind three-month-old MetaEditor output. +""" +from __future__ import annotations + +import os +import time + +import pytest + +from mt5api.backtest import handler + + +def _write_utf16(path, text): + with open(path, "w", encoding="utf-16-le") as handle: + handle.write(text) + + +@pytest.fixture +def terminal_logs(monkeypatch, tmp_path): + terminal_dir = tmp_path / "terminal" + log_dir = terminal_dir / "logs" + log_dir.mkdir(parents=True) + monkeypatch.setattr(handler, "TERMINAL_DIR", str(terminal_dir)) + return log_dir + + +def _age(path, seconds_ago): + when = time.time() - seconds_ago + os.utime(path, (when, when)) + + +def test_prefers_the_run_log_over_metaeditor_log(terminal_logs): + _write_utf16(terminal_logs / "20260808.log", "Tester\tautomatic testing started\n") + _write_utf16(terminal_logs / "metaeditor.log", "compiling ancient stuff\n") + # metaeditor.log is both alphabetically last AND, here, newer on disk — + # it must still never be chosen. + _age(terminal_logs / "20260808.log", 3600) + _age(terminal_logs / "metaeditor.log", 1) + + tail = handler._tail_terminal_log() + assert "automatic testing started" in tail + assert "ancient" not in tail + + +def test_picks_the_newest_dated_log(terminal_logs): + _write_utf16(terminal_logs / "20260501.log", "old run\n") + _write_utf16(terminal_logs / "20260808.log", "current run\n") + _age(terminal_logs / "20260501.log", 90 * 86400) + _age(terminal_logs / "20260808.log", 5) + + assert "current run" in handler._tail_terminal_log() + + +def test_newest_wins_even_when_it_sorts_first_by_name(terminal_logs): + # A log rotated across a year boundary sorts before last year's file. + _write_utf16(terminal_logs / "20261231.log", "last year\n") + _write_utf16(terminal_logs / "20270101.log", "this year\n") + _age(terminal_logs / "20261231.log", 86400) + _age(terminal_logs / "20270101.log", 5) + + assert "this year" in handler._tail_terminal_log() + + +def test_returns_empty_when_only_metaeditor_log_exists(terminal_logs): + _write_utf16(terminal_logs / "metaeditor.log", "compiling\n") + assert handler._tail_terminal_log() == "" + + +def test_returns_empty_when_there_is_no_log_dir(monkeypatch, tmp_path): + monkeypatch.setattr(handler, "TERMINAL_DIR", str(tmp_path / "nothing-here")) + assert handler._tail_terminal_log() == "" + + +def test_tail_is_limited_to_the_requested_line_count(terminal_logs): + _write_utf16(terminal_logs / "20260808.log", "".join(f"line {i}\n" for i in range(50))) + tail = handler._tail_terminal_log(lines=5) + assert tail.splitlines() == [f"line {i}" for i in range(45, 50)] diff --git a/tests/test_backtest_process_cleanup.py b/tests/test_backtest_process_cleanup.py new file mode 100644 index 0000000..da4e5cb --- /dev/null +++ b/tests/test_backtest_process_cleanup.py @@ -0,0 +1,182 @@ +"""Tester process cleanup: timeout kill and boot sweep. + +MT5 runs a test as terminal64.exe plus one metatester64.exe per agent, and the +agents are what bind the localhost ports. subprocess.run kills the process it +started but neither the agents nor a terminal MT5 relaunched in place of ours, +so a timed-out run used to leave those ports held and every later run on that +terminal died with "bind error [10048]" until the host was rebooted. +""" +from __future__ import annotations + +import psutil +import pytest + +from mt5api.backtest import handler + + +class FakeProc: + def __init__(self, name, exe, dies_on_terminate=True): + self.info = {"name": name, "exe": exe} + self.dies_on_terminate = dies_on_terminate + self.terminated = False + self.killed = False + + def terminate(self): + self.terminated = True + + def kill(self): + self.killed = True + + +@pytest.fixture +def terminal_dir(monkeypatch, tmp_path): + directory = tmp_path / "terminals" / "darwinex" / "live" / "a" + directory.mkdir(parents=True) + monkeypatch.setattr(handler, "TERMINAL_DIR", str(directory)) + return str(directory) + + +def _install(monkeypatch, procs, still_alive=()): + monkeypatch.setattr(psutil, "process_iter", lambda attrs=None: list(procs)) + monkeypatch.setattr( + psutil, "wait_procs", lambda victims, timeout=None: ([], list(still_alive)) + ) + + +def test_kills_the_terminal_and_its_agents(monkeypatch, terminal_dir): + terminal = FakeProc("terminal64.exe", f"{terminal_dir}\\terminal64.exe") + agent_one = FakeProc("metatester64.exe", f"{terminal_dir}\\metatester64.exe") + agent_two = FakeProc("metatester64.exe", f"{terminal_dir}\\metatester64.exe") + _install(monkeypatch, [terminal, agent_one, agent_two]) + + assert handler.kill_terminal_processes() == 3 + assert all(p.terminated for p in (terminal, agent_one, agent_two)) + + +def test_never_touches_a_sibling_terminals_processes(monkeypatch, terminal_dir): + sibling = terminal_dir[:-1] + "b" + mine = FakeProc("terminal64.exe", f"{terminal_dir}\\terminal64.exe") + theirs = FakeProc("terminal64.exe", f"{sibling}\\terminal64.exe") + theirs_agent = FakeProc("metatester64.exe", f"{sibling}\\metatester64.exe") + _install(monkeypatch, [mine, theirs, theirs_agent]) + + assert handler.kill_terminal_processes() == 1 + assert mine.terminated + assert not theirs.terminated and not theirs_agent.terminated + + +def test_never_touches_a_sibling_whose_dir_shares_our_prefix(monkeypatch, terminal_dir): + # `a` is a component boundary: a sibling at `a2` or `aa` sorts/prefixes + # after us but is a different terminal directory. A substring match would + # terminate theirs too. + for suffix in ("2", "a"): + sibling = terminal_dir + suffix + theirs = FakeProc("terminal64.exe", f"{sibling}\\terminal64.exe") + theirs_agent = FakeProc("metatester64.exe", f"{sibling}\\metatester64.exe") + mine = FakeProc("terminal64.exe", f"{terminal_dir}\\terminal64.exe") + _install(monkeypatch, [mine, theirs, theirs_agent]) + + assert handler.kill_terminal_processes() == 1 + assert mine.terminated + assert not theirs.terminated and not theirs_agent.terminated + + +def test_ignores_unrelated_processes(monkeypatch, terminal_dir): + noise = FakeProc("chrome.exe", f"{terminal_dir}\\chrome.exe") + _install(monkeypatch, [noise]) + assert handler.kill_terminal_processes() == 0 + assert not noise.terminated + + +def test_escalates_to_kill_when_terminate_is_ignored(monkeypatch, terminal_dir): + stubborn = FakeProc("terminal64.exe", f"{terminal_dir}\\terminal64.exe") + _install(monkeypatch, [stubborn], still_alive=[stubborn]) + + assert handler.kill_terminal_processes() == 1 + assert stubborn.terminated and stubborn.killed + + +def test_reports_nothing_to_do_when_the_terminal_is_idle(monkeypatch, terminal_dir): + _install(monkeypatch, []) + assert handler.kill_terminal_processes() == 0 + + +def test_alive_check_ignores_agents(monkeypatch, terminal_dir): + # Agents outliving their terminal must not read as "a run is still going", + # or _await_self_relaunch would wait out the full job timeout on them. + agent = FakeProc("metatester64.exe", f"{terminal_dir}\\metatester64.exe") + _install(monkeypatch, [agent]) + assert handler._terminal_process_alive() is False + + terminal = FakeProc("terminal64.exe", f"{terminal_dir}\\terminal64.exe") + _install(monkeypatch, [terminal]) + assert handler._terminal_process_alive() is True + + +def test_survives_a_process_vanishing_mid_scan(monkeypatch, terminal_dir): + class Vanishing(FakeProc): + def terminate(self): + raise psutil.NoSuchProcess(pid=1) + + gone = Vanishing("terminal64.exe", f"{terminal_dir}\\terminal64.exe") + _install(monkeypatch, [gone]) + assert handler.kill_terminal_processes() == 1 + + +# ── Startup cleanup ────────────────────────────────────────────────────────── + + +@pytest.fixture +def startup(monkeypatch): + """_run_backtest_startup_cleanup with its collaborators stubbed.""" + # mt5api.main pulls in the WSGI/MCP stack; skip where those are not + # installed rather than fail (the container test image installs them). + pytest.importorskip("a2wsgi") + from mt5api import main + + calls = {"killed": 0} + + def fake_kill(*_args, **_kwargs): + calls["killed"] += 1 + return 1 + + monkeypatch.setattr(handler, "kill_terminal_processes", fake_kill) + monkeypatch.setattr(main.backtest_jobs, "prune_old_jobs", lambda *a, **k: 0) + return main, calls + + +def test_kills_leftovers_even_when_nothing_was_swept(startup, monkeypatch): + """The case a sweep-gated kill misses. + + A run whose state file is older than the sweep lookback — an API killed + while a long test was live — is never swept, so gating the kill on the + sweep leaves that process holding this terminal's agent ports forever. + """ + main, calls = startup + monkeypatch.setattr(main, "MODE", "backtest") + monkeypatch.setattr(main.backtest_jobs, "sweep_orphans", lambda *a, **k: 0) + + main._run_backtest_startup_cleanup() + + assert calls["killed"] == 1 + + +def test_kills_leftovers_when_jobs_were_swept(startup, monkeypatch): + main, calls = startup + monkeypatch.setattr(main, "MODE", "backtest") + monkeypatch.setattr(main.backtest_jobs, "sweep_orphans", lambda *a, **k: 3) + + main._run_backtest_startup_cleanup() + + assert calls["killed"] == 1 + + +def test_never_kills_in_live_mode(startup, monkeypatch): + """The live terminal is meant to stay up, and this thread runs in every mode.""" + main, calls = startup + monkeypatch.setattr(main, "MODE", "live") + monkeypatch.setattr(main.backtest_jobs, "sweep_orphans", lambda *a, **k: 2) + + main._run_backtest_startup_cleanup() + + assert calls["killed"] == 0 diff --git a/tests/test_config_generation.py b/tests/test_config_generation.py index ebe82cc..9b03310 100644 --- a/tests/test_config_generation.py +++ b/tests/test_config_generation.py @@ -20,6 +20,23 @@ {"name": "bulk", "service": "mt5-b", "container_name": "mt5-b", "novnc_port": 8007}, ] +TWO_VMS_WITH_WICKWORKS = [ + { + "name": "fast", + "service": "mt5", + "container_name": "mt5", + "novnc_port": 8006, + "wickworks_service": "wickworks", + }, + { + "name": "bulk", + "service": "mt5-b", + "container_name": "mt5-b", + "novnc_port": 8007, + "wickworks_service": "wickworks-b", + }, +] + def _load_config_helper_module(): module_path = Path(__file__).resolve().parents[1] / "scripts" / "config_helper.py" @@ -216,3 +233,35 @@ def test_generate_compose_emits_one_service_per_vm(tmp_path, monkeypatch): port for name in ("mt5", "mt5-b") for port in (services[name].get("ports") or []) ] assert len(set(host_ports)) == len(host_ports), f"VMs share a host port: {host_ports}" + + +def test_generate_compose_gives_wickworks_a_self_healing_healthcheck( + tmp_path, monkeypatch +): + """The wickworks TA sidecar shares the mt5 netns. When the mt5 container is + recreated it is orphaned in the old netns while its own loopback healthcheck + still passes, so the generated compose must mount and run the self-heal + healthcheck that detects the orphan and forces a restart. + """ + helper = _load_config_helper_module() + config_path = _write_config( + tmp_path, [{"broker": "acme", "account": "main", "port": 5001}] + ) + vms_path = _write_vms(tmp_path, TWO_VMS_WITH_WICKWORKS) + outpath = tmp_path / "docker-compose.yml" + template_path = Path(__file__).resolve().parents[1] / "docker-compose.yml.j2" + monkeypatch.setattr(helper, "CONFIG_PATH", str(config_path)) + monkeypatch.setattr(helper, "VMS_PATH", str(vms_path)) + monkeypatch.setattr(helper, "COMPOSE_TEMPLATE_PATH", str(template_path)) + monkeypatch.setattr(helper, "COMPOSE_OUTPUT_PATH", str(outpath)) + monkeypatch.setattr("sys.argv", ["config_helper.py", "generate_compose"]) + + helper.main() + + services = yaml.safe_load(outpath.read_text(encoding="utf-8"))["services"] + for name in ("wickworks", "wickworks-b"): + svc = services[name] + assert svc["network_mode"] == "service:" + {"wickworks": "mt5", "wickworks-b": "mt5-b"}[name] + assert "./scripts/wickworks-healthcheck.py:/wickworks-healthcheck.py:ro" in svc["volumes"] + hc = svc["healthcheck"]["test"] + assert hc == ["CMD", "python", "/wickworks-healthcheck.py"] diff --git a/tests/test_wickworks_healthcheck.py b/tests/test_wickworks_healthcheck.py new file mode 100644 index 0000000..ba6094b --- /dev/null +++ b/tests/test_wickworks_healthcheck.py @@ -0,0 +1,179 @@ +"""Tests for the wickworks sidecar self-heal healthcheck. + +The wickworks TA sidecar shares the mt5 VM container's netns via compose +``network_mode: service:mt5``. When the mt5 container is recreated, Docker +leaves wickworks in the old, now-orphaned netns: its loopback /health still +answers (so the image's built-in check stays green) while the VM can no +longer reach it. The healthcheck must (a) stay green while the namespaces are +shared, (b) go red AND kill wickworks when the dockurr gateway becomes +unreachable, so compose's restart policy recreates it into the current netns. +""" + +import importlib.util +from pathlib import Path +from unittest.mock import patch + +import pytest + +SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "wickworks-healthcheck.py" + + +def _load(): + spec = importlib.util.spec_from_file_location("wickworks_hc_test", SCRIPT) + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(module) + return module + + +class _Resp: + def __init__(self, status): + self.status = status + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + +def _fake_sock(opens): + """Returns a connect_ex-spoofing socket whose result depends on `opens`.""" + class Sock: + def __init__(self, *args, **kwargs): + pass + + def settimeout(self, t): + pass + + def connect_ex(self, addr): + return 0 if opens else 111 + + def close(self): + pass + + return Sock + + +@pytest.fixture +def hc(): + return _load() + + +class _SlowSock: + """connect_ex sleeps before refusing, as an unreachable host would.""" + + def __init__(self, *args, **kwargs): + pass + + def settimeout(self, t): + pass + + def connect_ex(self, addr): + import time as _time + _time.sleep(0.4) + return 111 + + def close(self): + pass + + +def test_gateway_probes_run_concurrently(hc): + """The all-unreachable orphan path must be bounded by one probe timeout, + not len(GATEWAY_PORTS) * PROBE_TIMEOUT — Docker kills this healthcheck + after its compose timeout, so a serial sweep could outlive it and the + self-heal would never fire exactly when orphaned. + """ + import time + + hc.GATEWAY_PORTS = [445, 139, 5900, 5700] + hc.PROBE_TIMEOUT = 5 + with patch("socket.socket", _SlowSock): + start = time.monotonic() + result = hc._shares_live_mt5_netns() + elapsed = time.monotonic() - start + assert result is False + # Four serial 0.4s probes would take ~1.6s; concurrent takes ~0.4s. + assert elapsed < 1.0, f"gateway sweep took {elapsed:.2f}s — probes are serial" + + +def test_script_worst_case_fits_inside_the_compose_timeout(hc): + """The rendered wickworks healthcheck timeout must exceed the script's + worst-case runtime (self-health probe + concurrent gateway sweep), or + Docker would abort the orphan path before it can kill the process. + """ + import re + from pathlib import Path + + j2 = Path(__file__).resolve().parents[1] / "docker-compose.yml.j2" + match = re.search( + r"wickworks-healthcheck\.py.*?\btimeout:\s+(\d+)s", + j2.read_text(encoding="utf-8"), + re.DOTALL, + ) + assert match, "wickworks healthcheck timeout not found in docker-compose.yml.j2" + compose_timeout = int(match.group(1)) + worst_case = hc.PROBE_TIMEOUT + hc.PROBE_TIMEOUT + assert ( + compose_timeout > worst_case + ), f"compose timeout {compose_timeout}s does not exceed worst case {worst_case}s" + + +def test_healthy_when_self_and_gateway_reachable(hc): + with patch("urllib.request.urlopen", return_value=_Resp(200)), \ + patch("socket.socket", _fake_sock(True)), \ + patch("os.kill") as kill: + assert hc.main() == 0 + kill.assert_not_called() + + +def test_unhealthy_when_self_down_no_restart(hc): + with patch("urllib.request.urlopen", side_effect=OSError("refused")), \ + patch("os.kill") as kill: + assert hc.main() == 1 + kill.assert_not_called() + + +def test_orphan_kills_main_process(hc): + """Gateway unreachable + self up => orphaned => kill uvicorn, exit 1.""" + with patch("urllib.request.urlopen", return_value=_Resp(200)), \ + patch("socket.socket", _fake_sock(False)), \ + patch("os.listdir", return_value=["1", "7", "self"]), \ + patch("builtins.open", create=True) as mock_open: + # os.listdir drives the /proc scan in _kill_main_process. + def fake_open(path, *args, **kwargs): + if str(path).startswith("/proc/7/cmdline"): + fh = type("FH", (), {"__enter__": lambda s: s, "__exit__": lambda *a: False})() + fh.read = lambda: b"/opt/venv/bin/python /opt/venv/bin/uvicorn wickworks.server:app" + return fh + raise FileNotFoundError(path) + + mock_open.side_effect = fake_open + with patch("os.kill") as kill: + assert hc.main() == 1 + # One kill call: the uvicorn main process (pid 7). + assert kill.call_count == 1 + assert kill.call_args[0][0] == 7 + + +def test_orphan_skips_uvicorn_workers(hc): + """Only the uvicorn MAIN process is killed, not a --multiprocessing-fork + worker, so the worker-set shutdown path is untouched.""" + with patch("urllib.request.urlopen", return_value=_Resp(200)), \ + patch("socket.socket", _fake_sock(False)), \ + patch("os.listdir", return_value=["1", "9"]), \ + patch("builtins.open", create=True) as mock_open: + def fake_open(path, *args, **kwargs): + if str(path).startswith("/proc/9/cmdline"): + fh = type("FH", (), {"__enter__": lambda s: s, "__exit__": lambda *a: False})() + fh.read = lambda: b"python -B -c from multiprocessing.spawn import spawn_main ... --multiprocessing-fork" + return fh + raise FileNotFoundError(path) + + mock_open.side_effect = fake_open + with patch("os.kill") as kill: + assert hc.main() == 1 + # Worker is skipped; fallback SIGTERM to PID 1 happens instead. + assert kill.call_count == 1 + assert kill.call_args[0][0] == 1 + assert kill.call_args[0][1] == 15