src/xrpld/overlay/Slot.h

#pragma once
 
#include <xrpld/core/Config.h>
#include <xrpld/overlay/Peer.h>
#include <xrpld/overlay/ReduceRelayCommon.h>
 
#include <xrpl/basics/Log.h>
#include <xrpl/basics/Slice.h>
#include <xrpl/basics/UnorderedContainers.h>
#include <xrpl/basics/base_uint.h>
#include <xrpl/basics/hardened_hash.h>
#include <xrpl/basics/random.h>
#include <xrpl/beast/container/aged_unordered_map.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/core/ServiceRegistry.h>
#include <xrpl/protocol/PublicKey.h>
 
#include <xrpl.pb.h>
 
#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <iterator>
#include <optional>
#include <set>
#include <sstream>
#include <tuple>
#include <unordered_map>
#include <unordered_set>
#include <vector>
 
namespace xrpl::reduce_relay {
 
template <typename ClockType>
class Slots;

enum class PeerState : uint8_t {
    Counting,   // counting messages
    Selected,   // selected to relay, counting if Slot in Counting
    Squelched,  // squelched, doesn't relay
};

enum class SlotState : uint8_t {
    Counting,  // counting messages
    Selected,  // peers selected, stop counting
};
 
template <typename Unit, typename TP>
Unit
epoch(TP const& t)
{
    return std::chrono::duration_cast<Unit>(t.time_since_epoch());
}

class SquelchHandler
{
public:
    virtual ~SquelchHandler() = default;
    virtual void
    squelch(PublicKey const& validator, Peer::id_t id, std::uint32_t duration) const = 0;
    virtual void
    unsquelch(PublicKey const& validator, Peer::id_t id) const = 0;
};

template <typename ClockType>
class Slot final
{
private:
    friend class Slots<ClockType>;
    using id_t = Peer::id_t;
    using time_point = ClockType::time_point;
 
    // a callback to report ignored squelches
    using ignored_squelch_callback = std::function<void()>;

    Slot(SquelchHandler const& handler, beast::Journal journal, uint16_t maxSelectedPeers)
        : lastSelected_(ClockType::now())
        , handler_(handler)
        , journal_(journal)
        , maxSelectedPeers_(maxSelectedPeers)
    {
    }

    void
    update(
        PublicKey const& validator,
        id_t id,
        protocol::MessageType type,
        ignored_squelch_callback callback);

    void
    deletePeer(PublicKey const& validator, id_t id, bool erase);

    [[nodiscard]] time_point const&
    getLastSelected() const
    {
        return lastSelected_;
    }

    [[nodiscard]] std::uint16_t
    inState(PeerState state) const;

    [[nodiscard]] std::uint16_t
    notInState(PeerState state) const;

    [[nodiscard]] SlotState
    getState() const
    {
        return state_;
    }

    [[nodiscard]] std::set<id_t>
    getSelected() const;

    [[nodiscard]] std::unordered_map<id_t, std::tuple<PeerState, uint16_t, uint32_t, uint32_t>>
    getPeers() const;

    void
    deleteIdlePeer(PublicKey const& validator);

    std::chrono::seconds
    getSquelchDuration(std::size_t npeers);
 
private:
    void
    resetCounts();

    void
    initCounting();

    struct PeerInfo
    {
        PeerState state;         // peer's state
        std::size_t count;       // message count
        time_point expire;       // squelch expiration time
        time_point lastMessage;  // time last message received
    };
 
    std::unordered_map<id_t, PeerInfo> peers_;  // peer's data
 
    // pool of peers considered as the source of messages
    // from validator - peers that reached kMinMessageThreshold
    std::unordered_set<id_t> considered_;
 
    // number of peers that reached kMaxMessageThreshold
    std::uint16_t reachedThreshold_{0};
 
    // last time peers were selected, used to age the slot
    ClockType::time_point lastSelected_;
 
