From 8ebeb9d4ecc270414581b795c0f9500536e4c7d4 Mon Sep 17 00:00:00 2001 From: "A.Shakhmatov" Date: Sun, 9 Aug 2026 21:02:01 +0300 Subject: [PATCH] commands: $out and $merge, written by the dispatch epilogue 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. --- docs/M2_DESIGN_REVIEW.md | 18 ++- src/commands.zig | 277 +++++++++++++++++++++++++++++++++++++++ tests/spec/scorecard.txt | 17 +-- 3 files changed, 299 insertions(+), 13 deletions(-) diff --git a/docs/M2_DESIGN_REVIEW.md b/docs/M2_DESIGN_REVIEW.md index 8f1f87f..9be3cca 100644 --- a/docs/M2_DESIGN_REVIEW.md +++ b/docs/M2_DESIGN_REVIEW.md @@ -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. diff --git a/src/commands.zig b/src/commands.zig index b95efe0..225470e 100644 --- a/src/commands.zig +++ b/src/commands.zig @@ -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* diff --git a/tests/spec/scorecard.txt b/tests/spec/scorecard.txt index e690acb..e73dc88 100644 --- a/tests/spec/scorecard.txt +++ b/tests/spec/scorecard.txt @@ -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