feat: ETLng monitor (#1898)

For #1594

tests/common/util/MockNetworkValidatedLedgers.hpp

@@ -19,8 +19,9 @@
 
 #pragma once
 
-#include "etl/ETLHelpers.hpp"
+#include "etl/NetworkValidatedLedgersInterface.hpp"
 
+#include <boost/signals2/connection.hpp>
 #include <gmock/gmock.h>
 
 #include <cstdint>
@@ -31,6 +32,12 @@ struct MockNetworkValidatedLedgers : public etl::NetworkValidatedLedgersInterfac
     MOCK_METHOD(void, push, (uint32_t), (override));
     MOCK_METHOD(std::optional<uint32_t>, getMostRecent, (), (override));
     MOCK_METHOD(bool, waitUntilValidatedByNetwork, (uint32_t, std::optional<uint32_t>), (override));
+    MOCK_METHOD(
+        boost::signals2::scoped_connection,
+        subscribe,
+        (etl::NetworkValidatedLedgersInterface::SignalType::slot_type const& subscriber),
+        (override)
+    );
 };
 
 template <template <typename> typename MockType>

tests/common/util/MockOperation.hpp

@@ -48,4 +48,5 @@ struct MockRepeatingOperation {
     MOCK_METHOD(void, requestStop, (), (const));
     MOCK_METHOD(void, wait, (), (const));
     MOCK_METHOD(ValueType, get, (), (const));
+    MOCK_METHOD(void, invoke, (), (const));
 };

tests/common/util/MockStrand.hpp

@@ -41,6 +41,9 @@ struct MockStrand {
     template <typename T>
     using StoppableOperation = MockStoppableOperation<T>;
 
+    template <typename T>
+    using RepeatingOperation = MockRepeatingOperation<T>;
+
     MOCK_METHOD(Operation<std::any> const&, execute, (std::function<std::any()>), (const));
     MOCK_METHOD(
         Operation<std::any> const&,
@@ -60,4 +63,10 @@ struct MockStrand {
         (std::function<std::any(util::async::AnyStopToken)>, std::optional<std::chrono::milliseconds>),
         (const)
     );
+    MOCK_METHOD(
+        RepeatingOperation<std::any> const&,
+        executeRepeatedly,
+        (std::chrono::milliseconds, std::function<std::any()>),
+        (const)
+    );
 };

tests/unit/CMakeLists.txt

@@ -42,6 +42,8 @@ target_sources(
           etlng/SchedulingTests.cpp
           etlng/TaskManagerTests.cpp
           etlng/LoadingTests.cpp
+          etlng/NetworkValidatedLedgersTests.cpp
+          etlng/MonitorTests.cpp
           # Feed
           util/BytesConverterTests.cpp
           feed/BookChangesFeedTests.cpp

tests/unit/LoggerTests.cpp

@@ -215,14 +215,18 @@ TEST_F(LoggerInitTest, LogSizeAndHourRotationCannotBeZero)
         "log_rotation_hour_interval", "log_directory_max_size", "log_rotation_size"
     };
 
-    auto const jsonStr = fmt::format(R"json({{
+    auto const jsonStr = fmt::format(
+        R"json({{
         "{}": 0,
         "{}": 0,
         "{}": 0
-    }})json", keys[0], keys[1], keys[2]);
+    }})json",
+        keys[0],
+        keys[1],
+        keys[2]
+    );
 
