chore: Add counter for total messages waiting to be sent (#1691)

Fixes #1507

.github/workflows/nightly.yml

@@ -13,12 +13,15 @@ jobs:
         include:
           - os: macos14
             build_type: Release
+            static: false
           - os: heavy
             build_type: Release
+            static: true
             container:
               image: rippleci/clio_ci:latest
           - os: heavy
             build_type: Debug
+            static: true
             container:
               image: rippleci/clio_ci:latest
     runs-on: [self-hosted, "${{ matrix.os }}"]
@@ -50,6 +53,7 @@ jobs:
           conan_profile: ${{ steps.conan.outputs.conan_profile }}
           conan_cache_hit: ${{ steps.restore_cache.outputs.conan_cache_hit }}
           build_type: ${{ matrix.build_type }}
+          static: ${{ matrix.static }}
 
       - name: Build Clio
         uses: ./.github/actions/build_clio

src/etl/LoadBalancer.cpp

@@ -109,10 +109,13 @@ LoadBalancer::LoadBalancer(
             validatedLedgers,
             forwardingTimeout,
             [this]() {
-                if (not hasForwardingSource_)
+                if (not hasForwardingSource_.lock().get())
+                    chooseForwardingSource();
+            },
+            [this](bool wasForwarding) {
+                if (wasForwarding)
                     chooseForwardingSource();
             },
-            [this]() { chooseForwardingSource(); },
             [this]() {
                 if (forwardingCache_.has_value())
                     forwardingCache_->invalidate();
@@ -314,11 +317,13 @@ LoadBalancer::getETLState() noexcept
 void
 LoadBalancer::chooseForwardingSource()
 {
-    hasForwardingSource_ = false;
+    LOG(log_.info()) << "Choosing a new source to forward subscriptions";
+    auto hasForwardingSourceLock = hasForwardingSource_.lock();
+    hasForwardingSourceLock.get() = false;
     for (auto& source : sources_) {
-        if (not hasForwardingSource_ and source->isConnected()) {
+        if (not hasForwardingSourceLock.get() and source->isConnected()) {
             source->setForwarding(true);
-            hasForwardingSource_ = true;
+            hasForwardingSourceLock.get() = true;
         } else {
             source->setForwarding(false);
         }

src/etl/LoadBalancer.hpp

@@ -25,6 +25,7 @@
 #include "etl/Source.hpp"
 #include "etl/impl/ForwardingCache.hpp"
 #include "feed/SubscriptionManagerInterface.hpp"
+#include "util/Mutex.hpp"
 #include "util/config/Config.hpp"
 #include "util/log/Logger.hpp"
 
@@ -38,7 +39,6 @@
 #include <org/xrpl/rpc/v1/ledger.pb.h>
 #include <ripple/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
 
-#include <atomic>
 #include <chrono>
 #include <cstdint>
 #include <memory>
@@ -74,7 +74,10 @@ private:
     std::optional<ETLState> etlState_;
     std::uint32_t downloadRanges_ =
         DEFAULT_DOWNLOAD_RANGES; /*< The number of markers to use when downloading initial ledger */
-    std::atomic_bool hasForwardingSource_{false};
+
+    // Using mutext instead of atomic_bool because choosing a new source to
+    // forward messages should be done with a mutual exclusion otherwise there will be a race condition
+    util::Mutex<bool> hasForwardingSource_{false};
 
 public:
     /**

src/etl/Source.hpp

@@ -51,7 +51,7 @@ namespace etl {
 class SourceBase {
 public:
     using OnConnectHook = std::function<void()>;
-    using OnDisconnectHook = std::function<void()>;
+    using OnDisconnectHook = std::function<void(bool)>;
     using OnLedgerClosedHook = std::function<void()>;
 
     virtual ~SourceBase() = default;

src/etl/impl/GrpcSource.cpp

@@ -46,7 +46,7 @@
 namespace etl::impl {
 
 GrpcSource::GrpcSource(std::string const& ip, std::string const& grpcPort, std::shared_ptr<BackendInterface> backend)
-    : log_(fmt::format("ETL_Grpc[{}:{}]", ip, grpcPort)), backend_(std::move(backend))
+    : log_(fmt::format("GrpcSource[{}:{}]", ip, grpcPort)), backend_(std::move(backend))
 {
     try {
         boost::asio::ip::tcp::endpoint const endpoint{boost::asio::ip::make_address(ip), std::stoi(grpcPort)};

src/etl/impl/SubscriptionSource.cpp

@@ -24,6 +24,8 @@
 #include "rpc/JS.hpp"
 #include "util/Retry.hpp"
 #include "util/log/Logger.hpp"
+#include "util/prometheus/Label.hpp"
+#include "util/prometheus/Prometheus.hpp"
 #include "util/requests/Types.hpp"
 
 #include <boost/algorithm/string/classification.hpp>
@@ -66,22 +68,28 @@ SubscriptionSource::SubscriptionSource(
     OnConnectHook onConnect,
     OnDisconnectHook onDisconnect,
     OnLedgerClosedHook onLedgerClosed,
-    std::chrono::steady_clock::duration const connectionTimeout,
+    std::chrono::steady_clock::duration const wsTimeout,
     std::chrono::steady_clock::duration const retryDelay
 )
-    : log_(fmt::format("GrpcSource[{}:{}]", ip, wsPort))
+    : log_(fmt::format("SubscriptionSource[{}:{}]", ip, wsPort))
     , wsConnectionBuilder_(ip, wsPort)
     , validatedLedgers_(std::move(validatedLedgers))
     , subscriptions_(std::move(subscriptions))
     , strand_(boost::asio::make_strand(ioContext))
+    , wsTimeout_(wsTimeout)
     , retry_(util::makeRetryExponentialBackoff(retryDelay, RETRY_MAX_DELAY, strand_))
     , onConnect_(std::move(onConnect))
     , onDisconnect_(std::move(onDisconnect))
     , onLedgerClosed_(std::move(onLedgerClosed))
+    , lastMessageTimeSecondsSinceEpoch_(PrometheusService::gaugeInt(
+          "subscription_source_last_message_time",
+          util::prometheus::Labels({{"source", fmt::format("{}:{}", ip, wsPort)}}),
+          "Seconds since epoch of the last message received from rippled subscription streams"
+      ))
 {
     wsConnectionBuilder_.addHeader({boost::beast::http::field::user_agent, "clio-client"})
         .addHeader({"X-User", "clio-client"})
-        .setConnectionTimeout(connectionTimeout);
+        .setConnectionTimeout(wsTimeout_);
 }
 
 SubscriptionSource::~SubscriptionSource()
@@ -133,6 +141,7 @@ void
 SubscriptionSource::setForwarding(bool isForwarding)
 {
     isForwarding_ = isForwarding;
+    LOG(log_.info()) << "Forwarding set to " << isForwarding_;
 }
 
 std::chrono::steady_clock::time_point
@@ -166,20 +175,22 @@ SubscriptionSource::subscribe()
             }
 
             wsConnection_ = std::move(connection).value();
-            isConnected_ = true;
-            onConnect_();
 
             auto const& subscribeCommand = getSubscribeCommandJson();
-            auto const writeErrorOpt = wsConnection_->write(subscribeCommand, yield);
+            auto const writeErrorOpt = wsConnection_->write(subscribeCommand, yield, wsTimeout_);
             if (writeErrorOpt) {
                 handleError(writeErrorOpt.value(), yield);
                 return;
             }
 
+            isConnected_ = true;
+            LOG(log_.info()) << "Connected";
+            onConnect_();
+
             retry_.reset();
 
             while (!stop_) {
-                auto const message = wsConnection_->read(yield);
+                auto const message = wsConnection_->read(yield, wsTimeout_);
                 if (not message) {
                     handleError(message.error(), yield);
                     return;
@@ -224,10 +235,11 @@ SubscriptionSource::handleMessage(std::string const& message)
                 auto validatedLedgers = boost::json::value_to<std::string>(result.at(JS(validated_ledgers)));
                 setValidatedRange(std::move(validatedLedgers));
             }
-            LOG(log_.info()) << "Received a message on ledger subscription stream. Message : " << object;
+            LOG(log_.debug()) << "Received a message on ledger subscription stream. Message: " << object;
 
         } else if (object.contains(JS(type)) && object.at(JS(type)) == JS_LedgerClosed) {
-            LOG(log_.info()) << "Received a message on ledger subscription stream. Message : " << object;
+            LOG(log_.debug()) << "Received a message of type 'ledgerClosed' on ledger subscription stream. Message: "
+                              << object;
             if (object.contains(JS(ledger_index))) {
                 ledgerIndex = object.at(JS(ledger_index)).as_int64();
             }
@@ -245,10 +257,13 @@ SubscriptionSource::handleMessage(std::string const& message)
                 // 2 - Validated transaction
                 // Only forward proposed transaction, validated transactions are sent by Clio itself
                 if (object.contains(JS(transaction)) and !object.contains(JS(meta))) {
+                    LOG(log_.debug()) << "Forwarding proposed transaction: " << object;
                     subscriptions_->forwardProposedTransaction(object);
                 } else if (object.contains(JS(type)) && object.at(JS(type)) == JS_ValidationReceived) {
+                    LOG(log_.debug()) << "Forwarding validation: " << object;
                     subscriptions_->forwardValidation(object);
                 } else if (object.contains(JS(type)) && object.at(JS(type)) == JS_ManifestReceived) {
+                    LOG(log_.debug()) << "Forwarding manifest: " << object;
                     subscriptions_->forwardManifest(object);
                 }
             }
@@ -261,7 +276,7 @@ SubscriptionSource::handleMessage(std::string const& message)
 
         return std::nullopt;
     } catch (std::exception const& e) {
-        LOG(log_.error()) << "Exception in handleMessage : " << e.what();
+        LOG(log_.error()) << "Exception in handleMessage: " << e.what();
         return util::requests::RequestError{fmt::format("Error handling message: {}", e.what())};
     }
 }
@@ -270,16 +285,14 @@ void
 SubscriptionSource::handleError(util::requests::RequestError const& error, boost::asio::yield_context yield)
 {
     isConnected_ = false;
-    isForwarding_ = false;
+    bool const wasForwarding = isForwarding_.exchange(false);
     if (not stop_) {
-        onDisconnect_();
+        LOG(log_.info()) << "Disconnected";
+        onDisconnect_(wasForwarding);
     }
 
     if (wsConnection_ != nullptr) {
-        auto const err = wsConnection_->close(yield);
-        if (err) {
-            LOG(log_.error()) << "Error closing websocket connection: " << err->message();
-        }
+        wsConnection_->close(yield);
         wsConnection_.reset();
     }
 
@@ -306,7 +319,11 @@ SubscriptionSource::logError(util::requests::RequestError const& error) const
 void
 SubscriptionSource::setLastMessageTime()
 {
-    lastMessageTime_.lock().get() = std::chrono::steady_clock::now();
+    lastMessageTimeSecondsSinceEpoch_.get().set(
+        std::chrono::duration_cast<std::chrono::seconds>(std::chrono::system_clock::now().time_since_epoch()).count()
+    );
+    auto lock = lastMessageTime_.lock();
+    lock.get() = std::chrono::steady_clock::now();
 }
 
 void

src/etl/impl/SubscriptionSource.hpp

@@ -19,12 +19,13 @@
 
 #pragma once
 
-#include "etl/ETLHelpers.hpp"
+#include "etl/NetworkValidatedLedgersInterface.hpp"
 #include "etl/Source.hpp"
 #include "feed/SubscriptionManagerInterface.hpp"
 #include "util/Mutex.hpp"
 #include "util/Retry.hpp"
 #include "util/log/Logger.hpp"
+#include "util/prometheus/Gauge.hpp"
 #include "util/requests/Types.hpp"
 #include "util/requests/WsConnection.hpp"
 
@@ -37,6 +38,7 @@
 #include <atomic>
 #include <chrono>
 #include <cstdint>
+#include <functional>
 #include <future>
 #include <memory>
 #include <optional>
@@ -71,6 +73,8 @@ private:
 
     boost::asio::strand<boost::asio::io_context::executor_type> strand_;
 
+    std::chrono::steady_clock::duration wsTimeout_;
+
     util::Retry retry_;
 
     OnConnectHook onConnect_;
@@ -83,9 +87,11 @@ private:
 
     util::Mutex<std::chrono::steady_clock::time_point> lastMessageTime_;
 
+    std::reference_wrapper<util::prometheus::GaugeInt> lastMessageTimeSecondsSinceEpoch_;
+
     std::future<void> runFuture_;
 
-    static constexpr std::chrono::seconds CONNECTION_TIMEOUT{30};
+    static constexpr std::chrono::seconds WS_TIMEOUT{30};
     static constexpr std::chrono::seconds RETRY_MAX_DELAY{30};
     static constexpr std::chrono::seconds RETRY_DELAY{1};
 
@@ -103,7 +109,7 @@ public:
      * @param onNewLedger The onNewLedger hook. Called when a new ledger is received
      * @param onLedgerClosed The onLedgerClosed hook. Called when the ledger is closed but only if the source is
      * forwarding
-     * @param connectionTimeout The connection timeout. Defaults to 30 seconds
+     * @param wsTimeout A timeout for websocket operations. Defaults to 30 seconds
      * @param retryDelay The retry delay. Defaults to 1 second
      */
     SubscriptionSource(
@@ -115,7 +121,7 @@ public:
         OnConnectHook onConnect,
         OnDisconnectHook onDisconnect,
         OnLedgerClosedHook onLedgerClosed,
-        std::chrono::steady_clock::duration const connectionTimeout = CONNECTION_TIMEOUT,
+        std::chrono::steady_clock::duration const wsTimeout = WS_TIMEOUT,
         std::chrono::steady_clock::duration const retryDelay = RETRY_DELAY
     );
 

src/feed/impl/ProposedTransactionFeed.cpp

@@ -104,13 +104,17 @@ ProposedTransactionFeed::pub(boost::json::object const& receivedTxJson)
     boost::asio::post(strand_, [this, pubMsg = std::move(pubMsg), affectedAccounts = std::move(affectedAccounts)]() {
         notified_.clear();
         signal_.emit(pubMsg);
+
         // Prevent the same connection from receiving the same message twice if it is subscribed to multiple accounts
         // However, if the same connection subscribe both stream and account, it will still receive the message twice.
         // notified_ can be cleared before signal_ emit to improve this, but let's keep it as is for now, since rippled
         // acts like this.
         notified_.clear();
+
         for (auto const& account : affectedAccounts)
             accountSignal_.emit(account, pubMsg);
+
+        ++pubCount_.get();
     });
 }
 

src/feed/impl/ProposedTransactionFeed.hpp

@@ -24,6 +24,7 @@
 #include "feed/impl/TrackableSignalMap.hpp"
 #include "feed/impl/Util.hpp"
 #include "util/log/Logger.hpp"
+#include "util/prometheus/Counter.hpp"
 #include "util/prometheus/Gauge.hpp"
 
 #include <boost/asio/io_context.hpp>
@@ -54,6 +55,7 @@ class ProposedTransactionFeed {
     boost::asio::strand<boost::asio::io_context::executor_type> strand_;
     std::reference_wrapper<util::prometheus::GaugeInt> subAllCount_;
     std::reference_wrapper<util::prometheus::GaugeInt> subAccountCount_;
+    std::reference_wrapper<util::prometheus::CounterInt> pubCount_;
 
     TrackableSignalMap<ripple::AccountID, Subscriber, std::shared_ptr<std::string>> accountSignal_;
     TrackableSignal<Subscriber, std::shared_ptr<std::string>> signal_;
@@ -67,7 +69,7 @@ public:
         : strand_(boost::asio::make_strand(ioContext))
         , subAllCount_(getSubscriptionsGaugeInt("tx_proposed"))
         , subAccountCount_(getSubscriptionsGaugeInt("account_proposed"))
-
+        , pubCount_(getPublishedMessagesCounterInt("tx_proposed"))
     {
     }
 

src/feed/impl/SingleFeedBase.cpp

@@ -36,7 +36,10 @@
 namespace feed::impl {
 
 SingleFeedBase::SingleFeedBase(boost::asio::io_context& ioContext, std::string const& name)
-    : strand_(boost::asio::make_strand(ioContext)), subCount_(getSubscriptionsGaugeInt(name)), name_(name)
+    : strand_(boost::asio::make_strand(ioContext))
+    , subCount_(getSubscriptionsGaugeInt(name))
+    , pubCount_(getPublishedMessagesCounterInt(name))
+    , name_(name)
 {
 }
 
@@ -70,6 +73,7 @@ SingleFeedBase::pub(std::string msg) const
     boost::asio::post(strand_, [this, msg = std::move(msg)]() mutable {
         auto const msgPtr = std::make_shared<std::string>(std::move(msg));
         signal_.emit(msgPtr);
+        ++pubCount_.get();
     });
 }
 

src/feed/impl/SingleFeedBase.hpp

@@ -22,6 +22,7 @@
 #include "feed/Types.hpp"
 #include "feed/impl/TrackableSignal.hpp"
 #include "util/log/Logger.hpp"
+#include "util/prometheus/Counter.hpp"
 #include "util/prometheus/Gauge.hpp"
 
 #include <boost/asio/io_context.hpp>
@@ -40,6 +41,7 @@ namespace feed::impl {
 class SingleFeedBase {
     boost::asio::strand<boost::asio::io_context::executor_type> strand_;
     std::reference_wrapper<util::prometheus::GaugeInt> subCount_;
+    std::reference_wrapper<util::prometheus::CounterInt> pubCount_;
     TrackableSignal<Subscriber, std::shared_ptr<std::string> const&> signal_;
     util::Logger logger_{"Subscriptions"};
     std::string name_;

src/feed/impl/TransactionFeed.cpp

@@ -284,23 +284,29 @@ TransactionFeed::pub(
          affectedBooks = std::move(affectedBooks)]() {
             notified_.clear();
             signal_.emit(allVersionsMsgs);
+
             // clear the notified set. If the same connection subscribes both transactions + proposed_transactions,
             // rippled SENDS the same message twice
             notified_.clear();
             txProposedsignal_.emit(allVersionsMsgs);
             notified_.clear();
+
             // check duplicate for account and proposed_account, this prevents sending the same message multiple times
             // if it affects multiple accounts watched by the same connection
             for (auto const& account : affectedAccounts) {
                 accountSignal_.emit(account, allVersionsMsgs);
                 accountProposedSignal_.emit(account, allVersionsMsgs);
             }
+
             notified_.clear();
+
             // check duplicate for books, this prevents sending the same message multiple times if it affects multiple
             // books watched by the same connection
             for (auto const& book : affectedBooks) {
                 bookSignal_.emit(book, allVersionsMsgs);
             }
+
+            ++pubCount_.get();
         }
     );
 }

src/feed/impl/TransactionFeed.hpp

@@ -26,6 +26,7 @@
 #include "feed/impl/TrackableSignalMap.hpp"
 #include "feed/impl/Util.hpp"
 #include "util/log/Logger.hpp"
+#include "util/prometheus/Counter.hpp"
 #include "util/prometheus/Gauge.hpp"
 
 #include <boost/asio/io_context.hpp>
@@ -67,6 +68,7 @@ class TransactionFeed {
     std::reference_wrapper<util::prometheus::GaugeInt> subAllCount_;
     std::reference_wrapper<util::prometheus::GaugeInt> subAccountCount_;
     std::reference_wrapper<util::prometheus::GaugeInt> subBookCount_;
+    std::reference_wrapper<util::prometheus::CounterInt> pubCount_;
 
     TrackableSignalMap<ripple::AccountID, Subscriber, AllVersionTransactionsType const&> accountSignal_;
     TrackableSignalMap<ripple::Book, Subscriber, AllVersionTransactionsType const&> bookSignal_;
@@ -89,6 +91,7 @@ public:
         , subAllCount_(getSubscriptionsGaugeInt("tx"))
         , subAccountCount_(getSubscriptionsGaugeInt("account"))
         , subBookCount_(getSubscriptionsGaugeInt("book"))
+        , pubCount_(getPublishedMessagesCounterInt("tx"))
     {
     }
 

src/feed/impl/Util.hpp

@@ -19,6 +19,7 @@
 
 #pragma once
 
+#include "util/prometheus/Counter.hpp"
 #include "util/prometheus/Gauge.hpp"
 #include "util/prometheus/Label.hpp"
 #include "util/prometheus/Prometheus.hpp"
@@ -38,4 +39,14 @@ getSubscriptionsGaugeInt(std::string const& counterName)
         fmt::format("Current subscribers number on the {} stream", counterName)
     );
 }
+
+inline util::prometheus::CounterInt&
+getPublishedMessagesCounterInt(std::string const& counterName)
+{
+    return PrometheusService::counterInt(
+        "subscriptions_published_count",
+        util::prometheus::Labels({util::prometheus::Label{"stream", counterName}}),
+        fmt::format("Total published messages on the {} stream", counterName)
+    );
+}
 }  // namespace feed::impl

src/main/CMakeLists.txt

@@ -11,18 +11,16 @@ target_sources(clio_server PRIVATE Main.cpp)
 target_link_libraries(clio_server PRIVATE clio)
 
 if (static)
-  target_link_options(clio_server PRIVATE -static)
-
-  if (is_gcc AND NOT san)
+  if (san)
+    message(FATAL_ERROR "Static linkage not allowed when using sanitizers")
+  elseif (is_appleclang)
+    message(FATAL_ERROR "Static linkage not supported on AppleClang")
+  else ()
     target_link_options(
-      # For now let's assume that we only using libstdc++ under gcc.
+      # Note: -static-libstdc++ can statically link both libstdc++ and libc++
       clio_server PRIVATE -static-libstdc++ -static-libgcc
     )
   endif ()
-
-  if (is_appleclang)
-    message(FATAL_ERROR "Static linkage not supported on AppleClang")
-  endif ()
 endif ()
 
 set_target_properties(clio_server PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR})

src/rpc/RPCEngine.hpp

@@ -22,6 +22,7 @@
 #include "data/BackendInterface.hpp"
 #include "rpc/Counters.hpp"
 #include "rpc/Errors.hpp"
+#include "rpc/RPCHelpers.hpp"
 #include "rpc/WorkQueue.hpp"
 #include "rpc/common/HandlerProvider.hpp"
 #include "rpc/common/Types.hpp"
@@ -134,8 +135,13 @@ public:
     Result
     buildResponse(web::Context const& ctx)
     {
-        if (forwardingProxy_.shouldForward(ctx))
+        if (forwardingProxy_.shouldForward(ctx)) {
+            // Disallow forwarding of the admin api, only user api is allowed for security reasons.
+            if (isAdminCmd(ctx.method, ctx.params))
+                return Result{Status{RippledError::rpcNO_PERMISSION}};
+
             return forwardingProxy_.forward(ctx);
+        }
 
         if (backend_->isTooBusy()) {
             LOG(log_.error()) << "Database is too busy. Rejecting request";

src/rpc/RPCHelpers.cpp

@@ -24,6 +24,7 @@
 #include "rpc/Errors.hpp"
 #include "rpc/JS.hpp"
 #include "rpc/common/Types.hpp"
+#include "util/AccountUtils.hpp"
 #include "util/Profiler.hpp"
 #include "util/log/Logger.hpp"
 #include "web/Context.hpp"
@@ -35,6 +36,7 @@
 #include <boost/json/array.hpp>
 #include <boost/json/object.hpp>
 #include <boost/json/parse.hpp>
+#include <boost/json/serialize.hpp>
 #include <boost/json/string.hpp>
 #include <boost/json/value.hpp>
 #include <boost/json/value_to.hpp>
@@ -48,6 +50,7 @@
 #include <ripple/basics/chrono.h>
 #include <ripple/basics/strHex.h>
 #include <ripple/beast/utility/Zero.h>
+#include <ripple/json/json_reader.h>
 #include <ripple/json/json_value.h>
 #include <ripple/protocol/AccountID.h>
 #include <ripple/protocol/Book.h>
@@ -186,14 +189,14 @@ accountFromStringStrict(std::string const& account)
     if (blob && ripple::publicKeyType(ripple::makeSlice(*blob))) {
         publicKey = ripple::PublicKey(ripple::Slice{blob->data(), blob->size()});
     } else {
-        publicKey = ripple::parseBase58<ripple::PublicKey>(ripple::TokenType::AccountPublic, account);
+        publicKey = util::parseBase58Wrapper<ripple::PublicKey>(ripple::TokenType::AccountPublic, account);
     }
 
     std::optional<ripple::AccountID> result;
     if (publicKey) {
         result = ripple::calcAccountID(*publicKey);
     } else {
-        result = ripple::parseBase58<ripple::AccountID>(account);
+        result = util::parseBase58Wrapper<ripple::AccountID>(account);
     }
 
     return result;
@@ -799,7 +802,7 @@ getAccountsFromTransaction(boost::json::object const& transaction)
             auto inObject = getAccountsFromTransaction(value.as_object());
             accounts.insert(accounts.end(), inObject.begin(), inObject.end());
         } else if (value.is_string()) {
-            auto const account = ripple::parseBase58<ripple::AccountID>(boost::json::value_to<std::string>(value));
+            auto const account = util::parseBase58Wrapper<ripple::AccountID>(boost::json::value_to<std::string>(value));
             if (account) {
                 accounts.push_back(*account);
             }
@@ -1272,6 +1275,31 @@ specifiesCurrentOrClosedLedger(boost::json::object const& request)
     return false;
 }
 
+bool
+isAdminCmd(std::string const& method, boost::json::object const& request)
+{
+    if (method == JS(ledger)) {
+        auto const requestStr = boost::json::serialize(request);
+        Json::Value jv;
+        Json::Reader{}.parse(requestStr, jv);
+        // rippled considers string/non-zero int/non-empty array/ non-empty json as true.
+        // Use rippled's API asBool to get the same result.
+        // https://github.com/XRPLF/rippled/issues/5119
+        auto const isFieldSet = [&jv](auto const field) { return jv.isMember(field) and jv[field].asBool(); };
+
+        // According to doc
+        // https://xrpl.org/docs/references/http-websocket-apis/public-api-methods/ledger-methods/ledger,
+        // full/accounts/type are admin only, but type only works when full/accounts are set, so we don't need to check
+        // type.
+        if (isFieldSet(JS(full)) or isFieldSet(JS(accounts)))
+            return true;
+    }
+
+    if (method == JS(feature) and request.contains(JS(vetoed)))
+        return true;
+    return false;
+}
+
 std::variant<ripple::uint256, Status>
 getNFTID(boost::json::object const& request)
 {

src/rpc/RPCHelpers.hpp

@@ -557,6 +557,16 @@ parseIssue(boost::json::object const& issue);
 bool
 specifiesCurrentOrClosedLedger(boost::json::object const& request);
 
+/**
+ * @brief Check whether a request requires administrative privileges on rippled side.
+ *
+ * @param method The method name to check
+ * @param request The request to check
+ * @return true if the request requires ADMIN role
+ */
+bool
+isAdminCmd(std::string const& method, boost::json::object const& request);
+
 /**
  * @brief Get the NFTID from the request
  *

src/rpc/common/Types.hpp

@@ -94,7 +94,7 @@ struct ReturnType {
      * @param warnings The warnings generated by the RPC call
      */
     ReturnType(std::expected<boost::json::value, Status> result, boost::json::array warnings = {})
-        : result{std::move(result)}, warnings{std::move(warnings)}
+        : result{std::move(result)}, warnings(std::move(warnings))
     {
     }
 

src/rpc/common/Validators.cpp

@@ -22,6 +22,7 @@
 #include "rpc/Errors.hpp"
 #include "rpc/RPCHelpers.hpp"
 #include "rpc/common/Types.hpp"
+#include "util/AccountUtils.hpp"
 
 #include <boost/json/object.hpp>
 #include <boost/json/value.hpp>
@@ -29,6 +30,7 @@
 #include <fmt/core.h>
 #include <ripple/basics/base_uint.h>
 #include <ripple/protocol/AccountID.h>
+#include <ripple/protocol/ErrorCodes.h>
 #include <ripple/protocol/UintTypes.h>
 #include <ripple/protocol/tokens.h>
 
@@ -113,7 +115,7 @@ CustomValidator AccountBase58Validator =
         if (!value.is_string())
             return Error{Status{RippledError::rpcINVALID_PARAMS, std::string(key) + "NotString"}};
 
-        auto const account = ripple::parseBase58<ripple::AccountID>(boost::json::value_to<std::string>(value));
+        auto const account = util::parseBase58Wrapper<ripple::AccountID>(boost::json::value_to<std::string>(value));
         if (!account || account->isZero())
             return Error{Status{ClioError::rpcMALFORMED_ADDRESS}};
 
@@ -140,8 +142,12 @@ CustomValidator CurrencyValidator =
         if (!value.is_string())
             return Error{Status{RippledError::rpcINVALID_PARAMS, std::string(key) + "NotString"}};
 
+        auto const currencyStr = boost::json::value_to<std::string>(value);
+        if (currencyStr.empty())
+            return Error{Status{RippledError::rpcINVALID_PARAMS, std::string(key) + "IsEmpty"}};
+
         ripple::Currency currency;
-        if (!ripple::to_currency(currency, boost::json::value_to<std::string>(value)))
+        if (!ripple::to_currency(currency, currencyStr))
             return Error{Status{ClioError::rpcMALFORMED_CURRENCY, "malformedCurrency"}};
 
         return MaybeError{};

src/rpc/handlers/AccountTx.hpp

@@ -39,7 +39,6 @@
 #include <ripple/protocol/jss.h>
 
 #include <cstdint>
-#include <limits>
 #include <memory>
 #include <optional>
 #include <string>
@@ -57,8 +56,8 @@ class AccountTxHandler {
     std::shared_ptr<BackendInterface> sharedPtrBackend_;
 
 public:
-    // no max limit
     static auto constexpr LIMIT_MIN = 1;
+    static auto constexpr LIMIT_MAX = 1000;
     static auto constexpr LIMIT_DEFAULT = 200;
 
     /**
@@ -133,7 +132,7 @@ public:
             {JS(limit),
              validation::Type<uint32_t>{},
              validation::Min(1u),
-             modifiers::Clamp<int32_t>{LIMIT_MIN, std::numeric_limits<int32_t>::max()}},
+             modifiers::Clamp<int32_t>{LIMIT_MIN, LIMIT_MAX}},
             {JS(marker),
              meta::WithCustomError{
                  validation::Type<boost::json::object>{},

src/rpc/handlers/GatewayBalances.hpp

@@ -25,6 +25,7 @@
 #include "rpc/common/Specs.hpp"
 #include "rpc/common/Types.hpp"
 #include "rpc/common/Validators.hpp"
+#include "util/AccountUtils.hpp"
 
 #include <boost/json/array.hpp>
 #include <boost/json/conversion.hpp>
@@ -116,14 +117,14 @@ public:
                 auto const wallets = value.is_array() ? value.as_array() : boost::json::array{value};
                 auto const getAccountID = [](auto const& j) -> std::optional<ripple::AccountID> {
                     if (j.is_string()) {
-                        auto const pk = ripple::parseBase58<ripple::PublicKey>(
+                        auto const pk = util::parseBase58Wrapper<ripple::PublicKey>(
                             ripple::TokenType::AccountPublic, boost::json::value_to<std::string>(j)
                         );
 
                         if (pk)
                             return ripple::calcAccountID(*pk);
 
-                        return ripple::parseBase58<ripple::AccountID>(boost::json::value_to<std::string>(j));
+                        return util::parseBase58Wrapper<ripple::AccountID>(boost::json::value_to<std::string>(j));
                     }
 
                     return {};

src/rpc/handlers/GetAggregatePrice.cpp

@@ -23,6 +23,7 @@
 #include "rpc/JS.hpp"
 #include "rpc/RPCHelpers.hpp"
 #include "rpc/common/Types.hpp"
+#include "util/AccountUtils.hpp"
 
 #include <boost/asio/spawn.hpp>
 #include <boost/bimap/bimap.hpp>
@@ -263,7 +264,7 @@ tag_invoke(boost::json::value_to_tag<GetAggregatePriceHandler::Input>, boost::js
     for (auto const& oracle : jsonObject.at(JS(oracles)).as_array()) {
         input.oracles.push_back(GetAggregatePriceHandler::Oracle{
             .documentId = boost::json::value_to<std::uint64_t>(oracle.as_object().at(JS(oracle_document_id))),
-            .account = *ripple::parseBase58<ripple::AccountID>(
+            .account = *util::parseBase58Wrapper<ripple::AccountID>(
                 boost::json::value_to<std::string>(oracle.as_object().at(JS(account)))
             )
         });

src/rpc/handlers/GetAggregatePrice.hpp

@@ -163,10 +163,12 @@ public:
         static auto const rpcSpec = RpcSpec{
             {JS(ledger_hash), validation::Uint256HexStringValidator},
             {JS(ledger_index), validation::LedgerIndexValidator},
-            // validate quoteAsset in accordance to the currency code found in XRPL doc:
+            // validate quoteAsset and base_asset in accordance to the currency code found in XRPL doc:
             // https://xrpl.org/docs/references/protocol/data-types/currency-formats#currency-codes
             // usually Clio returns rpcMALFORMED_CURRENCY , return InvalidParam here just to mimic rippled
-            {JS(base_asset), validation::Required{}, validation::Type<std::string>{}},
+            {JS(base_asset),
+             validation::Required{},
+             meta::WithCustomError{validation::CurrencyValidator, Status(RippledError::rpcINVALID_PARAMS)}},
             {JS(quote_asset),
              validation::Required{},
              meta::WithCustomError{validation::CurrencyValidator, Status(RippledError::rpcINVALID_PARAMS)}},

src/rpc/handlers/Ledger.hpp

@@ -70,9 +70,8 @@ public:
      * - ledger
      * - type
      *
-     * Clio will throw an error when `queue` is set to `true`
-     * or if `full` or `accounts` are used.
-     * @see https://github.com/XRPLF/clio/issues/603
+     * Clio will throw an error when `queue`, `full` or `accounts` is set to `true`.
+     * @see https://github.com/XRPLF/clio/issues/603 and https://github.com/XRPLF/clio/issues/1537
      */
     struct Input {
         std::optional<std::string> ledgerHash;
@@ -105,9 +104,9 @@ public:
     spec([[maybe_unused]] uint32_t apiVersion)
     {
         static auto const rpcSpec = RpcSpec{
-            {JS(full), validation::NotSupported{}},
+            {JS(full), validation::Type<bool>{}, validation::NotSupported{true}},
             {JS(full), check::Deprecated{}},
-            {JS(accounts), validation::NotSupported{}},
+            {JS(accounts), validation::Type<bool>{}, validation::NotSupported{true}},
             {JS(accounts), check::Deprecated{}},
             {JS(owner_funds), validation::Type<bool>{}},
             {JS(queue), validation::Type<bool>{}, validation::NotSupported{true}},

src/rpc/handlers/LedgerEntry.cpp

@@ -23,6 +23,7 @@
 #include "rpc/JS.hpp"
 #include "rpc/RPCHelpers.hpp"
 #include "rpc/common/Types.hpp"
+#include "util/AccountUtils.hpp"
 
 #include <boost/json/conversion.hpp>
 #include <boost/json/object.hpp>
@@ -62,9 +63,9 @@ LedgerEntryHandler::process(LedgerEntryHandler::Input input, Context const& ctx)
     if (input.index) {
         key = ripple::uint256{std::string_view(*(input.index))};
     } else if (input.accountRoot) {
-        key = ripple::keylet::account(*ripple::parseBase58<ripple::AccountID>(*(input.accountRoot))).key;
+        key = ripple::keylet::account(*util::parseBase58Wrapper<ripple::AccountID>(*(input.accountRoot))).key;
     } else if (input.did) {
-        key = ripple::keylet::did(*ripple::parseBase58<ripple::AccountID>(*(input.did))).key;
+        key = ripple::keylet::did(*util::parseBase58Wrapper<ripple::AccountID>(*(input.did))).key;
     } else if (input.directory) {
         auto const keyOrStatus = composeKeyFromDirectory(*input.directory);
         if (auto const status = std::get_if<Status>(&keyOrStatus))
@@ -73,13 +74,14 @@ LedgerEntryHandler::process(LedgerEntryHandler::Input input, Context const& ctx)
         key = std::get<ripple::uint256>(keyOrStatus);
     } else if (input.offer) {
         auto const id =
-            ripple::parseBase58<ripple::AccountID>(boost::json::value_to<std::string>(input.offer->at(JS(account))));
+            util::parseBase58Wrapper<ripple::AccountID>(boost::json::value_to<std::string>(input.offer->at(JS(account)))
+            );
         key = ripple::keylet::offer(*id, boost::json::value_to<std::uint32_t>(input.offer->at(JS(seq)))).key;
     } else if (input.rippleStateAccount) {
-        auto const id1 = ripple::parseBase58<ripple::AccountID>(
+        auto const id1 = util::parseBase58Wrapper<ripple::AccountID>(
             boost::json::value_to<std::string>(input.rippleStateAccount->at(JS(accounts)).as_array().at(0))
         );
-        auto const id2 = ripple::parseBase58<ripple::AccountID>(
+        auto const id2 = util::parseBase58Wrapper<ripple::AccountID>(
             boost::json::value_to<std::string>(input.rippleStateAccount->at(JS(accounts)).as_array().at(1))
         );
         auto const currency =
@@ -88,20 +90,22 @@ LedgerEntryHandler::process(LedgerEntryHandler::Input input, Context const& ctx)
         key = ripple::keylet::line(*id1, *id2, currency).key;
     } else if (input.escrow) {
         auto const id =
-            ripple::parseBase58<ripple::AccountID>(boost::json::value_to<std::string>(input.escrow->at(JS(owner))));
+            util::parseBase58Wrapper<ripple::AccountID>(boost::json::value_to<std::string>(input.escrow->at(JS(owner)))
+            );
         key = ripple::keylet::escrow(*id, input.escrow->at(JS(seq)).as_int64()).key;
     } else if (input.depositPreauth) {
-        auto const owner = ripple::parseBase58<ripple::AccountID>(
+        auto const owner = util::parseBase58Wrapper<ripple::AccountID>(
             boost::json::value_to<std::string>(input.depositPreauth->at(JS(owner)))
         );
-        auto const authorized = ripple::parseBase58<ripple::AccountID>(
+        auto const authorized = util::parseBase58Wrapper<ripple::AccountID>(
             boost::json::value_to<std::string>(input.depositPreauth->at(JS(authorized)))
         );
 
         key = ripple::keylet::depositPreauth(*owner, *authorized).key;
     } else if (input.ticket) {
         auto const id =
-            ripple::parseBase58<ripple::AccountID>(boost::json::value_to<std::string>(input.ticket->at(JS(account))));
+            util::parseBase58Wrapper<ripple::AccountID>(boost::json::value_to<std::string>(input.ticket->at(JS(account))
+            ));
 
         key = ripple::getTicketIndex(*id, input.ticket->at(JS(ticket_seq)).as_int64());
     } else if (input.amm) {
@@ -112,7 +116,8 @@ LedgerEntryHandler::process(LedgerEntryHandler::Input input, Context const& ctx)
                 return ripple::xrpIssue();
             }
             auto const issuer =
-                ripple::parseBase58<ripple::AccountID>(boost::json::value_to<std::string>(assetJson.at(JS(issuer))));
+                util::parseBase58Wrapper<ripple::AccountID>(boost::json::value_to<std::string>(assetJson.at(JS(issuer)))
+                );
             return ripple::Issue{currency, *issuer};
         };
 
@@ -125,7 +130,7 @@ LedgerEntryHandler::process(LedgerEntryHandler::Input input, Context const& ctx)
             return Error{Status{ClioError::rpcMALFORMED_REQUEST}};
 
         if (input.bridgeAccount) {
-            auto const bridgeAccount = ripple::parseBase58<ripple::AccountID>(*(input.bridgeAccount));
+            auto const bridgeAccount = util::parseBase58Wrapper<ripple::AccountID>(*(input.bridgeAccount));
             auto const chainType = ripple::STXChainBridge::srcChain(bridgeAccount == input.bridge->lockingChainDoor());
 
             if (bridgeAccount != input.bridge->door(chainType))
@@ -201,7 +206,7 @@ LedgerEntryHandler::composeKeyFromDirectory(boost::json::object const& directory
     }
 
     auto const ownerID =
-        ripple::parseBase58<ripple::AccountID>(boost::json::value_to<std::string>(directory.at(JS(owner))));
+        util::parseBase58Wrapper<ripple::AccountID>(boost::json::value_to<std::string>(directory.at(JS(owner))));
     return ripple::keylet::page(ripple::keylet::ownerDir(*ownerID), subIndex).key;
 }
 
@@ -262,10 +267,10 @@ tag_invoke(boost::json::value_to_tag<LedgerEntryHandler::Input>, boost::json::va
     };
 
     auto const parseBridgeFromJson = [](boost::json::value const& bridgeJson) {
-        auto const lockingDoor = *ripple::parseBase58<ripple::AccountID>(
+        auto const lockingDoor = *util::parseBase58Wrapper<ripple::AccountID>(
             boost::json::value_to<std::string>(bridgeJson.at(ripple::sfLockingChainDoor.getJsonName().c_str()))
         );
-        auto const issuingDoor = *ripple::parseBase58<ripple::AccountID>(
+        auto const issuingDoor = *util::parseBase58Wrapper<ripple::AccountID>(
             boost::json::value_to<std::string>(bridgeJson.at(ripple::sfIssuingChainDoor.getJsonName().c_str()))
         );
         auto const lockingIssue =
@@ -278,7 +283,7 @@ tag_invoke(boost::json::value_to_tag<LedgerEntryHandler::Input>, boost::json::va
 
     auto const parseOracleFromJson = [](boost::json::value const& json) {
         auto const account =
-            ripple::parseBase58<ripple::AccountID>(boost::json::value_to<std::string>(json.at(JS(account))));
+            util::parseBase58Wrapper<ripple::AccountID>(boost::json::value_to<std::string>(json.at(JS(account))));
         auto const documentId = boost::json::value_to<uint32_t>(json.at(JS(oracle_document_id)));
 
         return ripple::keylet::oracle(*account, documentId).key;

src/rpc/handlers/LedgerEntry.hpp

@@ -28,6 +28,7 @@
 #include "rpc/common/Specs.hpp"
 #include "rpc/common/Types.hpp"
 #include "rpc/common/Validators.hpp"
+#include "util/AccountUtils.hpp"
 
 #include <boost/json/conversion.hpp>
 #include <boost/json/object.hpp>
@@ -136,9 +137,11 @@ public:
                 }
 
                 auto const id1 =
-                    ripple::parseBase58<ripple::AccountID>(boost::json::value_to<std::string>(value.as_array()[0]));
+                    util::parseBase58Wrapper<ripple::AccountID>(boost::json::value_to<std::string>(value.as_array()[0])
+                    );
                 auto const id2 =
-                    ripple::parseBase58<ripple::AccountID>(boost::json::value_to<std::string>(value.as_array()[1]));
+                    util::parseBase58Wrapper<ripple::AccountID>(boost::json::value_to<std::string>(value.as_array()[1])
+                    );
 
                 if (!id1 || !id2)
                     return Error{Status{ClioError::rpcMALFORMED_ADDRESS, "malformedAddresses"}};

src/util/AccountUtils.hpp

@@ -0,0 +1,67 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2024, 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 <ripple/protocol/tokens.h>
+
+#include <cctype>
+#include <optional>
+#include <string>
+
+namespace util {
+
+/**
+ * @brief A wrapper of parseBase58 function. It adds the check if all characters in the input string are alphanumeric.
+ * If not, it returns an empty optional, instead of calling the parseBase58 function.
+ *
+ * @tparam T The type of the value to parse to.
+ * @param str The string to parse.
+ * @return An optional with the parsed value, or an empty optional if the parse fails.
+ */
+template <class T>
+[[nodiscard]] std::optional<T>
+parseBase58Wrapper(std::string const& str)
+{
+    if (!std::all_of(std::begin(str), std::end(str), [](unsigned char c) { return std::isalnum(c); }))
+        return std::nullopt;
+
+    return ripple::parseBase58<T>(str);
+}
+
+/**
+ * @brief A wrapper of parseBase58 function. It add the check if all characters in the input string are alphanumeric. If
+ * not, it returns an empty optional, instead of calling the parseBase58 function.
+ *
+ * @tparam T The type of the value to parse to.
+ * @param type The type of the token to parse.
+ * @param str The string to parse.
+ * @return An optional with the parsed value, or an empty optional if the parse fails.
+ */
+template <class T>
+[[nodiscard]] std::optional<T>
+parseBase58Wrapper(ripple::TokenType type, std::string const& str)
+{
+    if (!std::all_of(std::begin(str), std::end(str), [](unsigned char c) { return std::isalnum(c); }))
+        return std::nullopt;
+
+    return ripple::parseBase58<T>(type, str);
+}
+
+}  // namespace util

src/util/requests/impl/WsConnectionImpl.hpp

@@ -39,8 +39,10 @@
 #include <boost/beast/websocket/stream_base.hpp>
 #include <boost/system/errc.hpp>
 
+#include <atomic>
 #include <chrono>
 #include <expected>
+#include <memory>
 #include <optional>
 #include <string>
 #include <utility>
@@ -123,15 +125,20 @@ private:
     static void
     withTimeout(Operation&& operation, boost::asio::yield_context yield, std::chrono::steady_clock::duration timeout)
     {
+        auto isCompleted = std::make_shared<bool>(false);
         boost::asio::cancellation_signal cancellationSignal;
         auto cyield = boost::asio::bind_cancellation_slot(cancellationSignal.slot(), yield);
 
         boost::asio::steady_timer timer{boost::asio::get_associated_executor(cyield), timeout};
-        timer.async_wait([&cancellationSignal](boost::system::error_code errorCode) {
-            if (!errorCode)
+
+        // The timer below can be called with no error code even if the operation is completed before the timeout, so we
+        // need an additional flag here
+        timer.async_wait([&cancellationSignal, isCompleted](boost::system::error_code errorCode) {
+            if (!errorCode and not*isCompleted)
                 cancellationSignal.emit(boost::asio::cancellation_type::terminal);
         });
         operation(cyield);
+        *isCompleted = true;
     }
 
     static boost::system::error_code

src/web/Server.hpp

@@ -69,10 +69,8 @@ namespace web {
  * @tparam HandlerType The executor to handle the requests
  */
 template <
-    template <typename>
-    class PlainSessionType,
-    template <typename>
-    class SslSessionType,
+    template <typename> class PlainSessionType,
+    template <typename> class SslSessionType,
     SomeServerHandler HandlerType>
 class Detector : public std::enable_shared_from_this<Detector<PlainSessionType, SslSessionType, HandlerType>> {
     using std::enable_shared_from_this<Detector<PlainSessionType, SslSessionType, HandlerType>>::shared_from_this;
@@ -191,10 +189,8 @@ public:
  * @tparam HandlerType The handler to process the request and return response.
  */
 template <
-    template <typename>
-    class PlainSessionType,
-    template <typename>
-    class SslSessionType,
+    template <typename> class PlainSessionType,
+    template <typename> class SslSessionType,
     SomeServerHandler HandlerType>
 class Server : public std::enable_shared_from_this<Server<PlainSessionType, SslSessionType, HandlerType>> {
     using std::enable_shared_from_this<Server<PlainSessionType, SslSessionType, HandlerType>>::shared_from_this;

src/web/impl/WsBase.hpp

@@ -23,6 +23,9 @@
 #include "rpc/common/Types.hpp"
 #include "util/Taggable.hpp"
 #include "util/log/Logger.hpp"
+#include "util/prometheus/Gauge.hpp"
+#include "util/prometheus/Label.hpp"
+#include "util/prometheus/Prometheus.hpp"
 #include "web/DOSGuard.hpp"
 #include "web/interface/Concepts.hpp"
 #include "web/interface/ConnectionBase.hpp"
@@ -71,6 +74,8 @@ template <template <typename> typename Derived, SomeServerHandler HandlerType>
 class WsBase : public ConnectionBase, public std::enable_shared_from_this<WsBase<Derived, HandlerType>> {
     using std::enable_shared_from_this<WsBase<Derived, HandlerType>>::shared_from_this;
 
+    std::reference_wrapper<util::prometheus::GaugeInt> messagesLength_;
+
     boost::beast::flat_buffer buffer_;
     std::reference_wrapper<web::DOSGuard> dosGuard_;
     bool sending_ = false;
@@ -103,15 +108,26 @@ public:
         std::shared_ptr<HandlerType> const& handler,
         boost::beast::flat_buffer&& buffer
     )
-        : ConnectionBase(tagFactory, ip), buffer_(std::move(buffer)), dosGuard_(dosGuard), handler_(handler)
+        : ConnectionBase(tagFactory, ip)
+        , messagesLength_(PrometheusService::gaugeInt(
+              "ws_messages_length",
+              util::prometheus::Labels(),
+              "The total length of messages in the queue"
+          ))
+        , buffer_(std::move(buffer))
+        , dosGuard_(dosGuard)
+        , handler_(handler)
     {
         upgraded = true;  // NOLINT (cppcoreguidelines-pro-type-member-init)
+
         LOG(perfLog_.debug()) << tag() << "session created";
     }
 
     ~WsBase() override
     {
         LOG(perfLog_.debug()) << tag() << "session closed";
+        if (!messages_.empty())
+            messagesLength_.get() -= messages_.size();
         dosGuard_.get().decrement(clientIp);
     }
 
@@ -135,6 +151,7 @@ public:
     onWrite(boost::system::error_code ec, std::size_t)
     {
         messages_.pop();
+        --messagesLength_.get();
         sending_ = false;
         if (ec) {
             wsFail(ec, "Failed to write");
@@ -165,6 +182,7 @@ public:
             derived().ws().get_executor(),
             [this, self = derived().shared_from_this(), msg = std::move(msg)]() {
                 messages_.push(msg);
+                ++messagesLength_.get();
                 maybeSendNext();
             }
         );

tests/common/CMakeLists.txt

@@ -2,6 +2,7 @@ add_library(clio_testing_common)
 
 target_sources(
   clio_testing_common PRIVATE util/StringUtils.cpp util/TestHttpServer.cpp util/TestWsServer.cpp util/TestObject.cpp
+                              util/AssignRandomPort.cpp
 )
 
 include(deps/gtest)

tests/common/util/AssignRandomPort.cpp

@@ -0,0 +1,45 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2024, 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 "util/AssignRandomPort.hpp"
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+
+#include <cstdint>
+
+using tcp = boost::asio::ip::tcp;
+
+namespace tests::util {
+
+uint32_t
+generateFreePort()
+{
+    boost::asio::io_context io_context;
+    tcp::acceptor acceptor(io_context);
+    tcp::endpoint const endpoint(tcp::v4(), 0);
+
+    acceptor.open(endpoint.protocol());
+    acceptor.set_option(tcp::acceptor::reuse_address(true));
+    acceptor.bind(endpoint);
+
+    return acceptor.local_endpoint().port();
+}
+
+}  // namespace tests::util

tests/common/util/AssignRandomPort.hpp

@@ -0,0 +1,29 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2024, 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 <cstdint>
+
+namespace tests::util {
+
+uint32_t
+generateFreePort();
+
+}  // namespace tests::util

tests/common/util/MockXrpLedgerAPIService.hpp

@@ -82,7 +82,7 @@ struct WithMockXrpLedgerAPIService : virtual ::testing::Test {
     WithMockXrpLedgerAPIService(std::string serverAddress)
     {
         grpc::ServerBuilder builder;
-        builder.AddListeningPort(serverAddress, grpc::InsecureServerCredentials());
+        builder.AddListeningPort(serverAddress, grpc::InsecureServerCredentials(), &port_);
         builder.RegisterService(&mockXrpLedgerAPIService);
         server_ = builder.BuildAndStart();
         serverThread_ = std::thread([this] { server_->Wait(); });
@@ -94,11 +94,17 @@ struct WithMockXrpLedgerAPIService : virtual ::testing::Test {
         serverThread_.join();
     }
 
+    int
+    getXRPLMockPort() const
+    {
+        return port_;
+    }
     MockXrpLedgerAPIService mockXrpLedgerAPIService;
 
 private:
     std::unique_ptr<grpc::Server> server_;
     std::thread serverThread_;
+    int port_{};
 };
 
 }  // namespace tests::util

tests/common/util/TestHttpServer.cpp

@@ -105,9 +105,9 @@ doSession(
 
 }  // namespace
 
-TestHttpServer::TestHttpServer(boost::asio::io_context& context, std::string host, int const port) : acceptor_(context)
+TestHttpServer::TestHttpServer(boost::asio::io_context& context, std::string host) : acceptor_(context)
 {
-    boost::asio::ip::tcp::endpoint const endpoint(boost::asio::ip::make_address(host), port);
+    boost::asio::ip::tcp::endpoint const endpoint(boost::asio::ip::make_address(host), 0);
     acceptor_.open(endpoint.protocol());
     acceptor_.set_option(asio::socket_base::reuse_address(true));
     acceptor_.bind(endpoint);
@@ -134,3 +134,9 @@ TestHttpServer::handleRequest(TestHttpServer::RequestHandler handler, bool const
         boost::asio::detached
     );
 }
+
+std::string
+TestHttpServer::port() const
+{
+    return std::to_string(acceptor_.local_endpoint().port());
+}

tests/common/util/TestHttpServer.hpp

@@ -41,9 +41,8 @@ public:
      *
      * @param context boost::asio::io_context to use for networking
      * @param host host to bind to
-     * @param port port to bind to
      */
-    TestHttpServer(boost::asio::io_context& context, std::string host, int port);
+    TestHttpServer(boost::asio::io_context& context, std::string host);
 
     /**
      * @brief Start the server
@@ -56,6 +55,14 @@ public:
     void
     handleRequest(RequestHandler handler, bool allowToFail = false);
 
+    /**
+     * @brief Return the port HTTP server is connected to
+     *
+     * @return string port number
+     */
+    std::string
+    port() const;
+
 private:
     boost::asio::ip::tcp::acceptor acceptor_;
 };

tests/common/util/TestObject.cpp

@@ -21,6 +21,7 @@
 
 #include "data/DBHelpers.hpp"
 #include "data/Types.hpp"
+#include "util/AccountUtils.hpp"
 #include "util/Assert.hpp"
 
 #include <ripple/basics/Blob.h>
@@ -60,7 +61,7 @@ constexpr static auto INDEX1 = "1B8590C01B0006EDFA9ED60296DD052DC5E90F99659B2501
 ripple::AccountID
 GetAccountIDWithString(std::string_view id)
 {
-    return ripple::parseBase58<ripple::AccountID>(std::string(id)).value();
+    return util::parseBase58Wrapper<ripple::AccountID>(std::string(id)).value();
 }
 
 ripple::uint256
@@ -154,11 +155,11 @@ CreatePaymentTransactionObject(
 {
     ripple::STObject obj(ripple::sfTransaction);
     obj.setFieldU16(ripple::sfTransactionType, ripple::ttPAYMENT);
-    auto account = ripple::parseBase58<ripple::AccountID>(std::string(accountId1));
+    auto account = util::parseBase58Wrapper<ripple::AccountID>(std::string(accountId1));
     obj.setAccountID(ripple::sfAccount, account.value());
     obj.setFieldAmount(ripple::sfAmount, ripple::STAmount(amount, false));
     obj.setFieldAmount(ripple::sfFee, ripple::STAmount(fee, false));
-    auto account2 = ripple::parseBase58<ripple::AccountID>(std::string(accountId2));
+    auto account2 = util::parseBase58Wrapper<ripple::AccountID>(std::string(accountId2));
     obj.setAccountID(ripple::sfDestination, account2.value());
     obj.setFieldU32(ripple::sfSequence, seq);
     char const* key = "test";
@@ -258,14 +259,14 @@ CreateCreateOfferTransactionObject(
 {
     ripple::STObject obj(ripple::sfTransaction);
     obj.setFieldU16(ripple::sfTransactionType, ripple::ttOFFER_CREATE);
-    auto account = ripple::parseBase58<ripple::AccountID>(std::string(accountId));
+    auto account = util::parseBase58Wrapper<ripple::AccountID>(std::string(accountId));
     obj.setAccountID(ripple::sfAccount, account.value());
     auto amount = ripple::STAmount(fee, false);
     obj.setFieldAmount(ripple::sfFee, amount);
     obj.setFieldU32(ripple::sfSequence, seq);
     // add amount
     ripple::Issue const issue1(
-        ripple::Currency{currency}, ripple::parseBase58<ripple::AccountID>(std::string(issuer)).value()
+        ripple::Currency{currency}, util::parseBase58Wrapper<ripple::AccountID>(std::string(issuer)).value()
     );
     if (reverse) {
         obj.setFieldAmount(ripple::sfTakerPays, ripple::STAmount(issue1, takerGets));
@@ -288,11 +289,11 @@ GetIssue(std::string_view currency, std::string_view issuerId)
     if (currency.size() == 3) {
         return ripple::Issue(
             ripple::to_currency(std::string(currency)),
-            ripple::parseBase58<ripple::AccountID>(std::string(issuerId)).value()
+            util::parseBase58Wrapper<ripple::AccountID>(std::string(issuerId)).value()
         );
     }
     return ripple::Issue(
-        ripple::Currency{currency}, ripple::parseBase58<ripple::AccountID>(std::string(issuerId)).value()
+        ripple::Currency{currency}, util::parseBase58Wrapper<ripple::AccountID>(std::string(issuerId)).value()
     );
 }
 
@@ -636,7 +637,7 @@ CreateMintNFTTxWithMetadata(
     // tx
     ripple::STObject tx(ripple::sfTransaction);
     tx.setFieldU16(ripple::sfTransactionType, ripple::ttNFTOKEN_MINT);
-    auto account = ripple::parseBase58<ripple::AccountID>(std::string(accountId));
+    auto account = util::parseBase58Wrapper<ripple::AccountID>(std::string(accountId));
     tx.setAccountID(ripple::sfAccount, account.value());
     auto amount = ripple::STAmount(fee, false);
     tx.setFieldAmount(ripple::sfFee, amount);
@@ -693,7 +694,7 @@ CreateAcceptNFTOfferTxWithMetadata(std::string_view accountId, uint32_t seq, uin
     // tx
     ripple::STObject tx(ripple::sfTransaction);
     tx.setFieldU16(ripple::sfTransactionType, ripple::ttNFTOKEN_ACCEPT_OFFER);
-    auto account = ripple::parseBase58<ripple::AccountID>(std::string(accountId));
+    auto account = util::parseBase58Wrapper<ripple::AccountID>(std::string(accountId));
     tx.setAccountID(ripple::sfAccount, account.value());
     auto amount = ripple::STAmount(fee, false);
     tx.setFieldAmount(ripple::sfFee, amount);
@@ -737,7 +738,7 @@ CreateCancelNFTOffersTxWithMetadata(
     // tx
     ripple::STObject tx(ripple::sfTransaction);
     tx.setFieldU16(ripple::sfTransactionType, ripple::ttNFTOKEN_CANCEL_OFFER);
-    auto account = ripple::parseBase58<ripple::AccountID>(std::string(accountId));
+    auto account = util::parseBase58Wrapper<ripple::AccountID>(std::string(accountId));
     tx.setAccountID(ripple::sfAccount, account.value());
     auto amount = ripple::STAmount(fee, false);
     tx.setFieldAmount(ripple::sfFee, amount);
@@ -791,7 +792,7 @@ CreateCreateNFTOfferTxWithMetadata(
     // tx
     ripple::STObject tx(ripple::sfTransaction);
     tx.setFieldU16(ripple::sfTransactionType, ripple::ttNFTOKEN_CREATE_OFFER);
-    auto account = ripple::parseBase58<ripple::AccountID>(std::string(accountId));
+    auto account = util::parseBase58Wrapper<ripple::AccountID>(std::string(accountId));
     tx.setAccountID(ripple::sfAccount, account.value());
     auto amount = ripple::STAmount(fee, false);
     tx.setFieldAmount(ripple::sfFee, amount);
@@ -839,7 +840,7 @@ CreateOracleSetTxWithMetadata(
     // tx
     ripple::STObject tx(ripple::sfTransaction);
     tx.setFieldU16(ripple::sfTransactionType, ripple::ttORACLE_SET);
-    auto account = ripple::parseBase58<ripple::AccountID>(std::string(accountId));
+    auto account = util::parseBase58Wrapper<ripple::AccountID>(std::string(accountId));
     tx.setAccountID(ripple::sfAccount, account.value());
     auto amount = ripple::STAmount(fee, false);
     tx.setFieldAmount(ripple::sfFee, amount);
@@ -909,7 +910,7 @@ CreateAMMObject(
     amm.setFieldIssue(ripple::sfAsset2, ripple::STIssue{ripple::sfAsset2, GetIssue(asset2Currency, asset2Issuer)});
     ripple::Issue const issue1(
         ripple::Currency{lpTokenBalanceIssueCurrency},
-        ripple::parseBase58<ripple::AccountID>(std::string(accountId)).value()
+        util::parseBase58Wrapper<ripple::AccountID>(std::string(accountId)).value()
     );
     amm.setFieldAmount(ripple::sfLPTokenBalance, ripple::STAmount(issue1, lpTokenBalanceIssueAmount));
     amm.setFieldU32(ripple::sfFlags, 0);

tests/common/util/TestWsServer.cpp

@@ -105,14 +105,20 @@ TestWsConnection::headers() const
     return headers_;
 }
 
-TestWsServer::TestWsServer(asio::io_context& context, std::string const& host, int port) : acceptor_(context)
+TestWsServer::TestWsServer(asio::io_context& context, std::string const& host) : acceptor_(context)
 {
-    auto endpoint = asio::ip::tcp::endpoint(boost::asio::ip::make_address(host), port);
+    auto endpoint = asio::ip::tcp::endpoint(boost::asio::ip::make_address(host), 0);
     acceptor_.open(endpoint.protocol());
     acceptor_.set_option(asio::socket_base::reuse_address(true));
     acceptor_.bind(endpoint);
 }
 
+std::string
+TestWsServer::port() const
+{
+    return std::to_string(this->acceptor_.local_endpoint().port());
+}
+
 std::expected<TestWsConnection, util::requests::RequestError>
 TestWsServer::acceptConnection(asio::yield_context yield)
 {

tests/common/util/TestWsServer.hpp

@@ -70,7 +70,10 @@ class TestWsServer {
     boost::asio::ip::tcp::acceptor acceptor_;
 
 public:
-    TestWsServer(boost::asio::io_context& context, std::string const& host, int port);
+    TestWsServer(boost::asio::io_context& context, std::string const& host);
+
+    std::string
+    port() const;
 
     std::expected<TestWsConnection, util::requests::RequestError>
     acceptConnection(boost::asio::yield_context yield);

tests/unit/CMakeLists.txt

@@ -92,6 +92,7 @@ target_sources(
           rpc/JsonBoolTests.cpp
           rpc/RPCHelpersTests.cpp
           rpc/WorkQueueTests.cpp
+          util/AccountUtilsTests.cpp
           util/AssertTests.cpp
           # Async framework
           util/async/AnyExecutionContextTests.cpp

tests/unit/etl/ForwardingSourceTests.cpp

@@ -37,8 +37,13 @@
 using namespace etl::impl;
 
 struct ForwardingSourceTests : SyncAsioContextTest {
-    TestWsServer server_{ctx, "0.0.0.0", 11114};
-    ForwardingSource forwardingSource{"127.0.0.1", "11114", std::chrono::milliseconds{1}, std::chrono::milliseconds{1}};
+    TestWsServer server_{ctx, "0.0.0.0"};
+    ForwardingSource forwardingSource{
+        "127.0.0.1",
+        server_.port(),
+        std::chrono::milliseconds{1},
+        std::chrono::milliseconds{1}
+    };
 };
 
 TEST_F(ForwardingSourceTests, ConnectionFailed)

tests/unit/etl/GrpcSourceTests.cpp

@@ -42,9 +42,9 @@ using namespace etl::impl;
 
 struct GrpcSourceTests : NoLoggerFixture, util::prometheus::WithPrometheus, tests::util::WithMockXrpLedgerAPIService {
     GrpcSourceTests()
-        : WithMockXrpLedgerAPIService("localhost:55051")
+        : WithMockXrpLedgerAPIService("localhost:0")
         , mockBackend_(std::make_shared<testing::StrictMock<MockBackend>>(util::Config{}))
-        , grpcSource_("127.0.0.1", "55051", mockBackend_)
+        , grpcSource_("127.0.0.1", std::to_string(getXRPLMockPort()), mockBackend_)
     {
     }
 

tests/unit/etl/LoadBalancerTests.cpp

@@ -271,15 +271,12 @@ TEST_F(LoadBalancerOnDisconnectHookTests, source0Disconnects)
     EXPECT_CALL(sourceFactory_.sourceAt(0), setForwarding(false));
     EXPECT_CALL(sourceFactory_.sourceAt(1), isConnected()).WillOnce(Return(true));
     EXPECT_CALL(sourceFactory_.sourceAt(1), setForwarding(true));
-    sourceFactory_.callbacksAt(0).onDisconnect();
+    sourceFactory_.callbacksAt(0).onDisconnect(true);
 }
 
 TEST_F(LoadBalancerOnDisconnectHookTests, source1Disconnects)
 {
-    EXPECT_CALL(sourceFactory_.sourceAt(0), isConnected()).WillOnce(Return(true));
-    EXPECT_CALL(sourceFactory_.sourceAt(0), setForwarding(true));
-    EXPECT_CALL(sourceFactory_.sourceAt(1), setForwarding(false));
-    sourceFactory_.callbacksAt(1).onDisconnect();
+    sourceFactory_.callbacksAt(1).onDisconnect(false);
 }
 
 TEST_F(LoadBalancerOnDisconnectHookTests, source0DisconnectsAndConnectsBack)
@@ -288,29 +285,25 @@ TEST_F(LoadBalancerOnDisconnectHookTests, source0DisconnectsAndConnectsBack)
     EXPECT_CALL(sourceFactory_.sourceAt(0), setForwarding(false));
     EXPECT_CALL(sourceFactory_.sourceAt(1), isConnected()).WillOnce(Return(true));
     EXPECT_CALL(sourceFactory_.sourceAt(1), setForwarding(true));
-    sourceFactory_.callbacksAt(0).onDisconnect();
+    sourceFactory_.callbacksAt(0).onDisconnect(true);
 
     sourceFactory_.callbacksAt(0).onConnect();
 }
 
 TEST_F(LoadBalancerOnDisconnectHookTests, source1DisconnectsAndConnectsBack)
 {
-    EXPECT_CALL(sourceFactory_.sourceAt(0), isConnected()).WillOnce(Return(true));
-    EXPECT_CALL(sourceFactory_.sourceAt(0), setForwarding(true));
-    EXPECT_CALL(sourceFactory_.sourceAt(1), setForwarding(false));
-    sourceFactory_.callbacksAt(1).onDisconnect();
-
+    sourceFactory_.callbacksAt(1).onDisconnect(false);
     sourceFactory_.callbacksAt(1).onConnect();
 }
 
 TEST_F(LoadBalancerOnConnectHookTests, bothSourcesDisconnectAndConnectBack)
 {
-    EXPECT_CALL(sourceFactory_.sourceAt(0), isConnected()).Times(2).WillRepeatedly(Return(false));
-    EXPECT_CALL(sourceFactory_.sourceAt(0), setForwarding(false)).Times(2);
-    EXPECT_CALL(sourceFactory_.sourceAt(1), isConnected()).Times(2).WillRepeatedly(Return(false));
-    EXPECT_CALL(sourceFactory_.sourceAt(1), setForwarding(false)).Times(2);
-    sourceFactory_.callbacksAt(0).onDisconnect();
-    sourceFactory_.callbacksAt(1).onDisconnect();
+    EXPECT_CALL(sourceFactory_.sourceAt(0), isConnected()).WillOnce(Return(false));
+    EXPECT_CALL(sourceFactory_.sourceAt(0), setForwarding(false));
+    EXPECT_CALL(sourceFactory_.sourceAt(1), isConnected()).WillOnce(Return(false));
+    EXPECT_CALL(sourceFactory_.sourceAt(1), setForwarding(false));
+    sourceFactory_.callbacksAt(0).onDisconnect(true);
+    sourceFactory_.callbacksAt(1).onDisconnect(false);
 
     EXPECT_CALL(sourceFactory_.sourceAt(0), isConnected()).WillOnce(Return(true));
     EXPECT_CALL(sourceFactory_.sourceAt(0), setForwarding(true));
@@ -353,12 +346,7 @@ TEST_F(LoadBalancer3SourcesTests, forwardingUpdate)
     sourceFactory_.callbacksAt(1).onConnect();
 
     // Source 0 got disconnected
-    EXPECT_CALL(sourceFactory_.sourceAt(0), isConnected()).WillOnce(Return(false));
-    EXPECT_CALL(sourceFactory_.sourceAt(0), setForwarding(false));
-    EXPECT_CALL(sourceFactory_.sourceAt(1), isConnected()).WillOnce(Return(true));
-    EXPECT_CALL(sourceFactory_.sourceAt(1), setForwarding(true));
-    EXPECT_CALL(sourceFactory_.sourceAt(2), setForwarding(false));  // only source 1 must be forwarding
-    sourceFactory_.callbacksAt(0).onDisconnect();
+    sourceFactory_.callbacksAt(0).onDisconnect(false);
 }
 
 struct LoadBalancerLoadInitialLedgerTests : LoadBalancerOnConnectHookTests {

tests/unit/etl/SubscriptionSourceTests.cpp

@@ -20,53 +20,58 @@
 #include "etl/impl/SubscriptionSource.hpp"
 #include "util/Fixtures.hpp"
 #include "util/MockNetworkValidatedLedgers.hpp"
+#include "util/MockPrometheus.hpp"
 #include "util/MockSubscriptionManager.hpp"
 #include "util/TestWsServer.hpp"
+#include "util/prometheus/Gauge.hpp"
 
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/spawn.hpp>
 #include <boost/json/object.hpp>
 #include <boost/json/serialize.hpp>
+#include <fmt/core.h>
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
 #include <chrono>
+#include <cstdint>
+#include <cstdlib>
 #include <optional>
 #include <string>
+#include <thread>
 #include <utility>
 
 using namespace etl::impl;
 using testing::MockFunction;
 using testing::StrictMock;
 
-struct SubscriptionSourceConnectionTests : public NoLoggerFixture {
-    SubscriptionSourceConnectionTests()
+struct SubscriptionSourceConnectionTestsBase : public NoLoggerFixture {
+    SubscriptionSourceConnectionTestsBase()
     {
         subscriptionSource_.run();
     }
 
     boost::asio::io_context ioContext_;
-
-    TestWsServer wsServer_{ioContext_, "0.0.0.0", 11113};
+    TestWsServer wsServer_{ioContext_, "0.0.0.0"};
 
     StrictMockNetworkValidatedLedgersPtr networkValidatedLedgers_;
     StrictMockSubscriptionManagerSharedPtr subscriptionManager_;
 
     StrictMock<MockFunction<void()>> onConnectHook_;
-    StrictMock<MockFunction<void()>> onDisconnectHook_;
+    StrictMock<MockFunction<void(bool)>> onDisconnectHook_;
     StrictMock<MockFunction<void()>> onLedgerClosedHook_;
 
     SubscriptionSource subscriptionSource_{
         ioContext_,
         "127.0.0.1",
-        "11113",
+        wsServer_.port(),
         networkValidatedLedgers_,
         subscriptionManager_,
         onConnectHook_.AsStdFunction(),
         onDisconnectHook_.AsStdFunction(),
         onLedgerClosedHook_.AsStdFunction(),
-        std::chrono::milliseconds(1),
-        std::chrono::milliseconds(1)
+        std::chrono::milliseconds(5),
+        std::chrono::milliseconds(5)
     };
 
     [[maybe_unused]] TestWsConnection
@@ -91,15 +96,17 @@ struct SubscriptionSourceConnectionTests : public NoLoggerFixture {
     }
 };
 
+struct SubscriptionSourceConnectionTests : util::prometheus::WithPrometheus, SubscriptionSourceConnectionTestsBase {};
+
 TEST_F(SubscriptionSourceConnectionTests, ConnectionFailed)
 {
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 }
 
 TEST_F(SubscriptionSourceConnectionTests, ConnectionFailed_Retry_ConnectionFailed)
 {
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([]() {}).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([]() {}).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 }
 
@@ -111,7 +118,19 @@ TEST_F(SubscriptionSourceConnectionTests, ReadError)
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() { subscriptionSource_.stop(); });
+    ioContext_.run();
+}
+
+TEST_F(SubscriptionSourceConnectionTests, ReadTimeout)
+{
+    boost::asio::spawn(ioContext_, [this](boost::asio::yield_context yield) {
+        auto connection = serverConnection(yield);
+        std::this_thread::sleep_for(std::chrono::milliseconds{10});
+    });
+
+    EXPECT_CALL(onConnectHook_, Call());
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 }
 
@@ -125,7 +144,7 @@ TEST_F(SubscriptionSourceConnectionTests, ReadError_Reconnect)
     });
 
     EXPECT_CALL(onConnectHook_, Call()).Times(2);
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([]() {}).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([]() {}).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 }
 
@@ -138,14 +157,14 @@ TEST_F(SubscriptionSourceConnectionTests, IsConnected)
     });
 
     EXPECT_CALL(onConnectHook_, Call()).WillOnce([this]() { EXPECT_TRUE(subscriptionSource_.isConnected()); });
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() {
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() {
         EXPECT_FALSE(subscriptionSource_.isConnected());
         subscriptionSource_.stop();
     });
     ioContext_.run();
 }
 
-struct SubscriptionSourceReadTests : public SubscriptionSourceConnectionTests {
+struct SubscriptionSourceReadTestsBase : public SubscriptionSourceConnectionTestsBase {
     [[maybe_unused]] TestWsConnection
     connectAndSendMessage(std::string const message, boost::asio::yield_context yield)
     {
@@ -156,6 +175,8 @@ struct SubscriptionSourceReadTests : public SubscriptionSourceConnectionTests {
     }
 };
 
+struct SubscriptionSourceReadTests : util::prometheus::WithPrometheus, SubscriptionSourceReadTestsBase {};
+
 TEST_F(SubscriptionSourceReadTests, GotWrongMessage_Reconnect)
 {
     boost::asio::spawn(ioContext_, [this](boost::asio::yield_context yield) {
@@ -166,7 +187,7 @@ TEST_F(SubscriptionSourceReadTests, GotWrongMessage_Reconnect)
     });
 
     EXPECT_CALL(onConnectHook_, Call()).Times(2);
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([]() {}).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([]() {}).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 }
 
@@ -178,7 +199,7 @@ TEST_F(SubscriptionSourceReadTests, GotResult)
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 }
 
@@ -190,7 +211,7 @@ TEST_F(SubscriptionSourceReadTests, GotResultWithLedgerIndex)
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() { subscriptionSource_.stop(); });
     EXPECT_CALL(*networkValidatedLedgers_, push(123));
     ioContext_.run();
 }
@@ -205,7 +226,7 @@ TEST_F(SubscriptionSourceReadTests, GotResultWithLedgerIndexAsString_Reconnect)
     });
 
     EXPECT_CALL(onConnectHook_, Call()).Times(2);
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([]() {}).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([]() {}).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 }
 
@@ -219,7 +240,7 @@ TEST_F(SubscriptionSourceReadTests, GotResultWithValidatedLedgersAsNumber_Reconn
     });
 
     EXPECT_CALL(onConnectHook_, Call()).Times(2);
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([]() {}).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([]() {}).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 }
 
@@ -241,7 +262,7 @@ TEST_F(SubscriptionSourceReadTests, GotResultWithValidatedLedgers)
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 
     EXPECT_TRUE(subscriptionSource_.hasLedger(123));
@@ -267,7 +288,7 @@ TEST_F(SubscriptionSourceReadTests, GotResultWithValidatedLedgersWrongValue_Reco
     });
 
     EXPECT_CALL(onConnectHook_, Call()).Times(2);
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([]() {}).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([]() {}).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 }
 
@@ -285,7 +306,7 @@ TEST_F(SubscriptionSourceReadTests, GotResultWithLedgerIndexAndValidatedLedgers)
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() { subscriptionSource_.stop(); });
     EXPECT_CALL(*networkValidatedLedgers_, push(123));
     ioContext_.run();
 
@@ -305,7 +326,7 @@ TEST_F(SubscriptionSourceReadTests, GotLedgerClosed)
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 }
 
@@ -320,7 +341,7 @@ TEST_F(SubscriptionSourceReadTests, GotLedgerClosedForwardingIsSet)
 
     EXPECT_CALL(onConnectHook_, Call());
     EXPECT_CALL(onLedgerClosedHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() {
+    EXPECT_CALL(onDisconnectHook_, Call(true)).WillOnce([this]() {
         EXPECT_FALSE(subscriptionSource_.isForwarding());
         subscriptionSource_.stop();
     });
@@ -335,7 +356,7 @@ TEST_F(SubscriptionSourceReadTests, GotLedgerClosedWithLedgerIndex)
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() { subscriptionSource_.stop(); });
     EXPECT_CALL(*networkValidatedLedgers_, push(123));
     ioContext_.run();
 }
@@ -350,7 +371,7 @@ TEST_F(SubscriptionSourceReadTests, GotLedgerClosedWithLedgerIndexAsString_Recon
     });
 
     EXPECT_CALL(onConnectHook_, Call()).Times(2);
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([]() {}).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([]() {}).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 }
 
@@ -364,7 +385,7 @@ TEST_F(SubscriptionSourceReadTests, GorLedgerClosedWithValidatedLedgersAsNumber_
     });
 
     EXPECT_CALL(onConnectHook_, Call()).Times(2);
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([]() {}).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([]() {}).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 }
 
@@ -381,7 +402,7 @@ TEST_F(SubscriptionSourceReadTests, GotLedgerClosedWithValidatedLedgers)
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 
     EXPECT_FALSE(subscriptionSource_.hasLedger(0));
@@ -405,7 +426,7 @@ TEST_F(SubscriptionSourceReadTests, GotLedgerClosedWithLedgerIndexAndValidatedLe
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() { subscriptionSource_.stop(); });
     EXPECT_CALL(*networkValidatedLedgers_, push(123));
     ioContext_.run();
 
@@ -424,7 +445,7 @@ TEST_F(SubscriptionSourceReadTests, GotTransactionIsForwardingFalse)
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 }
 
@@ -439,7 +460,7 @@ TEST_F(SubscriptionSourceReadTests, GotTransactionIsForwardingTrue)
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(true)).WillOnce([this]() { subscriptionSource_.stop(); });
     EXPECT_CALL(*subscriptionManager_, forwardProposedTransaction(message));
     ioContext_.run();
 }
@@ -455,7 +476,7 @@ TEST_F(SubscriptionSourceReadTests, GotTransactionWithMetaIsForwardingFalse)
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(true)).WillOnce([this]() { subscriptionSource_.stop(); });
     EXPECT_CALL(*subscriptionManager_, forwardProposedTransaction(message)).Times(0);
     ioContext_.run();
 }
@@ -468,7 +489,7 @@ TEST_F(SubscriptionSourceReadTests, GotValidationReceivedIsForwardingFalse)
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 }
 
@@ -483,7 +504,7 @@ TEST_F(SubscriptionSourceReadTests, GotValidationReceivedIsForwardingTrue)
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(true)).WillOnce([this]() { subscriptionSource_.stop(); });
     EXPECT_CALL(*subscriptionManager_, forwardValidation(message));
     ioContext_.run();
 }
@@ -496,7 +517,7 @@ TEST_F(SubscriptionSourceReadTests, GotManiefstReceivedIsForwardingFalse)
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 }
 
@@ -511,7 +532,7 @@ TEST_F(SubscriptionSourceReadTests, GotManifestReceivedIsForwardingTrue)
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(true)).WillOnce([this]() { subscriptionSource_.stop(); });
     EXPECT_CALL(*subscriptionManager_, forwardManifest(message));
     ioContext_.run();
 }
@@ -524,7 +545,7 @@ TEST_F(SubscriptionSourceReadTests, LastMessageTime)
     });
 
     EXPECT_CALL(onConnectHook_, Call());
