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

commands: $out and $merge, written by the dispatch epilogue

Browse Source 
The seven reachable failures of M2, and the first commit of the milestone to
move the scorecard: 194/97/196 -> 201/90/196, with `aggregate-*.json` going
9 pass / 13 fail to 16 pass / 6 fail. The seven that moved are exactly the
seven priced as reachable, and the six that remain are exactly the six
attributed to M2.5 ($addFields, the expression engine), M4 ($listLocalSessions)
and M8 (collation).

Both stages write to a collection the pipeline is not reading, and three things
stood against doing that in the handler: `aggregate` is a `.read` command,
dispatch takes locks from a static table keyed on the command name before the
handler runs, and `Collection.lock` allows exactly one collection lock at a
time. So the handler computes the output under the locks it has and leaves it
in `Context.pending_write`; the epilogue applies it with nothing held, beside
the commit and the checkpoint already there. The `.read`/`.write` contract is
amended in its own comment rather than quietly broken.

`pending_write` is cleared at the top of every dispatch, so a handler that
errors before setting one cannot leave the previous command's write to fire. A
failed write replaces the pipeline's `ok: 1` with the failure, because a client
told the aggregation succeeded would believe the collection had been written.

What the stages do not implement is refused, not ignored: `$merge`'s
`whenMatched`, `whenNotMatched`, `on` and `let` all select behaviour this
server does not have, and a `whenMatched: "fail"` that silently merged would be
the same lie Tier 0 spent three commits removing. Codes measured against
mongod 8.3.7. `$out` and `$merge` answer byte-identically to it on both the
replace and the upsert case.

NOT ATOMIC, and said out loud in the code rather than left to be discovered.
mongod replaces an `$out` target atomically; this engine has no
cross-collection atomicity and no rename to build one from, so a crash between
the drop and the last insert leaves the target holding part of the new output
where MongoDB would leave the whole of the old. The fix is
write-to-temp-and-rename and rename is a command that does not exist here.

The test's mutation is the argument for the epilogue in one line: apply the
write inside the `$out` branch and it deadlocks rather than fails.

191/191 unit tests in ReleaseFast and ReleaseSafe, 83/83 fuzz, e2e 49, e2e2
concurrent, e2e3 16, e2e4 17, e2e6 72, e2e7 86.
This commit was merged in pull request #2.
This commit is contained in:
A.Shakhmatov
2026-08-09 21:02:01 +03:00
parent 89eae1cd9c
commit 8ebeb9d4ec
3 changed files with 299 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
docs/M2_DESIGN_REVIEW.md
View File

@@ -339,5 +339,21 @@ chosen: mongod's `$out` replaces the target collection *atomically*, and this
engine has no cross-collection atomicity. A pipeline that fails after writing
half its output must not leave the target half-replaced.
**Nothing is implemented past Tier 0 until that is decided.**
**Decision: (b), the epilogue.** Taken after the options were set out. The
handler computes the output under the locks it already holds and leaves it in
`Context.pending_write`; the epilogue applies it once every lock is released,
beside the commit and the checkpoint that already live there. The
`.read`/`.write` contract's comment was amended to say so and why.
Measured after: `aggregate-*.json` went 9 pass / 13 fail to 16 pass / 6 fail,
and the whole corpus 194/97/196 to 201/90/196. The seven that moved are exactly
the seven priced as reachable above, and the six that remain are exactly the
six attributed to M2.5, M4 and M8.
**The durability question is answered honestly rather than solved.** `$out` is
*not* atomic here: it drops the target and inserts, so a crash in between
leaves part of the new output where mongod would leave the whole of the old.
The shape that fixes it is write-to-temp-and-rename, and this server has no
rename command. Recorded in `apply_pending_write`'s own comment as well, since
that is where somebody will be standing when it matters.

277
src/commands.zig
View File

