feat: Choose writer by cluster communication

src/app/ClioApplication.cpp

@@ -29,6 +29,8 @@
 #include "etl/ETLService.hpp"
 #include "etl/LoadBalancer.hpp"
 #include "etl/NetworkValidatedLedgers.hpp"
+#include "etl/SystemState.hpp"
+#include "etl/WriterState.hpp"
 #include "feed/SubscriptionManager.hpp"
 #include "migration/MigrationInspectorFactory.hpp"
 #include "rpc/Counters.hpp"
@@ -121,7 +123,11 @@ ClioApplication::run(bool const useNgWebServer)
     // Interface to the database
     auto backend = data::makeBackend(config_, cache);
 
-    cluster::ClusterCommunicationService clusterCommunicationService{backend};
+    auto systemState = etl::SystemState::makeSystemState(config_);
+
+    cluster::ClusterCommunicationService clusterCommunicationService{
+        backend, std::make_unique<etl::WriterState>(systemState)
+    };
     clusterCommunicationService.run();
 
     auto const amendmentCenter = std::make_shared<data::AmendmentCenter const>(backend);
@@ -151,7 +157,9 @@ ClioApplication::run(bool const useNgWebServer)
     );
 
     // ETL is responsible for writing and publishing to streams. In read-only mode, ETL only publishes
-    auto etl = etl::ETLService::makeETLService(config_, ctx, backend, subscriptions, balancer, ledgers);
+    auto etl = etl::ETLService::makeETLService(
+        config_, std::move(systemState), ctx, backend, subscriptions, balancer, ledgers
+    );
 
     auto workQueue = rpc::WorkQueue::makeWorkQueue(config_);
     auto counters = rpc::Counters::makeCounters(workQueue);
@@ -197,7 +205,16 @@ ClioApplication::run(bool const useNgWebServer)
         }
 
         appStopper_.setOnStop(
-            Stopper::makeOnStopCallback(httpServer.value(), *balancer, *etl, *subscriptions, *backend, cacheSaver, ioc)
+            Stopper::makeOnStopCallback(
+                httpServer.value(),
+                *balancer,
+                *etl,
+                *subscriptions,
+                *backend,
+                cacheSaver,
+                clusterCommunicationService,
+                ioc
+            )
         );
 
         // Blocks until stopped.
@@ -213,7 +230,9 @@ ClioApplication::run(bool const useNgWebServer)
 
     auto const httpServer = web::makeHttpServer(config_, ioc, dosGuard, handler, cache);
     appStopper_.setOnStop(
-        Stopper::makeOnStopCallback(*httpServer, *balancer, *etl, *subscriptions, *backend, cacheSaver, ioc)
+        Stopper::makeOnStopCallback(
+            *httpServer, *balancer, *etl, *subscriptions, *backend, cacheSaver, clusterCommunicationService, ioc
+        )
     );
 
     // Blocks until stopped.

src/app/Stopper.hpp

@@ -19,6 +19,7 @@
 
 #pragma once
 
+#include "cluster/Concepts.hpp"
 #include "data/BackendInterface.hpp"
 #include "data/LedgerCacheSaver.hpp"
 #include "etl/ETLServiceInterface.hpp"
