refactor: Add etl system state command

src/etl/ETLService.cpp

@@ -173,6 +173,7 @@ ETLService::ETLService(
     , state_(std::move(state))
     , startSequence_(config.get().maybeValue<uint32_t>("start_sequence"))
     , finishSequence_(config.get().maybeValue<uint32_t>("finish_sequence"))
+    , writeCommandStrand_(ctx_.makeStrand())
 {
     ASSERT(not state_->isWriting, "ETL should never start in writer mode");
 
@@ -232,6 +233,13 @@ ETLService::stop()
 {
     LOG(log_.info()) << "Stop called";
 
+    systemStateWriteCommandSubscription_.disconnect();
+    auto count = runningWriteCommandHandlers_.load();
+    while (count != 0) {
+        runningWriteCommandHandlers_.wait(count);  // Blocks until value changes
+        count = runningWriteCommandHandlers_.load();
+    }
+
     if (mainLoop_)
         mainLoop_->wait();
     if (taskMan_)
@@ -348,21 +356,40 @@ ETLService::startMonitor(uint32_t seq)
 {
     monitor_ = monitorProvider_->make(ctx_, backend_, ledgers_, seq);
 
+    systemStateWriteCommandSubscription_ =
+        state_->writeCommandSignal.connect([this](SystemState::WriteCommand command) {
+            ++runningWriteCommandHandlers_;
+            writeCommandStrand_.submit([this, command]() {
+                switch (command) {
+                    case etl::SystemState::WriteCommand::StartWriting:
+                        attemptTakeoverWriter();
+                        break;
+                    case etl::SystemState::WriteCommand::StopWriting:
+                        giveUpWriter();
+                        break;
+                }
+                --runningWriteCommandHandlers_;
+                runningWriteCommandHandlers_.notify_one();
+            });
+        });
+
     monitorNewSeqSubscription_ = monitor_->subscribeToNewSequence([this](uint32_t seq) {
         LOG(log_.info()) << "ETLService (via Monitor) got new seq from db: " << seq;
 
-        if (state_->writeConflict) {
-            LOG(log_.info()) << "Got a write conflict; Giving up writer seat immediately";
-            giveUpWriter();
-        }
+        auto const cacheNeedsUpdate = backend_->cache().latestLedgerSequence() < seq;
+        auto const backendRange = backend_->fetchLedgerRange();
+        auto const backendNeedsUpdate = backendRange.has_value() and backendRange->maxSequence < seq;
 
-        if (not state_->isWriting) {
+        if (cacheNeedsUpdate or backendNeedsUpdate) {
             auto const diff = data::synchronousAndRetryOnTimeout([this, seq](auto yield) {
                 return backend_->fetchLedgerDiff(seq, yield);
             });
 
-            cacheUpdater_->update(seq, diff);
-            backend_->updateRange(seq);
+            if (cacheNeedsUpdate)
+                cacheUpdater_->update(seq, diff);
+
+            if (backendNeedsUpdate)
+                backend_->updateRange(seq);
         }
 
         publisher_->publish(seq, {});
@@ -371,7 +398,7 @@ ETLService::startMonitor(uint32_t seq)
     monitorDbStalledSubscription_ = monitor_->subscribeToDbStalled([this]() {
         LOG(log_.warn()) << "ETLService received DbStalled signal from Monitor";
         if (not state_->isStrictReadonly and not state_->isWriting)
-            attemptTakeoverWriter();
+            state_->writeCommandSignal(SystemState::WriteCommand::StartWriting);
     });
 
     monitor_->run();
@@ -404,7 +431,6 @@ ETLService::giveUpWriter()
 {
     ASSERT(not state_->isStrictReadonly, "This should only happen on writer nodes");
     state_->isWriting = false;
-    state_->writeConflict = false;
     taskMan_ = nullptr;
 }
 

src/etl/ETLService.hpp

@@ -52,6 +52,7 @@
 #include "feed/SubscriptionManagerInterface.hpp"
 #include "util/async/AnyExecutionContext.hpp"
 #include "util/async/AnyOperation.hpp"
+#include "util/async/AnyStrand.hpp"
 #include "util/config/ConfigDefinition.hpp"
 #include "util/log/Logger.hpp"
 
@@ -69,12 +70,12 @@
 #include <xrpl/protocol/TxFormats.h>
 #include <xrpl/protocol/TxMeta.h>
 
+#include <atomic>
 #include <cstddef>
 #include <cstdint>
 #include <functional>
 #include <memory>
 #include <optional>
-#include <string>
 
 namespace etl {
 
@@ -117,6 +118,9 @@ class ETLService : public ETLServiceInterface {
 
     boost::signals2::scoped_connection monitorNewSeqSubscription_;
     boost::signals2::scoped_connection monitorDbStalledSubscription_;
+    boost::signals2::scoped_connection systemStateWriteCommandSubscription_;
+    util::async::AnyStrand writeCommandStrand_;
+    std::atomic<size_t> runningWriteCommandHandlers_{0};
 
     std::optional<util::async::AnyOperation<void>> mainLoop_;
 

src/etl/SystemState.hpp

@@ -19,11 +19,13 @@
 
 #pragma once
 
+#include "util/Channel.hpp"
 #include "util/prometheus/Bool.hpp"
 #include "util/prometheus/Label.hpp"
 #include "util/prometheus/Prometheus.hpp"
 
-#include <atomic>
+#include <boost/signals2/signal.hpp>
+#include <boost/signals2/variadic_signal.hpp>
 
 namespace etl {
 
@@ -50,8 +52,24 @@ struct SystemState {
         "Whether the process is writing to the database"
     );
 
-    std::atomic_bool isStopping = false;    /**< @brief Whether the software is stopping. */
-    std::atomic_bool writeConflict = false; /**< @brief Whether a write conflict was detected. */
+    /**
+     * @brief Commands for controlling the ETL writer state.
+     *
+     * These commands are emitted via writeCommandSignal to coordinate writer state transitions across components.
+     */
+    enum class WriteCommand {
+        StartWriting, /**< Request to attempt taking over as the ETL writer */
+        StopWriting   /**< Request to give up the ETL writer role (e.g., due to write conflict) */
+    };
+
+    /**
+     * @brief Signal for coordinating ETL writer state transitions.
+     *
+     * This signal allows components to request changes to the writer state without direct coupling.
+     * - Emitted with StartWriting when database stalls and node should attempt to become writer
+     * - Emitted with StopWriting when write conflicts are detected
+     */
+    boost::signals2::signal<void(WriteCommand)> writeCommandSignal;
 
     /**
      * @brief Whether clio detected an amendment block.

src/etl/impl/LedgerPublisher.hpp

@@ -45,6 +45,7 @@
 #include <xrpl/protocol/Serializer.h>
 
 #include <algorithm>
+#include <atomic>
 #include <chrono>
 #include <cstddef>
 #include <cstdint>
@@ -76,6 +77,8 @@ class LedgerPublisher : public LedgerPublisherInterface {
 
     util::async::AnyStrand publishStrand_;
 
+    std::atomic_bool stop_{false};
+
     std::shared_ptr<BackendInterface> backend_;
     std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions_;
     std::reference_wrapper<SystemState const> state_;  // shared state for ETL
@@ -125,7 +128,7 @@ class LedgerPublisher : public LedgerPublisherInterface {
     {
         LOG(log_.info()) << "Attempting to publish ledger = " << ledgerSequence;
         size_t numAttempts = 0;
-        while (not state_.get().isStopping) {
+        while (not stop_) {
             auto range = backend_->hardFetchLedgerRangeNoThrow();
 
             if (!range || range->maxSequence < ledgerSequence) {
@@ -258,6 +261,18 @@ class LedgerPublisher : public LedgerPublisherInterface {
         return *lastPublishedSequence_.lock();
     }
 
+    /**
+     * @brief Stops publishing
+     *
+     * @note This is a basic implementation to satisfy tests. This will be improved in
+     * https://github.com/XRPLF/clio/issues/2833
+     */
+    void
+    stop()
+    {
+        stop_ = true;
+    }
+
 private:
     void
     setLastClose(std::chrono::time_point<ripple::NetClock> lastCloseTime)

src/etl/impl/Loading.cpp

@@ -75,7 +75,7 @@ Loader::load(model::LedgerData const& data)
                              << "; took " << duration << "ms";
 
             if (not success) {
-                state_->writeConflict = true;
+                state_->writeCommandSignal(SystemState::WriteCommand::StopWriting);
                 LOG(log_.warn()) << "Another node wrote a ledger into the DB - we have a write conflict";
                 return std::unexpected(LoaderError::WriteConflict);
             }