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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
|
/*
* Test that a retryable write or retryable internal transaction that is initiated while the
* session has an open retryable internal transaction (i.e. one that has not committed or aborted)
* is blocked until the transaction commits or aborts and does not cause any write statements that
* have already executed to execute again.
*
* @tags: [requires_fcv_60, uses_transactions]
*/
(function() {
"use strict";
load("jstests/libs/fail_point_util.js");
load("jstests/libs/uuid_util.js");
load("jstests/replsets/rslib.js");
load("jstests/sharding/libs/sharded_transactions_helpers.js");
const st = new ShardingTest({shards: 1, rs: {nodes: 2}});
let shard0Primary = st.rs0.getPrimary();
const kDbName = "testDb";
const kCollName = "testColl";
const mongosTestDB = st.s.getDB(kDbName);
const mongosTestColl = mongosTestDB.getCollection(kCollName);
let shard0TestDB = st.rs0.getPrimary().getDB(kDbName);
let shard0TestColl = shard0TestDB.getCollection(kCollName);
assert.commandWorked(shard0TestDB.createCollection(kCollName));
function stepDownShard0Primary() {
const oldPrimary = st.rs0.getPrimary();
const oldSecondary = st.rs0.getSecondary();
assert.commandWorked(oldSecondary.adminCommand({replSetFreeze: 0}));
assert.commandWorked(
oldPrimary.adminCommand({replSetStepDown: ReplSetTest.kForeverSecs, force: true}));
const newPrimary = st.rs0.getPrimary();
assert.neq(oldPrimary, newPrimary);
shard0Primary = newPrimary;
shard0TestDB = shard0Primary.getDB(kDbName);
shard0TestColl = shard0TestDB.getCollection(kCollName);
}
const parentLsid = {
id: UUID()
};
let currentParentTxnNumber = 35;
/*
* Runs a write statement inside a retryable internal transaction and then runs 'retryFunc' to
* resend the write statement while the transaction is still open. Verifies that the retry is
* blocked until the transaction commits or aborts and does not cause the write statement to execute
* more than once.
*/
function testBlockingRetry(retryFunc, testOpts = {
prepareBeforeRetry,
abortAfterBlockingRetry,
stepDownPrimaryAfterBlockingRetry
}) {
jsTest.log("Test blocking retry with test options " + tojson(testOpts));
const parentTxnNumber = currentParentTxnNumber++;
const docToInsert = {x: 1};
const stmtId = 1;
// Start a retryable internal transaction.
const childLsid = {id: parentLsid.id, txnNumber: NumberLong(parentTxnNumber), txnUUID: UUID()};
const childTxnNumber = NumberLong(0);
const originalWriteCmdObj = {
insert: kCollName,
documents: [docToInsert],
lsid: childLsid,
txnNumber: NumberLong(childTxnNumber),
startTransaction: true,
autocommit: false,
stmtId: NumberInt(stmtId),
};
const prepareCmdObj = makePrepareTransactionCmdObj(childLsid, childTxnNumber);
const commitCmdObj = makeCommitTransactionCmdObj(childLsid, childTxnNumber);
const abortCmdObj = makeAbortTransactionCmdObj(childLsid, childTxnNumber);
assert.commandWorked(shard0TestDB.runCommand(originalWriteCmdObj));
if (testOpts.prepareBeforeRetry) {
const preparedTxnRes = assert.commandWorked(shard0Primary.adminCommand(prepareCmdObj));
commitCmdObj.commitTimestamp = preparedTxnRes.prepareTimestamp;
}
// Retry and wait for it to block behind the internal transaction above.
const fp = configureFailPoint(
shard0Primary, "waitAfterNewStatementBlocksBehindOpenInternalTransactionForRetryableWrite");
const retryThread = new Thread(retryFunc, {
shard0RstArgs: createRstArgs(st.rs0),
parentSessionUUIDString: extractUUIDFromObject(parentLsid.id),
parentTxnNumber,
docToInsert,
stmtId,
dbName: kDbName,
collName: kCollName,
stepDownPrimaryAfterBlockingRetry: testOpts.stepDownPrimaryAfterBlockingRetry
});
retryThread.start();
fp.wait();
fp.off();
if (testOpts.stepDownPrimaryAfterBlockingRetry) {
stepDownShard0Primary();
}
// Commit or abort the internal transaction, and verify that the write statement executed
// exactly once despite the concurrent retry.
assert.commandWorked(
shard0TestDB.adminCommand(testOpts.abortAfterBlockingRetry ? abortCmdObj : commitCmdObj));
retryThread.join();
assert.eq(shard0TestColl.count(docToInsert), 1);
assert.commandWorked(mongosTestColl.remove({}));
}
function runTests(retryFunc) {
testBlockingRetry(retryFunc, {
prepareBeforeRetry: false,
abortAfterBlockingRetry: false,
stepDownPrimaryAfterBlockingRetry: false
});
testBlockingRetry(retryFunc, {
prepareBeforeRetry: false,
abortAfterBlockingRetry: true,
stepDownPrimaryAfterBlockingRetry: false
});
testBlockingRetry(retryFunc, {
prepareBeforeRetry: true,
abortAfterBlockingRetry: false,
stepDownPrimaryAfterBlockingRetry: false
});
testBlockingRetry(retryFunc, {
prepareBeforeRetry: true,
abortAfterBlockingRetry: false,
stepDownPrimaryAfterBlockingRetry: true
});
testBlockingRetry(retryFunc, {
prepareBeforeRetry: true,
abortAfterBlockingRetry: true,
stepDownPrimaryAfterBlockingRetry: false
});
testBlockingRetry(retryFunc, {
prepareBeforeRetry: true,
abortAfterBlockingRetry: true,
stepDownPrimaryAfterBlockingRetry: true
});
}
{
jsTest.log(
"Test retrying write statement executed in a retryable internal transaction as a " +
"retryable write in the parent session while the transaction has not been committed " +
"or aborted");
let retryFunc = (testOpts) => {
load("jstests/replsets/rslib.js");
load("jstests/sharding/libs/sharded_transactions_helpers.js");
const shard0Rst = createRst(testOpts.shard0RstArgs);
let shard0Primary = shard0Rst.getPrimary();
const retryWriteCmdObj = {
insert: testOpts.collName,
documents: [testOpts.docToInsert],
lsid: {id: UUID(testOpts.parentSessionUUIDString)},
txnNumber: NumberLong(testOpts.parentTxnNumber),
stmtId: NumberInt(testOpts.stmtId)
};
let retryRes = shard0Primary.getDB(testOpts.dbName).runCommand(retryWriteCmdObj);
if (testOpts.stepDownPrimaryAfterBlockingRetry) {
assert(ErrorCodes.isNotPrimaryError(retryRes.code));
shard0Primary = shard0Rst.getPrimary();
retryRes = shard0Primary.getDB(testOpts.dbName).runCommand(retryWriteCmdObj);
}
assert.commandWorked(retryRes);
};
runTests(retryFunc);
}
{
jsTest.log(
"Test retrying write statement executed in a retryable internal transaction in a " +
"different retryable internal transaction while the original transaction has not been " +
"committed or aborted");
let retryFunc = (testOpts) => {
load("jstests/replsets/rslib.js");
load("jstests/sharding/libs/sharded_transactions_helpers.js");
const shard0Rst = createRst(testOpts.shard0RstArgs);
let shard0Primary = shard0Rst.getPrimary();
const retryChildLsid = {
id: UUID(testOpts.parentSessionUUIDString),
txnNumber: NumberLong(testOpts.parentTxnNumber),
txnUUID: UUID()
};
const retryChildTxnNumber = NumberLong(0);
const retryWriteCmdObj = {
insert: testOpts.collName,
documents: [testOpts.docToInsert],
lsid: retryChildLsid,
txnNumber: NumberLong(retryChildTxnNumber),
startTransaction: true,
autocommit: false,
stmtId: NumberInt(testOpts.stmtId),
};
const retryCommitCmdObj = makeCommitTransactionCmdObj(retryChildLsid, retryChildTxnNumber);
let retryRes = shard0Primary.getDB(testOpts.dbName).runCommand(retryWriteCmdObj);
if (testOpts.stepDownPrimaryAfterBlockingRetry) {
assert(ErrorCodes.isNotPrimaryError(retryRes.code));
shard0Primary = shard0Rst.getPrimary();
retryRes = shard0Primary.getDB(testOpts.dbName).runCommand(retryWriteCmdObj);
}
assert.commandWorked(retryRes);
assert.commandWorked(shard0Primary.getDB(testOpts.dbName).adminCommand(retryCommitCmdObj));
};
runTests(retryFunc);
}
st.stop();
})();
|
