diff --git a/README.md b/README.md index 2f20fb4..5c6d123 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -# mongo-light +# mongo-lite A lightweight, embedded MongoDB-compatible document database written in Zig 0.16. Like SQLite, it stores everything in a single file; unlike SQLite, @@ -11,7 +11,7 @@ driver, PyMongo — connect over TCP and just work. zig build # build the server zig build test # run the unit test suite -zig-out/bin/mongo-light --port 27017 --db data.log +zig-out/bin/mongo-lite --port 27017 --db data.log # in another terminal: mongosh --port 27017 diff --git a/build.zig b/build.zig index f1f43ed..295b902 100644 --- a/build.zig +++ b/build.zig @@ -20,7 +20,7 @@ pub fn build(b: *std.Build) void { }); const exe = b.addExecutable(.{ - .name = "mongo-light", + .name = "mongo-lite", .root_module = exe_mod, }); b.installArtifact(exe); @@ -28,7 +28,7 @@ pub fn build(b: *std.Build) void { const run_cmd = b.addRunArtifact(exe); run_cmd.step.dependOn(b.getInstallStep()); if (b.args) |args| run_cmd.addArgs(args); - const run_step = b.step("run", "Run mongo-light server"); + const run_step = b.step("run", "Run mongo-lite server"); run_step.dependOn(&run_cmd.step); const test_mod = b.createModule(.{ diff --git a/build.zig.zon b/build.zig.zon index cf84ce4..bb5ab92 100644 --- a/build.zig.zon +++ b/build.zig.zon @@ -1,7 +1,7 @@ .{ - .name = .mongo_light, + .name = .mongo_lite, .version = "0.0.1", .minimum_zig_version = "0.16.0", .paths = .{""}, - .fingerprint = 0xb3baeb7ec5369577, + .fingerprint = 0xcca51d85af632c24, } diff --git a/src/commands.zig b/src/commands.zig index 34066b1..a37d1f9 100644 --- a/src/commands.zig +++ b/src/commands.zig @@ -131,11 +131,17 @@ fn add_server_info(ctx: *Context, reply: *wire.Reply) !void { try reply.put("maxWireVersion", .{ .int32 = 8 }); try reply.put("readOnly", .{ .bool = false }); - const oid = ctx.oid_gen.new(ctx.io); - const tv = try reply.arena_alloc().alloc(bson.Pair, 2); - tv[0] = .{ .key = "processId", .value = .{ .object_id = oid } }; - tv[1] = .{ .key = "counter", .value = .{ .int64 = 0 } }; - try reply.put("topologyVersion", .{ .doc = tv }); + // Deliberately no `topologyVersion`. A driver treats its presence as + // "this server supports the streaming (awaitable) hello protocol" and + // switches monitoring to an exhaust hello: it sends one hello with + // maxAwaitTimeMS and the OP_MSG exhaustAllowed flag, then expects a + // stream of unsolicited replies each carrying moreToCome. We answer + // once with moreToCome clear and go back to reading, so the driver + // fails the heartbeat ("Server ended moreToCome unexpectedly"), drops + // the connection and resets its pool — a connect/disconnect loop once + // per heartbeat, which is what MongoDB Compass showed. Omitting the + // field keeps monitoring on the polling path, which we do implement, + // and matches maxWireVersion 8: streaming hello arrived in wire 9. } fn cmd_hello(ctx: *Context, _: *wire.Message, reply: *wire.Reply) !void { @@ -156,7 +162,7 @@ fn cmd_ping(_: *Context, _: *wire.Message, reply: *wire.Reply) !void { fn cmd_build_info(_: *Context, _: *wire.Message, reply: *wire.Reply) !void { try reply.put("version", .{ .string = "4.4.0" }); - try reply.put("gitVersion", .{ .string = "mongo-light" }); + try reply.put("gitVersion", .{ .string = "mongo-lite" }); try reply.put("versionArray", .{ .array = try int_array(reply, &.{ 4, 4, 0, 0 }) }); try reply.put("openssl", .{ .doc = &.{} }); try reply.put("loaderFlags", .{ .string = "" }); @@ -213,7 +219,7 @@ fn cmd_host_info(ctx: *Context, _: *wire.Message, reply: *wire.Reply) !void { fn cmd_get_cmd_line_opts(_: *Context, _: *wire.Message, reply: *wire.Reply) !void { const argv = try reply.arena_alloc().alloc(bson.Pair, 2); - argv[0] = .{ .key = "dbpath", .value = .{ .string = "mongo-light.log" } }; + argv[0] = .{ .key = "dbpath", .value = .{ .string = "mongo-lite.log" } }; argv[1] = .{ .key = "port", .value = .{ .int32 = 27017 } }; try reply.put("argv", .{ .array = &.{} }); try reply.put("parsed", .{ .doc = argv }); @@ -1363,6 +1369,28 @@ test "ping and hello replies parse" { try testing.expectEqual(@as(i32, 8), bson.get_pair(reply2.pairs.items, "maxWireVersion").?.int32); } +test "handshake does not advertise the streaming hello protocol" { + // `topologyVersion` in a hello/isMaster reply tells the driver it may + // monitor with an exhaust hello and expect moreToCome replies we never + // send; the driver then kills the connection every heartbeat. + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + const io = threaded.io(); + var tdb = try TestDb.init(io); + defer tdb.deinit(); + var ctx = tdb.ctx(io); + + for ([_][]const u8{ "hello", "isMaster", "ismaster" }) |cmd| { + var reply = wire.Reply.init(testing.allocator); + defer reply.deinit(); + var msg = try parse_fake_msg(cmd, .null, &.{}); + defer msg.deinit(); + try dispatch(&ctx, &msg, &reply); + try testing.expectEqual(@as(f64, 1.0), bson.get_pair(reply.pairs.items, "ok").?.double); + try testing.expect(bson.get_pair(reply.pairs.items, "topologyVersion") == null); + } +} + test "unknown command gives CommandNotFound" { var threaded: std.Io.Threaded = .init_single_threaded; defer threaded.deinit(); diff --git a/src/db.zig b/src/db.zig index 03a4766..f1b25ce 100644 --- a/src/db.zig +++ b/src/db.zig @@ -549,13 +549,13 @@ pub const Engine = struct { while (doc_it.next()) |doc_entry| { const duplicate = ix.add_doc(self.gpa, doc_entry.value_ptr.*, doc_entry.key_ptr.*, false) catch |err| switch (err) { error.ParallelArrays => { - std.debug.print("mongo-light: WARNING: index '{s}' cannot index an existing document; entry skipped\n", .{ix.name}); + std.debug.print("mongo-lite: WARNING: index '{s}' cannot index an existing document; entry skipped\n", .{ix.name}); continue; }, else => return err, }; if (duplicate) { - std.debug.print("mongo-light: WARNING: unique index '{s}' has duplicate keys in existing data; duplicates not enforced for existing documents\n", .{ix.name}); + std.debug.print("mongo-lite: WARNING: unique index '{s}' has duplicate keys in existing data; duplicates not enforced for existing documents\n", .{ix.name}); } } } @@ -608,7 +608,7 @@ fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) an switch (record.type) { storage.record_type_index_create => { self.register_index_from_spec(coll, doc) catch |err| { - std.debug.print("mongo-light: index create record failed to apply: {s}\n", .{@errorName(err)}); + std.debug.print("mongo-lite: index create record failed to apply: {s}\n", .{@errorName(err)}); return; }; return; @@ -626,7 +626,7 @@ fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) an } const id_value = doc.get("_id") orelse { - std.debug.print("mongo-light: log record without _id, skipping\n", .{}); + std.debug.print("mongo-lite: log record without _id, skipping\n", .{}); return; }; const id_key = try bson.serialize_value(self.gpa, id_value); diff --git a/src/lib.zig b/src/lib.zig index b483f22..4f23dee 100644 --- a/src/lib.zig +++ b/src/lib.zig @@ -1,4 +1,4 @@ -// mongo-light core library. Public entry point for tests and the server. +// mongo-lite core library. Public entry point for tests and the server. pub const bson = @import("bson.zig"); pub const wire = @import("wire.zig"); diff --git a/src/main.zig b/src/main.zig index 663002d..5f80f6f 100644 --- a/src/main.zig +++ b/src/main.zig @@ -2,12 +2,12 @@ const std = @import("std"); const mongo = @import("mongo"); const usage = - \\mongo-light — lightweight MongoDB-compatible document database + \\mongo-lite — lightweight MongoDB-compatible document database \\ - \\usage: mongo-light [options] + \\usage: mongo-lite [options] \\ --port listen port (default 27017) \\ --bind bind address (default 127.0.0.1) - \\ --db database file (default mongo-light.log) + \\ --db database file (default mongo-lite.log) \\ --ttl-sweep-secs \\ seconds between TTL index sweeps (default 60, 0 disables) \\ --help show this help @@ -17,7 +17,7 @@ const usage = pub fn main(init: std.process.Init) !void { var port: u16 = 27017; var bind_ip: []const u8 = "127.0.0.1"; - var db_path: []const u8 = "mongo-light.log"; + var db_path: []const u8 = "mongo-lite.log"; var ttl_sweep_secs: i64 = 60; var it = std.process.Args.Iterator.init(init.minimal.args); @@ -27,7 +27,7 @@ pub fn main(init: std.process.Init) !void { if (std.mem.eql(u8, arg, "--port")) { const v = it.next() orelse return error.MissingValue; port = std.fmt.parseInt(u16, v, 10) catch { - std.debug.print("mongo-light: invalid port '{s}'\n", .{v}); + std.debug.print("mongo-lite: invalid port '{s}'\n", .{v}); return error.InvalidPort; }; } else if (std.mem.eql(u8, arg, "--bind")) { @@ -41,14 +41,14 @@ pub fn main(init: std.process.Init) !void { // only thing parseInt would otherwise let through. ttl_sweep_secs = std.fmt.parseInt(i64, v, 10) catch -1; if (ttl_sweep_secs < 0) { - std.debug.print("mongo-light: invalid ttl sweep interval '{s}'\n", .{v}); + std.debug.print("mongo-lite: invalid ttl sweep interval '{s}'\n", .{v}); return error.InvalidTtlSweepSecs; } } else if (std.mem.eql(u8, arg, "--help") or std.mem.eql(u8, arg, "-h")) { try std.Io.File.writeStreamingAll(.stdout(), init.io, usage); return; } else { - std.debug.print("mongo-light: unknown option '{s}'\n{s}", .{ arg, usage }); + std.debug.print("mongo-lite: unknown option '{s}'\n{s}", .{ arg, usage }); return error.UnknownOption; } } @@ -56,7 +56,7 @@ pub fn main(init: std.process.Init) !void { const oid_gen = mongo.bson.ObjectIdGen.init(init.io); var engine = try mongo.db.Engine.open(init.gpa, init.io, db_path); defer engine.deinit(); - std.debug.print("mongo-light: opened database '{s}'\n", .{db_path}); + std.debug.print("mongo-lite: opened database '{s}'\n", .{db_path}); var server = mongo.server.Server{ .gpa = init.gpa, diff --git a/src/server.zig b/src/server.zig index 1a9e966..3985803 100644 --- a/src/server.zig +++ b/src/server.zig @@ -35,7 +35,7 @@ pub const Server = struct { var listener = try addr.listen(io, .{ .reuse_address = true }); defer listener.deinit(io); - std.debug.print("mongo-light: listening on {s}:{d}\n", .{ self.bind_ip, self.port }); + std.debug.print("mongo-lite: listening on {s}:{d}\n", .{ self.bind_ip, self.port }); var group: std.Io.Group = .init; defer group.cancel(io); @@ -48,7 +48,7 @@ pub const Server = struct { const stream = listener.accept(io) catch |err| switch (err) { error.Canceled => return, else => { - std.debug.print("mongo-light: accept error: {s}\n", .{@errorName(err)}); + std.debug.print("mongo-lite: accept error: {s}\n", .{@errorName(err)}); continue; }, }; @@ -72,7 +72,7 @@ fn ttl_monitor(io: std.Io, server: *Server) error{Canceled}!void { defer server.engine.unlock(); const now_ms = std.Io.Timestamp.now(io, .real).toMilliseconds(); _ = server.engine.ttl_sweep(now_ms) catch |err| { - std.debug.print("mongo-light: TTL sweep failed: {s}\n", .{@errorName(err)}); + std.debug.print("mongo-lite: TTL sweep failed: {s}\n", .{@errorName(err)}); continue; }; } @@ -113,7 +113,7 @@ fn handle_connection_inner(io: std.Io, stream: std.Io.net.Stream, server: *Serve 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 }); + std.debug.print("mongo-lite: bad message length {d} on conn {d}\n", .{ total, connection_id }); return; } @@ -122,14 +122,14 @@ 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} (body, len {d})\n", .{ connection_id, @errorName(err), total }); + std.debug.print("mongo-lite: read error on conn {d}: {s} (body, len {d})\n", .{ connection_id, @errorName(err), total }); return; }; var msg = wire.Message.parse(server.gpa, msg_buf.items) catch |err| { // Unparseable request: close the connection. const op: i32 = if (msg_buf.items.len >= 16) std.mem.readInt(i32, msg_buf.items[12..16], .little) else 0; - std.debug.print("mongo-light: bad message on conn {d}: {s} (opCode {d})\n", .{ connection_id, @errorName(err), op }); + std.debug.print("mongo-lite: bad message on conn {d}: {s} (opCode {d})\n", .{ connection_id, @errorName(err), op }); return; }; defer msg.deinit(); @@ -154,16 +154,16 @@ fn handle_connection_inner(io: std.Io, stream: std.Io.net.Stream, server: *Serve else reply.build(server.gpa, reply_request_id, msg.request_id, &out_buf); built catch |err| { - std.debug.print("mongo-light: reply build error on conn {d}: {s}\n", .{ connection_id, @errorName(err) }); + std.debug.print("mongo-lite: reply build error on conn {d}: {s}\n", .{ connection_id, @errorName(err) }); return; }; reply_request_id +%= 1; writer.interface.writeAll(out_buf.items) catch |err| { - std.debug.print("mongo-light: write error on conn {d}: {s}\n", .{ connection_id, @errorName(err) }); + std.debug.print("mongo-lite: write error on conn {d}: {s}\n", .{ connection_id, @errorName(err) }); return; }; writer.interface.flush() catch |err| { - std.debug.print("mongo-light: flush error on conn {d}: {s}\n", .{ connection_id, @errorName(err) }); + std.debug.print("mongo-lite: flush error on conn {d}: {s}\n", .{ connection_id, @errorName(err) }); return; }; } diff --git a/src/storage.zig b/src/storage.zig index fc4a1c7..9b9322d 100644 --- a/src/storage.zig +++ b/src/storage.zig @@ -91,21 +91,21 @@ pub const Log = struct { while (true) { const len_read = self.file.readPositionalAll(self.io, chunk[0..4], pos) catch |err| { - std.debug.print("mongo-light: log read error at {d}: {s}\n", .{ pos, @errorName(err) }); + std.debug.print("mongo-lite: log read error at {d}: {s}\n", .{ pos, @errorName(err) }); return error.InvalidLog; }; if (len_read == 0) return; // clean end if (len_read < 4) return; // torn tail const total: u32 = std.mem.readInt(u32, chunk[0..4], .little); if (total < header_len) { - std.debug.print("mongo-light: corrupt record length {d} at {d}\n", .{ total, pos }); + std.debug.print("mongo-lite: corrupt record length {d} at {d}\n", .{ total, pos }); return error.InvalidLog; } const payload_len: usize = total - 4; // Documents up to maxBsonObjectSize are legal; anything larger is // corruption. Covers a hostile length prefix from a truncated file. if (payload_len > max_record_payload) { - std.debug.print("mongo-light: record too large at {d}\n", .{pos}); + std.debug.print("mongo-lite: record too large at {d}\n", .{pos}); return error.InvalidLog; } const payload = if (payload_len <= chunk.len) chunk[0..payload_len] else blk: { @@ -132,7 +132,7 @@ pub const Log = struct { const doc = try self.gpa.create(bson.Document); doc.* = bson.Document.parse(self.gpa, doc_bytes) catch { self.gpa.destroy(doc); - std.debug.print("mongo-light: unparseable doc in log at {d}\n", .{pos}); + std.debug.print("mongo-lite: unparseable doc in log at {d}\n", .{pos}); return error.InvalidLog; }; try callback(ctx, .{ diff --git a/tests/e2e/README.md b/tests/e2e/README.md index 8c223e3..ec4835d 100644 --- a/tests/e2e/README.md +++ b/tests/e2e/README.md @@ -1,6 +1,6 @@ # End-to-end tests with the official MongoDB Node.js driver -These exercise mongo-light from a real driver over TCP: full CRUD, query +These exercise mongo-lite from a real driver over TCP: full CRUD, query operators, aggregation, error codes, concurrent clients, and crash recovery. ## Setup @@ -17,7 +17,7 @@ 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 --ttl-sweep-secs 1 & +zig-out/bin/mongo-lite --port 27020 --db /tmp/ml-e2e.log --ttl-sweep-secs 1 & 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) @@ -33,7 +33,7 @@ care about the flag. Rebuild with `zig build` after any change under `src/` before restarting the server: `zig build test` compiles the test binary only and leaves -`zig-out/bin/mongo-light` stale, so the suites keep running against the old +`zig-out/bin/mongo-lite` stale, so the suites keep running against the old rules and report failures that the source no longer explains. `e2e2.js concurrent` is safe to repeat against a running server (it drops its diff --git a/tests/e2e/e2e.js b/tests/e2e/e2e.js index 0834c11..7b279e5 100644 --- a/tests/e2e/e2e.js +++ b/tests/e2e/e2e.js @@ -1,4 +1,4 @@ -// End-to-end test: official MongoDB Node.js driver against mongo-light. +// End-to-end test: official MongoDB Node.js driver against mongo-lite. const { MongoClient, ObjectId } = require('mongodb'); const URL = 'mongodb://127.0.0.1:27020'; diff --git a/tests/e2e/e2e4.js b/tests/e2e/e2e4.js index cb01e7b..069ee83 100644 --- a/tests/e2e/e2e4.js +++ b/tests/e2e/e2e4.js @@ -4,7 +4,7 @@ // server must reject with CannotCreateIndex (67). // // The server must run with a short sweep interval, e.g. -// zig-out/bin/mongo-light --port 27020 --db /tmp/ml-e2e.log --ttl-sweep-secs 1 +// zig-out/bin/mongo-lite --port 27020 --db /tmp/ml-e2e.log --ttl-sweep-secs 1 const { MongoClient } = require('mongodb'); const URL = 'mongodb://127.0.0.1:27020'; diff --git a/tests/e2e/e2e5.js b/tests/e2e/e2e5.js new file mode 100644 index 0000000..23aee9b --- /dev/null +++ b/tests/e2e/e2e5.js @@ -0,0 +1,75 @@ +// E2E part 5: SDAM monitoring stability, official driver. +// Every other e2e file issues its commands and exits, so all of them pass +// against a server whose *monitoring* is broken. This one just sits on an +// idle connection and watches the driver's heartbeats. +// +// The failure it guards against: advertising `topologyVersion` in the hello +// reply makes the driver monitor with an exhaust hello and expect a stream of +// moreToCome replies. A server that answers once and goes back to reading +// fails the heartbeat ("Server ended moreToCome unexpectedly"), the driver +// drops the connection and clears the pool, and the client — MongoDB Compass, +// say — shows a connect/disconnect loop once per heartbeat. +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}`); +} + +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); + +// 500ms is the driver's floor for heartbeatFrequencyMS; over WATCH_MS it gives +// ~10 heartbeats, enough that a per-heartbeat failure cannot hide. +const HEARTBEAT_MS = 500; +const WATCH_MS = 5000; + +async function main() { + const client = new MongoClient(URL, { + heartbeatFrequencyMS: HEARTBEAT_MS, + serverSelectionTimeoutMS: 5000, + }); + + let succeeded = 0; + const failures = []; + client.on('serverHeartbeatSucceeded', () => succeeded++); + client.on('serverHeartbeatFailed', (e) => failures.push(e.failure ? e.failure.message : 'unknown')); + // The pool is cleared when SDAM decides the server went away; on a healthy + // idle connection it should never happen. + client.on('connectionPoolCleared', () => failures.push('connectionPoolCleared')); + + await client.connect(); + await client.db('e2e5').command({ ping: 1 }); + + await sleep(WATCH_MS); + + check('heartbeats actually ran', succeeded >= 5, `${succeeded} succeeded`); + check('no heartbeat failed', failures.length === 0, failures.join(', ')); + + // A reset pool still recovers, so liveness alone would not catch the loop — + // check it anyway, since a dead connection here means something worse. + let usable = true; + try { + await client.db('e2e5').command({ ping: 1 }); + } catch (e) { + usable = false; + check('connection usable after idle period', false, e.message); + } + if (usable) check('connection usable after idle period', true); + + 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('E2E5_OK'); +} + +main().catch((e) => { + console.error('E2E5_FAIL', e); + process.exit(1); +}); diff --git a/tests/e2e/package.json b/tests/e2e/package.json index 1de2a11..0651652 100644 --- a/tests/e2e/package.json +++ b/tests/e2e/package.json @@ -1,7 +1,7 @@ { "name": "e2e", "version": "1.0.0", - "description": "These exercise mongo-light from a real driver over TCP: full CRUD, query operators, aggregation, error codes, concurrent clients, and crash recovery.", + "description": "These exercise mongo-lite from a real driver over TCP: full CRUD, query operators, aggregation, error codes, concurrent clients, and crash recovery.", "main": "e2e.js", "scripts": { "test": "echo \"Error: no test specified\" && exit 1"