Rust MongoDB server: fix-one batch snapshot of rm-batch (298bf6b8, 1 commits)

changelog.d/crash-test-evidence.md

@@ -0,0 +1,7 @@
+### A crash-recovery test records why it fails, the next time it does
+
+`test_hard_kill_mid_transactions_keeps_every_acked_commit_and_its_history` failed once during a local gate. After the hard kill, a snapshot read at the first acknowledged commit's cluster time was refused as too old, because that time was 16 minutes before the run began. It has not been reproduced. The test now checks that its ack file is new and that every ack's cluster time falls inside the run's own wall-clock window. A repeat will show whether the ack file came from another run or the server minted a stale timestamp. The case is filed as RM-401.
+
+#### Changed
+
+- `tests/test_crash_recovery.py`: two checks that capture the cause of RM-401's failure. Filed RM-401.

changelog.d/rm-325-per-platform.md

@@ -0,0 +1,7 @@
+### Spill sizes are compared with each platform's own mongod
+
+mongod 8.2 writes spill files of different sizes on arm64 and x86_64. The same in-order sort stores 130,545 bytes on Apple silicon and 131,252 on CI's x86_64 Linux. Each platform's mongod is the reference. RM-325's regression test therefore compares every case on arm64. On x86_64 it leaves out the one case that still differs there, a sort after `$project` (122,288 bytes against mongod's 122,253), which is filed as RM-402 and checked on x86_64 by CI. RM-397's `$group` gap exists only on arm64, so its gap test now runs there only.
+
+#### Changed
+
+- `tests/backlog/test_rm_found_2026_10_26.py`: RM-325's regression test compares by platform, RM-402 has a new x86_64-only gap test, and RM-397's gap test runs on arm64 only. Filed RM-402.

changelog.d/rm-325-spill-file-format.md

@@ -0,0 +1,14 @@
+### A spilled sort takes the bytes on disk that mongod's does
+
+When a sort passes its memory budget, mongod writes its sorted runs to a spill file, and the profiler, `explain` and `serverStatus` report the file's size as `spilledDataStorageSize`. The Rust MongoDB server spilled the same runs, records and bytes, but wrote them as uncompressed BSON, so it reported about twenty times mongod's figure: 2480400 bytes where mongod 8.2 reports 130545. Its sorter now writes mongod's file. Each record is in mongod's own `serializeForSorter` form, the records go into 64 KiB chunks, and each chunk is snappy-compressed by a port of the compressor mongod builds in (snappy 1.1.10, with its CRC32-C hash), or stored as it is when compression saves less than a tenth. A query-layer sort, a pipeline `$sort` and `$bucketAuto` now report mongod's size to the byte whenever the sorter sees its documents in the order mongod's does. The size was checked by decoding mongod's own spill file, copied out of `_tmp` while a cursor held it open.
+
+Some spills still differ, and each is on the backlog. A classic `$group` writes this server's own partial states, and a `$setWindowFields` uses a file where mongod uses a WiredTiger table (RM-325). A sort that reorders the documents a scan reads is fed them already sorted, so it spills different runs (RM-395).
+
+#### Fixed
+
+- `secantus-core` (`sorter`, new `spill_file`, new `snappy`): a sort's runs are written as mongod 8.2's `SortedFileWriter` writes them. A record is a `Value` key (an array for a compound sort), then the document: as BSON for the query layer's sort, as a `Document` of the fields kept for a pipeline `$sort`, and inside `{output, generatedSortKey}` for `$bucketAuto`. Records go into 64 KiB chunks, each prefixed with its signed `int32` length and snappy-compressed when that saves a tenth. `spilledBytes` is now the bytes before compression, and `spilledDataStorageSize` and `metrics.query.sort.spillToDiskBytes` are the bytes written. Records convert from this crate's BSON and back byte for byte, so strings that are not UTF-8 and escaped field names survive a spill. The file's size on disk is checked against what was written.
+- Partly fixes RM-325 (tests/backlog/test_rm_found_2026_10_26.py::test_spill_file_storage_size). The item stays open for the `$group` and `$setWindowFields` spills (`test_spill_storage_size`).
+- Regression tests: `sorter::tests::spills_and_merges_count_as_mongods_do` (130545 bytes, merges included), `sorter::tests::a_spill_file_takes_what_mongods_takes`, `spill_file::tests::*` and `snappy::tests::*` (the CRC matches the arm64 `__crc32cw` instruction).
+- Filed RM-395: a query-layer sort is handed documents already sorted, so it spills different runs from mongod's (`test_leading_sort_spill_storage_size`).
+- Filed RM-396: a `$sort` after `$addFields` spills at slightly different record boundaries (`test_added_fields_sort_spill_storage_size`).
+- Filed RM-397: a `$group` of `$sum` and `$count` counts 16 bytes a group less memory than mongod (`test_group_sum_spilled_bytes`).

