Validations.h

#pragma once
 
#include <xrpld/consensus/LedgerTrie.h>
 
#include <xrpl/basics/Log.h>
#include <xrpl/basics/UnorderedContainers.h>
#include <xrpl/basics/chrono.h>
#include <xrpl/beast/container/aged_container_utility.h>
#include <xrpl/beast/container/aged_unordered_map.h>
#include <xrpl/protocol/PublicKey.h>
 
#include <mutex>
#include <optional>
#include <type_traits>
#include <utility>
#include <vector>
 
namespace xrpl {
 
struct ValidationParms
{
    explicit ValidationParms() = default;
 
    std::chrono::seconds validationCURRENT_WALL = std::chrono::minutes{5};
 
    std::chrono::seconds validationCURRENT_LOCAL = std::chrono::minutes{3};
 
    std::chrono::seconds validationCURRENT_EARLY = std::chrono::minutes{3};
 
    std::chrono::seconds validationSET_EXPIRES = std::chrono::minutes{10};
 
    std::chrono::seconds validationFRESHNESS = std::chrono::seconds{20};
};
 
template <class Seq>
class SeqEnforcer
{
    using time_point = std::chrono::steady_clock::time_point;
    Seq seq_{0};
    time_point when_;
 
public:
    bool
    operator()(time_point now, Seq s, ValidationParms const& p)
    {
        if (now > (when_ + p.validationSET_EXPIRES))
            seq_ = Seq{0};
        if (s <= seq_)
            return false;
        seq_ = s;
        when_ = now;
        return true;
    }
 
    Seq
    largest() const
    {
        return seq_;
    }
};
 
inline bool
isCurrent(
    ValidationParms const& p,
    NetClock::time_point now,
    NetClock::time_point signTime,
    NetClock::time_point seenTime)
{
    // Because this can be called on untrusted, possibly
    // malicious validations, we do our math in a way
    // that avoids any chance of overflowing or underflowing
    // the signing time.  All of the expressions below are
    // promoted from unsigned 32 bit to signed 64 bit prior
    // to computation.
 
    return (signTime > (now - p.validationCURRENT_EARLY)) && (signTime < (now + p.validationCURRENT_WALL)) &&
        ((seenTime == NetClock::time_point{}) || (seenTime < (now + p.validationCURRENT_LOCAL)));
}
 
enum class ValStatus {
    current,
    stale,
    badSeq,
    multiple,
    conflicting
};
 
inline std::string
to_string(ValStatus m)
{
    switch (m)
    {
        case ValStatus::current:
            return "current";
        case ValStatus::stale:
            return "stale";
        case ValStatus::badSeq:
            return "badSeq";
        case ValStatus::multiple:
            return "multiple";
        case ValStatus::conflicting:
            return "conflicting";
        default:
            return "unknown";
    }
}
 
template <class Adaptor>
class Validations
{
    using Mutex = typename Adaptor::Mutex;
    using Validation = typename Adaptor::Validation;
    using Ledger = typename Adaptor::Ledger;
    using ID = typename Ledger::ID;
    using Seq = typename Ledger::Seq;
    using NodeID = typename Validation::NodeID;
    using NodeKey = typename Validation::NodeKey;
 
    using WrappedValidationType = std::decay_t<std::invoke_result_t<decltype(&Validation::unwrap), Validation>>;
 
    // Manages concurrent access to members
    mutable Mutex mutex_;
 
    // Validations from currently listed and trusted nodes (partial and full)
    hash_map<NodeID, Validation> current_;
 
    // Used to enforce the largest validation invariant for the local node
    SeqEnforcer<Seq> localSeqEnforcer_;
 
    // Sequence of the largest validation received from each node
    hash_map<NodeID, SeqEnforcer<Seq>> seqEnforcers_;
 
    beast::aged_unordered_map<ID, hash_map<NodeID, Validation>, std::chrono::steady_clock, beast::uhash<>> byLedger_;
 
    // Partial and full validations indexed by sequence
    beast::aged_unordered_map<Seq, hash_map<NodeID, Validation>, std::chrono::steady_clock, beast::uhash<>> bySequence_;
 
    // A range [low_, high_) of validations to keep from expire
    struct KeepRange
    {
        Seq low_;
        Seq high_;
    };
    std::optional<KeepRange> toKeep_;
 
    // Represents the ancestry of validated ledgers
    LedgerTrie<Ledger> trie_;
 
    // Last (validated) ledger successfully acquired. If in this map, it is
    // accounted for in the trie.
    hash_map<NodeID, Ledger> lastLedger_;
 
