SERVER-24767 Replicate views

24 files changed, 312 insertions, 84 deletions

diff --git a/src/mongo/base/error_codes.err b/src/mongo/base/error_codes.err
index 0871fe48a1f..7c8f58de4fd 100644
--- a/src/mongo/base/error_codes.err
+++ b/src/mongo/base/error_codes.err
@@ -178,6 +178,7 @@ error_code("RangeOverlapConflict", 176)
 error_code("WindowsPdhError", 177)
 error_code("BadPerfCounterPath", 178)
 error_code("AmbiguousIndexKeyPattern", 179)
+error_code("InvalidViewDefinition", 180);
 
 # Non-sequential error codes (for compatibility only)
 error_code("SocketException", 9001)
diff --git a/src/mongo/db/catalog/coll_mod.cpp b/src/mongo/db/catalog/coll_mod.cpp
index 8d142b25a12..81ca126e64d 100644
--- a/src/mongo/db/catalog/coll_mod.cpp
+++ b/src/mongo/db/catalog/coll_mod.cpp
@@ -56,7 +56,7 @@ Status collMod(OperationContext* txn,
     Collection* coll = db ? db->getCollection(nss) : nullptr;
 
     // May also modify a view instead of a collection.
-    const ViewDefinition* view = db ? db->getViewCatalog()->lookup(nss.ns()) : nullptr;
+    const ViewDefinition* view = db ? db->getViewCatalog()->lookup(txn, nss.ns()) : nullptr;
     boost::optional<ViewDefinition> newView;
     if (view)
         newView = {*view};
@@ -259,25 +259,28 @@ Status collMod(OperationContext* txn,
         }
     }
 
-    // Actually update the view if it was parsed successfully.
-    if (view && errorStatus.isOK()) {
+    if (!errorStatus.isOK()) {
+        return errorStatus;
+    }
+
+    // Actually update the view if it was parsed successfully. Only observe non-view collMods,
+    // as view operations are observed as operations on the system.views collection.
+    if (view) {
         ViewCatalog* catalog = db->getViewCatalog();
-        catalog->dropView(txn, nss);
 
         BSONArrayBuilder pipeline;
         for (auto& item : newView->pipeline()) {
             pipeline.append(item);
         }
-        errorStatus = catalog->createView(txn, nss, newView->viewOn(), pipeline.obj());
-    }
-
-    if (!errorStatus.isOK()) {
-        return errorStatus;
+        errorStatus = catalog->modifyView(txn, nss, newView->viewOn(), BSONArray(pipeline.obj()));
+        if (!errorStatus.isOK()) {
+            return errorStatus;
+        }
+    } else {
+        getGlobalServiceContext()->getOpObserver()->onCollMod(
+            txn, (dbName.toString() + ".$cmd").c_str(), cmdObj);
     }
 
-    getGlobalServiceContext()->getOpObserver()->onCollMod(
-        txn, (dbName.toString() + ".$cmd").c_str(), cmdObj);
-
     wunit.commit();
     return Status::OK();
 }