changelog.d/rm-326-sort-counts-per-load.md

@@ -0,0 +1,12 @@
+### A spilled sort is counted the way mongod's cursor stage counts it
+
+When the stages after a pipeline's sort read it through mongod's cursor stage, mongod adds the sort's counts again for every buffer it loads from the query layer. A `$setWindowFields` over 300 documents whose sort spilled 13 times is therefore profiled with `sortSpills: 52`, and a first batch of 101 documents behind an `$addFields` with 39. The Rust MongoDB server now reports the same numbers in the profile and the slow-query line. It works out how many buffers mongod loads, of 32 documents at first and doubling, from the batch the client asked for. That count includes the `$skip` and `$limit` mongod moves into the query layer, and the 4 MB a buffer stops at. `metrics.query.sort.spillToDisk` now counts what mongod counts: one for each 64 KB chunk a sorter writes to its spill file, `$bucketAuto`'s sorter included.
+
+#### Fixed
+
+- `secantus-core` (`aggregate`, `memlimit`, `sorter`), `secantus-commands` (`aggregate`, `agg_executor`): the profile's and the slow-query line's `sortSpills`, `sortSpilledRecords`, `sortSpilledBytes`, `sortSpilledDataStorageSize` and `sortTotalDataSizeBytes` count a query-layer sort once per cursor-stage load, as mongod 8.2.11 does. `spillToDisk` counts sorter file chunks, and `$bucketAuto`'s spills now count in `metrics.query.sort` too.
+- Fixes RM-326 in part (tests/backlog/test_rm_found_2026_10_26.py::test_sort_profile_per_batch, tests/backlog/test_rm_326_sort_counts.py::test_sort_counts_per_load). RM-326 stays open for stages whose reads depend on the documents (`$unwind`, `$match`, a bounded or partitioned window): `test_sort_counts_per_load_through_stages_that_filter`.
+- Regression tests: `agg_executor::tests::the_cursor_stage_loads_as_mongod_does`, `agg_executor::tests::reads_that_depend_on_the_documents_are_not_followed`.
+- Filed RM-400: a `getMore` reports none of its cursor's sort counts, and `totalKeysSorted` / `totalBytesSorted` count each sort once (`test_rm_326_sort_counts.py::test_get_more_sort_counts`).
+- Filed RM-398: an SBE sort counts its bytes in `totalBytesSorted`, which mongod's does not (`test_rm_326_sort_counts.py::test_sbe_sort_bytes`).
+- Filed RM-399: a sort with a limit does not spill as mongod's top-k sorter does (`test_rm_326_sort_counts.py::test_top_k_sort_spills`).

crates/secantus-commands/src/agg_executor.rs

@@ -217,6 +217,202 @@ pub(crate) fn holds_executor(
         .map_or(true, |b| b.len() as i64 >= FIRST_LOAD)
 }
 