-    EXPECT_CALL(onDisconnectHook_, Call()).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() { subscriptionSource_.stop(); });
     ioContext_.run();
 
     auto const actualLastTimeMessage = subscriptionSource_.lastMessageTime();
@@ -532,3 +553,27 @@ TEST_F(SubscriptionSourceReadTests, LastMessageTime)
     auto const diff = std::chrono::duration_cast<std::chrono::milliseconds>(now - actualLastTimeMessage);
     EXPECT_LT(diff, std::chrono::milliseconds(100));
 }
+
+struct SubscriptionSourcePrometheusCounterTests : util::prometheus::WithMockPrometheus,
+                                                  SubscriptionSourceReadTestsBase {};
+
+TEST_F(SubscriptionSourcePrometheusCounterTests, LastMessageTime)
+{
+    auto& lastMessageTimeMock = makeMock<util::prometheus::GaugeInt>(
+        "subscription_source_last_message_time", fmt::format("{{source=\"127.0.0.1:{}\"}}", wsServer_.port())
+    );
+    boost::asio::spawn(ioContext_, [this](boost::asio::yield_context yield) {
+        auto connection = connectAndSendMessage("some_message", yield);
+        connection.close(yield);
+    });
+
+    EXPECT_CALL(onConnectHook_, Call());
+    EXPECT_CALL(onDisconnectHook_, Call(false)).WillOnce([this]() { subscriptionSource_.stop(); });
+    EXPECT_CALL(lastMessageTimeMock, set).WillOnce([](int64_t value) {
+        auto const now =
+            std::chrono::duration_cast<std::chrono::seconds>(std::chrono::system_clock::now().time_since_epoch())
+                .count();
+        EXPECT_LE(now - value, 1);
+    });
+    ioContext_.run();
+}

