fix(NODE-3515): do proper opTime merging in bulk results

lib/bulk/common.js

@@ -414,6 +414,15 @@ class WriteError {
   }
 }
 
+/**
+ * Converts the number to a Long or returns it.
+ *
+ * @ignore
+ */
+function longOrConvert(value) {
+  return typeof value === 'number' ? Long.fromNumber(value) : value;
+}
+
 /**
  * Merges results into shared data structure
  * @ignore
@@ -445,44 +454,41 @@ function mergeBatchResults(batch, bulkResult, err, result) {
     return;
   }
 
-  // Deal with opTime if available
+  // The server write command specification states that lastOp is an optional
+  // mongod only field that has a type of timestamp. Across various scare specs
+  // where opTime is mentioned, it is an "opaque" object that can have a "ts" and
+  // "t" field with Timestamp and Long as their types respectively.
+  // The "lastOp" field of the bulk write result is never mentioned in the driver
+  // specifications or the bulk write spec, so we should probably just keep its
+  // value consistent since it seems to vary.
+  // See: https://github.com/mongodb/specifications/blob/master/source/driver-bulk-update.rst#results-object
   if (result.opTime || result.lastOp) {
-    const opTime = result.lastOp || result.opTime;
-    let lastOpTS = null;
-    let lastOpT = null;
-
-    // We have a time stamp
-    if (opTime && opTime._bsontype === 'Timestamp') {
-      if (bulkResult.lastOp == null) {
-        bulkResult.lastOp = opTime;
-      } else if (opTime.greaterThan(bulkResult.lastOp)) {
-        bulkResult.lastOp = opTime;
-      }
-    } else {
-      // Existing TS
-      if (bulkResult.lastOp) {
-        lastOpTS =
-          typeof bulkResult.lastOp.ts === 'number'
-            ? Long.fromNumber(bulkResult.lastOp.ts)
-            : bulkResult.lastOp.ts;
-        lastOpT =
-          typeof bulkResult.lastOp.t === 'number'
-            ? Long.fromNumber(bulkResult.lastOp.t)
-            : bulkResult.lastOp.t;
+    let opTime = result.lastOp || result.opTime;
+
+    if (opTime) {
+      // If the opTime is a Timestamp, convert it to a consistent format to be
+      // able to compare easily. Converting to the object from a timestamp is
+      // much more straightforward than the other direction.
+      if (opTime._bsontype === 'Timestamp') {
+        opTime = { ts: opTime, t: Long.ZERO };
       }
 
-      // Current OpTime TS
-      const opTimeTS = typeof opTime.ts === 'number' ? Long.fromNumber(opTime.ts) : opTime.ts;
-      const opTimeT = typeof opTime.t === 'number' ? Long.fromNumber(opTime.t) : opTime.t;
-
-      // Compare the opTime's
-      if (bulkResult.lastOp == null) {
-        bulkResult.lastOp = opTime;
-      } else if (opTimeTS.greaterThan(lastOpTS)) {
+      // If there's no lastOp, just set it.
+      if (!bulkResult.lastOp) {
         bulkResult.lastOp = opTime;
-      } else if (opTimeTS.equals(lastOpTS)) {
-        if (opTimeT.greaterThan(lastOpT)) {
+      } else {
+        // First compare the ts values and set if the opTimeTS value is greater.
+        const lastOpTS = longOrConvert(bulkResult.lastOp.ts);
+        const opTimeTS = longOrConvert(opTime.ts);
+        if (opTimeTS.greaterThan(lastOpTS)) {
           bulkResult.lastOp = opTime;
+        } else if (opTimeTS.equals(lastOpTS)) {
+          // If the ts values are equal, then compare using the t values.
+          const lastOpT = longOrConvert(bulkResult.lastOp.t);
+          const opTimeT = longOrConvert(opTime.t);
+          if (opTimeT.greaterThan(lastOpT)) {
+            bulkResult.lastOp = opTime;
+          }
         }
       }
     }
@@ -1387,6 +1393,7 @@ Object.defineProperty(BulkOperationBase.prototype, 'length', {
 module.exports = {
   Batch,
   BulkOperationBase,
+  mergeBatchResults,
   bson,
   INSERT: INSERT,
   UPDATE: UPDATE,

test/unit/bulk_write.test.js

@@ -2,7 +2,11 @@
 
 const expect = require('chai').expect;
 const mock = require('mongodb-mock-server');
-const BulkWriteResult = require('../../lib/bulk/common').BulkWriteResult;
+const Long = require('../../lib/core').BSON.Long;
+const Timestamp = require('../../lib/core').BSON.Timestamp;
+const common = require('../../lib/bulk/common');
+const BulkWriteResult = common.BulkWriteResult;
+const mergeBatchResults = common.mergeBatchResults;
 
 describe('Bulk Writes', function() {
   const test = {};
@@ -131,4 +135,48 @@ describe('Bulk Writes', function() {
 
     expect(() => result.insertedIds).to.not.throw();
   });
+
+  describe('#mergeBatchResults', function() {
+    context('when opTime is an object', function() {
+      context('when the lastOp is a Timestamp', function() {
+        const batch = [];
+        const bulkResult = {
+          ok: 1,
+          writeErrors: [],
+          writeConcernErrors: [],
+          insertedIds: [],
+          nInserted: 0,
+          nUpserted: 0,
+          nMatched: 0,
+          nModified: 0,
+          nRemoved: 1,
+          upserted: [],
+          lastOp: {
+            ts: 7020546605669417496,
+            t: 10
+          }
+        };
+        const result = {
+          n: 8,
+          nModified: 8,
+          opTime: Timestamp.fromNumber(8020546605669417496),
+          electionId: '7fffffff0000000000000028',
+          ok: 1,
+          $clusterTime: {
+            clusterTime: '7020546605669417498',
+            signature: {
+              hash: 'AAAAAAAAAAAAAAAAAAAAAAAAAAA=',
+              keyId: 0
+            }
+          },
+          operationTime: '7020546605669417498'
+        };
+
+        it('replaces the lastOp with the properly formatted timestamp', function() {
+          mergeBatchResults(batch, bulkResult, null, result);
+          expect(bulkResult.lastOp.t).to.equal(Long.ZERO);
+        });
+      });
+    });
+  });
 });