From 411a380d386dc67d3d3f02d831ac729633e013c0 Mon Sep 17 00:00:00 2001 From: Aleksey Shakhmatov Date: Mon, 3 Aug 2026 17:09:21 +0300 Subject: [PATCH] commands: fix a remote invalid free in aggregate $sort Present since at least d4c9b04, found by the new spec-test harness on its first run. The $sort stage's materialization branch built its document list with the reply arena and then handed it to `trees`, whose scope-exit deinit -- and the $match branch above it -- free with the gpa. So a gpa free was handed an arena-owned pointer. macOS malloc catches it and aborts with SIGTRAP and no panic text, which is why the symptom read as "the connection closed": mfm_free <- Allocator.rawFree <- array_list.Aligned(*const bson.Document).deinit <- commands.cmd_aggregate Any pipeline with $sort and no preceding $group reached it, e.g. aggregate([{$sort: {x: 1}}]) -- so a client could kill the server with one ordinary query. With a $group first the stream is already in tree form and the branch is skipped, which is precisely why it survived: every aggregate case in e2e.js and e2e6.js sorts *after* grouping. The list buffer now comes from ctx.gpa. The documents stay in the arena on purpose -- it outlives the command, and only the ArrayList's own allocator has to match its deinit. Tests. The unit test uses a bare $sort pipeline, since a $group first would not reach the branch, and leans on testing.allocator detecting the invalid free itself rather than on the host allocator noticing -- mutation-checked by restoring `arena` on the append, which gives `panic: Invalid free`. The e2e case adds a second command afterwards, because the assertion that matters is not that the sort returned rows but that the connection is still there. --- src/commands.zig | 207 ++++++++++++++++++++++++++++++++++++++++++----- tests/e2e/e2e.js | 17 ++++ 2 files changed, 204 insertions(+), 20 deletions(-) diff --git a/src/commands.zig b/src/commands.zig index 6bce82b..3499028 100644 --- a/src/commands.zig +++ b/src/commands.zig @@ -424,7 +424,8 @@ fn cmd_create_indexes(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !vo return reply.put_error( @intFromEnum(ErrorCode.invalid_index_specification_option), "InvalidIndexSpecificationOption", - "the field 'expireAfterSeconds' is not valid for an _id index specification", + "the field 'expireAfterSeconds' is not valid for an _id " ++ + "index specification", ); } continue; @@ -447,7 +448,8 @@ fn cmd_create_indexes(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !vo error.InvalidExpireAfterSeconds => return reply.put_error( @intFromEnum(ErrorCode.cannot_create_index), "CannotCreateIndex", - "TTL index 'expireAfterSeconds' option must be a whole number between 0 and 2147483647", + "TTL index 'expireAfterSeconds' option must be a whole number " ++ + "between 0 and 2147483647", ), error.IndexOptionsConflict => return reply.put_error(@intFromEnum(ErrorCode.index_options_conflict), "IndexOptionsConflict", "index already exists with a different specification"), error.DuplicateKeyIndex => { @@ -491,7 +493,11 @@ fn cmd_list_indexes(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void } /// The _id_ index entry: {v, key: {_id: 1}, name: "_id_"}. -fn index_pairs_append(reply: *wire.Reply, pairs: []const bson.Pair, name: []const u8) ![]const bson.Pair { +fn index_pairs_append( + reply: *wire.Reply, + pairs: []const bson.Pair, + name: []const u8, +) ![]const bson.Pair { const arena = reply.arena_alloc(); const with_name = try arena.alloc(bson.Pair, pairs.len + 1); @memcpy(with_name[0..pairs.len], pairs); @@ -541,7 +547,13 @@ fn cmd_drop_indexes(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void /// The E11000 text drivers parse. One definition for both create-time and /// write-time conflicts; they differ only in how the dup key is rendered. -fn e11000_message(reply: *wire.Reply, db_name: []const u8, coll_name: []const u8, index_name: []const u8, key_text: []const u8) ![]const u8 { +fn e11000_message( + reply: *wire.Reply, + db_name: []const u8, + coll_name: []const u8, + index_name: []const u8, + key_text: []const u8, +) ![]const u8 { return std.fmt.allocPrint( reply.arena_alloc(), "E11000 duplicate key error collection: {s}.{s} index: {s} dup key: {s}", @@ -761,7 +773,13 @@ fn doc_tree(arena: std.mem.Allocator, coll: *const Collection, off: u64) !*const return doc; } -fn emit_docs_tree(reply: *wire.Reply, db_name: []const u8, coll_name: []const u8, proj_pairs: ?[]const bson.Pair, docs: []const *const bson.Document) !void { +fn emit_docs_tree( + reply: *wire.Reply, + db_name: []const u8, + coll_name: []const u8, + proj_pairs: ?[]const bson.Pair, + docs: []const *const bson.Document, +) !void { const values = try reply.arena_alloc().alloc(bson.Value, docs.len); for (docs, 0..) |d, i| { values[i] = try project_doc(reply, d, proj_pairs); @@ -770,7 +788,11 @@ fn emit_docs_tree(reply: *wire.Reply, db_name: []const u8, coll_name: []const u8 } /// Project a stored doc (or deep-copy it) into the reply arena. -fn project_doc(reply: *wire.Reply, doc: *const bson.Document, proj_pairs: ?[]const bson.Pair) !bson.Value { +fn project_doc( + reply: *wire.Reply, + doc: *const bson.Document, + proj_pairs: ?[]const bson.Pair, +) !bson.Value { if (proj_pairs) |pp| { var out: std.ArrayListUnmanaged(bson.Pair) = .empty; errdefer out.deinit(reply.arena_alloc()); @@ -1091,8 +1113,23 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void { if (!in_trees) { // Sorting needs the values; materialize and switch the // stream to tree form for the rest of the pipeline. + // + // The list buffer must come from `ctx.gpa`, because that is + // what frees it: ownership moves to `trees`, and `trees` is + // released by this function's `defer trees.deinit(ctx.gpa)` + // and by the $match branch above. Building it from the + // reply arena instead handed a gpa-free an arena-owned + // pointer -- a remote, client-triggerable invalid free that + // macOS malloc turns into SIGTRAP with no panic text, so it + // read as "the connection closed". Any pipeline with $sort + // and no preceding $group reached it. + // + // The *documents* stay in the arena on purpose: it outlives + // the command, and only the ArrayList's own allocator has + // to match its deinit. var all: std.ArrayListUnmanaged(*const bson.Document) = .empty; - for (offs.items) |off| try all.append(arena, try doc_tree(arena, coll, off)); + errdefer all.deinit(ctx.gpa); + for (offs.items) |off| try all.append(ctx.gpa, try doc_tree(arena, coll, off)); trees.deinit(ctx.gpa); trees = all; all = .empty; @@ -1219,7 +1256,13 @@ fn count_only_pipeline(reply: *wire.Reply, stages: []const bson.Value) !?CountSh /// Minimal $group: supports `_id` of null/literal/"$field" and `$sum` /// accumulators (constant or "$field"). -fn run_group(ctx: *Context, reply: *wire.Reply, coll: *const Collection, group_pairs: []const bson.Pair, docs: []const u64) !?std.ArrayListUnmanaged(*const bson.Document) { +fn run_group( + ctx: *Context, + reply: *wire.Reply, + coll: *const Collection, + group_pairs: []const bson.Pair, + docs: []const u64, +) !?std.ArrayListUnmanaged(*const bson.Document) { const arena = reply.arena_alloc(); const id_expr = bson.get_pair(group_pairs, "_id") orelse { try bad_value(reply, "$group requires _id"); @@ -1326,7 +1369,11 @@ fn run_group(ctx: *Context, reply: *wire.Reply, coll: *const Collection, group_p } /// Resolve a simple "$field" path expression inside a document. -fn query_path_value_bytes(gpa: std.mem.Allocator, bytes: []const u8, path: []const u8) !?bson.Value { +fn query_path_value_bytes( + gpa: std.mem.Allocator, + bytes: []const u8, + path: []const u8, +) !?bson.Value { var cur: bson.Value = undefined; var it = std.mem.splitScalar(u8, path, '.'); const first = it.next() orelse return null; @@ -1371,7 +1418,11 @@ fn clone_doc(reply: *wire.Reply, doc: *const bson.Document) !*bson.Document { /// Build the document for an upsert: equality fields from the filter, then /// the update operators applied. Owned by the reply arena. -fn build_upsert_doc(reply: *wire.Reply, q: []const bson.Pair, u_doc: []const bson.Pair) !*bson.Document { +fn build_upsert_doc( + reply: *wire.Reply, + q: []const bson.Pair, + u_doc: []const bson.Pair, +) !*bson.Document { const arena = reply.arena_alloc(); var pairs: std.ArrayListUnmanaged(bson.Pair) = .empty; defer pairs.deinit(arena); @@ -1463,7 +1514,12 @@ fn stage_count(reply: *wire.Reply, v: bson.Value, stage: []const u8) !?usize { /// Fetch a batch argument, writing the standard error reply and returning /// null when it is missing or malformed. -fn batch_arg(msg: *wire.Message, reply: *wire.Reply, cmd: []const u8, name: []const u8) !?[]const bson.Document { +fn batch_arg( + msg: *wire.Message, + reply: *wire.Reply, + cmd: []const u8, + name: []const u8, +) !?[]const bson.Document { return msg.batch(name) catch |err| { const arena = reply.arena_alloc(); const text = switch (err) { @@ -1490,7 +1546,13 @@ fn bad_value(reply: *wire.Reply, msg: []const u8) !void { /// dup_index (set by a rejected unique-index write) when the conflict came /// from a secondary index; otherwise it is the _id_ index. Per-collection /// so concurrent writers on other collections cannot clobber it. -fn duplicate_key_message(ctx: *Context, reply: *wire.Reply, db_name: []const u8, coll_name: []const u8, doc: *const bson.Document) ![]const u8 { +fn duplicate_key_message( + ctx: *Context, + reply: *wire.Reply, + db_name: []const u8, + coll_name: []const u8, + doc: *const bson.Document, +) ![]const u8 { var index_name: []const u8 = "_id_"; var key_text: []const u8 = undefined; const coll = ctx.engine.get_collection(db_name, coll_name); @@ -1507,7 +1569,14 @@ fn duplicate_key_message(ctx: *Context, reply: *wire.Reply, db_name: []const u8, /// Render the dup key of a secondary index from the offending document: the /// document's values for the index key pattern. -fn render_dup_key(ctx: *Context, reply: *wire.Reply, db_name: []const u8, coll_name: []const u8, index_name: []const u8, doc: *const bson.Document) ![]const u8 { +fn render_dup_key( + ctx: *Context, + reply: *wire.Reply, + db_name: []const u8, + coll_name: []const u8, + index_name: []const u8, + doc: *const bson.Document, +) ![]const u8 { const arena = reply.arena_alloc(); const coll = ctx.engine.get_collection(db_name, coll_name) orelse // Index not found (defensive): fall back to the document's _id. @@ -1532,7 +1601,13 @@ fn render_dup_key(ctx: *Context, reply: *wire.Reply, db_name: []const u8, coll_n return out.toOwnedSlice(arena); } -fn duplicate_key_error(ctx: *Context, reply: *wire.Reply, db_name: []const u8, coll_name: []const u8, doc: *const bson.Document) !void { +fn duplicate_key_error( + ctx: *Context, + reply: *wire.Reply, + db_name: []const u8, + coll_name: []const u8, + doc: *const bson.Document, +) !void { const msg_text = try duplicate_key_message(ctx, reply, db_name, coll_name, doc); return reply.put_error(@intFromEnum(ErrorCode.duplicate_key), "DuplicateKey", msg_text); } @@ -1555,7 +1630,13 @@ fn serialize_value_compact(reply: *wire.Reply, v: bson.Value) ![]const u8 { } /// Build a { id: , ns: "...", : [...] } cursor document. -pub fn cursor_doc(reply: *wire.Reply, cursor_id: i64, ns: []const u8, batch_key: []const u8, docs: []const bson.Value) ![]const bson.Pair { +pub fn cursor_doc( + reply: *wire.Reply, + cursor_id: i64, + ns: []const u8, + batch_key: []const u8, + docs: []const bson.Value, +) ![]const bson.Pair { const c = try reply.arena_alloc().alloc(bson.Pair, 3); c[0] = .{ .key = "id", .value = .{ .int64 = cursor_id } }; c[1] = .{ .key = "ns", .value = .{ .string = ns } }; @@ -1784,7 +1865,14 @@ test "concurrent insert/find commands on a threaded Io" { var remaining = std.atomic.Value(usize).init(@intCast(total)); const Worker = struct { - fn writer(iow: std.Io, eng: *db.Engine, id_counter: *std.atomic.Value(i32), pending: *std.atomic.Value(usize), fiber_id: u32, total_writes: i32) error{Canceled}!void { + fn writer( + iow: std.Io, + eng: *db.Engine, + id_counter: *std.atomic.Value(i32), + pending: *std.atomic.Value(usize), + fiber_id: u32, + total_writes: i32, + ) error{Canceled}!void { var wgen = bson.ObjectIdGen.init(iow); var ctx = test_ctx(iow, eng, &wgen, fiber_id); while (true) { @@ -1802,7 +1890,13 @@ test "concurrent insert/find commands on a threaded Io" { } } - fn reader(iow: std.Io, eng: *db.Engine, pending: *std.atomic.Value(usize), fiber_id: u32, total_writes: i32) error{Canceled}!void { + fn reader( + iow: std.Io, + eng: *db.Engine, + pending: *std.atomic.Value(usize), + fiber_id: u32, + total_writes: i32, + ) error{Canceled}!void { var rgen = bson.ObjectIdGen.init(iow); var ctx = test_ctx(iow, eng, &rgen, fiber_id); while (pending.load(.acquire) > 0) { @@ -1835,7 +1929,12 @@ test "concurrent insert/find commands on a threaded Io" { // -- index command tests ---------------------------------------------------- /// Dispatch an insert of the given documents (each a bson.Value .doc). -fn dispatch_insert(tdb: *TestDb, io: std.Io, coll_name: []const u8, docs: []const bson.Value) !void { +fn dispatch_insert( + tdb: *TestDb, + io: std.Io, + coll_name: []const u8, + docs: []const bson.Value, +) !void { var ctx = tdb.ctx(io); var msg = try parse_fake_msg("insert", .{ .string = coll_name }, &.{ .{ .key = "documents", .value = .{ .array = docs } }, @@ -1862,7 +1961,13 @@ fn dispatch_create_index(tdb: *TestDb, io: std.Io, coll_name: []const u8, spec: } /// Dispatch find and append the serialized _id of every result to `out`. -fn dispatch_find_ids(tdb: *TestDb, io: std.Io, coll_name: []const u8, filter: []const bson.Pair, out: *std.ArrayListUnmanaged([]u8)) !void { +fn dispatch_find_ids( + tdb: *TestDb, + io: std.Io, + coll_name: []const u8, + filter: []const bson.Pair, + out: *std.ArrayListUnmanaged([]u8), +) !void { var ctx = tdb.ctx(io); var msg = try parse_fake_msg("find", .{ .string = coll_name }, &.{ .{ .key = "filter", .value = .{ .doc = filter } }, @@ -1987,7 +2092,13 @@ test "TTL index round-trips through createIndexes/listIndexes; bad specs give 67 // A driver sends expireAfterSeconds as a double. try dispatch_create_index(&tdb, io, "sessions", .{ .doc = &.{ - .{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } }, + .{ + .key = "key", + .value = .{ .doc = &.{ + .{ .key = "expireAt", .value = .{ .int32 = 1 } }, + } }, + }, + .{ .key = "name", .value = .{ .string = "expireAt_1" } }, .{ .key = "expireAfterSeconds", .value = .{ .double = 60.0 } }, } }); @@ -2128,6 +2239,62 @@ fn clear_id_list(gpa: std.mem.Allocator, list: *std.ArrayListUnmanaged([]u8)) vo list.clearRetainingCapacity(); } +test "aggregate $sort without a preceding $group sorts and frees correctly" { + // Regression test for a remote, client-triggerable invalid free: the + // $sort stage materialized its document list from the reply arena and + // handed it to `trees`, which is freed with the gpa. Two things made it + // survive for so long, and this test is shaped to close both: + // + // - every existing aggregate test sorts *after* a $group, which leaves + // the stream already materialized so the guilty branch never runs. + // So this pipeline must have $sort with NO $group before it. + // - the symptom was allocator-dependent (macOS malloc aborted; other + // allocators may not notice). testing.allocator detects an invalid + // free itself, which is what gives this teeth in every mode. + // + // Mutation check: change the `all.append(ctx.gpa, ...)` back to + // `all.append(arena, ...)` in cmd_aggregate and this goes red. + var threaded = std.Io.Threaded.init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var tdb = try TestDb.init(io); + defer tdb.deinit(); + + // Insert out of order so a missing sort is visible, not coincidental. + try dispatch_insert(&tdb, io, "agg", &.{ + .{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "x", .value = .{ .int32 = 30 } } } }, + .{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 2 } }, .{ .key = "x", .value = .{ .int32 = 10 } } } }, + .{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 3 } }, .{ .key = "x", .value = .{ .int32 = 20 } } } }, + }); + + const sort_stage = bson.Value{ .doc = &.{.{ .key = "$sort", .value = .{ .doc = &.{.{ .key = "x", .value = .{ .int32 = 1 } }} } }} }; + const stages = [_]bson.Value{sort_stage}; + + var ctx = tdb.ctx(io); + var msg = try parse_fake_msg("aggregate", .{ .string = "agg" }, &.{ + .{ .key = "pipeline", .value = .{ .array = &stages } }, + .{ .key = "cursor", .value = .{ .doc = &.{} } }, + }); + defer msg.deinit(); + var reply = wire.Reply.init(testing.allocator); + defer reply.deinit(); + try dispatch(&ctx, &msg, &reply); + + try testing.expectEqual(@as(f64, 1.0), bson.get_pair(reply.pairs.items, "ok").?.double); + const cursor = bson.get_pair(reply.pairs.items, "cursor") orelse return error.TestUnexpectedResult; + const batch = switch (bson.get_pair(cursor.doc, "firstBatch") orelse return error.TestUnexpectedResult) { + .array => |a| a, + else => return error.TestUnexpectedResult, + }; + try testing.expectEqual(@as(usize, 3), batch.len); + // Ascending by x means _id order 2, 3, 1. + const want = [_]i32{ 2, 3, 1 }; + for (batch, want) |d, id| { + try testing.expectEqual(id, bson.get_pair(d.doc, "_id").?.int32); + } +} + test "count_only_pipeline accepts only shapes a count can answer" { // The fast path skips materializing documents, so mis-accepting a // pipeline would silently return a wrong aggregate rather than a slow diff --git a/tests/e2e/e2e.js b/tests/e2e/e2e.js index 474485e..d74fdb9 100644 --- a/tests/e2e/e2e.js +++ b/tests/e2e/e2e.js @@ -93,6 +93,23 @@ async function main() { const cnt = await users.aggregate([{ $match: {} }, { $count: 'n' }]).toArray(); check('aggregate $count', cnt[0]?.n === 4, JSON.stringify(cnt)); + // $sort with NO preceding $group. This shape used to kill the server: the + // stage materialized its document list from the reply arena and it was then + // freed with the general allocator. Every aggregate case above happens to + // sort after a $group, which leaves the stream already materialized and the + // guilty branch unreached -- so the bug survived the whole suite. The second + // aggregate is the part that actually proves recovery: if the server died, + // this connection is gone. + const sorted = await users.aggregate([{ $sort: { age: -1 } }]).toArray(); + const ages = sorted.map((d) => d.age); + check( + 'aggregate bare $sort (no $group) returns sorted docs', + ages.length === 4 && ages.every((a, i) => i === 0 || ages[i - 1] >= a), + JSON.stringify(ages), + ); + const stillAlive = await users.aggregate([{ $match: {} }, { $count: 'n' }]).toArray(); + check('server survives a bare $sort pipeline', stillAlive[0]?.n === 4, JSON.stringify(stillAlive)); + // --- duplicate key --- let dupErr = null; try {