MultiforaDB/tests/e2e/big.js

// Big-collection harness: how multiforadb behaves with multi-GB collections.
//
// Spawns its own server, bulk-inserts up to ~5 GB of documents, measures
// insert throughput, log/compaction behavior and server RSS, benchmarks
// find/count/update/delete against the full dataset, then kills the server
// and measures reopen (replay) time and crash durability.
//
//   node tests/e2e/big.js [options]
//     --size <n>      target collection size; k/m/g suffixes (default 5g)
//     --doc-size <n>  approximate bytes per document (default 32k)
//     --batch <n>     docs per insertMany call (default 500)
//     --index <f>     create a secondary index on field f *before* inserting
//                     (entry insert is O(n), so this makes the load quadratic)
//     --oid           use ObjectId _ids (driver-generated): O(1) _id lookups.
//                     Without it, int _ids fall back to a full scan (the _id
//                     fast path is skipped for serialization-ambiguous
//                     numeric classes), so findOne({_id}) costs a scan.
//     --compact-threshold <n>
//                     pass through to the server: log bytes between
//                     compactions (default 16m). Raise for bulk loads.
//     --port <n>      server port (default 27221)
//     --keep          keep the db file after the run
//     --quick         tiny run (256m, 16k docs)
//
// Env: MFDB_BIN  server binary (default ../../zig-out/bin/multiforadb)
//      BIG_DB  db file path (default .zig-cache/big.log)
const { MongoClient } = require('mongodb');
const { spawn } = require('child_process');
const fs = require('fs');
const path = require('path');

const BIN = process.env.MFDB_BIN || path.resolve(__dirname, '../../zig-out/bin/multiforadb');
const PORT = Number(process.env.BIG_PORT || 27221);
const DBFILE = process.env.BIG_DB || path.resolve(__dirname, '../../.zig-cache/big.log');
const URL = `mongodb://127.0.0.1:${PORT}`;

function parseSize(s) {
  const m = /^(\d+(?:\.\d+)?)([kmgt]?)$/i.exec(String(s).trim());
  if (!m) throw new Error(`bad size '${s}'`);
  const mult = { '': 1, k: 1 << 10, m: 1 << 20, g: 1 << 30, t: 1 << 40 }[m[2].toLowerCase()];
  return Math.round(parseFloat(m[1]) * mult);
}

let opt = { size: '5g', docSize: '32k', batch: 500, index: null, port: PORT, keep: false, oid: false, compactThreshold: null };
for (let i = 2; i < process.argv.length; i++) {
  const a = process.argv[i];
  if (a === '--quick') { opt.size = '256m'; opt.docSize = '16k'; }
  else if (a === '--keep') opt.keep = true;
  else if (a === '--oid') opt.oid = true;
  else if (a === '--index') opt.index = process.argv[++i];
  else if (a.startsWith('--size=')) opt.size = a.slice(7);
  else if (a.startsWith('--doc-size=')) opt.docSize = a.slice(11);
  else if (a.startsWith('--batch=')) opt.batch = Number(a.slice(8));
  else if (a.startsWith('--port=')) opt.port = Number(a.slice(7));
  else if (a.startsWith('--compact-threshold=')) opt.compactThreshold = a.slice(20);
  else if (a === '--size') opt.size = process.argv[++i];
  else if (a === '--doc-size') opt.docSize = process.argv[++i];
  else if (a === '--batch') opt.batch = Number(process.argv[++i]);
  else if (a === '--compact-threshold') opt.compactThreshold = process.argv[++i];
  else { console.error(`unknown option ${a}`); process.exit(2); }
}

const SIZE = parseSize(opt.size);
const DOC_SIZE = parseSize(opt.docSize);
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