tests/unit/rpc/BaseTests.cpp

@@ -426,6 +426,9 @@ TEST_F(RPCBaseTest, AccountValidator)
     failingInput = json::parse(R"({ "account": "02000000000000000000000000000000000000000000000000000000000000000" })");
     ASSERT_FALSE(spec.process(failingInput));
 
+    failingInput = json::parse(R"({ "account": "rf1BiGeXwwQoi8Z2ueFYTEXSwuJYfV2Jp?" })");
+    ASSERT_FALSE(spec.process(failingInput));
+
     auto passingInput = json::parse(R"({ "account": "rf1BiGeXwwQoi8Z2ueFYTEXSwuJYfV2Jpn" })");
     ASSERT_TRUE(spec.process(passingInput));
 
@@ -434,6 +437,28 @@ TEST_F(RPCBaseTest, AccountValidator)
     ASSERT_TRUE(spec.process(passingInput));
 }
 
+TEST_F(RPCBaseTest, AccountBase58Validator)
+{
+    auto spec = RpcSpec{
+        {"account", validation::AccountBase58Validator},
+    };
+    auto failingInput = json::parse(R"({ "account": 256 })");
+    ASSERT_FALSE(spec.process(failingInput));
+
+    failingInput = json::parse(R"({ "account": "rf1BiGeXwwQoi8Z2ueFYTEXSwuJYfV2Jp" })");
+    ASSERT_FALSE(spec.process(failingInput));
+
+    failingInput =
+        json::parse(R"({ "account": "020000000000000000000000000000000000000000000000000000000000000000" })");
+    ASSERT_FALSE(spec.process(failingInput));
+
+    failingInput = json::parse(R"({ "account": "rf1BiGeXwwQoi8Z2ueFYTEXSwuJYfV2Jp?" })");
+    ASSERT_FALSE(spec.process(failingInput));
+
+    auto passingInput = json::parse(R"({ "account": "rf1BiGeXwwQoi8Z2ueFYTEXSwuJYfV2Jpn" })");
+    ASSERT_TRUE(spec.process(passingInput));
+}
+
 TEST_F(RPCBaseTest, AccountMarkerValidator)
 {
     auto spec = RpcSpec{

tests/unit/rpc/RPCHelpersTests.cpp

@@ -24,6 +24,7 @@
 #include "rpc/common/Types.hpp"
 #include "util/Fixtures.hpp"
 #include "util/MockPrometheus.hpp"
+#include "util/NameGenerator.hpp"
 #include "util/TestObject.hpp"
 
 #include <boost/asio/impl/spawn.hpp>
@@ -538,3 +539,67 @@ TEST_F(RPCHelpersTest, ParseIssue)
         std::runtime_error
     );
 }
