update: $push's modifiers are modifiers

`$slice`, `$position` and `$sort` were parsed, accepted and dropped.
`{$each: [3, 4], $slice: -3}` appended both values, sliced nothing and
answered ok: 1 with modifiedCount: 1 -- the same class of wrong answer the
positional operators were, and the reason `tests/spec/operators/` exists
rather than a list of TODOs.

Measured, and the order is the whole of it: insert at `$position`, then
`$sort` the array *including* the new elements, then `$slice` the result.

  - `$slice: n` keeps the first n, `$slice: -n` the **last** n. That half is
    what a capped log depends on and is the one easy to write backwards; the
    test mutates exactly it.
  - `$position` counts back from the end when negative, and clamps at the
    front rather than wrapping.
  - `$sort: 1` orders whole elements in BSON order; `$sort: {a: 1}` orders on
    a field of them, and an element without it sorts as null -- the rank a
    missing field has everywhere else here.
  - **without `$each` there are no modifiers at all**: `{$push: {t: {$slice:
    1}}}` pushes the document `{$slice: 1}` as a value. That is what makes
    `$each` the flag rather than a member of the set, and it is measured, not
    reasoned.

An unknown `$`-prefixed key beside `$each`, or a `$slice`/`$position` that is
not a number, is BadValue -- refused rather than ignored, which is the point.

push-modifiers.json 6/21 -> 21/21. 240/240 unit tests.

src/commands.zig

@@ -4308,6 +4308,11 @@ fn update_refusal(reply: *wire.Reply, err: anyerror, diag: update.Diagnostic) !v
             "TypeMismatch",
             "The argument to $each in $addToSet must be an array",
         ) else return bad_value(reply, "The argument to $each in $push must be an array"),