-    auto const parsingErrors =
-        config_.parse(ConfigFileJson{boost::json::parse(jsonStr).as_object()});
+    auto const parsingErrors = config_.parse(ConfigFileJson{boost::json::parse(jsonStr).as_object()});
     ASSERT_EQ(parsingErrors->size(), 3);
     for (std::size_t i = 0; i < parsingErrors->size(); ++i) {
         EXPECT_EQ(

tests/unit/etlng/MonitorTests.cpp

@@ -0,0 +1,112 @@
+//------------------------------------------------------------------------------
+/*
+    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/Types.hpp"
+#include "etlng/impl/AmendmentBlockHandler.hpp"
+#include "etlng/impl/Monitor.hpp"
+#include "util/MockBackendTestFixture.hpp"
+#include "util/MockNetworkValidatedLedgers.hpp"
+#include "util/MockPrometheus.hpp"
+#include "util/async/context/BasicExecutionContext.hpp"
+
+#include <boost/signals2/connection.hpp>
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <functional>
+#include <semaphore>
+
+using namespace etlng::impl;
+
+namespace {
+constexpr auto kSTART_SEQ = 123u;
+}  // namespace
+
+struct MonitorTests : util::prometheus::WithPrometheus, MockBackendTest {
+protected:
+    util::async::CoroExecutionContext ctx_;
+    StrictMockNetworkValidatedLedgersPtr ledgers_;
+    testing::StrictMock<testing::MockFunction<void(uint32_t)>> actionMock_;
+
+    etlng::impl::Monitor monitor_ = etlng::impl::Monitor(ctx_, backend_, ledgers_, kSTART_SEQ);
+};
+
+TEST_F(MonitorTests, ConsumesAndNotifiesForAllOutstandingSequencesAtOnce)
+{
+    uint8_t count = 3;
+    LedgerRange range(kSTART_SEQ, kSTART_SEQ + count - 1);
+
+    std::binary_semaphore unblock(0);
+
+    EXPECT_CALL(*ledgers_, subscribe(testing::_));
+    EXPECT_CALL(*backend_, hardFetchLedgerRange(testing::_)).WillOnce(testing::Return(range));
+    EXPECT_CALL(actionMock_, Call).Times(count).WillRepeatedly([&] {
+        if (--count == 0u)
+            unblock.release();
+    });
+
+    auto subscription = monitor_.subscribe(actionMock_.AsStdFunction());
+    monitor_.run(std::chrono::milliseconds{1});
+    unblock.acquire();
+}
+
+TEST_F(MonitorTests, NotifiesForEachSequence)
+{
+    uint8_t count = 3;
+    LedgerRange range(kSTART_SEQ, kSTART_SEQ);
+
+    std::binary_semaphore unblock(0);
+
+    EXPECT_CALL(*ledgers_, subscribe(testing::_));
+    EXPECT_CALL(*backend_, hardFetchLedgerRange(testing::_)).Times(count).WillRepeatedly([&] {
+        auto tmp = range;
+        ++range.maxSequence;
+        return tmp;
+    });
+    EXPECT_CALL(actionMock_, Call).Times(count).WillRepeatedly([&] {
+        if (--count == 0u)
+            unblock.release();
+    });
+
+    auto subscription = monitor_.subscribe(actionMock_.AsStdFunction());
+    monitor_.run(std::chrono::milliseconds{1});
+    unblock.acquire();
+}
+
+TEST_F(MonitorTests, NotifiesWhenForcedByNewSequenceAvailableFromNetwork)
+{
+    LedgerRange range(kSTART_SEQ, kSTART_SEQ);
+    std::binary_semaphore unblock(0);
+    std::function<void(uint32_t)> pusher;
+
+    EXPECT_CALL(*ledgers_, subscribe(testing::_)).WillOnce([&](auto&& subscriber) {
+        pusher = subscriber;
+        return boost::signals2::scoped_connection();  // to keep the compiler happy
+    });
+    EXPECT_CALL(*backend_, hardFetchLedgerRange(testing::_)).WillOnce(testing::Return(range));
+    EXPECT_CALL(actionMock_, Call).WillOnce([&] { unblock.release(); });
+
+    auto subscription = monitor_.subscribe(actionMock_.AsStdFunction());
+    monitor_.run(std::chrono::seconds{10});  // expected to be force-invoked sooner than in 10 sec
+    pusher(kSTART_SEQ);                      // pretend network validated a new ledger
+    unblock.acquire();
+}

tests/unit/etlng/NetworkValidatedLedgersTests.cpp

@@ -0,0 +1,91 @@
+//------------------------------------------------------------------------------
+/*
+    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 "etl/NetworkValidatedLedgers.hpp"
+#include "etl/NetworkValidatedLedgersInterface.hpp"
+#include "etlng/impl/AmendmentBlockHandler.hpp"
+#include "util/LoggerFixtures.hpp"
+#include "util/async/context/BasicExecutionContext.hpp"
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+
+using namespace etlng::impl;
+
+struct NetworkValidatedLedgersTests : NoLoggerFixture {
+protected:
+    util::async::CoroExecutionContext ctx_{2};
+    std::shared_ptr<etl::NetworkValidatedLedgersInterface> ledgers_ =
+        etl::NetworkValidatedLedgers::makeValidatedLedgers();
+};
+
+TEST_F(NetworkValidatedLedgersTests, WaitUntilValidatedByNetworkWithoutTimeout)
+{
+    auto awaitable = ctx_.execute([this] { return ledgers_->waitUntilValidatedByNetwork(123u); });
+
+    ledgers_->push(122u);
+    ledgers_->push(123u);
+
+    EXPECT_TRUE(awaitable.get().value());
+}
+
+TEST_F(NetworkValidatedLedgersTests, WaitUntilValidatedByNetworkWithTimeout)
+{
+    static constexpr auto kTIMEOUT_MILLIS = 10u;
+    auto awaitable = ctx_.execute([this] { return ledgers_->waitUntilValidatedByNetwork(123u, kTIMEOUT_MILLIS); });
+
+    ledgers_->push(122u);
+
+    EXPECT_FALSE(awaitable.get().value());
+}
+
+TEST_F(NetworkValidatedLedgersTests, GetMostRecent)
+{
+    ledgers_->push(122u);
+    ledgers_->push(123u);
+
+    auto awaitable = ctx_.execute([this] { return ledgers_->getMostRecent(); });
+
+    EXPECT_EQ(awaitable.get().value(), 123u);
+
+    ledgers_->push(124u);
+    EXPECT_EQ(ledgers_->getMostRecent(), 124u);
+}
+
+TEST_F(NetworkValidatedLedgersTests, SubscribersGetNotifiedWhileConnectionIsAlive)
+{
+    testing::StrictMock<testing::MockFunction<void(uint32_t)>> actionMock1, actionMock2;
+    EXPECT_CALL(actionMock1, Call).Times(2);
+    EXPECT_CALL(actionMock2, Call).Times(2);
+
+    {
+        auto connection1 = ledgers_->subscribe(actionMock1.AsStdFunction());
+        auto connection2 = ledgers_->subscribe(actionMock2.AsStdFunction());
+
+        ledgers_->push(123u);
+        ledgers_->push(124u);
+    }
+
+    ledgers_->push(125u);
+    ledgers_->push(126u);
+}

tests/unit/util/async/AnyOperationTests.cpp

@@ -128,7 +128,18 @@ TEST_F(AnyOperationTests, RepeatingOpRequestStopCallPropagated)
     repeatingOp.abort();
 }
 
+TEST_F(AnyOperationTests, RepeatingOpInvokeCallPropagated)
+{
+    EXPECT_CALL(mockRepeatingOp, invoke());
+    repeatingOp.invoke();
+}
+
 TEST_F(AnyOperationDeathTest, CallAbortOnNonStoppableOrCancellableOperation)
 {
     EXPECT_DEATH(voidOp.abort(), ".*");
 }
+
+TEST_F(AnyOperationDeathTest, CallInvokeOnNonForceInvocableOperation)
+{
+    EXPECT_DEATH(voidOp.invoke(), ".*");
+}

tests/unit/util/async/AnyStrandTests.cpp

@@ -43,6 +43,9 @@ struct AnyStrandTests : ::testing::Test {
     template <typename T>
     using StoppableOperationType = ::testing::NiceMock<MockStoppableOperation<T>>;
 
+    template <typename T>
+    using RepeatingOperationType = NiceMock<MockRepeatingOperation<T>>;
+
     ::testing::NaggyMock<MockStrand> mockStrand;
     AnyStrand strand{static_cast<MockStrand&>(mockStrand)};
 };
@@ -146,3 +149,15 @@ TEST_F(AnyStrandTests, ExecuteWithTimoutAndStopTokenAndReturnValueThrowsExceptio
         [[maybe_unused]] auto unused = strand.execute([](auto) { return 42; }, std::chrono::milliseconds{1})
     );
 }
+
+TEST_F(AnyStrandTests, RepeatingOperation)
+{
+    auto mockRepeatingOp = RepeatingOperationType<std::any>{};
+    EXPECT_CALL(mockRepeatingOp, wait());
+    EXPECT_CALL(mockStrand, executeRepeatedly(std::chrono::milliseconds{1}, A<std::function<std::any()>>()))
+        .WillOnce([&mockRepeatingOp] -> RepeatingOperationType<std::any> const& { return mockRepeatingOp; });
+
+    auto res = strand.executeRepeatedly(std::chrono::milliseconds{1}, [] -> void { throw 0; });
+    static_assert(std::is_same_v<decltype(res), AnyOperation<void>>);
+    res.wait();
+}

tests/unit/util/async/AsyncExecutionContextTests.cpp

@@ -24,8 +24,10 @@
 
 #include <gtest/gtest.h>
 
+#include <atomic>
 #include <chrono>
 #include <cstddef>
+#include <ranges>
 #include <semaphore>
 #include <stdexcept>
 #include <string>
@@ -220,6 +222,24 @@ TYPED_TEST(ExecutionContextTests, repeatingOperation)
     EXPECT_LE(callCount, expectedActualCount);     // never should be called more times than possible before timeout
 }
 
+TYPED_TEST(ExecutionContextTests, repeatingOperationForceInvoke)
+{
+    std::atomic_size_t callCount = 64uz;
+    std::binary_semaphore unblock(0);
+
+    auto res = this->ctx.executeRepeatedly(std::chrono::seconds{10}, [&] {
+        if (--callCount == 0uz)
+            unblock.release();
+    });
+    for ([[maybe_unused]] auto unused : std::views::iota(0uz, callCount.load()))
+        res.invoke();
+
+    unblock.acquire();
+    res.abort();
+
+    EXPECT_EQ(callCount, 0uz);
+}
+
 TYPED_TEST(ExecutionContextTests, strandMove)
 {
     auto strand = this->ctx.makeStrand();
@@ -273,6 +293,43 @@ TYPED_TEST(ExecutionContextTests, strandWithTimeout)
     EXPECT_EQ(res.get().value(), 42);
 }
 
+TYPED_TEST(ExecutionContextTests, strandedRepeatingOperation)
+{
+    auto strand = this->ctx.makeStrand();
+    auto const repeatDelay = std::chrono::milliseconds{1};
+    auto const timeout = std::chrono::milliseconds{15};
+    auto callCount = 0uz;
+
+    auto res = strand.executeRepeatedly(repeatDelay, [&] { ++callCount; });
+    auto timeSpent = util::timed([timeout] { std::this_thread::sleep_for(timeout); });  // calculate actual time spent
+
+    res.abort();  // outside of the above stopwatch because it blocks and can take arbitrary time
+    auto const expectedPureCalls = timeout.count() / repeatDelay.count();
+    auto const expectedActualCount = timeSpent / repeatDelay.count();
+
+    EXPECT_GE(callCount, expectedPureCalls / 2u);  // expect at least half of the scheduled calls
+    EXPECT_LE(callCount, expectedActualCount);     // never should be called more times than possible before timeout
+}
+
+TYPED_TEST(ExecutionContextTests, strandedRepeatingOperationForceInvoke)
+{
+    auto strand = this->ctx.makeStrand();
+    auto callCount = 64uz;  // does not need to be atomic since we are on a strand
+    std::binary_semaphore unblock(0);
+
+    auto res = strand.executeRepeatedly(std::chrono::seconds{10}, [&] {
+        if (--callCount == 0uz)
+            unblock.release();
+    });
+    for ([[maybe_unused]] auto unused : std::views::iota(0uz, callCount))
+        res.invoke();
+
+    unblock.acquire();
+    res.abort();
+
+    EXPECT_EQ(callCount, 0uz);
+}
+
 TYPED_TEST(AsyncExecutionContextTests, executeAutoAborts)
 {
     auto value = 0;

tests/unit/util/newconfig/ArrayTests.cpp

@@ -126,4 +126,3 @@ TEST(ArrayTest, addNullRequired)
     auto const error = arr.addNull();
     EXPECT_TRUE(error.has_value());
 }
-

tests/unit/util/newconfig/ClioConfigDefinitionTests.cpp

@@ -31,8 +31,8 @@
 #include <boost/json/object.hpp>
 #include <boost/json/parse.hpp>
 #include <boost/json/value.hpp>
-#include <gtest/gtest.h>
 #include <gmock/gmock.h>
+#include <gtest/gtest.h>
 
 #include <algorithm>
 #include <cstdint>
@@ -334,7 +334,8 @@ TEST_F(ClioConfigDefinitionParseArrayTest, emptyArray)
 {
     auto const configJson = boost::json::parse(R"json({
         "array": []
-    })json").as_object();
+    })json")
+                                .as_object();
 
     auto const result = config.parse(ConfigFileJson{configJson});
     EXPECT_FALSE(result.has_value());
@@ -355,20 +356,23 @@ TEST_F(ClioConfigDefinitionParseArrayTest, fullArray)
             {"int": 1, "string": "one"},
             {"int": 2, "string": "two"}
         ]
-    })json").as_object();
+    })json")
+                                .as_object();
 
     auto const result = config.parse(ConfigFileJson{configJson});
     EXPECT_FALSE(result.has_value());
     EXPECT_EQ(config.arraySize("array.[]"), 2);
 }
 
-TEST_F(ClioConfigDefinitionParseArrayTest, onlyRequiredFields) {
+TEST_F(ClioConfigDefinitionParseArrayTest, onlyRequiredFields)
+{
     auto const configJson = boost::json::parse(R"json({
         "array": [
             {"int": 1},
             {"int": 2}
         ]
-    })json").as_object();
+    })json")
+                                .as_object();
 
     auto const configFile = ConfigFileJson{configJson};
     auto const result = config.parse(configFile);
@@ -388,7 +392,8 @@ TEST_F(ClioConfigDefinitionParseArrayTest, someOptionalFieldsMissing)
             {"int": 1, "string": "one"},
             {"int": 2}
         ]
-    })json").as_object();
+    })json")
+                                .as_object();
 
     auto const configFile = ConfigFileJson{configJson};
     auto const result = config.parse(configFile);
@@ -401,13 +406,15 @@ TEST_F(ClioConfigDefinitionParseArrayTest, someOptionalFieldsMissing)
     EXPECT_FALSE(config.getArray("array.[].string").valueAt(1).hasValue());
 }
 
-TEST_F(ClioConfigDefinitionParseArrayTest, optionalFieldMissingAtFirstPosition) {
+TEST_F(ClioConfigDefinitionParseArrayTest, optionalFieldMissingAtFirstPosition)
+{
     auto const configJson = boost::json::parse(R"json({
         "array": [
             {"int": 1},
             {"int": 2, "string": "two"}
         ]
-    })json").as_object();
+    })json")
+                                .as_object();
 
     auto const configFile = ConfigFileJson{configJson};
     auto const result = config.parse(configFile);
@@ -421,13 +428,15 @@ TEST_F(ClioConfigDefinitionParseArrayTest, optionalFieldMissingAtFirstPosition)
     EXPECT_EQ(config.getArray("array.[].string").valueAt(1).asString(), "two");
 }
 
-TEST_F(ClioConfigDefinitionParseArrayTest, missingRequiredFields) {
+TEST_F(ClioConfigDefinitionParseArrayTest, missingRequiredFields)
+{
     auto const configJson = boost::json::parse(R"json({
         "array": [
             {"int": 1},
             {"string": "two"}
         ]
-    })json").as_object();
+    })json")
+                                .as_object();
 
     auto const configFile = ConfigFileJson{configJson};
     auto const result = config.parse(configFile);
@@ -436,13 +445,15 @@ TEST_F(ClioConfigDefinitionParseArrayTest, missingRequiredFields) {
     EXPECT_THAT(result->at(0).error, testing::StartsWith("array.[].int"));
 }
 
-TEST_F(ClioConfigDefinitionParseArrayTest, missingAllRequiredFields) {
+TEST_F(ClioConfigDefinitionParseArrayTest, missingAllRequiredFields)
+{
     auto const configJson = boost::json::parse(R"json({
         "array": [
             {"string": "one"},
             {"string": "two"}
         ]
-    })json").as_object();
+    })json")
+                                .as_object();
 
     auto const configFile = ConfigFileJson{configJson};
     auto const result = config.parse(configFile);

tests/unit/util/newconfig/ConfigFileJsonTests.cpp

@@ -420,7 +420,8 @@ TEST_F(ConfigFileJsonTest, getArrayObjectInArray)
     EXPECT_EQ(std::get<std::string>(strings.at(1).value()), "some string");
 }
 
-TEST_F(ConfigFileJsonTest, getArrayOptionalInArray) {
+TEST_F(ConfigFileJsonTest, getArrayOptionalInArray)
+{
     auto const jsonStr = R"json({
         "array": [
             { "int": 42 },