+
+struct IsAdminCmdParamTestCaseBundle {
+    std::string testName;
+    std::string method;
+    std::string testJson;
+    bool expected;
+};
+
+struct IsAdminCmdParameterTest : public TestWithParam<IsAdminCmdParamTestCaseBundle> {};
+
+static auto
+generateTestValuesForParametersTest()
+{
+    return std::vector<IsAdminCmdParamTestCaseBundle>{
+        {"ledgerEntry", "ledger_entry", R"({"type": false})", false},
+
+        {"featureVetoedTrue", "feature", R"({"vetoed": true, "feature": "foo"})", true},
+        {"featureVetoedFalse", "feature", R"({"vetoed": false, "feature": "foo"})", true},
+        {"featureVetoedIsStr", "feature", R"({"vetoed": "String"})", true},
+
+        {"ledger", "ledger", R"({})", false},
+        {"ledgerWithType", "ledger", R"({"type": "fee"})", false},
+        {"ledgerFullTrue", "ledger", R"({"full": true})", true},
+        {"ledgerFullFalse", "ledger", R"({"full": false})", false},
+        {"ledgerFullIsStr", "ledger", R"({"full": "String"})", true},
+        {"ledgerFullIsEmptyStr", "ledger", R"({"full": ""})", false},
+        {"ledgerFullIsNumber1", "ledger", R"({"full": 1})", true},
+        {"ledgerFullIsNumber0", "ledger", R"({"full": 0})", false},
+        {"ledgerFullIsNull", "ledger", R"({"full": null})", false},
+        {"ledgerFullIsFloat0", "ledger", R"({"full": 0.0})", false},
+        {"ledgerFullIsFloat1", "ledger", R"({"full": 0.1})", true},
+        {"ledgerFullIsArray", "ledger", R"({"full": [1]})", true},
+        {"ledgerFullIsEmptyArray", "ledger", R"({"full": []})", false},
+        {"ledgerFullIsObject", "ledger", R"({"full": {"key": 1}})", true},
+        {"ledgerFullIsEmptyObject", "ledger", R"({"full": {}})", false},
+
+        {"ledgerAccountsTrue", "ledger", R"({"accounts": true})", true},
+        {"ledgerAccountsFalse", "ledger", R"({"accounts": false})", false},
+        {"ledgerAccountsIsStr", "ledger", R"({"accounts": "String"})", true},
+        {"ledgerAccountsIsEmptyStr", "ledger", R"({"accounts": ""})", false},
+        {"ledgerAccountsIsNumber1", "ledger", R"({"accounts": 1})", true},
+        {"ledgerAccountsIsNumber0", "ledger", R"({"accounts": 0})", false},
+        {"ledgerAccountsIsNull", "ledger", R"({"accounts": null})", false},
+        {"ledgerAccountsIsFloat0", "ledger", R"({"accounts": 0.0})", false},
+        {"ledgerAccountsIsFloat1", "ledger", R"({"accounts": 0.1})", true},
+        {"ledgerAccountsIsArray", "ledger", R"({"accounts": [1]})", true},
+        {"ledgerAccountsIsEmptyArray", "ledger", R"({"accounts": []})", false},
+        {"ledgerAccountsIsObject", "ledger", R"({"accounts": {"key": 1}})", true},
+        {"ledgerAccountsIsEmptyObject", "ledger", R"({"accounts": {}})", false},
+    };
+}
+
+INSTANTIATE_TEST_CASE_P(
+    IsAdminCmdTest,
+    IsAdminCmdParameterTest,
+    ValuesIn(generateTestValuesForParametersTest()),
+    tests::util::NameGenerator
+);
+
+TEST_P(IsAdminCmdParameterTest, Test)
+{
+    auto const testBundle = GetParam();
+    EXPECT_EQ(isAdminCmd(testBundle.method, boost::json::parse(testBundle.testJson).as_object()), testBundle.expected);
+}

