commands: drop does not unlock the collection it just freed

A use-after-free. Dispatch held `drop`'s collection lock across the handler,
`drop_collection` freed the Collection the lock lives in, and dispatch then
ran `unlock_collection` on freed memory -- an atomic read-modify-write
inside `Io.RwLock.unlock`. One insert and one drop was enough.

Two things kept it hidden for this long. `drop` had no unit test at all:
before this commit every `parse_fake_msg("drop", ...)` in the tree was in a
test written to hunt it. And over the wire it does not fault -- 25
insert/drop cycles against a live server pass -- because the general
allocator leaves the freed page mapped and the atomic write lands somewhere
harmless. That was luck, not safety: the same undefined behaviour either
way, and testing.allocator is what makes it visible, which is why the
regression test is a unit test rather than an e2e script.

Fixed by giving `drop` no collection lock at all. The catalog lock is what
actually excludes here: every collection lock in this engine -- dispatch,
the TTL sweep, `compact`'s rebuild, `write_catalog`, `slab_stats`,
reclamation -- is taken while holding the catalog at least shared, so
holding it exclusively already keeps every one of them out. The collection
lock was buying exclusion that was already there and paying for it by
locking an object about to cease existing.

That exposed a second bug rather than creating one. The dispatch epilogue --
commit, then maybe checkpoint, then maybe compact -- fired on
`locks.coll == .exclusive` as a stand-in for "this was a write". It is now
keyed on `kind == .write`, because the two agreed only by accident:
`dropDatabase` is the one write that never held a collection lock, so it has
never reached that epilogue, and a dropped database waited for some later
write to trigger the checkpoint that records it.

Mutation-checked: restoring `.coll = .exclusive` on the drop row reproduces
the original SIGSEGV in the new test.

208/208 unit (2 new) in ReleaseFast and ReleaseSafe, 83/83 fuzz, crud
scorecard unchanged at 204/87, aggregation corpus 70/0, full e2e matrix,
crash-fuzz and the 25-cycle wire drop probe green.

Left open and recorded in PLAN §6: `drop_collection` still writes no log
record, so a dropped collection resurrects on reopen unless a checkpoint ran
(pre-existing, with its own test at db.zig:5542); and `apply_pending_write`
drops under `engine.rwlock` rather than the catalog lock, so the two drop
paths disagree about which lock protects a namespace.

PLAN.md

@@ -1045,19 +1045,43 @@ has to be its own commit with its own re-recorded scorecard.
   `$out`/`$merge` durability semantics, whether the expression evaluator is
   shared with M3's pipeline updates, and whether `allowDiskUse` has to stop
   being a lie.
