From e38d7d07a22225784735deb7f6795cf60a39ef8e Mon Sep 17 00:00:00 2001 From: Fisto Date: Tue, 22 Sep 2026 19:31:19 +0200 Subject: [PATCH] Add UDP discovery and heartbeat-miss notices to remchild --- remchild/README.md | 11 ++- remchild/client/main.c | 170 +++++++++++++++++++++++++++++++++---- remchild/plan.md | 26 +++++- remchild/server/server.mjs | 93 ++++++++++++++++++++ 4 files changed, 278 insertions(+), 22 deletions(-) diff --git a/remchild/README.md b/remchild/README.md index 5d133dd..b3c0c59 100644 --- a/remchild/README.md +++ b/remchild/README.md @@ -26,10 +26,15 @@ node remchild/server/server.mjs --port 8799 --ips 192.168.1.5 ./remchild/client/remchild --server 192.168.1.5:8799 ``` -The client asks for the address when `--server` is absent. It prints the assigned name, then -accepts `p` (ping test) and `q` (quit). +With no `--server`, the client probes for a server for five seconds: one answer connects +automatically, several show a pick list, none asks for `host:port`. `--server` skips discovery. It +prints the assigned name, then accepts `p` (ping test) and `q` (quit). -Non-LAN addresses in the allowlist need the launch argument: +If the server misses a heartbeat during the fast phases (the first few minutes), it sends a +`notice`; the client prints a warning that the connection may be unstable. Nothing else changes. + +Discovery uses UDP 8798 (`--discovery-port`); `--beacon-ms 0` keeps probe replies but stops the +beacon. Non-LAN addresses in the allowlist need the launch argument: ```sh node remchild/server/server.mjs --port 8799 --ips 203.0.113.7 --allow-non-lan diff --git a/remchild/client/main.c b/remchild/client/main.c index 62ef3df..d7568a7 100644 --- a/remchild/client/main.c +++ b/remchild/client/main.c @@ -31,6 +31,10 @@ #include "words.h" +#define DISCOVERY_PORT 8798 +#define DISCOVERY_WINDOW_MS 5000 +#define DISCOVERY_MAX 8 + #define HEARTBEAT_FAST_MS 10000 #define HEARTBEAT_MID_MS 30000 #define HEARTBEAT_SLOW_MS 300000 @@ -38,6 +42,12 @@ #define MID_UNTIL_MS 360000 #define LINE_MAX 4096 +static long heartbeat_interval(long elapsed_ms) { + if (elapsed_ms < FAST_UNTIL_MS) return HEARTBEAT_FAST_MS; + if (elapsed_ms < MID_UNTIL_MS) return HEARTBEAT_MID_MS; + return HEARTBEAT_SLOW_MS; +} + static long now_ms(void) { struct timespec ts; clock_gettime(CLOCK_MONOTONIC, &ts); @@ -104,6 +114,100 @@ static void name_for_id(const unsigned char *id, size_t len, char *out, size_t c snprintf(out, cap, "%s %s", REMCHILD_FIRST[first], REMCHILD_SECOND[second]); } +struct Discovered { + char ip[64]; + char name[64]; + char id[64]; + int port; +}; + +/* + * Ask the LAN who is there. Probes go out as broadcast and to loopback; a + * server answers with its beacon, and the address comes from the reply. + * Waits up to DISCOVERY_WINDOW_MS. Returns the number of servers found. + */ +static int discover(struct Discovered *found, int cap) { + int fd = socket(AF_INET, SOCK_DGRAM, 0); + if (fd < 0) return 0; + + int one = 1; + setsockopt(fd, SOL_SOCKET, SO_BROADCAST, &one, sizeof(one)); + + struct sockaddr_in local = {0}; + local.sin_family = AF_INET; + local.sin_addr.s_addr = htonl(INADDR_ANY); + if (bind(fd, (struct sockaddr *)&local, sizeof(local)) != 0) { + close(fd); + return 0; + } + + struct sockaddr_in broadcast = {0}; + broadcast.sin_family = AF_INET; + broadcast.sin_port = htons(DISCOVERY_PORT); + broadcast.sin_addr.s_addr = htonl(INADDR_BROADCAST); + + struct sockaddr_in loopback = broadcast; + loopback.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + + const char *probe = "{\"type\":\"veskslop-remchild-probe\",\"version\":\"0.1\"}"; + long deadline = now_ms() + DISCOVERY_WINDOW_MS; + long next_probe = 0; + int count = 0; + + while (now_ms() < deadline) { + if (now_ms() >= next_probe) { + sendto(fd, probe, strlen(probe), 0, (struct sockaddr *)&broadcast, sizeof(broadcast)); + sendto(fd, probe, strlen(probe), 0, (struct sockaddr *)&loopback, sizeof(loopback)); + next_probe = now_ms() + 1000; + } + + struct pollfd pfd = {fd, POLLIN, 0}; + if (poll(&pfd, 1, 200) <= 0) continue; + if (!(pfd.revents & POLLIN)) continue; + + char buffer[1024]; + struct sockaddr_in from; + socklen_t from_len = sizeof(from); + ssize_t got = recvfrom(fd, buffer, sizeof(buffer) - 1, 0, (struct sockaddr *)&from, &from_len); + if (got <= 0) continue; + buffer[got] = 0; + + if (!strstr(buffer, "veskslop-remchild") || strstr(buffer, "probe")) continue; + + char ip[64]; + if (!inet_ntop(AF_INET, &from.sin_addr, ip, sizeof(ip))) continue; + + int port = 0; + char name[64] = "veskslop"; + char id[64] = ""; + char *match = strstr(buffer, "\"port\":"); + if (match) port = atoi(match + 7); + match = strstr(buffer, "\"name\":\""); + if (match) sscanf(match + 8, "%63[^\"]", name); + match = strstr(buffer, "\"id\":\""); + if (match) sscanf(match + 6, "%63[^\"]", id); + if (port <= 0) continue; + + // One server can answer from both loopback and its LAN address; the + // beacon id identifies it. + int duplicate = 0; + for (int i = 0; i < count; i++) { + if (id[0] && found[i].id[0] && !strcmp(found[i].id, id)) duplicate = 1; + else if (!id[0] && !found[i].id[0] && !strcmp(found[i].name, name) && found[i].port == port) duplicate = 1; + } + if (duplicate || count >= cap) continue; + + snprintf(found[count].ip, sizeof(found[count].ip), "%s", ip); + snprintf(found[count].name, sizeof(found[count].name), "%s", name); + snprintf(found[count].id, sizeof(found[count].id), "%s", id); + found[count].port = port; + count++; + } + + close(fd); + return count; +} + static int connect_to(const char *host, int port) { struct sockaddr_in addr = {0}; addr.sin_family = AF_INET; @@ -159,17 +263,43 @@ int main(int argc, char **argv) { int interactive = isatty(STDIN_FILENO); if (!host[0]) { - printf("Server address (host:port): "); + struct Discovered found[DISCOVERY_MAX]; + printf("[remchild] looking for a server on the LAN for %d seconds...\n", DISCOVERY_WINDOW_MS / 1000); fflush(stdout); - char line[128]; - if (!fgets(line, sizeof(line), stdin)) return 1; - line[strcspn(line, "\n")] = 0; - char *colon = strrchr(line, ':'); - if (colon) { - *colon = 0; - port = atoi(colon + 1); + int count = discover(found, DISCOVERY_MAX); + + if (count == 1) { + printf("[remchild] found %s at %s:%d\n", found[0].name, found[0].ip, found[0].port); + snprintf(host, sizeof(host), "%s", found[0].ip); + port = found[0].port; + } else if (count > 1) { + printf("[remchild] found %d servers:\n", count); + for (int i = 0; i < count; i++) { + printf(" %d) %s at %s:%d\n", i + 1, found[i].name, found[i].ip, found[i].port); + } + printf("pick one [1]: "); + fflush(stdout); + char pick_line[32]; + int pick = 1; + if (fgets(pick_line, sizeof(pick_line), stdin)) pick = atoi(pick_line); + if (pick < 1 || pick > count) pick = 1; + printf("[remchild] using %s at %s:%d\n", found[pick - 1].name, found[pick - 1].ip, found[pick - 1].port); + snprintf(host, sizeof(host), "%s", found[pick - 1].ip); + port = found[pick - 1].port; + } else { + printf("[remchild] no server answered discovery.\n"); + printf("Server address (host:port): "); + fflush(stdout); + char line[128]; + if (!fgets(line, sizeof(line), stdin)) return 1; + line[strcspn(line, "\n")] = 0; + char *colon = strrchr(line, ':'); + if (colon) { + *colon = 0; + port = atoi(colon + 1); + } + snprintf(host, sizeof(host), "%s", line); } - snprintf(host, sizeof(host), "%s", line); } unsigned char id[32]; @@ -196,6 +326,7 @@ int main(int argc, char **argv) { printf("[remchild] identity %s (name: %s)\n", id_len == 6 ? "MAC" : "machine-id", name); printf("[remchild] connecting to %s:%d\n", host, port); + fflush(stdout); int fd = connect_to(host, port); if (fd < 0) { @@ -214,8 +345,9 @@ int main(int argc, char **argv) { char hello[512]; snprintf(hello, sizeof(hello), - "{\"type\":\"hello\",\"id\":\"%s\",\"name\":\"%s\",\"platform\":\"linux\",\"version\":\"0.1\"}", - id_hex, name); + "{\"type\":\"hello\",\"id\":\"%s\",\"name\":\"%s\",\"platform\":\"linux\"," + "\"version\":\"0.1\",\"heartbeatMs\":%ld}", + id_hex, name, heartbeat_interval(0)); send_line(fd, hello); long started = now_ms(); @@ -234,13 +366,12 @@ int main(int argc, char **argv) { for (;;) { long now = now_ms(); - long interval = now - started < FAST_UNTIL_MS ? HEARTBEAT_FAST_MS - : now - started < MID_UNTIL_MS ? HEARTBEAT_MID_MS - : HEARTBEAT_SLOW_MS; + long interval = heartbeat_interval(now - started); if (now - last_beat >= interval) { char beat[128]; - snprintf(beat, sizeof(beat), "{\"type\":\"heartbeat\",\"ts\":%ld}", (long)time(NULL)); + snprintf(beat, sizeof(beat), "{\"type\":\"heartbeat\",\"ts\":%ld,\"heartbeatMs\":%ld}", + (long)time(NULL), interval); send_line(fd, beat); last_beat = now; } @@ -272,9 +403,16 @@ int main(int argc, char **argv) { } else if (strstr(line, "\"type\":\"pong\"")) { if (ping_sent_at) printf("[remchild] pong seq=%d rtt=%ldms\n", ping_seq, now_ms() - ping_sent_at); answered_ping = 1; + } else if (strstr(line, "\"type\":\"notice\"")) { + int missed = 0; + char *value = strstr(line, "\"missed\":"); + if (value) missed = atoi(value + 9); + printf("[remchild] notice: the server missed %d heartbeat(s); the connection may be unstable.\n", + missed); } else if (strstr(line, "\"type\":\"ping\"")) { int seq = 0; - sscanf(line, "%*[^\"]\"seq\":%d", &seq); + char *value = strstr(line, "\"seq\":"); + if (value) seq = atoi(value + 6); char pong[64]; snprintf(pong, sizeof(pong), "{\"type\":\"pong\",\"seq\":%d}", seq); send_line(fd, pong); diff --git a/remchild/plan.md b/remchild/plan.md index b51f0e7..94ead9c 100644 --- a/remchild/plan.md +++ b/remchild/plan.md @@ -51,6 +51,11 @@ The client sends `heartbeat` on this schedule after boot: The server keeps three states from `lastSeen`: +If the server misses a heartbeat during the fast phases (10 s or 30 s; roughly the first six +minutes), it sends the client a `notice` with the miss count. The client prints it as a warning. +Nothing else changes: it is information for the user, not a reconnect trigger. The steady 5 minute +phase relies on the stale and gone states instead. + | State | Age | Effect | | --- | --- | --- | | `active` | under 12 min | Jobs and commands may be sent | @@ -59,10 +64,22 @@ The server keeps three states from `lastSeen`: First and last seen are kept in memory and written to `clients.jsonl`. `SIGINT` prints the table. +## Discovery + +The server answers probes and beacons on UDP 8798 (`--discovery-port`, `--beacon-ms`, 0 disables +the beacon). Probes go to the subnet broadcast and to loopback; the server replies unicast, so the +client reads the address from the reply. Beacons carry `{type, name, port, id}` and go to every +interface's broadcast address, for passive listeners. + +`remchild` with no `--server` spends five seconds probing. One server connects automatically, +several show a pick list, none falls back to the typed `host:port` prompt. `--server` always wins +and skips discovery. Discovery only crosses one subnet: WiFi client isolation, VLANs, and VPNs +block it, and the Web UI cannot use it at all. + ## Transport -Newline-delimited JSON over TCP on a dedicated port (PoC: 8799). No Discord connection, no HTTP -framework. +Newline-delimited JSON over TCP on a dedicated port (PoC: 8799), plus the UDP discovery above. No +Discord connection, no HTTP framework. | Direction | Message | Meaning | | --- | --- | --- | @@ -70,6 +87,9 @@ framework. | server → client | `welcome` | Assigned name, heartbeat interval, server name | | client → server | `heartbeat` | `ts` | | both | `ping` / `pong` | `seq` echo, for the test button and latency | +| client → server | `probe` | UDP discovery ask (`veskslop-remchild-probe`) | +| server → client | `beacon` | UDP discovery answer and broadcast (`veskslop-remchild`) | +| server → client | `notice` | Information, e.g. `heartbeat-missed` with a count | | server → client | `job` | Later: a unit of work with an ID and a payload | | client → server | `job-result` | Later: `id`, `ok`, `data` or `error` | @@ -91,7 +111,7 @@ The PoC server moves into the daemon as `poc/src/remchild-server.ts`: | Phase | Work | | --- | --- | -| M1 (done) | PoC: naming, listen, allowlist, heartbeat, ping test, history | +| M1 (done) | PoC: naming, listen, allowlist, heartbeat, ping test, history, UDP discovery, heartbeat-miss notices | | M2 | Move the server into the daemon, add settings, the launch-arg unlock, and tests. Add `job` dispatch and `job-result` with retry and dedupe. | | M3 | Audio: the daemon sends call audio to the client, the client runs VAD and STT (SAM/VOX or an AI model), and returns annotated text and voice commands. The same path carries TTS back. | | M4 | Video: stream a window or play a file with MPV, and return frames or a stream URL. | diff --git a/remchild/server/server.mjs b/remchild/server/server.mjs index 9f68342..a2a3bfa 100644 --- a/remchild/server/server.mjs +++ b/remchild/server/server.mjs @@ -9,10 +9,15 @@ * Run: node remchild/server/server.mjs --port 8799 --ips 127.0.0.1,192.168.1.5 * node remchild/server/server.mjs --allow-non-lan --ips 203.0.113.7 * node remchild/server/server.mjs --allow-all # testing: any address + * + * Discovery: a UDP beacon on 8798 answers probes and advertises the server on + * every interface, so remchild can find it without a typed address. */ +import dgram from "node:dgram"; import { appendFileSync, existsSync, mkdirSync, readFileSync } from "node:fs"; import net from "node:net"; +import { hostname, networkInterfaces } from "node:os"; import { dirname, join } from "node:path"; import { fileURLToPath } from "node:url"; @@ -30,6 +35,8 @@ function arg(name, fallback = undefined) { const port = Number(arg("port", process.env.REMCHILD_PORT ?? 8799)); const allowNonLan = arg("allow-non-lan", false) !== false; const allowAll = arg("allow-all", false) !== false; +const discoveryPort = Number(arg("discovery-port", process.env.REMCHILD_DISCOVERY_PORT ?? 8798)); +const beaconMs = Number(arg("beacon-ms", 3000)); const historyFile = arg("history", join(HERE, "clients.jsonl")); const configFile = arg("config", join(HERE, "config.json")); @@ -62,6 +69,8 @@ function allowed(address) { return configured.includes(address); } +const HEARTBEAT_GRACE_MS = 5000; + const clients = new Map(); mkdirSync(dirname(historyFile), { recursive: true }); @@ -92,6 +101,32 @@ function touch(client, reason) { } } +function heartbeatInterval(client) { + return Math.min(Math.max(Number(client.heartbeatMs) || 300000, 5000), 300000); +} + +/** + * Heartbeat misses are informational. Only the fast phases (10 s and 30 s + * heartbeats) are watched; the 5 minute steady state relies on the stale and + * gone states instead. + */ +setInterval(() => { + const now = Date.now(); + + for (const client of clients.values()) { + if (!client.socket) continue; + if (heartbeatInterval(client) > 30000) continue; + if (now <= client.nextDue) continue; + + client.missed++; + client.nextDue = now + heartbeatInterval(client) + HEARTBEAT_GRACE_MS; + console.log(`[remchild] ${client.name} missed heartbeat ${client.missed}`); + client.socket.write( + JSON.stringify({ type: "notice", code: "heartbeat-missed", missed: client.missed }) + "\n" + ); + } +}, 2000).unref(); + setInterval(() => { const now = Date.now(); for (const client of clients.values()) { @@ -105,6 +140,55 @@ setInterval(() => { } }, 30_000).unref(); +/* ── Discovery beacon ─────────────────────────────────────────────────────── */ + +function beacon() { + return JSON.stringify({ + type: "veskslop-remchild", + name: hostname() || "veskslop", + port, + id: `veskslop-${process.pid}` + }); +} + +const discovery = dgram.createSocket({ type: "udp4", reuseAddr: true }); + +discovery.on("message", (message, remote) => { + if (!message.toString("utf8").includes("probe")) return; + const payload = Buffer.from(beacon()); + discovery.send(payload, remote.port, remote.address, () => {}); +}); + +discovery.on("error", error => { + console.error(`[remchild] discovery failed: ${error.message}`); +}); + +if (beaconMs > 0) { + setInterval(() => { + const payload = Buffer.from(beacon()); + + // Per-interface subnet broadcast, plus loopback for clients on this host. + for (const list of Object.values(networkInterfaces())) { + for (const info of list ?? []) { + if (info.family !== "IPv4" || info.internal || !info.netmask) continue; + const address = info.address.split(".").map(Number); + const mask = info.netmask.split(".").map(Number); + const broadcast = address.map((octet, index) => (octet | (~mask[index] & 255)) >>> 0).join("."); + discovery.send(payload, discoveryPort, broadcast, () => {}); + } + } + + discovery.send(payload, discoveryPort, "127.0.0.1", () => {}); + }, beaconMs).unref(); +} + +discovery.bind(discoveryPort, () => { + discovery.setBroadcast(true); + console.log( + `[remchild] discovery on udp/${discoveryPort}${beaconMs > 0 ? ` (beacon every ${beaconMs} ms)` : " (probe replies only)"}` + ); +}); + const server = net.createServer(socket => { const address = socket.remoteAddress?.replace(/^::ffff:/, "") ?? "unknown"; @@ -151,6 +235,13 @@ const server = net.createServer(socket => { socket, firstSeen: existing?.firstSeen ?? Date.now(), lastSeen: Date.now(), + connectedAt: Date.now(), + heartbeatMs: Number(message.heartbeatMs) > 0 ? Number(message.heartbeatMs) : 300000, + missed: 0, + nextDue: + Date.now() + + (Number(message.heartbeatMs) > 0 ? Number(message.heartbeatMs) : 300000) + + HEARTBEAT_GRACE_MS, heartbeats: 0, state: "active" }; @@ -173,7 +264,9 @@ const server = net.createServer(socket => { if (message.type === "heartbeat") { const client = [...clients.values()].find(item => item.socket === socket); if (client) { + if (Number(message.heartbeatMs) > 0) client.heartbeatMs = Number(message.heartbeatMs); client.heartbeats++; + client.nextDue = Date.now() + heartbeatInterval(client) + HEARTBEAT_GRACE_MS; touch(client, "heartbeat"); } continue;