jstests/noPassthrough/libs/change_stream_pre_image_time_based_expiration_utils.js
// 'primary' represents the connection to the shard primary node.
'primary' represents the connection to the shard primary node.
' primary ' represents the connection to the shard primary node .
function testTimeBasedPreImageRetentionPolicy(conn, primary) {
    const shouldExpire = "shouldExpire";
    const shouldRetain = "shouldRetain";
    const docsStatePerCollection = [
        [shouldRetain],
        [shouldExpire],
        [shouldExpire, shouldRetain],
        [shouldExpire, shouldExpire],
        [shouldExpire],
        [shouldExpire, shouldExpire],
        [shouldRetain, shouldRetain],
        [shouldRetain],
        [shouldExpire]
    ];
    const collectionCount = docsStatePerCollection.length;
    const testDB = primary.getDB("test");
    for (let collIdx = 0; collIdx < collectionCount; collIdx++) {
        const collName = "coll" + collIdx;
        assert.commandWorked(
            testDB.createCollection(collName, {changeStreamPreAndPostImages: {enabled: true}}));
    }
    let collectionsInfo = testDB.getCollectionInfos();
    assert.eq(collectionsInfo.length, collectionCount);
    collectionsInfo.sort((coll1, coll2) => {
        return coll1.info.uuid <= coll2.info.uuid ? -1 : 1;
    });
    assert.commandWorked(conn.getDB("admin").runCommand({
        setClusterParameter: {changeStreamOptions: {preAndPostImages: {expireAfterSeconds: "off"}}}
    }));
    let shouldRetainDocs = [];
    let shouldExpireDocs = [];
    let allDocs = [];
    let documentId = 0;
    for (let collIdx = 0; collIdx < collectionCount; collIdx++) {
        const collName = collectionsInfo[collIdx]["name"];
        const coll = testDB.getCollection(collName);
        const docs = docsStatePerCollection[collIdx];
        for (let docIdx = 0; docIdx < docs.length; docIdx++) {
            assert.commandWorked(
                coll.insert({_id: documentId}, {$set: {documentState: "inserted"}}));
            const documentState = docs[docIdx];
            allDocs.push([collIdx, documentId]);
            if (documentState !== shouldExpire) {
                shouldRetainDocs.push([collIdx, documentId]);
            } else {
                shouldExpireDocs.push([collIdx, documentId]);
            }
            ++documentId;
        }
    }
    const updateDocument = (docInfo) => {
        const [collIdx, docId] = docInfo;
        const collName = collectionsInfo[collIdx]["name"];
        const coll = testDB.getCollection(collName);
        assert.commandWorked(coll.updateOne({_id: docId}, {$set: {documentState: "updated"}}));
    };
    shouldExpireDocs.forEach(updateDocument);
    const preImageColl = primary.getDB("config").getCollection("system.preimages");
    const expireAfterSeconds = 1;
    verifyPreImages(preImageColl, shouldExpireDocs, collectionsInfo);
    const lastPreImageToExpire = preImageColl.find().sort({"_id.ts": -1}).limit(1).toArray();
    assert.eq(lastPreImageToExpire.length, 1, lastPreImageToExpire);
    const preImageShouldExpireAfter =
        lastPreImageToExpire[0].operationTime.getTime() + expireAfterSeconds * 1000;
    const currentTime = new Date(preImageShouldExpireAfter + 1);
    sleep(1);
    shouldRetainDocs.forEach(updateDocument);
    verifyPreImages(preImageColl, allDocs, collectionsInfo);
    const currentTimeFailPoint =
        configureFailPoint(primary,
                           "changeStreamPreImageRemoverCurrentTime",
                           {currentTimeForTimeBasedExpiration: currentTime});
    currentTimeFailPoint.wait({timesEntered: 2});
    verifyPreImages(preImageColl, allDocs, collectionsInfo);
    assert.commandWorked(conn.getDB("admin").runCommand({
        setClusterParameter:
            {changeStreamOptions: {preAndPostImages: {expireAfterSeconds: expireAfterSeconds}}}
    }));
    assert.soon(() => {
        return preImageColl.find().toArray().length == shouldRetainDocs.length;
    });
    verifyPreImages(preImageColl, shouldRetainDocs, collectionsInfo);
    currentTimeFailPoint.off();
}
function testTimeBasedPreImageRetentionPolicy ( conn , primary ) { const shouldExpire = "shouldExpire" ; const shouldRetain = "shouldRetain" ; const docsStatePerCollection = [ [ shouldRetain ] , [ shouldExpire ] , [ shouldExpire , shouldRetain ] , [ shouldExpire , shouldExpire ] , [ shouldExpire ] , [ shouldExpire , shouldExpire ] , [ shouldRetain , shouldRetain ] , [ shouldRetain ] , [ shouldExpire ] ] ; const collectionCount = docsStatePerCollection . length ; const testDB = primary . getDB ( "test" ) ; for ( let collIdx = 0 ; collIdx < collectionCount ; collIdx ++ ) { const collName = "coll" + collIdx ; assert . commandWorked ( testDB . createCollection ( collName , { changeStreamPreAndPostImages : { enabled : true } } ) ) ; } let collectionsInfo = testDB . getCollectionInfos ( ) ; assert . eq ( collectionsInfo . length , collectionCount ) ; collectionsInfo . sort ( ( coll1 , coll2 ) => { return coll1 . info . uuid <= coll2 . info . uuid ? - 1 : 1 ; } ) ; assert . commandWorked ( conn . getDB ( "admin" ) . runCommand ( { setClusterParameter : { changeStreamOptions : { preAndPostImages : { expireAfterSeconds : "off" } } } } ) ) ; let shouldRetainDocs = [ ] ; let shouldExpireDocs = [ ] ; let allDocs = [ ] ; let documentId = 0 ; for ( let collIdx = 0 ; collIdx < collectionCount ; collIdx ++ ) { const collName = collectionsInfo [ collIdx ] [ "name" ] ; const coll = testDB . getCollection ( collName ) ; const docs = docsStatePerCollection [ collIdx ] ; for ( let docIdx = 0 ; docIdx < docs . length ; docIdx ++ ) { assert . commandWorked ( coll . insert ( { _id : documentId } , { $set : { documentState : "inserted" } } ) ) ; const documentState = docs [ docIdx ] ; allDocs . push ( [ collIdx , documentId ] ) ; if ( documentState !== shouldExpire ) { shouldRetainDocs . push ( [ collIdx , documentId ] ) ; } else { shouldExpireDocs . push ( [ collIdx , documentId ] ) ; } ++ documentId ; } } const updateDocument = ( docInfo ) => { const [ collIdx , docId ] = docInfo ; const collName = collectionsInfo [ collIdx ] [ "name" ] ; const coll = testDB . getCollection ( collName ) ; assert . commandWorked ( coll . updateOne ( { _id : docId } , { $set : { documentState : "updated" } } ) ) ; } ; shouldExpireDocs . forEach ( updateDocument ) ; const preImageColl = primary . getDB ( "config" ) . getCollection ( "system.preimages" ) ; const expireAfterSeconds = 1 ; verifyPreImages ( preImageColl , shouldExpireDocs , collectionsInfo ) ; const lastPreImageToExpire = preImageColl . find ( ) . sort ( { "_id.ts" : - 1 } ) . limit ( 1 ) . toArray ( ) ; assert . eq ( lastPreImageToExpire . length , 1 , lastPreImageToExpire ) ; const preImageShouldExpireAfter = lastPreImageToExpire [ 0 ] . operationTime . getTime ( ) + expireAfterSeconds * 1000 ; const currentTime = new Date ( preImageShouldExpireAfter + 1 ) ; sleep ( 1 ) ; shouldRetainDocs . forEach ( updateDocument ) ; verifyPreImages ( preImageColl , allDocs , collectionsInfo ) ; const currentTimeFailPoint = configureFailPoint ( primary , "changeStreamPreImageRemoverCurrentTime" , { currentTimeForTimeBasedExpiration : currentTime } ) ; currentTimeFailPoint . wait ( { timesEntered : 2 } ) ; verifyPreImages ( preImageColl , allDocs , collectionsInfo ) ; assert . commandWorked ( conn . getDB ( "admin" ) . runCommand ( { setClusterParameter : { changeStreamOptions : { preAndPostImages : { expireAfterSeconds : expireAfterSeconds } } } } ) ) ; assert . soon ( ( ) => { return preImageColl . find ( ) . toArray ( ) . length == shouldRetainDocs . length ; } ) ; verifyPreImages ( preImageColl , shouldRetainDocs , collectionsInfo ) ; currentTimeFailPoint . off ( ) ; }
'primary' represents the connection to the shard primary node.
' primary ' represents the connection to the shard primary node .
// Status for pre-images that define if pre-image is expected to expire or not. // Each element defines a sequence of documents belonging to one collection. // Each documents has has associated status - 'expire' or 'retain'. This structure models the // state of the pre-images associated with a particular document of the collection, ie. the // pre-images corresponding to the documents 'shouldExpire' will be deleted by the remover job // and those with 'shouldRetain' will be retained. // Create several collections with pre- and post-images enabled. // Get the collections information and sort them by uuid. The pre-image documents are naturally // sorted first by uuid and then by timestamp in a pre-images collection. Sorting of collection // helps in setting up of the pre-images in the required order. The 'collectionsInfo' has // one-to-one mapping with 'docsStatePerCollection'. // Disable pre-image time-based expiration policy. // Insert documents to each collection. Iterate through the documents and group them as // 'shouldExpire' or 'shouldRetain' based on the document states. Each element of these groups // is an array, where the first element is the collection index and the second is the document // id. // Helper to update the document in the collection. // Update each document that should expire, this will create pre-images. // Verify that pre-images to be expired is recorded. // Get the last pre-image that should expire and compute the current time using that. The // current time is computed by adding (expireAfterSeconds + 0.001) seconds to the operation time // of the last recorded pre-image. // Sleep for 1 ms before doing the next updates. The will ensure that the difference in // operation time between pre-images to be retained and pre-images to be expired is at least // 1 ms. // Update each document for which pre-images will be retained. These pre-images will be ordered // after pre-images that should expire for a particular collection. // Verify that all pre-images are recorded. // Configure the current time for the pre-image remover job. At this point, the time-based // pre-image expiration is still disabled. // Wait until at least 1 complete cycle of pre-image removal job is completed. // Verify that when time-based pre-image expiration disabled, no pre-images are not deleted. // Enable time-based pre-image expiration and configure the 'expireAfterSeconds' to 1 seconds. // Verify that at some point in time, all expired pre-images will be deleted. // Verify that pre-images corresponding to documents with document states 'shouldRetain' are // present.