tests/unit/rpc/common/TypesTests.cpp

@@ -20,6 +20,8 @@
 #include "rpc/Errors.hpp"
 #include "rpc/common/Types.hpp"
 
+#include <boost/json/array.hpp>
+#include <boost/json/value.hpp>
 #include <gtest/gtest.h>
 
 #include <expected>
@@ -34,3 +36,46 @@ TEST(MaybeErrorTest, OperatorEquals)
     EXPECT_EQ(MaybeError{std::unexpected{Status{"Error"}}}, MaybeError{std::unexpected{Status{"Error"}}});
     EXPECT_NE(MaybeError{std::unexpected{Status{"Error"}}}, MaybeError{std::unexpected{Status{"Another_error"}}});
 }
+
+TEST(ReturnTypeTests, Constructor)
+{
+    boost::json::value const value{42};
+
+    {
+        ReturnType const r{value};
+        ASSERT_TRUE(r.result);
+        EXPECT_EQ(r.result.value(), value);
+        EXPECT_EQ(r.warnings, boost::json::array{});
+    }
+
+    {
+        boost::json::array const warnings{1, 2, 3};
+        ReturnType const r{value, warnings};
+        ASSERT_TRUE(r.result);
+        EXPECT_EQ(r.result.value(), value);
+        EXPECT_EQ(r.warnings, warnings);
+    }
+
+    {
+        Status const status{"Error"};
+
+        ReturnType const r{std::unexpected{status}};
+        ASSERT_FALSE(r.result);
+        EXPECT_EQ(r.result.error(), status);
+        EXPECT_EQ(r.warnings, boost::json::array{});
+    }
+}
+
+TEST(ReturnTypeTests, operatorBool)
+{
+    {
+        boost::json::value const value{42};
+        ReturnType const r{value};
+        EXPECT_TRUE(r);
+    }
+    {
+        Status const status{"Error"};
+        ReturnType const r{std::unexpected{status}};
+        EXPECT_FALSE(r);
+    }
+}