diff --git a/src/mongo/db/catalog/database.cpp b/src/mongo/db/catalog/database.cpp
index 75fa8db90bf..733ed4a3725 100644
--- a/src/mongo/db/catalog/database.cpp
+++ b/src/mongo/db/catalog/database.cpp
@@ -203,9 +203,9 @@ Database::Database(OperationContext* txn, StringData name, DatabaseCatalogEntry*
       _dbEntry(dbEntry),
       _profileName(_name + ".system.profile"),
       _indexesName(_name + ".system.indexes"),
-      _viewsName(_name + ".system.views"),
+      _viewsName(_name + "." + DurableViewCatalog::viewsCollectionName().toString()),
       _durableViews(DurableViewCatalogImpl(this)),
-      _views(txn, &_durableViews) {
+      _views(&_durableViews) {
     Status status = validateDBName(_name);
     if (!status.isOK()) {
         warning() << "tried to open invalid db: " << _name << endl;
@@ -220,9 +220,14 @@ Database::Database(OperationContext* txn, StringData name, DatabaseCatalogEntry*
         const string ns = *it;
         _collections[ns] = _getOrCreateCollectionInstance(txn, ns);
     }
+    // At construction time of the viewCatalog, the _collections map wasn't initialized yet, so no
+    // system.views collection would be found. Now we're sufficiently initialized, signal a version
+    // change. Also force a reload, so if there are problems with the catalog contents as might be
+    // caused by incorrect mongod versions or similar, they are found right away.
+    getViewCatalog()->invalidate();
+    uassertStatusOK(_views.reloadIfNeeded(txn));
 }
 
-
 /*static*/
 string Database::duplicateUncasedName(const string& name, set<string>* duplicates) {
     if (duplicates) {
@@ -350,8 +355,8 @@ void Database::getStats(OperationContext* opCtx, BSONObjBuilder* output, double
     _dbEntry->appendExtraStats(opCtx, output, scale);
 }
 
-void Database::dropView(OperationContext* txn, StringData fullns) {
-    _views.dropView(txn, NamespaceString(fullns));
+Status Database::dropView(OperationContext* txn, StringData fullns) {
+    return (_views.dropView(txn, NamespaceString(fullns)));
 }
 
 Status Database::dropCollection(OperationContext* txn, StringData fullns) {
@@ -527,7 +532,7 @@ Status Database::createView(OperationContext* txn,
         return Status(ErrorCodes::InvalidNamespace,
                       str::stream() << "invalid namespace name for a view: " + nss.toString());
 
-    return _views.createView(txn, nss, viewOnNss, options.pipeline);
+    return _views.createView(txn, nss, viewOnNss, BSONArray(options.pipeline));
 }
 
 
@@ -648,7 +653,7 @@ Status userCreateNS(OperationContext* txn,
         return Status(ErrorCodes::NamespaceExists,
                       str::stream() << "a collection '" << ns.toString() << "' already exists");
 
-    if (db->getViewCatalog()->lookup(ns))
+    if (db->getViewCatalog()->lookup(txn, ns))
         return Status(ErrorCodes::NamespaceExists,
                       str::stream() << "a view '" << ns.toString() << "' already exists");
 
diff --git a/src/mongo/db/catalog/database.h b/src/mongo/db/catalog/database.h
index b04584b7361..bb5741cb498 100644
--- a/src/mongo/db/catalog/database.h
+++ b/src/mongo/db/catalog/database.h
@@ -151,7 +151,7 @@ public:
 
     Status dropCollection(OperationContext* txn, StringData fullns);
 
-    void dropView(OperationContext* txn, StringData fullns);
+    Status dropView(OperationContext* txn, StringData fullns);
 
     Collection* createCollection(OperationContext* txn,
                                  StringData ns,
diff --git a/src/mongo/db/catalog/drop_collection.cpp b/src/mongo/db/catalog/drop_collection.cpp
index c436f758b76..52df162eb05 100644
--- a/src/mongo/db/catalog/drop_collection.cpp
+++ b/src/mongo/db/catalog/drop_collection.cpp
@@ -64,7 +64,8 @@ Status dropCollection(OperationContext* txn,
         AutoGetDb autoDb(txn, dbname, MODE_X);
         Database* const db = autoDb.getDb();
         Collection* coll = db ? db->getCollection(collectionName) : nullptr;
-        ViewDefinition* view = db ? db->getViewCatalog()->lookup(collectionName.ns()) : nullptr;
+        ViewDefinition* view =
+            db ? db->getViewCatalog()->lookup(txn, collectionName.ns()) : nullptr;
 
         if (!db || (!coll && !view)) {
             return Status(ErrorCodes::NamespaceNotFound, "ns not found");
@@ -100,7 +101,10 @@ Status dropCollection(OperationContext* txn,
             result.append("nIndexesWas", numIndexes);
         } else {
             invariant(view);
-            db->dropView(txn, collectionName.ns());
+            Status status = db->dropView(txn, collectionName.ns());
+            if (!status.isOK()) {
+                return status;
+            }
         }
         wunit.commit();
     }
diff --git a/src/mongo/db/catalog/drop_indexes.cpp b/src/mongo/db/catalog/drop_indexes.cpp
index 3def51b36ed..cdeeae8cede 100644
--- a/src/mongo/db/catalog/drop_indexes.cpp
+++ b/src/mongo/db/catalog/drop_indexes.cpp
@@ -155,7 +155,7 @@ Status dropIndexes(OperationContext* txn,
                     str::stream() << "Not primary while dropping indexes in " << nss.ns()};
         }
 
-        if (db && db->getViewCatalog()->lookup(nss.ns())) {
+        if (db && db->getViewCatalog()->lookup(txn, nss.ns())) {
             return {ErrorCodes::CommandNotSupportedOnView,
                     str::stream() << "Cannot drop indexes on view " << nss.ns()};
         }
diff --git a/src/mongo/db/cloner.cpp b/src/mongo/db/cloner.cpp
index 9d4a5165424..e924fa7312b 100644
--- a/src/mongo/db/cloner.cpp
+++ b/src/mongo/db/cloner.cpp
@@ -455,7 +455,7 @@ StatusWith<std::vector<BSONObj>> Cloner::filterCollectionsForClone(
         const NamespaceString ns(opts.fromDB, collectionName.c_str());
 
         if (ns.isSystem()) {
-            if (legalClientSystemNS(ns.ns(), true) == 0) {
+            if (legalClientSystemNS(ns.ns()) == 0) {
                 LOG(2) << "\t\t not cloning because system collection" << endl;
                 continue;
             }
diff --git a/src/mongo/db/commands/compact.cpp b/src/mongo/db/commands/compact.cpp
index 4c68ff88fe9..bf768fe131b 100644
--- a/src/mongo/db/commands/compact.cpp
+++ b/src/mongo/db/commands/compact.cpp
@@ -149,7 +149,7 @@ public:
         Database* const collDB = autoDb.getDb();
 
         Collection* collection = collDB ? collDB->getCollection(nss) : nullptr;
-        auto view = collDB ? collDB->getViewCatalog()->lookup(nss.ns()) : nullptr;
+        auto view = collDB ? collDB->getViewCatalog()->lookup(txn, nss.ns()) : nullptr;
 
         // If db/collection does not exist, short circuit and return.
         if (!collDB || !collection) {
diff --git a/src/mongo/db/commands/create_indexes.cpp b/src/mongo/db/commands/create_indexes.cpp
index e37eb45233d..a3dcd815ebb 100644
--- a/src/mongo/db/commands/create_indexes.cpp
+++ b/src/mongo/db/commands/create_indexes.cpp
@@ -188,7 +188,7 @@ public:
 
         Database* db = dbHolder().get(txn, ns.db());
 
-        if (db && db->getViewCatalog()->lookup(ns.ns())) {
+        if (db && db->getViewCatalog()->lookup(txn, ns.ns())) {
             errmsg = "cannot create indexes on a view";
             return appendCommandStatus(result,
                                        Status(ErrorCodes::CommandNotSupportedOnView, errmsg));
diff --git a/src/mongo/db/commands/drop_indexes.cpp b/src/mongo/db/commands/drop_indexes.cpp
index fb98444202e..17fefcfa1ba 100644
--- a/src/mongo/db/commands/drop_indexes.cpp
+++ b/src/mongo/db/commands/drop_indexes.cpp
@@ -133,7 +133,7 @@ public:
         OldClientContext ctx(txn, toReIndexNs.ns());
 
         Collection* collection = ctx.db()->getCollection(toReIndexNs.ns());
-        auto view = ctx.db()->getViewCatalog()->lookup(toReIndexNs.ns());
+        auto view = ctx.db()->getViewCatalog()->lookup(txn, toReIndexNs.ns());
 
         if (!collection) {
             if (view)
diff --git a/src/mongo/db/commands/find_and_modify.cpp b/src/mongo/db/commands/find_and_modify.cpp
index 07655e32572..5a2a1f45435 100644
--- a/src/mongo/db/commands/find_and_modify.cpp
+++ b/src/mongo/db/commands/find_and_modify.cpp
@@ -395,7 +395,7 @@ public:
                 }
 
                 AutoGetOrCreateDb autoDb(txn, dbName, MODE_IX);
-                if (autoDb.getDb()->getViewCatalog()->lookup(nsString.ns())) {
+                if (autoDb.getDb()->getViewCatalog()->lookup(txn, nsString.ns())) {
                     return appendCommandStatus(result,
                                                {ErrorCodes::CommandNotSupportedOnView,
                                                 "findAndModify not supported on views"});
@@ -472,7 +472,7 @@ public:
                 }
 
                 AutoGetOrCreateDb autoDb(txn, dbName, MODE_IX);
-                if (autoDb.getDb()->getViewCatalog()->lookup(nsString.ns())) {
+                if (autoDb.getDb()->getViewCatalog()->lookup(txn, nsString.ns())) {
                     return appendCommandStatus(result,
                                                {ErrorCodes::CommandNotSupportedOnView,
                                                 "findAndModify not supported on views"});
diff --git a/src/mongo/db/db_raii.cpp b/src/mongo/db/db_raii.cpp
index 160de246216..2808c7e0e35 100644
--- a/src/mongo/db/db_raii.cpp
+++ b/src/mongo/db/db_raii.cpp
@@ -54,7 +54,9 @@ AutoGetCollection::AutoGetCollection(OperationContext* txn,
       _collLock(txn->lockState(), nss.ns(), modeColl),
       _coll(_autoDb.getDb() ? _autoDb.getDb()->getCollection(nss) : nullptr) {
     Database* db = _autoDb.getDb();
-    if (_viewMode == ViewMode::kViewsForbidden && db && db->getViewCatalog()->lookup(nss.ns()))
+    // If the database exists, but not the collection, check for views.
+    if (_viewMode == ViewMode::kViewsForbidden && db && !_coll &&
+        db->getViewCatalog()->lookup(txn, nss.ns()))
         uasserted(ErrorCodes::CommandNotSupportedOnView,
                   str::stream() << "Namespace " << nss.ns() << " is a view, not a collection");
 }
@@ -157,8 +159,9 @@ void AutoGetCollectionForRead::_ensureMajorityCommittedSnapshotIsValid(const Nam
 AutoGetCollectionOrViewForRead::AutoGetCollectionOrViewForRead(OperationContext* txn,
                                                                const NamespaceString& nss)
     : AutoGetCollectionForRead(txn, nss, AutoGetCollection::ViewMode::kViewsPermitted),
-      _view(_autoColl->getDb() ? _autoColl->getDb()->getViewCatalog()->lookup(nss.ns()) : nullptr) {
-}
+      _view(_autoColl->getDb() && !getCollection()
+                ? _autoColl->getDb()->getViewCatalog()->lookup(txn, nss.ns())
+                : nullptr) {}
 
 void AutoGetCollectionOrViewForRead::releaseLocksForView() noexcept {
     invariant(_view);
diff --git a/src/mongo/db/namespace_string.cpp b/src/mongo/db/namespace_string.cpp
index 5b53b509c5f..43f02e26bb5 100644
--- a/src/mongo/db/namespace_string.cpp
+++ b/src/mongo/db/namespace_string.cpp
@@ -74,7 +74,7 @@ constexpr auto listIndexesCursorNSPrefix = "$cmd.listIndexes."_sd;
 
 }  // namespace
 
-bool legalClientSystemNS(StringData ns, bool write) {
+bool legalClientSystemNS(StringData ns) {
     if (ns == "local.system.replset")
         return true;
 
@@ -93,6 +93,9 @@ bool legalClientSystemNS(StringData ns, bool write) {
     if (ns.find(".system.js") != string::npos)
         return true;
 
+    if (nsToCollectionSubstring(ns) == "system.views")
+        return true;
+
     return false;
 }
 
diff --git a/src/mongo/db/namespace_string.h b/src/mongo/db/namespace_string.h
index 7ffb802cb1e..bfab52bb92e 100644
--- a/src/mongo/db/namespace_string.h
+++ b/src/mongo/db/namespace_string.h
@@ -45,9 +45,8 @@ const size_t MaxDatabaseNameLen = 128;  // max str len for the db name, includin
 
 /** @return true if a client can modify this namespace even though it is under ".system."
     For example <dbname>.system.users is ok for regular clients to update.
-    @param write used when .system.js
 */
-bool legalClientSystemNS(StringData ns, bool write);
+bool legalClientSystemNS(StringData ns);
 
 /* e.g.
    NamespaceString ns("acme.orders");
diff --git a/src/mongo/db/op_observer.cpp b/src/mongo/db/op_observer.cpp
index 14b66943063..83f8b463945 100644
--- a/src/mongo/db/op_observer.cpp
+++ b/src/mongo/db/op_observer.cpp
@@ -37,6 +37,7 @@
 #include "mongo/db/operation_context.h"
 #include "mongo/db/repl/oplog.h"
 #include "mongo/db/s/collection_sharding_state.h"
+#include "mongo/db/views/durable_view_catalog.h"
 #include "mongo/scripting/engine.h"
 
 namespace mongo {
@@ -80,6 +81,9 @@ void OpObserver::onInserts(OperationContext* txn,
     if (strstr(ns, ".system.js")) {
         Scope::storedFuncMod(txn);
     }
+    if (nss.coll() == DurableViewCatalog::viewsCollectionName()) {
+        DurableViewCatalog::onExternalChange(txn, nss);
+    }
 }
 
 void OpObserver::onUpdate(OperationContext* txn, const OplogUpdateEntryArgs& args) {
@@ -101,6 +105,11 @@ void OpObserver::onUpdate(OperationContext* txn, const OplogUpdateEntryArgs& arg
     if (strstr(args.ns.c_str(), ".system.js")) {
         Scope::storedFuncMod(txn);
     }
+
+    NamespaceString nss(args.ns);
+    if (nss.coll() == DurableViewCatalog::viewsCollectionName()) {
+        DurableViewCatalog::onExternalChange(txn, nss);
+    }
 }
 
 CollectionShardingState::DeleteState OpObserver::aboutToDelete(OperationContext* txn,
@@ -138,6 +147,9 @@ void OpObserver::onDelete(OperationContext* txn,
     if (ns.coll() == "system.js") {
         Scope::storedFuncMod(txn);
     }
+    if (ns.coll() == DurableViewCatalog::viewsCollectionName()) {
+        DurableViewCatalog::onExternalChange(txn, ns);
+    }
 }
 
 void OpObserver::onOpMessage(OperationContext* txn, const BSONObj& msgObj) {
@@ -195,6 +207,9 @@ void OpObserver::onDropCollection(OperationContext* txn, const NamespaceString&
         repl::logOp(txn, "c", dbName.c_str(), cmdObj, nullptr, false);
     }
 
+    if (collectionName.coll() == DurableViewCatalog::viewsCollectionName()) {
+        DurableViewCatalog::onExternalChange(txn, collectionName);
+    }
     getGlobalAuthorizationManager()->logOp(txn, "c", dbName.c_str(), cmdObj, nullptr);
     logOpForDbHash(txn, dbName.c_str());
 }
@@ -221,6 +236,11 @@ void OpObserver::onRenameCollection(OperationContext* txn,
                                 << dropTarget);
 
     repl::logOp(txn, "c", dbName.c_str(), cmdObj, nullptr, false);
+    if (fromCollection.coll() == DurableViewCatalog::viewsCollectionName() ||
+        toCollection.coll() == DurableViewCatalog::viewsCollectionName()) {
+        DurableViewCatalog::onExternalChange(
+            txn, NamespaceString(DurableViewCatalog::viewsCollectionName()));
+    }
 
     getGlobalAuthorizationManager()->logOp(txn, "c", dbName.c_str(), cmdObj, nullptr);
     logOpForDbHash(txn, dbName.c_str());
diff --git a/src/mongo/db/ops/insert.cpp b/src/mongo/db/ops/insert.cpp
index c7648b0f254..db4a490392a 100644
--- a/src/mongo/db/ops/insert.cpp
+++ b/src/mongo/db/ops/insert.cpp
@@ -30,6 +30,7 @@
 
 #include "mongo/db/ops/insert.h"
 #include "mongo/db/global_timestamp.h"
+#include "mongo/db/views/durable_view_catalog.h"
 #include "mongo/util/mongoutils/str.h"
 
 namespace mongo {
@@ -185,6 +186,8 @@ Status userAllowedCreateNS(StringData db, StringData coll) {
             return Status::OK();
         if (coll == "system.users")
             return Status::OK();
+        if (coll == DurableViewCatalog::viewsCollectionName())
+            return Status::OK();
         if (db == "admin") {
             if (coll == "system.version")
                 return Status::OK();
diff --git a/src/mongo/db/query/get_executor.cpp b/src/mongo/db/query/get_executor.cpp
index d45c6145abb..5d31d2e214a 100644
--- a/src/mongo/db/query/get_executor.cpp
+++ b/src/mongo/db/query/get_executor.cpp
@@ -693,8 +693,7 @@ StatusWith<unique_ptr<PlanExecutor>> getExecutorDelete(OperationContext* txn,
     const NamespaceString& nss(request->getNamespaceString());
     if (!request->isGod()) {
         if (nss.isSystem()) {
-            uassert(
-                12050, "cannot delete from system namespace", legalClientSystemNS(nss.ns(), true));
+            uassert(12050, "cannot delete from system namespace", legalClientSystemNS(nss.ns()));
         }
         if (nss.isVirtualized()) {
             log() << "cannot delete from a virtual collection: " << nss;
@@ -833,7 +832,7 @@ inline void validateUpdate(const char* ns, const BSONObj& updateobj, const BSONO
                 str::stream() << "cannot update system collection: " << ns << " q: " << patternOrig
                               << " u: "
                               << updateobj,
-                legalClientSystemNS(ns, true));
+                legalClientSystemNS(ns));
     }
 }
 
diff --git a/src/mongo/db/repl/databases_cloner.cpp b/src/mongo/db/repl/databases_cloner.cpp
index 4542db17e55..b668374513c 100644
--- a/src/mongo/db/repl/databases_cloner.cpp
+++ b/src/mongo/db/repl/databases_cloner.cpp
@@ -219,7 +219,7 @@ void DatabasesCloner::_onListDatabaseFinish(const CommandCallbackArgs& cbd) {
         const auto collectionFilterPred = [dbName](const BSONObj& collInfo) {
             const auto collName = collInfo["name"].str();
             const NamespaceString ns(dbName, collName);
-            if (ns.isSystem() && !legalClientSystemNS(ns.ns(), true)) {
+            if (ns.isSystem() && !legalClientSystemNS(ns.ns())) {
                 LOG(1) << "Skipping 'system' collection: " << ns.ns();
                 return false;
             }
diff --git a/src/mongo/db/s/set_shard_version_command.cpp b/src/mongo/db/s/set_shard_version_command.cpp
index 5a0a8afd8c1..b19e72c9489 100644
--- a/src/mongo/db/s/set_shard_version_command.cpp
+++ b/src/mongo/db/s/set_shard_version_command.cpp
@@ -187,7 +187,7 @@ public:
             autoDb.emplace(txn, nss.db(), MODE_IS);
 
             // Views do not require a shard version check.
-            if (autoDb->getDb() && autoDb->getDb()->getViewCatalog()->lookup(nss.ns())) {
+            if (autoDb->getDb() && autoDb->getDb()->getViewCatalog()->lookup(txn, nss.ns())) {
                 return true;
             }
 
diff --git a/src/mongo/db/views/durable_view_catalog.cpp b/src/mongo/db/views/durable_view_catalog.cpp
index 2fa3de3364d..76eb693478a 100644
--- a/src/mongo/db/views/durable_view_catalog.cpp
+++ b/src/mongo/db/views/durable_view_catalog.cpp
@@ -33,24 +33,46 @@
 #include "mongo/db/views/durable_view_catalog.h"
 
 #include <string>
+#include <unordered_set>
 
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/database.h"
+#include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/dbhelpers.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/storage/record_data.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/log.h"
+#include "mongo/util/string_map.h"
 
 namespace mongo {
 
-void DurableViewCatalogImpl::iterate(OperationContext* txn, Callback callback) {
-    dassert(txn->lockState()->isDbLockedForMode(_db->name(), MODE_X));
+// DurableViewCatalog
+
+void DurableViewCatalog::onExternalChange(OperationContext* txn, const NamespaceString& name) {
+    dassert(txn->lockState()->isDbLockedForMode(name.db(), MODE_IX));
+    Database* db = dbHolder().get(txn, name.db());
+
+    if (db) {
+        txn->recoveryUnit()->onCommit([db]() { db->getViewCatalog()->invalidate(); });
+    }
+}
+
+// DurableViewCatalogImpl
+
+const std::string& DurableViewCatalogImpl::getName() const {
+    return _db->name();
+}
+
+Status DurableViewCatalogImpl::iterate(OperationContext* txn, Callback callback) {
+    dassert(txn->lockState()->isDbLockedForMode(_db->name(), MODE_IS) ||
+            txn->lockState()->isDbLockedForMode(_db->name(), MODE_IX));
     Collection* systemViews = _db->getCollection(_db->getSystemViewsName());
     if (!systemViews)
-        return;
+        return Status::OK();
 
+    Lock::CollectionLock lk(txn->lockState(), _db->getSystemViewsName(), MODE_IS);
     auto cursor = systemViews->getCursor(txn);
     while (auto record = cursor->next()) {
         RecordData& data = record->data;
@@ -59,27 +81,62 @@ void DurableViewCatalogImpl::iterate(OperationContext* txn, Callback callback) {
         fassertStatusOK(40224, validateBSON(data.data(), data.size()));
         BSONObj viewDef = data.toBson();
 
-        // Make sure we fail when new fields get added to the definition, so we fail safe in case
-        // of future format upgrades.
+        // Check read definitions for correct structure, and refuse reading past invalid
+        // definitions. Complain loudly, but otherwise ignore any further view definitions.
+        bool valid = true;
         for (const BSONElement& e : viewDef) {
             std::string name(e.fieldName());
-            fassert(40225, name == "_id" || name == "viewOn" || name == "pipeline");
+            valid &= name == "_id" || name == "viewOn" || name == "pipeline";
         }
         NamespaceString viewName(viewDef["_id"].str());
-        fassert(40226, viewName.db() == _db->name());
+        valid &= viewName.isValid() && viewName.db() == _db->name();
+        valid &= NamespaceString::validCollectionName(viewDef["viewOn"].str());
+
+        if (!valid) {
+            return {ErrorCodes::InvalidViewDefinition,
+                    str::stream() << "invalid view definitions reading '"
+                                  << _db->getSystemViewsName()
+                                  << "'"};
+        }
 
         callback(viewDef);
     }
+    return Status::OK();
 }
 
-void DurableViewCatalogImpl::insert(OperationContext* txn, const BSONObj& view) {
+void DurableViewCatalogImpl::upsert(OperationContext* txn,
+                                    const NamespaceString& name,
+                                    const BSONObj& view) {
     dassert(txn->lockState()->isDbLockedForMode(_db->name(), MODE_X));
-    Collection* systemViews = _db->getOrCreateCollection(txn, _db->getSystemViewsName());
+    NamespaceString systemViewsNs(_db->getSystemViewsName());
+    Collection* systemViews = _db->getOrCreateCollection(txn, systemViewsNs.ns());
+
+    const bool requireIndex = false;
+    RecordId id = Helpers::findOne(txn, systemViews, BSON("_id" << name.ns()), requireIndex);
 
-    OpDebug* const opDebug = nullptr;
-    const bool enforceQuota = false;
-    LOG(2) << "insert view " << view << " in " << _db->getSystemViewsName();
-    uassertStatusOK(systemViews->insertDocument(txn, view, opDebug, enforceQuota));
+    const bool enforceQuota = true;
+    Snapshotted<BSONObj> oldView;
+    if (!id.isNormal() || !systemViews->findDoc(txn, id, &oldView)) {
+        LOG(2) << "insert view " << view << " into " << _db->getSystemViewsName();
+        uassertStatusOK(
+            systemViews->insertDocument(txn, view, &CurOp::get(txn)->debug(), enforceQuota));
+    } else {
+        OplogUpdateEntryArgs args;
+        args.ns = systemViewsNs.ns();
+        args.update = view;
+        args.criteria = BSON("_id" << name.ns());
+
+        const bool assumeIndexesAreAffected = true;
+        auto res = systemViews->updateDocument(txn,
+                                               id,
+                                               oldView,
+                                               view,
+                                               enforceQuota,
+                                               assumeIndexesAreAffected,
+                                               &CurOp::get(txn)->debug(),
+                                               &args);
+        uassertStatusOK(res);
+    }
 }
 
 void DurableViewCatalogImpl::remove(OperationContext* txn, const NamespaceString& name) {
@@ -93,7 +150,6 @@ void DurableViewCatalogImpl::remove(OperationContext* txn, const NamespaceString
         return;
 
     LOG(2) << "remove view " << name << " from " << _db->getSystemViewsName();
-    OpDebug* const opDebug = nullptr;
-    systemViews->deleteDocument(txn, id, opDebug);
+    systemViews->deleteDocument(txn, id, &CurOp::get(txn)->debug());
 }
 }  // namespace mongo
diff --git a/src/mongo/db/views/durable_view_catalog.h b/src/mongo/db/views/durable_view_catalog.h
index 68dec43255f..1ea7268f2ee 100644
--- a/src/mongo/db/views/durable_view_catalog.h
+++ b/src/mongo/db/views/durable_view_catalog.h
@@ -28,6 +28,10 @@
 
 #pragma once
 
+#include <string>
+
+#include "mongo/base/status.h"
+#include "mongo/base/string_data.h"
 #include "mongo/stdx/functional.h"
 
 namespace mongo {
@@ -39,15 +43,28 @@ class OperationContext;
 
 /**
  * Interface for system.views collection operations associated with view catalog management.
- * Methods must be called from within a WriteUnitOfWork, and with the DBLock held.
+ * All methods must be called from within a WriteUnitOfWork, and with the DBLock held in at
+ * least intent mode.
  */
 class DurableViewCatalog {
 public:
-    using Callback = stdx::function<void(const BSONObj& view)>;
+    static constexpr StringData viewsCollectionName() {
+        return "system.views"_sd;
+    }
 
-    virtual void iterate(OperationContext* txn, Callback callback) = 0;
-    virtual void insert(OperationContext* txn, const BSONObj& view) = 0;
+    /**
+     * Thread-safe method to mark a catalog name was changed. This will cause the in-memory
+     * view catalog to be marked invalid
+     */
+    static void onExternalChange(OperationContext* txn, const NamespaceString& name);
+
+    using Callback = stdx::function<void(const BSONObj& view)>;
+    virtual Status iterate(OperationContext* txn, Callback callback) = 0;
+    virtual void upsert(OperationContext* txn,
+                        const NamespaceString& name,
+                        const BSONObj& view) = 0;
     virtual void remove(OperationContext* txn, const NamespaceString& name) = 0;
+    virtual const std::string& getName() const = 0;
 };
 
 /**
@@ -58,9 +75,10 @@ class DurableViewCatalogImpl final : public DurableViewCatalog {
 public:
     explicit DurableViewCatalogImpl(Database* db) : _db(db) {}
 
-    void iterate(OperationContext* txn, Callback callback);
-    void insert(OperationContext* txn, const BSONObj& view);
+    Status iterate(OperationContext* txn, Callback callback);
+    void upsert(OperationContext* txn, const NamespaceString& name, const BSONObj& view);
     void remove(OperationContext* txn, const NamespaceString& name);
+    const std::string& getName() const;
 
 private:
     Database* const _db;
diff --git a/src/mongo/db/views/view_catalog.cpp b/src/mongo/db/views/view_catalog.cpp
index fd738cb2f6d..edfc152278f 100644
--- a/src/mongo/db/views/view_catalog.cpp
+++ b/src/mongo/db/views/view_catalog.cpp
@@ -41,6 +41,7 @@
 #include "mongo/db/server_parameters.h"
 #include "mongo/db/storage/recovery_unit.h"
 #include "mongo/db/views/view.h"
+#include "mongo/util/log.h"
 
 namespace {
 bool enableViews = false;
@@ -52,28 +53,63 @@ ExportedServerParameter<bool, ServerParameterType::kStartupOnly> enableViewsPara
 
 const std::uint32_t ViewCatalog::kMaxViewDepth = 20;
 
-ViewCatalog::ViewCatalog(OperationContext* txn, DurableViewCatalog* durable) : _durable(durable) {
-    durable->iterate(txn, [&](const BSONObj& view) {
+Status ViewCatalog::reloadIfNeeded(OperationContext* txn) {
+    stdx::lock_guard<stdx::mutex> lk(_mutex);
+    return _reloadIfNeeded_inlock(txn);
+}
+
+Status ViewCatalog::_reloadIfNeeded_inlock(OperationContext* txn) {
+    if (_valid.load())
+        return Status::OK();
+
+    LOG(1) << "reloading view catalog for database " << _durable->getName();
+
+    // Need to reload, first clear our cache.
+    _viewMap.clear();
+
+    Status status = _durable->iterate(txn, [&](const BSONObj& view) {
         NamespaceString viewName(view["_id"].str());
         ViewDefinition def(
             viewName.db(), viewName.coll(), view["viewOn"].str(), view["pipeline"].Obj());
         _viewMap[viewName.ns()] = std::make_shared<ViewDefinition>(def);
     });
+    _valid.store(status.isOK());
+    return status;
+}
+
+void ViewCatalog::_createOrUpdateView_inlock(OperationContext* txn,
+                                             const NamespaceString& viewName,
+                                             const NamespaceString& viewOn,
+                                             const BSONArray& pipeline) {
+    invariant(_valid.load());
+    BSONObj viewDef =
+        BSON("_id" << viewName.ns() << "viewOn" << viewOn.coll() << "pipeline" << pipeline);
+    _durable->upsert(txn, viewName, viewDef);
+
+    BSONObj ownedPipeline = pipeline.getOwned();
+    _viewMap[viewName.ns()] = std::make_shared<ViewDefinition>(
+        viewName.db(), viewName.coll(), viewOn.coll(), ownedPipeline);
+    txn->recoveryUnit()->onRollback([this, viewName]() { this->_viewMap.erase(viewName.ns()); });
+
+    // We may get invalidated, but we're exclusively locked, so the change must be ours.
+    txn->recoveryUnit()->onCommit([this]() { this->_valid.store(true); });
 }
 
+
 Status ViewCatalog::createView(OperationContext* txn,
                                const NamespaceString& viewName,
                                const NamespaceString& viewOn,
-                               const BSONObj& pipeline) {
+                               const BSONArray& pipeline) {
+    stdx::lock_guard<stdx::mutex> lk(_mutex);
+
     if (!enableViews)
         return Status(ErrorCodes::CommandNotSupported, "View support not enabled");
 
-
     if (viewName.db() != viewOn.db())
         return Status(ErrorCodes::BadValue,
                       "View must be created on a view or collection in the same database");
 
-    if (lookup(StringData(viewName.ns())))
+    if (_lookup_inlock(txn, StringData(viewName.ns())))
         return Status(ErrorCodes::NamespaceExists, "Namespace already exists");
 
     if (!NamespaceString::validCollectionName(viewOn.coll()))
@@ -82,28 +118,58 @@ Status ViewCatalog::createView(OperationContext* txn,
 
     // TODO(SERVER-24768): Need to ensure view is correct and doesn't introduce a cycle.
 
-    BSONObj viewDef =
-        BSON("_id" << viewName.ns() << "viewOn" << viewOn.coll() << "pipeline" << pipeline);
-    _durable->insert(txn, viewDef);
+    _createOrUpdateView_inlock(txn, viewName, viewOn, pipeline);
+    return Status::OK();
+}
 
-    BSONObj ownedPipeline = pipeline.getOwned();
-    _viewMap[viewName.ns()] = std::make_shared<ViewDefinition>(
-        viewName.db(), viewName.coll(), viewOn.coll(), ownedPipeline);
-    txn->recoveryUnit()->onRollback([this, viewName]() { this->_viewMap.erase(viewName.ns()); });
+Status ViewCatalog::modifyView(OperationContext* txn,
+                               const NamespaceString& viewName,
+                               const NamespaceString& viewOn,
+                               const BSONArray& pipeline) {
+    stdx::lock_guard<stdx::mutex> lk(_mutex);
+
+    if (viewName.db() != viewOn.db())
+        return Status(ErrorCodes::BadValue,
+                      "View must be created on a view or collection in the same database");
+
+    if (!_lookup_inlock(txn, StringData(viewName.ns())))
+        return Status(ErrorCodes::NamespaceNotFound,
+                      str::stream() << "cannot modify missing view " << viewName.ns());
+
+    if (!NamespaceString::validCollectionName(viewOn.coll()))
+        return Status(ErrorCodes::InvalidNamespace,
+                      str::stream() << "invalid name for 'viewOn': " << viewOn.coll());
+
+    _createOrUpdateView_inlock(txn, viewName, viewOn, pipeline);
     return Status::OK();
 }
 
-void ViewCatalog::dropView(OperationContext* txn, const NamespaceString& viewName) {
-    _durable->remove(txn, viewName);
+Status ViewCatalog::dropView(OperationContext* txn, const NamespaceString& viewName) {
+    stdx::lock_guard<stdx::mutex> lk(_mutex);
+
     // Save a copy of the view definition in case we need to roll back.
-    ViewDefinition savedDefinition = *lookup(viewName.ns());
+    ViewDefinition* viewPtr = _lookup_inlock(txn, viewName.ns());
+    if (!viewPtr) {
+        return {ErrorCodes::NamespaceNotFound,
+                str::stream() << "cannot drop missing view: " << viewName.ns()};
+    }
+
+    ViewDefinition savedDefinition = *viewPtr;
+
+    invariant(_valid.load());
+    _durable->remove(txn, viewName);
     _viewMap.erase(viewName.ns());
     txn->recoveryUnit()->onRollback([this, viewName, savedDefinition]() {
         this->_viewMap[viewName.ns()] = std::make_shared<ViewDefinition>(savedDefinition);
     });
+
+    // We may get invalidated, but we're exclusively locked, so the change must be ours.
+    txn->recoveryUnit()->onCommit([this]() { this->_valid.store(true); });
+    return Status::OK();
 }
 
-ViewDefinition* ViewCatalog::lookup(StringData ns) {
+ViewDefinition* ViewCatalog::_lookup_inlock(OperationContext* txn, StringData ns) {
+    uassertStatusOK(_reloadIfNeeded_inlock(txn));
     ViewMap::const_iterator it = _viewMap.find(ns);
     if (it != _viewMap.end()) {
         return it->second.get();
@@ -111,13 +177,19 @@ ViewDefinition* ViewCatalog::lookup(StringData ns) {
     return nullptr;
 }
 
+ViewDefinition* ViewCatalog::lookup(OperationContext* txn, StringData ns) {
+    stdx::lock_guard<stdx::mutex> lk(_mutex);
+    return _lookup_inlock(txn, ns);
+}
+
 StatusWith<ResolvedView> ViewCatalog::resolveView(OperationContext* txn,
                                                   const NamespaceString& nss) {
+    stdx::lock_guard<stdx::mutex> lk(_mutex);
     const NamespaceString* resolvedNss = &nss;
     std::vector<BSONObj> resolvedPipeline;
 
     for (std::uint32_t i = 0; i < ViewCatalog::kMaxViewDepth; i++) {
-        ViewDefinition* view = lookup(resolvedNss->ns());
+        ViewDefinition* view = _lookup_inlock(txn, resolvedNss->ns());
         if (!view)
             return StatusWith<ResolvedView>({*resolvedNss, resolvedPipeline});
 
diff --git a/src/mongo/db/views/view_catalog.h b/src/mongo/db/views/view_catalog.h
index c3cbb505f99..e96278c8ce9 100644
--- a/src/mongo/db/views/view_catalog.h
+++ b/src/mongo/db/views/view_catalog.h
@@ -41,14 +41,15 @@
 #include "mongo/db/views/durable_view_catalog.h"
 #include "mongo/db/views/resolved_view.h"
 #include "mongo/db/views/view.h"
+#include "mongo/stdx/mutex.h"
 #include "mongo/util/string_map.h"
 
 namespace mongo {
 class OperationContext;
 
 /**
- * In-memory data structure for view definitions. Note that this structure is not thread-safe; you
- * must be holding a database lock to access a database's view catalog.
+ * In-memory data structure for view definitions. This datastructure is thread-safe. This is needed
+ * as concurrent updates may happen through direct writes to the views catalog collection.
  */
 class ViewCatalog {
     MONGO_DISALLOW_COPYING(ViewCatalog);
@@ -58,7 +59,7 @@ public:
     using ViewMap = StringMap<std::shared_ptr<ViewDefinition>>;
     static const std::uint32_t kMaxViewDepth;
 
-    ViewCatalog(OperationContext* txn, DurableViewCatalog* durable);
+    explicit ViewCatalog(DurableViewCatalog* durable) : _durable(durable) {}
 
     ViewMap::const_iterator begin() const {
         return _viewMap.begin();
@@ -80,14 +81,25 @@ public:
     Status createView(OperationContext* txn,
                       const NamespaceString& viewName,
                       const NamespaceString& viewOn,
-                      const BSONObj& pipeline);
+                      const BSONArray& pipeline);
 
     /**
      * Drop the view named 'viewName'.
      *
      * Must be in WriteUnitOfWork. The drop rolls back if the unit of work aborts.
      */
-    void dropView(OperationContext* txn, const NamespaceString& viewName);
+    Status dropView(OperationContext* txn, const NamespaceString& viewName);
+
+    /**
+     * Modify the view named 'viewName' to have the new 'viewOn' and 'pipeline'.
+     *
+     * Must be in WriteUnitOfWork. The modification rolls back if the unit of work aborts.
+     */
+    Status modifyView(OperationContext* txn,
+                      const NamespaceString& viewName,
+                      const NamespaceString& viewOn,
+                      const BSONArray& pipeline);
+
 
     /**
      * Look up the namespace in the view catalog, returning a pointer to a View definition, or
@@ -96,7 +108,7 @@ public:
      * @param ns The full namespace string of the view.
      * @return A bare pointer to a view definition if ns is a valid view with a backing namespace.
      */
-    ViewDefinition* lookup(StringData ns);
+    ViewDefinition* lookup(OperationContext* txn, StringData ns);
 
     /**
      * Resolve the views on 'ns', transforming the pipeline appropriately. This function returns a
@@ -107,8 +119,34 @@ public:
      */
     StatusWith<ResolvedView> resolveView(OperationContext* txn, const NamespaceString& nss);
 
+    /**
+     * Reload the views catalog if marked invalid. No-op if already valid. Does only minimal
+     * validation, namely that the view definitions are valid BSON and have no unknown fields.
+     * No cycle detection etc. This is implicitly called by other methods when the ViewCatalog is
+     * marked invalid, and on first opening a database.
+     */
+    Status reloadIfNeeded(OperationContext* txn);
+
+    /**
+     * To be called when direct modifications to the DurableViewCatalog have been committed, so
+     * subsequent lookups will reload the catalog and make the changes visible.
+     */
+    void invalidate() {
+        _valid.store(false);
+    }
+
 private:
+    void _createOrUpdateView_inlock(OperationContext* txn,
+                                    const NamespaceString& viewName,
+                                    const NamespaceString& viewOn,
+                                    const BSONArray& pipeline);
+
+    ViewDefinition* _lookup_inlock(OperationContext* txn, StringData ns);
+    Status _reloadIfNeeded_inlock(OperationContext* txn);
+
+    stdx::mutex _mutex;  // Protects all members, except for _valid.
     ViewMap _viewMap;
     DurableViewCatalog* _durable;
+    AtomicBool _valid;
 };
 }  // namespace mongo
diff --git a/src/mongo/util/unordered_fast_key_table.h b/src/mongo/util/unordered_fast_key_table.h
index 71a67355b7a..763ac0e4d57 100644
--- a/src/mongo/util/unordered_fast_key_table.h
+++ b/src/mongo/util/unordered_fast_key_table.h
@@ -170,6 +170,10 @@ public:
         return get(HashedKey(key));
     }
 
+    void clear() {
+        *this = {};
+    }
+
     /**
      * @return number of elements removed
      */