    // Set of ledgers being acquired from the network
    hash_map<std::pair<Seq, ID>, hash_set<NodeID>> acquiring_;
 
    // Parameters to determine validation staleness
    ValidationParms const parms_;
 
    // Adaptor instance
    // Is NOT managed by the mutex_ above
    Adaptor adaptor_;
 
private:
    // Remove support of a validated ledger
    void
    removeTrie(std::lock_guard<Mutex> const&, NodeID const& nodeID, Validation const& val)
    {
        {
            auto it = acquiring_.find(std::make_pair(val.seq(), val.ledgerID()));
            if (it != acquiring_.end())
            {
                it->second.erase(nodeID);
                if (it->second.empty())
                    acquiring_.erase(it);
            }
        }
        {
            auto it = lastLedger_.find(nodeID);
            if (it != lastLedger_.end() && it->second.id() == val.ledgerID())
            {
                trie_.remove(it->second);
                lastLedger_.erase(nodeID);
            }
        }
    }
 
    // Check if any pending acquire ledger requests are complete
    void
    checkAcquired(std::lock_guard<Mutex> const& lock)
    {
        for (auto it = acquiring_.begin(); it != acquiring_.end();)
        {
            if (std::optional<Ledger> ledger = adaptor_.acquire(it->first.second))
            {
                for (NodeID const& nodeID : it->second)
                    updateTrie(lock, nodeID, *ledger);
 
                it = acquiring_.erase(it);
            }
            else
                ++it;
        }
    }
 
    // Update the trie to reflect a new validated ledger
    void
    updateTrie(std::lock_guard<Mutex> const&, NodeID const& nodeID, Ledger ledger)
    {
        auto const [it, inserted] = lastLedger_.emplace(nodeID, ledger);
        if (!inserted)
        {
            trie_.remove(it->second);
            it->second = ledger;
        }
        trie_.insert(ledger);
    }
 
    void
    updateTrie(
        std::lock_guard<Mutex> const& lock,
        NodeID const& nodeID,
        Validation const& val,
        std::optional<std::pair<Seq, ID>> prior)
    {
        XRPL_ASSERT(val.trusted(), "xrpl::Validations::updateTrie : trusted input validation");
 
        // Clear any prior acquiring ledger for this node
        if (prior)
        {
            auto it = acquiring_.find(*prior);
            if (it != acquiring_.end())
            {
                it->second.erase(nodeID);
                if (it->second.empty())
                    acquiring_.erase(it);
            }
        }
 
        checkAcquired(lock);
 
        std::pair<Seq, ID> valPair{val.seq(), val.ledgerID()};
        auto it = acquiring_.find(valPair);
        if (it != acquiring_.end())
        {
            it->second.insert(nodeID);
        }
        else
        {
            if (std::optional<Ledger> ledger = adaptor_.acquire(val.ledgerID()))
                updateTrie(lock, nodeID, *ledger);
            else
                acquiring_[valPair].insert(nodeID);
        }
    }
 
    template <class F>
    auto
    withTrie(std::lock_guard<Mutex> const& lock, F&& f)
    {
        // Call current to flush any stale validations
        current(lock, [](auto) {}, [](auto, auto) {});
        checkAcquired(lock);
        return f(trie_);
    }
 
    template <class Pre, class F>
    void
    current(std::lock_guard<Mutex> const& lock, Pre&& pre, F&& f)
    {
        NetClock::time_point t = adaptor_.now();
        pre(current_.size());
        auto it = current_.begin();
        while (it != current_.end())
        {
            // Check for staleness
            if (!isCurrent(parms_, t, it->second.signTime(), it->second.seenTime()))
            {
                removeTrie(lock, it->first, it->second);
                it = current_.erase(it);
            }
            else
            {
                auto cit = typename decltype(current_)::const_iterator{it};
                // contains a live record
                f(cit->first, cit->second);
                ++it;
            }
        }
    }
 
    template <class Pre, class F>
    void
    byLedger(std::lock_guard<Mutex> const&, ID const& ledgerID, Pre&& pre, F&& f)
    {
        auto it = byLedger_.find(ledgerID);
        if (it != byLedger_.end())
        {
            // Update set time since it is being used
            byLedger_.touch(it);
            pre(it->second.size());
            for (auto const& [key, val] : it->second)
                f(key, val);
        }
    }
 
public:
    template <class... Ts>
    Validations(ValidationParms const& p, beast::abstract_clock<std::chrono::steady_clock>& c, Ts&&... ts)
        : byLedger_(c), bySequence_(c), parms_(p), adaptor_(std::forward<Ts>(ts)...)
    {
    }
 
