SERVER-66283 Correct tailable cursor support on unreplicated capped collections

This corrects previously incorrect behavior where tailable cursors on unreplicated capped
collections like the profile collection could miss writes

This pushes non-oplog capped collection visibility logic outside of the storage engine layer

1 files changed, 148 insertions, 0 deletions

diff --git a/jstests/core/local_tail_capped.js b/jstests/core/local_tail_capped.js
new file mode 100644
index 00000000000..613eb58e669
--- /dev/null
+++ b/jstests/core/local_tail_capped.js
@@ -0,0 +1,148 @@
+/**
+ * This test tests concurrent read and write behavior for tailable cursors on unreplicated capped
+ * collections. These collections accept concurrent writes and thus must ensure no documents are
+ * skipped for forward cursors.
+ *
+ * This test sets up a single capped collection with many concurrent writers. Concurrent readers
+ * open tailable cursors and clone the contents into their own collection copies. The readers then
+ * assert that the contents match the source.
+ *
+ * @tags: [
+ *   assumes_against_mongod_not_mongos,
+ *   does_not_support_retryable_writes,
+ *   requires_capped,
+ *   requires_non_retryable_writes,
+ *   # Tailable cursors do not work correctly on previous versions.
+ *   requires_fcv_63,
+ * ]
+ */
+
+load("jstests/libs/parallelTester.js");  // For Thread
+
+(function() {
+'use strict';
+
+function insertWorker(host, collName, tid, nInserts) {
+    const conn = new Mongo(host);
+    const db = conn.getDB('local');
+
+    for (let i = 0; i < nInserts;) {
+        const bulk = db[collName].initializeUnorderedBulkOp();
+        for (let j = 0; j < 10; j++) {
+            bulk.insert({t: tid, i: i++});
+        }
+        assert.commandWorked(bulk.execute());
+    }
+    print(tid + ": done");
+}
+
+function tailWorker(host, collName, tid, expectedDocs) {
+    // Rewrite the connection string as a mongo URI so that we can add an 'appName' to make
+    // debugging easier. When run against a standalone, 'host' is in the form '<host>:<port>'. When
+    // run against a replica set, 'host' is in the form '<rs name>/<host1>:<port1>,...'
+    const iSlash = host.indexOf('/');
+    let connString = 'mongodb://';
+    if (iSlash > 0) {
+        connString += host.substr(iSlash + 1) + '/?appName=tid' + tid +
+            '&replicaSet=' + host.substr(0, iSlash);
+    } else {
+        connString += host + '/?appName=tid' + tid;
+    }
+    const conn = new Mongo(connString);
+    const db = conn.getDB('local');
+    const cloneColl = db[collName + "_clone_" + tid];
+    cloneColl.drop();
+
+    let res = db.runCommand({find: collName, batchSize: 0, awaitData: true, tailable: true});
+    assert.commandWorked(res);
+    assert.gt(res.cursor.id, NumberLong(0));
+    assert.eq(res.cursor.firstBatch.length, 0);
+
+    const curId = res.cursor.id;
+
+    let myCount = 0;
+    let emptyGetMores = 0;
+    let nonEmptyGetMores = 0;
+    assert.soon(() => {
+        res = db.runCommand({getMore: curId, collection: collName, maxTimeMS: 1000});
+        assert.commandWorked(res);
+
+        const batchLen = res.cursor.nextBatch.length;
+        if (batchLen > 0) {
+            nonEmptyGetMores++;
+        } else {
+            emptyGetMores++;
+        }
+
+        print(tid + ': got batch of size ' + batchLen +
+              '. first doc: ' + tojson(res.cursor.nextBatch[0]) +
+              '. last doc: ' + tojson(res.cursor.nextBatch[batchLen - 1]) +
+              '. empty getMores so far: ' + emptyGetMores +
+              '. non-empty getMores so far: ' + nonEmptyGetMores);
+        myCount += batchLen;
+
+        const bulk = cloneColl.initializeUnorderedBulkOp();
+        for (let i = 0; i < batchLen; i++) {
+            bulk.insert(res.cursor.nextBatch[i]);
+        }
+        assert.commandWorked(bulk.execute());
+
+        // The writers are done, so we are draining until we see as many docs as we
+        // expect.
+        if (myCount == expectedDocs) {
+            return true;
+        } else {
+            print(tid + ": waiting. my count: " + myCount + " expected: " + expectedDocs);
+        }
+        return false;
+    }, "failed to return all documents within timeout");
+
+    print(tid + ": validating");
+    const expected = db[collName].find().sort({_id: 1}).toArray();
+    const actual = cloneColl.find().sort({_id: 1}).toArray();
+    assert.eq(expected.length, actual.length, function() {
+        return "number of documents do not match. expected: " + tojson(expected) +
+            " actual: " + tojson(actual);
+    });
+    for (let i = 0; i < actual.length; i++) {
+        assert.docEq(actual[i], expected[i], function() {
+            return "mismatched documents. expected: " + tojson(expected) +
+                " actual: " + tojson(actual);
+        });
+    }
+    print(tid + ": done");
+}
+
+const collName = 'capped';
+const localDb = db.getSiblingDB('local');
+localDb[collName].drop();
+
+assert.commandWorked(localDb.runCommand({create: collName, capped: true, size: 10 * 1024 * 1024}));
+assert.commandWorked(localDb[collName].insert({firstDoc: 1, i: -1}));
+
+const nWriters = 5;
+const nReaders = 5;
+
+const insertsPerThread = 1000;
+const expectedDocs = nWriters * insertsPerThread + 1;
+
+let threads = [];
+
+for (let i = 0; i < nReaders; i++) {
+    const thread =
+        new Thread(tailWorker, db.getMongo().host, collName, threads.length, expectedDocs);
+    thread.start();
+    threads.push(thread);
+}
+
+for (let i = 0; i < nWriters; i++) {
+    const thread =
+        new Thread(insertWorker, db.getMongo().host, collName, threads.length, insertsPerThread);
+    thread.start();
+    threads.push(thread);
+}
+
+for (let i = 0; i < threads.length; i++) {
+    threads[i].join();
+}
+})();