SERVER-35798 preallocate prepare timestamp

2 files changed, 257 insertions, 7 deletions

diff --git a/jstests/core/txns/prepare_conflict.js b/jstests/core/txns/prepare_conflict.js
index 5fc81d0cdff..870af78909a 100644
--- a/jstests/core/txns/prepare_conflict.js
+++ b/jstests/core/txns/prepare_conflict.js
@@ -32,9 +32,16 @@
         assert(prepareConflicted);
     }
 
-    // Insert the document.
-    const doc1 = {_id: 1, x: 1};
-    assert.commandWorked(testColl.insert(doc1));
+    // Insert a document modified by the transaction.
+    const txnDoc = {_id: 1, x: 1};
+    assert.commandWorked(testColl.insert(txnDoc));
+
+    // Insert a document unmodified by the transaction.
+    const otherDoc = {_id: 2, y: 2};
+    assert.commandWorked(testColl.insert(otherDoc));
+
+    // Create an index on 'y' to avoid conflicts on the field.
+    assert.commandWorked(testColl.createIndex({y: 1}));
 
     // Enable the profiler to log slow queries. We expect a 'find' to hang until the prepare
     // conflict is resolved.
@@ -45,19 +52,31 @@
     session.startTransaction({readConcern: {level: "snapshot"}});
     assert.commandWorked(sessionDB.runCommand({
         update: collName,
-        updates: [{q: doc1, u: {$inc: {x: 1}}}],
+        updates: [{q: txnDoc, u: {$inc: {x: 1}}}],
     }));
 
     assert.commandWorked(sessionDB.adminCommand({prepareTransaction: 1}));
-    assertPrepareConflict({_id: 1});
+
+    // Conflict on _id of prepared document.
+    assertPrepareConflict({_id: txnDoc._id});
+
+    // Conflict on field that could be added to a prepared document.
+    assertPrepareConflict({randomField: "random"});
+
+    // No conflict on _id of a non-prepared document.
+    assert.commandWorked(testDB.runCommand({find: collName, filter: {_id: otherDoc._id}}));
+
+    // No conflict on indexed field of a non-prepared document.
+    assert.commandWorked(testDB.runCommand({find: collName, filter: {y: otherDoc.y}}));
 
     // At this point, we can guarantee all subsequent reads will conflict. Do a read in a parallel
     // shell, abort the transaction, then ensure the read succeeded with the old document.
     TestData.collName = collName;
     TestData.dbName = dbName;
+    TestData.txnDoc = txnDoc;
     const findAwait = startParallelShell(function() {
         const it = db.getSiblingDB(TestData.dbName)
-                       .runCommand({find: TestData.collName, filter: {_id: 1}});
+                       .runCommand({find: TestData.collName, filter: {_id: TestData.txnDoc._id}});
     }, db.getMongo().port);
 
     session.abortTransaction();
@@ -66,5 +85,5 @@
     findAwait({checkExitSuccess: true});
 
     // The document should be unmodified, because we aborted.
-    assert.eq(doc1, testColl.findOne(doc1));
+    assert.eq(txnDoc, testColl.findOne(txnDoc));
 })();
