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
|
/**
* Tests that the client can retry commitTransaction on the tenant migration recipient.
*
* @tags: [requires_fcv_47, requires_majority_read_concern, incompatible_with_eft]
*/
(function() {
"use strict";
// Direct writes to config.transactions cannot be part of a session.
TestData.disableImplicitSessions = true;
load("jstests/replsets/libs/tenant_migration_test.js");
load("jstests/replsets/libs/tenant_migration_util.js");
load("jstests/replsets/rslib.js");
load("jstests/libs/uuid_util.js");
const donorRst = new ReplSetTest({
nodes: 1,
name: "donor",
nodeOptions: {
setParameter: {
enableTenantMigrations: true,
// Set the delay before a donor state doc is garbage collected to be short to speed up
// the test.
tenantMigrationGarbageCollectionDelayMS: 3 * 1000,
// Set the TTL monitor to run at a smaller interval to speed up the test.
ttlMonitorSleepSecs: 1,
}
}
});
const recipientRst = new ReplSetTest({
nodes: [{}, {rsConfig: {priority: 0}}, {rsConfig: {priority: 0}}],
name: "recipient",
nodeOptions: {
setParameter: {
enableTenantMigrations: true,
// TODO SERVER-51734: Remove the failpoint 'returnResponseOkForRecipientSyncDataCmd'.
'failpoint.returnResponseOkForRecipientSyncDataCmd': tojson({mode: 'alwaysOn'})
}
}
});
donorRst.startSet();
donorRst.initiate();
recipientRst.startSet();
recipientRst.initiate();
const tenantMigrationTest = new TenantMigrationTest({name: jsTestName(), donorRst, recipientRst});
const kTenantId = "testTenantId";
const kDbName = tenantMigrationTest.tenantDB(kTenantId, "testDB");
const kCollName = "testColl";
const kNs = `${kDbName}.${kCollName}`;
const donorPrimary = donorRst.getPrimary();
const recipientPrimary = recipientRst.getPrimary();
assert.commandWorked(donorPrimary.getCollection(kNs).insert(
[{_id: 0, x: 0}, {_id: 1, x: 1}, {_id: 2, x: 2}], {writeConcern: {w: "majority"}}));
{
jsTest.log("Run a transaction prior to the migration");
const session = donorPrimary.startSession({causalConsistency: false});
const sessionDb = session.getDatabase(kDbName);
const sessionColl = sessionDb[kCollName];
session.startTransaction({writeConcern: {w: "majority"}});
const findAndModifyRes0 = sessionColl.findAndModify({query: {x: 0}, remove: true});
assert.eq({_id: 0, x: 0}, findAndModifyRes0);
assert.commandWorked(session.commitTransaction_forTesting());
assert.sameMembers(sessionColl.find({}).toArray(), [{_id: 1, x: 1}, {_id: 2, x: 2}]);
session.endSession();
}
let txnEntryOnDonor = donorPrimary.getCollection("config.transactions").find().toArray()[0];
jsTest.log("Run a migration to completion");
const migrationId = UUID();
const migrationOpts = {
migrationIdString: extractUUIDFromObject(migrationId),
tenantId: kTenantId,
};
assert.commandWorked(tenantMigrationTest.runMigration(migrationOpts));
const donorDoc =
donorPrimary.getCollection(TenantMigrationTest.kConfigDonorsNS).findOne({tenantId: kTenantId});
assert.commandWorked(tenantMigrationTest.forgetMigration(migrationOpts.migrationIdString));
tenantMigrationTest.waitForMigrationGarbageCollection(donorRst.nodes, migrationId, kTenantId);
{
jsTest.log("Run another transaction after the migration");
const session = donorPrimary.startSession({causalConsistency: false});
const sessionDb = session.getDatabase(kDbName);
const sessionColl = sessionDb[kCollName];
session.startTransaction({writeConcern: {w: "majority"}});
const findAndModifyRes1 = sessionColl.findAndModify({query: {x: 1}, remove: true});
assert.eq({_id: 1, x: 1}, findAndModifyRes1);
assert.commandWorked(session.commitTransaction_forTesting());
assert.sameMembers(sessionColl.find({}).toArray(), [{_id: 2, x: 2}]);
session.endSession();
}
// Test the aggregation pipeline the recipient would use for getting the config.transactions entry
// on the donor. The recipient will use the real startFetchingTimestamp, but this test uses the
// donor's commit timestamp as a substitute.
const startFetchingTimestamp = donorDoc.commitOrAbortOpTime.ts;
const aggRes = donorPrimary.getDB("config").runCommand({
aggregate: "transactions",
pipeline: [
{$match: {"lastWriteOpTime.ts": {$lt: startFetchingTimestamp}, "state": "committed"}},
],
readConcern: {level: "majority", afterClusterTime: startFetchingTimestamp},
hint: "_id_",
cursor: {},
});
assert.eq(1, aggRes.cursor.firstBatch.length);
assert.eq(txnEntryOnDonor, aggRes.cursor.firstBatch[0]);
// Test the client can retry commitTransaction for that transaction that committed prior to the
// migration.
// Insert the config.transactions entry on the recipient, but with a dummy lastWriteOpTime. The
// recipient should not need a real lastWriteOpTime to support a commitTransaction retry.
txnEntryOnDonor.lastWriteOpTime.ts = new Timestamp(0, 0);
assert.commandWorked(
recipientPrimary.getCollection("config.transactions").insert([txnEntryOnDonor]));
recipientRst.awaitLastOpCommitted();
recipientRst.getSecondaries().forEach(node => {
assert.eq(1, node.getCollection("config.transactions").count(txnEntryOnDonor));
});
assert.commandWorked(recipientPrimary.adminCommand({
commitTransaction: 1,
lsid: txnEntryOnDonor._id,
txnNumber: txnEntryOnDonor.txnNum,
autocommit: false
}));
donorRst.stopSet();
recipientRst.stopSet();
})();
|
