kimi-code / packages /minidb /bench /reader-catchup.ts

Add files using upload-large-folder tool

4e23b01 verified about 2 months ago

11 kB

// bench/reader-catchup.ts
//
// Reader-side catch-up benchmark for ClusterDb: one writer child process
// hammers a single hot shard while this process reads keys on that shard
// through a second, read-only ClusterDb. Measures per-read latency and how
// the reader pool keeps its cached shard instance current.
//
//   BEFORE (full reopen per WAL change):
//     every read whose fingerprint check notices a WAL append closes the
//     cached reader and replays snapshot+WAL from scratch — O(shard size).
//   AFTER (incremental WAL catch-up):
//     only the appended WAL frames are scanned and applied — O(delta).
//
// Run:  node --import tsx bench/reader-catchup.ts
// Env:  READER_BENCH_KEYS=10000,50000  READER_BENCH_WINDOW_MS=8000
//       READER_BENCH_VALUE_BYTES=150   READER_BENCH_SHARDS=4
//       READER_BENCH_PACE_MS=5 (writer sleep between ops ~190 ops/s; the WAL
//       then grows steadily enough that a full-reopen-per-read reader can
//       never catch a quiet fingerprint, which pins the disputed cost)
//
// ---------------------------------------------------------------------------
// RESULTS (this machine, node v24.15.0, valueBytes=150, poll=20ms, window
// 8000ms, 4 shards, hot shard = 1, writer pace=5ms):
//
// BEFORE (2026-07-17, pre-change; run: node --import tsx bench/reader-catchup.ts):
//       keys |    build | reads | qps |  p50 ms |  p95 ms |  p99 ms | reopens | catchups | frames
//     10,000 |  1,054ms |    52 | 6/s |   130.5 |   157.2 |   238.7 |      52 |        0 |      0
//     50,000 |  5,054ms |    14 | 2/s |   532.2 |   860.8 |   860.8 |      14 |        0 |      0
//   Every 20ms poll observes fresh WAL appends, so every read pays a full
//   replay of the whole shard (snapshot + rebuild of all derived indexes):
//   read QPS collapses to ~1/reopen-cost and p50 = the reopen cost itself.
//
// AFTER (same command, with incremental catch-up):
//       keys |    build | reads | qps |  p50 ms |  p95 ms |  p99 ms | reopens | catchups | frames
//     10,000 |    868ms |   374 | 47/s |     0.4 |     1.3 |     5.7 |       0 |      374 |  1,388
//     50,000 |  5,671ms |   368 | 46/s |     0.4 |     2.7 |     6.2 |       0 |      368 |  1,345
//   Every poll applies only the appended WAL frames (catchupFrames == the
//   writer's ops) and never reopens: p50 drops ~330x (@10k) / ~1,330x (@50k);
//   read QPS rises 6→47/s and 2→46/s (bounded only by the 20ms poll grid).
// ---------------------------------------------------------------------------

import { spawn } from 'node:child_process';
import fs from 'node:fs/promises';
import os from 'node:os';
import path from 'node:path';
import { fileURLToPath } from 'node:url';
import { MiniDb } from '../src/index.js';
import { ClusterDb, shardDirName } from '../src/cluster/index.js';
import { shardFor } from '../src/cluster/utils.js';

const __dirname = path.dirname(fileURLToPath(import.meta.url));
const WORKER = path.join(__dirname, 'reader-worker.ts');

const SIZES = (process.env.READER_BENCH_KEYS ?? '10000,50000').split(',').map(Number);
const WINDOW_MS = Number(process.env.READER_BENCH_WINDOW_MS ?? 8000);
const VALUE_BYTES = Number(process.env.READER_BENCH_VALUE_BYTES ?? 150);
const SHARDS = Number(process.env.READER_BENCH_SHARDS ?? 4);
const PACE_MS = Number(process.env.READER_BENCH_PACE_MS ?? 5);
const POLL_MS = Number(process.env.READER_BENCH_POLL_MS ?? 20);
const HOT_SHARD = 1;

const fmt = (n: number) => n.toLocaleString('en-US', { maximumFractionDigits: 0 });

interface Report {
  ok: number;
  mode: string;
  n: number;
  ms: number;
  compactMs?: number;
  error?: string;
}