tests/unit/rpc/handlers/AccountTxTests.cpp

@@ -769,14 +769,13 @@ TEST_F(RPCAccountTxHandlerTest, LimitAndMarker)
 
     auto const transactions = genTransactions(MINSEQ + 1, MAXSEQ - 1);
     auto const transCursor = TransactionsAndCursor{transactions, TransactionsCursor{12, 34}};
-    ON_CALL(*backend, fetchAccountTransactions).WillByDefault(Return(transCursor));
     EXPECT_CALL(
         *backend,
         fetchAccountTransactions(
             testing::_, testing::_, false, testing::Optional(testing::Eq(TransactionsCursor{10, 11})), testing::_
         )
     )
-        .Times(1);
+        .WillOnce(Return(transCursor));
 
     runSpawn([&, this](auto yield) {
         auto const handler = AnyHandler{AccountTxHandler{backend}};
@@ -804,6 +803,73 @@ TEST_F(RPCAccountTxHandlerTest, LimitAndMarker)
     });
 }
 
+TEST_F(RPCAccountTxHandlerTest, LimitIsCapped)
+{
+    backend->setRange(MINSEQ, MAXSEQ);
+
+    auto const transactions = genTransactions(MINSEQ + 1, MAXSEQ - 1);
+    auto const transCursor = TransactionsAndCursor{transactions, TransactionsCursor{12, 34}};
+    EXPECT_CALL(*backend, fetchAccountTransactions(testing::_, testing::_, false, testing::_, testing::_))
+        .WillOnce(Return(transCursor));
+
+    runSpawn([&, this](auto yield) {
+        auto const handler = AnyHandler{AccountTxHandler{backend}};
+        auto static const input = json::parse(fmt::format(
+            R"({{
+                "account": "{}",
+                "ledger_index_min": {},
+                "ledger_index_max": {},
+                "limit": 100000,
+                "forward": false
+            }})",
+            ACCOUNT,
+            -1,
+            -1
+        ));
+        auto const output = handler.process(input, Context{yield});
+        ASSERT_TRUE(output);
+        EXPECT_EQ(output.result->at("account").as_string(), ACCOUNT);
+        EXPECT_EQ(output.result->at("ledger_index_min").as_uint64(), MINSEQ);
+        EXPECT_EQ(output.result->at("ledger_index_max").as_uint64(), MAXSEQ);
+        EXPECT_EQ(output.result->at("limit").as_uint64(), AccountTxHandler::LIMIT_MAX);
+        EXPECT_EQ(output.result->at("transactions").as_array().size(), 2);
+    });
+}
+
+TEST_F(RPCAccountTxHandlerTest, LimitAllowedUpToCap)
+{
+    backend->setRange(MINSEQ, MAXSEQ);
+
+    auto const transactions = genTransactions(MINSEQ + 1, MAXSEQ - 1);
+    auto const transCursor = TransactionsAndCursor{transactions, TransactionsCursor{12, 34}};
+    EXPECT_CALL(*backend, fetchAccountTransactions(testing::_, testing::_, false, testing::_, testing::_))
+        .WillOnce(Return(transCursor));
+
+    runSpawn([&, this](auto yield) {
+        auto const handler = AnyHandler{AccountTxHandler{backend}};
+        auto static const input = json::parse(fmt::format(
+            R"({{
+                "account": "{}",
+                "ledger_index_min": {},
+                "ledger_index_max": {},
+                "limit": {},
+                "forward": false
+            }})",
+            ACCOUNT,
+            -1,
+            -1,
+            AccountTxHandler::LIMIT_MAX - 1
+        ));
+        auto const output = handler.process(input, Context{yield});
+        ASSERT_TRUE(output);
+        EXPECT_EQ(output.result->at("account").as_string(), ACCOUNT);
+        EXPECT_EQ(output.result->at("ledger_index_min").as_uint64(), MINSEQ);
+        EXPECT_EQ(output.result->at("ledger_index_max").as_uint64(), MAXSEQ);
+        EXPECT_EQ(output.result->at("limit").as_uint64(), AccountTxHandler::LIMIT_MAX - 1);
+        EXPECT_EQ(output.result->at("transactions").as_array().size(), 2);
+    });
+}
+
 TEST_F(RPCAccountTxHandlerTest, SpecificLedgerIndex)
 {
     backend->setRange(MINSEQ, MAXSEQ);

tests/unit/rpc/handlers/GetAggregatePriceTests.cpp

@@ -144,6 +144,54 @@ generateTestValuesForParametersTest()
             "invalidParams",
             "Required field 'base_asset' missing"
         },
+        GetAggregatePriceParamTestCaseBundle{
+            "invalid_base_asset",
+            R"({
+                    "quote_asset" : "USD",
+                    "base_asset": "asdf",
+                    "oracles": 
+                    [
+                        {
+                            "account": "rGh1VZCRBJY6rJiaFpD4LZtyHiuCkC8aeD",
+                            "oracle_document_id": 2
+                        }
+                    ]
+                })",
+            "invalidParams",
+            "Invalid parameters."
+        },
+        GetAggregatePriceParamTestCaseBundle{
+            "emtpy_base_asset",
+            R"({
+                    "quote_asset" : "USD",
+                    "base_asset": "",
+                    "oracles": 
+                    [
+                        {
+                            "account": "rGh1VZCRBJY6rJiaFpD4LZtyHiuCkC8aeD",
+                            "oracle_document_id": 2
+                        }
+                    ]
+                })",
+            "invalidParams",
+            "Invalid parameters."
+        },
+        GetAggregatePriceParamTestCaseBundle{
+            "invalid_base_asset2",
+            R"({
+                    "quote_asset" : "USD",
+                    "base_asset": "+aa",
+                    "oracles": 
+                    [
+                        {
+                            "account": "rGh1VZCRBJY6rJiaFpD4LZtyHiuCkC8aeD",
+                            "oracle_document_id": 2
+                        }
+                    ]
+                })",
+            "invalidParams",
+            "Invalid parameters."
+        },
         GetAggregatePriceParamTestCaseBundle{
             "no_quote_asset",
             R"({
@@ -175,6 +223,22 @@ generateTestValuesForParametersTest()
             "invalidParams",
             "Invalid parameters."
         },
+        GetAggregatePriceParamTestCaseBundle{
+            "empty_quote_asset",
+            R"({
+                    "quote_asset" : "",
+                    "base_asset": "USD",
+                    "oracles": 
+                    [
+                        {
+                            "account": "rGh1VZCRBJY6rJiaFpD4LZtyHiuCkC8aeD",
+                            "oracle_document_id": 2
+                        }
+                    ]
+                })",
+            "invalidParams",
+            "Invalid parameters."
+        },
         GetAggregatePriceParamTestCaseBundle{
             "invalid_quote_asset2",
             R"({

tests/unit/rpc/handlers/LedgerTests.cpp

@@ -79,25 +79,25 @@ generateTestValuesForParametersTest()
             "AccountsInvalidBool",
             R"({"accounts": true})",
             "notSupported",
-            "Not supported field 'accounts'",
+            "Not supported field 'accounts's value 'true'",
         },
         {
             "AccountsInvalidInt",
             R"({"accounts": 123})",
-            "notSupported",
-            "Not supported field 'accounts'",
+            "invalidParams",
+            "Invalid parameters.",
         },
         {
             "FullInvalidBool",
             R"({"full": true})",
             "notSupported",
-            "Not supported field 'full'",
+            "Not supported field 'full's value 'true'",
         },
         {
             "FullInvalidInt",
             R"({"full": 123})",
-            "notSupported",
-            "Not supported field 'full'",
+            "invalidParams",
+            "Invalid parameters.",
         },
         {
             "QueueExist",
@@ -304,6 +304,8 @@ TEST_F(RPCLedgerHandlerTest, ConditionallyNotSupportedFieldsDefaultValue)
         auto const handler = AnyHandler{LedgerHandler{backend}};
         auto const req = json::parse(
             R"({
+                "full": false,
+                "accounts": false,
                 "queue": false
             })"
         );

tests/unit/util/AccountUtilsTests.cpp

