diff --git a/scripts/agent/load-test-relay.mjs b/scripts/agent/load-test-relay.mjs new file mode 100644 index 00000000..bff55206 --- /dev/null +++ b/scripts/agent/load-test-relay.mjs @@ -0,0 +1,100 @@ +import { WebSocketServer } from "ws"; +import { matchFilters } from "nostr-tools/filter"; +import { fileURLToPath } from "url"; + +export function startRelay(port = 8889) { + const wss = new WebSocketServer({ port }); + const events = new Map(); // id -> event + const maxEvents = 10000; + const subs = new Map(); // ws -> Map(subId -> filters) + + console.log(`Load Test Relay starting on port ${port}`); + + wss.on("connection", (ws) => { + subs.set(ws, new Map()); + + ws.on("message", (message) => { + try { + const data = JSON.parse(message); + if (!Array.isArray(data)) return; + + const [type, ...payload] = data; + + if (type === "EVENT") { + const event = payload[0]; + if (!event || !event.id) { + ws.send(JSON.stringify(["NOTICE", "Invalid event"])); + return; + } + + if (!events.has(event.id)) { + // Eviction logic + if (events.size >= maxEvents) { + const firstKey = events.keys().next().value; + events.delete(firstKey); + } + + events.set(event.id, event); + + // Broadcast to matching subs + for (const [client, clientSubs] of subs.entries()) { + if (client.readyState !== 1) continue; // OPEN + for (const [subId, filters] of clientSubs.entries()) { + if (matchFilters(filters, event)) { + client.send(JSON.stringify(["EVENT", subId, event])); + } + } + } + } + + ws.send(JSON.stringify(["OK", event.id, true, "saved"])); + } else if (type === "REQ") { + const subId = payload[0]; + const filters = payload.slice(1); + + if (typeof subId !== "string") return; + + subs.get(ws).set(subId, filters); + + // Send stored events + for (const event of events.values()) { + if (matchFilters(filters, event)) { + ws.send(JSON.stringify(["EVENT", subId, event])); + } + } + ws.send(JSON.stringify(["EOSE", subId])); + } else if (type === "CLOSE") { + const subId = payload[0]; + if (typeof subId === "string") { + subs.get(ws).delete(subId); + } + } + } catch (err) { + // console.error("Relay error processing message:", err); + } + }); + + ws.on("close", () => { + subs.delete(ws); + }); + + ws.on("error", (err) => { + console.error("Relay connection error:", err); + }); + }); + + return { + close: () => { + return new Promise((resolve) => { + wss.close(resolve); + }); + }, + port + }; +} + +// Allow running standalone +if (process.argv[1] === fileURLToPath(import.meta.url)) { + const port = process.env.PORT ? parseInt(process.env.PORT) : 8889; + startRelay(port); +} diff --git a/scripts/agent/load-test.mjs b/scripts/agent/load-test.mjs index 600b4d41..5b0c9f4c 100644 --- a/scripts/agent/load-test.mjs +++ b/scripts/agent/load-test.mjs @@ -1,303 +1,362 @@ +import './setup-test-env.js'; +import { parseArgs } from 'node:util'; +import { spawn } from 'node:child_process'; +import { fileURLToPath } from 'url'; +import NodeWebSocket from 'ws'; +import * as NostrTools from 'nostr-tools'; +import fs from 'node:fs'; +import path from 'node:path'; +import { performance } from 'node:perf_hooks'; -import { WebSocket } from "ws"; -import { generateSecretKey, getPublicKey, finalizeEvent } from "nostr-tools"; -import { startRelay } from "./simple-relay.mjs"; -import fs from "fs"; -import path from "path"; -import { performance } from "perf_hooks"; +// Polyfill global NostrTools for app modules +global.NostrTools = NostrTools; -// Parse args -const args = process.argv.slice(2); -const getArg = (name, def) => { - const idx = args.indexOf(name); - return idx !== -1 ? args[idx + 1] : def; +// Import Schema Builders +import { buildVideoPostEvent, buildViewEvent } from '../../js/nostrEventSchemas.js'; + +// --- Configuration --- +const ARTIFACTS_DIR = 'artifacts'; +if (!fs.existsSync(ARTIFACTS_DIR)) { + fs.mkdirSync(ARTIFACTS_DIR, { recursive: true }); +} + +// Default Config +const DEFAULT_CONFIG = { + duration: 600, // seconds + clients: 1000, + rate: 0.1, // events per second per client? No, that would be 100 events/sec. + // Let's interpret 'rate' as TOTAL events per second to target, + // OR rate per client. + // If I have 1000 clients and rate 0.1, that's 100 events/sec. That's reasonable. + relay: 'ws://localhost:8889', }; -const DURATION_MS = parseInt(getArg("--duration", "60000")); // 1 minute -const N_CLIENTS = parseInt(getArg("--clients", "100")); // Default 100 for safety -const RELAY_URL = getArg("--relay", null); -const RATE_PER_CLIENT = parseFloat(getArg("--rate", "0.1")); // events/sec +// Parse Args +const { values: args } = parseArgs({ + options: { + duration: { type: 'string', short: 'd' }, + clients: { type: 'string', short: 'c' }, + rate: { type: 'string', short: 'r' }, + relay: { type: 'string', short: 'u' }, // u for url + }, +}); -console.log(`Load Test Config: - Duration: ${DURATION_MS}ms - Clients: ${N_CLIENTS} - Relay: ${RELAY_URL || "Internal (localhost:8888)"} - Rate: ${RATE_PER_CLIENT} ev/s/client -`); +const config = { + duration: args.duration ? parseInt(args.duration) : DEFAULT_CONFIG.duration, + clients: args.clients ? parseInt(args.clients) : DEFAULT_CONFIG.clients, + rate: args.rate ? parseFloat(args.rate) : DEFAULT_CONFIG.rate, + relay: args.relay || DEFAULT_CONFIG.relay, +}; -async function main() { - let relayServer = null; - let targetUrl = RELAY_URL; +console.log('--- Load Test Configuration ---'); +console.log(JSON.stringify(config, null, 2)); - if (!targetUrl) { - relayServer = startRelay(8888); - targetUrl = "ws://localhost:8888"; +// --- State --- +const activeClients = []; // { ws, sk, pk, pendingEvents: Map } +let relayProcess = null; +const metrics = { + latencies: [], + errors: 0, + sent: 0, + received: 0, + errorCounts: {}, + operationTimes: { + build: [], + sign: [], + }, + resourceUsage: [], +}; + +// --- Helpers --- + +async function startRelay() { + console.log('[Setup] Starting local relay...'); + const relayLog = fs.openSync(path.join(ARTIFACTS_DIR, 'load-relay.log'), 'w'); + + // Assuming the relay script is in the same directory + const relayScript = path.join(path.dirname(fileURLToPath(import.meta.url)), 'load-test-relay.mjs'); + + const port = new URL(config.relay).port || '8889'; + + relayProcess = spawn('node', [relayScript], { + stdio: ['ignore', relayLog, relayLog], + env: { ...process.env, PORT: port } + }); + + // Wait for port + await waitForPort(parseInt(port)); + console.log('[Setup] Relay started.'); +} + +async function waitForPort(port) { + const retryInterval = 200; + const maxRetries = 50; // 10 seconds + for (let i = 0; i < maxRetries; i++) { + try { + const controller = new AbortController(); + setTimeout(() => controller.abort(), 100); + await fetch(`http://127.0.0.1:${port}`, { signal: controller.signal }).catch(() => {}); + + await new Promise((resolve, reject) => { + const ws = new NodeWebSocket(`ws://127.0.0.1:${port}`); + ws.on('open', () => { ws.close(); resolve(); }); + ws.on('error', reject); + }); + return; + } catch (e) { + // console.log('Retrying connection...', e.message); + await new Promise(r => setTimeout(r, retryInterval)); + } } + throw new Error(`Relay did not start on port ${port}`); +} - // Setup metrics - const latencies = []; - let sentCount = 0; - let okCount = 0; - let errorCount = 0; - const errors = {}; - - // Hot functions profiling - const buildTimes = []; - const signTimes = []; - - const clients = []; - - // Start clients - console.log("Connecting clients..."); - for (let i = 0; i < N_CLIENTS; i++) { - clients.push(createClient(targetUrl, i)); - } - - await Promise.all(clients.map((c) => c.connect())); - console.log("All clients connected."); - - // Start load - console.log("Starting load..."); - const startTime = Date.now(); - const interval = setInterval(() => { - reportMetrics(); - }, 5000); - - clients.forEach((c) => c.startLoad()); - - // Wait for duration - await new Promise((resolve) => setTimeout(resolve, DURATION_MS)); - - // Stop load - console.log("Stopping load..."); - clearInterval(interval); - clients.forEach((c) => c.stopLoad()); - - // Wait a bit for pending OKs - await new Promise((resolve) => setTimeout(resolve, 2000)); - - // Disconnect - clients.forEach((c) => c.close()); - if (relayServer) { - await relayServer.close(); - } - - // Generate Report - const endTime = Date.now(); - const actualDuration = (endTime - startTime) / 1000; - - latencies.sort((a, b) => a - b); - const avgLatency = - latencies.reduce((a, b) => a + b, 0) / latencies.length || 0; - const p95Latency = latencies[Math.floor(latencies.length * 0.95)] || 0; - const p99Latency = latencies[Math.floor(latencies.length * 0.99)] || 0; - - const avgBuildTime = buildTimes.reduce((a, b) => a + b, 0) / buildTimes.length || 0; - const avgSignTime = signTimes.reduce((a, b) => a + b, 0) / signTimes.length || 0; - - const report = { - timestamp: new Date().toISOString(), - config: { - durationMs: DURATION_MS, - clients: N_CLIENTS, - relay: targetUrl, - rate: RATE_PER_CLIENT - }, - metrics: { - totalSent: sentCount, - totalOk: okCount, - totalErrors: errorCount, - throughput: okCount / actualDuration, - latency: { - avg: avgLatency, - p95: p95Latency, - p99: p99Latency, - min: latencies[0] || 0, - max: latencies[latencies.length - 1] || 0 - }, - resources: { - cpu: process.cpuUsage(), - memory: process.memoryUsage() - } - }, - hot_functions: [ - { name: 'buildEvent', avg_ms: avgBuildTime }, - { name: 'signEvent', avg_ms: avgSignTime } - ], - errors: errors, - recommendations: [] +function createClient() { + const sk = NostrTools.generateSecretKey(); + const pk = NostrTools.getPublicKey(sk); + const client = { + sk, + pk, + ws: new NodeWebSocket(config.relay), + pendingEvents: new Map(), }; - if (p95Latency > 200) { - report.recommendations.push( - "High latency detected (>200ms p95). Consider scaling relay or reducing batch size." - ); - } - if (errorCount > 0) { - report.recommendations.push( - "Errors detected. Check error logs and relay stability." - ); - } - if (report.metrics.throughput < N_CLIENTS * RATE_PER_CLIENT * 0.8) { - report.recommendations.push( - "Throughput is significantly lower than target rate. Possible bottleneck." - ); - } - if (avgSignTime > 5) { // 5ms per signature is quite high for main thread loop - report.recommendations.push( - "Cryptographic bottleneck detected: signEvent avg > 5ms. Consider offloading to workers." - ); - } + return new Promise((resolve) => { + client.ws.on('open', () => { + resolve(client); + }); + client.ws.on('message', (data) => { + try { + const msg = JSON.parse(data); + if (msg[0] === 'OK') { + const eventId = msg[1]; + const accepted = msg[2]; + const reason = msg[3]; - const artifactsDir = path.resolve("artifacts"); - if (!fs.existsSync(artifactsDir)) { - fs.mkdirSync(artifactsDir); - } + if (client.pendingEvents.has(eventId)) { + const startTime = client.pendingEvents.get(eventId); + const duration = performance.now() - startTime; + metrics.latencies.push(duration); + metrics.received++; + client.pendingEvents.delete(eventId); - const dateStr = new Date().toISOString().split("T")[0].replace(/-/g, ""); - const reportPath = path.join(artifactsDir, `load-report-${dateStr}.json`); - - fs.writeFileSync(reportPath, JSON.stringify(report, null, 2)); - console.log(`Report written to ${reportPath}`); - - // Exit - process.exit(0); - - function reportMetrics() { - const mem = process.memoryUsage(); - console.log( - `[Status] Sent: ${sentCount}, OK: ${okCount}, Errs: ${errorCount}, Heap: ${(mem.heapUsed / 1024 / 1024).toFixed(2)}MB` - ); - } - - function createClient(url, index) { - let ws; - let timer; - let connected = false; - const sk = generateSecretKey(); - const pk = getPublicKey(sk); - const pending = new Map(); // eventId -> timestamp - - return { - connect: () => - new Promise((resolve) => { - ws = new WebSocket(url); - ws.on("open", () => { - connected = true; - resolve(); - }); - ws.on("message", (data) => { - try { - const msg = JSON.parse(data); - if (msg[0] === "OK") { - const eventId = msg[1]; - const ok = msg[2]; - const reason = msg[3]; - - if (pending.has(eventId)) { - const start = pending.get(eventId); - const lat = Date.now() - start; - latencies.push(lat); - pending.delete(eventId); - if (ok) okCount++; - else { - errorCount++; - errors[reason] = (errors[reason] || 0) + 1; - } - } - } else if (msg[0] === "NOTICE") { - errorCount++; - errors["NOTICE: " + msg[1]] = - (errors["NOTICE: " + msg[1]] || 0) + 1; - } - } catch (e) { - // ignore + if (!accepted) { + metrics.errors++; + metrics.errorCounts[reason] = (metrics.errorCounts[reason] || 0) + 1; } - }); - ws.on("error", (e) => { - errorCount++; - errors[e.message] = (errors[e.message] || 0) + 1; - resolve(); // resolve anyway to not block - }); - ws.on("close", () => { - connected = false; - }); - }), - - startLoad: () => { - if (!connected) return; - const intervalMs = 1000 / RATE_PER_CLIENT; - timer = setInterval(() => { - if (ws.readyState === WebSocket.OPEN) { - const isLarge = Math.random() > 0.8; // 20% large events - const event = isLarge - ? createVideoPost(pk, sk) - : createViewEvent(pk, sk); - - pending.set(event.id, Date.now()); - ws.send(JSON.stringify(["EVENT", event])); - sentCount++; } - }, intervalMs); - }, - - stopLoad: () => { - if (timer) clearInterval(timer); - }, - - close: () => { - if (ws) ws.close(); + } + } catch (e) { + // ignore malformed } + }); + client.ws.on('error', (err) => { + metrics.errors++; + const msg = err.message || 'connection_error'; + metrics.errorCounts[msg] = (metrics.errorCounts[msg] || 0) + 1; + }); + }); +} + +// --- Main --- + +async function run() { + try { + // 1. Start Relay + await startRelay(); + + // 2. Init Clients + console.log(`[Setup] Connecting ${config.clients} clients...`); + const connections = []; + for (let i = 0; i < config.clients; i++) { + connections.push(createClient()); + if (i % 100 === 0 && i > 0) process.stdout.write('.'); + } + const results = await Promise.all(connections); + activeClients.push(...results); + console.log('\n[Setup] Clients connected.'); + + // 3. Load Loop + console.log('[Load] Starting load generation...'); + const startTime = performance.now(); + const endTime = startTime + (config.duration * 1000); + + // Total expected events per second + const totalRate = config.clients * config.rate; + const intervalMs = 1000 / totalRate; + + console.log(`[Load] Target rate: ${totalRate.toFixed(2)} events/sec`); + + let running = true; + + // Monitoring Loop + const monitorInterval = setInterval(() => { + const cpu = process.cpuUsage(); + const mem = process.memoryUsage(); + metrics.resourceUsage.push({ + timestamp: Date.now(), + cpuUser: cpu.user, + cpuSystem: cpu.system, + rss: mem.rss, + heapUsed: mem.heapUsed, + }); + console.log(`[Monitor] Sent: ${metrics.sent}, Recv: ${metrics.received}, Errors: ${metrics.errors}, Latency p50: ${calculatePercentile(metrics.latencies, 50).toFixed(2)}ms`); + }, 5000); + + // Event Loop + const loadLoop = async () => { + while (performance.now() < endTime && running) { + const iterationStart = performance.now(); + + // Pick random client + const clientIndex = Math.floor(Math.random() * activeClients.length); + const client = activeClients[clientIndex]; + + // Pick event type (10% heavy, 90% light) + const isHeavy = Math.random() < 0.1; + + try { + let event; + const buildStart = performance.now(); + const now = Math.floor(Date.now() / 1000); + const hexPk = client.pk; // getPublicKey returns hex in v2 + + if (isHeavy) { + // Video Post + const content = { + version: 3, + title: `Load Test Video ${Date.now()}`, + description: 'A description for load testing purposes. '.repeat(50), // make it larger + magnet: `magnet:?xt=urn:btih:${Math.random().toString(16).slice(2).repeat(10)}`, + mode: 'live', + videoRootId: `load-${Date.now()}-${Math.random()}`, + isNsfw: false, + isForKids: false, + }; + event = buildVideoPostEvent({ + pubkey: hexPk, + created_at: now, + content, + dTagValue: content.videoRootId + }); + } else { + // View Event + event = buildViewEvent({ + pubkey: hexPk, + created_at: now, + pointerValue: `load-video-${Math.random()}`, + pointerTag: ['d', `load-video-${Math.random()}`] + }); + } + metrics.operationTimes.build.push(performance.now() - buildStart); + + // Sign + const signStart = performance.now(); + // We need to pass event to finalizeEvent. + // Note: build*Event returns a plain object. + // nostr-tools v2 finalizeEvent takes (t, secretKey) + const signedEvent = NostrTools.finalizeEvent(event, client.sk); + metrics.operationTimes.sign.push(performance.now() - signStart); + + // Publish + client.pendingEvents.set(signedEvent.id, performance.now()); + client.ws.send(JSON.stringify(['EVENT', signedEvent])); + metrics.sent++; + + } catch (e) { + console.error('Error generating/sending event:', e); + metrics.errors++; + } + + // Throttle + const elapsed = performance.now() - iterationStart; + const delay = Math.max(0, intervalMs - elapsed); + if (delay > 0) await new Promise(r => setTimeout(r, delay)); + else { + // Yield to event loop if we are falling behind + await new Promise(r => setImmediate(r)); + } + } + running = false; }; - } - function createViewEvent(pk, sk) { - const startBuild = performance.now(); - const event = { - kind: 30079, - created_at: Math.floor(Date.now() / 1000), - tags: [["d", "test-view"]], - content: "", - pubkey: pk - }; - const endBuild = performance.now(); - buildTimes.push(endBuild - startBuild); + await loadLoop(); - const startSign = performance.now(); - const signed = finalizeEvent(event, sk); - const endSign = performance.now(); - signTimes.push(endSign - startSign); + clearInterval(monitorInterval); + console.log('[Load] Test finished.'); - return signed; - } + // 4. Report + generateReport(); - function createVideoPost(pk, sk) { - const startBuild = performance.now(); - const event = { - kind: 30078, - created_at: Math.floor(Date.now() / 1000), - tags: [ - ["d", "video-1"], - ["title", "Test Video"], - ["t", "video"] - ], - content: JSON.stringify({ - title: "Test Video", - description: "A large description ".repeat(100), - url: "https://example.com/video.mp4" - }), - pubkey: pk - }; - const endBuild = performance.now(); - buildTimes.push(endBuild - startBuild); - - const startSign = performance.now(); - const signed = finalizeEvent(event, sk); - const endSign = performance.now(); - signTimes.push(endSign - startSign); - - return signed; + } catch (err) { + console.error('Fatal error:', err); + } finally { + // Teardown + console.log('[Teardown] Closing clients...'); + for (const c of activeClients) { + c.ws.terminate(); + } + if (relayProcess) { + console.log('[Teardown] Killing relay...'); + relayProcess.kill(); + } } } -main().catch(console.error); +function calculatePercentile(data, percentile) { + if (data.length === 0) return 0; + data.sort((a, b) => a - b); + const index = Math.floor(data.length * (percentile / 100)); + return data[index]; +} + +function generateReport() { + const report = { + timestamp: new Date().toISOString(), + config, + metrics: { + total_sent: metrics.sent, + total_received: metrics.received, + errors: metrics.errors, + throughput_sent: metrics.sent / config.duration, + throughput_recv: metrics.received / config.duration, + latency_ms: { + p50: calculatePercentile(metrics.latencies, 50), + p95: calculatePercentile(metrics.latencies, 95), + p99: calculatePercentile(metrics.latencies, 99), + max: calculatePercentile(metrics.latencies, 100), + }, + operation_times_ms: { + build_avg: metrics.operationTimes.build.reduce((a, b) => a + b, 0) / metrics.operationTimes.build.length || 0, + sign_avg: metrics.operationTimes.sign.reduce((a, b) => a + b, 0) / metrics.operationTimes.sign.length || 0, + }, + resource_usage: metrics.resourceUsage, + error_breakdown: metrics.errorCounts, + }, + bottlenecks: [], + remediation: [] + }; + + // Simple heuristics for bottlenecks + if (report.metrics.latency_ms.p99 > 1000) { + report.bottlenecks.push('High P99 Latency (>1s)'); + report.remediation.push('Relay might be overloaded or network saturated. Investigate relay event loop lag.'); + } + if (report.metrics.errors > 0) { + report.bottlenecks.push('Errors detected'); + report.remediation.push('Check error breakdown.'); + } + + // Hot functions proxy + if (report.metrics.operation_times_ms.sign_avg > 10) { + report.bottlenecks.push('Signing is slow'); + report.remediation.push('Optimize signing or offload to worker.'); + } + + const filename = `load-report-${new Date().toISOString().replace(/[:.]/g, '-')}.json`; + const reportPath = path.join(ARTIFACTS_DIR, filename); + fs.writeFileSync(reportPath, JSON.stringify(report, null, 2)); + console.log(`[Report] Saved to ${reportPath}`); +} + +run();