function spawnWorker(args: string[]): { child: ReturnType<typeof spawn>; done: Promise<Report> } {
  const child = spawn(process.execPath, ['--import', 'tsx', WORKER, ...args], { stdio: ['ignore', 'pipe', 'pipe'] });
  let stdout = '';
  let stderr = '';
  child.stdout.on('data', (d) => (stdout += d));
  child.stderr.on('data', (d) => (stderr += d));
  const done = new Promise<Report>((resolve, reject) => {
    child.on('error', reject);
    child.on('close', (code) => {
      const lines = stdout.trim().split('\n').filter((l) => l.startsWith('{'));
      if (code !== 0 || lines.length === 0) {
        reject(new Error(`worker failed (code=${code}) args=${args.join(' ')}\n${stderr}`));
        return;
      }
      const report = JSON.parse(lines[lines.length - 1]!) as Report;
      if (!report.ok) reject(new Error(`worker error args=${args.join(' ')}: ${report.error}`));
      else resolve(report);
    });
  });
  return { child, done };
}

async function runWorker(args: string[]): Promise<Report> {
  const { done } = spawnWorker(args);
  return done;
}

function sleep(ms: number): Promise<void> {
  return new Promise((r) => setTimeout(r, ms));
}

function percentile(sorted: number[], p: number): number {
  if (!sorted.length) return NaN;
  const idx = Math.min(sorted.length - 1, Math.ceil((p / 100) * sorted.length) - 1);
  return sorted[Math.max(0, idx)]!;
}

/** The exact preload key list (same generator as reader-worker 'preload'). */
function preloadKeys(n: number): string[] {
  const keys: string[] = [];
  for (let seq = 0; keys.length < n; seq++) {
    const key = `pre:${seq}`;
    if (shardFor(key, SHARDS) === HOT_SHARD) keys.push(key);
  }
  return keys;
}

interface Row {
  keys: number;
  buildMs: number;
  fullReopenMs: number;
  reads: number;
  qps: number;
  p50: number;
  p95: number;
  p99: number;
  writerOps: number;
  readerReopens: number;
  incrementalCatchups: number;
  catchupFramesApplied: number;
}

async function benchSize(n: number): Promise<Row> {
  const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'minidb-reader-bench-'));
  const keys = preloadKeys(n);
  let db: ClusterDb | null = null;
  try {
    // Build the shard (writes + compaction), then start the hot writer.
    const t0 = performance.now();
    await runWorker(['preload', dir, String(SHARDS), String(HOT_SHARD), String(n), String(VALUE_BYTES)]);
    const buildMs = performance.now() - t0;

    // Full-reopen cost of the built shard: what a reader pays when incremental
    // catch-up cannot serve (first attach, fingerprint mismatch, forced
    // resync). Median of three cold read-only opens of the hot shard.
    const shardDir = path.join(dir, shardDirName(HOT_SHARD, SHARDS));
    const reopens: number[] = [];
    for (let r = 0; r < 3; r++) {
      const t = performance.now();
      const shard = await MiniDb.open({ dir: shardDir, valueCodec: 'json', readOnly: true });
      reopens.push(performance.now() - t);
      await shard.close();
    }
    reopens.sort((a, b) => a - b);
    const fullReopenMs = reopens[1]!;

    db = await ClusterDb.open({ dir, readOnly: true });
    // Warm the reader cache (this first open pays the full replay once).
    for (let i = 0; i < 3; i++) await db.get(keys[i]!);
    const stats0 = db.stats();

    const hammer = spawnWorker(['hammer', dir, String(SHARDS), String(HOT_SHARD), String(VALUE_BYTES), String(PACE_MS)]);
    // The hammer's own shard open replays the (large) shard first, so it only
    // starts appending hundreds of ms after spawn. Wait for the WAL to grow
    // once before measuring, or the early window samples a quiet file.
    const walPath = path.join(dir, shardDirName(HOT_SHARD, SHARDS), 'db.wal');
    const walBase = await fs.stat(walPath).then((s) => s.size, () => -1);
    for (let waited = 0; ; waited += 50) {
      const cur = await fs.stat(walPath).then((s) => s.size, () => -1);
      if (cur !== walBase || waited > 15_000) break;
      await sleep(50);
    }

    // Poll at a fixed interval rather than busy-looping: with a hot writer the
    // WAL changes between any two polls, so every poll pays exactly one
    // refresh — a full reopen before, an incremental catch-up after. A busy
    // loop instead degenerates into a scheduling lottery between the two
    // processes and measures mostly stale-fingerprint fast paths.
    const lat: number[] = [];
    const w0 = performance.now();
    const deadline = w0 + WINDOW_MS;
    let i = 0;
    while (performance.now() < deadline) {
      await sleep(POLL_MS);
      const t = performance.now();
      await db.get(keys[i % keys.length]!);
      lat.push(performance.now() - t);
      i++;
    }
    const windowMs = performance.now() - w0;

    hammer.child.kill('SIGTERM');
    const report = await Promise.race([
      hammer.done,
      sleep(10_000).then(() => {
        hammer.child.kill('SIGKILL');
        return hammer.done;
      }),
    ]);

    lat.sort((a, b) => a - b);
    const stats1 = db.stats();
    return {
      keys: n,
      buildMs,
      fullReopenMs,
      reads: lat.length,
      qps: (lat.length / windowMs) * 1000,
      p50: percentile(lat, 50),
      p95: percentile(lat, 95),
      p99: percentile(lat, 99),
      writerOps: report.n,
      readerReopens: stats1.readerReopens - stats0.readerReopens,
      incrementalCatchups: (stats1.incrementalCatchups ?? 0) - (stats0.incrementalCatchups ?? 0),
      catchupFramesApplied: (stats1.catchupFramesApplied ?? 0) - (stats0.catchupFramesApplied ?? 0),
    };
  } finally {
    if (db) await db.close().catch(() => {});
    await fs.rm(dir, { recursive: true, force: true });
  }
}