    SlotState state_{SlotState::Counting};  // slot's state
    SquelchHandler const& handler_;         // squelch/unsquelch handler
    beast::Journal const journal_;          // logging
 
    // the maximum number of peers that should be selected as a validator
    // message source
    uint16_t const maxSelectedPeers_;
};
 
template <typename ClockType>
void
Slot<ClockType>::deleteIdlePeer(PublicKey const& validator)
{
    using namespace std::chrono;
    auto now = ClockType::now();
    for (auto it = peers_.begin(); it != peers_.end();)
    {
        auto& peer = it->second;
        auto id = it->first;
        ++it;
        if (now - peer.lastMessage > kIdled)
        {
            JLOG(journal_.trace())
                << "deleteIdlePeer: " << Slice(validator) << " " << id << " idled "
                << duration_cast<seconds>(now - peer.lastMessage).count() << " selected "
                << (peer.state == PeerState::Selected);
            deletePeer(validator, id, false);
        }
    }
}
 
template <typename ClockType>
void
Slot<ClockType>::update(
    PublicKey const& validator,
    id_t id,
    protocol::MessageType type,
    ignored_squelch_callback callback)
{
    using namespace std::chrono;
    auto now = ClockType::now();
    auto it = peers_.find(id);
    // First message from this peer
    if (it == peers_.end())
    {
        JLOG(journal_.trace()) << "update: adding peer " << Slice(validator) << " " << id;
        peers_.emplace(std::make_pair(id, PeerInfo{PeerState::Counting, 0, now, now}));
        initCounting();
        return;
    }
    // Message from a peer with expired squelch
    if (it->second.state == PeerState::Squelched && now > it->second.expire)
    {
        JLOG(journal_.trace()) << "update: squelch expired " << Slice(validator) << " " << id;
        it->second.state = PeerState::Counting;
        it->second.lastMessage = now;
        initCounting();
        return;
    }
 
    auto& peer = it->second;
 
    JLOG(journal_.trace()) << "update: existing peer " << Slice(validator) << " " << id
                           << " slot state " << static_cast<int>(state_) << " peer state "
                           << static_cast<int>(peer.state) << " count " << peer.count << " last "
                           << duration_cast<milliseconds>(now - peer.lastMessage).count()
                           << " pool " << considered_.size() << " threshold " << reachedThreshold_
                           << " " << (type == protocol::mtVALIDATION ? "validation" : "proposal");
 
    peer.lastMessage = now;
 
    // report if we received a message from a squelched peer
    if (peer.state == PeerState::Squelched)
        callback();
 
    if (state_ != SlotState::Counting || peer.state == PeerState::Squelched)
        return;
 
    if (++peer.count > kMinMessageThreshold)
        considered_.insert(id);
    if (peer.count == (kMaxMessageThreshold + 1))
        ++reachedThreshold_;
 
    if (now - lastSelected_ > 2 * kMaxUnsquelchExpireDefault)
    {
        JLOG(journal_.trace()) << "update: resetting due to inactivity " << Slice(validator) << " "
                               << id << " " << duration_cast<seconds>(now - lastSelected_).count();
        initCounting();
        return;
    }
 
    if (reachedThreshold_ == maxSelectedPeers_)
    {
        // Randomly select maxSelectedPeers_ peers from considered.
        // Exclude peers that have been idling > IDLED -
        // it's possible that deleteIdlePeer() has not been called yet.
        // If number of remaining peers != maxSelectedPeers_
        // then reset the Counting state and let deleteIdlePeer() handle
        // idled peers.
        std::unordered_set<id_t> selected;
        auto const consideredPoolSize = considered_.size();
        while (selected.size() != maxSelectedPeers_ && !considered_.empty())
        {
            auto i = considered_.size() == 1 ? 0 : randInt(considered_.size() - 1);
            auto it = std::next(considered_.begin(), i);
            auto id = *it;
            considered_.erase(it);
            auto const& itPeers = peers_.find(id);
            if (itPeers == peers_.end())
            {
                JLOG(journal_.error())
                    << "update: peer not found " << Slice(validator) << " " << id;
                continue;
            }
            if (now - itPeers->second.lastMessage < kIdled)
                selected.insert(id);
        }
 
        if (selected.size() != maxSelectedPeers_)
        {
            JLOG(journal_.trace()) << "update: selection failed " << Slice(validator) << " " << id;
            initCounting();
            return;
        }
 
        lastSelected_ = now;
 
        auto s = selected.begin();
        JLOG(journal_.trace()) << "update: " << Slice(validator) << " " << id << " pool size "
                               << consideredPoolSize << " selected " << *s << " "
                               << *std::next(s, 1) << " " << *std::next(s, 2);
 
        XRPL_ASSERT(
            peers_.size() >= maxSelectedPeers_, "xrpl::reduce_relay::Slot::update : minimum peers");
 
        // squelch peers which are not selected and
        // not already squelched
        std::stringstream str;
        for (auto& [k, v] : peers_)
        {
            v.count = 0;
 
            if (selected.find(k) != selected.end())
            {
                v.state = PeerState::Selected;
            }
            else if (v.state != PeerState::Squelched)
            {
                if (journal_.trace())
                    str << k << " ";
                v.state = PeerState::Squelched;
                std::chrono::seconds const duration =
                    getSquelchDuration(peers_.size() - maxSelectedPeers_);
                v.expire = now + duration;
                handler_.squelch(validator, k, duration.count());
            }
        }
        JLOG(journal_.trace()) << "update: squelching " << Slice(validator) << " " << id << " "
                               << str.str();
        considered_.clear();
        reachedThreshold_ = 0;
        state_ = SlotState::Selected;
    }
}
 
template <typename ClockType>
std::chrono::seconds
Slot<ClockType>::getSquelchDuration(std::size_t npeers)
{
    using namespace std::chrono;
    auto m = std::max(kMaxUnsquelchExpireDefault, seconds{kSquelchPerPeer * npeers});
    if (m > kMaxUnsquelchExpirePeers)
    {
        m = kMaxUnsquelchExpirePeers;
        JLOG(journal_.warn()) << "getSquelchDuration: unexpected squelch duration " << npeers;
    }
    return seconds{xrpl::randInt(kMinUnsquelchExpire / 1s, m / 1s)};
}
 
template <typename ClockType>
void
Slot<ClockType>::deletePeer(PublicKey const& validator, id_t id, bool erase)
{
    auto it = peers_.find(id);
    if (it != peers_.end())
    {
        std::vector<Peer::id_t> toUnsquelch;
 
        JLOG(journal_.trace()) << "deletePeer: " << Slice(validator) << " " << id << " selected "
                               << (it->second.state == PeerState::Selected) << " considered "
                               << (considered_.contains(id)) << " erase " << erase;
        auto now = ClockType::now();
        if (it->second.state == PeerState::Selected)
        {
            for (auto& [k, v] : peers_)
            {
                if (v.state == PeerState::Squelched)
                    toUnsquelch.push_back(k);
                v.state = PeerState::Counting;
                v.count = 0;
                v.expire = now;
            }
 
            considered_.clear();
            reachedThreshold_ = 0;
            state_ = SlotState::Counting;
        }
        else if (considered_.contains(id))
        {
            if (it->second.count > kMaxMessageThreshold)
                --reachedThreshold_;
            considered_.erase(id);
        }
 
        it->second.lastMessage = now;
        it->second.count = 0;
 
        if (erase)
            peers_.erase(it);
 
        // Must be after peers_.erase(it)
        for (auto const& k : toUnsquelch)
            handler_.unsquelch(validator, k);
    }
}
 
template <typename ClockType>
void
Slot<ClockType>::resetCounts()
{
    for (auto& [_, peer] : peers_)
    {
        (void)_;
        peer.count = 0;
    }
}
 
template <typename ClockType>
void
Slot<ClockType>::initCounting()
{
    state_ = SlotState::Counting;
    considered_.clear();
    reachedThreshold_ = 0;
    resetCounts();
}
 
template <typename ClockType>
std::uint16_t
Slot<ClockType>::inState(PeerState state) const
{
    return std::count_if(
        peers_.begin(), peers_.end(), [&](auto const& it) { return (it.second.state == state); });
}
 
template <typename ClockType>
std::uint16_t
Slot<ClockType>::notInState(PeerState state) const
{
    return std::count_if(
        peers_.begin(), peers_.end(), [&](auto const& it) { return (it.second.state != state); });
}
 
template <typename ClockType>
std::set<Peer::id_t>
Slot<ClockType>::getSelected() const
{
    std::set<id_t> r;
    for (auto const& [id, info] : peers_)
    {
        if (info.state == PeerState::Selected)
            r.insert(id);
    }
    return r;
}
 
template <typename ClockType>
std::unordered_map<Peer::id_t, std::tuple<PeerState, uint16_t, uint32_t, uint32_t>>
Slot<ClockType>::getPeers() const
{
    using namespace std::chrono;
    auto r = std::
        unordered_map<id_t, std::tuple<PeerState, std::uint16_t, std::uint32_t, std::uint32_t>>();
 
    for (auto const& [id, info] : peers_)
    {
        r.emplace(
            std::make_pair(
                id,
                std::move(
                    std::make_tuple(
                        info.state,
                        info.count,
                        epoch<milliseconds>(info.expire).count(),
                        epoch<milliseconds>(info.lastMessage).count()))));
    }
 
    return r;
}

template <typename ClockType>
class Slots final
{
    using time_point = ClockType::time_point;
    using id_t = Peer::id_t;
    using messages = beast::aged_unordered_map<
        uint256,
        std::unordered_set<Peer::id_t>,
        ClockType,
        HardenedHash<strong_hash>>;
 
public:
    Slots(ServiceRegistry& registry, SquelchHandler const& handler, Config const& config)
        : handler_(handler)
        , logs_(registry.getLogs())
        , journal_(registry.getJournal("Slots"))
        , baseSquelchEnabled_(config.vpReduceRelayBaseSquelchEnable)
        , maxSelectedPeers_(config.vpReduceRelaySquelchMaxSelectedPeers)
    {
    }
    ~Slots() = default;