diff --git a/jstests/core/txns/timestamped_reads_wait_for_prepare_oplog_visibility.js b/jstests/core/txns/timestamped_reads_wait_for_prepare_oplog_visibility.js
new file mode 100644
index 00000000000..74671ee8439
--- /dev/null
+++ b/jstests/core/txns/timestamped_reads_wait_for_prepare_oplog_visibility.js
@@ -0,0 +1,231 @@
+/**
+ * Tests that timestamped reads, reads with snapshot and afterClusterTime, wait for the prepare
+ * transaction oplog entry to be visible before choosing a read timestamp.
+ *
+ * @tags: [uses_transactions]
+ */
+(function() {
+    'use strict';
+    load("jstests/libs/check_log.js");
+    load('jstests/core/txns/libs/prepare_helpers.js');
+    load('jstests/libs/parallel_shell_helpers.js');
+
+    TestData.dbName = 'test';
+    const baseCollName = 'timestamped_reads_wait_for_prepare_oplog_visibility';
+    const testDB = db.getSiblingDB(TestData.dbName);
+    TestData.failureTimeout = 1 * 1000;       // 1 second.
+    TestData.successTimeout = 5 * 60 * 1000;  // 5 minutes.
+    TestData.txnDoc = {_id: 1, x: 1};
+    TestData.otherDoc = {_id: 2, y: 7};
+    TestData.txnDocFilter = {_id: TestData.txnDoc._id};
+    TestData.otherDocFilter = {_id: TestData.otherDoc._id};
+
+    /**
+     * A function that accepts a 'readFunc' and a collection name. 'readFunc' accepts a collection
+     * name and returns an object with an 'oplogVisibility' test field and a 'prepareConflict' test
+     * field. This function is run in a separate thread and tests that oplog visibility blocks
+     * certain reads and that prepare conflicts block other types of reads.
+     */
+    const readThreadFunc = function(readFunc, _collName) {
+        load("jstests/libs/check_log.js");
+
+        // Do not start reads until we are blocked in 'prepareTransaction'.
+        checkLog.contains(db.getMongo(), "hangAfterReservingPrepareTimestamp fail point enabled");
+
+        // Create a 'readFuncObj' from the 'readFunc'.
+        const readFuncObj = readFunc(_collName);
+        readFuncObj.oplogVisibility();
+
+        // Let the transaction finish preparing and wait for 'prepareTransaction' to complete.
+        assert.commandWorked(db.adminCommand(
+            {configureFailPoint: 'hangAfterReservingPrepareTimestamp', mode: 'off'}));
+        checkLog.contains(db.getMongo(), "command: prepareTransaction");
+
+        readFuncObj.prepareConflict();
+    };
+
+    function runTest(prefix, readFunc) {
+        // Reset the log history between tests.
+        assert.commandWorked(db.adminCommand({clearLog: 'global'}));
+
+        jsTestLog('Testing oplog visibility for ' + prefix);
+        const collName = baseCollName + '_' + prefix;
+        const testColl = testDB.getCollection(collName);
+
+        testColl.drop({writeConcern: {w: "majority"}});
+        assert.commandWorked(testDB.runCommand({create: collName, writeConcern: {w: 'majority'}}));
+
+        assert.commandWorked(testDB.adminCommand(
+            {configureFailPoint: 'hangAfterReservingPrepareTimestamp', mode: 'alwaysOn'}));
+
+        // Insert a document for the transaction.
+        assert.commandWorked(testColl.insert(TestData.txnDoc));
+        // Insert a document untouched by the transaction.
+        assert.commandWorked(testColl.insert(TestData.otherDoc));
+
+        // Start a transaction with a single update on the 'txnDoc'.
+        const session = db.getMongo().startSession({causalConsistency: false});
+        const sessionDB = session.getDatabase(TestData.dbName);
+        session.startTransaction({readConcern: {level: 'snapshot'}});
+        assert.commandWorked(sessionDB[collName].update(TestData.txnDoc, {$inc: {x: 1}}));
+
+        // We set the log level up to know when 'prepareTransaction' completes.
+        db.setLogLevel(1);
+
+        // Clear the log history to ensure we only see the most recent 'prepareTransaction'
+        // failpoint log message.
+        assert.commandWorked(db.adminCommand({clearLog: 'global'}));
+        const joinReadThread = startParallelShell(funWithArgs(readThreadFunc, readFunc, collName));
+
+        jsTestLog("Preparing the transaction for " + prefix);
+        const prepareTimestamp = PrepareHelpers.prepareTransaction(session);
+
+        db.setLogLevel(0);
+        joinReadThread({checkExitSuccess: true});
+
+        PrepareHelpers.commitTransaction(session, prepareTimestamp);
+    }
+
+    const snapshotRead = function(_collName) {
+        const _db = db.getSiblingDB(TestData.dbName);
+
+        const session = db.getMongo().startSession({causalConsistency: false});
+        const sessionDB = session.getDatabase(TestData.dbName);
+        session.startTransaction({readConcern: {level: 'snapshot'}});
+
+        const oplogVisibility = function() {
+            jsTestLog("Snapshot reads should block on oplog visibility.");
+            assert.commandFailedWithCode(sessionDB.runCommand({
+                find: _collName,
+                filter: TestData.txnDocFilter,
+                maxTimeMS: TestData.failureTimeout
+            }),
+                                         ErrorCodes.MaxTimeMSExpired);
+            assert.commandFailedWithCode(sessionDB.runCommand({
+                find: _collName,
+                filter: TestData.otherDocFilter,
+                maxTimeMS: TestData.failureTimeout
+            }),
+                                         ErrorCodes.MaxTimeMSExpired);
+        };
+
+        const prepareConflict = function() {
+            jsTestLog("Snapshot reads should block on prepared transactions for " +
+                      "conflicting documents.");
+            assert.commandFailedWithCode(sessionDB.runCommand({
+                find: _collName,
+                filter: TestData.txnDocFilter,
+                maxTimeMS: TestData.failureTimeout
+            }),
+                                         ErrorCodes.MaxTimeMSExpired);
+
+            jsTestLog("Snapshot reads should succeed on non-conflicting documents while a " +
+                      "transaction is in prepare.");
+            let cursor = assert.commandWorked(sessionDB.runCommand({
+                find: _collName,
+                filter: TestData.otherDocFilter,
+                maxTimeMS: TestData.successTimeout
+            }));
+            assert.docEq(cursor.cursor.firstBatch, [TestData.otherDoc], tojson(cursor));
+        };
+
+        return {oplogVisibility: oplogVisibility, prepareConflict: prepareConflict};
+    };
+
+    const afterClusterTime = function(_collName) {
+        const _db = db.getSiblingDB(TestData.dbName);
+
+        // Advance the cluster time with an arbitrary other insert.
+        let res = assert.commandWorked(
+            _db.runCommand({insert: _collName, documents: [{advanceClusterTime: 1}]}));
+        assert(res.hasOwnProperty("$clusterTime"), tojson(res));
+        assert(res.$clusterTime.hasOwnProperty("clusterTime"), tojson(res));
+        const clusterTime = res.$clusterTime.clusterTime;
+        jsTestLog("Using afterClusterTime: " + clusterTime);
+
+        const oplogVisibility = function() {
+            jsTestLog("afterClusterTime reads should block on oplog visibility.");
+            assert.commandFailedWithCode(_db.runCommand({
+                find: _collName,
+                filter: TestData.txnDocFilter,
+                readConcern: {afterClusterTime: clusterTime},
+                maxTimeMS: TestData.failureTimeout
+            }),
+                                         ErrorCodes.MaxTimeMSExpired);
+            assert.commandFailedWithCode(_db.runCommand({
+                find: _collName,
+                filter: TestData.otherDocFilter,
+                readConcern: {afterClusterTime: clusterTime},
+                maxTimeMS: TestData.failureTimeout
+            }),
+                                         ErrorCodes.MaxTimeMSExpired);
+        };
+
+        const prepareConflict = function() {
+            jsTestLog("afterClusterTime reads should block on prepared transactions for " +
+                      "conflicting documents.");
+            assert.commandFailedWithCode(_db.runCommand({
+                find: _collName,
+                filter: TestData.txnDocFilter,
+                readConcern: {afterClusterTime: clusterTime},
+                maxTimeMS: TestData.failureTimeout
+            }),
+                                         ErrorCodes.MaxTimeMSExpired);
+
+            jsTestLog("afterClusterTime reads should succeed on non-conflicting documents " +
+                      "while transaction is in prepare.");
+            let cursor = assert.commandWorked(_db.runCommand({
+                find: _collName,
+                filter: TestData.otherDocFilter,
+                readConcern: {afterClusterTime: clusterTime},
+                maxTimeMS: TestData.successTimeout
+            }));
+            assert.docEq(cursor.cursor.firstBatch, [TestData.otherDoc], tojson(cursor));
+        };
+
+        return {oplogVisibility: oplogVisibility, prepareConflict: prepareConflict};
+    };
+
+    const normalRead = function(_collName) {
+        const _db = db.getSiblingDB(TestData.dbName);
+
+        const oplogVisibility = function() {
+            jsTestLog("Ordinary reads should not block on oplog visibility.");
+            let cursor = assert.commandWorked(_db.runCommand({
+                find: _collName,
+                filter: TestData.txnDocFilter,
+                maxTimeMS: TestData.successTimeout
+            }));
+            assert.docEq(cursor.cursor.firstBatch, [TestData.txnDoc], tojson(cursor));
+            cursor = assert.commandWorked(_db.runCommand({
+                find: _collName,
+                filter: TestData.otherDocFilter,
+                maxTimeMS: TestData.successTimeout
+            }));
+            assert.docEq(cursor.cursor.firstBatch, [TestData.otherDoc], tojson(cursor));
+        };
+
+        const prepareConflict = function() {
+            jsTestLog("Ordinary reads should not block on prepared transactions.");
+            // TODO (SERVER-36382): Uncomment this block when local reads don't cause prepare
+            // conflicts.
+            // cursor = assert.commandWorked(_db.runCommand(
+            //    {find: _collName, filter: TestData.txnDocFilter, maxTimeMS:
+            //  TestData.successTimeout}));
+            // assert.docEq(cursor.cursor.firstBatch, [TestData.txnDoc], tojson(cursor));
+            let cursor = assert.commandWorked(_db.runCommand({
+                find: _collName,
+                filter: TestData.otherDocFilter,
+                maxTimeMS: TestData.successTimeout
+            }));
+            assert.docEq(cursor.cursor.firstBatch, [TestData.otherDoc], tojson(cursor));
+        };
+
+        return {oplogVisibility: oplogVisibility, prepareConflict: prepareConflict};
+    };
+
+    runTest('normal_reads', normalRead);
+    // TODO (SERVER-35821): Unblacklist this snapshot reads test.
+    // runTest('snapshot_reads', snapshotRead);
+    runTest('afterClusterTime', afterClusterTime);
+})();