diff --git a/src/commands.zig b/src/commands.zig index 2e7692f..3730019 100644 --- a/src/commands.zig +++ b/src/commands.zig @@ -123,6 +123,28 @@ pub fn dispatch(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void { // the target collection, then run the handler. The collection lock is // taken while the catalog lock is held, so a concurrent drop can never // free the collection out from under us. + // Resolve the namespace *before* taking any lock. It used to happen after + // the catalog lock, with `orelse return` on both parts -- and a plain + // `return` is not an error return, so it ran neither the errdefer below nor + // the explicit unlocks after the handler. The catalog lock was simply + // leaked, shared, forever. + // + // A database-level command reaches it: `db.aggregate(...)` sends + // `{aggregate: 1}`, whose value is not a string, so str_arg returns null. + // The symptom was baffling because a leaked *shared* lock is invisible to + // readers -- ping and listDatabases kept answering in microseconds -- while + // the next write that needs the catalog exclusive to create a collection + // blocks forever. It presented as an unrelated client-side timeout one + // command later. + var ns: ?struct { db: []const u8, coll: []const u8 } = null; + if (cmd.locks.coll != .none) { + const db_name = msg.db_name() orelse + return bad_value(reply, "command requires a $db"); + const coll_name = str_arg(msg.body.get(name)) orelse + return bad_value(reply, "command requires a collection name"); + ns = .{ .db = db_name, .coll = coll_name }; + } + switch (cmd.locks.catalog) { .none => {}, .shared => try ctx.engine.lock_catalog(false), @@ -135,13 +157,11 @@ pub fn dispatch(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void { if (catalog_held) ctx.engine.unlock_catalog(cmd.locks.catalog == .exclusive); } - if (cmd.locks.coll != .none) { - const db_name = msg.db_name() orelse return; - const coll_name = str_arg(msg.body.get(name)) orelse return; + if (ns) |n| { // Write commands may create the collection on first use; the catalog // lock is upgraded to exclusive for that, then restored to shared. const create = cmd.kind == .write and cmd.locks.catalog == .shared; - if (try ctx.engine.lock_collection(db_name, coll_name, cmd.locks.coll == .exclusive, create)) |c| { + if (try ctx.engine.lock_collection(n.db, n.coll, cmd.locks.coll == .exclusive, create)) |c| { coll = c; } } @@ -1055,7 +1075,16 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void { var trees: std.ArrayListUnmanaged(*const bson.Document) = .empty; defer trees.deinit(ctx.gpa); var in_trees = false; - const coll = ctx.engine.get_collection(db_name, coll_name) orelse return; + // No such collection is an empty result, not an absent one. A bare + // `return` here sent a reply with no `ok` field at all, which the driver + // reports as the uninformative `MongoServerError: n/a` -- and it is what + // `db.aggregate(...)` hits, because a database-level aggregate names no + // collection. MongoDB answers an aggregate over a missing collection with + // an empty cursor. + const coll = ctx.engine.get_collection(db_name, coll_name) orelse { + try emit_docs_tree(reply, db_name, coll_name, null, &.{}); + return reply.put_ok(); + }; // A leading $match is pushed down into an indexed candidate scan; the // stage is then dropped from the pipeline so it is not applied twice. if (stages.len > 0 and stages[0] == .doc and stages[0].doc.len > 0 and std.mem.eql(u8, stages[0].doc[0].key, "$match")) { @@ -2351,6 +2380,56 @@ test "count_only_pipeline accepts only shapes a count can answer" { try testing.expect(try count_only_pipeline(&reply, &.{}) == null); } +test "a command with no collection name errors and holds no lock" { + // Regression for a leaked catalog lock. dispatch resolved the namespace + // *after* taking the catalog lock, with `orelse return` -- and a plain + // return runs neither the errdefer nor the explicit unlocks, so the lock + // was held shared forever. `db.aggregate(...)` sends {aggregate: 1}, whose + // value is not a string, so it reached exactly that path. + // + // Two assertions, because the first alone would have passed before the fix + // for the wrong reason: the reply must be a real error (it used to be an + // empty document, which a driver reports as the useless "n/a"), and a + // subsequent write that has to take the catalog exclusive to create a + // collection must still complete. The second is the lock check. + // + // Mutation check: restore the `orelse return` pair after the lock + // acquisition and this test hangs on the insert rather than failing -- the + // suite times out. That is a deadlock, so it cannot be asserted more + // politely from a single fiber; the e2e suite covers the same sequence + // where a hang surfaces as a client timeout instead. + 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); + // {aggregate: 1} -- a database-level aggregate, no collection named. + var msg = try parse_fake_msg("aggregate", .{ .int32 = 1 }, &.{ + .{ .key = "pipeline", .value = .{ .array = &.{} } }, + .{ .key = "cursor", .value = .{ .doc = &.{} } }, + }); + defer msg.deinit(); + var reply = wire.Reply.init(testing.allocator); + defer reply.deinit(); + try dispatch(&ctx, &msg, &reply); + // A proper error, not an empty reply. + try testing.expectEqual(@as(f64, 0.0), bson.get_pair(reply.pairs.items, "ok").?.double); + try testing.expectEqual( + @as(i32, @intFromEnum(ErrorCode.bad_value)), + bson.get_pair(reply.pairs.items, "code").?.int32, + ); + } + + // The lock assertion: this insert creates a collection, which upgrades the + // catalog lock to exclusive. With the lock leaked it never returns. + try dispatch_insert(&tdb, io, "after", &.{ + .{ .doc = &.{.{ .key = "_id", .value = .{ .int32 = 1 } }} }, + }); +} + test "compare-equal _id encodings collide under the unique _id_ index" { // _id uniqueness moved from a docs-map probe keyed on serialize_value to // the _id_ B+tree, keyed on the canonical bson.encode_key (PLAN A3/A4).