fix: not forward admin API (#1629 )

To merge to 2.2.3 branch. It is different from the #1628 . I think this branch still forward feature API to rippled.
fix: Subscription source bugs fix (#1626 )
2025-11-04 20:05:51 +00:00 · 2024-09-06 10:37:08 +01:00 · 2024-09-05 14:58:06 +01:00 · 2024-08-29 16:48:04 +01:00 · 2024-08-29 15:59:02 +01:00 · 2024-07-18 15:38:24 +01:00
32 changed files with 462 additions and 141 deletions
--- a/src/etl/LoadBalancer.cpp
+++ b/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);
        }
--- a/src/etl/LoadBalancer.hpp
+++ b/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:
    /**
--- a/src/etl/Source.hpp
+++ b/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;
--- a/src/etl/impl/GrpcSource.cpp
+++ b/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)};
--- a/src/etl/impl/SubscriptionSource.cpp
+++ b/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
--- a/src/etl/impl/SubscriptionSource.hpp
+++ b/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
    );

--- a/src/feed/impl/ProposedTransactionFeed.cpp
+++ b/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();
    });
 }

--- a/src/feed/impl/ProposedTransactionFeed.hpp
+++ b/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"))
    {
    }

--- a/src/feed/impl/SingleFeedBase.cpp
+++ b/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();
    });
 }

--- a/src/feed/impl/SingleFeedBase.hpp
+++ b/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_;
--- a/src/feed/impl/TransactionFeed.cpp
+++ b/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();
        }
    );
 }
--- a/src/feed/impl/TransactionFeed.hpp
+++ b/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"))
    {
    }

--- a/src/feed/impl/Util.hpp
+++ b/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
--- a/src/rpc/RPCEngine.hpp
+++ b/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";
--- a/src/rpc/RPCHelpers.cpp
+++ b/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"
@@ -186,14 +187,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 +800,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 +1273,23 @@ specifiesCurrentOrClosedLedger(boost::json::object const& request)
    return false;
 }

+bool
+isAdminCmd(std::string const& method, boost::json::object const& request)
+{
+    auto const isFieldSet = [&request](auto const field) {
+        return request.contains(field) and request.at(field).is_bool() and request.at(field).as_bool();
+    };
+
+    if (method == JS(ledger)) {
+        if (isFieldSet(JS(full)) or isFieldSet(JS(accounts)) or isFieldSet(JS(type)))
+            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)
 {
--- a/src/rpc/RPCHelpers.hpp
+++ b/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
 *
--- a/src/rpc/common/Validators.cpp
+++ b/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>
@@ -114,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}};

--- a/src/rpc/handlers/GatewayBalances.hpp
+++ b/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 {};
--- a/src/rpc/handlers/GetAggregatePrice.cpp
+++ b/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)))
            )
        });
--- a/src/rpc/handlers/Ledger.hpp
+++ b/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}},
--- a/src/rpc/handlers/LedgerEntry.cpp
+++ b/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;
--- a/src/rpc/handlers/LedgerEntry.hpp
+++ b/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"}};
--- a/src/util/AccountUtils.hpp
+++ b/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
--- a/src/util/requests/impl/WsConnectionImpl.hpp
+++ b/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
--- a/tests/common/util/TestObject.cpp
+++ b/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);
--- a/tests/unit/CMakeLists.txt
+++ b/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
--- a/tests/unit/etl/LoadBalancerTests.cpp
+++ b/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 {
--- a/tests/unit/etl/SubscriptionSourceTests.cpp
+++ b/tests/unit/etl/SubscriptionSourceTests.cpp
@@ -18,31 +18,35 @@
 //==============================================================================

 #include "etl/impl/SubscriptionSource.hpp"
-#include "util/AssignRandomPort.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();
    }
@@ -54,7 +58,7 @@ struct SubscriptionSourceConnectionTests : public NoLoggerFixture {
    StrictMockSubscriptionManagerSharedPtr subscriptionManager_;

    StrictMock<MockFunction<void()>> onConnectHook_;
-    StrictMock<MockFunction<void()>> onDisconnectHook_;
+    StrictMock<MockFunction<void(bool)>> onDisconnectHook_;
    StrictMock<MockFunction<void()>> onLedgerClosedHook_;

    SubscriptionSource subscriptionSource_{
@@ -66,8 +70,8 @@ struct SubscriptionSourceConnectionTests : public NoLoggerFixture {
        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
@@ -92,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();
 }

@@ -112,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();
 }

@@ -126,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();
 }

@@ -139,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)
    {
@@ -157,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) {
@@ -167,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();
 }

@@ -179,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();
 }

@@ -191,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();
 }
@@ -206,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();
 }

@@ -220,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();
 }

@@ -242,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));
@@ -268,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();
 }

@@ -286,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();

@@ -306,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();
 }

@@ -321,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();
    });
@@ -336,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();
 }
@@ -351,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();
 }

@@ -365,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();
 }

@@ -382,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));
@@ -406,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();

@@ -425,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();
 }

@@ -440,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();
 }
@@ -456,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();
 }
@@ -469,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();
 }

@@ -484,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();
 }
@@ -497,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();
 }

@@ -512,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();
 }
@@ -525,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();
@@ -533,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();
+}
--- a/tests/unit/rpc/BaseTests.cpp
+++ b/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{
--- a/tests/unit/rpc/RPCHelpersTests.cpp
+++ b/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,41 @@ 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>{
+        {"featureVetoedTrue", "feature", R"({"vetoed": true, "feature": "foo"})", true},
+        {"featureVetoedFalse", "feature", R"({"vetoed": false, "feature": "foo"})", true},
+        {"ledgerFullTrue", "ledger", R"({"full": true})", true},
+        {"ledgerAccountsTrue", "ledger", R"({"accounts": true})", true},
+        {"ledgerTypeTrue", "ledger", R"({"type": true})", true},
+        {"ledgerFullFalse", "ledger", R"({"full": false})", false},
+        {"ledgerAccountsFalse", "ledger", R"({"accounts": false})", false},
+        {"ledgerTypeFalse", "ledger", R"({"type": false})", false},
+        {"ledgerEntry", "ledger_entry", R"({"type": false})", 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);
+}
--- a/tests/unit/rpc/handlers/LedgerTests.cpp
+++ b/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
            })"
        );
--- a/tests/unit/util/AccountUtilsTests.cpp
+++ b/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"
+    ));
+}
Author	SHA1	Message	Date
cyan317	0054e4b64c	fix: not forward admin API (#1629 ) To merge to 2.2.3 branch. It is different from the #1628 . I think this branch still forward feature API to rippled.	2024-09-06 10:37:08 +01:00
Sergey Kuznetsov	9fe9e7c9d2	fix: Subscription source bugs fix (#1626 ) For #1620. - Add timeouts for websocket operations for connections to rippled. Without these timeouts if connection hangs for some reason, clio wouldn't know the connection is hanging. - Fix potential data race in choosing new subscription source which will forward messages to users. - Optimise switching between subscription sources.	2024-09-05 14:58:06 +01:00
Alex Kremer	2e2740d4c5	feat: Published subscription message counters (#1618 ) This PR adds counters to track the amount of published messages for each subscription stream.	2024-08-29 16:48:04 +01:00
Sergey Kuznetsov	5004dc4e15	fix: Fix logging in SubscriptionSource (#1617 ) For #1616. Later should be ported to develop as well.	2024-08-29 15:59:02 +01:00
cyan317	665fab183a	fix: Add more account check (#1543 ) Make sure all char is alphanumeric for account	2024-07-18 15:38:24 +01:00
Alex Kremer	b65ac67d17	fix: Relax error when `full` or `accounts` set to `false` (#1540 ) Fixes #1537	2024-07-18 15:20:46 +01:00