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
|
// Basic tests for resuming a $changeStream that is open against all collections in a database.
// Do not run in whole-cluster passthrough since this test assumes that the change stream will be
// invalidated by a database drop.
// @tags: [do_not_run_in_whole_cluster_passthrough]
(function() {
"use strict";
load("jstests/libs/collection_drop_recreate.js"); // For assert[Drop|Create]Collection.
load("jstests/libs/change_stream_util.js"); // For ChangeStreamTest.
load("jstests/libs/fixture_helpers.js"); // For FixtureHelpers.
// Drop and recreate the collections to be used in this set of tests.
const testDB = db.getSiblingDB(jsTestName());
let coll = assertDropAndRecreateCollection(testDB, "resume_coll");
const otherColl = assertDropAndRecreateCollection(testDB, "resume_coll_other");
let cst = new ChangeStreamTest(testDB);
let resumeCursor = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: 1});
// Insert a single document to each collection and save the resume token from the first insert.
assert.commandWorked(coll.insert({_id: 1}));
assert.commandWorked(otherColl.insert({_id: 2}));
const firstInsertChangeDoc = cst.getOneChange(resumeCursor);
assert.docEq({_id: 1}, firstInsertChangeDoc.fullDocument);
assert.eq(firstInsertChangeDoc.ns, {db: testDB.getName(), coll: coll.getName()});
// Test resuming the change stream after the first insert should pick up the insert on the
// second collection.
resumeCursor = cst.startWatchingChanges({
pipeline: [{$changeStream: {resumeAfter: firstInsertChangeDoc._id}}],
collection: 1,
aggregateOptions: {cursor: {batchSize: 0}},
});
const secondInsertChangeDoc = cst.getOneChange(resumeCursor);
assert.docEq({_id: 2}, secondInsertChangeDoc.fullDocument);
assert.eq(secondInsertChangeDoc.ns, {db: testDB.getName(), coll: otherColl.getName()});
// Insert a third document to the first collection and test that the change stream picks it up.
assert.commandWorked(coll.insert({_id: 3}));
const thirdInsertChangeDoc = cst.getOneChange(resumeCursor);
assert.docEq({_id: 3}, thirdInsertChangeDoc.fullDocument);
assert.eq(thirdInsertChangeDoc.ns, {db: testDB.getName(), coll: coll.getName()});
// Test resuming after the first insert again.
resumeCursor = cst.startWatchingChanges({
pipeline: [{$changeStream: {resumeAfter: firstInsertChangeDoc._id}}],
collection: 1,
aggregateOptions: {cursor: {batchSize: 0}},
});
assert.docEq(secondInsertChangeDoc, cst.getOneChange(resumeCursor));
assert.docEq(thirdInsertChangeDoc, cst.getOneChange(resumeCursor));
// Test resume after second insert.
resumeCursor = cst.startWatchingChanges({
pipeline: [{$changeStream: {resumeAfter: secondInsertChangeDoc._id}}],
collection: 1,
aggregateOptions: {cursor: {batchSize: 0}},
});
assert.docEq(thirdInsertChangeDoc, cst.getOneChange(resumeCursor));
// Rename the collection and attempt to resume from the 'rename' notification. Skip this
// test when running on a sharded collection, since these cannot be renamed.
if (!FixtureHelpers.isSharded(coll)) {
assertDropAndRecreateCollection(coll.getDB(), coll.getName());
const renameColl = coll.getDB().getCollection("rename_coll");
assertDropCollection(renameColl.getDB(), renameColl.getName());
resumeCursor = cst.startWatchingChanges({collection: 1, pipeline: [{$changeStream: {}}]});
assert.commandWorked(coll.renameCollection(renameColl.getName()));
const renameChanges = cst.assertNextChangesEqual({
cursor: resumeCursor,
expectedChanges: [
{
operationType: "rename",
ns: {db: coll.getDB().getName(), coll: coll.getName()},
to: {db: renameColl.getDB().getName(), coll: renameColl.getName()}
},
]
});
const resumeTokenRename = renameChanges[0]._id;
// Insert into the renamed collection.
assert.commandWorked(renameColl.insert({_id: "after rename"}));
// Resume from the rename notification using 'resumeAfter' and verify that the change stream
// returns the next insert.
let expectedInsert = {
operationType: "insert",
ns: {db: renameColl.getDB().getName(), coll: renameColl.getName()},
fullDocument: {_id: "after rename"},
documentKey: {_id: "after rename"}
};
resumeCursor = cst.startWatchingChanges(
{collection: 1, pipeline: [{$changeStream: {resumeAfter: resumeTokenRename}}]});
cst.assertNextChangesEqual({cursor: resumeCursor, expectedChanges: expectedInsert});
// Resume from the rename notification using 'startAfter' and verify that the change stream
// returns the next insert.
expectedInsert = {
operationType: "insert",
ns: {db: renameColl.getDB().getName(), coll: renameColl.getName()},
fullDocument: {_id: "after rename"},
documentKey: {_id: "after rename"}
};
resumeCursor = cst.startWatchingChanges(
{collection: 1, pipeline: [{$changeStream: {startAfter: resumeTokenRename}}]});
cst.assertNextChangesEqual({cursor: resumeCursor, expectedChanges: expectedInsert});
// Rename back to the original collection for reliability of the collection drops when
// dropping the database.
assert.commandWorked(renameColl.renameCollection(coll.getName()));
}
// Explicitly drop one collection to ensure reliability of the order of notifications from the
// dropDatabase command.
resumeCursor = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: 1});
assertDropCollection(testDB, otherColl.getName());
const firstCollDrop = cst.getOneChange(resumeCursor);
assert.eq(firstCollDrop.operationType, "drop", tojson(firstCollDrop));
assert.eq(firstCollDrop.ns, {db: testDB.getName(), coll: otherColl.getName()});
// Dropping a database should generate a 'drop' notification for each collection, a
// 'dropDatabase' notification, and finally an 'invalidate'.
assert.commandWorked(testDB.dropDatabase());
const dropDbChanges = cst.assertDatabaseDrop({cursor: resumeCursor, db: testDB});
const secondCollDrop = dropDbChanges[0];
// For sharded passthrough suites, we know that the last entry will be a 'dropDatabase' however
// there may be multiple collection drops in 'dropDbChanges' depending on the number of involved
// shards.
const resumeTokenDropDb = dropDbChanges[dropDbChanges.length - 1]._id;
const resumeTokenInvalidate =
cst.assertNextChangesEqual(
{cursor: resumeCursor, expectedChanges: [{operationType: "invalidate"}]})[0]
._id;
// Test resuming from the first collection drop and the second collection drop as a result of
// dropping the database.
[firstCollDrop, secondCollDrop].forEach(token => {
resumeCursor = cst.startWatchingChanges({
pipeline: [{$changeStream: {resumeAfter: token._id}}],
collection: 1,
aggregateOptions: {cursor: {batchSize: 0}},
});
cst.assertDatabaseDrop({cursor: resumeCursor, db: testDB});
cst.assertNextChangesEqual(
{cursor: resumeCursor, expectedChanges: [{operationType: "invalidate"}]});
});
// Recreate the test collection.
assert.commandWorked(coll.insert({_id: "after recreate"}));
// Test resuming from the 'dropDatabase' entry using 'resumeAfter'.
resumeCursor = cst.startWatchingChanges({
pipeline: [{$changeStream: {resumeAfter: resumeTokenDropDb}}],
collection: 1,
aggregateOptions: {cursor: {batchSize: 0}},
});
cst.assertNextChangesEqual(
{cursor: resumeCursor, expectedChanges: [{operationType: "invalidate"}]});
// Test resuming from the 'invalidate' entry using 'resumeAfter'.
assert.commandFailedWithCode(db.runCommand({
aggregate: 1,
pipeline: [{$changeStream: {resumeAfter: resumeTokenInvalidate}}],
cursor: {},
collation: {locale: "simple"},
}),
ErrorCodes.InvalidResumeToken);
// Test resuming from the 'dropDatabase' entry using 'startAfter'.
resumeCursor = cst.startWatchingChanges({
pipeline: [{$changeStream: {startAfter: resumeTokenDropDb}}],
collection: 1,
aggregateOptions: {cursor: {batchSize: 0}},
});
cst.assertNextChangesEqual(
{cursor: resumeCursor, expectedChanges: [{operationType: "invalidate"}]});
// Test resuming from the 'invalidate' entry using 'startAfter' and verifies it picks up the
// insert after recreating the db/collection.
const expectedInsert = {
operationType: "insert",
ns: {db: testDB.getName(), coll: coll.getName()},
fullDocument: {_id: "after recreate"},
documentKey: {_id: "after recreate"}
};
resumeCursor = cst.startWatchingChanges({
pipeline: [{$changeStream: {startAfter: resumeTokenInvalidate}}],
collection: 1,
aggregateOptions: {cursor: {batchSize: 0}},
});
cst.consumeDropUpTo({
cursor: resumeCursor,
dropType: "dropDatabase",
expectedNext: expectedInsert,
});
cst.cleanUp();
})();
|
