summaryrefslogtreecommitdiff
path: root/jstests/replsets/change_stream_speculative_majority_latest_oplog_timestamp.js
blob: 2016cf0c6ead3c12f8cbbf2e5b9cb1550b55f763 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
/**
 * Test that change streams using speculative majority wait for the latest observed oplog timestamp
 * to majority commit.
 *
 * If a change stream query returns a batch containing oplog entries no newer than timestamp T, the
 * server may still report a high-water-mark postBatchResumeToken representing the latest majority
 * committed oplog timestamp that it observed while scanning the oplog, which may be greater than T.
 * A mongoS will use this PBRT as a guarantee that no new change events will occur at a lesser
 * timestamp. This guarantee is only valid if the timestamp is actually majority committed, so we
 * need to make sure that guarantee holds, even when using speculative majority.
 *
 * @tags: [uses_speculative_majority]
 */
(function() {
    "use strict";

    load("jstests/libs/write_concern_util.js");  // for [stop|restart]ServerReplication.

    const name = "change_stream_speculative_majority_latest_oplog_timestamp";
    const replTest = new ReplSetTest({
        name: name,
        nodes: [{}, {rsConfig: {priority: 0}}],
        nodeOptions: {enableMajorityReadConcern: 'false'}
    });
    replTest.startSet();
    replTest.initiate();

    const dbName = name;
    const collName = "coll";
    const otherCollName = "coll_other";

    const primary = replTest.getPrimary();
    const secondary = replTest.getSecondary();

    const primaryDB = primary.getDB(dbName);
    const primaryColl = primaryDB[collName];

    assert.commandWorked(primaryColl.insert({_id: 0}, {writeConcern: {w: "majority"}}));

    let res = primaryDB.runCommand(
        {aggregate: collName, pipeline: [{$changeStream: {}}], cursor: {}, maxTimeMS: 5000});

    assert.commandWorked(res);
    let cursorId = res.cursor.id;

    // Insert a document on primary and let it majority commit.
    assert.commandWorked(primaryColl.insert({_id: 1}, {writeConcern: {w: "majority"}}));

    // Pause replication on the secondary so that further writes won't majority commit.
    jsTestLog("Stopping replication to secondary.");
    stopServerReplication(secondary);

    // Receive the first change event.
    res = primary.getDB(dbName).runCommand({getMore: cursorId, collection: collName});
    let changes = res.cursor.nextBatch;
    assert.eq(changes.length, 1);
    assert.eq(changes[0]["fullDocument"], {_id: 1});
    assert.eq(changes[0]["operationType"], "insert");

    // Extract the postBatchResumeToken from the first batch.
    const initialPostBatchResumeToken = res.cursor.postBatchResumeToken;
    assert.neq(initialPostBatchResumeToken, undefined);

    // Do a write on a collection that we are not watching changes for.
    let otherWriteRes = primaryDB.runCommand({insert: otherCollName, documents: [{_id: 1}]});
    let otherWriteOpTime = otherWriteRes.operationTime;

    // Replication to the secondary is paused, so the write to 'otherCollName' cannot majority
    // commit. A change stream getMore is expected to return the "latest oplog timestamp" which it
    // scanned and this timestamp must be majority committed. So, this getMore should time out
    // waiting for the previous write to majority commit, even though it's on a collection that is
    // not being watched.
    res = primary.getDB(dbName).runCommand(
        {getMore: cursorId, collection: collName, maxTimeMS: 5000});
    assert.commandFailedWithCode(res, ErrorCodes.MaxTimeMSExpired);

    jsTestLog("Restarting replication to secondary.");
    restartServerReplication(secondary);
    replTest.awaitReplication();

    // Now that writes can replicate again, the previous operation should have majority committed,
    // making it safe to advance the postBatchResumeToken. Note that no further events are returned,
    // indicating that the new PBRT is a high water mark generated at the latest oplog timestamp.
    res = primary.getDB(dbName).runCommand(
        {getMore: cursorId, collection: collName, maxTimeMS: 5000});
    assert.commandWorked(res);
    assert.eq(res.cursor.nextBatch, []);
    assert.gt(bsonWoCompare(res.cursor.postBatchResumeToken, initialPostBatchResumeToken), 0);

    replTest.stopSet();
})();