+/// The bytes a cursor-stage buffer stops at.
+const LOAD_BYTES: usize = 4 * 1024 * 1024;
+
+/// What the stages read of their input to give `pulls` documents of
+/// output: some number of documents, or everything to its end.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Reads {
+    Some(usize),
+    All,
+}
+
+/// How many buffers mongod's cursor stage loads from its executor while
+/// `stages` (the pipeline after the stages the query layer took) give
+/// `pulls` documents. `sizes`: the BSON size of each document the executor
+/// hands over, in order, before the `$skip` and `$limit` it takes from
+/// `stages`. `limited`: the executor already took a `$limit`. `None`: what
+/// the stages read depends on the documents (a `$match`, an `$unwind`, a
+/// `$setWindowFields` window), which is not followed here.
+///
+/// mongod 8.2.11: the `$skip`s and the least `$limit` that only stages
+/// passing each document on stand before are moved into the executor
+/// (`extractSkipForPushdown`, `extractLimitForPushdown`). Its cursor stage
+/// (`DocumentSourceCursor::loadBatch`) then loads a buffer of 32 documents
+/// at first and twice as many after each that filled by count -- without a
+/// count when the executor took a `$limit` -- and stops a buffer once it
+/// holds more than 4 MB; a load that reaches the end of the executor is its
+/// last. Asking for the document after the last one loads again.
+pub(crate) fn cursor_loads(
+    stages: &[Bson],
+    sizes: &[usize],
+    limited: bool,
+    pulls: usize,
+) -> Option<usize> {
+    let count = |stage: &Bson, name: &str| {
+        stage
+            .as_document()
+            .and_then(|d| d.get(name))
+            .and_then(crate::util::as_i64)
+            .and_then(|n| usize::try_from(n).ok())
+    };
+    let (mut skip, mut limit) = (0usize, None::<usize>);
+    let mut left: Vec<&Bson> = Vec::new();
+    let mut moving = true;
+    for (i, stage) in stages.iter().enumerate() {
+        if moving {
+            if let Some(k) = count(stage, "$skip") {
+                skip = skip.saturating_add(k);
+                continue;
+            }
+            if let Some(k) = count(stage, "$limit") {
+                let k = k.saturating_add(skip);
+                limit = Some(limit.map_or(k, |l| l.min(k)));
+                continue;
+            }
+            moving = swaps_with_skip_and_limit(stage, stages.get(i + 1));
+        }
+        left.push(stage);
+    }
+    let sizes = &sizes[..limit.map_or(sizes.len(), |l| l.min(sizes.len()))];
+    let sizes = &sizes[skip.min(sizes.len())..];
+    let mut cap = if limited || limit.is_some() {
+        usize::MAX
+    } else {
+        FIRST_LOAD as usize
+    };
+    let mut reads = Reads::Some(pulls);
+    for stage in left.into_iter().rev() {
+        let (name, spec) = stage.as_document().and_then(|d| d.iter().next())?;
+        let n = || crate::util::as_i64(spec).and_then(|n| usize::try_from(n).ok());
+        reads = match (name.as_str(), reads) {
+            (_, Reads::Some(0)) => Reads::Some(0),
+            (secantus_core::sbe::ENGINE_STAGE, r) => r,
+            // A `$limit` asks for no more once it has passed its count.
+            ("$limit", Reads::All) => Reads::Some(n()?),
+            ("$limit", Reads::Some(p)) => Reads::Some(p.min(n()?)),
+            ("$skip", Reads::Some(p)) => Reads::Some(p + n()?),
+            (
+                "$project" | "$addFields" | "$set" | "$unset" | "$replaceRoot" | "$replaceWith"
+                | "$lookup" | "$graphLookup" | "$unionWith",
+                r,
+            ) => r,
+            ("$setWindowFields", Reads::Some(_)) if whole_input_window(spec) => Reads::All,
+            (_, Reads::Some(_)) if blocks_bson(stage) => Reads::All,
+            (_, Reads::All) => Reads::All,
+            _ => return None,
+        };
+    }
+    let wanted = match reads {
+        Reads::Some(p) if p <= sizes.len() => Some(p),
+        _ => None,
+    };
+    let (mut at, mut loads) = (0usize, 0usize);
+    loop {
+        if wanted.is_some_and(|w| at >= w) {
+            return Some(loads);
+        }
+        loads += 1;
+        let (mut count, mut bytes) = (0usize, 0usize);
+        loop {
+            let Some(size) = sizes.get(at) else {
+                return Some(loads);
+            };
+            at += 1;
+            count += 1;
+            bytes += size;
+            if count >= cap {
+                cap = cap.saturating_mul(2);
+                break;
+            }
+            if bytes > LOAD_BYTES {
+                break;
+            }
+        }
+    }
+}
+
+/// mongod's `canSwapWithSkippingOrLimitingStage`: a stage that hands on
+/// each document it is given, one for one. A `$lookup` / `$graphLookup`
+/// that takes in the `$unwind` after it does not.
+fn swaps_with_skip_and_limit(stage: &Bson, next: Option<&Bson>) -> bool {
+    match name_of(stage) {
+        "$project"
+        | "$addFields"
+        | "$set"
+        | "$unset"
+        | "$replaceRoot"
+        | "$replaceWith"
+        | secantus_core::sbe::ENGINE_STAGE => true,
+        name @ ("$lookup" | "$graphLookup") => {
+            let as_field = stage
+                .as_document()
+                .and_then(|d| d.get_document(name).ok())
+                .and_then(|spec| spec.get_str("as").ok());
+            let unwound = next
+                .and_then(Bson::as_document)
+                .and_then(|d| match d.get("$unwind") {
+                    Some(Bson::String(path)) => Some(path.clone()),
+                    Some(Bson::Document(spec)) => spec.get_str("path").ok().map(str::to_string),
+                    _ => None,
+                });
+            !matches!((as_field, unwound), (Some(a), Some(u)) if u == format!("${a}"))
+        }
+        _ => false,
+    }
+}
+
+fn blocks_bson(stage: &Bson) -> bool {
+    stage.as_document().is_some_and(blocks) || matches!(name_of(stage), "$out" | "$merge")
+}
+
+fn name_of(stage: &Bson) -> &str {
+    stage
+        .as_document()
+        .and_then(|d| d.keys().next())
+        .map_or("", String::as_str)
+}
+
+/// A `$setWindowFields` over one partition with an output whose window
+/// reaches its end: it reads everything before its first document.
+fn whole_input_window(spec: &Bson) -> bool {
+    let Some(spec) = spec.as_document() else {
+        return false;
+    };
+    if spec.contains_key("partitionBy") {
+        return false;
+    }
+    let Ok(output) = spec.get_document("output") else {
+        return false;
+    };
+    // The functions that take no window, each of which reads at most a
+    // bounded distance ahead.
+    const UNWINDOWED: &[&str] = &[
+        "$rank",
+        "$denseRank",
+        "$documentNumber",
+        "$shift",
+        "$locf",
+        "$linearFill",
+        "$expMovingAvg",
+    ];
+    output.values().filter_map(Bson::as_document).any(|out| {
+        let function = out.keys().find(|k| *k != "window").map(String::as_str);
+        match out.get_document("window") {
+            Ok(window) => ["documents", "range"].iter().any(|kind| {
+                window
+                    .get_array(kind)
+                    .ok()
+                    .and_then(|b| b.get(1))
+                    .is_some_and(|upper| matches!(upper, Bson::String(s) if s == "unbounded"))
+            }),
+            // A windowed function with no window: the whole partition.
+            Err(_) => function.is_some_and(|f| !UNWINDOWED.contains(&f)),
+        }
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -302,4 +498,92 @@ mod tests {
         let d = vec![m, doc! { "$group": { "_id": "$a" } }];
         assert_eq!(executor(&d, &on_a), Executor::HeldIfLarge(2));
     }
+
+    fn b(d: Document) -> Bson {
+        Bson::Document(d)
+    }
+
+    /// The loads measured on mongod 8.2.11 (2026-11-07) for 300 documents
+    /// of 1 KB behind a sort that spilled: `sortSpills` was 13 times these.
+    #[test]
+    fn the_cursor_stage_loads_as_mongod_does() {
+        let sizes = vec![1060; 300];
+        let af = b(doc! { "$addFields": { "z": 1 } });
+        let group = b(doc! { "$group": { "_id": "$k" } });
+        let window = b(doc! { "$setWindowFields": {
+        "sortBy": { "n": 1 }, "output": { "t": { "$sum": "$n" } } } });
+        let cases: Vec<(Vec<Bson>, usize, usize)> = vec![
+            // A first batch of 101: 32 + 64 + 128 documents read.
+            (vec![af.clone()], 101, 3),
+            (vec![af.clone()], 10, 1),
+            (vec![af.clone()], 40, 2),
+            // All 300, and the end found: 32 + 64 + 128 + 76.
+            (vec![af.clone()], 301, 4),
+            (vec![af.clone(), group.clone()], 8, 4),
+            (vec![window.clone()], 102, 4),
+            // The `$skip` moves into the executor, which hands over 200.
+            (vec![af.clone(), b(doc! { "$skip": 100 })], 101, 3),
+            (vec![af.clone(), b(doc! { "$skip": 150 })], 101, 3),
+            // A `$limit` in the executor lifts the count: one buffer.
+            (vec![af.clone(), b(doc! { "$limit": 50 })], 51, 1),
+            (
+                vec![
+                    af.clone(),
+                    b(doc! { "$skip": 10 }),
+                    b(doc! { "$limit": 150 }),
+                ],
+                101,
+                1,
+            ),
+        ];
+        for (stages, pulls, loads) in cases {
+            assert_eq!(
+                cursor_loads(&stages, &sizes, false, pulls),
+                Some(loads),
+                "{stages:?} {pulls}"
+            );
+        }
+        // A `$limit` the executor took before the stages.
+        assert_eq!(
+            cursor_loads(std::slice::from_ref(&af), &sizes[..200], true, 101),
+            Some(1)
+        );
+        // 32 documents fill the first buffer; the end is a second load.
+        assert_eq!(
+            cursor_loads(std::slice::from_ref(&window), &sizes[..32], false, 33),
+            Some(2)
+        );
+        assert_eq!(cursor_loads(&[window], &sizes[..33], false, 34), Some(2));
+        // A buffer stops past 4 MB, without doubling.
+        let big = vec![1 << 20; 10];
+        assert_eq!(
+            cursor_loads(std::slice::from_ref(&group), &big, true, 1),
+            Some(3)
+        );
+        assert_eq!(cursor_loads(&[group], &big, false, 1), Some(3));
+    }
+
+    #[test]
+    fn reads_that_depend_on_the_documents_are_not_followed() {
+        let sizes = vec![1060; 300];
+        for stage in [
+            doc! { "$unwind": "$s" },
+            doc! { "$match": { "k": 1 } },
+            doc! { "$setWindowFields": { "sortBy": { "n": 1 }, "output": { "r": { "$rank": {} } } } },
+            doc! { "$setWindowFields": {
+            "partitionBy": "$k", "sortBy": { "n": 1 }, "output": { "t": { "$sum": "$n" } } } },
+        ] {
+            assert_eq!(
+                cursor_loads(&[b(stage.clone())], &sizes, false, 101),
+                None,
+                "{stage:?}"
+            );
+        }
+        // Read to the end, they read everything.
+        let then_group = [
+            b(doc! { "$match": { "k": 1 } }),
+            b(doc! { "$group": { "_id": "$k" } }),
+        ];
+        assert_eq!(cursor_loads(&then_group, &sizes, false, 2), Some(4));
+    }
 }