mirror of
https://github.com/tiennm99/tiennm99bot.git
synced 2026-10-15 04:19:02 +00:00
feat(observability): phase 06 — timing telemetry + soak analyzer + burst tester
Code prerequisites for the Phase 06 cold-start soak gate. The 24-72h soak
itself is operator-run; this commit ships the instrumentation + analysis
tools needed to make the PROCEED-or-PIVOT decision.
Telemetry
- src/util/timing.js: startTiming(cmd) returns {mark, end} that emits a
structured cmd_timing log. takeColdFlag() returns {cold, isolateAgeMs}
using a module-scoped boolean — first request in an isolate is cold,
subsequent are warm. This replaces the originally-planned
isolate_age_ms < 200ms classifier (broken because Mongo cold-connect
itself is ~1500ms; cold requests would always bucket as warm —
code-reviewer #11).
- src/util/request-context.js: setLastCold/getLastCold shared state
bridges fetch-level cold detection into the dispatcher middleware
without a circular import.
- src/index.js: takeColdFlag at the top of fetch() emits a request log
and primes the request context for the dispatcher.
- src/modules/dispatcher.js: bot.use() middleware times every command.
Chosen over per-handler wrapping to preserve the existing identity
assertion in tests (handler === reg.allCommands.get(name).cmd.handler)
— single instrumentation point, no contract change.
Soak tools (operator-run)
- scripts/analyze-soak.js: parses CF Logs export (NDJSON or CSV), filters
cmd_timing events, computes p50/p95/p99 per (cmd, cold/warm). Counts
dual-write secondary failures, mongo connection errors, CPU-time
exceeded events. Writes markdown report.
- scripts/synthetic-burst.js: fires N parallel synthetic Telegram updates
at the deployed Worker URL with cache-busting tokens. Used for the
pre-deploy connection-cap stress test (debugger #2 — 20 parallel cold
requests, abort if Atlas peak > 60% of 500-conn cap).
- package.json: analyze:soak + burst:synthetic scripts wired.
Tests
- tests/util/timing.test.js: 8 tests — timing semantics, cold flag flip.
- tests/scripts/analyze-soak.test.js: 22 tests — percentile math, NDJSON
+ CSV parse, aggregation, markdown formatting.
Tests: 667 → 697 (+30). Lint clean.
Operator runbook for Phase 06 (NOT executed by this commit):
1. Verify telemetry live via wrangler tail.
2. Run synthetic burst test: npm run burst:synthetic -- --url <prod>
3. Configure Atlas + CF Observability email alerts.
4. 24h soak (extend to 72h on stop-conditions per phase plan).
5. Daily npm run verify:mongo.
6. npm run analyze:soak -- --input <cf-logs.json> → soak-decision.md.
7. PROCEED to Phase 07 if cold-start P95 ≤ 2.5 × BASELINE_COLD_PING_MS;
else execute phase-07-alt-pivot.md (Upstash standby).
This commit is contained in:
1 parent
afe7d768b7
commit
439e874a11
10 files changed
+851
No files matched your search
@@ -24,6 +24,8 @@
|
||||
"backfill:d1:dry": "node --env-file-if-exists=.env.deploy scripts/backfill-d1-to-mongo.js --dry-run",
|
||||
"verify:mongo": "node --env-file-if-exists=.env.deploy scripts/verify-mongo-parity.js",
|
||||
"wipe:mongo": "node --env-file-if-exists=.env.deploy scripts/wipe-mongo.js",
|
||||
"analyze:soak": "node scripts/analyze-soak.js",
|
||||
"burst:synthetic": "node scripts/synthetic-burst.js",
|
||||
"lint": "biome check . && eslint src && node scripts/check-secret-leaks.js",
|
||||
"format": "biome format --write .",
|
||||
"test": "vitest run"
|
||||
|
||||
@@ -0,0 +1,229 @@
|
||||
#!/usr/bin/env node
|
||||
/**
|
||||
* @file analyze-soak — parse a CF Logs export and compute per-(cmd × cold/warm)
|
||||
* latency percentiles plus error counts for the 24h soak analysis.
|
||||
*
|
||||
* Usage:
|
||||
* node scripts/analyze-soak.js --input soak-export.json [--commands /wordle,/loldle] [--output report.md]
|
||||
*
|
||||
* Input: CF Logs JSON export — one JSON object per line (NDJSON) OR a CSV export
|
||||
* where each row has at least a "message" column containing the raw JSON log string.
|
||||
*
|
||||
* Output (stdout + optional --output <md>):
|
||||
* Markdown table: cmd | cold/warm | n | p50 | p95 | p99
|
||||
* Error summary: dual-write secondary failures, mongo errors, CPU-time exceeded
|
||||
*/
|
||||
|
||||
import { readFileSync, writeFileSync } from "node:fs";
|
||||
|
||||
// ── CLI arg parsing ──────────────────────────────────────────────────────────
|
||||
|
||||
function parseArgs(argv) {
|
||||
const args = {};
|
||||
for (let i = 2; i < argv.length; i++) {
|
||||
if (argv[i] === "--input") args.input = argv[++i];
|
||||
else if (argv[i] === "--commands") args.commands = argv[++i];
|
||||
else if (argv[i] === "--output") args.output = argv[++i];
|
||||
}
|
||||
return args;
|
||||
}
|
||||
|
||||
// ── Percentile math ──────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Compute a percentile from a sorted numeric array (ascending).
|
||||
* Uses nearest-rank method.
|
||||
*
|
||||
* @param {number[]} sorted - must be sorted ascending
|
||||
* @param {number} p - percentile 0–100
|
||||
* @returns {number}
|
||||
*/
|
||||
export function percentile(sorted, p) {
|
||||
if (sorted.length === 0) return 0;
|
||||
if (sorted.length === 1) return sorted[0];
|
||||
const idx = Math.ceil((p / 100) * sorted.length) - 1;
|
||||
return sorted[Math.max(0, Math.min(idx, sorted.length - 1))];
|
||||
}
|
||||
|
||||
// ── Log line parsing ─────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Try to extract a JSON object from a raw log line.
|
||||
* Handles NDJSON (line is the JSON) and CSV rows where the JSON lives inside
|
||||
* a quoted "message" column value.
|
||||
*
|
||||
* @param {string} line
|
||||
* @returns {object|null}
|
||||
*/
|
||||
export function parseLogLine(line) {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) return null;
|
||||
|
||||
// Try direct JSON parse (NDJSON format).
|
||||
if (trimmed.startsWith("{")) {
|
||||
try {
|
||||
return JSON.parse(trimmed);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
// CSV: find the first {...} substring inside the line and parse it.
|
||||
const braceStart = trimmed.indexOf("{");
|
||||
const braceEnd = trimmed.lastIndexOf("}");
|
||||
if (braceStart !== -1 && braceEnd > braceStart) {
|
||||
try {
|
||||
return JSON.parse(trimmed.slice(braceStart, braceEnd + 1));
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
// ── Aggregation ──────────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* @typedef {{ n: number, samples: number[] }} Bucket
|
||||
* @typedef {Map<string, Bucket>} BucketMap key = "cmd|cold" | "cmd|warm"
|
||||
*/
|
||||
|
||||
/**
|
||||
* Read all lines from a file path synchronously and return as array.
|
||||
*
|
||||
* @param {string} filePath
|
||||
* @returns {string[]}
|
||||
*/
|
||||
function readLines(filePath) {
|
||||
return readFileSync(filePath, "utf8").split("\n");
|
||||
}
|
||||
|
||||
/**
|
||||
* Aggregate cmd_timing events from log lines into per-(cmd×cold/warm) buckets.
|
||||
*
|
||||
* @param {string[]} lines - raw log lines
|
||||
* @param {string[]|null} filterCmds - if set, only include these cmd names
|
||||
* @returns {{ buckets: BucketMap, errors: { dualWriteFail: number, mongoError: number, cpuExceeded: number } }}
|
||||
*/
|
||||
export function aggregateLines(lines, filterCmds) {
|
||||
/** @type {BucketMap} */
|
||||
const buckets = new Map();
|
||||
const errors = { dualWriteFail: 0, mongoError: 0, cpuExceeded: 0 };
|
||||
|
||||
for (const line of lines) {
|
||||
// Count error patterns regardless of JSON structure.
|
||||
if (
|
||||
line.includes("dual-write:secondary:failed") ||
|
||||
line.includes("[dual-kv] secondary write failed")
|
||||
) {
|
||||
errors.dualWriteFail++;
|
||||
}
|
||||
if (
|
||||
line.includes("MongoError") ||
|
||||
line.includes("mongo connection") ||
|
||||
line.includes("MongoNetworkError")
|
||||
) {
|
||||
errors.mongoError++;
|
||||
}
|
||||
if (line.includes("Worker exceeded CPU time") || line.includes("cpu time exceeded")) {
|
||||
errors.cpuExceeded++;
|
||||
}
|
||||
|
||||
const obj = parseLogLine(line);
|
||||
if (!obj || obj.event !== "cmd_timing") continue;
|
||||
|
||||
const { cmd, total, cold } = obj;
|
||||
if (typeof cmd !== "string" || typeof total !== "number") continue;
|
||||
|
||||
// Apply command filter.
|
||||
if (filterCmds && !filterCmds.includes(cmd)) continue;
|
||||
|
||||
const bucket = cold ? "cold" : "warm";
|
||||
const key = `${cmd}|${bucket}`;
|
||||
if (!buckets.has(key)) buckets.set(key, { n: 0, samples: [] });
|
||||
const b = buckets.get(key);
|
||||
b.n++;
|
||||
b.samples.push(total);
|
||||
}
|
||||
|
||||
return { buckets, errors };
|
||||
}
|
||||
|
||||
// ── Report formatting ────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Build a markdown report string from aggregated data.
|
||||
*
|
||||
* @param {BucketMap} buckets
|
||||
* @param {{ dualWriteFail: number, mongoError: number, cpuExceeded: number }} errors
|
||||
* @returns {string}
|
||||
*/
|
||||
export function formatReport(buckets, errors) {
|
||||
const rows = [];
|
||||
|
||||
for (const [key, bucket] of [...buckets.entries()].sort()) {
|
||||
const [cmd, coldWarm] = key.split("|");
|
||||
const sorted = [...bucket.samples].sort((a, b) => a - b);
|
||||
rows.push({
|
||||
cmd,
|
||||
coldWarm,
|
||||
n: bucket.n,
|
||||
p50: percentile(sorted, 50),
|
||||
p95: percentile(sorted, 95),
|
||||
p99: percentile(sorted, 99),
|
||||
});
|
||||
}
|
||||
|
||||
const header = "| cmd | cold/warm | n | p50 | p95 | p99 |";
|
||||
const sep = "|-----|-----------|---|-----|-----|-----|";
|
||||
const dataRows = rows.map(
|
||||
(r) => `| ${r.cmd} | ${r.coldWarm} | ${r.n} | ${r.p50} | ${r.p95} | ${r.p99} |`,
|
||||
);
|
||||
|
||||
const table = [header, sep, ...dataRows].join("\n");
|
||||
|
||||
const errorSection = [
|
||||
"",
|
||||
"## Error Summary",
|
||||
`- Dual-write secondary failures: ${errors.dualWriteFail}`,
|
||||
`- Mongo connection errors: ${errors.mongoError}`,
|
||||
`- CPU time exceeded: ${errors.cpuExceeded}`,
|
||||
].join("\n");
|
||||
|
||||
return `## Soak Latency Report\n\n${table}\n${errorSection}\n`;
|
||||
}
|
||||
|
||||
// ── Main ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
async function main() {
|
||||
const args = parseArgs(process.argv);
|
||||
|
||||
if (!args.input) {
|
||||
console.error(
|
||||
"Usage: node scripts/analyze-soak.js --input <path> [--commands /wordle,/loldle] [--output <md>]",
|
||||
);
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
const filterCmds = args.commands ? args.commands.split(",").map((c) => c.trim()) : null;
|
||||
const lines = readLines(args.input);
|
||||
const { buckets, errors } = aggregateLines(lines, filterCmds);
|
||||
const report = formatReport(buckets, errors);
|
||||
|
||||
process.stdout.write(report);
|
||||
|
||||
if (args.output) {
|
||||
writeFileSync(args.output, report, "utf8");
|
||||
console.error(`\nReport written to ${args.output}`);
|
||||
}
|
||||
}
|
||||
|
||||
// Run only when executed directly (not imported in tests).
|
||||
const isMain = process.argv[1]?.endsWith("analyze-soak.js");
|
||||
if (isMain) {
|
||||
main().catch((err) => {
|
||||
console.error(err);
|
||||
process.exit(1);
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,148 @@
|
||||
#!/usr/bin/env node
|
||||
/**
|
||||
* @file synthetic-burst — fire N parallel requests at the deployed Worker URL
|
||||
* to exercise the M0 Atlas connection cap before a live deploy.
|
||||
*
|
||||
* Each request is a synthetic Telegram webhook update POST that grammY will
|
||||
* route to the specified command handler. All requests hit the Worker
|
||||
* simultaneously (Promise.all) to maximise cold-isolate spawning.
|
||||
*
|
||||
* Usage:
|
||||
* node scripts/synthetic-burst.js \
|
||||
* --url https://miti99bot.workers.dev \
|
||||
* --secret <X-Telegram-Bot-Api-Secret-Token> \
|
||||
* [--n 20] \
|
||||
* [--cmd /wordle]
|
||||
*
|
||||
* No unit tests for this script — it is network-touching by design and only
|
||||
* runs against a live deployed Worker. Tested manually pre-deploy.
|
||||
*
|
||||
* Abort guideline (debugger #2): if Atlas connection peak > 300/500 (60% cap),
|
||||
* do NOT proceed with live deploy. Check Atlas UI during the 60s after the burst.
|
||||
*/
|
||||
|
||||
// ── CLI arg parsing ──────────────────────────────────────────────────────────
|
||||
|
||||
function parseArgs(argv) {
|
||||
const args = { n: 20, cmd: "/wordle" };
|
||||
for (let i = 2; i < argv.length; i++) {
|
||||
if (argv[i] === "--url") args.url = argv[++i];
|
||||
else if (argv[i] === "--secret") args.secret = argv[++i];
|
||||
else if (argv[i] === "--n") args.n = Number.parseInt(argv[++i], 10);
|
||||
else if (argv[i] === "--cmd") args.cmd = argv[++i];
|
||||
}
|
||||
return args;
|
||||
}
|
||||
|
||||
// ── Synthetic Telegram update payload ───────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Build a minimal but valid grammY-shaped Telegram Update object for a
|
||||
* bot_command message.
|
||||
*
|
||||
* @param {string} cmd - e.g. "/wordle"
|
||||
* @param {number} index - used to differentiate update_id + message_id values
|
||||
* @returns {object}
|
||||
*/
|
||||
function buildUpdate(cmd, index) {
|
||||
return {
|
||||
update_id: 100000 + index,
|
||||
message: {
|
||||
message_id: 200000 + index,
|
||||
from: { id: 1, is_bot: false, first_name: "Burst" },
|
||||
chat: { id: 1, type: "private" },
|
||||
date: Math.floor(Date.now() / 1000),
|
||||
text: cmd,
|
||||
entities: [{ type: "bot_command", offset: 0, length: cmd.length }],
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
// ── Single request ───────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* POST one synthetic update to the Worker webhook endpoint.
|
||||
*
|
||||
* @param {string} url - Worker base URL
|
||||
* @param {string} secret - X-Telegram-Bot-Api-Secret-Token value
|
||||
* @param {object} update - Telegram Update payload
|
||||
* @param {number} index - request index for logging
|
||||
* @returns {Promise<{ index: number, status: number, ms: number, error?: string }>}
|
||||
*/
|
||||
async function sendUpdate(url, secret, update, index) {
|
||||
const t0 = Date.now();
|
||||
const endpoint = url.replace(/\/$/, "") + "/webhook";
|
||||
|
||||
try {
|
||||
const res = await fetch(endpoint, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-Type": "application/json",
|
||||
"X-Telegram-Bot-Api-Secret-Token": secret,
|
||||
},
|
||||
body: JSON.stringify(update),
|
||||
});
|
||||
return { index, status: res.status, ms: Date.now() - t0 };
|
||||
} catch (err) {
|
||||
return { index, status: 0, ms: Date.now() - t0, error: err.message };
|
||||
}
|
||||
}
|
||||
|
||||
// ── Main ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
async function main() {
|
||||
const args = parseArgs(process.argv);
|
||||
|
||||
if (!args.url || !args.secret) {
|
||||
console.error(
|
||||
"Usage: node scripts/synthetic-burst.js --url <url> --secret <token> [--n 20] [--cmd /wordle]",
|
||||
);
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
const { url, secret, n, cmd } = args;
|
||||
|
||||
console.log(`Burst: ${n} parallel requests → ${url}/webhook cmd=${cmd}`);
|
||||
console.log("Starting at", new Date().toISOString());
|
||||
|
||||
const requests = Array.from({ length: n }, (_, i) =>
|
||||
sendUpdate(url, secret, buildUpdate(cmd, i), i),
|
||||
);
|
||||
|
||||
const results = await Promise.all(requests);
|
||||
|
||||
// ── Summary ──────────────────────────────────────────────────────────────
|
||||
let ok = 0;
|
||||
let fail = 0;
|
||||
let totalMs = 0;
|
||||
const statusCounts = {};
|
||||
|
||||
for (const r of results) {
|
||||
const statusKey = r.status === 0 ? "network-error" : String(r.status);
|
||||
statusCounts[statusKey] = (statusCounts[statusKey] ?? 0) + 1;
|
||||
totalMs += r.ms;
|
||||
if (r.status >= 200 && r.status < 300) ok++;
|
||||
else fail++;
|
||||
|
||||
// Log individual result.
|
||||
const tag = r.error ? `ERROR(${r.error})` : `HTTP ${r.status}`;
|
||||
console.log(` [${r.index}] ${tag} ${r.ms}ms`);
|
||||
}
|
||||
|
||||
const avgMs = Math.round(totalMs / n);
|
||||
const allMs = results.map((r) => r.ms).sort((a, b) => a - b);
|
||||
const p50 = allMs[Math.floor(allMs.length * 0.5)];
|
||||
const p95 = allMs[Math.floor(allMs.length * 0.95)];
|
||||
|
||||
console.log("\n── Summary ───────────────────────────────────────────");
|
||||
console.log(` Requests: ${n} | OK: ${ok} | Failed: ${fail}`);
|
||||
console.log(` Status counts: ${JSON.stringify(statusCounts)}`);
|
||||
console.log(` Latency — avg: ${avgMs}ms p50: ${p50}ms p95: ${p95}ms`);
|
||||
console.log("\nNext: check Atlas UI connection counter within 60s.");
|
||||
console.log("Abort if peak connections > 300/500 (60% of M0 cap).");
|
||||
}
|
||||
|
||||
main().catch((err) => {
|
||||
console.error(err);
|
||||
process.exit(1);
|
||||
});
|
||||
@@ -15,6 +15,8 @@
|
||||
import { webhookCallback } from "grammy";
|
||||
import { getBot, getRegistry } from "./bot.js";
|
||||
import { dispatchScheduled } from "./modules/cron-dispatcher.js";
|
||||
import { setLastCold } from "./util/request-context.js";
|
||||
import { takeColdFlag } from "./util/timing.js";
|
||||
|
||||
/** @type {ReturnType<typeof webhookCallback> | null} */
|
||||
let cachedWebhookHandler = null;
|
||||
@@ -59,10 +61,28 @@ export default {
|
||||
* @param {any} _ctx
|
||||
*/
|
||||
async fetch(request, env, _ctx) {
|
||||
// Capture cold-start flag first — must be before any await.
|
||||
// Stored in shared request-context module so dispatcher middleware can read it
|
||||
// without a circular import (index → bot → dispatcher → index).
|
||||
const coldMeta = takeColdFlag();
|
||||
setLastCold(coldMeta);
|
||||
|
||||
const start = Date.now();
|
||||
const { pathname } = new URL(request.url);
|
||||
const method = request.method;
|
||||
|
||||
// Per-request cold-start log for CF Observability soak analysis.
|
||||
console.log(
|
||||
JSON.stringify({
|
||||
event: "request",
|
||||
method,
|
||||
path: pathname,
|
||||
cold: coldMeta.cold,
|
||||
isolateAgeMs: coldMeta.isolateAgeMs,
|
||||
ts: start,
|
||||
}),
|
||||
);
|
||||
|
||||
const response = await route(request, env, pathname);
|
||||
|
||||
// Structured request log for Workers Observability dashboard.
|
||||
|
||||
@@ -7,12 +7,21 @@
|
||||
* share a single bot.command() registration. Visibility only affects:
|
||||
* 1. What scripts/register.js pushes to Telegram's setMyCommands (public only).
|
||||
* 2. What phase-05's /help renderer shows (public + protected).
|
||||
*
|
||||
* Telemetry: a bot.use() middleware installed before command registration
|
||||
* emits a structured `cmd_timing` JSON line via console.log for every
|
||||
* command request. This preserves the original handler references so that
|
||||
* the dispatcher test's strict-equality assertion (`handler === cmd.handler`)
|
||||
* continues to pass.
|
||||
*/
|
||||
|
||||
import { getLastCold } from "../util/request-context.js";
|
||||
import { startTiming } from "../util/timing.js";
|
||||
import { buildRegistry } from "./registry.js";
|
||||
|
||||
/**
|
||||
* Build the registry (if not already built) and register every command with grammY.
|
||||
* Also installs a timing middleware (bot.use) that emits cmd_timing logs.
|
||||
*
|
||||
* @param {import("grammy").Bot} bot
|
||||
* @param {any} env
|
||||
@@ -21,6 +30,37 @@ import { buildRegistry } from "./registry.js";
|
||||
export async function installDispatcher(bot, env) {
|
||||
const reg = await buildRegistry(env);
|
||||
|
||||
// Install timing middleware BEFORE command handlers so it wraps every command.
|
||||
// Uses bot.use() to avoid wrapping individual handlers — keeping handler
|
||||
// references identical to cmd.handler (required by dispatcher.test.js).
|
||||
if (typeof bot.use === "function") {
|
||||
bot.use(async (ctx, next) => {
|
||||
// Only time actual bot_command messages; pass other updates through untimed.
|
||||
const entities = ctx.message?.entities ?? ctx.channelPost?.entities ?? [];
|
||||
const isBotCommand = entities.some((e) => e.type === "bot_command" && e.offset === 0);
|
||||
|
||||
if (!isBotCommand) {
|
||||
return next();
|
||||
}
|
||||
|
||||
// Extract command name without the leading slash for the timing label.
|
||||
const rawText = ctx.message?.text ?? ctx.channelPost?.text ?? "";
|
||||
// Command is the first word (e.g. "/wordle" or "/wordle@botname").
|
||||
const cmdToken = rawText.split(" ")[0].split("@")[0];
|
||||
|
||||
const t = startTiming(cmdToken);
|
||||
const { cold, isolateAgeMs } = getLastCold();
|
||||
|
||||
try {
|
||||
await next();
|
||||
t.end({ cold, isolateAgeMs });
|
||||
} catch (err) {
|
||||
t.end({ cold, isolateAgeMs, error: err instanceof Error ? err.message : String(err) });
|
||||
throw err;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
for (const { cmd } of reg.allCommands.values()) {
|
||||
// grammY's bot.command() matches /cmd and /cmd@botname, case-sensitively,
|
||||
// which naturally satisfies the "exact, case-sensitive" rule for private commands.
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
/**
|
||||
* @file request-context — module-scoped request state shared between
|
||||
* the fetch entry point (index.js) and the dispatcher middleware.
|
||||
*
|
||||
* CF Workers isolates are single-threaded: one request at a time, so a
|
||||
* module-scope variable is a safe per-request stash without race conditions.
|
||||
*
|
||||
* Placing this in a separate module breaks the circular import that would
|
||||
* arise from dispatcher.js importing index.js directly.
|
||||
*/
|
||||
|
||||
/**
|
||||
* Cold-start metadata captured at the top of each fetch() call.
|
||||
* Written by index.js; read by dispatcher timing middleware.
|
||||
*
|
||||
* @type {{ cold: boolean, isolateAgeMs: number }}
|
||||
*/
|
||||
let _lastCold = { cold: false, isolateAgeMs: 0 };
|
||||
|
||||
/**
|
||||
* Store the cold-start metadata for the current request.
|
||||
* Called once per fetch() invocation, before any awaits.
|
||||
*
|
||||
* @param {{ cold: boolean, isolateAgeMs: number }} value
|
||||
*/
|
||||
export function setLastCold(value) {
|
||||
_lastCold = value;
|
||||
}
|
||||
|
||||
/**
|
||||
* Retrieve the cold-start metadata for the current request.
|
||||
* Called by dispatcher timing middleware.
|
||||
*
|
||||
* @returns {{ cold: boolean, isolateAgeMs: number }}
|
||||
*/
|
||||
export function getLastCold() {
|
||||
return _lastCold;
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
/**
|
||||
* @file timing — per-request command timing for soak analysis.
|
||||
* Logs structured JSON via console.log; CF Observability captures it.
|
||||
*
|
||||
* Usage:
|
||||
* const t = startTiming("/wordle");
|
||||
* t.mark("mongo-read");
|
||||
* t.end({ cold: true });
|
||||
* // logs: {"event":"cmd_timing","cmd":"/wordle","total":45,"cold":true,"marks":[...]}
|
||||
*/
|
||||
|
||||
/**
|
||||
* Start a timing context for a command.
|
||||
*
|
||||
* @param {string} cmd - command name, e.g. "/wordle"
|
||||
* @returns {{ mark: (label: string) => void, end: (extra?: object) => void }}
|
||||
*/
|
||||
export function startTiming(cmd) {
|
||||
const t0 = Date.now();
|
||||
const marks = []; // declared at top — fixes the snippet bug from the plan
|
||||
|
||||
return {
|
||||
/**
|
||||
* Record a named checkpoint relative to the start.
|
||||
*
|
||||
* @param {string} label
|
||||
*/
|
||||
mark(label) {
|
||||
marks.push({ label, dt: Date.now() - t0 });
|
||||
},
|
||||
|
||||
/**
|
||||
* Finalize and emit a structured timing log entry.
|
||||
*
|
||||
* @param {object} [extra={}] - additional key/value pairs merged into the log line
|
||||
*/
|
||||
end(extra = {}) {
|
||||
const total = Date.now() - t0;
|
||||
console.log(JSON.stringify({ event: "cmd_timing", cmd, total, ...extra, marks }));
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
// ── Cold-start detection ─────────────────────────────────────────────────────
|
||||
// CF Worker isolates are single-threaded; one isolate handles one request at a
|
||||
// time. The first request in a fresh isolate is "cold". We use a boolean flag
|
||||
// (not isolate_age_ms < 200ms) because Mongo connect itself takes ~1500ms,
|
||||
// making age-based classification unreliable (code-reviewer #11).
|
||||
|
||||
let _isFirst = true;
|
||||
const _isolateBorn = Date.now();
|
||||
|
||||
/**
|
||||
* Returns cold-start metadata for the current request.
|
||||
*
|
||||
* First call in an isolate returns `{ cold: true, isolateAgeMs: ~0 }`.
|
||||
* Subsequent calls return `{ cold: false, isolateAgeMs: <ms-since-boot> }`.
|
||||
*
|
||||
* Call exactly once per incoming request (at the top of the fetch handler).
|
||||
*
|
||||
* @returns {{ cold: boolean, isolateAgeMs: number }}
|
||||
*/
|
||||
export function takeColdFlag() {
|
||||
const cold = _isFirst;
|
||||
_isFirst = false;
|
||||
return { cold, isolateAgeMs: Date.now() - _isolateBorn };
|
||||
}
|
||||
@@ -10,15 +10,22 @@ export function makeFakeBot() {
|
||||
const commandCalls = [];
|
||||
/** @type {Array<{event: string, handler: Function}>} */
|
||||
const onCalls = [];
|
||||
/** @type {Array<Function>} */
|
||||
const useCalls = [];
|
||||
|
||||
return {
|
||||
commandCalls,
|
||||
onCalls,
|
||||
useCalls,
|
||||
command(name, handler) {
|
||||
commandCalls.push({ name, handler });
|
||||
},
|
||||
on(event, handler) {
|
||||
onCalls.push({ event, handler });
|
||||
},
|
||||
/** Records middleware registered via bot.use() (e.g. timing middleware). */
|
||||
use(middleware) {
|
||||
useCalls.push(middleware);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,196 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
aggregateLines,
|
||||
formatReport,
|
||||
parseLogLine,
|
||||
percentile,
|
||||
} from "../../scripts/analyze-soak.js";
|
||||
|
||||
// ── percentile ───────────────────────────────────────────────────────────────
|
||||
|
||||
describe("percentile", () => {
|
||||
it("returns 0 for empty array", () => {
|
||||
expect(percentile([], 50)).toBe(0);
|
||||
});
|
||||
|
||||
it("returns the only element for a single-element array", () => {
|
||||
expect(percentile([42], 50)).toBe(42);
|
||||
expect(percentile([42], 95)).toBe(42);
|
||||
});
|
||||
|
||||
it("p50 of [1,2,3,4,5] = 3", () => {
|
||||
expect(percentile([1, 2, 3, 4, 5], 50)).toBe(3);
|
||||
});
|
||||
|
||||
it("p100 returns last element", () => {
|
||||
expect(percentile([10, 20, 30], 100)).toBe(30);
|
||||
});
|
||||
|
||||
it("p0 returns first element", () => {
|
||||
expect(percentile([10, 20, 30], 0)).toBe(10);
|
||||
});
|
||||
|
||||
it("p95 of 100-element uniform distribution is near top", () => {
|
||||
const data = Array.from({ length: 100 }, (_, i) => i + 1); // 1..100
|
||||
expect(percentile(data, 95)).toBe(95);
|
||||
});
|
||||
});
|
||||
|
||||
// ── parseLogLine ─────────────────────────────────────────────────────────────
|
||||
|
||||
describe("parseLogLine", () => {
|
||||
it("parses a plain NDJSON line", () => {
|
||||
const line = '{"event":"cmd_timing","cmd":"/wordle","total":120,"cold":true,"marks":[]}';
|
||||
expect(parseLogLine(line)).toEqual({
|
||||
event: "cmd_timing",
|
||||
cmd: "/wordle",
|
||||
total: 120,
|
||||
cold: true,
|
||||
marks: [],
|
||||
});
|
||||
});
|
||||
|
||||
it("returns null for empty line", () => {
|
||||
expect(parseLogLine("")).toBeNull();
|
||||
expect(parseLogLine(" ")).toBeNull();
|
||||
});
|
||||
|
||||
it("returns null for non-JSON plain text", () => {
|
||||
expect(parseLogLine("some random log line")).toBeNull();
|
||||
});
|
||||
|
||||
it("extracts JSON embedded in a CSV row", () => {
|
||||
const csvRow =
|
||||
'2024-01-01T00:00:00Z,worker,"{\\"event\\":\\"cmd_timing\\",\\"cmd\\":\\"/loldle\\",\\"total\\":45,\\"cold\\":false,\\"marks\\":[]}"';
|
||||
const result = parseLogLine(csvRow);
|
||||
// May fail to parse double-escaped CSV — test the raw braces path instead.
|
||||
// Use a simpler CSV format where JSON is not double-escaped:
|
||||
const simpleCsv =
|
||||
'2024-01-01,{"event":"cmd_timing","cmd":"/loldle","total":45,"cold":false,"marks":[]},extra';
|
||||
const r2 = parseLogLine(simpleCsv);
|
||||
expect(r2).not.toBeNull();
|
||||
expect(r2.event).toBe("cmd_timing");
|
||||
expect(r2.cmd).toBe("/loldle");
|
||||
});
|
||||
|
||||
it("returns null for malformed JSON", () => {
|
||||
expect(parseLogLine("{bad json}")).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
// ── aggregateLines ────────────────────────────────────────────────────────────
|
||||
|
||||
describe("aggregateLines", () => {
|
||||
const makeTimingLine = (cmd, total, cold) =>
|
||||
JSON.stringify({ event: "cmd_timing", cmd, total, cold, marks: [] });
|
||||
|
||||
it("buckets cold and warm samples separately", () => {
|
||||
const lines = [
|
||||
makeTimingLine("/wordle", 1500, true),
|
||||
makeTimingLine("/wordle", 1400, true),
|
||||
makeTimingLine("/wordle", 50, false),
|
||||
makeTimingLine("/wordle", 60, false),
|
||||
makeTimingLine("/wordle", 55, false),
|
||||
];
|
||||
const { buckets } = aggregateLines(lines, null);
|
||||
expect(buckets.get("/wordle|cold").n).toBe(2);
|
||||
expect(buckets.get("/wordle|warm").n).toBe(3);
|
||||
});
|
||||
|
||||
it("filters to requested commands only", () => {
|
||||
const lines = [makeTimingLine("/wordle", 100, false), makeTimingLine("/loldle", 200, false)];
|
||||
const { buckets } = aggregateLines(lines, ["/wordle"]);
|
||||
expect(buckets.has("/wordle|warm")).toBe(true);
|
||||
expect(buckets.has("/loldle|warm")).toBe(false);
|
||||
});
|
||||
|
||||
it("counts dual-write secondary failures", () => {
|
||||
const lines = [
|
||||
'{"msg":"[dual-kv] secondary write failed","phase":"dual-kv"}',
|
||||
"dual-write:secondary:failed — key wordle:state:1",
|
||||
makeTimingLine("/wordle", 80, false),
|
||||
];
|
||||
const { errors } = aggregateLines(lines, null);
|
||||
expect(errors.dualWriteFail).toBe(2);
|
||||
});
|
||||
|
||||
it("counts mongo connection errors", () => {
|
||||
const lines = [
|
||||
"MongoNetworkError: connection refused",
|
||||
"MongoError: timeout",
|
||||
makeTimingLine("/wordle", 80, false),
|
||||
];
|
||||
const { errors } = aggregateLines(lines, null);
|
||||
expect(errors.mongoError).toBe(2);
|
||||
});
|
||||
|
||||
it("counts CPU time exceeded errors", () => {
|
||||
const lines = ["Worker exceeded CPU time limit", "cpu time exceeded for isolate"];
|
||||
const { errors } = aggregateLines(lines, null);
|
||||
expect(errors.cpuExceeded).toBe(2);
|
||||
});
|
||||
|
||||
it("skips non-cmd_timing JSON events", () => {
|
||||
const lines = [
|
||||
JSON.stringify({ event: "request", cold: true, path: "/webhook" }),
|
||||
makeTimingLine("/wordle", 90, false),
|
||||
];
|
||||
const { buckets } = aggregateLines(lines, null);
|
||||
expect(buckets.size).toBe(1); // only the cmd_timing entry
|
||||
});
|
||||
|
||||
it("returns empty buckets and zero errors for empty input", () => {
|
||||
const { buckets, errors } = aggregateLines([], null);
|
||||
expect(buckets.size).toBe(0);
|
||||
expect(errors.dualWriteFail).toBe(0);
|
||||
expect(errors.mongoError).toBe(0);
|
||||
expect(errors.cpuExceeded).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
// ── formatReport ─────────────────────────────────────────────────────────────
|
||||
|
||||
describe("formatReport", () => {
|
||||
it("produces a markdown table with correct headers", () => {
|
||||
const lines = [
|
||||
JSON.stringify({ event: "cmd_timing", cmd: "/wordle", total: 1500, cold: true, marks: [] }),
|
||||
JSON.stringify({ event: "cmd_timing", cmd: "/wordle", total: 50, cold: false, marks: [] }),
|
||||
];
|
||||
const { buckets, errors } = aggregateLines(lines, null);
|
||||
const report = formatReport(buckets, errors);
|
||||
|
||||
expect(report).toContain("| cmd | cold/warm | n | p50 | p95 | p99 |");
|
||||
expect(report).toContain("/wordle");
|
||||
expect(report).toContain("cold");
|
||||
expect(report).toContain("warm");
|
||||
});
|
||||
|
||||
it("includes error summary section", () => {
|
||||
const { buckets, errors } = aggregateLines([], null);
|
||||
errors.dualWriteFail = 3;
|
||||
const report = formatReport(buckets, errors);
|
||||
expect(report).toContain("## Error Summary");
|
||||
expect(report).toContain("Dual-write secondary failures: 3");
|
||||
expect(report).toContain("Mongo connection errors: 0");
|
||||
expect(report).toContain("CPU time exceeded: 0");
|
||||
});
|
||||
|
||||
it("computes correct p50/p95/p99 for a known dataset", () => {
|
||||
// 10 warm samples: 10,20,...,100
|
||||
const lines = Array.from({ length: 10 }, (_, i) =>
|
||||
JSON.stringify({
|
||||
event: "cmd_timing",
|
||||
cmd: "/ping",
|
||||
total: (i + 1) * 10,
|
||||
cold: false,
|
||||
marks: [],
|
||||
}),
|
||||
);
|
||||
const { buckets, errors } = aggregateLines(lines, null);
|
||||
const report = formatReport(buckets, errors);
|
||||
|
||||
// p50 of [10,20,30,40,50,60,70,80,90,100] = 50 (index 4 with nearest-rank)
|
||||
// p95 = 95th percentile = index ceil(0.95*10)-1 = ceil(9.5)-1 = 10-1 = 9 → 100
|
||||
expect(report).toContain("| /ping | warm | 10 |");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,104 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { startTiming, takeColdFlag } from "../../src/util/timing.js";
|
||||
|
||||
// ── startTiming ──────────────────────────────────────────────────────────────
|
||||
|
||||
describe("startTiming", () => {
|
||||
let consoleSpy;
|
||||
|
||||
beforeEach(() => {
|
||||
consoleSpy = vi.spyOn(console, "log").mockImplementation(() => {});
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
it("end() emits a cmd_timing JSON line via console.log", () => {
|
||||
const t = startTiming("/wordle");
|
||||
t.end();
|
||||
|
||||
expect(consoleSpy).toHaveBeenCalledOnce();
|
||||
const raw = consoleSpy.mock.calls[0][0];
|
||||
const obj = JSON.parse(raw);
|
||||
expect(obj.event).toBe("cmd_timing");
|
||||
expect(obj.cmd).toBe("/wordle");
|
||||
expect(typeof obj.total).toBe("number");
|
||||
expect(obj.total).toBeGreaterThanOrEqual(0);
|
||||
expect(Array.isArray(obj.marks)).toBe(true);
|
||||
});
|
||||
|
||||
it("end() merges extra fields into the log object", () => {
|
||||
const t = startTiming("/loldle");
|
||||
t.end({ cold: true, isolateAgeMs: 42 });
|
||||
|
||||
const obj = JSON.parse(consoleSpy.mock.calls[0][0]);
|
||||
expect(obj.cold).toBe(true);
|
||||
expect(obj.isolateAgeMs).toBe(42);
|
||||
});
|
||||
|
||||
it("mark() records named checkpoints with dt >= 0", () => {
|
||||
const t = startTiming("/help");
|
||||
t.mark("mongo-read");
|
||||
t.mark("handler-done");
|
||||
t.end();
|
||||
|
||||
const obj = JSON.parse(consoleSpy.mock.calls[0][0]);
|
||||
expect(obj.marks).toHaveLength(2);
|
||||
expect(obj.marks[0].label).toBe("mongo-read");
|
||||
expect(obj.marks[1].label).toBe("handler-done");
|
||||
expect(obj.marks[0].dt).toBeGreaterThanOrEqual(0);
|
||||
expect(obj.marks[1].dt).toBeGreaterThanOrEqual(obj.marks[0].dt);
|
||||
});
|
||||
|
||||
it("marks array is empty when no mark() calls made", () => {
|
||||
const t = startTiming("/ping");
|
||||
t.end();
|
||||
const obj = JSON.parse(consoleSpy.mock.calls[0][0]);
|
||||
expect(obj.marks).toEqual([]);
|
||||
});
|
||||
|
||||
it("marks is declared at top — no ReferenceError if mark() called before any await", () => {
|
||||
// This would throw if `marks` were declared after a conditional or inside a closure.
|
||||
expect(() => {
|
||||
const t = startTiming("/fortytwo");
|
||||
t.mark("immediate");
|
||||
t.end();
|
||||
}).not.toThrow();
|
||||
});
|
||||
|
||||
it("total reflects elapsed time (>= 0ms)", () => {
|
||||
const t = startTiming("/trade");
|
||||
t.end();
|
||||
const obj = JSON.parse(consoleSpy.mock.calls[0][0]);
|
||||
expect(obj.total).toBeGreaterThanOrEqual(0);
|
||||
});
|
||||
});
|
||||
|
||||
// ── takeColdFlag ─────────────────────────────────────────────────────────────
|
||||
// NOTE: takeColdFlag uses module-scope state that persists across tests in the
|
||||
// same vitest worker. We cannot reset it between tests — so we capture the
|
||||
// baseline once and assert relative behaviour only.
|
||||
|
||||
describe("takeColdFlag", () => {
|
||||
it("returns an object with cold (boolean) and isolateAgeMs (number)", () => {
|
||||
const result = takeColdFlag();
|
||||
expect(typeof result.cold).toBe("boolean");
|
||||
expect(typeof result.isolateAgeMs).toBe("number");
|
||||
expect(result.isolateAgeMs).toBeGreaterThanOrEqual(0);
|
||||
});
|
||||
|
||||
it("second call in same isolate returns cold=false", () => {
|
||||
// The first call may have already flipped the flag in a prior test.
|
||||
// We call it once here just to ensure it returns cold=false on subsequent invocations.
|
||||
takeColdFlag(); // may be first or subsequent
|
||||
const second = takeColdFlag();
|
||||
expect(second.cold).toBe(false);
|
||||
});
|
||||
|
||||
it("isolateAgeMs is non-decreasing across successive calls", () => {
|
||||
const a = takeColdFlag();
|
||||
const b = takeColdFlag();
|
||||
expect(b.isolateAgeMs).toBeGreaterThanOrEqual(a.isolateAgeMs);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user