jstests/replsets/inmemory_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({
        nodeOptions: {syncdelay: 1, setParameter: {logComponentVerbosity: tojson({storage: 1})}},
        nodes: [
            {storageEngine: "wiredTiger"},
            {storageEngine: "inMemory", rsConfig: {priority: 0, votes: 0}},
        ],
        waitForKeys: false
    });
    replSet.startSet(PrepareHelpers.replSetStartSetOptions);
    replSet.initiateWithAnyNodeAsPrimary(
        null, "replSetInitiate", {doNotWaitForStableRecoveryTimestamp: true});
    const primary = replSet.getPrimary();
    const secondary = replSet.getSecondary();
    const primaryOplog = primary.getDB("local").oplog.rs;
    assert.lte(primaryOplog.dataSize(), PrepareHelpers.oplogSizeBytes);
    const secondaryOplog = secondary.getDB("local").oplog.rs;
    assert.lte(secondaryOplog.dataSize(), PrepareHelpers.oplogSizeBytes);
    const coll = primary.getDB("test").test;
    assert.commandWorked(coll.insert({}));
    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 oldestRequiredTimestampForCrashRecovery =
        getOldestRequiredTimestampForCrashRecovery(primary.getDB("test"));
    assert.lte(oldestRequiredTimestampForCrashRecovery, prepareTimestamp);
    jsTestLog("Get transaction entry from config.transactions");
    const txnEntry = primary.getDB("config").transactions.findOne();
    assert.lte(txnEntry.startOpTime.ts, prepareTimestamp, tojson(txnEntry));
    assert.soonNoExcept(() => {
        const secondaryTxnEntry = secondary.getDB("config").transactions.findOne();
        assert(secondaryTxnEntry);
        assert.eq(secondaryTxnEntry, txnEntry, tojson(secondaryTxnEntry));
        return true;
    });
    jsTestLog("Find prepare oplog entry");
    const oplogEntry = PrepareHelpers.findPrepareEntry(primaryOplog);
    assert.eq(oplogEntry.ts, prepareTimestamp, tojson(oplogEntry));
    const secondaryOplogEntry = PrepareHelpers.findPrepareEntry(secondaryOplog);
    assert.eq(secondaryOplogEntry.ts, prepareTimestamp, tojson(secondaryOplogEntry));
    jsTestLog("Insert documents until oplog exceeds oplogSize");
    PrepareHelpers.growOplogPastMaxSize(replSet);
    jsTestLog(`Oplog dataSize = ${primaryOplog.dataSize()}, check the prepare entry still exists`);
    assert.eq(oplogEntry, PrepareHelpers.findPrepareEntry(primaryOplog));
    assert.soon(() => {
        return secondaryOplog.dataSize() > PrepareHelpers.oplogSizeBytes;
    });
    assert.eq(oplogEntry, PrepareHelpers.findPrepareEntry(secondaryOplog));
    if (commitOrAbort === "commit") {
        jsTestLog("Commit prepared transaction and wait for oplog to shrink to max oplogSize");
        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}`);
    }
    PrepareHelpers.awaitOplogTruncation(replSet);
    replSet.stopSet();
}
function doTest ( commitOrAbort ) { const replSet = new ReplSetTest ( { nodeOptions : { syncdelay : 1 , setParameter : { logComponentVerbosity : tojson ( { storage : 1 } ) } } , nodes : [ { storageEngine : "wiredTiger" } , { storageEngine : "inMemory" , rsConfig : { priority : 0 , votes : 0 } } , ] , waitForKeys : false } ) ; replSet . startSet ( PrepareHelpers . replSetStartSetOptions ) ; replSet . initiateWithAnyNodeAsPrimary ( null , "replSetInitiate" , { doNotWaitForStableRecoveryTimestamp : true } ) ; const primary = replSet . getPrimary ( ) ; const secondary = replSet . getSecondary ( ) ; const primaryOplog = primary . getDB ( "local" ) . oplog . rs ; assert . lte ( primaryOplog . dataSize ( ) , PrepareHelpers . oplogSizeBytes ) ; const secondaryOplog = secondary . getDB ( "local" ) . oplog . rs ; assert . lte ( secondaryOplog . dataSize ( ) , PrepareHelpers . oplogSizeBytes ) ; const coll = primary . getDB ( "test" ) . test ; assert . commandWorked ( coll . insert ( { } ) ) ; 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 oldestRequiredTimestampForCrashRecovery = getOldestRequiredTimestampForCrashRecovery ( primary . getDB ( "test" ) ) ; assert . lte ( oldestRequiredTimestampForCrashRecovery , prepareTimestamp ) ; jsTestLog ( "Get transaction entry from config.transactions" ) ; const txnEntry = primary . getDB ( "config" ) . transactions . findOne ( ) ; assert . lte ( txnEntry . startOpTime . ts , prepareTimestamp , tojson ( txnEntry ) ) ; assert . soonNoExcept ( ( ) => { const secondaryTxnEntry = secondary . getDB ( "config" ) . transactions . findOne ( ) ; assert ( secondaryTxnEntry ) ; assert . eq ( secondaryTxnEntry , txnEntry , tojson ( secondaryTxnEntry ) ) ; return true ; } ) ; jsTestLog ( "Find prepare oplog entry" ) ; const oplogEntry = PrepareHelpers . findPrepareEntry ( primaryOplog ) ; assert . eq ( oplogEntry . ts , prepareTimestamp , tojson ( oplogEntry ) ) ; const secondaryOplogEntry = PrepareHelpers . findPrepareEntry ( secondaryOplog ) ; assert . eq ( secondaryOplogEntry . ts , prepareTimestamp , tojson ( secondaryOplogEntry ) ) ; jsTestLog ( "Insert documents until oplog exceeds oplogSize" ) ; PrepareHelpers . growOplogPastMaxSize ( replSet ) ; jsTestLog ( ` ${ primaryOplog . dataSize ( ) } ` ) ; assert . eq ( oplogEntry , PrepareHelpers . findPrepareEntry ( primaryOplog ) ) ; assert . soon ( ( ) => { return secondaryOplog . dataSize ( ) > PrepareHelpers . oplogSizeBytes ; } ) ; assert . eq ( oplogEntry , PrepareHelpers . findPrepareEntry ( secondaryOplog ) ) ; if ( commitOrAbort === "commit" ) { jsTestLog ( "Commit prepared transaction and wait for oplog to shrink to max oplogSize" ) ; 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 } ` ) ; } 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. // inMemory node must not be a voter, otherwise lastCommitted never advances // The prepare oplog entry may or may not be the first oplog entry depending on packing. // Must already be written on secondary, since the config.transactions entry is. // Oplog with prepared txn grows indefinitely - let it reach twice its supposed max size.