jstests/replsets/tenant_migration_donor_rollback_during_cloning.js
// 7. The migration is allowed to completed, and a 'forgetMigration' command is issued.
7. The migration is allowed to completed, and a 'forgetMigration' command is issued.
7 . The migration is allowed to completed and a ' forgetMigration ' command is issued .
function runTest(tenantId,
                 setupFunction,
                 whilePausedFunction,
                 postMigrationFunction,
                 firstFailpointData,
                 secondFailpoint) {
    const donorRst = new ReplSetTest({
        name: "donorRst",
        nodes: 5,
        nodeOptions: Object.assign(migrationX509Options.donor, {
            setParameter: {
                'failpoint.tenantMigrationDonorAllowsNonTimestampedReads':
                    tojson({mode: 'alwaysOn'}),
            }
        }),
        settings: {catchUpTimeoutMillis: 0, chainingAllowed: false}
    });
    donorRst.startSet();
    donorRst.initiateWithHighElectionTimeout();
    const recipientPrimary = recipientRst.getPrimary();
    const tenantMigrationTest = new TenantMigrationTest(
        {name: jsTestName(), donorRst: donorRst, recipientRst: recipientRst});
    const migrationId = UUID();
    const migrationOpts = {
        migrationIdString: extractUUIDFromObject(migrationId),
        recipientConnString: tenantMigrationTest.getRecipientConnString(),
        tenantId: tenantId,
        readPreference: {mode: 'secondary'}
    };
    firstFailpointData.database = tenantMigrationTest.tenantDB(tenantId, "testDB");
    const fpBeforeListCall =
        configureFailPoint(recipientPrimary, "hangBeforeClonerStage", firstFailpointData);
    const fpAfterListCall = configureFailPoint(recipientPrimary, secondFailpoint);
    const fpPauseOplogFetcher =
        configureFailPoint(recipientPrimary, "hangBeforeStartingOplogFetcher");
    setupFunction(tenantId, tenantMigrationTest);
    donorRst.awaitReplication();
    jsTestLog(`Starting tenant migration with migrationId: ${
        migrationOpts.migrationIdString}, tenantId: ${tenantId}`);
    assert.commandWorked(tenantMigrationTest.startMigration(migrationOpts));
    fpBeforeListCall.wait();
    jsTestLog("Stopping donor replication.");
    let res = recipientPrimary.adminCommand({currentOp: true, desc: "tenant recipient migration"});
    let currOp = res.inprog.find(op => bsonWoCompare(op.instanceID, migrationId) === 0);
    assert.eq(bsonWoCompare(currOp.instanceID, migrationId), 0, res);
    assert.eq(currOp.numRestartsDueToDonorConnectionFailure, 0, res);
    let syncSourceNode = undefined;
    let otherNodes = [];
    donorRst.getSecondaries().forEach(node => {
        if (node.host != currOp.donorSyncSource) {
            otherNodes.push(node);
        } else {
            syncSourceNode = node;
        }
    });
    assert.eq(otherNodes.length, 3, otherNodes);
    jsTestLog(`Sync source node: ${syncSourceNode}, other secondaries: ${otherNodes}`);
    stopServerReplication(otherNodes);
    jsTestLog("Performing work that will be rolled back.");
    whilePausedFunction(tenantId, syncSourceNode, tenantMigrationTest);
    fpBeforeListCall.off();
    jsTestLog("Stepping a new node up.");
    fpAfterListCall.wait();
    const newDonorPrimary = otherNodes[0];
    newDonorPrimary.adminCommand({replSetStepUp: 1});
    restartServerReplication(otherNodes);
    tenantMigrationTest.insertDonorDB(
        tenantMigrationTest.nonTenantDB(tenantId, 'alternateDB'),
        'alternateColl',
        [{x: "Tom Petty", y: "Free Fallin"}, {x: "Sushin Shyam", y: "Cherathukal"}]);
    fpAfterListCall.off();
    jsTestLog("Make sure that the recipient has had to restart the migration.");
    assert.soon(() => {
        res = recipientPrimary.adminCommand({currentOp: true, desc: "tenant recipient migration"});
        currOp = res.inprog.find(op => bsonWoCompare(op.instanceID, migrationId) === 0);
        assert.eq(bsonWoCompare(currOp.instanceID, migrationId), 0, res);
        return currOp.numRestartsDueToDonorConnectionFailure == 1;
    }, "Expected the recipient to have restarted: " + tojson(res));
    fpPauseOplogFetcher.off();
    jsTestLog("Waiting for migration to complete.");
    TenantMigrationTest.assertCommitted(
        tenantMigrationTest.waitForMigrationToComplete(migrationOpts));
    donorRst.awaitReplication();
    postMigrationFunction(tenantId, tenantMigrationTest);
    assert.commandWorked(tenantMigrationTest.forgetMigration(migrationOpts.migrationIdString));
    donorRst.stopSet();
}
function runTest ( tenantId , setupFunction , whilePausedFunction , postMigrationFunction , firstFailpointData , secondFailpoint ) { const donorRst = new ReplSetTest ( { name : "donorRst" , nodes : 5 , nodeOptions : Object . assign ( migrationX509Options . donor , { setParameter : { 'failpoint.tenantMigrationDonorAllowsNonTimestampedReads' : tojson ( { mode : 'alwaysOn' } ) , } } ) , settings : { catchUpTimeoutMillis : 0 , chainingAllowed : false } } ) ; donorRst . startSet ( ) ; donorRst . initiateWithHighElectionTimeout ( ) ; const recipientPrimary = recipientRst . getPrimary ( ) ; const tenantMigrationTest = new TenantMigrationTest ( { name : jsTestName ( ) , donorRst : donorRst , recipientRst : recipientRst } ) ; const migrationId = UUID ( ) ; const migrationOpts = { migrationIdString : extractUUIDFromObject ( migrationId ) , recipientConnString : tenantMigrationTest . getRecipientConnString ( ) , tenantId : tenantId , readPreference : { mode : 'secondary' } } ; firstFailpointData . database = tenantMigrationTest . tenantDB ( tenantId , "testDB" ) ; const fpBeforeListCall = configureFailPoint ( recipientPrimary , "hangBeforeClonerStage" , firstFailpointData ) ; const fpAfterListCall = configureFailPoint ( recipientPrimary , secondFailpoint ) ; const fpPauseOplogFetcher = configureFailPoint ( recipientPrimary , "hangBeforeStartingOplogFetcher" ) ; setupFunction ( tenantId , tenantMigrationTest ) ; donorRst . awaitReplication ( ) ; jsTestLog ( ` ${ migrationOpts . migrationIdString } ${ tenantId } ` ) ; assert . commandWorked ( tenantMigrationTest . startMigration ( migrationOpts ) ) ; fpBeforeListCall . wait ( ) ; jsTestLog ( "Stopping donor replication." ) ; let res = recipientPrimary . adminCommand ( { currentOp : true , desc : "tenant recipient migration" } ) ; let currOp = res . inprog . find ( op => bsonWoCompare ( op . instanceID , migrationId ) === 0 ) ; assert . eq ( bsonWoCompare ( currOp . instanceID , migrationId ) , 0 , res ) ; assert . eq ( currOp . numRestartsDueToDonorConnectionFailure , 0 , res ) ; let syncSourceNode = undefined ; let otherNodes = [ ] ; donorRst . getSecondaries ( ) . forEach ( node => { if ( node . host != currOp . donorSyncSource ) { otherNodes . push ( node ) ; } else { syncSourceNode = node ; } } ) ; assert . eq ( otherNodes . length , 3 , otherNodes ) ; jsTestLog ( ` ${ syncSourceNode } ${ otherNodes } ` ) ; stopServerReplication ( otherNodes ) ; jsTestLog ( "Performing work that will be rolled back." ) ; whilePausedFunction ( tenantId , syncSourceNode , tenantMigrationTest ) ; fpBeforeListCall . off ( ) ; jsTestLog ( "Stepping a new node up." ) ; fpAfterListCall . wait ( ) ; const newDonorPrimary = otherNodes [ 0 ] ; newDonorPrimary . adminCommand ( { replSetStepUp : 1 } ) ; restartServerReplication ( otherNodes ) ; tenantMigrationTest . insertDonorDB ( tenantMigrationTest . nonTenantDB ( tenantId , 'alternateDB' ) , 'alternateColl' , [ { x : "Tom Petty" , y : "Free Fallin" } , { x : "Sushin Shyam" , y : "Cherathukal" } ] ) ; fpAfterListCall . off ( ) ; jsTestLog ( "Make sure that the recipient has had to restart the migration." ) ; assert . soon ( ( ) => { res = recipientPrimary . adminCommand ( { currentOp : true , desc : "tenant recipient migration" } ) ; currOp = res . inprog . find ( op => bsonWoCompare ( op . instanceID , migrationId ) === 0 ) ; assert . eq ( bsonWoCompare ( currOp . instanceID , migrationId ) , 0 , res ) ; return currOp . numRestartsDueToDonorConnectionFailure == 1 ; } , "Expected the recipient to have restarted: " + tojson ( res ) ) ; fpPauseOplogFetcher . off ( ) ; jsTestLog ( "Waiting for migration to complete." ) ; TenantMigrationTest . assertCommitted ( tenantMigrationTest . waitForMigrationToComplete ( migrationOpts ) ) ; donorRst . awaitReplication ( ) ; postMigrationFunction ( tenantId , tenantMigrationTest ) ; assert . commandWorked ( tenantMigrationTest . forgetMigration ( migrationOpts . migrationIdString ) ) ; donorRst . stopSet ( ) ; }
7.
7 .
// Allow non-timestamped reads on donor after migration completes for testing. // Set the 'catchUpTimeoutMillis' to 0, so that the new primary doesn't fetch operations // that we want rolled back. Turn off chaining to make sure that the secondary the recipient // initially syncs from is able to keep in sync with the primary. // The failpoints correspond to the instants right before and after the 'list*' call that the // recipient cloners make. // Pause the oplog fetcher to make sure that the cloner's failure rather than the fetcher's // failure results in restarting the migration. // Perform any initial setup on the donor before running the tenant migration. // Wait until right before the 'list*' call is made, and then stop replication on the donor RST. // Figure out which donor node the recipient is syncing from. // 'syncSourceNode' is the donor secondary from which the recipient is syncing. 'otherNodes' // are the other secondaries in the donor RST. // Perform some work on the donor primary while replication is paused. This work will not be // replicated, and will later be rolled back due to donor primary step down. // Once the 'list*' call has been made, the tenant migration can proceed. The 'list*' call will // have returned information that wasn't majority committed. Step up a new primary to expose // this situation. Allow replication once again. // Advance the cluster time by applying new operations on the new primary. We insert documents // into a non-tenant DB, so this data will not be migrated but will still advance the cluster // time. // Test to make sure some conditions post migration have been met.