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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
|
/**
* Test speculative majority reads using the 'find' command.
*
* Speculative majority reads allow the server to provide "majority" read guarantees without storage
* engine support for reading from a historical snapshot. Instead of reading historical, majority
* committed data, we just read the newest data available on a node, and then, before returning to a
* client, block until we know the data has become majority committed. Currently this is an internal
* feature used only by change streams.
*
* @tags: [uses_speculative_majority]
*/
(function() {
"use strict";
load("jstests/libs/write_concern_util.js"); // for [stop|restart]ServerReplication.
load("jstests/libs/parallelTester.js"); // for ScopedThread.
let name = "speculative_majority_find";
let replTest = new ReplSetTest({
name: name,
nodes: [{}, {rsConfig: {priority: 0}}],
nodeOptions: {enableMajorityReadConcern: 'false'}
});
replTest.startSet();
replTest.initiate();
let dbName = name;
let collName = "coll";
let primary = replTest.getPrimary();
let secondary = replTest.getSecondary();
let primaryDB = primary.getDB(dbName);
let secondaryDB = secondary.getDB(dbName);
let primaryColl = primaryDB[collName];
// Create a collection.
assert.commandWorked(primaryColl.insert({}, {writeConcern: {w: "majority"}}));
//
// Test basic reads with speculative majority.
//
// Pause replication on the secondary so that writes won't majority commit.
stopServerReplication(secondary);
assert.commandWorked(primaryColl.insert({_id: 1}));
jsTestLog("Do a speculative majority read that should time out.");
let res = primaryDB.runCommand({
find: collName,
readConcern: {level: "majority"},
filter: {_id: 1},
allowSpeculativeMajorityRead: true,
maxTimeMS: 5000
});
assert.commandFailedWithCode(res, ErrorCodes.MaxTimeMSExpired);
restartServerReplication(secondary);
replTest.awaitReplication();
jsTestLog("Do a speculative majority read that should succeed.");
res = primaryDB.runCommand({
find: collName,
readConcern: {level: "majority"},
filter: {_id: 1},
allowSpeculativeMajorityRead: true
});
assert.commandWorked(res);
assert.eq(res.cursor.firstBatch.length, 1);
assert.eq(res.cursor.firstBatch[0], {_id: 1});
//
// Test that blocked reads can succeed when a write majority commits.
//
// Pause replication on the secondary so that writes won't majority commit.
stopServerReplication(secondary);
assert.commandWorked(primaryColl.insert({_id: 2}));
jsTestLog("Do a speculative majority that should block until write commits.");
let speculativeRead = new ScopedThread(function(host, dbName, collName) {
const nodeDB = new Mongo(host).getDB(dbName);
return nodeDB.runCommand({
find: collName,
readConcern: {level: "majority"},
filter: {_id: 2},
allowSpeculativeMajorityRead: true
});
}, primary.host, dbName, collName);
speculativeRead.start();
// Wait for the read to start on the server.
assert.soon(() => primaryDB.currentOp({ns: primaryColl.getFullName(), "command.find": collName})
.inprog.length === 1);
// Let the previous write commit.
restartServerReplication(secondary);
assert.commandWorked(
primaryColl.insert({_id: "commit_last_write"}, {writeConcern: {w: "majority"}}));
// Make sure the read finished and returned correct results.
speculativeRead.join();
res = speculativeRead.returnData();
assert.commandWorked(res);
assert.eq(res.cursor.firstBatch.length, 1);
assert.eq(res.cursor.firstBatch[0], {_id: 2});
//
// Test 'afterClusterTime' reads with speculative majority.
//
stopServerReplication(secondary);
// Insert a document on the primary and record the response.
let writeRes = primaryDB.runCommand({insert: collName, documents: [{_id: 3}]});
assert.commandWorked(writeRes);
jsTestLog(
"Do a speculative majority read on primary with 'afterClusterTime' that should time out.");
res = primaryDB.runCommand({
find: collName,
readConcern: {level: "majority", afterClusterTime: writeRes.operationTime},
filter: {_id: 3},
$clusterTime: writeRes.$clusterTime,
allowSpeculativeMajorityRead: true,
maxTimeMS: 5000
});
assert.commandFailedWithCode(res, ErrorCodes.MaxTimeMSExpired);
jsTestLog(
"Do a speculative majority read on secondary with 'afterClusterTime' that should time out.");
res = secondaryDB.runCommand({
find: collName,
readConcern: {level: "majority", afterClusterTime: writeRes.operationTime},
filter: {_id: 3},
$clusterTime: writeRes.$clusterTime,
allowSpeculativeMajorityRead: true,
maxTimeMS: 5000
});
assert.commandFailedWithCode(res, ErrorCodes.MaxTimeMSExpired);
// Let the previous write majority commit.
restartServerReplication(secondary);
replTest.awaitReplication();
jsTestLog("Do a speculative majority read with 'afterClusterTime' that should succeed.");
res = primaryDB.runCommand({
find: collName,
readConcern: {level: "majority", afterClusterTime: writeRes.operationTime},
filter: {_id: 3},
$clusterTime: res.$clusterTime,
allowSpeculativeMajorityRead: true
});
assert.commandWorked(res);
assert.eq(res.cursor.firstBatch.length, 1);
assert.eq(res.cursor.firstBatch[0], {_id: 3});
replTest.stopSet();
})();
|
