refactor: Change recursive_mutex to mutex in DatabaseRotatingImp (#5276)

Rewrites the code so that the lock is not held during the callback. Instead it locks twice, once before, and once after. This is safe due to the structure of the code, but is checked after the second lock. This allows mutex_ to be changed back to a regular mutex.

src/test/app/SHAMapStore_test.cpp

@@ -20,9 +20,11 @@
 #include <test/jtx.h>
 #include <test/jtx/envconfig.h>
 #include <xrpld/app/main/Application.h>
+#include <xrpld/app/main/NodeStoreScheduler.h>
 #include <xrpld/app/misc/SHAMapStore.h>
 #include <xrpld/app/rdb/backend/SQLiteDatabase.h>
 #include <xrpld/core/ConfigSections.h>
+#include <xrpld/nodestore/detail/DatabaseRotatingImp.h>
 #include <xrpl/protocol/jss.h>
 
 namespace ripple {
@@ -518,12 +520,136 @@ public:
         lastRotated = ledgerSeq - 1;
     }
 
+    std::unique_ptr<NodeStore::Backend>
+    makeBackendRotating(
+        jtx::Env& env,
+        NodeStoreScheduler& scheduler,
+        std::string path)
+    {
+        Section section{
+            env.app().config().section(ConfigSection::nodeDatabase())};
+        boost::filesystem::path newPath;
+
+        if (!BEAST_EXPECT(path.size()))
+            return {};
+        newPath = path;
+        section.set("path", newPath.string());
+
+        auto backend{NodeStore::Manager::instance().make_Backend(
+            section,
+            megabytes(env.app().config().getValueFor(
+                SizedItem::burstSize, std::nullopt)),
+            scheduler,
+            env.app().logs().journal("NodeStoreTest"))};
+        backend->open();
+        return backend;
+    }
+
+    void
+    testRotate()
+    {
+        // The only purpose of this test is to ensure that if something that
+        // should never happen happens, we don't get a deadlock.
+        testcase("rotate with lock contention");
+
+        using namespace jtx;
+        Env env(*this, envconfig(onlineDelete));
+
+        /////////////////////////////////////////////////////////////
+        // Create the backend. Normally, SHAMapStoreImp handles all these
+        // details
+        auto nscfg = env.app().config().section(ConfigSection::nodeDatabase());
+
+        // Provide default values:
+        if (!nscfg.exists("cache_size"))
+            nscfg.set(
+                "cache_size",
+                std::to_string(env.app().config().getValueFor(
+                    SizedItem::treeCacheSize, std::nullopt)));
+
+        if (!nscfg.exists("cache_age"))
+            nscfg.set(
+                "cache_age",
+                std::to_string(env.app().config().getValueFor(
+                    SizedItem::treeCacheAge, std::nullopt)));
+
+        NodeStoreScheduler scheduler(env.app().getJobQueue());
+
+        std::string const writableDb = "write";
+        std::string const archiveDb = "archive";
+        auto writableBackend = makeBackendRotating(env, scheduler, writableDb);
+        auto archiveBackend = makeBackendRotating(env, scheduler, archiveDb);
+
+        // Create NodeStore with two backends to allow online deletion of
+        // data
+        constexpr int readThreads = 4;
+        auto dbr = std::make_unique<NodeStore::DatabaseRotatingImp>(
+            scheduler,
+            readThreads,
+            std::move(writableBackend),
+            std::move(archiveBackend),
+            nscfg,
+            env.app().logs().journal("NodeStoreTest"));
+
+        /////////////////////////////////////////////////////////////
+        // Check basic functionality
+        using namespace std::chrono_literals;
+        std::atomic<int> threadNum = 0;
+
+        {
+            auto newBackend = makeBackendRotating(
+                env, scheduler, std::to_string(++threadNum));
+
+            auto const cb = [&](std::string const& writableName,
+                                std::string const& archiveName) {
+                BEAST_EXPECT(writableName == "1");
+                BEAST_EXPECT(archiveName == "write");
+                // Ensure that dbr functions can be called from within the
+                // callback
+                BEAST_EXPECT(dbr->getName() == "1");
+            };
+
+            dbr->rotate(std::move(newBackend), cb);
+        }
+        BEAST_EXPECT(threadNum == 1);
+        BEAST_EXPECT(dbr->getName() == "1");
+
+        /////////////////////////////////////////////////////////////
+        // Do something stupid. Try to re-enter rotate from inside the callback.
+        {
+            auto const cb = [&](std::string const& writableName,
+                                std::string const& archiveName) {
+                BEAST_EXPECT(writableName == "3");
+                BEAST_EXPECT(archiveName == "2");
+                // Ensure that dbr functions can be called from within the
+                // callback
+                BEAST_EXPECT(dbr->getName() == "3");
+            };
+            auto const cbReentrant = [&](std::string const& writableName,
+                                         std::string const& archiveName) {
+                BEAST_EXPECT(writableName == "2");
+                BEAST_EXPECT(archiveName == "1");
+                auto newBackend = makeBackendRotating(
+                    env, scheduler, std::to_string(++threadNum));
+                // Reminder: doing this is stupid and should never happen
+                dbr->rotate(std::move(newBackend), cb);
+            };
+            auto newBackend = makeBackendRotating(
+                env, scheduler, std::to_string(++threadNum));
+            dbr->rotate(std::move(newBackend), cbReentrant);
+        }
+
+        BEAST_EXPECT(threadNum == 3);
+        BEAST_EXPECT(dbr->getName() == "3");
+    }
+
     void
     run() override
     {
         testClear();
         testAutomatic();
         testCanDelete();
+        testRotate();
     }
 };