mongo/jstests/sharding/resharding_change_streams_resumability.js

// Tests that change streams on a collection can be resumed during and after the given collection is
// resharded.
//
// @tags: [
//   requires_majority_read_concern,
//   uses_atclustertime,
// ]
import {ChangeStreamTest} from "jstests/libs/query/change_stream_util.js";
import {ReshardingTest} from "jstests/sharding/libs/resharding_test_fixture.js";

// Use a higher frequency for periodic noops to speed up the test.
const reshardingTest = new ReshardingTest({
    numDonors: 2,
    numRecipients: 1,
    reshardInPlace: false,
    periodicNoopIntervalSecs: 1,
    writePeriodicNoops: true,
});
reshardingTest.setup();

const kDbName = "reshardingDb";
const collName = "coll";
const ns = kDbName + "." + collName;

const donorShardNames = reshardingTest.donorShardNames;
const sourceCollection = reshardingTest.createShardedCollection({
    ns: ns,
    shardKeyPattern: {oldKey: 1},
    chunks: [
        {min: {oldKey: MinKey}, max: {oldKey: 0}, shard: donorShardNames[0]},
        {min: {oldKey: 0}, max: {oldKey: MaxKey}, shard: donorShardNames[1]},
    ],
    primaryShardName: donorShardNames[0],
});

const mongos = sourceCollection.getMongo();
const reshardingDb = mongos.getDB(kDbName);

const cst = new ChangeStreamTest(reshardingDb);

// Open a change streams cursor on the collection that will be resharded.
let changeStreamsCursor = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: collName});
assert.eq([], changeStreamsCursor.firstBatch, "Expected cursor not to have changes, but it did");

// We want to confirm that change streams can see events before, during, and after the resharding
// operation. Note in particular that this test confirms that a regular user change stream does
// NOT see internal resharding events such as reshardBegin.
const expectedChanges = [
    {
        documentKey: {_id: 0, oldKey: 0},
        fullDocument: {_id: 0, oldKey: 0},
        ns: {db: kDbName, coll: collName},
        operationType: "insert",
    },
    {
        documentKey: {oldKey: 1, _id: 1},
        fullDocument: {_id: 1, oldKey: 1},
        ns: {db: kDbName, coll: collName},
        operationType: "insert",
    },
    {
        documentKey: {oldKey: 2, _id: 2},
        fullDocument: {_id: 2, oldKey: 2},
        ns: {db: kDbName, coll: collName},
        operationType: "insert",
    },
    {
        documentKey: {newKey: 3, _id: 3},
        fullDocument: {_id: 3, newKey: 3, oldKey: 3},
        ns: {db: kDbName, coll: collName},
        operationType: "insert",
    },
    {
        documentKey: {newKey: 4, _id: 4},
        fullDocument: {_id: 4, newKey: 4, oldKey: 4},
        ns: {db: kDbName, coll: collName},
        operationType: "insert",
    },
];
const preReshardCollectionChange = expectedChanges[0];
const midReshardCollectionChanges = expectedChanges.slice(1, 3);
const postReshardCollectionChanges = expectedChanges.slice(3);

// Verify that the cursor sees changes before the collection is resharded.
assert.commandWorked(sourceCollection.insert({_id: 0, oldKey: 0}));
const preReshardCollectionResumeToken = cst.assertNextChangesEqual({
    cursor: changeStreamsCursor,
    expectedChanges: [preReshardCollectionChange],
})[0]._id;

const recipientShardNames = reshardingTest.recipientShardNames;
let midReshardCollectionResumeToken;
let changeStreamsCursor2;
reshardingTest.withReshardingInBackground(
    //
    {
        // If a donor is also a recipient, the donor state machine will run renameCollection with
        // {dropTarget : true} rather than running drop and letting the recipient state machine run
        // rename at the end of the resharding operation. So, we ensure that only one of the donor
        // shards will also be a recipient shard order to verify that neither the rename with
        // {dropTarget : true} nor the drop command are picked up by the change streams cursor.
        newShardKeyPattern: {newKey: 1},
        newChunks: [{min: {newKey: MinKey}, max: {newKey: MaxKey}, shard: recipientShardNames[0]}],
    },
    () => {
        // Wait until participants are aware of the resharding operation.
        reshardingTest.awaitCloneTimestampChosen();

        // Open another change streams cursor while the collection is being resharded.
        changeStreamsCursor2 = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: collName});

        assert.commandWorked(sourceCollection.insert({_id: 1, oldKey: 1}));
        assert.commandWorked(sourceCollection.insert({_id: 2, oldKey: 2}));

        // Assert that both the cursors see the two new inserts.
        cst.assertNextChangesEqual({cursor: changeStreamsCursor, expectedChanges: midReshardCollectionChanges});
        cst.assertNextChangesEqual({cursor: changeStreamsCursor2, expectedChanges: midReshardCollectionChanges});

        // Check that we can resume from the token returned before resharding began.
        let resumedCursor = cst.startWatchingChanges({
            pipeline: [{$changeStream: {resumeAfter: preReshardCollectionResumeToken}}],
            collection: collName,
        });
        midReshardCollectionResumeToken = cst.assertNextChangesEqual({
            cursor: resumedCursor,
            expectedChanges: midReshardCollectionChanges,
        })[1]._id;
    },
);

assert.commandWorked(sourceCollection.insert({_id: 3, newKey: 3, oldKey: 3}));

// Assert that both the cursor opened before resharding started and the one opened during
// resharding see the insert after resharding has finished.
cst.assertNextChangesEqual({cursor: changeStreamsCursor, expectedChanges: [postReshardCollectionChanges[0]]});
cst.assertNextChangesEqual({cursor: changeStreamsCursor2, expectedChanges: [postReshardCollectionChanges[0]]});

// Check that we can resume from both the token returned before resharding began and the token
// returned during resharding.
let resumedCursorFromPreOperation = cst.startWatchingChanges({
    pipeline: [{$changeStream: {resumeAfter: preReshardCollectionResumeToken}}],
    collection: collName,
});
let midAndPostReshardCollectionChanges = midReshardCollectionChanges.concat(postReshardCollectionChanges);

let resumedCursorFromMidOperation = cst.startWatchingChanges({
    pipeline: [{$changeStream: {resumeAfter: midReshardCollectionResumeToken}}],
    collection: collName,
});

assert.commandWorked(sourceCollection.insert({_id: 4, newKey: 4, oldKey: 4}));

cst.assertNextChangesEqual({
    cursor: resumedCursorFromPreOperation,
    expectedChanges: midAndPostReshardCollectionChanges,
});
cst.assertNextChangesEqual({cursor: resumedCursorFromMidOperation, expectedChanges: postReshardCollectionChanges});

reshardingTest.teardown();