diff --git a/.claude/rules/turma.md b/.claude/rules/turma.md index bb1c34cc..84613181 100644 --- a/.claude/rules/turma.md +++ b/.claude/rules/turma.md @@ -283,6 +283,81 @@ working-status bar, ready-for-review, ended sessions, the composer and the termi working. It is hub-derived and **stripped from the fleet payload** — putting it on the wire would make it a client contract. - Tests: the `XERK-268:` cases in `server.test.js`. + +### 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. + - **XERK-268 shrank who can do this; it did not bound it.** `device` is now PROVED by the + credential, so this is no longer any-token-holder — but a compromised or buggy host still mints + names under its own token, the `legacy` master a mid-rollover fleet accepts is not yet retired, + and a host deriving its name from something unstable grows records with no attacker at all. + Per-agent tokens and this cap are complementary, not alternatives. +- 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 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. 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. +- 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 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. 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 + 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. 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. +- 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` (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 bed56fb3..61a93cdd 100644 --- a/turma/server.js +++ b/turma/server.js @@ -430,6 +430,325 @@ 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) --------------------------------- +// +// Nothing capped how many DISTINCT `device` names the registry could hold, and +// every one of them is 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. +// +// XERK-268 binds `device` to the credential, so this is no longer reachable by +// anyone holding the fleet token — but it does not BOUND anything: a +// compromised or buggy host still mints names under its own proved token, the +// `legacy` master a mid-rollover fleet accepts is not yet retired, and a host +// whose name derives from something unstable grows records with no attacker at +// all. The two are complementary; neither replaces the other. +// +// 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 +// 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. + +/** + * 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. + // 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); +// 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 = 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 +// 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 = 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 +// 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. +// +// 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 +// 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 +// 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. +// +// 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(); + +// 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 +// 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; +} + +// 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) + .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 ${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); + shareWarned.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] = the record awaiting the // agent's data-WS dial-back for channel `ch`. @@ -452,6 +771,28 @@ 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) { + // 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`; + 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 ` + + (movedTo ? `kept at ${movedTo}` : `left in place at ${STATE_FILE} (could not move it)`) + ); + } 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 @@ -461,9 +802,20 @@ 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 */ +} 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 @@ -2031,22 +2383,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. // @@ -2069,7 +2408,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]; @@ -4166,6 +4505,29 @@ const server = http.createServer(async (req, res) => { // Which credential it used decides which token its ttyd is running with // (see ttydAuth). Hub-derived, never read off the payload. const tokenBound = agentBearerKind(req, key) === "proved"; + // Admission control (XERK-272), ordered AFTER the binding above so an + // unbound beat can never spend a registry slot. 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. + // + // XERK-268 makes `device` PROVED rather than self-asserted, which shrinks + // this from an anyone-with-the-token attack to a compromised-or-buggy host + // and the `legacy` master credential a mid-rollover fleet still accepts — + // neither of which is nothing, and a host deriving its name from something + // unstable grows records with no attacker at all. + const known = Object.prototype.hasOwnProperty.call(agents, key); + if (!known && !makeRegistryRoom(0, 1)) { + // 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, + }); + } 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. @@ -4265,7 +4627,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 }); @@ -4290,7 +4652,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" }); @@ -4305,12 +4667,54 @@ 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` ); } 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 + // 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) && 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)); + } else { + delete agents[key]; + recordBytes.delete(key); + } + 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, share: AGENT_FAIR_SHARE, + }); + } ingestHistory(next, historyResults); ingestSubagentHistory(next, subagentHistoryResults); ingestJiraIssues(next, jiraIssueResults); @@ -4345,6 +4749,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); @@ -6069,6 +6481,13 @@ 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, AGENT_FAIR_SHARE, + 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: // Android decodes /api/agents atomically, so one host's wrong-typed field // hides the WHOLE fleet from that phone (XERK-246). `normalizeRecord` is @@ -6180,6 +6599,14 @@ if (process.env.TURMA_TEST) { } 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 new file mode 100644 index 00000000..9cc67424 --- /dev/null +++ b/turma/tests/registry-cap.test.js @@ -0,0 +1,548 @@ +// Unit tests for the agent registry's own ceiling (XERK-272). +// +// 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 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"; + +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"; +// 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); + +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, AGENT_FAIR_SHARE, STATE_FILE_MAX, + registryBytes, makeRegistryRoom, agentRecordSize, positiveEnv, logName, + 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. 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(refused.body.share, AGENT_FAIR_SHARE); + assert.equal("fat-c" in agents, false); +}); + +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: 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); + 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"])); + // 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" }]); +}); + +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("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("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. 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\u001b[2J\u0085NEL"; + 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. + 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. + 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; + } + // 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) { + // 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)}`); + } +}); + +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("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 + // 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 }; + 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/registry-restore.test.js b/turma/tests/registry-restore.test.js new file mode 100644 index 00000000..3546fba7 --- /dev/null +++ b/turma/tests/registry-restore.test.js @@ -0,0 +1,170 @@ +// 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"; +// 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"); +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 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)); + // 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("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("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. + 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"); +}); diff --git a/turma/tests/server.test.js b/turma/tests/server.test.js index 6e380c83..1c4a2aa3 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`