let server = null;
let serverLog = '';
let serverDead = false;
function cleanup() {
  if (server && !serverDead) {
    try { server.kill('SIGKILL'); } catch {}
  }
}
process.on('exit', cleanup);
process.on('SIGINT', () => { cleanup(); process.exit(130); });
process.on('SIGTERM', () => { cleanup(); process.exit(143); });
function startServer() {
  return new Promise((resolve, reject) => {
    const t0 = Date.now();
    serverDead = false;
    const args = ['--port', String(opt.port), '--db', DBFILE];
    if (opt.compactThreshold) args.push('--compact-threshold', opt.compactThreshold);
    server = spawn(BIN, args, { stdio: ['ignore', 'pipe', 'pipe'] });
    server.stdout.on('data', (d) => (serverLog += d));
    server.stderr.on('data', (d) => (serverLog += d));
    server.on('error', (e) => reject(new Error(`cannot start ${BIN}: ${e.message}`)));
    server.on('exit', (code, sig) => {
      serverDead = true;
      if (code !== null && sig === null) serverLog += `\n[child exited rc=${code}]`;
    });
    const deadline = Date.now() + 120000;
    (async () => {
      while (Date.now() < deadline) {
        if (serverDead) {
          reject(new Error(`server child exited during start (port ${opt.port} busy?)\n${serverLog}`));
          return;
        }
        const c = new MongoClient(URL, { serverSelectionTimeoutMS: 1000 });
        try {
          await c.connect();
          await c.db('admin').command({ ping: 1 });
          await c.close();
          return resolve(Date.now() - t0);
        } catch {
          try { await c.close(); } catch {}
          await sleep(100);
        }
      }
      reject(new Error(`server did not come up on :${opt.port}\n${serverLog}`));
    })();
  });
}
async function stopServer(sig = 'SIGKILL') {
  if (!server) return;
  const exited = new Promise((r) => server.once('exit', r));
  server.kill(sig);
  await Promise.race([exited, sleep(5000)]);
  serverDead = true;
  server = null;
}
// Synchronous on purpose. The async version sampled inside setInterval and a
// short run could finish before any sample landed, reporting RSS 0 -- which reads
// as "measured and tiny" rather than "not measured".
function rssMB() {
  if (!server) return 0;
  try {
    const out = require('child_process').execSync(`ps -o rss= -p ${server.pid}`, { encoding: 'utf8' }).trim();
    return Math.round(Number(out) / 1024);
  } catch { return 0; }
}

const report = [];
function row(label, value) { report.push([label, value]); console.log(`  ${String(label).padEnd(46)} ${value}`); }