-- **`drop` segfaults when dispatched in-process** — found while writing the
+- **`drop` unlocked the collection it had just freed** — *fixed.* Found while
-  positional-refusal tests, unrelated to them, and **reproduces at `3c5eee2`
+  writing the positional-refusal tests and not caused by them; it reproduced
-  with the change stashed**, so it is not caused by that work. A single
+  at `3c5eee2` with that work stashed.
-  `insert` followed by a single `drop` against a `TestDb` engine terminates
+
-  with SIGSEGV and no stack trace; dropping a collection that never existed is
+  A use-after-free, not an allocator quirk. Dispatch held `drop`'s collection
-  fine. The wire path is clean — 25 insert / refused-update / distinct / drop
+  lock across the handler, the handler freed the `Collection` the lock lives
-  cycles against a live server on `:27020` all passed — so what differs is the
+  in, and dispatch then ran `unlock_collection` on freed memory — an atomic
-  in-process caller, not the command. That matters more than it looks: D1's
+  read-modify-write inside `Io.RwLock.unlock`. One insert and one drop was
-  whole architecture is "in-process server now, library + C API later", and
+  enough.
-  this is the embedding path. **`drop` has no unit test at all** — the only
+
-  `parse_fake_msg("drop", ...)` calls in the tree were the scratch ones written
+  The wire path not faulting was luck, not safety: 25 insert/drop cycles
-  to find this — which is why it went unseen. Needs its own commit: reproduce
+  against a live server pass because the general allocator leaves the freed
-  it in a committed test first, then fix.
+  page mapped, so the atomic write lands somewhere harmless. It was the same
+  undefined behaviour either way, and `testing.allocator` is what made it
+  visible. **`drop` had no unit test at all**, which is why it went unseen.
+
+  Fixed by giving `drop` no collection lock. The catalog lock is what actually
+  excludes: every collection lock in the engine — dispatch, the TTL sweep,
+  `compact`'s rebuild, `write_catalog`, `slab_stats`, reclamation — is taken
+  while holding the catalog at least shared, so `drop` holding it exclusively
+  already keeps all of them out. The collection lock bought exclusion that was
+  already there and paid for it by locking an object about to cease existing.
+
+  That change also made the dispatch epilogue's predicate wrong, and it turned
+  out to have been wrong already: it fired on `locks.coll == .exclusive` as a
+  stand-in for "this was a write", and **`dropDatabase` is the one write that
+  never held a collection lock**, so it had never reached the commit and
+  checkpoint epilogue at all. Now keyed on `kind == .write`.
+
+  Still open, and deliberately not fixed here: `drop_collection` writes no log
+  record, so a dropped collection resurrects on reopen unless a checkpoint
+  happened to run — a pre-existing limitation with its own test at
+  `db.zig:5542`. Separately, `apply_pending_write` (`$out`/`$merge`) calls
+  `drop_collection` under `engine.rwlock` rather than the catalog lock, so it
+  is not excluded by the reasoning above; it holds no collection lock, so it
+  is not this crash, but the two drops disagree about which lock protects the
+  namespace and that wants one answer.
 - **M3 update operators** — open. `distinct` landed first because it was a
   whole missing command with no dependencies, and measuring it turned up three
   things worth keeping, none of which are `distinct`'s to fix:

src/commands.zig

@@ -198,7 +198,19 @@ const command_table = [_]Command{
     // Writes: the target collection exclusively; create/drop take the
     // catalog exclusively (they mutate the maps).
     .{ .name = "create", .kind = .write, .locks = .{ .catalog = .exclusive, .coll = .exclusive }, .handler = cmd_create },
-    .{ .name = "drop", .kind = .write, .locks = .{ .catalog = .exclusive, .coll = .exclusive }, .handler = cmd_drop },
+    // `drop` takes the catalog exclusively and **no collection lock**: it frees
+    // the very Collection a lock would live in, and dispatch then unlocked the
+    // freed memory. That was a use-after-free on an `Io.RwLock`, and under
+    // testing.allocator it is a hard SIGSEGV on the first insert-then-drop.
+    //
+    // Nothing is lost by dropping the lock, because the catalog lock is what
+    // actually excludes here: every collection lock in this engine -- dispatch,
+    // the TTL sweep, `compact`'s rebuild, `write_catalog`, `slab_stats`,
+    // reclamation -- is taken while holding the catalog at least shared, so
+    // holding it exclusively already keeps every one of them out. The
+    // collection lock was buying exclusion that was already there, and paying
+    // for it by locking an object about to cease existing.
+    .{ .name = "drop", .kind = .write, .locks = .{ .catalog = .exclusive }, .handler = cmd_drop },
     .{ .name = "dropDatabase", .kind = .write, .locks = .{ .catalog = .exclusive }, .handler = cmd_drop_database },
     .{ .name = "createIndexes", .kind = .write, .locks = .{ .catalog = .shared, .coll = .exclusive }, .handler = cmd_create_indexes },
     .{ .name = "dropIndexes", .kind = .write, .locks = .{ .catalog = .shared, .coll = .exclusive }, .handler = cmd_drop_indexes },
@@ -313,7 +325,15 @@ pub fn dispatch(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
         };
         try ctx.engine.commit();
     }
