mongo/jstests/replsets/primary_commit_split_prepare_transactions.js

/**
 * Tests split prepared transaction commit on primary.
 *
 * This test does the following:
 * 1) Prepare a transaction on the old primary.
 * 2) Step up the secondary as the new primary after it applies the prepared transaction.
 * 3) Commit the (split) prepared transaction on the new primary, but block it on a failpoint part
 *    way through.
 * 4) Test that reads (under certain read concerns) would be blocked on prepare conflicts.
 * 5) Restart the new primary node and wait for it to be a primary again.
 * 6) Commit the (split) prepared transaction again.
 * And it also tests the integrity of the prepared transaction along the way.
 *
 * @tags: [
 *   requires_fcv_70,
 *   uses_transactions,
 *   uses_prepare_transaction,
 *]
 */
import {PrepareHelpers} from "jstests/core/txns/libs/prepare_helpers.js";
import {configureFailPoint} from "jstests/libs/fail_point_util.js";
import {funWithArgs} from "jstests/libs/parallel_shell_helpers.js";
import {ReplSetTest} from "jstests/libs/replsettest.js";

// Verify that the documents updated in the transaction are found or not, depending on expectOld.
const checkDocuments = function (docCount, testColl, expectOld, readConcern = null) {
    for (let i = 1; i <= docCount; ++i) {
        const doc = {_id: i, x: 1};
        const expected = expectOld ? doc : {_id: i, x: 1, y: 1};

        assert.eq(expected, testColl.findOne(doc), {}, {}, readConcern);
    }
};

const replTest = new ReplSetTest({
    nodes: 2,
    nodeOptions: {
        // Set the 'syncdelay' to 1s to speed up checkpointing.
        syncdelay: 1,
        setParameter: {
            logComponentVerbosity: tojsononeline({replication: 3, command: 2}),
        },
    },
});
replTest.startSet();

// initiate makes replTest.waitForPrimary() below very slow
// and adding a replTest.stepUp(primary) does not help.
replTest.initiate(null, null, {initiateWithDefaultElectionTimeout: true});

const primary = replTest.getPrimary();
const secondary = replTest.getSecondary();

const dbName = "test";
const collName = jsTestName();
let testDB = primary.getDB(dbName);
let testColl = testDB.getCollection(collName);

// Verify that we can't insert in the transaction if it is in prepared/committed state.
// Also checks the config.transactions entry.
const checkTransaction = function (sessionDB, lsid, txnNumber, expectedState) {
    const expectedError =
        expectedState == "prepared" ? ErrorCodes.PreparedTransactionInProgress : ErrorCodes.TransactionCommitted;
    assert.commandFailedWithCode(
        sessionDB.runCommand({
            insert: collName,
            documents: [{x: 2}],
            txnNumber: NumberLong(txnNumber),
            autocommit: false,
        }),
        expectedError,
    );

    const res = replTest
        .getPrimary()
        .getDB("config")
        .getCollection("transactions")
        .find({"_id.id": lsid["id"], "txnNum": txnNumber})
        .readConcern("majority")
        .toArray();
    assert.eq(1, res.length);
    assert.eq(expectedState, res[0]["state"]);
};

testColl.drop({writeConcern: {w: "majority"}});
assert.commandWorked(testDB.runCommand({create: collName, writeConcern: {w: "majority"}}));

let session = primary.startSession({causalConsistency: false});
const lsid = session.getSessionId();
let sessionDB = session.getDatabase(dbName);
let sessionColl = sessionDB.getCollection(collName);

jsTestLog("Inserting documents before the transaction.");

const docCount = 100;
for (let i = 1; i <= docCount; ++i) {
    assert.commandWorked(testColl.insert({_id: i, x: 1}));
}

session.startTransaction();
const txnNumber = session.getTxnNumber_forTesting();

jsTestLog("Updating documents in the transaction.");

for (let i = 1; i <= docCount; ++i) {
    assert.commandWorked(sessionColl.update({_id: i}, {$set: {y: 1}}));
}