async function main() {
  if (!fs.existsSync(BIN)) {
    console.error(`server binary not found at ${BIN} — run \`zig build\` first`);
    process.exit(1);
  }
  fs.rmSync(DBFILE, { force: true });
  fs.rmSync(DBFILE + '.data', { force: true });
  const fmt = (n) => (n >= 1e9 ? (n / 1e9).toFixed(2) + ' GB' : n >= 1e6 ? (n / 1e6).toFixed(1) + ' MB' : n >= 1e3 ? (n / 1e3).toFixed(1) + ' KB' : n + ' B');
  console.log(`multiforadb big-collection harness`);
  console.log(`  size ${fmt(SIZE)} · doc ~${fmt(DOC_SIZE)} · batch ${opt.batch} · index ${opt.index || 'none'} · ids ${opt.oid ? 'ObjectId' : 'int'} · compact-threshold ${opt.compactThreshold || '16m'} · db ${DBFILE}`);

  console.log('\n== server start (fresh log) ==');
  const openMs = await startServer();
  row('open + first ping (fresh log)', `${openMs} ms`);

  const client = new MongoClient(URL, { serverSelectionTimeoutMS: 10000 });
  await client.connect();
  const db = client.db('big');
  const coll = db.collection('items');
  await coll.drop().catch(() => {});

  if (opt.index) {
    const t0 = Date.now();
    await coll.createIndex({ [opt.index]: 1 });
    row(`createIndex({${opt.index}: 1}) on empty coll`, `${Date.now() - t0} ms`);
  }

  // ---- insert ------------------------------------------------------------
  console.log('\n== insert ==');
  const nDocs = Math.max(1, Math.ceil(SIZE / DOC_SIZE));
  const payloadLen = Math.max(1, DOC_SIZE - 130); // bson overhead for _id/k/p/ts/payload
  const payload = 'x'.repeat(payloadLen);
  const { ObjectId } = require('mongodb');
  const t0 = Date.now();
  const logSamples = [{ t: 0, size: fs.existsSync(DBFILE) ? fs.statSync(DBFILE).size : 0, rss: rssMB() }];
  const sampler = setInterval(() => {
    logSamples.push({ t: Date.now() - t0, size: fs.existsSync(DBFILE) ? fs.statSync(DBFILE).size : 0, rss: rssMB() });
  }, 2000);

  // Rate curve: avg MB/s per progress chunk, to show how throughput changes
  // as the log grows (compaction rewrites + fsync-per-record dominate).
  const rates = [];
  let lastProgressT = t0;
  let lastProgressDocs = 0;
  let midDocId = null; // ObjectId of the mid doc (oid mode), captured at insert
  let lastDocId = null;
  let inserted = 0;
  const midIndex = Math.floor(nDocs / 2);
  try {
    while (inserted < nDocs) {
      const n = Math.min(opt.batch, nDocs - inserted);
      const docs = new Array(n);
      for (let i = 0; i < n; i++) {
        const id = inserted + i + 1;
        const doc = opt.oid
          ? { _id: new ObjectId(), k: id % 1000, p: id % 5000, ts: new Date(Date.UTC(2024, 0, 1) + id * 1000), payload }
          : { _id: id, k: id % 1000, p: id % 5000, ts: new Date(Date.UTC(2024, 0, 1) + id * 1000), payload };
        if (id === midIndex) midDocId = doc._id;
        if (id === nDocs) lastDocId = doc._id;
        docs[i] = doc;
      }
      await coll.insertMany(docs, { ordered: false });
      inserted += n;
      if (inserted % Math.max(1000, Math.floor(nDocs / 20)) < n) {
        const dt = (Date.now() - t0) / 1000;
        const mb = inserted * DOC_SIZE / 1e6;
        const chunkMb = (inserted - lastProgressDocs) * DOC_SIZE / 1e6;
        const chunkT = (Date.now() - lastProgressT) / 1000;
        rates.push(+(chunkMb / chunkT).toFixed(1));
        lastProgressT = Date.now();
        lastProgressDocs = inserted;
        console.log(`    ${inserted.toLocaleString()} docs · ${fmt(inserted * DOC_SIZE)} · ${(mb / dt).toFixed(1)} MB/s avg`);
      }
    }
  } finally {
    clearInterval(sampler);
  }
  const insertMs = Date.now() - t0;
  const bytes = inserted * DOC_SIZE;
  row('docs inserted', inserted.toLocaleString());
  row('approx bytes', fmt(bytes));
  row('wall time', `${(insertMs / 1000).toFixed(1)} s`);
  row('throughput', `${(bytes / 1e6 / (insertMs / 1000)).toFixed(1)} MB/s (${(inserted / (insertMs / 1000)).toFixed(0)} docs/s)`);
  row('rate curve (MB/s per chunk)', rates.join(' → ') || 'n/a');

  // A shrinking log means a *checkpoint* now, not a compaction: the checkpoint
  // publishes the data file and truncates the log to its header. Compaction
  // (which rewrites the data file) leaves no signature in the log's size, so
  // this counter cannot see it and must not claim to.
  const truncations = logSamples.filter((s, i) => i > 0 && s.size < logSamples[i - 1].size - 2 * 1024 * 1024).length;
  const sizes = logSamples.map((s) => s.size);
  row('log file size (min → final)', `${fmt(Math.min(...sizes))} → ${fmt(sizes[sizes.length - 1])}`);
  const dataFileSize = fs.existsSync(DBFILE + '.data') ? fs.statSync(DBFILE + '.data').size : 0;
  row('data file size', fmt(dataFileSize));
  row('log truncations (checkpoints)', truncations, '(log shrank by >2MB between samples)');
  const peakRss = Math.max(...logSamples.map((s) => s.rss));
  row('peak server RSS', `${peakRss} MB`, '(a write touches its pages; see the after-reopen row)');

  // ---- find / read -------------------------------------------------------
  console.log('\n== find / read on full dataset ==');
  const bench = async (label, fn, min = 1) => {
    const a = Date.now();
    const res = await fn();
    const ms = Date.now() - a;
    row(`${label}`, `${ms < 1000 ? ms + ' ms' : (ms / 1000).toFixed(2) + ' s'}${res !== undefined ? `  (${res})` : ''}`);
    return ms;
  };
  await bench('countDocuments({})', async () => {
    const n = await coll.countDocuments({});
    if (n !== inserted) throw new Error(`count ${n} != ${inserted}`);
    return `${n.toLocaleString()} docs`;
  });
  await bench('findOne({_id: mid}) — ' + (opt.oid ? 'docs-map fast path' : 'scan (int _id: fast path skipped)'), async () => {
    const d = await coll.findOne({ _id: opt.oid ? midDocId : midIndex });
    if (!d) throw new Error('miss');
  });
  await bench('findOne({_id: last})', async () => {
    const d = await coll.findOne({ _id: opt.oid ? lastDocId : nDocs });
    if (!d) throw new Error('miss');
  });
  if (opt.index) {
    await bench(`find({${opt.index}: 4242}).count() — via index`, async () => {
      const n = await coll.countDocuments({ k: 4242 });
      if (n < 1) throw new Error('no hits');
      return `${n} hits`;
    });
  }
  await bench('find({k: 4242}).count() — scan', async () => {
    const n = await coll.countDocuments({ k: 4242 });
    return `${n} hits`;
  });
  await bench('find({p: {$gte, $lt}}).count() — range scan', async () => {
    const lo = 1000, hi = 2000;
    const n = await coll.countDocuments({ p: { $gte: lo, $lt: hi } });
    return `${n} hits`;
  });
  await bench('find({}).sort({_id:-1}).limit(20) — full scan + sort', async () => {
    const docs = await coll.find({}).sort({ _id: -1 }).limit(20).toArray();
    if (docs.length !== 20) throw new Error('bad page');
  });
  await bench('find({}, {proj: _id,k,p}).limit(500) — page', async () => {
    const docs = await coll.find({}, { projection: { payload: 0 } }).limit(500).toArray();
    if (docs.length !== 500) throw new Error('short page');
  });

  // ---- write ops against the full dataset ---------------------------------
  console.log('\n== point write ops ==');
  const midId = opt.oid ? midDocId : midIndex;
  await bench('updateOne({_id: mid}, {$set}) — 1 fsync', async () => {
    const r = await coll.updateOne({ _id: midId }, { $set: { touch: Date.now() } });
    if (r.modifiedCount !== 1) throw new Error('miss');
  });
  await bench('updateMany({k: 7}, {$inc}) — ~N/1000 fsyncs', async () => {
    const r = await coll.updateMany({ k: 7 }, { $inc: { hits: 1 } });
    return `${r.modifiedCount} modified`;
  });
  await bench('deleteOne({_id: mid}) + re-insert', async () => {
    await coll.deleteOne({ _id: midId });
    if (opt.oid) {
      await coll.insertOne({ _id: new ObjectId(), k: midIndex % 1000, p: midIndex % 5000, payload });
    } else {
      await coll.insertOne({ _id: midIndex, k: midIndex % 1000, p: midIndex % 5000, payload });
    }
  });
  await bench('aggregate $group by k', async () => {
    const out = await coll.aggregate([{ $group: { _id: '$k', n: { $sum: 1 } } }]).toArray();
    return `${out.length} groups`;
  });

  // ---- reopen (replay) ----------------------------------------------------
  console.log('\n== durability ==');
  await client.close();
  await stopServer('SIGKILL');
  const reopenMs = await startServer();
  row('kill -9 then reopen', `${(reopenMs / 1000).toFixed(1)} s`);
  // The point of the mmap work: what an open pays for is the working set, not
  // the size of the database. Sampled right after the server answers its first
  // ping, before any query has touched a document.
  const rssAfterReopen = rssMB();
  row('RSS after reopen (working set)', `${rssAfterReopen} MB`);
  const c2 = new MongoClient(URL, { serverSelectionTimeoutMS: 10000 });
  await c2.connect();
  const db2 = c2.db('big');
  const coll2 = db2.collection('items');
  const afterRestart = await coll2.countDocuments({});
  row('count after restart', `${afterRestart.toLocaleString()} (${afterRestart === inserted ? 'OK' : 'MISMATCH!'})`);
  const spot = await coll2.findOne({ _id: opt.oid ? lastDocId : nDocs });
  row('last doc intact after restart', spot ? `payload ${spot.payload.length}B` : 'MISSING!');
  if (afterRestart !== inserted || !spot) throw new Error('durability check failed');

  // Crash-durability: every write is fsynced before it is acknowledged, so a
  // kill -9 right after an insert must not lose it.
  const crash = db2.collection('crash');
  await crash.drop().catch(() => {});
  let committed = 0;
  for (let i = 1; i <= 200; i++) {
    await crash.insertOne({ _id: i, seq: i });
    committed = i;
  }
  await c2.close();
  await stopServer('SIGKILL');
  await startServer();
  const c3 = new MongoClient(URL, { serverSelectionTimeoutMS: 10000 });
  await c3.connect();
  const db3 = c3.db('big');
  const crashN = await db3.collection('crash').countDocuments({});
  row('kill -9 after 200 committed writes', `${crashN}/200 survived (${crashN === 200 ? 'OK' : 'MISMATCH!'})`);
  await c3.close();

  if (!opt.keep) {
    fs.rmSync(DBFILE, { force: true });
    fs.rmSync(DBFILE + '.data', { force: true });
  }
  await stopServer('SIGKILL');

  console.log('\n== summary ==');
  // Values captured while the server was alive: by the time the summary prints,
  // it has been stopped and its files removed, so sampling here reports zero.
  console.log(`  multiforadb handles a ${fmt(bytes)} collection in a ${fmt(dataFileSize)} data file`);
  console.log(`  peak RSS during load ${peakRss} MB; after reopen ${rssAfterReopen} MB — the working set, not the data size`);
  console.log(`  insert: ${(bytes / 1e6 / (insertMs / 1000)).toFixed(1)} MB/s — fsync per write is by design (crash safety)`);
  if (truncations > 0) {
    console.log(`  ${truncations} checkpoints reclaimed the log during the load, which is why it ends at ${fmt(sizes[sizes.length - 1])} rather than ${fmt(bytes)}`);
  }
  console.log('BIG_OK');
}

main().catch((e) => {
  console.error('BIG_FAIL', e);
  console.log('--- server log tail ---');
  console.log(serverLog.split('\n').slice(-40).join('\n'));
  process.exit(1);
});