From c29c09d6e8cae0108d373d2d5b69b7a3e4f4b9f0 Mon Sep 17 00:00:00 2001 From: mongo-light Date: Sun, 2 Aug 2026 10:56:43 +0300 Subject: [PATCH] query: support bare regex filter values and array-index dot paths The Node driver sends {field: /re/} as a BSON regex element (type 0x0B), which the matcher previously only handled via the $regex operator form; and dot paths with numeric segments (tags.0) were ignored because array descent only recursed into embedded docs. Both are part of standard MongoDB query semantics and were caught by the driver e2e suite. server: unbounded Io async limit so the accept loop never wedges, and treat header-read failures (client RST on pool teardown) as clean disconnects. With the default cpu_count-1 limit, groupAsync's eager fallback ran connection handlers inline on the accept-loop fiber once that many connections were alive, stalling accept() and timing out handshakes for further clients. Add tests/e2e/: official driver CRUD, concurrency, and kill -9 recovery suites (29 + 2 + 3 checks), plus unit tests for the query fixes. --- src/query.zig | 39 +++++++++++++ src/server.zig | 17 +++--- tests/e2e/README.md | 29 ++++++++++ tests/e2e/e2e.js | 132 ++++++++++++++++++++++++++++++++++++++++++++ tests/e2e/e2e2.js | 107 +++++++++++++++++++++++++++++++++++ 5 files changed, 315 insertions(+), 9 deletions(-) create mode 100644 tests/e2e/README.md create mode 100644 tests/e2e/e2e.js create mode 100644 tests/e2e/e2e2.js diff --git a/src/query.zig b/src/query.zig index 4479098..93ea36d 100644 --- a/src/query.zig +++ b/src/query.zig @@ -98,6 +98,13 @@ fn field_matches(gpa: std.mem.Allocator, path: []const u8, expected: bson.Value, } return true; } + // Bare BSON regex value: {field: /re/} behaves like {$regex: "re"}. + if (expected == .regex) { + for (candidates.items) |a| { + if (a == .string and regex_match(expected.regex.pattern, expected.regex.options, a.string)) return true; + } + return false; + } // Bare equality — matches if any candidate equals the expected value. for (candidates.items) |actual| { if (bson.compare(actual, expected) == .eq) return true; @@ -274,6 +281,21 @@ fn collect_from_value(gpa: std.mem.Allocator, v: bson.Value, path: []const u8, o switch (v) { .doc => |pairs| try collect_values(gpa, pairs, path, out, depth), .array => |items| { + // Numeric first segment: address an element by index ("tags.0"). + var pit = std.mem.splitScalar(u8, path, '.'); + const seg = pit.next() orelse return; + if (std.fmt.parseInt(usize, seg, 10)) |idx| { + if (idx < items.len) { + const rest = pit.rest(); + if (rest.len == 0) { + if (depth < 8) try out.append(gpa, items[idx]); + } else { + try collect_from_value(gpa, items[idx], rest, out, depth + 1); + } + } + return; + } else |_| {} + // Multikey semantics: descend into embedded documents of the array. for (items) |item| { switch (item) { .doc => try collect_values(gpa, item.doc, path, out, depth), @@ -828,6 +850,23 @@ test "dot path filters" { try testing.expect(try matches(testing.allocator, &doc_of(&.{.{ .key = "items.sku", .value = .{ .string = "y" } }}), &d)); } +test "array index dot path and bare regex value" { + const d = doc_of(&.{ + .{ .key = "tags", .value = .{ .array = &.{ + .{ .string = "c" }, + .{ .string = "a" }, + } } }, + .{ .key = "name", .value = .{ .string = "carol" } }, + }); + // tags.0 addresses the first element. + try testing.expect(try matches(testing.allocator, &doc_of(&.{.{ .key = "tags.0", .value = .{ .string = "c" } }}), &d)); + try testing.expect(!try matches(testing.allocator, &doc_of(&.{.{ .key = "tags.1", .value = .{ .string = "c" } }}), &d)); + try testing.expect(!try matches(testing.allocator, &doc_of(&.{.{ .key = "tags.5", .value = .{ .string = "c" } }}), &d)); + // Bare BSON regex value behaves like {$regex: "re"}. + try testing.expect(try matches(testing.allocator, &doc_of(&.{.{ .key = "name", .value = .{ .regex = .{ .pattern = "^c", .options = "" } } }}), &d)); + try testing.expect(!try matches(testing.allocator, &doc_of(&.{.{ .key = "name", .value = .{ .regex = .{ .pattern = "^z", .options = "" } } }}), &d)); +} + test "sort compares by BSON order" { const a = doc_of(&.{ .{ .key = "n", .value = .{ .int32 = 2 } }, .{ .key = "x", .value = .{ .string = "a" } } }); const b = doc_of(&.{ .{ .key = "n", .value = .{ .int32 = 10 } }, .{ .key = "x", .value = .{ .string = "b" } } }); diff --git a/src/server.zig b/src/server.zig index 547e2e8..94c6b42 100644 --- a/src/server.zig +++ b/src/server.zig @@ -18,7 +18,12 @@ pub const Server = struct { start_time: std.Io.Timestamp, pub fn run(self: *Server) !void { - var threaded: std.Io.Threaded = std.Io.Threaded.init(self.gpa, .{}); + // Unbounded async limit: connection handlers otherwise fall back to + // running inline on the accept-loop fiber once busy_count hits the + // default cpu_count-1, which blocks accept() for the handler's + // lifetime and stalls new connections (handshake timeouts). With an + // unlimited limit the pool spawns a thread per live connection. + var threaded: std.Io.Threaded = std.Io.Threaded.init(self.gpa, .{ .async_limit = .unlimited }); defer threaded.deinit(); const io = threaded.io(); @@ -76,13 +81,7 @@ fn handle_connection_inner(io: std.Io, stream: std.Io.net.Stream, server: *Serve while (true) { var len_bytes: [4]u8 = undefined; - reader.interface.readSliceAll(&len_bytes) catch |err| switch (err) { - error.EndOfStream => return, // clean client disconnect - else => { - std.debug.print("mongo-light: read error on conn {d}: {s}\n", .{ connection_id, @errorName(err) }); - return; - }, - }; + reader.interface.readSliceAll(&len_bytes) catch return; // clean client disconnect (EOF or RST) const total: u32 = std.mem.readInt(u32, &len_bytes, .little); if (total < 16 or total > wire.max_message_size) { std.debug.print("mongo-light: bad message length {d} on conn {d}\n", .{ total, connection_id }); @@ -94,7 +93,7 @@ fn handle_connection_inner(io: std.Io, stream: std.Io.net.Stream, server: *Serve msg_buf.items.len = total; std.mem.writeInt(u32, msg_buf.items[0..4], total, .little); reader.interface.readSliceAll(msg_buf.items[4..]) catch |err| { - std.debug.print("mongo-light: read error on conn {d}: {s}\n", .{ connection_id, @errorName(err) }); + std.debug.print("mongo-light: read error on conn {d}: {s} (body, len {d})\n", .{ connection_id, @errorName(err), total }); return; }; diff --git a/tests/e2e/README.md b/tests/e2e/README.md new file mode 100644 index 0000000..a4b0e3d --- /dev/null +++ b/tests/e2e/README.md @@ -0,0 +1,29 @@ +# End-to-end tests with the official MongoDB Node.js driver + +These exercise mongo-light from a real driver over TCP: full CRUD, query +operators, aggregation, error codes, concurrent clients, and crash recovery. + +## Setup + +```sh +cd tests/e2e +npm init -y >/dev/null +npm install mongodb +``` + +## Run + +Start the server, then run the suites against it (defaults to port 27020): + +```sh +zig build +zig-out/bin/mongo-light --port 27020 --db /tmp/ml-e2e.log & + +node tests/e2e/e2e.js # CRUD + operators + aggregate + errors (29 checks) +node tests/e2e/e2e2.js concurrent # 8 clients: 4 writers + 4 readers (2 checks) +node tests/e2e/e2e2.js crash-a # write 50 docs, then kill -9 the server +node tests/e2e/e2e2.js crash-b # restart and verify all 50 survived +``` + +`e2e2.js concurrent` is safe to repeat against a running server (it drops its +collection first); `crash-a`/`crash-b` are two halves of one scenario. diff --git a/tests/e2e/e2e.js b/tests/e2e/e2e.js new file mode 100644 index 0000000..0834c11 --- /dev/null +++ b/tests/e2e/e2e.js @@ -0,0 +1,132 @@ +// End-to-end test: official MongoDB Node.js driver against mongo-light. +const { MongoClient, ObjectId } = require('mongodb'); + +const URL = 'mongodb://127.0.0.1:27020'; +const results = []; +function check(name, cond, detail = '') { + results.push({ name, ok: !!cond, detail: String(detail) }); + if (!cond) console.error(` ✗ ${name} ${detail}`); +} +async function main() { + const client = new MongoClient(URL, { serverSelectionTimeoutMS: 5000 }); + await client.connect(); + const db = client.db('e2e'); + const users = db.collection('users'); + await users.drop().catch(() => {}); + + // --- insert --- + await users.insertOne({ name: 'alice', age: 30, tags: ['a', 'b'] }); + const many = await users.insertMany([ + { name: 'bob', age: 25, tags: ['b'] }, + { name: 'carol', age: 35, tags: ['c', 'a'] }, + { name: 'dave', age: 40, tags: [] }, + ]); + check('insertMany acknowledged', many.acknowledged === true, many); + check('auto _id assigned', ObjectId.isValid(many.insertedIds[0])); + + // --- find: filters, operators, sort, skip, limit, projection --- + const gt = await users.find({ age: { $gt: 28 } }).sort({ age: -1 }).toArray(); + check('find $gt + sort desc', gt.map((d) => d.name).join(',') === 'dave,carol,alice', gt.map((d) => d.name)); + + const inq = await users.find({ name: { $in: ['alice', 'bob'] } }).count(); + check('find $in count', inq === 2, inq); + + const rgx = await users.find({ name: /^[bc]/ }).toArray(); + check('find $regex', rgx.length === 2, rgx.map((d) => d.name)); + + const exists = await users.find({ tags: { $exists: true } }).count(); + check('find $exists', exists === 4, exists); + + const lim = await users.find({}).sort({ age: 1 }).skip(1).limit(2).toArray(); + check('find skip+limit+sort', lim.map((d) => d.name).join(',') === 'alice,carol', lim.map((d) => d.name)); + + const proj = await users.findOne({ name: 'alice' }, { projection: { _id: 0, name: 1 } }); + check('projection', proj.name === 'alice' && proj.age === undefined, JSON.stringify(proj)); + + const dot = await users.findOne({ 'tags.0': 'c' }); + check('dot path + array', dot?.name === 'carol'); + + // --- count --- + check('countDocuments', (await users.countDocuments({})) === 4); + check('countDocuments with filter', (await users.countDocuments({ age: { $gte: 30 } })) === 3); + 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({ name: 'alice' }); + check('updateOne $set+$inc applied', alice.vip === true && alice.age === 31, JSON.stringify(alice)); + + const um = await users.updateMany({}, { $set: { seen: true } }); + check('updateMany', um.modifiedCount === 4, um); + + const push = await users.updateOne({ name: 'dave' }, { $push: { tags: 'x' } }); + check('$push', push.modifiedCount === 1); + check('$push visible', (await users.findOne({ name: 'dave' })).tags.length === 1); + + const ups = await users.updateOne({ name: 'erin' }, { $set: { age: 28 } }, { upsert: true }); + check('upsert', ups.upsertedCount === 1 && ups.matchedCount === 0, ups); + check('upsert doc exists', (await users.findOne({ name: 'erin' }))?.age === 28); + + // --- findOneAndUpdate (findAndModify) --- + const fam = await users.findOneAndUpdate( + { name: 'bob' }, + { $set: { lucky: true } }, + { returnDocument: 'after' }, + ); + const famDoc = fam.value ?? fam; + check('findOneAndUpdate returns new', famDoc.lucky === true, JSON.stringify(fam)); + const famRemove = await users.findOneAndDelete({ name: 'erin' }); + const famDelDoc = famRemove.value ?? famRemove; + check('findOneAndDelete', famDelDoc?.name === 'erin', JSON.stringify(famRemove)); + + // --- 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+$sort', grp.length >= 1 && grp.some((g) => g.total > 0), JSON.stringify(grp)); + + const cnt = await users.aggregate([{ $match: {} }, { $count: 'n' }]).toArray(); + check('aggregate $count', cnt[0]?.n === 4, JSON.stringify(cnt)); + + // --- duplicate key --- + let dupErr = null; + try { + await users.insertOne({ _id: many.insertedIds[0], name: 'clobber' }); + } catch (e) { + dupErr = e; + } + check('duplicate key rejected', dupErr?.code === 11000, dupErr?.message); + + // --- listCollections / listDatabases --- + const colls = await db.listCollections({}, { nameOnly: true }).toArray(); + check('listCollections', colls.some((c) => c.name === 'users'), JSON.stringify(colls)); + const dbs = await client.db('admin').admin().listDatabases(); + check('listDatabases', dbs.databases.some((d) => d.name === 'e2e'), JSON.stringify(dbs.databases.map((d) => d.name))); + + // --- delete --- + const del1 = await users.deleteOne({ name: 'dave' }); + check('deleteOne', del1.deletedCount === 1, del1); + const delMany = await users.deleteMany({}); + check('deleteMany', delMany.deletedCount === 3, delMany); + check('empty after delete', (await users.countDocuments({})) === 0); + + await client.close(); + + 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(', ')); + process.exit(1); + } + console.log('E2E_OK'); +} + +main().catch((e) => { + console.error('E2E_FAIL', e); + process.exit(1); +}); diff --git a/tests/e2e/e2e2.js b/tests/e2e/e2e2.js new file mode 100644 index 0000000..3021a55 --- /dev/null +++ b/tests/e2e/e2e2.js @@ -0,0 +1,107 @@ +// E2E part 2: concurrent clients + crash recovery, official driver. +const { MongoClient } = require('mongodb'); + +const URL = 'mongodb://127.0.0.1:27020'; +const results = []; +function check(name, cond, detail = '') { + results.push({ name, ok: !!cond, detail: String(detail) }); + if (!cond) console.error(` ✗ ${name} ${detail}`); +} + +async function concurrentClients() { + // 8 clients: 4 inserting unique docs, 4 reading concurrently. + const WRITERS = 4; + const READERS = 4; + const PER = 100; + const TOTAL = WRITERS * PER; + + const clients = await Promise.all( + Array.from({ length: WRITERS + READERS }, () => new MongoClient(URL, { serverSelectionTimeoutMS: 5000 }).connect()), + ); + await clients[0].db('conc').collection('items').drop().catch(() => {}); + + let nextId = 1; + const pending = { n: TOTAL }; + const writerJobs = []; + for (let w = 0; w < WRITERS; w++) { + writerJobs.push((async () => { + const db = clients[w].db('conc'); + const coll = db.collection('items'); + while (true) { + const id = nextId++; + if (id > TOTAL) break; + await coll.insertOne({ _id: id, w: w, payload: 'x'.repeat(64) }); + pending.n--; + } + })()); + } + const readerJobs = []; + for (let r = 0; r < READERS; r++) { + const idx = WRITERS + r; + readerJobs.push((async () => { + const db = clients[idx].db('conc'); + const coll = db.collection('items'); + while (pending.n > 0) { + const n = await coll.countDocuments({}); + if (n > TOTAL) throw new Error('reader saw impossible count ' + n); + await coll.findOne({ _id: Math.floor(Math.random() * TOTAL) + 1 }); + } + })()); + } + await Promise.all([...writerJobs, ...readerJobs]); + + const final = await clients[0].db('conc').collection('items').countDocuments({}); + check('concurrent writers+readers final count', final === TOTAL, final); + const spot = await clients[0].db('conc').collection('items').findOne({ _id: TOTAL }); + check('concurrent insert visible', spot?._id === TOTAL, JSON.stringify(spot)); + + await Promise.all(clients.map((c) => c.close())); +} + +async function crashRecovery() { + // Phase A: write and record state. + const c1 = await new MongoClient(URL, { serverSelectionTimeoutMS: 5000 }).connect(); + const coll = c1.db('crash').collection('docs'); + await coll.deleteMany({}); + const docs = []; + for (let i = 1; i <= 50; i++) { + docs.push({ _id: i, name: 'doc-' + i }); + } + await coll.insertMany(docs); + await c1.close(); + check('crash: committed before kill', true); + + // (The harness kills -9 the server between phases.) + await new Promise((r) => setTimeout(r, 200)); +} + +async function crashVerify() { + const c2 = await new MongoClient(URL, { serverSelectionTimeoutMS: 5000 }).connect(); + const coll = c2.db('crash').collection('docs'); + const n = await coll.countDocuments({}); + check('crash recovery: all 50 docs survived kill -9', n === 50, n); + const one = await coll.findOne({ _id: 37 }); + check('crash recovery: doc content intact', one?.name === 'doc-37', JSON.stringify(one)); + // keep writing after recovery (log reopened correctly) + await coll.insertOne({ _id: 51, name: 'post-recovery' }); + const n2 = await coll.countDocuments({}); + check('crash recovery: writes continue', n2 === 51, n2); + await c2.close(); +} + +async function main() { + const phase = process.argv[2]; + if (phase === 'concurrent') await concurrentClients(); + if (phase === 'crash-a') await crashRecovery(); + if (phase === 'crash-b') await crashVerify(); + + 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(', ')); + process.exit(1); + } + console.log(phase === 'crash-a' ? 'CRASH_PHASE_A_OK' : 'E2E2_OK'); +} + +main().catch((e) => { console.error('E2E2_FAIL', e); process.exit(1); });