ARCHIVE/clio
1
0
Fork 0
mirror of https://github.com/XRPLF/clio.git synced 2025-11-04 11:55:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: server info cache (#1671)

Browse Source 
fix: #1181
This commit is contained in:
cyan317
2024-09-30 17:00:23 +01:00
committed by GitHub
parent 1b42466a0d
commit da76907279
14 changed files with 697 additions and 264 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
docs/examples/config/example-config.json
View File

@@ -39,6 +39,9 @@
"cache_timeout": 0.250, // in seconds, could be 0, which means no cache
"request_timeout": 10.0 // time for Clio to wait for rippled to reply on a forwarded request (default is 10 seconds)
},
"rpc": {
"cache_timeout": 0.5 // in seconds, could be 0, which means no cache for rpc
}
"dos_guard": {
// Comma-separated list of IPs to exclude from rate limiting
"whitelist": [

6
src/app/ClioApplication.cpp
View File

@@ -121,12 +121,14 @@ ClioApplication::run()
auto const handlerProvider = std::make_shared<rpc::impl::ProductionHandlerProvider const>(
config_, backend, subscriptions, balancer, etl, amendmentCenter, counters
);
using RPCEngineType = rpc::RPCEngine<etl::LoadBalancer, rpc::Counters>;
auto const rpcEngine =
rpc::RPCEngine::make_RPCEngine(backend, balancer, dosGuard, workQueue, counters, handlerProvider);
RPCEngineType::make_RPCEngine(config_, backend, balancer, dosGuard, workQueue, counters, handlerProvider);
// Init the web server
auto handler =
std::make_shared<web::RPCServerHandler<rpc::RPCEngine, etl::ETLService>>(config_, backend, rpcEngine, etl);
std::make_shared<web::RPCServerHandler<RPCEngineType, etl::ETLService>>(config_, backend, rpcEngine, etl);
auto const httpServer = web::make_HttpServer(config_, ioc, dosGuard, handler);
// Blocks until stopped.

1
src/etl/CMakeLists.txt
View File

@@ -11,7 +11,6 @@ target_sources(
NFTHelpers.cpp
Source.cpp
impl/AmendmentBlockHandler.cpp
impl/ForwardingCache.cpp
impl/ForwardingSource.cpp
impl/GrpcSource.cpp
impl/SubscriptionSource.cpp

15
src/etl/LoadBalancer.cpp
View File

@@ -27,6 +27,7 @@
#include "rpc/Errors.hpp"
#include "util/Assert.hpp"
#include "util/Random.hpp"
#include "util/ResponseExpirationCache.hpp"
#include "util/log/Logger.hpp"
#include <boost/asio/io_context.hpp>
@@ -34,6 +35,7 @@
#include <boost/json/array.hpp>
#include <boost/json/object.hpp>
#include <boost/json/value.hpp>
#include <boost/json/value_to.hpp>
#include <fmt/core.h>
#include <algorithm>
@@ -79,7 +81,10 @@ LoadBalancer::LoadBalancer(
{
auto const forwardingCacheTimeout = config.valueOr<float>("forwarding.cache_timeout", 0.f);
if (forwardingCacheTimeout > 0.f) {
forwardingCache_ = impl::ForwardingCache{Config::toMilliseconds(forwardingCacheTimeout)};
forwardingCache_ = util::ResponseExpirationCache{
Config::toMilliseconds(forwardingCacheTimeout),
{"server_info", "server_state", "server_definitions", "fee", "ledger_closed"}
};
}
static constexpr std::uint32_t MAX_DOWNLOAD = 256;
@@ -224,8 +229,12 @@ LoadBalancer::forwardToRippled(
boost::asio::yield_context yield
)
{
if (not request.contains("command"))
return std::unexpected{rpc::ClioError::rpcCOMMAND_IS_MISSING};
auto const cmd = boost::json::value_to<std::string>(request.at("command"));
if (forwardingCache_) {
if (auto cachedResponse = forwardingCache_->get(request); cachedResponse) {
if (auto cachedResponse = forwardingCache_->get(cmd); cachedResponse) {
return std::move(cachedResponse).value();
}
}
@@ -253,7 +262,7 @@ LoadBalancer::forwardToRippled(
if (response) {
if (forwardingCache_ and not response->contains("error"))
forwardingCache_->put(request, *response);
forwardingCache_->put(cmd, *response);
return std::move(response).value();
}

4
src/etl/LoadBalancer.hpp
View File

@@ -23,9 +23,9 @@
#include "etl/ETLState.hpp"
#include "etl/NetworkValidatedLedgersInterface.hpp"
#include "etl/Source.hpp"
#include "etl/impl/ForwardingCache.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "util/Mutex.hpp"
#include "util/ResponseExpirationCache.hpp"
#include "util/config/Config.hpp"
#include "util/log/Logger.hpp"
@@ -68,7 +68,7 @@ private:
util::Logger log_{"ETL"};
// Forwarding cache must be destroyed after sources because sources have a callback to invalidate cache
std::optional<impl::ForwardingCache> forwardingCache_;
std::optional<util::ResponseExpirationCache> forwardingCache_;
std::optional<std::string> forwardingXUserValue_;
std::vector<SourcePtr> sources_;

52
src/rpc/RPCEngine.hpp
View File

@@ -20,20 +20,22 @@
#pragma once
#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"
#include "rpc/common/impl/ForwardingProxy.hpp"
#include "util/ResponseExpirationCache.hpp"
#include "util/log/Logger.hpp"
#include "web/Context.hpp"
#include "web/dosguard/DOSGuardInterface.hpp"
#include <boost/asio/spawn.hpp>
#include <boost/iterator/transform_iterator.hpp>
#include <boost/json.hpp>
#include <fmt/core.h>
#include <fmt/format.h>
#include <xrpl/protocol/ErrorCodes.h>
#include <chrono>
@@ -42,14 +44,9 @@
#include <memory>
#include <optional>
#include <string>
#include <unordered_set>
#include <utility>
// forward declarations
namespace etl {
class LoadBalancer;
class ETLService;
} // namespace etl
/**
* @brief This namespace contains all the RPC logic and handlers.
*/
@@ -58,6 +55,7 @@ namespace rpc {
/**
* @brief The RPC engine that ties all RPC-related functionality together.
*/
template <typename LoadBalancerType, typename CountersType>
class RPCEngine {
util::Logger perfLog_{"Performance"};
util::Logger log_{"RPC"};
@@ -65,16 +63,19 @@ class RPCEngine {
std::shared_ptr<BackendInterface> backend_;
std::reference_wrapper<web::dosguard::DOSGuardInterface const> dosGuard_;
std::reference_wrapper<WorkQueue> workQueue_;
std::reference_wrapper<Counters> counters_;
std::reference_wrapper<CountersType> counters_;
std::shared_ptr<HandlerProvider const> handlerProvider_;
impl::ForwardingProxy<etl::LoadBalancer, Counters, HandlerProvider> forwardingProxy_;
impl::ForwardingProxy<LoadBalancerType, CountersType, HandlerProvider> forwardingProxy_;
std::optional<util::ResponseExpirationCache> responseCache_;
public:
/**
* @brief Construct a new RPCEngine object
*
* @param config The config to use
* @param backend The backend to use
* @param balancer The load balancer to use
* @param dosGuard The DOS guard to use
@@ -83,11 +84,12 @@ public:
* @param handlerProvider The handler provider to use
*/
RPCEngine(
util::Config const& config,
std::shared_ptr<BackendInterface> const& backend,
std::shared_ptr<etl::LoadBalancer> const& balancer,
std::shared_ptr<LoadBalancerType> const& balancer,
web::dosguard::DOSGuardInterface const& dosGuard,
WorkQueue& workQueue,
Counters& counters,
CountersType& counters,
std::shared_ptr<HandlerProvider const> const& handlerProvider
)
: backend_{backend}
@@ -97,11 +99,22 @@ public:
, handlerProvider_{handlerProvider}
, forwardingProxy_{balancer, counters, handlerProvider}
{
// Let main thread catch the exception if config type is wrong
auto const cacheTimeout = config.valueOr<float>("rpc.cache_timeout", 0.f);
if (cacheTimeout > 0.f) {
LOG(log_.info()) << fmt::format("Init RPC Cache, timeout: {} seconds", cacheTimeout);
responseCache_.emplace(
util::Config::toMilliseconds(cacheTimeout), std::unordered_set<std::string>{"server_info"}
);
}
}
/**
* @brief Factory function to create a new instance of the RPC engine.
*
* @param config The config to use
* @param backend The backend to use
* @param balancer The load balancer to use
* @param dosGuard The DOS guard to use
@@ -112,15 +125,16 @@ public:
*/
static std::shared_ptr<RPCEngine>
make_RPCEngine(
util::Config const& config,
std::shared_ptr<BackendInterface> const& backend,
std::shared_ptr<etl::LoadBalancer> const& balancer,
std::shared_ptr<LoadBalancerType> const& balancer,
web::dosguard::DOSGuardInterface const& dosGuard,
WorkQueue& workQueue,
Counters& counters,
CountersType& counters,
std::shared_ptr<HandlerProvider const> const& handlerProvider
)
{
return std::make_shared<RPCEngine>(backend, balancer, dosGuard, workQueue, counters, handlerProvider);
return std::make_shared<RPCEngine>(config, backend, balancer, dosGuard, workQueue, counters, handlerProvider);
}
/**
@@ -140,6 +154,11 @@ public:
return forwardingProxy_.forward(ctx);
}
if (not ctx.isAdmin and responseCache_) {
if (auto res = responseCache_->get(ctx.method); res.has_value())
return Result{std::move(res).value()};
}
if (backend_->isTooBusy()) {
LOG(log_.error()) << "Database is too busy. Rejecting request";
notifyTooBusy(); // TODO: should we add ctx.method if we have it?
@@ -160,8 +179,11 @@ public:
LOG(perfLog_.debug()) << ctx.tag() << " finish executing rpc `" << ctx.method << '`';
if (not v)
if (not v) {
notifyErrored(ctx.method);
} else if (not ctx.isAdmin and responseCache_) {
responseCache_->put(ctx.method, v.result->as_object());
}
return Result{std::move(v)};
} catch (data::DatabaseTimeout const& t) {

1
src/util/CMakeLists.txt
View File

@@ -19,6 +19,7 @@ target_sources(
requests/Types.cpp
requests/WsConnection.cpp
requests/impl/SslContext.cpp
ResponseExpirationCache.cpp
SignalsHandler.cpp
Taggable.cpp
TerminationHandler.cpp

66
src/etl/impl/ForwardingCache.cpp → src/util/ResponseExpirationCache.cpp
View File

@@ -17,88 +17,55 @@
*/
//==============================================================================
#include "etl/impl/ForwardingCache.hpp"
#include "util/ResponseExpirationCache.hpp"
#include "util/Assert.hpp"
#include <boost/json/object.hpp>
#include <boost/json/value_to.hpp>
#include <chrono>
#include <mutex>
#include <optional>
#include <shared_mutex>
#include <string>
#include <unordered_set>
#include <utility>
namespace etl::impl {
namespace {
std::optional<std::string>
getCommand(boost::json::object const& request)
{
if (not request.contains("command")) {
return std::nullopt;
}
return boost::json::value_to<std::string>(request.at("command"));
}
} // namespace
namespace util {
void
CacheEntry::put(boost::json::object response)
ResponseExpirationCache::Entry::put(boost::json::object response)
{
response_ = std::move(response);
lastUpdated_ = std::chrono::steady_clock::now();
}
std::optional<boost::json::object>
CacheEntry::get() const
ResponseExpirationCache::Entry::get() const
{
return response_;
}
std::chrono::steady_clock::time_point
CacheEntry::lastUpdated() const
ResponseExpirationCache::Entry::lastUpdated() const
{
return lastUpdated_;
}
void
CacheEntry::invalidate()
ResponseExpirationCache::Entry::invalidate()
{
response_.reset();
}
std::unordered_set<std::string> const
ForwardingCache::CACHEABLE_COMMANDS{"server_info", "server_state", "server_definitions", "fee", "ledger_closed"};
ForwardingCache::ForwardingCache(std::chrono::steady_clock::duration const cacheTimeout) : cacheTimeout_{cacheTimeout}
{
for (auto const& command : CACHEABLE_COMMANDS) {
cache_.emplace(command, CacheEntry{});
}
}
bool
ForwardingCache::shouldCache(boost::json::object const& request)
ResponseExpirationCache::shouldCache(std::string const& cmd)
{
auto const command = getCommand(request);
return command.has_value() and CACHEABLE_COMMANDS.contains(*command);
return cache_.contains(cmd);
}
std::optional<boost::json::object>
ForwardingCache::get(boost::json::object const& request) const
ResponseExpirationCache::get(std::string const& cmd) const
{
auto const command = getCommand(request);
if (not command.has_value()) {
return std::nullopt;
}
auto it = cache_.find(*command);
auto it = cache_.find(cmd);
if (it == cache_.end())
return std::nullopt;
@@ -110,20 +77,19 @@ ForwardingCache::get(boost::json::object const& request) const
}
void
ForwardingCache::put(boost::json::object const& request, boost::json::object const& response)
ResponseExpirationCache::put(std::string const& cmd, boost::json::object const& response)
{
auto const command = getCommand(request);
if (not command.has_value() or not shouldCache(request))
if (not shouldCache(cmd))
return;
ASSERT(cache_.contains(*command), "Command is not in the cache: {}", *command);
ASSERT(cache_.contains(cmd), "Command is not in the cache: {}", cmd);
auto entry = cache_[*command].lock<std::unique_lock>();
auto entry = cache_[cmd].lock<std::unique_lock>();
entry->put(response);
}
void
ForwardingCache::invalidate()
ResponseExpirationCache::invalidate()
{
for (auto& [_, entry] : cache_) {
auto entryLock = entry.lock<std::unique_lock>();
@@ -131,4 +97,4 @@ ForwardingCache::invalidate()
}
}
} // namespace etl::impl
} // namespace util

113
src/etl/impl/ForwardingCache.hpp → src/util/ResponseExpirationCache.hpp
View File

@@ -30,90 +30,92 @@
#include <unordered_map>
#include <unordered_set>
namespace etl::impl {
namespace util {
/**
* @brief A class to store a cache entry.
* @brief Cache of requests' responses with TTL support and configurable cachable commands
*/
class CacheEntry {
std::chrono::steady_clock::time_point lastUpdated_;
std::optional<boost::json::object> response_;
public:
class ResponseExpirationCache {
/**
* @brief Put a response into the cache
*
* @param response The response to store
* @brief A class to store a cache entry.
*/
void
put(boost::json::object response);
class Entry {
std::chrono::steady_clock::time_point lastUpdated_;
std::optional<boost::json::object> response_;
/**
* @brief Get the response from the cache
*
* @return The response
*/
std::optional<boost::json::object>
get() const;
public:
/**
* @brief Put a response into the cache
*
* @param response The response to store
*/
void
put(boost::json::object response);
/**
* @brief Get the last time the cache was updated
*
* @return The last time the cache was updated
*/
std::chrono::steady_clock::time_point
lastUpdated() const;
/**
* @brief Get the response from the cache
*
* @return The response
*/
std::optional<boost::json::object>
get() const;
/**
* @brief Invalidate the cache entry
*/
void
invalidate();
};
/**
* @brief Get the last time the cache was updated
*
* @return The last time the cache was updated
*/
std::chrono::steady_clock::time_point
lastUpdated() const;
/**
* @brief Invalidate the cache entry
*/
void
invalidate();
};
/**
* @brief A class to store a cache of forwarding responses
*/
class ForwardingCache {
std::chrono::steady_clock::duration cacheTimeout_;
std::unordered_map<std::string, util::Mutex<CacheEntry, std::shared_mutex>> cache_;
std::unordered_map<std::string, util::Mutex<Entry, std::shared_mutex>> cache_;
bool
shouldCache(std::string const& cmd);
public:
static std::unordered_set<std::string> const CACHEABLE_COMMANDS;
/**
* @brief Construct a new Forwarding Cache object
* @brief Construct a new Cache object
*
* @param cacheTimeout The time for cache entries to expire
* @param cmds The commands that should be cached
*/
ForwardingCache(std::chrono::steady_clock::duration cacheTimeout);
/**
* @brief Check if a request should be cached
*
* @param request The request to check
* @return true if the request should be cached and false otherwise
*/
[[nodiscard]] static bool
shouldCache(boost::json::object const& request);
ResponseExpirationCache(
std::chrono::steady_clock::duration cacheTimeout,
std::unordered_set<std::string> const& cmds
)
: cacheTimeout_(cacheTimeout)
{
for (auto const& command : cmds) {
cache_.emplace(command, Entry{});
}
}
/**
* @brief Get a response from the cache
*
* @param request The request to get the response for
* @param cmd The command to get the response for
* @return The response if it exists or std::nullopt otherwise
*/
[[nodiscard]] std::optional<boost::json::object>
get(boost::json::object const& request) const;
get(std::string const& cmd) const;
/**
* @brief Put a response into the cache if the request should be cached
*
* @param request The request to store the response for
* @param cmd The command to store the response for
* @param response The response to store
*/
void
put(boost::json::object const& request, boost::json::object const& response);
put(std::string const& cmd, boost::json::object const& response);
/**
* @brief Invalidate all entries in the cache
@@ -121,5 +123,4 @@ public:
void
invalidate();
};
} // namespace etl::impl
} // namespace util

3
tests/unit/CMakeLists.txt
View File

@@ -23,7 +23,6 @@ target_sources(
etl/ETLStateTests.cpp
etl/ExtractionDataPipeTests.cpp
etl/ExtractorTests.cpp
etl/ForwardingCacheTests.cpp
etl/ForwardingSourceTests.cpp
etl/GrpcSourceTests.cpp
etl/LedgerPublisherTests.cpp
@@ -92,6 +91,7 @@ target_sources(
rpc/handlers/UnsubscribeTests.cpp
rpc/handlers/VersionHandlerTests.cpp
rpc/JsonBoolTests.cpp
rpc/RPCEngineTests.cpp
rpc/RPCHelpersTests.cpp
rpc/WorkQueueTests.cpp
util/AccountUtilsTests.cpp
@@ -121,6 +121,7 @@ target_sources(
util/RandomTests.cpp
util/RetryTests.cpp
util/RepeatTests.cpp
util/ResponseExpirationCacheTests.cpp
util/SignalsHandlerTests.cpp
util/TimeUtilsTests.cpp
util/TxUtilTests.cpp

133
tests/unit/etl/ForwardingCacheTests.cpp
View File

@@ -1,133 +0,0 @@
//------------------------------------------------------------------------------
/*
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 "etl/impl/ForwardingCache.hpp"
#include <boost/json/object.hpp>
#include <gtest/gtest.h>
#include <chrono>
#include <thread>
using namespace etl::impl;
struct CacheEntryTests : public ::testing::Test {
CacheEntry entry_;
boost::json::object const object_ = {{"key", "value"}};
};
TEST_F(CacheEntryTests, PutAndGet)
{
EXPECT_FALSE(entry_.get());
entry_.put(object_);
auto result = entry_.get();
ASSERT_TRUE(result);
EXPECT_EQ(*result, object_);
}
TEST_F(CacheEntryTests, LastUpdated)
{
EXPECT_EQ(entry_.lastUpdated().time_since_epoch().count(), 0);
entry_.put(object_);
auto const lastUpdated = entry_.lastUpdated();
EXPECT_GE(
std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - lastUpdated).count(), 0
);
entry_.put(boost::json::object{{"key", "new value"}});
auto const newLastUpdated = entry_.lastUpdated();
EXPECT_GT(newLastUpdated, lastUpdated);
EXPECT_GE(
std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - newLastUpdated)
.count(),
0
);
}
TEST_F(CacheEntryTests, Invalidate)
{
entry_.put(object_);
entry_.invalidate();
EXPECT_FALSE(entry_.get());
}
TEST(ForwardingCacheTests, ShouldCache)
{
for (auto const& command : ForwardingCache::CACHEABLE_COMMANDS) {
auto const request = boost::json::object{{"command", command}};
EXPECT_TRUE(ForwardingCache::shouldCache(request));
}
auto const request = boost::json::object{{"command", "tx"}};
EXPECT_FALSE(ForwardingCache::shouldCache(request));
auto const requestWithoutCommand = boost::json::object{{"key", "value"}};
EXPECT_FALSE(ForwardingCache::shouldCache(requestWithoutCommand));
}
TEST(ForwardingCacheTests, Get)
{
ForwardingCache cache{std::chrono::seconds{100}};
auto const request = boost::json::object{{"command", "server_info"}};
auto const response = boost::json::object{{"key", "value"}};
cache.put(request, response);
auto const result = cache.get(request);
ASSERT_TRUE(result);
EXPECT_EQ(*result, response);
}
TEST(ForwardingCacheTests, GetExpired)
{
ForwardingCache cache{std::chrono::milliseconds{1}};
auto const request = boost::json::object{{"command", "server_info"}};
auto const response = boost::json::object{{"key", "value"}};
cache.put(request, response);
std::this_thread::sleep_for(std::chrono::milliseconds{2});
auto const result = cache.get(request);
EXPECT_FALSE(result);
}
TEST(ForwardingCacheTests, GetAndPutNotCommand)
{
ForwardingCache cache{std::chrono::seconds{100}};
auto const request = boost::json::object{{"key", "value"}};
auto const response = boost::json::object{{"key", "value"}};
cache.put(request, response);
auto const result = cache.get(request);
EXPECT_FALSE(result);
}
TEST(ForwardingCache, Invalidate)
{
ForwardingCache cache{std::chrono::seconds{100}};
auto const request = boost::json::object{{"command", "server_info"}};
auto const response = boost::json::object{{"key", "value"}};
cache.put(request, response);
cache.invalidate();
EXPECT_FALSE(cache.get(request));
}

17
tests/unit/etl/LoadBalancerTests.cpp
View File

@@ -519,7 +519,7 @@ struct LoadBalancerForwardToRippledTests : LoadBalancerConstructorTests, SyncAsi
EXPECT_CALL(sourceFactory_.sourceAt(1), run);
}
boost::json::object const request_{{"request", "value"}};
boost::json::object const request_{{"command", "value"}};
std::optional<std::string> const clientIP_ = "some_ip";
boost::json::object const response_{{"response", "other_value"}};
};
@@ -699,6 +699,21 @@ TEST_F(LoadBalancerForwardToRippledTests, onLedgerClosedHookInvalidatesCache)
});
}
TEST_F(LoadBalancerForwardToRippledTests, commandLineMissing)
{
EXPECT_CALL(sourceFactory_, makeSource).Times(2);
auto loadBalancer = makeLoadBalancer();
auto const request = boost::json::object{{"command2", "server_info"}};
runSpawn([&](boost::asio::yield_context yield) {
EXPECT_EQ(
loadBalancer->forwardToRippled(request, clientIP_, false, yield).error(),
rpc::ClioError::rpcCOMMAND_IS_MISSING
);
});
}
struct LoadBalancerToJsonTests : LoadBalancerOnConnectHookTests {};
TEST_F(LoadBalancerToJsonTests, toJson)

477
tests/unit/rpc/RPCEngineTests.cpp Normal file
View File

@@ -0,0 +1,477 @@
//------------------------------------------------------------------------------
/*
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 "data/BackendInterface.hpp"
#include "data/Types.hpp"
#include "rpc/Errors.hpp"
#include "rpc/FakesAndMocks.hpp"
#include "rpc/RPCEngine.hpp"
#include "rpc/WorkQueue.hpp"
#include "rpc/common/AnyHandler.hpp"
#include "util/AsioContextTestFixture.hpp"
#include "util/MockBackendTestFixture.hpp"
#include "util/MockCounters.hpp"
#include "util/MockCountersFixture.hpp"
#include "util/MockETLServiceTestFixture.hpp"
#include "util/MockHandlerProvider.hpp"
#include "util/MockLoadBalancer.hpp"
#include "util/MockPrometheus.hpp"
#include "util/NameGenerator.hpp"
#include "util/Taggable.hpp"
#include "util/config/Config.hpp"
#include "web/Context.hpp"
#include "web/dosguard/DOSGuard.hpp"
#include "web/dosguard/WhitelistHandler.hpp"
#include <boost/json/object.hpp>
#include <boost/json/parse.hpp>
#include <gmock/gmock.h>
#include <gtest/gtest.h>
#include <xrpl/protocol/ErrorCodes.h>
#include <exception>
#include <memory>
#include <optional>
#include <string>
#include <variant>
#include <vector>
using namespace rpc;
using namespace util;
namespace json = boost::json;
using namespace testing;
namespace {
constexpr auto FORWARD_REPLY = R"JSON({
"result":
{
"status": "success",
"forwarded": true
}
})JSON";
} // namespace
struct RPCEngineTest : util::prometheus::WithPrometheus,
MockBackendTest,
MockCountersTest,
MockLoadBalancerTest,
SyncAsioContextTest {
Config cfg = Config{json::parse(R"JSON({
"server": {"max_queue_size": 2},
"workers": 4
})JSON")};
util::TagDecoratorFactory tagFactory{cfg};
WorkQueue queue = WorkQueue::make_WorkQueue(cfg);
web::dosguard::WhitelistHandler whitelistHandler{cfg};
web::dosguard::DOSGuard dosGuard{cfg, whitelistHandler};
std::shared_ptr<MockHandlerProvider> handlerProvider = std::make_shared<MockHandlerProvider>();
};
struct RPCEngineFlowTestCaseBundle {
std::string testName;
bool isAdmin;
std::string method;
std::string params;
bool forwarded;
std::optional<bool> isTooBusy;
std::optional<bool> isUnknownCmd;
bool handlerReturnError;
std::optional<rpc::Status> status;
std::optional<boost::json::object> response;
};
struct RPCEngineFlowParameterTest : public RPCEngineTest, WithParamInterface<RPCEngineFlowTestCaseBundle> {};
static auto
generateTestValuesForParametersTest()
{
auto const neverCalled = std::nullopt;
return std::vector<RPCEngineFlowTestCaseBundle>{
{.testName = "ForwardedSubmit",
.isAdmin = true,
.method = "submit",
.params = "{}",
.forwarded = true,
.isTooBusy = neverCalled,
.isUnknownCmd = neverCalled,
.handlerReturnError = false,
.status = rpc::Status{},
.response = boost::json::parse(FORWARD_REPLY).as_object()},
{.testName = "ForwardAdminCmd",
.isAdmin = false,
.method = "ledger",
.params = R"JSON({"full": true, "ledger_index": "current"})JSON",
.forwarded = false,
.isTooBusy = neverCalled,
.isUnknownCmd = neverCalled,
.handlerReturnError = false,
.status = rpc::Status{RippledError::rpcNO_PERMISSION},
.response = std::nullopt},
{.testName = "BackendTooBusy",
.isAdmin = false,
.method = "ledger",
.params = "{}",
.forwarded = false,
.isTooBusy = true,
.isUnknownCmd = neverCalled,
.handlerReturnError = false,
.status = rpc::Status{RippledError::rpcTOO_BUSY},
.response = std::nullopt},
{.testName = "HandlerUnknown",
.isAdmin = false,
.method = "ledger",
.params = "{}",
.forwarded = false,
.isTooBusy = false,
.isUnknownCmd = true,
.handlerReturnError = false,
.status = rpc::Status{RippledError::rpcUNKNOWN_COMMAND},
.response = std::nullopt},
{.testName = "HandlerReturnError",
.isAdmin = false,
.method = "ledger",
.params = R"JSON({"hello": "world", "limit": 50})JSON",
.forwarded = false,
.isTooBusy = false,
.isUnknownCmd = false,
.handlerReturnError = true,
.status = rpc::Status{"Very custom error"},
.response = std::nullopt},
{.testName = "HandlerReturnResponse",
.isAdmin = false,
.method = "ledger",
.params = R"JSON({"hello": "world", "limit": 50})JSON",
.forwarded = false,
.isTooBusy = false,
.isUnknownCmd = false,
.handlerReturnError = false,
.status = std::nullopt,
.response = boost::json::parse(R"JSON({"computed": "world_50"})JSON").as_object()},
};
}
INSTANTIATE_TEST_CASE_P(
RPCEngineFlow,
RPCEngineFlowParameterTest,
ValuesIn(generateTestValuesForParametersTest()),
tests::util::NameGenerator
);
TEST_P(RPCEngineFlowParameterTest, Test)
{
auto const& testBundle = GetParam();
std::shared_ptr<RPCEngine<MockLoadBalancer, MockCounters>> engine =
RPCEngine<MockLoadBalancer, MockCounters>::make_RPCEngine(
Config{}, backend, mockLoadBalancerPtr, dosGuard, queue, *mockCountersPtr, handlerProvider
);
if (testBundle.forwarded) {
EXPECT_CALL(*mockLoadBalancerPtr, forwardToRippled)
.WillOnce(Return(std::expected<boost::json::object, rpc::ClioError>(json::parse(FORWARD_REPLY).as_object()))
);
EXPECT_CALL(*handlerProvider, contains).WillOnce(Return(true));
EXPECT_CALL(*mockCountersPtr, rpcForwarded(testBundle.method));
}
if (testBundle.isTooBusy.has_value()) {
EXPECT_CALL(*backend, isTooBusy).WillOnce(Return(*testBundle.isTooBusy));
if (testBundle.isTooBusy.value())
EXPECT_CALL(*mockCountersPtr, onTooBusy);
}
EXPECT_CALL(*handlerProvider, isClioOnly).WillOnce(Return(false));
if (testBundle.isUnknownCmd.has_value()) {
if (testBundle.isUnknownCmd.value()) {
EXPECT_CALL(*handlerProvider, getHandler).WillOnce(Return(std::nullopt));
EXPECT_CALL(*mockCountersPtr, onUnknownCommand);
} else {
if (testBundle.handlerReturnError) {
EXPECT_CALL(*handlerProvider, getHandler)
.WillOnce(Return(AnyHandler{tests::common::FailingHandlerFake{}}));
EXPECT_CALL(*mockCountersPtr, rpcErrored(testBundle.method));
EXPECT_CALL(*handlerProvider, contains(testBundle.method)).WillOnce(Return(true));
} else {
EXPECT_CALL(*handlerProvider, getHandler(testBundle.method))
.WillOnce(Return(AnyHandler{tests::common::HandlerFake{}}));
}
}
}
runSpawn([&](auto yield) {
auto const ctx = web::Context(
yield,
testBundle.method,
1, // api version
boost::json::parse(testBundle.params).as_object(),
nullptr,
tagFactory,
LedgerRange{0, 30},
"127.0.0.2",
testBundle.isAdmin
);
auto const res = engine->buildResponse(ctx);
auto const status = std::get_if<rpc::Status>(&res.response);
auto const response = std::get_if<boost::json::object>(&res.response);
ASSERT_EQ(status == nullptr, testBundle.response.has_value());
if (testBundle.response.has_value()) {
EXPECT_EQ(*response, testBundle.response.value());
} else {
EXPECT_TRUE(*status == testBundle.status.value());
}
});
}
TEST_F(RPCEngineTest, ThrowDatabaseError)
{
auto const method = "subscribe";
std::shared_ptr<RPCEngine<MockLoadBalancer, MockCounters>> engine =
RPCEngine<MockLoadBalancer, MockCounters>::make_RPCEngine(
cfg, backend, mockLoadBalancerPtr, dosGuard, queue, *mockCountersPtr, handlerProvider
);
EXPECT_CALL(*backend, isTooBusy).WillOnce(Return(false));
EXPECT_CALL(*handlerProvider, getHandler(method)).WillOnce(Return(AnyHandler{tests::common::FailingHandlerFake{}}));
EXPECT_CALL(*mockCountersPtr, rpcErrored(method)).WillOnce(Throw(data::DatabaseTimeout{}));
EXPECT_CALL(*handlerProvider, contains(method)).WillOnce(Return(true));
EXPECT_CALL(*mockCountersPtr, onTooBusy());
runSpawn([&](auto yield) {
auto const ctx = web::Context(
yield,
method,
1,
boost::json::parse("{}").as_object(),
nullptr,
tagFactory,
LedgerRange{0, 30},
"127.0.0.2",
false
);
auto const res = engine->buildResponse(ctx);
auto const status = std::get_if<rpc::Status>(&res.response);
ASSERT_TRUE(status != nullptr);
EXPECT_TRUE(*status == Status{RippledError::rpcTOO_BUSY});
});
}
TEST_F(RPCEngineTest, ThrowException)
{
auto const method = "subscribe";
std::shared_ptr<RPCEngine<MockLoadBalancer, MockCounters>> engine =
RPCEngine<MockLoadBalancer, MockCounters>::make_RPCEngine(
cfg, backend, mockLoadBalancerPtr, dosGuard, queue, *mockCountersPtr, handlerProvider
);
EXPECT_CALL(*backend, isTooBusy).WillOnce(Return(false));
EXPECT_CALL(*handlerProvider, getHandler(method)).WillOnce(Return(AnyHandler{tests::common::FailingHandlerFake{}}));
EXPECT_CALL(*mockCountersPtr, rpcErrored(method)).WillOnce(Throw(std::exception{}));
EXPECT_CALL(*handlerProvider, contains(method)).WillOnce(Return(true));
EXPECT_CALL(*mockCountersPtr, onInternalError());
runSpawn([&](auto yield) {
auto const ctx = web::Context(
yield,
method,
1,
boost::json::parse("{}").as_object(),
nullptr,
tagFactory,
LedgerRange{0, 30},
"127.0.0.2",
false
);
auto const res = engine->buildResponse(ctx);
auto const status = std::get_if<rpc::Status>(&res.response);
ASSERT_TRUE(status != nullptr);
EXPECT_TRUE(*status == Status{RippledError::rpcINTERNAL});
});
}
struct RPCEngineCacheTestCaseBundle {
std::string testName;
std::string config;
std::string method;
bool isAdmin;
bool expectedCacheEnabled;
};
struct RPCEngineCacheParameterTest : public RPCEngineTest, WithParamInterface<RPCEngineCacheTestCaseBundle> {};
static auto
generateCacheTestValuesForParametersTest()
{
return std::vector<RPCEngineCacheTestCaseBundle>{
{.testName = "CacheEnabled",
.config = R"JSON({
"server": {"max_queue_size": 2},
"workers": 4,
"rpc":
{"cache_timeout": 10}
})JSON",
.method = "server_info",
.isAdmin = false,
.expectedCacheEnabled = true},
{.testName = "CacheDisabledWhenNoConfig",
.config = R"JSON({
"server": {"max_queue_size": 2},
"workers": 4,
"rpc": {}
})JSON",
.method = "server_info",
.isAdmin = false,
.expectedCacheEnabled = false},
{.testName = "CacheDisabledWhenNoTimeout",
.config = R"JSON({
"server": {"max_queue_size": 2},
"workers": 4,
"rpc": {}
})JSON",
.method = "server_info",
.isAdmin = false,
.expectedCacheEnabled = false},
{.testName = "CacheDisabledWhenTimeoutIsZero",
.config = R"JSON({
"server": {"max_queue_size": 2},
"workers": 4,
"rpc": {"cache_timeout": 0}
})JSON",
.method = "server_info",
.isAdmin = false,
.expectedCacheEnabled = false},
{.testName = "CacheNotWorkForAdmin",
.config = R"JSON({
"server": {"max_queue_size": 2},
"workers": 4,
"rpc": { "cache_timeout": 10}
})JSON",
.method = "server_info",
.isAdmin = true,
.expectedCacheEnabled = false},
{.testName = "CacheDisabledWhenCmdNotMatch",
.config = R"JSON({
"server": {"max_queue_size": 2},
"workers": 4,
"rpc": {"cache_timeout": 10}
})JSON",
.method = "server_info2",
.isAdmin = false,
.expectedCacheEnabled = false},
};
}
INSTANTIATE_TEST_CASE_P(
RPCEngineCache,
RPCEngineCacheParameterTest,
ValuesIn(generateCacheTestValuesForParametersTest()),
tests::util::NameGenerator
);
TEST_P(RPCEngineCacheParameterTest, Test)
{
auto const& testParam = GetParam();
auto const cfgCache = Config{json::parse(testParam.config)};
auto const admin = testParam.isAdmin;
auto const method = testParam.method;
std::shared_ptr<RPCEngine<MockLoadBalancer, MockCounters>> engine =
RPCEngine<MockLoadBalancer, MockCounters>::make_RPCEngine(
cfgCache, backend, mockLoadBalancerPtr, dosGuard, queue, *mockCountersPtr, handlerProvider
);
int callTime = 2;
EXPECT_CALL(*handlerProvider, isClioOnly).Times(callTime).WillRepeatedly(Return(false));
if (testParam.expectedCacheEnabled) {
EXPECT_CALL(*backend, isTooBusy).WillOnce(Return(false));
EXPECT_CALL(*handlerProvider, getHandler).WillOnce(Return(AnyHandler{tests::common::HandlerFake{}}));
} else {
EXPECT_CALL(*backend, isTooBusy).Times(callTime).WillRepeatedly(Return(false));
EXPECT_CALL(*handlerProvider, getHandler)
.Times(callTime)
.WillRepeatedly(Return(AnyHandler{tests::common::HandlerFake{}}));
}
while (callTime-- != 0) {
runSpawn([&](auto yield) {
auto const ctx = web::Context(
yield,
method,
1,
boost::json::parse(R"JSON({"hello": "world", "limit": 50})JSON").as_object(),
nullptr,
tagFactory,
LedgerRange{0, 30},
"127.0.0.2",
admin
);
auto const res = engine->buildResponse(ctx);
auto const response = std::get_if<boost::json::object>(&res.response);
EXPECT_TRUE(*response == boost::json::parse(R"JSON({ "computed": "world_50"})JSON").as_object());
});
}
}
TEST_F(RPCEngineTest, NotCacheIfErrorHappen)
{
auto const cfgCache = Config{json::parse(R"JSON({
"server": {"max_queue_size": 2},
"workers": 4,
"rpc": {"cache_timeout": 10}
})JSON")};
auto const notAdmin = false;
auto const method = "server_info";
std::shared_ptr<RPCEngine<MockLoadBalancer, MockCounters>> engine =
RPCEngine<MockLoadBalancer, MockCounters>::make_RPCEngine(
cfgCache, backend, mockLoadBalancerPtr, dosGuard, queue, *mockCountersPtr, handlerProvider
);
int callTime = 2;
EXPECT_CALL(*backend, isTooBusy).Times(callTime).WillRepeatedly(Return(false));
EXPECT_CALL(*handlerProvider, getHandler)
.Times(callTime)
.WillRepeatedly(Return(AnyHandler{tests::common::FailingHandlerFake{}}));
EXPECT_CALL(*mockCountersPtr, rpcErrored(method)).Times(callTime);
EXPECT_CALL(*handlerProvider, isClioOnly).Times(callTime).WillRepeatedly(Return(false));
EXPECT_CALL(*handlerProvider, contains).Times(callTime).WillRepeatedly(Return(true));
while (callTime-- != 0) {
runSpawn([&](auto yield) {
auto const ctx = web::Context(
yield,
method,
1,
boost::json::parse(R"JSON({"hello": "world","limit": 50})JSON").as_object(),
nullptr,
tagFactory,
LedgerRange{0, 30},
"127.0.0.2",
notAdmin
);
auto const res = engine->buildResponse(ctx);
auto const error = std::get_if<rpc::Status>(&res.response);
EXPECT_TRUE(*error == rpc::Status{"Very custom error"});
});
}
}

70
tests/unit/util/ResponseExpirationCacheTests.cpp Normal file
View File

@@ -0,0 +1,70 @@
//------------------------------------------------------------------------------
/*
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/ResponseExpirationCache.hpp"
#include <boost/json/object.hpp>
#include <boost/json/parse.hpp>
#include <gtest/gtest.h>
#include <chrono>
#include <thread>
using namespace util;
struct ResponseExpirationCacheTests : public ::testing::Test {
protected:
ResponseExpirationCache cache_{std::chrono::seconds{100}, {"key"}};
boost::json::object object_{{"key", "value"}};
};
TEST_F(ResponseExpirationCacheTests, PutAndGetNotExpired)
{
EXPECT_FALSE(cache_.get("key").has_value());
cache_.put("key", object_);
auto result = cache_.get("key");
ASSERT_TRUE(result.has_value());
EXPECT_EQ(*result, object_);
result = cache_.get("key2");
ASSERT_FALSE(result.has_value());
cache_.put("key2", object_);
result = cache_.get("key2");
ASSERT_FALSE(result.has_value());
}
TEST_F(ResponseExpirationCacheTests, Invalidate)
{
cache_.put("key", object_);
cache_.invalidate();
EXPECT_FALSE(cache_.get("key").has_value());
}
TEST_F(ResponseExpirationCacheTests, GetExpired)
{
ResponseExpirationCache cache{std::chrono::milliseconds{1}, {"key"}};
auto const response = boost::json::object{{"key", "value"}};
cache.put("key", response);
std::this_thread::sleep_for(std::chrono::milliseconds{2});
auto const result = cache.get("key");
EXPECT_FALSE(result);
}