RPCEngine.hpp

#pragma once
 
#include "data/BackendInterface.hpp"
#include "etl/LoadBalancerInterface.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 <boost/json/object.hpp>
#include <fmt/format.h>
#include <xrpl/protocol/ErrorCodes.h>
 
#include <chrono>
#include <cstdint>
#include <exception>
#include <functional>
#include <memory>
#include <optional>
#include <string>
#include <unordered_set>
#include <utility>

namespace rpc {

template <typename CountersType>
class RPCEngine {
    util::Logger perfLog_{"Performance"};
    util::Logger log_{"RPC"};
 
    std::shared_ptr<BackendInterface> backend_;
    std::reference_wrapper<web::dosguard::DOSGuardInterface const> dosGuard_;
    std::reference_wrapper<WorkQueue> workQueue_;
    std::reference_wrapper<CountersType> counters_;
 
    std::shared_ptr<HandlerProvider const> handlerProvider_;
 
    impl::ForwardingProxy<CountersType, HandlerProvider> forwardingProxy_;
 
    std::optional<util::ResponseExpirationCache> responseCache_;
 
public:
    RPCEngine(
        util::config::ClioConfigDefinition const& config,
        std::shared_ptr<BackendInterface> backend,
        std::shared_ptr<etl::LoadBalancerInterface> const& balancer,
        web::dosguard::DOSGuardInterface const& dosGuard,
        WorkQueue& workQueue,
        CountersType& counters,
        std::shared_ptr<HandlerProvider const> const& handlerProvider
    )
        : backend_{std::move(backend)}
        , dosGuard_{std::cref(dosGuard)}
        , workQueue_{std::ref(workQueue)}
        , counters_{std::ref(counters)}
        , handlerProvider_{handlerProvider}
        , forwardingProxy_{balancer, counters, handlerProvider}
    {
        // Let main thread catch the exception if config type is wrong
        auto const cacheTimeout = config.get<float>("rpc.cache_timeout");
 
        if (cacheTimeout > 0.f) {
            LOG(log_.info()) << fmt::format("Init RPC Cache, timeout: {} seconds", cacheTimeout);
 
            responseCache_.emplace(
                util::config::ClioConfigDefinition::toMilliseconds(cacheTimeout),
                std::unordered_set<std::string>{"server_info"}
            );
        }
    }

    static std::shared_ptr<RPCEngine>
    makeRPCEngine(
        util::config::ClioConfigDefinition const& config,
        std::shared_ptr<BackendInterface> const& backend,
        std::shared_ptr<etl::LoadBalancerInterface> const& balancer,
        web::dosguard::DOSGuardInterface const& dosGuard,
        WorkQueue& workQueue,
        CountersType& counters,
        std::shared_ptr<HandlerProvider const> const& handlerProvider
    )
    {
        return std::make_shared<RPCEngine>(
            config, backend, balancer, dosGuard, workQueue, counters, handlerProvider
        );
    }

    Result
    buildResponse(web::Context const& 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 (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?
            return Result{Status{RippledError::rpcTOO_BUSY}};
        }
 
        auto const method = handlerProvider_->getHandler(ctx.method);
        if (!method) {
            notifyUnknownCommand();
            return Result{Status{RippledError::rpcUNKNOWN_COMMAND}};
        }
 
        try {
            LOG(perfLog_.debug()) << ctx.tag() << " start executing rpc `" << ctx.method << '`';
 
            auto const context = Context{
                .yield = ctx.yield,
                .session = ctx.session,
                .isAdmin = ctx.isAdmin,
                .clientIp = ctx.clientIp,
                .apiVersion = ctx.apiVersion
            };
            auto v = (*method).process(ctx.params, context);
 
            LOG(perfLog_.debug()) << ctx.tag() << " finish executing rpc `" << ctx.method << '`';
 
            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) {
            LOG(log_.error()) << "Database timeout";
            notifyTooBusy();
 
            return Result{Status{RippledError::rpcTOO_BUSY}};
        } catch (std::exception const& ex) {
            LOG(log_.error()) << ctx.tag() << "Caught exception: " << ex.what();
            notifyInternalError();
 
            return Result{Status{RippledError::rpcINTERNAL}};
        }
    }

    template <typename FnType>
    bool
    post(FnType&& func, std::string const& ip)
    {
        return workQueue_.get().postCoro(
            std::forward<FnType>(func), dosGuard_.get().isWhiteListed(ip)
        );
    }

    void
    notifyComplete(
        web::Context const& context,
        std::chrono::microseconds const& duration,
        bool isForwarded
    )
    {
        if (validHandler(context.method)) {
            counters_.get().rpcComplete(context.method, duration);
            if (not isForwarded) {
                counters_.get().recordLedgerRequest(context.params, context.range.maxSequence);
            }
        }
    }

    void
    recordLedgerMetrics(boost::json::object const& params, std::uint32_t currentLedgerSequence)
    {
        counters_.get().recordLedgerRequest(params, currentLedgerSequence);
    }

    void
    notifyFailed(std::string const& method)
    {
        // FIXME: seems like this is not used?
        if (validHandler(method))
            counters_.get().rpcFailed(method);
    }

    void
    notifyErrored(std::string const& method)
    {
        if (validHandler(method))
            counters_.get().rpcErrored(method);
    }

    void
    notifyTooBusy()
    {
        counters_.get().onTooBusy();
    }

    void
    notifyNotReady()
    {
        counters_.get().onNotReady();
    }

    void
    notifyBadSyntax()
    {
        counters_.get().onBadSyntax();
    }

    void
    notifyUnknownCommand()
    {
        counters_.get().onUnknownCommand();
    }

    void
    notifyInternalError()
    {
        counters_.get().onInternalError();
    }
 
private:
    bool
    validHandler(std::string const& method) const
    {
        return handlerProvider_->contains(method) || forwardingProxy_.isProxied(method);
    }
 
    Result
    buildResponseImpl(web::Context const& ctx)
    {
        if (backend_->isTooBusy()) {
            LOG(log_.error()) << "Database is too busy. Rejecting request";
            notifyTooBusy();  // TODO: should we add ctx.method if we have it?
            return Result{Status{RippledError::rpcTOO_BUSY}};
        }
 
        auto const method = handlerProvider_->getHandler(ctx.method);
        if (!method) {
            notifyUnknownCommand();
            return Result{Status{RippledError::rpcUNKNOWN_COMMAND}};
        }
 
        try {
            LOG(perfLog_.debug()) << ctx.tag() << " start executing rpc `" << ctx.method << '`';
 
            auto const context = Context{
                .yield = ctx.yield,
                .session = ctx.session,
                .isAdmin = ctx.isAdmin,
                .clientIp = ctx.clientIp,
                .apiVersion = ctx.apiVersion
            };
            auto v = (*method).process(ctx.params, context);
 
            LOG(perfLog_.debug()) << ctx.tag() << " finish executing rpc `" << ctx.method << '`';
 
            if (not v) {
                notifyErrored(ctx.method);
            }
 
            return Result{std::move(v)};
        } catch (data::DatabaseTimeout const& t) {
            LOG(log_.error()) << "Database timeout";
            notifyTooBusy();
 
            return Result{Status{RippledError::rpcTOO_BUSY}};
        } catch (std::exception const& ex) {
            LOG(log_.error()) << ctx.tag() << "Caught exception: " << ex.what();
            notifyInternalError();
 
            return Result{Status{RippledError::rpcINTERNAL}};
        }
    }
};
 
}  // namespace rpc