test: Modularize Peerfinder component and migrate Peerfinder tests from Beast to GTest and GMock (#7054)

Co-authored-by: Alex Kremer <akremer@ripple.com>

include/xrpl/beast/container/detail/aged_unordered_container.h

@@ -5,6 +5,7 @@
 #include <xrpl/beast/container/detail/aged_associative_container.h>
 #include <xrpl/beast/container/detail/aged_container_iterator.h>
 #include <xrpl/beast/container/detail/empty_base_optimization.h>
+#include <xrpl/beast/utility/instrumentation.h>
 
 #include <boost/intrusive/list.hpp>
 #include <boost/intrusive/unordered_set.hpp>

include/xrpl/peerfinder/Config.h

@@ -0,0 +1,163 @@
+#pragma once
+
+#include <xrpl/beast/utility/PropertyStream.h>
+#include <xrpl/peerfinder/detail/Tuning.h>
+
+#include <cstddef>
+#include <cstdint>
+#include <optional>
+#include <string>
+#include <string_view>
+
+namespace xrpl::PeerFinder {
+
+struct PeerLimitConfig
+{
+    std::optional<std::size_t> maxPeers;
+    std::optional<std::size_t> inPeers;
+    std::optional<std::size_t> outPeers;
+};
+
+/**
+ * PeerFinder configuration settings.
+ */
+struct Config
+{
+    /**
+     * The largest number of public peer slots to allow.
+     * This includes both inbound and outbound, but does not include
+     * fixed peers.
+     */
+    std::size_t maxPeers{Tuning::kDefaultMaxPeers};
+
+    /**
+     * The number of automatic outbound connections to maintain.
+     * Outbound connections are only maintained if autoConnect
+     * is `true`.
+     */
+    std::size_t outPeers = calcOutPeers();  // Note: relies on `maxPeers` being initialized
+
+    /**
+     * The number of automatic inbound connections to maintain.
+     * Inbound connections are only maintained if wantIncoming
+     * is `true`.
+     */
+    std::size_t inPeers{0};
+
+    /**
+     * `true` if we want our IP address kept private.
+     */
+    bool peerPrivate = true;
+
+    /**
+     * `true` if we want to accept incoming connections.
+     */
+    bool wantIncoming{true};
+
+    /**
+     * `true` if we want to establish connections automatically
+     */
+    bool autoConnect{true};
+
+    /**
+     * The listening port number.
+     */
+    std::uint16_t listeningPort{0};
+
+    /**
+     * The set of features we advertise.
+     */
+    std::string features;
+
+    /**
+     * Limit how many incoming connections we allow per IP
+     */
+    int ipLimit{0};
+
+    /**
+     * `true` if we want to verify endpoints in TMEndpoints messages
+     */
+    bool verifyEndpoints = true;
+
+    //--------------------------------------------------------------------------
+
+    /**
+     * Returns a suitable value for outPeers according to the rules.
+     */
+    [[nodiscard]] std::size_t
+    calcOutPeers() const;
+
+    /**
+     * Adjusts the values so they follow the business rules.
+     */
+    void
+    applyTuning();
+
+    /**
+     * Write the configuration into a property stream
+     */
+    void
+    onWrite(beast::PropertyStream::Map& map) const;
+
+    /**
+     * Make PeerFinder::Config from peer limit and server mode parameters.
+     */
+    static Config
+    makeConfig(
+        bool peerPrivate,
+        bool standalone,
+        PeerLimitConfig const& limits,
+        std::uint16_t port,
+        bool validationPublicKey,
+        int ipLimit,
+        bool verifyEndpoints);
+
+    /**
+     * Compares two configurations for equality field by field.
+     */
+    friend bool
+    operator==(Config const& lhs, Config const& rhs) = default;
+};
+
+//------------------------------------------------------------------------------
+
+/**
+ * Possible results from activating a slot.
+ */
+enum class Result { InboundDisabled, DuplicatePeer, IpLimitExceeded, Full, Success };
+
+/**
+ * @brief Converts a `Result` enum value to its string representation.
+ *
+ * This function provides a human-readable string for a given `Result` enum,
+ * which is useful for logging, debugging, or displaying status messages.
+ *
+ * @param result The `Result` enum value to convert.
+ * @return A `std::string_view` representing the enum value. Returns "unknown"
+ * if the enum value is not explicitly handled.
+ *
+ * @note This function returns a `std::string_view` for performance.
+ * A `std::string` would need to allocate memory on the heap and copy the
+ * string literal into it every time the function is called.
+ */
+inline std::string_view
+to_string(Result result) noexcept
+{
+    switch (result)
+    {
+        case Result::InboundDisabled:
+            return "inbound disabled";
+        case Result::DuplicatePeer:
+            return "peer already connected";
+        case Result::IpLimitExceeded:
+            return "ip limit exceeded";
+        case Result::Full:
+            return "slots full";
+        case Result::Success:
+            return "success";
+    }
+
+    return "unknown";
+}
+
+}  // namespace xrpl::PeerFinder

include/xrpl/peerfinder/PeerfinderManager.h

@@ -0,0 +1,179 @@
+#pragma once
+
+#include <xrpl/beast/net/IPEndpoint.h>
+#include <xrpl/beast/utility/PropertyStream.h>
+#include <xrpl/peerfinder/Config.h>
+#include <xrpl/peerfinder/Slot.h>
+#include <xrpl/peerfinder/Types.h>
+#include <xrpl/protocol/PublicKey.h>
+
+#include <boost/asio/ip/tcp.hpp>
+
+#include <memory>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <vector>
+
+namespace xrpl::PeerFinder {
+
+/**
+ * Maintains a set of IP addresses used for getting into the network.
+ */
+class Manager : public beast::PropertyStream::Source
+{
+protected:
+    Manager() noexcept;
+
+public:
+    /**
+     * Destroy the object.
+     * Any pending source fetch operations are aborted.
+     * There may be some listener calls made before the
+     * destructor returns.
+     */
+    ~Manager() override = default;
+
+    /**
+     * Set the configuration for the manager.
+     * The new settings will be applied asynchronously.
+     * Thread safety:
+     *     Can be called from any threads at any time.
+     */
+    virtual void
+    setConfig(Config const& config) = 0;
+
+    /**
+     * Transition to the started state, synchronously.
+     */
+    virtual void
+    start() = 0;
+
+    /**
+     * Transition to the stopped state, synchronously.
+     */
+    virtual void
+    stop() = 0;
+
+    /**
+     * Returns the configuration for the manager.
+     */
+    virtual Config
+    config() = 0;
+
+    /**
+     * Add a peer that should always be connected.
+     * This is useful for maintaining a private cluster of peers.
+     * The string is the name as specified in the configuration
+     * file, along with the set of corresponding IP addresses.
+     */
+    virtual void
+    addFixedPeer(std::string_view name, std::vector<beast::IP::Endpoint> const& addresses) = 0;
+
+    /**
+     * Add a set of strings as fallback IP::Endpoint sources.
+     * @param name A label used for diagnostics.
+     */
+    virtual void
+    addFallbackStrings(std::string const& name, std::vector<std::string> const& strings) = 0;
+
+    /**
+     * Add a URL as a fallback location to obtain IP::Endpoint sources.
+     * @param name A label used for diagnostics.
+     */
+    /* VFALCO NOTE Unimplemented
+    virtual void addFallbackURL (std::string const& name,
+        std::string const& url) = 0;
+    */
+
+    //--------------------------------------------------------------------------
+
+    /**
+     * Create a new inbound slot with the specified remote endpoint.
+     * If nullptr is returned, then the slot could not be assigned.
+     * Usually this is because of a detected self-connection.
+     */
+    virtual std::pair<std::shared_ptr<Slot>, Result>
+    newInboundSlot(
+        beast::IP::Endpoint const& localEndpoint,
+        beast::IP::Endpoint const& remoteEndpoint) = 0;
+
+    /**
+     * Create a new outbound slot with the specified remote endpoint.
+     * If nullptr is returned, then the slot could not be assigned.
+     * Usually this is because of a duplicate connection.
+     */
+    virtual std::pair<std::shared_ptr<Slot>, Result>
+    newOutboundSlot(beast::IP::Endpoint const& remoteEndpoint) = 0;
+
+    /**
+     * Called when mtENDPOINTS is received.
+     */
+    virtual void
+    onEndpoints(std::shared_ptr<Slot> const& slot, Endpoints const& endpoints) = 0;
+
+    /**
+     * Called when the slot is closed.
+     * This always happens when the socket is closed, unless the socket
+     * was canceled.
+     */
+    virtual void
+    onClosed(std::shared_ptr<Slot> const& slot) = 0;
+
+    /**
+     * Called when an outbound connection is deemed to have failed
+     */
+    virtual void
+    onFailure(std::shared_ptr<Slot> const& slot) = 0;
+
+    /**
+     * Called when we received redirect IPs from a busy peer.
+     */
+    virtual void
+    onRedirects(
+        boost::asio::ip::tcp::endpoint const& remoteAddress,
+        std::vector<boost::asio::ip::tcp::endpoint> const& eps) = 0;
+
+    //--------------------------------------------------------------------------
+
+    /**
+     * Called when an outbound connection attempt succeeds.
+     * The local endpoint must be valid. If the caller receives an error
+     * when retrieving the local endpoint from the socket, it should
+     * proceed as if the connection attempt failed by calling on_closed
+     * instead of on_connected.
+     * @return `true` if the connection should be kept
+     */
+    virtual bool
+    onConnected(std::shared_ptr<Slot> const& slot, beast::IP::Endpoint const& localEndpoint) = 0;
+
+    /**
+     * Request an active slot type.
+     */
+    virtual Result
+    activate(std::shared_ptr<Slot> const& slot, PublicKey const& key, bool reserved) = 0;
+
+    /**
+     * Returns a set of endpoints suitable for redirection.
+     */
+    virtual std::vector<Endpoint>
+    redirect(std::shared_ptr<Slot> const& slot) = 0;
+
+    /**
+     * Return a set of addresses we should connect to.
+     */
+    virtual std::vector<beast::IP::Endpoint>
+    autoconnect() = 0;
+
+    virtual std::vector<std::pair<std::shared_ptr<Slot>, std::vector<Endpoint>>>
+    buildEndpointsForPeers() = 0;
+
+    /**
+     * Perform periodic activity.
+     * This should be called once per second.
+     */
+    virtual void
+    oncePerSecond() = 0;
+};
+
+}  // namespace xrpl::PeerFinder

include/xrpl/peerfinder/Slot.h

@@ -0,0 +1,75 @@
+#pragma once
+
+#include <xrpl/beast/net/IPEndpoint.h>
+#include <xrpl/protocol/PublicKey.h>
+
+#include <cstdint>
+#include <memory>
+#include <optional>
+
+namespace xrpl::PeerFinder {
+
+/**
+ * Properties and state associated with a peer to peer overlay connection.
+ */
+class Slot
+{
+public:
+    using ptr = std::shared_ptr<Slot>;
+
+    enum class State { Accept, Connect, Connected, Active, Closing };
+
+    virtual ~Slot() = 0;
+
+    /**
+     * Returns `true` if this is an inbound connection.
+     */
+    [[nodiscard]] virtual bool
+    inbound() const = 0;
+
+    /**
+     * Returns `true` if this is a fixed connection.
+     * A connection is fixed if its remote endpoint is in the list of
+     * remote endpoints for fixed connections.
+     */
+    [[nodiscard]] virtual bool
+    fixed() const = 0;
+
+    /**
+     * Returns `true` if this is a reserved connection.
+     * It might be a cluster peer, or a peer with a reservation.
+     * This is only known after then handshake completes.
+     */
+    [[nodiscard]] virtual bool
+    reserved() const = 0;
+
+    /**
+     * Returns the state of the connection.
+     */
+    [[nodiscard]] virtual State
+    state() const = 0;
+
+    /**
+     * The remote endpoint of socket.
+     */
+    [[nodiscard]] virtual beast::IP::Endpoint const&
+    remoteEndpoint() const = 0;
+
+    /**
+     * The local endpoint of the socket, when known.
+     */
+    [[nodiscard]] virtual std::optional<beast::IP::Endpoint> const&
+    localEndpoint() const = 0;
+
+    [[nodiscard]] virtual std::optional<std::uint16_t>
+    listeningPort() const = 0;
+
+    /**
+     * The peer's public key, when known.
+     * The public key is established when the handshake is complete.
+     */
+    [[nodiscard]] virtual std::optional<PublicKey> const&
+    publicKey() const = 0;
+};
+
+}  // namespace xrpl::PeerFinder

include/xrpl/peerfinder/Types.h

@@ -0,0 +1,46 @@
+#pragma once
+
+#include <xrpl/beast/clock/abstract_clock.h>
+#include <xrpl/beast/net/IPEndpoint.h>
+#include <xrpl/peerfinder/detail/Tuning.h>
+
+#include <chrono>
+#include <cstdint>
+#include <vector>
+
+namespace xrpl::PeerFinder {
+
+using clock_type = beast::AbstractClock<std::chrono::steady_clock>;
+
+/**
+ * Represents a set of addresses.
+ */
+using IPAddresses = std::vector<beast::IP::Endpoint>;
+
+//------------------------------------------------------------------------------
+
+/**
+ * Describes a connectable peer address along with some metadata.
+ */
+struct Endpoint
+{
+    Endpoint() = default;
+
+    Endpoint(beast::IP::Endpoint ep, std::uint32_t hops);
+
+    std::uint32_t hops = 0;
+    beast::IP::Endpoint address;
+};
+
+inline bool
+operator<(Endpoint const& lhs, Endpoint const& rhs)
+{
+    return lhs.address < rhs.address;
+}
+
+/**
+ * A set of Endpoint used for connecting.
+ */
+using Endpoints = std::vector<Endpoint>;
+
+}  // namespace xrpl::PeerFinder

include/xrpl/peerfinder/detail/Bootcache.h

@@ -0,0 +1,192 @@
+#pragma once
+
+#include <xrpl/beast/net/IPEndpoint.h>
+#include <xrpl/beast/utility/Journal.h>
+#include <xrpl/beast/utility/PropertyStream.h>
+#include <xrpl/peerfinder/Types.h>
+#include <xrpl/peerfinder/detail/Store.h>
+#include <xrpl/peerfinder/detail/Tuning.h>
+
+#include <boost/bimap.hpp>
+#include <boost/bimap/multiset_of.hpp>
+#include <boost/bimap/unordered_set_of.hpp>
+#include <boost/iterator/transform_iterator.hpp>
+
+#include <functional>
+
+namespace xrpl::PeerFinder {
+
+/**
+ * Stores IP addresses useful for gaining initial connections.
+ *
+ * This is one of the caches that is consulted when additional outgoing
+ * connections are needed. Along with the address, each entry has this
+ * additional metadata:
+ *
+ * Valence
+ *     A signed integer which represents the number of successful
+ *     consecutive connection attempts when positive, and the number of
+ *     failed consecutive connection attempts when negative.
+ *
+ * When choosing addresses from the boot cache for the purpose of
+ * establishing outgoing connections, addresses are ranked in decreasing
+ * order of high uptime, with valence as the tie breaker.
+ */
+class Bootcache
+{
+private:
+    class Entry
+    {
+    public:
+        Entry(int valence) : valence_(valence)
+        {
+        }
+
+        int&
+        valence()
+        {
+            return valence_;
+        }
+
+        [[nodiscard]] int
+        valence() const
+        {
+            return valence_;
+        }
+
+        friend bool
+        operator<(Entry const& lhs, Entry const& rhs)
+        {
+            return lhs.valence() > rhs.valence();
+        }
+
+    private:
+        int valence_;
+    };
+
+    using left_t = boost::bimaps::
+        unordered_set_of<beast::IP::Endpoint, boost::hash<beast::IP::Endpoint>, std::equal_to<>>;
+    using right_t = boost::bimaps::multiset_of<Entry, std::less<>>;
+    using map_type = boost::bimap<left_t, right_t>;
+    using value_type = map_type::value_type;
+
+    struct Transform
+    {
+        using first_argument_type = map_type::right_map::const_iterator::value_type const&;
+        using result_type = beast::IP::Endpoint const&;
+
+        explicit Transform() = default;
+
+        beast::IP::Endpoint const&
+        operator()(map_type::right_map::const_iterator::value_type const& v) const
+        {
+            return v.get_left();
+        }
+    };
+
+private:
+    map_type map_;
+
+    Store& store_;
+    clock_type& clock_;
+    beast::Journal journal_;
+
+    // Time after which we can update the database again
+    clock_type::time_point whenUpdate_;
+
+    // Set to true when a database update is needed
+    bool needsUpdate_{false};
+
+public:
+    static constexpr int kStaticValence = 32;
+
+    using iterator = boost::transform_iterator<Transform, map_type::right_map::const_iterator>;
+
+    using const_iterator = iterator;
+
+    Bootcache(Store& store, clock_type& clock, beast::Journal journal);
+
+    ~Bootcache();
+
+    /**
+     * Returns `true` if the cache is empty.
+     */
+    [[nodiscard]] bool
+    empty() const;
+
+    /**
+     * Returns the number of entries in the cache.
+     */
+    [[nodiscard]] map_type::size_type
+    size() const;
+
+    /**
+     * IP::Endpoint iterators that traverse in decreasing valence.
+     */
+    /** @{ */
+    [[nodiscard]] const_iterator
+    begin() const;
+    [[nodiscard]] const_iterator
+    cbegin() const;
+    [[nodiscard]] const_iterator
+    end() const;
+    [[nodiscard]] const_iterator
+    cend() const;
+    void
+    clear();
+    /** @} */
+
+    /**
+     * Load the persisted data from the Store into the container.
+     */
+    void
+    load();
+
+    /**
+     * Add a newly-learned address to the cache.
+     */
+    bool
+    insert(beast::IP::Endpoint const& endpoint);
+
+    /**
+     * Add a staticallyconfigured address to the cache.
+     */
+    bool
+    insertStatic(beast::IP::Endpoint const& endpoint);
+
+    /**
+     * Called when an outbound connection handshake completes.
+     */
+    void
+    onSuccess(beast::IP::Endpoint const& endpoint);
+
+    /**
+     * Called when an outbound connection attempt fails to handshake.
+     */
+    void
+    onFailure(beast::IP::Endpoint const& endpoint);
+
+    /**
+     * Stores the cache in the persistent database on a timer.
+     */
+    void
+    periodicActivity();
+
+    /**
+     * Write the cache state to the property stream.
+     */
+    void
+    onWrite(beast::PropertyStream::Map& map);
+
+private:
+    void
+    prune();
+    void
+    update();
+    void
+    checkUpdate();
+    void
+    flagForUpdate();
+};
+
+}  // namespace xrpl::PeerFinder

include/xrpl/peerfinder/detail/Checker.h

@@ -0,0 +1,205 @@
+#pragma once
+
+#include <xrpl/beast/net/IPAddressConversion.h>
+#include <xrpl/beast/net/IPEndpoint.h>
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/intrusive/list.hpp>
+
+#include <condition_variable>
+#include <memory>
+#include <mutex>
+
+namespace xrpl::PeerFinder {
+
+/**
+ * Tests remote listening sockets to make sure they are connectable.
+ */
+template <class Protocol = boost::asio::ip::tcp>
+class Checker
+{
+private:
+    using error_code = boost::system::error_code;
+
+    struct BasicAsyncOp : boost::intrusive::list_base_hook<
+                              boost::intrusive::link_mode<boost::intrusive::normal_link>>
+    {
+        virtual ~BasicAsyncOp() = default;
+
+        virtual void
+        stop() = 0;
+
+        virtual void
+        operator()(error_code const& ec) = 0;
+    };
+
+    template <class Handler>
+    struct AsyncOp : BasicAsyncOp
+    {
+        using socket_type = Protocol::socket;
+        using endpoint_type = Protocol::endpoint;
+
+        Checker& checker;
+        socket_type socket;
+        Handler handler;
+
+        AsyncOp(Checker& owner, boost::asio::io_context& ioContext, Handler&& handler);
+
+        ~AsyncOp() override
+        {
+            checker.remove(*this);
+        }
+
+        void
+        stop() override;
+
+        void
+        operator()(error_code const& ec) override;  // NOLINT(readability-identifier-naming)
+    };
+
+    //--------------------------------------------------------------------------
+
+    using list_type =
+        boost::intrusive::make_list<BasicAsyncOp, boost::intrusive::constant_time_size<true>>::type;
+
+    std::mutex mutex_;
+    std::condition_variable cond_;
+    boost::asio::io_context& ioContext_;
+    list_type list_;
+    bool stop_ = false;
+
+public:
+    explicit Checker(boost::asio::io_context& ioContext);
+
+    /**
+     * Destroy the service.
+     * Any pending I/O operations will be canceled. This call blocks until
+     * all pending operations complete (either with success or with
+     * operation_aborted) and the associated thread and io_context have
+     * no more work remaining.
+     */
+    ~Checker();
+
+    /**
+     * Stop the service.
+     * Pending I/O operations will be canceled.
+     * This issues cancel orders for all pending I/O operations and then
+     * returns immediately. Handlers will receive operation_aborted errors,
+     * or if they were already queued they will complete normally.
+     */
+    void
+    stop();
+
+    /**
+     * Block until all pending I/O completes.
+     */
+    void
+    wait();
+
+    /**
+     * Performs an async connection test on the specified endpoint.
+     * The port must be non-zero. Note that the execution guarantees
+     * offered by asio handlers are NOT enforced.
+     */
+    template <class Handler>
+    void
+    asyncConnect(beast::IP::Endpoint const& endpoint, Handler&& handler);
+
+private:
+    void
+    remove(BasicAsyncOp& op);
+};
+
+//------------------------------------------------------------------------------
+
+template <class Protocol>
+template <class Handler>
+Checker<Protocol>::AsyncOp<Handler>::AsyncOp(
+    Checker& owner,
+    boost::asio::io_context& ioContext,
+    Handler&&
+        handler)  // NOLINT(cppcoreguidelines-rvalue-reference-param-not-moved) -- forwarded in init
+    : checker(owner), socket(ioContext), handler(std::forward<Handler>(handler))
+{
+}
+
+template <class Protocol>
+template <class Handler>
+void
+Checker<Protocol>::AsyncOp<Handler>::stop()
+{
+    error_code ec;
+    socket.cancel(ec);
+}
+
+template <class Protocol>
+template <class Handler>
+void
+Checker<Protocol>::AsyncOp<Handler>::operator()(error_code const& ec)
+{
+    handler(ec);
+}
+
+//------------------------------------------------------------------------------
+
+template <class Protocol>
+Checker<Protocol>::Checker(boost::asio::io_context& ioContext) : ioContext_(ioContext)
+{
+}
+
+template <class Protocol>
+Checker<Protocol>::~Checker()
+{
+    wait();
+}
+
+template <class Protocol>
+void
+Checker<Protocol>::stop()
+{
+    std::scoped_lock const lock(mutex_);
+    if (!stop_)
+    {
+        stop_ = true;
+        for (auto& c : list_)
+            c.stop();
+    }
+}
+
+template <class Protocol>
+void
+Checker<Protocol>::wait()
+{
+    std::unique_lock<std::mutex> lock(mutex_);
+    while (!list_.empty())
+        cond_.wait(lock);
+}
+
+template <class Protocol>
+template <class Handler>
+void
+Checker<Protocol>::asyncConnect(beast::IP::Endpoint const& endpoint, Handler&& handler)
+{
+    auto const op =
+        std::make_shared<AsyncOp<Handler>>(*this, ioContext_, std::forward<Handler>(handler));
+    {
+        std::scoped_lock const lock(mutex_);
+        list_.push_back(*op);
+    }
+    op->socket.async_connect(
+        beast::IPAddressConversion::toAsioEndpoint(endpoint),
+        [op](error_code const& ec) { (*op)(ec); });
+}
+
+template <class Protocol>
+void
+Checker<Protocol>::remove(BasicAsyncOp& op)
+{
+    std::scoped_lock const lock(mutex_);
+    list_.erase(list_.iterator_to(op));
+    if (list_.size() == 0)
+        cond_.notify_all();
+}
+
+}  // namespace xrpl::PeerFinder

include/xrpl/peerfinder/detail/Counts.h

@@ -0,0 +1,394 @@
+#pragma once
+
+#include <xrpl/beast/utility/PropertyStream.h>
+#include <xrpl/beast/utility/instrumentation.h>
+#include <xrpl/peerfinder/Config.h>
+#include <xrpl/peerfinder/Slot.h>
+#include <xrpl/peerfinder/detail/Tuning.h>
+
+#include <cstddef>
+#include <sstream>
+#include <string>
+
+namespace xrpl::PeerFinder {
+
+/**
+ * Direction of a slot count adjustment.
+ */
+enum class CountAdjustment : int { Decrement = -1, Increment = 1 };
+
+/**
+ * Manages the count of available connections for the various slots.
+ */
+class Counts
+{
+public:
+    /**
+     * Adds the slot state and properties to the slot counts.
+     */
+    void
+    add(Slot const& s)
+    {
+        adjust(s, CountAdjustment::Increment);
+    }
+
+    /**
+     * Removes the slot state and properties from the slot counts.
+     */
+    void
+    remove(Slot const& s)
+    {
+        adjust(s, CountAdjustment::Decrement);
+    }
+
+    /**
+     * Returns `true` if the slot can become active.
+     */
+    [[nodiscard]] bool
+    canActivate(Slot const& s) const
+    {
+        // Must be handshaked and in the right state
+        XRPL_ASSERT(
+            s.state() == Slot::State::Connected || s.state() == Slot::State::Accept,
+            "xrpl::PeerFinder::Counts::can_activate : valid input state");
+
+        if (s.fixed() || s.reserved())
+            return true;
+
+        if (s.inbound())
+            return inActive_ < inMax_;
+
+        return outActive_ < outMax_;
+    }
+
+    /**
+     * Returns the number of attempts needed to bring us to the max.
+     */
+    [[nodiscard]] std::size_t
+    attemptsNeeded() const
+    {
+        if (attempts_ >= Tuning::kMaxConnectAttempts)
+            return 0;
+        return Tuning::kMaxConnectAttempts - attempts_;
+    }
+
+    /**
+     * Returns the number of outbound connection attempts.
+     */
+    [[nodiscard]] std::size_t
+    attempts() const
+    {
+        return attempts_;
+    }
+
+    /**
+     * Returns the total number of outbound slots.
+     */
+    [[nodiscard]] int
+    outMax() const
+    {
+        return outMax_;
+    }
+
+    /**
+     * Returns the number of outbound peers assigned an open slot.
+     * Fixed peers do not count towards outbound slots used.
+     */
+    [[nodiscard]] int
+    outActive() const
+    {
+        return outActive_;
+    }
+
+    /**
+     * Returns the number of fixed connections.
+     */
+    [[nodiscard]] std::size_t
+    fixed() const
+    {
+        return fixed_;
+    }
+
+    /**
+     * Returns the number of active fixed connections.
+     */
+    [[nodiscard]] std::size_t
+    fixedActive() const
+    {
+        return fixedActive_;
+    }
+
+    //--------------------------------------------------------------------------
+
+    /**
+     * Called when the config is set or changed.
+     */
+    void
+    onConfig(Config const& config)
+    {
+        outMax_ = config.outPeers;
+        if (config.wantIncoming)
+            inMax_ = config.inPeers;
+    }
+
+    /**
+     * Returns the number of accepted connections that haven't handshaked.
+     */
+    [[nodiscard]] int
+    acceptCount() const
+    {
+        return acceptCount_;
+    }
+
+    /**
+     * Returns the number of connection attempts currently active.
+     */
+    [[nodiscard]] int
+    connectCount() const
+    {
+        return attempts_;
+    }
+
+    /**
+     * Returns the number of connections that are gracefully closing.
+     */
+    [[nodiscard]] int
+    closingCount() const
+    {
+        return closingCount_;
+    }
+
+    /**
+     * Returns the total number of inbound slots.
+     */
+    [[nodiscard]] int
+    inMax() const
+    {
+        return inMax_;
+    }
+
+    /**
+     * Returns the number of inbound peers assigned an open slot.
+     */
+    [[nodiscard]] int
+    inboundActive() const
+    {
+        return inActive_;
+    }
+
+    /**
+     * Returns the total number of active peers excluding fixed peers.
+     */
+    [[nodiscard]] int
+    totalActive() const
+    {
+        return inActive_ + outActive_;
+    }
+
+    /**
+     * Returns the number of unused inbound slots.
+     * Fixed peers do not deduct from inbound slots or count towards totals.
+     */
+    [[nodiscard]] int
+    inboundSlotsFree() const
+    {
+        if (inActive_ < inMax_)
+            return inMax_ - inActive_;
+        return 0;
+    }
+
+    /**
+     * Returns the number of unused outbound slots.
+     * Fixed peers do not deduct from outbound slots or count towards totals.
+     */
+    [[nodiscard]] int
+    outboundSlotsFree() const
+    {
+        if (outActive_ < outMax_)
+            return outMax_ - outActive_;
+        return 0;
+    }
+
+    //--------------------------------------------------------------------------
+
+    /**
+     * Returns true if the slot logic considers us "connected" to the network.
+     */
+    [[nodiscard]] bool
+    isConnectedToNetwork() const
+    {
+        // We will consider ourselves connected if we have reached
+        // the number of outgoing connections desired, or if connect
+        // automatically is false.
+        //
+        // Fixed peers do not count towards the active outgoing total.
+
+        return outMax_ <= 0;
+    }
+
+    /**
+     * Output statistics.
+     */
+    void
+    onWrite(beast::PropertyStream::Map& map) const
+    {
+        map["accept"] = acceptCount();
+        map["connect"] = connectCount();
+        map["close"] = closingCount();
+        map["in"] << inActive_ << "/" << inMax_;
+        map["out"] << outActive_ << "/" << outMax_;
+        map["fixed"] = fixedActive_;
+        map["reserved"] = reserved_;
+        map["total"] = active_;
+    }
+
+    /**
+     * Records the state for diagnostics.
+     */
+    [[nodiscard]] std::string
+    stateString() const
+    {
+        std::stringstream ss;
+        ss << outActive_ << "/" << outMax_ << " out, " << inActive_ << "/" << inMax_ << " in, "
+           << connectCount() << " connecting, " << closingCount() << " closing";
+        return ss.str();
+    }
+
+    //--------------------------------------------------------------------------
+private:
+    /**
+     * Increments or decrements a counter based on the adjustment direction.
+     */
+    template <typename T>
+    static void
+    adjustCounter(T& counter, CountAdjustment dir)
+    {
+        switch (dir)
+        {
+            case CountAdjustment::Increment:
+                ++counter;
+                break;
+            case CountAdjustment::Decrement:
+                --counter;
+                break;
+        }
+    }
+
+    // Adjusts counts based on the specified slot, in the direction indicated.
+    //
+    // IMPORTANT: All std::size_t counters MUST be adjusted via adjustCounter()
+    // and NEVER via `+= n` where n = static_cast<int>(dir).  When dir is
+    // Decrement, n == -1; adding -1 to a std::size_t implicitly converts -1 to
+    // SIZE_MAX, which UBSan flags as unsigned-integer-overflow and masks real
+    // underflow bugs (decrementing a counter already at zero).  Plain int
+    // counters (acceptCount_, attempts_, closingCount_) are safe with += n.
+    void
+    adjust(Slot const& s, CountAdjustment const dir)
+    {
+        int const n = static_cast<int>(dir);
+        if (s.fixed())
+            adjustCounter(fixed_, dir);
+
+        if (s.reserved())
+            adjustCounter(reserved_, dir);
+
+        switch (s.state())
+        {
+            case Slot::State::Accept:
+                XRPL_ASSERT(s.inbound(), "xrpl::PeerFinder::Counts::adjust : input is inbound");
+                acceptCount_ += n;
+                break;
+
+            case Slot::State::Connect:
+            case Slot::State::Connected:
+                XRPL_ASSERT(
+                    !s.inbound(),
+                    "xrpl::PeerFinder::Counts::adjust : input is not "
+                    "inbound");
+                attempts_ += n;
+                break;
+
+            case Slot::State::Active:
+                if (s.fixed())
+                    adjustCounter(fixedActive_, dir);
+                if (!s.fixed() && !s.reserved())
+                {
+                    if (s.inbound())
+                    {
+                        adjustCounter(inActive_, dir);
+                    }
+                    else
+                    {
+                        adjustCounter(outActive_, dir);
+                    }
+                }
+                adjustCounter(active_, dir);
+                break;
+
+            case Slot::State::Closing:
+                closingCount_ += n;
+                break;
+
+            // LCOV_EXCL_START
+            default:
+                UNREACHABLE("xrpl::PeerFinder::Counts::adjust : invalid input state");
+                break;
+                // LCOV_EXCL_STOP
+        };
+    }
+
+private:
+    /**
+     * Outbound connection attempts.
+     */
+    int attempts_{0};
+
+    /**
+     * Active connections, including fixed and reserved.
+     */
+    std::size_t active_{0};
+
+    /**
+     * Total number of inbound slots.
+     */
+    std::size_t inMax_{0};
+
+    /**
+     * Number of inbound slots assigned to active peers.
+     */
+    std::size_t inActive_{0};
+
+    /**
+     * Maximum desired outbound slots.
+     */
+    std::size_t outMax_{0};
+
+    /**
+     * Active outbound slots.
+     */
+    std::size_t outActive_{0};
+
+    /**
+     * Fixed connections.
+     */
+    std::size_t fixed_{0};
+
+    /**
+     * Active fixed connections.
+     */
+    std::size_t fixedActive_{0};
+
+    /**
+     * Reserved connections.
+     */
+    std::size_t reserved_{0};
+
+    // Number of inbound connections that are
+    // not active or gracefully closing.
+    int acceptCount_{0};
+
+    // Number of connections that are gracefully closing.
+    int closingCount_{0};
+};
+
+}  // namespace xrpl::PeerFinder

include/xrpl/peerfinder/detail/Fixed.h

@@ -0,0 +1,58 @@
+#pragma once
+
+#include <xrpl/peerfinder/Types.h>
+#include <xrpl/peerfinder/detail/Tuning.h>
+
+#include <algorithm>
+#include <chrono>
+#include <cstddef>
+
+namespace xrpl::PeerFinder {
+
+/**
+ * Metadata for a Fixed slot.
+ */
+class Fixed
+{
+public:
+    explicit Fixed(clock_type& clock) : when_(clock.now())
+    {
+    }
+
+    Fixed(Fixed const&) = default;
+
+    /**
+     * Returns the time after which we should allow a connection attempt.
+     */
+    [[nodiscard]] clock_type::time_point const&
+    when() const
+    {
+        return when_;
+    }
+
+    /**
+     * Updates metadata to reflect a failed connection.
+     */
+    void
+    failure(clock_type::time_point const& now)
+    {
+        failures_ = std::min(failures_ + 1, Tuning::kConnectionBackoff.size() - 1);
+        when_ = now + std::chrono::minutes(Tuning::kConnectionBackoff[failures_]);
+    }
+
+    /**
+     * Updates metadata to reflect a successful connection.
+     */
+    void
+    success(clock_type::time_point const& now)
+    {
+        failures_ = 0;
+        when_ = now;
+    }
+
+private:
+    clock_type::time_point when_;
+    std::size_t failures_{0};
+};
+
+}  // namespace xrpl::PeerFinder

include/xrpl/peerfinder/detail/Handouts.h

@@ -0,0 +1,344 @@
+#pragma once
+
+#include <xrpl/beast/container/aged_set.h>
+#include <xrpl/beast/net/IPAddress.h>
+#include <xrpl/beast/utility/instrumentation.h>
+#include <xrpl/peerfinder/Types.h>
+#include <xrpl/peerfinder/detail/SlotImp.h>
+#include <xrpl/peerfinder/detail/Tuning.h>
+
+#include <algorithm>
+#include <cstddef>
+#include <utility>
+#include <vector>
+
+namespace xrpl::PeerFinder {
+
+namespace detail {
+
+/**
+ * Try to insert one object in the target.
+ * When an item is handed out it is moved to the end of the container.
+ * @return The number of objects inserted
+ */
+// VFALCO TODO specialization that handles std::list for SequenceContainer
+//             using splice for optimization over erase/push_back
+//
+template <class Target, class HopContainer>
+std::size_t
+handoutOne(Target& t, HopContainer& h)
+{
+    XRPL_ASSERT(!t.full(), "xrpl::PeerFinder::detail::handoutOne : target is not full");
+    for (auto it = h.begin(); it != h.end(); ++it)
+    {
+        auto const& e = *it;
+        if (t.tryInsert(e))
+        {
+            h.moveBack(it);
+            return 1;
+        }
+    }
+    return 0;
+}
+
+}  // namespace detail
+
+/**
+ * Distributes objects to targets according to business rules.
+ * A best effort is made to evenly distribute items in the sequence
+ * container list into the target sequence list.
+ */
+template <class TargetFwdIter, class SeqFwdIter>
+void
+handout(TargetFwdIter first, TargetFwdIter last, SeqFwdIter seqFirst, SeqFwdIter seqLast)
+{
+    for (;;)
+    {
+        std::size_t n(0);
+        for (auto si = seqFirst; si != seqLast; ++si)
+        {
+            auto c = *si;
+            bool allFull(true);
+            for (auto ti = first; ti != last; ++ti)
+            {
+                auto& t = *ti;
+                if (!t.full())
+                {
+                    n += detail::handoutOne(t, c);
+                    allFull = false;
+                }
+            }
+            if (allFull)
+                return;
+        }
+        if (!n)
+            break;
+    }
+}
+
+//------------------------------------------------------------------------------
+
+/**
+ * Receives handouts for redirecting a connection.
+ * An incoming connection request is redirected when we are full on slots.
+ */
+class RedirectHandouts
+{
+public:
+    template <class = void>
+    explicit RedirectHandouts(SlotImp::ptr slot);
+
+    template <class = void>
+    bool
+    tryInsert(Endpoint const& ep);
+
+    [[nodiscard]] bool
+    full() const
+    {
+        return list_.size() >= Tuning::kRedirectEndpointCount;
+    }
+
+    [[nodiscard]] SlotImp::ptr const&
+    slot() const
+    {
+        return slot_;
+    }
+
+    std::vector<Endpoint>&
+    list()
+    {
+        return list_;
+    }
+
+    [[nodiscard]] std::vector<Endpoint> const&
+    list() const
+    {
+        return list_;
+    }
+
+private:
+    SlotImp::ptr slot_;
+    std::vector<Endpoint> list_;
+};
+
+template <class>
+RedirectHandouts::RedirectHandouts(SlotImp::ptr slot) : slot_(std::move(slot))
+{
+    list_.reserve(Tuning::kRedirectEndpointCount);
+}
+
+template <class>
+bool
+RedirectHandouts::tryInsert(Endpoint const& ep)
+{
+    if (full())
+        return false;
+
+    // VFALCO NOTE This check can be removed when we provide the
+    //             addresses in a peer HTTP handshake instead of
+    //             the tmENDPOINTS message.
+    //
+    if (ep.hops > Tuning::kMaxHops)
+        return false;
+
+    // Don't send them our address
+    if (ep.hops == 0)
+        return false;
+
+    // Don't send them their own address
+    if (slot_->remoteEndpoint().address() == ep.address.address())
+        return false;
+
+    // Make sure the address isn't already in our list
+    if (std::ranges::any_of(list_, [&ep](Endpoint const& other) {
+            // Ignore port for security reasons
+            return other.address.address() == ep.address.address();
+        }))
+    {
+        return false;
+    }
+
+    list_.emplace_back(ep.address, ep.hops);
+
+    return true;
+}
+
+//------------------------------------------------------------------------------
+
+/**
+ * Receives endpoints for a slot during periodic handouts.
+ */
+class SlotHandouts
+{
+public:
+    template <class = void>
+    explicit SlotHandouts(SlotImp::ptr slot);
+
+    template <class = void>
+    bool
+    tryInsert(Endpoint const& ep);
+
+    [[nodiscard]] bool
+    full() const
+    {
+        return list_.size() >= Tuning::kNumberOfEndpoints;
+    }
+
+    void
+    insert(Endpoint const& ep)
+    {
+        list_.push_back(ep);
+    }
+
+    [[nodiscard]] SlotImp::ptr const&
+    slot() const
+    {
+        return slot_;
+    }
+
+    [[nodiscard]] std::vector<Endpoint> const&
+    list() const
+    {
+        return list_;
+    }
+
+private:
+    SlotImp::ptr slot_;
+    std::vector<Endpoint> list_;
+};
+
+template <class>
+SlotHandouts::SlotHandouts(SlotImp::ptr slot) : slot_(std::move(slot))
+{
+    list_.reserve(Tuning::kNumberOfEndpoints);
+}
+
+template <class>
+bool
+SlotHandouts::tryInsert(Endpoint const& ep)
+{
+    if (full())
+        return false;
+
+    if (ep.hops > Tuning::kMaxHops)
+        return false;
+
+    if (slot_->recent.filter(ep.address, ep.hops))
+        return false;
+
+    // Don't send them their own address
+    if (slot_->remoteEndpoint().address() == ep.address.address())
+        return false;
+
+    // Make sure the address isn't already in our list
+    if (std::ranges::any_of(list_, [&ep](Endpoint const& other) {
+            // Ignore port for security reasons
+            return other.address.address() == ep.address.address();
+        }))
+        return false;
+
+    list_.emplace_back(ep.address, ep.hops);
+
+    // Insert into this slot's recent table. Although the endpoint
+    // didn't come from the slot, adding it to the slot's table
+    // prevents us from sending it again until it has expired from
+    // the other end's cache.
+    //
+    slot_->recent.insert(ep.address, ep.hops);
+
+    return true;
+}
+
+//------------------------------------------------------------------------------
+
+/**
+ * Receives handouts for making automatic connections.
+ */
+class ConnectHandouts
+{
+public:
+    // Keeps track of addresses we have made outgoing connections
+    // to, for the purposes of not connecting to them too frequently.
+    using Squelches = beast::aged_set<beast::IP::Address>;
+
+    using list_type = std::vector<beast::IP::Endpoint>;
+
+private:
+    std::size_t needed_;
+    Squelches& squelches_;
+    list_type list_;
+
+public:
+    template <class = void>
+    ConnectHandouts(std::size_t needed, Squelches& squelches);
+
+    template <class = void>
+    bool
+    tryInsert(beast::IP::Endpoint const& endpoint);
+
+    [[nodiscard]] bool
+    empty() const
+    {
+        return list_.empty();
+    }
+
+    [[nodiscard]] bool
+    full() const
+    {
+        return list_.size() >= needed_;
+    }
+
+    bool
+    tryInsert(Endpoint const& endpoint)
+    {
+        return tryInsert(endpoint.address);
+    }
+
+    list_type&
+    list()
+    {
+        return list_;
+    }
+
+    [[nodiscard]] list_type const&
+    list() const
+    {
+        return list_;
+    }
+};
+
+template <class>
+ConnectHandouts::ConnectHandouts(std::size_t needed, Squelches& squelches)
+    : needed_(needed), squelches_(squelches)
+{
+    list_.reserve(needed);
+}
+
+template <class>
+bool
+ConnectHandouts::tryInsert(beast::IP::Endpoint const& endpoint)
+{
+    if (full())
+        return false;
+
+    // Make sure the address isn't already in our list
+    if (std::ranges::any_of(list_, [&endpoint](beast::IP::Endpoint const& other) {
+            // Ignore port for security reasons
+            return other.address() == endpoint.address();
+        }))
+    {
+        return false;
+    }
+
+    // Add to squelch list so we don't try it too often.
+    // If its already there, then make try_insert fail.
+    auto const result(squelches_.insert(endpoint.address()));
+    if (!result.second)
+        return false;
+
+    list_.push_back(endpoint);
+
+    return true;
+}
+
+}  // namespace xrpl::PeerFinder

include/xrpl/peerfinder/detail/Livecache.h

@@ -0,0 +1,564 @@
+#pragma once
+
+#include <xrpl/basics/Log.h>
+#include <xrpl/basics/random.h>
+#include <xrpl/beast/container/aged_map.h>
+#include <xrpl/beast/net/IPEndpoint.h>
+#include <xrpl/beast/utility/Journal.h>
+#include <xrpl/beast/utility/PropertyStream.h>
+#include <xrpl/beast/utility/instrumentation.h>
+#include <xrpl/beast/utility/maybe_const.h>
+#include <xrpl/peerfinder/Types.h>
+#include <xrpl/peerfinder/detail/Tuning.h>
+
+#include <boost/intrusive/list.hpp>
+#include <boost/iterator/transform_iterator.hpp>
+
+#include <algorithm>
+#include <array>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <functional>
+#include <iomanip>
+#include <ios>
+#include <iterator>
+#include <memory>
+#include <sstream>
+#include <string>
+#include <utility>
+#include <vector>
+
+namespace xrpl::PeerFinder {
+
+template <class>
+class Livecache;
+
+namespace detail {
+
+class LivecacheBase
+{
+public:
+    explicit LivecacheBase() = default;
+
+protected:
+    struct Element : boost::intrusive::list_base_hook<>
+    {
+        Element(Endpoint endpoint) : endpoint(std::move(endpoint))
+        {
+        }
+
+        Endpoint endpoint;
+    };
+
+    using list_type =
+        boost::intrusive::make_list<Element, boost::intrusive::constant_time_size<false>>::type;
+
+public:
+    /**
+     * A list of Endpoint at the same hops
+     * This is a lightweight wrapper around a reference to the underlying
+     * container.
+     */
+    template <bool IsConst>
+    class Hop
+    {
+    public:
+        // Iterator transformation to extract the endpoint from Element
+        struct Transform
+        {
+            using first_argument = Element;
+            using result_type = Endpoint;
+
+            explicit Transform() = default;
+
+            Endpoint const&
+            operator()(Element const& e) const
+            {
+                return e.endpoint;
+            }
+        };
+
+    public:
+        using iterator = boost::transform_iterator<Transform, list_type::const_iterator>;
+
+        using const_iterator = iterator;
+
+        using reverse_iterator =
+            boost::transform_iterator<Transform, list_type::const_reverse_iterator>;
+
+        using const_reverse_iterator = reverse_iterator;
+
+        [[nodiscard]] iterator
+        begin() const
+        {
+            return iterator(list_.get().cbegin(), Transform());
+        }
+
+        [[nodiscard]] iterator
+        cbegin() const
+        {
+            return iterator(list_.get().cbegin(), Transform());
+        }
+
+        [[nodiscard]] iterator
+        end() const
+        {
+            return iterator(list_.get().cend(), Transform());
+        }
+
+        [[nodiscard]] iterator
+        cend() const
+        {
+            return iterator(list_.get().cend(), Transform());
+        }
+
+        [[nodiscard]] reverse_iterator
+        rbegin() const
+        {
+            return reverse_iterator(list_.get().crbegin(), Transform());
+        }
+
+        [[nodiscard]] reverse_iterator
+        crbegin() const
+        {
+            return reverse_iterator(list_.get().crbegin(), Transform());
+        }
+
+        [[nodiscard]] reverse_iterator
+        rend() const
+        {
+            return reverse_iterator(list_.get().crend(), Transform());
+        }
+
+        [[nodiscard]] reverse_iterator
+        crend() const
+        {
+            return reverse_iterator(list_.get().crend(), Transform());
+        }
+
+        // move the element to the end of the container
+        void
+        moveBack(const_iterator pos)
+        {
+            auto& e(const_cast<Element&>(*pos.base()));
+            list_.get().erase(list_.get().iterator_to(e));
+            list_.get().push_back(e);
+        }
+
+    private:
+        explicit Hop(beast::MaybeConst<IsConst, list_type>::type& list) : list_(list)
+        {
+        }
+
+        friend class LivecacheBase;
+
+        std::reference_wrapper<typename beast::MaybeConst<IsConst, list_type>::type> list_;
+    };
+
+protected:
+    // Work-around to call Hop's private constructor from Livecache
+    template <bool IsConst>
+    static Hop<IsConst>
+    makeHop(beast::MaybeConst<IsConst, list_type>::type& list)
+    {
+        return Hop<IsConst>(list);
+    }
+};
+
+}  // namespace detail
+
+//------------------------------------------------------------------------------
+
+/**
+ * The Livecache holds the short-lived relayed Endpoint messages.
+ *
+ * Since peers only advertise themselves when they have open slots,
+ * we want these messages to expire rather quickly after the peer becomes
+ * full.
+ *
+ * Addresses added to the cache are not connection-tested to see if
+ * they are connectable (with one small exception regarding neighbors).
+ * Therefore, these addresses are not suitable for persisting across
+ * launches or for bootstrapping, because they do not have verifiable
+ * and locally observed uptime and connectability information.
+ */
+template <class Allocator = std::allocator<char>>
+class Livecache : protected detail::LivecacheBase
+{
+private:
+    using cache_type = beast::aged_map<
+        beast::IP::Endpoint,
+        Element,
+        std::chrono::steady_clock,
+        std::less<beast::IP::Endpoint>,
+        Allocator>;
+
+    beast::Journal journal_;
+    cache_type cache_;
+
+public:
+    using allocator_type = Allocator;
+
+    /**
+     * Create the cache.
+     */
+    Livecache(clock_type& clock, beast::Journal journal, Allocator alloc = Allocator());
+
+    //
+    // Iteration by hops
+    //
+    // The range [begin, end) provides a sequence of list_type
+    // where each list contains endpoints at a given hops.
+    //
+
+    class HopsT
+    {
+    private:
+        // An endpoint at hops=0 represents the local node.
+        // Endpoints coming in at maxHops are stored at maxHops +1,
+        // but not given out (since they would exceed maxHops). They
+        // are used for automatic connection attempts.
+        //
+        using Histogram = std::array<int, 1 + Tuning::kMaxHops + 1>;
+        using lists_type = std::array<list_type, 1 + Tuning::kMaxHops + 1>;
+
+        template <bool IsConst>
+        struct Transform
+        {
+            using first_argument = lists_type::value_type;
+            using result_type = Hop<IsConst>;
+
+            explicit Transform() = default;
+
+            Hop<IsConst>
+            operator()(beast::MaybeConst<IsConst, lists_type::value_type>::type& list) const
+            {
+                return makeHop<IsConst>(list);
+            }
+        };
+
+    public:
+        using iterator = boost::transform_iterator<Transform<false>, lists_type::iterator>;
+
+        using const_iterator =
+            boost::transform_iterator<Transform<true>, lists_type::const_iterator>;
+
+        using reverse_iterator =
+            boost::transform_iterator<Transform<false>, lists_type::reverse_iterator>;
+
+        using const_reverse_iterator =
+            boost::transform_iterator<Transform<true>, lists_type::const_reverse_iterator>;
+
+        iterator
+        begin()
+        {
+            return iterator(lists_.begin(), Transform<false>());
+        }
+
+        [[nodiscard]] const_iterator
+        begin() const
+        {
+            return const_iterator(lists_.cbegin(), Transform<true>());
+        }
+
+        [[nodiscard]] const_iterator
+        cbegin() const
+        {
+            return const_iterator(lists_.cbegin(), Transform<true>());
+        }
+
+        iterator
+        end()
+        {
+            return iterator(lists_.end(), Transform<false>());
+        }
+
+        [[nodiscard]] const_iterator
+        end() const
+        {
+            return const_iterator(lists_.cend(), Transform<true>());
+        }
+
+        [[nodiscard]] const_iterator
+        cend() const
+        {
+            return const_iterator(lists_.cend(), Transform<true>());
+        }
+
+        reverse_iterator
+        rbegin()
+        {
+            return reverse_iterator(lists_.rbegin(), Transform<false>());
+        }
+
+        [[nodiscard]] const_reverse_iterator
+        rbegin() const
+        {
+            return const_reverse_iterator(lists_.crbegin(), Transform<true>());
+        }
+
+        [[nodiscard]] const_reverse_iterator
+        crbegin() const
+        {
+            return const_reverse_iterator(lists_.crbegin(), Transform<true>());
+        }
+
+        reverse_iterator
+        rend()
+        {
+            return reverse_iterator(lists_.rend(), Transform<false>());
+        }
+
+        [[nodiscard]] const_reverse_iterator
+        rend() const
+        {
+            return const_reverse_iterator(lists_.crend(), Transform<true>());
+        }
+
+        [[nodiscard]] const_reverse_iterator
+        crend() const
+        {
+            return const_reverse_iterator(lists_.crend(), Transform<true>());
+        }
+
+        /**
+         * Shuffle each hop list.
+         */
+        void
+        shuffle();
+
+        [[nodiscard]] std::string
+        histogram() const;
+
+    private:
+        explicit HopsT(Allocator const& alloc);
+
+        void
+        insert(Element& e);
+
+        // Reinsert e at a new hops
+        void
+        reinsert(Element& e, std::uint32_t hops);
+
+        void
+        remove(Element& e);
+
+        friend class Livecache;
+        lists_type lists_;
+        Histogram hist_{};
+    } hops;
+
+    /**
+     * Returns `true` if the cache is empty.
+     */
+    [[nodiscard]] bool
+    empty() const
+    {
+        return cache_.empty();
+    }
+
+    /**
+     * Returns the number of entries in the cache.
+     */
+    cache_type::size_type
+    size() const
+    {
+        return cache_.size();
+    }
+
+    /**
+     * Erase entries whose time has expired.
+     */
+    void
+    expire();
+
+    /**
+     * Creates or updates an existing Element based on a new message.
+     */
+    void
+    insert(Endpoint const& ep);
+
+    /**
+     * Output statistics.
+     */
+    void
+    onWrite(beast::PropertyStream::Map& map);
+};
+
+//------------------------------------------------------------------------------
+
+template <class Allocator>
+Livecache<Allocator>::Livecache(clock_type& clock, beast::Journal journal, Allocator alloc)
+    : journal_(journal), cache_(clock, alloc), hops(alloc)
+{
+}
+
+template <class Allocator>
+void
+Livecache<Allocator>::expire()
+{
+    std::size_t n(0);
+    typename cache_type::time_point const expired(
+        cache_.clock().now() - Tuning::kLiveCacheSecondsToLive);
+    for (auto iter(cache_.chronological.begin());
+         iter != cache_.chronological.end() && iter.when() <= expired;)
+    {
+        Element& e(iter->second);
+        hops.remove(e);
+        iter = cache_.erase(iter);
+        ++n;
+    }
+    if (n > 0)
+    {
+        JLOG(journal_.debug()) << std::left << std::setw(18) << "Livecache expired " << n
+                               << ((n > 1) ? " entries" : " entry");
+    }
+}
+
+template <class Allocator>
+void
+Livecache<Allocator>::insert(Endpoint const& ep)
+{
+    // The caller already incremented hop, so if we got a
+    // message at maxHops we will store it at maxHops + 1.
+    // This means we won't give out the address to other peers
+    // but we will use it to make connections and hand it out
+    // when redirecting.
+    //
+    XRPL_ASSERT(
+        ep.hops <= (Tuning::kMaxHops + 1),
+        "xrpl::PeerFinder::Livecache::insert : maximum input hops");
+    auto result = cache_.emplace(ep.address, ep);
+    Element& e(result.first->second);
+    if (result.second)
+    {
+        hops.insert(e);
+        JLOG(journal_.debug()) << std::left << std::setw(18) << "Livecache insert " << ep.address
+                               << " at hops " << ep.hops;
+        return;
+    }
+    if (!result.second && (ep.hops > e.endpoint.hops))
+    {
+        // Drop duplicates at higher hops
+        std::size_t const excess(ep.hops - e.endpoint.hops);
+        JLOG(journal_.trace()) << std::left << std::setw(18) << "Livecache drop " << ep.address
+                               << " at hops +" << excess;
+        return;
+    }
+
+    cache_.touch(result.first);
+
+    // Address already in the cache so update metadata
+    if (ep.hops < e.endpoint.hops)
+    {
+        hops.reinsert(e, ep.hops);
+        JLOG(journal_.debug()) << std::left << std::setw(18) << "Livecache update " << ep.address
+                               << " at hops " << ep.hops;
+    }
+    else
+    {
+        JLOG(journal_.trace()) << std::left << std::setw(18) << "Livecache refresh " << ep.address
+                               << " at hops " << ep.hops;
+    }
+}
+
+template <class Allocator>
+void
+Livecache<Allocator>::onWrite(beast::PropertyStream::Map& map)
+{
+    typename cache_type::time_point const expired(
+        cache_.clock().now() - Tuning::kLiveCacheSecondsToLive);
+    map["size"] = size();
+    map["hist"] = hops.histogram();
+    beast::PropertyStream::Set set("entries", map);
+    for (auto iter(cache_.cbegin()); iter != cache_.cend(); ++iter)
+    {
+        auto const& e(iter->second);
+        beast::PropertyStream::Map item(set);
+        item["hops"] = e.endpoint.hops;
+        item["address"] = e.endpoint.address.toString();
+        std::stringstream ss;
+        ss << (iter.when() - expired).count();
+        item["expires"] = ss.str();
+    }
+}
+
+//------------------------------------------------------------------------------
+
+template <class Allocator>
+void
+Livecache<Allocator>::HopsT::shuffle()
+{
+    for (auto& list : lists_)
+    {
+        std::vector<std::reference_wrapper<Element>> v;
+        v.reserve(list.size());
+        std::ranges::copy(list, std::back_inserter(v));
+        std::shuffle(v.begin(), v.end(), defaultPrng());
+        list.clear();
+        for (auto& e : v)
+            list.push_back(e);
+    }
+}
+
+template <class Allocator>
+std::string
+Livecache<Allocator>::HopsT::histogram() const
+{
+    std::string s;
+    for (auto const& h : hist_)
+    {
+        if (!s.empty())
+            s += ", ";
+        s += std::to_string(h);
+    }
+    return s;
+}
+
+template <class Allocator>
+Livecache<Allocator>::HopsT::HopsT(Allocator const& alloc)
+{
+    std::ranges::fill(hist_, 0);
+}
+
+template <class Allocator>
+void
+Livecache<Allocator>::HopsT::insert(Element& e)
+{
+    XRPL_ASSERT(
+        e.endpoint.hops <= Tuning::kMaxHops + 1,
+        "xrpl::PeerFinder::Livecache::HopsT::insert : maximum input hops");
+    // This has security implications without a shuffle
+    lists_[e.endpoint.hops].push_front(e);
+    ++hist_[e.endpoint.hops];
+}
+
+template <class Allocator>
+void
+Livecache<Allocator>::HopsT::reinsert(Element& e, std::uint32_t numHops)
+{
+    XRPL_ASSERT(
+        numHops <= Tuning::kMaxHops + 1,
+        "xrpl::PeerFinder::Livecache::HopsT::reinsert : maximum hops input");
+
+    auto& list = lists_[e.endpoint.hops];
+    list.erase(list.iterator_to(e));
+
+    --hist_[e.endpoint.hops];
+
+    e.endpoint.hops = numHops;
+    insert(e);
+}
+
+template <class Allocator>
+void
+Livecache<Allocator>::HopsT::remove(Element& e)
+{
+    --hist_[e.endpoint.hops];
+
+    auto& list = lists_[e.endpoint.hops];
+    list.erase(list.iterator_to(e));
+}
+
+}  // namespace xrpl::PeerFinder

include/xrpl/peerfinder/detail/SlotImp.h

@@ -0,0 +1,199 @@
+#pragma once
+
+#include <xrpl/beast/container/aged_unordered_map.h>
+#include <xrpl/beast/net/IPEndpoint.h>
+#include <xrpl/peerfinder/Slot.h>
+#include <xrpl/peerfinder/Types.h>
+#include <xrpl/protocol/PublicKey.h>
+
+#include <atomic>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+
+namespace xrpl::PeerFinder {
+
+class SlotImp : public Slot
+{
+public:
+    using ptr = std::shared_ptr<SlotImp>;
+
+    // inbound
+    SlotImp(
+        beast::IP::Endpoint const& localEndpoint,
+        beast::IP::Endpoint remoteEndpoint,
+        bool fixed,
+        clock_type& clock);
+
+    // outbound
+    SlotImp(beast::IP::Endpoint remoteEndpoint, bool fixed, clock_type& clock);
+
+    bool
+    inbound() const override
+    {
+        return inbound_;
+    }
+
+    bool
+    fixed() const override
+    {
+        return fixed_;
+    }
+
+    bool
+    reserved() const override
+    {
+        return reserved_;
+    }
+
+    State
+    state() const override
+    {
+        return state_;
+    }
+
+    beast::IP::Endpoint const&
+    remoteEndpoint() const override
+    {
+        return remoteEndpoint_;
+    }
+
+    std::optional<beast::IP::Endpoint> const&
+    localEndpoint() const override
+    {
+        return localEndpoint_;
+    }
+
+    std::optional<PublicKey> const&
+    publicKey() const override
+    {
+        return publicKey_;
+    }
+
+    std::string
+    prefix() const
+    {
+        return "[" + getFingerprint(remoteEndpoint(), publicKey()) + "] ";
+    }
+
+    std::optional<std::uint16_t>
+    listeningPort() const override
+    {
+        std::uint32_t const value = listeningPort_;
+        if (value == kUnknownPort)
+            return std::nullopt;
+        return value;
+    }
+
+    void
+    setListeningPort(std::uint16_t port)
+    {
+        listeningPort_ = port;
+    }
+
+    void
+    localEndpoint(beast::IP::Endpoint const& endpoint)
+    {
+        localEndpoint_ = endpoint;
+    }
+
+    void
+    remoteEndpoint(beast::IP::Endpoint const& endpoint)
+    {
+        remoteEndpoint_ = endpoint;
+    }
+
+    void
+    publicKey(PublicKey const& key)
+    {
+        publicKey_ = key;
+    }
+
+    void
+    reserved(bool reserved)
+    {
+        reserved_ = reserved;
+    }
+
+    //--------------------------------------------------------------------------
+
+    void
+    state(State state);
+
+    void
+    activate(clock_type::time_point const& now);
+
+    // "Memberspace"
+    //
+    // The set of all recent addresses that we have seen from this peer.
+    // We try to avoid sending a peer the same addresses they gave us.
+    //
+    class RecentT
+    {
+    public:
+        explicit RecentT(clock_type& clock);
+
+        /**
+         * Called for each valid endpoint received for a slot.
+         * We also insert messages that we send to the slot to prevent
+         * sending a slot the same address too frequently.
+         */
+        void
+        insert(beast::IP::Endpoint const& ep, std::uint32_t hops);
+
+        /**
+         * Returns `true` if we should not send endpoint to the slot.
+         */
+        bool
+        filter(beast::IP::Endpoint const& ep, std::uint32_t hops);
+
+    private:
+        void
+        expire();
+
+        friend class SlotImp;
+        beast::aged_unordered_map<beast::IP::Endpoint, std::uint32_t> cache_;
+    } recent;
+
+    void
+    expire()
+    {
+        recent.expire();
+    }
+
+private:
+    bool const inbound_;
+    bool const fixed_;
+    bool reserved_;
+    State state_;
+    beast::IP::Endpoint remoteEndpoint_;
+    std::optional<beast::IP::Endpoint> localEndpoint_;
+    std::optional<PublicKey> publicKey_;
+
+    static std::int32_t constexpr kUnknownPort = -1;
+    std::atomic<std::int32_t> listeningPort_;
+
+public:
+    // DEPRECATED public data members
+
+    // Tells us if we checked the connection. Outbound connections
+    // are always considered checked since we successfully connected.
+    bool checked;
+
+    // Set to indicate if the connection can receive incoming at the
+    // address advertised in mtENDPOINTS. Only valid if checked is true.
+    bool canAccept;
+
+    // Set to indicate that a connection check for this peer is in
+    // progress. Valid always.
+    bool connectivityCheckInProgress;
+
+    // The time after which we will accept mtENDPOINTS from the peer
+    // This is to prevent flooding or spamming. Receipt of mtENDPOINTS
+    // sooner than the allotted time should impose a load charge.
+    //
+    clock_type::time_point whenAcceptEndpoints;
+};
+
+}  // namespace xrpl::PeerFinder

include/xrpl/peerfinder/detail/Source.h

@@ -0,0 +1,49 @@
+#pragma once
+
+#include <xrpl/beast/utility/Journal.h>
+#include <xrpl/peerfinder/Types.h>
+
+#include <boost/system/error_code.hpp>
+
+#include <string>
+
+namespace xrpl::PeerFinder {
+
+/**
+ * A static or dynamic source of peer addresses.
+ * These are used as fallbacks when we are bootstrapping and don't have
+ * a local cache, or when none of our addresses are functioning. Typically
+ * sources will represent things like static text in the config file, a
+ * separate local file with addresses, or a remote HTTPS URL that can
+ * be updated automatically. Another solution is to use a custom DNS server
+ * that hands out peer IP addresses when name lookups are performed.
+ */
+class Source
+{
+public:
+    /**
+     * The results of a fetch.
+     */
+    struct Results
+    {
+        explicit Results() = default;
+
+        // error_code on a failure
+        boost::system::error_code error;
+
+        // list of fetched endpoints
+        IPAddresses addresses;
+    };
+
+    virtual ~Source() = default;
+    virtual std::string const&
+    name() = 0;
+    virtual void
+    cancel()
+    {
+    }
+    virtual void
+    fetch(Results& results, beast::Journal journal) = 0;
+};
+
+}  // namespace xrpl::PeerFinder

include/xrpl/peerfinder/detail/SourceStrings.h

@@ -0,0 +1,25 @@
+#pragma once
+
+#include <xrpl/peerfinder/detail/Source.h>
+
+#include <memory>
+#include <string>
+#include <vector>
+
+namespace xrpl::PeerFinder {
+
+/**
+ * Provides addresses from a static set of strings.
+ */
+class SourceStrings : public Source
+{
+public:
+    explicit SourceStrings() = default;
+
+    using Strings = std::vector<std::string>;
+
+    static std::shared_ptr<Source>
+    make(std::string const& name, Strings const& strings);
+};
+
+}  // namespace xrpl::PeerFinder

include/xrpl/peerfinder/detail/Store.h

@@ -0,0 +1,36 @@
+#pragma once
+
+#include <xrpl/beast/net/IPEndpoint.h>
+
+#include <cstddef>
+#include <functional>
+#include <vector>
+
+namespace xrpl::PeerFinder {
+
+/**
+ * Abstract persistence for PeerFinder data.
+ */
+class Store
+{
+public:
+    virtual ~Store() = default;
+
+    // load the bootstrap cache
+    using load_callback = std::function<void(beast::IP::Endpoint, int)>;
+    virtual std::size_t
+    load(load_callback const& cb) = 0;
+
+    // save the bootstrap cache
+    struct Entry
+    {
+        explicit Entry() = default;
+
+        beast::IP::Endpoint endpoint;
+        int valence{};
+    };
+    virtual void
+    save(std::vector<Entry> const& v) = 0;
+};
+
+}  // namespace xrpl::PeerFinder

include/xrpl/peerfinder/detail/Tuning.h

@@ -0,0 +1,115 @@
+#pragma once
+
+#include <algorithm>
+#include <array>
+#include <chrono>
+#include <cstdint>
+
+/**
+ * Heuristically tuned constants.
+ */
+/** @{ */
+namespace xrpl::PeerFinder::Tuning {
+
+//---------------------------------------------------------
+//
+// Automatic Connection Policy
+//
+//---------------------------------------------------------
+
+/**
+ * Time to wait between making batches of connection attempts
+ */
+static constexpr auto kSecondsPerConnect = 10;
+
+/**
+ * Maximum number of simultaneous connection attempts.
+ */
+static constexpr auto kMaxConnectAttempts = 20;
+
+/**
+ * The percentage of total peer slots that are outbound.
+ * The number of outbound peers will be the larger of the
+ * minOutCount and outPercent * Config::maxPeers specially
+ * rounded.
+ */
+static constexpr auto kOutPercent = 15;
+
+/**
+ * A hard minimum on the number of outgoing connections.
+ * This is enforced outside the Logic, so that the unit test
+ * can use any settings it wants.
+ */
+static constexpr auto kMinOutCount = 10;
+
+/**
+ * The default value of Config::maxPeers.
+ */
+static constexpr auto kDefaultMaxPeers = 21;
+
+/**
+ * Max redirects we will accept from one connection.
+ * Redirects are limited for security purposes, to prevent
+ * the address caches from getting flooded.
+ */
+static constexpr auto kMaxRedirects = 30;
+
+//------------------------------------------------------------------------------
+//
+// Fixed
+//
+//------------------------------------------------------------------------------
+
+static std::array<int, 10> const kConnectionBackoff{{1, 1, 2, 3, 5, 8, 13, 21, 34, 55}};
+
+//------------------------------------------------------------------------------
+//
+// Bootcache
+//
+//------------------------------------------------------------------------------
+
+// Threshold of cache entries above which we trim.
+static constexpr auto kBootcacheSize = 1000;
+
+// The percentage of addresses we prune when we trim the cache.
+static constexpr auto kBootcachePrunePercent = 10;
+
+// The cool down wait between database updates
+// Ideally this should be larger than the time it takes a full
+// peer to send us a set of addresses and then disconnect.
+//
+static std::chrono::seconds const kBootcacheCooldownTime(60);
+
+//------------------------------------------------------------------------------
+//
+// Livecache
+//
+//------------------------------------------------------------------------------
+
+// Drop incoming messages with hops greater than this number
+constexpr std::uint32_t kMaxHops = 6;
+
+// How many Endpoint to send in each mtENDPOINTS
+constexpr std::uint32_t kNumberOfEndpoints = 2 * kMaxHops;
+
+// The most Endpoint we will accept in mtENDPOINTS
+constexpr std::uint32_t kNumberOfEndpointsMax =
+    std::max<decltype(kNumberOfEndpoints)>(kNumberOfEndpoints * 2, 64);
+
+// Number of addresses we provide when redirecting.
+constexpr std::uint32_t kRedirectEndpointCount = 10;
+
+// How often we send or accept mtENDPOINTS messages per peer
+// (we use a prime number of purpose)
+constexpr std::chrono::seconds kSecondsPerMessage(151);
+
+// How long an Endpoint will stay in the cache
+// This should be a small multiple of the broadcast frequency
+constexpr std::chrono::seconds kLiveCacheSecondsToLive(30);
+
+// How much time to wait before trying an outgoing address again.
+// Note that we ignore the port for purposes of comparison.
+constexpr std::chrono::seconds kRecentAttemptDuration(60);
+
+}  // namespace xrpl::PeerFinder::Tuning
+/** @} */

include/xrpl/peerfinder/make_Manager.h

@@ -0,0 +1,36 @@
+#pragma once
+
+#include <xrpl/beast/insight/Collector.h>
+#include <xrpl/beast/utility/Journal.h>
+#include <xrpl/peerfinder/PeerfinderManager.h>
+#include <xrpl/peerfinder/Types.h>
+#include <xrpl/peerfinder/detail/Store.h>
+
+#include <boost/asio/io_context.hpp>
+
+#include <memory>
+
+namespace xrpl::PeerFinder {
+
+/**
+ * @brief Create a new Manager.
+ *
+ * @param ioContext The io_context used to schedule asynchronous work.
+ * @param clock The clock used for timekeeping.
+ * @param journal The journal used for logging.
+ * @param store The persistence backend for the bootstrap cache. The caller
+ *     retains ownership and must keep it alive (and opened) for the lifetime of
+ *     the returned Manager. This lets consumers supply their own Store
+ *     implementation (e.g. the SQLite-backed StoreSqdb in xrpld).
+ * @param collector The collector used to report metrics.
+ * @return The newly created Manager.
+ */
+std::unique_ptr<Manager>
+makeManager(
+    boost::asio::io_context& ioContext,
+    clock_type& clock,
+    beast::Journal journal,
+    Store& store,
+    beast::insight::Collector::ptr const& collector);
+
+}  // namespace xrpl::PeerFinder