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

@@ -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" } } });

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;
         };
 

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.

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);
+});

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); });