From f4cd9a7e37213574f9f98b72c328d922b4630c8c Mon Sep 17 00:00:00 2001 From: Malcolm Habeeb Date: Wed, 12 Aug 2026 15:43:45 -0400 Subject: [PATCH 1/5] XERK-272: bound the agent registry itself, not just one record MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `agents` is keyed by the heartbeat's `device`, which every agent mints with one shared TURMA_AGENT_TOKEN — so the number of RETAINED records was unbounded. AGENT_RECORD_MAX bounds one record and prune() only reclaims at seven days, so 512 beats of 0.9 MiB under 512 names OOM-killed a 256 MiB hub while the same 512 beats under ONE name peaked at 169 MiB. The per-record ceiling was never the aggregate. Two budgets, bounding different things: * AGENTS_TOTAL_MAX — aggregate record bytes, derived from the container's own cgroup limit (an eighth, clamped 8-64 MiB) rather than picked, since a ceiling above the limit the kernel kills on is not a ceiling. * AGENTS_MAX — record count (64). Not redundant: the byte budget measures what agentRecordSize measures, which excludes the on-demand caches, so only a cap on hosts bounds their multiple. A host already in the registry is always admitted — turning the cap into a wall for the fleet's own hosts is the same outage from the other side. A NEW device gets a slot only if there is room, or one can be reclaimed from a host unseen for AGENT_EVICT_IDLE_MS (1h); otherwise it is refused 429 and logged. A record holds an offline host's last known sessions, PR chips and usage, so a rebooting or updating host is never displaced by a newcomer, and eviction never runs when it could not satisfy the request anyway. The state.json restore enforces the same budget (keep-newest): a flood that landed before a restart is on disk, and a bound the loading path skips is not a bound — the same reason normalizeRecord runs on the restore too. Byte accounting is a side map, never a field on the record, and re-measures unknown keys and forgets dead ones so the many `delete agents[key]` sites need not remember it. Tests: turma/tests/registry-cap.test.js (own process, tiny caps). --- .claude/rules/turma.md | 32 ++++ turma/server.js | 264 ++++++++++++++++++++++++++-- turma/tests/registry-cap.test.js | 287 +++++++++++++++++++++++++++++++ turma/tests/server.test.js | 7 + 4 files changed, 574 insertions(+), 16 deletions(-) create mode 100644 turma/tests/registry-cap.test.js diff --git a/.claude/rules/turma.md b/.claude/rules/turma.md index fba40f95..56215ddf 100644 --- a/.claude/rules/turma.md +++ b/.claude/rules/turma.md @@ -253,6 +253,38 @@ working-status bar, ready-for-review, ended sessions, the composer and the termi session's ttyd by port) sit behind single-user HTTP Basic auth (`TURMA_USER`/`TURMA_PASSWORD`). Agents authenticate heartbeats, tunnel WebSockets, and ttyd with one shared token (`TURMA_TOKEN` in the agent's env = `TURMA_AGENT_TOKEN` on the hub). + - **One token means the heartbeat's `device` is attacker-chosen**, so it names a record but never + authorizes one. Anything keyed on it needs its own bound. + +### The agent registry's ceiling (XERK-272) + +- **`agents` is bounded as an AGGREGATE, not just per record.** `AGENT_RECORD_MAX` (8 MiB) bounds + ONE record and `prune()` only reclaims at 7 days, so an unbounded NUMBER of `device` names was an + unbounded amount of retained memory — 512 beats of 0.9 MiB under 512 names OOM-killed a 256 MiB + hub, while the same 512 beats under ONE name peaked at 169 MiB. +- Two budgets, bounding different things: **`AGENTS_TOTAL_MAX`** (aggregate record bytes, defaulting + to an eighth of the container's own cgroup limit, clamped 8–64 MiB) and **`AGENTS_MAX`** (record + count, default 64). The count cap is not redundant — the byte budget measures what + `agentRecordSize` measures, which EXCLUDES the on-demand caches, so only a cap on hosts bounds + their multiple. A ceiling above the limit the kernel kills on is not a ceiling: size it from the + container, never pick a number. +- **A newcomer never displaces a host that is still around.** Past the cap a record is reclaimed + only if it has been unseen for `AGENT_EVICT_IDLE_MS` (1h, ≫ `OFFLINE_AFTER_MS`) — a record holds + an offline host's last known sessions, PR chips and usage, so a host rebooting or updating keeps + its slot and the new `device` gets a 429 instead. The accepted cost is that a flood of names can + squat slots and block onboarding a genuinely new host until it stops or `AGENTS_MAX` is raised; + per-agent tokens are the real fix and are not this. +- A host **already in the registry is always admitted** — turning the cap into a wall for the fleet's + own hosts is the same outage from the other side. The aggregate can still refuse its beat, which + is contention (it lands on a later beat), not a wall, and never damages the record being served. +- **The state.json restore enforces the same budget** (`trimRestoredAgents`, keep-newest): a flood + that landed before a restart is on disk, and a bound the loading path skips is not a bound. Same + reason `normalizeRecord` runs on the restore as well as the ingest. +- Byte accounting is a side map (`recordBytes`), never a field on the record — anything on a record + is served to every client — and `registryBytes()` re-measures unknown keys and forgets dead ones, + so the many places that `delete agents[key]` need not remember it. +- Tests: `registry-cap.test.js`, which pins tiny caps in its own process (`server.test.js` lifts + `AGENTS_MAX` because ~100 synthetic host names is not a fleet). - The hub also serves the `glasses/` client: a CORS'd `/api/*` surface for that cross-origin WebView; per-session `input`/`history` endpoints; `GET /api/ws-token` for short-lived WebSocket auth; an `/audio` STT WebSocket (G2-mic PCM to the LiteLLM instance's transcription endpoint); and diff --git a/turma/server.js b/turma/server.js index 73955abd..77a3de8a 100644 --- a/turma/server.js +++ b/turma/server.js @@ -356,6 +356,192 @@ const PR_ALERT_MAX_TRACKED = 20; // bookkeeping. One container per host, so the host name is the stable identity. let agents = {}; +// ---- the registry's own ceiling (XERK-272) --------------------------------- +// +// `device` is attacker-chosen: every agent shares one TURMA_AGENT_TOKEN, so any +// buggy or compromised host can mint an unlimited number of DISTINCT keys, and +// every one of them becomes a retained record that also lands in state.json, +// every /api/agents body and every SSE frame. `prune()` only reclaims at seven +// days and AGENT_RECORD_MAX bounds ONE record, so 512 beats of 0.9 MiB under +// 512 names OOM-killed a 256 MiB hub while the same 512 beats under ONE name +// peaked at 169 MiB. The per-record bound was never the aggregate. +// +// Two budgets, because they bound different things: +// * AGENTS_TOTAL_MAX — the aggregate BYTES of every record, sized from the +// container the hub actually runs in. A ceiling above the limit the kernel +// kills on is not a ceiling (XERK-258), so this is derived from the cgroup +// rather than picked. +// * AGENTS_MAX — the record COUNT. The byte budget measures what +// `agentRecordSize` measures, which deliberately EXCLUDES the on-demand +// caches (history, subagent history, Jira issues, create results); those are +// bounded per host by count, so only a cap on hosts bounds their multiple. +// +// Both are env-overridable: an operator who grows the fleet past the cap should +// raise it in compose, not discover the hub silently dropping a host. +const AGENTS_MAX = Number(process.env.AGENTS_MAX) || 64; +// How long a record must have gone unseen before the registry may reclaim its +// slot for a host it has never met. Deliberately far longer than +// OFFLINE_AFTER_MS: a record holds an offline host's last known sessions, PR +// chips and usage, so a host rebooting, updating, or off for the afternoon must +// never be displaced by a newcomer — a flood is refused instead. Only a host +// that has been gone long enough to be uninteresting is evictable. +const AGENT_EVICT_IDLE_MS = Number(process.env.AGENT_EVICT_IDLE_MS) || 60 * 60 * 1000; + +// The container's memory limit in bytes, or null when it cannot be read (not +// containerised, no cgroupfs, or explicitly unlimited). cgroup v2 first, then +// v1, whose "unlimited" is a near-2^63 sentinel rather than a word. +function containerMemoryLimit() { + for (const p of ["/sys/fs/cgroup/memory.max", + "/sys/fs/cgroup/memory/memory.limit_in_bytes"]) { + let raw; + try { raw = fs.readFileSync(p, "utf8").trim(); } catch { continue; } + if (raw === "max") continue; + const n = Number(raw); + // 2 ** 40, not `1 << 40` — JS bitwise is 32-bit and would wrap to 0. + if (Number.isFinite(n) && n > 0 && n < 2 ** 40) return n; + } + return null; +} + +// An eighth of the container, clamped. The deployed hub is `mem_limit: 256m`, +// so that is 32 MiB retained — against a measured real fleet whose LARGEST +// record is 0.30 MiB, i.e. ~100 hosts of headroom. The eighth is what leaves +// room for the copies the registry implies: the memoized /api/agents body, the +// state.json blob the save timer builds, and the per-record SSE frames all +// materialize alongside it. +function defaultRegistryBudget() { + const limit = containerMemoryLimit(); + if (!limit) return 32 << 20; + return Math.max(8 << 20, Math.min(64 << 20, Math.floor(limit / 8))); +} +const AGENTS_TOTAL_MAX = Number(process.env.AGENTS_TOTAL_MAX) || defaultRegistryBudget(); + +// The cache keys `serializeAgent` strips; see AGENT_RECORD_MAX. +// +// Declared UP HERE, above the state.json restore, for the reason LIVE_AGENTS_MAX +// documents: the restore runs at module init and enforces the budget below, so +// every constant that path reads has to exist by then or the `const` is in its +// temporal dead zone and the restore's own `catch {}` swallows the throw. +const AGENT_CACHE_KEYS = [ + "history", "subagentHistory", "jiraIssues", "statusResults", + "createMeta", "createTypes", "createResults", "resultWaits", +]; + +// The serialized size of what this record contributes to /api/agents. +function agentRecordSize(record) { + try { + return JSON.stringify(record, (k, v) => + AGENT_CACHE_KEYS.includes(k) && v && typeof v === "object" ? undefined : v + ).length; + } catch { + return Infinity; // circular or unserializable — it cannot be persisted anyway + } +} + +// Last measured size per host, so the aggregate costs a sum over numbers rather +// than re-serializing the whole registry on every beat. A side map, not a field +// on the record: anything stored on the record is served to every client. +const recordBytes = new Map(); + +// The aggregate `agentRecordSize` of the whole registry. Measures lazily for a +// key it has not seen (the state.json restore, and the tests, install records +// without going through the heartbeat) and forgets keys that are gone, so it +// stays correct without every `delete agents[key]` site having to remember it. +function registryBytes() { + let total = 0; + for (const [key, a] of Object.entries(agents)) { + let n = recordBytes.get(key); + if (n === undefined) recordBytes.set(key, (n = agentRecordSize(a))); + total += n; + } + if (recordBytes.size > Object.keys(agents).length) { + for (const key of recordBytes.keys()) { + if (!Object.prototype.hasOwnProperty.call(agents, key)) recordBytes.delete(key); + } + } + return total; +} + +// Hosts whose slot may be reclaimed for a newcomer, least-recently-seen first. +function evictableAgents(now) { + return Object.entries(agents) + .filter(([, a]) => now - (a.lastSeen || 0) > AGENT_EVICT_IDLE_MS) + .sort((a, b) => (a[1].lastSeen || 0) - (b[1].lastSeen || 0)) + .map(([key]) => key); +} + +// Reclaim long-idle records until the registry has room for `addSlots` more +// hosts and `addBytes` more bytes; returns whether it does. `false` is the +// caller's cue to REFUSE the beat rather than take a live host's slot — see +// AGENT_EVICT_IDLE_MS for why a newcomer never wins that trade. +function makeRegistryRoom(addBytes, addSlots) { + let bytes = registryBytes(); + let count = Object.keys(agents).length; + const overBudget = () => + bytes + addBytes > AGENTS_TOTAL_MAX || count + addSlots > AGENTS_MAX; + if (!overBudget()) return true; + const now = Date.now(); + const evictable = evictableAgents(now); + // Would evicting ALL of them even be enough? If not, evict none: dropping + // records the caller is going to be refused anyway loses an offline host's + // last known state and buys nothing. + const reclaimable = evictable.reduce((n, key) => n + (recordBytes.get(key) || 0), 0); + if (bytes - reclaimable + addBytes > AGENTS_TOTAL_MAX || + count - evictable.length + addSlots > AGENTS_MAX) return false; + const before = evictable.length; + while (overBudget() && evictable.length) { + const key = evictable.shift(); + bytes -= recordBytes.get(key) || 0; + count -= 1; + console.warn( + `registry at its limit — evicting ${key}, unseen for ` + + `${Math.round((now - (agents[key].lastSeen || 0)) / 60000)}m` + ); + delete agents[key]; + recordBytes.delete(key); + invalidateAgentsCache(); + sseBroadcast("removed", { key }); + } + // An eviction has to reach state.json, or a restart brings the record back. + if (evictable.length !== before) scheduleSave(); + return !overBudget(); +} + +// Hold the restored registry to the same budget. A hub that was flooded before +// it restarted has the flood ON DISK, and loading all of it is an OOM before +// the first request is served — a bound that the path which LOADS the state +// doesn't enforce is not a bound (the same reason `normalizeRecord` runs on the +// restore as well as the ingest, XERK-259). +// +// Unconditional keep-newest, NOT the idle rule above: nothing is live at boot, +// every record is by definition from before the restart, and the alternative to +// dropping the stalest is not booting at all. +function trimRestoredAgents() { + const keys = Object.keys(agents); + if (!keys.length) return; + const newestFirst = keys.sort((a, b) => (agents[b].lastSeen || 0) - (agents[a].lastSeen || 0)); + let bytes = 0; + const dropped = []; + newestFirst.forEach((key, i) => { + const size = agentRecordSize(agents[key]); + if (i < AGENTS_MAX && bytes + size <= AGENTS_TOTAL_MAX) { + bytes += size; + recordBytes.set(key, size); + return; + } + dropped.push(key); + delete agents[key]; + recordBytes.delete(key); + }); + if (dropped.length) { + console.warn( + `dropped ${dropped.length} restored agent record(s) over the registry ` + + `budget (${AGENTS_MAX} hosts / ${AGENTS_TOTAL_MAX} bytes); kept the ` + + `${keys.length - dropped.length} most recently seen` + ); + } +} + // Reverse-tunnel state. controlChannels[name] = the live control connection for // that container's tunnel-agent; pendingChannels[ch] = resolver awaiting the // agent's data-WS dial-back for channel `ch`. @@ -380,6 +566,10 @@ try { // with the ingest path so the two cannot drift; adding a coercion in one // place covers both. Tests: `the state.json restore coerces too`. for (const a of Object.values(agents)) normalizeRecord(a); + // Hold what we LOAD to the registry budget too — a flood that landed before + // the restart is on disk, and restoring all of it is an OOM before the first + // request (XERK-272). + trimRestoredAgents(); console.log(`loaded ${Object.keys(agents).length} agents from ${STATE_FILE}`); } catch { /* first boot or no volume mounted */ @@ -1851,22 +2041,9 @@ const AGENT_RECORD_MAX = 8 << 20; // 8 MiB // the crossing rather than on every beat. const recordSizeWarned = new Map(); -// The cache keys `serializeAgent` strips; see AGENT_RECORD_MAX. -const AGENT_CACHE_KEYS = [ - "history", "subagentHistory", "jiraIssues", "statusResults", - "createMeta", "createTypes", "createResults", "resultWaits", -]; - -// The serialized size of what this record contributes to /api/agents. -function agentRecordSize(record) { - try { - return JSON.stringify(record, (k, v) => - AGENT_CACHE_KEYS.includes(k) && v && typeof v === "object" ? undefined : v - ).length; - } catch { - return Infinity; // circular or unserializable — it cannot be persisted anyway - } -} +// `AGENT_CACHE_KEYS` and `agentRecordSize` are declared with the registry +// budget instead (see `let agents`), because the state.json restore enforces +// that budget at module init and so has to be able to measure a record. // Drop unrecognised keys that are too large to be a plausible new field. // @@ -3847,6 +4024,24 @@ const server = http.createServer(async (req, res) => { key === "__proto__" || key === "constructor" || key === "prototype") { return json(res, 400, { error: "device must be a plain host name" }); } + const known = Object.prototype.hasOwnProperty.call(agents, key); + // Admission control (XERK-272). A host already in the registry always gets + // in — its record is bounded below and replacing it frees what it held — + // but a name the hub has never seen only gets a slot if there is room, or + // one can be reclaimed from a host gone longer than AGENT_EVICT_IDLE_MS. + // Refusing here, BEFORE the record is built, is the point: `device` is + // attacker-chosen, so an unbounded number of names is an unbounded number + // of retained records. + if (!known && !makeRegistryRoom(0, 1)) { + console.error( + `heartbeat from ${key}: registry is full ` + + `(${Object.keys(agents).length}/${AGENTS_MAX} hosts, ` + + `${registryBytes()}/${AGENTS_TOTAL_MAX} bytes) — new host refused` + ); + return json(res, 429, { + error: "agent registry full", limit: AGENTS_MAX, bytes: AGENTS_TOTAL_MAX, + }); + } const prev = agents[key] || {}; // At-least-once command delivery: drop any queued command the agent // reports as executed; keep re-sending the rest until acked. @@ -3978,6 +4173,29 @@ const server = http.createServer(async (req, res) => { } recordSizeWarned.set(key, overHalf); if (recordSize > AGENT_RECORD_MAX) return refuseOversized(recordSize); + // The AGGREGATE budget (XERK-272). One record under the per-record ceiling + // is not the bound: AGENTS_MAX records AT that ceiling is 512 MiB on a + // 256 MiB hub. Measured after the coercion, because the coerced record is + // the one that gets retained, served and saved. + recordBytes.set(key, recordSize); + if (!makeRegistryRoom(0, 0)) { + // This host is never what gets evicted — it was just seen, so the idle + // rule excludes it. And refusing here is contention, not a wall: the + // host beats again in ~20s and lands the moment there is room, rather + // than being stuck at its old record forever. + if (prev && Object.keys(prev).length) { + agents[key] = prev; + recordBytes.set(key, agentRecordSize(prev)); + } else { + delete agents[key]; + recordBytes.delete(key); + } + console.error( + `heartbeat from ${key}: record is ${recordSize} bytes and the registry ` + + `is at its ${AGENTS_TOTAL_MAX}-byte budget — beat refused` + ); + return json(res, 429, { error: "agent registry full", bytes: AGENTS_TOTAL_MAX }); + } ingestHistory(next, historyResults); ingestSubagentHistory(next, subagentHistoryResults); ingestJiraIssues(next, jiraIssueResults); @@ -4001,6 +4219,14 @@ const server = http.createServer(async (req, res) => { const reply = publicCommands(commands); // strip AFTER stamping, or the scheduleSave(); // no-op copy hands back the // same objects and leaks it + // Re-measure what the record ACTUALLY ended up as. The gate above runs + // before the ingests on purpose (a refused beat must never reach the + // caches, XERK-235), so it measures the record before the alert/PR + // bookkeeping lands on it — and an aggregate that only ever sees the + // pre-bookkeeping size drifts low, which is a budget that quietly grows. + // Settling it here costs the next beat's gate nothing and keeps the + // number honest; a beat that ends slightly over is caught on that gate. + recordBytes.set(key, agentRecordSize(next)); // A fresh beat landed — refresh the memoized fleet payload and push the // updated record to open dashboards so the UI reflects it near-instantly. publishAgent(key); @@ -5716,6 +5942,12 @@ if (process.env.TURMA_TEST) { // and the suite stayed green, so they are exported to be pinned. sanitizeHeartbeat, agentRecordSize, safeAgentsCache, HEARTBEAT_UNKNOWN_MAX, AGENT_RECORD_MAX, + // XERK-272 registry bounds. Exported for the same reason as the group above: + // the per-record ceiling stayed green while an unbounded NUMBER of records + // OOM-killed the hub, so the aggregate has to be pinned by name too. + AGENTS_MAX, AGENTS_TOTAL_MAX, AGENT_EVICT_IDLE_MS, + registryBytes, makeRegistryRoom, trimRestoredAgents, containerMemoryLimit, + defaultRegistryBudget, recordBytes, // Ingest coercion, exported for the same reason as the rest of this group: // Android decodes /api/agents atomically, so one host's wrong-typed field // hides the WHOLE fleet from that phone (XERK-246). `normalizeRecord` is diff --git a/turma/tests/registry-cap.test.js b/turma/tests/registry-cap.test.js new file mode 100644 index 00000000..b422587a --- /dev/null +++ b/turma/tests/registry-cap.test.js @@ -0,0 +1,287 @@ +// Unit tests for the agent registry's own ceiling (XERK-272). +// +// `device` is attacker-chosen — every agent shares one TURMA_AGENT_TOKEN — so +// the number of RETAINED records was unbounded: 512 beats of 0.9 MiB under 512 +// names OOM-killed a 256 MiB hub, while the same 512 beats under ONE name +// peaked at 169 MiB. `AGENT_RECORD_MAX` bounds one record and `prune()` only +// reclaims at seven days, so neither is the aggregate. +// +// This gets its OWN process because the caps are process-wide constants read at +// require time, and the numbers that make the behavior testable (4 hosts, 64 +// KiB) are nothing like the fleet's. server.test.js lifts `AGENTS_MAX` for the +// opposite reason: it invents ~100 synthetic hosts and is not a fleet either. +// node:test, no npm. + +"use strict"; + +const os = require("os"); +const fs = require("fs"); +const path = require("path"); +const http = require("http"); +const test = require("node:test"); +const assert = require("node:assert/strict"); + +// Environment must be pinned BEFORE the module under test loads. +process.env.TURMA_TEST = "1"; +process.env.TURMA_USER = "hubuser"; +process.env.TURMA_PASSWORD = "hubpass"; +process.env.TURMA_AGENT_TOKEN = "agenttok"; +process.env.AGENTS_MAX = "4"; +process.env.AGENTS_TOTAL_MAX = String(64 << 10); +process.env.AGENT_EVICT_IDLE_MS = String(60 * 1000); + +const tmp = (name) => path.join(os.tmpdir(), `turma-regcap-${name}-${process.pid}.json`); +process.env.DEVICES_FILE = tmp("devices"); +process.env.TICKET_AGENTS_FILE = tmp("ticket-agents"); +process.env.AUTOSTART_ORGS_FILE = tmp("autostart-orgs"); +process.env.TICKET_MODELS_FILE = tmp("ticket-models"); +process.env.ORG_COLORS_FILE = tmp("org-colors"); +process.env.MIGRATE_SPOOL_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "turma-regcap-migrations-")); +process.env.ARCHIVE_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "turma-regcap-archive-")); +process.env.ARCHIVE_DB = path.join(process.env.ARCHIVE_DIR, "index.db"); + +// A state.json holding SIX hosts against a cap of four — the shape a hub that +// was flooded before it restarted has on disk. Written before the require so +// the restore path sees it. +process.env.STATE_FILE = tmp("state"); +const RESTORE_NOW = Date.now(); +const restoredSeen = { + "keep-newest": RESTORE_NOW - 1000, + "keep-2": RESTORE_NOW - 2000, + "keep-3": RESTORE_NOW - 3000, + "keep-4": RESTORE_NOW - 4000, + "drop-stale": RESTORE_NOW - 5000, + "drop-stalest": RESTORE_NOW - 6000, +}; +fs.writeFileSync( + process.env.STATE_FILE, + JSON.stringify( + Object.fromEntries( + Object.entries(restoredSeen).map(([key, lastSeen]) => [ + key, { device: key, lastSeen, repos: [{ name: "r1" }], sessions: [] }, + ]) + ) + ) +); + +const hub = require("../server.js"); +const { + server, agents, recordBytes, + AGENTS_MAX, AGENTS_TOTAL_MAX, AGENT_EVICT_IDLE_MS, + registryBytes, makeRegistryRoom, agentRecordSize, + containerMemoryLimit, defaultRegistryBudget, +} = hub; + +// What the restore left behind, snapshotted before any test mutates the +// registry (the trim runs at require time, once). +const restoredKeys = Object.keys(agents).slice().sort(); + +// ---- the restore trim ------------------------------------------------------- + +test("restore: a state.json over the cap loads only the most recently seen", () => { + // A bound the LOADING path doesn't enforce is not a bound — restoring the + // whole flood is an OOM before the first request is served. + assert.equal(restoredKeys.length, AGENTS_MAX); + assert.deepEqual(restoredKeys, ["keep-2", "keep-3", "keep-4", "keep-newest"]); +}); + +test("restore: the trim seeds the byte accounting for what it kept", () => { + // Not cosmetic: the aggregate check on the first beat after a restart reads + // these, and a registry that measures as 0 admits a flood. + for (const key of restoredKeys) { + assert.equal(typeof recordBytes.get(key), "number", `${key} unmeasured`); + assert.ok(recordBytes.get(key) > 0); + } +}); + +// ---- HTTP ------------------------------------------------------------------ + +let baseUrl; +test.before(async () => { + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + baseUrl = `http://127.0.0.1:${server.address().port}`; +}); +test.after(() => server.close()); + +function request(method, pathName, { body, headers } = {}) { + return new Promise((resolve, reject) => { + const req = http.request(baseUrl + pathName, { method, headers }, (res) => { + let data = ""; + res.on("data", (c) => (data += c)); + res.on("end", () => { + let parsed = null; + try { parsed = JSON.parse(data); } catch { /* not JSON */ } + resolve({ status: res.statusCode, body: parsed, raw: data }); + }); + }); + req.on("error", reject); + if (body) req.write(JSON.stringify(body)); + req.end(); + }); +} + +const agentHeaders = { authorization: "Bearer agenttok", "content-type": "application/json" }; +const beat = (body) => request("POST", "/api/heartbeat", { body, headers: agentHeaders }); + +// Each test starts from an empty registry: the caps are small enough that one +// test's hosts would otherwise be the next one's flood. +function resetRegistry() { + for (const key of Object.keys(agents)) delete agents[key]; + recordBytes.clear(); +} + +// A record big enough to matter against the 64 KiB aggregate but far under the +// 8 MiB per-record ceiling, so what refuses it is unambiguously the aggregate. +const chunky = (device, kib) => ({ + device, + sessions: [{ id: "s1", status: "running", label: "L".repeat(kib << 10) }], +}); + +test("http: past the cap, a NEW device is refused rather than admitted", async () => { + resetRegistry(); + for (let i = 0; i < AGENTS_MAX; i++) { + assert.equal((await beat({ device: `full-${i}` })).status, 200); + } + const refused = await beat({ device: "one-too-many" }); + assert.equal(refused.status, 429); + assert.equal(refused.body.error, "agent registry full"); + assert.equal(refused.body.limit, AGENTS_MAX); + // Refused means REFUSED: no record, no slot taken, nothing to serve. + assert.equal("one-too-many" in agents, false); + assert.equal(Object.keys(agents).length, AGENTS_MAX); +}); + +test("http: a host already in the registry keeps beating at the cap", async () => { + resetRegistry(); + for (let i = 0; i < AGENTS_MAX; i++) await beat({ device: `full-${i}` }); + // The cap is admission control on NEW names. Turning it into a wall for the + // fleet's own hosts would make every one of them go offline at once — the + // exact outage the ticket is about, arrived at from the other side. + const again = await beat({ device: "full-0", repos: [{ name: "r2" }] }); + assert.equal(again.status, 200); + assert.deepEqual(agents["full-0"].repos, [{ name: "r2" }]); +}); + +test("http: a live host is never evicted to seat a newcomer", async () => { + resetRegistry(); + for (let i = 0; i < AGENTS_MAX; i++) await beat({ device: `live-${i}` }); + // All four were seen just now, so none is reclaimable however old the newcomer + // makes the registry look. + assert.equal((await beat({ device: "newcomer" })).status, 429); + for (let i = 0; i < AGENTS_MAX; i++) { + assert.ok(agents[`live-${i}`], `live-${i} was evicted for a newcomer`); + } +}); + +test("http: a newcomer reclaims the LONG-idle slot, oldest first", async () => { + resetRegistry(); + for (let i = 0; i < AGENTS_MAX; i++) await beat({ device: `mix-${i}` }); + // Two hosts gone long past the idle window; the rest were seen a moment ago. + agents["mix-0"].lastSeen = Date.now() - AGENT_EVICT_IDLE_MS * 3; + agents["mix-1"].lastSeen = Date.now() - AGENT_EVICT_IDLE_MS * 2; + + assert.equal((await beat({ device: "newcomer" })).status, 200); + assert.ok(agents["newcomer"]); + // Least-recently-seen goes first, and ONLY as many as the newcomer needs — + // a record holds an offline host's last known sessions, PR chips and usage, + // so eviction is not free and is never done in bulk. + assert.equal("mix-0" in agents, false, "the stalest slot should have been taken"); + assert.ok(agents["mix-1"], "only one slot was needed"); + assert.ok(agents["mix-2"] && agents["mix-3"]); + assert.equal(recordBytes.has("mix-0"), false, "an evicted host's bytes must be released"); +}); + +test("http: a host just short of the idle window is NOT evictable", async () => { + resetRegistry(); + for (let i = 0; i < AGENTS_MAX; i++) await beat({ device: `grace-${i}` }); + // Offline (well past OFFLINE_AFTER_MS) but not yet idle enough to reclaim: + // a host rebooting or updating must not lose its record to a newcomer. + agents["grace-0"].lastSeen = Date.now() - (AGENT_EVICT_IDLE_MS - 5000); + assert.equal((await beat({ device: "newcomer" })).status, 429); + assert.ok(agents["grace-0"]); +}); + +// ---- the aggregate byte budget --------------------------------------------- + +test("http: records that fit the per-record ceiling still can't sum past the aggregate", async () => { + resetRegistry(); + // AGENTS_MAX records at AGENT_RECORD_MAX is 512 MiB on a 256 MiB hub, so the + // per-record ceiling was never the bound. Both of these pass it easily. + assert.equal((await beat(chunky("fat-a", 40))).status, 200); + const refused = await beat(chunky("fat-b", 40)); + assert.equal(refused.status, 429); + assert.equal(refused.body.error, "agent registry full"); + assert.equal(refused.body.bytes, AGENTS_TOTAL_MAX); + assert.equal("fat-b" in agents, false); + assert.ok(registryBytes() <= AGENTS_TOTAL_MAX); +}); + +test("http: an aggregate refusal leaves the host's PREVIOUS record intact", async () => { + resetRegistry(); + assert.equal((await beat({ device: "grower", repos: [{ name: "r1" }] })).status, 200); + assert.equal((await beat(chunky("ballast", 40))).status, 200); + // `grower` is known, so admission lets it through — the aggregate is what + // refuses it, and a refused beat must not damage what is being served. + const refused = await beat(chunky("grower", 40)); + assert.equal(refused.status, 429); + assert.deepEqual(agents["grower"].repos, [{ name: "r1" }]); + assert.equal(recordBytes.get("grower"), agentRecordSize(agents["grower"])); + // Not a wall: it beats again in ~20s and lands as soon as there is room. + assert.equal((await beat({ device: "grower", repos: [{ name: "r3" }] })).status, 200); + assert.deepEqual(agents["grower"].repos, [{ name: "r3" }]); +}); + +test("registryBytes tracks deletes it never saw, and matches a full re-measure", async () => { + resetRegistry(); + // Every route that drops a host (DELETE /api/agents/, prune(), the + // tests) mutates `agents` directly, so the accounting has to self-heal rather + // than depend on each of those sites remembering it. + await beat({ device: "gone", sessions: [{ id: "s1", status: "running" }] }); + await beat({ device: "stays", sessions: [{ id: "s2", status: "running" }] }); + const both = registryBytes(); + delete agents["gone"]; + const after = registryBytes(); + assert.ok(after < both); + assert.equal(after, agentRecordSize(agents["stays"])); + assert.equal(recordBytes.has("gone"), false); +}); + +// ---- sizing ---------------------------------------------------------------- + +test("the default budget is derived from the container, and clamped", () => { + // "A ceiling above the limit the kernel kills on is not a ceiling" (XERK-258) + // — so the default is read from the cgroup rather than picked, and clamped so + // a hostile/absent cgroup value can't produce a budget of nothing or of + // everything. + const budget = defaultRegistryBudget(); + assert.ok(budget >= (8 << 20) && budget <= (64 << 20), `budget ${budget} out of range`); + const limit = containerMemoryLimit(); + assert.ok(limit === null || (typeof limit === "number" && limit > 0)); + if (limit && limit / 8 >= (8 << 20) && limit / 8 <= (64 << 20)) { + assert.equal(budget, Math.floor(limit / 8)); + } +}); + +test("makeRegistryRoom reports failure instead of over-evicting", () => { + resetRegistry(); + agents["fresh"] = { device: "fresh", lastSeen: Date.now() }; + // Nothing reclaimable and no room: the answer is `false` (which the caller + // turns into a 429), never "evict the live host anyway". + assert.equal(makeRegistryRoom(AGENTS_TOTAL_MAX + 1, 0), false); + assert.ok(agents["fresh"]); + assert.equal(makeRegistryRoom(0, AGENTS_MAX), false); + assert.ok(agents["fresh"]); + // Room for one more host is exactly what a beat from a new device asks for. + assert.equal(makeRegistryRoom(0, 1), true); +}); + +test("makeRegistryRoom spends no record on a request it cannot satisfy", () => { + resetRegistry(); + agents["idle"] = { device: "idle", lastSeen: Date.now() - AGENT_EVICT_IDLE_MS * 2 }; + recordBytes.set("idle", 100); + // Evicting everything reclaimable still would not fit this, so evict nothing: + // the caller is refused either way, and an evicted record is an offline host's + // last known sessions, PR chips and usage, gone. + assert.equal(makeRegistryRoom(AGENTS_TOTAL_MAX + 1, 0), false); + assert.ok(agents["idle"], "an idle record was spent on a refusal"); +}); diff --git a/turma/tests/server.test.js b/turma/tests/server.test.js index 2dc0c80f..0070baaa 100644 --- a/turma/tests/server.test.js +++ b/turma/tests/server.test.js @@ -36,6 +36,13 @@ process.env.CONTROL_DEAD_AFTER_MS = "400"; // Same trick for the create single-flight's expiry (XERK-241): the fleet gives // an unresolved create 60s to rejoin a retry, which is only testable wound down. process.env.CREATE_INFLIGHT_TTL_MS = "300"; +// The registry cap (XERK-272) is sized for a FLEET — the deployed one is a +// handful of hosts. This suite is not a fleet: it invents ~100 synthetic host +// names in one process and never removes them, so it is lifted here rather than +// having every later test refused. The cap itself, its eviction rule and the +// restore trim get their own process in registry-cap.test.js, which pins tiny +// values and drives them over the wire. +process.env.AGENTS_MAX = "1000"; process.env.STATE_FILE = path.join( os.tmpdir(), `turma-test-state-${process.pid}.json` From 3401f72a8130bf6bba300370a47ae49ef6008a4a Mon Sep 17 00:00:00 2001 From: Malcolm Habeeb Date: Wed, 12 Aug 2026 16:24:54 -0400 Subject: [PATCH 2/5] XERK-272: fix what the QA pass found in the registry cap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five defects, all from the adversarial pass on the first commit. A full registry must not read as an outage. The aggregate gate refused whichever host beat next, and rolling a known host back to its previous record rolls back `lastSeen` with it — so under sustained pressure a live, normal-sized host was refused every beat, aged past OFFLINE_AFTER_MS and showed offline while it was up, indistinguishable from a network failure and invisible to the operator (measured: 57/57 refused over 85s). Only a host OVER its share (AGENT_FAIR_SHARE = total/count, 512 KiB deployed, against a measured largest-real-record of 0.30 MiB) is refused now. The cost is a bounded overshoot (~2x the budget) instead of a hard total; the claim that a refusal was "contention, not a wall" was wrong and is gone from the docs. The restore trim could not protect a restore it never reached: readFileSync + JSON.parse materialize the whole file, so a 264 MiB state.json left by a flood killed a 256 MiB hub at init — before any log line, every boot, forever under `restart: unless-stopped`. The file is now measured with statSync first, moved to .oversized, and the hub boots empty and says so. The refusal log was one line per refused beat, so surviving a flood cost the host its disk instead: throttled to one a minute carrying the suppressed count. `device` is agent-supplied and unvalidated for content, so a newline in it forged a line reading exactly like the hub's own — every log naming a host goes through logName now. The new env knobs went through `Number(x) || default`, which silently obeys a negative and would refuse the whole fleet on its first beat; positiveEnv announces and ignores a bad value, and the derived budget is printed at boot. Also: eviction now spends no record on a request it could not satisfy anyway, and releases the evicted host's recordSizeWarned entry. Filed XERK-292 for the one finding out of scope here: `historyResults` and the five other on-demand caches are capped by COUNT and never by bytes, so ONE device name and no concurrency at all still OOM-kills the hub (4 sequential 30 MiB beats). Pre-existing, kills origin/main identically. Tests: registry-cap.test.js (share exemption, throttle, log forging, env knobs), registry-restore.test.js (the oversized state.json, own process). --- .claude/rules/turma.md | 44 +++++-- turma/server.js | 170 +++++++++++++++++++++++---- turma/tests/registry-cap.test.js | 133 ++++++++++++++++++--- turma/tests/registry-restore.test.js | 80 +++++++++++++ 4 files changed, 377 insertions(+), 50 deletions(-) create mode 100644 turma/tests/registry-restore.test.js diff --git a/.claude/rules/turma.md b/.claude/rules/turma.md index 56215ddf..ea30c79d 100644 --- a/.claude/rules/turma.md +++ b/.claude/rules/turma.md @@ -265,26 +265,44 @@ working-status bar, ready-for-review, ended sessions, the composer and the termi - Two budgets, bounding different things: **`AGENTS_TOTAL_MAX`** (aggregate record bytes, defaulting to an eighth of the container's own cgroup limit, clamped 8–64 MiB) and **`AGENTS_MAX`** (record count, default 64). The count cap is not redundant — the byte budget measures what - `agentRecordSize` measures, which EXCLUDES the on-demand caches, so only a cap on hosts bounds - their multiple. A ceiling above the limit the kernel kills on is not a ceiling: size it from the - container, never pick a number. + `agentRecordSize` measures, which EXCLUDES the on-demand caches, so it bounds their MULTIPLE and + nothing bounds their SIZE. A ceiling above the limit the kernel kills on is not a ceiling: size it + from the container, never pick a number. - **A newcomer never displaces a host that is still around.** Past the cap a record is reclaimed only if it has been unseen for `AGENT_EVICT_IDLE_MS` (1h, ≫ `OFFLINE_AFTER_MS`) — a record holds an offline host's last known sessions, PR chips and usage, so a host rebooting or updating keeps - its slot and the new `device` gets a 429 instead. The accepted cost is that a flood of names can - squat slots and block onboarding a genuinely new host until it stops or `AGENTS_MAX` is raised; - per-agent tokens are the real fix and are not this. + its slot and the new `device` gets a 429 instead. Nothing is evicted when eviction could not + satisfy the request anyway. The accepted cost is that a flood of names can squat slots and block + onboarding a genuinely new host until it stops or `AGENTS_MAX` is raised; per-agent tokens are the + real fix and are not this. - A host **already in the registry is always admitted** — turning the cap into a wall for the fleet's - own hosts is the same outage from the other side. The aggregate can still refuse its beat, which - is contention (it lands on a later beat), not a wall, and never damages the record being served. -- **The state.json restore enforces the same budget** (`trimRestoredAgents`, keep-newest): a flood - that landed before a restart is on disk, and a bound the loading path skips is not a bound. Same - reason `normalizeRecord` runs on the restore as well as the ingest. + own hosts is the same outage from the other side. +- **The aggregate refuses only a host OVER its share** (`AGENT_FAIR_SHARE` = total/count, floored at + 64 KiB; 512 KiB deployed, against a measured largest-real-record of 0.30 MiB). Refusing a KNOWN + host rolls it back to its previous record — `lastSeen` included — so a host refused every beat + ages past `OFFLINE_AFTER_MS` and **reads offline while it is up**, indistinguishable from a + network failure and invisible to the operator. A host inside its share is not why the registry is + full, so it never pays; the refusal lands on the host the operator needs named. The cost is a + bounded overshoot (~2× the budget worst case) rather than a hard total. +- **The state.json restore enforces the same budget** (`trimRestoredAgents`, keep-newest), and the + file is **measured with `statSync` before it is opened** (`STATE_FILE_MAX`, container/4): the trim + cannot protect a restore it never reaches, and `readFileSync` + `JSON.parse` of a flooded file + killed the hub at init with no log line, every boot, forever. An oversized file is moved to + `.oversized` and the hub boots empty — losing that cache is documented as harmless; not booting + is not. +- **Every log line naming a host goes through `logName`** — `device` is agent-supplied and validated + only for length and prototype keys, so a newline in it forged a line reading exactly like the + hub's own. Refusal logs are throttled to one a minute with the suppressed count, because the flood + the cap exists to survive is precisely the traffic that writes them. - Byte accounting is a side map (`recordBytes`), never a field on the record — anything on a record is served to every client — and `registryBytes()` re-measures unknown keys and forgets dead ones, so the many places that `delete agents[key]` need not remember it. -- Tests: `registry-cap.test.js`, which pins tiny caps in its own process (`server.test.js` lifts - `AGENTS_MAX` because ~100 synthetic host names is not a fleet). +- New env knobs go through `positiveEnv`: a silently-obeyed negative cap refuses the whole fleet on + its first beat, so a bad value is announced and ignored. The effective budget is printed at boot + because it is DERIVED, not configured. +- Tests: `registry-cap.test.js` and `registry-restore.test.js`, each pinning tiny caps in its own + process (`server.test.js` lifts `AGENTS_MAX` because ~100 synthetic host names is not a fleet, so + the cap's interaction with other routes is covered only in those two files). - The hub also serves the `glasses/` client: a CORS'd `/api/*` surface for that cross-origin WebView; per-session `input`/`history` endpoints; `GET /api/ws-token` for short-lived WebSocket auth; an `/audio` STT WebSocket (G2-mic PCM to the LiteLLM instance's transcription endpoint); and diff --git a/turma/server.js b/turma/server.js index 77a3de8a..4e90e460 100644 --- a/turma/server.js +++ b/turma/server.js @@ -373,19 +373,51 @@ let agents = {}; // rather than picked. // * AGENTS_MAX — the record COUNT. The byte budget measures what // `agentRecordSize` measures, which deliberately EXCLUDES the on-demand -// caches (history, subagent history, Jira issues, create results); those are -// bounded per host by count, so only a cap on hosts bounds their multiple. +// caches (history, subagent history, Jira issues, create results). Those are +// capped per host by COUNT, not by bytes, so this bounds their multiple and +// nothing here bounds their size — one host can still park a lot in them. +// +// The aggregate is what a NEWCOMER is admitted against, and what an OVER-SHARE +// host is refused against — never a host beating a normal-sized record, whose +// refusal would be indistinguishable from an outage. See AGENT_FAIR_SHARE. // // Both are env-overridable: an operator who grows the fleet past the cap should // raise it in compose, not discover the hub silently dropping a host. -const AGENTS_MAX = Number(process.env.AGENTS_MAX) || 64; + +/** + * A positive-integer env knob, or the default. A knob that silently accepts a + * negative refuses the WHOLE fleet on its first beat with nothing but a + * per-beat 429 to explain it, so a bad value is announced and ignored rather + * than obeyed. + */ +function positiveEnv(name, fallback) { + const raw = process.env[name]; + if (raw === undefined || raw === "") return fallback; + const n = Number(raw); + if (Number.isFinite(n) && n > 0) return Math.floor(n); + console.warn(`WARNING: ${name}=${JSON.stringify(raw)} is not a positive number — using ${fallback}`); + return fallback; +} + +// A host name is agent-supplied and reaches the hub's log; it is validated for +// length and prototype keys but NOT for content, so a newline in it forges a +// log line that reads exactly like one of the hub's own. Every log that names a +// host goes through this. +function logName(key) { + // JSON.stringify does the line-forging half (a newline becomes the two + // characters \ and n, so it can never start a line); the sweep after it + // covers the control characters JSON leaves intact, which a terminal acts on. + return JSON.stringify(String(key).slice(0, 200)).replace(/[\u0000-\u001f\u007f]/g, "?"); +} + +const AGENTS_MAX = positiveEnv("AGENTS_MAX", 64); // How long a record must have gone unseen before the registry may reclaim its // slot for a host it has never met. Deliberately far longer than // OFFLINE_AFTER_MS: a record holds an offline host's last known sessions, PR // chips and usage, so a host rebooting, updating, or off for the afternoon must // never be displaced by a newcomer — a flood is refused instead. Only a host // that has been gone long enough to be uninteresting is evictable. -const AGENT_EVICT_IDLE_MS = Number(process.env.AGENT_EVICT_IDLE_MS) || 60 * 60 * 1000; +const AGENT_EVICT_IDLE_MS = positiveEnv("AGENT_EVICT_IDLE_MS", 60 * 60 * 1000); // The container's memory limit in bytes, or null when it cannot be read (not // containerised, no cgroupfs, or explicitly unlimited). cgroup v2 first, then @@ -414,7 +446,40 @@ function defaultRegistryBudget() { if (!limit) return 32 << 20; return Math.max(8 << 20, Math.min(64 << 20, Math.floor(limit / 8))); } -const AGENTS_TOTAL_MAX = Number(process.env.AGENTS_TOTAL_MAX) || defaultRegistryBudget(); +const AGENTS_TOTAL_MAX = positiveEnv("AGENTS_TOTAL_MAX", defaultRegistryBudget()); + +// One host's share of the aggregate — 512 KiB at the deployed sizing, against a +// measured real fleet whose LARGEST record is 0.30 MiB. +// +// This is what keeps a full registry from reading as an outage. The aggregate +// gate refuses whoever happens to beat next, and rolling a known host back to +// its previous record rolls back its `lastSeen` too — so under sustained +// pressure a live host is refused on every beat and crosses OFFLINE_AFTER_MS +// while it is up, silently, with no way for the operator to tell that from a +// network failure. A host inside its share is never the reason the registry is +// full, so it is never the one that pays: the refusal lands on the host whose +// record is over its share, which is also the host the operator needs named. +// +// The cost is a bounded overshoot rather than a hard total: fat records are +// still admitted only while the aggregate has slack, and every host inside its +// share sums to at most AGENTS_TOTAL_MAX again — so worst-case retained is +// ~2x the budget (a quarter of the container at the deployed sizing), not the +// unbounded growth this replaces. +const AGENT_FAIR_SHARE = Math.max(64 << 10, Math.floor(AGENTS_TOTAL_MAX / AGENTS_MAX)); + +// The most state.json may be before the restore refuses to open it at all. Sized +// like the registry budget and generously above it, because the saved blob +// carries the on-demand caches the budget deliberately excludes; the point is +// to catch a file that cannot fit in the container's memory, not to second-guess +// a legitimate one. See the restore for why measuring beats parsing. +const STATE_FILE_MAX = positiveEnv( + "STATE_FILE_MAX", + (() => { + const limit = containerMemoryLimit(); + if (!limit) return 64 << 20; + return Math.max(32 << 20, Math.min(128 << 20, Math.floor(limit / 4))); + })() +); // The cache keys `serializeAgent` strips; see AGENT_RECORD_MAX. // @@ -462,6 +527,25 @@ function registryBytes() { return total; } +// A refusal is one line per window, not one per beat: a flood is exactly the +// traffic that triggers it, so an unthrottled log turns a survived attack into +// disk pressure. The suppressed count rides the next line so nothing is hidden. +const REFUSED_LOG_EVERY_MS = 60 * 1000; +let refusedLogAt = 0; +let refusedSinceLog = 0; +function logRegistryFull(detail) { + refusedSinceLog += 1; + const now = Date.now(); + if (now - refusedLogAt < REFUSED_LOG_EVERY_MS) return; + const also = refusedSinceLog > 1 ? ` (+${refusedSinceLog - 1} more refused since the last line)` : ""; + refusedLogAt = now; + refusedSinceLog = 0; + console.error( + `registry is full (${Object.keys(agents).length}/${AGENTS_MAX} hosts, ` + + `${registryBytes()}/${AGENTS_TOTAL_MAX} bytes): ${detail}${also}` + ); +} + // Hosts whose slot may be reclaimed for a newcomer, least-recently-seen first. function evictableAgents(now) { return Object.entries(agents) @@ -494,11 +578,14 @@ function makeRegistryRoom(addBytes, addSlots) { bytes -= recordBytes.get(key) || 0; count -= 1; console.warn( - `registry at its limit — evicting ${key}, unseen for ` + + `registry at its limit — evicting ${logName(key)}, unseen for ` + `${Math.round((now - (agents[key].lastSeen || 0)) / 60000)}m` ); delete agents[key]; recordBytes.delete(key); + // Everything else keyed by host name has to let go too, or a registry that + // admits and evicts forever leaks a per-host entry per name it ever saw. + recordSizeWarned.delete(key); invalidateAgentsCache(); sseBroadcast("removed", { key }); } @@ -557,6 +644,23 @@ const liveClients = {}; // ---- persistence (best-effort: survives hub restarts so the UI isn't blank // for the first heartbeat interval; losing it is harmless) ------------------- try { + // The trim below cannot protect a restore it never reaches: `readFileSync` + + // `JSON.parse` materialize the WHOLE file first, so a 264 MiB state.json left + // by a flood kills a 256 MiB hub at init — before a single log line, on every + // boot, which `restart: unless-stopped` turns into a permanent crash loop + // with nothing to read. So the file is measured before it is opened + // (XERK-272). Losing the cache is documented as harmless; not booting is not, + // and the file is preserved beside itself rather than deleted so the flood + // can still be examined. + const stateSize = fs.statSync(STATE_FILE).size; + if (stateSize > STATE_FILE_MAX) { + const aside = `${STATE_FILE}.oversized`; + try { fs.renameSync(STATE_FILE, aside); } catch { /* read-only volume */ } + throw new Error( + `state file is ${stateSize} bytes, over the ${STATE_FILE_MAX} limit — ` + + `starting with an empty registry; the file is kept at ${aside}` + ); + } agents = JSON.parse(fs.readFileSync(STATE_FILE, "utf8")); // Records written before a coercion existed — and any host that is OFFLINE, // so no beat will ever rewrite its record — carry whatever shape was current @@ -571,8 +675,15 @@ try { // request (XERK-272). trimRestoredAgents(); console.log(`loaded ${Object.keys(agents).length} agents from ${STATE_FILE}`); -} catch { - /* first boot or no volume mounted */ +} catch (e) { + // A missing file is first boot / no volume mounted and says nothing worth + // saying. Anything else is a state file the hub HAS and could not use — an + // oversized one, or corrupt JSON — which the operator has to be told about, + // and which leaves a partially-built registry behind unless it is cleared. + if (!e || e.code !== "ENOENT") { + agents = {}; + console.error(`state restore skipped: ${(e && e.message) || e}`); + } } // The state blob, or null when it cannot be produced. JSON.stringify throws // RangeError once the aggregate passes V8's ~512 MiB string ceiling, and it runs @@ -4033,11 +4144,11 @@ const server = http.createServer(async (req, res) => { // attacker-chosen, so an unbounded number of names is an unbounded number // of retained records. if (!known && !makeRegistryRoom(0, 1)) { - console.error( - `heartbeat from ${key}: registry is full ` + - `(${Object.keys(agents).length}/${AGENTS_MAX} hosts, ` + - `${registryBytes()}/${AGENTS_TOTAL_MAX} bytes) — new host refused` - ); + // Throttled like the over-half warning, and for the same reason: the + // flood this exists to survive is exactly the traffic that would write + // this line, so a per-beat log turns a refused attack into disk + // pressure on the host. Once per REFUSED_LOG_EVERY_MS, with the count. + logRegistryFull(`new host ${logName(key)} refused`); return json(res, 429, { error: "agent registry full", limit: AGENTS_MAX, bytes: AGENTS_TOTAL_MAX, }); @@ -4178,11 +4289,13 @@ const server = http.createServer(async (req, res) => { // 256 MiB hub. Measured after the coercion, because the coerced record is // the one that gets retained, served and saved. recordBytes.set(key, recordSize); - if (!makeRegistryRoom(0, 0)) { - // This host is never what gets evicted — it was just seen, so the idle - // rule excludes it. And refusing here is contention, not a wall: the - // host beats again in ~20s and lands the moment there is room, rather - // than being stuck at its old record forever. + if (!makeRegistryRoom(0, 0) && recordSize > AGENT_FAIR_SHARE) { + // Only a host OVER its share is refused here. A host inside it is not + // the reason the registry is full, and refusing it would roll back its + // `lastSeen` with its record — reading as offline while it is up, every + // beat, with nothing to distinguish that from a network failure. See + // AGENT_FAIR_SHARE. This host is never what gets EVICTED either: it was + // just seen, so the idle rule excludes it. if (prev && Object.keys(prev).length) { agents[key] = prev; recordBytes.set(key, agentRecordSize(prev)); @@ -4190,11 +4303,13 @@ const server = http.createServer(async (req, res) => { delete agents[key]; recordBytes.delete(key); } - console.error( - `heartbeat from ${key}: record is ${recordSize} bytes and the registry ` + - `is at its ${AGENTS_TOTAL_MAX}-byte budget — beat refused` + logRegistryFull( + `${logName(key)} refused — its record is ${recordSize} bytes, over the ` + + `${AGENT_FAIR_SHARE}-byte per-host share` ); - return json(res, 429, { error: "agent registry full", bytes: AGENTS_TOTAL_MAX }); + return json(res, 429, { + error: "agent registry full", bytes: AGENTS_TOTAL_MAX, share: AGENT_FAIR_SHARE, + }); } ingestHistory(next, historyResults); ingestSubagentHistory(next, subagentHistoryResults); @@ -5945,7 +6060,8 @@ if (process.env.TURMA_TEST) { // XERK-272 registry bounds. Exported for the same reason as the group above: // the per-record ceiling stayed green while an unbounded NUMBER of records // OOM-killed the hub, so the aggregate has to be pinned by name too. - AGENTS_MAX, AGENTS_TOTAL_MAX, AGENT_EVICT_IDLE_MS, + AGENTS_MAX, AGENTS_TOTAL_MAX, AGENT_EVICT_IDLE_MS, AGENT_FAIR_SHARE, + STATE_FILE_MAX, positiveEnv, logName, registryBytes, makeRegistryRoom, trimRestoredAgents, containerMemoryLimit, defaultRegistryBudget, recordBytes, // Ingest coercion, exported for the same reason as the rest of this group: @@ -6025,6 +6141,14 @@ if (process.env.TURMA_TEST) { if (!TURMA_TRIGGER_TOKEN) console.warn("WARNING: TURMA_TRIGGER_TOKEN not set — POST /api/trigger accepts only the user login (no dedicated token)"); server.listen(PORT, () => { console.log(`turma listening on :${PORT}`); + // The effective registry budget, printed because it is DERIVED (from this + // container's own cgroup limit) rather than configured — without this the + // only way to learn what the hub is enforcing is to be refused by it. + console.log( + `agent registry: <=${AGENTS_MAX} hosts, <=${AGENTS_TOTAL_MAX} bytes ` + + `(${AGENT_FAIR_SHARE}/host), container limit ` + + `${containerMemoryLimit() ?? "unknown"}` + ); if (push.fcmEnabled()) console.log("FCM push alerts -> Android devices"); // A warning, not an info line: a hub running without FCM delivers ZERO mobile // notifications (every notify() is a no-op), and that has silently bitten us diff --git a/turma/tests/registry-cap.test.js b/turma/tests/registry-cap.test.js index b422587a..858d2579 100644 --- a/turma/tests/registry-cap.test.js +++ b/turma/tests/registry-cap.test.js @@ -27,7 +27,10 @@ process.env.TURMA_USER = "hubuser"; process.env.TURMA_PASSWORD = "hubpass"; process.env.TURMA_AGENT_TOKEN = "agenttok"; process.env.AGENTS_MAX = "4"; -process.env.AGENTS_TOTAL_MAX = String(64 << 10); +// Chosen so the derived per-host share (total / max = 160 KiB) sits clear of +// BOTH the 64 KiB floor and the record sizes below — the whole point of the +// share is that a small host and a fat one land on opposite sides of it. +process.env.AGENTS_TOTAL_MAX = String(640 << 10); process.env.AGENT_EVICT_IDLE_MS = String(60 * 1000); const tmp = (name) => path.join(os.tmpdir(), `turma-regcap-${name}-${process.pid}.json`); @@ -67,8 +70,8 @@ fs.writeFileSync( const hub = require("../server.js"); const { server, agents, recordBytes, - AGENTS_MAX, AGENTS_TOTAL_MAX, AGENT_EVICT_IDLE_MS, - registryBytes, makeRegistryRoom, agentRecordSize, + AGENTS_MAX, AGENTS_TOTAL_MAX, AGENT_EVICT_IDLE_MS, AGENT_FAIR_SHARE, STATE_FILE_MAX, + registryBytes, makeRegistryRoom, agentRecordSize, positiveEnv, logName, containerMemoryLimit, defaultRegistryBudget, } = hub; @@ -206,27 +209,56 @@ test("http: a host just short of the idle window is NOT evictable", async () => test("http: records that fit the per-record ceiling still can't sum past the aggregate", async () => { resetRegistry(); // AGENTS_MAX records at AGENT_RECORD_MAX is 512 MiB on a 256 MiB hub, so the - // per-record ceiling was never the bound. Both of these pass it easily. - assert.equal((await beat(chunky("fat-a", 40))).status, 200); - const refused = await beat(chunky("fat-b", 40)); + // per-record ceiling was never the bound. All three pass it easily. + assert.ok(300 << 10 > AGENT_FAIR_SHARE, "the rig's fat record must be over-share"); + assert.equal((await beat(chunky("fat-a", 300))).status, 200); + assert.equal((await beat(chunky("fat-b", 300))).status, 200); + const refused = await beat(chunky("fat-c", 300)); assert.equal(refused.status, 429); assert.equal(refused.body.error, "agent registry full"); assert.equal(refused.body.bytes, AGENTS_TOTAL_MAX); - assert.equal("fat-b" in agents, false); - assert.ok(registryBytes() <= AGENTS_TOTAL_MAX); + assert.equal(refused.body.share, AGENT_FAIR_SHARE); + assert.equal("fat-c" in agents, false); }); -test("http: an aggregate refusal leaves the host's PREVIOUS record intact", async () => { +test("http: a host INSIDE its share is not refused for someone else's bulk", async () => { + // The regression this exists for: the aggregate gate refuses whoever beats + // next, and rolling a known host back to its previous record rolls back + // `lastSeen` too — so a live, normal-sized host was refused every beat, aged + // past OFFLINE_AFTER_MS, and read as offline while it was up, with nothing to + // tell that apart from a network failure. + resetRegistry(); + assert.equal((await beat({ device: "small", repos: [{ name: "r1" }] })).status, 200); + assert.equal((await beat(chunky("hog-a", 300))).status, 200); + assert.equal((await beat(chunky("hog-b", 300))).status, 200); + assert.ok(registryBytes() > AGENTS_TOTAL_MAX - (300 << 10), "the rig must be under pressure"); + + const before = agents["small"].lastSeen; + await new Promise((r) => setTimeout(r, 5)); + for (let i = 0; i < 3; i++) { + const ok = await beat({ device: "small", repos: [{ name: `r${i}` }] }); + assert.equal(ok.status, 200, "a host inside its share must keep beating"); + } + assert.deepEqual(agents["small"].repos, [{ name: "r2" }], "its content must be current"); + assert.ok(agents["small"].lastSeen > before, "and its liveness must advance"); + // The exemption is bounded, not a hole: every exempt host is under its share, + // so they sum to at most the budget again. + assert.ok(registryBytes() <= AGENTS_TOTAL_MAX + AGENTS_MAX * AGENT_FAIR_SHARE); +}); + +test("http: an over-share refusal leaves the host's PREVIOUS record intact", async () => { resetRegistry(); assert.equal((await beat({ device: "grower", repos: [{ name: "r1" }] })).status, 200); - assert.equal((await beat(chunky("ballast", 40))).status, 200); - // `grower` is known, so admission lets it through — the aggregate is what - // refuses it, and a refused beat must not damage what is being served. - const refused = await beat(chunky("grower", 40)); + assert.equal((await beat(chunky("ballast-a", 300))).status, 200); + assert.equal((await beat(chunky("ballast-b", 300))).status, 200); + // `grower` is known, so admission lets it through — going over its share + // while the registry is full is what refuses it, and a refused beat must not + // damage what is being served. + const refused = await beat(chunky("grower", 300)); assert.equal(refused.status, 429); assert.deepEqual(agents["grower"].repos, [{ name: "r1" }]); assert.equal(recordBytes.get("grower"), agentRecordSize(agents["grower"])); - // Not a wall: it beats again in ~20s and lands as soon as there is room. + // A normal-sized beat from the same host still lands. assert.equal((await beat({ device: "grower", repos: [{ name: "r3" }] })).status, 200); assert.deepEqual(agents["grower"].repos, [{ name: "r3" }]); }); @@ -275,6 +307,79 @@ test("makeRegistryRoom reports failure instead of over-evicting", () => { assert.equal(makeRegistryRoom(0, 1), true); }); +test("a bad env knob is announced and ignored, never obeyed", () => { + // `Number(x) || default` accepted a negative silently, and a negative cap + // refuses the WHOLE fleet on its first beat with only a per-beat 429 to + // explain it — a compose typo taking every host offline. + const prev = process.env.__REGCAP_PROBE; + const warned = []; + const realWarn = console.warn; + console.warn = (m) => warned.push(String(m)); + try { + for (const bad of ["-1", "0", "abc", "-1e9"]) { + process.env.__REGCAP_PROBE = bad; + assert.equal(positiveEnv("__REGCAP_PROBE", 64), 64, `${bad} was obeyed`); + } + process.env.__REGCAP_PROBE = "128"; + assert.equal(positiveEnv("__REGCAP_PROBE", 64), 128); + delete process.env.__REGCAP_PROBE; + assert.equal(positiveEnv("__REGCAP_PROBE", 64), 64); + } finally { + console.warn = realWarn; + if (prev === undefined) delete process.env.__REGCAP_PROBE; + else process.env.__REGCAP_PROBE = prev; + } + assert.ok(warned.some((m) => m.includes("__REGCAP_PROBE")), "a bad value must be announced"); +}); + +test("a host name cannot forge a hub log line", async () => { + // `device` is agent-supplied and validated only for length and prototype + // keys, so a newline in it wrote a line indistinguishable from the hub's own. + const forged = "evil\n2026-01-01T00:00:00Z FORGED: all clear"; + assert.equal(logName(forged).includes("\n"), false); + assert.equal(logName("hx").includes(""), false); + assert.ok(logName("plain-host").includes("plain-host")); + // And it reaches the log through a real refusal, not just the helper. + resetRegistry(); + for (let i = 0; i < AGENTS_MAX; i++) await beat({ device: `forge-${i}` }); + const lines = []; + const realErr = console.error; + console.error = (m) => lines.push(String(m)); + try { + assert.equal((await beat({ device: forged })).status, 429); + } finally { + console.error = realErr; + } + assert.equal(lines.some((l) => l.includes("\n")), false, "a refusal log carried a raw newline"); +}); + +test("the refusal log is throttled, and says how many it swallowed", async () => { + // The flood this cap exists to survive is exactly the traffic that writes + // this line — unthrottled, surviving the attack costs the host its disk. + resetRegistry(); + for (let i = 0; i < AGENTS_MAX; i++) await beat({ device: `noisy-${i}` }); + const lines = []; + const realErr = console.error; + console.error = (m) => lines.push(String(m)); + try { + for (let i = 0; i < 60; i++) { + assert.equal((await beat({ device: `flood-${i}` })).status, 429); + } + } finally { + console.error = realErr; + } + assert.ok(lines.length <= 2, `60 refusals wrote ${lines.length} log lines`); +}); + +test("the state.json ceiling is measured before the file is opened", () => { + // The restore trim cannot protect a restore it never reaches: readFileSync + + // JSON.parse materialize the whole file, so a flooded state.json killed the + // hub at init — before any log line, on every boot, forever. + assert.ok(STATE_FILE_MAX >= (32 << 20), "the ceiling must clear a legitimate state file"); + const limit = containerMemoryLimit(); + if (limit) assert.ok(STATE_FILE_MAX <= limit, "and must not exceed the container itself"); +}); + test("makeRegistryRoom spends no record on a request it cannot satisfy", () => { resetRegistry(); agents["idle"] = { device: "idle", lastSeen: Date.now() - AGENT_EVICT_IDLE_MS * 2 }; diff --git a/turma/tests/registry-restore.test.js b/turma/tests/registry-restore.test.js new file mode 100644 index 00000000..da16139c --- /dev/null +++ b/turma/tests/registry-restore.test.js @@ -0,0 +1,80 @@ +// The state.json restore's own ceiling (XERK-272). +// +// `trimRestoredAgents()` holds the RESTORED registry to the same budget as a +// live one, but it cannot protect a restore it never reaches: the restore does +// `readFileSync` + `JSON.parse` on the WHOLE file first, so a state.json left +// behind by a flood killed a 256 MiB hub at module init — before a single log +// line, on every boot, which `restart: unless-stopped` turns into a permanent +// crash loop with nothing to read and no recovery short of deleting the file by +// hand. So the file is measured before it is opened. +// +// Its own process (and its own file) because the restore runs once, at require +// time — registry-cap.test.js has already loaded the module with a good one. +// node:test, no npm. + +"use strict"; + +const os = require("os"); +const fs = require("fs"); +const path = require("path"); +const test = require("node:test"); +const assert = require("node:assert/strict"); + +process.env.TURMA_TEST = "1"; +process.env.TURMA_USER = "hubuser"; +process.env.TURMA_PASSWORD = "hubpass"; +process.env.TURMA_AGENT_TOKEN = "agenttok"; +// Wound right down so an "oversized" file is a few KiB rather than the hundreds +// of MiB it takes to reproduce the real kill. +process.env.STATE_FILE_MAX = "4096"; + +const tmp = (name) => path.join(os.tmpdir(), `turma-regrestore-${name}-${process.pid}.json`); +process.env.DEVICES_FILE = tmp("devices"); +process.env.TICKET_AGENTS_FILE = tmp("ticket-agents"); +process.env.AUTOSTART_ORGS_FILE = tmp("autostart-orgs"); +process.env.TICKET_MODELS_FILE = tmp("ticket-models"); +process.env.ORG_COLORS_FILE = tmp("org-colors"); +process.env.MIGRATE_SPOOL_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "turma-regrestore-migrations-")); +process.env.ARCHIVE_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "turma-regrestore-archive-")); +process.env.ARCHIVE_DB = path.join(process.env.ARCHIVE_DIR, "index.db"); + +process.env.STATE_FILE = tmp("state"); +const ASIDE = `${process.env.STATE_FILE}.oversized`; +try { fs.unlinkSync(ASIDE); } catch { /* first run */ } +// 40 records of ~4 KiB — comfortably over the 4096-byte ceiling above. +fs.writeFileSync( + process.env.STATE_FILE, + JSON.stringify( + Object.fromEntries( + Array.from({ length: 40 }, (_, i) => [ + `flood-${i}`, + { device: `flood-${i}`, lastSeen: Date.now() - i, sessions: [{ id: `s${i}`, label: "L".repeat(4000) }] }, + ]) + ) + ) +); +const WROTE = fs.statSync(process.env.STATE_FILE).size; + +const errors = []; +const realError = console.error; +console.error = (m) => { errors.push(String(m)); realError(m); }; +const hub = require("../server.js"); +console.error = realError; + +test("an oversized state.json does not get parsed, and the hub still boots", () => { + assert.ok(WROTE > Number(process.env.STATE_FILE_MAX)); + // Booting with an empty registry is the documented-harmless outcome (the + // state file is a best-effort cache); not booting is not. + assert.deepEqual(Object.keys(hub.agents), []); + assert.equal(typeof hub.server.listen, "function"); +}); + +test("it says so, and keeps the file rather than deleting it", () => { + // Silence here is the whole failure mode being fixed: the old behaviour was a + // crash loop that logged nothing at all. + assert.ok(errors.some((m) => m.includes("state restore skipped")), errors.join("\n")); + assert.ok(errors.some((m) => m.includes(String(hub.STATE_FILE_MAX)))); + assert.ok(fs.existsSync(ASIDE), "the oversized file must be preserved for forensics"); + assert.equal(fs.statSync(ASIDE).size, WROTE); + assert.equal(fs.existsSync(process.env.STATE_FILE), false, "and moved out of the way"); +}); From 4fc4d7e56eef6eb3bc0b180ffb3f5316ccf22cae Mon Sep 17 00:00:00 2001 From: Malcolm Habeeb Date: Wed, 12 Aug 2026 18:31:51 -0400 Subject: [PATCH 3/5] XERK-272: pin the share exemption's overshoot bound MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The exemption admits an at-or-under-share host regardless of the budget, so the aggregate is a soft total. Two things bound it: a fat beat lands only while the whole registry fits, and a NEW device is admitted only while the registry is inside the budget — so the flood path cannot reach the exemption at all, and only an already-seated host can overshoot, by at most its share. Worst case is 2x the budget, held by a test that actually drives it. --- turma/tests/registry-cap.test.js | 38 ++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/turma/tests/registry-cap.test.js b/turma/tests/registry-cap.test.js index 858d2579..c2f3c1b9 100644 --- a/turma/tests/registry-cap.test.js +++ b/turma/tests/registry-cap.test.js @@ -246,6 +246,44 @@ test("http: a host INSIDE its share is not refused for someone else's bulk", asy assert.ok(registryBytes() <= AGENTS_TOTAL_MAX + AGENTS_MAX * AGENT_FAIR_SHARE); }); +test("http: the share exemption's overshoot is bounded at 2x the budget", async () => { + // The exemption admits an at-or-under-share host REGARDLESS of the budget, so + // the aggregate is a soft total. Drive the worst case and hold the bound: + // * a fat beat is accepted only while the whole registry fits, so the fat + // records sum to at most the budget; + // * every exempt host is inside its share and there are at most AGENTS_MAX + // of them, so they sum to at most AGENTS_MAX x share = the budget again. + // Hence 2x, and never the container. + resetRegistry(); + // Only a GROWING host can overshoot. A new device is admitted only while the + // registry is inside the budget (`makeRegistryRoom(0, 1)`), so the flood path + // cannot reach past it — the exemption is reachable only by hosts that were + // already seated. Seat the worst case first: one host holding most of the + // budget, and the rest of the slots. + assert.equal((await beat(chunky("fat-1", 560))).status, 200); + for (let i = 0; i < AGENTS_MAX - 1; i++) { + assert.equal((await beat({ device: `snug-${i}` })).status, 200); + } + assert.equal(Object.keys(agents).length, AGENTS_MAX); + // A further fat beat is capped by the budget, which is what bounds the fat + // half at one budget's worth. + assert.equal((await beat(chunky("fat-1", 900))).status, 429); + + // Now grow every seated host to just inside its share. Each is exempt, so all + // of them land — this is the overshoot, and it is the whole of it. + const snugKiB = Math.floor((AGENT_FAIR_SHARE - 4096) / 1024); + for (let i = 0; i < AGENTS_MAX - 1; i++) { + assert.equal((await beat(chunky(`snug-${i}`, snugKiB))).status, 200, `snug-${i}`); + } + assert.ok(registryBytes() > AGENTS_TOTAL_MAX, "the rig must actually be overshooting"); + assert.ok( + registryBytes() <= 2 * AGENTS_TOTAL_MAX, + `overshoot ${registryBytes()} exceeded 2x the ${AGENTS_TOTAL_MAX} budget` + ); + // And no new host can be seated while it overshoots, so it cannot compound. + assert.equal((await beat({ device: "latecomer" })).status, 429); +}); + test("http: an over-share refusal leaves the host's PREVIOUS record intact", async () => { resetRegistry(); assert.equal((await beat({ device: "grower", repos: [{ name: "r1" }] })).status, 200); From 7df911bdcbef44518b506c6d5f5a91e87e48c989 Mon Sep 17 00:00:00 2001 From: Malcolm Habeeb Date: Wed, 12 Aug 2026 19:01:00 -0400 Subject: [PATCH 4/5] XERK-272: fix the fair share's floor, which broke the bound it exists for MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit QA pass 2 cleared D2-D6 and found one blocker plus two smaller defects. F1 (blocker). The overshoot bound is an identity — worst-case retained is AGENTS_TOTAL_MAX + AGENTS_MAX * AGENT_FAIR_SHARE — so a FLOOR under the share makes the second term unbounded in AGENTS_MAX. And raising AGENTS_MAX is exactly what the comment and the rules file tell an operator with a growing fleet to do. At AGENTS_MAX=2000 against the deployed 32 MiB budget the 64 KiB floor gave 3.9x the budget and the hub was OOM-killed at -m 256m, exit 137, /healthz dead. The share is now derived and never floored, so the bound is 2x at any AGENTS_MAX; a share too small to be sane warns at load instead, since the config — not the bound — is what is wrong there. The warning moved to module scope so it is reachable without binding a port. F2. On a read-only /data the rename fails and the message still sent the operator to a .oversized that was never created. It now says which happened. D4 was only half fixed: logName() went on the two new 429 lines while refuseOversized (ONE request, no registry pressure) and sanitizeHeartbeat's drop line (rides a 200 beat) stayed injectable — both easier to reach than the path that was fixed. All four heartbeat log sites now go through it. F3 was a coverage gap, and QA's mutation run proved it: the fair-share floor, the recordSizeWarned eviction sweep, and the restore's agents={} clear could all be deleted with the suite still green. All three are now caught — the floor needs the degenerate config, so registry-restore.test.js carries AGENTS_MAX=2000, and the half-built restore needs a file that PARSES and then throws, so it boots a child process on a poisoned state.json. Filed, not fixed here: XERK-298 (no heartbeat refusal — 413 or 429 — reaches the operator; the host just freezes or vanishes, and closing it spans web + Android), XERK-299 (hub-agent post() clears a delivered beat's staged work inside the transport try, so a late exception loses it and discards the reply; latent). --- .claude/rules/turma.md | 23 ++++++-- turma/server.js | 49 ++++++++++++---- turma/tests/registry-cap.test.js | 83 ++++++++++++++++++++++++---- turma/tests/registry-restore.test.js | 65 ++++++++++++++++++++++ 4 files changed, 194 insertions(+), 26 deletions(-) diff --git a/.claude/rules/turma.md b/.claude/rules/turma.md index 5d5001ed..5651606c 100644 --- a/.claude/rules/turma.md +++ b/.claude/rules/turma.md @@ -314,8 +314,18 @@ working-status bar, ready-for-review, ended sessions, the composer and the termi host rolls it back to its previous record — `lastSeen` included — so a host refused every beat ages past `OFFLINE_AFTER_MS` and **reads offline while it is up**, indistinguishable from a network failure and invisible to the operator. A host inside its share is not why the registry is - full, so it never pays; the refusal lands on the host the operator needs named. The cost is a - bounded overshoot (~2× the budget worst case) rather than a hard total. + full, so it never pays; the refusal lands on the host the operator needs named. An OVER-share host + is still refused silently — it freezes and ages to offline, or (if new) never appears at all, with + only the throttled log to say why. That is the accepted cost, and the headroom is 1.7× the largest + measured real record. +- **The cost of the exemption is a bounded overshoot, and the bound is an identity**: worst-case + retained is `AGENTS_TOTAL_MAX + AGENTS_MAX × AGENT_FAIR_SHARE`. **So the share is DERIVED and + never floored** — a floor makes the second term unbounded in `AGENTS_MAX`, and raising + `AGENTS_MAX` is exactly what an operator with a growing fleet is told to do (at 2000 hosts a + 64 KiB floor was 3.9× the budget and OOM-killed the hub). **Raising the count means raising the + budget with it**; a share under `AGENT_SHARE_SANE_MIN` warns at load rather than letting the two + contradict silently. The flood path cannot reach the exemption at all — a new device is admitted + only while the registry is inside the budget, so only an already-seated host can overshoot. - **The state.json restore enforces the same budget** (`trimRestoredAgents`, keep-newest), and the file is **measured with `statSync` before it is opened** (`STATE_FILE_MAX`, container/4): the trim cannot protect a restore it never reaches, and `readFileSync` + `JSON.parse` of a flooded file @@ -332,9 +342,12 @@ working-status bar, ready-for-review, ended sessions, the composer and the termi - New env knobs go through `positiveEnv`: a silently-obeyed negative cap refuses the whole fleet on its first beat, so a bad value is announced and ignored. The effective budget is printed at boot because it is DERIVED, not configured. -- Tests: `registry-cap.test.js` and `registry-restore.test.js`, each pinning tiny caps in its own - process (`server.test.js` lifts `AGENTS_MAX` because ~100 synthetic host names is not a fleet, so - the cap's interaction with other routes is covered only in those two files). +- Tests: `registry-cap.test.js` (small caps) and `registry-restore.test.js` (the restore, plus the + DEGENERATE `AGENTS_MAX=2000` config — the overshoot bound only breaks when the derived share falls + below what a floor would impose, which the small-cap rig never does). Each needs its own process + because the caps are read at require time; `server.test.js` lifts `AGENTS_MAX` because ~100 + synthetic host names is not a fleet, so the cap's interaction with other routes lives only in + those two files. - The hub also serves the `glasses/` client: a CORS'd `/api/*` surface for that cross-origin WebView; per-session `input`/`history` endpoints; `GET /api/ws-token` for short-lived WebSocket auth; an `/audio` STT WebSocket (G2-mic PCM to the LiteLLM instance's transcription endpoint); and diff --git a/turma/server.js b/turma/server.js index d819b63a..c43c8091 100644 --- a/turma/server.js +++ b/turma/server.js @@ -543,9 +543,33 @@ const AGENTS_TOTAL_MAX = positiveEnv("AGENTS_TOTAL_MAX", defaultRegistryBudget() // The cost is a bounded overshoot rather than a hard total: fat records are // still admitted only while the aggregate has slack, and every host inside its // share sums to at most AGENTS_TOTAL_MAX again — so worst-case retained is -// ~2x the budget (a quarter of the container at the deployed sizing), not the -// unbounded growth this replaces. -const AGENT_FAIR_SHARE = Math.max(64 << 10, Math.floor(AGENTS_TOTAL_MAX / AGENTS_MAX)); +// exactly AGENTS_TOTAL_MAX + AGENTS_MAX * AGENT_FAIR_SHARE, i.e. 2x the budget +// (a quarter of the container at the deployed sizing), not the unbounded growth +// this replaces. +// +// **That identity is the whole bound, so the share is DERIVED and never +// floored.** A floor under it (there was a 64 KiB one) makes the second term +// `AGENTS_MAX * 64 KiB`, which is unbounded in AGENTS_MAX — and raising +// AGENTS_MAX is exactly what an operator with a growing fleet is told to do. +// At AGENTS_MAX=2000 against the deployed 32 MiB budget that is 3.9x, and the +// hub is OOM-killed. Raising the count means raising the budget with it; the +// boot check below says so rather than letting the two contradict silently. +const AGENT_FAIR_SHARE = Math.max(1, Math.floor(AGENTS_TOTAL_MAX / AGENTS_MAX)); +// A share this small means the count cap and the byte budget disagree about how +// big a fleet this hub is for: hosts would be refused on size long before the +// slots ran out. The memory bound still holds — it is the CONFIGURATION that is +// wrong, so this warns rather than adjusting either number, and it warns HERE +// (module scope) rather than in the listen banner, so it is reachable to a test +// and to anything that loads the module without binding a port. +const AGENT_SHARE_SANE_MIN = 64 << 10; +if (AGENT_FAIR_SHARE < AGENT_SHARE_SANE_MIN) { + console.warn( + `WARNING: AGENTS_MAX=${AGENTS_MAX} leaves each host only ${AGENT_FAIR_SHARE} ` + + `bytes of the ${AGENTS_TOTAL_MAX}-byte registry budget — hosts will be ` + + `refused on record size long before the slots run out. Raise ` + + `AGENTS_TOTAL_MAX alongside AGENTS_MAX.` + ); +} // The most state.json may be before the restore refuses to open it at all. Sized // like the registry budget and generously above it, because the saved blob @@ -741,11 +765,16 @@ try { // can still be examined. const stateSize = fs.statSync(STATE_FILE).size; if (stateSize > STATE_FILE_MAX) { + // The rename can fail (a read-only /data), and the message has to say what + // actually happened: an operator sent to a `.oversized` that was never + // created finds nothing and concludes the hub ate their state. const aside = `${STATE_FILE}.oversized`; - try { fs.renameSync(STATE_FILE, aside); } catch { /* read-only volume */ } + let movedTo = null; + try { fs.renameSync(STATE_FILE, aside); movedTo = aside; } catch { /* read-only volume */ } throw new Error( `state file is ${stateSize} bytes, over the ${STATE_FILE_MAX} limit — ` + - `starting with an empty registry; the file is kept at ${aside}` + `starting with an empty registry; the file is ` + + (movedTo ? `kept at ${movedTo}` : `left in place at ${STATE_FILE} (could not move it)`) ); } agents = JSON.parse(fs.readFileSync(STATE_FILE, "utf8")); @@ -2363,7 +2392,7 @@ function sanitizeHeartbeat(payload, key) { } if (size > HEARTBEAT_UNKNOWN_MAX) { console.error( - `heartbeat from ${key}: dropped unknown field ${JSON.stringify(k)} ` + + `heartbeat from ${logName(key)}: dropped unknown field ${JSON.stringify(k)} ` + `(${size} bytes, limit ${HEARTBEAT_UNKNOWN_MAX})` ); delete payload[k]; @@ -4582,7 +4611,7 @@ const server = http.createServer(async (req, res) => { if (prev && Object.keys(prev).length) agents[key] = prev; else delete agents[key]; console.error( - `heartbeat from ${key}: record is ${size} bytes, over the ` + + `heartbeat from ${logName(key)}: record is ${size} bytes, over the ` + `${AGENT_RECORD_MAX} limit — beat refused` ); return json(res, 413, { error: "agent record too large", limit: AGENT_RECORD_MAX }); @@ -4607,7 +4636,7 @@ const server = http.createServer(async (req, res) => { try { recordCoercion.normalize(next); } catch (e) { - console.error(`heartbeat from ${key}: coercion failed (${e.message}) — beat refused`); + console.error(`heartbeat from ${logName(key)}: coercion failed (${e.message}) — beat refused`); if (prev && Object.keys(prev).length) agents[key] = prev; else delete agents[key]; return json(res, 400, { error: "malformed heartbeat" }); @@ -4622,7 +4651,7 @@ const server = http.createServer(async (req, res) => { const overHalf = recordSize > AGENT_RECORD_MAX / 2 && recordSize <= AGENT_RECORD_MAX; if (overHalf && !recordSizeWarned.get(key)) { console.warn( - `heartbeat from ${key}: record is ${recordSize} bytes, over half the ` + + `heartbeat from ${logName(key)}: record is ${recordSize} bytes, over half the ` + `${AGENT_RECORD_MAX} limit` ); } @@ -6425,7 +6454,7 @@ if (process.env.TURMA_TEST) { // the per-record ceiling stayed green while an unbounded NUMBER of records // OOM-killed the hub, so the aggregate has to be pinned by name too. AGENTS_MAX, AGENTS_TOTAL_MAX, AGENT_EVICT_IDLE_MS, AGENT_FAIR_SHARE, - STATE_FILE_MAX, positiveEnv, logName, + STATE_FILE_MAX, positiveEnv, logName, recordSizeWarned, registryBytes, makeRegistryRoom, trimRestoredAgents, containerMemoryLimit, defaultRegistryBudget, recordBytes, // Ingest coercion, exported for the same reason as the rest of this group: diff --git a/turma/tests/registry-cap.test.js b/turma/tests/registry-cap.test.js index c2f3c1b9..9a804c60 100644 --- a/turma/tests/registry-cap.test.js +++ b/turma/tests/registry-cap.test.js @@ -1,15 +1,17 @@ // Unit tests for the agent registry's own ceiling (XERK-272). // -// `device` is attacker-chosen — every agent shares one TURMA_AGENT_TOKEN — so -// the number of RETAINED records was unbounded: 512 beats of 0.9 MiB under 512 -// names OOM-killed a 256 MiB hub, while the same 512 beats under ONE name -// peaked at 169 MiB. `AGENT_RECORD_MAX` bounds one record and `prune()` only -// reclaims at seven days, so neither is the aggregate. +// Nothing capped how many DISTINCT `device` names the registry could retain: +// 512 beats of 0.9 MiB under 512 names OOM-killed a 256 MiB hub, while the same +// 512 beats under ONE name peaked at 169 MiB. `AGENT_RECORD_MAX` bounds one +// record and `prune()` only reclaims at seven days, so neither is the aggregate. +// XERK-268 binds `device` to the credential, which changes WHO can do this +// (a compromised or buggy host, or the `legacy` master) but bounds nothing. // // This gets its OWN process because the caps are process-wide constants read at -// require time, and the numbers that make the behavior testable (4 hosts, 64 -// KiB) are nothing like the fleet's. server.test.js lifts `AGENTS_MAX` for the -// opposite reason: it invents ~100 synthetic hosts and is not a fleet either. +// require time, and the numbers that make the behavior testable are nothing +// like the fleet's. server.test.js lifts `AGENTS_MAX` for the opposite reason: +// it invents ~100 synthetic hosts and is not a fleet either. The DEGENERATE +// config (a fleet cap far past the byte budget) lives in registry-restore. // node:test, no npm. "use strict"; @@ -27,9 +29,9 @@ process.env.TURMA_USER = "hubuser"; process.env.TURMA_PASSWORD = "hubpass"; process.env.TURMA_AGENT_TOKEN = "agenttok"; process.env.AGENTS_MAX = "4"; -// Chosen so the derived per-host share (total / max = 160 KiB) sits clear of -// BOTH the 64 KiB floor and the record sizes below — the whole point of the -// share is that a small host and a fat one land on opposite sides of it. +// Chosen so the derived per-host share (total / max = 160 KiB) sits clear of the +// record sizes below — the whole point of the share is that a small host and a +// fat one land on opposite sides of it. process.env.AGENTS_TOTAL_MAX = String(640 << 10); process.env.AGENT_EVICT_IDLE_MS = String(60 * 1000); @@ -391,6 +393,37 @@ test("a host name cannot forge a hub log line", async () => { assert.equal(lines.some((l) => l.includes("\n")), false, "a refusal log carried a raw newline"); }); +test("EVERY heartbeat log naming a host is safe, not just the newest ones", async () => { + // The 429 paths were fixed first and the older ones left; that is the wrong + // way round. `refuseOversized` is ONE request with no registry pressure at + // all, and the unknown-field drop rides a beat that returns 200 — both are + // easier to reach than the throttled refusal above. + resetRegistry(); + const forged = "evil\r\n2026-01-01T00:00:00Z FORGED: hub healthy"; + const lines = []; + const realErr = console.error; + const realWarn = console.warn; + console.error = (m) => lines.push(String(m)); + console.warn = (m) => lines.push(String(m)); + try { + // Over AGENT_RECORD_MAX -> refuseOversized's 413. + const fat = await beat({ device: forged, sessions: "A".repeat((8 << 20) + 1024) }); + assert.equal(fat.status, 413); + // An oversized UNKNOWN field -> sanitizeHeartbeat's drop line, on a beat + // that is otherwise accepted. + await beat({ device: forged, bogusField: "B".repeat((64 << 10) + 512) }); + } finally { + console.error = realErr; + console.warn = realWarn; + } + assert.ok(lines.length >= 2, `expected both log paths, got ${lines.length}`); + for (const l of lines) { + assert.equal(/[\u0000-\u0009\u000b-\u001f\u007f]/.test(l), false, + `a raw control character reached the log: ${JSON.stringify(l)}`); + assert.equal(l.includes("\n"), false, `a forged line break reached the log: ${JSON.stringify(l)}`); + } +}); + test("the refusal log is throttled, and says how many it swallowed", async () => { // The flood this cap exists to survive is exactly the traffic that writes // this line — unthrottled, surviving the attack costs the host its disk. @@ -418,6 +451,34 @@ test("the state.json ceiling is measured before the file is opened", () => { if (limit) assert.ok(STATE_FILE_MAX <= limit, "and must not exceed the container itself"); }); +test("the per-host share is DERIVED, so the overshoot cannot grow with AGENTS_MAX", () => { + // The bound is exactly AGENTS_TOTAL_MAX + AGENTS_MAX * AGENT_FAIR_SHARE. A + // FLOOR under the share (there was a 64 KiB one) makes the second term + // unbounded in AGENTS_MAX — and raising AGENTS_MAX is what an operator with a + // growing fleet is told to do. At AGENTS_MAX=2000 against the deployed 32 MiB + // budget that was 3.9x the budget and the hub was OOM-killed. + assert.ok( + AGENTS_MAX * AGENT_FAIR_SHARE <= AGENTS_TOTAL_MAX, + `${AGENTS_MAX} hosts x ${AGENT_FAIR_SHARE} bytes exceeds the ${AGENTS_TOTAL_MAX} budget` + ); + assert.equal(AGENT_FAIR_SHARE, Math.max(1, Math.floor(AGENTS_TOTAL_MAX / AGENTS_MAX))); +}); + +test("an evicted host is forgotten by everything keyed on its name", async () => { + resetRegistry(); + for (let i = 0; i < AGENTS_MAX; i++) await beat({ device: `sweep-${i}` }); + agents["sweep-0"].lastSeen = Date.now() - AGENT_EVICT_IDLE_MS * 2; + // Populate the size-warning ledger for the host about to go, the way a beat + // over half the per-record ceiling would. + hub.recordSizeWarned.set("sweep-0", true); + assert.equal((await beat({ device: "replacement" })).status, 200); + assert.equal("sweep-0" in agents, false); + // A registry that admits and evicts forever must not leak a per-host entry + // for every name it has ever seen. + assert.equal(recordBytes.has("sweep-0"), false); + assert.equal(hub.recordSizeWarned.has("sweep-0"), false, "recordSizeWarned leaked an evicted host"); +}); + test("makeRegistryRoom spends no record on a request it cannot satisfy", () => { resetRegistry(); agents["idle"] = { device: "idle", lastSeen: Date.now() - AGENT_EVICT_IDLE_MS * 2 }; diff --git a/turma/tests/registry-restore.test.js b/turma/tests/registry-restore.test.js index da16139c..8f15f6c7 100644 --- a/turma/tests/registry-restore.test.js +++ b/turma/tests/registry-restore.test.js @@ -27,6 +27,12 @@ process.env.TURMA_AGENT_TOKEN = "agenttok"; // Wound right down so an "oversized" file is a few KiB rather than the hundreds // of MiB it takes to reproduce the real kill. process.env.STATE_FILE_MAX = "4096"; +// This process ALSO carries the degenerate registry config (a big fleet cap +// against the default byte budget), because the overshoot bound only breaks +// when the DERIVED per-host share falls below what a floor would have imposed — +// at the caps registry-cap.test.js uses it never does. A hub at these numbers +// was OOM-killed at -m 256m before the share stopped being floored. +process.env.AGENTS_MAX = "2000"; const tmp = (name) => path.join(os.tmpdir(), `turma-regrestore-${name}-${process.pid}.json`); process.env.DEVICES_FILE = tmp("devices"); @@ -56,10 +62,39 @@ fs.writeFileSync( const WROTE = fs.statSync(process.env.STATE_FILE).size; const errors = []; +const warns = []; const realError = console.error; +const realWarn = console.warn; console.error = (m) => { errors.push(String(m)); realError(m); }; +console.warn = (m) => { warns.push(String(m)); realWarn(m); }; const hub = require("../server.js"); console.error = realError; +console.warn = realWarn; + +test("a big fleet cap cannot inflate the overshoot bound", () => { + // The retained worst case is AGENTS_TOTAL_MAX + AGENTS_MAX x AGENT_FAIR_SHARE. + // A FLOOR under the share makes the second term unbounded in AGENTS_MAX, and + // raising AGENTS_MAX is exactly what an operator with a growing fleet is told + // to do: at these numbers the floored share was 65536, i.e. 2000 x 64 KiB = + // 3.9x the budget, and the hub was OOM-killed at -m 256m. Derived, it is 2x + // at any AGENTS_MAX. + assert.equal(hub.AGENTS_MAX, 2000); + assert.ok(hub.AGENT_FAIR_SHARE < (64 << 10), "the rig must put the share under the old floor"); + assert.ok( + hub.AGENTS_MAX * hub.AGENT_FAIR_SHARE <= hub.AGENTS_TOTAL_MAX, + `${hub.AGENTS_MAX} x ${hub.AGENT_FAIR_SHARE} exceeds the ${hub.AGENTS_TOTAL_MAX} budget` + ); +}); + +test("and the hub says so, because the two numbers now disagree", () => { + // The bound holds, but the CONFIG is wrong — hosts get refused on record size + // long before the slots run out. Silence here is how the boot banner ends up + // printing two numbers whose product contradicts the third. + assert.ok( + warns.some((m) => m.includes("AGENTS_MAX=2000") && m.includes("Raise AGENTS_TOTAL_MAX")), + warns.join("\n") + ); +}); test("an oversized state.json does not get parsed, and the hub still boots", () => { assert.ok(WROTE > Number(process.env.STATE_FILE_MAX)); @@ -69,6 +104,36 @@ test("an oversized state.json does not get parsed, and the hub still boots", () assert.equal(typeof hub.server.listen, "function"); }); +test("a restore that fails PART WAY through serves nothing, not half a registry", () => { + // The size ceiling throws BEFORE the parse, so it cannot reach this state. + // What can: a file small enough to open whose CONTENT breaks the walk — + // `agents = JSON.parse(...)` installs the whole thing before `normalizeRecord` + // or `trimRestoredAgents` ever looks at it, so a throw in either used to leave + // the raw, uncoerced, unbounded parse installed and being served. That is the + // one state the restore exists to prevent, so it needs its own boot. + const poisoned = tmp("state-poisoned"); + // Null records: the trim's sort reads `.lastSeen` off each and throws. + fs.writeFileSync(poisoned, JSON.stringify({ h1: null, h2: null, h3: null })); + const probe = ` + process.env.STATE_FILE = ${JSON.stringify(poisoned)}; + const hub = require(${JSON.stringify(path.join(__dirname, "..", "server.js"))}); + process.stdout.write(JSON.stringify({ + keys: Object.keys(hub.agents), bytes: hub.registryBytes(), + })); + `; + const env = { ...process.env, STATE_FILE: poisoned }; + const r = require("child_process").spawnSync(process.execPath, ["-e", probe], { + env, encoding: "utf8", + }); + assert.equal(r.status, 0, `the hub must still boot:\n${r.stderr}`); + const out = JSON.parse(r.stdout); + assert.deepEqual(out.keys, [], "a failed restore must leave an EMPTY registry, not the raw parse"); + // And the accounting agrees, so the first beat is measured against an empty + // registry rather than a phantom one. + assert.equal(out.bytes, 0); + assert.ok(r.stderr.includes("state restore skipped"), r.stderr); +}); + test("it says so, and keeps the file rather than deleting it", () => { // Silence here is the whole failure mode being fixed: the old behaviour was a // crash loop that logged nothing at all. From 62f7a02dc4135e41f240669f57e3ac9a4de6d1f7 Mon Sep 17 00:00:00 2001 From: Malcolm Habeeb Date: Wed, 12 Aug 2026 19:17:50 -0400 Subject: [PATCH 5/5] XERK-272: close the mutation escapes QA's PASS still listed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The verdict was PASS with four mutations escaping, all new surface from the previous commit — including the read-only-volume message, which was itself the fix for the finding before it and had nothing pinning it. * The share is now `fairShare(total, max)`, a function, so the extremes are reachable to a test without a process pinned to a degenerate config. A share of 0 (reachable when AGENTS_MAX passes the budget in BYTES, which positiveEnv accepts) makes every host over-share and brings back the silent-offline regression; nothing was watching that. * The oversized-state message's two branches are both driven, on a real read-only directory in a child process. * All four converted log sites are driven — the earlier test only reached two of them, so the over-half warn and the coercion-failure line could have reverted to interpolating the raw device name. * `logName` also strips C1 now: JSON.stringify escapes none of that block, and NEL (U+0085) reads as a line break to some log viewers. Also QA's answer to whether the budget should be a larger fraction of the container: no — the gap is the silence, not the size. The per-record ceiling warns at 4 MiB while a share is 512 KiB, so a record drifts past its share, starts being refused, and the first thing the operator sees is the host vanishing, with the older warning still eight times away. So a host now gets the same crossing-edge warning against HALF ITS SHARE. That is what makes the eighth-of-the-container default defensible; raising the fraction would reserve half the container, since the 193.8 MiB flood peak is mostly serving overhead that scales with the registry rather than the registry itself. QA verdict: PASS (pass 3). 512 distinct devices survive at -m 256m in both strict and legacy auth (36x200/476x429, peak 193.8 MiB); the AGENTS_MAX=2000 config that was OOM-killed now lands at 54.8 MiB against a 64 MiB bound; dashboard/sessions/board/usage byte-identical to origin/main. --- .claude/rules/turma.md | 14 +++++- turma/server.js | 37 +++++++++++++-- turma/tests/registry-cap.test.js | 71 +++++++++++++++++++++++++--- turma/tests/registry-restore.test.js | 25 ++++++++++ 4 files changed, 135 insertions(+), 12 deletions(-) diff --git a/.claude/rules/turma.md b/.claude/rules/turma.md index 5651606c..84613181 100644 --- a/.claude/rules/turma.md +++ b/.claude/rules/turma.md @@ -334,8 +334,18 @@ working-status bar, ready-for-review, ended sessions, the composer and the termi is not. - **Every log line naming a host goes through `logName`** — `device` is agent-supplied and validated only for length and prototype keys, so a newline in it forged a line reading exactly like the - hub's own. Refusal logs are throttled to one a minute with the suppressed count, because the flood - the cap exists to survive is precisely the traffic that writes them. + hub's own. It strips C0, DEL **and C1** (`JSON.stringify` escapes none of the C1 block). All FIVE + sites go through it: converting only the new ones left the two cheapest to reach — the 413, which + is one request needing no registry pressure, and the unknown-field drop, which rides a 200. + Refusal logs are throttled to one a minute with the suppressed count, because the flood the cap + exists to survive is precisely the traffic that writes them. +- **A host is warned on the crossing into half its share** (`shareWarned`, the `recordSizeWarned` + pattern). Without it the first signal is the host vanishing: the per-record ceiling's warning is at + 4 MiB and a share is 512 KiB, so a record drifts past its share — and starts being refused — with + that warning still eight times away. This is what makes the eighth-of-the-container default safe; + a record grows over weeks, which is ample notice provided somebody is told. If real records ever + approach the share, raise `AGENTS_TOTAL_MAX` in the DockerOps compose beside `mem_limit` rather + than moving the derived default for every deployment at once. - Byte accounting is a side map (`recordBytes`), never a field on the record — anything on a record is served to every client — and `registryBytes()` re-measures unknown keys and forgets dead ones, so the many places that `delete agents[key]` need not remember it. diff --git a/turma/server.js b/turma/server.js index c43c8091..61a93cdd 100644 --- a/turma/server.js +++ b/turma/server.js @@ -487,7 +487,10 @@ function logName(key) { // JSON.stringify does the line-forging half (a newline becomes the two // characters \ and n, so it can never start a line); the sweep after it // covers the control characters JSON leaves intact, which a terminal acts on. - return JSON.stringify(String(key).slice(0, 200)).replace(/[\u0000-\u001f\u007f]/g, "?"); + // C0, DEL **and C1** — JSON.stringify escapes none of the C1 block, and NEL + // (U+0085) is a line break to some readers. + return JSON.stringify(String(key).slice(0, 200)) + .replace(/[\u0000-\u001f\u007f-\u009f]/g, "?"); } const AGENTS_MAX = positiveEnv("AGENTS_MAX", 64); @@ -554,7 +557,15 @@ const AGENTS_TOTAL_MAX = positiveEnv("AGENTS_TOTAL_MAX", defaultRegistryBudget() // At AGENTS_MAX=2000 against the deployed 32 MiB budget that is 3.9x, and the // hub is OOM-killed. Raising the count means raising the budget with it; the // boot check below says so rather than letting the two contradict silently. -const AGENT_FAIR_SHARE = Math.max(1, Math.floor(AGENTS_TOTAL_MAX / AGENTS_MAX)); +// +// A function, not an expression, so the extremes are reachable to a test +// without a whole process pinned to a degenerate config: the interesting cases +// (a count past the budget in BYTES, a count of one) are exactly the ones a +// realistic rig never reaches. +function fairShare(total, max) { + return Math.max(1, Math.floor(total / max)); +} +const AGENT_FAIR_SHARE = fairShare(AGENTS_TOTAL_MAX, AGENTS_MAX); // A share this small means the count cap and the byte budget disagree about how // big a fleet this hub is for: hosts would be refused on size long before the // slots ran out. The memory bound still holds — it is the CONFIGURATION that is @@ -612,6 +623,10 @@ function agentRecordSize(record) { // on the record: anything stored on the record is served to every client. const recordBytes = new Map(); +// Which hosts are already over half their share, so that warning fires on the +// crossing rather than on every beat — same discipline as recordSizeWarned. +const shareWarned = new Map(); + // The aggregate `agentRecordSize` of the whole registry. Measures lazily for a // key it has not seen (the state.json restore, and the tests, install records // without going through the heartbeat) and forgets keys that are gone, so it @@ -690,6 +705,7 @@ function makeRegistryRoom(addBytes, addSlots) { // Everything else keyed by host name has to let go too, or a registry that // admits and evicts forever leaks a per-host entry per name it ever saw. recordSizeWarned.delete(key); + shareWarned.delete(key); invalidateAgentsCache(); sseBroadcast("removed", { key }); } @@ -4656,6 +4672,21 @@ const server = http.createServer(async (req, res) => { ); } recordSizeWarned.set(key, overHalf); + // The same crossing-edge warning against the host's SHARE, which is the + // line that actually bites: the ceiling above is 8 MiB and a share is + // 512 KiB, so a host drifts past its share — and starts being refused + // once the registry is also full — with the ceiling's warning still + // eight times away. A record grows over weeks, which is plenty of notice + // if anyone is told; without this the first signal is the host vanishing. + const overHalfShare = recordSize > AGENT_FAIR_SHARE / 2; + if (overHalfShare && !shareWarned.get(key)) { + console.warn( + `heartbeat from ${logName(key)}: record is ${recordSize} bytes, over half ` + + `its ${AGENT_FAIR_SHARE}-byte share of the registry budget — past the ` + + `whole share it is refused whenever the registry is full` + ); + } + shareWarned.set(key, overHalfShare); if (recordSize > AGENT_RECORD_MAX) return refuseOversized(recordSize); // The AGGREGATE budget (XERK-272). One record under the per-record ceiling // is not the bound: AGENTS_MAX records AT that ceiling is 512 MiB on a @@ -6454,7 +6485,7 @@ if (process.env.TURMA_TEST) { // the per-record ceiling stayed green while an unbounded NUMBER of records // OOM-killed the hub, so the aggregate has to be pinned by name too. AGENTS_MAX, AGENTS_TOTAL_MAX, AGENT_EVICT_IDLE_MS, AGENT_FAIR_SHARE, - STATE_FILE_MAX, positiveEnv, logName, recordSizeWarned, + STATE_FILE_MAX, positiveEnv, logName, recordSizeWarned, shareWarned, fairShare, registryBytes, makeRegistryRoom, trimRestoredAgents, containerMemoryLimit, defaultRegistryBudget, recordBytes, // Ingest coercion, exported for the same reason as the rest of this group: diff --git a/turma/tests/registry-cap.test.js b/turma/tests/registry-cap.test.js index 9a804c60..9cc67424 100644 --- a/turma/tests/registry-cap.test.js +++ b/turma/tests/registry-cap.test.js @@ -397,9 +397,10 @@ test("EVERY heartbeat log naming a host is safe, not just the newest ones", asyn // The 429 paths were fixed first and the older ones left; that is the wrong // way round. `refuseOversized` is ONE request with no registry pressure at // all, and the unknown-field drop rides a beat that returns 200 — both are - // easier to reach than the throttled refusal above. + // easier to reach than the throttled refusal above. All four sites that name + // a host are driven here, so none of them can quietly revert. resetRegistry(); - const forged = "evil\r\n2026-01-01T00:00:00Z FORGED: hub healthy"; + const forged = "evil\r\n2026-01-01T00:00:00Z FORGED: hub healthy\u001b[2J\u0085NEL"; const lines = []; const realErr = console.error; const realWarn = console.warn; @@ -407,18 +408,37 @@ test("EVERY heartbeat log naming a host is safe, not just the newest ones", asyn console.warn = (m) => lines.push(String(m)); try { // Over AGENT_RECORD_MAX -> refuseOversized's 413. - const fat = await beat({ device: forged, sessions: "A".repeat((8 << 20) + 1024) }); - assert.equal(fat.status, 413); + assert.equal( + (await beat({ device: forged, sessions: "A".repeat((8 << 20) + 1024) })).status, 413); // An oversized UNKNOWN field -> sanitizeHeartbeat's drop line, on a beat - // that is otherwise accepted. + // that is otherwise ACCEPTED. await beat({ device: forged, bogusField: "B".repeat((64 << 10) + 512) }); + // Between half and all of AGENT_RECORD_MAX -> the over-half warn (and the + // over-half-SHARE warn, which this rig crosses far earlier). + await beat(chunky(forged, (8 << 20) / 2 / 1024 + 64)); + // And the coercion-failure line, which no wire input can reach. + const realNormalize = hub.recordCoercion.normalize; + hub.recordCoercion.normalize = () => { throw new Error("coercion blew up"); }; + try { + assert.equal((await beat({ device: forged })).status, 400); + } finally { + hub.recordCoercion.normalize = realNormalize; + } } finally { console.error = realErr; console.warn = realWarn; } - assert.ok(lines.length >= 2, `expected both log paths, got ${lines.length}`); + // 413, the drop line, over-half, over-half-share, coercion failure. + assert.ok(lines.length >= 5, `only ${lines.length} of the log paths were driven`); + assert.ok(lines.some((l) => l.includes("over the")), "the 413 line"); + assert.ok(lines.some((l) => l.includes("dropped unknown field")), "the drop line"); + assert.ok(lines.some((l) => l.includes("over half the")), "the over-half warn"); + assert.ok(lines.some((l) => l.includes("share of the registry budget")), "the share warn"); + assert.ok(lines.some((l) => l.includes("coercion failed")), "the coercion-failure line"); for (const l of lines) { - assert.equal(/[\u0000-\u0009\u000b-\u001f\u007f]/.test(l), false, + // C0, DEL and C1 — JSON.stringify escapes none of the C1 block, and NEL + // (U+0085) reads as a line break to some log viewers. + assert.equal(/[\u0000-\u0009\u000b-\u001f\u007f-\u009f]/.test(l), false, `a raw control character reached the log: ${JSON.stringify(l)}`); assert.equal(l.includes("\n"), false, `a forged line break reached the log: ${JSON.stringify(l)}`); } @@ -451,6 +471,43 @@ test("the state.json ceiling is measured before the file is opened", () => { if (limit) assert.ok(STATE_FILE_MAX <= limit, "and must not exceed the container itself"); }); +test("fairShare never returns zero, however absurd the caps are", () => { + // A share of 0 makes `recordSize > share` true for everyone, so the exemption + // never applies and every host is refused the moment the registry is full — + // the silent-offline regression, arrived at from a config `positiveEnv` + // accepts. Held on the function so the extremes are reachable without a whole + // process pinned to a degenerate config. + assert.equal(hub.fairShare(1 << 20, 4), (1 << 20) / 4); + assert.equal(hub.fairShare(100, 100), 1); + assert.equal(hub.fairShare(100, 1000), 1, "a count past the budget in BYTES must still leave 1"); + assert.equal(hub.fairShare(1, 1 << 30), 1); + assert.equal(hub.fairShare(1 << 20, 1), 1 << 20); + // Floor, not round or ceil: AGENTS_MAX x share must never exceed the budget. + assert.equal(hub.fairShare(10, 3), 3); +}); + +test("a host is warned BEFORE its share starts refusing it, once per crossing", async () => { + // The per-record ceiling's warning is at 4 MiB and a share is 512 KiB, so a + // host drifts past its share with the older warning still eight times away — + // and the first thing the operator sees is the host vanishing. + resetRegistry(); + const warns = []; + const realWarn = console.warn; + console.warn = (m) => warns.push(String(m)); + try { + const halfShareKiB = Math.ceil(AGENT_FAIR_SHARE / 2 / 1024) + 2; + assert.equal((await beat(chunky("drifter", halfShareKiB))).status, 200); + assert.equal((await beat(chunky("drifter", halfShareKiB))).status, 200); + assert.equal((await beat(chunky("drifter", halfShareKiB))).status, 200); + } finally { + console.warn = realWarn; + } + const share = warns.filter((m) => m.includes("over half") && m.includes("share")); + assert.equal(share.length, 1, `warned ${share.length} times, want exactly one crossing`); + assert.ok(share[0].includes("drifter")); + assert.ok(hub.shareWarned.get("drifter"), "the crossing must be remembered, not re-warned"); +}); + test("the per-host share is DERIVED, so the overshoot cannot grow with AGENTS_MAX", () => { // The bound is exactly AGENTS_TOTAL_MAX + AGENTS_MAX * AGENT_FAIR_SHARE. A // FLOOR under the share (there was a 64 KiB one) makes the second term diff --git a/turma/tests/registry-restore.test.js b/turma/tests/registry-restore.test.js index 8f15f6c7..3546fba7 100644 --- a/turma/tests/registry-restore.test.js +++ b/turma/tests/registry-restore.test.js @@ -134,6 +134,31 @@ test("a restore that fails PART WAY through serves nothing, not half a registry" assert.ok(r.stderr.includes("state restore skipped"), r.stderr); }); +test("when it CANNOT move the file, it says that instead of naming one that isn't there", () => { + // The message is the operator's only lead. On a read-only /data the rename + // fails, and pointing them at a `.oversized` that was never created sends + // them looking for a file the hub did not write — reading as "the hub ate my + // state". + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "turma-regrestore-ro-")); + const locked = path.join(dir, "state.json"); + fs.writeFileSync(locked, JSON.stringify({ h: { device: "h", lastSeen: 1, pad: "x".repeat(8000) } })); + fs.chmodSync(dir, 0o555); // read-only DIRECTORY: the file is readable, unrenamable + try { + const probe = `require(${JSON.stringify(path.join(__dirname, "..", "server.js"))});`; + const r = require("child_process").spawnSync(process.execPath, ["-e", probe], { + env: { ...process.env, STATE_FILE: locked }, encoding: "utf8", + }); + assert.equal(r.status, 0, `the hub must still boot on a read-only volume:\n${r.stderr}`); + assert.ok(r.stderr.includes("could not move it"), r.stderr); + assert.equal(r.stderr.includes(".oversized"), false, + "it must not name a file it failed to create"); + assert.ok(fs.existsSync(locked), "and the original must still be there"); + } finally { + fs.chmodSync(dir, 0o755); + fs.rmSync(dir, { recursive: true, force: true }); + } +}); + test("it says so, and keeps the file rather than deleting it", () => { // Silence here is the whole failure mode being fixed: the old behaviour was a // crash loop that logged nothing at all.