fix(model): preserve unsaved documents after ordered bulkSave errors

lib/model.js

@@ -3676,7 +3676,8 @@ Model.bulkSave = async function bulkSave(documents, options) {
 
   await Promise.all(documents.map(doc => buildPreSavePromise(doc, options)));
 
-  const writeOperations = this.buildBulkWriteOperations(documents, options);
+  const documentIndexes = [];
+  const writeOperations = this.buildBulkWriteOperations(documents, options, documentIndexes);
   const opts = { skipValidation: true, _skipCastBulkWrite: true, ...options };
   const { bulkWriteResult, bulkWriteError } = await this.bulkWrite(writeOperations, opts).then(
     (res) => ({ bulkWriteResult: res, bulkWriteError: null }),
@@ -3697,20 +3698,22 @@ Model.bulkSave = async function bulkSave(documents, options) {
     );
   }
 
-  const failedDocumentIds = documents.length >= 25 ? new Set(bulkWriteError?.writeErrors?.map(writeError => {
-    const writeErrorDocumentId = writeError.err.op._id || writeError.err.op.q._id;
-    return writeErrorDocumentId.toString();
-  })) : null;
-  const successfulDocuments = [];
-  for (let i = 0; i < documents.length; i++) {
-    const document = documents[i];
-    const documentId = document._doc._id.toString();
-    const failed = failedDocumentIds == null ? bulkWriteError?.writeErrors.find(writeError => {
-      const writeErrorDocumentId = writeError.err.op._id || writeError.err.op.q._id;
-      return writeErrorDocumentId.toString() === documentId;
-    }) != null : failedDocumentIds.has(documentId);
-    if (!failed) {
-      successfulDocuments.push(document);
+  const writeErrors = bulkWriteError?.writeErrors || [];
+  let successfulDocuments = documents;
+  if (writeErrors.length > 0) {
+    if (options.ordered !== false) {
+      // Unchanged documents don't generate operations, so map back to the input before slicing.
+      successfulDocuments = documents.slice(0, documentIndexes[writeErrors[0].index]);
+    } else {
+      const failedDocumentIndexes = documents.length >= 25 ? {} : null;
+      if (failedDocumentIndexes != null) {
+        for (const writeError of writeErrors) {
+          failedDocumentIndexes[documentIndexes[writeError.index]] = true;
+        }
+      }
+      successfulDocuments = documents.filter((document, i) => failedDocumentIndexes == null ?
+        !writeErrors.some(writeError => documentIndexes[writeError.index] === i) :
+        !failedDocumentIndexes[i]);
     }
   }
   await Promise.all(successfulDocuments.map(document => handleSuccessfulWrite(document, options)));
@@ -3976,11 +3979,12 @@ Model.castObject = function castObject(obj, options) {
  * @param {object} options
  * @param {boolean} options.skipValidation defaults to `false`, when set to true, building the write operations will bypass validating the documents.
  * @param {boolean} options.timestamps defaults to `null`, when set to false, mongoose will not add/update timestamps to the documents.
+ * @param {number[]} [documentIndexes] If provided, receives the input document index for each generated operation in order.
  * @return {Promise[]} Returns a array of all Promises the function executes to be awaited.
  * @api private
  */
 
-Model.buildBulkWriteOperations = function buildBulkWriteOperations(documents, options) {
+Model.buildBulkWriteOperations = function buildBulkWriteOperations(documents, options, documentIndexes) {
   if (!Array.isArray(documents)) {
     throw new MongooseError(`bulkSave expects an array of documents to be passed, received \`${documents}\` instead`);
   }
@@ -4010,6 +4014,7 @@ Model.buildBulkWriteOperations = function buildBulkWriteOperations(documents, op
       }
       const writeOperation = { insertOne: { document } };
       utils.injectTimestampsOption(writeOperation.insertOne, options.timestamps);
+      documentIndexes?.push(i);
       return writeOperation;
     }
 
@@ -4036,6 +4041,7 @@ Model.buildBulkWriteOperations = function buildBulkWriteOperations(documents, op
       document.$__version(where, delta);
       const writeOperation = { updateOne: { filter: where, update: changes } };
       utils.injectTimestampsOption(writeOperation.updateOne, options.timestamps);
+      documentIndexes?.push(i);
       return writeOperation;
     }
 

test/model.test.js

@@ -8207,6 +8207,158 @@ describe('Model', function() {
       assert.deepEqual(user2.getChanges(), { $set: { age: 27, name: 'Sam' } });
 
     });
+    it('preserves unattempted inserts after an ordered bulkSave error', async() => {
+      const userSchema = new Schema({ name: { type: String, unique: true } });
+      const savedNames = [];
+      userSchema.post('save', function() {
+        savedNames.push(this.name);
+      });
+      const User = db.model('User', userSchema);
+      await User.init();
+
+      const users = [
+        new User({ name: 'duplicate' }),
+        new User({ name: 'duplicate' }),
+        new User({ name: 'last' })
+      ];
+      await assert.rejects(User.bulkSave(users), { name: 'MongoBulkWriteError', code: 11000 });
+
+      assert.equal(await User.countDocuments(), 1);
+      assert.deepStrictEqual(users.map(user => user.isNew), [false, true, true]);
+      assert.deepStrictEqual(users.map(user => user.isModified('name')), [false, true, true]);
+      assert.deepStrictEqual(savedNames, ['duplicate']);
+
+      users[1].name = 'recovered';
+      await User.bulkSave(users.slice(1));
+      assert.equal(await User.countDocuments(), users.length);
+      assert.ok(users.every(user => !user.isNew && !user.isModified()));
+    });
+
+    it('preserves unattempted inserts after an ordered bulkSave error with 25 documents', async() => {
+      const userSchema = new Schema({ name: { type: String, unique: true } });
+      const savedNames = [];
+      userSchema.post('save', function() {
+        savedNames.push(this.name);
+      });
+      const User = db.model('User', userSchema);
+      await User.init();
+
+      const documentCount = 25;
+      const users = Array.from({ length: documentCount }, (_, index) => new User({
+        name: index < 2 ? 'duplicate' : `user-${index}`
+      }));
+      await assert.rejects(User.bulkSave(users), { name: 'MongoBulkWriteError', code: 11000 });
+
+      assert.equal(await User.countDocuments(), 1);
+      assert.deepStrictEqual(users.map(user => user.isNew), [false, ...Array(documentCount - 1).fill(true)]);
+      assert.deepStrictEqual(users.map(user => user.isModified('name')), [false, ...Array(documentCount - 1).fill(true)]);
+      assert.deepStrictEqual(savedNames, ['duplicate']);
+
+      users[1].name = 'recovered';
+      await User.bulkSave(users.slice(1));
+      assert.equal(await User.countDocuments(), documentCount);
+      assert.ok(users.every(user => !user.isNew && !user.isModified()));
+    });
+
+    it('preserves unattempted updates after an ordered bulkSave error with an unchanged document', async() => {
+      const userSchema = new Schema({ name: { type: String, unique: true } });
+      const savedNames = [];
+      userSchema.post('save', function() {
+        savedNames.push(this.name);
+      });
+      const User = db.model('User', userSchema);
+      await User.init();
+      const users = await User.insertMany([
+        { name: 'unchanged' },
+        { name: 'first' },
+        { name: 'second' },
+        { name: 'third' }
+      ]);
+
+      users[1].name = 'duplicate';
+      users[2].name = 'duplicate';
+      users[3].name = 'pending';
+      await assert.rejects(User.bulkSave(users, { ordered: true }), { name: 'MongoBulkWriteError', code: 11000 });
+
+      assert.equal((await User.findById(users[1]._id)).name, 'duplicate');
+      assert.equal((await User.findById(users[3]._id)).name, 'third');
+      assert.deepStrictEqual(users.map(user => user.isModified('name')), [false, false, true, true]);
+      assert.deepStrictEqual(savedNames, ['unchanged', 'duplicate']);
+
+      users[2].name = 'recovered';
+      await User.bulkSave(users.slice(2));
+      assert.equal((await User.findById(users[2]._id)).name, 'recovered');
+      assert.equal((await User.findById(users[3]._id)).name, 'pending');
+    });
+
+    it('uses operation order when an ordered bulkSave fails on a duplicate _id', async() => {
+      const User = db.model('User', new Schema({ name: String }));
+      const sharedId = new mongoose.Types.ObjectId();
+      const users = [
+        new User({ _id: sharedId, name: 'first' }),
+        new User({ name: 'second' }),
+        new User({ _id: sharedId, name: 'duplicate' }),
+        new User({ name: 'unattempted' })
+      ];
+
+      await assert.rejects(User.bulkSave(users), { name: 'MongoBulkWriteError', code: 11000 });
+
+      assert.equal(await User.countDocuments(), 2);
+      assert.deepStrictEqual(users.map(user => user.isNew), [false, false, true, true]);
+      assert.deepStrictEqual(users.map(user => user.isModified('name')), [false, false, true, true]);
+    });
+
+    it('preserves unattempted writes when ordered bulkSave mixes inserts and updates', async() => {
+      const User = db.model('User', new Schema({ name: { type: String, unique: true } }));
+      await User.init();
+      const existing = await User.insertMany([{ name: 'unchanged' }, { name: 'original' }]);
+      existing[1].name = 'pending';
+      const inserted = new User({ name: 'new' });
+      const duplicate = new User({ name: 'new' });
+      const unattempted = new User({ name: 'last' });
+
+      await assert.rejects(User.bulkSave([existing[0], inserted, duplicate, existing[1], unattempted]), {
+        name: 'MongoBulkWriteError', code: 11000
+      });
+
+      assert.equal(inserted.isNew, false);
+      assert.equal(duplicate.isNew, true);
+      assert.equal(existing[1].isModified('name'), true);
+      assert.equal(unattempted.isNew, true);
+      await User.bulkSave([existing[1], unattempted]);
+      assert.equal((await User.findById(existing[1]._id)).name, 'pending');
+      assert.ok(await User.exists({ _id: unattempted._id }));
+    });
+
+    it('preserves unordered failures while saving later documents with a repeated _id', async() => {
+      const savedNames = [];
+      const userSchema = new Schema({ name: String });
+      userSchema.post('save', function() {
+        savedNames.push(this.name);
+      });
+      const User = db.model('User', userSchema);
+      const existing = await User.create({ name: 'existing' });
+      savedNames.length = 0;
+      const sharedId = new mongoose.Types.ObjectId();
+      const users = [
+        existing,
+        new User({ _id: sharedId, name: 'first' }),
+        new User({ _id: sharedId, name: 'duplicate' }),
+        new User({ name: 'last' })
+      ];
+
+      await assert.rejects(User.bulkSave(users, { ordered: false }), { name: 'MongoBulkWriteError', code: 11000 });
+
+      assert.equal(await User.countDocuments(), 3);
+      assert.deepStrictEqual(users.map(user => user.isNew), [false, false, true, false]);
+      assert.deepStrictEqual(users.map(user => user.isModified('name')), [false, false, true, false]);
+      assert.deepStrictEqual(savedNames, ['existing', 'first', 'last']);
+
+      users[2]._id = new mongoose.Types.ObjectId();
+      await User.bulkSave([users[2]], { ordered: false });
+      assert.equal(await User.countDocuments(), users.length);
+    });
+
     it('updates successful document state with multiple unordered write errors and 25 documents', async() => {
       const userSchema = new Schema({
         name: { type: String, unique: true }