commands/e2e: drop topologyVersion from the handshake; rename to mongo-lite
Advertising topologyVersion in the hello reply is what tells a driver the server speaks the streaming (awaitable) hello protocol — in the Node driver it is the only condition checked. From the second heartbeat on, the driver then monitored with an exhaust hello (exhaustAllowed + maxAwaitTimeMS) and waited for a stream of replies carrying moreToCome. We answered once with the flag clear and went back to reading, so every heartbeat failed with "Server ended moreToCome unexpectedly", destroying the connection and clearing the pool. MongoDB Compass showed this as a connect/disconnect loop once per heartbeat. We do not implement streaming hello, so we must not claim to. Omitting the field keeps monitoring on the polling path, and agrees with the maxWireVersion 8 we report: streaming hello arrived in wire version 9. The existing e2e files all passed against the broken server — they issue their commands and exit before the second heartbeat — so e2e5 watches SDAM heartbeats on an idle connection instead. Also renames mongo-light to mongo-lite throughout (binary, log messages, docs, gitVersion). Unrelated to the fix above, but squashed in at request rather than left as a commit whose message described only the fix.
This commit is contained in:
@@ -1,4 +1,4 @@
|
|||||||
# mongo-light
|
# mongo-lite
|
||||||
|
|
||||||
A lightweight, embedded MongoDB-compatible document database written in
|
A lightweight, embedded MongoDB-compatible document database written in
|
||||||
Zig 0.16. Like SQLite, it stores everything in a single file; unlike SQLite,
|
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 # build the server
|
||||||
zig build test # run the unit test suite
|
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:
|
# in another terminal:
|
||||||
mongosh --port 27017
|
mongosh --port 27017
|
||||||
|
|||||||
@@ -20,7 +20,7 @@ pub fn build(b: *std.Build) void {
|
|||||||
});
|
});
|
||||||
|
|
||||||
const exe = b.addExecutable(.{
|
const exe = b.addExecutable(.{
|
||||||
.name = "mongo-light",
|
.name = "mongo-lite",
|
||||||
.root_module = exe_mod,
|
.root_module = exe_mod,
|
||||||
});
|
});
|
||||||
b.installArtifact(exe);
|
b.installArtifact(exe);
|
||||||
@@ -28,7 +28,7 @@ pub fn build(b: *std.Build) void {
|
|||||||
const run_cmd = b.addRunArtifact(exe);
|
const run_cmd = b.addRunArtifact(exe);
|
||||||
run_cmd.step.dependOn(b.getInstallStep());
|
run_cmd.step.dependOn(b.getInstallStep());
|
||||||
if (b.args) |args| run_cmd.addArgs(args);
|
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);
|
run_step.dependOn(&run_cmd.step);
|
||||||
|
|
||||||
const test_mod = b.createModule(.{
|
const test_mod = b.createModule(.{
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
.{
|
.{
|
||||||
.name = .mongo_light,
|
.name = .mongo_lite,
|
||||||
.version = "0.0.1",
|
.version = "0.0.1",
|
||||||
.minimum_zig_version = "0.16.0",
|
.minimum_zig_version = "0.16.0",
|
||||||
.paths = .{""},
|
.paths = .{""},
|
||||||
.fingerprint = 0xb3baeb7ec5369577,
|
.fingerprint = 0xcca51d85af632c24,
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -131,11 +131,17 @@ fn add_server_info(ctx: *Context, reply: *wire.Reply) !void {
|
|||||||
try reply.put("maxWireVersion", .{ .int32 = 8 });
|
try reply.put("maxWireVersion", .{ .int32 = 8 });
|
||||||
try reply.put("readOnly", .{ .bool = false });
|
try reply.put("readOnly", .{ .bool = false });
|
||||||
|
|
||||||
const oid = ctx.oid_gen.new(ctx.io);
|
// Deliberately no `topologyVersion`. A driver treats its presence as
|
||||||
const tv = try reply.arena_alloc().alloc(bson.Pair, 2);
|
// "this server supports the streaming (awaitable) hello protocol" and
|
||||||
tv[0] = .{ .key = "processId", .value = .{ .object_id = oid } };
|
// switches monitoring to an exhaust hello: it sends one hello with
|
||||||
tv[1] = .{ .key = "counter", .value = .{ .int64 = 0 } };
|
// maxAwaitTimeMS and the OP_MSG exhaustAllowed flag, then expects a
|
||||||
try reply.put("topologyVersion", .{ .doc = tv });
|
// 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 {
|
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 {
|
fn cmd_build_info(_: *Context, _: *wire.Message, reply: *wire.Reply) !void {
|
||||||
try reply.put("version", .{ .string = "4.4.0" });
|
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("versionArray", .{ .array = try int_array(reply, &.{ 4, 4, 0, 0 }) });
|
||||||
try reply.put("openssl", .{ .doc = &.{} });
|
try reply.put("openssl", .{ .doc = &.{} });
|
||||||
try reply.put("loaderFlags", .{ .string = "" });
|
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 {
|
fn cmd_get_cmd_line_opts(_: *Context, _: *wire.Message, reply: *wire.Reply) !void {
|
||||||
const argv = try reply.arena_alloc().alloc(bson.Pair, 2);
|
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 } };
|
argv[1] = .{ .key = "port", .value = .{ .int32 = 27017 } };
|
||||||
try reply.put("argv", .{ .array = &.{} });
|
try reply.put("argv", .{ .array = &.{} });
|
||||||
try reply.put("parsed", .{ .doc = argv });
|
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);
|
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" {
|
test "unknown command gives CommandNotFound" {
|
||||||
var threaded: std.Io.Threaded = .init_single_threaded;
|
var threaded: std.Io.Threaded = .init_single_threaded;
|
||||||
defer threaded.deinit();
|
defer threaded.deinit();
|
||||||
|
|||||||
@@ -549,13 +549,13 @@ pub const Engine = struct {
|
|||||||
while (doc_it.next()) |doc_entry| {
|
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) {
|
const duplicate = ix.add_doc(self.gpa, doc_entry.value_ptr.*, doc_entry.key_ptr.*, false) catch |err| switch (err) {
|
||||||
error.ParallelArrays => {
|
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;
|
continue;
|
||||||
},
|
},
|
||||||
else => return err,
|
else => return err,
|
||||||
};
|
};
|
||||||
if (duplicate) {
|
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) {
|
switch (record.type) {
|
||||||
storage.record_type_index_create => {
|
storage.record_type_index_create => {
|
||||||
self.register_index_from_spec(coll, doc) catch |err| {
|
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;
|
||||||
};
|
};
|
||||||
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 {
|
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;
|
return;
|
||||||
};
|
};
|
||||||
const id_key = try bson.serialize_value(self.gpa, id_value);
|
const id_key = try bson.serialize_value(self.gpa, id_value);
|
||||||
|
|||||||
@@ -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 bson = @import("bson.zig");
|
||||||
pub const wire = @import("wire.zig");
|
pub const wire = @import("wire.zig");
|
||||||
|
|||||||
16
src/main.zig
16
src/main.zig
@@ -2,12 +2,12 @@ const std = @import("std");
|
|||||||
const mongo = @import("mongo");
|
const mongo = @import("mongo");
|
||||||
|
|
||||||
const usage =
|
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 <n> listen port (default 27017)
|
\\ --port <n> listen port (default 27017)
|
||||||
\\ --bind <ip> bind address (default 127.0.0.1)
|
\\ --bind <ip> bind address (default 127.0.0.1)
|
||||||
\\ --db <path> database file (default mongo-light.log)
|
\\ --db <path> database file (default mongo-lite.log)
|
||||||
\\ --ttl-sweep-secs <n>
|
\\ --ttl-sweep-secs <n>
|
||||||
\\ seconds between TTL index sweeps (default 60, 0 disables)
|
\\ seconds between TTL index sweeps (default 60, 0 disables)
|
||||||
\\ --help show this help
|
\\ --help show this help
|
||||||
@@ -17,7 +17,7 @@ const usage =
|
|||||||
pub fn main(init: std.process.Init) !void {
|
pub fn main(init: std.process.Init) !void {
|
||||||
var port: u16 = 27017;
|
var port: u16 = 27017;
|
||||||
var bind_ip: []const u8 = "127.0.0.1";
|
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 ttl_sweep_secs: i64 = 60;
|
||||||
|
|
||||||
var it = std.process.Args.Iterator.init(init.minimal.args);
|
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")) {
|
if (std.mem.eql(u8, arg, "--port")) {
|
||||||
const v = it.next() orelse return error.MissingValue;
|
const v = it.next() orelse return error.MissingValue;
|
||||||
port = std.fmt.parseInt(u16, v, 10) catch {
|
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;
|
return error.InvalidPort;
|
||||||
};
|
};
|
||||||
} else if (std.mem.eql(u8, arg, "--bind")) {
|
} 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.
|
// only thing parseInt would otherwise let through.
|
||||||
ttl_sweep_secs = std.fmt.parseInt(i64, v, 10) catch -1;
|
ttl_sweep_secs = std.fmt.parseInt(i64, v, 10) catch -1;
|
||||||
if (ttl_sweep_secs < 0) {
|
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;
|
return error.InvalidTtlSweepSecs;
|
||||||
}
|
}
|
||||||
} else if (std.mem.eql(u8, arg, "--help") or std.mem.eql(u8, arg, "-h")) {
|
} else if (std.mem.eql(u8, arg, "--help") or std.mem.eql(u8, arg, "-h")) {
|
||||||
try std.Io.File.writeStreamingAll(.stdout(), init.io, usage);
|
try std.Io.File.writeStreamingAll(.stdout(), init.io, usage);
|
||||||
return;
|
return;
|
||||||
} else {
|
} 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;
|
return error.UnknownOption;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -56,7 +56,7 @@ pub fn main(init: std.process.Init) !void {
|
|||||||
const oid_gen = mongo.bson.ObjectIdGen.init(init.io);
|
const oid_gen = mongo.bson.ObjectIdGen.init(init.io);
|
||||||
var engine = try mongo.db.Engine.open(init.gpa, init.io, db_path);
|
var engine = try mongo.db.Engine.open(init.gpa, init.io, db_path);
|
||||||
defer engine.deinit();
|
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{
|
var server = mongo.server.Server{
|
||||||
.gpa = init.gpa,
|
.gpa = init.gpa,
|
||||||
|
|||||||
@@ -35,7 +35,7 @@ pub const Server = struct {
|
|||||||
var listener = try addr.listen(io, .{ .reuse_address = true });
|
var listener = try addr.listen(io, .{ .reuse_address = true });
|
||||||
defer listener.deinit(io);
|
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;
|
var group: std.Io.Group = .init;
|
||||||
defer group.cancel(io);
|
defer group.cancel(io);
|
||||||
@@ -48,7 +48,7 @@ pub const Server = struct {
|
|||||||
const stream = listener.accept(io) catch |err| switch (err) {
|
const stream = listener.accept(io) catch |err| switch (err) {
|
||||||
error.Canceled => return,
|
error.Canceled => return,
|
||||||
else => {
|
else => {
|
||||||
std.debug.print("mongo-light: accept error: {s}\n", .{@errorName(err)});
|
std.debug.print("mongo-lite: accept error: {s}\n", .{@errorName(err)});
|
||||||
continue;
|
continue;
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
@@ -72,7 +72,7 @@ fn ttl_monitor(io: std.Io, server: *Server) error{Canceled}!void {
|
|||||||
defer server.engine.unlock();
|
defer server.engine.unlock();
|
||||||
const now_ms = std.Io.Timestamp.now(io, .real).toMilliseconds();
|
const now_ms = std.Io.Timestamp.now(io, .real).toMilliseconds();
|
||||||
_ = server.engine.ttl_sweep(now_ms) catch |err| {
|
_ = 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;
|
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)
|
reader.interface.readSliceAll(&len_bytes) catch return; // clean client disconnect (EOF or RST)
|
||||||
const total: u32 = std.mem.readInt(u32, &len_bytes, .little);
|
const total: u32 = std.mem.readInt(u32, &len_bytes, .little);
|
||||||
if (total < 16 or total > wire.max_message_size) {
|
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;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -122,14 +122,14 @@ fn handle_connection_inner(io: std.Io, stream: std.Io.net.Stream, server: *Serve
|
|||||||
msg_buf.items.len = total;
|
msg_buf.items.len = total;
|
||||||
std.mem.writeInt(u32, msg_buf.items[0..4], total, .little);
|
std.mem.writeInt(u32, msg_buf.items[0..4], total, .little);
|
||||||
reader.interface.readSliceAll(msg_buf.items[4..]) catch |err| {
|
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;
|
return;
|
||||||
};
|
};
|
||||||
|
|
||||||
var msg = wire.Message.parse(server.gpa, msg_buf.items) catch |err| {
|
var msg = wire.Message.parse(server.gpa, msg_buf.items) catch |err| {
|
||||||
// Unparseable request: close the connection.
|
// 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;
|
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;
|
return;
|
||||||
};
|
};
|
||||||
defer msg.deinit();
|
defer msg.deinit();
|
||||||
@@ -154,16 +154,16 @@ fn handle_connection_inner(io: std.Io, stream: std.Io.net.Stream, server: *Serve
|
|||||||
else
|
else
|
||||||
reply.build(server.gpa, reply_request_id, msg.request_id, &out_buf);
|
reply.build(server.gpa, reply_request_id, msg.request_id, &out_buf);
|
||||||
built catch |err| {
|
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;
|
return;
|
||||||
};
|
};
|
||||||
reply_request_id +%= 1;
|
reply_request_id +%= 1;
|
||||||
writer.interface.writeAll(out_buf.items) catch |err| {
|
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;
|
return;
|
||||||
};
|
};
|
||||||
writer.interface.flush() catch |err| {
|
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;
|
return;
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -91,21 +91,21 @@ pub const Log = struct {
|
|||||||
|
|
||||||
while (true) {
|
while (true) {
|
||||||
const len_read = self.file.readPositionalAll(self.io, chunk[0..4], pos) catch |err| {
|
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;
|
return error.InvalidLog;
|
||||||
};
|
};
|
||||||
if (len_read == 0) return; // clean end
|
if (len_read == 0) return; // clean end
|
||||||
if (len_read < 4) return; // torn tail
|
if (len_read < 4) return; // torn tail
|
||||||
const total: u32 = std.mem.readInt(u32, chunk[0..4], .little);
|
const total: u32 = std.mem.readInt(u32, chunk[0..4], .little);
|
||||||
if (total < header_len) {
|
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;
|
return error.InvalidLog;
|
||||||
}
|
}
|
||||||
const payload_len: usize = total - 4;
|
const payload_len: usize = total - 4;
|
||||||
// Documents up to maxBsonObjectSize are legal; anything larger is
|
// Documents up to maxBsonObjectSize are legal; anything larger is
|
||||||
// corruption. Covers a hostile length prefix from a truncated file.
|
// corruption. Covers a hostile length prefix from a truncated file.
|
||||||
if (payload_len > max_record_payload) {
|
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;
|
return error.InvalidLog;
|
||||||
}
|
}
|
||||||
const payload = if (payload_len <= chunk.len) chunk[0..payload_len] else blk: {
|
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);
|
const doc = try self.gpa.create(bson.Document);
|
||||||
doc.* = bson.Document.parse(self.gpa, doc_bytes) catch {
|
doc.* = bson.Document.parse(self.gpa, doc_bytes) catch {
|
||||||
self.gpa.destroy(doc);
|
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;
|
return error.InvalidLog;
|
||||||
};
|
};
|
||||||
try callback(ctx, .{
|
try callback(ctx, .{
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
# End-to-end tests with the official MongoDB Node.js driver
|
# 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.
|
operators, aggregation, error codes, concurrent clients, and crash recovery.
|
||||||
|
|
||||||
## Setup
|
## Setup
|
||||||
@@ -17,7 +17,7 @@ Start the server, then run the suites against it (defaults to port 27020):
|
|||||||
|
|
||||||
```sh
|
```sh
|
||||||
zig build
|
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/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 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
|
Rebuild with `zig build` after any change under `src/` before restarting the
|
||||||
server: `zig build test` compiles the test binary only and leaves
|
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.
|
rules and report failures that the source no longer explains.
|
||||||
|
|
||||||
`e2e2.js concurrent` is safe to repeat against a running server (it drops its
|
`e2e2.js concurrent` is safe to repeat against a running server (it drops its
|
||||||
|
|||||||
@@ -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 { MongoClient, ObjectId } = require('mongodb');
|
||||||
|
|
||||||
const URL = 'mongodb://127.0.0.1:27020';
|
const URL = 'mongodb://127.0.0.1:27020';
|
||||||
|
|||||||
@@ -4,7 +4,7 @@
|
|||||||
// server must reject with CannotCreateIndex (67).
|
// server must reject with CannotCreateIndex (67).
|
||||||
//
|
//
|
||||||
// The server must run with a short sweep interval, e.g.
|
// 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 { MongoClient } = require('mongodb');
|
||||||
|
|
||||||
const URL = 'mongodb://127.0.0.1:27020';
|
const URL = 'mongodb://127.0.0.1:27020';
|
||||||
|
|||||||
75
tests/e2e/e2e5.js
Normal file
75
tests/e2e/e2e5.js
Normal file
@@ -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);
|
||||||
|
});
|
||||||
@@ -1,7 +1,7 @@
|
|||||||
{
|
{
|
||||||
"name": "e2e",
|
"name": "e2e",
|
||||||
"version": "1.0.0",
|
"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",
|
"main": "e2e.js",
|
||||||
"scripts": {
|
"scripts": {
|
||||||
"test": "echo \"Error: no test specified\" && exit 1"
|
"test": "echo \"Error: no test specified\" && exit 1"
|
||||||
|
|||||||
Reference in New Issue
Block a user