commands: an update may be written as a pipeline

`u` (and `findAndModify`'s `update`) may be an array of aggregation stages
instead of a document of operators. The stages themselves already existed --
M2 built `compile_rewrite` and `apply_rewrite` for `aggregate` -- so this is
mostly about which of them are allowed here and what happens around them.

Six are: `$addFields`/`$set`, `$project`, `$unset`, `$replaceRoot`/
`$replaceWith`. Each rewrites one document into one document, which is the
property that makes them usable: `$match` could drop it, `$group` and
`$unwind` could change how many there are, and `$sort` means nothing to one.
mongod refuses those four by name with InvalidOptions (72) and a name that is
no stage at all with 40324, and the difference is worth keeping -- "not here"
and "not at all" are different things to be told.

`$replaceWith` did not exist here at all and now does, in `compile_rewrite`,
so `aggregate` gets it too: it is `$replaceRoot` with the expression in place
of the `{newRoot: ...}` wrapper, one stage under two spellings.

**The `_id` survives every stage.** `$replaceRoot: {newRoot: "$t"}` drops it
and `$project: {_id: 0}` asks to, and it comes back either way, because a
pipeline update rewrites a document rather than replacing one document with
another. A stage that sets it to a *different* value is ImmutableField (66);
restating the same one is fine. Measured on all six stages, and the mutation
that reddens the test is deleting the restore -- the document becomes
unfindable by the id it is stored under.

`EvalCtx.coll` becomes optional, which is what lets the upsert path run a
pipeline over a document held in memory before the collection it will be
inserted into exists. Only an `.offsets` stream reads the slab; a `.docs` one
never touched the field.

`arrayFilters` beside a pipeline is FailedToParse (9), not ignored: an
identifier a pipeline cannot spell would make the update silently different
from the one the client wrote.

pipeline.json 0/23 -> 23/23, operator corpus 125/125.
Pinned crud scorecard **218 -> 228 pass, 73 -> 63 fail**: the five pipeline
files, the four `-rawdata` files that use pipelines, and
`findOneAndUpdate-comment`, which is two more pipeline cases.
249/249 unit tests in ReleaseFast and ReleaseSafe, 83/83 fuzz.

src/commands.zig

@@ -74,6 +74,12 @@ pub const ErrorCode = enum(i32) {
     /// "Unrecognized pipeline stage name". A `Location` code, so mongod names it
     /// `Location40324` rather than after any symbol.
     location_unrecognized_stage = 40324,
+    /// `Location40323`, measured on mongod 8.3.7: an array element that packs
+    /// two stages into one document, where a pipeline stage is one field.
+    location_stage_needs_one_field = 40323,
+    /// `ImmutableField`, measured: what a pipeline-style update answers when a
+    /// stage leaves the document with a different `_id` than it started with.
+    immutable_field = 66,
     index_options_conflict = 85,
     cannot_create_index = 67,
     invalid_index_specification_option = 197,
@@ -1954,7 +1960,18 @@ fn cmd_update(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
     // by the dispatch epilogue once the collection lock is released.
     for (specs, 0..) |*spec, si| {
         const q = doc_arg(spec.get("q")) orelse return bad_value(reply, "update spec requires q");
-        const u_doc = doc_arg(spec.get("u")) orelse return bad_value(reply, "update spec requires u");
+        // `u` is a document of operators, a replacement document, or an array
+        // of aggregation stages. The third is a different machine from the
+        // first two and stays separate all the way down.
+        const u_value = spec.get("u") orelse return bad_value(reply, "update spec requires u");
+        const u_pipeline: ?[]const bson.Value = switch (u_value) {
+            .array => |a| a,
+            else => null,
+        };
+        const u_doc = if (u_pipeline == null)
+            doc_arg(u_value) orelse return bad_value(reply, "update spec requires u")
+        else
+            &.{};
         const multi = bool_arg(spec.get("multi")) orelse false;
         const upsert = bool_arg(spec.get("upsert")) orelse false;
         // `sort` on an update spec picks *which* match to write when the filter
@@ -1969,7 +1986,7 @@ fn cmd_update(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
         // A replacement describes one document, so there is no coherent meaning
         // for applying it to many: every match would end up identical apart from
         // its `_id`. MongoDB rejects the combination rather than doing that.
-        if (multi and update.is_replacement(u_doc)) {
+        if (u_pipeline == null and multi and update.is_replacement(u_doc)) {
             return failed_to_parse(reply, "multi update is not supported for replacement-style update");
         }
 
@@ -1988,7 +2005,13 @@ fn cmd_update(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
         // binds is refused whether or not it would have matched anything, and
         // an array filter the update never uses is refused even when the whole
         // command was a no-op. Both measured.
-        update.validate(u_doc, opts) catch |err| return update_refusal(reply, err, diag);
+        if (u_pipeline == null) {
+            update.validate(u_doc, opts) catch |err| return update_refusal(reply, err, diag);
+        } else if (opts.array_filters.len > 0) {
+            // Not "ignored": an identifier a pipeline cannot spell would be a
+            // silently different update from the one the client wrote.
+            return failed_to_parse(reply, "arrayFilters may not be specified for pipeline-style updates");
+        }
 
         var matched: std.ArrayListUnmanaged(u64) = .empty;
         defer matched.deinit(ctx.gpa);
@@ -1996,8 +2019,7 @@ fn cmd_update(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
 
         if (matched.items.len == 0) {
             if (upsert) {
-                const new_doc = build_upsert_doc(reply, ctx, q, u_doc, opts) catch |err|
+                const new_doc = (try build_upsert_doc(reply, ctx, q, u_doc, u_pipeline, opts, &diag)) orelse return;
-                    return update_refusal(reply, err, diag);
                 ctx.engine.insert(db_name, coll_name, new_doc, ctx.oid_gen) catch |err| switch (err) {
                     error.DuplicateKey, error.DuplicateKeyIndex => return duplicate_key_error(ctx, reply, db_name, coll_name, new_doc),
                     else => return err,
@@ -2018,9 +2040,14 @@ fn cmd_update(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
             // Work on a copy: the log write must precede any visible change,
             // and a rejected update must not corrupt the stored document.
             const doc = try doc_tree(reply.arena_alloc(), coll, off);
-            const copy = try clone_doc(reply, doc);
+            const copy = if (u_pipeline) |stages|
-            update.apply(copy, &.{ .arena = undefined, .pairs = u_doc }, opts) catch |err|
+                (try apply_update_pipeline(ctx, reply, stages, doc, coll)) orelse return
-                return update_refusal(reply, err, diag);
+            else blk: {
+                const c = try clone_doc(reply, doc);
+                update.apply(c, &.{ .arena = undefined, .pairs = u_doc }, opts) catch |err|
+                    return update_refusal(reply, err, diag);
+                break :blk c;
+            };
             const written = ctx.engine.replace(db_name, coll_name, copy, ctx.oid_gen) catch |err| switch (err) {
                 error.DuplicateKey, error.DuplicateKeyIndex => {
                     const e = try reply.arena_alloc().alloc(bson.Pair, 3);
@@ -2110,8 +2137,19 @@ fn cmd_find_and_modify(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !v
         .now_ms = std.Io.Timestamp.now(ctx.io, .real).toMilliseconds(),
         .diag = &diag,
     };
-    if (doc_arg(msg.body.get("update"))) |u| {
+    // The same three shapes `update`'s `u` takes, read once here.
-        update.validate(u, opts) catch |err| return update_refusal(reply, err, diag);
+    const u_pipeline: ?[]const bson.Value = switch (msg.body.get("update") orelse bson.Value.null) {
+        .array => |a| a,
+        else => null,
+    };
+    const u_doc: []const bson.Pair = if (u_pipeline != null) &.{} else doc_arg(msg.body.get("update")) orelse &.{};
+    if (u_pipeline == null) {
+        if (do_update and doc_arg(msg.body.get("update")) == null) {
+            return bad_value(reply, "update must be a document");
+        }
+        update.validate(u_doc, opts) catch |err| return update_refusal(reply, err, diag);
+    } else if (opts.array_filters.len > 0) {
+        return failed_to_parse(reply, "arrayFilters may not be specified for pipeline-style updates");
     }
 
     var matched: std.ArrayListUnmanaged(u64) = .empty;
@@ -2137,9 +2175,7 @@ fn cmd_find_and_modify(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !v
     var value: bson.Value = .null;
 
     if (target == null and do_update and upsert) {
-        const u_doc = doc_arg(msg.body.get("update")) orelse return bad_value(reply, "update must be a document");
+        const new_doc = (try build_upsert_doc(reply, ctx, q, u_doc, u_pipeline, opts, &diag)) orelse return;
-        const new_doc = build_upsert_doc(reply, ctx, q, u_doc, opts) catch |err|
-            return update_refusal(reply, err, diag);
         ctx.engine.insert(db_name, coll_name, new_doc, ctx.oid_gen) catch |err| switch (err) {
             error.DuplicateKey, error.DuplicateKeyIndex => return duplicate_key_error(ctx, reply, db_name, coll_name, new_doc),
             else => return err,
@@ -2153,11 +2189,15 @@ fn cmd_find_and_modify(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !v
         value = try project_doc(reply, target.?, proj_pairs);
         _ = try ctx.engine.remove_by_id(db_name, coll_name, target.?.get("_id") orelse unreachable);
     } else if (target != null and do_update) {
-        const u_doc = doc_arg(msg.body.get("update")) orelse return bad_value(reply, "update must be a document");
         const before = try bson.copy_pairs(arena, target.?.pairs);
-        const copy = try clone_doc(reply, target.?);
+        const copy = if (u_pipeline) |stages|
-        update.apply(copy, &.{ .arena = undefined, .pairs = u_doc }, opts) catch |err|
+            (try apply_update_pipeline(ctx, reply, stages, target.?, coll)) orelse return
-            return update_refusal(reply, err, diag);
+        else blk: {
+            const c = try clone_doc(reply, target.?);
+            update.apply(c, &.{ .arena = undefined, .pairs = u_doc }, opts) catch |err|
+                return update_refusal(reply, err, diag);
+            break :blk c;
+        };
         // findAndModify reports `n` (matched) and `updatedExisting`, neither of
         // which distinguishes a no-op, so whether it wrote is not needed here.
         _ = try ctx.engine.replace(db_name, coll_name, copy, ctx.oid_gen);
@@ -2487,6 +2527,7 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
             end = trees.items.len;
         } else if (std.mem.eql(u8, stage_name, "$addFields") or std.mem.eql(u8, stage_name, "$set") or
             std.mem.eql(u8, stage_name, "$unset") or std.mem.eql(u8, stage_name, "$replaceRoot") or
+            std.mem.eql(u8, stage_name, "$replaceWith") or
             std.mem.eql(u8, stage_name, "$unwind") or std.mem.eql(u8, stage_name, "$project"))
         {
             // The stages that rewrite a document, all one shape: read the
@@ -2841,13 +2882,17 @@ fn path_in_pairs(pairs: []const bson.Pair, path: []const u8) ?bson.Value {
 /// not build a million trees to read one field.
 fn stream_path(
     gpa: std.mem.Allocator,
-    coll: *const Collection,
+    /// Null when the stream is materialized. An `.offsets` stream is the only
+    /// form that needs one, because only it reads the slab -- which is what
+    /// lets a pipeline-style update evaluate expressions against a document
+    /// held in memory, before the collection it will be inserted into exists.
+    coll: ?*const Collection,
     src: Stream,
     i: usize,
     path: []const u8,
 ) !?bson.Value {
     return switch (src) {
-        .offsets => |o| try query_path_value_bytes(gpa, coll.doc_bytes(o[i]), path),
+        .offsets => |o| try query_path_value_bytes(gpa, coll.?.doc_bytes(o[i]), path),
         .docs => |d| path_in_pairs(d[i].pairs, path),
     };
 }
@@ -3232,6 +3277,13 @@ fn compile_rewrite(
         };
         return Rewrite{ .replace_root = (try compile_expr(reply, arena, new_root, "$replaceRoot's newRoot")) orelse return null };
     }
+    // `$replaceWith` is `$replaceRoot` with the expression in place of the
+    // `{newRoot: ...}` wrapper -- one stage under two spellings, like
+    // `$addFields` and `$set`. Reached from `aggregate` as well as from a
+    // pipeline-style update, since the compiler is shared.
+    if (std.mem.eql(u8, name, "$replaceWith")) {
+        return Rewrite{ .replace_root = (try compile_expr(reply, arena, spec, "$replaceWith's expression")) orelse return null };
+    }
     const d = doc_arg(spec) orelse {
         try bad_value(reply, "$addFields requires a document");
         return null;
@@ -3498,7 +3550,7 @@ fn report_eval_error(reply: *wire.Reply, err: EvalError) !void {
 /// Where an expression reads its document from.
 const EvalCtx = struct {
     arena: std.mem.Allocator,
-    coll: *const Collection,
+    coll: ?*const Collection,
     src: Stream,
     i: usize,
 };
@@ -4426,6 +4478,117 @@ fn wrong_type_dynamic(
     return reply.put_error(@intFromEnum(ErrorCode.type_mismatch), "TypeMismatch", text);
 }
 
+/// The stages an update may be written out of. Every one rewrites a single
+/// document into a single document, which is what makes them usable here:
+/// `$match` could drop it, `$group` and `$unwind` could change how many there
+/// are, and `$sort` means nothing to one. mongod refuses those four by name
+/// rather than by shape, and so does this.
+const update_pipeline_stages = [_][]const u8{
+    "$addFields", "$set", "$project", "$unset", "$replaceRoot", "$replaceWith",
+};
+
+/// Apply a pipeline-style update to one document, returning the new one.
+///
+/// Returns null having written the error reply, like the other `*_arg`
+/// helpers. `coll` may be null: the stages are fed a materialized document, so
+/// nothing reads the slab, and the upsert path has no collection yet.
+fn apply_update_pipeline(
+    ctx: *Context,
+    reply: *wire.Reply,
+    stages: []const bson.Value,
+    doc: *const bson.Document,
+    coll: ?*const Collection,
+) !?*const bson.Document {
+    const arena = reply.arena_alloc();
+    // The `_id` the document came in with. Every stage may drop it -- a
+    // `$replaceRoot` almost always does -- and it comes back afterwards,
+    // because a pipeline update rewrites a document rather than replacing one
+    // document with another. Measured on all six stages.
+    const original_id = doc.get("_id");
+    var current = doc;
+    for (stages) |stage| {
+        const spec = doc_arg(stage) orelse {
+            try failed_to_parse(reply, "each element of a pipeline update must be a document");
+            return null;
+        };
+        if (spec.len != 1) {
+            try reply.put_error(
+                @intFromEnum(ErrorCode.location_stage_needs_one_field),
+                "Location40323",
+                "A pipeline stage specification object must contain exactly one field.",
+            );
+            return null;
+        }
+        const name = spec[0].key;
+        if (!contains_name(&update_pipeline_stages, name)) {
+            if (is_known_pipeline_stage(name)) {
+                try reply.put_error(
+                    @intFromEnum(ErrorCode.invalid_options),
+                    "InvalidOptions",
+                    try std.fmt.allocPrint(arena, "{s} is not allowed to be used within an update", .{name}),
+                );
+            } else {
+                try reply.put_error(
+                    @intFromEnum(ErrorCode.location_unrecognized_stage),
+                    "Location40324",
+                    try std.fmt.allocPrint(arena, "Unrecognized pipeline stage name: '{s}'", .{name}),
+                );
+            }
+            return null;
+        }
+        const rewrite = try compile_rewrite(ctx, reply, arena, name, spec[0].value) orelse return null;
+        var built: std.ArrayListUnmanaged(*const bson.Document) = .empty;
+        defer built.deinit(ctx.gpa);
+        const ec: EvalCtx = .{ .arena = arena, .coll = coll, .src = .{ .docs = (&current)[0..1] }, .i = 0 };
+        apply_rewrite(ec, rewrite, current, &built, ctx.gpa) catch |err| {
+            try report_eval_error(reply, err);
+            return null;
+        };
+        // Every stage here is 1->1, so this holds by construction; it is an
+        // assertion rather than a branch because a stage that broke it would
+        // silently drop or duplicate the document being updated.
+        assert(built.items.len == 1);
+        current = built.items[0];
+    }
+
+    if (original_id) |id| {
+        if (current.get("_id")) |now| {
+            if (bson.compare(id, now) != .eq) {
+                try reply.put_error(
+                    @intFromEnum(ErrorCode.immutable_field),
+                    "ImmutableField",
+                    "After applying the update, the (immutable) field '_id' was found to have been altered",
+                );
+                return null;
+            }
+        } else {
+            const with_id = try arena.alloc(bson.Pair, current.pairs.len + 1);
+            with_id[0] = .{ .key = "_id", .value = id };
+            @memcpy(with_id[1..], current.pairs);
+            const owned = try arena.create(bson.Document);
+            owned.* = .{ .arena = std.heap.ArenaAllocator.init(arena), .pairs = with_id };
+            current = owned;
+        }
+    }
+    return current;
+}
+
+fn contains_name(names: []const []const u8, name: []const u8) bool {
+    for (names) |n| if (std.mem.eql(u8, n, name)) return true;
+    return false;
+}
+
+/// Whether a name is a stage this server knows anywhere, which is the whole of
+/// the difference between "not here" (72) and "not at all" (40324).
+fn is_known_pipeline_stage(name: []const u8) bool {
+    const known = [_][]const u8{
+        "$match",  "$group", "$sort",  "$limit", "$skip",
+        "$unwind", "$count", "$out",   "$merge", "$lookup",
+        "$facet",  "$sample", "$sortByCount", "$documents",
+    };
+    return contains_name(&known, name);
+}
+
 /// Build the document for an upsert: equality fields from the filter, then
 /// the update operators applied. Owned by the reply arena.
 ///
@@ -4441,8 +4604,10 @@ fn build_upsert_doc(
     ctx: *Context,
     q: []const bson.Pair,
     u_doc: []const bson.Pair,
+    u_pipeline: ?[]const bson.Value,
     opts: update.Options,
-) !*bson.Document {
+    diag: *update.Diagnostic,
+) !?*const bson.Document {
     const arena = reply.arena_alloc();
     var pairs: std.ArrayListUnmanaged(bson.Pair) = .empty;
     defer pairs.deinit(arena);
@@ -4458,20 +4623,34 @@ fn build_upsert_doc(
     // Apply update operators to build the final doc; _id handled by insert.
     // `inserting` is what `$setOnInsert` asks about, and this is the only
     // caller that answers yes.
-    var insert_opts = opts;
+    // A pipeline runs over the document the filter implies, exactly as it
-    insert_opts.inserting = true;
+    // would over a stored one; operators get the `inserting` bit, which is
-    try update.apply(owned, &.{ .arena = undefined, .pairs = u_doc }, insert_opts);
+    // what `$setOnInsert` asks about and this is the only caller that answers
-    // After the operators, because `$setOnInsert` may supply the `_id` itself
+    // yes.
-    // and a generated one would then be the wrong answer. At the front,
+    var built: *const bson.Document = owned;
-    // because that is where MongoDB stores it and where the `_id_` index
+    if (u_pipeline) |stages| {
-    // descends on it.
+        built = (try apply_update_pipeline(ctx, reply, stages, owned, null)) orelse return null;
-    if (owned.get("_id") == null) {
+    } else {
-        const with_id = try arena.alloc(bson.Pair, owned.pairs.len + 1);
+        var insert_opts = opts;
-        with_id[0] = .{ .key = "_id", .value = .{ .object_id = ctx.oid_gen.new(ctx.io) } };
+        insert_opts.inserting = true;
-        @memcpy(with_id[1..], owned.pairs);
+        update.apply(owned, &.{ .arena = undefined, .pairs = u_doc }, insert_opts) catch |err| {
-        owned.pairs = with_id;
+            try update_refusal(reply, err, diag.*);
+            return null;
+        };
     }
-    return owned;
+    // After the update, because `$setOnInsert` may supply the `_id` itself and
+    // a generated one would then be the wrong answer. At the front, because
+    // that is where MongoDB stores it and where the `_id_` index descends on
+    // it.
+    if (built.get("_id") == null) {
+        const with_id = try arena.alloc(bson.Pair, built.pairs.len + 1);
+        with_id[0] = .{ .key = "_id", .value = .{ .object_id = ctx.oid_gen.new(ctx.io) } };
+        @memcpy(with_id[1..], built.pairs);
+        const owned_id = try arena.create(bson.Document);
+        owned_id.* = .{ .arena = std.heap.ArenaAllocator.init(arena), .pairs = with_id };
+        built = owned_id;
+    }
+    return built;
 }
 
 fn parse_sort_keys(reply: *wire.Reply, value: ?bson.Value) ![]const query.SortKey {
@@ -7155,3 +7334,143 @@ test "an upsert reports the _id it generated" {
     const upserted2 = bson.get_pair(reply2.pairs.items, "upserted").?.array;
     try testing.expectEqual(@as(i32, 9), bson.get_pair(upserted2[0].doc, "_id").?.int32);
 }
+
+test "a pipeline-style update keeps the _id its stages dropped" {
+    // `$replaceRoot` almost always drops the `_id`, and a pipeline update
+    // rewrites a document rather than replacing one document with another --
+    // so the `_id` comes back. Mutation check: delete the restore block in
+    // `apply_update_pipeline` and the document becomes unfindable by the id
+    // it was stored under.
+    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, "pl", &.{
+        .{ .doc = &.{
+            .{ .key = "_id", .value = .{ .int32 = 1 } },
+            .{ .key = "x", .value = .{ .int32 = 1 } },
+            .{ .key = "t", .value = .{ .doc = &.{.{ .key = "u", .value = .{ .int32 = 7 } }} } },
+        } },
+    });
+
+    const stages = [_]bson.Value{
+        .{ .doc = &.{.{ .key = "$replaceRoot", .value = .{ .doc = &.{
+            .{ .key = "newRoot", .value = .{ .string = "$t" } },
+        } } }} },
+        .{ .doc = &.{.{ .key = "$addFields", .value = .{ .doc = &.{
+            .{ .key = "foo", .value = .{ .int32 = 1 } },
+        } } }} },
+    };
+    const updates = [_]bson.Value{.{ .doc = &.{
+        .{ .key = "q", .value = .{ .doc = &.{} } },
+        .{ .key = "u", .value = .{ .array = &stages } },
+    } }};
+    try testing.expectEqual(@as(?i32, null), try run_for_code(&ctx, "update", .{ .string = "pl" }, &.{
+        .{ .key = "updates", .value = .{ .array = &updates } },
+    }));
+
+    var reply = wire.Reply.init(testing.allocator);
+    defer reply.deinit();
+    const ids = try distinct_values(&tdb, io, &reply, "pl", &.{
+        .{ .key = "key", .value = .{ .string = "_id" } },
+    });
+    try testing.expectEqual(@as(usize, 1), ids.len);
+    try testing.expectEqual(@as(i32, 1), ids[0].int32);
+    // And the stages did run, in order: `u` came from `$t`, `foo` from the
+    // stage after it. A second reply, because `put` appends and a reused one
+    // would answer with the first call's `values`.
+    var reply2 = wire.Reply.init(testing.allocator);
+    defer reply2.deinit();
+    const us = try distinct_values(&tdb, io, &reply2, "pl", &.{
+        .{ .key = "key", .value = .{ .string = "u" } },
+    });
+    try testing.expectEqual(@as(i32, 7), us[0].int32);
+}
+
+test "a pipeline-style update refuses what it cannot be written out of" {
+    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, "plr", &.{
+        .{ .doc = &.{
+            .{ .key = "_id", .value = .{ .int32 = 1 } },
+            .{ .key = "x", .value = .{ .int32 = 1 } },
+        } },
+    });
+
+    // A real stage refused here (72) reads differently from a name that is no
+    // stage at all (40324), and mongod distinguishes them -- so this does.
+    const cases = [_]struct { stage: bson.Pair, code: i32 }{
+        .{ .stage = .{ .key = "$match", .value = .{ .doc = &.{} } }, .code = 72 },
+        .{ .stage = .{ .key = "$group", .value = .{ .doc = &.{} } }, .code = 72 },
+        .{ .stage = .{ .key = "$unwind", .value = .{ .string = "$x" } }, .code = 72 },
+        .{ .stage = .{ .key = "$bogus", .value = .{ .doc = &.{} } }, .code = 40324 },
+    };
+    for (cases) |c| {
+        const one = [_]bson.Pair{c.stage};
+        const stages = [_]bson.Value{.{ .doc = &one }};
+        const updates = [_]bson.Value{.{ .doc = &.{
+            .{ .key = "q", .value = .{ .doc = &.{} } },
+            .{ .key = "u", .value = .{ .array = &stages } },
+        } }};
+        try testing.expectEqual(@as(?i32, c.code), try run_for_code(&ctx, "update", .{ .string = "plr" }, &.{
+            .{ .key = "updates", .value = .{ .array = &updates } },
+        }));
+    }
+
+    // Two stages in one element, and a stage that changes the `_id`.
+    const packed_stage = [_]bson.Value{.{ .doc = &.{
+        .{ .key = "$addFields", .value = .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} } },
+        .{ .key = "$unset", .value = .{ .string = "x" } },
+    } }};
+    const packed_updates = [_]bson.Value{.{ .doc = &.{
+        .{ .key = "q", .value = .{ .doc = &.{} } },
+        .{ .key = "u", .value = .{ .array = &packed_stage } },
+    } }};
+    try testing.expectEqual(@as(?i32, 40323), try run_for_code(&ctx, "update", .{ .string = "plr" }, &.{
+        .{ .key = "updates", .value = .{ .array = &packed_updates } },
+    }));
+
+    const reid = [_]bson.Value{.{ .doc = &.{.{ .key = "$addFields", .value = .{ .doc = &.{
+        .{ .key = "_id", .value = .{ .int32 = 9 } },
+    } } }} }};
+    const reid_updates = [_]bson.Value{.{ .doc = &.{
+        .{ .key = "q", .value = .{ .doc = &.{} } },
+        .{ .key = "u", .value = .{ .array = &reid } },
+    } }};
+    try testing.expectEqual(@as(?i32, 66), try run_for_code(&ctx, "update", .{ .string = "plr" }, &.{
+        .{ .key = "updates", .value = .{ .array = &reid_updates } },
+    }));
+
+    // `arrayFilters` has nothing to bind to in a pipeline, and is refused
+    // rather than ignored.
+    const ok_stage = [_]bson.Value{.{ .doc = &.{.{ .key = "$addFields", .value = .{ .doc = &.{
+        .{ .key = "a", .value = .{ .int32 = 1 } },
+    } } }} }};
+    const af_updates = [_]bson.Value{.{ .doc = &.{
+        .{ .key = "q", .value = .{ .doc = &.{} } },
+        .{ .key = "u", .value = .{ .array = &ok_stage } },
+        .{ .key = "arrayFilters", .value = .{ .array = &.{.{ .doc = &.{
+            .{ .key = "i.b", .value = .{ .int32 = 1 } },
+        } }} } },
+    } }};
+    try testing.expectEqual(@as(?i32, 9), try run_for_code(&ctx, "update", .{ .string = "plr" }, &.{
+        .{ .key = "updates", .value = .{ .array = &af_updates } },
+    }));
+
+    // Nothing above wrote: the document is the one that was inserted.
+    var reply = wire.Reply.init(testing.allocator);
+    defer reply.deinit();
+    const xs = try distinct_values(&tdb, io, &reply, "plr", &.{
+        .{ .key = "key", .value = .{ .string = "x" } },
+    });
+    try testing.expectEqual(@as(usize, 1), xs.len);
+    try testing.expectEqual(@as(i32, 1), xs[0].int32);
+}

tests/spec/scorecard.txt

@@ -18,7 +18,7 @@
 # hasServerConnectionId, and maxTimeMS in an expected command (CSOT rewrites
 # it -- the only assertion this runner declines to make).
 
-total	218 pass	73 fail	196 skip	175 files	0 errored
+total	228 pass	63 fail	196 skip	175 files	0 errored
 
 # per-file: name	pass	fail	skip
 aggregate-allowdiskuse.json	3	0	0
@@ -54,13 +54,13 @@ bulkWrite-update-validation.json	3	0	0
 bulkWrite-updateMany-dots_and_dollars.json	0	0	4
 bulkWrite-updateMany-hint-unacknowledged.json	2	0	0
 bulkWrite-updateMany-let.json	0	1	1
-bulkWrite-updateMany-pipeline.json	0	1	0
+bulkWrite-updateMany-pipeline.json	1	0	0
-bulkWrite-updateMany-rawdata.json	0	1	1
+bulkWrite-updateMany-rawdata.json	1	0	1
 bulkWrite-updateOne-dots_and_dollars.json	0	0	4
 bulkWrite-updateOne-hint-unacknowledged.json	2	0	0
 bulkWrite-updateOne-let.json	0	1	1
-bulkWrite-updateOne-pipeline.json	0	1	0
+bulkWrite-updateOne-pipeline.json	1	0	0
-bulkWrite-updateOne-rawdata.json	0	1	1
+bulkWrite-updateOne-rawdata.json	1	0	1
 bulkWrite-updateOne-sort.json	1	0	1
 bulkWrite.json	10	0	0
 bypassDocumentValidation.json	4	5	0
@@ -143,15 +143,15 @@ findOneAndReplace-upsert.json	2	2	0
 findOneAndReplace.json	4	2	0
 findOneAndUpdate-arrayFilters.json	3	0	0
 findOneAndUpdate-collation.json	0	1	0
-findOneAndUpdate-comment.json	0	2	1
+findOneAndUpdate-comment.json	2	0	1
 findOneAndUpdate-dots_and_dollars.json	0	0	4
 findOneAndUpdate-errorResponse.json	0	1	1
 findOneAndUpdate-hint-serverError.json	0	0	2
 findOneAndUpdate-hint-unacknowledged.json	2	0	2
 findOneAndUpdate-hint.json	2	0	0
 findOneAndUpdate-let.json	0	1	1
-findOneAndUpdate-pipeline.json	0	1	0
+findOneAndUpdate-pipeline.json	1	0	0
-findOneAndUpdate-rawdata.json	0	1	1
+findOneAndUpdate-rawdata.json	1	0	1
 findOneAndUpdate.json	5	3	0
 insertMany-comment.json	2	0	1
 insertMany-dots_and_dollars.json	3	1	1
@@ -179,7 +179,7 @@ updateMany-dots_and_dollars.json	0	0	4
 updateMany-hint-unacknowledged.json	2	0	0
 updateMany-hint.json	2	0	0
 updateMany-let.json	0	1	1
-updateMany-pipeline.json	0	1	0
+updateMany-pipeline.json	1	0	0
 updateMany-rawdata.json	1	0	1
 updateMany-validation.json	1	0	0
 updateMany.json	4	0	0
@@ -191,7 +191,7 @@ updateOne-errorResponse.json	0	0	1
 updateOne-hint-unacknowledged.json	2	0	0
 updateOne-hint.json	2	0	0
 updateOne-let.json	0	1	1
-updateOne-pipeline.json	0	1	0
+updateOne-pipeline.json	1	0	0
 updateOne-rawdata.json	1	0	1
 updateOne-sort.json	1	0	1
 updateOne-validation.json	1	0	0
@@ -237,19 +237,15 @@ bulkWrite-updateMany-dots_and_dollars.json	SKIP	Updating document to set top-lev
 bulkWrite-updateMany-dots_and_dollars.json	SKIP	Updating document to set dollar-prefixed key in embedded doc on 5.0+ server	needs server >= 5.0
 bulkWrite-updateMany-dots_and_dollars.json	SKIP	Updating document to set dotted key in embedded doc on 5.0+ server	needs server >= 5.0
 bulkWrite-updateMany-let.json	SKIP	BulkWrite updateMany with let option	needs server >= 5.0
-bulkWrite-updateMany-let.json	FAIL	BulkWrite updateMany with let option unsupported (server-side error)	bulkWrite: error message "update spec requires u" does not contain "'update.let' is an unknown field"
+bulkWrite-updateMany-let.json	FAIL	BulkWrite updateMany with let option unsupported (server-side error)	bulkWrite: expected an error, the operation succeeded
-bulkWrite-updateMany-pipeline.json	FAIL	UpdateMany in bulk write using pipelines	MongoBulkWriteError: update spec requires u
 bulkWrite-updateMany-rawdata.json	SKIP	BulkWrite updateMany with rawData option	needs server >= 8.2.0
-bulkWrite-updateMany-rawdata.json	FAIL	BulkWrite updateMany with rawData option on less than 8.2.0 - ignore argument	MongoBulkWriteError: update spec requires u
 bulkWrite-updateOne-dots_and_dollars.json	SKIP	Updating document to set top-level dollar-prefixed key on 5.0+ server	needs server >= 5.0
 bulkWrite-updateOne-dots_and_dollars.json	SKIP	Updating document to set top-level dotted key on 5.0+ server	needs server >= 5.0
 bulkWrite-updateOne-dots_and_dollars.json	SKIP	Updating document to set dollar-prefixed key in embedded doc on 5.0+ server	needs server >= 5.0
 bulkWrite-updateOne-dots_and_dollars.json	SKIP	Updating document to set dotted key in embedded doc on 5.0+ server	needs server >= 5.0
 bulkWrite-updateOne-let.json	SKIP	BulkWrite updateOne with let option	needs server >= 5.0
-bulkWrite-updateOne-let.json	FAIL	BulkWrite updateOne with let option unsupported (server-side error)	bulkWrite: error message "update spec requires u" does not contain "'update.let' is an unknown field"
+bulkWrite-updateOne-let.json	FAIL	BulkWrite updateOne with let option unsupported (server-side error)	bulkWrite: expected an error, the operation succeeded
-bulkWrite-updateOne-pipeline.json	FAIL	UpdateOne in bulk write using pipelines	MongoBulkWriteError: update spec requires u
 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	events client0[0].command.bypassDocumentValidation: missing from actual
 bypassDocumentValidation.json	FAIL	BulkWrite passes bypassDocumentValidation: false	events client0[0].command.bypassDocumentValidation: missing from actual
@@ -359,8 +355,6 @@ findOneAndReplace-upsert.json	FAIL	FindOneAndReplace when no documents match wit
 findOneAndReplace.json	FAIL	FindOneAndReplace when many documents match returning the document after modification	findOneAndReplace.x: expected 32, got 22
 findOneAndReplace.json	FAIL	FindOneAndReplace when one document matches returning the document after modification	findOneAndReplace.x: expected 32, got 22
 findOneAndUpdate-collation.json	FAIL	FindOneAndUpdate when many documents match with collation returning the document before modification	findOneAndUpdate: expected a document, got null
-findOneAndUpdate-comment.json	FAIL	findOneAndUpdate with string comment	MongoServerError: update must be a document
-findOneAndUpdate-comment.json	FAIL	findOneAndUpdate with document comment	MongoServerError: update must be a document
 findOneAndUpdate-comment.json	SKIP	findOneAndUpdate with comment - pre 4.4	needs server <= 4.2.99
 findOneAndUpdate-dots_and_dollars.json	SKIP	Updating document to set top-level dollar-prefixed key on 5.0+ server	needs server >= 5.0
 findOneAndUpdate-dots_and_dollars.json	SKIP	Updating document to set top-level dotted key on 5.0+ server	needs server >= 5.0
@@ -373,9 +367,7 @@ findOneAndUpdate-hint-unacknowledged.json	SKIP	Unacknowledged findOneAndUpdate w
 findOneAndUpdate-hint-unacknowledged.json	SKIP	Unacknowledged findOneAndUpdate with hint document fails with client-side error on pre-4.4 server	needs server <= 4.2.99
 findOneAndUpdate-let.json	SKIP	findOneAndUpdate with let option	needs server >= 5.0
 findOneAndUpdate-let.json	FAIL	findOneAndUpdate with let option unsupported (server-side error)	findOneAndUpdate: expected an error, the operation succeeded
-findOneAndUpdate-pipeline.json	FAIL	FindOneAndUpdate using pipelines	MongoServerError: update must be a document
 findOneAndUpdate-rawdata.json	SKIP	findOneAndUpdate with rawData option	needs server >= 8.2.0
-findOneAndUpdate-rawdata.json	FAIL	findOneAndUpdate with rawData option on less than 8.2.0 - ignore argument	MongoServerError: update must be a document
 findOneAndUpdate.json	FAIL	FindOneAndUpdate when many documents match returning the document after modification	findOneAndUpdate.x: expected 23, got 22
 findOneAndUpdate.json	FAIL	FindOneAndUpdate when one document matches returning the document after modification	findOneAndUpdate.x: expected 23, got 22
 findOneAndUpdate.json	FAIL	FindOneAndUpdate when no documents match with upsert returning the document after modification	findOneAndUpdate: expected a document, got null
@@ -404,8 +396,7 @@ updateMany-dots_and_dollars.json	SKIP	Updating document to set top-level dotted
 updateMany-dots_and_dollars.json	SKIP	Updating document to set dollar-prefixed key in embedded doc on 5.0+ server	needs server >= 5.0
 updateMany-dots_and_dollars.json	SKIP	Updating document to set dotted key in embedded doc on 5.0+ server	needs server >= 5.0
 updateMany-let.json	SKIP	updateMany with let option	needs server >= 5.0
-updateMany-let.json	FAIL	updateMany with let option unsupported (server-side error)	updateMany: error message "update spec requires u" does not contain "'update.let' is an unknown field"
+updateMany-let.json	FAIL	updateMany with let option unsupported (server-side error)	updateMany: expected an error, the operation succeeded
-updateMany-pipeline.json	FAIL	UpdateMany using pipelines	MongoServerError: update spec requires u
 updateMany-rawdata.json	SKIP	updateMany with rawData option	needs server >= 8.2.0
 updateOne-collation.json	FAIL	UpdateOne when one document matches with collation	updateOne.matchedCount: expected 1, got 0
 updateOne-comment.json	SKIP	UpdateOne with comment - pre 4.4	needs server <= 4.2.99
@@ -415,7 +406,6 @@ updateOne-dots_and_dollars.json	SKIP	Updating document to set dollar-prefixed ke
 updateOne-dots_and_dollars.json	SKIP	Updating document to set dotted key in embedded doc on 5.0+ server	needs server >= 5.0
 updateOne-errorResponse.json	SKIP	update operations support errorResponse assertions	runner: failPoint
 updateOne-let.json	SKIP	UpdateOne with let option	needs server >= 5.0
-updateOne-let.json	FAIL	UpdateOne with let option unsupported (server-side error)	updateOne: error message "update spec requires u" does not contain "'update.let' is an unknown field"
+updateOne-let.json	FAIL	UpdateOne with let option unsupported (server-side error)	updateOne: expected an error, the operation succeeded
-updateOne-pipeline.json	FAIL	UpdateOne using pipelines	MongoServerError: update spec requires u
 updateOne-rawdata.json	SKIP	UpdateOne with rawData option	needs server >= 8.2.0
 updateOne-sort.json	SKIP	UpdateOne with sort option	needs server >= 8.0