test(loadtest): 100-guest / 1000-image stress harness
HTTP-level load driver simulating ~100 guests uploading ~1000 images in bursts over a window, plus SSE viewers and one real browser on /diashow. Correlates upload→upload-processed (pipeline latency), waits for the compression backlog to drain against DB ground truth, and emits per-status/latency metrics with pass/fail flags. Includes a realistic-JPEG generator and a diashow-SSE regression check (confirm-diashow-fix.mjs). Run artifacts are gitignored. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
MechaCat02
620
e2e/loadtest/driver.mjs
Normal file
620
e2e/loadtest/driver.mjs
Normal file
@@ -0,0 +1,620 @@
|
||||
#!/usr/bin/env node
|
||||
/**
|
||||
* EventSnap load / stress driver.
|
||||
*
|
||||
* Simulates ~100 guests joining an event and uploading ~1000 images in bursts
|
||||
* (10-20 at a time) spread across a compressed time window, plus a pool of
|
||||
* viewers holding live SSE connections (like phones with the feed open) and one
|
||||
* "diashow" connection (the showcase display). Measures the metrics that matter
|
||||
* for a live event — especially the *pipeline backlog*: how long between an
|
||||
* upload succeeding and its preview being ready (SSE `upload-processed`), which
|
||||
* is exactly "how long until the photo shows up on the diashow".
|
||||
*
|
||||
* This is an HTTP-level driver on purpose: 100 real browsers would bottleneck
|
||||
* the test box, not the server. One real browser watches /diashow separately
|
||||
* (see diashow-watch.mjs).
|
||||
*
|
||||
* METHODOLOGY NOTE — what we change vs. the shipping config:
|
||||
* We ONLY disable rate limits. They are per-IP/per-user anti-abuse guards; a
|
||||
* synthetic test from one IP would trip them in a way real guests (distinct
|
||||
* IPs, phones) never would, so leaving them on would measure the rate limiter
|
||||
* instead of the pipeline. We DELIBERATELY leave compression concurrency,
|
||||
* DB pool size and quotas at their real defaults — validating those is the
|
||||
* whole point. (Runtime can't change compression concurrency anyway; it's a
|
||||
* boot-time env var.)
|
||||
*
|
||||
* Related real-world finding to keep in mind: the default upload_rate_per_hour
|
||||
* is 10, so a real guest uploading a burst of 10-20 would be throttled by the
|
||||
* SHIPPING config too. That's a genuine event-day issue worth its own report,
|
||||
* independent of this pipeline test.
|
||||
*
|
||||
* Usage:
|
||||
* node driver.mjs # full run (100 guests / 1000 imgs / 15 min)
|
||||
* LT_GUESTS=5 LT_IMAGES=50 LT_WINDOW_SEC=60 node driver.mjs # smoke
|
||||
*
|
||||
* Env knobs (all optional; defaults target the full run):
|
||||
* LT_BASE http://localhost:3101 frontend/caddy base URL
|
||||
* LT_GUESTS 100 number of virtual guests
|
||||
* LT_IMAGES 1000 total uploads to perform
|
||||
* LT_WINDOW_SEC 900 spread uploads over this window
|
||||
* LT_BURST_MIN/MAX 10 / 20 images per burst
|
||||
* LT_BURST_CONC 3 parallel uploads within one burst (phone-like)
|
||||
* LT_VIEWERS 20 extra guests holding SSE + polling feed
|
||||
* LT_PHOTOS_DIR /tmp/eventsnap-loadtest/photos
|
||||
* LT_TRUNCATE 1 wipe event data before the run
|
||||
* LT_DRAIN_TIMEOUT_SEC 600 max wait for compression backlog to drain
|
||||
* LT_APP_CONTAINER e2e-app-1 docker container for CPU/mem sampling
|
||||
* LT_DB_CONTAINER e2e-db-1 docker container for psql ground-truth
|
||||
* LT_ADMIN_PW admin-test-pw
|
||||
* LT_KEEP_RATELIMITS 0 set 1 to NOT disable rate limits
|
||||
* LT_OUT_DIR ./results
|
||||
*/
|
||||
|
||||
import { readFileSync, readdirSync, writeFileSync, mkdirSync, statSync } from 'node:fs';
|
||||
import { execFile } from 'node:child_process';
|
||||
import { promisify } from 'node:util';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
import { dirname, join } from 'node:path';
|
||||
|
||||
const execFileAsync = promisify(execFile);
|
||||
const __dirname = dirname(fileURLToPath(import.meta.url));
|
||||
|
||||
// ── Config ──────────────────────────────────────────────────────────────────
|
||||
const cfg = {
|
||||
base: process.env.LT_BASE ?? 'http://localhost:3101',
|
||||
guests: int('LT_GUESTS', 100),
|
||||
images: int('LT_IMAGES', 1000),
|
||||
windowSec: int('LT_WINDOW_SEC', 900),
|
||||
burstMin: int('LT_BURST_MIN', 10),
|
||||
burstMax: int('LT_BURST_MAX', 20),
|
||||
burstConc: int('LT_BURST_CONC', 3),
|
||||
viewers: int('LT_VIEWERS', 20),
|
||||
photosDir: process.env.LT_PHOTOS_DIR ?? '/tmp/eventsnap-loadtest/photos',
|
||||
truncate: process.env.LT_TRUNCATE !== '0',
|
||||
drainTimeoutSec: int('LT_DRAIN_TIMEOUT_SEC', 600),
|
||||
appContainer: process.env.LT_APP_CONTAINER ?? 'e2e-app-1',
|
||||
dbContainer: process.env.LT_DB_CONTAINER ?? 'e2e-db-1',
|
||||
adminPw: process.env.LT_ADMIN_PW ?? 'admin-test-pw',
|
||||
keepRateLimits: process.env.LT_KEEP_RATELIMITS === '1',
|
||||
outDir: process.env.LT_OUT_DIR ?? join(__dirname, 'results'),
|
||||
};
|
||||
|
||||
function int(name, def) {
|
||||
const v = process.env[name];
|
||||
return v === undefined ? def : parseInt(v, 10);
|
||||
}
|
||||
|
||||
const API = `${cfg.base}/api/v1`;
|
||||
const now = () => Date.now();
|
||||
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
|
||||
const rand = (min, max) => min + Math.floor(Math.random() * (max - min + 1));
|
||||
const pick = (arr) => arr[Math.floor(Math.random() * arr.length)];
|
||||
|
||||
// ── HTTP helpers ──────────────────────────────────────────────────────────────
|
||||
async function api(path, { method = 'GET', token, json, expect } = {}) {
|
||||
const headers = {};
|
||||
if (token) headers.Authorization = `Bearer ${token}`;
|
||||
if (json !== undefined) headers['Content-Type'] = 'application/json';
|
||||
const res = await fetch(`${API}${path}`, {
|
||||
method,
|
||||
headers,
|
||||
body: json !== undefined ? JSON.stringify(json) : undefined,
|
||||
});
|
||||
let body;
|
||||
if (res.status !== 204) {
|
||||
const text = await res.text();
|
||||
try {
|
||||
body = text.length ? JSON.parse(text) : undefined;
|
||||
} catch {
|
||||
body = text;
|
||||
}
|
||||
}
|
||||
if (expect && !expect.includes(res.status)) {
|
||||
throw new Error(`${method} ${path} → ${res.status}: ${JSON.stringify(body)}`);
|
||||
}
|
||||
return { status: res.status, body };
|
||||
}
|
||||
|
||||
const adminLogin = () =>
|
||||
api('/admin/login', { method: 'POST', json: { password: cfg.adminPw }, expect: [200] }).then(
|
||||
(r) => r.body.jwt
|
||||
);
|
||||
|
||||
const joinGuest = (name) =>
|
||||
api('/join', { method: 'POST', json: { display_name: name }, expect: [201] }).then((r) => r.body);
|
||||
|
||||
const patchConfig = (adminJwt, patch) =>
|
||||
api('/admin/config', { method: 'PATCH', token: adminJwt, json: patch, expect: [204] });
|
||||
|
||||
const getConfig = (adminJwt) => api('/admin/config', { token: adminJwt }).then((r) => r.body);
|
||||
|
||||
const truncate = (adminJwt) =>
|
||||
api('/admin/__truncate', { method: 'POST', token: adminJwt, expect: [204] });
|
||||
|
||||
const CAPTIONS = [
|
||||
'Was für ein magischer Tag 💍', 'Der erste Tanz 🕺', 'Prost! 🥂', 'Die Torte 🍰',
|
||||
'Feuerwerk 🎆', 'Beste Freunde 💕', 'Was für eine Stimmung! 🎉', 'Details 🌸',
|
||||
'Sonnenuntergang 🌅', 'Tanzfläche brennt 🔥', null, null, null,
|
||||
];
|
||||
const TAGS = ['hochzeit', 'liebe', 'party', 'tanzen', 'natur', 'feier', 'freunde', 'dessert'];
|
||||
|
||||
async function uploadPhoto(jwt, file, buf) {
|
||||
const form = new FormData();
|
||||
form.append('file', new Blob([buf], { type: 'image/jpeg' }), 'photo.jpg');
|
||||
const cap = pick(CAPTIONS);
|
||||
if (cap) form.append('caption', cap);
|
||||
form.append('hashtags', `${pick(TAGS)},${pick(TAGS)}`);
|
||||
const t0 = now();
|
||||
const res = await fetch(`${API}/upload`, {
|
||||
method: 'POST',
|
||||
headers: { Authorization: `Bearer ${jwt}` },
|
||||
body: form,
|
||||
});
|
||||
const t1 = now();
|
||||
let id, errText;
|
||||
if (res.status === 201) {
|
||||
id = (await res.json()).id;
|
||||
} else {
|
||||
errText = (await res.text()).slice(0, 200);
|
||||
}
|
||||
return { status: res.status, id, ms: t1 - t0, endTs: t1, bytes: buf.length, errText };
|
||||
}
|
||||
|
||||
// ── SSE client (auto-reconnecting) ────────────────────────────────────────────
|
||||
class SseClient {
|
||||
constructor(jwt, label, onEvent) {
|
||||
this.jwt = jwt;
|
||||
this.label = label;
|
||||
this.onEvent = onEvent;
|
||||
this.stop = false;
|
||||
this.reconnects = 0;
|
||||
this.resyncs = 0;
|
||||
this.controller = null;
|
||||
}
|
||||
start() {
|
||||
this._loop();
|
||||
return this;
|
||||
}
|
||||
async _loop() {
|
||||
while (!this.stop) {
|
||||
try {
|
||||
const tkt = await api('/stream/ticket', { method: 'POST', token: this.jwt, expect: [200] });
|
||||
this.controller = new AbortController();
|
||||
const res = await fetch(`${API}/stream?ticket=${tkt.body.ticket}`, {
|
||||
headers: { Accept: 'text/event-stream' },
|
||||
signal: this.controller.signal,
|
||||
});
|
||||
if (!res.ok || !res.body) throw new Error(`stream ${res.status}`);
|
||||
const reader = res.body.getReader();
|
||||
const dec = new TextDecoder();
|
||||
let buf = '';
|
||||
while (!this.stop) {
|
||||
const { value, done } = await reader.read();
|
||||
if (done) break;
|
||||
buf += dec.decode(value, { stream: true });
|
||||
let idx;
|
||||
while ((idx = buf.indexOf('\n\n')) !== -1) {
|
||||
const frame = buf.slice(0, idx);
|
||||
buf = buf.slice(idx + 2);
|
||||
this._parseFrame(frame);
|
||||
}
|
||||
}
|
||||
} catch (e) {
|
||||
if (this.stop) return;
|
||||
this.reconnects++;
|
||||
await sleep(500 + Math.random() * 500);
|
||||
}
|
||||
}
|
||||
}
|
||||
_parseFrame(frame) {
|
||||
let event = 'message';
|
||||
let data = '';
|
||||
for (const line of frame.split('\n')) {
|
||||
if (line.startsWith('event:')) event = line.slice(6).trim();
|
||||
else if (line.startsWith('data:')) data += line.slice(5).trim();
|
||||
}
|
||||
if (event === 'resync') this.resyncs++;
|
||||
if (!data) return;
|
||||
let payload;
|
||||
try {
|
||||
payload = JSON.parse(data);
|
||||
} catch {
|
||||
payload = data;
|
||||
}
|
||||
this.onEvent(event, payload, this.label);
|
||||
}
|
||||
close() {
|
||||
this.stop = true;
|
||||
try {
|
||||
this.controller?.abort();
|
||||
} catch {
|
||||
/* noop */
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Resource sampler (docker stats + psql ground truth) ───────────────────────
|
||||
async function sampleResources() {
|
||||
const out = { ts: now() };
|
||||
try {
|
||||
const { stdout } = await execFileAsync('docker', [
|
||||
'stats', '--no-stream', '--format',
|
||||
'{{.Name}}|{{.CPUPerc}}|{{.MemUsage}}',
|
||||
cfg.appContainer, cfg.dbContainer,
|
||||
]);
|
||||
out.docker = stdout.trim();
|
||||
} catch (e) {
|
||||
out.dockerErr = String(e).slice(0, 120);
|
||||
}
|
||||
try {
|
||||
const { stdout } = await psql(
|
||||
`select count(*) from pg_stat_activity where datname='eventsnap_test'`
|
||||
);
|
||||
out.dbConns = parseInt(stdout.trim(), 10);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
function psql(sql) {
|
||||
return execFileAsync('docker', [
|
||||
'exec', cfg.dbContainer, 'psql', '-U', 'eventsnap_test', '-d', 'eventsnap_test',
|
||||
'-tAc', sql,
|
||||
]);
|
||||
}
|
||||
|
||||
async function compressionCounts() {
|
||||
try {
|
||||
const { stdout } = await psql(
|
||||
`select compression_status, count(*) from upload where deleted_at is null group by 1`
|
||||
);
|
||||
const map = {};
|
||||
for (const line of stdout.trim().split('\n')) {
|
||||
if (!line) continue;
|
||||
const [status, n] = line.split('|');
|
||||
map[status] = parseInt(n, 10);
|
||||
}
|
||||
return map;
|
||||
} catch (e) {
|
||||
return { error: String(e).slice(0, 120) };
|
||||
}
|
||||
}
|
||||
|
||||
// ── Stats helpers ─────────────────────────────────────────────────────────────
|
||||
function pct(sorted, p) {
|
||||
if (!sorted.length) return null;
|
||||
const i = Math.min(sorted.length - 1, Math.floor((p / 100) * sorted.length));
|
||||
return sorted[i];
|
||||
}
|
||||
function summarize(nums) {
|
||||
if (!nums.length) return null;
|
||||
const s = [...nums].sort((a, b) => a - b);
|
||||
const sum = s.reduce((a, b) => a + b, 0);
|
||||
return {
|
||||
n: s.length, min: s[0], max: s[s.length - 1], mean: Math.round(sum / s.length),
|
||||
p50: pct(s, 50), p95: pct(s, 95), p99: pct(s, 99),
|
||||
};
|
||||
}
|
||||
|
||||
// ── Image pool ────────────────────────────────────────────────────────────────
|
||||
function loadPool() {
|
||||
let files;
|
||||
try {
|
||||
files = readdirSync(cfg.photosDir).filter((f) => f.endsWith('.jpg'));
|
||||
} catch {
|
||||
files = [];
|
||||
}
|
||||
if (!files.length) {
|
||||
console.error(
|
||||
`\n✗ No images in ${cfg.photosDir}. Run first:\n` +
|
||||
` LT_PHOTOS_DIR=${cfg.photosDir} python3 ${join(__dirname, 'gen-images.py')}\n`
|
||||
);
|
||||
process.exit(1);
|
||||
}
|
||||
const pool = files.map((f) => {
|
||||
const p = join(cfg.photosDir, f);
|
||||
return { buf: readFileSync(p), size: statSync(p).size, name: f };
|
||||
});
|
||||
return pool;
|
||||
}
|
||||
|
||||
// ── Burst schedule ────────────────────────────────────────────────────────────
|
||||
// Build bursts until we reach the image target, assign each to a random guest at
|
||||
// a random time in the window. Naturally gives some guests several bursts and
|
||||
// some none — like a real event. Guests all join before their first burst.
|
||||
function buildSchedule() {
|
||||
const bursts = [];
|
||||
let remaining = cfg.images;
|
||||
while (remaining > 0) {
|
||||
const size = Math.min(remaining, rand(cfg.burstMin, cfg.burstMax));
|
||||
bursts.push({
|
||||
guest: rand(0, cfg.guests - 1),
|
||||
atMs: Math.floor(Math.random() * cfg.windowSec * 1000),
|
||||
size,
|
||||
});
|
||||
remaining -= size;
|
||||
}
|
||||
bursts.sort((a, b) => a.atMs - b.atMs);
|
||||
return bursts;
|
||||
}
|
||||
|
||||
// ── Main ──────────────────────────────────────────────────────────────────────
|
||||
async function main() {
|
||||
console.log('━'.repeat(72));
|
||||
console.log('EventSnap load driver');
|
||||
console.log(
|
||||
` target: ${cfg.guests} guests, ${cfg.images} images, ` +
|
||||
`bursts ${cfg.burstMin}-${cfg.burstMax}, window ${cfg.windowSec}s, ${cfg.viewers} viewers`
|
||||
);
|
||||
console.log(` base: ${cfg.base}`);
|
||||
console.log('━'.repeat(72));
|
||||
|
||||
const pool = loadPool();
|
||||
const avgMb = pool.reduce((a, p) => a + p.size, 0) / pool.length / 1e6;
|
||||
console.log(
|
||||
`[pool] ${pool.length} images, avg ${avgMb.toFixed(2)} MB, ` +
|
||||
`projected originals ~${((avgMb * cfg.images) / 1000).toFixed(1)} GB`
|
||||
);
|
||||
|
||||
// Preflight: disk headroom
|
||||
try {
|
||||
const { stdout } = await execFileAsync('df', ['-BG', '--output=avail', '/']);
|
||||
console.log(`[preflight] disk avail:${stdout.trim().split('\n').pop().trim()}`);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
|
||||
// Admin: reset + config
|
||||
const admin = await adminLogin();
|
||||
const cfgBefore = await getConfig(admin);
|
||||
if (cfg.truncate) {
|
||||
console.log('[setup] truncating event data…');
|
||||
await truncate(admin);
|
||||
}
|
||||
const admin2 = await adminLogin();
|
||||
if (!cfg.keepRateLimits) {
|
||||
console.log('[setup] disabling rate limits (methodology: isolate pipeline, not limiter)');
|
||||
await patchConfig(admin2, {
|
||||
rate_limits_enabled: 'false',
|
||||
upload_rate_enabled: 'false',
|
||||
feed_rate_enabled: 'false',
|
||||
join_rate_enabled: 'false',
|
||||
recover_rate_enabled: 'false',
|
||||
});
|
||||
}
|
||||
console.log(
|
||||
`[setup] compression concurrency / quotas left at SHIPPING defaults ` +
|
||||
`(compression_worker_concurrency is a boot env var, not runtime)`
|
||||
);
|
||||
|
||||
// Metrics stores
|
||||
const uploads = []; // {status, ms, endTs, bytes, id, guest}
|
||||
const processedAt = new Map(); // upload_id -> ts of upload-processed
|
||||
const uploadEndTs = new Map(); // upload_id -> endTs
|
||||
const errorEvents = []; // upload-error payloads
|
||||
const resources = [];
|
||||
let newUploadEvents = 0;
|
||||
|
||||
// Join guests (staggered, small concurrency)
|
||||
console.log(`[join] joining ${cfg.guests} guests…`);
|
||||
const accounts = new Array(cfg.guests);
|
||||
const joinConc = 10;
|
||||
for (let i = 0; i < cfg.guests; i += joinConc) {
|
||||
await Promise.all(
|
||||
Array.from({ length: Math.min(joinConc, cfg.guests - i) }, (_, k) =>
|
||||
joinGuest(`LoadGuest ${i + k}`).then((a) => (accounts[i + k] = a))
|
||||
)
|
||||
);
|
||||
}
|
||||
console.log('[join] done');
|
||||
|
||||
// Live connections: 1 diashow + N viewers
|
||||
const onEvent = (event, payload) => {
|
||||
if (event === 'new-upload') newUploadEvents++;
|
||||
else if (event === 'upload-processed' && payload?.upload_id) {
|
||||
if (!processedAt.has(payload.upload_id)) processedAt.set(payload.upload_id, now());
|
||||
} else if (event === 'upload-error' && payload?.upload_id) {
|
||||
errorEvents.push(payload);
|
||||
}
|
||||
};
|
||||
const sseClients = [];
|
||||
sseClients.push(new SseClient(accounts[0].jwt, 'diashow', onEvent).start());
|
||||
for (let i = 0; i < Math.min(cfg.viewers, cfg.guests); i++) {
|
||||
sseClients.push(new SseClient(accounts[i].jwt, `viewer${i}`, onEvent).start());
|
||||
}
|
||||
console.log(`[sse] opened ${sseClients.length} live connections (1 diashow + viewers)`);
|
||||
|
||||
// Viewers also poll the feed periodically (viewing load)
|
||||
let feedPolls = 0;
|
||||
const feedPoller = setInterval(async () => {
|
||||
const g = accounts[rand(0, Math.min(cfg.viewers, cfg.guests) - 1)];
|
||||
try {
|
||||
await api('/feed', { token: g.jwt });
|
||||
feedPolls++;
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}, 1500);
|
||||
|
||||
// Resource sampler
|
||||
const sampler = setInterval(async () => resources.push(await sampleResources()), 5000);
|
||||
resources.push(await sampleResources());
|
||||
|
||||
// Run the burst schedule
|
||||
const schedule = buildSchedule();
|
||||
console.log(
|
||||
`[run] ${schedule.length} bursts scheduled over ${cfg.windowSec}s ` +
|
||||
`(≈${(cfg.images / (cfg.windowSec / 60)).toFixed(0)} img/min avg). Starting…`
|
||||
);
|
||||
const startTs = now();
|
||||
let done = 0;
|
||||
|
||||
const runBurst = async (burst) => {
|
||||
const jwt = accounts[burst.guest].jwt;
|
||||
const items = Array.from({ length: burst.size }, () => pick(pool));
|
||||
// upload with phone-like small concurrency inside the burst
|
||||
for (let i = 0; i < items.length; i += cfg.burstConc) {
|
||||
const chunk = items.slice(i, i + cfg.burstConc);
|
||||
const results = await Promise.all(chunk.map((it) => uploadPhoto(jwt, it.name, it.buf)));
|
||||
for (const r of results) {
|
||||
uploads.push({ ...r, guest: burst.guest });
|
||||
if (r.id) uploadEndTs.set(r.id, r.endTs);
|
||||
done++;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
const timers = [];
|
||||
const burstPromises = [];
|
||||
for (const burst of schedule) {
|
||||
timers.push(
|
||||
setTimeout(() => {
|
||||
burstPromises.push(runBurst(burst));
|
||||
}, burst.atMs)
|
||||
);
|
||||
}
|
||||
|
||||
// progress ticker
|
||||
const ticker = setInterval(() => {
|
||||
const elapsed = ((now() - startTs) / 1000).toFixed(0);
|
||||
const ok = uploads.filter((u) => u.status === 201).length;
|
||||
console.log(
|
||||
`[t+${elapsed}s] uploads ${done}/${cfg.images} (ok ${ok}), ` +
|
||||
`processed ${processedAt.size}, new-upload evts ${newUploadEvents}, ` +
|
||||
`feedPolls ${feedPolls}`
|
||||
);
|
||||
}, 10000);
|
||||
|
||||
// wait for the window to elapse, then for all scheduled bursts to finish
|
||||
await sleep(cfg.windowSec * 1000 + 500);
|
||||
await Promise.all(burstPromises);
|
||||
clearInterval(ticker);
|
||||
console.log(`[run] all bursts issued. uploaded ${done}, ok ${uploads.filter((u) => u.status === 201).length}`);
|
||||
|
||||
// Drain: wait for compression backlog to clear (SSE processed ⊇ successful ids)
|
||||
console.log('[drain] waiting for compression backlog to clear…');
|
||||
const drainStart = now();
|
||||
const successIds = new Set(uploads.filter((u) => u.id).map((u) => u.id));
|
||||
let lastLog = 0;
|
||||
let drainReason = 'timeout';
|
||||
while (now() - drainStart < cfg.drainTimeoutSec * 1000) {
|
||||
// SSE view (what the diashow actually "sees")
|
||||
const pendingSse = [...successIds].filter((id) => !processedAt.has(id)).length;
|
||||
if (pendingSse === 0) {
|
||||
console.log('[drain] backlog cleared (all upload-processed events received)');
|
||||
drainReason = 'sse-complete';
|
||||
break;
|
||||
}
|
||||
// DB ground truth — authoritative even if an SSE reconnect dropped events
|
||||
const counts = await compressionCounts();
|
||||
const dbPending = Object.entries(counts)
|
||||
.filter(([s]) => s !== 'done' && s !== 'error')
|
||||
.reduce((a, [, n]) => a + n, 0);
|
||||
if (!counts.error && dbPending === 0) {
|
||||
console.log(`[drain] backlog cleared per DB (db status: ${JSON.stringify(counts)})`);
|
||||
drainReason = 'db-complete';
|
||||
break;
|
||||
}
|
||||
if (now() - lastLog > 5000) {
|
||||
console.log(
|
||||
`[drain] pending(sse) ${pendingSse}, pending(db) ${dbPending} — db: ${JSON.stringify(counts)}`
|
||||
);
|
||||
lastLog = now();
|
||||
}
|
||||
await sleep(1000);
|
||||
}
|
||||
const drainMs = now() - drainStart;
|
||||
|
||||
// Stop background work
|
||||
clearInterval(sampler);
|
||||
clearInterval(feedPoller);
|
||||
timers.forEach(clearTimeout);
|
||||
resources.push(await sampleResources());
|
||||
const finalCounts = await compressionCounts();
|
||||
await sleep(200);
|
||||
sseClients.forEach((c) => c.close());
|
||||
|
||||
// ── Build report ────────────────────────────────────────────────────────────
|
||||
const byStatus = {};
|
||||
for (const u of uploads) byStatus[u.status] = (byStatus[u.status] ?? 0) + 1;
|
||||
const okUploads = uploads.filter((u) => u.status === 201);
|
||||
const uploadLatency = summarize(okUploads.map((u) => u.ms));
|
||||
const pipelineLatency = summarize(
|
||||
[...processedAt.entries()]
|
||||
.filter(([id]) => uploadEndTs.has(id))
|
||||
.map(([id, ts]) => ts - uploadEndTs.get(id))
|
||||
);
|
||||
const totalReconnects = sseClients.reduce((a, c) => a + c.reconnects, 0);
|
||||
const totalResyncs = sseClients.reduce((a, c) => a + c.resyncs, 0);
|
||||
|
||||
const report = {
|
||||
config: cfg,
|
||||
startedAt: new Date(startTs).toISOString(),
|
||||
durationSec: Math.round((now() - startTs) / 1000),
|
||||
totals: {
|
||||
uploadsAttempted: uploads.length,
|
||||
uploadsOk: okUploads.length,
|
||||
byStatus,
|
||||
newUploadEvents,
|
||||
processedEvents: processedAt.size,
|
||||
uploadErrorEvents: errorEvents.length,
|
||||
feedPolls,
|
||||
},
|
||||
uploadLatencyMs: uploadLatency,
|
||||
pipelineLatencyMs: pipelineLatency,
|
||||
drain: {
|
||||
ms: drainMs,
|
||||
cleared: drainReason !== 'timeout',
|
||||
reason: drainReason,
|
||||
ssePending: [...successIds].filter((id) => !processedAt.has(id)).length,
|
||||
finalDbCounts: finalCounts,
|
||||
},
|
||||
sse: {
|
||||
connections: sseClients.length,
|
||||
reconnects: totalReconnects,
|
||||
resyncs: totalResyncs,
|
||||
},
|
||||
resources,
|
||||
rateLimitsDisabled: !cfg.keepRateLimits,
|
||||
configBefore: cfgBefore,
|
||||
};
|
||||
|
||||
mkdirSync(cfg.outDir, { recursive: true });
|
||||
const stamp = new Date(startTs).toISOString().replace(/[:.]/g, '-');
|
||||
const outPath = join(cfg.outDir, `run-${stamp}.json`);
|
||||
writeFileSync(outPath, JSON.stringify(report, null, 2));
|
||||
|
||||
// ── Print verdict ─────────────────────────────────────────────────────────
|
||||
console.log('\n' + '━'.repeat(72));
|
||||
console.log('RESULTS');
|
||||
console.log('━'.repeat(72));
|
||||
console.log(`uploads: ${okUploads.length}/${uploads.length} ok — byStatus ${JSON.stringify(byStatus)}`);
|
||||
console.log(`upload latency ms: ${JSON.stringify(uploadLatency)}`);
|
||||
console.log(`pipeline latency ms (upload→preview ready): ${JSON.stringify(pipelineLatency)}`);
|
||||
console.log(`backlog drain: ${(drainMs / 1000).toFixed(1)}s, cleared=${report.drain.cleared}`);
|
||||
console.log(`final db compression status: ${JSON.stringify(finalCounts)}`);
|
||||
console.log(`sse: ${sseClients.length} conns, ${totalReconnects} reconnects, ${totalResyncs} resyncs`);
|
||||
console.log(`\nfull report → ${outPath}`);
|
||||
|
||||
// Heuristic pass/fail flags (validate shipping config)
|
||||
const flags = [];
|
||||
const err5xx = Object.entries(byStatus).filter(([s]) => +s >= 500).reduce((a, [, n]) => a + n, 0);
|
||||
if (err5xx > 0) flags.push(`✗ ${err5xx} server errors (5xx)`);
|
||||
if (byStatus['507']) flags.push(`✗ ${byStatus['507']} quota rejections (507)`);
|
||||
if (byStatus['413']) flags.push(`⚠ ${byStatus['413']} too-large (413)`);
|
||||
if (byStatus['429']) flags.push(`⚠ ${byStatus['429']} rate-limited (429) — unexpected with limits off`);
|
||||
if (!report.drain.cleared) flags.push(`✗ backlog did NOT drain within ${cfg.drainTimeoutSec}s`);
|
||||
if (totalResyncs > sseClients.length) flags.push(`⚠ ${totalResyncs} SSE resyncs (consumer lag)`);
|
||||
if (pipelineLatency && pipelineLatency.p95 > 60000)
|
||||
flags.push(`⚠ pipeline p95 ${(pipelineLatency.p95 / 1000).toFixed(0)}s (diashow lags >1min)`);
|
||||
console.log('\nflags:');
|
||||
if (flags.length) flags.forEach((f) => console.log(' ' + f));
|
||||
else console.log(' ✓ no red flags — default config handled the load');
|
||||
console.log('━'.repeat(72));
|
||||
}
|
||||
|
||||
main().catch((e) => {
|
||||
console.error('\n✗ driver failed:', e);
|
||||
process.exit(1);
|
||||
});
|
||||
Reference in New Issue
Block a user