async function main(): Promise<void> {
  console.log(
    `\nClusterDb reader catch-up benchmark  (value=${VALUE_BYTES}B pad, window=${WINDOW_MS}ms, poll=${POLL_MS}ms, S=${SHARDS}, hot shard=${HOT_SHARD}, writer pace=${PACE_MS}ms, fsync=no, node ${process.version})`,
  );
  const rows: Row[] = [];
  for (const n of SIZES) {
    process.stdout.write(`building shard with ${fmt(n)} keys ... `);
    const row = await benchSize(n);
    rows.push(row);
    console.log(
      `built in ${fmt(row.buildMs)}ms; fullReopen=${row.fullReopenMs.toFixed(1)}ms; reads=${fmt(row.reads)} (${fmt(row.qps)}/s), ` +
        `p50=${row.p50.toFixed(1)}ms p95=${row.p95.toFixed(1)}ms p99=${row.p99.toFixed(1)}ms, ` +
        `writerOps=${fmt(row.writerOps)}, fullReopens=${fmt(row.readerReopens)}, ` +
        `incrementalCatchups=${fmt(row.incrementalCatchups)}, catchupFrames=${fmt(row.catchupFramesApplied)}`,
    );
  }

  console.log(`\n  ${'keys'.padStart(8)} | ${'build'.padStart(8)} | ${'reopen'.padStart(8)} | ${'reads'.padStart(7)} | ${'qps'.padStart(8)} | ${'p50 ms'.padStart(8)} | ${'p95 ms'.padStart(8)} | ${'p99 ms'.padStart(8)} | ${'reopens'.padStart(8)} | ${'catchups'.padStart(9)} | ${'frames'.padStart(9)}`);
  for (const r of rows) {
    console.log(
      `  ${fmt(r.keys).padStart(8)} | ${fmt(r.buildMs).padStart(8)} | ${r.fullReopenMs.toFixed(1).padStart(8)} | ${fmt(r.reads).padStart(7)} | ${fmt(r.qps).padStart(8)} | ${r.p50.toFixed(1).padStart(8)} | ${r.p95.toFixed(1).padStart(8)} | ${r.p99.toFixed(1).padStart(8)} | ${fmt(r.readerReopens).padStart(8)} | ${fmt(r.incrementalCatchups).padStart(9)} | ${fmt(r.catchupFramesApplied).padStart(9)}`,
    );
  }
  console.log('');
}