// Updates should not be visible outside the session.
checkDocuments(docCount, testColl, true /* expectOld */);

// Updates should be visible in this session.
checkDocuments(docCount, sessionColl, false /* expectOld */);

// 1) Prepare a transaction on the old primary.
jsTestLog("Preparing the transaction.");
const prepareTimestamp = PrepareHelpers.prepareTransaction(session);

// Wait until lastStableRecoveryTimestamp >= prepareTimestamp on all nodes.
assert.soon(() => {
    const primaryLastStableRecoveryTimestamp = assert.commandWorked(primary.adminCommand({replSetGetStatus: 1}))[
        "lastStableRecoveryTimestamp"
    ];
    const secondaryLastStableRecoveryTimestamp = assert.commandWorked(secondary.adminCommand({replSetGetStatus: 1}))[
        "lastStableRecoveryTimestamp"
    ];
    jsTestLog(
        "Awaiting last stable recovery timestamp " +
            `(primary last stable recovery: ${tojson(primaryLastStableRecoveryTimestamp)}, ` +
            `secondary last stable recovery: ${tojson(secondaryLastStableRecoveryTimestamp)}) ` +
            `prepareTimestamp: ${tojson(prepareTimestamp)}`,
    );
    return (
        timestampCmp(primaryLastStableRecoveryTimestamp, prepareTimestamp) >= 0 &&
        timestampCmp(secondaryLastStableRecoveryTimestamp, prepareTimestamp) >= 0
    );
});

checkTransaction(sessionDB, lsid, txnNumber, "prepared");

// 2) Step up the secondary as the new primary after it applies the prepared transaction.
jsTestLog("Forcing secondary to become primary.");
replTest.stepUp(secondary);
replTest.waitForState(secondary, ReplSetTest.State.PRIMARY);
const newPrimary = replTest.getPrimary();
assert.eq(newPrimary, secondary);

// Freezing the old primary so it will no longer be a primary.
jsTestLog("Freezing old primary node.");
assert.commandWorked(primary.adminCommand({replSetFreeze: ReplSetTest.kForeverSecs}));

// Force the session to use the same lsid and txnNumber as from before the stepUp.
// This ensures that we're working with the same session and transaction.
session = PrepareHelpers.createSessionWithGivenId(newPrimary, lsid);
session.setTxnNumber_forTesting(txnNumber);
sessionDB = session.getDatabase(dbName);

checkTransaction(sessionDB, lsid, txnNumber, "prepared");

// Inserts are not seen outside the transaction.
checkDocuments(docCount, testColl, true /* expectOld */);

testDB = newPrimary.getDB(dbName);
testColl = testDB.getCollection(collName);

// 3) Commit the (split) prepared transaction on the new primary, but block it on a failpoint part
//    way through.
jsTestLog("Committing transaction (with failpoint to pause split transaction commit).");

// Now the new primary will have to commit the split prepared transaction,
// so it will enter _commitSplitPreparedTxnOnPrimary.

// Set the failpoint with skip so part of the split transaction is committed and part is not.
const failPointName = "hangInCommitSplitPreparedTxnOnPrimary";
const failPoint = configureFailPoint(testDB, failPointName, {}, {skip: 2});

const commitTxnFunc = async function (dbName, prepareTimestamp, lsid, txnNumber) {
    const {PrepareHelpers} = await import("jstests/core/txns/libs/prepare_helpers.js");
    const session = PrepareHelpers.createSessionWithGivenId(db.getMongo(), lsid);
    session.setTxnNumber_forTesting(txnNumber);
    const sessionDB = session.getDatabase(dbName);

    const err = assert.throws(() =>
        sessionDB.adminCommand({
            commitTransaction: 1,
            commitTimestamp: prepareTimestamp,
            txnNumber: txnNumber,
            autocommit: false,
        }),
    );

    assert(isNetworkError(err), tojson(err));
};

