1 // record_store_v1_capped_test.cpp
2 
3 
4 /**
5  *    Copyright (C) 2018-present MongoDB, Inc.
6  *
7  *    This program is free software: you can redistribute it and/or modify
8  *    it under the terms of the Server Side Public License, version 1,
9  *    as published by MongoDB, Inc.
10  *
11  *    This program is distributed in the hope that it will be useful,
12  *    but WITHOUT ANY WARRANTY; without even the implied warranty of
13  *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14  *    Server Side Public License for more details.
15  *
16  *    You should have received a copy of the Server Side Public License
17  *    along with this program. If not, see
18  *    <http://www.mongodb.com/licensing/server-side-public-license>.
19  *
20  *    As a special exception, the copyright holders give permission to link the
21  *    code of portions of this program with the OpenSSL library under certain
22  *    conditions as described in each individual source file and distribute
23  *    linked combinations including the program with the OpenSSL library. You
24  *    must comply with the Server Side Public License in all respects for
25  *    all of the code used other than as permitted herein. If you modify file(s)
26  *    with this exception, you may extend this exception to your version of the
27  *    file(s), but you are not obligated to do so. If you do not wish to do so,
28  *    delete this exception statement from your version. If you delete this
29  *    exception statement from all source files in the program, then also delete
30  *    it in the license file.
31  */
32 
33 #include "mongo/db/storage/mmap_v1/record_store_v1_capped.h"
34 #include "mongo/db/storage/mmap_v1/record_store_v1_capped_iterator.h"
35 
36 #include "mongo/db/operation_context_noop.h"
37 #include "mongo/db/storage/mmap_v1/extent.h"
38 #include "mongo/db/storage/mmap_v1/record.h"
39 #include "mongo/db/storage/mmap_v1/record_store_v1_test_help.h"
40 
41 #include "mongo/unittest/unittest.h"
42 
43 using namespace mongo;
44 
45 namespace {
46 
47 using std::string;
48 using std::vector;
49 
50 // Provides data to be inserted. Must be large enough for largest possible record.
51 // Should be in BSS so unused portions should be free.
52 char zeros[20 * 1024 * 1024] = {};
53 
54 class DummyCappedCallback : public CappedCallback {
55 public:
aboutToDeleteCapped(OperationContext * opCtx,const RecordId & loc,RecordData data)56     Status aboutToDeleteCapped(OperationContext* opCtx, const RecordId& loc, RecordData data) {
57         deleted.push_back(DiskLoc::fromRecordId(loc));
58         return Status::OK();
59     }
60 
haveCappedWaiters()61     bool haveCappedWaiters() {
62         return false;
63     }
notifyCappedWaitersIfNeeded()64     void notifyCappedWaitersIfNeeded() {}
65 
66     vector<DiskLoc> deleted;
67 };
68 
simpleInsertTest(const char * buf,int size)69 void simpleInsertTest(const char* buf, int size) {
70     OperationContextNoop opCtx;
71     DummyExtentManager em;
72     DummyRecordStoreV1MetaData* md = new DummyRecordStoreV1MetaData(true, 0);
73     DummyCappedCallback cb;
74 
75     string myns = "test.simple1";
76     CappedRecordStoreV1 rs(&opCtx, &cb, myns, md, &em, false);
77 
78     rs.increaseStorageSize(&opCtx, 1024, false);
79 
80     ASSERT_NOT_OK(rs.insertRecord(&opCtx, buf, 3, Timestamp(), true).getStatus());
81 
82     ASSERT_OK(rs.insertRecord(&opCtx, buf, size, Timestamp(), true).getStatus());
83 
84     {
85         BSONObjBuilder b;
86         int64_t storageSize = rs.storageSize(&opCtx, &b);
87         BSONObj obj = b.obj();
88         ASSERT_EQUALS(1, obj["numExtents"].numberInt());
89         ASSERT_EQUALS(storageSize, em.quantizeExtentSize(1024));
90     }
91 
92     for (int i = 0; i < 1000; i++) {
93         ASSERT_OK(rs.insertRecord(&opCtx, buf, size, Timestamp(), true).getStatus());
94     }
95 
96     long long start = md->numRecords();
97     for (int i = 0; i < 1000; i++) {
98         ASSERT_OK(rs.insertRecord(&opCtx, buf, size, Timestamp(), true).getStatus());
99     }
100     ASSERT_EQUALS(start, md->numRecords());
101     ASSERT_GREATER_THAN(start, 100);
102     ASSERT_LESS_THAN(start, 1000);
103 }
104 
TEST(CappedRecordStoreV1,SimpleInsertSize4)105 TEST(CappedRecordStoreV1, SimpleInsertSize4) {
106     simpleInsertTest("abcd", 4);
107 }
TEST(CappedRecordStoreV1,SimpleInsertSize8)108 TEST(CappedRecordStoreV1, SimpleInsertSize8) {
109     simpleInsertTest("abcdefgh", 8);
110 }
111 
TEST(CappedRecordStoreV1,EmptySingleExtent)112 TEST(CappedRecordStoreV1, EmptySingleExtent) {
113     OperationContextNoop opCtx;
114     DummyExtentManager em;
115     DummyRecordStoreV1MetaData* md = new DummyRecordStoreV1MetaData(true, 0);
116     DummyCappedCallback cb;
117     CappedRecordStoreV1 rs(&opCtx, &cb, "test.foo", md, &em, false);
118 
119     {
120         LocAndSize records[] = {{}};
121         LocAndSize drecs[] = {{DiskLoc(0, 1000), 1000}, {}};
122         md->setCapExtent(&opCtx, DiskLoc(0, 0));
123         md->setCapFirstNewRecord(&opCtx, DiskLoc().setInvalid());
124         initializeV1RS(&opCtx, records, drecs, NULL, &em, md);
125     }
126 
127     ASSERT_OK(
128         rs.insertRecord(&opCtx, zeros, 100 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
129             .getStatus());
130 
131 
132     {
133         LocAndSize recs[] = {{DiskLoc(0, 1000), 100}, {}};
134         LocAndSize drecs[] = {{DiskLoc(0, 1100), 900}, {}};
135         assertStateV1RS(&opCtx, recs, drecs, NULL, &em, md);
136         ASSERT_EQUALS(md->capExtent(), DiskLoc(0, 0));
137         ASSERT_EQUALS(md->capFirstNewRecord(), DiskLoc().setInvalid());  // unlooped
138     }
139 }
140 
TEST(CappedRecordStoreV1,FirstLoopWithSingleExtentExactSize)141 TEST(CappedRecordStoreV1, FirstLoopWithSingleExtentExactSize) {
142     OperationContextNoop opCtx;
143     DummyExtentManager em;
144     DummyRecordStoreV1MetaData* md = new DummyRecordStoreV1MetaData(true, 0);
145     DummyCappedCallback cb;
146     CappedRecordStoreV1 rs(&opCtx, &cb, "test.foo", md, &em, false);
147 
148     {
149         LocAndSize records[] = {{DiskLoc(0, 1000), 100},
150                                 {DiskLoc(0, 1100), 100},
151                                 {DiskLoc(0, 1200), 100},
152                                 {DiskLoc(0, 1300), 100},
153                                 {DiskLoc(0, 1400), 100},
154                                 {}};
155         LocAndSize drecs[] = {{DiskLoc(0, 1500), 50}, {}};
156         md->setCapExtent(&opCtx, DiskLoc(0, 0));
157         md->setCapFirstNewRecord(&opCtx, DiskLoc().setInvalid());  // unlooped
158         initializeV1RS(&opCtx, records, drecs, NULL, &em, md);
159     }
160 
161     ASSERT_OK(
162         rs.insertRecord(&opCtx, zeros, 100 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
163             .getStatus());
164 
165     {
166         LocAndSize recs[] = {{DiskLoc(0, 1200), 100},  // first old record
167                              {DiskLoc(0, 1300), 100},
168                              {DiskLoc(0, 1400), 100},  // last old record
169                              {DiskLoc(0, 1000), 100},  // first new record
170                              {}};
171         LocAndSize drecs[] = {
172             {DiskLoc(0, 1100), 100},  // gap after newest record XXX this is probably a bug
173             {DiskLoc(0, 1500), 50},   // gap at end of extent
174             {}};
175         assertStateV1RS(&opCtx, recs, drecs, NULL, &em, md);
176         ASSERT_EQUALS(md->capExtent(), DiskLoc(0, 0));
177         ASSERT_EQUALS(md->capFirstNewRecord(), DiskLoc(0, 1000));
178     }
179 }
180 
TEST(CappedRecordStoreV1,NonFirstLoopWithSingleExtentExactSize)181 TEST(CappedRecordStoreV1, NonFirstLoopWithSingleExtentExactSize) {
182     OperationContextNoop opCtx;
183     DummyExtentManager em;
184     DummyRecordStoreV1MetaData* md = new DummyRecordStoreV1MetaData(true, 0);
185     DummyCappedCallback cb;
186     CappedRecordStoreV1 rs(&opCtx, &cb, "test.foo", md, &em, false);
187 
188     {
189         LocAndSize records[] = {{DiskLoc(0, 1000), 100},
190                                 {DiskLoc(0, 1100), 100},
191                                 {DiskLoc(0, 1200), 100},
192                                 {DiskLoc(0, 1300), 100},
193                                 {DiskLoc(0, 1400), 100},
194                                 {}};
195         LocAndSize drecs[] = {{DiskLoc(0, 1500), 50}, {}};
196         md->setCapExtent(&opCtx, DiskLoc(0, 0));
197         md->setCapFirstNewRecord(&opCtx, DiskLoc(0, 1000));
198         initializeV1RS(&opCtx, records, drecs, NULL, &em, md);
199     }
200 
201     ASSERT_OK(
202         rs.insertRecord(&opCtx, zeros, 100 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
203             .getStatus());
204 
205     {
206         LocAndSize recs[] = {{DiskLoc(0, 1200), 100},  // first old record
207                              {DiskLoc(0, 1300), 100},
208                              {DiskLoc(0, 1400), 100},  // last old record
209                              {DiskLoc(0, 1000), 100},  // first new record
210                              {}};
211         LocAndSize drecs[] = {
212             {DiskLoc(0, 1100), 100},  // gap after newest record XXX this is probably a bug
213             {DiskLoc(0, 1500), 50},   // gap at end of extent
214             {}};
215         assertStateV1RS(&opCtx, recs, drecs, NULL, &em, md);
216         ASSERT_EQUALS(md->capExtent(), DiskLoc(0, 0));
217         ASSERT_EQUALS(md->capFirstNewRecord(), DiskLoc(0, 1000));
218     }
219 }
220 
221 /**
222  * Current code always tries to leave 24 bytes to create a DeletedRecord.
223  */
TEST(CappedRecordStoreV1,WillLoopWithout24SpareBytes)224 TEST(CappedRecordStoreV1, WillLoopWithout24SpareBytes) {
225     OperationContextNoop opCtx;
226     DummyExtentManager em;
227     DummyRecordStoreV1MetaData* md = new DummyRecordStoreV1MetaData(true, 0);
228     DummyCappedCallback cb;
229     CappedRecordStoreV1 rs(&opCtx, &cb, "test.foo", md, &em, false);
230 
231     {
232         LocAndSize records[] = {{DiskLoc(0, 1000), 100},
233                                 {DiskLoc(0, 1100), 100},
234                                 {DiskLoc(0, 1200), 100},
235                                 {DiskLoc(0, 1300), 100},
236                                 {DiskLoc(0, 1400), 100},
237                                 {}};
238         LocAndSize drecs[] = {{DiskLoc(0, 1500), 123}, {}};
239         md->setCapExtent(&opCtx, DiskLoc(0, 0));
240         md->setCapFirstNewRecord(&opCtx, DiskLoc(0, 1000));
241         initializeV1RS(&opCtx, records, drecs, NULL, &em, md);
242     }
243 
244     ASSERT_OK(
245         rs.insertRecord(&opCtx, zeros, 100 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
246             .getStatus());
247 
248     {
249         LocAndSize recs[] = {{DiskLoc(0, 1200), 100},  // first old record
250                              {DiskLoc(0, 1300), 100},
251                              {DiskLoc(0, 1400), 100},  // last old record
252                              {DiskLoc(0, 1000), 100},  // first new record
253                              {}};
254         LocAndSize drecs[] = {{DiskLoc(0, 1100), 100},  // gap after newest record
255                               {DiskLoc(0, 1500), 123},  // gap at end of extent
256                               {}};
257         assertStateV1RS(&opCtx, recs, drecs, NULL, &em, md);
258         ASSERT_EQUALS(md->capExtent(), DiskLoc(0, 0));
259         ASSERT_EQUALS(md->capFirstNewRecord(), DiskLoc(0, 1000));
260     }
261 }
262 
TEST(CappedRecordStoreV1,WontLoopWith24SpareBytes)263 TEST(CappedRecordStoreV1, WontLoopWith24SpareBytes) {
264     OperationContextNoop opCtx;
265     DummyExtentManager em;
266     DummyRecordStoreV1MetaData* md = new DummyRecordStoreV1MetaData(true, 0);
267     DummyCappedCallback cb;
268     CappedRecordStoreV1 rs(&opCtx, &cb, "test.foo", md, &em, false);
269 
270     {
271         LocAndSize records[] = {{DiskLoc(0, 1000), 100},
272                                 {DiskLoc(0, 1100), 100},
273                                 {DiskLoc(0, 1200), 100},
274                                 {DiskLoc(0, 1300), 100},
275                                 {DiskLoc(0, 1400), 100},
276                                 {}};
277         LocAndSize drecs[] = {{DiskLoc(0, 1500), 124}, {}};
278         md->setCapExtent(&opCtx, DiskLoc(0, 0));
279         md->setCapFirstNewRecord(&opCtx, DiskLoc(0, 1000));
280         initializeV1RS(&opCtx, records, drecs, NULL, &em, md);
281     }
282 
283     ASSERT_OK(
284         rs.insertRecord(&opCtx, zeros, 100 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
285             .getStatus());
286 
287     {
288         LocAndSize recs[] = {{DiskLoc(0, 1000), 100},
289                              {DiskLoc(0, 1100), 100},
290                              {DiskLoc(0, 1200), 100},
291                              {DiskLoc(0, 1300), 100},
292                              {DiskLoc(0, 1400), 100},
293                              {DiskLoc(0, 1500), 100},
294                              {}};
295         LocAndSize drecs[] = {{DiskLoc(0, 1600), 24},  // gap at end of extent
296                               {}};
297         assertStateV1RS(&opCtx, recs, drecs, NULL, &em, md);
298         ASSERT_EQUALS(md->capExtent(), DiskLoc(0, 0));
299         ASSERT_EQUALS(md->capFirstNewRecord(), DiskLoc(0, 1000));
300     }
301 }
302 
TEST(CappedRecordStoreV1,MoveToSecondExtentUnLooped)303 TEST(CappedRecordStoreV1, MoveToSecondExtentUnLooped) {
304     OperationContextNoop opCtx;
305     DummyExtentManager em;
306     DummyRecordStoreV1MetaData* md = new DummyRecordStoreV1MetaData(true, 0);
307     DummyCappedCallback cb;
308     CappedRecordStoreV1 rs(&opCtx, &cb, "test.foo", md, &em, false);
309 
310     {
311         // Two extents, each with 1000 bytes.
312         LocAndSize records[] = {
313             {DiskLoc(0, 1000), 500}, {DiskLoc(0, 1500), 300}, {DiskLoc(0, 1800), 100}, {}};
314         LocAndSize drecs[] = {{DiskLoc(0, 1900), 100}, {DiskLoc(1, 1000), 1000}, {}};
315         md->setCapExtent(&opCtx, DiskLoc(0, 0));
316         md->setCapFirstNewRecord(&opCtx, DiskLoc().setInvalid());
317         initializeV1RS(&opCtx, records, drecs, NULL, &em, md);
318     }
319 
320     ASSERT_OK(
321         rs.insertRecord(&opCtx, zeros, 100 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
322             .getStatus());
323 
324     {
325         LocAndSize recs[] = {{DiskLoc(0, 1000), 500},
326                              {DiskLoc(0, 1500), 300},
327                              {DiskLoc(0, 1800), 100},
328 
329                              {DiskLoc(1, 1000), 100},
330                              {}};
331         LocAndSize drecs[] = {{DiskLoc(0, 1900), 100}, {DiskLoc(1, 1100), 900}, {}};
332         assertStateV1RS(&opCtx, recs, drecs, NULL, &em, md);
333         ASSERT_EQUALS(md->capExtent(), DiskLoc(1, 0));
334         ASSERT_EQUALS(md->capFirstNewRecord(), DiskLoc().setInvalid());  // unlooped
335     }
336 }
337 
TEST(CappedRecordStoreV1,MoveToSecondExtentLooped)338 TEST(CappedRecordStoreV1, MoveToSecondExtentLooped) {
339     OperationContextNoop opCtx;
340     DummyExtentManager em;
341     DummyRecordStoreV1MetaData* md = new DummyRecordStoreV1MetaData(true, 0);
342     DummyCappedCallback cb;
343     CappedRecordStoreV1 rs(&opCtx, &cb, "test.foo", md, &em, false);
344 
345     {
346         // Two extents, each with 1000 bytes.
347         LocAndSize records[] = {{DiskLoc(0, 1800), 100},  // old
348                                 {DiskLoc(0, 1000), 500},  // first new
349                                 {DiskLoc(0, 1500), 400},
350 
351                                 {DiskLoc(1, 1000), 300},
352                                 {DiskLoc(1, 1300), 600},
353                                 {}};
354         LocAndSize drecs[] = {{DiskLoc(0, 1900), 100}, {DiskLoc(1, 1900), 100}, {}};
355         md->setCapExtent(&opCtx, DiskLoc(0, 0));
356         md->setCapFirstNewRecord(&opCtx, DiskLoc(0, 1000));
357         initializeV1RS(&opCtx, records, drecs, NULL, &em, md);
358     }
359 
360     ASSERT_OK(
361         rs.insertRecord(&opCtx, zeros, 200 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
362             .getStatus());
363 
364     {
365         LocAndSize recs[] = {{DiskLoc(0, 1000), 500},
366                              {DiskLoc(0, 1500), 400},
367 
368                              {DiskLoc(1, 1300), 600},  // old
369                              {DiskLoc(1, 1000), 200},  // first new
370                              {}};
371         LocAndSize drecs[] = {
372             {DiskLoc(0, 1800), 200}, {DiskLoc(1, 1200), 100}, {DiskLoc(1, 1900), 100}, {}};
373         assertStateV1RS(&opCtx, recs, drecs, NULL, &em, md);
374         ASSERT_EQUALS(md->capExtent(), DiskLoc(1, 0));
375         ASSERT_EQUALS(md->capFirstNewRecord(), DiskLoc(1, 1000));
376     }
377 }
378 
379 // Larger than storageSize (fails early)
TEST(CappedRecordStoreV1,OversizedRecordHuge)380 TEST(CappedRecordStoreV1, OversizedRecordHuge) {
381     OperationContextNoop opCtx;
382     DummyExtentManager em;
383     DummyRecordStoreV1MetaData* md = new DummyRecordStoreV1MetaData(true, 0);
384     DummyCappedCallback cb;
385     CappedRecordStoreV1 rs(&opCtx, &cb, "test.foo", md, &em, false);
386 
387     {
388         LocAndSize records[] = {{}};
389         LocAndSize drecs[] = {{DiskLoc(0, 1000), 1000}, {}};
390         md->setCapExtent(&opCtx, DiskLoc(0, 0));
391         md->setCapFirstNewRecord(&opCtx, DiskLoc().setInvalid());
392         initializeV1RS(&opCtx, records, drecs, NULL, &em, md);
393     }
394 
395     StatusWith<RecordId> status = rs.insertRecord(&opCtx, zeros, 16000, Timestamp(), false);
396     ASSERT_EQUALS(status.getStatus(), ErrorCodes::DocTooLargeForCapped);
397     ASSERT_STRING_CONTAINS(status.getStatus().reason(), "larger than capped size");
398 }
399 
400 // Smaller than storageSize, but larger than usable space (fails late)
TEST(CappedRecordStoreV1,OversizedRecordMedium)401 TEST(CappedRecordStoreV1, OversizedRecordMedium) {
402     OperationContextNoop opCtx;
403     DummyExtentManager em;
404     DummyRecordStoreV1MetaData* md = new DummyRecordStoreV1MetaData(true, 0);
405     DummyCappedCallback cb;
406     CappedRecordStoreV1 rs(&opCtx, &cb, "test.foo", md, &em, false);
407 
408     {
409         LocAndSize records[] = {{}};
410         LocAndSize drecs[] = {{DiskLoc(0, 1000), 1000}, {}};
411         md->setCapExtent(&opCtx, DiskLoc(0, 0));
412         md->setCapFirstNewRecord(&opCtx, DiskLoc().setInvalid());
413         initializeV1RS(&opCtx, records, drecs, NULL, &em, md);
414     }
415 
416     StatusWith<RecordId> status =
417         rs.insertRecord(&opCtx, zeros, 1004 - MmapV1RecordHeader::HeaderSize, Timestamp(), false);
418     ASSERT_EQUALS(status.getStatus(), ErrorCodes::DocTooLargeForCapped);
419     ASSERT_STRING_CONTAINS(status.getStatus().reason(), "doesn't fit");
420 }
421 
422 //
423 // XXX The CappedRecordStoreV1Scrambler suite of tests describe existing behavior that is less
424 // than ideal. Any improved implementation will need to be able to handle a collection that has
425 // been scrambled like this.
426 //
427 
428 /**
429  * This is a minimal example that shows the current allocator laying out records out-of-order.
430  */
TEST(CappedRecordStoreV1Scrambler,Minimal)431 TEST(CappedRecordStoreV1Scrambler, Minimal) {
432     OperationContextNoop opCtx;
433     DummyExtentManager em;
434     DummyRecordStoreV1MetaData* md = new DummyRecordStoreV1MetaData(true, 0);
435     DummyCappedCallback cb;
436     CappedRecordStoreV1 rs(&opCtx, &cb, "test.foo", md, &em, false);
437 
438     {
439         // Starting with a single empty 1000 byte extent.
440         LocAndSize records[] = {{}};
441         LocAndSize drecs[] = {{DiskLoc(0, 1000), 1000}, {}};
442         md->setCapExtent(&opCtx, DiskLoc(0, 0));
443         md->setCapFirstNewRecord(&opCtx, DiskLoc().setInvalid());  // unlooped
444         initializeV1RS(&opCtx, records, drecs, NULL, &em, md);
445     }
446 
447     ASSERT_OK(
448         rs.insertRecord(&opCtx, zeros, 500 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
449             .getStatus());
450     ASSERT_OK(
451         rs.insertRecord(&opCtx, zeros, 300 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
452             .getStatus());
453     ASSERT_OK(
454         rs.insertRecord(&opCtx, zeros, 400 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
455             .getStatus());  // won't fit at end so wraps
456     ASSERT_OK(
457         rs.insertRecord(&opCtx, zeros, 120 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
458             .getStatus());  // fits at end
459     ASSERT_OK(
460         rs.insertRecord(&opCtx, zeros, 60 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
461             .getStatus());  // fits in earlier hole
462 
463     {
464         LocAndSize recs[] = {{DiskLoc(0, 1500), 300},  // 2nd insert
465                              {DiskLoc(0, 1000), 400},  // 3rd (1st new)
466                              {DiskLoc(0, 1800), 120},  // 4th
467                              {DiskLoc(0, 1400), 60},   // 5th
468                              {}};
469         LocAndSize drecs[] = {{DiskLoc(0, 1460), 40}, {DiskLoc(0, 1920), 80}, {}};
470         assertStateV1RS(&opCtx, recs, drecs, NULL, &em, md);
471         ASSERT_EQUALS(md->capExtent(), DiskLoc(0, 0));
472         ASSERT_EQUALS(md->capFirstNewRecord(), DiskLoc(0, 1000));
473     }
474 }
475 
476 /**
477  * This tests a specially crafted set of inserts that scrambles a capped collection in a way
478  * that leaves 4 deleted records in a single extent.
479  */
TEST(CappedRecordStoreV1Scrambler,FourDeletedRecordsInSingleExtent)480 TEST(CappedRecordStoreV1Scrambler, FourDeletedRecordsInSingleExtent) {
481     OperationContextNoop opCtx;
482     DummyExtentManager em;
483     DummyRecordStoreV1MetaData* md = new DummyRecordStoreV1MetaData(true, 0);
484     DummyCappedCallback cb;
485     CappedRecordStoreV1 rs(&opCtx, &cb, "test.foo", md, &em, false);
486 
487     {
488         // Starting with a single empty 1000 byte extent.
489         LocAndSize records[] = {{}};
490         LocAndSize drecs[] = {{DiskLoc(0, 1000), 1000}, {}};
491         md->setCapExtent(&opCtx, DiskLoc(0, 0));
492         md->setCapFirstNewRecord(&opCtx, DiskLoc().setInvalid());  // unlooped
493         initializeV1RS(&opCtx, records, drecs, NULL, &em, md);
494     }
495 
496     // This list of sizes was empirically generated to achieve this outcome. Don't think too
497     // much about them.
498     ASSERT_OK(
499         rs.insertRecord(&opCtx, zeros, 500 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
500             .getStatus());
501     ASSERT_OK(
502         rs.insertRecord(&opCtx, zeros, 300 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
503             .getStatus());
504     ASSERT_OK(
505         rs.insertRecord(&opCtx, zeros, 304 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
506             .getStatus());
507     ASSERT_OK(
508         rs.insertRecord(&opCtx, zeros, 76 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
509             .getStatus());
510     ASSERT_OK(
511         rs.insertRecord(&opCtx, zeros, 100 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
512             .getStatus());
513     ASSERT_OK(
514         rs.insertRecord(&opCtx, zeros, 96 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
515             .getStatus());
516     ASSERT_OK(
517         rs.insertRecord(&opCtx, zeros, 76 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
518             .getStatus());
519     ASSERT_OK(
520         rs.insertRecord(&opCtx, zeros, 200 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
521             .getStatus());
522     ASSERT_OK(
523         rs.insertRecord(&opCtx, zeros, 100 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
524             .getStatus());
525     ASSERT_OK(
526         rs.insertRecord(&opCtx, zeros, 100 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
527             .getStatus());
528     ASSERT_OK(
529         rs.insertRecord(&opCtx, zeros, 200 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
530             .getStatus());
531     ASSERT_OK(
532         rs.insertRecord(&opCtx, zeros, 56 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
533             .getStatus());
534     ASSERT_OK(
535         rs.insertRecord(&opCtx, zeros, 100 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
536             .getStatus());
537     ASSERT_OK(
538         rs.insertRecord(&opCtx, zeros, 96 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
539             .getStatus());
540     ASSERT_OK(
541         rs.insertRecord(&opCtx, zeros, 104 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
542             .getStatus());
543     ASSERT_OK(
544         rs.insertRecord(&opCtx, zeros, 96 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
545             .getStatus());
546     ASSERT_OK(
547         rs.insertRecord(&opCtx, zeros, 60 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
548             .getStatus());
549     ASSERT_OK(
550         rs.insertRecord(&opCtx, zeros, 60 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
551             .getStatus());
552     ASSERT_OK(
553         rs.insertRecord(&opCtx, zeros, 146 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
554             .getStatus());
555     ASSERT_OK(
556         rs.insertRecord(&opCtx, zeros, 146 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
557             .getStatus());
558     ASSERT_OK(
559         rs.insertRecord(&opCtx, zeros, 40 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
560             .getStatus());
561     ASSERT_OK(
562         rs.insertRecord(&opCtx, zeros, 40 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
563             .getStatus());
564     ASSERT_OK(
565         rs.insertRecord(&opCtx, zeros, 36 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
566             .getStatus());
567     ASSERT_OK(
568         rs.insertRecord(&opCtx, zeros, 100 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
569             .getStatus());
570     ASSERT_OK(
571         rs.insertRecord(&opCtx, zeros, 96 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
572             .getStatus());
573     ASSERT_OK(
574         rs.insertRecord(&opCtx, zeros, 200 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
575             .getStatus());
576     ASSERT_OK(
577         rs.insertRecord(&opCtx, zeros, 60 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
578             .getStatus());
579     ASSERT_OK(
580         rs.insertRecord(&opCtx, zeros, 64 - MmapV1RecordHeader::HeaderSize, Timestamp(), false)
581             .getStatus());
582 
583     {
584         LocAndSize recs[] = {{DiskLoc(0, 1148), 148},
585                              {DiskLoc(0, 1936), 40},
586                              {DiskLoc(0, 1712), 40},
587                              {DiskLoc(0, 1296), 36},
588                              {DiskLoc(0, 1752), 100},
589                              {DiskLoc(0, 1332), 96},
590                              {DiskLoc(0, 1428), 200},
591                              {DiskLoc(0, 1852), 60},
592                              {DiskLoc(0, 1000), 64},  // (1st new)
593                              {}};
594         LocAndSize drecs[] = {{DiskLoc(0, 1064), 84},
595                               {DiskLoc(0, 1976), 24},
596                               {DiskLoc(0, 1912), 24},
597                               {DiskLoc(0, 1628), 84},
598                               {}};
599         assertStateV1RS(&opCtx, recs, drecs, NULL, &em, md);
600         ASSERT_EQUALS(md->capExtent(), DiskLoc(0, 0));
601         ASSERT_EQUALS(md->capFirstNewRecord(), DiskLoc(0, 1000));
602     }
603 }
604 
605 //
606 // The CappedRecordStoreV1QueryStage tests some nitty-gritty capped
607 //  collection details.  Ported and polished from pdfiletests.cpp.
608 //
609 
610 class CollscanHelper {
611 public:
CollscanHelper(int nExtents)612     CollscanHelper(int nExtents)
613         : md(new DummyRecordStoreV1MetaData(true, 0)), rs(&opCtx, &cb, ns(), md, &em, false) {
614         LocAndSize recs[] = {{}};
615         LocAndSize drecs[8];
616         ASSERT_LESS_THAN(nExtents, 8);
617         for (int j = 0; j < nExtents; ++j) {
618             drecs[j].loc = DiskLoc(j, 1000);
619             drecs[j].size = 1000;
620         }
621         drecs[nExtents].loc = DiskLoc();
622         drecs[nExtents].size = 0;
623 
624         md->setCapExtent(&opCtx, DiskLoc(0, 0));
625         md->setCapFirstNewRecord(&opCtx, DiskLoc().setInvalid());  // unlooped
626         initializeV1RS(&opCtx, recs, drecs, NULL, &em, md);
627     }
628 
629     // Insert bypasses standard alloc/insert routines to use the extent we want.
630     // TODO: Directly declare resulting record store state instead of procedurally creating it
insert(const DiskLoc & ext,int i)631     DiskLoc insert(const DiskLoc& ext, int i) {
632         // Copied verbatim.
633         BSONObjBuilder b;
634         b.append("a", i);
635         BSONObj o = b.done();
636         int len = o.objsize();
637         Extent* e = em.getExtent(ext);
638         e = opCtx.recoveryUnit()->writing(e);
639         int ofs;
640         if (e->lastRecord.isNull()) {
641             ofs = ext.getOfs() + (e->_extentData - (char*)e);
642         } else {
643             ofs = e->lastRecord.getOfs() + em.recordForV1(e->lastRecord)->lengthWithHeaders();
644         }
645         DiskLoc dl(ext.a(), ofs);
646         MmapV1RecordHeader* r = em.recordForV1(dl);
647         r = (MmapV1RecordHeader*)opCtx.recoveryUnit()->writingPtr(
648             r, MmapV1RecordHeader::HeaderSize + len);
649         r->lengthWithHeaders() = MmapV1RecordHeader::HeaderSize + len;
650         r->extentOfs() = e->myLoc.getOfs();
651         r->nextOfs() = DiskLoc::NullOfs;
652         r->prevOfs() = e->lastRecord.isNull() ? DiskLoc::NullOfs : e->lastRecord.getOfs();
653         memcpy(r->data(), o.objdata(), len);
654         if (e->firstRecord.isNull())
655             e->firstRecord = dl;
656         else
657             opCtx.recoveryUnit()->writingInt(em.recordForV1(e->lastRecord)->nextOfs()) = ofs;
658         e->lastRecord = dl;
659         return dl;
660     }
661 
662     // TODO: Directly assert the desired record store state instead of just walking it
walkAndCount(int expectedCount)663     void walkAndCount(int expectedCount) {
664         // Walk the collection going forward.
665         {
666             CappedRecordStoreV1Iterator cursor(&opCtx, &rs, /*forward=*/true);
667             int resultCount = 0;
668             while (auto record = cursor.next()) {
669                 ++resultCount;
670             }
671 
672             ASSERT_EQUALS(resultCount, expectedCount);
673         }
674 
675         // Walk the collection going backwards.
676         {
677             CappedRecordStoreV1Iterator cursor(&opCtx, &rs, /*forward=*/false);
678             int resultCount = expectedCount;
679             while (auto record = cursor.next()) {
680                 --resultCount;
681             }
682 
683             ASSERT_EQUALS(resultCount, 0);
684         }
685     }
686 
ns()687     static const char* ns() {
688         return "unittests.QueryStageCollectionScanCapped";
689     }
690 
691     OperationContextNoop opCtx;
692     DummyRecordStoreV1MetaData* md;
693     DummyExtentManager em;
694 
695 private:
696     DummyCappedCallback cb;
697     CappedRecordStoreV1 rs;
698 };
699 
700 
TEST(CappedRecordStoreV1QueryStage,CollscanCappedBase)701 TEST(CappedRecordStoreV1QueryStage, CollscanCappedBase) {
702     CollscanHelper h(1);
703     h.walkAndCount(0);
704 }
705 
TEST(CappedRecordStoreV1QueryStage,CollscanEmptyLooped)706 TEST(CappedRecordStoreV1QueryStage, CollscanEmptyLooped) {
707     CollscanHelper h(1);
708     h.md->setCapFirstNewRecord(&h.opCtx, DiskLoc());
709     h.walkAndCount(0);
710 }
711 
TEST(CappedRecordStoreV1QueryStage,CollscanEmptyMultiExtentLooped)712 TEST(CappedRecordStoreV1QueryStage, CollscanEmptyMultiExtentLooped) {
713     CollscanHelper h(3);
714     h.md->setCapFirstNewRecord(&h.opCtx, DiskLoc());
715     h.walkAndCount(0);
716 }
717 
TEST(CappedRecordStoreV1QueryStage,CollscanSingle)718 TEST(CappedRecordStoreV1QueryStage, CollscanSingle) {
719     CollscanHelper h(1);
720 
721     h.md->setCapFirstNewRecord(&h.opCtx, h.insert(h.md->capExtent(), 0));
722     h.walkAndCount(1);
723 }
724 
TEST(CappedRecordStoreV1QueryStage,CollscanNewCapFirst)725 TEST(CappedRecordStoreV1QueryStage, CollscanNewCapFirst) {
726     CollscanHelper h(1);
727     DiskLoc x = h.insert(h.md->capExtent(), 0);
728     h.md->setCapFirstNewRecord(&h.opCtx, x);
729     h.insert(h.md->capExtent(), 1);
730     h.walkAndCount(2);
731 }
732 
TEST(CappedRecordStoreV1QueryStage,CollscanNewCapMiddle)733 TEST(CappedRecordStoreV1QueryStage, CollscanNewCapMiddle) {
734     CollscanHelper h(1);
735     h.insert(h.md->capExtent(), 0);
736     h.md->setCapFirstNewRecord(&h.opCtx, h.insert(h.md->capExtent(), 1));
737     h.insert(h.md->capExtent(), 2);
738     h.walkAndCount(3);
739 }
740 
TEST(CappedRecordStoreV1QueryStage,CollscanFirstExtent)741 TEST(CappedRecordStoreV1QueryStage, CollscanFirstExtent) {
742     CollscanHelper h(2);
743     h.insert(h.md->capExtent(), 0);
744     h.insert(h.md->lastExtent(&h.opCtx), 1);
745     h.md->setCapFirstNewRecord(&h.opCtx, h.insert(h.md->capExtent(), 2));
746     h.insert(h.md->capExtent(), 3);
747     h.walkAndCount(4);
748 }
749 
TEST(CappedRecordStoreV1QueryStage,CollscanLastExtent)750 TEST(CappedRecordStoreV1QueryStage, CollscanLastExtent) {
751     CollscanHelper h(2);
752     h.md->setCapExtent(&h.opCtx, h.md->lastExtent(&h.opCtx));
753     h.insert(h.md->capExtent(), 0);
754     h.insert(h.md->firstExtent(&h.opCtx), 1);
755     h.md->setCapFirstNewRecord(&h.opCtx, h.insert(h.md->capExtent(), 2));
756     h.insert(h.md->capExtent(), 3);
757     h.walkAndCount(4);
758 }
759 
TEST(CappedRecordStoreV1QueryStage,CollscanMidExtent)760 TEST(CappedRecordStoreV1QueryStage, CollscanMidExtent) {
761     CollscanHelper h(3);
762     h.md->setCapExtent(&h.opCtx, h.em.getExtent(h.md->firstExtent(&h.opCtx))->xnext);
763     h.insert(h.md->capExtent(), 0);
764     h.insert(h.md->lastExtent(&h.opCtx), 1);
765     h.insert(h.md->firstExtent(&h.opCtx), 2);
766     h.md->setCapFirstNewRecord(&h.opCtx, h.insert(h.md->capExtent(), 3));
767     h.insert(h.md->capExtent(), 4);
768     h.walkAndCount(5);
769 }
770 
TEST(CappedRecordStoreV1QueryStage,CollscanAloneInExtent)771 TEST(CappedRecordStoreV1QueryStage, CollscanAloneInExtent) {
772     CollscanHelper h(3);
773     h.md->setCapExtent(&h.opCtx, h.em.getExtent(h.md->firstExtent(&h.opCtx))->xnext);
774     h.insert(h.md->lastExtent(&h.opCtx), 0);
775     h.insert(h.md->firstExtent(&h.opCtx), 1);
776     h.md->setCapFirstNewRecord(&h.opCtx, h.insert(h.md->capExtent(), 2));
777     h.walkAndCount(3);
778 }
779 
TEST(CappedRecordStoreV1QueryStage,CollscanFirstInExtent)780 TEST(CappedRecordStoreV1QueryStage, CollscanFirstInExtent) {
781     CollscanHelper h(3);
782     h.md->setCapExtent(&h.opCtx, h.em.getExtent(h.md->firstExtent(&h.opCtx))->xnext);
783     h.insert(h.md->lastExtent(&h.opCtx), 0);
784     h.insert(h.md->firstExtent(&h.opCtx), 1);
785     h.md->setCapFirstNewRecord(&h.opCtx, h.insert(h.md->capExtent(), 2));
786     h.insert(h.md->capExtent(), 3);
787     h.walkAndCount(4);
788 }
789 
TEST(CappedRecordStoreV1QueryStage,CollscanLastInExtent)790 TEST(CappedRecordStoreV1QueryStage, CollscanLastInExtent) {
791     CollscanHelper h(3);
792     h.md->setCapExtent(&h.opCtx, h.em.getExtent(h.md->firstExtent(&h.opCtx))->xnext);
793     h.insert(h.md->capExtent(), 0);
794     h.insert(h.md->lastExtent(&h.opCtx), 1);
795     h.insert(h.md->firstExtent(&h.opCtx), 2);
796     h.md->setCapFirstNewRecord(&h.opCtx, h.insert(h.md->capExtent(), 3));
797     h.walkAndCount(4);
798 }
799 }
800