    bool
    baseSquelchReady()
    {
        return baseSquelchEnabled_ && reduceRelayReady();
    }

    bool
    reduceRelayReady()
    {
        if (!reduceRelayReady_)
        {
            reduceRelayReady_ = reduce_relay::epoch<std::chrono::minutes>(ClockType::now()) >
                reduce_relay::kWaitOnBootup;
        }
 
        return reduceRelayReady_;
    }

    void
    updateSlotAndSquelch(
        uint256 const& key,
        PublicKey const& validator,
        id_t id,
        protocol::MessageType type)
    {
        updateSlotAndSquelch(key, validator, id, type, []() {});
    }

    void
    updateSlotAndSquelch(
        uint256 const& key,
        PublicKey const& validator,
        id_t id,
        protocol::MessageType type,
        Slot<ClockType>::ignored_squelch_callback callback);

    void
    deleteIdlePeers();

    [[nodiscard]] std::optional<std::uint16_t>
    inState(PublicKey const& validator, PeerState state) const
    {
        auto const& it = slots_.find(validator);
        if (it != slots_.end())
            return it->second.inState(state);
        return {};
    }

    [[nodiscard]] std::optional<std::uint16_t>
    notInState(PublicKey const& validator, PeerState state) const
    {
        auto const& it = slots_.find(validator);
        if (it != slots_.end())
            return it->second.notInState(state);
        return {};
    }

