From 2647061646c4c5f84b1c41c73fcf371452689fdc Mon Sep 17 00:00:00 2001 From: Dushnilin Date: Sat, 8 Aug 2026 11:01:09 +0300 Subject: [PATCH] =?UTF-8?q?refactor(watchdog):=20=D1=80=D0=B0=D0=B7=D0=B4?= =?UTF-8?q?=D0=B5=D0=BB=D0=B5=D0=BD=D0=B8=D0=B5=20=D0=BD=D0=B0=D0=B1=D0=BB?= =?UTF-8?q?=D1=8E=D0=B4=D0=B5=D0=BD=D0=B8=D1=8F=20=D0=B8=20=D0=B2=D0=BE?= =?UTF-8?q?=D1=81=D1=81=D1=82=D0=B0=D0=BD=D0=BE=D0=B2=D0=BB=D0=B5=D0=BD?= =?UTF-8?q?=D0=B8=D1=8F=20=D1=87=D0=B5=D1=80=D0=B5=D0=B7=20=D1=88=D0=B8?= =?UTF-8?q?=D0=BD=D1=83=20=D1=81=D0=BE=D0=B1=D1=8B=D1=82=D0=B8=D0=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ранее вочдог опрашивал состояние системы и устранял проблемы в едином смешанном цикле. Добавление новой проверки или действия по восстановлению требовало изменения монолитной функции, где мониторинг и ремонт были переплетены, что затрудняло аудит порядка выполнения и изолированное тестирование. Вводится `usr/lib/service/event_controller.uc` — шина событий pub/sub для вочдога. Теперь зонды публикуют факты (`EV.PROXY_DOWN`, `EV.DNS_DOWN`, `EV.SINGBOX_STOPPED`, `EV.OOM_DETECTED`, `EV.WAN_DOWN` и т.д.), не зная, кто их обрабатывает. Восстановители (healers) подписываются на эти факты и независимо выполняют ремонт. Ключевые изменения: - `event_controller.uc` управляет словарём событий, регистрацией подписчиков, сортировкой по приоритетам и синхронной отправкой. - Зонды (`probe_proxy`, `probe_dns`, `probe_singbox`, `probe_oom`, `probe_wan`) публикуют события вместо прямого вызова функций ремонта. - Восстановители регистрируются как подписчики с явными приоритетами, чтобы первопричины устранялись раньше вторичных сбоев. - `tests/event_bus.sh` добавляет интеграционные тесты для отправки событий, сортировки приоритетов и обработки подписчиков. --- tachyon/files/usr/lib/core/events.uc | 237 ++++ .../files/usr/lib/service/event_controller.uc | 957 ++++++++++++++ tachyon/files/usr/lib/service/watchdog.uc | 1123 ++++++----------- tests/event_bus.sh | 240 ++++ tests/event_controller.sh | 204 +++ tests/forward_references.sh | 104 ++ tests/watchdog_subscriptions.sh | 144 +++ 7 files changed, 2279 insertions(+), 730 deletions(-) create mode 100644 tachyon/files/usr/lib/core/events.uc create mode 100644 tachyon/files/usr/lib/service/event_controller.uc create mode 100644 tests/event_bus.sh create mode 100644 tests/event_controller.sh create mode 100644 tests/forward_references.sh create mode 100644 tests/watchdog_subscriptions.sh diff --git a/tachyon/files/usr/lib/core/events.uc b/tachyon/files/usr/lib/core/events.uc new file mode 100644 index 00000000..eafce640 --- /dev/null +++ b/tachyon/files/usr/lib/core/events.uc @@ -0,0 +1,237 @@ +#!/usr/bin/env ucode +// +// Event bus: publish/subscribe with per-subscriber cooldown, publish-side +// deduplication, deterministic priority ordering and handler error isolation. +// +// The bus is deliberately domain-agnostic: it knows nothing about sing-box, +// UCI or reload locks. Observation belongs to service/event_controller.uc, +// remediation to service/watchdog.uc. Keeping this file free of I/O is what +// makes it directly unit-testable (see tests/event_bus.sh). +// +// Timing uses clock(true) (CLOCK_MONOTONIC) rather than time(): cooldown and +// dedup windows must not be skewed by an NTP step, which is a real event on +// routers that boot before the network is up. + +// Monotonic milliseconds. Independent of wall-clock jumps. +function now_ms() { + let c = clock(true); + if (type(c) != "array" || length(c) < 2) + return 0; + return c[0] * 1000 + int(c[1] / 1000000); +} + +function as_string(value) { + return value == null ? "" : "" + value; +} + +// Creates an isolated bus. Each instance owns its subscriber table, dedup +// state and counters, so tests can build a fresh bus per assertion. +function bus() { + let handlers = {}; // type -> array of subscriber records + let last_emit = {}; // dedup key -> monotonic ms of last accepted emit + let counters = { + emitted: 0, // events accepted onto the bus + suppressed: 0, // emit_once calls dropped as duplicates + delivered: 0, // handler invocations that ran to completion + skipped: 0, // handler invocations skipped by cooldown + failed: 0 // handler invocations that threw + }; + let error_sink = null; // optional function(name, type, err) + + let self = {}; + + // Registers a handler. Options: + // name — label used in error reports and stats (defaults to type) + // cooldown — minimum seconds between two invocations of THIS handler. + // Replaces the hand-rolled last_*_time globals. + // priority — lower runs first; ties keep registration order. + self.on = function(event_type, handler, opts) { + event_type = as_string(event_type); + if (event_type == "" || type(handler) != "function") + return false; + + let options = (type(opts) == "object") ? opts : {}; + let record = { + handler: handler, + name: as_string(options.name || event_type), + cooldown: options.cooldown != null ? int(options.cooldown) : 0, + priority: options.priority != null ? int(options.priority) : 50, + last_run: -1, + runs: 0 + }; + + if (type(handlers[event_type]) != "array") + handlers[event_type] = []; + push(handlers[event_type], record); + + // Stable sort keeps registration order within one priority level, so + // subscriber ordering stays reproducible across runs. + sort(handlers[event_type], function(a, b) { return a.priority - b.priority; }); + return true; + }; + + // Reports a handler failure without letting it abort the remaining + // subscribers. This is the isolation that safe_call() used to provide + // around every individual check in the old watchdog loop. + function report_error(name, event_type, err) { + counters.failed++; + if (type(error_sink) == "function") { + try { + error_sink(name, event_type, err); + } + catch (nested) { + } + } + } + + // Installs the sink used to log handler failures. Kept injectable so the + // bus itself never depends on a logging backend. + self.on_error = function(sink) { + error_sink = (type(sink) == "function") ? sink : null; + }; + + function dispatch(event_type, event) { + let subscribers = handlers[event_type]; + if (type(subscribers) != "array") + return 0; + + let ran = 0; + let stamp = now_ms(); + + for (let record in subscribers) { + // Cooldown is per subscriber, not per event type: two handlers on + // the same fact may legitimately want different pacing. + if (record.cooldown > 0 && record.last_run >= 0 && + (stamp - record.last_run) < record.cooldown * 1000) { + counters.skipped++; + continue; + } + + record.last_run = stamp; + record.runs++; + try { + record.handler(event); + counters.delivered++; + ran++; + } + catch (err) { + report_error(record.name, event_type, err); + } + } + return ran; + } + + // Publishes a fact. Returns the number of handlers that completed. + self.emit = function(event_type, payload) { + event_type = as_string(event_type); + if (event_type == "") + return 0; + + counters.emitted++; + return dispatch(event_type, { + type: event_type, + payload: (type(payload) == "object") ? payload : {}, + ts: time(), + ms: now_ms() + }); + }; + + // Publishes a fact at most once per window_seconds. The dedup key is the + // event type plus an optional caller-supplied discriminator, so + // "same problem" can be defined per source (e.g. one key per domain). + // Returns -1 when the emit was suppressed as a duplicate. + self.emit_once = function(event_type, payload, window_seconds, key) { + event_type = as_string(event_type); + if (event_type == "") + return -1; + + let window = window_seconds != null ? int(window_seconds) : 0; + let dedup_key = event_type + "" + as_string(key); + let stamp = now_ms(); + + if (window > 0) { + let previous = last_emit[dedup_key]; + if (previous != null && (stamp - int(previous)) < window * 1000) { + counters.suppressed++; + return -1; + } + } + last_emit[dedup_key] = stamp; + + return self.emit(event_type, payload); + }; + + // True when at least one handler is registered for the type. Lets sources + // skip an expensive probe nobody is listening to. + self.has = function(event_type) { + let subscribers = handlers[as_string(event_type)]; + return type(subscribers) == "array" && length(subscribers) > 0; + }; + + // Drops dedup and cooldown state without unsubscribing. Used when the + // world changed under us (config reload) and past timing is meaningless. + self.reset_timers = function() { + last_emit = {}; + for (let event_type in keys(handlers)) { + for (let record in handlers[event_type]) + record.last_run = -1; + } + }; + + // Counter snapshot for ai-status-full. + self.stats = function() { + let subscriber_count = 0; + for (let event_type in keys(handlers)) + subscriber_count += length(handlers[event_type]); + + return { + emitted: counters.emitted, + suppressed: counters.suppressed, + delivered: counters.delivered, + skipped: counters.skipped, + failed: counters.failed, + types: length(keys(handlers)), + subscribers: subscriber_count + }; + }; + + // Per-handler run counts, ordered by event type. Diagnostic aid: shows at + // a glance which subscriber is actually firing in production. + self.subscriber_runs = function() { + let out = {}; + for (let event_type in keys(handlers)) { + for (let record in handlers[event_type]) + out[record.name] = record.runs; + } + return out; + }; + + return self; +} + +function module_exports() { + return { + bus, + now_ms + }; +} + +if (sourcepath(1) != null && sourcepath(1) != "") + return module_exports(); + +// CLI surface exists for the shell test suite, which has no way to import a +// ucode module directly. +let mode = ARGV[0] || ""; + +if (mode == "selftest") { + let b = bus(); + let seen = []; + b.on("t", function(ev) { push(seen, ev.type); }, { name: "probe" }); + b.emit("t", {}); + print(length(seen) == 1 && seen[0] == "t" ? "ok\n" : "fail\n"); + exit(length(seen) == 1 ? 0 : 1); +} +else { + warn("Usage: core/events.uc \n"); + exit(1); +} diff --git a/tachyon/files/usr/lib/service/event_controller.uc b/tachyon/files/usr/lib/service/event_controller.uc new file mode 100644 index 00000000..18f760a4 --- /dev/null +++ b/tachyon/files/usr/lib/service/event_controller.uc @@ -0,0 +1,957 @@ +#!/usr/bin/env ucode +// +// Event controller: the single place that observes the system and turns +// observations into facts on the event bus. It never repairs anything — +// remediation lives in service/watchdog.uc, which subscribes to these facts. +// +// Two kinds of source feed the bus: +// +// push — the kernel or procd tells us something happened (ubus service +// events, syslog lines, the honeypot FIFO). Zero cost when idle. +// probe — nothing pushes "DNS stopped resolving", so a probe tier asks. +// Probing lives here, in one loop, instead of being duplicated +// inside every healer: one tick collects each observation exactly +// once and publishes it, where the old code called is_dns_working() +// from two healers and get_sing_box_pid() from eight. +// +// Probe tiers and their intervals are unchanged from the timer-based +// watchdog: fast 15s, normal adaptive 120/300s, slow 300s. + +let fs = require("fs"); +let uci_core = require("core.uci"); +let common = require("core.common"); +let helpers = require("core.helpers"); + +const CONFIG_NAME = getenv("TACHYON_CONFIG_NAME") || "tachyon"; + +// Minimum spacing between two increments of the DNS stall streak. Matches the +// floor of the old adaptive normal tier (120s), which is the rate at which +// ai_heal_dns() used to step that counter before probing moved to the fast tier. +const DNS_STREAK_INTERVAL = 120; + +let as_string = common.as_string; +let shell_quote = common.shell_quote; +let command_from_args = common.command_from_args; +let command_success_from_args = common.command_success_from_args; +let is_process_name_running = helpers.is_process_name_running; + +// ─── Event type vocabulary ──────────────────────────────────────────────────── +// Every fact the controller can publish. Subscribers reference these constants +// rather than bare strings so a typo is a load-time error, not silent silence. +const EV = { + SINGBOX_STOPPED: "singbox.stopped", + FIREWALL_RELOADED: "firewall.reloaded", + OOM_DETECTED: "oom.detected", + OOM_RECOVERABLE: "oom.recoverable", + SMARTDETECT_CANDIDATE: "smartdetect.candidate", + URLTEST_SWITCHED: "urltest.switched", + HONEYPOT_HIT: "honeypot.hit", + PROXY_DOWN: "proxy.down", + PROXY_UP: "proxy.up", + DNS_DOWN: "dns.down", + DNS_UP: "dns.up", + MEMORY_LOW: "memory.low", + NFT_MISSING: "nft.missing", + QOS_MISSING: "qos.missing", + TPROXY_DOWN: "tproxy.down", + RPCD_FD_LEAK: "rpcd.fd_leak", + WAN_DOWN: "wan.down", + SUBNETS_EMPTY: "subnets.empty", + SUBSCRIPTION_UNREACHABLE: "subscription.unreachable", + CONFIG_CORRUPT: "config.corrupt", + SECTIONS_EMPTY: "sections.empty", + ANOMALY_RECONNECTS: "anomaly.reconnects", + PAUSE_EXPIRED: "pause.expired", + TICK: "tick" +}; + +// ─── Shell helpers ──────────────────────────────────────────────────────────── +function command_capture(command) { + let pipe = fs.popen(command, "r"); + if (!pipe) + return { status: 1, output: "" }; + let data = pipe.read("all"); + let status = pipe.close(); + if (status > 255) status = int(status / 256); + return { status, output: data == null ? "" : as_string(data) }; +} + +function command_output_from_args(args) { + let result = command_capture(command_from_args(args) + " 2>/dev/null"); + return result.status == 0 ? result.output : ""; +} + +function settings() { + return common.object_or_empty(uci_core.get_all(CONFIG_NAME, "settings")); +} + +function setting(key, default_val) { + let cfg = settings(); + let val = cfg[key]; + return val != null ? as_string(val) : as_string(default_val); +} + +function enabled(key, default_val) { + return setting(key, default_val) == "1"; +} + +function process_running(pid, expected_name) { + if (match(as_string(pid), /^[0-9]+$/) == null) + return false; + if (expected_name != null && expected_name != "") + return is_process_name_running(pid, expected_name); + return fs.stat("/proc/" + pid) != null; +} + +// ─── Shared observations ────────────────────────────────────────────────────── +// Resolved once per tick and handed to every probe that needs them. This is +// what removes the repeated get_sing_box_pid() / is_dns_working() calls that +// the per-healer detection prologues used to make. + +function get_sing_box_pid() { + for (let path in [ "/var/run/sing-box.pid", "/var/run/sing-box/sing-box.pid" ]) { + let pid = trim(fs.readfile(path) || ""); + if (pid != "" && process_running(pid, "sing-box")) return pid; + } + + let ubus_res = command_capture("ubus call service list '{\"name\":\"sing-box\"}' 2>/dev/null"); + if (ubus_res.status == 0 && ubus_res.output != "") { + let matched = match(ubus_res.output, /"pid":\s*([0-9]+)/); + if (matched && matched[1] != "") { + let pid = matched[1]; + if (process_running(pid, "sing-box")) return pid; + } + } + + let pidof_res = command_capture("pidof sing-box 2>/dev/null"); + if (pidof_res.status == 0 && pidof_res.output != "") { + let fields = split(trim(pidof_res.output), /[ \t]+/); + if (length(fields) > 0 && fields[0] != "") { + let pid = fields[0]; + if (process_running(pid, "sing-box")) return pid; + } + } + + return ""; +} + +// Scans /proc for a sing-box executable. Slower than get_sing_box_pid(), used +// only to confirm a suspected stop before publishing the fact. +function sing_box_in_proc() { + let proc = fs.opendir("/proc"); + if (!proc) return ""; + let entry; + let pid = ""; + while ((entry = proc.read()) != null) { + if (match(entry, /^[0-9]+$/)) { + let exe = fs.readlink("/proc/" + entry + "/exe") || ""; + let slash = rindex(exe, "/"); + if ((slash >= 0 ? substr(exe, slash + 1) : exe) == "sing-box") { + pid = entry; + break; + } + } + } + proc.close(); + return pid; +} + +function is_dns_working() { + return command_success_from_args([ "nslookup", "-timeout=2", "yandex.ru", "127.0.0.1" ]) || + command_success_from_args([ "nslookup", "-timeout=2", "connectivitycheck.gstatic.com", "127.0.0.1" ]) || + command_success_from_args([ "nslookup", "-timeout=2", "google.com", "127.0.0.1" ]); +} + +function check_tachyon_cli_running() { + let running = false; + let proc = fs.opendir("/proc"); + if (proc) { + let entry; + while ((entry = proc.read()) != null) { + if (match(entry, /^[0-9]+$/)) { + let cmdline = fs.readfile("/proc/" + entry + "/cmdline") || ""; + if (index(cmdline, "/usr/bin/tachyon") >= 0) { + if (index(cmdline, "start") >= 0 || index(cmdline, "restart") >= 0 || + index(cmdline, "reload") >= 0 || index(cmdline, "stop") >= 0) { + running = true; + break; + } + } + } + } + proc.close(); + } + return running; +} + +function is_list_update_running() { + let pid = trim(fs.readfile("/var/run/tachyon_list_update.pid") || ""); + return process_running(pid, "ucode"); +} + +function is_reload_in_progress() { + return fs.stat("/var/run/tachyon.reload.lock") != null || check_tachyon_cli_running(); +} + +// Reads the http/mixed inbound port from the generated sing-box config. +// Cached because the config only changes across a reload. +let cached_proxy_port = null; +function proxy_port() { + if (cached_proxy_port !== null) + return as_string(cached_proxy_port); + + let port = "4534"; + let sb_cfg_data = fs.readfile("/etc/sing-box/config.json"); + if (sb_cfg_data) { + try { + let sb_cfg = json(sb_cfg_data); + if (sb_cfg.inbounds) { + for (let inb in sb_cfg.inbounds) { + if (inb.type == "http" || inb.type == "mixed") { + port = as_string(inb.listen_port || 4534); + break; + } + } + } + } catch(e) {} + } + cached_proxy_port = port; + return port; +} + +function forget_proxy_port() { + cached_proxy_port = null; +} + +function tproxy_port() { + let port = 4530; + let sb_cfg_data = fs.readfile("/etc/sing-box/config.json"); + if (sb_cfg_data) { + try { + let sb_cfg = json(sb_cfg_data); + if (sb_cfg.inbounds) { + for (let inb in sb_cfg.inbounds) { + if (inb.type == "tproxy") { + port = int(inb.listen_port || port); + break; + } + } + } + } catch(e) {} + } + return port; +} + +// ─── Log line classification ────────────────────────────────────────────────── +// Pure functions: a syslog line in, a fact (or null) out. No I/O, so the +// classifier is directly testable — see tests/event_controller.sh. + +// Extract a candidate domain from a sing-box log line. Returns null when the +// line carries no usable hostname. Kept separate from log handling so the +// pattern can be exercised directly by tests. +function smart_detect_extract_domain(line) { + if (line == null) return null; + let text = as_string(line); + let m = match(text, /"([a-zA-Z0-9][a-zA-Z0-9.-]{1,60}\.[a-zA-Z]{2,})(:[0-9]+)?"/); + if (!m) m = match(text, /target[= ]([a-zA-Z0-9][a-zA-Z0-9.-]{1,60}\.[a-zA-Z]{2,})/); + if (!m || !m[1]) return null; + let domain = m[1]; + if (length(domain) < 5) return null; + if (index(domain, "*") >= 0 || index(domain, "?") >= 0) return null; + if (index(domain, "..") >= 0) return null; + if (index(domain, "-") == 0 || substr(domain, length(domain) - 1) == "-") return null; + return domain; +} + +// Classifies one syslog line into zero or more facts, mirroring the branch +// structure of the old handle_log_line() exactly: +// +// - the keyword pre-filter rejects the overwhelming majority of lines before +// any lowercasing or regex work happens; +// - an OOM line is terminal — the original returned right after it, so a line +// that is both an OOM and a URLTest line yields only the OOM fact; +// - smart-detect and URLTest are NOT exclusive: the original fell through +// from one to the other, so both facts can come from a single line. Hence +// an array return rather than a single fact. +// +// The netlink/nlbwmon discrimination is preserved: misreading a netlink +// warning as OOM would shrink GOMEMLIMIT for nothing. +function classify_log_line(line) { + if (!line || line == "") return []; + + // Fast keyword pre-filter: skip 95%+ of irrelevant log lines instantly + if (index(line, "direct") < 0 && index(line, "DIRECT") < 0 && + index(line, "memory") < 0 && index(line, "oom") < 0 && index(line, "OOM") < 0 && + index(line, "URLTest") < 0 && index(line, "proxy") < 0) { + return []; + } + + let line_lower = lc(line); + + let is_oom = (index(line_lower, "oom-killer") >= 0 || + index(line_lower, "out of memory: kill process") >= 0 || + (index(line_lower, "kernel:") >= 0 && index(line_lower, "out of memory") >= 0) || + (index(line_lower, "sing-box") >= 0 && index(line_lower, "out of memory") >= 0) || + index(line_lower, "fatal error: out of memory") >= 0); + let is_netlink_warning = (index(line_lower, "netlink") >= 0 || index(line_lower, "nlbwmon") >= 0); + + // Terminal, as in the original: no further classification after an OOM. + if (is_oom && !is_netlink_warning) + return [ { type: EV.OOM_DETECTED, payload: {} } ]; + + let facts = []; + + if ((index(line_lower, "direct") >= 0 || index(line_lower, "DIRECT") >= 0) && + (index(line_lower, "failed") >= 0 || index(line_lower, "timeout") >= 0 || + index(line_lower, "reset") >= 0)) { + let domain = smart_detect_extract_domain(line); + if (domain != null) + push(facts, { type: EV.SMARTDETECT_CANDIDATE, payload: { domain: domain } }); + } + + if (index(line, "URLTest") >= 0 || index(line_lower, "selected proxy") >= 0 || + index(line_lower, "switch proxy") >= 0) + push(facts, { type: EV.URLTEST_SWITCHED, payload: {} }); + + return facts; +} + +// ─── Controller ─────────────────────────────────────────────────────────────── +// Owns streak counters and latency history: both describe the *observation*, +// not the repair, so they belong on this side of the boundary. Subscribers +// receive the streak in the payload and compare it against their own +// threshold, exactly as the old healers compared their private counters. + +function controller(bus, opts) { + let options = (type(opts) == "object") ? opts : {}; + let log = (type(options.log) == "function") ? options.log : function(msg, lvl) {}; + + let state = { + proxy_consecutive_fails: 0, + dns_consecutive_fails: 0, + dns_fail_streak: 0, + dns_streak_stamp: 0, + proxy_latency_history: [], + dns_latency_history: [], + syslog_start_time: 0, + logread_pipe_fd: -1, + last_oom_time: 0, + healthy_streak: 0 + }; + + let self = { EV: EV, state: state }; + + // Keeps the last 20 samples. Bounded so a long-lived watchdog cannot grow + // its heap through the history array. + function push_history(history, entry) { + push(history, entry); + if (length(history) > 20) { + let trimmed = []; + for (let i = length(history) - 20; i < length(history); i++) + push(trimmed, history[i]); + return trimmed; + } + return history; + } + + // ── Push source: syslog ─────────────────────────────────────────────────── + // Returns the list of facts actually published (after the OOM replay guard, + // the smart_detect setting and the URLTest throttle), so tests can assert + // on publication rather than on classification alone. + self.handle_log_line = function(line) { + let published = []; + + for (let fact in classify_log_line(line)) { + if (fact.type == EV.OOM_DETECTED) { + let now = time(); + // Ignore replay of the historical log buffer that logread -f + // dumps on start, and collapse an OOM storm into one fact. + if (state.syslog_start_time > 0 && (now - state.syslog_start_time > 3) && + (now - state.last_oom_time > 60)) { + state.last_oom_time = now; + bus.emit(EV.OOM_DETECTED, {}); + push(published, EV.OOM_DETECTED); + } + continue; + } + + if (fact.type == EV.SMARTDETECT_CANDIDATE) { + if (setting("smart_detect", "0") != "1") continue; + bus.emit(EV.SMARTDETECT_CANDIDATE, fact.payload); + push(published, EV.SMARTDETECT_CANDIDATE); + continue; + } + + // URLTest switches are chatty; the old check_urltest_switches() + // throttled itself to one run per 5s, so the fact carries the same. + if (fact.type == EV.URLTEST_SWITCHED) { + if (bus.emit_once(EV.URLTEST_SWITCHED, {}, 5) < 0) continue; + push(published, EV.URLTEST_SWITCHED); + } + } + + return published; + }; + + // ── Push source: ubus ───────────────────────────────────────────────────── + self.handle_ubus_service_stop = function(name, reason) { + if (as_string(name) != "sing-box") return false; + bus.emit(EV.SINGBOX_STOPPED, { reason: as_string(reason) }); + return true; + }; + + // handle_ubus_firewall_reload is defined further down, next to the probes it + // re-runs. Declaration order in this file is load-bearing: ucode captures a + // closure's upvalues when the closure is created, so a local declared later + // in the same scope is not captured at all — the name resolves as a global + // and comes back null. That fails at call time, which neither `ucode -c` nor + // `ucode -S -c` can see. Every helper here precedes its callers. + + // ── Push source: honeypot FIFO ──────────────────────────────────────────── + self.handle_honeypot_line = function(line) { + let ip = trim(as_string(line)); + if (ip == "" || match(ip, /^[0-9a-fA-F:.]+$/) == null) return false; + bus.emit(EV.HONEYPOT_HIT, { ip: ip }); + return true; + }; + + // ── Probe: expired pause ────────────────────────────────────────────────── + // Declared ahead of probe_singbox, which consults it on every fast tick. + function probe_pause() { + let val = trim(fs.readfile("/tmp/tachyon_paused_until") || ""); + if (val == "") return false; + if (int(val) <= time()) { + bus.emit(EV.PAUSE_EXPIRED, {}); + return false; + } + return true; // still paused + } + self.is_paused = probe_pause; + + // ── Probe: sing-box liveness ────────────────────────────────────────────── + // The expensive /proc scan only runs once the cheap pidfile/ubus/pidof + // path has already failed, preserving the original fast-path ordering. + function probe_singbox() { + if (setting("recovery_bypass", "0") == "1") return; + // An active pause means sing-box is stopped on purpose. The original + // check_singbox_process() consulted the pause on every fast tick, so it + // is consulted here too: leaving it to the normal tier alone would open + // a window of up to one normal interval in which a deliberate pause is + // reported as a stop. + if (probe_pause()) return; + if (check_tachyon_cli_running()) return; + if (is_list_update_running()) return; + + let pid = get_sing_box_pid(); + if (pid != "" && process_running(pid, "sing-box")) return; + + // No configured sections means sing-box is legitimately absent. + let has_sections = false; + let uci_sections = uci_core.get_all(CONFIG_NAME); + if (uci_sections) { + for (let k in keys(uci_sections)) { + if (uci_sections[k][".type"] == "section") { + has_sections = true; + break; + } + } + } + if (!has_sections) return; + + if (sing_box_in_proc() == "") + bus.emit(EV.SINGBOX_STOPPED, { reason: "process missing from /proc" }); + } + + // ── Probe: proxy reachability ───────────────────────────────────────────── + // One measurement serves both proxy subscribers (the old + // ai_heal_proxy_health and ai_heal_proxy_connectivity each ran their own + // curl). Feature flags are deliberately NOT checked here: the two + // subscribers are gated by different settings, so the flag belongs to the + // subscriber. The probe only skips when measuring is pointless or its + // result would be misleading. + function probe_proxy() { + if (!bus.has(EV.PROXY_DOWN) && !bus.has(EV.PROXY_UP)) return; + // A reload tears the proxy down on purpose; a sample taken now would + // land in the latency history as a fault that never happened. + if (is_reload_in_progress()) return; + + let pid = get_sing_box_pid(); + if (pid == "" || !process_running(pid, "sing-box")) return; + + let port = proxy_port(); + let check_url = setting("ai_proxy_health_url", "https://cp.cloudflare.com/generate_204"); + + let started = time(); + let ok = command_success_from_args([ + "curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", + "--connect-timeout", "3", "--max-time", "5", + "--proxy", "http://127.0.0.1:" + port, + check_url + ]); + let elapsed = (time() - started) * 1000; + + state.proxy_latency_history = push_history(state.proxy_latency_history, + { ok: ok, ms: elapsed, ts: time() }); + + if (ok) { + state.proxy_consecutive_fails = 0; + bus.emit(EV.PROXY_UP, { port: port, ms: elapsed }); + return; + } + + state.proxy_consecutive_fails++; + + // Distinguish "proxy is broken" from "the whole uplink is down": only + // the former is worth restarting sing-box over. + let direct_ok = command_success_from_args([ + "curl", "-s", "-I", "--connect-timeout", "3", "--max-time", "5", + "https://cp.cloudflare.com/generate_204" + ]); + + bus.emit(EV.PROXY_DOWN, { + port: port, + streak: state.proxy_consecutive_fails, + direct_ok: direct_ok, + ms: elapsed + }); + } + + // Clearing the streak also clears its pacing stamp, so the first failure + // after a recovery is counted immediately rather than waiting out the rest + // of an interval that started before the resolver came back. + function clear_dns_streak() { + state.dns_fail_streak = 0; + state.dns_streak_stamp = 0; + } + + // ── Probe: DNS resolution ───────────────────────────────────────────────── + // As with the proxy probe, one resolution attempt serves both DNS + // subscribers; each carries its own enable flag and its own threshold. + function probe_dns() { + if (!bus.has(EV.DNS_DOWN) && !bus.has(EV.DNS_UP)) return; + if (setting("recovery_bypass", "0") == "1") return; + + // A reload restarts the resolver; failures during it are expected and + // must not accumulate, so the streak resets exactly as before. + if (is_reload_in_progress()) { + clear_dns_streak(); + return; + } + + let pid = get_sing_box_pid(); + if (pid == "" || !process_running(pid, "sing-box")) { + clear_dns_streak(); + return; + } + + let started = time(); + let ok = is_dns_working(); + let elapsed = (time() - started) * 1000; + + state.dns_latency_history = push_history(state.dns_latency_history, + { ok: ok, ms: elapsed, ts: time() }); + + if (ok) { + state.dns_consecutive_fails = 0; + clear_dns_streak(); + bus.emit(EV.DNS_UP, { ms: elapsed }); + return; + } + + state.dns_consecutive_fails++; + + // Two counters at two rates, because the two DNS healers were measured on + // two different tiers before the split: dns_consecutive_fails came from + // the fast tier (15s), dns_fail_streak from the normal-tier audit (120s+). + // The probe now runs once at the fast rate, so the streak is only stepped + // once per DNS_STREAK_INTERVAL to keep "3 failures" meaning the same + // ~6 minutes of trouble it meant before, not 45 seconds. + let now = time(); + if (now - state.dns_streak_stamp >= DNS_STREAK_INTERVAL) { + state.dns_streak_stamp = now; + state.dns_fail_streak++; + log("Watchdog: DNS check failed (" + as_string(state.dns_fail_streak) + "/3 failures)", "warn"); + } + + bus.emit(EV.DNS_DOWN, { + streak: state.dns_fail_streak, + consecutive: state.dns_consecutive_fails, + ms: elapsed + }); + } + + // ── Probe: memory ───────────────────────────────────────────────────────── + function read_mem_available_mb() { + let mem_info = fs.readfile("/proc/meminfo") || ""; + for (let line in split(mem_info, "\n")) { + if (index(line, "MemAvailable:") == 0) { + let fields = split(trim(line), /[ \t]+/); + if (length(fields) >= 2) + return int(fields[1]) / 1024; + break; + } + } + return -1; + } + self.read_mem_available_mb = read_mem_available_mb; + + function probe_memory() { + let free_mb = read_mem_available_mb(); + if (free_mb >= 0 && free_mb < 15) + bus.emit(EV.MEMORY_LOW, { free_mb: free_mb }); + } + + // ── Probe: nftables table and QoS marks ─────────────────────────────────── + function probe_nftables() { + if (is_list_update_running()) return; + if (is_reload_in_progress()) return; + + let nft_table = getenv("NFT_TABLE_NAME") || "TachyonTable"; + let routing_mode = setting("routing_mode", "nftables"); + // One `nft list` serves both the table check and the QoS check. + let out_nft = command_output_from_args(["nft", "list", "table", "inet", nft_table]); + + if (routing_mode == "nftables") { + if (index(out_nft, "tproxy") < 0 || index(out_nft, "priority_rules") < 0) { + bus.emit(EV.NFT_MISSING, { table: nft_table }); + return; + } + } + + if (setting("qos_priority_engine", "1") == "0") return; + if ((index(out_nft, "dscp set 0x2e") < 0 && index(out_nft, "dscp set ef") < 0) || + (index(out_nft, "dscp set 0x22") < 0 && index(out_nft, "dscp set af41") < 0)) + bus.emit(EV.QOS_MISSING, { table: nft_table }); + } + + // ── Probe: TPROXY port liveness ─────────────────────────────────────────── + function probe_tproxy() { + if (setting("recovery_bypass", "0") == "1") return; + if (is_reload_in_progress()) return; + + let pid = get_sing_box_pid(); + if (pid == "" || !process_running(pid, "sing-box")) return; + + let port = tproxy_port(); + // /proc/net/tcp encodes the local port in hex; 0A is TCP_LISTEN. + let hex_port = sprintf("%04X", port); + let listening = false; + for (let proc_file in ["/proc/net/tcp6", "/proc/net/tcp"]) { + let tcp_data = fs.readfile(proc_file) || ""; + for (let line in split(tcp_data, "\n")) { + if (index(line, ":" + hex_port + " ") >= 0 && index(line, " 0A ") >= 0) { + listening = true; + break; + } + } + if (listening) break; + } + + if (!listening) + bus.emit(EV.TPROXY_DOWN, { port: port }); + } + + // ── Probe: rpcd file descriptor leak ────────────────────────────────────── + // rpcd accumulates FDs from LuCI API calls. Past ~1024 it can no longer + // fork /usr/bin/tachyon and LuCI reports everything as stopped. 512 is + // half the limit — early enough to restart without user-visible impact. + const RPCD_FD_THRESHOLD = 512; + + function probe_rpcd() { + let rpcd_pid = null; + let proc_dir = fs.opendir("/proc"); + if (!proc_dir) return; + let entry; + while ((entry = proc_dir.read()) != null) { + if (!match(entry, /^[0-9]+$/)) continue; + let comm = trim(fs.readfile("/proc/" + entry + "/comm") || ""); + if (comm == "rpcd") { rpcd_pid = entry; break; } + } + proc_dir.close(); + if (!rpcd_pid) return; + + let fd_dir = fs.opendir("/proc/" + rpcd_pid + "/fd"); + if (!fd_dir) return; + let fd_count = 0; + while ((entry = fd_dir.read()) != null) { + if (match(entry, /^[0-9]+$/)) fd_count++; + } + fd_dir.close(); + + if (fd_count > RPCD_FD_THRESHOLD) + bus.emit(EV.RPCD_FD_LEAK, { fd_count: fd_count, threshold: RPCD_FD_THRESHOLD }); + } + + // ── Probe: WAN address and default route ────────────────────────────────── + function probe_wan() { + if (is_reload_in_progress()) return; + + let proto = uci_core.get("network", "wan", "proto") || "pppoe"; + let device = trim(uci_core.get("network", "wan", "device") || "eth0"); + let iface = (proto == "pppoe") ? "pppoe-wan" : device; + + let addr_out = command_capture("ip addr show " + shell_quote(iface) + " 2>/dev/null").output; + let no_address = index(addr_out, "inet ") < 0; + + let route_out = command_capture("ip route 2>/dev/null").output; + let no_gateway = index(route_out, "default") < 0; + + if (no_address || no_gateway) + bus.emit(EV.WAN_DOWN, { iface: iface, no_address: no_address, no_gateway: no_gateway }); + } + + // ── Probe: community subnet nft sets ────────────────────────────────────── + function probe_subnet_sets() { + if (setting("recovery_bypass", "0") == "1") return; + if (is_reload_in_progress()) return; + if (is_list_update_running()) return; + + let pid = get_sing_box_pid(); + if (pid == "" || !process_running(pid, "sing-box")) return; + + let nft_table = getenv("NFT_TABLE_NAME") || "TachyonTable"; + let all_sections = uci_core.get_all(CONFIG_NAME); + if (!all_sections) return; + + let empty = []; + for (let sec_name in keys(all_sections)) { + let s = all_sections[sec_name]; + if (s[".type"] != "section" || s.enabled != "1") continue; + if (!s.community_lists) continue; + + // A non-zero exit means the set does not exist for this section + // (no subnet community configured) — not an emptiness problem. + let set_name = "tachyon_rule_" + sec_name + "_subnets"; + let result = command_capture( + command_from_args(["nft", "list", "set", "inet", nft_table, set_name]) + " 2>/dev/null"); + if (result.status != 0 || result.output == "") continue; + if (index(result.output, "elements") < 0) + push(empty, set_name); + } + + if (length(empty) > 0) + bus.emit(EV.SUBNETS_EMPTY, { sets: empty }); + } + + // ── Probe: subscription URL reachability ────────────────────────────────── + function probe_subscription() { + let sub_url = trim(setting("subscription_url", "")); + if (sub_url == "") return; + + let res = command_capture( + "curl -s -o /dev/null -w %{http_code} --connect-timeout 10 " + shell_quote(sub_url) + " 2>&1"); + let code = int(res.output); + if (res.status == 0 && code >= 200 && code < 400) return; + + bus.emit(EV.SUBSCRIPTION_UNREACHABLE, { url: sub_url, code: code }); + } + + // ── Probe: UCI config integrity ─────────────────────────────────────────── + function probe_config() { + let data = fs.readfile("/etc/config/tachyon"); + if (data == null || data == "") + bus.emit(EV.CONFIG_CORRUPT, {}); + } + + // ── Probe: proxy sections with subscriptions but no usable outbounds ────── + function probe_empty_sections() { + if (!enabled("ai_section_failover_enabled", "1")) return; + if (is_reload_in_progress()) return; + + let connections = require("config.connections"); + let sections = common.object_or_empty(uci_core.get_all(CONFIG_NAME)); + let empty = []; + + for (let name in sections) { + let s = common.object_or_empty(sections[name]); + let action = common.as_string(s.action || ""); + if (!connections.is_connections_action(action)) continue; + let sub_urls = common.list_option(s, "subscription_url"); + if (length(sub_urls) == 0) continue; + + let cache = common.object_or_empty( + common.read_json_file("/var/run/tachyon/section-cache/" + name + ".json")); + let usable = length(common.array_or_empty(cache.servers)) + + length(common.array_or_empty(cache.urls)) + + length(common.array_or_empty(cache.selector_urls)) + + length(common.array_or_empty(cache.domain)) + + length(common.array_or_empty(cache.domain_suffix)) + + length(common.array_or_empty(cache.ip_cidr)); + if (usable > 0) continue; + + push(empty, name); + } + + if (length(empty) > 0) + bus.emit(EV.SECTIONS_EMPTY, { sections: empty }); + } + + // ── Probe: reconnect rate anomaly ───────────────────────────────────────── + function probe_anomalies() { + if (!enabled("ai_anomaly_detection_enabled", "1")) return; + + let threshold = int(setting("ai_anomaly_reconnect_threshold", "10")); + let reconnects = int(trim(fs.readfile("/tmp/tachyon_reconnect_count") || "0")); + + if (reconnects > threshold) + bus.emit(EV.ANOMALY_RECONNECTS, { count: reconnects, threshold: threshold }); + } + + // ── Probe: OOM scale recovery window ────────────────────────────────────── + // Half an hour without an OOM means the shrunk GOMEMLIMIT can start + // creeping back up; the healer decides by how much. + // A scale of 1.0 means there is nothing to restore, so the marker file is + // dropped here rather than published as a fact: removing a stale file is an + // observation cleanup, not a repair a subscriber should have to schedule. + function probe_oom_recovery() { + let now = time(); + if (state.last_oom_time == 0) return; + if (now - state.last_oom_time < 1800) return; + + let scale_path = "/etc/tachyon/mem_scale"; + let scale_data = fs.readfile(scale_path); + if (scale_data == null) return; + let current_scale = double(trim(as_string(scale_data))); + if (current_scale >= 1.0) { + try { fs.unlink(scale_path); } catch (e) {} + return; + } + if (current_scale < 0.1) return; + + bus.emit(EV.OOM_RECOVERABLE, { last_oom: state.last_oom_time, scale: current_scale }); + } + + // ── Tier drivers ────────────────────────────────────────────────────────── + // Each probe is wrapped so one failing observation cannot stop the tier; + // this mirrors the safe_call() isolation the old per-check loop had. + function run_probe(fn, name) { + try { + fn(); + } + catch (e) { + log("Probe " + name + " failed: " + as_string(e), "err"); + } + } + + // ── Push source: ubus firewall.reload ───────────────────────────────────── + // Declared here, after run_probe and the two probes it drives, because a + // closure cannot capture a local declared below it (see the note above + // handle_honeypot_line). A firewall reload can wipe our table and flush the + // community sets, so those two probes re-run immediately instead of waiting + // for the next tier tick — this is what check_firewall_rules() did on this + // event. + self.handle_ubus_firewall_reload = function() { + bus.emit(EV.FIREWALL_RELOADED, {}); + run_probe(probe_nftables, "nftables"); + run_probe(probe_subnet_sets, "subnet_sets"); + return true; + }; + + self.probe_fast = function() { + run_probe(probe_singbox, "singbox"); + run_probe(probe_proxy, "proxy"); + run_probe(probe_dns, "dns"); }; + + self.probe_normal = function() { + run_probe(probe_pause, "pause"); + run_probe(probe_memory, "memory"); + run_probe(probe_rpcd, "rpcd"); + run_probe(probe_nftables, "nftables"); + run_probe(probe_tproxy, "tproxy"); + }; + + self.probe_slow = function() { + run_probe(probe_subnet_sets, "subnet_sets"); + run_probe(probe_wan, "wan"); + run_probe(probe_subscription, "subscription"); + run_probe(probe_config, "config"); + run_probe(probe_empty_sections, "empty_sections"); + run_probe(probe_anomalies, "anomalies"); + run_probe(probe_oom_recovery, "oom_recovery"); + }; + + // Health is graded by how long we go without publishing a fault: after a + // long clean streak the normal tier can afford to slow down. + self.adaptive_normal_interval = function() { + if (!enabled("ai_adaptive_intervals_enabled", "1")) return 120; + return state.healthy_streak > 25 ? 300 : 120; + }; + + self.note_healthy = function() { state.healthy_streak++; }; + self.note_incident = function() { state.healthy_streak = 0; }; + self.note_oom = function(ts) { state.last_oom_time = ts != null ? int(ts) : time(); }; + self.set_syslog_start = function(ts) { state.syslog_start_time = int(ts); }; + + // Streak resets belong to whoever acted on the fact: the old healers zeroed + // their counter right after restarting, so that a restart-in-flight would + // not immediately trip the same threshold again on the next tick. + self.reset_dns_streak = clear_dns_streak; + self.reset_dns_consecutive = function() { state.dns_consecutive_fails = 0; }; + self.reset_proxy_consecutive = function() { state.proxy_consecutive_fails = 0; }; + + // Latency history and streak counters feed ai-status-full and the metrics + // export, which are still owned by the watchdog. + self.proxy_latency_history = function() { return state.proxy_latency_history; }; + self.dns_latency_history = function() { return state.dns_latency_history; }; + self.proxy_consecutive_fails = function() { return state.proxy_consecutive_fails; }; + self.dns_consecutive_fails = function() { return state.dns_consecutive_fails; }; + self.healthy_streak = function() { return state.healthy_streak; }; + + self.forget_proxy_port = forget_proxy_port; + self.get_sing_box_pid = get_sing_box_pid; + self.is_dns_working = is_dns_working; + self.is_reload_in_progress = is_reload_in_progress; + self.is_list_update_running = is_list_update_running; + self.check_tachyon_cli_running = check_tachyon_cli_running; + self.proxy_port = proxy_port; + + return self; +} + +function module_exports() { + return { + controller, + classify_log_line, + smart_detect_extract_domain, + EV + }; +} + +if (sourcepath(1) != null && sourcepath(1) != "") + return module_exports(); + +// CLI surface for the shell test suite, which cannot import a ucode module. +let mode = ARGV[0] || ""; + +if (mode == "classify") { + let facts = classify_log_line(ARGV[1]); + if (length(facts) == 0) exit(1); + // One type per line, in publication order. + for (let fact in facts) + print(fact.type + "\n"); + exit(0); +} +else if (mode == "classify-domain") { + for (let fact in classify_log_line(ARGV[1])) { + if (fact.payload.domain != null) { + print(fact.payload.domain + "\n"); + exit(0); + } + } + exit(1); +} +else if (mode == "extract-domain") { + let extracted = smart_detect_extract_domain(ARGV[1]); + if (extracted == null) exit(1); + print(extracted + "\n"); + exit(0); +} +else if (mode == "event-types") { + for (let k in sort(keys(EV))) + print(k + "=" + EV[k] + "\n"); + exit(0); +} +else { + warn("Usage: service/event_controller.uc ...\n"); + exit(1); +} diff --git a/tachyon/files/usr/lib/service/watchdog.uc b/tachyon/files/usr/lib/service/watchdog.uc index cc512d87..18b5f030 100644 --- a/tachyon/files/usr/lib/service/watchdog.uc +++ b/tachyon/files/usr/lib/service/watchdog.uc @@ -5,6 +5,8 @@ let uci_core = require("core.uci"); let common = require("core.common"); let helpers = require("core.helpers"); let connections = require("config.connections"); +let events = require("core.events"); +let event_controller = require("service.event_controller"); const CONFIG_NAME = getenv("TACHYON_CONFIG_NAME") || "tachyon"; const LIB_DIR = getenv("TACHYON_LIB") || "/usr/lib/tachyon"; @@ -24,7 +26,6 @@ let telegram_msg_count = 0; let telegram_msg_window = time(); // FD-cascade prevention: track logread pipe FD to close it in background spawns let logread_pipe_fd = -1; -let syslog_start_time = 0; let command_from_args = common.command_from_args; let command_status = common.command_status; @@ -41,15 +42,6 @@ function command_capture(command) { return { status, output: data == null ? "" : as_string(data) }; } -function command_output(command) { - let result = command_capture(command); - return result.status == 0 ? result.output : ""; -} - -function command_output_from_args(args) { - return command_output(command_from_args(args) + " 2>/dev/null"); -} - // Run a command in background, explicitly closing the logread pipe FD to // prevent FD-cascade: each restart/reload inherits read-end of logread pipe, // keeping orphaned logread -f processes alive across watchdog generations. @@ -303,28 +295,42 @@ let ubus = null; try { uloop = require("uloop"); } catch (e) {} try { ubus = require("ubus"); } catch (e) {} -let last_oom_time = 0; +// ─── Event bus and observation controller ───────────────────────────────────── +// The controller observes and publishes facts; everything below subscribes and +// repairs. Splitting the two is what removes the detection prologue that used +// to be copy-pasted into the head of every ai_heal_* function. +let EV = event_controller.EV; +let bus = events.bus(); +let controller = event_controller.controller(bus, { + log: function(message, level) { log_message(message, level); } +}); + +// A subscriber that throws must not silence the subscribers behind it. This is +// the isolation safe_call() gave each check in the old loop, now applied once +// by the bus to every handler. +bus.on_error(function(name, event_type, err) { + log_message("Graceful degradation: " + name + " (" + event_type + ") failed: " + as_string(err), "err"); +}); + +// Repair-side pacing. Unlike the observation counters below, these belong to +// the action: they debounce how often a healer is allowed to act, not how often +// the world is measured. let last_oom_recovery_time = 0; let last_restart_time = 0; -let last_urltest_check = 0; +let last_reload_time = 0; let pending_smart_domains = {}; let smart_detect_last_run = 0; -let last_subnet_heal_time = 0; -let last_wan_heal_time = 0; -let last_gateway_heal_time = 0; -let last_reload_time = 0; -let proxy_consecutive_fails = 0; -let dns_consecutive_fails = 0; -let ai_healthy_streak = 0; -let cached_proxy_port = null; -let proxy_latency_history = []; -let dns_latency_history = []; -let last_anomaly_check = 0; -let last_metrics_export = 0; let last_fast_check = 0; let last_normal_check = 0; let last_slow_check = 0; +// Observation state lives in the controller; these read through to it so the +// status and metrics contracts keep reporting exactly the same numbers. +function proxy_consecutive_fails() { return controller.proxy_consecutive_fails(); } +function dns_consecutive_fails() { return controller.dns_consecutive_fails(); } +function proxy_latency_history() { return controller.proxy_latency_history(); } +function dns_latency_history() { return controller.dns_latency_history(); } + function check_tachyon_cli_running() { let running = false; let proc = fs.opendir("/proc"); @@ -346,7 +352,9 @@ function check_tachyon_cli_running() { return running; } -function handle_singbox_stop_event(reason) { +// The 30s floor is the original restart debounce: procd emits several stop +// events for one crash, and each would otherwise queue its own restart. +function heal_singbox_stopped(ev) { let now = time(); if (now - last_restart_time < 30) return; last_restart_time = now; @@ -359,7 +367,7 @@ function handle_singbox_stop_event(reason) { let list_update_pid = trim(fs.readfile("/var/run/tachyon_list_update.pid") || ""); if (process_running(list_update_pid, "ucode")) return; - log_message("sing-box is stopped (" + as_string(reason || "health check") + "). Restarting Tachyon...", "warn"); + log_message("sing-box is stopped (" + as_string(ev.payload.reason || "health check") + "). Restarting Tachyon...", "warn"); increment_reconnect_count(); let tcfg = common.object_or_empty(uci_core.get_all(CONFIG_NAME, "telegram")); if (tcfg.notify_crash != "0") { @@ -369,98 +377,7 @@ function handle_singbox_stop_event(reason) { } function get_sing_box_pid() { - for (let path in [ "/var/run/sing-box.pid", "/var/run/sing-box/sing-box.pid" ]) { - let pid = trim(fs.readfile(path) || ""); - if (pid != "" && process_running(pid, "sing-box")) return pid; - } - - let ubus_res = command_capture("ubus call service list '{\"name\":\"sing-box\"}' 2>/dev/null"); - if (ubus_res.status == 0 && ubus_res.output != "") { - let matched = match(ubus_res.output, /"pid":\s*([0-9]+)/); - if (matched && matched[1] != "") { - let pid = matched[1]; - if (process_running(pid, "sing-box")) return pid; - } - } - - let pidof_res = command_capture("pidof sing-box 2>/dev/null"); - if (pidof_res.status == 0 && pidof_res.output != "") { - let fields = split(trim(pidof_res.output), /[ \t]+/); - if (length(fields) > 0 && fields[0] != "") { - let pid = fields[0]; - if (process_running(pid, "sing-box")) return pid; - } - } - - return ""; -} - -function check_singbox_process() { - let cfg = settings(); - if (cfg.recovery_bypass == "1") return; - if (check_auto_resume_pause()) return; - if (check_tachyon_cli_running()) return; - - let list_update_pid = trim(fs.readfile("/var/run/tachyon_list_update.pid") || ""); - if (process_running(list_update_pid, "ucode")) return; - - // Fast-path check: verify if sing-box process is active - let sb_pid = get_sing_box_pid(); - if (sb_pid != "" && process_running(sb_pid, "sing-box")) { - return; - } - - let has_sections = false; - let uci_sections = uci_core.get_all(CONFIG_NAME); - if (uci_sections) { - for (let k in keys(uci_sections)) { - if (uci_sections[k][".type"] == "section") { - has_sections = true; - break; - } - } - } - if (!has_sections) return; - - let binary_name = "sing-box"; - let pid = ""; - let proc = fs.opendir("/proc"); - if (proc) { - let entry; - while ((entry = proc.read()) != null) { - if (match(entry, /^[0-9]+$/)) { - let exe = fs.readlink("/proc/" + entry + "/exe") || ""; - let slash = rindex(exe, "/"); - if ((slash >= 0 ? substr(exe, slash + 1) : exe) == binary_name) { - pid = entry; - break; - } - } - } - proc.close(); - } - - if (pid == "") { - handle_singbox_stop_event("process missing from /proc"); - } -} - -function check_memory() { - let free_mb = -1; - let mem_info = fs.readfile("/proc/meminfo") || ""; - for (let line in split(mem_info, "\n")) { - if (index(line, "MemAvailable:") == 0) { - let fields = split(trim(line), /[ \t]+/); - if (length(fields) >= 2) { - free_mb = int(fields[1]) / 1024; - } - break; - } - } - if (free_mb >= 0 && free_mb < 15) { - log_message("Low memory detected (" + free_mb + "MB). Clearing caches...", "warn"); - system("echo 3 > /proc/sys/vm/drop_caches"); - } + return controller.get_sing_box_pid(); } // ─── AI Watchdog Self-Healing Matrix ────────────────────────────────────────── @@ -469,7 +386,7 @@ let last_ai_incident = null; function ai_export_status() { let is_healthy = last_ai_incident == null || (time() - last_ai_incident.timestamp >= 300); - if (is_healthy) ai_healthy_streak++; + if (is_healthy) controller.note_healthy(); let status_obj = { timestamp: time(), status: is_healthy ? "healthy" : "repaired", @@ -482,7 +399,7 @@ function ai_export_status() { function ai_heal_report(event_type, description, resolution, status_code) { ai_incidents_count++; - ai_healthy_streak = 0; + controller.note_incident(); last_ai_incident = { type: event_type, description: description, @@ -530,7 +447,9 @@ function safe_proxy_restart(reason) { } try { fs.writefile(PROXY_RESTART_LOCK, as_string(now)); } catch(e) {} proxy_restart_count++; - cached_proxy_port = null; + // The proxy port can change across a restart, so the controller's cached + // lookup is invalidated here rather than after the fact. + controller.forget_proxy_port(); let lock_path = shell_quote(PROXY_RESTART_LOCK); bg_system("/etc/init.d/tachyon restart /dev/null 2>&1 & rm -f " + lock_path + " &"); return true; @@ -545,87 +464,44 @@ function safe_reload_firewall() { bg_system("/usr/bin/tachyon reload_firewall /dev/null 2>&1 &"); } -function ai_heal_nftables() { - let cfg = settings(); - let routing_mode = cfg.routing_mode || "nftables"; - let nft_table = getenv("NFT_TABLE_NAME") || "TachyonTable"; +// ─── Repairs ────────────────────────────────────────────────────────────────── +// Each function below is the *action* half of a former ai_heal_* function: the +// detection it used to perform now happens in the controller and arrives as a +// fact. The repair bodies themselves are unchanged, so thresholds, messages and +// side effects stay identical. - let list_update_pid = trim(fs.readfile("/var/run/tachyon_list_update.pid") || ""); - if (process_running(list_update_pid, "ucode")) return true; - if (is_reload_in_progress()) return true; - - if (routing_mode == "nftables") { - let out_nft = command_output_from_args(["nft", "list", "table", "inet", nft_table]); - if (index(out_nft, "tproxy") < 0 || index(out_nft, "priority_rules") < 0) { - ai_heal_report( - "nftables", - "Таблица правил nftables очищена или повреждена", - "Выполнена быстрая регенерация правил TachyonTable и цепочки TPROXY", - "fixed" - ); - safe_reload_firewall(); - return false; - } - } - return true; +function heal_nftables(ev) { + ai_heal_report( + "nftables", + "Таблица правил nftables очищена или повреждена", + "Выполнена быстрая регенерация правил TachyonTable и цепочки TPROXY", + "fixed" + ); + safe_reload_firewall(); } -function ai_heal_qos() { - let cfg = settings(); - if (cfg.qos_priority_engine == "0") return true; - if (is_reload_in_progress()) return true; - - let nft_table = getenv("NFT_TABLE_NAME") || "TachyonTable"; - let out_nft = command_output_from_args(["nft", "list", "table", "inet", nft_table]); - if ((index(out_nft, "dscp set 0x2e") < 0 && index(out_nft, "dscp set ef") < 0) || (index(out_nft, "dscp set 0x22") < 0 && index(out_nft, "dscp set af41") < 0)) { - ai_heal_report( - "qos_priority", - "Правила Игрового & Голосового QoS Ускорителя не найдены в nftables", - "Применены высокоприоритетные метки DSCP EF (0x2e) для Voice/RTC и DSCP AF41 (0x22) для Gaming", - "fixed" - ); - safe_reload_firewall(); - return false; - } - return true; +function heal_qos(ev) { + ai_heal_report( + "qos_priority", + "Правила Игрового & Голосового QoS Ускорителя не найдены в nftables", + "Применены высокоприоритетные метки DSCP EF (0x2e) для Voice/RTC и DSCP AF41 (0x22) для Gaming", + "fixed" + ); + safe_reload_firewall(); } -let dns_fail_streak = 0; - function is_dns_working() { - return command_success_from_args([ "nslookup", "-timeout=2", "yandex.ru", "127.0.0.1" ]) || - command_success_from_args([ "nslookup", "-timeout=2", "connectivitycheck.gstatic.com", "127.0.0.1" ]) || - command_success_from_args([ "nslookup", "-timeout=2", "google.com", "127.0.0.1" ]); + return controller.is_dns_working(); } -function ai_heal_dns() { - let cfg = settings(); - if (cfg.recovery_bypass == "1") return true; +// Threshold 3 is the original ai_heal_dns() streak: two isolated failures are +// noise, three in a row is a stall. The streak reset after acting prevents the +// in-flight restart from tripping the same threshold on the next tick. +function heal_dns_stall(ev) { + if (settings().recovery_bypass == "1") return; + if (int(ev.payload.streak) < 3) return; - if (is_reload_in_progress()) { - dns_fail_streak = 0; - return true; - } - - let sb_pid = get_sing_box_pid(); - if (sb_pid == "" || !process_running(sb_pid, "sing-box")) { - dns_fail_streak = 0; - return true; - } - - if (is_dns_working()) { - dns_fail_streak = 0; - return true; - } - - dns_fail_streak++; - log_message("Watchdog: DNS check failed (" + as_string(dns_fail_streak) + "/3 failures)", "warn"); - - if (dns_fail_streak < 3) { - return true; - } - - dns_fail_streak = 0; + controller.reset_dns_streak(); log_message("Watchdog: DNS stalled after 3 attempts, soft-reloading proxy runtime", "warn"); safe_proxy_restart("dns_stalled"); @@ -635,62 +511,24 @@ function ai_heal_dns() { "Soft-reloaded proxy runtime safely without breaking active TCP/RDP connections", "fixed" ); - return false; } -function ai_heal_proxy_connectivity() { - let cfg = settings(); - if (cfg.recovery_bypass == "1") return true; +// Only a proxy that fails while the uplink itself works is worth restarting +// over: when direct access is down too, the fault is upstream and a restart +// would drop live connections for nothing. +function heal_proxy_connectivity(ev) { + if (settings().recovery_bypass == "1") return; + if (!ev.payload.direct_ok) return; - let sb_pid = get_sing_box_pid(); - if (sb_pid == "" || !process_running(sb_pid, "sing-box")) return true; - - let now = time(); - let proxy_addr = "127.0.0.1:4534"; - if (cached_proxy_port !== null) { - proxy_addr = "127.0.0.1:" + as_string(cached_proxy_port); - } else { - let sb_cfg_data = fs.readfile("/etc/sing-box/config.json"); - if (sb_cfg_data) { - try { - let sb_cfg = json(sb_cfg_data); - if (sb_cfg.inbounds) { - for (let inb in sb_cfg.inbounds) { - if (inb.type == "http" || inb.type == "mixed") { - cached_proxy_port = inb.listen_port || 4534; - proxy_addr = "127.0.0.1:" + as_string(cached_proxy_port); - break; - } - } - } - } catch(e) {} - } - } - - let proxy_ok = command_success_from_args([ - "curl", "-s", "-I", "--connect-timeout", "3", "--max-time", "5", - "--proxy", "http://" + proxy_addr, - "https://cp.cloudflare.com/generate_204" - ]); - - if (!proxy_ok) { - let direct_ok = command_success_from_args([ - "curl", "-s", "-I", "--connect-timeout", "3", "--max-time", "5", - "https://cp.cloudflare.com/generate_204" - ]); - if (direct_ok) { - ai_heal_report( - "proxy", - "Зависание или неполный отклик прокси-порту sing-box (" + proxy_addr + ")", - "Очищена база cache.db и выполнен перезапуск sing-box", - "fixed" - ); - remove_file("/tmp/sing-box/cache.db"); - safe_proxy_restart("proxy_connectivity"); - return false; - } - } - return true; + let proxy_addr = "127.0.0.1:" + as_string(ev.payload.port); + ai_heal_report( + "proxy", + "Зависание или неполный отклик прокси-порту sing-box (" + proxy_addr + ")", + "Очищена база cache.db и выполнен перезапуск sing-box", + "fixed" + ); + remove_file("/tmp/sing-box/cache.db"); + safe_proxy_restart("proxy_connectivity"); } // ─── Subnet cache restore: /etc/tachyon/rulesets/ → /tmp/sing-box/rulesets/ ── @@ -729,144 +567,37 @@ function ai_heal_subnet_cache() { } // ─── Check nft sets are populated (community subnets) ───────────────────────── -function ai_heal_community_subnet_sets() { - let cfg = settings(); - if (cfg.recovery_bypass == "1") return true; +function heal_community_subnet_sets(ev) { + if (settings().recovery_bypass == "1") return; - // Don't run more than once every 5 minutes - let now = time(); - if (now - last_subnet_heal_time < 300) return true; - if (is_reload_in_progress()) return true; + for (let set_name in common.array_or_empty(ev.payload.sets)) + log_message("Community subnet set " + set_name + " is empty — will repopulate", "warn"); - // Skip if list-update is running (it will populate sets itself) - let list_update_pid = trim(fs.readfile("/var/run/tachyon_list_update.pid") || ""); - if (process_running(list_update_pid, "ucode")) return true; - - let sb_pid = get_sing_box_pid(); - if (sb_pid == "" || !process_running(sb_pid, "sing-box")) return true; - - let nft_table = getenv("NFT_TABLE_NAME") || "TachyonTable"; - let all_sections = uci_core.get_all(CONFIG_NAME); - if (!all_sections) return true; - - let empty_sets_found = false; - - for (let sec_name in keys(all_sections)) { - let s = all_sections[sec_name]; - if (s[".type"] != "section" || s.enabled != "1") continue; - if (!s.community_lists) continue; - - // Check if the _subnets set exists for this section. - // If nft returns a valid set definition but no 'elements =' block → it's empty. - // If nft exits non-zero (set doesn't exist) → section has no subnet community → skip. - let set_name = "tachyon_rule_" + sec_name + "_subnets"; - let result = command_capture(command_from_args(["nft", "list", "set", "inet", nft_table, set_name]) + " 2>/dev/null"); - if (result.status != 0 || result.output == "") continue; // set doesn't exist for this section - if (index(result.output, "elements") < 0) { - log_message("Community subnet set " + set_name + " is empty — will repopulate", "warn"); - empty_sets_found = true; - } - } - - if (empty_sets_found) { - last_subnet_heal_time = now; - ai_heal_subnet_cache(); - ai_heal_report( - "nft_community_sets", - "Пустые nftables sets подсетей (community) — данные не были загружены при reload", - "Восстановлены nftables sets из persistent кеша (/etc/tachyon/rulesets/)", - "fixed" - ); - bg_system("/usr/bin/tachyon reload_firewall /dev/null 2>&1 &"); - return false; - } - return true; + ai_heal_subnet_cache(); + ai_heal_report( + "nft_community_sets", + "Пустые nftables sets подсетей (community) — данные не были загружены при reload", + "Восстановлены nftables sets из persistent кеша (/etc/tachyon/rulesets/)", + "fixed" + ); + bg_system("/usr/bin/tachyon reload_firewall /dev/null 2>&1 &"); } -// ─── TPROXY port liveness check ─────────────────────────────────────────────── -function ai_heal_tproxy_port() { - let cfg = settings(); - if (cfg.recovery_bypass == "1") return true; +// ─── TPROXY port liveness ───────────────────────────────────────────────────── +function heal_tproxy_port(ev) { + if (settings().recovery_bypass == "1") return; - let sb_pid = get_sing_box_pid(); - if (sb_pid == "" || !process_running(sb_pid, "sing-box")) return true; - if (is_reload_in_progress()) return true; - - // Read TPROXY port from sing-box config - let tproxy_port = 4530; - let sb_cfg_data = fs.readfile("/etc/sing-box/config.json"); - if (sb_cfg_data) { - try { - let sb_cfg = json(sb_cfg_data); - if (sb_cfg.inbounds) { - for (let inb in sb_cfg.inbounds) { - if (inb.type == "tproxy") { - tproxy_port = int(inb.listen_port || tproxy_port); - break; - } - } - } - } catch(e) {} - } - - // Check /proc/net/tcp6 and /proc/net/tcp for the TPROXY port (hex format) - let hex_port = sprintf("%04X", tproxy_port); - let listening = false; - for (let proc_file in ["/proc/net/tcp6", "/proc/net/tcp"]) { - let tcp_data = fs.readfile(proc_file) || ""; - // Format: local_address (0.0.0.0:PORT in hex), state 0A = LISTEN - for (let line in split(tcp_data, "\n")) { - if (index(line, ":" + hex_port + " ") >= 0 && index(line, " 0A ") >= 0) { - listening = true; - break; - } - } - if (listening) break; - } - - if (!listening) { - ai_heal_report( - "tproxy_port", - sprintf("TPROXY порт %d не слушает — правила перехвата трафика не работают", tproxy_port), - "Выполнен reload_firewall для восстановления TPROXY правил", - "fixed" - ); - safe_reload_firewall(); - return false; - } - return true; + ai_heal_report( + "tproxy_port", + sprintf("TPROXY порт %d не слушает — правила перехвата трафика не работают", int(ev.payload.port)), + "Выполнен reload_firewall для восстановления TPROXY правил", + "fixed" + ); + safe_reload_firewall(); } -// ─── Fixes undefined function referenced from ubus firewall.reload handler ──── -function check_firewall_rules() { - ai_heal_nftables(); - ai_heal_community_subnet_sets(); -} - -function ai_heal_wan_and_gateway() { - let now = time(); - if (now - last_wan_heal_time < 300) return; - if (is_reload_in_progress()) return; - - let need_restart = false; - - // Check WAN interface - let proto = uci_core.get("network", "wan", "proto") || "pppoe"; - let device = trim(uci_core.get("network", "wan", "device") || "eth0"); - let iface_to_check = device; - if (proto == "pppoe") iface_to_check = "pppoe-wan"; - - let out = command_capture("ip addr show " + shell_quote(iface_to_check) + " 2>/dev/null").output; - if (index(out, "inet ") < 0) need_restart = true; - - // Check gateway - let route_out = command_capture("ip route 2>/dev/null").output; - if (index(route_out, "default") < 0) need_restart = true; - - if (!need_restart) return; - - last_wan_heal_time = now; +function heal_wan_and_gateway(ev) { let tcfg = common.object_or_empty(uci_core.get_all(CONFIG_NAME, "telegram")); if (tcfg.notify_crash != "0") { send_telegram_notification("⚠️ *Watchdog:* WAN/Gateway проблема. Перезапуск wan..."); @@ -874,43 +605,32 @@ function ai_heal_wan_and_gateway() { bg_system("/sbin/ifdown wan >/dev/null 2>&1 && /sbin/ifup wan >/dev/null 2>&1 &"); } -function ai_heal_subscriptions() { - let cfg = settings(); - let sub_url = trim(cfg.subscription_url || ""); - if (sub_url == "") return; - - let res = command_capture("curl -s -o /dev/null -w %{http_code} --connect-timeout 10 " + shell_quote(sub_url) + " 2>&1"); - let code = int(res.output); - if (res.status == 0 && code >= 200 && code < 400) return; - +function heal_subscriptions(ev) { let tcfg = common.object_or_empty(uci_core.get_all(CONFIG_NAME, "telegram")); if (tcfg.notify_crash != "0") { - send_telegram_notification("⚠️ *Watchdog:* Подписка недоступна (HTTP " + code + "). Обновите подписку вручную."); + send_telegram_notification("⚠️ *Watchdog:* Подписка недоступна (HTTP " + int(ev.payload.code) + "). Обновите подписку вручную."); } } -function ai_heal_uci_config() { - let data = fs.readfile("/etc/config/tachyon"); - if (data == null || data == "") { - let tcfg = common.object_or_empty(uci_core.get_all(CONFIG_NAME, "telegram")); - if (tcfg.notify_crash != "0") { - send_telegram_notification("⚠️ *Watchdog:* Конфигурация Tachyon повреждена! Восстановление из backup..."); - } - let backup = fs.readfile("/etc/backup/tachyon_config"); - if (backup != null && backup != "") { - let valid = command_success_from_args([ "uci", "-c", "/etc/config", "valid", CONFIG_NAME ]) || - command_success_from_args([ "/sbin/uci", "valid", CONFIG_NAME ]); - if (!valid) { - log_message("Backup config also invalid, skipping restore", "warn"); - return; - } - let tmp = "/etc/config/tachyon.restore-tmp"; - if (fs.writefile(tmp, backup) != null) { - fs.rename(tmp, "/etc/config/tachyon"); - system("chmod 0600 /etc/config/tachyon 2>/dev/null"); - safe_proxy_restart("uci_config_restore"); - } - } +function heal_uci_config(ev) { + let tcfg = common.object_or_empty(uci_core.get_all(CONFIG_NAME, "telegram")); + if (tcfg.notify_crash != "0") { + send_telegram_notification("⚠️ *Watchdog:* Конфигурация Tachyon повреждена! Восстановление из backup..."); + } + let backup = fs.readfile("/etc/backup/tachyon_config"); + if (backup == null || backup == "") return; + + let valid = command_success_from_args([ "uci", "-c", "/etc/config", "valid", CONFIG_NAME ]) || + command_success_from_args([ "/sbin/uci", "valid", CONFIG_NAME ]); + if (!valid) { + log_message("Backup config also invalid, skipping restore", "warn"); + return; + } + let tmp = "/etc/config/tachyon.restore-tmp"; + if (fs.writefile(tmp, backup) != null) { + fs.rename(tmp, "/etc/config/tachyon"); + system("chmod 0600 /etc/config/tachyon 2>/dev/null"); + safe_proxy_restart("uci_config_restore"); } } @@ -930,154 +650,68 @@ function validate_singbox_config() { return command_success_from_args(["sing-box", "check", "-c", "/etc/sing-box/config.json"]); } -// ─── Proxy Health Monitor (fast tier) ───────────────────────────────────────── -function ai_heal_proxy_health() { +// ─── Proxy Health Monitor ───────────────────────────────────────────────────── +// The second proxy subscriber: where heal_proxy_connectivity() reacts to a +// single failure with a working uplink, this one waits for a configurable run +// of failures and validates the sing-box config before restarting. +function heal_proxy_health(ev) { if (!ai_enabled("ai_proxy_health_enabled", "1")) return; - if (is_reload_in_progress()) return; - let sb_pid = get_sing_box_pid(); - if (sb_pid == "" || !process_running(sb_pid, "sing-box")) return; - - let proxy_port = "4534"; - let sb_cfg_data = fs.readfile("/etc/sing-box/config.json"); - if (sb_cfg_data) { - try { - let sb_cfg = json(sb_cfg_data); - if (sb_cfg.inbounds) { - for (let inb in sb_cfg.inbounds) { - if (inb.type == "http" || inb.type == "mixed") { - proxy_port = as_string(inb.listen_port || 4534); - break; - } - } - } - } catch(e) {} - } - - let check_url = ai_setting("ai_proxy_health_url", "https://cp.cloudflare.com/generate_204"); let threshold = int(ai_setting("ai_proxy_health_fail_threshold", "3")); + let fails = int(ev.payload.streak); + if (fails < threshold) return; - let start_time = time(); - let proxy_ok = command_success_from_args([ - "curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", "--connect-timeout", "3", "--max-time", "5", - "--proxy", "http://127.0.0.1:" + proxy_port, - check_url - ]); - let elapsed = (time() - start_time) * 1000; - - push(proxy_latency_history, { ok: proxy_ok, ms: elapsed, ts: time() }); - if (length(proxy_latency_history) > 20) { - let new_arr = []; - for (let i = length(proxy_latency_history) - 20; i < length(proxy_latency_history); i++) - push(new_arr, proxy_latency_history[i]); - proxy_latency_history = new_arr; - } - - if (!proxy_ok) { - proxy_consecutive_fails++; - if (proxy_consecutive_fails >= threshold) { - ai_heal_report( - "proxy_health", - sprintf("Proxy health check failed %d times consecutively (port %s)", proxy_consecutive_fails, proxy_port), - "Restarting Tachyon to restore proxy connectivity", - "fixed" - ); - if (ai_enabled("ai_config_validation_enabled", "1") && !validate_singbox_config()) { - log_message("sing-box config validation failed before proxy restart", "err"); - return; - } - safe_proxy_restart("proxy_health"); - proxy_consecutive_fails = 0; - } - } else { - proxy_consecutive_fails = 0; + ai_heal_report( + "proxy_health", + sprintf("Proxy health check failed %d times consecutively (port %s)", fails, as_string(ev.payload.port)), + "Restarting Tachyon to restore proxy connectivity", + "fixed" + ); + // A restart on a config sing-box will refuse to load leaves the proxy down + // for good, so validation gates the restart, not the report. + if (ai_enabled("ai_config_validation_enabled", "1") && !validate_singbox_config()) { + log_message("sing-box config validation failed before proxy restart", "err"); + return; } + safe_proxy_restart("proxy_health"); + controller.reset_proxy_consecutive(); } -// ─── DNS Continuous Check (fast tier) ───────────────────────────────────────── -function ai_heal_dns_continuous() { +// ─── DNS Continuous Check ───────────────────────────────────────────────────── +// The second DNS subscriber: restores dnsmasq's own configuration rather than +// restarting the proxy, so it is complementary to heal_dns_stall(). +function heal_dns_continuous(ev) { if (!ai_enabled("ai_dns_continuous_enabled", "1")) return; - if (is_reload_in_progress()) return; + if (int(ev.payload.consecutive) < 3) return; - let sb_pid = get_sing_box_pid(); - if (sb_pid == "" || !process_running(sb_pid, "sing-box")) return; - - let start_time = time(); - let dns_ok = is_dns_working(); - let elapsed = (time() - start_time) * 1000; - - push(dns_latency_history, { ok: dns_ok, ms: elapsed, ts: time() }); - if (length(dns_latency_history) > 20) { - let new_arr = []; - for (let i = length(dns_latency_history) - 20; i < length(dns_latency_history); i++) - push(new_arr, dns_latency_history[i]); - dns_latency_history = new_arr; - } - - if (!dns_ok) { - dns_consecutive_fails++; - if (dns_consecutive_fails >= 3) { - ai_heal_report( - "dns_continuous", - "DNS resolution failed 3 times consecutively", - "Restoring dnsmasq config and reloading", - "fixed" - ); - system("/sbin/uci set dhcp.@dnsmasq[0].noresolv='1' >/dev/null 2>&1"); - system("/sbin/uci commit dhcp >/dev/null 2>&1"); - system("/etc/init.d/dnsmasq reload >/dev/null 2>&1"); - dns_consecutive_fails = 0; - } - } else { - dns_consecutive_fails = 0; - } + ai_heal_report( + "dns_continuous", + "DNS resolution failed 3 times consecutively", + "Restoring dnsmasq config and reloading", + "fixed" + ); + system("/sbin/uci set dhcp.@dnsmasq[0].noresolv='1' >/dev/null 2>&1"); + system("/sbin/uci commit dhcp >/dev/null 2>&1"); + system("/etc/init.d/dnsmasq reload >/dev/null 2>&1"); + controller.reset_dns_consecutive(); } // ─── AI Empty Sections Recovery ───────────────────────────────────────────── -let last_section_heal_attempt = 0; -function ai_heal_empty_sections() { - if (!ai_enabled("ai_section_failover_enabled", "1")) return; - if (is_reload_in_progress()) return; - let now = time(); - if (now - last_section_heal_attempt < 120) return; - last_section_heal_attempt = now; +function heal_empty_sections(ev) { + let recovered = common.array_or_empty(ev.payload.sections); + if (length(recovered) == 0) return; - let cfg = settings(); - let sections = common.object_or_empty(uci_core.get_all(CONFIG_NAME)); - let recovered = []; - - for (let name in sections) { - let s = common.object_or_empty(sections[name]); - let action = common.as_string(s.action || ""); - if (!connections.is_connections_action(action)) continue; - let sub_urls = common.list_option(s, "subscription_url"); - if (length(sub_urls) == 0) continue; - - let cache_path = "/var/run/tachyon/section-cache/" + name + ".json"; - let cache = common.object_or_empty(common.read_json_file(cache_path)); - let servers = common.array_or_empty(cache.servers); - let urls = common.array_or_empty(cache.urls); - let selector_urls = common.array_or_empty(cache.selector_urls); - let domains = common.array_or_empty(cache.domain); - let domain_suffixes = common.array_or_empty(cache.domain_suffix); - let ip_cidrs = common.array_or_empty(cache.ip_cidr); - let usable = length(servers) + length(urls) + length(selector_urls) + length(domains) + length(domain_suffixes) + length(ip_cidrs); - if (usable > 0) continue; - - // Section with subscriptions but 0 usable outbounds — trigger async reload + for (let name in recovered) { log_message("Empty proxy section '" + name + "' — triggering subscription update", "warn"); bg_system("/usr/bin/tachyon subscription_update_async " + common.shell_quote(name) + " /dev/null 2>&1 1000<&- &"); - push(recovered, name); } - if (length(recovered) > 0) { - ai_heal_report( - "empty_proxy_sections", - "Empty proxy sections detected: " + join(", ", recovered), - "Auto-triggering subscription update for " + as_string(length(recovered)) + " section(s)", - "fixed" - ); - } + ai_heal_report( + "empty_proxy_sections", + "Empty proxy sections detected: " + join(", ", recovered), + "Auto-triggering subscription update for " + as_string(length(recovered)) + " section(s)", + "fixed" + ); } // ─── DNS Loop Recovery (dead proxy + DNS detour = total DNS loss) ──────────── @@ -1101,13 +735,14 @@ function remove_dns_recovery_state() { try { fs.unlink(DNS_RECOVERY_STATE_FILE); } catch(e) {} } -let last_dns_loop_heal_attempt = 0; +// Kept as a self-contained check rather than a fact subscriber: it needs the +// *current* resolver state at the moment it acts (it re-tests DNS to decide +// whether to re-enable the detour), so acting on a fact observed earlier in the +// tick could re-enable a detour against a resolver that has since died. function ai_heal_dns_loop() { if (!ai_enabled("ai_dns_loop_heal_enabled", "1")) return; if (is_reload_in_progress()) return; let now = time(); - if (now - last_dns_loop_heal_attempt < 60) return; - last_dns_loop_heal_attempt = now; let cfg = settings(); let detour_enabled = common.bool_option(cfg, "dns_detour_enabled", false); @@ -1210,16 +845,16 @@ function export_metrics() { let hour_bucket = int(now / 3600) * 3600; let last_bucket = length(data.hours) > 0 ? data.hours[length(data.hours) - 1] : null; if (last_bucket && int(last_bucket.ts) == hour_bucket) { - last_bucket.proxy_ok = proxy_consecutive_fails == 0; - last_bucket.proxy_lat_ms = average_latency(proxy_latency_history); - last_bucket.dns_lat_ms = average_latency(dns_latency_history); + last_bucket.proxy_ok = proxy_consecutive_fails() == 0; + last_bucket.proxy_lat_ms = average_latency(proxy_latency_history()); + last_bucket.dns_lat_ms = average_latency(dns_latency_history()); last_bucket.incidents = ai_incidents_count; } else { push(data.hours, { ts: hour_bucket, - proxy_ok: proxy_consecutive_fails == 0, - proxy_lat_ms: average_latency(proxy_latency_history), - dns_lat_ms: average_latency(dns_latency_history), + proxy_ok: proxy_consecutive_fails() == 0, + proxy_lat_ms: average_latency(proxy_latency_history()), + dns_lat_ms: average_latency(dns_latency_history()), incidents: ai_incidents_count }); } @@ -1231,28 +866,18 @@ function export_metrics() { fs.writefile(metrics_path, sprintf("%J\n", data)); } -// ─── Anomaly Detection (slow tier) ──────────────────────────────────────────── -function analyze_anomalies() { +// ─── Anomaly Detection ──────────────────────────────────────────────────────── +function heal_anomaly_reconnects(ev) { if (!ai_enabled("ai_anomaly_detection_enabled", "1")) return; - let now = time(); - if (now - last_anomaly_check < 300) return; - last_anomaly_check = now; - - let threshold = int(ai_setting("ai_anomaly_reconnect_threshold", "10")); - let count_file = "/tmp/tachyon_reconnect_count"; - let count_data = fs.readfile(count_file) || "0"; - let reconnects = int(trim(count_data)); - - if (reconnects > threshold) { - ai_heal_report( - "anomaly_reconnects", - sprintf("sing-box reconnected %d times in the last hour (threshold: %d)", reconnects, threshold), - "High reconnect rate detected. Check proxy server health or ISP stability.", - "warn" - ); - fs.writefile(count_file, "0\n"); - } + ai_heal_report( + "anomaly_reconnects", + sprintf("sing-box reconnected %d times in the last hour (threshold: %d)", + int(ev.payload.count), int(ev.payload.threshold)), + "High reconnect rate detected. Check proxy server health or ISP stability.", + "warn" + ); + fs.writefile("/tmp/tachyon_reconnect_count", "0\n"); } function increment_reconnect_count() { @@ -1263,11 +888,13 @@ function increment_reconnect_count() { // ─── Adaptive Intervals ─────────────────────────────────────────────────────── function adaptive_normal_interval() { - if (!ai_enabled("ai_adaptive_intervals_enabled", "1")) return 120; - return ai_healthy_streak > 25 ? 300 : 120; + return controller.adaptive_normal_interval(); } // ─── Graceful Degradation wrapper ───────────────────────────────────────────── +// Still used for the few actions that are not bus subscribers (the metrics +// export, smart-detect batch processing, the ai-heal CLI audit). Subscribers +// get the same isolation from the bus itself. function safe_call(fn, name) { if (!ai_enabled("ai_graceful_degradation_enabled", "1")) { fn(); @@ -1280,70 +907,29 @@ function safe_call(fn, name) { } } -// ─── rpcd FD leak watchdog ──────────────────────────────────────────────────── -// rpcd accumulates file descriptors over time from LuCI API calls. -// When FD count approaches the process limit (~1024), it can no longer -// fork to execute /usr/bin/tachyon → LuCI shows everything as "stopped". -// Threshold 512 = 50% of limit, safe to restart without user impact. -const RPCD_FD_THRESHOLD = 512; - -function ai_heal_rpcd() { - // Find rpcd PID via /proc scan (avoid shell fork for pidof) - let rpcd_pid = null; - let proc_dir = fs.opendir("/proc"); - if (!proc_dir) return true; - let entry; - while ((entry = proc_dir.read()) != null) { - if (!match(entry, /^[0-9]+$/)) continue; - let comm = trim(fs.readfile("/proc/" + entry + "/comm") || ""); - if (comm == "rpcd") { rpcd_pid = entry; break; } - } - proc_dir.close(); - if (!rpcd_pid) return true; - - // Count open file descriptors - let fd_dir = fs.opendir("/proc/" + rpcd_pid + "/fd"); - if (!fd_dir) return true; - let fd_count = 0; - while ((entry = fd_dir.read()) != null) { - if (match(entry, /^[0-9]+$/)) fd_count++; - } - fd_dir.close(); - - if (fd_count > RPCD_FD_THRESHOLD) { - ai_heal_report( - "rpcd_fd_leak", - sprintf("rpcd накопил %d открытых FD (порог %d/1024) — LuCI не может запускать команды", fd_count, RPCD_FD_THRESHOLD), - "Выполнен перезапуск rpcd для освобождения файловых дескрипторов", - "fixed" - ); - system("/etc/init.d/rpcd restart /dev/null 2>&1"); - return false; - } - return true; +// ─── rpcd FD leak ───────────────────────────────────────────────────────────── +function heal_rpcd(ev) { + ai_heal_report( + "rpcd_fd_leak", + sprintf("rpcd накопил %d открытых FD (порог %d/1024) — LuCI не может запускать команды", + int(ev.payload.fd_count), int(ev.payload.threshold)), + "Выполнен перезапуск rpcd для освобождения файловых дескрипторов", + "fixed" + ); + system("/etc/init.d/rpcd restart /dev/null 2>&1"); } -// ─── Full health audit ──────────────────────────────────────────────────────── -function ai_full_health_audit() { - safe_call(check_memory, "check_memory"); - safe_call(ai_heal_rpcd, "ai_heal_rpcd"); - safe_call(ai_heal_nftables, "ai_heal_nftables"); - safe_call(ai_heal_qos, "ai_heal_qos"); - safe_call(ai_heal_dns, "ai_heal_dns"); - safe_call(ai_heal_proxy_connectivity, "ai_heal_proxy_connectivity"); - safe_call(ai_heal_community_subnet_sets, "ai_heal_community_subnet_sets"); - safe_call(ai_heal_wan_and_gateway, "ai_heal_wan_and_gateway"); - safe_call(ai_heal_subscriptions, "ai_heal_subscriptions"); - safe_call(ai_heal_uci_config, "ai_heal_uci_config"); - safe_call(ai_heal_tproxy_port, "ai_heal_tproxy_port"); - safe_call(ai_export_status, "ai_export_status"); +// ─── Low memory ─────────────────────────────────────────────────────────────── +// Deliberately silent: dropping caches is routine housekeeping, not an +// incident, so it never called ai_heal_report() and still does not. +function heal_low_memory(ev) { + log_message("Low memory detected (" + as_string(ev.payload.free_mb) + "MB). Clearing caches...", "warn"); + system("echo 3 > /proc/sys/vm/drop_caches"); } -function check_urltest_switches() { - let now = time(); - if (now - last_urltest_check < 5) return; - last_urltest_check = now; - +// The 5s throttle that used to live here is now the controller's emit_once +// window on urltest.switched, so a burst of log lines still costs one poll. +function notify_urltest_switch(ev) { let tcfg = common.object_or_empty(uci_core.get_all(CONFIG_NAME, "telegram")); if (tcfg.enabled != "1" || tcfg.notify_crash == "0") return; @@ -1489,65 +1075,43 @@ function smart_detect_process_pending() { fs.writefile(SMART_DETECT_SEEN_FILE, sprintf("%J", clean)); } -function handle_log_line(line) { - if (!line || line == "") return; - - // Fast keyword pre-filter: skip 95%+ of irrelevant log lines instantly - if (index(line, "direct") < 0 && index(line, "DIRECT") < 0 && - index(line, "memory") < 0 && index(line, "oom") < 0 && index(line, "OOM") < 0 && - index(line, "URLTest") < 0 && index(line, "proxy") < 0) { - return; +// ─── OOM response ───────────────────────────────────────────────────────────── +// Shrinks the GOMEMLIMIT scale by 20% per OOM, floored at 0.2 so sing-box is +// never starved into permanent failure, then restarts to apply it. +function heal_oom(ev) { + log_message("OOM event detected from syslog! Reducing GOMEMLIMIT scaling...", "err"); + send_telegram_notification("🚨 *Watchdog:* Обнаружено событие OOM (Out Of Memory)! Уменьшаю GOMEMLIMIT и перезапускаю службы..."); + let scale = 1.0; + let scale_path = "/etc/tachyon/mem_scale"; + let scale_data = fs.readfile(scale_path); + if (scale_data != null) { + let parsed_scale = double(trim(as_string(scale_data))); + if (parsed_scale > 0.1) scale = parsed_scale; } - let line_lower = lc(line); + let new_scale = scale * 0.8; + if (new_scale < 0.2) new_scale = 0.2; + fs.mkdir("/etc/tachyon"); + fs.writefile(scale_path, sprintf("%.2f", new_scale)); + system("logread -c >/dev/null 2>&1"); + command_status("/usr/bin/tachyon restart >/dev/null 2>&1"); +} - // 1. OOM Detection (Kernel OOM-killer or sing-box process OOM crash only) - let is_oom = (index(line_lower, "oom-killer") >= 0 || - index(line_lower, "out of memory: kill process") >= 0 || - (index(line_lower, "kernel:") >= 0 && index(line_lower, "out of memory") >= 0) || - (index(line_lower, "sing-box") >= 0 && index(line_lower, "out of memory") >= 0) || - index(line_lower, "fatal error: out of memory") >= 0); - let is_netlink_warning = (index(line_lower, "netlink") >= 0 || index(line_lower, "nlbwmon") >= 0); +// Candidate domains are queued here and probed in a batch by +// smart_detect_process_pending(): probing inside the log handler would block +// the event loop on curl for every failing connection. +function collect_smart_detect_candidate(ev) { + pending_smart_domains[ev.payload.domain] = time(); +} - if (is_oom && !is_netlink_warning) { - let now = time(); - // Ignore replay of old historical log buffer dumped when logread -f starts - if (syslog_start_time > 0 && (now - syslog_start_time > 3) && (now - last_oom_time > 60)) { - last_oom_time = now; - log_message("OOM event detected from syslog! Reducing GOMEMLIMIT scaling...", "err"); - send_telegram_notification("🚨 *Watchdog:* Обнаружено событие OOM (Out Of Memory)! Уменьшаю GOMEMLIMIT и перезапускаю службы..."); - let scale = 1.0; - let scale_path = "/etc/tachyon/mem_scale"; - let scale_data = fs.readfile(scale_path); - if (scale_data != null) { - let parsed_scale = double(trim(as_string(scale_data))); - if (parsed_scale > 0.1) scale = parsed_scale; - } - let new_scale = scale * 0.8; - if (new_scale < 0.2) new_scale = 0.2; - fs.mkdir("/etc/tachyon"); - fs.writefile(scale_path, sprintf("%.2f", new_scale)); - system("logread -c >/dev/null 2>&1"); - command_status("/usr/bin/tachyon restart >/dev/null 2>&1"); - } - return; - } - - // 2. Smart Detect candidate domain extraction +// ─── Honeypot ───────────────────────────────────────────────────────────────── +// The controller has already validated the address shape, so nothing +// unvalidated reaches the nft command line here. +function heal_honeypot_hit(ev) { let cfg = settings(); - if (cfg.smart_detect == "1") { - if ((index(line_lower, "direct") >= 0 || index(line_lower, "DIRECT") >= 0) && - (index(line_lower, "failed") >= 0 || index(line_lower, "timeout") >= 0 || index(line_lower, "reset") >= 0)) { - let domain = smart_detect_extract_domain(line); - if (domain != null) { - pending_smart_domains[domain] = time(); - } - } - } - - // 3. URLTest proxy switch notifications - if (index(line, "URLTest") >= 0 || index(line_lower, "selected proxy") >= 0 || index(line_lower, "switch proxy") >= 0) { - check_urltest_switches(); - } + let ttl = cfg.honeypot_ttl || "86400"; + let nft_table = getenv("NFT_TABLE_NAME") || "TachyonTable"; + command_success_from_args(["nft", "add", "element", "inet", nft_table, "tachyon_honeypot", + "{", ev.payload.ip, "timeout", ttl + "s", "}"]); } function setup_honeypot_listener() { @@ -1576,13 +1140,7 @@ function setup_honeypot_listener() { uloop.handle(fifo_fd.fileno(), function(events) { let line; while ((line = fifo_fd.read("line")) != null) { - let ip = trim(as_string(line)); - if (ip != "" && match(ip, /^[0-9a-fA-F:.]+$/) != null) { - let cfg = settings(); - let ttl = cfg.honeypot_ttl || "86400"; - let nft_table = getenv("NFT_TABLE_NAME") || "TachyonTable"; - command_success_from_args(["nft", "add", "element", "inet", nft_table, "tachyon_honeypot", "{", ip, "timeout", ttl + "s", "}"]); - } + controller.handle_honeypot_line(line); } }, uloop.ULOOP_READ); } catch (e) { @@ -1600,21 +1158,16 @@ function setup_honeypot_listener() { } } -function recover_oom_scale() { +// The "OOM was long enough ago" and "scale is below 1.0" tests are the +// controller's oom.recovery probe; what is left here is the write-back. +// The 600s self-cooldown stays local because it paces the healer, not the fact. +function recover_oom_scale(ev) { let now = time(); - if (now - last_oom_time < 1800) return; - if (last_oom_time == 0) return; if (now - last_oom_recovery_time < 600) return; last_oom_recovery_time = now; let scale_path = "/etc/tachyon/mem_scale"; - let scale_data = fs.readfile(scale_path); - if (scale_data == null) return; - let current_scale = double(trim(as_string(scale_data))); - if (current_scale >= 1.0 || current_scale < 0.1) { - if (current_scale >= 1.0) try { fs.unlink(scale_path); } catch(e) {} - return; - } + let current_scale = double(ev.payload.scale); let new_scale = current_scale + 0.05; if (new_scale > 1.0) new_scale = 1.0; log_message("OOM recovery: restoring GOMEMLIMIT scale from " + sprintf("%.2f", current_scale) + " to " + sprintf("%.2f", new_scale), "info"); @@ -1623,7 +1176,7 @@ function recover_oom_scale() { function setup_syslog_listener() { if (!uloop) return null; - syslog_start_time = time(); + controller.set_syslog_start(time()); // Kill orphaned logread -f processes from previous watchdog instances. // Without this, every restart cascades: new watchdog inherits old watchdog's // logread pipe read-end, keeping old logread alive. Over N restarts, @@ -1639,7 +1192,7 @@ function setup_syslog_listener() { uloop.handle(log_pipe.fileno(), function(events) { let line; while ((line = log_pipe.read("line")) != null) { - handle_log_line(trim(as_string(line))); + controller.handle_log_line(trim(as_string(line))); } }, uloop.ULOOP_READ); } catch (e) { @@ -1656,17 +1209,15 @@ function setup_ubus_listener() { try { conn.listener("service.instance.stop", function(ev, msg) { - if (type(msg) == "object" && msg.name == "sing-box") { - handle_singbox_stop_event("ubus service.instance.stop event"); - } + if (type(msg) == "object") + controller.handle_ubus_service_stop(msg.name, "ubus service.instance.stop event"); }); conn.listener("service.stop", function(ev, msg) { - if (type(msg) == "object" && msg.name == "sing-box") { - handle_singbox_stop_event("ubus service.stop event"); - } + if (type(msg) == "object") + controller.handle_ubus_service_stop(msg.name, "ubus service.stop event"); }); conn.listener("firewall.reload", function(ev, msg) { - check_firewall_rules(); + controller.handle_ubus_firewall_reload(); }); } catch (e) { log_message("Failed to register ubus listeners: " + as_string(e), "warn"); @@ -1674,9 +1225,127 @@ function setup_ubus_listener() { return conn; } +// Idempotent: the ai-heal CLI mode and worker() both need the subscriptions in +// place, and registering twice would double every repair. +let subscribers_registered = false; + +// ─── Subscriptions ──────────────────────────────────────────────────────────── +// One table, one place: what the watchdog reacts to and in what order. Priority +// encodes the old tier ordering — restore the uplink before blaming the proxy, +// restore the proxy before touching DNS — so a single tick that observes several +// faults repairs them bottom-up like the sequential check list used to. +// +// `cooldown` values are the module globals the old code hand-rolled: +// heal_wan_and_gateway ← last_wan_heal_time (300s) +// heal_community_subnet… ← last_subnet_heal_time (300s) +// heal_empty_sections ← last_section_heal_attempt (120s) +// heal_anomaly_reconnects ← last_anomaly_check (300s) +// Debounces that belong to the repair rather than the fact (the 30s sing-box +// restart floor, the 600s OOM-recovery pacing) stay inside their handlers. +// A handler that is not a function makes bus.on() return false and register +// nothing — the repair would then silently never run, with no error anywhere. +// ucode does not hoist function declarations, so that is exactly what happens if +// a heal_* is moved below this function. subscribe() turns that silent loss into +// a startup failure. +function subscribe(event_type, handler, opts) { + if (!bus.on(event_type, handler, opts)) { + let label = (type(opts) == "object" && opts.name != null) ? opts.name : as_string(event_type); + log_message("FATAL: subscriber " + label + " for " + as_string(event_type) + + " could not be registered; this repair would never run", "err"); + die("watchdog: failed to register subscriber " + label); + } +} + +function register_subscribers() { + if (subscribers_registered) return; + subscribers_registered = true; + + subscribe(EV.WAN_DOWN, heal_wan_and_gateway, + { name: "heal_wan_and_gateway", priority: 10, cooldown: 300 }); + + subscribe(EV.SINGBOX_STOPPED, heal_singbox_stopped, + { name: "heal_singbox_stopped", priority: 20 }); + + subscribe(EV.CONFIG_CORRUPT, heal_uci_config, + { name: "heal_uci_config", priority: 25 }); + + subscribe(EV.NFT_MISSING, heal_nftables, + { name: "heal_nftables", priority: 30 }); + subscribe(EV.QOS_MISSING, heal_qos, + { name: "heal_qos", priority: 30 }); + subscribe(EV.TPROXY_DOWN, heal_tproxy_port, + { name: "heal_tproxy_port", priority: 30 }); + subscribe(EV.SUBNETS_EMPTY, heal_community_subnet_sets, + { name: "heal_community_subnet_sets", priority: 30, cooldown: 300 }); + + // Both proxy subscribers see the same fact, but they used to be measured on + // different tiers, and their thresholds are calibrated to those rates: + // ai_heal_proxy_health ran fast (15s), ai_heal_proxy_connectivity ran inside + // the normal-tier audit (120s+). The probe now runs once at the fast rate, + // so the connectivity healer carries a 120s cooldown to keep its original + // pacing. Without it a single stall would restart sing-box eight times as + // often — the aggressive-restart regression fixed in c8052ee6. + subscribe(EV.PROXY_DOWN, heal_proxy_connectivity, + { name: "heal_proxy_connectivity", priority: 40, cooldown: 120 }); + subscribe(EV.PROXY_DOWN, heal_proxy_health, + { name: "heal_proxy_health", priority: 41 }); + + // Same split for DNS: ai_heal_dns_continuous was fast, ai_heal_dns was part + // of the normal-tier audit. The streak the stall healer thresholds on is + // already paced at the normal rate inside the probe (DNS_STREAK_INTERVAL); + // the cooldown here is a second bound on how often a restart can be issued, + // so a streak that stays at 3 cannot restart the proxy every 15 seconds. + subscribe(EV.DNS_DOWN, heal_dns_stall, + { name: "heal_dns_stall", priority: 50, cooldown: 120 }); + subscribe(EV.DNS_DOWN, heal_dns_continuous, + { name: "heal_dns_continuous", priority: 51 }); + + subscribe(EV.SECTIONS_EMPTY, heal_empty_sections, + { name: "heal_empty_sections", priority: 60, cooldown: 120 }); + subscribe(EV.SUBSCRIPTION_UNREACHABLE, heal_subscriptions, + { name: "heal_subscriptions", priority: 60 }); + + subscribe(EV.OOM_DETECTED, heal_oom, + { name: "heal_oom", priority: 70 }); + subscribe(EV.OOM_RECOVERABLE, recover_oom_scale, + { name: "recover_oom_scale", priority: 70 }); + subscribe(EV.MEMORY_LOW, heal_low_memory, + { name: "heal_low_memory", priority: 70 }); + subscribe(EV.RPCD_FD_LEAK, heal_rpcd, + { name: "heal_rpcd", priority: 70 }); + + subscribe(EV.ANOMALY_RECONNECTS, heal_anomaly_reconnects, + { name: "heal_anomaly_reconnects", priority: 80, cooldown: 300 }); + + // Observers, not repairs: they must run after the healers above. + subscribe(EV.SMARTDETECT_CANDIDATE, collect_smart_detect_candidate, + { name: "collect_smart_detect_candidate", priority: 90 }); + subscribe(EV.URLTEST_SWITCHED, notify_urltest_switch, + { name: "notify_urltest_switch", priority: 90 }); + subscribe(EV.HONEYPOT_HIT, heal_honeypot_hit, + { name: "heal_honeypot_hit", priority: 90 }); + subscribe(EV.PAUSE_EXPIRED, function(ev) { check_auto_resume_pause(); }, + { name: "resume_after_pause", priority: 5 }); +} + +// One synchronous sweep of every tier, used by the `ai-heal` CLI mode where no +// worker is running: probe, let the subscribers repair, then publish status. +function ai_full_health_audit() { + register_subscribers(); + controller.probe_fast(); + controller.probe_normal(); + controller.probe_slow(); + safe_call(ai_heal_dns_loop, "ai_heal_dns_loop"); + ai_export_status(); +} + function worker() { log_message("Watchdog daemon started.", "info"); + // Subscribe before any source is wired up, so no fact observed during + // startup is published into an empty bus and lost. + register_subscribers(); + setup_honeypot_listener(); // Restore subnet cache from persistent storage at startup (in case /tmp was cleared) ai_heal_subnet_cache(); @@ -1706,27 +1375,19 @@ function worker() { let ubus_conn = setup_ubus_listener(); function perform_fast_checks() { - safe_call(check_singbox_process, "check_singbox_process"); - safe_call(ai_heal_proxy_health, "ai_heal_proxy_health"); - safe_call(ai_heal_dns_continuous, "ai_heal_dns_continuous"); + controller.probe_fast(); } function perform_normal_checks() { - safe_call(check_auto_resume_pause, "check_auto_resume_pause"); - safe_call(ai_full_health_audit, "ai_full_health_audit"); + controller.probe_normal(); safe_call(smart_detect_process_pending, "smart_detect_process_pending"); safe_call(export_metrics, "export_metrics"); + safe_call(ai_export_status, "ai_export_status"); } function perform_slow_checks() { - safe_call(ai_heal_community_subnet_sets, "ai_heal_community_subnet_sets"); - safe_call(ai_heal_tproxy_port, "ai_heal_tproxy_port"); - safe_call(ai_heal_subscriptions, "ai_heal_subscriptions"); - safe_call(ai_heal_uci_config, "ai_heal_uci_config"); - safe_call(ai_heal_empty_sections, "ai_heal_empty_sections"); + controller.probe_slow(); safe_call(ai_heal_dns_loop, "ai_heal_dns_loop"); - safe_call(analyze_anomalies, "analyze_anomalies"); - safe_call(recover_oom_scale, "recover_oom_scale"); } if (uloop) { @@ -1794,8 +1455,10 @@ function print_ai_status() { function print_ai_status_full() { let now = time(); - let proxy_ok = proxy_consecutive_fails == 0; - let dns_ok = dns_consecutive_fails == 0; + let proxy_fails = proxy_consecutive_fails(); + let dns_fails = dns_consecutive_fails(); + let proxy_ok = proxy_fails == 0; + let dns_ok = dns_fails == 0; let sb_uptime = 0; let sb_pid = get_sing_box_pid(); @@ -1833,14 +1496,14 @@ function print_ai_status_full() { uptime_s: sb_uptime, memory_mb: mem_mb, proxy_ok: proxy_ok, - proxy_latency_ms: average_latency(proxy_latency_history), - proxy_consecutive_fails: proxy_consecutive_fails, + proxy_latency_ms: average_latency(proxy_latency_history()), + proxy_consecutive_fails: proxy_fails, dns_ok: dns_ok, - dns_latency_ms: average_latency(dns_latency_history), - dns_consecutive_fails: dns_consecutive_fails, + dns_latency_ms: average_latency(dns_latency_history()), + dns_consecutive_fails: dns_fails, incidents_total: ai_incidents_count, last_incident: last_ai_incident, - healthy_streak: ai_healthy_streak, + healthy_streak: controller.healthy_streak(), adaptive_interval_s: adaptive_normal_interval(), reconnects_hour: int(trim(fs.readfile("/tmp/tachyon_reconnect_count") || "0")) }; diff --git a/tests/event_bus.sh b/tests/event_bus.sh new file mode 100644 index 00000000..4efa8d33 --- /dev/null +++ b/tests/event_bus.sh @@ -0,0 +1,240 @@ +#!/usr/bin/env bash +set -eo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +TACHYON_LIB="$ROOT_DIR/tachyon/files/usr/lib" +WORK_DIR="$(mktemp -d)" + +cleanup() { + rm -rf "$WORK_DIR" +} +trap cleanup EXIT + +fail() { + printf 'FAIL: %s\n' "$1" >&2 + exit 1 +} + +# Runs a ucode snippet with the event bus already required as `events`. +run_case() { + local body="$1" + { + printf '%s\n' 'let events = require("core.events");' + printf '%s\n' "$body" + } >"$WORK_DIR/case.uc" + ucode -L "$TACHYON_LIB" "$WORK_DIR/case.uc" +} + +assert_eq() { + local actual="$1" + local expected="$2" + local label="$3" + [ "$actual" = "$expected" ] || fail "$label: expected '$expected', got '$actual'" +} + +# --- delivery: a subscriber sees the fact it subscribed to --- +actual="$(run_case ' +let b = events.bus(); +let got = []; +b.on("dns.down", function(ev) { push(got, ev.type + ":" + ev.payload.streak); }); +b.emit("dns.down", { streak: 3 }); +print(join(",", got)); +')" +assert_eq "$actual" "dns.down:3" "handler receives type and payload" + +# --- isolation: an unrelated type is not delivered --- +actual="$(run_case ' +let b = events.bus(); +let n = 0; +b.on("dns.down", function(ev) { n++; }); +b.emit("proxy.down", {}); +print(n); +')" +assert_eq "$actual" "0" "unrelated event type is not delivered" + +# --- priority: lower priority number runs first --- +actual="$(run_case ' +let b = events.bus(); +let order = []; +b.on("x", function(ev) { push(order, "late"); }, { priority: 90 }); +b.on("x", function(ev) { push(order, "early"); }, { priority: 10 }); +b.on("x", function(ev) { push(order, "mid"); }, { priority: 50 }); +b.emit("x", {}); +print(join(",", order)); +')" +assert_eq "$actual" "early,mid,late" "handlers run in priority order" + +# --- priority ties keep registration order (stable sort) --- +actual="$(run_case ' +let b = events.bus(); +let order = []; +b.on("x", function(ev) { push(order, "first"); }, { priority: 50 }); +b.on("x", function(ev) { push(order, "second"); }, { priority: 50 }); +b.on("x", function(ev) { push(order, "third"); }, { priority: 50 }); +b.emit("x", {}); +print(join(",", order)); +')" +assert_eq "$actual" "first,second,third" "equal priorities keep registration order" + +# --- error isolation: a throwing handler must not abort the rest --- +# This is the property safe_call() used to give each individual check; losing +# it would mean one broken healer silences every healer after it. +actual="$(run_case ' +let b = events.bus(); +let survived = 0; +b.on("x", function(ev) { die("boom"); }, { priority: 10, name: "exploder" }); +b.on("x", function(ev) { survived++; }, { priority: 20 }); +b.emit("x", {}); +print(survived + "," + b.stats().failed); +')" +assert_eq "$actual" "1,1" "throwing handler is isolated and counted" + +# --- error sink receives the failing handler name --- +actual="$(run_case ' +let b = events.bus(); +let reported = ""; +b.on_error(function(name, event_type, err) { reported = name + "@" + event_type; }); +b.on("x", function(ev) { die("boom"); }, { name: "heal_dns" }); +b.emit("x", {}); +print(reported); +')" +assert_eq "$actual" "heal_dns@x" "error sink gets handler name and event type" + +# --- cooldown: second invocation inside the window is skipped --- +# Replaces the hand-rolled `now - last_*_time < N` guards. The window is +# generous (60s) so the two emits are unambiguously inside it. +actual="$(run_case ' +let b = events.bus(); +let n = 0; +b.on("x", function(ev) { n++; }, { cooldown: 60 }); +b.emit("x", {}); +b.emit("x", {}); +b.emit("x", {}); +print(n + "," + b.stats().skipped); +')" +assert_eq "$actual" "1,2" "cooldown suppresses repeat invocations" + +# --- cooldown is per subscriber, not per event type --- +actual="$(run_case ' +let b = events.bus(); +let cooled = 0; +let eager = 0; +b.on("x", function(ev) { cooled++; }, { cooldown: 60 }); +b.on("x", function(ev) { eager++; }, { cooldown: 0 }); +b.emit("x", {}); +b.emit("x", {}); +print(cooled + "," + eager); +')" +assert_eq "$actual" "1,2" "cooldown applies per subscriber" + +# --- emit_once: duplicate inside the window is suppressed at publish time --- +actual="$(run_case ' +let b = events.bus(); +let n = 0; +b.on("oom.detected", function(ev) { n++; }); +b.emit_once("oom.detected", {}, 60); +b.emit_once("oom.detected", {}, 60); +print(n + "," + b.stats().suppressed); +')" +assert_eq "$actual" "1,1" "emit_once suppresses duplicate within window" + +# --- emit_once with a zero window never suppresses --- +actual="$(run_case ' +let b = events.bus(); +let n = 0; +b.on("x", function(ev) { n++; }); +b.emit_once("x", {}, 0); +b.emit_once("x", {}, 0); +print(n); +')" +assert_eq "$actual" "2" "zero window disables deduplication" + +# --- emit_once dedup key discriminates distinct subjects --- +# One blocked domain must not mask a different blocked domain. +actual="$(run_case ' +let b = events.bus(); +let seen = []; +b.on("smartdetect.candidate", function(ev) { push(seen, ev.payload.domain); }); +b.emit_once("smartdetect.candidate", { domain: "a.example.com" }, 600, "a.example.com"); +b.emit_once("smartdetect.candidate", { domain: "a.example.com" }, 600, "a.example.com"); +b.emit_once("smartdetect.candidate", { domain: "b.example.com" }, 600, "b.example.com"); +print(join(",", seen)); +')" +assert_eq "$actual" "a.example.com,b.example.com" "dedup key separates distinct subjects" + +# --- has(): sources can skip probes nobody listens to --- +actual="$(run_case ' +let b = events.bus(); +b.on("x", function(ev) {}); +print(b.has("x") + "," + b.has("y")); +')" +assert_eq "$actual" "true,false" "has() reports subscriber presence" + +# --- reset_timers clears cooldown and dedup state --- +actual="$(run_case ' +let b = events.bus(); +let n = 0; +b.on("x", function(ev) { n++; }, { cooldown: 60 }); +b.emit("x", {}); +b.emit("x", {}); +b.reset_timers(); +b.emit("x", {}); +print(n); +')" +assert_eq "$actual" "2" "reset_timers clears cooldown state" + +# --- stats and subscriber_runs report real activity --- +actual="$(run_case ' +let b = events.bus(); +b.on("x", function(ev) {}, { name: "alpha" }); +b.on("y", function(ev) {}, { name: "beta" }); +b.emit("x", {}); +b.emit("x", {}); +b.emit("y", {}); +let s = b.stats(); +let r = b.subscriber_runs(); +print(s.emitted + "," + s.delivered + "," + s.types + "," + s.subscribers + "," + r.alpha + "," + r.beta); +')" +assert_eq "$actual" "3,3,2,2,2,1" "stats and subscriber_runs track activity" + +# --- malformed registrations are rejected, not crashed on --- +actual="$(run_case ' +let b = events.bus(); +let ok_empty_type = b.on("", function(ev) {}); +let ok_no_handler = b.on("x", null); +print(ok_empty_type + "," + ok_no_handler + "," + b.stats().subscribers); +')" +assert_eq "$actual" "false,false,0" "invalid subscriptions are rejected" + +# --- emitting an unknown type is a no-op, not an error --- +actual="$(run_case ' +let b = events.bus(); +print(b.emit("nobody.listens", {}) + "," + b.emit("", {})); +')" +assert_eq "$actual" "0,0" "emit without subscribers is a safe no-op" + +# --- non-object payload is normalised so handlers can always index it --- +actual="$(run_case ' +let b = events.bus(); +let t = ""; +b.on("x", function(ev) { t = type(ev.payload); }); +b.emit("x", null); +print(t); +')" +assert_eq "$actual" "object" "payload is always an object" + +# --- events carry both wall-clock and monotonic stamps --- +actual="$(run_case ' +let b = events.bus(); +let ok = false; +b.on("x", function(ev) { ok = (ev.ts > 0 && ev.ms >= 0); }); +b.emit("x", {}); +print(ok); +')" +assert_eq "$actual" "true" "events carry ts and monotonic ms" + +# --- module CLI selftest stays wired --- +actual="$(ucode -L "$TACHYON_LIB" "$TACHYON_LIB/core/events.uc" selftest)" +assert_eq "$actual" "ok" "module selftest passes" + +printf 'event bus checks passed\n' diff --git a/tests/event_controller.sh b/tests/event_controller.sh new file mode 100644 index 00000000..fc195371 --- /dev/null +++ b/tests/event_controller.sh @@ -0,0 +1,204 @@ +#!/usr/bin/env bash +set -eo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +TACHYON_LIB="$ROOT_DIR/tachyon/files/usr/lib" +CONTROLLER_UC="$TACHYON_LIB/service/event_controller.uc" +WATCHDOG_UC="$TACHYON_LIB/service/watchdog.uc" + +fail() { + printf 'FAIL: %s\n' "$1" >&2 + exit 1 +} + +uc() { + command ucode -L "$TACHYON_LIB" "$@" +} + +# Prints the fact types a line classifies into, one per line. +classify() { + uc "$CONTROLLER_UC" classify "$1" 2>/dev/null || true +} + +# Prints the classified types joined by '+' so a multi-fact line is one token. +classify_joined() { + classify "$1" | paste -sd '+' - +} + +assert_eq() { + local actual="$1" expected="$2" label="$3" + [ "$actual" = "$expected" ] || fail "$label: expected '$expected', got '$actual'" +} + +# --- OOM detection: the real thing must be recognised --- +for line in \ + 'kernel: sing-box invoked oom-killer: gfp_mask=0x100cca' \ + 'kernel: Out of memory: Kill process 1234 (sing-box) score 900' \ + 'daemon.err sing-box[123]: fatal error: out of memory' +do + assert_eq "$(classify_joined "$line")" "oom.detected" "OOM line: $line" +done + +# --- OOM discrimination: netlink/nlbwmon noise must NOT read as OOM --- +# This is the guard at watchdog.uc:1509. Misreading it would shrink +# GOMEMLIMIT for nothing, degrading throughput on every router that logs it. +assert_eq "$(classify_joined 'nlbwmon: netlink: out of memory allocating buffer')" \ + "" "netlink out-of-memory warning is not an OOM event" +assert_eq "$(classify_joined 'kernel: netlink: 12 bytes leftover after parsing, out of memory')" \ + "" "kernel netlink warning is not an OOM event" + +# --- OOM is terminal: no further facts from the same line --- +# The original returned immediately after handling OOM. +assert_eq "$(classify_joined 'kernel: sing-box invoked oom-killer, URLTest switch proxy')" \ + "oom.detected" "OOM short-circuits URLTest classification" + +# --- keyword pre-filter: unrelated lines cost nothing and yield nothing --- +assert_eq "$(classify_joined 'daemon.info dnsmasq[1]: reading /tmp/resolv.conf')" \ + "" "unrelated line yields no fact" +assert_eq "$(classify_joined '')" "" "empty line yields no fact" + +# --- smart detect: a direct-connection failure carries the domain --- +assert_eq "$(classify_joined 'outbound/direct: failed to connect to "blocked.example.com:443"')" \ + "smartdetect.candidate" "direct failure classifies as smart-detect candidate" +assert_eq "$(uc "$CONTROLLER_UC" classify-domain \ + 'outbound/direct: failed to connect to "blocked.example.com:443"' 2>/dev/null)" \ + "blocked.example.com" "smart-detect fact carries the extracted domain" + +# A direct line with no usable host is not a candidate. +assert_eq "$(classify_joined 'outbound/direct: failed to connect to 192.168.1.1:443')" \ + "" "direct failure without a domain yields no candidate" + +# A direct line that did not fail is not a candidate. +assert_eq "$(classify_joined 'outbound/direct: connected to "example.com:443"')" \ + "" "successful direct connection yields no candidate" + +# --- URLTest switches --- +assert_eq "$(classify_joined 'URLTest: selected proxy outbound-2')" \ + "urltest.switched" "URLTest line classifies as a switch" +assert_eq "$(classify_joined 'group: switch proxy to node-tokyo')" \ + "urltest.switched" "switch proxy line classifies as a switch" + +# --- smart detect and URLTest are not exclusive --- +# handle_log_line() fell through from one branch to the other, so a line that +# satisfies both must still produce both facts. +assert_eq "$(classify_joined 'URLTest: direct failed "blocked.example.com:443", switch proxy')" \ + "smartdetect.candidate+urltest.switched" "one line can yield both facts, in order" + +# --- extractor parity: the controller must match watchdog byte for byte --- +# tests/smart_detect_domain_extraction.sh pins the watchdog CLI; this pins the +# controller against it, so the two cannot drift apart silently. +for line in \ + 'outbound/direct: failed to connect to "example.com:443"' \ + 'direct connection failed: "sub.domain.example.org"' \ + 'DIRECT timeout "xn--80ak6aa92e.com:443"' \ + 'direct reset "a-b-c.example.co.uk:8443"' \ + 'direct failed target=blocked.example.net' \ + 'direct failed target blocked2.example.net' \ + 'direct failed to connect to 192.168.1.1:443' \ + 'direct connection failed, no host in line' \ + 'direct failed "localhost"' \ + 'direct failed "a.co"' \ + 'direct failed "*.example.com"' \ + 'direct failed "example..com"' \ + 'direct failed "-example.com"' +do + from_watchdog="$(uc "$WATCHDOG_UC" smart-detect-extract-domain "$line" 2>/dev/null || true)" + from_controller="$(uc "$CONTROLLER_UC" extract-domain "$line" 2>/dev/null || true)" + [ "$from_watchdog" = "$from_controller" ] || \ + fail "extractor drift on '$line': watchdog='$from_watchdog' controller='$from_controller'" +done + +# --- event vocabulary is exported and non-empty --- +types="$(uc "$CONTROLLER_UC" event-types)" +for expected in \ + "SINGBOX_STOPPED=singbox.stopped" \ + "DNS_DOWN=dns.down" \ + "PROXY_DOWN=proxy.down" \ + "OOM_DETECTED=oom.detected" \ + "SMARTDETECT_CANDIDATE=smartdetect.candidate" +do + printf '%s\n' "$types" | grep -qx "$expected" || fail "event vocabulary missing $expected" +done + +# --- publication layer: the OOM replay guard --- +# logread -f replays the historical buffer on start. Facts derived from that +# replay describe the past, not the present, and must not be published. +run_case() { + local body="$1" + local tmp + tmp="$(mktemp "${TMPDIR:-/tmp}/tachyon_ec_case.XXXXXX.uc")" + { + printf '%s\n' 'let events = require("core.events");' + printf '%s\n' 'let ec = require("service.event_controller");' + printf '%s\n' "$body" + } >"$tmp" + uc "$tmp" + rm -f "$tmp" +} + +actual="$(run_case ' +let b = events.bus(); +let c = ec.controller(b, {}); +let seen = 0; +b.on("oom.detected", function(ev) { seen++; }); +// syslog_start_time still 0: we are inside the historical replay window. +print(length(c.handle_log_line("kernel: sing-box invoked oom-killer")) + "," + seen); +')" +assert_eq "$actual" "0,0" "OOM from the log replay window is not published" + +actual="$(run_case ' +let b = events.bus(); +let c = ec.controller(b, {}); +let seen = 0; +b.on("oom.detected", function(ev) { seen++; }); +// Started well in the past: live lines, not replay. +c.set_syslog_start(time() - 3600); +c.handle_log_line("kernel: sing-box invoked oom-killer"); +c.handle_log_line("kernel: sing-box invoked oom-killer"); +print(seen); +')" +assert_eq "$actual" "1" "an OOM storm collapses into one published fact" + +# --- publication layer: URLTest throttle --- +actual="$(run_case ' +let b = events.bus(); +let c = ec.controller(b, {}); +let seen = 0; +b.on("urltest.switched", function(ev) { seen++; }); +c.handle_log_line("URLTest: selected proxy a"); +c.handle_log_line("URLTest: selected proxy b"); +c.handle_log_line("URLTest: selected proxy c"); +print(seen); +')" +assert_eq "$actual" "1" "URLTest facts are throttled to one per window" + +# --- publication layer: ubus service stop is scoped to sing-box --- +actual="$(run_case ' +let b = events.bus(); +let c = ec.controller(b, {}); +let seen = []; +b.on("singbox.stopped", function(ev) { push(seen, ev.payload.reason); }); +let ours = c.handle_ubus_service_stop("sing-box", "procd"); +let theirs = c.handle_ubus_service_stop("dnsmasq", "procd"); +print(ours + "," + theirs + "," + join(",", seen)); +')" +assert_eq "$actual" "true,false,procd" "only sing-box stops publish singbox.stopped" + +# --- publication layer: honeypot input is validated before publishing --- +# The FIFO is world-writable by design; a malformed line must not become a fact +# that a subscriber later interpolates into an nft or iptables command. +actual="$(run_case ' +let b = events.bus(); +let c = ec.controller(b, {}); +let seen = []; +b.on("honeypot.hit", function(ev) { push(seen, ev.payload.ip); }); +let ok4 = c.handle_honeypot_line("203.0.113.7\n"); +let ok6 = c.handle_honeypot_line("2001:db8::1"); +let bad = c.handle_honeypot_line("203.0.113.7; nft flush ruleset"); +let empty = c.handle_honeypot_line(""); +print(ok4 + "," + ok6 + "," + bad + "," + empty + "|" + join(",", seen)); +')" +assert_eq "$actual" "true,true,false,false|203.0.113.7,2001:db8::1" \ + "honeypot publishes only well-formed addresses" + +printf 'event_controller checks passed\n' diff --git a/tests/forward_references.sh b/tests/forward_references.sh new file mode 100644 index 00000000..e5e7c4e6 --- /dev/null +++ b/tests/forward_references.sh @@ -0,0 +1,104 @@ +#!/usr/bin/env bash +set -eo pipefail + +# ucode captures a closure's upvalues when the closure is CREATED. A local +# declared later in the same scope was never captured, so the name resolves as a +# global and evaluates to null — "left-hand side is not a function" at call time. +# +# Neither `ucode -c` nor `ucode -S -c` catches this, and a unit test only catches +# it if it actually calls the offending path. Both halves are needed: +# 1. a static scan for declaration-after-use across every module +# 2. a runtime invocation of the push handlers, which are pure closures over +# probes declared elsewhere in the same scope + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +TACHYON_LIB="$ROOT_DIR/tachyon/files/usr/lib" + +fail() { + printf 'FAIL: %s\n' "$1" >&2 + exit 1 +} + +# ── 1. static scan ────────────────────────────────────────────────────────── +# Comments and string literals are stripped first so that prose mentioning a +# function name is not mistaken for a call. +scan_forward_refs() { + local file="$1" stripped + stripped="$(mktemp)" + sed -E 's://.*::; s:"[^"]*":"":g' "$file" > "$stripped" + + grep -nE '^[[:space:]]+function [a-z_0-9]+\(' "$stripped" \ + | sed -E 's/^([0-9]+):[[:space:]]*function ([a-z_0-9]+).*/\1 \2/' \ + | while read -r decl name; do + local first + first="$(grep -nE "(^|[^a-zA-Z0-9_.])$name([^a-zA-Z0-9_]|$)" "$stripped" \ + | grep -vE "function $name" | head -1 | cut -d: -f1)" + if [ -n "$first" ] && [ "$first" -lt "$decl" ]; then + printf '%s used at line %s but declared at line %s\n' "$name" "$first" "$decl" + fi + done + + rm -f "$stripped" +} + +found=0 +while IFS= read -r uc; do + hits="$(scan_forward_refs "$uc")" + if [ -n "$hits" ]; then + printf 'FAIL: forward reference in %s:\n%s\n' "${uc#"$ROOT_DIR/"}" "$hits" >&2 + found=1 + fi +done < <(find "$TACHYON_LIB" -name '*.uc' | sort) +[ "$found" -eq 0 ] || fail "nested functions are used before their declaration (ucode does not hoist them)" + +# The scan is only worth trusting if it fires on a known-bad file. +canary="$(mktemp "${TMPDIR:-/tmp}/tachyon_fwd_canary.XXXXXX.uc")" +cat > "$canary" <<'EOF' +function outer() { + let self = {}; + self.go = function() { return later(); }; + function later() { return 1; } + return self; +} +EOF +[ -n "$(scan_forward_refs "$canary")" ] \ + || { rm -f "$canary"; fail "the forward-reference scan does not detect a known-bad file"; } +rm -f "$canary" + +# ── 2. runtime invocation ─────────────────────────────────────────────────── +# Calling the handlers is what actually proves the bindings resolve. The probes +# they drive touch nft and /proc, which do not exist in the test container, so +# run_probe's isolation absorbs the failures — a null binding would still raise +# "left-hand side is not a function" before any probe body runs. +driver="$(mktemp "${TMPDIR:-/tmp}/tachyon_fwd.XXXXXX.uc")" +cat > "$driver" <<'EOF' +let events = require("core.events"); +let ec = require("service.event_controller"); + +let b = events.bus(); +let seen = []; +for (let t in ["singbox.stopped", "firewall.reloaded", "honeypot.hit"]) + b.on(t, function(ev) { push(seen, ev.type); }, { name: "collect" }); + +let c = ec.controller(b, {}); +c.handle_ubus_service_stop("sing-box", "test"); +c.handle_ubus_firewall_reload(); +c.handle_honeypot_line("10.0.0.1"); +print(join(",", seen) + "\n"); +EOF + +out="$(ucode -L "$TACHYON_LIB" "$driver" 2>&1)" || { + rm -f "$driver" + fail "invoking the push handlers raised an error: +$out" +} +rm -f "$driver" + +printf '%s\n' "$out" | grep -q 'singbox.stopped' \ + || fail "handle_ubus_service_stop published nothing (got: $out)" +printf '%s\n' "$out" | grep -q 'firewall.reloaded' \ + || fail "handle_ubus_firewall_reload published nothing (got: $out)" +printf '%s\n' "$out" | grep -q 'honeypot.hit' \ + || fail "handle_honeypot_line published nothing (got: $out)" + +printf 'forward reference checks passed\n' diff --git a/tests/watchdog_subscriptions.sh b/tests/watchdog_subscriptions.sh new file mode 100644 index 00000000..8ce5c8fb --- /dev/null +++ b/tests/watchdog_subscriptions.sh @@ -0,0 +1,144 @@ +#!/usr/bin/env bash +set -eo pipefail + +# Pins the wiring between the two halves of the split: event_controller.uc +# observes and emits, watchdog.uc subscribes and repairs. A fact that nobody +# subscribes to is a repair that silently stopped happening — exactly the +# regression this refactor could introduce and that no other test would catch. + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +TACHYON_LIB="$ROOT_DIR/tachyon/files/usr/lib" +CONTROLLER_UC="$TACHYON_LIB/service/event_controller.uc" +WATCHDOG_UC="$TACHYON_LIB/service/watchdog.uc" + +fail() { + printf 'FAIL: %s\n' "$1" >&2 + exit 1 +} + +# Facts that are deliberately published without a repair attached. +# PROXY_UP / DNS_UP are the recovery side of a fault pair, kept so a future +# subscriber can act on recovery; FIREWALL_RELOADED and TICK are informational. +is_informational() { + case "$1" in + PROXY_UP|DNS_UP|FIREWALL_RELOADED|TICK) return 0 ;; + *) return 1 ;; + esac +} + +emitted="$(grep -oE 'bus\.emit(_once)?\(EV\.[A-Z_]+' "$CONTROLLER_UC" \ + | grep -oE 'EV\.[A-Z_]+' | sed 's/^EV\.//' | sort -u)" +[ -n "$emitted" ] || fail "no emit sites found in event_controller.uc" + +# Registrations go through subscribe(), a wrapper that turns a bus.on() rejection +# into a startup failure instead of a silently missing repair. +subscribed="$(grep -oE 'subscribe\(EV\.[A-Z_]+' "$WATCHDOG_UC" \ + | grep -oE 'EV\.[A-Z_]+' | sed 's/^EV\.//' | sort -u)" +[ -n "$subscribed" ] || fail "no subscribe() registrations found in watchdog.uc" + +# --- every emitted fault has a subscriber --- +for ev in $emitted; do + is_informational "$ev" && continue + printf '%s\n' "$subscribed" | grep -qx "$ev" \ + || fail "event_controller emits EV.$ev but watchdog subscribes to nothing for it" +done + +# --- every subscription has a matching emit site --- +# A dead subscription means a repair that can never run. +for ev in $subscribed; do + printf '%s\n' "$emitted" | grep -qx "$ev" \ + || fail "watchdog subscribes to EV.$ev but event_controller never emits it" +done + +# --- every referenced EV name exists in the vocabulary --- +vocabulary="$(ucode -L "$TACHYON_LIB" "$CONTROLLER_UC" event-types \ + | sed 's/=.*//' | sort -u)" +for ev in $emitted $subscribed; do + printf '%s\n' "$vocabulary" | grep -qx "$ev" \ + || fail "EV.$ev is used but missing from the exported event vocabulary" +done + +# --- the faults that used to be repaired on a timer are still covered --- +# Each name here was an ai_heal_* function called from a worker() tier before the +# refactor. If one loses its subscription, the router stops self-healing it. +for required in \ + SINGBOX_STOPPED WAN_DOWN CONFIG_CORRUPT NFT_MISSING QOS_MISSING TPROXY_DOWN \ + SUBNETS_EMPTY PROXY_DOWN DNS_DOWN SECTIONS_EMPTY SUBSCRIPTION_UNREACHABLE \ + OOM_DETECTED OOM_RECOVERABLE MEMORY_LOW RPCD_FD_LEAK ANOMALY_RECONNECTS \ + SMARTDETECT_CANDIDATE URLTEST_SWITCHED HONEYPOT_HIT +do + printf '%s\n' "$subscribed" | grep -qx "$required" \ + || fail "no subscriber for EV.$required (was repaired on a timer before the refactor)" +done + +# --- subscriptions are registered exactly once --- +# register_subscribers() is called from both worker() and the ai-heal CLI mode; +# without the idempotence guard every repair would run twice per fact. +grep -q 'if (subscribers_registered) return;' "$WATCHDOG_UC" \ + || fail "register_subscribers() lost its idempotence guard" + +# --- a rejected registration is fatal, not silent --- +# bus.on() returns false for a non-function handler rather than raising, and +# ucode does not hoist function declarations: moving a heal_* below +# register_subscribers() would register nothing, emit nothing and throw nothing. +# subscribe() exists to make that loud. Registering directly through bus.on() +# would reopen the hole. +grep -q 'function subscribe(event_type, handler, opts)' "$WATCHDOG_UC" \ + || fail "the subscribe() wrapper is gone; a lost handler would fail silently" +grep -q 'if (!bus.on(event_type, handler, opts))' "$WATCHDOG_UC" \ + || fail "subscribe() no longer checks the bus.on() return value" +if grep -qE '^\s+bus\.on\(EV\.' "$WATCHDOG_UC"; then + fail "a subscriber registers via bus.on() directly, bypassing the subscribe() guard" +fi + +# --- cooldowns that replaced the old last_*_time globals are still declared --- +# These four were the only per-repair timers in the old code; the bus owns them +# now, so the values have to survive here. +check_cooldown() { + local handler="$1" seconds="$2" + grep -A1 -E "subscribe\(EV\.[A-Z_]+, $handler," "$WATCHDOG_UC" \ + | grep -q "cooldown: $seconds" \ + || fail "$handler lost its cooldown of ${seconds}s" +} +check_cooldown heal_wan_and_gateway 300 +check_cooldown heal_community_subnet_sets 300 +check_cooldown heal_empty_sections 120 +check_cooldown heal_anomaly_reconnects 300 + +# --- the two-tier repairs keep their original pacing --- +# Before the split, proxy and DNS were measured twice at two different rates: +# ai_heal_proxy_health / ai_heal_dns_continuous on the fast tier (15s), and +# ai_heal_proxy_connectivity / ai_heal_dns inside the normal-tier audit (120s+). +# One probe now serves both, at the fast rate. Without these bounds a single +# stall would restart sing-box eight times as often as it used to — the +# aggressive-restart regression fixed in c8052ee6. +check_cooldown heal_proxy_connectivity 120 +check_cooldown heal_dns_stall 120 + +# The DNS stall threshold counts a streak of 3. That streak has to be stepped at +# the old normal-tier rate, or "3 failures" would mean 45 seconds of trouble +# instead of ~6 minutes. +grep -q 'const DNS_STREAK_INTERVAL = 120;' "$CONTROLLER_UC" \ + || fail "DNS_STREAK_INTERVAL is missing or no longer 120s" +grep -q 'now - state.dns_streak_stamp >= DNS_STREAK_INTERVAL' "$CONTROLLER_UC" \ + || fail "the DNS stall streak is no longer paced by DNS_STREAK_INTERVAL" + +# The fast-tier counters must NOT be paced — they are the ones the old fast +# healers incremented every 15s, and ai-status-full reports them. +grep -q 'state.dns_consecutive_fails++;' "$CONTROLLER_UC" \ + || fail "dns_consecutive_fails is no longer incremented every probe" +grep -q 'state.proxy_consecutive_fails++;' "$CONTROLLER_UC" \ + || fail "proxy_consecutive_fails is no longer incremented every probe" + +# --- the observation layer does not repair --- +# The whole point of the split: no restart, reload or nft mutation may leak back +# into event_controller.uc. +for forbidden in 'safe_proxy_restart' 'safe_reload_firewall' 'ai_heal_report' \ + '/etc/init.d/tachyon' 'service sing-box' 'nft add' 'nft flush' +do + if grep -q -- "$forbidden" "$CONTROLLER_UC"; then + fail "event_controller.uc performs a repair action ($forbidden); observation must not heal" + fi +done + +printf 'watchdog subscription checks passed\n'