@@ -82,10 +83,14 @@ class Stopper {
      * @param subscriptions The subscription manager to stop.
      * @param backend The backend to stop.
      * @param cacheSaver The ledger cache saver
+     * @param clusterCommunicationService The cluster communication service to stop.
      * @param ioc The io_context to stop.
      * @return The callback to be called on application stop.
      */
-    template <web::SomeServer ServerType, data::SomeLedgerCacheSaver LedgerCacheSaverType>
+    template <
+        web::SomeServer ServerType,
+        data::SomeLedgerCacheSaver LedgerCacheSaverType,
+        cluster::SomeClusterCommunicationService ClusterCommunicationServiceType>
     static std::function<void(boost::asio::yield_context)>
     makeOnStopCallback(
         ServerType& server,
@@ -94,6 +99,7 @@ class Stopper {
         feed::SubscriptionManagerInterface& subscriptions,
         data::BackendInterface& backend,
         LedgerCacheSaverType& cacheSaver,
+        ClusterCommunicationServiceType& clusterCommunicationService,
         boost::asio::io_context& ioc
     )
     {
@@ -111,6 +117,8 @@ class Stopper {
             });
             coroutineGroup.asyncWait(yield);
 
+            clusterCommunicationService.stop();
+
             etl.stop();
             LOG(util::LogService::info()) << "ETL stopped";
 

src/cluster/Backend.cpp

@@ -0,0 +1,141 @@
+//------------------------------------------------------------------------------
+/*
+    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 "cluster/Backend.hpp"
+
+#include "cluster/ClioNode.hpp"
+#include "data/BackendInterface.hpp"
+#include "etl/WriterState.hpp"
+
+#include <boost/asio/bind_cancellation_slot.hpp>
+#include <boost/asio/cancellation_type.hpp>
+#include <boost/asio/error.hpp>
+#include <boost/asio/execution_context.hpp>
+#include <boost/asio/executor.hpp>
+#include <boost/asio/spawn.hpp>
+#include <boost/asio/steady_timer.hpp>
+#include <boost/asio/thread_pool.hpp>
+#include <boost/asio/use_future.hpp>
+#include <boost/json/parse.hpp>
+#include <boost/json/serialize.hpp>
+#include <boost/json/value.hpp>
+#include <boost/json/value_from.hpp>
+#include <boost/json/value_to.hpp>
+#include <boost/uuid/random_generator.hpp>
+#include <boost/uuid/uuid.hpp>
+#include <fmt/format.h>
+
+#include <chrono>
+#include <memory>
+#include <utility>
+#include <vector>
+
+namespace cluster {
+
+Backend::Backend(
+    boost::asio::thread_pool& ctx,
+    std::shared_ptr<data::BackendInterface> backend,
+    std::unique_ptr<etl::WriterStateInterface const> writerState,
+    std::chrono::steady_clock::duration readInterval,
+    std::chrono::steady_clock::duration writeInterval
+)
+    : backend_(std::move(backend))
+    , writerState_(std::move(writerState))
+    , readerTask_(readInterval, ctx)
+    , writerTask_(writeInterval, ctx)
+    , selfUuid_(std::make_shared<boost::uuids::uuid>(boost::uuids::random_generator{}()))
+{
+}
+
+void
+Backend::run()
+{
+    readerTask_.run([this](boost::asio::yield_context yield) {
+        auto clusterData = doRead(yield);
+        onNewState_(selfUuid_, std::make_shared<ClusterData>(std::move(clusterData)));
+    });
+
+    writerTask_.run([this]() { doWrite(); });
+}
+
+Backend::~Backend()
+{
+    stop();
+}
+
+void
+Backend::stop()
+{
+    readerTask_.stop();
+    writerTask_.stop();
+}
+
+ClioNode::CUuid
+Backend::selfId() const
+{
+    return selfUuid_;
+}
+
+Backend::ClusterData
+Backend::doRead(boost::asio::yield_context yield)
+{
+    BackendInterface::ClioNodesDataFetchResult expectedResult;
+    try {
+        expectedResult = backend_->fetchClioNodesData(yield);
+    } catch (...) {
+        expectedResult = std::unexpected{"Failed to fetch Clio nodes data"};
+    }
+
+    if (!expectedResult.has_value()) {
+        return std::unexpected{std::move(expectedResult).error()};
+    }
+
+    std::vector<ClioNode> otherNodesData;
+    for (auto const& [uuid, nodeDataStr] : expectedResult.value()) {
+        if (uuid == *selfUuid_) {
+            continue;
+        }
+
+        boost::system::error_code errorCode;
+        auto const json = boost::json::parse(nodeDataStr, errorCode);
+        if (errorCode.failed()) {
+            return std::unexpected{fmt::format("Error parsing json from DB: {}", nodeDataStr)};
+        }
+
+        auto expectedNodeData = boost::json::try_value_to<ClioNode>(json);
+        if (expectedNodeData.has_error()) {
+            return std::unexpected{fmt::format("Error converting json to ClioNode: {}", nodeDataStr)};
+        }
+        *expectedNodeData->uuid = uuid;
+        otherNodesData.push_back(std::move(expectedNodeData).value());
+    }
+    otherNodesData.push_back(ClioNode::from(selfUuid_, *writerState_));
+    return otherNodesData;
+}
+
+void
+Backend::doWrite()
+{
+    auto const selfData = ClioNode::from(selfUuid_, *writerState_);
+    boost::json::value jsonValue{};
+    boost::json::value_from(selfData, jsonValue);
+    backend_->writeNodeMessage(*selfData.uuid, boost::json::serialize(jsonValue.as_object()));
+}
+
+}  // namespace cluster

src/cluster/Backend.hpp

@@ -0,0 +1,147 @@
+//------------------------------------------------------------------------------
+/*
+    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 "cluster/ClioNode.hpp"
+#include "cluster/impl/RepeatedTask.hpp"
+#include "data/BackendInterface.hpp"
+#include "etl/WriterState.hpp"
+#include "util/log/Logger.hpp"
+
+#include <boost/asio/any_io_executor.hpp>
+#include <boost/asio/cancellation_signal.hpp>
+#include <boost/asio/execution_context.hpp>
+#include <boost/asio/executor.hpp>
+#include <boost/asio/spawn.hpp>
+#include <boost/asio/strand.hpp>
+#include <boost/asio/thread_pool.hpp>
+#include <boost/signals2/connection.hpp>
+#include <boost/signals2/signal.hpp>
+#include <boost/signals2/variadic_signal.hpp>
+#include <boost/uuid/uuid.hpp>
+
+#include <chrono>
+#include <concepts>
+#include <memory>
+#include <string>
+#include <vector>
+
+namespace cluster {
+
+/**
+ * @brief Backend communication handler for cluster state synchronization.
+ *
+ * This class manages reading and writing cluster state information to/from the backend database.
+ * It periodically reads the state of other nodes in the cluster and writes the current node's state,
+ * enabling cluster-wide coordination and awareness.
+ */
+class Backend {
+public:
+    /** @brief Type representing cluster data result - either a vector of nodes or an error message */
+    using ClusterData = std::expected<std::vector<ClioNode>, std::string>;
+
+private:
+    util::Logger log_{"ClusterCommunication"};
+
+    std::shared_ptr<data::BackendInterface> backend_;
+    std::unique_ptr<etl::WriterStateInterface const> writerState_;
+
+    impl::RepeatedTask<boost::asio::thread_pool> readerTask_;
+    impl::RepeatedTask<boost::asio::thread_pool> writerTask_;
+
+    ClioNode::Uuid selfUuid_;
+
+    boost::signals2::signal<void(ClioNode::CUuid, std::shared_ptr<ClusterData const>)> onNewState_;
+
+public:
+    /**
+     * @brief Construct a Backend communication handler.
+     *
+     * @param ctx The execution context for asynchronous operations
+     * @param backend Interface to the backend database
+     * @param writerState State indicating whether this node is writing to the database
+     * @param readInterval How often to read cluster state from the backend
+     * @param writeInterval How often to write this node's state to the backend
+     */
+    Backend(
+        boost::asio::thread_pool& ctx,
+        std::shared_ptr<data::BackendInterface> backend,
+        std::unique_ptr<etl::WriterStateInterface const> writerState,
+        std::chrono::steady_clock::duration readInterval,
+        std::chrono::steady_clock::duration writeInterval
+    );
+
+    ~Backend();
+
+    Backend(Backend&&) = delete;
+    Backend&
+    operator=(Backend&&) = delete;
+    Backend(Backend const&) = delete;
+    Backend&
+    operator=(Backend const&) = delete;
+
+    /**
+     * @brief Start the backend read and write tasks.
+     *
+     * Begins periodic reading of cluster state from the backend and writing of this node's state.
+     */
+    void
+    run();
+
+    /**
+     * @brief Stop the backend read and write tasks.
+     *
+     * Stops all periodic tasks and waits for them to complete.
+     */
+    void
+    stop();
+
+    /**
+     * @brief Subscribe to new cluster state notifications.
+     *
+     * @tparam S Callable type accepting (ClioNode::cUUID, ClusterData)
+     * @param s Subscriber callback to be invoked when new cluster state is available
+     * @return A connection object that can be used to unsubscribe
+     */
+    template <typename S>
+        requires std::invocable<S, ClioNode::CUuid, std::shared_ptr<ClusterData const>>
+    boost::signals2::connection
+    subscribeToNewState(S&& s)
+    {
+        return onNewState_.connect(s);
+    }
+
+    /**
+     * @brief Get the UUID of this node in the cluster.
+     *
+     * @return The UUID of this node.
+     */
+    ClioNode::CUuid
+    selfId() const;
+
+private:
+    ClusterData
+    doRead(boost::asio::yield_context yield);
+
+    void
+    doWrite();
+};
+
+}  // namespace cluster