From 187b8f91d34f88e087af04994974a696759f2564 Mon Sep 17 00:00:00 2001 From: Josh Winters Date: Fri, 6 Nov 2026 18:50:55 -0500 Subject: [PATCH] Require sync convergence and preserve failed profiler cleanup --- scripts/profile-hot-sync.mjs | 24 +++++++++++--- tests/runtime/capacity-profile.test.ts | 46 ++++++++++++++++++++++++++ 2 files changed, 66 insertions(+), 4 deletions(-) create mode 100644 tests/runtime/capacity-profile.test.ts diff --git a/scripts/profile-hot-sync.mjs b/scripts/profile-hot-sync.mjs index c34f012..a2778d4 100644 --- a/scripts/profile-hot-sync.mjs +++ b/scripts/profile-hot-sync.mjs @@ -1,7 +1,7 @@ #!/usr/bin/env node // Bounded workload using the shipped manager, disposable identities and synthetic findings. import { createRequire } from 'node:module'; -import { mkdtempSync, rmSync } from 'node:fs'; +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'; import os from 'node:os'; import path from 'node:path'; import { performance } from 'node:perf_hooks'; @@ -15,7 +15,9 @@ if (!process.argv.includes('--allow-production') && !['localhost', '127.0.0.1']. const count = Number(option('--clients', '6')), seconds = Number(option('--seconds', '65')), captureMs = Number(option('--capture-ms', '3000')); if (!Number.isSafeInteger(count) || count < 1 || count > 10 || seconds < 10 || seconds > 180 || captureMs < 1500) throw new Error('Budget: 1–10 clients, 10–180 seconds, captures at least 1500 ms apart'); const root = mkdtempSync(path.join(os.tmpdir(), 'crosshead-sync-profile-')); -const clients = [], managers = [], observations = [], captureLatencies = []; +const clients = [], managers = [], identities = [], observations = [], captureLatencies = []; +const recovery = path.join(root, 'owned-recovery.json'); +const saveRecovery = () => writeFileSync(recovery, JSON.stringify({ origin, root, identities: identities.map((identity, i) => ({ ...identity, session: clients[i]?.session() })) }) + '\n', { mode: 0o600 }); const originalFetch = globalThis.fetch; let abort = null, phase = 'workload'; globalThis.fetch = async (...args) => { @@ -33,8 +35,10 @@ try { for (let i = 0; i < count; i++) { const key = generateSecretKey(), identity = { nsec: nip19.nsecEncode(key), npub: nip19.npubEncode(getPublicKey(key)) }; const client = new HostedOnlyBotsClient({ apiBase: origin, ...identity }); + identities.push(identity); summary.npubs.push(identity.npub); clients.push(client); managers.push(new HotSyncManager({ stateDir: path.join(root, String(i)), origin, ...identity, profileName: 'launch-capacity-fixture' }, client)); + saveRecovery(); } const start = performance.now(); let nextCapture = 0, event = 0; while (!abort && performance.now() - start < seconds*1000) { @@ -49,13 +53,22 @@ try { for (const manager of managers) manager.tick(); await sleep(100); } + // A transport failure can be swallowed by the background tick. Success + // requires the captures to reach the server, not merely the absence of 5xx. + const convergeUntil = performance.now() + 10_000; + while (!abort && performance.now() < convergeUntil && managers.some(m => m.status().pendingCount || m.status().state !== 'synced')) { + for (const manager of managers) manager.tick(); + await sleep(100); + } summary.elapsedSeconds = Number(((performance.now()-start)/1000).toFixed(2)); summary.abort = abort; summary.captures = captureLatencies.length; summary.captureP95Ms = Number(quantile(captureLatencies,.95).toFixed(2)); - summary.devices = managers.map(m => { const s = m.status(); return { state: s.state, pending: s.pendingCount, error: s.error }; }); + summary.devices = managers.map(m => { const s = m.status(); return { state: s.state, pending: s.pendingCount, error: s.error, discoveryCompleted: !s.initialSyncPending }; }); + if (!abort && summary.devices.some(s => s.state !== 'synced' || s.pending !== 0 || s.error || !s.discoveryCompleted)) summary.abort = { reason: 'sync_did_not_converge' }; } finally { phase = 'cleanup'; + saveRecovery(); for (const [device,result] of (await Promise.allSettled(managers.map(m => m.stop()))).entries()) if(result.status === "rejected") summary.cleanup.push({device,stopFailed:true}); for (const [i, manager] of managers.entries()) { const client = clients[i]; @@ -73,7 +86,10 @@ try { } summary.requests = Object.values(observations.reduce((all, x) => { const key = `${x.phase} ${x.method} ${x.route}`; const row = all[key] ??= { route: key, count: 0, statuses: {}, latencies: [] }; row.count++; row.statuses[x.status] = (row.statuses[x.status] ?? 0)+1; row.latencies.push(x.ms); return all; }, {})).map(({latencies, ...row}) => ({ ...row, p95Ms: Math.round(quantile(latencies,.95)) })); summary.finishedAt = new Date().toISOString(); - globalThis.fetch = originalFetch; rmSync(root, { recursive: true, force: true }); + globalThis.fetch = originalFetch; + if (summary.cleanup.some(x => x.scopeCleared === false || x.sessionRevoked === false || x.stopFailed)) { + saveRecovery(); summary.retainedRecovery = recovery; + } else rmSync(root, { recursive: true, force: true }); console.log(JSON.stringify(summary, null, 2)); } if (summary.abort || summary.cleanup.some(x => x.scopeCleared === false || x.sessionRevoked === false || x.stopFailed)) process.exitCode = 2; diff --git a/tests/runtime/capacity-profile.test.ts b/tests/runtime/capacity-profile.test.ts new file mode 100644 index 0000000..f83a7d2 --- /dev/null +++ b/tests/runtime/capacity-profile.test.ts @@ -0,0 +1,46 @@ +import { describe, expect, it } from "vitest"; +import { createServer } from "node:http"; +import { spawn } from "node:child_process"; +import { mkdtempSync, readFileSync, rmSync, statSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; + +describe("capacity profiler acceptance", () => { + it("fails on swallowed transport errors and retains private recovery after failed cleanup", async () => { + const home = mkdtempSync(path.join(tmpdir(), "xhd-profile-test-")); + const server = createServer((_request, response) => response.destroy()); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + const port = (server.address() as { port: number }).port; + const child = spawn(process.execPath, ["scripts/profile-hot-sync.mjs", "--base-url", `http://127.0.0.1:${port}`, "--clients", "1", "--seconds", "10", "--capture-ms", "2000"], { + env: { PATH: process.env.PATH, TMPDIR: home, TEMP: home, TMP: home }, stdio: ["ignore", "pipe", "pipe"], + }); + let stdout = "", stderr = ""; + child.stdout!.on("data", data => stdout += data.toString()); + child.stderr!.on("data", data => stderr += data.toString()); + const ended = new Promise<{ code: number | null; signal: string | null }>((resolve, reject) => { + child.once("close", (code, signal) => resolve({ code, signal })); + child.once("error", reject); + }); + const timer = setTimeout(() => child.kill("SIGKILL"), 45000); + try { + const result = await ended; + expect(result, stderr).toEqual({ code: 2, signal: null }); + const report = JSON.parse(stdout); + expect(report.abort).toEqual({ reason: "sync_did_not_converge" }); + expect(report.devices[0].pending).toBeGreaterThan(0); + expect(report.cleanup.some((entry: { scopeCleared?: boolean }) => entry.scopeCleared === false)).toBe(true); + expect(path.relative(home, report.retainedRecovery).startsWith("..")).toBe(false); + const recovery = JSON.parse(readFileSync(report.retainedRecovery, "utf8")); + expect(recovery.identities[0].npub).toBe(report.npubs[0]); + expect(recovery.identities[0].nsec).toMatch(/^nsec1/); + expect(statSync(report.retainedRecovery).mode & 0o777).toBe(0o600); + expect(stdout).not.toContain(recovery.identities[0].nsec); + } finally { + clearTimeout(timer); + if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL"); + await ended; + await new Promise(resolve => server.close(() => resolve())); + rmSync(home, { recursive: true, force: true }); + } + }, 55000); +});