SERVER-36768 Use namespace orphan for the recovered orphaned collections

(cherry picked from commit 238af0b7739e6c4702e234cfb28e3a714f119523)

5 files changed, 27 insertions, 11 deletions

diff --git a/src/mongo/db/namespace_string.h b/src/mongo/db/namespace_string.h
index 7174e00b3bd..398a24716ba 100644
--- a/src/mongo/db/namespace_string.h
+++ b/src/mongo/db/namespace_string.h
@@ -61,7 +61,7 @@ public:
     static constexpr StringData kSystemDotViewsCollectionName = "system.views"_sd;
 
     // Prefix for orphan collections
-    static constexpr StringData kOrphanCollectionPrefix = "system.orphan-"_sd;
+    static constexpr StringData kOrphanCollectionPrefix = "orphan."_sd;
     static constexpr StringData kOrphanCollectionDb = "local"_sd;
 
     // Namespace for storing configuration data, which needs to be replicated if the server is
diff --git a/src/mongo/db/storage/kv/kv_catalog.cpp b/src/mongo/db/storage/kv/kv_catalog.cpp
index 7223189062c..6dfc952e505 100644
--- a/src/mongo/db/storage/kv/kv_catalog.cpp
+++ b/src/mongo/db/storage/kv/kv_catalog.cpp
@@ -633,14 +633,19 @@ bool KVCatalog::isCollectionIdent(StringData ident) const {
 }
 
 StatusWith<std::string> KVCatalog::newOrphanedIdent(OperationContext* opCtx, std::string ident) {
-    // The collection will be named local.system.orphan-xxxxx.
+    // The collection will be named local.orphan.xxxxx.
+    std::string identNs = ident;
+    std::replace(identNs.begin(), identNs.end(), '-', '_');
     std::string ns = NamespaceString(NamespaceString::kOrphanCollectionDb,
-                                     NamespaceString::kOrphanCollectionPrefix + ident)
+                                     NamespaceString::kOrphanCollectionPrefix + identNs)
                          .ns();
 
     stdx::lock_guard<stdx::mutex> lk(_identsLock);
     Entry& old = _idents[ns];
-    invariant(old.ident.empty());
+    if (!old.ident.empty()) {
+        return Status(ErrorCodes::NamespaceExists,
+                      str::stream() << ns << " already exists in the catalog");
+    }
     opCtx->recoveryUnit()->registerChange(new AddIdentChange(this, ns));
 
     // Generate a new UUID for the orphaned collection.
diff --git a/src/mongo/db/storage/kv/kv_catalog.h b/src/mongo/db/storage/kv/kv_catalog.h
index d8a1a6e91d3..a9827a1fce8 100644
--- a/src/mongo/db/storage/kv/kv_catalog.h
+++ b/src/mongo/db/storage/kv/kv_catalog.h
@@ -106,6 +106,8 @@ public:
     /**
      * Create an entry in the catalog for an orphaned collection found in the
      * storage engine. Return the generated ns of the collection.
+     * Note that this function does not recreate the _id index on the collection because it does not
+     * have access to index catalog.
      */
     StatusWith<std::string> newOrphanedIdent(OperationContext* opCtx, std::string ident);
 
diff --git a/src/mongo/db/storage/kv/kv_storage_engine.cpp b/src/mongo/db/storage/kv/kv_storage_engine.cpp
index 6222ce811e0..c49a0dfe776 100644
--- a/src/mongo/db/storage/kv/kv_storage_engine.cpp
+++ b/src/mongo/db/storage/kv/kv_storage_engine.cpp
@@ -155,7 +155,7 @@ void KVStorageEngine::loadCatalog(OperationContext* opCtx) {
     if (_options.forRepair) {
         // It's possible that there are collection files on disk that are unknown to the catalog. In
         // a repair context, if we can't find an ident in the catalog, we generate a catalog entry
-        // 'local.system.orphan-xxxxx' for it. However, in a nonrepair context, the orphaned idents
+        // 'local.orphan.xxxxx' for it. However, in a nonrepair context, the orphaned idents
         // will be dropped in reconcileCatalogAndIdents().
         for (const auto& ident : identsKnownToStorageEngine) {
             if (_catalog->isCollectionIdent(ident)) {
@@ -171,9 +171,13 @@ void KVStorageEngine::loadCatalog(OperationContext* opCtx) {
                     StatusWith<std::string> statusWithNs = _catalog->newOrphanedIdent(opCtx, ident);
                     if (statusWithNs.isOK()) {
                         wuow.commit();
+                        auto orphanCollNs = statusWithNs.getValue();
                         log() << "Successfully created an entry in the catalog for the orphaned "
                                  "collection: "
-                              << statusWithNs.getValue();
+                              << orphanCollNs;
+                        warning() << orphanCollNs
+                                  << " does not have the _id index. Please manually "
+                                     "build the index.";
 
                         StorageRepairObserver::get(getGlobalServiceContext())
                             ->onModification(str::stream() << "Orphan collection created: "
@@ -182,9 +186,10 @@ void KVStorageEngine::loadCatalog(OperationContext* opCtx) {
                     } else {
                         // Log an error message if we cannot create the entry.
                         // reconcileCatalogAndIdents() will later drop this ident.
-                        error()
-                            << "Cannot create an entry in the catalog for the orphaned collection: "
-                            << ident << " due to " << statusWithNs.getStatus().reason();
+                        error() << "Cannot create an entry in the catalog for the orphaned "
+                                   "collection ident: "
+                                << ident << " due to " << statusWithNs.getStatus().reason();
+                        error() << "Restarting the server will remove this ident.";
                     }
                 }
             }
diff --git a/src/mongo/db/storage/kv/kv_storage_engine_test.cpp b/src/mongo/db/storage/kv/kv_storage_engine_test.cpp
index 1355ec8e3f2..456b8a9ccc6 100644
--- a/src/mongo/db/storage/kv/kv_storage_engine_test.cpp
+++ b/src/mongo/db/storage/kv/kv_storage_engine_test.cpp
@@ -336,7 +336,9 @@ TEST_F(KVStorageEngineRepairTest, LoadCatalogRecoversOrphansInCatalog) {
 
     // When in a repair context, loadCatalog() recreates catalog entries for orphaned idents.
     _storageEngine->loadCatalog(opCtx.get());
-    NamespaceString orphanNs = NamespaceString("local.system.orphan-" + swIdentName.getValue());
+    auto identNs = swIdentName.getValue();
+    std::replace(identNs.begin(), identNs.end(), '-', '_');
+    NamespaceString orphanNs = NamespaceString("local.orphan." + identNs);
 
     ASSERT(identExists(opCtx.get(), swIdentName.getValue()));
     ASSERT(collectionExists(opCtx.get(), orphanNs));
@@ -367,7 +369,9 @@ TEST_F(KVStorageEngineTest, LoadCatalogDropsOrphans) {
     ASSERT_OK(reconcile(opCtx.get()).getStatus());
 
     ASSERT(!identExists(opCtx.get(), swIdentName.getValue()));
-    NamespaceString orphanNs = NamespaceString("local.system.orphan-" + swIdentName.getValue());
+    auto identNs = swIdentName.getValue();
+    std::replace(identNs.begin(), identNs.end(), '-', '_');
+    NamespaceString orphanNs = NamespaceString("local.orphan." + identNs);
     ASSERT(!collectionExists(opCtx.get(), orphanNs));
 }
 }  // namespace