feat: Graceful shutdown with old web server (#2786)

- Stop accepting connections during graceful shutdown in the old web server
- Stop all the services before Clio exits
- Move cache saving into stop callback

Author: kuznetsss

src/app/ClioApplication.cpp

@@ -25,6 +25,7 @@
 #include "data/AmendmentCenter.hpp"
 #include "data/BackendFactory.hpp"
 #include "data/LedgerCache.hpp"
+#include "data/LedgerCacheSaver.hpp"
 #include "etl/ETLService.hpp"
 #include "etl/LoadBalancer.hpp"
 #include "etl/NetworkValidatedLedgers.hpp"
@@ -98,36 +99,23 @@ ClioApplication::run(bool const useNgWebServer)
     auto const threads = config_.get<uint16_t>("io_threads");
     LOG(util::LogService::info()) << "Number of io threads = " << threads;
 
-    // IO context to handle all incoming requests, as well as other things.
-    // This is not the only io context in the application.
-    boost::asio::io_context ioc{threads};
-
     // Similarly we need a context to run ETL on
     // In the future we can remove the raw ioc and use ctx instead
+    // This context should be above ioc because its reference is getting into tasks inside ioc
     util::async::CoroExecutionContext ctx{threads};
 
+    // IO context to handle all incoming requests, as well as other things.
+    // This is not the only io context in the application.
+    boost::asio::io_context ioc{threads};
+
     // Rate limiter, to prevent abuse
     auto whitelistHandler = web::dosguard::WhitelistHandler{config_};
     auto const dosguardWeights = web::dosguard::Weights::make(config_);
     auto dosGuard = web::dosguard::DOSGuard{config_, whitelistHandler, dosguardWeights};
     auto sweepHandler = web::dosguard::IntervalSweepHandler{config_, ioc, dosGuard};
 
     auto cache = data::LedgerCache{};
-    appStopper_.setOnStop([&cache, this](auto&&) {
-        // TODO(kuznetsss): move this into Stopper::makeOnStopCallback()
-        auto const cacheFilePath = config_.maybeValue<std::string>("cache.file.path");
-        if (not cacheFilePath.has_value()) {
-            return;
-        }
-
-        LOG(util::LogService::info()) << "Saving ledger cache to " << *cacheFilePath;
-        if (auto const [success, duration_ms] = util::timed([&]() { return cache.saveToFile(*cacheFilePath); });
-            success.has_value()) {
-            LOG(util::LogService::info()) << "Successfully saved ledger cache in " << duration_ms << " ms";
-        } else {
-            LOG(util::LogService::error()) << "Error saving LedgerCache to file";
-        }
-    });
+    auto cacheSaver = data::LedgerCacheSaver{config_, cache};
 
     // Interface to the database
     auto backend = data::makeBackend(config_, cache);
@@ -208,7 +196,7 @@ ClioApplication::run(bool const useNgWebServer)
         }
 
         appStopper_.setOnStop(
-            Stopper::makeOnStopCallback(httpServer.value(), *balancer, *etl, *subscriptions, *backend, ioc)
+            Stopper::makeOnStopCallback(httpServer.value(), *balancer, *etl, *subscriptions, *backend, cacheSaver, ioc)
         );
 
         // Blocks until stopped.
@@ -223,6 +211,9 @@ ClioApplication::run(bool const useNgWebServer)
     auto handler = std::make_shared<web::RPCServerHandler<RPCEngineType>>(config_, backend, rpcEngine, etl, dosGuard);
 
     auto const httpServer = web::makeHttpServer(config_, ioc, dosGuard, handler, cache);
+    appStopper_.setOnStop(
+        Stopper::makeOnStopCallback(*httpServer, *balancer, *etl, *subscriptions, *backend, cacheSaver, ioc)
+    );
 
     // Blocks until stopped.
     // When stopped, shared_ptrs fall out of scope

src/app/Stopper.hpp

@@ -20,12 +20,13 @@
 #pragma once
 
 #include "data/BackendInterface.hpp"
+#include "data/LedgerCacheSaver.hpp"
 #include "etl/ETLServiceInterface.hpp"
 #include "etl/LoadBalancerInterface.hpp"
 #include "feed/SubscriptionManagerInterface.hpp"
 #include "util/CoroutineGroup.hpp"
 #include "util/log/Logger.hpp"
-#include "web/ng/Server.hpp"
+#include "web/interface/Concepts.hpp"
 
 #include <boost/asio/executor_work_guard.hpp>
 #include <boost/asio/io_context.hpp>