-    if (cmd.locks.coll == .exclusive) {
+    // Keyed on what the command *is*, not on which lock it happened to take.
+    // These two agreed for every command until `drop` gave up its collection
+    // lock, at which point the old `locks.coll == .exclusive` would have
+    // silently stopped committing and checkpointing after a drop. It was
+    // already wrong for `dropDatabase`, the one write that never held a
+    // collection lock: it has been skipping this epilogue all along, so a
+    // dropped database waited for some later write to trigger a checkpoint
+    // before the catalog recording it was written.
+    if (cmd.kind == .write) {
         // Durability (seal + fsync) coalesces across concurrent writers. A
         // commit error deliberately wins over the handler's captured `result`:
         // whether the write reached disk matters more to the client than why
@@ -5533,6 +5553,67 @@ test "distinct applies its filter before collecting" {
     try testing.expectEqual(@as(i32, 33), values[1].int32);
 }
 
+test "drop does not unlock the collection it just freed" {
+    // Regression test for a use-after-free: dispatch held `drop`'s collection
+    // lock across the handler, the handler freed the Collection the lock lives
+    // in, and dispatch then ran `unlock_collection` on freed memory. One
+    // insert and one drop was enough -- SIGSEGV inside `Io.RwLock.unlock`.
+    //
+    // Two things kept it hidden. `drop` had no unit test at all: before this
+    // one, every `parse_fake_msg("drop", ...)` in the tree was inside a test
+    // written to hunt it. And over the wire it did not fault -- 25
+    // insert/drop cycles against a live server pass -- because the general
+    // allocator leaves the freed page mapped, so the atomic write lands
+    // somewhere harmless. testing.allocator is what makes it visible, which
+    // is exactly why this test belongs here rather than in an e2e script.
+    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);
+
+    try dispatch_insert(&tdb, io, "arr", &.{
+        .{ .doc = &.{.{ .key = "_id", .value = .{ .int32 = 1 } }} },
+    });
+    try testing.expectEqual(@as(?i32, null), try run_for_code(&ctx, "drop", .{ .string = "arr" }, &.{}));
+    try testing.expect(ctx.engine.get_collection("test", "arr") == null);
+
+    // The namespace is reusable afterwards, and dropping it again is
+    // NamespaceNotFound rather than a second free.
+    try dispatch_insert(&tdb, io, "arr", &.{
+        .{ .doc = &.{.{ .key = "_id", .value = .{ .int32 = 2 } }} },
+    });
+    try testing.expectEqual(@as(i32, 1), try doc_count(&ctx, "arr"));
+    try testing.expectEqual(@as(?i32, null), try run_for_code(&ctx, "drop", .{ .string = "arr" }, &.{}));
+    try testing.expectEqual(
+        @as(?i32, @intFromEnum(ErrorCode.namespace_not_found)),
+        try run_for_code(&ctx, "drop", .{ .string = "arr" }, &.{}),
+    );
+}
+
+test "dropDatabase frees its collections without unlocking them" {
+    // The same shape one level up, and the reason the epilogue is now keyed on
+    // `kind == .write`: dropDatabase is the one write that never held a
+    // collection lock, so it never reached the commit/checkpoint epilogue.
+    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);
+
+    try dispatch_insert(&tdb, io, "one", &.{
+        .{ .doc = &.{.{ .key = "_id", .value = .{ .int32 = 1 } }} },
+    });
+    try dispatch_insert(&tdb, io, "two", &.{
+        .{ .doc = &.{.{ .key = "_id", .value = .{ .int32 = 2 } }} },
+    });
+    try testing.expectEqual(@as(?i32, null), try run_for_code(&ctx, "dropDatabase", .{ .int32 = 1 }, &.{}));
+    try testing.expect(ctx.engine.get_collection("test", "one") == null);
+    try testing.expect(ctx.engine.get_collection("test", "two") == null);
+}
+
 test "a positional update is refused on the wire and stores nothing" {
     // The end of the chain the unit tests start: the refusal has to reach the
     // client as a code, and the stored document -- not just the working copy