Add UDP discovery and heartbeat-miss notices to remchild

This commit is contained in:
Fisto
2026-09-22 19:31:19 +02:00
parent a7ab2a22f7
commit e38d7d07a2
4 changed files with 278 additions and 22 deletions
+8 -3
View File
@@ -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 ./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 With no `--server`, the client probes for a server for five seconds: one answer connects
accepts `p` (ping test) and `q` (quit). 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 ```sh
node remchild/server/server.mjs --port 8799 --ips 203.0.113.7 --allow-non-lan node remchild/server/server.mjs --port 8799 --ips 203.0.113.7 --allow-non-lan
+154 -16
View File
@@ -31,6 +31,10 @@
#include "words.h" #include "words.h"
#define DISCOVERY_PORT 8798
#define DISCOVERY_WINDOW_MS 5000
#define DISCOVERY_MAX 8
#define HEARTBEAT_FAST_MS 10000 #define HEARTBEAT_FAST_MS 10000
#define HEARTBEAT_MID_MS 30000 #define HEARTBEAT_MID_MS 30000
#define HEARTBEAT_SLOW_MS 300000 #define HEARTBEAT_SLOW_MS 300000
@@ -38,6 +42,12 @@
#define MID_UNTIL_MS 360000 #define MID_UNTIL_MS 360000
#define LINE_MAX 4096 #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) { static long now_ms(void) {
struct timespec ts; struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &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]); 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) { static int connect_to(const char *host, int port) {
struct sockaddr_in addr = {0}; struct sockaddr_in addr = {0};
addr.sin_family = AF_INET; addr.sin_family = AF_INET;
@@ -159,17 +263,43 @@ int main(int argc, char **argv) {
int interactive = isatty(STDIN_FILENO); int interactive = isatty(STDIN_FILENO);
if (!host[0]) { 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); fflush(stdout);
char line[128]; int count = discover(found, DISCOVERY_MAX);
if (!fgets(line, sizeof(line), stdin)) return 1;
line[strcspn(line, "\n")] = 0; if (count == 1) {
char *colon = strrchr(line, ':'); printf("[remchild] found %s at %s:%d\n", found[0].name, found[0].ip, found[0].port);
if (colon) { snprintf(host, sizeof(host), "%s", found[0].ip);
*colon = 0; port = found[0].port;
port = atoi(colon + 1); } 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]; 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] identity %s (name: %s)\n", id_len == 6 ? "MAC" : "machine-id", name);
printf("[remchild] connecting to %s:%d\n", host, port); printf("[remchild] connecting to %s:%d\n", host, port);
fflush(stdout);
int fd = connect_to(host, port); int fd = connect_to(host, port);
if (fd < 0) { if (fd < 0) {
@@ -214,8 +345,9 @@ int main(int argc, char **argv) {
char hello[512]; char hello[512];
snprintf(hello, sizeof(hello), snprintf(hello, sizeof(hello),
"{\"type\":\"hello\",\"id\":\"%s\",\"name\":\"%s\",\"platform\":\"linux\",\"version\":\"0.1\"}", "{\"type\":\"hello\",\"id\":\"%s\",\"name\":\"%s\",\"platform\":\"linux\","
id_hex, name); "\"version\":\"0.1\",\"heartbeatMs\":%ld}",
id_hex, name, heartbeat_interval(0));
send_line(fd, hello); send_line(fd, hello);
long started = now_ms(); long started = now_ms();
@@ -234,13 +366,12 @@ int main(int argc, char **argv) {
for (;;) { for (;;) {
long now = now_ms(); long now = now_ms();
long interval = now - started < FAST_UNTIL_MS ? HEARTBEAT_FAST_MS long interval = heartbeat_interval(now - started);
: now - started < MID_UNTIL_MS ? HEARTBEAT_MID_MS
: HEARTBEAT_SLOW_MS;
if (now - last_beat >= interval) { if (now - last_beat >= interval) {
char beat[128]; 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); send_line(fd, beat);
last_beat = now; last_beat = now;
} }
@@ -272,9 +403,16 @@ int main(int argc, char **argv) {
} else if (strstr(line, "\"type\":\"pong\"")) { } 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); if (ping_sent_at) printf("[remchild] pong seq=%d rtt=%ldms\n", ping_seq, now_ms() - ping_sent_at);
answered_ping = 1; 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\"")) { } else if (strstr(line, "\"type\":\"ping\"")) {
int seq = 0; int seq = 0;
sscanf(line, "%*[^\"]\"seq\":%d", &seq); char *value = strstr(line, "\"seq\":");
if (value) seq = atoi(value + 6);
char pong[64]; char pong[64];
snprintf(pong, sizeof(pong), "{\"type\":\"pong\",\"seq\":%d}", seq); snprintf(pong, sizeof(pong), "{\"type\":\"pong\",\"seq\":%d}", seq);
send_line(fd, pong); send_line(fd, pong);
+23 -3
View File
@@ -51,6 +51,11 @@ The client sends `heartbeat` on this schedule after boot:
The server keeps three states from `lastSeen`: 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 | | State | Age | Effect |
| --- | --- | --- | | --- | --- | --- |
| `active` | under 12 min | Jobs and commands may be sent | | `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. 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 ## Transport
Newline-delimited JSON over TCP on a dedicated port (PoC: 8799). No Discord connection, no HTTP Newline-delimited JSON over TCP on a dedicated port (PoC: 8799), plus the UDP discovery above. No
framework. Discord connection, no HTTP framework.
| Direction | Message | Meaning | | Direction | Message | Meaning |
| --- | --- | --- | | --- | --- | --- |
@@ -70,6 +87,9 @@ framework.
| server → client | `welcome` | Assigned name, heartbeat interval, server name | | server → client | `welcome` | Assigned name, heartbeat interval, server name |
| client → server | `heartbeat` | `ts` | | client → server | `heartbeat` | `ts` |
| both | `ping` / `pong` | `seq` echo, for the test button and latency | | 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 | | server → client | `job` | Later: a unit of work with an ID and a payload |
| client → server | `job-result` | Later: `id`, `ok`, `data` or `error` | | 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 | | 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. | | 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. | | 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. | | M4 | Video: stream a window or play a file with MPV, and return frames or a stream URL. |
+93
View File
@@ -9,10 +9,15 @@
* Run: node remchild/server/server.mjs --port 8799 --ips 127.0.0.1,192.168.1.5 * 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-non-lan --ips 203.0.113.7
* node remchild/server/server.mjs --allow-all # testing: any address * 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 { appendFileSync, existsSync, mkdirSync, readFileSync } from "node:fs";
import net from "node:net"; import net from "node:net";
import { hostname, networkInterfaces } from "node:os";
import { dirname, join } from "node:path"; import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url"; 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 port = Number(arg("port", process.env.REMCHILD_PORT ?? 8799));
const allowNonLan = arg("allow-non-lan", false) !== false; const allowNonLan = arg("allow-non-lan", false) !== false;
const allowAll = arg("allow-all", 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 historyFile = arg("history", join(HERE, "clients.jsonl"));
const configFile = arg("config", join(HERE, "config.json")); const configFile = arg("config", join(HERE, "config.json"));
@@ -62,6 +69,8 @@ function allowed(address) {
return configured.includes(address); return configured.includes(address);
} }
const HEARTBEAT_GRACE_MS = 5000;
const clients = new Map(); const clients = new Map();
mkdirSync(dirname(historyFile), { recursive: true }); 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(() => { setInterval(() => {
const now = Date.now(); const now = Date.now();
for (const client of clients.values()) { for (const client of clients.values()) {
@@ -105,6 +140,55 @@ setInterval(() => {
} }
}, 30_000).unref(); }, 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 server = net.createServer(socket => {
const address = socket.remoteAddress?.replace(/^::ffff:/, "") ?? "unknown"; const address = socket.remoteAddress?.replace(/^::ffff:/, "") ?? "unknown";
@@ -151,6 +235,13 @@ const server = net.createServer(socket => {
socket, socket,
firstSeen: existing?.firstSeen ?? Date.now(), firstSeen: existing?.firstSeen ?? Date.now(),
lastSeen: 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, heartbeats: 0,
state: "active" state: "active"
}; };
@@ -173,7 +264,9 @@ const server = net.createServer(socket => {
if (message.type === "heartbeat") { if (message.type === "heartbeat") {
const client = [...clients.values()].find(item => item.socket === socket); const client = [...clients.values()].find(item => item.socket === socket);
if (client) { if (client) {
if (Number(message.heartbeatMs) > 0) client.heartbeatMs = Number(message.heartbeatMs);
client.heartbeats++; client.heartbeats++;
client.nextDue = Date.now() + heartbeatInterval(client) + HEARTBEAT_GRACE_MS;
touch(client, "heartbeat"); touch(client, "heartbeat");
} }
continue; continue;