+        error.BadPushModifier => return bad_value(reply, try std.fmt.allocPrint(
+            arena,
+            "Unrecognized or invalid $push modifier '{s}' at field '{s}'",
+            .{ diag.segment, diag.path },
+        )),
         error.PathNotViable => return reply.put_error(
             @intFromEnum(ErrorCode.path_not_viable),
             "PathNotViable",

src/update.zig

@@ -34,6 +34,10 @@ pub const UpdateError = error{
     PullAllNeedsArray,
     /// `$each` that is not an array.
     BadEach,
+    /// A `$push` modifier that is unknown, or whose operand is not what it
+    /// takes: `$slice`/`$position` want a number, `$sort` a number or a
+    /// one-field document.
+    BadPushModifier,
     /// A path segment that is not a field to create: a non-numeric name
     /// applied to an array, or any name applied to a scalar element a
     /// positional segment selected. mongod's `PathNotViable`.
@@ -729,6 +733,62 @@ fn op_extremum(
     }
 }
 
+/// `$push`'s modifiers, in the order mongod applies them.
+///
+/// Absent `$each` there are no modifiers at all: `{$push: {t: {$slice: 1}}}`
+/// pushes the document `{$slice: 1}` as a value. Measured, and it is what
+/// makes `$each` the flag rather than a member of the set.
+const PushModifiers = struct {
+    each: []const bson.Value,
+    /// Where the new elements go. Negative counts back from the end.
+    position: ?i64 = null,
+    /// Ascending/descending over whole elements, or a document naming a field
+    /// of them.
+    sort: ?bson.Value = null,
+    /// Keep the first n, or with a negative n the last -n. Applied last, and
+    /// on its own it truncates without adding anything.
+    slice: ?i64 = null,
+};
+
+fn parse_push_modifiers(spec: []const bson.Pair, opts: Options, path: []const u8) UpdateError!PushModifiers {
+    var m = PushModifiers{ .each = &.{} };
+    for (spec) |p| {
+        if (std.mem.eql(u8, p.key, "$each")) {
+            m.each = switch (p.value) {
+                .array => |a| a,
+                else => {
+                    note(opts.diag, path, "$push");
+                    return error.BadEach;
+                },
+            };
+            continue;
+        }
+        const numeric: ?i64 = if (p.value.is_number()) @intFromFloat(@trunc(p.value.as_f128())) else null;
+        if (std.mem.eql(u8, p.key, "$position")) {
+            m.position = numeric orelse return bad_modifier(opts, path, "$position");
+            continue;
+        }
+        if (std.mem.eql(u8, p.key, "$slice")) {
+            m.slice = numeric orelse return bad_modifier(opts, path, "$slice");
+            continue;
+        }
+        if (std.mem.eql(u8, p.key, "$sort")) {
+            m.sort = p.value;
+            continue;
+        }
+        // An unknown `$`-prefixed key beside `$each` is a typo, not a field of
+        // a document being pushed: the document is `$each`'s elements, not
+        // this one.
+        return bad_modifier(opts, path, p.key);
+    }
+    return m;
+}
+
+fn bad_modifier(opts: Options, path: []const u8, name: []const u8) UpdateError {
+    note(opts.diag, path, name);
+    return error.BadPushModifier;
+}
+
 fn op_push(
     arena: std.mem.Allocator,
     pairs: *std.ArrayListUnmanaged(bson.Pair),
@@ -736,34 +796,110 @@ fn op_push(
     opts: Options,
 ) UpdateError!void {
     for (ops) |p| {
+        const mods: ?PushModifiers = if (each_of(p.value) != null)
+            try parse_push_modifiers(p.value.doc, opts, p.key)
+        else
+            null;
         for (try resolve(arena, pairs.items, p.key, opts)) |segs| {
-            const current_opt = get_value(pairs.items, segs);
             var items: std.ArrayListUnmanaged(bson.Value) = .empty;
             defer items.deinit(arena);
-            if (current_opt) |current| {
+            if (get_value(pairs.items, segs)) |current| switch (current) {
-                switch (current) {
+                .array => |arr| try items.appendSlice(arena, arr),
-                    .array => |arr| try items.appendSlice(arena, arr),
+                .null => {},
-                    .null => {},
+                else => return error.InvalidUpdate, // non-array field
-                    else => return error.InvalidUpdate, // non-array field
+            };
+            if (mods) |m| {
+                try insert_each(arena, &items, m);
+                if (m.sort) |key| try sort_elements(items.items, key);
+                if (m.slice) |n| {
+                    const kept = slice_range(items.items.len, n);
+                    std.mem.copyForwards(bson.Value, items.items[0..kept.len], kept.of(items.items));
+                    items.shrinkRetainingCapacity(kept.len);
                 }
+            } else {
+                try items.append(arena, try bson.copy_value(arena, p.value));
             }
-            if (p.value == .doc) {
-                if (bson.get_pair(p.value.doc, "$each")) |each| {
-                    const arr = switch (each) {
-                        .array => |a| a,
-                        else => return error.InvalidUpdate,
-                    };
-                    for (arr) |item| try items.append(arena, try bson.copy_value(arena, item));
-                    try set_path(arena, pairs, segs, .{ .array = try items.toOwnedSlice(arena) }, p.key, opts.diag);
-                    continue;
-                }
-            }
-            try items.append(arena, try bson.copy_value(arena, p.value));
             try set_path(arena, pairs, segs, .{ .array = try items.toOwnedSlice(arena) }, p.key, opts.diag);
         }
     }
 }
 
+fn insert_each(
+    arena: std.mem.Allocator,
+    items: *std.ArrayListUnmanaged(bson.Value),
+    m: PushModifiers,
+) UpdateError!void {
+    const at: usize = if (m.position) |pos| blk: {
+        if (pos >= 0) break :blk @min(@as(usize, @intCast(pos)), items.items.len);
+        // Counted back from the end, and clamped at the front rather than
+        // wrapping: `$position: -99` on a two-element array is 0.
+        const back: usize = @intCast(-pos);
+        break :blk items.items.len -| back;
+    } else items.items.len;
+    for (m.each, 0..) |item, i| {
+        try items.insert(arena, at + i, try bson.copy_value(arena, item));
+    }
+}
+
+/// Which elements a `$slice` keeps. A non-negative `n` keeps the first `n`; a
+/// negative one keeps the **last** `-n`, which is the shape a capped log uses
+/// and the half that is easy to get backwards.
+fn slice_range(len: usize, n: i64) struct {
+    start: usize,
+    len: usize,
+
+    fn of(self: @This(), items: []bson.Value) []bson.Value {
+        return items[self.start .. self.start + self.len];
+    }
+} {
+    if (n >= 0) return .{ .start = 0, .len = @min(len, @as(usize, @intCast(n))) };
+    const keep = @min(len, @as(usize, @intCast(-n)));
+    return .{ .start = len - keep, .len = keep };
+}
+
+/// `$sort: 1` orders whole elements; `$sort: {a: 1}` orders on a field of
+/// them, which is only meaningful when they are documents -- an element that
+/// is not one, or that lacks the field, sorts as null, the same rank a missing
+/// field has everywhere else here.
+const ElementSort = struct {
+    path: ?[]const u8,
+    descending: bool,
+
+    fn key_of(self: ElementSort, v: bson.Value) bson.Value {
+        const path = self.path orelse return v;
+        const sub = switch (v) {
+            .doc => |d| d,
+            else => return .null,
+        };
+        var segs: [max_path_segments][]const u8 = undefined;
+        const n = split_path(path, &segs) orelse return .null;
+        return get_value(sub, segs[0..n]) orelse .null;
+    }
+
+    fn less(self: ElementSort, a: bson.Value, b: bson.Value) bool {
+        const order = bson.compare(self.key_of(a), self.key_of(b));
+        return if (self.descending) order == .gt else order == .lt;
+    }
+};
+
+fn sort_elements(items: []bson.Value, key: bson.Value) UpdateError!void {
+    const spec: ElementSort = switch (key) {
+        .doc => |pairs| blk: {
+            if (pairs.len != 1) return error.BadPushModifier;
+            break :blk .{ .path = pairs[0].key, .descending = is_descending(pairs[0].value) };
+        },
+        else => blk: {
+            if (!key.is_number()) return error.BadPushModifier;
+            break :blk .{ .path = null, .descending = is_descending(key) };
+        },
+    };
+    std.mem.sort(bson.Value, items, spec, ElementSort.less);
+}
+
+fn is_descending(v: bson.Value) bool {
+    return v.is_number() and v.as_f128() < 0;
+}
+
 fn op_pull(
     arena: std.mem.Allocator,
     pairs: *std.ArrayListUnmanaged(bson.Pair),
@@ -2150,3 +2286,141 @@ test "the array operators refuse a field that is not an array" {
     }), .{}));
     try testing.expectEqual(@as(i32, 5), doc.get("t").?.int32);
 }
+
+test "$push applies its modifiers in mongod's order: position, sort, slice" {
+    // The order is the whole of it. `$sort` runs over the array *after* the
+    // new elements are in it, and `$slice` runs last, over the sorted result.
+    var doc = try doc_with(testing.allocator, &.{
+        .{ .key = "t", .value = .{ .array = &.{ .{ .int32 = 3 }, .{ .int32 = 1 } } } },
+    });
+    defer doc.arena.deinit();
+    try apply(&doc, &doc_of(&.{
+        .{ .key = "$push", .value = .{ .doc = &.{.{ .key = "t", .value = .{ .doc = &.{
+            .{ .key = "$each", .value = .{ .array = &.{.{ .int32 = 2 }} } },
+            .{ .key = "$sort", .value = .{ .int32 = 1 } },
+            .{ .key = "$slice", .value = .{ .int32 = 2 } },
+        } } }} } },
+    }), .{});
+    const t = doc.get("t").?.array;
+    try testing.expectEqual(@as(usize, 2), t.len);
+    try testing.expectEqual(@as(i32, 1), t[0].int32);
+    try testing.expectEqual(@as(i32, 2), t[1].int32);
+}
+
+test "$slice keeps the first n, or with a negative n the last" {
+    // Mutation check: keep the first `-n` elements for a negative slice and
+    // the second half goes red. It is the half a capped log depends on.
+    const cases = [_]struct { n: i32, want: [3]i32, len: usize }{
+        .{ .n = 3, .want = .{ 1, 2, 3 }, .len = 3 },
+        .{ .n = -3, .want = .{ 2, 3, 4 }, .len = 3 },
+        .{ .n = 0, .want = .{ 0, 0, 0 }, .len = 0 },
+        .{ .n = 10, .want = .{ 1, 2, 3 }, .len = 4 },
+    };
+    for (cases) |c| {
+        var doc = try doc_with(testing.allocator, &.{
+            .{ .key = "t", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .int32 = 2 } } } },
+        });
+        defer doc.arena.deinit();
+        const spec = [2]bson.Pair{
+            .{ .key = "$each", .value = .{ .array = &.{ .{ .int32 = 3 }, .{ .int32 = 4 } } } },
+            .{ .key = "$slice", .value = .{ .int32 = c.n } },
+        };
+        const arg = [1]bson.Pair{.{ .key = "t", .value = .{ .doc = &spec } }};
+        try apply(&doc, &doc_of(&.{.{ .key = "$push", .value = .{ .doc = &arg } }}), .{});
+        const t = doc.get("t").?.array;
+        try testing.expectEqual(c.len, t.len);
+        for (t, 0..) |v, i| {
+            if (i < 3) try testing.expectEqual(c.want[i], v.int32);
+        }
+    }
+}
+
+test "$position inserts where it says, counting back from the end when negative" {
+    const cases = [_]struct { pos: i32, want: [3]i32 }{
+        .{ .pos = 0, .want = .{ 9, 1, 2 } },
+        .{ .pos = 1, .want = .{ 1, 9, 2 } },
+        .{ .pos = 99, .want = .{ 1, 2, 9 } },
+        .{ .pos = -1, .want = .{ 1, 9, 2 } },
+        .{ .pos = -99, .want = .{ 9, 1, 2 } }, // clamped at the front, not wrapped
+    };
+    for (cases) |c| {
+        var doc = try doc_with(testing.allocator, &.{
+            .{ .key = "t", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .int32 = 2 } } } },
+        });
+        defer doc.arena.deinit();
+        const spec = [2]bson.Pair{
+            .{ .key = "$each", .value = .{ .array = &.{.{ .int32 = 9 }} } },
+            .{ .key = "$position", .value = .{ .int32 = c.pos } },
+        };
+        const arg = [1]bson.Pair{.{ .key = "t", .value = .{ .doc = &spec } }};
+        try apply(&doc, &doc_of(&.{.{ .key = "$push", .value = .{ .doc = &arg } }}), .{});
+        const t = doc.get("t").?.array;
+        try testing.expectEqual(@as(usize, 3), t.len);
+        for (t, 0..) |v, i| try testing.expectEqual(c.want[i], v.int32);
+    }
+}
+
+test "$sort orders on a field of the elements" {
+    var doc = try doc_with(testing.allocator, &.{
+        .{ .key = "t", .value = .{ .array = &.{
+            .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 3 } }} },
+            .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} },
+        } } },
+    });
+    defer doc.arena.deinit();
+    try apply(&doc, &doc_of(&.{
+        .{ .key = "$push", .value = .{ .doc = &.{.{ .key = "t", .value = .{ .doc = &.{
+            .{ .key = "$each", .value = .{ .array = &.{
+                .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 2 } }} },
+            } } },
+            .{ .key = "$sort", .value = .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} } },
+        } } }} } },
+    }), .{});
+    const t = doc.get("t").?.array;
+    try testing.expectEqual(@as(usize, 3), t.len);
+    for (t, 1..) |v, want| try testing.expectEqual(@as(i32, @intCast(want)), v.doc[0].value.int32);
+}
+
+test "without $each there are no modifiers, only a value" {
+    // The measured rule that makes `$each` the flag: `{$slice: 1}` on its own
+    // is a document to push, not an instruction. Mutation check: treat any
+    // `$`-prefixed key as a modifier and this stores nothing where mongod
+    // stores a document.
+    var doc = try doc_with(testing.allocator, &.{
+        .{ .key = "t", .value = .{ .array = &.{.{ .int32 = 1 }} } },
+    });
+    defer doc.arena.deinit();
+    try apply(&doc, &doc_of(&.{
+        .{ .key = "$push", .value = .{ .doc = &.{.{ .key = "t", .value = .{ .doc = &.{
+            .{ .key = "$slice", .value = .{ .int32 = 1 } },
+        } } }} } },
+    }), .{});
+    const t = doc.get("t").?.array;
+    try testing.expectEqual(@as(usize, 2), t.len);
+    try testing.expect(t[1] == .doc);
+}
+
+test "a $push modifier that is not one, or is handed the wrong thing, is refused" {
+    var doc = try doc_with(testing.allocator, &.{
+        .{ .key = "t", .value = .{ .array = &.{.{ .int32 = 1 }} } },
+    });
+    defer doc.arena.deinit();
+    const bad = [_]bson.Pair{
+        .{ .key = "$bogus", .value = .{ .int32 = 1 } },
+        .{ .key = "$slice", .value = .{ .string = "x" } },
+        .{ .key = "$position", .value = .{ .string = "x" } },
+    };
+    for (bad) |m| {
+        const spec = [2]bson.Pair{
+            .{ .key = "$each", .value = .{ .array = &.{.{ .int32 = 3 }} } },
+            m,
+        };
+        const arg = [1]bson.Pair{.{ .key = "t", .value = .{ .doc = &spec } }};
+        try testing.expectError(error.BadPushModifier, apply(
+            &doc,
+            &doc_of(&.{.{ .key = "$push", .value = .{ .doc = &arg } }}),
+            .{},
+        ));
+    }
+    try testing.expectEqual(@as(usize, 1), doc.get("t").?.array.len);
+}