    Adaptor const&
    adaptor() const
    {
        return adaptor_;
    }
 
    ValidationParms const&
    parms() const
    {
        return parms_;
    }
 
    bool
    canValidateSeq(Seq const s)
    {
        std::lock_guard lock{mutex_};
        return localSeqEnforcer_(byLedger_.clock().now(), s, parms_);
    }
 
    ValStatus
    add(NodeID const& nodeID, Validation const& val)
    {
        if (!isCurrent(parms_, adaptor_.now(), val.signTime(), val.seenTime()))
            return ValStatus::stale;
 
        {
            std::lock_guard lock{mutex_};
 
            // Check that validation sequence is greater than any non-expired
            // validations sequence from that validator; if it's not, perform
            // additional work to detect Byzantine validations
            auto const now = byLedger_.clock().now();
 
            auto const [seqit, seqinserted] = bySequence_[val.seq()].emplace(nodeID, val);
 
            if (!seqinserted)
            {
                // Check if the entry we're already tracking was signed
                // long enough ago that we can disregard it.
                auto const diff = std::max(seqit->second.signTime(), val.signTime()) -
                    std::min(seqit->second.signTime(), val.signTime());
 
                if (diff > parms_.validationCURRENT_WALL && val.signTime() > seqit->second.signTime())
                    seqit->second = val;
            }
 
            // Enforce monotonically increasing sequences for validations
            // by a given node, and run the active Byzantine detector:
            if (auto& enf = seqEnforcers_[nodeID]; !enf(now, val.seq(), parms_))
            {
                // If the validation is for the same sequence as one we are
                // tracking, check it closely:
                if (seqit->second.seq() == val.seq())
                {
                    // Two validations for the same sequence but for different
                    // ledgers. This could be the result of misconfiguration
                    // but it can also mean a Byzantine validator.
                    if (seqit->second.ledgerID() != val.ledgerID())
                        return ValStatus::conflicting;
 
                    // Two validations for the same sequence and for the same
                    // ledger with different sign times. This could be the
                    // result of a misconfiguration but it can also mean a
                    // Byzantine validator.
                    if (seqit->second.signTime() != val.signTime())
                        return ValStatus::conflicting;
 
                    // Two validations for the same sequence but with different
                    // cookies. This is probably accidental misconfiguration.
                    if (seqit->second.cookie() != val.cookie())
                        return ValStatus::multiple;
                }
 
                return ValStatus::badSeq;
            }
 
            byLedger_[val.ledgerID()].insert_or_assign(nodeID, val);
 
            auto const [it, inserted] = current_.emplace(nodeID, val);
            if (!inserted)
            {
                // Replace existing only if this one is newer
                Validation& oldVal = it->second;
                if (val.signTime() > oldVal.signTime())
                {
                    std::pair<Seq, ID> old(oldVal.seq(), oldVal.ledgerID());
                    it->second = val;
                    if (val.trusted())
                        updateTrie(lock, nodeID, val, old);
                }
                else
                    return ValStatus::stale;
            }
            else if (val.trusted())
            {
                updateTrie(lock, nodeID, val, std::nullopt);
            }
        }
 
        return ValStatus::current;
    }
 
    void
    setSeqToKeep(Seq const& low, Seq const& high)
    {
        std::lock_guard lock{mutex_};
        XRPL_ASSERT(low < high, "xrpl::Validations::setSeqToKeep : valid inputs");
        toKeep_ = {low, high};
    }
 
    void
    expire(beast::Journal& j)
    {
        auto const start = std::chrono::steady_clock::now();
        {
            std::lock_guard lock{mutex_};
            if (toKeep_)
            {
                // We only need to refresh the keep range when it's just about
                // to expire. Track the next time we need to refresh.
                static std::chrono::steady_clock::time_point refreshTime;
                if (auto const now = byLedger_.clock().now(); refreshTime <= now)
                {
                    // The next refresh time is shortly before the expiration
                    // time from now.
                    refreshTime = now + parms_.validationSET_EXPIRES - parms_.validationFRESHNESS;
 
                    for (auto i = byLedger_.begin(); i != byLedger_.end(); ++i)
                    {
                        auto const& validationMap = i->second;
                        if (!validationMap.empty())
                        {
                            auto const seq = validationMap.begin()->second.seq();
                            if (toKeep_->low_ <= seq && seq < toKeep_->high_)
                            {
                                byLedger_.touch(i);
                            }
                        }
                    }
 
                    for (auto i = bySequence_.begin(); i != bySequence_.end(); ++i)
                    {
                        if (toKeep_->low_ <= i->first && i->first < toKeep_->high_)
                        {
                            bySequence_.touch(i);
                        }
                    }
                }
            }
 
            beast::expire(byLedger_, parms_.validationSET_EXPIRES);
            beast::expire(bySequence_, parms_.validationSET_EXPIRES);
        }
        JLOG(j.debug())
            << "Validations sets sweep lock duration "
            << std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - start).count()
            << "ms";
    }
 
