diff --git a/config/scripts/relay-gc-host.cjs b/config/scripts/relay-gc-host.cjs new file mode 100644 index 00000000000..eca72e5ab29 --- /dev/null +++ b/config/scripts/relay-gc-host.cjs @@ -0,0 +1,41 @@ +// 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. +// +// 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. + global.gc() + global.gc() + } catch { + /* --expose-gc absent; report raw numbers so the caller can tell */ + } + let snapshot = null + try { + 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 */ + } +}) + +require('./relay.js') 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-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..46035fcb774 --- /dev/null +++ b/config/scripts/relay-pty-gc-cycles-probe.mjs @@ -0,0 +1,247 @@ +// 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 +// 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 }) + +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-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) +}) 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) +})