    [[nodiscard]] bool
    inState(PublicKey const& validator, SlotState state) const
    {
        auto const& it = slots_.find(validator);
        if (it != slots_.end())
            return it->second.state_ == state;
        return false;
    }

    std::set<id_t>
    getSelected(PublicKey const& validator)
    {
        auto const& it = slots_.find(validator);
        if (it != slots_.end())
            return it->second.getSelected();
        return {};
    }

    std::unordered_map<Peer::id_t, std::tuple<PeerState, uint16_t, uint32_t, std::uint32_t>>
    getPeers(PublicKey const& validator)
    {
        auto const& it = slots_.find(validator);
        if (it != slots_.end())
            return it->second.getPeers();
        return {};
    }

    std::optional<SlotState>
    getState(PublicKey const& validator)
    {
        auto const& it = slots_.find(validator);
        if (it != slots_.end())
            return it->second.getState();
        return {};
    }

    void
    deletePeer(id_t id, bool erase);
 
private:
    bool
    addPeerMessage(uint256 const& key, id_t id);
 
    std::atomic_bool reduceRelayReady_{false};
 
    hash_map<PublicKey, Slot<ClockType>> slots_;
    SquelchHandler const& handler_;  // squelch/unsquelch handler
    Logs& logs_;
    beast::Journal const journal_;
 
