SERVER-25268 OplogBufferCollection enforces order of timestamps in documents passed to pushAllNonBlocking with invariant

2 files changed, 29 insertions, 7 deletions

diff --git a/src/mongo/db/repl/oplog_buffer_collection.cpp b/src/mongo/db/repl/oplog_buffer_collection.cpp
index 7b59bdcd3c6..fe27a6870f3 100644
--- a/src/mongo/db/repl/oplog_buffer_collection.cpp
+++ b/src/mongo/db/repl/oplog_buffer_collection.cpp
@@ -60,6 +60,7 @@ std::pair<BSONObj, Timestamp> OplogBufferCollection::addIdToDocument(const BSONO
     invariant(!orig.isEmpty());
     BSONObjBuilder bob;
     Timestamp ts = orig["ts"].timestamp();
+    invariant(!ts.isNull());
     bob.append("_id", ts);
     bob.append(kOplogEntryFieldName, orig);
     return std::pair<BSONObj, Timestamp>{bob.obj(), ts};
@@ -120,6 +121,7 @@ void OplogBufferCollection::pushAllNonBlocking(OperationContext* txn,
     Timestamp ts;
     std::transform(begin, end, docsToInsert.begin(), [&ts](const Value& value) {
         auto pair = addIdToDocument(value);
+        invariant(ts.isNull() || pair.second > ts);
         ts = pair.second;
         return pair.first;
     });
diff --git a/src/mongo/db/repl/oplog_buffer_collection_test.cpp b/src/mongo/db/repl/oplog_buffer_collection_test.cpp
index f3366f32c3c..af5a6c9e965 100644
--- a/src/mongo/db/repl/oplog_buffer_collection_test.cpp
+++ b/src/mongo/db/repl/oplog_buffer_collection_test.cpp
@@ -197,6 +197,12 @@ TEST_F(OplogBufferCollectionTest, addIdToDocumentChangesTimestampToId) {
     ASSERT_EQUALS(Timestamp(1, 1), testOpPair.second);
 }
 
+DEATH_TEST_F(OplogBufferCollectionTest,
+             addIdToDocumentWithMissingTimestampFieldTriggersInvariantFailure,
+             "Invariant failure !ts.isNull()") {
+    OplogBufferCollection::addIdToDocument(BSON("x" << 1));
+}
+
 TEST_F(OplogBufferCollectionTest, PushOneDocumentWithPushAllNonBlockingAddsDocument) {
     auto nss = makeNamespace(_agent);
     OplogBufferCollection oplogBuffer(_storageInterface, nss);
@@ -363,7 +369,7 @@ TEST_F(OplogBufferCollectionTest, PopAndPeekReturnDocumentsInOrder) {
 
     oplogBuffer.startup(_txn.get());
     const std::vector<BSONObj> oplog = {
-        makeOplogEntry(2), makeOplogEntry(1), makeOplogEntry(3),
+        makeOplogEntry(1), makeOplogEntry(2), makeOplogEntry(3),
     };
     ASSERT_EQUALS(oplogBuffer.getCount(), 0UL);
     oplogBuffer.pushAllNonBlocking(_txn.get(), oplog.begin(), oplog.end());
@@ -386,19 +392,19 @@ TEST_F(OplogBufferCollectionTest, PopAndPeekReturnDocumentsInOrder) {
 
     BSONObj doc;
     ASSERT_TRUE(oplogBuffer.peek(_txn.get(), &doc));
-    ASSERT_BSONOBJ_EQ(doc, oplog[1]);
+    ASSERT_BSONOBJ_EQ(doc, oplog[0]);
     ASSERT_EQUALS(oplogBuffer.getCount(), 3UL);
 
     ASSERT_TRUE(oplogBuffer.tryPop(_txn.get(), &doc));
-    ASSERT_BSONOBJ_EQ(doc, oplog[1]);
+    ASSERT_BSONOBJ_EQ(doc, oplog[0]);
     ASSERT_EQUALS(oplogBuffer.getCount(), 2UL);
 
     ASSERT_TRUE(oplogBuffer.peek(_txn.get(), &doc));
-    ASSERT_BSONOBJ_EQ(doc, oplog[0]);
+    ASSERT_BSONOBJ_EQ(doc, oplog[1]);
     ASSERT_EQUALS(oplogBuffer.getCount(), 2UL);
 
     ASSERT_TRUE(oplogBuffer.tryPop(_txn.get(), &doc));
-    ASSERT_BSONOBJ_EQ(doc, oplog[0]);
+    ASSERT_BSONOBJ_EQ(doc, oplog[1]);
     ASSERT_EQUALS(oplogBuffer.getCount(), 1UL);
 
     ASSERT_TRUE(oplogBuffer.peek(_txn.get(), &doc));
@@ -416,14 +422,14 @@ TEST_F(OplogBufferCollectionTest, LastObjectPushedReturnsNewestOplogEntry) {
 
     oplogBuffer.startup(_txn.get());
     const std::vector<BSONObj> oplog = {
-        makeOplogEntry(1), makeOplogEntry(3), makeOplogEntry(2),
+        makeOplogEntry(1), makeOplogEntry(2), makeOplogEntry(3),
     };
     ASSERT_EQUALS(oplogBuffer.getCount(), 0UL);
     oplogBuffer.pushAllNonBlocking(_txn.get(), oplog.begin(), oplog.end());
     ASSERT_EQUALS(oplogBuffer.getCount(), 3UL);
 
     auto doc = oplogBuffer.lastObjectPushed(_txn.get());
-    ASSERT_BSONOBJ_EQ(*doc, oplog[1]);
+    ASSERT_BSONOBJ_EQ(*doc, oplog[2]);
     ASSERT_EQUALS(oplogBuffer.getCount(), 3UL);
 }
 
@@ -650,6 +656,20 @@ TEST_F(OplogBufferCollectionTest, PushEvenIfFullPushesOnSentinelsProperly) {
 }
 
 DEATH_TEST_F(OplogBufferCollectionTest,
+             PushAllNonBlockingWithOutOfOrderDocumentsTriggersInvariantFailure,
+             "Invariant failure ts.isNull() || pair.second > ts") {
+    auto nss = makeNamespace(_agent);
+    OplogBufferCollection oplogBuffer(_storageInterface, nss);
+
+    oplogBuffer.startup(_txn.get());
+    const std::vector<BSONObj> oplog = {
+        makeOplogEntry(2), makeOplogEntry(1),
+    };
+    ASSERT_EQUALS(oplogBuffer.getCount(), 0UL);
+    oplogBuffer.pushAllNonBlocking(_txn.get(), oplog.begin(), oplog.end());
+}
+
+DEATH_TEST_F(OplogBufferCollectionTest,
              PushAllNonBlockingWithSentinelTriggersInvariantFailure,
              "Invariant failure !orig.isEmpty()") {
     auto nss = makeNamespace(_agent);