jstests/replsets/libs/initial_sync_test.js
/**
 * This fixture allows the user to optionally pass in a custom ReplSetTest to be used for the test.
 * The underlying replica set must have exactly two nodes: a primary, and a secondary.
 *
 * If the caller does not provide their own replica set, a two node replset will be initialized
 * instead, with all nodes running the latest version.
 *
 * @param {string} [name] the name of the test being run
 * @param {Object} [replSet] the ReplSetTest instance to adopt
 * @param {int} [timeout] how long to wait for initial sync to start
 */
This fixture allows the user to optionally pass in a custom ReplSetTest to be used for the test.
The underlying replica set must have exactly two nodes: a primary, and a secondary.

If the caller does not provide their own replica set, a two node replset will be initialized
instead, with all nodes running the latest version.

@param {string} [name] the name of the test being run
@param {Object} [replSet] the ReplSetTest instance to adopt
@param {int} [timeout] how long to wait for initial sync to start
This fixture allows the user to optionally pass in a custom ReplSetTest to be used for the test . The underlying replica set must have exactly two nodes : a primary and a secondary . If the caller does not provide their own replica set a two node replset will be initialized instead with all nodes running the latest version . @param { string } [ name ] the name of the test being run @param { Object } [ replSet ] the ReplSetTest instance to adopt @param { int } [ timeout ] how long to wait for initial sync to start
function InitialSyncTest(
    name = "InitialSyncTest", replSet, timeout, replBatchLimitBytes = 100 * 1024 * 1024) {
    this.replBatchLimitBytes = replBatchLimitBytes;
    const State = {
        kBeforeInitialSync: "kBeforeInitialSync",
        kDuringInitialSync: "kDuringInitialSync",
        kInitialSyncCompleted: "kInitialSyncCompleted",
        kStopped: "kStopped"
    };
    const AcceptableTransitions = {
        [State.kBeforeInitialSync]: [State.kDuringInitialSync],
        [State.kDuringInitialSync]: [State.kInitialSyncCompleted],
        [State.kInitialSyncCompleted]: [State.kStopped],
        [State.kStopped]: []
    };
    let currState = State.kBeforeInitialSync;
    replSet = (replSet === undefined) ? performSetup() : replSet;
    assert.eq(2, replSet.nodes.length, "Replica set must contain exactly two nodes.");
    let initialSyncTimeout = timeout || replSet.kDefaultTimeoutMS;
    const primary = replSet.getPrimary();
    let secondary = replSet.getSecondary();
    replSet.waitForState(secondary, ReplSetTest.State.SECONDARY);
    function performSetup() {
        let nodeOptions = {};
        if (TestData.logComponentVerbosity) {
            nodeOptions["setParameter"] = {
                "logComponentVerbosity": tojsononeline(TestData.logComponentVerbosity),
                "replBatchLimitBytes": replBatchLimitBytes,
            };
        }
        let replSet = new ReplSetTest({
            name: name,
            nodes: [{}, {rsConfig: {priority: 0, votes: 0}}],
            nodeOptions: nodeOptions
        });
        replSet.startSet();
        replSet.initiate();
        return replSet;
    }
    function transitionIfAllowed(newState) {
        if (AcceptableTransitions[currState].includes(newState)) {
            jsTestLog(`Transitioning to: "${newState}"`, true);
            currState = newState;
        } else {
            throw new Error(`Can't transition to State "${newState}" from State "${currState}"`);
        }
    }
    function isNodeInState(node, state) {
        return state ===
            assert
                .commandWorkedOrFailedWithCode(
                    node.adminCommand({replSetGetStatus: 1, initialSync: 0}),
                    ErrorCodes.NotYetInitialized)
                .myState;
    }
    function hasStartedInitialSync() {
        return isNodeInState(secondary, ReplSetTest.State.STARTUP_2);
    }
    function hasCompletedInitialSync() {
        assert.eq(currState,
                  State.kDuringInitialSync,
                  "Should not check if initial sync completed before node is restarted");
        return isNodeInState(secondary, ReplSetTest.State.SECONDARY);
    }
    function assertNoOpenTxns() {
        const status = assert.commandWorked(primary.adminCommand('serverStatus'));
        assert(typeof status.transactions === "object", status);
        assert.eq(0, status.transactions.currentOpen, status.transactions);
    }
    function restartNodeWithoutData() {
        replSet.stop(secondary, undefined, {skipValidation: true});
        const nodeOptions = {
            startClean: true,
            setParameter: {
                'failpoint.initialSyncFuzzerSynchronizationPoint1': tojson({mode: 'alwaysOn'}),
                'replBatchLimitBytes': replBatchLimitBytes,
            },
        };
        if (TestData.logComponentVerbosity) {
            nodeOptions.setParameter.logComponentVerbosity =
                tojsononeline(TestData.logComponentVerbosity);
        }
        secondary = replSet.start(secondary, nodeOptions, true);
    }
    function waitUntilInitialSyncPausedOrCompleted() {
        assert.soon(function() {
            if (checkLog.checkContainsOnce(
                    secondary, "initialSyncFuzzerSynchronizationPoint1 fail point enabled")) {
                return true;
            }
            return hasCompletedInitialSync();
        }, "initial sync did not pause or complete");
    }
    function pauseBeforeSyncSourceCommand() {
        assert.commandWorked(secondary.adminCommand(
            {"configureFailPoint": 'initialSyncFuzzerSynchronizationPoint2', "mode": 'alwaysOn'}));
        assert.commandWorked(secondary.adminCommand(
            {"configureFailPoint": 'initialSyncFuzzerSynchronizationPoint1', "mode": 'off'}));
        checkLog.contains(secondary, "initialSyncFuzzerSynchronizationPoint2 fail point enabled");
    }
    function resumeAndPauseBeforeNextSyncSourceCommand() {
        assert.commandWorked(secondary.adminCommand(
            {"configureFailPoint": 'initialSyncFuzzerSynchronizationPoint1', "mode": 'alwaysOn'}));
        assert.commandWorked(secondary.adminCommand(
            {"configureFailPoint": 'initialSyncFuzzerSynchronizationPoint2', "mode": 'off'}));
        waitUntilInitialSyncPausedOrCompleted();
    }
    this.step = function() {
        if (currState === State.kBeforeInitialSync) {
            restartNodeWithoutData();
            assert.soon(hasStartedInitialSync, "failed to start initial sync", initialSyncTimeout);
            transitionIfAllowed(State.kDuringInitialSync);
            checkLog.contains(secondary,
                              "initialSyncFuzzerSynchronizationPoint1 fail point enabled");
            return false;
        }
        assert.neq(currState, State.kStopped, "Cannot call step() if the test fixture was stopped");
        assert.neq(currState,
                   State.kInitialSyncCompleted,
                   "Cannot call step() if initial sync already completed");
        pauseBeforeSyncSourceCommand();
        assert.commandWorked(secondary.adminCommand({clearLog: 'global'}));
        resumeAndPauseBeforeNextSyncSourceCommand();
        if (hasCompletedInitialSync()) {
            transitionIfAllowed(State.kInitialSyncCompleted);
            assert.commandWorked(secondary.adminCommand(
                {"configureFailPoint": 'initialSyncFuzzerSynchronizationPoint1', "mode": 'off'}));
            return true;
        }
        return false;
    };
    this.getPrimary = function() {
        return primary;
    };
    this.getSecondary = function() {
        return secondary;
    };
    this.fail = function() {
        assert.commandWorked(secondary.adminCommand(
            {"configureFailPoint": 'initialSyncFuzzerSynchronizationPoint1', "mode": 'off'}));
        assert.commandWorked(secondary.adminCommand(
            {"configureFailPoint": 'initialSyncFuzzerSynchronizationPoint2', "mode": 'off'}));
    };
    this.stop = function() {
        transitionIfAllowed(State.kStopped);
        assertNoOpenTxns();
        return replSet.stopSet();
    };
}
function InitialSyncTest ( name = "InitialSyncTest" , replSet , timeout , replBatchLimitBytes = 100 * 1024 * 1024 ) { this . replBatchLimitBytes = replBatchLimitBytes ; const State = { kBeforeInitialSync : "kBeforeInitialSync" , kDuringInitialSync : "kDuringInitialSync" , kInitialSyncCompleted : "kInitialSyncCompleted" , kStopped : "kStopped" } ; const AcceptableTransitions = { [ State . kBeforeInitialSync ] : [ State . kDuringInitialSync ] , [ State . kDuringInitialSync ] : [ State . kInitialSyncCompleted ] , [ State . kInitialSyncCompleted ] : [ State . kStopped ] , [ State . kStopped ] : [ ] } ; let currState = State . kBeforeInitialSync ; replSet = ( replSet === undefined ) ? performSetup ( ) : replSet ; assert . eq ( 2 , replSet . nodes . length , "Replica set must contain exactly two nodes." ) ; let initialSyncTimeout = timeout || replSet . kDefaultTimeoutMS ; const primary = replSet . getPrimary ( ) ; let secondary = replSet . getSecondary ( ) ; replSet . waitForState ( secondary , ReplSetTest . State . SECONDARY ) ; function performSetup ( ) { let nodeOptions = { } ; if ( TestData . logComponentVerbosity ) { nodeOptions [ "setParameter" ] = { "logComponentVerbosity" : tojsononeline ( TestData . logComponentVerbosity ) , "replBatchLimitBytes" : replBatchLimitBytes , } ; } let replSet = new ReplSetTest ( { name : name , nodes : [ { } , { rsConfig : { priority : 0 , votes : 0 } } ] , nodeOptions : nodeOptions } ) ; replSet . startSet ( ) ; replSet . initiate ( ) ; return replSet ; } function transitionIfAllowed ( newState ) { if ( AcceptableTransitions [ currState ] . includes ( newState ) ) { jsTestLog ( ` ${ newState } ` , true ) ; currState = newState ; } else { throw new Error ( ` ${ newState } ${ currState } ` ) ; } } function isNodeInState ( node , state ) { return state === assert . commandWorkedOrFailedWithCode ( node . adminCommand ( { replSetGetStatus : 1 , initialSync : 0 } ) , ErrorCodes . NotYetInitialized ) . myState ; } function hasStartedInitialSync ( ) { return isNodeInState ( secondary , ReplSetTest . State . STARTUP_2 ) ; } function hasCompletedInitialSync ( ) { assert . eq ( currState , State . kDuringInitialSync , "Should not check if initial sync completed before node is restarted" ) ; return isNodeInState ( secondary , ReplSetTest . State . SECONDARY ) ; } function assertNoOpenTxns ( ) { const status = assert . commandWorked ( primary . adminCommand ( 'serverStatus' ) ) ; assert ( typeof status . transactions === "object" , status ) ; assert . eq ( 0 , status . transactions . currentOpen , status . transactions ) ; } function restartNodeWithoutData ( ) { replSet . stop ( secondary , undefined , { skipValidation : true } ) ; const nodeOptions = { startClean : true , setParameter : { 'failpoint.initialSyncFuzzerSynchronizationPoint1' : tojson ( { mode : 'alwaysOn' } ) , 'replBatchLimitBytes' : replBatchLimitBytes , } , } ; if ( TestData . logComponentVerbosity ) { nodeOptions . setParameter . logComponentVerbosity = tojsononeline ( TestData . logComponentVerbosity ) ; } secondary = replSet . start ( secondary , nodeOptions , true ) ; } function waitUntilInitialSyncPausedOrCompleted ( ) { assert . soon ( function ( ) { if ( checkLog . checkContainsOnce ( secondary , "initialSyncFuzzerSynchronizationPoint1 fail point enabled" ) ) { return true ; } return hasCompletedInitialSync ( ) ; } , "initial sync did not pause or complete" ) ; } function pauseBeforeSyncSourceCommand ( ) { assert . commandWorked ( secondary . adminCommand ( { "configureFailPoint" : 'initialSyncFuzzerSynchronizationPoint2' , "mode" : 'alwaysOn' } ) ) ; assert . commandWorked ( secondary . adminCommand ( { "configureFailPoint" : 'initialSyncFuzzerSynchronizationPoint1' , "mode" : 'off' } ) ) ; checkLog . contains ( secondary , "initialSyncFuzzerSynchronizationPoint2 fail point enabled" ) ; } function resumeAndPauseBeforeNextSyncSourceCommand ( ) { assert . commandWorked ( secondary . adminCommand ( { "configureFailPoint" : 'initialSyncFuzzerSynchronizationPoint1' , "mode" : 'alwaysOn' } ) ) ; assert . commandWorked ( secondary . adminCommand ( { "configureFailPoint" : 'initialSyncFuzzerSynchronizationPoint2' , "mode" : 'off' } ) ) ; waitUntilInitialSyncPausedOrCompleted ( ) ; } this . step = function ( ) { if ( currState === State . kBeforeInitialSync ) { restartNodeWithoutData ( ) ; assert . soon ( hasStartedInitialSync , "failed to start initial sync" , initialSyncTimeout ) ; transitionIfAllowed ( State . kDuringInitialSync ) ; checkLog . contains ( secondary , "initialSyncFuzzerSynchronizationPoint1 fail point enabled" ) ; return false ; } assert . neq ( currState , State . kStopped , "Cannot call step() if the test fixture was stopped" ) ; assert . neq ( currState , State . kInitialSyncCompleted , "Cannot call step() if initial sync already completed" ) ; pauseBeforeSyncSourceCommand ( ) ; assert . commandWorked ( secondary . adminCommand ( { clearLog : 'global' } ) ) ; resumeAndPauseBeforeNextSyncSourceCommand ( ) ; if ( hasCompletedInitialSync ( ) ) { transitionIfAllowed ( State . kInitialSyncCompleted ) ; assert . commandWorked ( secondary . adminCommand ( { "configureFailPoint" : 'initialSyncFuzzerSynchronizationPoint1' , "mode" : 'off' } ) ) ; return true ; } return false ; } ; this . getPrimary = function ( ) { return primary ; } ; this . getSecondary = function ( ) { return secondary ; } ; this . fail = function ( ) { assert . commandWorked ( secondary . adminCommand ( { "configureFailPoint" : 'initialSyncFuzzerSynchronizationPoint1' , "mode" : 'off' } ) ) ; assert . commandWorked ( secondary . adminCommand ( { "configureFailPoint" : 'initialSyncFuzzerSynchronizationPoint2' , "mode" : 'off' } ) ) ; } ; this . stop = function ( ) { transitionIfAllowed ( State . kStopped ) ; assertNoOpenTxns ( ) ; return replSet . stopSet ( ) ; } ; }
This fixture allows the user to optionally pass in a custom ReplSetTest to be used for the test.
This fixture allows the user to optionally pass in a custom ReplSetTest to be used for the test .
// Make sure we have a replica set up and running. /**
     * Return an instance of ReplSetTest initialized with a standard two-node replica set running
     * with the latest version.
     */ /**
     * Transition from the current State to `newState` if it's a valid transition, otherwise throw
     * an error.
     */ // Transitioning to a disallowed State is likely a bug in the code, so we throw an // error here instead of silently failing. /**
     * Calls replSetGetStatus and checks if the node is in the provided state.
     */ // We suppress the initialSync field here, because initial sync is paused while holding the // mutex needed to report initial sync progress. // We know that initial sync has started once the node has transitioned to STARTUP2. // Make sure this isn't called before the secondary starts initial sync. // We know initial sync has completed if the node has transitioned to SECONDARY state. /**
     * Asserts that there are no open transactions.
     */ /**
     * Restarts the secondary with the first synchronization failpoint enabled so that we ensure
     * that initial sync pauses the first time the node reaches out to the sync source.
     */ // Skip validation when stopping the node in case there are transactions in prepare. // Restart the node with the first synchronization failpoint enabled so that initial sync // doesn't finish before we can pause it. /**
     * Wait until the first synchronization fail point is hit to show that initial sync is paused
     * or until initial sync has completed.
     */ /**
     * Flip failpoints and wait until the second synchronization failpoint is hit so that we know
     * it is safe to let initial sync resume again. This step is necessary before issuing the next
     * command to ensure that we only run one command before pausing at the first synchronization
     * failpoint again.
     */ /**
     * Flip failpoints and wait until the first synchronization failpoint is hit so that initial
     * sync can make progress by issuing the next command, but pausing before the following command
     * can be issued.
     */ /**
     * This function will resume initial sync and run the next command before using the
     * synchronization failpoints to make sure initial sync is either paused or completed. Other
     * than before initial sync has started and after initial sync has completed, when this
     * function is called the secondary should be paused at the first synchronization failpoint
     * before and after running the next command. This ensures that we can deterministically pause
     * initial sync and only one command is run each function call. During collection cloning, we
     * pause before running listDatabases, listCollections and listIndexes commands on the sync
     * source. During the oplog application phase, we pause before applying each batch of oplog
     * entries on the initial syncing node.
     *
     * If initial sync hasn't started yet, the function will restart the secondary without data to
     * cause it to go through initial sync. It will throw an exception if called after initial sync
     * has already completed or stop() has been called.
     *
     * @return true if initial sync has completed
     */ // If initial sync has not started yet, restart the node without data to cause it to go // through initial sync. // Wait until initial sync has started. // Make sure this wasn't called after the test fixture was stopped or this function already // returned that initial sync was completed. // Clear ramlog so checkLog can't find log messages from previous times either failpoint was // enabled. // If initial sync is completed, let the caller know. /**
     * Performs data consistency checks and then stops the replica set. Will fail if there is a
     * transaction that wasn't aborted or committed.
     */