Validations.h

#pragma once
 
#include <xrpl/basics/Log.h>
#include <xrpl/basics/UnorderedContainers.h>
#include <xrpl/basics/chrono.h>
#include <xrpl/beast/clock/abstract_clock.h>
#include <xrpl/beast/container/aged_container_utility.h>
#include <xrpl/beast/container/aged_unordered_map.h>
#include <xrpl/beast/hash/uhash.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/consensus/LedgerTrie.h>
#include <xrpl/json/json_value.h>
 
#include <algorithm>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <mutex>
#include <optional>
#include <string>
#include <type_traits>
#include <utility>
#include <vector>
 
namespace xrpl {

struct ValidationParms
{
    explicit ValidationParms() = default;

    std::chrono::seconds validationCurrentWall = std::chrono::minutes{5};

    std::chrono::seconds validationCurrentLocal = std::chrono::minutes{3};

    std::chrono::seconds validationCurrentEarly = std::chrono::minutes{3};

    std::chrono::seconds validationSetExpires = 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.validationSetExpires))
            seq_ = Seq{0};
        if (s <= seq_)
            return false;
        seq_ = s;
        when_ = now;
        return true;
    }
 
    [[nodiscard]] 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.validationCurrentEarly)) &&
        (signTime < (now + p.validationCurrentWall)) &&
        ((seenTime == NetClock::time_point{}) || (seenTime < (now + p.validationCurrentLocal)));
}

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 = Adaptor::Mutex;
    using Validation = Adaptor::Validation;
    using Ledger = Adaptor::Ledger;
    using ID = Ledger::ID;
    using Seq = Ledger::Seq;
    using NodeID = Validation::NodeID;
    using NodeKey = 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::scoped_lock<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::scoped_lock<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::scoped_lock<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::scoped_lock<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> const 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::scoped_lock<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::scoped_lock<Mutex> const& lock, Pre&& pre, F&& f)
    {
        NetClock::time_point const 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::scoped_lock<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::AbstractClock<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::scoped_lock const 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::scoped_lock const 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_.validationCurrentWall &&
                    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 const& 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::scoped_lock const lock{mutex_};
        XRPL_ASSERT(low < high, "xrpl::Validations::setSeqToKeep : valid inputs");
        toKeep_ = {low, high};
    }

    void
    expire(beast::Journal const& j)
    {
        auto const start = std::chrono::steady_clock::now();
        {
            std::scoped_lock const 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 kRefreshTime;
                if (auto const now = byLedger_.clock().now(); kRefreshTime <= now)
                {
                    // The next refresh time is shortly before the expiration
                    // time from now.
                    kRefreshTime = now + parms_.validationSetExpires - 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_.validationSetExpires);
            beast::expire(bySequence_, parms_.validationSetExpires);
        }
        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::scoped_lock const 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::scoped_lock const lock{mutex_};
        return trie_.getJson();
    }

    std::optional<std::pair<Seq, ID>>
    getPreferred(Ledger const& curr)
    {
        std::scoped_lock const 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::ranges::max_element(acquiring_, [](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::ranges::max_element(peerCounts, [](auto const& a, auto const& 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::scoped_lock const 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::ranges::count_if(lastLedger_, [&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::scoped_lock const 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::scoped_lock const 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::scoped_lock const 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::scoped_lock const 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::scoped_lock const 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::scoped_lock const lock{mutex_};
        current_.clear();
    }

    std::size_t
    laggards(Seq const seq, hash_set<NodeKey>& trustedKeys)
    {
        std::size_t laggards = 0;
 
        current(
            std::scoped_lock{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::scoped_lock const lock{mutex_};
        return current_.size();
    }
 
    std::size_t
    sizeOfSeqEnforcersCache() const
    {
        std::scoped_lock const lock{mutex_};
        return seqEnforcers_.size();
    }
 
    std::size_t
    sizeOfByLedgerCache() const
    {
        std::scoped_lock const lock{mutex_};
        return byLedger_.size();
    }
 
    std::size_t
    sizeOfBySequenceCache() const
    {
        std::scoped_lock const lock{mutex_};
        return bySequence_.size();
    }
};
 
}  // namespace xrpl