// E2E part 6: the full lifecycle, self-contained. // // Spawns its own multiforadb server on a fresh log file and drives the whole // feature surface through the official driver: CRUD + query operators + // aggregation + error codes + secondary indexes + TTL expiry + admin // commands, then restarts the server twice — once gracefully, once with // kill -9 mid-write — and verifies that everything (data, indexes, TTL // state) survives both. // // Unlike the other e2e files it needs no server running beforehand: // // node tests/e2e/e2e6.js // // Env: E2E6_PORT listen port (default 27220) // MFDB_BIN server binary (default ../../zig-out/bin/multiforadb) // E2E6_KEEP keep the log file after the run const { MongoClient, ObjectId } = require('mongodb'); const { spawn } = require('child_process'); const fs = require('fs'); const path = require('path'); const PORT = Number(process.env.E2E6_PORT || 27220); const BIN = process.env.MFDB_BIN || path.resolve(__dirname, '../../zig-out/bin/multiforadb'); const DBFILE = process.env.E2E6_DB || path.resolve(__dirname, '../../.zig-cache/e2e6-full.log'); const URL = `mongodb://127.0.0.1:${PORT}`; const results = []; function check(name, cond, detail = '') { results.push({ name, ok: !!cond, detail: String(detail) }); if (!cond) console.error(` ✗ ${name} ${detail}`); } const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); async function waitFor(fn, timeoutMs = 20000, stepMs = 200) { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { if (await fn()) return true; await sleep(stepMs); } return false; } let server = null; let serverLog = ''; let serverDead = false; // Never leak the spawned server: kill it no matter how the test exits. 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(fresh = false) { return new Promise((resolve, reject) => { // Only the very first start must wipe the log; restarts must reuse it. if (fresh) fs.rmSync(DBFILE, { force: true }); serverDead = false; server = spawn(BIN, ['--port', String(PORT), '--db', DBFILE, '--ttl-sweep-secs', '1'], { 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) => { // A child that dies (e.g. address already in use) must fail the start; // otherwise the ping poll below would talk to a *stale* server on the // same port and the whole run would go against the wrong database. serverDead = true; if (code !== null && sig === null) serverLog += `\n[child exited rc=${code}]`; }); // Replay happens before the listener opens, so a successful connect is // also the reopen benchmark. Poll until the server answers ping. const deadline = Date.now() + 15000; (async () => { while (Date.now() < deadline) { if (serverDead) { reject(new Error(`server child exited during start (port ${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(); } catch { try { await c.close(); } catch {} await sleep(100); } } reject(new Error(`server did not come up on :${PORT}\n${serverLog}`)); })(); }); } async function stopServer(sig = 'SIGTERM') { 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; } async function expectCode(fn, code, name) { let err = null; try { await fn(); } catch (e) { err = e; } check(name, err && err.code === code, err ? `code ${err.code}: ${err.message}` : 'no error'); } async function phase1(client, db) { const users = db.collection('users'); await users.drop().catch(() => {}); // ---- insert ----------------------------------------------------------- await users.insertOne({ name: 'alice', age: 30, tags: ['a', 'b'], scores: [3, 6, 9], email: 'a@x.io' }); const many = await users.insertMany([ { name: 'bob', age: 25, tags: ['b'], scores: [1], email: 'b@x.io' }, { name: 'carol', age: 35, tags: ['c', 'a'], scores: [8, 5], email: 'c@x.io' }, { name: 'dave', age: 40, tags: [], scores: [4, 4, 4], email: 'd@x.io' }, ]); check('insertMany acknowledged', many.acknowledged === true, many); check('auto _id assigned', ObjectId.isValid(many.insertedIds[0])); const aliceId = (await users.findOne({ name: 'alice' }))._id; check('explicit _id round-trips', (await users.findOne({ _id: many.insertedIds[0] })).name === 'bob'); // ---- find: operators --------------------------------------------------- check('$gt + sort desc', (await users.find({ age: { $gt: 28 } }).sort({ age: -1 }).toArray()).map((d) => d.name).join(',') === 'dave,carol,alice'); check('$gte', (await users.countDocuments({ age: { $gte: 30 } })) === 3); check('$lt', (await users.countDocuments({ age: { $lt: 30 } })) === 1); check('$lte', (await users.countDocuments({ age: { $lte: 30 } })) === 2); check('$ne', (await users.countDocuments({ age: { $ne: 30 } })) === 3); check('$in', (await users.find({ name: { $in: ['alice', 'bob'] } }).count()) === 2); check('$nin', (await users.countDocuments({ name: { $nin: ['alice', 'bob', 'carol', 'dave'] } })) === 0); check('$exists true', (await users.countDocuments({ tags: { $exists: true } })) === 4); check('$exists false', (await users.countDocuments({ ghost: { $exists: false } })) === 4); check('$regex anchors', (await users.find({ name: /^[bc]/ }).toArray()).length === 2); check('$regex case-insensitive', (await users.countDocuments({ name: /^ALICE$/i })) === 1); check('$not', (await users.countDocuments({ age: { $not: { $gt: 30 } } })) === 2); check('$and', (await users.countDocuments({ $and: [{ age: { $gte: 25 } }, { age: { $lt: 40 } }] })) === 3); check('$or', (await users.countDocuments({ $or: [{ name: 'alice' }, { name: 'dave' }] })) === 2); check('$nor', (await users.countDocuments({ $nor: [{ name: 'alice' }, { name: 'dave' }] })) === 2); check('$size', (await users.countDocuments({ tags: { $size: 2 } })) === 2); check('$all', (await users.countDocuments({ tags: { $all: ['a', 'b'] } })) === 1); check('$elemMatch', (await users.countDocuments({ scores: { $elemMatch: { $gte: 5, $lt: 9 } } })) === 2); check('dot path', (await users.findOne({ 'tags.0': 'c' })).name === 'carol'); check('dot path numeric index', (await users.findOne({ 'scores.0': 3 })).name === 'alice'); check('array equality', (await users.countDocuments({ scores: [4, 4, 4] })) === 1); // ---- find: sort / skip / limit / projection ---------------------------- check('skip+limit+sort', (await users.find({}).sort({ age: 1 }).skip(1).limit(2).toArray()).map((d) => d.name).join(',') === 'alice,carol'); const proj = await users.findOne({ name: 'alice' }, { projection: { _id: 0, name: 1 } }); check('projection', proj.name === 'alice' && proj.age === undefined, JSON.stringify(proj)); // ---- counts ------------------------------------------------------------ check('countDocuments', (await users.countDocuments({})) === 4); check('estimatedDocumentCount', (await users.estimatedDocumentCount()) === 4); // ---- update ------------------------------------------------------------ const u1 = await users.updateOne({ name: 'alice' }, { $set: { vip: true }, $inc: { age: 1 } }); check('updateOne nModified', u1.modifiedCount === 1, u1); const alice = await users.findOne({ _id: aliceId }); check('$set+$inc applied', alice.vip === true && alice.age === 31, JSON.stringify(alice)); await users.updateOne({ name: 'carol' }, { $unset: { tags: '' } }); check('$unset', (await users.findOne({ name: 'carol' })).tags === undefined); await users.updateOne({ name: 'dave' }, { $push: { tags: 'x' }, $rename: { vip: 'member' } }); const dave = await users.findOne({ name: 'dave' }); check('$push+$rename', dave.tags.length === 1 && dave.member === undefined, JSON.stringify(dave)); await users.updateOne({ name: 'dave' }, { $pull: { tags: 'x' } }); check('$pull', (await users.findOne({ name: 'dave' })).tags.length === 0); const um = await users.updateMany({}, { $set: { seen: true } }); check('updateMany', um.modifiedCount === 4, um); const ups = await users.updateOne({ name: 'erin' }, { $set: { age: 28 } }, { upsert: true }); check('upsert create', ups.upsertedCount === 1 && ups.matchedCount === 0, ups); const ups2 = await users.updateOne({ name: 'erin' }, { $set: { age: 29 } }, { upsert: true }); check('upsert match', ups2.upsertedCount === 0 && ups2.modifiedCount === 1, ups2); // ---- findOneAndUpdate / Delete ------------------------------------------ const fam = await users.findOneAndUpdate({ name: 'bob' }, { $set: { lucky: true } }, { returnDocument: 'after' }); check('findOneAndUpdate returns new', (fam.value ?? fam).lucky === true); const famDel = await users.findOneAndDelete({ name: 'erin' }); check('findOneAndDelete', (famDel.value ?? famDel).name === 'erin'); // ---- aggregate ----------------------------------------------------------- const grp = await users .aggregate([ { $match: { age: { $gte: 25 } } }, { $group: { _id: '$tags.length', total: { $sum: '$age' } } }, { $sort: { _id: 1 } }, ]) .toArray(); check('aggregate $match+$group+$sum', grp.length >= 1 && grp.some((g) => g.total > 0), JSON.stringify(grp)); check('aggregate $count', (await users.aggregate([{ $match: {} }, { $count: 'n' }]).toArray())[0].n === 4); // $project here is field selection (inclusion/exclusion); computed fields // like `who: '$name'` are outside the documented surface. const projAgg = await users.aggregate([{ $match: { name: 'alice' } }, { $project: { _id: 0, name: 1 } }]).toArray(); check('aggregate $project', projAgg.length === 1 && projAgg[0].name === 'alice' && projAgg[0].age === undefined, JSON.stringify(projAgg)); // ---- duplicate key ------------------------------------------------------- await expectCode(() => users.insertOne({ _id: many.insertedIds[0], name: 'clobber' }), 11000, 'duplicate _id rejected (11000)'); // ---- indexes ------------------------------------------------------------- await users.createIndex({ email: 1 }, { unique: true }); await users.createIndex({ name: 1, age: 1 }, { name: 'name_1_age_1' }); await users.createIndex({ nickname: 1 }, { sparse: true, name: 'nickname_1_sparse' }); await users.createIndex({ score: -1 }); const idxs = await users.indexes(); const names = idxs.map((i) => i.name).sort(); check('indexes listed', names.join(',') === '_id_,email_1,name_1_age_1,nickname_1_sparse,score_-1', names.join(',')); await users.updateOne({ name: 'dave' }, { $set: { email: 'dave@x.io', nickname: 'davie' } }); check('unique index find', (await users.find({ email: 'dave@x.io' }).toArray()).length === 1); await expectCode(() => users.insertOne({ name: 'dup', email: 'dave@x.io' }), 11000, 'unique index rejects dup (11000)'); check('compound prefix find', (await users.find({ name: 'alice', age: 31 }).toArray()).length === 1); check('descending index created', idxs.some((i) => i.key && i.key.score === -1)); await users.dropIndex('name_1_age_1'); check('dropIndex', !(await users.indexes()).some((i) => i.name === 'name_1_age_1')); await users.dropIndexes(); const afterAll = await users.indexes(); check('dropIndexes keeps only _id_', afterAll.length === 1 && afterAll[0].name === '_id_', JSON.stringify(afterAll.map((i) => i.name))); // Recreate the unique index: the restart phase depends on it surviving. // Every doc has a distinct email (inserted above), so this is legal — a // unique index over docs that *lack* the field would be E11000 (duplicate // null), exactly as in MongoDB. await users.createIndex({ email: 1 }, { unique: true }); // ---- TTL ----------------------------------------------------------------- const sessions = db.collection('sessions'); await sessions.drop().catch(() => {}); await sessions.createIndex({ expireAt: 1 }, { expireAfterSeconds: 1 }); const now = Date.now(); await sessions.insertMany([ { _id: 'past', expireAt: new Date(now - 60_000) }, { _id: 'future', expireAt: new Date(now + 3_600_000) }, ]); check('TTL index listed', (await sessions.indexes()).some((i) => i.name === 'expireAt_1' && Number(i.expireAfterSeconds) === 1)); check('TTL past-doc expired', await waitFor(async () => (await sessions.countDocuments({ _id: 'past' })) === 0)); check('TTL future-doc survives', (await sessions.countDocuments({ _id: 'future' })) === 1); // ---- admin ---------------------------------------------------------------- const colls = await db.listCollections({}, { nameOnly: true }).toArray(); check('listCollections', colls.some((c) => c.name === 'users') && colls.some((c) => c.name === 'sessions')); const dbs = await db.admin().listDatabases(); check('listDatabases has e2e6', dbs.databases.some((d) => d.name === 'e2e6')); // dropDatabase must run against a scratch db so it can't nuke the data the // later phases depend on. const scratch = client.db('e2e6_scratch'); await scratch.collection('scratchme').insertOne({ x: 1 }); await scratch.dropDatabase(); const dbs2 = await db.admin().listDatabases(); check('dropDatabase removes it', !dbs2.databases.some((d) => d.name === 'e2e6_scratch'), JSON.stringify(dbs2.databases.map((d) => d.name))); return { users, sessions, aliceId }; } async function phase2(client, { users, aliceId }) { // Compaction only means something when writes *discard* data: the log is // append-only, so replace/delete records pile up as junk until the // 16 MiB threshold triggers a rewrite of just the live documents. // // insert 2000 x 12KB (~24 MB) // replace all 2000 (+24 MB junk) // delete half (+12 MB junk) // // ~60 MB written; a working compactor leaves the file near the live // size (~12 MB + one 16 MB epoch), a broken one leaves ~60 MB. const bulk = client.db('e2e6').collection('bulk'); await bulk.drop().catch(() => {}); const payload = 'z'.repeat(12 * 1024); for (let b = 0; b < 4; b++) { const docs = Array.from({ length: 500 }, (_, i) => ({ _id: b * 500 + i, g: (b * 500 + i) % 2, payload })); await bulk.insertMany(docs); } if (process.env.E2E6_DEBUG) { const dbg = await users.findOne({ name: 'alice' }); console.log('DEBUG phase2 after bulk-insert alice _id:', String(dbg._id), 'same:', String(dbg._id) === String(aliceId)); } // Watch the log file while the replace junk accumulates. Compaction is // fast (one fsync for the whole rewrite), so a sampled drop can be tiny; // the deterministic signals are that the file *peaked* well above where // it ended (junk accumulated) and *ended* near the live size (compaction // reclaimed it). ~48 MB of records are written here and ~12 MB of it // survives, so without compaction the file would end near 48 MB. // // Both bounds are relative to the live size on purpose: compaction now // triggers on the share of the log that is garbage rather than on bytes // appended, so the absolute peak depends on when that share crosses the // threshold and is not a stable number to assert on. let peakSize = fs.statSync(DBFILE).size; const watcher = setInterval(() => { const s = fs.statSync(DBFILE).size; if (s > peakSize) peakSize = s; }, 10); const payload2 = 'q'.repeat(12 * 1024); const replaced = await bulk.updateMany({}, { $set: { payload: payload2 } }); check('bulk replace logged per doc', replaced.modifiedCount === 2000, replaced); const deleted = await bulk.deleteMany({ g: 0 }); check('bulk delete half', deleted.deletedCount === 1000, deleted); clearInterval(watcher); const logSize = fs.statSync(DBFILE).size; if (process.env.E2E6_DEBUG) { const dbg = await users.findOne({ name: 'alice' }); console.log('DEBUG phase2 after replace+delete alice _id:', String(dbg._id), 'same:', String(dbg._id) === String(aliceId)); console.log('DEBUG phase2 log size:', logSize); } check( 'compaction reclaimed junk (file ~ live size)', logSize < 24 * 1024 * 1024 && peakSize > logSize * 1.4, `file peaked at ${(peakSize / 1e6).toFixed(1)}MB, ended at ${(logSize / 1e6).toFixed(1)}MB after ~48MB of records were written (~12MB live)`, ); check('bulk survivors intact', (await bulk.findOne({ _id: 1999 })).payload === payload2); check('bulk count after delete', (await bulk.countDocuments({})) === 1000); // Phase-1 data survives in memory (erin was deleted in phase 1, so 4). check('users count after bulk', (await users.countDocuments({})) === 4); // ---- graceful restart ----------------------------------------------------- await stopServer('SIGTERM'); await startServer(); const client2 = new MongoClient(URL, { serverSelectionTimeoutMS: 5000 }); await client2.connect(); const db2 = client2.db('e2e6'); const users2 = db2.collection('users'); check('after restart: users count', (await users2.countDocuments({})) === 4); if (process.env.E2E6_DEBUG) { console.log('DEBUG aliceId:', String(aliceId), 'isOID', aliceId instanceof ObjectId); console.log('DEBUG all users:', JSON.stringify(await users2.find({}).toArray())); console.log('DEBUG bulk:', JSON.stringify(await db2.collection('bulk').find({}, { projection: { payload: 0 } }).limit(5).toArray())); console.log('DEBUG sessions:', JSON.stringify(await db2.collection('sessions').find({}).toArray())); console.log('DEBUG colls:', JSON.stringify(await db2.listCollections({}, { nameOnly: true }).toArray())); console.log('DEBUG dbs:', JSON.stringify((await db2.admin().listDatabases()).databases.map((d) => d.name))); console.log('DEBUG log file size:', fs.statSync(DBFILE).size); } const alice2 = await users2.findOne({ _id: aliceId }); check('after restart: doc content', alice2.name === 'alice' && alice2.age === 31 && alice2.seen === true, JSON.stringify(alice2)); check('after restart: unique index find', (await users2.find({ email: 'dave@x.io' }).toArray()).length === 1); check('after restart: unique index still enforced', await (async () => { try { await users2.insertOne({ name: 'dup2', email: 'dave@x.io' }); return false; } catch (e) { return e.code === 11000; } })()); check('after restart: bulk count', (await db2.collection('bulk').countDocuments({})) === 1000); check('after restart: TTL index listed', (await db2.collection('sessions').indexes()).some((i) => i.name === 'expireAt_1')); return client2; } async function phase3(client) { // ---- kill -9 mid-write ---------------------------------------------------- // Every write is logged + fsynced before it becomes visible, so whatever // count we see before the kill must be there after the restart. const crash = client.db('e2e6').collection('crash'); await crash.drop().catch(() => {}); let committed = 0; for (let i = 1; i <= 150; i++) { await crash.insertOne({ _id: i, seq: i }); committed = i; } check('crash: committed before kill', committed === 150); await client.close(); await stopServer('SIGKILL'); await startServer(); const c3 = new MongoClient(URL, { serverSelectionTimeoutMS: 5000 }); await c3.connect(); const db3 = c3.db('e2e6'); const n = await db3.collection('crash').countDocuments({}); check('crash recovery: all committed docs survived kill -9', n === 150, n); check('crash recovery: last doc intact', (await db3.collection('crash').findOne({ _id: 150 })).seq === 150); check('crash recovery: pre-crash data intact', (await db3.collection('users').countDocuments({})) === 4); await db3.collection('crash').insertOne({ _id: 151, seq: 151 }); check('crash recovery: writes continue', (await db3.collection('crash').countDocuments({})) === 151); await c3.close(); } async function main() { if (!fs.existsSync(BIN)) { console.error(`server binary not found at ${BIN} — run \`zig build\` first`); process.exit(1); } await startServer(true); const client = new MongoClient(URL, { serverSelectionTimeoutMS: 5000 }); await client.connect(); const db = client.db('e2e6'); console.log('phase 1: feature surface'); const state = await phase1(client, db); console.log('phase 2: compaction + graceful restart'); const client2 = await phase2(client, state); // The phase-1 client's connection died with the restart; close it so the // process can exit (and so the exit handler can reap the server child). await client.close().catch(() => {}); console.log('phase 3: kill -9 crash recovery'); await phase3(client2); if (process.env.E2E6_KEEP !== '1') fs.rmSync(DBFILE, { force: true }); await stopServer('SIGTERM'); const failed = results.filter((r) => !r.ok); console.log(`\n${results.length - failed.length}/${results.length} checks passed`); if (failed.length) { console.log('FAILED:', failed.map((f) => f.name).join(', ')); console.log('--- server log tail ---'); console.log(serverLog.split('\n').slice(-30).join('\n')); process.exit(1); } console.log('E2E6_OK'); } main().catch((e) => { console.error('E2E6_FAIL', e); console.log('--- server log tail ---'); console.log(serverLog.split('\n').slice(-40).join('\n')); process.exit(1); });