Fail capacity profiles with unsynced captures and retain cleanup recovery

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;

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<void>(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<void>(resolve => server.close(() => resolve()));
+			rmSync(home, { recursive: true, force: true });
+		}
+	}, 55000);
+});