    bool const baseSquelchEnabled_;
    uint16_t const maxSelectedPeers_;
 
    // Maintain aged container of message/peers. This is required
    // to discard duplicate message from the same peer. A message
    // is aged after IDLED seconds. A message received IDLED seconds
    // after it was relayed is ignored by PeerImp.
    inline static messages peersWithMessage{beast::getAbstractClock<ClockType>()};
};
 
template <typename ClockType>
bool
Slots<ClockType>::addPeerMessage(uint256 const& key, id_t id)
{
    beast::expire(peersWithMessage, reduce_relay::kIdled);
 
    if (key.isNonZero())
    {
        auto it = peersWithMessage.find(key);
        if (it == peersWithMessage.end())
        {
            JLOG(journal_.trace()) << "addPeerMessage: new " << to_string(key) << " " << id;
            peersWithMessage.emplace(key, std::unordered_set<id_t>{id});
            return true;
        }
 
        if (it->second.find(id) != it->second.end())
        {
            JLOG(journal_.trace())
                << "addPeerMessage: duplicate message " << to_string(key) << " " << id;
            return false;
        }
 
        JLOG(journal_.trace()) << "addPeerMessage: added " << to_string(key) << " " << id;
 
        it->second.insert(id);
    }
 
    return true;
}
 
template <typename ClockType>
void
Slots<ClockType>::updateSlotAndSquelch(
    uint256 const& key,
    PublicKey const& validator,
    id_t id,
    protocol::MessageType type,
    Slot<ClockType>::ignored_squelch_callback callback)
{
    if (!addPeerMessage(key, id))
        return;
 
    auto it = slots_.find(validator);
    if (it == slots_.end())
    {
        JLOG(journal_.trace()) << "updateSlotAndSquelch: new slot " << Slice(validator);
        auto it = slots_
                      .emplace(
                          std::make_pair(
                              validator,
                              Slot<ClockType>(handler_, logs_.journal("Slot"), maxSelectedPeers_)))
                      .first;
        it->second.update(validator, id, type, callback);
    }
    else
    {
        it->second.update(validator, id, type, callback);
    }
}
 
template <typename ClockType>
void
Slots<ClockType>::deletePeer(id_t id, bool erase)
{
    for (auto& [validator, slot] : slots_)
        slot.deletePeer(validator, id, erase);
}
 
template <typename ClockType>
void
Slots<ClockType>::deleteIdlePeers()
{
    auto now = ClockType::now();
 
    for (auto it = slots_.begin(); it != slots_.end();)
    {
        it->second.deleteIdlePeer(it->first);
        if (now - it->second.getLastSelected() > kMaxUnsquelchExpireDefault)
        {
            JLOG(journal_.trace()) << "deleteIdlePeers: deleting idle slot " << Slice(it->first);
            it = slots_.erase(it);
        }
        else
        {
            ++it;
        }
    }
}
 
}  // namespace xrpl::reduce_relay