    void
    trustChanged(hash_set<NodeID> const& added, hash_set<NodeID> const& removed)
    {
        std::lock_guard lock{mutex_};
 
        for (auto& [nodeId, validation] : current_)
        {
            if (added.find(nodeId) != added.end())
            {
                validation.setTrusted();
                updateTrie(lock, nodeId, validation, std::nullopt);
            }
            else if (removed.find(nodeId) != removed.end())
            {
                validation.setUntrusted();
                removeTrie(lock, nodeId, validation);
            }
        }
 
        for (auto& [_, validationMap] : byLedger_)
        {
            (void)_;
            for (auto& [nodeId, validation] : validationMap)
            {
                if (added.find(nodeId) != added.end())
                {
                    validation.setTrusted();
                }
                else if (removed.find(nodeId) != removed.end())
                {
                    validation.setUntrusted();
                }
            }
        }
    }
 
    Json::Value
    getJsonTrie() const
    {
        std::lock_guard lock{mutex_};
        return trie_.getJson();
    }
 
    std::optional<std::pair<Seq, ID>>
    getPreferred(Ledger const& curr)
    {
        std::lock_guard lock{mutex_};
        std::optional<SpanTip<Ledger>> preferred =
            withTrie(lock, [this](LedgerTrie<Ledger>& trie) { return trie.getPreferred(localSeqEnforcer_.largest()); });
        // No trusted validations to determine branch
        if (!preferred)
        {
            // fall back to majority over acquiring ledgers
            auto it = std::max_element(acquiring_.begin(), acquiring_.end(), [](auto const& a, auto const& b) {
                std::pair<Seq, ID> const& aKey = a.first;
                typename hash_set<NodeID>::size_type const& aSize = a.second.size();
                std::pair<Seq, ID> const& bKey = b.first;
                typename hash_set<NodeID>::size_type const& bSize = b.second.size();
                // order by number of trusted peers validating that ledger
                // break ties with ledger ID
                return std::tie(aSize, aKey.second) < std::tie(bSize, bKey.second);
            });
            if (it != acquiring_.end())
                return it->first;
            return std::nullopt;
        }
 
        // If we are the parent of the preferred ledger, stick with our
        // current ledger since we might be about to generate it
        if (preferred->seq == curr.seq() + Seq{1} && preferred->ancestor(curr.seq()) == curr.id())
            return std::make_pair(curr.seq(), curr.id());
 
        // A ledger ahead of us is preferred regardless of whether it is
        // a descendant of our working ledger or it is on a different chain
        if (preferred->seq > curr.seq())
            return std::make_pair(preferred->seq, preferred->id);
 
        // Only switch to earlier or same sequence number
        // if it is a different chain.
        if (curr[preferred->seq] != preferred->id)
            return std::make_pair(preferred->seq, preferred->id);
 
        // Stick with current ledger
        return std::make_pair(curr.seq(), curr.id());
    }
 
    ID
    getPreferred(Ledger const& curr, Seq minValidSeq)
    {
        std::optional<std::pair<Seq, ID>> preferred = getPreferred(curr);
        if (preferred && preferred->first >= minValidSeq)
            return preferred->second;
        return curr.id();
    }
 
    ID
    getPreferredLCL(Ledger const& lcl, Seq minSeq, hash_map<ID, std::uint32_t> const& peerCounts)
    {
        std::optional<std::pair<Seq, ID>> preferred = getPreferred(lcl);
 
        // Trusted validations exist, but stick with local preferred ledger if
        // preferred is in the past
        if (preferred)
            return (preferred->first >= minSeq) ? preferred->second : lcl.id();
 
        // Otherwise, rely on peer ledgers
        auto it = std::max_element(peerCounts.begin(), peerCounts.end(), [](auto& a, auto& b) {
            // Prefer larger counts, then larger ids on ties
            // (max_element expects this to return true if a < b)
            return std::tie(a.second, a.first) < std::tie(b.second, b.first);
        });
 
        if (it != peerCounts.end())
            return it->first;
        return lcl.id();
    }
 
