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
158
159
|
/**
* Test the waiting logic of $backupCursorExtend. Given a timestamp T, when
* $backupCursorExtend returns, oplog with T should be majority committed and
* persisent on the disk of that node.
*
* @tags: [
* requires_journaling,
* requires_persistence,
* requires_sharding,
* requires_wiredtiger,
* ]
*/
(function() {
"use strict";
load("jstests/replsets/rslib.js"); // For reconfig, isConfigCommitted and
// safeReconfigShouldFail.
load("jstests/libs/backup_utils.js");
load("jstests/libs/write_concern_util.js");
const DEBUG = false;
const dbName = "test";
const collName = "coll";
const restorePath = MongoRunner.dataPath + "forRestore/";
const numDocs = 2;
let addNodeConfig = function(rst, nodeId, conn, arbiter) {
const config = rst.getReplSetConfigFromNode();
if (arbiter) {
config.members.push({_id: nodeId, host: conn.host, arbiterOnly: true});
} else {
config.members.push({_id: nodeId, host: conn.host});
}
return config;
};
let removeNodeConfig = function(rst, conn) {
const config = rst.getReplSetConfigFromNode();
for (var i = 0; i < config.members.length; i++) {
if (config.members[i].host == conn.host) {
config.members.splice(i, 1);
break;
}
}
return config;
};
function testReconfig(rst, config, shouldSucceed, errCode, errMsg) {
if (shouldSucceed) {
reconfig(rst, config);
assert.soon(() => isConfigCommitted(rst.getPrimary()));
rst.waitForConfigReplication(rst.getPrimary());
rst.awaitReplication();
// rst.await
} else {
safeReconfigShouldFail(rst, config, false /* force */, errCode, errMsg);
// A force reconfig should also fail.
safeReconfigShouldFail(rst, config, true /* force */, errCode, errMsg);
}
}
function insertDoc(db, collName, doc) {
let res = assert.commandWorked(db.runCommand({insert: collName, documents: [doc]}));
assert(res.hasOwnProperty("operationTime"), tojson(res));
return res.operationTime;
}
/*
* Assert that lagged secondary will block when Timestamp T has not been majority committed yet.
*/
function assertLaggedSecondaryGetBlocked() {
resetDbpath(restorePath);
let rst = new ReplSetTest({name: "test", nodes: 1});
rst.startSet();
rst.initiateWithHighElectionTimeout();
const primaryDB = rst.getPrimary().getDB(dbName);
print("Ahoo0 ==> Insert Docs to Primary");
for (let i = 0; i < 1000; i++) {
insertDoc(primaryDB, collName, {k: i});
}
print("Ahoo0 ==> AddSecondary");
testReconfig(rst,
addNodeConfig(rst, 1 /* nodeId */, rst.add() /* conn */, false /* arbiter */),
true /* shouldSucceed */);
rst.stopSet();
return;
let cursor = openBackupCursor(rst.getSecondary());
// let firstBatch = cursor.next();
let firstBatch = undefined;
while(cursor.hasNext()) {
let batch = cursor.next();
print("Ahoo1 ==> ", tojson(batch));
if (!firstBatch) {
firstBatch = batch;
}
}
print("Ahoo2 --> first batch: ", tojson(firstBatch));
let checkpointTimestamp = firstBatch.metadata["checkpointTimestamp"];
const backupId = firstBatch.metadata.backupId;
print("ahoo2 -> ", tojson(checkpointTimestamp), " "+ backupId);
jsTestLog("Start writes on primary");
let clusterTime;
for (let i = 0; i < numDocs - 1; i++) {
clusterTime = insertDoc(primaryDB, collName, {a: i});
}
print("Ahoo3 ==> clusterTime: ", tojson(clusterTime));
let extendCursor = extendBackupCursor(rst.getSecondary(), backupId, clusterTime);
while(extendCursor.hasNext()) {
let batch = extendCursor.next();
print("Ahoo3 ==> ", tojson(batch));
}
jsTestLog("Start writes on primary");
for (let i = 0; i < numDocs - 1; i++) {
clusterTime = insertDoc(primaryDB, collName, {b: i});
}
print("Ahoo4 ==> clusterTime: ", tojson(clusterTime));
extendCursor = extendBackupCursor(rst.getSecondary(), backupId, clusterTime);
while(extendCursor.hasNext()) {
let batch = extendCursor.next();
print("Ahoo4 ==> ", tojson(batch));
}
jsTestLog("Start writes on primary");
for (let i = 0; i < numDocs - 1; i++) {
clusterTime = insertDoc(primaryDB, collName, {c: i});
}
print("Ahoo5 ==> clusterTime: ", tojson(clusterTime));
extendCursor = extendBackupCursor(rst.getSecondary(), backupId, clusterTime);
while(extendCursor.hasNext()) {
let batch = extendCursor.next();
print("Ahoo5 ==> ", tojson(batch));
}
cursor.close();
cursor = openBackupCursor(rst.getSecondary());
// let firstBatch = cursor.next();
while(cursor.hasNext()) {
let batch = cursor.next();
print("Ahoo6 ==> ", tojson(batch));
}
cursor.close();
rst.stopSet();
}
assertLaggedSecondaryGetBlocked();
})();
|