@@ -0,0 +1,40 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2024, 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 "util/AccountUtils.hpp"
+
+#include <gtest/gtest.h>
+#include <ripple/protocol/AccountID.h>
+#include <ripple/protocol/SecretKey.h>
+#include <ripple/protocol/tokens.h>
+
+constexpr static auto ACCOUNT = "rf1BiGeXwwQoi8Z2ueFYTEXSwuJYfV2Jpn";
+
+TEST(AccountUtils, parseBase58Wrapper)
+{
+    EXPECT_FALSE(util::parseBase58Wrapper<ripple::AccountID>("rf1BiGeXwwQoi8Z2ueFYTEXSwuJYfV2Jp!"));
+    EXPECT_TRUE(util::parseBase58Wrapper<ripple::AccountID>(ACCOUNT));
+
+    EXPECT_TRUE(util::parseBase58Wrapper<ripple::SecretKey>(
+        ripple::TokenType::NodePrivate, "paQmjZ37pKKPMrgadBLsuf9ab7Y7EUNzh27LQrZqoexpAs31nJi"
+    ));
+    EXPECT_FALSE(util::parseBase58Wrapper<ripple::SecretKey>(
+        ripple::TokenType::NodePrivate, "??paQmjZ37pKKPMrgadBLsuf9ab7Y7EUNzh27LQrZqoexpAs31n"
+    ));
+}

tests/unit/util/requests/RequestBuilderTests.cpp

@@ -17,6 +17,7 @@
 */
 //==============================================================================
 
+#include "util/AssignRandomPort.hpp"
 #include "util/Fixtures.hpp"
 #include "util/TestHttpServer.hpp"
 #include "util/requests/RequestBuilder.hpp"
@@ -31,6 +32,7 @@
 #include <gtest/gtest.h>
 
 #include <chrono>
+#include <cstdint>
 #include <expected>
 #include <optional>
 #include <string>
@@ -50,8 +52,8 @@ struct RequestBuilderTestBundle {
 };
 
 struct RequestBuilderTestBase : SyncAsioContextTest {
-    TestHttpServer server{ctx, "0.0.0.0", 11111};
-    RequestBuilder builder{"localhost", "11111"};
+    TestHttpServer server{ctx, "0.0.0.0"};
+    RequestBuilder builder{"localhost", server.port()};
 };
 
 struct RequestBuilderTest : RequestBuilderTestBase, testing::WithParamInterface<RequestBuilderTestBundle> {};
@@ -182,7 +184,7 @@ TEST_F(RequestBuilderTest, ResolveError)
 
 TEST_F(RequestBuilderTest, ConnectionError)
 {
-    builder = RequestBuilder{"localhost", "11112"};
+    builder = RequestBuilder{"localhost", std::to_string(tests::util::generateFreePort())};
     builder.setTimeout(std::chrono::milliseconds{1});
     runSpawn([this](asio::yield_context yield) {
         auto const response = builder.getPlain(yield);

tests/unit/util/requests/WsConnectionTests.cpp

@@ -17,6 +17,7 @@
 */
 //==============================================================================
 
+#include "util/AssignRandomPort.hpp"
 #include "util/Fixtures.hpp"
 #include "util/TestWsServer.hpp"
 #include "util/requests/Types.hpp"
@@ -29,6 +30,7 @@
 
 #include <chrono>
 #include <cstddef>
+#include <cstdint>
 #include <expected>
 #include <memory>
 #include <optional>
@@ -41,8 +43,8 @@ namespace asio = boost::asio;
 namespace http = boost::beast::http;
 
 struct WsConnectionTestsBase : SyncAsioContextTest {
-    WsConnectionBuilder builder{"localhost", "11112"};
-    TestWsServer server{ctx, "0.0.0.0", 11112};
+    TestWsServer server{ctx, "0.0.0.0"};
+    WsConnectionBuilder builder{"localhost", server.port()};
 
     template <typename T, typename E>
     T

tests/unit/web/ServerTests.cpp

@@ -17,10 +17,12 @@
 */
 //==============================================================================
 
+#include "util/AssignRandomPort.hpp"
 #include "util/Fixtures.hpp"
 #include "util/MockPrometheus.hpp"
 #include "util/TestHttpSyncClient.hpp"
 #include "util/config/Config.hpp"
+#include "util/prometheus/Gauge.hpp"
 #include "util/prometheus/Label.hpp"
 #include "util/prometheus/Prometheus.hpp"
 #include "web/DOSGuard.hpp"
@@ -41,15 +43,20 @@
 #include <boost/json/parse.hpp>
 #include <boost/system/system_error.hpp>
 #include <fmt/core.h>
+#include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
+#include <chrono>
 #include <condition_variable>
+#include <cstdint>
 #include <functional>
+#include <iostream>
 #include <memory>
 #include <mutex>
 #include <optional>
 #include <stdexcept>
 #include <string>
+#include <string_view>
 #include <thread>
 #include <utility>
 #include <vector>
@@ -57,37 +64,48 @@
 using namespace util;
 using namespace web::impl;
 using namespace web;
+using namespace boost::json;
 
-constexpr static auto JSONData = R"JSON(
-    {
-        "server":{
-            "ip":"0.0.0.0",
-            "port":8888
-        },
-        "dos_guard": {
-            "max_fetches": 100,
-            "sweep_interval": 1000,
-            "max_connections": 2,
-            "max_requests": 3,
-            "whitelist": ["127.0.0.1"]
-        }
-    }
-)JSON";
+std::string
+generateJSONWithDynamicPort(std::string_view port)
+{
+    return fmt::format(
+        R"JSON({{
+            "server": {{
+                "ip": "0.0.0.0",
+                "port": {}
+            }},
+            "dos_guard": {{
+                "max_fetches": 100,
+                "sweep_interval": 1000,
+                "max_connections": 2,
+                "max_requests": 3,
+                "whitelist": ["127.0.0.1"]
+            }}
+        }})JSON",
+        port
+    );
+}
 
-constexpr static auto JSONDataOverload = R"JSON(
-    {
-        "server":{
-            "ip":"0.0.0.0",
-            "port":8888
-        },
-        "dos_guard": {
-            "max_fetches": 100,
-            "sweep_interval": 1000,
-            "max_connections": 2,
-            "max_requests": 1
-        }
-    }
-)JSON";
+std::string
+generateJSONDataOverload(std::string_view port)
+{
+    return fmt::format(
+        R"JSON({{
+            "server": {{
+                "ip": "0.0.0.0",
+                "port": {}
+            }},
+            "dos_guard": {{
+                "max_fetches": 100,
+                "sweep_interval": 1000,
+                "max_connections": 2,
+                "max_requests": 1
+            }}
+        }})JSON",
+        port
+    );
+}
 
 // for testing, we use a self-signed certificate
 std::optional<ssl::context>
@@ -168,12 +186,13 @@ protected:
 
     // this ctx is for dos timer
     boost::asio::io_context ctxSync;
-    Config cfg{boost::json::parse(JSONData)};
+    std::string const port = std::to_string(tests::util::generateFreePort());
+    Config cfg{parse(generateJSONWithDynamicPort(port))};
     IntervalSweepHandler sweepHandler = web::IntervalSweepHandler{cfg, ctxSync};
     WhitelistHandler whitelistHandler = web::WhitelistHandler{cfg};
     DOSGuard dosGuard = web::DOSGuard{cfg, whitelistHandler, sweepHandler};
 
-    Config cfgOverload{boost::json::parse(JSONDataOverload)};
+    Config cfgOverload{parse(generateJSONDataOverload(port))};
     IntervalSweepHandler sweepHandlerOverload = web::IntervalSweepHandler{cfgOverload, ctxSync};
     WhitelistHandler whitelistHandlerOverload = web::WhitelistHandler{cfgOverload};
     DOSGuard dosGuardOverload = web::DOSGuard{cfgOverload, whitelistHandlerOverload, sweepHandlerOverload};
@@ -185,6 +204,8 @@ private:
     std::optional<std::thread> runner;
 };
 
+struct WebServerTestsWithMockPrometheus : WebServerTest, prometheus::WithMockPrometheus {};
+
 class EchoExecutor {
 public:
     void
@@ -246,42 +267,52 @@ makeServerSync(
 
 }  // namespace
 
-TEST_F(WebServerTest, Http)
+TEST_F(WebServerTestsWithMockPrometheus, Http)
 {
     auto e = std::make_shared<EchoExecutor>();
     auto const server = makeServerSync(cfg, ctx, std::nullopt, dosGuard, e);
-    auto const res = HttpSyncClient::syncPost("localhost", "8888", R"({"Hello":1})");
+    auto const res = HttpSyncClient::syncPost("localhost", port, R"({"Hello":1})");
     EXPECT_EQ(res, R"({"Hello":1})");
 }
 
-TEST_F(WebServerTest, Ws)
+TEST_F(WebServerTestsWithMockPrometheus, Ws)
 {
+    ::testing::StrictMock<util::prometheus::MockCounterImplInt>& wsMessagesCounterMock =
+        makeMock<util::prometheus::GaugeInt>("ws_messages_length", "");
+    EXPECT_CALL(wsMessagesCounterMock, add(1));
+    EXPECT_CALL(wsMessagesCounterMock, add(-1));
+
     auto e = std::make_shared<EchoExecutor>();
     auto const server = makeServerSync(cfg, ctx, std::nullopt, dosGuard, e);
     WebSocketSyncClient wsClient;
-    wsClient.connect("localhost", "8888");
+    wsClient.connect("localhost", port);
     auto const res = wsClient.syncPost(R"({"Hello":1})");
     EXPECT_EQ(res, R"({"Hello":1})");
     wsClient.disconnect();
 }
 
-TEST_F(WebServerTest, HttpInternalError)
+TEST_F(WebServerTestsWithMockPrometheus, HttpInternalError)
 {
     auto e = std::make_shared<ExceptionExecutor>();
     auto const server = makeServerSync(cfg, ctx, std::nullopt, dosGuard, e);
-    auto const res = HttpSyncClient::syncPost("localhost", "8888", R"({})");
+    auto const res = HttpSyncClient::syncPost("localhost", port, R"({})");
     EXPECT_EQ(
         res,
         R"({"error":"internal","error_code":73,"error_message":"Internal error.","status":"error","type":"response"})"
     );
 }
 
-TEST_F(WebServerTest, WsInternalError)
+TEST_F(WebServerTestsWithMockPrometheus, WsInternalError)
 {
+    ::testing::StrictMock<util::prometheus::MockCounterImplInt>& wsMessagesCounterMock =
+        makeMock<util::prometheus::GaugeInt>("ws_messages_length", "");
+    EXPECT_CALL(wsMessagesCounterMock, add(1));
+    EXPECT_CALL(wsMessagesCounterMock, add(-1));
+
     auto e = std::make_shared<ExceptionExecutor>();
     auto const server = makeServerSync(cfg, ctx, std::nullopt, dosGuard, e);
     WebSocketSyncClient wsClient;
-    wsClient.connect("localhost", "8888");
+    wsClient.connect("localhost", port);
     auto const res = wsClient.syncPost(R"({"id":"id1"})");
     wsClient.disconnect();
     EXPECT_EQ(
@@ -290,12 +321,17 @@ TEST_F(WebServerTest, WsInternalError)
     );
 }
 
-TEST_F(WebServerTest, WsInternalErrorNotJson)
+TEST_F(WebServerTestsWithMockPrometheus, WsInternalErrorNotJson)
 {
+    ::testing::StrictMock<util::prometheus::MockCounterImplInt>& wsMessagesCounterMock =
+        makeMock<util::prometheus::GaugeInt>("ws_messages_length", "");
+    EXPECT_CALL(wsMessagesCounterMock, add(1));
+    EXPECT_CALL(wsMessagesCounterMock, add(-1));
+
     auto e = std::make_shared<ExceptionExecutor>();
     auto const server = makeServerSync(cfg, ctx, std::nullopt, dosGuard, e);
     WebSocketSyncClient wsClient;
-    wsClient.connect("localhost", "8888");
+    wsClient.connect("localhost", port);
     auto const res = wsClient.syncPost("not json");
     wsClient.disconnect();
     EXPECT_EQ(
@@ -304,54 +340,64 @@ TEST_F(WebServerTest, WsInternalErrorNotJson)
     );
 }
 
-TEST_F(WebServerTest, Https)
+TEST_F(WebServerTestsWithMockPrometheus, Https)
 {
     auto e = std::make_shared<EchoExecutor>();
     auto sslCtx = parseCertsForTest();
     auto const ctxSslRef = sslCtx ? std::optional<std::reference_wrapper<ssl::context>>{sslCtx.value()} : std::nullopt;
     auto const server = makeServerSync(cfg, ctx, ctxSslRef, dosGuard, e);
-    auto const res = HttpsSyncClient::syncPost("localhost", "8888", R"({"Hello":1})");
+    auto const res = HttpsSyncClient::syncPost("localhost", port, R"({"Hello":1})");
     EXPECT_EQ(res, R"({"Hello":1})");
 }
 
-TEST_F(WebServerTest, Wss)
+TEST_F(WebServerTestsWithMockPrometheus, Wss)
 {
+    ::testing::StrictMock<util::prometheus::MockCounterImplInt>& wsMessagesCounterMock =
+        makeMock<util::prometheus::GaugeInt>("ws_messages_length", "");
+    EXPECT_CALL(wsMessagesCounterMock, add(1));
+    EXPECT_CALL(wsMessagesCounterMock, add(-1));
+
     auto e = std::make_shared<EchoExecutor>();
     auto sslCtx = parseCertsForTest();
     auto const ctxSslRef = sslCtx ? std::optional<std::reference_wrapper<ssl::context>>{sslCtx.value()} : std::nullopt;
 
     auto server = makeServerSync(cfg, ctx, ctxSslRef, dosGuard, e);
     WebServerSslSyncClient wsClient;
-    wsClient.connect("localhost", "8888");
+    wsClient.connect("localhost", port);
     auto const res = wsClient.syncPost(R"({"Hello":1})");
     EXPECT_EQ(res, R"({"Hello":1})");
     wsClient.disconnect();
 }
 
-TEST_F(WebServerTest, HttpRequestOverload)
+TEST_F(WebServerTestsWithMockPrometheus, HttpRequestOverload)
 {
     auto e = std::make_shared<EchoExecutor>();
     auto const server = makeServerSync(cfg, ctx, std::nullopt, dosGuardOverload, e);
-    auto res = HttpSyncClient::syncPost("localhost", "8888", R"({})");
+    auto res = HttpSyncClient::syncPost("localhost", port, R"({})");
     EXPECT_EQ(res, "{}");
-    res = HttpSyncClient::syncPost("localhost", "8888", R"({})");
+    res = HttpSyncClient::syncPost("localhost", port, R"({})");
     EXPECT_EQ(
         res,
         R"({"error":"slowDown","error_code":10,"error_message":"You are placing too much load on the server.","status":"error","type":"response"})"
     );
 }
 
-TEST_F(WebServerTest, WsRequestOverload)
+TEST_F(WebServerTestsWithMockPrometheus, WsRequestOverload)
 {
+    ::testing::StrictMock<util::prometheus::MockCounterImplInt>& wsMessagesCounterMock =
+        makeMock<util::prometheus::GaugeInt>("ws_messages_length", "");
+    EXPECT_CALL(wsMessagesCounterMock, add(1)).Times(2);
+    EXPECT_CALL(wsMessagesCounterMock, add(-1)).Times(2);
+
     auto e = std::make_shared<EchoExecutor>();
     auto const server = makeServerSync(cfg, ctx, std::nullopt, dosGuardOverload, e);
     WebSocketSyncClient wsClient;
-    wsClient.connect("localhost", "8888");
+    wsClient.connect("localhost", port);
     auto res = wsClient.syncPost(R"({})");
     wsClient.disconnect();
     EXPECT_EQ(res, "{}");
     WebSocketSyncClient wsClient2;
-    wsClient2.connect("localhost", "8888");
+    wsClient2.connect("localhost", port);
     res = wsClient2.syncPost(R"({})");
     wsClient2.disconnect();
     EXPECT_EQ(
@@ -360,25 +406,30 @@ TEST_F(WebServerTest, WsRequestOverload)
     );
 }
 
-TEST_F(WebServerTest, HttpPayloadOverload)
+TEST_F(WebServerTestsWithMockPrometheus, HttpPayloadOverload)
 {
     std::string const s100(100, 'a');
     auto e = std::make_shared<EchoExecutor>();
     auto server = makeServerSync(cfg, ctx, std::nullopt, dosGuardOverload, e);
-    auto const res = HttpSyncClient::syncPost("localhost", "8888", fmt::format(R"({{"payload":"{}"}})", s100));
+    auto const res = HttpSyncClient::syncPost("localhost", port, fmt::format(R"({{"payload":"{}"}})", s100));
     EXPECT_EQ(
         res,
         R"({"payload":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","warning":"load","warnings":[{"id":2003,"message":"You are about to be rate limited"}]})"
     );
 }
 
