jstests/multiVersion/genericChangeStreams/change_streams_read_oplog_after_downgrade.js
/**
 * For each test case and change stream variation, generates the oplog entries to be tested and
 * creates an augmented test-case containing a resume token which marks the start of the test, and a
 * sentinel entry that marks the end of the test. These will be used post-downgrade to run each of
 * the test cases in isolation.
 */
For each test case and change stream variation, generates the oplog entries to be tested and
creates an augmented test-case containing a resume token which marks the start of the test, and a
sentinel entry that marks the end of the test. These will be used post-downgrade to run each of
the test cases in isolation.
For each test case and change stream variation generates the oplog entries to be tested and creates an augmented test - case containing a resume token which marks the start of the test and a sentinel entry that marks the end of the test . These will be used post - downgrade to run each of the test cases in isolation .
function writeOplogEntriesAndCreateResumePointsOnLatestVersion() {
    function createSentinelEntryAndGetTimeStamp(testNum) {
        const documentId = "sentinel_entry_" + testNum;
        assert.commandWorked(shardedColl.insert({_id: documentId}));
        const oplog = st.rs0.getPrimary().getCollection("local.oplog.rs");
        const opLogEntries =
            oplog.find({op: "i", "o._id": documentId, ns: shardedColl.getFullName()}).toArray();
        assert.eq(opLogEntries.length, 1);
        return opLogEntries[0].ts;
    }
    let testNum = 0;
    let testStartTime = createSentinelEntryAndGetTimeStamp(testNum);
    const outputChangeStreams = [];
    for (let testCase of testCases) {
        jsTestLog(`Opening a change stream for '${testCase.testName}' at startTime: ${
            tojson(testStartTime)}`);
        let resumeToken;
        const csCursor = changeStreamsVariants[0].watch({startAtOperationTime: testStartTime});
        assert.soon(
            () => {
                if (!csCursor.hasNext()) {
                    return false;
                }
                const nextEvent = csCursor.next();
                resumeToken = nextEvent._id;
                return (nextEvent.documentKey._id == "sentinel_entry_" + testNum);
            },
            () => {
                return tojson(dumpLatestOpLogEntries(st.rs0.getPrimary(), 100));
            });
        for (let changeStreamVariant of changeStreamsVariants) {
            const outputChangeStream = {
                watch: changeStreamVariant.watch,
                resumeToken: resumeToken,
                endSentinelEntry: "sentinel_entry_" + (testNum + 1),
                testName: testCases.testName
            };
            outputChangeStreams.push(outputChangeStream);
        }
        testCase.generateOpLogEntry(shardedColl);
        testStartTime = createSentinelEntryAndGetTimeStamp(++testNum);
    }
    return outputChangeStreams;
}
function writeOplogEntriesAndCreateResumePointsOnLatestVersion ( ) { function createSentinelEntryAndGetTimeStamp ( testNum ) { const documentId = "sentinel_entry_" + testNum ; assert . commandWorked ( shardedColl . insert ( { _id : documentId } ) ) ; const oplog = st . rs0 . getPrimary ( ) . getCollection ( "local.oplog.rs" ) ; const opLogEntries = oplog . find ( { op : "i" , "o._id" : documentId , ns : shardedColl . getFullName ( ) } ) . toArray ( ) ; assert . eq ( opLogEntries . length , 1 ) ; return opLogEntries [ 0 ] . ts ; } let testNum = 0 ; let testStartTime = createSentinelEntryAndGetTimeStamp ( testNum ) ; const outputChangeStreams = [ ] ; for ( let testCase of testCases ) { jsTestLog ( ` ${ testCase . testName } ${ tojson ( testStartTime ) } ` ) ; let resumeToken ; const csCursor = changeStreamsVariants [ 0 ] . watch ( { startAtOperationTime : testStartTime } ) ; assert . soon ( ( ) => { if ( ! csCursor . hasNext ( ) ) { return false ; } const nextEvent = csCursor . next ( ) ; resumeToken = nextEvent . _id ; return ( nextEvent . documentKey . _id == "sentinel_entry_" + testNum ) ; } , ( ) => { return tojson ( dumpLatestOpLogEntries ( st . rs0 . getPrimary ( ) , 100 ) ) ; } ) ; for ( let changeStreamVariant of changeStreamsVariants ) { const outputChangeStream = { watch : changeStreamVariant . watch , resumeToken : resumeToken , endSentinelEntry : "sentinel_entry_" + ( testNum + 1 ) , testName : testCases . testName } ; outputChangeStreams . push ( outputChangeStream ) ; } testCase . generateOpLogEntry ( shardedColl ) ; testStartTime = createSentinelEntryAndGetTimeStamp ( ++ testNum ) ; } return outputChangeStreams ; }
For each test case and change stream variation, generates the oplog entries to be tested and
 creates an augmented test-case containing a resume token which marks the start of the test, and a
 sentinel entry that marks the end of the test.
For each test case and change stream variation generates the oplog entries to be tested and creates an augmented test - case containing a resume token which marks the start of the test and a sentinel entry that marks the end of the test .
// Find the oplog entry for the document inserted above, and return its timestamp. // We write a sentinel entry before each test case so that the resumed changestreams will have a // known point at which to stop while running each test. // Capture the 'resumeToken' when the sentinel entry is found. We use the token to resume // the stream rather than the 'testStartTime' because resuming from a token adds more stages // to the $changeStream pipeline, which increases our coverage. // Start a change stream on the sentinel entry for each test case. // The termination for the change stream of the current test case will be the // sentinel entry for the next test case. // We copy this for debugging purposes only. // Run the test case's 'generateOpLogEntry' function, which will create the actual oplog // entry to be tested. // Insert a sentinel to separate this test-case from the next.