SERVER-27996 Invalidate stale system.namespaces record IDs

(cherry picked from commit c8afd7bba3aec74e49ce3467cce386f1d3f9b92f)

2 files changed, 65 insertions, 1 deletions

diff --git a/jstests/core/list_namespaces_invalidation.js b/jstests/core/list_namespaces_invalidation.js
new file mode 100644
index 00000000000..81f6c7d69f1
--- /dev/null
+++ b/jstests/core/list_namespaces_invalidation.js
@@ -0,0 +1,45 @@
+// SERVER-27996 Missing invalidation for system.namespaces writes
+(function() {
+    'use strict';
+    let dbInvalidName = 'system_namespaces_invalidations';
+    let dbInvalid = db.getSiblingDB(dbInvalidName);
+    let num_collections = 3;
+    function testNamespaceInvalidation(isRename) {
+        dbInvalid.dropDatabase();
+
+        // Create enough collections to necessitate multiple cursor batches.
+        for (let i = 0; i < num_collections; i++) {
+            assert.commandWorked(dbInvalid.createCollection('coll' + i.toString()));
+        }
+
+        // Get the first two namespaces. Use find on 'system.namespaces' on MMAPv1, listCollections
+        // otherwise.
+        let cmd = dbInvalid.system.indexes.count() ? {find: 'system.namespaces'}
+                                                   : {listCollections: dbInvalidName};
+        Object.extend(cmd, {batchSize: 2});
+        let res = dbInvalid.runCommand(cmd);
+        assert.commandWorked(res, 'could not run ' + tojson(cmd));
+        printjson(res);
+
+        // Ensure the cursor has data, drop or rename the collections, and exhaust the cursor.
+        let cursor = new DBCommandCursor(dbInvalid.getMongo(), res);
+        let errMsg =
+            'expected more data from command ' + tojson(cmd) + ', with result ' + tojson(res);
+        assert(cursor.hasNext(), errMsg);
+        for (let j = 0; j < num_collections; j++) {
+            if (isRename) {
+                // Rename the collection to something that does not fit in the previously allocated
+                // memory for the record.
+                assert.commandWorked(dbInvalid['coll' + j.toString()].renameCollection(
+                    'coll' + j.toString() + 'lkdsahflaksjdhfsdkljhfskladhfkahfsakfla' +
+                    'skfjhaslfaslfkhasklfjhsakljhdsjksahkldjslh'));
+            } else {
+                assert(dbInvalid['coll' + j.toString()].drop());
+            }
+        }
+        assert.gt(cursor.itcount(), 0, errMsg);
+    }
+    // Test that we invalidate namespaces for both collection drops and renames.
+    testNamespaceInvalidation(false);
+    testNamespaceInvalidation(true);
+}());
diff --git a/src/mongo/db/storage/mmap_v1/mmap_v1_database_catalog_entry.cpp b/src/mongo/db/storage/mmap_v1/mmap_v1_database_catalog_entry.cpp
index 60c3055f3cc..f6e0e6c185c 100644
--- a/src/mongo/db/storage/mmap_v1/mmap_v1_database_catalog_entry.cpp
+++ b/src/mongo/db/storage/mmap_v1/mmap_v1_database_catalog_entry.cpp
@@ -383,6 +383,15 @@ Status MMAPV1DatabaseCatalogEntry::_renameSingleNamespace(OperationContext* txn,
 
     RecordId rid = _addNamespaceToNamespaceCollection(txn, toNS, newSpec.isEmpty() ? 0 : &newSpec);
 
+    // Invalidate old namespace record
+    const NamespaceString nsn(name(), "system.namespaces");
+    StringData dbName(name());
+    invariant(txn->lockState()->isDbLockedForMode(dbName, MODE_X));
+    Database* db = dbHolder().get(txn, dbName);
+    Collection* systemNamespaces = db->getCollection(nsn);
+    systemNamespaces->getCursorManager()->invalidateDocument(
+        txn, oldSpecLocation, INVALIDATION_DELETION);
+
     _getNamespaceRecordStore()->deleteRecord(txn, oldSpecLocation);
 
     Entry*& entry = _collections[toNS.toString()];
@@ -811,7 +820,17 @@ void MMAPV1DatabaseCatalogEntry::_removeNamespaceFromNamespaceCollection(Operati
     RecordStoreV1Base* rs = _getNamespaceRecordStore();
     invariant(rs);
 
-    rs->deleteRecord(txn, entry->second->catalogEntry->getNamespacesRecordId());
+    // Invalidate old namespace record
+    RecordId oldSpecLocation = entry->second->catalogEntry->getNamespacesRecordId();
+    const NamespaceString nsn(name(), "system.namespaces");
+    StringData dbName(name());
+    invariant(txn->lockState()->isDbLockedForMode(dbName, MODE_X));
+    Database* db = dbHolder().get(txn, dbName);
+    Collection* systemNamespaces = db->getCollection(nsn);
+    systemNamespaces->getCursorManager()->invalidateDocument(
+        txn, oldSpecLocation, INVALIDATION_DELETION);
+
+    rs->deleteRecord(txn, oldSpecLocation);
 }
 
 CollectionOptions MMAPV1DatabaseCatalogEntry::getCollectionOptions(OperationContext* txn,