dev/MultiforaDB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

commands: fix a remote invalid free in aggregate $sort

Browse Source 
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.
This commit is contained in:
Alexey Shakhmatov
2026-08-03 17:09:21 +03:00
parent 06504127fb
commit 411a380d38
2 changed files with 204 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

207
src/commands.zig
View File

@@ -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: <n>, ns: "...", <batch_key>: [...] } 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

17
tests/e2e/e2e.js
View File

@@ -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 {