@@ -23,6 +23,36 @@ pub const Context = struct {
client_desc: []const u8,
engine: *db.Engine,
server_start: std.Io.Timestamp,
/// What an aggregation's last stage asked to be written, and where.
///
/// `$out` and `$merge` write to a collection the pipeline is not reading,
/// and three things in this file stand against doing that inside the
/// handler: `aggregate` is a `.read` command, dispatch takes locks from a
/// static table before the handler runs, and `Collection.lock` allows only
/// one collection lock at a time. So the handler computes the documents
/// under the locks it has and leaves them here; the epilogue applies them
/// once every lock is released, next to the commit and the checkpoint that
/// already live there.
///
/// Cleared at the top of every dispatch, so a request can never inherit the
/// one before it.
pending_write: ?PendingWrite = null,
};
/// A write an aggregation pipeline asked the epilogue to perform.
pub const PendingWrite = struct {
db: []const u8,
coll: []const u8,
/// Documents in the reply's arena, which outlives the epilogue.
docs: []const *const bson.Document,
mode: enum {
/// `$out`: the target holds the pipeline's output and nothing else.
replace,
/// `$merge`: each document replaces the one with its `_id`, or is
/// inserted. The default `whenMatched`/`whenNotMatched` pair, which is
/// the only one this server implements.
merge,
},
};
pub const ErrorCode = enum(i32) {
@@ -76,6 +106,7 @@ pub const ErrorCode = enum(i32) {
location_project_empty = 51272,
location_project_mixed = 31254,
location_project_unknown_expression = 31325,
location_write_stage_not_last = 40601,
location_unknown_group_operator = 15952,
location_group_needs_id = 15955,
location_accumulator_not_object = 40234,
@@ -202,6 +233,10 @@ pub fn dispatch(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
// 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.
// Nothing carries over: a handler that errors before it sets one must not
// leave the previous command's write to be applied below.
ctx.pending_write = null;
var ns: ?struct { db: []const u8, coll: []const u8 } = null;
if (cmd.locks.coll != .none) {
const db_name = msg.db_name() orelse
@@ -242,6 +277,26 @@ pub fn dispatch(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
coll = null;
if (catalog_held) ctx.engine.unlock_catalog(cmd.locks.catalog == .exclusive);
catalog_held = false;
// Here, and only here: no lock is held, so taking the target's is not a
// second one. See `Context.pending_write`.
if (ctx.pending_write) |pending| {
ctx.pending_write = null;
apply_pending_write(ctx, pending) catch |err| {
// The pipeline's own answer is already in `reply`; replace it with
// the failure, because a client told `ok: 1` would believe the
// collection had been written. The pairs go, the arena stays --
// resetting it would free the very strings this message is built
// from.
const detail = try std.fmt.allocPrint(
reply.arena_alloc(),
"the pipeline's output could not be written to {s}.{s}: {s}",
.{ pending.db, pending.coll, @errorName(err) },
);
reply.pairs.clearRetainingCapacity();
try reply.put_error(@intFromEnum(ErrorCode.operation_failed), "OperationFailed", detail);
};
try ctx.engine.commit();
}
if (cmd.locks.coll == .exclusive) {
// Durability (seal + fsync) coalesces across concurrent writers. A
// commit error deliberately wins over the handler's captured `result`:
@@ -2275,6 +2330,39 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
in_trees = true;
start = 0;
end = trees.items.len;
} else if (std.mem.eql(u8, stage_name, "$out") or std.mem.eql(u8, stage_name, "$merge")) {
const is_out = std.mem.eql(u8, stage_name, "$out");
// MongoDB requires either to be last, and this server needs it too:
// the stage does not produce a stream for a later one to read.
if (!std.mem.eql(u8, stage_name, stages[stages.len - 1].doc[0].key)) {
const detail = try std.fmt.allocPrint(
reply.arena_alloc(),
"{s} can only be the final stage in the pipeline",
.{stage_name},
);
return reply.put_error(@intFromEnum(ErrorCode.location_write_stage_not_last), "Location40601", detail);
}
const target = (try write_stage_target(reply, db_name, stage[0].value, is_out)) orelse return;
const arena = reply.arena_alloc();
var out_docs: std.ArrayListUnmanaged(*const bson.Document) = .empty;
if (in_trees) {
for (trees.items[start..end]) |d| try out_docs.append(arena, d);
} else {
for (offs.items[start..end]) |off| try out_docs.append(arena, try doc_tree(arena, coll, off));
}
// Handed to the epilogue rather than written here: see
// `Context.pending_write`. The documents live in the reply's arena,
// which outlives it.
ctx.pending_write = .{
.db = target.db,
.coll = target.coll,
.docs = out_docs.items,
.mode = if (is_out) .replace else .merge,
};
// Both stages answer an empty cursor, as mongod does: the output
// went to a collection, not to the client.
try emit_first_batch(ctx, reply, db_name, coll_name, null, &.{}, batch_size);
return reply.put_ok();
} else if (std.mem.eql(u8, stage_name, "$count")) {
count_stage = switch (stage[0].value) {
.string => |s| s,
@@ -2465,6 +2553,98 @@ fn projected_tree(
return projected;
}
/// Where a `$out` or `$merge` writes, or null once the client has been told
/// why not.
///
/// `$out` takes a collection name or `{db, coll}`; `$merge` takes `into` in
/// either of those shapes. Everything past that -- `whenMatched`,
/// `whenNotMatched`, `on`, `let` -- selects behaviour this server does not
/// have, so it is refused rather than ignored: a `whenMatched: "fail"` that
/// silently merged would be the same lie Tier 0 spent three commits removing.
fn write_stage_target(
reply: *wire.Reply,
db_name: []const u8,
v: bson.Value,
is_out: bool,
) !?struct { db: []const u8, coll: []const u8 } {
var spec = v;
if (!is_out) {
const d = doc_arg(v) orelse {
// mongod's IDL parser answers for the whole stage document.
try reply.put_error(
@intFromEnum(ErrorCode.idl_failed_to_parse),
"IDLFailedToParse",
"BSON field '$merge.into' is missing but a required field",
);
return null;
};
for (d) |p| {
if (std.mem.eql(u8, p.key, "into")) continue;
const detail = try std.fmt.allocPrint(
reply.arena_alloc(),
"$merge does not support '{s}' on this server: only the default " ++
"whenMatched/whenNotMatched behaviour is implemented",
.{p.key},
);
try reply.put_error(@intFromEnum(ErrorCode.idl_unknown_field), "IDLUnknownField", detail);
return null;
}
spec = bson.get_pair(d, "into") orelse {
try reply.put_error(
@intFromEnum(ErrorCode.idl_failed_to_parse),
"IDLFailedToParse",
"BSON field '$merge.into' is missing but a required field",
);
return null;
};
}
switch (spec) {
.string => |name| return .{ .db = db_name, .coll = name },
.doc => |d| {
const coll = str_arg(bson.get_pair(d, "coll")) orelse {
try bad_value(reply, "the target of a write stage needs a coll");
return null;
};
return .{ .db = str_arg(bson.get_pair(d, "db")) orelse db_name, .coll = coll };
},
else => {
try bad_value(reply, "the target of a write stage must be a string or a document");
return null;
},
}
}
/// Apply what an aggregation's last stage asked for. The caller holds no lock.
///
/// Not atomic, and that has to be said out loud: mongod replaces an `$out`
/// target atomically, and this engine has no cross-collection atomicity and no
/// rename to build one out of. A crash between the drop and the last insert
/// leaves the target holding part of the new output where MongoDB would leave
/// the whole of the old. Recorded in `docs/M2_DESIGN_REVIEW.md` as the open
/// half of this decision rather than papered over: the shape that fixes it is
/// write-to-temp-and-rename, and rename is a command this server does not have.
fn apply_pending_write(ctx: *Context, pending: PendingWrite) !void {
try ctx.engine.lock();
defer ctx.engine.unlock();
if (pending.mode == .replace) {
// `$out` means "the target holds this and nothing else".
_ = ctx.engine.drop_collection(pending.db, pending.coll) catch |err| switch (err) {
error.NamespaceNotFound => {},
else => return err,
};
}
for (pending.docs) |d| {
// The pairs belong to the reply's arena; this Document is only a
// carrier, so its own arena is empty and frees nothing that matters.
var doc: bson.Document = .{ .arena = std.heap.ArenaAllocator.init(ctx.gpa), .pairs = d.pairs };
defer doc.arena.deinit();
// `$out` writes into a collection it has just emptied, so every write
// is an insert; `$merge`'s default pair is "replace the document with
// this `_id`, or insert it", which is what `replace` already means.
_ = try ctx.engine.replace(pending.db, pending.coll, &doc, ctx.oid_gen);
}
}
/// Where a pipeline stage reads its input.
///
/// A pipeline starts as slab offsets -- matched and reordered in place, never
@@ -4149,6 +4329,103 @@ test "aggregate $sort without a preceding $group sorts and frees correctly" {
}
}
test "$out and $merge write through the epilogue" {
// The write stages are the reason `Context.pending_write` exists. They
// write to a collection the pipeline is not reading, and doing that inside
// the handler would take a second collection lock while the first is held
// -- which `Collection.lock`'s own comment forbids. So the handler computes
// and the epilogue writes, with nothing held.
//
// Mutation check: apply the write inside the `$out` branch instead of
// stashing it. Deadlocks rather than fails, which is the argument for the
// epilogue in one line.
var threaded = std.Io.Threaded.init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var tdb = try TestDb.init(io);
defer tdb.deinit();
try dispatch_insert(&tdb, io, "src", &.{
.{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "x", .value = .{ .int32 = 1 } } } },
.{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 2 } }, .{ .key = "x", .value = .{ .int32 = 2 } } } },
});
try dispatch_insert(&tdb, io, "dst", &.{
.{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 9 } }, .{ .key = "old", .value = .{ .bool = true } } } },
});
var ctx = tdb.ctx(io);
const run = struct {
fn go(c: *Context, stages: []const bson.Value) !wire.Reply {
var reply = wire.Reply.init(testing.allocator);
errdefer reply.deinit();
var msg = try parse_fake_msg("aggregate", .{ .string = "src" }, &.{
.{ .key = "pipeline", .value = .{ .array = stages } },
.{ .key = "cursor", .value = .{ .doc = &.{} } },
});
defer msg.deinit();
try dispatch(c, &msg, &reply);
return reply;
}
}.go;
// $out replaces the target outright: the pre-existing document is gone.
{
const stages = [_]bson.Value{.{ .doc = &.{.{ .key = "$out", .value = .{ .string = "dst" } }} }};
var reply = try run(&ctx, &stages);
defer reply.deinit();
try testing.expectEqual(@as(f64, 1.0), bson.get_pair(reply.pairs.items, "ok").?.double);
}
{
const coll = ctx.engine.get_collection("test", "dst").?;
try testing.expectEqual(@as(u64, 2), coll.doc_count);
}
// $merge keeps what it does not name and replaces what it does.
try dispatch_insert(&tdb, io, "dst", &.{
.{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 9 } }, .{ .key = "old", .value = .{ .bool = true } } } },
});
{
const stages = [_]bson.Value{.{ .doc = &.{.{ .key = "$merge", .value = .{ .doc = &.{
.{ .key = "into", .value = .{ .string = "dst" } },
} } }} }};
var reply = try run(&ctx, &stages);
defer reply.deinit();
try testing.expectEqual(@as(f64, 1.0), bson.get_pair(reply.pairs.items, "ok").?.double);
}
{
const coll = ctx.engine.get_collection("test", "dst").?;
try testing.expectEqual(@as(u64, 3), coll.doc_count);
}
// And the three shapes that are refused rather than half-honoured.
const Case = struct { name: []const u8, stages: []const bson.Value, code: i32 };
const out_then_match = [_]bson.Value{
.{ .doc = &.{.{ .key = "$out", .value = .{ .string = "dst" } }} },
.{ .doc = &.{.{ .key = "$match", .value = .{ .doc = &.{} } }} },
};
const merge_bare = [_]bson.Value{.{ .doc = &.{.{ .key = "$merge", .value = .{ .doc = &.{} } }} }};
const merge_when = [_]bson.Value{.{ .doc = &.{.{ .key = "$merge", .value = .{ .doc = &.{
.{ .key = "into", .value = .{ .string = "dst" } },
.{ .key = "whenMatched", .value = .{ .string = "fail" } },
} } }} }};
const cases = [_]Case{
.{ .name = "$out is not last", .stages = &out_then_match, .code = 40601 },
.{ .name = "$merge without into", .stages = &merge_bare, .code = 40414 },
.{ .name = "$merge with whenMatched", .stages = &merge_when, .code = 40415 },
};
for (cases) |c| {
var reply = try run(&ctx, c.stages);
defer reply.deinit();
testing.expectEqual(@as(f64, 0.0), bson.get_pair(reply.pairs.items, "ok").?.double) catch |err| {
std.debug.print(" {s}: answered ok:1\n", .{c.name});
return err;
};
try testing.expectEqual(c.code, bson.get_pair(reply.pairs.items, "code").?.int32);
// And nothing was written: a refused stage leaves the target alone.
try testing.expectEqual(@as(u64, 3), ctx.engine.get_collection("test", "dst").?.doc_count);
}
}
test "$project is a stage, not a note about how to print the answer" {
// It used to set a variable applied once, at the emit. Three consequences,
// all measured on a live server before this changed: only the *last*

17
tests/spec/scorecard.txt
View File

@@ -18,16 +18,16 @@
# hasServerConnectionId, and maxTimeMS in an expected command (CSOT rewrites
# it -- the only assertion this runner declines to make).
total 194 pass 97 fail 196 skip 175 files 0 errored
total 201 pass 90 fail 196 skip 175 files 0 errored
# per-file: name pass fail skip
aggregate-allowdiskuse.json 3 0 0
aggregate-collation.json 0 1 0
aggregate-let.json 0 2 2
aggregate-merge-errorResponse.json 0 0 1
aggregate-merge.json 0 5 0
aggregate-merge.json 5 0 0
aggregate-out-readConcern.json 0 0 4
aggregate-out.json 0 2 0
aggregate-out.json 2 0 0
aggregate-rawdata.json 1 0 1
aggregate-write-readPreference.json 0 0 4
aggregate.json 5 0 2
@@ -202,16 +202,9 @@ aggregate-collation.json FAIL Aggregate with collation aggregate: expected 1 ele
aggregate-let.json SKIP Aggregate with let option needs server >= 5.0
aggregate-let.json FAIL Aggregate with let option unsupported (server-side error) aggregate: expected an error, the operation succeeded
aggregate-let.json SKIP Aggregate to collection with let option needs server >= 5.0
aggregate-let.json FAIL Aggregate to collection with let option unsupported (server-side error) aggregate: error message "Unrecognized pipeline stage name: '$out'" does not contain "unrecognized field 'let'"
aggregate-let.json FAIL Aggregate to collection with let option unsupported (server-side error) aggregate: expected an error, the operation succeeded
aggregate-merge-errorResponse.json SKIP aggregate $merge DuplicateKey error is accessible needs server >= 5.1
aggregate-merge.json FAIL Aggregate with $merge MongoServerError: Unrecognized pipeline stage name: '$merge'
aggregate-merge.json FAIL Aggregate with $merge and batch size of 0 MongoServerError: Unrecognized pipeline stage name: '$merge'
aggregate-merge.json FAIL Aggregate with $merge and majority readConcern MongoServerError: Unrecognized pipeline stage name: '$merge'
aggregate-merge.json FAIL Aggregate with $merge and local readConcern MongoServerError: Unrecognized pipeline stage name: '$merge'
aggregate-merge.json FAIL Aggregate with $merge and available readConcern MongoServerError: Unrecognized pipeline stage name: '$merge'
aggregate-out-readConcern.json SKIP * needs topology replicaset/sharded
aggregate-out.json FAIL Aggregate with $out MongoServerError: Unrecognized pipeline stage name: '$out'
aggregate-out.json FAIL Aggregate with $out and batch size of 0 MongoServerError: Unrecognized pipeline stage name: '$out'
aggregate-rawdata.json SKIP Aggregate with rawData option needs server >= 8.2.0
aggregate-write-readPreference.json SKIP * needs topology replicaset/sharded/load-balanced
aggregate.json SKIP aggregate with a document comment - pre 4.4 needs server <= 4.2.99
@@ -261,7 +254,7 @@ bulkWrite-updateOne-pipeline.json FAIL UpdateOne in bulk write using pipelines M
bulkWrite-updateOne-rawdata.json SKIP BulkWrite updateOne with rawData option needs server >= 8.2.0
bulkWrite-updateOne-rawdata.json FAIL BulkWrite updateOne with rawData option on less than 8.2.0 - ignore argument MongoBulkWriteError: update spec requires u
bulkWrite-updateOne-sort.json SKIP BulkWrite updateOne with sort option needs server >= 8.0
bypassDocumentValidation.json FAIL Aggregate with $out passes bypassDocumentValidation: false MongoServerError: Unrecognized pipeline stage name: '$out'
bypassDocumentValidation.json FAIL Aggregate with $out passes bypassDocumentValidation: false events client0[0].command.bypassDocumentValidation: missing from actual
bypassDocumentValidation.json FAIL BulkWrite passes bypassDocumentValidation: false events client0[0].command.bypassDocumentValidation: missing from actual
bypassDocumentValidation.json FAIL FindOneAndReplace passes bypassDocumentValidation: false events client0[0].command.bypassDocumentValidation: missing from actual
bypassDocumentValidation.json FAIL FindOneAndUpdate passes bypassDocumentValidation: false events client0[0].command.bypassDocumentValidation: missing from actual