    std::size_t
    getNodesAfter(Ledger const& ledger, ID const& ledgerID)
    {
        std::lock_guard lock{mutex_};
 
        // Use trie if ledger is the right one
        if (ledger.id() == ledgerID)
            return withTrie(lock, [&ledger](LedgerTrie<Ledger>& trie) {
                return trie.branchSupport(ledger) - trie.tipSupport(ledger);
            });
 
        // Count parent ledgers as fallback
        return std::count_if(lastLedger_.begin(), lastLedger_.end(), [&ledgerID](auto const& it) {
            auto const& curr = it.second;
            return curr.seq() > Seq{0} && curr[curr.seq() - Seq{1}] == ledgerID;
        });
    }
 
    std::vector<WrappedValidationType>
    currentTrusted()
    {
        std::vector<WrappedValidationType> ret;
        std::lock_guard lock{mutex_};
        current(
            lock,
            [&](std::size_t numValidations) { ret.reserve(numValidations); },
            [&](NodeID const&, Validation const& v) {
                if (v.trusted() && v.full())
                    ret.push_back(v.unwrap());
            });
        return ret;
    }
 
    auto
    getCurrentNodeIDs() -> hash_set<NodeID>
    {
        hash_set<NodeID> ret;
        std::lock_guard lock{mutex_};
        current(
            lock,
            [&](std::size_t numValidations) { ret.reserve(numValidations); },
            [&](NodeID const& nid, Validation const&) { ret.insert(nid); });
 
        return ret;
    }
 
    std::size_t
    numTrustedForLedger(ID const& ledgerID)
    {
        std::size_t count = 0;
        std::lock_guard lock{mutex_};
        byLedger(
            lock,
            ledgerID,
            [&](std::size_t) {},  // nothing to reserve
            [&](NodeID const&, Validation const& v) {
                if (v.trusted() && v.full())
                    ++count;
            });
        return count;
    }
 
    std::vector<WrappedValidationType>
    getTrustedForLedger(ID const& ledgerID, Seq const& seq)
    {
        std::vector<WrappedValidationType> res;
        std::lock_guard lock{mutex_};
        byLedger(
            lock,
            ledgerID,
            [&](std::size_t numValidations) { res.reserve(numValidations); },
            [&](NodeID const&, Validation const& v) {
                if (v.trusted() && v.full() && v.seq() == seq)
                    res.emplace_back(v.unwrap());
            });
 
        return res;
    }
 
    std::vector<std::uint32_t>
    fees(ID const& ledgerID, std::uint32_t baseFee)
    {
        std::vector<std::uint32_t> res;
        std::lock_guard lock{mutex_};
        byLedger(
            lock,
            ledgerID,
            [&](std::size_t numValidations) { res.reserve(numValidations); },
            [&](NodeID const&, Validation const& v) {
                if (v.trusted() && v.full())
                {
                    std::optional<std::uint32_t> loadFee = v.loadFee();
                    if (loadFee)
                        res.push_back(*loadFee);
                    else
                        res.push_back(baseFee);
                }
            });
        return res;
    }
 
    void
    flush()
    {
        std::lock_guard lock{mutex_};
        current_.clear();
    }
 
    std::size_t
    laggards(Seq const seq, hash_set<NodeKey>& trustedKeys)
    {
        std::size_t laggards = 0;
 
        current(
            std::lock_guard{mutex_},
            [](std::size_t) {},
            [&](NodeID const&, Validation const& v) {
                if (adaptor_.now() < v.seenTime() + parms_.validationFRESHNESS &&
                    trustedKeys.find(v.key()) != trustedKeys.end())
                {
                    trustedKeys.erase(v.key());
                    if (seq > v.seq())
                        ++laggards;
                }
            });
 
        return laggards;
    }
 
    std::size_t
    sizeOfCurrentCache() const
    {
        std::lock_guard lock{mutex_};
        return current_.size();
    }
 
    std::size_t
    sizeOfSeqEnforcersCache() const
    {
        std::lock_guard lock{mutex_};
        return seqEnforcers_.size();
    }
 
    std::size_t
    sizeOfByLedgerCache() const
    {
        std::lock_guard lock{mutex_};
        return byLedger_.size();
    }
 
    std::size_t
    sizeOfBySequenceCache() const
    {
        std::lock_guard lock{mutex_};
        return bySequence_.size();
    }
};
 
}  // namespace xrpl