main().catch((e) => {
  console.error(e);
  process.exit(1);
});

	// bench/reader-catchup.ts
	//
	// Reader-side catch-up benchmark for ClusterDb: one writer child process
	// hammers a single hot shard while this process reads keys on that shard
	// through a second, read-only ClusterDb. Measures per-read latency and how
	// the reader pool keeps its cached shard instance current.
	//
	// BEFORE (full reopen per WAL change):
	// every read whose fingerprint check notices a WAL append closes the
	// cached reader and replays snapshot+WAL from scratch — O(shard size).
	// AFTER (incremental WAL catch-up):
	// only the appended WAL frames are scanned and applied — O(delta).
	//
	// Run: node --import tsx bench/reader-catchup.ts
	// Env: READER_BENCH_KEYS=10000,50000 READER_BENCH_WINDOW_MS=8000
	// READER_BENCH_VALUE_BYTES=150 READER_BENCH_SHARDS=4
	// READER_BENCH_PACE_MS=5 (writer sleep between ops ~190 ops/s; the WAL
	// then grows steadily enough that a full-reopen-per-read reader can
	// never catch a quiet fingerprint, which pins the disputed cost)
	//
	// ---------------------------------------------------------------------------
	// RESULTS (this machine, node v24.15.0, valueBytes=150, poll=20ms, window
	// 8000ms, 4 shards, hot shard = 1, writer pace=5ms):
	//
	// BEFORE (2026-07-17, pre-change; run: node --import tsx bench/reader-catchup.ts):
	// keys \| build \| reads \| qps \| p50 ms \| p95 ms \| p99 ms \| reopens \| catchups \| frames
	// 10,000 \| 1,054ms \| 52 \| 6/s \| 130.5 \| 157.2 \| 238.7 \| 52 \| 0 \| 0
	// 50,000 \| 5,054ms \| 14 \| 2/s \| 532.2 \| 860.8 \| 860.8 \| 14 \| 0 \| 0
	// Every 20ms poll observes fresh WAL appends, so every read pays a full
	// replay of the whole shard (snapshot + rebuild of all derived indexes):
	// read QPS collapses to ~1/reopen-cost and p50 = the reopen cost itself.
	//
	// AFTER (same command, with incremental catch-up):
	// keys \| build \| reads \| qps \| p50 ms \| p95 ms \| p99 ms \| reopens \| catchups \| frames
	// 10,000 \| 868ms \| 374 \| 47/s \| 0.4 \| 1.3 \| 5.7 \| 0 \| 374 \| 1,388
	// 50,000 \| 5,671ms \| 368 \| 46/s \| 0.4 \| 2.7 \| 6.2 \| 0 \| 368 \| 1,345
	// Every poll applies only the appended WAL frames (catchupFrames == the
	// writer's ops) and never reopens: p50 drops ~330x (@10k) / ~1,330x (@50k);
	// read QPS rises 6→47/s and 2→46/s (bounded only by the 20ms poll grid).
	// ---------------------------------------------------------------------------

	import { spawn } from 'node:child_process';
	import fs from 'node:fs/promises';
	import os from 'node:os';
	import path from 'node:path';
	import { fileURLToPath } from 'node:url';
	import { MiniDb } from '../src/index.js';
	import { ClusterDb, shardDirName } from '../src/cluster/index.js';
	import { shardFor } from '../src/cluster/utils.js';

	const __dirname = path.dirname(fileURLToPath(import.meta.url));
	const WORKER = path.join(__dirname, 'reader-worker.ts');

	const SIZES = (process.env.READER_BENCH_KEYS ?? '10000,50000').split(',').map(Number);
	const WINDOW_MS = Number(process.env.READER_BENCH_WINDOW_MS ?? 8000);
	const VALUE_BYTES = Number(process.env.READER_BENCH_VALUE_BYTES ?? 150);
	const SHARDS = Number(process.env.READER_BENCH_SHARDS ?? 4);
	const PACE_MS = Number(process.env.READER_BENCH_PACE_MS ?? 5);
	const POLL_MS = Number(process.env.READER_BENCH_POLL_MS ?? 20);
	const HOT_SHARD = 1;

	const fmt = (n: number) => n.toLocaleString('en-US', { maximumFractionDigits: 0 });

	interface Report {
	ok: number;
	mode: string;
	n: number;
	ms: number;
	compactMs?: number;
	error?: string;
	}

	function spawnWorker(args: string[]): { child: ReturnType<typeof spawn>; done: Promise<Report> } {
	const child = spawn(process.execPath, ['--import', 'tsx', WORKER, ...args], { stdio: ['ignore', 'pipe', 'pipe'] });
	let stdout = '';
	let stderr = '';
	child.stdout.on('data', (d) => (stdout += d));
	child.stderr.on('data', (d) => (stderr += d));
	const done = new Promise<Report>((resolve, reject) => {
	child.on('error', reject);
	child.on('close', (code) => {
	const lines = stdout.trim().split('\n').filter((l) => l.startsWith('{'));
	if (code !== 0 \|\| lines.length === 0) {
	reject(new Error(`worker failed (code=${code}) args=${args.join(' ')}\n${stderr}`));
	return;
	}
	const report = JSON.parse(lines[lines.length - 1]!) as Report;
	if (!report.ok) reject(new Error(`worker error args=${args.join(' ')}: ${report.error}`));
	else resolve(report);
	});
	});
	return { child, done };
	}

	async function runWorker(args: string[]): Promise<Report> {
	const { done } = spawnWorker(args);
	return done;
	}

	function sleep(ms: number): Promise<void> {
	return new Promise((r) => setTimeout(r, ms));
	}

	function percentile(sorted: number[], p: number): number {
	if (!sorted.length) return NaN;
	const idx = Math.min(sorted.length - 1, Math.ceil((p / 100) * sorted.length) - 1);
	return sorted[Math.max(0, idx)]!;
	}

	/** The exact preload key list (same generator as reader-worker 'preload'). */
	function preloadKeys(n: number): string[] {
	const keys: string[] = [];
	for (let seq = 0; keys.length < n; seq++) {
	const key = `pre:${seq}`;
	if (shardFor(key, SHARDS) === HOT_SHARD) keys.push(key);
	}
	return keys;
	}

	interface Row {
	keys: number;
	buildMs: number;
	fullReopenMs: number;
	reads: number;
	qps: number;
	p50: number;
	p95: number;
	p99: number;
	writerOps: number;
	readerReopens: number;
	incrementalCatchups: number;
	catchupFramesApplied: number;
	}

	async function benchSize(n: number): Promise<Row> {
	const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'minidb-reader-bench-'));
	const keys = preloadKeys(n);
	let db: ClusterDb \| null = null;
	try {
	// Build the shard (writes + compaction), then start the hot writer.
	const t0 = performance.now();
	await runWorker(['preload', dir, String(SHARDS), String(HOT_SHARD), String(n), String(VALUE_BYTES)]);
	const buildMs = performance.now() - t0;

	// Full-reopen cost of the built shard: what a reader pays when incremental
	// catch-up cannot serve (first attach, fingerprint mismatch, forced
	// resync). Median of three cold read-only opens of the hot shard.
	const shardDir = path.join(dir, shardDirName(HOT_SHARD, SHARDS));
	const reopens: number[] = [];
	for (let r = 0; r < 3; r++) {
	const t = performance.now();
	const shard = await MiniDb.open({ dir: shardDir, valueCodec: 'json', readOnly: true });
	reopens.push(performance.now() - t);
	await shard.close();
	}
	reopens.sort((a, b) => a - b);
	const fullReopenMs = reopens[1]!;

	db = await ClusterDb.open({ dir, readOnly: true });
	// Warm the reader cache (this first open pays the full replay once).
	for (let i = 0; i < 3; i++) await db.get(keys[i]!);
	const stats0 = db.stats();

	const hammer = spawnWorker(['hammer', dir, String(SHARDS), String(HOT_SHARD), String(VALUE_BYTES), String(PACE_MS)]);
	// The hammer's own shard open replays the (large) shard first, so it only
	// starts appending hundreds of ms after spawn. Wait for the WAL to grow
	// once before measuring, or the early window samples a quiet file.
	const walPath = path.join(dir, shardDirName(HOT_SHARD, SHARDS), 'db.wal');
	const walBase = await fs.stat(walPath).then((s) => s.size, () => -1);
	for (let waited = 0; ; waited += 50) {
	const cur = await fs.stat(walPath).then((s) => s.size, () => -1);
	if (cur !== walBase \|\| waited > 15_000) break;
	await sleep(50);
	}

	// Poll at a fixed interval rather than busy-looping: with a hot writer the
	// WAL changes between any two polls, so every poll pays exactly one
	// refresh — a full reopen before, an incremental catch-up after. A busy
	// loop instead degenerates into a scheduling lottery between the two
	// processes and measures mostly stale-fingerprint fast paths.
	const lat: number[] = [];
	const w0 = performance.now();
	const deadline = w0 + WINDOW_MS;
	let i = 0;
	while (performance.now() < deadline) {
	await sleep(POLL_MS);
	const t = performance.now();
	await db.get(keys[i % keys.length]!);
	lat.push(performance.now() - t);
	i++;
	}
	const windowMs = performance.now() - w0;

	hammer.child.kill('SIGTERM');
	const report = await Promise.race([
	hammer.done,
	sleep(10_000).then(() => {
	hammer.child.kill('SIGKILL');
	return hammer.done;
	}),
	]);

	lat.sort((a, b) => a - b);
	const stats1 = db.stats();
	return {
	keys: n,
	buildMs,
	fullReopenMs,
	reads: lat.length,
	qps: (lat.length / windowMs) * 1000,
	p50: percentile(lat, 50),
	p95: percentile(lat, 95),
	p99: percentile(lat, 99),
	writerOps: report.n,
	readerReopens: stats1.readerReopens - stats0.readerReopens,
	incrementalCatchups: (stats1.incrementalCatchups ?? 0) - (stats0.incrementalCatchups ?? 0),
	catchupFramesApplied: (stats1.catchupFramesApplied ?? 0) - (stats0.catchupFramesApplied ?? 0),
	};
	} finally {
	if (db) await db.close().catch(() => {});
	await fs.rm(dir, { recursive: true, force: true });
	}
	}

	async function main(): Promise<void> {
	console.log(
	`\nClusterDb reader catch-up benchmark (value=${VALUE_BYTES}B pad, window=${WINDOW_MS}ms, poll=${POLL_MS}ms, S=${SHARDS}, hot shard=${HOT_SHARD}, writer pace=${PACE_MS}ms, fsync=no, node ${process.version})`,
	);
	const rows: Row[] = [];
	for (const n of SIZES) {
	process.stdout.write(`building shard with ${fmt(n)} keys ... `);
	const row = await benchSize(n);
	rows.push(row);
	console.log(
	`built in ${fmt(row.buildMs)}ms; fullReopen=${row.fullReopenMs.toFixed(1)}ms; reads=${fmt(row.reads)} (${fmt(row.qps)}/s), ` +
	`p50=${row.p50.toFixed(1)}ms p95=${row.p95.toFixed(1)}ms p99=${row.p99.toFixed(1)}ms, ` +
	`writerOps=${fmt(row.writerOps)}, fullReopens=${fmt(row.readerReopens)}, ` +
	`incrementalCatchups=${fmt(row.incrementalCatchups)}, catchupFrames=${fmt(row.catchupFramesApplied)}`,
	);
	}

	console.log(`\n ${'keys'.padStart(8)} \| ${'build'.padStart(8)} \| ${'reopen'.padStart(8)} \| ${'reads'.padStart(7)} \| ${'qps'.padStart(8)} \| ${'p50 ms'.padStart(8)} \| ${'p95 ms'.padStart(8)} \| ${'p99 ms'.padStart(8)} \| ${'reopens'.padStart(8)} \| ${'catchups'.padStart(9)} \| ${'frames'.padStart(9)}`);
	for (const r of rows) {
	console.log(
	` ${fmt(r.keys).padStart(8)} \| ${fmt(r.buildMs).padStart(8)} \| ${r.fullReopenMs.toFixed(1).padStart(8)} \| ${fmt(r.reads).padStart(7)} \| ${fmt(r.qps).padStart(8)} \| ${r.p50.toFixed(1).padStart(8)} \| ${r.p95.toFixed(1).padStart(8)} \| ${r.p99.toFixed(1).padStart(8)} \| ${fmt(r.readerReopens).padStart(8)} \| ${fmt(r.incrementalCatchups).padStart(9)} \| ${fmt(r.catchupFramesApplied).padStart(9)}`,
	);
	}
	console.log('');
	}

	main().catch((e) => {
	console.error(e);
	process.exit(1);
	});