MDH: stable _id sort in download to prevent duplicate rows across batches

The collection download issues N concurrent aggregate calls with $skip/$limit
and no explicit sort, relying on MongoDB's natural order. That order is not
guaranteed to be stable across separate aggregate executions, so adjacent
windows can overlap (duplicate _ids in the output file) and other docs fall
in the gap and are missed entirely.

Append {$sort: {_id: 1}} to every per-batch pipeline unless the caller's
pipelineStages already end in $sort. The _id index is always present so the
sort is an index scan with no extra cost. Caller-supplied sorts (e.g. from
"Download filtered") are preserved; the header comment now recommends
including _id as a tie-breaker when sorting on a non-unique field.

Author: mrtnzlml

src/mdh/downloadCollection.js

@@ -31,6 +31,15 @@ import * as api from './api.js';
 // inside the buffer-room wait. The streaming file, if any, is aborted on
 // cancel so the partial file is discarded rather than left as half-valid
 // JSON on the user's disk.
+//
+// Stable ordering across batches: each worker issues its OWN aggregate
+// call, and MongoDB does not guarantee a stable natural order across
+// independent aggregations on the same collection. Without an explicit
+// sort, adjacent $skip/$limit windows can overlap (the same doc appears
+// in two batches) AND leave gaps (other docs missed entirely). We append
+// {$sort: {_id: 1}} to the pipeline when the caller hasn't provided
+// their own sort, so every worker scans in the same deterministic order.
+// {_id: 1} uses the always-present _id index, so this is free.
 
 export const BATCH_SIZE = 1000;
 export const CONCURRENCY = 10;
@@ -52,7 +61,10 @@ export async function downloadCollection(collectionName, opts = {}) {
     // Prepended to every batch's aggregate call. Default downloads the raw
     // collection; pass `[{$match: ...}, {$sort: ...}, ...]` to export the
     // result of a filtered/transformed pipeline. The downloader appends its
-    // own `$skip` / `$limit` per batch — callers should strip those.
+    // own `$sort` (if absent), `$skip`, and `$limit` per batch — callers
+    // should strip those. Callers sorting on a non-unique field should
+    // include `_id` as a tie-breaker (e.g. `{$sort: {name: 1, _id: 1}}`)
+    // to keep batch boundaries stable.
     pipelineStages = [{ $match: {} }],
     filename: filenameOpt,
   } = opts;
@@ -86,6 +98,14 @@ export async function downloadCollection(collectionName, opts = {}) {
     const offsets = [];
     for (let s = 0; s < total; s += batchSize) offsets.push(s);
 
+    // Inject a deterministic sort unless the caller already ended their
+    // pipeline with one. Without this, separate aggregate calls can iterate
+    // the collection in different orders and the workers' $skip/$limit
+    // windows overlap.
+    const stages = pipelineEndsWithSort(pipelineStages)
+      ? pipelineStages
+      : [...pipelineStages, { $sort: { _id: 1 } }];
+
     const parts = [];
     let docsWritten = 0;
     let fetched = 0;
@@ -150,7 +170,7 @@ export async function downloadCollection(collectionName, opts = {}) {
 
         try {
           const res = await api.aggregate(collectionName, [
-            ...pipelineStages,
+            ...stages,
             { $skip: myOffset },
             { $limit: batchSize },
           ]);
@@ -203,6 +223,12 @@ function formatDoc(doc) {
   return '  ' + JSON.stringify(doc, null, 2).replace(/\n/g, '\n  ');
 }
 
+function pipelineEndsWithSort(stages) {
+  if (!Array.isArray(stages) || stages.length === 0) return false;
+  const last = stages[stages.length - 1];
+  return last && typeof last === 'object' && Object.prototype.hasOwnProperty.call(last, '$sort');
+}
+
 async function safeAbort(writer) {
   if (!writer || typeof writer.abort !== 'function') return;
   try { await writer.abort('cancelled'); } catch { /* writer may already be closed */ }

tests/mdh-download-collection.test.js

@@ -73,13 +73,49 @@ describe('downloadCollection — streaming (FileSystem Access) path', () => {
 
     expect(result.fetched).toBe(2500);
     expect(api.aggregate).toHaveBeenCalledTimes(3);
-    expect(api.aggregate).toHaveBeenNthCalledWith(1, 'big', [{ $match: {} }, { $skip: 0 }, { $limit: 1000 }]);
-    expect(api.aggregate).toHaveBeenNthCalledWith(2, 'big', [{ $match: {} }, { $skip: 1000 }, { $limit: 1000 }]);
-    expect(api.aggregate).toHaveBeenNthCalledWith(3, 'big', [{ $match: {} }, { $skip: 2000 }, { $limit: 1000 }]);
+    // Every batch must include the {$sort: {_id: 1}} we inject — without it,
+    // MongoDB's natural order isn't stable across separate aggregate calls
+    // and adjacent windows overlap, producing duplicate _ids in the output.
+    expect(api.aggregate).toHaveBeenNthCalledWith(1, 'big', [{ $match: {} }, { $sort: { _id: 1 } }, { $skip: 0 }, { $limit: 1000 }]);
+    expect(api.aggregate).toHaveBeenNthCalledWith(2, 'big', [{ $match: {} }, { $sort: { _id: 1 } }, { $skip: 1000 }, { $limit: 1000 }]);
+    expect(api.aggregate).toHaveBeenNthCalledWith(3, 'big', [{ $match: {} }, { $sort: { _id: 1 } }, { $skip: 2000 }, { $limit: 1000 }]);
     const parsed = JSON.parse(writer.chunks.join(''));
     expect(parsed).toEqual(docs);
   });
 
+  it('preserves a caller-provided sort instead of overriding it', async () => {
+    api.aggregate.mockResolvedValueOnce({ result: [{ _id: 1 }] });
+    const writer = fakeWriter();
+    await downloadCollection('c', {
+      fetchCount: async () => 1,
+      pipelineStages: [{ $match: { status: 'open' } }, { $sort: { name: 1, _id: 1 } }],
+      pickFile: () => Promise.resolve(fakeHandle(writer)),
+    });
+    expect(api.aggregate).toHaveBeenCalledWith('c', [
+      { $match: { status: 'open' } },
+      { $sort: { name: 1, _id: 1 } },
+      { $skip: 0 },
+      { $limit: 1000 },
+    ]);
+  });
+
+  it('appends the trailing _id sort when the caller\'s pipeline ends with a non-sort stage', async () => {
+    api.aggregate.mockResolvedValueOnce({ result: [{ _id: 1 }] });
+    const writer = fakeWriter();
+    await downloadCollection('c', {
+      fetchCount: async () => 1,
+      pipelineStages: [{ $match: { active: true } }, { $project: { name: 1 } }],
+      pickFile: () => Promise.resolve(fakeHandle(writer)),
+    });
+    expect(api.aggregate).toHaveBeenCalledWith('c', [
+      { $match: { active: true } },
+      { $project: { name: 1 } },
+      { $sort: { _id: 1 } },
+      { $skip: 0 },
+      { $limit: 1000 },
+    ]);
+  });
+
   it('writes batches in source order even when later batches resolve first', async () => {
     const d0 = defer();
     const d1 = defer();
@@ -484,6 +520,7 @@ describe('downloadCollection — pipelineStages option (filtered download)', ()
     });
     expect(api.aggregate).toHaveBeenCalledWith('c', [
       { $match: {} },
+      { $sort: { _id: 1 } },
       { $skip: 0 },
       { $limit: 1000 },
     ]);