const awaitCommitTransaction = startParallelShell(
    funWithArgs(commitTxnFunc, dbName, prepareTimestamp, lsid, txnNumber),
    newPrimary.port,
);

failPoint.wait();

jsTestLog("Transaction is blocked on failpoint in the middle of a split transaction commit.");

// 4) Test that reads (under certain read concerns) would be blocked on prepare conflicts.
{
    const shortTimeout = 1 * 1000; // 1 second.
    const longTimeout = ReplSetTest.kForeverMillis;
    const read = function (readConcern, timeout) {
        return testDB.runCommand({
            find: collName,
            filter: {y: 1},
            readConcern: readConcern,
            maxTimeMS: timeout,
        });
    };

    jsTestLog("Test read with read concern 'local' doesn't block on prepared conflicts.");
    assert.commandWorked(read({level: "local"}, longTimeout));

    jsTestLog("Test read with read concern 'majority' doesn't block on prepared conflicts.");
    assert.commandWorked(read({level: "majority"}, longTimeout));

    jsTestLog("Test read with read concern 'linearizable' blocks on prepared conflicts.");
    assert.commandFailedWithCode(read({level: "linearizable"}, shortTimeout), ErrorCodes.MaxTimeMSExpired);

    jsTestLog("Test afterClusterTime read after prepareTimestamp blocks on prepare conflicts.");
    assert.commandFailedWithCode(
        read({level: "local", afterClusterTime: prepareTimestamp}, shortTimeout),
        ErrorCodes.MaxTimeMSExpired,
    );

    jsTestLog("Test snapshot read at prepareTimestamp blocks on prepare conflicts.");
    assert.commandFailedWithCode(
        read({level: "snapshot", atClusterTime: prepareTimestamp}, shortTimeout),
        ErrorCodes.MaxTimeMSExpired,
    );
}

// 5) Restart the new primary node and wait for it to be a primary again.
jsTestLog("Restarting the primary node");
// Restart newPrimary.
replTest.stop(newPrimary, 9 /* signal */, {forRestart: true, allowedExitCode: MongoRunner.EXIT_SIGKILL});
replTest.start(newPrimary, {waitForConnect: true}, true /* waitForHealth */);
jsTestLog("Restarted the primary node");

// Join the parallel thread.
awaitCommitTransaction();

// The restarted newPrimary node should be elected as the primary again since the other node (old
// primary) was set a high freeze timeout.
replTest.waitForPrimary();
replTest.awaitSecondaryNodes();

const newPrimary2 = replTest.getPrimary();
assert.eq(newPrimary2, newPrimary);

session = PrepareHelpers.createSessionWithGivenId(newPrimary2, lsid);
session.setTxnNumber_forTesting(txnNumber);
sessionDB = session.getDatabase(dbName);

checkTransaction(sessionDB, lsid, txnNumber, "prepared");

testDB = newPrimary2.getDB(dbName);
testColl = testDB.getCollection(collName);
const secondaryTestDB = replTest.getSecondary().getDB(dbName);
const secondaryTestColl = secondaryTestDB.getCollection(collName);

// Updates are not seen outside the transaction.
checkDocuments(docCount, testColl, true /* expectOld */, "local" /* readConcern */);
checkDocuments(docCount, secondaryTestColl, true /* expectOld */, "local" /* readConcern */);

testDB = newPrimary2.getDB(dbName);
testColl = testDB.getCollection(collName);

// 6) Commit the (split) prepared transaction again.
jsTestLog("Committing transaction (this one is expected to succeed)");
assert.commandWorked(
    sessionDB.adminCommand({
        commitTransaction: 1,
        commitTimestamp: prepareTimestamp,
        txnNumber: txnNumber,
        autocommit: false,
    }),
);

checkTransaction(sessionDB, lsid, txnNumber, "committed");

// After commit the updates become visible.
checkDocuments(docCount, testColl, false /* expectOld */);

replTest.stopSet();