/**
 * Tests that config.transactions documents are correctly modified on FCV upgrade/downgrade and that
 * retryability is preserved for transactions that don't use prepare after upgrade and always for
 * retryable writes.
 *
 * @tags: [uses_transactions, uses_prepare_transaction]
 */
(function() {
"use strict";
load("jstests/libs/feature_compatibility_version.js");
load('jstests/sharding/libs/sharded_transactions_helpers.js');

const dbName = "test";
const collName = "config_transactions_set_fcv";

// Define autocommit as a variable so it can be used in object literals w/o an explicit value.
const autocommit = false;

// Start a replica set with an odd number of members to verify nodes outside the majority behave
// correctly around setFeatureCompatibilityVersion, which uses majority writes to update the FCV
// document. The primary isn't expected to change, so each secondary is given priority 0.
const rst = new ReplSetTest({nodes: [{}, {rsConfig: {priority: 0}}, {rsConfig: {priority: 0}}]});
rst.startSet();
rst.initiate();

let testDB = rst.getPrimary().getDB(dbName);
let adminDB = rst.getPrimary().getDB("admin");

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

// Starts a dummy transaction, commits or aborts it with or without prepare, then returns the
// commit or abort response. Returns the response from prepare if it fails.
function runTxn({lsid, txnNumber}, {commit, prepare, leaveOpen}) {
    const startTransactionRes = testDB.runCommand({
        insert: collName,
        documents: [{x: "dummy_txn"}],
        txnNumber: NumberLong(txnNumber),
        startTransaction: true,
        lsid,
        autocommit,
    });
    if (!startTransactionRes.ok || leaveOpen) {
        return startTransactionRes;
    }

    if (prepare) {
        const prepareRes = testDB.adminCommand({
            prepareTransaction: 1,
            txnNumber: NumberLong(txnNumber),
            lsid,
            autocommit,
            writeConcern: {w: "majority"}
        });
        if (!prepareRes.ok) {
            return prepareRes;
        }

        if (commit) {
            // Add 1 to the increment so that the commitTimestamp is after the prepareTimestamp.
            const commitTimestamp = Timestamp(prepareRes.prepareTimestamp.getTime(),
                                              prepareRes.prepareTimestamp.getInc() + 1);
            return testDB.adminCommand({
                commitTransaction: 1,
                commitTimestamp,
                txnNumber: NumberLong(txnNumber),
                lsid,
                autocommit
            });
        } else {
            return testDB.adminCommand(
                {abortTransaction: 1, txnNumber: NumberLong(txnNumber), lsid, autocommit});
        }
    }

    if (commit) {
        return testDB.adminCommand(
            {commitTransaction: 1, txnNumber: NumberLong(txnNumber), lsid, autocommit});
    } else {
        return testDB.adminCommand(
            {abortTransaction: 1, txnNumber: NumberLong(txnNumber), lsid, autocommit});
    }
}

// Retries commitTransaction for the given txnId, returning the response.
function retryCommit({lsid, txnNumber}) {
    return testDB.adminCommand(
        {commitTransaction: 1, txnNumber: NumberLong(txnNumber), lsid, autocommit});
}

// Asserts aborting the given txnId returns NoSuchTransaction.
function assertTransactionAborted({lsid, txnNumber}) {
    assert.commandFailedWithCode(
        testDB.adminCommand(
            {abortTransaction: 1, txnNumber: NumberLong(txnNumber), lsid, autocommit}),
        ErrorCodes.NoSuchTransaction);
}

// Global counter for the number of retryable writes completed. Used to verify retried retryable
// writes aren't double applied.
let numRetryableWrites = 0;

// Runs a dummy retryable write and increments the retryable write counter.
function assertRetryableWriteWorked({lsid, txnNumber}) {
    numRetryableWrites += 1;
    assert.commandWorked(testDB.runCommand({
        insert: collName,
        documents: [{fromRetryableWrite: true}],
        txnNumber: NumberLong(txnNumber),
        lsid
    }));
}

// Verifies a txnId has already been used for a retryable write by running a dummy retryable
// write and asserting the write isn't applied.
function assertRetryableWriteCanBeRetried({lsid, txnNumber}) {
    assert.commandWorked(testDB.runCommand({
        insert: collName,
        documents: [{fromRetryableWrite: true}],
        txnNumber: NumberLong(txnNumber),
        lsid
    }));
    assert.eq(numRetryableWrites, testDB[collName].find({fromRetryableWrite: true}).itcount());
}

// Searches config.transactions for an entry for the given txnId on each node in the replica
// set, verifying the entry does / does not exist and has the expected state, if specified.
function checkConfigTransactionEntry(rst, {lsid, txnNumber}, {hasEntry, expectedState}) {
    rst.awaitReplication();
    rst.nodes.forEach((node) => {
        // Search for id since we don't know the uid, which is generated by the server.
        const entry = node.getDB("config").transactions.findOne({"_id.id": lsid.id});

        if (!hasEntry) {
            // There should be no entry for this session or it should be for an earlier
            // operation.
            if (entry) {
                assert.gt(txnNumber,
                          entry.txnNum,
                          "expected entry to have lower txnNumber, entry: " + tojson(entry) +
                              ", node: " + tojson(node));
            } else {
                assert.isnull(entry,
                              "expected entry to be null, entry: " + tojson(entry) +
                                  ", node: " + tojson(node));
            }
            return;
        }

        assert.eq(txnNumber,
                  entry.txnNum,
                  "expected entry to have the same txnNumber, entry: " + tojson(entry) +
                      ", node: " + tojson(node));

        if (expectedState) {
            assert.eq(
                expectedState, entry.state, "entry: " + tojson(entry) + ", node: " + tojson(node));
        } else {
            assert(!entry.hasOwnProperty("state"),
                   "expected entry to not have state, entry: " + tojson(entry) +
                       ", node: " + tojson(node));
        }
    });
}

function runTest({shouldRestart}) {
    // The test waits for failpoints to log a message when hit, so clear the program output
    // before starting so messages from previous iterations aren't in it.
    clearRawMongoProgramOutput();

    const txnIds = {
        write: {lsid: {id: UUID()}, txnNumber: 0},          // Retryable write.
        commit: {lsid: {id: UUID()}, txnNumber: 0},         // Committed transaction w/o prepare.
        commitPrepare: {lsid: {id: UUID()}, txnNumber: 0},  // Committed transaction w/ prepare.
        abort: {lsid: {id: UUID()}, txnNumber: 0},          // Aborted transaction w/o prepare.
        abortPrepare: {lsid: {id: UUID()}, txnNumber: 0},   // Aborted transaction after prepare.
        concurrentTxn: {lsid: {id: UUID()}, txnNumber: 0},  // Transaction concurrent w/ setFCV.
        concurrentWrite:
            {lsid: {id: UUID()}, txnNumber: 0},  // Retryable write concurrent w/ setFCV.
        upgradingTxn:
            {lsid: {id: UUID()}, txnNumber: 0},  // Transaction started during FCV upgrade.
    };

    //
    // In the latest FCV, verify the expected updates are made to config.transactions for each
    // case and the successful operations are retryable.
    //
    checkFCV(adminDB, latestFCV);

    assertRetryableWriteWorked(txnIds.write);
    assert.commandWorked(runTxn(txnIds.commit, {commit: true, prepare: false}));
    assert.commandWorked(runTxn(txnIds.commitPrepare, {commit: true, prepare: true}));
    assert.commandWorked(runTxn(txnIds.abort, {commit: false, prepare: false}));
    assert.commandWorked(runTxn(txnIds.abortPrepare, {commit: false, prepare: true}));

    checkConfigTransactionEntry(rst, txnIds.write, {hasEntry: true});
    checkConfigTransactionEntry(rst, txnIds.commit, {hasEntry: true, expectedState: "committed"});
    checkConfigTransactionEntry(
        rst, txnIds.commitPrepare, {hasEntry: true, expectedState: "committed"});
    checkConfigTransactionEntry(rst, txnIds.abort, {hasEntry: false});
    checkConfigTransactionEntry(
        rst, txnIds.abortPrepare, {hasEntry: true, expectedState: "aborted"});

    // The retryable write and the commit of both committed transactions should be retryable.
    // The aborted transactions should still be aborted.
    assertRetryableWriteCanBeRetried(txnIds.write);
    assert.commandWorked(retryCommit(txnIds.commit));
    assert.commandWorked(retryCommit(txnIds.commitPrepare));
    assertTransactionAborted(txnIds.abort);
    assertTransactionAborted(txnIds.abortPrepare);

    //
    // Downgrade to the last-stable FCV and verify config.transactions was updated as expected
    // for previously completed operations and operations concurrent with the downgrade.
    //

    if (shouldRestart) {
        // Restart to verify config.transactions entries for sessions not in-memory at the
        // beginning of FCV downgrade are updated correctly.
        jsTestLog("Restarting replica set before downgrading the featureCompatibilityVersion.");
        for (let i = 0; i < rst.nodes.length; i++) {
            rst.restart(i);
        }
        testDB = rst.getPrimary().getDB(dbName);
        adminDB = rst.getPrimary().getDB("admin");
    }

    // Make setFCV pause in the downgrading state after getting the list of sessions to
    // potentially modify.
    assert.commandWorked(rst.getPrimary().adminCommand(
        {configureFailPoint: "pauseBeforeDowngradingSessions", mode: "alwaysOn"}));

    // Downgrade FCV in a parallel shell and wait until it blocks at the failpoint above.
    const awaitDowngradeFCV = startParallelShell(() => {
        load("jstests/libs/feature_compatibility_version.js");
        jsTestLog("Downgrade the featureCompatibilityVersion in a parallel shell.");
        assert.commandWorked(db.adminCommand({setFeatureCompatibilityVersion: lastStableFCV}));
    }, rst.getPrimary().port);
    waitForFailpoint("Hit pauseBeforeDowngradingSessions failpoint", 1 /*numTimes*/);

    // Concurrent transactions that use prepare will fail.
    assert.commandFailedWithCode(runTxn(txnIds.concurrentTxn, {commit: true, prepare: true}),
                                 ErrorCodes.CommandNotSupported);
    txnIds.concurrentTxn.txnNumber += 1;

    // Concurrent transactions that do not use prepare and retryable writes succeed.
    assert.commandWorked(runTxn(txnIds.concurrentTxn, {commit: true, prepare: false}));
    assertRetryableWriteWorked(txnIds.concurrentWrite);

    // Unset the failpoint and wait for the downgrade to finish.
    assert.commandWorked(rst.getPrimary().adminCommand(
        {configureFailPoint: "pauseBeforeDowngradingSessions", mode: "off"}));

    awaitDowngradeFCV();
    checkFCV(adminDB, lastStableFCV);

    // The successful concurrent operations should have entries without state and be retryable.
    checkConfigTransactionEntry(rst, txnIds.concurrentTxn, {hasEntry: true});
    assert.commandWorked(retryCommit(txnIds.concurrentTxn));
    checkConfigTransactionEntry(rst, txnIds.concurrentWrite, {hasEntry: true});
    assertRetryableWriteCanBeRetried(txnIds.concurrentWrite);

    // Only the retryable write entry should remain.
    checkConfigTransactionEntry(rst, txnIds.write, {hasEntry: true});
    checkConfigTransactionEntry(rst, txnIds.commit, {hasEntry: false});
    checkConfigTransactionEntry(rst, txnIds.commitPrepare, {hasEntry: false});
    checkConfigTransactionEntry(rst, txnIds.abort, {hasEntry: false});
    checkConfigTransactionEntry(rst, txnIds.abortPrepare, {hasEntry: false});

    // The retryable write can be retried.
    assertRetryableWriteCanBeRetried(txnIds.write);

    // Neither of the commits can be retried.
    assert.commandFailedWithCode(retryCommit(txnIds.commit), ErrorCodes.NoSuchTransaction);
    assert.commandFailedWithCode(retryCommit(txnIds.commitPrepare), ErrorCodes.NoSuchTransaction);

    //
    // In the last-stable FCV, verify the expected updates are made to config.transactions for
    // each case and the successful operations are retryable.
    //

    // Reset each txnId to test upgrade with a clean slate.
    Object.keys(txnIds).forEach((txnIdKey) => {
        txnIds[txnIdKey].lsid = {id: UUID()};
        txnIds[txnIdKey].txnNumber = 0;
    });

    // Prepare can't be used in FCV 4.0, so only commit, abort, and retryable write should
    // succeed.
    assertRetryableWriteWorked(txnIds.write);
    assert.commandWorked(runTxn(txnIds.commit, {commit: true, prepare: false}));
    assert.commandFailedWithCode(runTxn(txnIds.commitPrepare, {commit: true, prepare: true}),
                                 ErrorCodes.CommandNotSupported);
    assert.commandWorked(runTxn(txnIds.abort, {commit: false, prepare: false}));
    assert.commandFailedWithCode(runTxn(txnIds.abortPrepare, {commit: false, prepare: true}),
                                 ErrorCodes.CommandNotSupported);

    // Only the retryable write and transaction that committed without prepare should have an
    // entry. Neither should have state.
    checkConfigTransactionEntry(rst, txnIds.write, {hasEntry: true});
    checkConfigTransactionEntry(rst, txnIds.commit, {hasEntry: true});
    checkConfigTransactionEntry(rst, txnIds.commitPrepare, {hasEntry: false});
    checkConfigTransactionEntry(rst, txnIds.abort, {hasEntry: false});
    checkConfigTransactionEntry(rst, txnIds.abortPrepare, {hasEntry: false});

    // The retryable write and successful commit can be retried.
    assertRetryableWriteCanBeRetried(txnIds.write);
    assert.commandWorked(retryCommit(txnIds.commit));

    if (shouldRestart) {
        // Restart to verify config.transactions entries for sessions not in-memory at the
        // beginning of FCV upgrade are updated correctly.
        jsTestLog("Restarting replica set before upgrading the featureCompatibilityVersion.");
        for (let i = 0; i < rst.nodes.length; i++) {
            rst.restart(i);
        }
        testDB = rst.getPrimary().getDB(dbName);
        adminDB = rst.getPrimary().getDB("admin");
    }

    //
    // Upgrade to the latest FCV and verify config.transactions was updated as expected for
    // previously completed operations and operations concurrent with the upgrade.
    //

    // Run a retryable write on the session that will be used during upgrade so it has a
    // transaction table entry and will be checked out by the upgrade.
    assertRetryableWriteWorked(txnIds.upgradingTxn);
    txnIds.upgradingTxn.txnNumber += 1;

    // Make setFCV pause in the upgrading state after getting the list of sessions to
    // potentially modify.
    assert.commandWorked(rst.getPrimary().adminCommand(
        {configureFailPoint: "pauseBeforeUpgradingSessions", mode: "alwaysOn"}));

    // Upgrade FCV in a parallel shell and wait until it blocks at the failpoint above.
    const awaitUpgradeFCV = startParallelShell(() => {
        load("jstests/libs/feature_compatibility_version.js");
        jsTestLog("Upgrade the featureCompatibilityVersion in a parallel shell.");
        assert.commandWorked(db.adminCommand({setFeatureCompatibilityVersion: latestFCV}));
    }, rst.getPrimary().port);
    waitForFailpoint("Hit pauseBeforeUpgradingSessions failpoint", 1 /*numTimes*/);

    // Concurrent transactions that use prepare will fail.
    assert.commandFailedWithCode(runTxn(txnIds.concurrentTxn, {commit: true, prepare: true}),
                                 ErrorCodes.CommandNotSupported);
    txnIds.concurrentTxn.txnNumber += 1;

    // Concurrent transactions that do not use prepare and retryable writes succeed.
    assert.commandWorked(runTxn(txnIds.concurrentTxn, {commit: true, prepare: false}));
    assertRetryableWriteWorked(txnIds.concurrentWrite);

    // Start a transaction in the upgrading state and verify that it doesn't get aborted by the
    // rest of the upgrade. Note that all sessions are killed and their transactions aborted for
    // writes to the FCV document except when it is set to the fully upgraded state, so this
    // can't be tested for downgrade.
    assert.commandWorked(runTxn(txnIds.upgradingTxn, {leaveOpen: true}));

    // Unset the failpoint and wait for the upgrade to finish.
    assert.commandWorked(rst.getPrimary().adminCommand(
        {configureFailPoint: "pauseBeforeUpgradingSessions", mode: "off"}));

    awaitUpgradeFCV();
    checkFCV(adminDB, latestFCV);

    // The transaction started while upgrading shouldn't have been killed and can be committed.
    assert.commandWorked(testDB.adminCommand({
        commitTransaction: 1,
        lsid: txnIds.upgradingTxn.lsid,
        txnNumber: NumberLong(txnIds.upgradingTxn.txnNumber),
        autocommit
    }));

    // The successful concurrent transaction should have "committed" state and be retryable, and
    // the concurrent retryable write should not have state and also be retryable.
    checkConfigTransactionEntry(
        rst, txnIds.concurrentTxn, {hasEntry: true, expectedState: "committed"});
    assert.commandWorked(retryCommit(txnIds.concurrentTxn));
    checkConfigTransactionEntry(rst, txnIds.concurrentWrite, {hasEntry: true});
    assertRetryableWriteCanBeRetried(txnIds.concurrentWrite);

    // There should still only be entries for the committed transaction and retryable write. The
    // committed transaction should now have a "state" field.
    checkConfigTransactionEntry(rst, txnIds.write, {hasEntry: true});
    checkConfigTransactionEntry(rst, txnIds.commit, {hasEntry: true, expectedState: "committed"});
    checkConfigTransactionEntry(rst, txnIds.commitPrepare, {hasEntry: false});
    checkConfigTransactionEntry(rst, txnIds.abort, {hasEntry: false});
    checkConfigTransactionEntry(rst, txnIds.abortPrepare, {hasEntry: false});

    // The retryable write and successful commit can be retried.
    assertRetryableWriteCanBeRetried(txnIds.write);
    assert.commandWorked(retryCommit(txnIds.commit));
}

runTest({shouldRestart: false});
runTest({shouldRestart: true});

//
// Verify setFCV is interruptible between modifying sessions.
//
clearRawMongoProgramOutput();
checkFCV(adminDB, latestFCV);

// Construct a config.transactions entry that would be modified by downgrade.
const txnIds = {
    interrupt: {lsid: {id: UUID()}, txnNumber: 0}
};
assert.commandWorked(runTxn(txnIds.interrupt, {commit: true, prepare: true}));
checkConfigTransactionEntry(rst, txnIds.interrupt, {hasEntry: true, expectedState: "committed"});

// Pause setFCV before it would modify the entry.
assert.commandWorked(rst.getPrimary().adminCommand(
    {configureFailPoint: "pauseBeforeDowngradingSessions", mode: "alwaysOn"}));

TestData.setFCVLsid = {
    id: UUID()
};
const awaitUpgradeFCV = startParallelShell(() => {
    load("jstests/libs/feature_compatibility_version.js");
    assert.commandFailedWithCode(
        db.adminCommand({setFeatureCompatibilityVersion: lastStableFCV, lsid: TestData.setFCVLsid}),
        ErrorCodes.Interrupted);
}, rst.getPrimary().port);
waitForFailpoint("Hit pauseBeforeDowngradingSessions failpoint", 1 /*numTimes*/);

// Kill the session running setFCV.
assert.commandWorked(rst.getPrimary().adminCommand({killSessions: [TestData.setFCVLsid]}));

// Unpause the failpoint and verify setFCV returns without modifying config.transactions.
assert.commandWorked(rst.getPrimary().adminCommand(
    {configureFailPoint: "pauseBeforeDowngradingSessions", mode: "off"}));

awaitUpgradeFCV();
checkConfigTransactionEntry(rst, txnIds.interrupt, {hasEntry: true, expectedState: "committed"});

rst.stopSet();
}());