commands: stop leaking the catalog lock on a nameless command

dispatch resolved the namespace *after* acquiring the catalog lock, and bailed
out with `orelse return` when either part was missing. A plain return is not an
error return, so it ran neither the errdefer nor the explicit unlocks after the
handler: the catalog lock was held, shared, for the life of the process.

`db.aggregate(...)` reaches it. That sends `{aggregate: 1}`, whose value is a
number, so str_arg returns null.

What made this hard to see is that a leaked *shared* lock is invisible to
readers. ping and listDatabases kept answering in microseconds, and the server
looked perfectly healthy from outside -- an external prober got `ok 15ms`
throughout. Only a write needing the catalog exclusive to create a collection
blocked, so the failure surfaced one command later, on a different connection,
as a client-side timeout with nothing to connect it to its cause. It cost three
invalid spec-test baselines before the driver's own command log showed an
insert sitting for exactly socketTimeoutMS against an idle engine.

Namespace resolution now happens before any lock is taken, and a missing name
is a BadValue reply instead of an empty document (which drivers render as the
uninformative "n/a").

Also fixes the aggregate path it exposed: a missing collection returned a reply
with no `ok` field, where MongoDB answers an empty cursor.

Tested by asserting both halves -- a real error reply, and that a following
write which creates a collection still completes. The second is the lock check.
Mutation-checked: reintroducing the leak reddens that test by name.

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) {
+    if (ns) |n| {
-        const db_name = msg.db_name() orelse return;
-        const coll_name = str_arg(msg.body.get(name)) orelse return;
         // 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).