@@ -71,21 +72,25 @@ class Stopper {
      * @param etl The ETL service to stop.
      * @param subscriptions The subscription manager to stop.
      * @param backend The backend to stop.
+     * @param cacheSaver The ledger cache saver
      * @param ioc The io_context to stop.
      * @return The callback to be called on application stop.
      */
-    template <web::ng::SomeServer ServerType>
+    template <web::SomeServer ServerType, data::SomeLedgerCacheSaver LedgerCacheSaverType>
     static std::function<void(boost::asio::yield_context)>
     makeOnStopCallback(
         ServerType& server,
         etl::LoadBalancerInterface& balancer,
         etl::ETLServiceInterface& etl,
         feed::SubscriptionManagerInterface& subscriptions,
         data::BackendInterface& backend,
+        LedgerCacheSaverType& cacheSaver,
         boost::asio::io_context& ioc
     )
     {
         return [&](boost::asio::yield_context yield) {
+            cacheSaver.save();
+
             util::CoroutineGroup coroutineGroup{yield};
             coroutineGroup.spawn(yield, [&server](auto innerYield) {
                 server.stop(innerYield);
@@ -106,6 +111,8 @@ class Stopper {
             backend.waitForWritesToFinish();
             LOG(util::LogService::info()) << "Backend writes finished";
 
+            cacheSaver.waitToFinish();
+
             ioc.stop();
             LOG(util::LogService::info()) << "io_context stopped";
 

src/data/CMakeLists.txt

@@ -5,6 +5,7 @@ target_sources(
           BackendCounters.cpp
           BackendInterface.cpp
           LedgerCache.cpp
+          LedgerCacheSaver.cpp
           LedgerHeaderCache.cpp
           cassandra/impl/Future.cpp
           cassandra/impl/Cluster.cpp

src/data/LedgerCache.cpp

@@ -265,7 +265,7 @@ LedgerCache::saveToFile(std::string const& path) const
     }
 
     impl::LedgerCacheFile file{path};
-    std::unique_lock const lock{mtx_};
+    std::shared_lock const lock{mtx_};
     impl::LedgerCacheFile::DataView const data{.latestSeq = latestSeq_, .map = map_, .deleted = deleted_};
     return file.write(data);
 }

src/data/LedgerCache.hpp

@@ -145,15 +145,8 @@ class LedgerCache : public LedgerCacheInterface {
     void
     waitUntilCacheContainsSeq(uint32_t seq) override;
 
-    /**
-     * @brief Save the cache to file
-     * @note This operation takes about 7 seconds and it keeps mtx_ exclusively locked
-     *
-     * @param path The file path to save the cache to
-     * @return An error as a string if any
-     */
     std::expected<void, std::string>
-    saveToFile(std::string const& path) const;
+    saveToFile(std::string const& path) const override;
 
     std::expected<void, std::string>
     loadFromFile(std::string const& path, uint32_t minLatestSequence) override;

src/data/LedgerCacheInterface.hpp

@@ -171,6 +171,16 @@ class LedgerCacheInterface {
     virtual void
     waitUntilCacheContainsSeq(uint32_t seq) = 0;
 
+    /**
+     * @brief Save the cache to file
+     * @note This operation takes about 7 seconds and it keeps a shared lock of mtx_
+     *
+     * @param path The file path to save the cache to
+     * @return An error as a string if any
+     */
+    [[nodiscard]] virtual std::expected<void, std::string>
+    saveToFile(std::string const& path) const = 0;
+
     /**
      * @brief Load the cache from file
      * @note This operation takes about 7 seconds and it keeps mtx_ exclusively locked

src/data/LedgerCacheSaver.cpp

@@ -0,0 +1,70 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2025, the clio developers.
+
+    Permission to use, copy, modify, and distribute this software for any
+    purpose with or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#include "data/LedgerCacheSaver.hpp"
+
+#include "data/LedgerCacheInterface.hpp"
+#include "util/Assert.hpp"
+#include "util/Profiler.hpp"
+#include "util/log/Logger.hpp"
+
+#include <string>
+#include <thread>
+
+namespace data {
+
+LedgerCacheSaver::LedgerCacheSaver(util::config::ClioConfigDefinition const& config, LedgerCacheInterface const& cache)
+    : cacheFilePath_(config.maybeValue<std::string>("cache.file.path")), cache_(cache)
+{
+}
+
+LedgerCacheSaver::~LedgerCacheSaver()
+{
+    waitToFinish();
+}
+
+void
+LedgerCacheSaver::save()
+{
+    ASSERT(not savingThread_.has_value(), "Multiple save() calls are not allowed");
+    savingThread_ = std::thread([this]() {
+        if (not cacheFilePath_.has_value()) {
+            return;
+        }
+
+        LOG(util::LogService::info()) << "Saving ledger cache to " << *cacheFilePath_;
+        if (auto const [success, durationMs] = util::timed([&]() { return cache_.get().saveToFile(*cacheFilePath_); });
+            success.has_value()) {
+            LOG(util::LogService::info()) << "Successfully saved ledger cache in " << durationMs << " ms";
+        } else {
+            LOG(util::LogService::error()) << "Error saving LedgerCache to file";
+        }
+    });
+}
+
+void
+LedgerCacheSaver::waitToFinish()
+{
+    if (savingThread_.has_value() and savingThread_->joinable()) {
+        savingThread_->join();
+    }
+    savingThread_.reset();
+}
+
+}  // namespace data

src/data/LedgerCacheSaver.hpp

@@ -0,0 +1,93 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2025, the clio developers.
+
+    Permission to use, copy, modify, and distribute this software for any
+    purpose with or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#pragma once
+
+#include "data/LedgerCacheInterface.hpp"
+#include "util/config/ConfigDefinition.hpp"
+
+#include <concepts>
+#include <functional>
+#include <optional>
+#include <string>
+#include <thread>
+
+namespace data {
+
+/**
+ * @brief A concept for a class that can save ledger cache asynchronously.
+ *
+ * This concept defines the interface requirements for any type that manages
+ * asynchronous saving of ledger cache to persistent storage.
+ */
+template <typename T>
+concept SomeLedgerCacheSaver = requires(T a) {
+    { a.save() } -> std::same_as<void>;
+    { a.waitToFinish() } -> std::same_as<void>;
+};
+
+/**
+ * @brief Manages asynchronous saving of ledger cache to a file.
+ *
+ * This class provides functionality to save the ledger cache to a file in a separate thread,
+ * allowing the main application to continue without blocking. The file path is configured
+ * through the application's configuration system.
+ */
+class LedgerCacheSaver {
+    std::optional<std::string> cacheFilePath_;
+    std::reference_wrapper<LedgerCacheInterface const> cache_;
+    std::optional<std::thread> savingThread_;
+
+public:
+    /**
+     * @brief Constructs a LedgerCacheSaver instance.
+     *
+     * @param config The configuration object containing the cache file path setting
+     * @param cache Reference to the ledger cache interface to be saved
+     */
+    LedgerCacheSaver(util::config::ClioConfigDefinition const& config, LedgerCacheInterface const& cache);
+
+    /**
+     * @brief Destructor that ensures the saving thread is properly joined.
+     *
+     * Waits for any ongoing save operation to complete before destruction.
+     */
+    ~LedgerCacheSaver();
+
+    /**
+     * @brief Initiates an asynchronous save operation of the ledger cache.
+     *
+     * Spawns a new thread that saves the ledger cache to the configured file path.
+     * If no file path is configured, the operation is skipped. Logs the progress
+     * and result of the save operation.
+     */
+    void
+    save();
+
+    /**
+     * @brief Waits for the saving thread to complete.
+     *
+     * Blocks until the saving operation finishes if a thread is currently active.
+     * Safe to call multiple times or when no save operation is in progress.
+     */
+    void
+    waitToFinish();
+};
+
+}  // namespace data

src/util/StopHelper.cpp

@@ -19,6 +19,8 @@
 
 #include "util/StopHelper.hpp"
 
+#include "util/Spawn.hpp"
+
 #include <boost/asio/spawn.hpp>
 #include <boost/asio/steady_timer.hpp>
 
@@ -37,7 +39,7 @@ void
 StopHelper::asyncWaitForStop(boost::asio::yield_context yield)
 {
     boost::asio::steady_timer timer{yield.get_executor(), std::chrono::steady_clock::duration::max()};
-    onStopReady_.connect([&timer]() { timer.cancel(); });
+    onStopReady_.connect([&]() { util::spawn(yield, [&timer](auto&&) { timer.cancel(); }); });
     boost::system::error_code error;
     if (!*stopped_)
         timer.async_wait(yield[error]);

src/util/StopHelper.hpp

@@ -36,6 +36,16 @@ class StopHelper {
     std::unique_ptr<std::atomic_bool> stopped_ = std::make_unique<std::atomic_bool>(false);
 
 public:
+    StopHelper() = default;
+    ~StopHelper() = default;
+
+    StopHelper(StopHelper&&) = delete;
+    StopHelper&
+    operator=(StopHelper&&) = delete;
+    StopHelper(StopHelper const&) = delete;
+    StopHelper&
+    operator=(StopHelper const&) = delete;
+
     /**
      * @brief Notify that the class is ready to stop.
      */