-TEST_F(WebServerTest, WsPayloadOverload)
+TEST_F(WebServerTestsWithMockPrometheus, WsPayloadOverload)
 {
+    ::testing::StrictMock<util::prometheus::MockCounterImplInt>& wsMessagesCounterMock =
+        makeMock<util::prometheus::GaugeInt>("ws_messages_length", "");
+    EXPECT_CALL(wsMessagesCounterMock, add(1));
+    EXPECT_CALL(wsMessagesCounterMock, add(-1));
+
     std::string const s100(100, 'a');
     auto e = std::make_shared<EchoExecutor>();
     auto server = makeServerSync(cfg, ctx, std::nullopt, dosGuardOverload, e);
     WebSocketSyncClient wsClient;
-    wsClient.connect("localhost", "8888");
+    wsClient.connect("localhost", port);
     auto const res = wsClient.syncPost(fmt::format(R"({{"payload":"{}"}})", s100));
     wsClient.disconnect();
     EXPECT_EQ(
@@ -387,19 +438,19 @@ TEST_F(WebServerTest, WsPayloadOverload)
     );
 }
 
-TEST_F(WebServerTest, WsTooManyConnection)
+TEST_F(WebServerTestsWithMockPrometheus, WsTooManyConnection)
 {
     auto e = std::make_shared<EchoExecutor>();
     auto server = makeServerSync(cfg, ctx, std::nullopt, dosGuardOverload, e);
     // max connection is 2, exception should happen when the third connection is made
     WebSocketSyncClient wsClient1;
-    wsClient1.connect("localhost", "8888");
+    wsClient1.connect("localhost", port);
     WebSocketSyncClient wsClient2;
-    wsClient2.connect("localhost", "8888");
+    wsClient2.connect("localhost", port);
     bool exceptionThrown = false;
     try {
         WebSocketSyncClient wsClient3;
-        wsClient3.connect("localhost", "8888");
+        wsClient3.connect("localhost", port);
     } catch (boost::system::system_error const& ex) {
         exceptionThrown = true;
         EXPECT_EQ(ex.code(), boost::beast::websocket::error::upgrade_declined);
@@ -409,45 +460,65 @@ TEST_F(WebServerTest, WsTooManyConnection)
     EXPECT_TRUE(exceptionThrown);
 }
 
-static auto constexpr JSONServerConfigWithAdminPassword = R"JSON(
-    {
-        "server":{
-            "ip": "0.0.0.0",
-            "port": 8888,
-            "admin_password": "secret"
-        }
-    }
-)JSON";
+std::string
+JSONServerConfigWithAdminPassword(uint32_t const port)
+{
+    return fmt::format(
+        R"JSON({{
+            "server": {{
+                "ip": "0.0.0.0",
+                "port": {},
+                "admin_password": "secret"
+            }}
+        }})JSON",
+        port
+    );
+}
 
-static auto constexpr JSONServerConfigWithLocalAdmin = R"JSON(
-    {
-        "server":{
-            "ip": "0.0.0.0",
-            "port": 8888,
-            "local_admin": true
-        }
-    }
-)JSON";
+std::string
+JSONServerConfigWithLocalAdmin(uint32_t const port)
+{
+    return fmt::format(
+        R"JSON({{
+            "server": {{
+                "ip": "0.0.0.0",
+                "port": {},
+                "local_admin": true
+            }}
+        }})JSON",
+        port
+    );
+}
 
-static auto constexpr JSONServerConfigWithBothAdminPasswordAndLocalAdminFalse = R"JSON(
-    {
-        "server":{
-            "ip": "0.0.0.0",
-            "port": 8888,
-            "admin_password": "secret",
-            "local_admin": false
-        }
-    }
-)JSON";
+std::string
+JSONServerConfigWithBothAdminPasswordAndLocalAdminFalse(uint32_t const port)
+{
+    return fmt::format(
+        R"JSON({{
+            "server": {{
+                "ip": "0.0.0.0",
+                "port": {},
+                "admin_password": "secret",
+                "local_admin": false
+            }}
+        }})JSON",
+        port
+    );
+}
 
-static auto constexpr JSONServerConfigWithNoSpecifiedAdmin = R"JSON(
-    {
-        "server":{
-            "ip": "0.0.0.0",
-            "port": 8888
-        }
-    }
-)JSON";
+std::string
+JSONServerConfigWithNoSpecifiedAdmin(uint32_t const port)
+{
+    return fmt::format(
+        R"JSON({{
+            "server": {{
+                "ip": "0.0.0.0",
+                "port": {}
+            }}
+        }})JSON",
+        port
+    );
+}
 
 // get this value from online sha256 generator
 static auto constexpr SecertSha256 = "2bb80d537b1da3e38bd30361aa855686bde0eacd7162fef6a25fe97bf527a25b";
@@ -473,15 +544,23 @@ struct WebServerAdminTestParams {
     std::string expectedResponse;
 };
 
-class WebServerAdminTest : public WebServerTest, public ::testing::WithParamInterface<WebServerAdminTestParams> {};
+class WebServerAdminTest : public WebServerTest,
+                           public ::testing::WithParamInterface<WebServerAdminTestParams>,
+                           public prometheus::WithMockPrometheus {};
 
 TEST_P(WebServerAdminTest, WsAdminCheck)
 {
+    ::testing::StrictMock<util::prometheus::MockCounterImplInt>& wsMessagesCounterMock =
+        makeMock<util::prometheus::GaugeInt>("ws_messages_length", "");
+    EXPECT_CALL(wsMessagesCounterMock, add(1));
+    EXPECT_CALL(wsMessagesCounterMock, add(-1));
+
     auto e = std::make_shared<AdminCheckExecutor>();
-    Config const serverConfig{boost::json::parse(GetParam().config)};
+    Config const serverConfig{parse(GetParam().config)};
     auto server = makeServerSync(serverConfig, ctx, std::nullopt, dosGuardOverload, e);
     WebSocketSyncClient wsClient;
-    wsClient.connect("localhost", "8888", GetParam().headers);
+    uint32_t webServerPort = serverConfig.value<uint32_t>("server.port");
+    wsClient.connect("localhost", std::to_string(webServerPort), GetParam().headers);
     std::string const request = "Why hello";
     auto const res = wsClient.syncPost(request);
     wsClient.disconnect();
@@ -491,10 +570,11 @@ TEST_P(WebServerAdminTest, WsAdminCheck)
 TEST_P(WebServerAdminTest, HttpAdminCheck)
 {
     auto e = std::make_shared<AdminCheckExecutor>();
-    Config const serverConfig{boost::json::parse(GetParam().config)};
+    Config const serverConfig{parse(GetParam().config)};
     auto server = makeServerSync(serverConfig, ctx, std::nullopt, dosGuardOverload, e);
     std::string const request = "Why hello";
-    auto const res = HttpSyncClient::syncPost("localhost", "8888", request, GetParam().headers);
+    uint32_t webServerPort = serverConfig.value<uint32_t>("server.port");
+    auto const res = HttpSyncClient::syncPost("localhost", std::to_string(webServerPort), request, GetParam().headers);
     EXPECT_EQ(res, fmt::format("{} {}", request, GetParam().expectedResponse));
 }
 
@@ -503,27 +583,27 @@ INSTANTIATE_TEST_CASE_P(
     WebServerAdminTest,
     ::testing::Values(
         WebServerAdminTestParams{
-            .config = JSONServerConfigWithAdminPassword,
+            .config = JSONServerConfigWithAdminPassword(tests::util::generateFreePort()),
             .headers = {},
             .expectedResponse = "user"
         },
         WebServerAdminTestParams{
-            .config = JSONServerConfigWithAdminPassword,
+            .config = JSONServerConfigWithAdminPassword(tests::util::generateFreePort()),
             .headers = {WebHeader(http::field::authorization, "")},
             .expectedResponse = "user"
         },
         WebServerAdminTestParams{
-            .config = JSONServerConfigWithAdminPassword,
+            .config = JSONServerConfigWithAdminPassword(tests::util::generateFreePort()),
             .headers = {WebHeader(http::field::authorization, "s")},
             .expectedResponse = "user"
         },
         WebServerAdminTestParams{
-            .config = JSONServerConfigWithAdminPassword,
+            .config = JSONServerConfigWithAdminPassword(tests::util::generateFreePort()),
             .headers = {WebHeader(http::field::authorization, SecertSha256)},
             .expectedResponse = "user"
         },
         WebServerAdminTestParams{
-            .config = JSONServerConfigWithAdminPassword,
+            .config = JSONServerConfigWithAdminPassword(tests::util::generateFreePort()),
             .headers = {WebHeader(
                 http::field::authorization,
                 fmt::format("{}{}", PasswordAdminVerificationStrategy::passwordPrefix, SecertSha256)
@@ -531,12 +611,12 @@ INSTANTIATE_TEST_CASE_P(
             .expectedResponse = "admin"
         },
         WebServerAdminTestParams{
-            .config = JSONServerConfigWithBothAdminPasswordAndLocalAdminFalse,
+            .config = JSONServerConfigWithBothAdminPasswordAndLocalAdminFalse(tests::util::generateFreePort()),
             .headers = {WebHeader(http::field::authorization, SecertSha256)},
             .expectedResponse = "user"
         },
         WebServerAdminTestParams{
-            .config = JSONServerConfigWithBothAdminPasswordAndLocalAdminFalse,
+            .config = JSONServerConfigWithBothAdminPasswordAndLocalAdminFalse(tests::util::generateFreePort()),
             .headers = {WebHeader(
                 http::field::authorization,
                 fmt::format("{}{}", PasswordAdminVerificationStrategy::passwordPrefix, SecertSha256)
@@ -544,16 +624,20 @@ INSTANTIATE_TEST_CASE_P(
             .expectedResponse = "admin"
         },
         WebServerAdminTestParams{
-            .config = JSONServerConfigWithAdminPassword,
+            .config = JSONServerConfigWithAdminPassword(tests::util::generateFreePort()),
             .headers = {WebHeader(
                 http::field::authentication_info,
                 fmt::format("{}{}", PasswordAdminVerificationStrategy::passwordPrefix, SecertSha256)
             )},
             .expectedResponse = "user"
         },
-        WebServerAdminTestParams{.config = JSONServerConfigWithLocalAdmin, .headers = {}, .expectedResponse = "admin"},
         WebServerAdminTestParams{
-            .config = JSONServerConfigWithNoSpecifiedAdmin,
+            .config = JSONServerConfigWithLocalAdmin(tests::util::generateFreePort()),
+            .headers = {},
+            .expectedResponse = "admin"
+        },
+        WebServerAdminTestParams{
+            .config = JSONServerConfigWithNoSpecifiedAdmin(tests::util::generateFreePort()),
             .headers = {},
             .expectedResponse = "admin"
         }
@@ -563,36 +647,40 @@ INSTANTIATE_TEST_CASE_P(
 
 TEST_F(WebServerTest, AdminErrorCfgTestBothAdminPasswordAndLocalAdminSet)
 {
-    static auto constexpr JSONServerConfigWithBothAdminPasswordAndLocalAdmin = R"JSON(
-        {
-            "server":{
+    uint32_t webServerPort = tests::util::generateFreePort();
+    std::string JSONServerConfigWithBothAdminPasswordAndLocalAdmin = fmt::format(
+        R"JSON({{
+        "server":{{
                 "ip": "0.0.0.0",
-                "port": 8888,
+                "port": {},
                 "admin_password": "secret",
                 "local_admin": true
-            }
-        }
-    )JSON";
+            }}
+    }})JSON",
+        webServerPort
+    );
 
     auto e = std::make_shared<AdminCheckExecutor>();
-    Config const serverConfig{boost::json::parse(JSONServerConfigWithBothAdminPasswordAndLocalAdmin)};
+    Config const serverConfig{parse(JSONServerConfigWithBothAdminPasswordAndLocalAdmin)};
     EXPECT_THROW(web::make_HttpServer(serverConfig, ctx, std::nullopt, dosGuardOverload, e), std::logic_error);
 }
 
 TEST_F(WebServerTest, AdminErrorCfgTestBothAdminPasswordAndLocalAdminFalse)
 {
-    static auto constexpr JSONServerConfigWithNoAdminPasswordAndLocalAdminFalse = R"JSON(
-        {
-            "server":{
-                "ip": "0.0.0.0",
-                "port": 8888,
-                "local_admin": false
-            }
-        }
-    )JSON";
+    uint32_t webServerPort = tests::util::generateFreePort();
+    std::string JSONServerConfigWithNoAdminPasswordAndLocalAdminFalse = fmt::format(
+        R"JSON({{
+        "server": {{
+            "ip": "0.0.0.0",
+            "port": {},
+            "local_admin": false
+        }}
+    }})JSON",
+        webServerPort
+    );
 
     auto e = std::make_shared<AdminCheckExecutor>();
-    Config const serverConfig{boost::json::parse(JSONServerConfigWithNoAdminPasswordAndLocalAdminFalse)};
+    Config const serverConfig{parse(JSONServerConfigWithNoAdminPasswordAndLocalAdminFalse)};
     EXPECT_THROW(web::make_HttpServer(serverConfig, ctx, std::nullopt, dosGuardOverload, e), std::logic_error);
 }
 
@@ -601,32 +689,35 @@ struct WebServerPrometheusTest : util::prometheus::WithPrometheus, WebServerTest
 TEST_F(WebServerPrometheusTest, rejectedWithoutAdminPassword)
 {
     auto e = std::make_shared<EchoExecutor>();
-    Config const serverConfig{boost::json::parse(JSONServerConfigWithAdminPassword)};
+    uint32_t webServerPort = tests::util::generateFreePort();
+    Config const serverConfig{parse(JSONServerConfigWithAdminPassword(webServerPort))};
     auto server = makeServerSync(serverConfig, ctx, std::nullopt, dosGuard, e);
-    auto const res = HttpSyncClient::syncGet("localhost", "8888", "", "/metrics");
+    auto const res = HttpSyncClient::syncGet("localhost", std::to_string(webServerPort), "", "/metrics");
     EXPECT_EQ(res, "Only admin is allowed to collect metrics");
 }
 
 TEST_F(WebServerPrometheusTest, rejectedIfPrometheusIsDisabled)
 {
-    static auto constexpr JSONServerConfigWithDisabledPrometheus = R"JSON(
-        {
-            "server": {
+    uint32_t webServerPort = tests::util::generateFreePort();
+    std::string JSONServerConfigWithDisabledPrometheus = fmt::format(
+        R"JSON({{
+        "server":{{
                 "ip": "0.0.0.0",
-                "port": 8888,
+                "port": {},
                 "admin_password": "secret"
-            },
-            "prometheus": { "enabled": false }
-        }
-    )JSON";
+            }},
+        "prometheus": {{ "enabled": false }}
+    }})JSON",
+        webServerPort
+    );
 
     auto e = std::make_shared<EchoExecutor>();
-    Config const serverConfig{boost::json::parse(JSONServerConfigWithDisabledPrometheus)};
+    Config const serverConfig{parse(JSONServerConfigWithDisabledPrometheus)};
     PrometheusService::init(serverConfig);
     auto server = makeServerSync(serverConfig, ctx, std::nullopt, dosGuard, e);
     auto const res = HttpSyncClient::syncGet(
         "localhost",
-        "8888",
+        std::to_string(webServerPort),
         "",
         "/metrics",
         {WebHeader(
@@ -639,14 +730,15 @@ TEST_F(WebServerPrometheusTest, rejectedIfPrometheusIsDisabled)
 
 TEST_F(WebServerPrometheusTest, validResponse)
 {
+    uint32_t webServerPort = tests::util::generateFreePort();
     auto& testCounter = PrometheusService::counterInt("test_counter", util::prometheus::Labels());
     ++testCounter;
     auto e = std::make_shared<EchoExecutor>();
-    Config const serverConfig{boost::json::parse(JSONServerConfigWithAdminPassword)};
+    Config const serverConfig{parse(JSONServerConfigWithAdminPassword(webServerPort))};
     auto server = makeServerSync(serverConfig, ctx, std::nullopt, dosGuard, e);
     auto const res = HttpSyncClient::syncGet(
         "localhost",
-        "8888",
+        std::to_string(webServerPort),
         "",
         "/metrics",
         {WebHeader(