jstests/replsets/initial_sync_preserves_active_txns.js
// A new replica set for both the commit and abort tests to ensure the same clean state.
A new replica set for both the commit and abort tests to ensure the same clean state.
A new replica set for both the commit and abort tests to ensure the same clean state .
function doTest(commitOrAbort) {
    const replSet = new ReplSetTest({
        oplogSize: PrepareHelpers.oplogSizeMB,
        nodeOptions: {syncdelay: 1, setParameter: {logComponentVerbosity: tojson({storage: 1})}},
        nodes: 1
    });
    replSet.startSet(PrepareHelpers.replSetStartSetOptions);
    replSet.initiate();
    const primary = replSet.getPrimary();
    const primaryOplog = primary.getDB("local").oplog.rs;
    assert.lte(primaryOplog.dataSize(), PrepareHelpers.oplogSizeBytes);
    const coll = primary.getDB("test").test;
    assert.commandWorked(coll.insert({}, {writeConcern: {w: "majority"}}));
    jsTestLog("Prepare a transaction");
    const session = primary.startSession();
    session.startTransaction();
    assert.commandWorked(session.getDatabase("test").test.insert({myTransaction: 1}));
    const prepareTimestamp = PrepareHelpers.prepareTransaction(session);
    const txnEntry = primary.getDB("config").transactions.findOne();
    const oldestRequiredTimestampForCrashRecovery =
        getOldestRequiredTimestampForCrashRecovery(primary.getDB("test"));
    assert.lte(oldestRequiredTimestampForCrashRecovery, prepareTimestamp);
    let oplog = primary.getDB("local").getCollection("oplog.rs");
    const txnNum = session.getTxnNumber_forTesting();
    const op = oplog.findOne({"txnNumber": txnNum, "lsid.id": session.getSessionId().id});
    assert.neq(op, null);
    const firstTxnOpTs = op.ts;
    assert.eq(txnEntry.startOpTime.ts, firstTxnOpTs, tojson(txnEntry));
    jsTestLog("Insert documents until oplog exceeds oplogSize");
    PrepareHelpers.growOplogPastMaxSize(replSet);
    jsTestLog("Make sure the transaction's first entry is still in the oplog");
    assert.eq(primaryOplog.find({ts: firstTxnOpTs}).itcount(), 1);
    jsTestLog("Add a secondary node");
    const secondary = replSet.add({rsConfig: {votes: 0, priority: 0}});
    replSet.reInitiate();
    jsTestLog("Reinitiated, awaiting secondary node");
    replSet.awaitSecondaryNodes();
    jsTestLog("Checking secondary oplog and config.transactions");
    const secondaryOplog = secondary.getDB("local").oplog.rs;
    assert.gt(secondaryOplog.dataSize(), PrepareHelpers.oplogSizeBytes);
    assert.eq(secondaryOplog.find({ts: firstTxnOpTs}).itcount(), 1);
    const secondaryTxnEntry = secondary.getDB("config").transactions.findOne();
    assert.eq(secondaryTxnEntry, txnEntry, tojson(secondaryTxnEntry));
    if (commitOrAbort === "commit") {
        jsTestLog("Commit prepared transaction and wait for oplog to shrink to max oplogSize");
        assert.commandWorked(PrepareHelpers.commitTransaction(session, prepareTimestamp));
    } else if (commitOrAbort === "abort") {
        jsTestLog("Abort prepared transaction and wait for oplog to shrink to max oplogSize");
        assert.commandWorked(session.abortTransaction_forTesting());
    } else {
        throw new Error(`Unrecognized value for commitOrAbort: ${commitOrAbort}`);
    }
    replSet.awaitReplication();
    PrepareHelpers.awaitOplogTruncation(replSet);
    replSet.stopSet();
}
function doTest ( commitOrAbort ) { const replSet = new ReplSetTest ( { oplogSize : PrepareHelpers . oplogSizeMB , nodeOptions : { syncdelay : 1 , setParameter : { logComponentVerbosity : tojson ( { storage : 1 } ) } } , nodes : 1 } ) ; replSet . startSet ( PrepareHelpers . replSetStartSetOptions ) ; replSet . initiate ( ) ; const primary = replSet . getPrimary ( ) ; const primaryOplog = primary . getDB ( "local" ) . oplog . rs ; assert . lte ( primaryOplog . dataSize ( ) , PrepareHelpers . oplogSizeBytes ) ; const coll = primary . getDB ( "test" ) . test ; assert . commandWorked ( coll . insert ( { } , { writeConcern : { w : "majority" } } ) ) ; jsTestLog ( "Prepare a transaction" ) ; const session = primary . startSession ( ) ; session . startTransaction ( ) ; assert . commandWorked ( session . getDatabase ( "test" ) . test . insert ( { myTransaction : 1 } ) ) ; const prepareTimestamp = PrepareHelpers . prepareTransaction ( session ) ; const txnEntry = primary . getDB ( "config" ) . transactions . findOne ( ) ; const oldestRequiredTimestampForCrashRecovery = getOldestRequiredTimestampForCrashRecovery ( primary . getDB ( "test" ) ) ; assert . lte ( oldestRequiredTimestampForCrashRecovery , prepareTimestamp ) ; let oplog = primary . getDB ( "local" ) . getCollection ( "oplog.rs" ) ; const txnNum = session . getTxnNumber_forTesting ( ) ; const op = oplog . findOne ( { "txnNumber" : txnNum , "lsid.id" : session . getSessionId ( ) . id } ) ; assert . neq ( op , null ) ; const firstTxnOpTs = op . ts ; assert . eq ( txnEntry . startOpTime . ts , firstTxnOpTs , tojson ( txnEntry ) ) ; jsTestLog ( "Insert documents until oplog exceeds oplogSize" ) ; PrepareHelpers . growOplogPastMaxSize ( replSet ) ; jsTestLog ( "Make sure the transaction's first entry is still in the oplog" ) ; assert . eq ( primaryOplog . find ( { ts : firstTxnOpTs } ) . itcount ( ) , 1 ) ; jsTestLog ( "Add a secondary node" ) ; const secondary = replSet . add ( { rsConfig : { votes : 0 , priority : 0 } } ) ; replSet . reInitiate ( ) ; jsTestLog ( "Reinitiated, awaiting secondary node" ) ; replSet . awaitSecondaryNodes ( ) ; jsTestLog ( "Checking secondary oplog and config.transactions" ) ; const secondaryOplog = secondary . getDB ( "local" ) . oplog . rs ; assert . gt ( secondaryOplog . dataSize ( ) , PrepareHelpers . oplogSizeBytes ) ; assert . eq ( secondaryOplog . find ( { ts : firstTxnOpTs } ) . itcount ( ) , 1 ) ; const secondaryTxnEntry = secondary . getDB ( "config" ) . transactions . findOne ( ) ; assert . eq ( secondaryTxnEntry , txnEntry , tojson ( secondaryTxnEntry ) ) ; if ( commitOrAbort === "commit" ) { jsTestLog ( "Commit prepared transaction and wait for oplog to shrink to max oplogSize" ) ; assert . commandWorked ( PrepareHelpers . commitTransaction ( session , prepareTimestamp ) ) ; } else if ( commitOrAbort === "abort" ) { jsTestLog ( "Abort prepared transaction and wait for oplog to shrink to max oplogSize" ) ; assert . commandWorked ( session . abortTransaction_forTesting ( ) ) ; } else { throw new Error ( ` ${ commitOrAbort } ` ) ; } replSet . awaitReplication ( ) ; PrepareHelpers . awaitOplogTruncation ( replSet ) ; replSet . stopSet ( ) ; }
A new replica set for both the commit and abort tests to ensure the same clean state.
A new replica set for both the commit and abort tests to ensure the same clean state .
// Oplog can be truncated each "sync" cycle. Increase its frequency to once per second. // Make sure that the timestamp of the first oplog entry for this transaction matches the // start timestamp in the transactions table. // Oplog with prepared txn grows indefinitely - let it reach twice its supposed max size. // Oplog grew past maxSize, and it includes the oldest active transaction's entry.