Merge pull request #1640 from PR0M3TH3AN/ai/load-test-20260124-3194345285315307704

Add Load Testing Scripts
This commit is contained in:
thePR0M3TH3AN authored and GitHub committed 2026-01-24 14:37:08 -05:00
commit 3bfe8d664f
2 files changed
+438 -279

No files matched your search

+100
View File
@@ -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);
}
+338 -279
View File
@@ -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<id, startTime> }
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();