From f41ca1b33b76b57d94d791967f715a5bd11bf347 Mon Sep 17 00:00:00 2001 From: Neil Date: Thu, 10 Sep 2026 20:37:47 -0700 Subject: [PATCH 1/2] test(relay): add the real-host PTY churn and forced-GC probes Harness behind the openclaw measurements, kept so the numbers are reproducible rather than quoted. Each reports exact counts or post-collection heap, never a duration. The invariant they establish is `fds == 25 + ptys.active + liveConnections`, which holds at every sample: a dropped connection releases its own descriptor, a PTY keeps its own until it is explicitly retired, and 200 PTY create/destroy cycles leave the count exactly at baseline. The retire probe exists because the churn probe used a fixed 1.5s settle and could not tell "not retired" from "not retired yet" -- retirement takes about 5.6s. It polls to a real baseline instead of guessing a window. relay-gc-host.cjs loads the relay unmodified and adds one SIGUSR2 handler that collects twice and writes process.memoryUsage(), so retained heap can be read after a real collection rather than inferred from an RSS curve. --- config/scripts/relay-gc-host.cjs | 21 ++ config/scripts/relay-pty-churn-probe.mjs | 316 +++++++++++++++++++ config/scripts/relay-pty-gc-cycles-probe.mjs | 244 ++++++++++++++ config/scripts/relay-pty-retire-probe.mjs | 217 +++++++++++++ 4 files changed, 798 insertions(+) create mode 100644 config/scripts/relay-gc-host.cjs create mode 100644 config/scripts/relay-pty-churn-probe.mjs create mode 100644 config/scripts/relay-pty-gc-cycles-probe.mjs create mode 100644 config/scripts/relay-pty-retire-probe.mjs diff --git a/config/scripts/relay-gc-host.cjs b/config/scripts/relay-gc-host.cjs new file mode 100644 index 00000000000..3c8395e30fb --- /dev/null +++ b/config/scripts/relay-gc-host.cjs @@ -0,0 +1,21 @@ +// Runs the real relay bundle under --expose-gc and reports post-collection memory on SIGUSR2. +// Nothing about the relay changes: this only adds a signal handler before loading it. +const { writeFileSync } = require('node:fs') + +const report = process.env.ORCA_GC_REPORT +process.on('SIGUSR2', () => { + try { + // Twice: the first pass can resurrect via finalizers, the second settles it. + global.gc() + global.gc() + } catch { + /* --expose-gc absent; report raw numbers so the caller can tell */ + } + try { + writeFileSync(report, JSON.stringify({ at: Date.now(), mem: process.memoryUsage() })) + } catch { + /* best-effort */ + } +}) + +require('./relay.js') diff --git a/config/scripts/relay-pty-churn-probe.mjs b/config/scripts/relay-pty-churn-probe.mjs new file mode 100644 index 00000000000..9ba349a7c28 --- /dev/null +++ b/config/scripts/relay-pty-churn-probe.mjs @@ -0,0 +1,316 @@ +// HOW TO RUN: copy this file, relay-gc-host.cjs, the built `relay.js` and its `.version` into one +// directory on the target host (they resolve each other via import.meta.dirname), install a +// matching `node-pty` beside them, then `node `. Results are exact counts and +// post-collection heap figures; nothing here asserts on a duration. +// Real-host churn probe with PTYs attached, plus forced-GC retained-heap readings. +// Every reported number is an exact count or a post-collection heap figure; no wall-clock bounds. +import { spawn } from 'node:child_process' +import net from 'node:net' +import { readFileSync, readdirSync, existsSync, rmSync, mkdirSync, writeFileSync } from 'node:fs' +import { join } from 'node:path' + +const HERE = import.meta.dirname +const VERSION = readFileSync(join(HERE, '.version'), 'utf8').trim() +const RUNDIR = '/tmp/orca-pty-churn-probe' +const SOCK = join(RUNDIR, 'relay.sock') +const GC_REPORT = join(RUNDIR, 'gc.json') +const HEADER = 13 +const CYCLES = 3 +const CONNS_PER_CYCLE = 20 + +rmSync(RUNDIR, { recursive: true, force: true }) +mkdirSync(RUNDIR, { recursive: true }) + +function encodeFrame(type, id, ack, payload) { + const h = Buffer.alloc(HEADER) + h[0] = type + h.writeUInt32BE(id, 1) + h.writeUInt32BE(ack, 5) + h.writeUInt32BE(payload.length, 9) + return Buffer.concat([h, payload]) +} +const handshakeFrame = () => + encodeFrame( + 2, + 0, + 0, + Buffer.from(JSON.stringify({ type: 'orca-relay-handshake', version: VERSION })) + ) + +function decodeFrames(buf) { + const out = [] + let off = 0 + while (buf.length - off >= HEADER) { + const len = buf.readUInt32BE(off + 9) + if (buf.length - off - HEADER < len) { + break + } + out.push({ type: buf[off], payload: buf.subarray(off + HEADER, off + HEADER + len) }) + off += HEADER + len + } + return { frames: out, rest: buf.subarray(off) } +} + +function connect() { + return new Promise((resolve, reject) => { + const sock = net.connect(SOCK) + let buf = Buffer.alloc(0) + let done = false + sock.on('error', (e) => !done && reject(e)) + sock.on('data', (d) => { + buf = Buffer.concat([buf, d]) + const { frames, rest } = decodeFrames(buf) + buf = rest + for (const f of frames) { + if (f.type !== 2) { + continue + } + const msg = JSON.parse(f.payload.toString()) + done = true + if (msg.type === 'orca-relay-handshake-ok') { + sock.removeAllListeners('data') + sock._carry = buf + sock._seq = 0 + resolve(sock) + } else { + reject(new Error(`handshake refused: ${msg.type}`)) + } + } + }) + sock.on('connect', () => sock.write(handshakeFrame())) + }) +} + +let rpcId = 1 +function rpc(sock, method, params, timeoutMs = 20000) { + return new Promise((resolve, reject) => { + const id = rpcId++ + let buf = sock._carry ?? Buffer.alloc(0) + sock._carry = Buffer.alloc(0) + const timer = setTimeout(() => { + sock.off('data', onData) + reject(new Error(`${method} timed out`)) + }, timeoutMs) + const onData = (d) => { + buf = Buffer.concat([buf, d]) + const { frames, rest } = decodeFrames(buf) + buf = rest + for (const f of frames) { + if (f.type !== 1) { + continue + } + let msg + try { + msg = JSON.parse(f.payload.toString()) + } catch { + continue + } + if (msg.id === id) { + clearTimeout(timer) + sock.off('data', onData) + sock._carry = buf + if (msg.error) { + reject(new Error(JSON.stringify(msg.error))) + } else { + resolve(msg.result) + } + } + } + } + sock.on('data', onData) + sock.write( + encodeFrame( + 1, + ++sock._seq, + 0, + Buffer.from(JSON.stringify({ jsonrpc: '2.0', id, method, params })) + ) + ) + }) +} + +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)) +const fdCount = (pid) => { + try { + return readdirSync(`/proc/${pid}/fd`).length + } catch { + return -1 + } +} + +async function forcedGcMem(pid) { + try { + writeFileSync(GC_REPORT, '') + } catch { + /* ignore */ + } + process.kill(pid, 'SIGUSR2') + for (let i = 0; i < 60; i++) { + await sleep(100) + try { + const raw = readFileSync(GC_REPORT, 'utf8') + if (raw.trim()) { + return JSON.parse(raw).mem + } + } catch { + /* not written yet */ + } + } + return null +} + +async function main() { + const child = spawn( + process.execPath, + [ + '--expose-gc', + join(HERE, 'gc-host.cjs'), + '--detached', + '--sock-path', + SOCK, + '--grace-time', + '0', + '--log-file', + join(RUNDIR, 'relay.log') + ], + { + cwd: RUNDIR, + stdio: ['ignore', 'pipe', 'pipe'], + env: { + ...process.env, + ORCA_GC_REPORT: GC_REPORT, + ORCA_RELAY_EMPTY_STARTUP_GRACE_MS: '3600000', + ORCA_RELAY_IDLE_GRACE_MS: '3600000' + } + } + ) + const relayLog = [] + child.stderr.on('data', (d) => relayLog.push(d.toString())) + child.stdout.on('data', (d) => relayLog.push(d.toString())) + + for (let i = 0; i < 150 && !existsSync(SOCK); i++) { + await sleep(100) + } + if (!existsSync(SOCK)) { + throw new Error('relay socket never appeared') + } + await sleep(500) + + const pid = child.pid + const obs = await connect() + const rows = [] + const snap = async (label) => { + const s = await rpc(obs, 'relay.status', {}) + const mem = await forcedGcMem(pid) + rows.push({ + label, + clients: s.socket.clients, + accepted: s.socket.acceptedConnections, + ptys: s.ptys.active, + fds: fdCount(pid), + heapAfterGcMb: mem ? +(mem.heapUsed / 1048576).toFixed(2) : null, + rssMb: mem ? +(mem.rss / 1048576).toFixed(1) : null, + externalMb: mem ? +(mem.external / 1048576).toFixed(2) : null, + session: JSON.stringify(s.ptySourceCredit.session) + }) + } + + await snap('baseline (observer only)') + + const ptyIds = [] + let spawnError = null + for (let cycle = 1; cycle <= CYCLES; cycle++) { + const socks = [] + for (let i = 0; i < CONNS_PER_CYCLE; i++) { + const s = await connect() + socks.push(s) + try { + const r = await rpc(s, 'pty.spawn', { + cwd: RUNDIR, + cols: 80, + rows: 24, + shell: '/bin/sh', + args: [] + }) + if (r?.id !== undefined) { + ptyIds.push(r.id) + } + } catch (e) { + spawnError ??= e.message + } + } + await snap(`cycle ${cycle}: ${CONNS_PER_CYCLE} conns holding PTYs`) + // Drop uncleanly -- no pty.shutdown, no socket end handshake. + for (const s of socks) { + s.destroy() + } + await sleep(1000) + await snap(`cycle ${cycle}: conns dropped uncleanly`) + } + + // PTYs are meant to SURVIVE a dropped connection; retire them explicitly and re-check. + let shutdownError = null + for (const id of ptyIds) { + try { + await rpc(obs, 'pty.shutdown', { id }) + } catch (e) { + shutdownError ??= e.message + } + } + await sleep(1500) + await snap('after explicit pty.shutdown of every PTY') + + console.log(JSON.stringify({ spawnError, shutdownError, ptysSpawned: ptyIds.length }, null, 1)) + console.log(JSON.stringify(rows, null, 1)) + + // --- reaper: a client that speaks once (keepalive) then goes silent --- + const beforeB = (await rpc(obs, 'relay.status', {})).socket.clients + const talker = await connect() + talker.write(encodeFrame(9, 1, 0, Buffer.alloc(0))) // KeepAlive => keepaliveObserved = true + await sleep(2000) + const duringB = (await rpc(obs, 'relay.status', {})).socket.clients + await sleep(32000) // TIMEOUT_MS is 20s; give the 5s keepalive tick room to judge it + const afterB = (await rpc(obs, 'relay.status', {})).socket.clients + console.log( + JSON.stringify({ reaperStoppedAnswering: { beforeB, duringB, afterB, fds: fdCount(pid) } }) + ) + talker.destroy() + + // --- reaper: a client that completes the handshake and never frames anything --- + await sleep(1000) + const beforeA = (await rpc(obs, 'relay.status', {})).socket.clients + const mute = await connect() + await sleep(2000) + const duringA = (await rpc(obs, 'relay.status', {})).socket.clients + await sleep(132000) // SILENT_CONNECT_TIMEOUT_MS = TIMEOUT_MS * 6 = 120s + const afterA = (await rpc(obs, 'relay.status', {})).socket.clients + console.log(JSON.stringify({ reaperNeverSpoke: { beforeA, duringA, afterA, fds: fdCount(pid) } })) + mute.destroy() + + const finalMem = await forcedGcMem(pid) + console.log( + JSON.stringify({ + finalHeapAfterGcMb: finalMem ? +(finalMem.heapUsed / 1048576).toFixed(2) : null, + finalRssMb: finalMem ? +(finalMem.rss / 1048576).toFixed(1) : null, + finalFds: fdCount(pid) + }) + ) + + const log = relayLog.join('') + console.log( + JSON.stringify({ + logStoppedAnswering: (log.match(/stopped answering/g) || []).length, + logNeverSpoke: (log.match(/never spoke/g) || []).length + }) + ) + + obs.destroy() + child.kill('SIGKILL') + await sleep(500) + console.log(`RELAY_PID=${pid}`) +} + +main().catch(async (e) => { + console.error(`PROBE_FAILED: ${e.message}`) + process.exit(1) +}) diff --git a/config/scripts/relay-pty-gc-cycles-probe.mjs b/config/scripts/relay-pty-gc-cycles-probe.mjs new file mode 100644 index 00000000000..963d0b85dd6 --- /dev/null +++ b/config/scripts/relay-pty-gc-cycles-probe.mjs @@ -0,0 +1,244 @@ +// HOW TO RUN: copy this file, relay-gc-host.cjs, the built `relay.js` and its `.version` into one +// directory on the target host (they resolve each other via import.meta.dirname), install a +// matching `node-pty` beside them, then `node `. Results are exact counts and +// post-collection heap figures; nothing here asserts on a duration. +// Discriminator: is retained-after-GC heap proportional to churn (a leak) or fixed (init cost)? +// Repeats spawn-20-PTYs / shutdown-all / forced-GC many times and reports the retained heap each +// cycle. A per-PTY or per-connection leak climbs with the cycle count; init cost plateaus. +import { spawn } from 'node:child_process' +import net from 'node:net' +import { readFileSync, readdirSync, existsSync, rmSync, mkdirSync, writeFileSync } from 'node:fs' +import { join } from 'node:path' + +const HERE = import.meta.dirname +const VERSION = readFileSync(join(HERE, '.version'), 'utf8').trim() +const RUNDIR = '/tmp/orca-pty-gc-cycles' +const SOCK = join(RUNDIR, 'relay.sock') +const GC_REPORT = join(RUNDIR, 'gc.json') +const HEADER = 13 +const PTYS = 20 +const CYCLES = 10 + +rmSync(RUNDIR, { recursive: true, force: true }) +mkdirSync(RUNDIR, { recursive: true }) + +const enc = (type, id, ack, payload) => { + const h = Buffer.alloc(HEADER) + h[0] = type + h.writeUInt32BE(id, 1) + h.writeUInt32BE(ack, 5) + h.writeUInt32BE(payload.length, 9) + return Buffer.concat([h, payload]) +} +const dec = (buf) => { + const out = [] + let off = 0 + while (buf.length - off >= HEADER) { + const len = buf.readUInt32BE(off + 9) + if (buf.length - off - HEADER < len) { + break + } + out.push({ type: buf[off], payload: buf.subarray(off + HEADER, off + HEADER + len) }) + off += HEADER + len + } + return { frames: out, rest: buf.subarray(off) } +} + +function connect() { + return new Promise((resolve, reject) => { + const sock = net.connect(SOCK) + let buf = Buffer.alloc(0) + sock.on('error', reject) + sock.on('data', (d) => { + buf = Buffer.concat([buf, d]) + const { frames, rest } = dec(buf) + buf = rest + for (const f of frames) { + if (f.type !== 2) { + continue + } + const m = JSON.parse(f.payload.toString()) + if (m.type === 'orca-relay-handshake-ok') { + sock.removeAllListeners('data') + sock._carry = buf + sock._seq = 0 + resolve(sock) + } else { + reject(new Error(m.type)) + } + } + }) + sock.on('connect', () => + sock.write( + enc( + 2, + 0, + 0, + Buffer.from(JSON.stringify({ type: 'orca-relay-handshake', version: VERSION })) + ) + ) + ) + }) +} + +let rpcId = 1 +function rpc(sock, method, params, timeoutMs = 25000) { + return new Promise((resolve, reject) => { + const id = rpcId++ + let buf = sock._carry ?? Buffer.alloc(0) + sock._carry = Buffer.alloc(0) + const timer = setTimeout(() => { + sock.off('data', onData) + reject(new Error(`${method} timed out`)) + }, timeoutMs) + const onData = (d) => { + buf = Buffer.concat([buf, d]) + const { frames, rest } = dec(buf) + buf = rest + for (const f of frames) { + if (f.type !== 1) { + continue + } + let m + try { + m = JSON.parse(f.payload.toString()) + } catch { + continue + } + if (m.id === id) { + clearTimeout(timer) + sock.off('data', onData) + sock._carry = buf + if (m.error) { + reject(new Error(JSON.stringify(m.error))) + } else { + resolve(m.result) + } + } + } + } + sock.on('data', onData) + sock.write( + enc(1, ++sock._seq, 0, Buffer.from(JSON.stringify({ jsonrpc: '2.0', id, method, params }))) + ) + }) +} + +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)) +const fds = (pid) => { + try { + return readdirSync(`/proc/${pid}/fd`).length + } catch { + return -1 + } +} + +async function forcedGc(pid) { + writeFileSync(GC_REPORT, '') + process.kill(pid, 'SIGUSR2') + for (let i = 0; i < 80; i++) { + await sleep(100) + const raw = readFileSync(GC_REPORT, 'utf8') + if (raw.trim()) { + return JSON.parse(raw).mem + } + } + return null +} + +async function main() { + const child = spawn( + process.execPath, + [ + '--expose-gc', + join(HERE, 'gc-host.cjs'), + '--detached', + '--sock-path', + SOCK, + '--grace-time', + '0', + '--log-file', + join(RUNDIR, 'relay.log') + ], + { + cwd: RUNDIR, + stdio: ['ignore', 'pipe', 'pipe'], + env: { + ...process.env, + ORCA_GC_REPORT: GC_REPORT, + ORCA_RELAY_EMPTY_STARTUP_GRACE_MS: '3600000', + ORCA_RELAY_IDLE_GRACE_MS: '3600000' + } + } + ) + child.stderr.on('data', () => {}) + child.stdout.on('data', () => {}) + for (let i = 0; i < 150 && !existsSync(SOCK); i++) { + await sleep(100) + } + await sleep(500) + const pid = child.pid + const obs = await connect() + const rows = [] + + const m0 = await forcedGc(pid) + rows.push({ + cycle: 0, + ptysCreatedSoFar: 0, + connsSoFar: 1, + ptys: 0, + fds: fds(pid), + heapAfterGcMb: +(m0.heapUsed / 1048576).toFixed(3) + }) + + let created = 0 + for (let c = 1; c <= CYCLES; c++) { + const holder = await connect() + const ids = [] + for (let i = 0; i < PTYS; i++) { + const r = await rpc(holder, 'pty.spawn', { + cwd: RUNDIR, + cols: 80, + rows: 24, + shell: '/bin/sh', + args: [] + }) + ids.push(r.id) + created++ + } + holder.destroy() + await sleep(500) + for (const id of ids) { + await rpc(obs, 'pty.shutdown', { id }) + } + // poll to a real baseline rather than guessing a settle + for (let i = 0; i < 120; i++) { + const st = await rpc(obs, 'relay.status', {}) + if (st.ptys.active === 0) { + break + } + await sleep(500) + } + const m = await forcedGc(pid) + const st = await rpc(obs, 'relay.status', {}) + rows.push({ + cycle: c, + ptysCreatedSoFar: created, + connsSoFar: 1 + c, + ptys: st.ptys.active, + fds: fds(pid), + heapAfterGcMb: +(m.heapUsed / 1048576).toFixed(3) + }) + } + + console.log(JSON.stringify(rows, null, 1)) + obs.destroy() + child.kill('SIGKILL') + await sleep(300) + console.log(`RELAY_PID=${pid}`) +} + +main().catch((e) => { + console.error(`PROBE_FAILED: ${e.message}`) + process.exit(1) +}) diff --git a/config/scripts/relay-pty-retire-probe.mjs b/config/scripts/relay-pty-retire-probe.mjs new file mode 100644 index 00000000000..8a59a520eaa --- /dev/null +++ b/config/scripts/relay-pty-retire-probe.mjs @@ -0,0 +1,217 @@ +// HOW TO RUN: copy this file, relay-gc-host.cjs, the built `relay.js` and its `.version` into one +// directory on the target host (they resolve each other via import.meta.dirname), install a +// matching `node-pty` beside them, then `node `. Results are exact counts and +// post-collection heap figures; nothing here asserts on a duration. +// Focused follow-up: after pty.shutdown, poll until PTY state actually returns to baseline. +// The churn probe used a fixed 1.5s settle, which was too short to tell "not retired" from +// "not retired yet". This polls instead of guessing. +import { spawn } from 'node:child_process' +import net from 'node:net' +import { readFileSync, readdirSync, existsSync, rmSync, mkdirSync } from 'node:fs' +import { join } from 'node:path' + +const HERE = import.meta.dirname +const VERSION = readFileSync(join(HERE, '.version'), 'utf8').trim() +const RUNDIR = '/tmp/orca-pty-retire-probe' +const SOCK = join(RUNDIR, 'relay.sock') +const HEADER = 13 +const PTYS = 20 + +rmSync(RUNDIR, { recursive: true, force: true }) +mkdirSync(RUNDIR, { recursive: true }) + +const enc = (type, id, ack, payload) => { + const h = Buffer.alloc(HEADER) + h[0] = type + h.writeUInt32BE(id, 1) + h.writeUInt32BE(ack, 5) + h.writeUInt32BE(payload.length, 9) + return Buffer.concat([h, payload]) +} +const dec = (buf) => { + const out = [] + let off = 0 + while (buf.length - off >= HEADER) { + const len = buf.readUInt32BE(off + 9) + if (buf.length - off - HEADER < len) { + break + } + out.push({ type: buf[off], payload: buf.subarray(off + HEADER, off + HEADER + len) }) + off += HEADER + len + } + return { frames: out, rest: buf.subarray(off) } +} + +function connect() { + return new Promise((resolve, reject) => { + const sock = net.connect(SOCK) + let buf = Buffer.alloc(0) + sock.on('error', reject) + sock.on('data', (d) => { + buf = Buffer.concat([buf, d]) + const { frames, rest } = dec(buf) + buf = rest + for (const f of frames) { + if (f.type !== 2) { + continue + } + const m = JSON.parse(f.payload.toString()) + if (m.type === 'orca-relay-handshake-ok') { + sock.removeAllListeners('data') + sock._carry = buf + sock._seq = 0 + resolve(sock) + } else { + reject(new Error(m.type)) + } + } + }) + sock.on('connect', () => + sock.write( + enc( + 2, + 0, + 0, + Buffer.from(JSON.stringify({ type: 'orca-relay-handshake', version: VERSION })) + ) + ) + ) + }) +} + +let rpcId = 1 +function rpc(sock, method, params, timeoutMs = 20000) { + return new Promise((resolve, reject) => { + const id = rpcId++ + let buf = sock._carry ?? Buffer.alloc(0) + sock._carry = Buffer.alloc(0) + const timer = setTimeout(() => { + sock.off('data', onData) + reject(new Error(`${method} timed out`)) + }, timeoutMs) + const onData = (d) => { + buf = Buffer.concat([buf, d]) + const { frames, rest } = dec(buf) + buf = rest + for (const f of frames) { + if (f.type !== 1) { + continue + } + let m + try { + m = JSON.parse(f.payload.toString()) + } catch { + continue + } + if (m.id === id) { + clearTimeout(timer) + sock.off('data', onData) + sock._carry = buf + if (m.error) { + reject(new Error(JSON.stringify(m.error))) + } else { + resolve(m.result) + } + } + } + } + sock.on('data', onData) + sock.write( + enc(1, ++sock._seq, 0, Buffer.from(JSON.stringify({ jsonrpc: '2.0', id, method, params }))) + ) + }) +} + +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)) +const fds = (pid) => { + try { + return readdirSync(`/proc/${pid}/fd`).length + } catch { + return -1 + } +} + +async function main() { + const child = spawn( + process.execPath, + [ + join(HERE, 'relay.js'), + '--detached', + '--sock-path', + SOCK, + '--grace-time', + '0', + '--log-file', + join(RUNDIR, 'relay.log') + ], + { + cwd: RUNDIR, + stdio: ['ignore', 'pipe', 'pipe'], + env: { + ...process.env, + ORCA_RELAY_EMPTY_STARTUP_GRACE_MS: '3600000', + ORCA_RELAY_IDLE_GRACE_MS: '3600000' + } + } + ) + child.stderr.on('data', () => {}) + child.stdout.on('data', () => {}) + for (let i = 0; i < 150 && !existsSync(SOCK); i++) { + await sleep(100) + } + await sleep(500) + const pid = child.pid + const obs = await connect() + + const base = await rpc(obs, 'relay.status', {}) + const out = { baselineFds: fds(pid), baselinePtys: base.ptys.active } + + const holder = await connect() + const ids = [] + for (let i = 0; i < PTYS; i++) { + const r = await rpc(holder, 'pty.spawn', { + cwd: RUNDIR, + cols: 80, + rows: 24, + shell: '/bin/sh', + args: [] + }) + ids.push(r.id) + } + out.afterSpawn = { fds: fds(pid), ptys: (await rpc(obs, 'relay.status', {})).ptys.active } + + holder.destroy() + await sleep(1500) + out.afterUncleanDrop = { fds: fds(pid), ptys: (await rpc(obs, 'relay.status', {})).ptys.active } + + for (const id of ids) { + await rpc(obs, 'pty.shutdown', { id }) + } + + // Poll rather than assume a settle window. + const started = Date.now() + let polls = 0 + let ptys = -1 + let f = -1 + while (Date.now() - started < 60000) { + polls++ + ptys = (await rpc(obs, 'relay.status', {})).ptys.active + f = fds(pid) + if (ptys === 0 && f === out.baselineFds) { + break + } + await sleep(500) + } + out.afterShutdown = { fds: f, ptys, polls, elapsedMs: Date.now() - started } + + console.log(JSON.stringify(out, null, 1)) + obs.destroy() + child.kill('SIGKILL') + await sleep(300) + console.log(`RELAY_PID=${pid}`) +} + +main().catch((e) => { + console.error(`PROBE_FAILED: ${e.message}`) + process.exit(1) +}) From b45c6bbb9a57ee15c3668fc879ffac8d45baad8b Mon Sep 17 00:00:00 2001 From: Neil Date: Thu, 10 Sep 2026 20:47:22 -0700 Subject: [PATCH 2/2] test(relay): attribute the post-GC residual with a heap-snapshot diff Adds snapshot capture to the GC host, a constructor-level snapshot diff, and a capture probe, then uses them to settle what survives a forced collection across PTY churn. It is not relay state. Diffing cycle 2 against cycle 10 over 200 PTY create/destroy cycles, the entire +0.467 MB delta is V8: compiled code objects, TrustedByteArray, ProtectedFixedArray, WeakArrayList, FeedbackVector, LoadHandler. No application object appears in the growth at all -- no PTY record, no replay buffer, no relay map or set. The replay-buffer retention at pty-handler.ts:2150 was a plausible suspect and is not the cause. Extending to 30 cycles shows why 10 was not enough to say so: retained heap climbs 0.064 MB/cycle through the first decade, 0.017 through the second, and is flat across the last three (12.834 / 12.834 / 12.835). It also falls twice, which monotonic retention cannot do. The cycle count is env-overridable and defaults to 30 so the next reader does not repeat the short-window mistake. File descriptors stayed at exactly 25 for all 31 samples across 600 PTYs. The diff reports instance count and total self size per constructor rather than true retained size, which would need a dominator tree; a constructor whose instance count tracks the churn is the signal being looked for. --- config/scripts/relay-gc-host.cjs | 24 +- config/scripts/relay-heap-snapshot-diff.mjs | 88 +++++++ config/scripts/relay-pty-gc-cycles-probe.mjs | 5 +- .../scripts/relay-pty-heap-capture-probe.mjs | 236 ++++++++++++++++++ 4 files changed, 350 insertions(+), 3 deletions(-) create mode 100644 config/scripts/relay-heap-snapshot-diff.mjs create mode 100644 config/scripts/relay-pty-heap-capture-probe.mjs diff --git a/config/scripts/relay-gc-host.cjs b/config/scripts/relay-gc-host.cjs index 3c8395e30fb..eca72e5ab29 100644 --- a/config/scripts/relay-gc-host.cjs +++ b/config/scripts/relay-gc-host.cjs @@ -1,8 +1,14 @@ // Runs the real relay bundle under --expose-gc and reports post-collection memory on SIGUSR2. // Nothing about the relay changes: this only adds a signal handler before loading it. -const { writeFileSync } = require('node:fs') +// +// If ORCA_HEAP_SNAPSHOT_REQUEST names a file and that file exists when the signal arrives, its +// contents are read as a destination path and a heap snapshot is written there after the +// collection. The request file is consumed, so one request yields one snapshot. +const { writeFileSync, readFileSync, existsSync, unlinkSync } = require('node:fs') const report = process.env.ORCA_GC_REPORT +const snapshotRequest = process.env.ORCA_HEAP_SNAPSHOT_REQUEST + process.on('SIGUSR2', () => { try { // Twice: the first pass can resurrect via finalizers, the second settles it. @@ -11,8 +17,22 @@ process.on('SIGUSR2', () => { } catch { /* --expose-gc absent; report raw numbers so the caller can tell */ } + let snapshot = null try { - writeFileSync(report, JSON.stringify({ at: Date.now(), mem: process.memoryUsage() })) + if (snapshotRequest && existsSync(snapshotRequest)) { + const dest = readFileSync(snapshotRequest, 'utf8').trim() + unlinkSync(snapshotRequest) + if (dest) { + require('node:v8').writeHeapSnapshot(dest) + snapshot = dest + } + } + } catch { + /* a failed snapshot must not stop the memory report */ + } + try { + // Written last: the caller polls for this, so it must not appear before the snapshot is done. + writeFileSync(report, JSON.stringify({ at: Date.now(), mem: process.memoryUsage(), snapshot })) } catch { /* best-effort */ } diff --git a/config/scripts/relay-heap-snapshot-diff.mjs b/config/scripts/relay-heap-snapshot-diff.mjs new file mode 100644 index 00000000000..86a1606a453 --- /dev/null +++ b/config/scripts/relay-heap-snapshot-diff.mjs @@ -0,0 +1,88 @@ +// Aggregates a V8 heap snapshot by constructor and diffs two of them. +// +// Reports object COUNT and total SELF size per constructor, not true retained size -- retained size +// needs a dominator tree, and for finding what churn retains, a constructor whose instance count +// climbs with the cycle count is the signal. Counts are exact. +// +// Usage: node relay-heap-snapshot-diff.mjs [topN] +import { readFileSync } from 'node:fs' + +function aggregate(path) { + const snap = JSON.parse(readFileSync(path, 'utf8')) + const fields = snap.snapshot.meta.node_fields + const typeNames = snap.snapshot.meta.node_types[0] + const width = fields.length + const iType = fields.indexOf('type') + const iName = fields.indexOf('name') + const iSelf = fields.indexOf('self_size') + const nodes = snap.nodes + const strings = snap.strings + const byCtor = new Map() + let totalSelf = 0 + for (let off = 0; off < nodes.length; off += width) { + const type = typeNames[nodes[off + iType]] + const name = strings[nodes[off + iName]] + const self = nodes[off + iSelf] + totalSelf += self + // Key on type+name: "object/Foo" and "string" land in distinct buckets. + const key = `${type}/${name}` + const cur = byCtor.get(key) + if (cur) { + cur.count++ + cur.self += self + } else { + byCtor.set(key, { count: 1, self }) + } + } + return { byCtor, totalSelf, nodeCount: nodes.length / width } +} + +const [beforePath, afterPath, topRaw] = process.argv.slice(2) +if (!beforePath || !afterPath) { + console.error('usage: relay-heap-snapshot-diff.mjs [topN]') + process.exit(1) +} +const top = Number.parseInt(topRaw ?? '25', 10) + +const a = aggregate(beforePath) +const b = aggregate(afterPath) + +const keys = new Set([...a.byCtor.keys(), ...b.byCtor.keys()]) +const rows = [] +for (const key of keys) { + const x = a.byCtor.get(key) ?? { count: 0, self: 0 } + const y = b.byCtor.get(key) ?? { count: 0, self: 0 } + const dCount = y.count - x.count + const dSelf = y.self - x.self + if (dCount === 0 && dSelf === 0) { + continue + } + rows.push({ + key, + beforeCount: x.count, + afterCount: y.count, + dCount, + dSelfKb: +(dSelf / 1024).toFixed(1) + }) +} + +console.log( + JSON.stringify( + { + before: { nodeCount: a.nodeCount, totalSelfMb: +(a.totalSelf / 1048576).toFixed(3) }, + after: { nodeCount: b.nodeCount, totalSelfMb: +(b.totalSelf / 1048576).toFixed(3) }, + deltaSelfMb: +((b.totalSelf - a.totalSelf) / 1048576).toFixed(3), + deltaNodeCount: b.nodeCount - a.nodeCount + }, + null, + 1 + ) +) +console.log('\n--- top growth by self size ---') +for (const r of rows.sort((p, q) => q.dSelfKb - p.dSelfKb).slice(0, top)) { + console.log(JSON.stringify(r)) +} +console.log('\n--- top growth by instance count ---') +for (const r of rows.sort((p, q) => q.dCount - p.dCount).slice(0, top)) { + console.log(JSON.stringify(r)) +} diff --git a/config/scripts/relay-pty-gc-cycles-probe.mjs b/config/scripts/relay-pty-gc-cycles-probe.mjs index 963d0b85dd6..46035fcb774 100644 --- a/config/scripts/relay-pty-gc-cycles-probe.mjs +++ b/config/scripts/relay-pty-gc-cycles-probe.mjs @@ -17,7 +17,10 @@ const SOCK = join(RUNDIR, 'relay.sock') const GC_REPORT = join(RUNDIR, 'gc.json') const HEADER = 13 const PTYS = 20 -const CYCLES = 10 +// Why 30 and not 10: the first ~10 cycles are still inside V8's JIT warmup, where retained heap +// climbs about 0.064 MB/cycle and looks like a linear leak. It decays to ~0.017 MB/cycle over the +// second decade and is flat across the last three. Reading 10 cycles alone produces a false leak. +const CYCLES = Number.parseInt(process.env.ORCA_PROBE_CYCLES ?? '30', 10) rmSync(RUNDIR, { recursive: true, force: true }) mkdirSync(RUNDIR, { recursive: true }) diff --git a/config/scripts/relay-pty-heap-capture-probe.mjs b/config/scripts/relay-pty-heap-capture-probe.mjs new file mode 100644 index 00000000000..f2b45f3b55c --- /dev/null +++ b/config/scripts/relay-pty-heap-capture-probe.mjs @@ -0,0 +1,236 @@ +// HOW TO RUN: copy this file, relay-gc-host.cjs, the built `relay.js` and its `.version` into one +// directory on the target host, install a matching `node-pty` beside them, then `node `. +// +// Same cycle shape as relay-pty-gc-cycles-probe.mjs, but captures a heap snapshot after the forced +// collection at two chosen cycles so the residual can be attributed to a constructor rather than +// guessed at. Feed the two files to relay-heap-snapshot-diff.mjs. +import { spawn } from 'node:child_process' +import net from 'node:net' +import { readFileSync, readdirSync, existsSync, rmSync, mkdirSync, writeFileSync } from 'node:fs' +import { join } from 'node:path' + +const HERE = import.meta.dirname +const VERSION = readFileSync(join(HERE, '.version'), 'utf8').trim() +const RUNDIR = '/tmp/orca-pty-heap-capture' +const SOCK = join(RUNDIR, 'relay.sock') +const GC_REPORT = join(RUNDIR, 'gc.json') +const SNAP_REQUEST = join(RUNDIR, 'snapshot.request') +const HEADER = 13 +const PTYS = 20 +const CYCLES = 10 +const SNAPSHOT_AT = new Set([2, 10]) + +rmSync(RUNDIR, { recursive: true, force: true }) +mkdirSync(RUNDIR, { recursive: true }) + +const enc = (type, id, ack, payload) => { + const h = Buffer.alloc(HEADER) + h[0] = type + h.writeUInt32BE(id, 1) + h.writeUInt32BE(ack, 5) + h.writeUInt32BE(payload.length, 9) + return Buffer.concat([h, payload]) +} +const dec = (buf) => { + const out = [] + let off = 0 + while (buf.length - off >= HEADER) { + const len = buf.readUInt32BE(off + 9) + if (buf.length - off - HEADER < len) { + break + } + out.push({ type: buf[off], payload: buf.subarray(off + HEADER, off + HEADER + len) }) + off += HEADER + len + } + return { frames: out, rest: buf.subarray(off) } +} + +function connect() { + return new Promise((resolve, reject) => { + const sock = net.connect(SOCK) + let buf = Buffer.alloc(0) + sock.on('error', reject) + sock.on('data', (d) => { + buf = Buffer.concat([buf, d]) + const { frames, rest } = dec(buf) + buf = rest + for (const f of frames) { + if (f.type !== 2) { + continue + } + const m = JSON.parse(f.payload.toString()) + if (m.type === 'orca-relay-handshake-ok') { + sock.removeAllListeners('data') + sock._carry = buf + sock._seq = 0 + resolve(sock) + } else { + reject(new Error(m.type)) + } + } + }) + sock.on('connect', () => + sock.write( + enc( + 2, + 0, + 0, + Buffer.from(JSON.stringify({ type: 'orca-relay-handshake', version: VERSION })) + ) + ) + ) + }) +} + +let rpcId = 1 +function rpc(sock, method, params, timeoutMs = 25000) { + return new Promise((resolve, reject) => { + const id = rpcId++ + let buf = sock._carry ?? Buffer.alloc(0) + sock._carry = Buffer.alloc(0) + const timer = setTimeout(() => { + sock.off('data', onData) + reject(new Error(`${method} timed out`)) + }, timeoutMs) + const onData = (d) => { + buf = Buffer.concat([buf, d]) + const { frames, rest } = dec(buf) + buf = rest + for (const f of frames) { + if (f.type !== 1) { + continue + } + let m + try { + m = JSON.parse(f.payload.toString()) + } catch { + continue + } + if (m.id === id) { + clearTimeout(timer) + sock.off('data', onData) + sock._carry = buf + if (m.error) { + reject(new Error(JSON.stringify(m.error))) + } else { + resolve(m.result) + } + } + } + } + sock.on('data', onData) + sock.write( + enc(1, ++sock._seq, 0, Buffer.from(JSON.stringify({ jsonrpc: '2.0', id, method, params }))) + ) + }) +} + +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)) +const fds = (pid) => { + try { + return readdirSync(`/proc/${pid}/fd`).length + } catch { + return -1 + } +} + +async function forcedGc(pid, snapshotPath) { + writeFileSync(GC_REPORT, '') + if (snapshotPath) { + writeFileSync(SNAP_REQUEST, snapshotPath) + } + process.kill(pid, 'SIGUSR2') + // A snapshot write is slow; the report is written last so it gates on completion. + for (let i = 0; i < 600; i++) { + await sleep(200) + const raw = readFileSync(GC_REPORT, 'utf8') + if (raw.trim()) { + return JSON.parse(raw) + } + } + return null +} + +async function main() { + const child = spawn( + process.execPath, + [ + '--expose-gc', + join(HERE, 'gc-host.cjs'), + '--detached', + '--sock-path', + SOCK, + '--grace-time', + '0', + '--log-file', + join(RUNDIR, 'relay.log') + ], + { + cwd: RUNDIR, + stdio: ['ignore', 'pipe', 'pipe'], + env: { + ...process.env, + ORCA_GC_REPORT: GC_REPORT, + ORCA_HEAP_SNAPSHOT_REQUEST: SNAP_REQUEST, + ORCA_RELAY_EMPTY_STARTUP_GRACE_MS: '3600000', + ORCA_RELAY_IDLE_GRACE_MS: '3600000' + } + } + ) + child.stderr.on('data', () => {}) + child.stdout.on('data', () => {}) + for (let i = 0; i < 150 && !existsSync(SOCK); i++) { + await sleep(100) + } + await sleep(500) + const pid = child.pid + const obs = await connect() + const rows = [] + + for (let c = 1; c <= CYCLES; c++) { + const holder = await connect() + const ids = [] + for (let i = 0; i < PTYS; i++) { + const r = await rpc(holder, 'pty.spawn', { + cwd: RUNDIR, + cols: 80, + rows: 24, + shell: '/bin/sh', + args: [] + }) + ids.push(r.id) + } + holder.destroy() + await sleep(500) + for (const id of ids) { + await rpc(obs, 'pty.shutdown', { id }) + } + for (let i = 0; i < 120; i++) { + const st = await rpc(obs, 'relay.status', {}) + if (st.ptys.active === 0) { + break + } + await sleep(500) + } + const wantSnapshot = SNAPSHOT_AT.has(c) ? join(RUNDIR, `cycle-${c}.heapsnapshot`) : null + const rep = await forcedGc(pid, wantSnapshot) + rows.push({ + cycle: c, + ptysCreatedSoFar: c * PTYS, + fds: fds(pid), + heapAfterGcMb: +(rep.mem.heapUsed / 1048576).toFixed(3), + snapshot: rep.snapshot + }) + } + + console.log(JSON.stringify(rows, null, 1)) + obs.destroy() + child.kill('SIGKILL') + await sleep(300) + console.log(`RELAY_PID=${pid}`) +} + +main().catch((e) => { + console.error(`PROBE_FAILED: ${e.message}`) + process.exit(1) +})