Merge branch 'dev' into udp-peer-v2

Builds/levelization/results/loops.txt

@@ -26,7 +26,7 @@ Loop: xrpld.app xrpld.nodestore
   xrpld.app > xrpld.nodestore
 
 Loop: xrpld.app xrpld.overlay
-  xrpld.overlay ~= xrpld.app
+  xrpld.overlay == xrpld.app
 
 Loop: xrpld.app xrpld.peerfinder
   xrpld.app > xrpld.peerfinder

Builds/levelization/results/ordering.txt

@@ -103,6 +103,7 @@ test.overlay > xrpld.overlay
 test.overlay > xrpld.peerfinder
 test.overlay > xrpld.shamap
 test.overlay > xrpl.protocol
+test.overlay > xrpl.resource
 test.peerfinder > test.beast
 test.peerfinder > test.unit_test
 test.peerfinder > xrpl.basics

cfg/xahaud-example.cfg

@@ -564,6 +564,30 @@
 #
 #
 #
+# [udp_highway]
+#
+#   0 or 1.
+#
+#   0: Disable the Xahau UDP Superhighway (XUSH). [default]
+#   1: Enable the Xahau UDP Superhighway (XUSH).
+#
+#   The UDP Superhighway is a best-effort transport that runs alongside the
+#   peer protocol. When it is enabled the server also listens for UDP
+#   datagrams on its peer port number, and every transaction it relays to its
+#   peers is also sent, over UDP, to a small random set of other servers that
+#   have the highway enabled. It never replaces the peer protocol: anything
+#   sent over the highway is also relayed normally.
+#
+#   To receive highway traffic, inbound UDP must be allowed to the peer port
+#   (in addition to inbound TCP).
+#
+#   A server that keeps its address private (peer_private = 1, or any server
+#   configured as a validator) only exchanges highway traffic with the peers
+#   listed in [ips_fixed], and never advertises highway endpoints. Fixed peers
+#   are never advertised to other servers.
+#
+#
+#
 # [peers_max]
 #
 #   The largest number of desired peer connections (incoming or outgoing).

include/xrpl/server/Port.h

@@ -100,8 +100,18 @@ struct Port
         return protocol.count("udp") > 0;
     }
 
+    bool
+    has_peer() const
+    {
+        return protocol.count("peer") > 0;
+    }
+
     // Maximum UDP packet size (default 64KB)
     std::size_t udp_packet_size = 65536;
+
+    // If this is a peer port, also listen for UDP Superhighway (XUSH)
+    // datagrams on the same port number.
+    bool udp_highway = false;
 };
 
 std::ostream&

include/xrpl/server/detail/ServerImpl.h

@@ -208,6 +208,16 @@ ServerImpl<Handler>::ports(std::vector<Port> const& ports)
 
                 sp->run();
             }
+
+            // UDP Superhighway (XUSH) datagrams arrive on the peer port
+            // number. TCP and UDP port numbers are independent, so this
+            // needs no socket sharing with the TCP door.
+            if (internalPort.has_peer() && internalPort.udp_highway)
+            {
+                if (auto sp = ios_.emplace<UDPDoor<Handler>>(
+                        handler_, io_context_, internalPort, j_))
+                    sp->run();
+            }
         }
     }
     return eps;

include/xrpl/server/detail/UDPDoor.h

@@ -78,10 +78,9 @@ public:
         error_code ec;
 
         // Create UDP endpoint from port configuration
-        auto const addr = port_.ip.to_v4();
-        boost::asio::ip::udp::endpoint udp_endpoint(addr, port_.port);
+        boost::asio::ip::udp::endpoint udp_endpoint(port_.ip, port_.port);
 
-        socket_.open(boost::asio::ip::udp::v4(), ec);
+        socket_.open(udp_endpoint.protocol(), ec);
         if (ec)
         {
             JLOG(j_.error()) << "UDP socket open failed: " << ec.message();
@@ -94,17 +93,22 @@ public:
         {
             JLOG(j_.error())
                 << "UDP set reuse_address failed: " << ec.message();
+            close();
             return;
         }
 
         socket_.bind(udp_endpoint, ec);
         if (ec)
         {
-            JLOG(j_.error()) << "UDP socket bind failed: " << ec.message();
+            // Close the socket so that run() does not start receiving on an
+            // unbound socket, which would silently bind an ephemeral port.
+            JLOG(j_.error()) << "UDP socket bind to " << udp_endpoint
+                             << " failed: " << ec.message();
+            close();
             return;
         }
 
-        JLOG(j_.info()) << "UDP-RPC listening on " << udp_endpoint;
+        JLOG(j_.info()) << "UDP listening on " << udp_endpoint;
     }
 
     endpoint_type
@@ -169,6 +173,7 @@ private:
         handler_.onUDPMessage(
             std::string(recv_buffer_.data(), bytes_transferred),
             tcp_endpoint,
+            port_,
             [this, tcp_endpoint](std::string const& response) {
                 do_send(response, tcp_endpoint);
             });

src/test/core/Config_test.cpp

@@ -454,6 +454,39 @@ port_wss_admin
         }
     }
 
+    void
+    testUdpHighway()
+    {
+        testcase("udp_highway");
+
+        // The UDP Superhighway is opt-in: off unless turned on.
+        {
+            Config c;
+            c.loadFromString("");
+            BEAST_EXPECT(!c.UDP_HIGHWAY);
+        }
+
+        std::vector<std::pair<std::string, bool>> const values{
+            {"0", false}, {"false", false}, {"1", true}, {"TRUE", true}};
+        for (auto const& [value, expected] : values)
+        {
+            Config c;
+            c.loadFromString("[udp_highway]\n" + value + "\n");
+            BEAST_EXPECT(c.UDP_HIGHWAY == expected);
+        }
+
+        try
+        {
+            Config c;
+            c.loadFromString("[udp_highway]\nsometimes\n");
+            fail();
+        }
+        catch (std::exception const&)
+        {
+            pass();
+        }
+    }
+
     void
     testNetworkID()
     {
@@ -1792,6 +1825,7 @@ r.ripple.com:51235
         testOverlay();
         testNetworkID();
         testEarlyManifestSync();
+        testUdpHighway();
     }
 };
 

src/test/overlay/OverlayXUSH_test.cpp

@@ -0,0 +1,542 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2026 XRPL Labs
+
+    Permission to use, copy, modify, and/or distribute this software for any
+    purpose  with  or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#include <test/jtx.h>
+#include <xrpld/app/misc/HashRouter.h>
+#include <xrpld/core/JobQueue.h>
+#include <xrpld/overlay/detail/OverlayImpl.h>
+#include <xrpld/overlay/detail/XUSH.h>
+#include <xrpl/beast/net/IPAddressConversion.h>
+#include <xrpl/protocol/Feature.h>
+#include <xrpl/protocol/STTx.h>
+#include <xrpl/resource/ResourceManager.h>
+#include <xrpl/resource/detail/Tuning.h>
+
+#include <boost/asio/ip/udp.hpp>
+
+#include <algorithm>
+#include <array>
+#include <chrono>
+#include <cstring>
+#include <optional>
+#include <set>
+#include <string>
+#include <thread>
+#include <vector>
+
+namespace ripple {
+namespace test {
+
+// Tests the UDP Superhighway (XUSH) handling in OverlayImpl: inbound datagram
+// processing, outbound publishing and the XUSHPEER advertisement timer.
+class OverlayXUSH_test : public beast::unit_test::suite
+{
+    using udp = boost::asio::ip::udp;
+
+    static std::unique_ptr<Config>
+    highway(std::unique_ptr<Config> cfg)
+    {
+        cfg->UDP_HIGHWAY = true;
+        return cfg;
+    }
+
+    static beast::IP::Endpoint
+    ep(std::string const& s)
+    {
+        return beast::IP::Endpoint::from_string(s);
+    }
+
+    static std::set<beast::IP::Endpoint>
+    asSet(std::vector<beast::IP::Endpoint> const& v)
+    {
+        return {v.begin(), v.end()};
+    }
+
+    static Slice
+    slice(Buffer const& b)
+    {
+        return Slice(b.data(), b.size());
+    }
+
+    static std::string
+    str(Buffer const& b)
+    {
+        return std::string(reinterpret_cast<char const*>(b.data()), b.size());
+    }
+
+    static Buffer
+    serialize(STTx const& stx)
+    {
+        Serializer s;
+        stx.add(s);
+        return Buffer(s.data(), s.size());
+    }
+
+    // A pseudo-transaction, which servers never submit
+    static STTx
+    pseudoTx()
+    {
+        return STTx(ttAMENDMENT, [](auto& obj) {
+            obj.setAccountID(sfAccount, AccountID());
+            obj.setFieldH256(sfAmendment, uint256(2));
+            obj.setFieldU32(sfLedgerSequence, 5);
+        });
+    }
+
+    // A local UDP socket that highway datagrams can be sent to
+    struct Sink
+    {
+        boost::asio::io_context io;
+        udp::socket socket;
+
+        explicit Sink(boost::asio::ip::address const& addr)
+            : socket(io, udp::endpoint(addr, 0))
+        {
+            socket.non_blocking(true);
+        }
+
+        beast::IP::Endpoint
+        endpoint() const
+        {
+            auto const local = socket.local_endpoint();
+            return beast::IP::Endpoint(local.address(), local.port());
+        }
+
+        void
+        sendTo(Buffer const& datagram, std::uint16_t port)
+        {
+            socket.send_to(
+                boost::asio::buffer(datagram.data(), datagram.size()),
+                udp::endpoint(socket.local_endpoint().address(), port));
+        }
+
+        std::optional<std::string>
+        receive(std::chrono::milliseconds timeout = std::chrono::seconds(5))
+        {
+            auto const deadline = std::chrono::steady_clock::now() + timeout;
+            std::array<char, 4096> buf;
+            while (std::chrono::steady_clock::now() < deadline)
+            {
+                udp::endpoint from;
+                boost::system::error_code ec;
+                auto const n =
+                    socket.receive_from(boost::asio::buffer(buf), from, 0, ec);
+                if (!ec)
+                    return std::string(buf.data(), n);
+                if (ec != boost::asio::error::would_block &&
+                    ec != boost::asio::error::try_again)
+                    return std::nullopt;
+                std::this_thread::sleep_for(std::chrono::milliseconds(5));
+            }
+            return std::nullopt;
+        }
+
+        // Receives datagrams until one of the given type arrives
+        std::optional<std::string>
+        receive(xush::MessageType type)
+        {
+            while (auto d = receive())
+            {
+                if (xush::classify(makeSlice(*d)) == type)
+                    return d;
+            }
+            return std::nullopt;
+        }
+    };
+
+    struct Fixture
+    {
+        jtx::Env env;
+        OverlayImpl& overlay;
+        PeerFinder::Manager& peerFinder;
+        Resource::Manager& resources;
+
+        Fixture(beast::unit_test::suite& suite, bool enabled)
+            : env(suite, enabled ? jtx::envconfig(highway) : jtx::envconfig())
+            , overlay(dynamic_cast<OverlayImpl&>(env.app().overlay()))
+            , peerFinder(overlay.peerFinder())
+            , resources(env.app().getResourceManager())
+        {
+        }
+
+        // Delivers a datagram as if it had been received from `remote`
+        void
+        deliver(Buffer const& datagram, beast::IP::Endpoint const& remote)
+        {
+            overlay.processXUSH(
+                str(datagram), beast::IP::to_asio_endpoint(remote));
+        }
+
+        void
+        deliver(
+            std::vector<Buffer> const& datagrams,
+            beast::IP::Endpoint const& remote)
+        {
+            for (auto const& d : datagrams)
+                deliver(d, remote);
+        }
+
+        // Tracks the resource balance charged to `remote`
+        Resource::Consumer
+        probe(beast::IP::Endpoint const& remote)
+        {
+            return resources.newInboundEndpoint(remote);
+        }
+
+        std::set<beast::IP::Endpoint>
+        targets()
+        {
+            return asSet(peerFinder.highway_targets(1000));
+        }
+
+        // Whether `probe` was charged at least `fee` since `before`. Balances
+        // are normalized by the decay window, so a charge of `fee` raises the
+        // balance by at least fee / decayWindowSeconds.
+        static bool
+        charged(Resource::Consumer& probe, int before, int fee)
+        {
+            return probe.balance() - before >=
+                fee / Resource::decayWindowSeconds;
+        }
+
+        bool
+        isBad(uint256 const& txid)
+        {
+            env.app().getJobQueue().rendezvous();
+            return env.app().getHashRouter().getFlags(txid) & SF_BAD;
+        }
+    };
+
+    void
+    testDisabled()
+    {
+        testcase("disabled");
+        using namespace jtx;
+        Fixture f(*this, false);
+        auto const remote = ep("203.0.113.1:51235");
+
+        // Nothing is learned or sent while the highway is disabled
+        f.deliver(xush::encodePeers({ep("8.8.8.8:51235")}), remote);
+        BEAST_EXPECT(f.targets().empty());
+
+        Sink sink(boost::asio::ip::make_address("127.0.0.1"));
+        f.peerFinder.addFixedPeer("sink", {sink.endpoint()});
+        BEAST_EXPECT(f.targets().empty());
+
+        Account const alice("alice");
+        f.env.fund(XRP(10000), alice);
+        f.env.close();
+
+        Buffer const blob(64);
+        f.overlay.publishTxXUSH(slice(blob), uint256(1));
+        BEAST_EXPECT(!sink.receive(std::chrono::milliseconds(200)));
+    }
+
+    void
+    testPeers()
+    {
+        testcase("XUSHPEER");
+        Fixture f(*this, true);
+        auto const remote = ep("203.0.113.2:51235");
+        auto probe = f.probe(remote);
+
+        BEAST_EXPECT(f.targets().empty());
+
+        f.deliver(
+            xush::encodePeers({ep("8.8.8.8:51235"), ep("1.1.1.1:51235")}),
+            remote);
+        BEAST_EXPECT(
+            f.targets() == asSet({ep("8.8.8.8:51235"), ep("1.1.1.1:51235")}));
+        auto const balance = probe.balance();
+
+        // Unknown datagram types are ignored and cost almost nothing
+        f.deliver(Buffer("XUSHPING", 8), remote);
+        BEAST_EXPECT(!Fixture::charged(probe, balance, 100));
+
+        // A malformed advertisement is charged for
+        Buffer malformed(xush::tagSize + 2);
+        std::memcpy(malformed.data(), "XUSHPEER\x05\x00", malformed.size());
+        f.deliver(malformed, remote);
+        BEAST_EXPECT(Fixture::charged(probe, balance, 200));
+        BEAST_EXPECT(f.targets().size() == 2);
+    }
+
+    void
+    testTransaction()
+    {
+        testcase("XUSHTXNF transaction");
+        using namespace jtx;
+        Fixture f(*this, true);
+        auto& env = f.env;
+        auto const remote = ep("203.0.113.3:51235");
+
+        Account const alice("alice");
+        Account const bob("bob");
+        env.fund(XRP(10000), alice, bob);
+        env.close();
+
+        auto const jt = env.jt(pay(alice, bob, XRP(100)));
+        auto const txid = jt.stx->getTransactionID();
+
+        // Fragments arrive out of order and are reassembled
+        auto datagrams = xush::encodeTxn(slice(serialize(*jt.stx)), txid, 40);
+        BEAST_EXPECT(datagrams.size() > 2);
+        std::reverse(datagrams.begin(), datagrams.end());
+        f.deliver(datagrams, remote);
+        env.app().getJobQueue().rendezvous();
+
+        BEAST_EXPECT(!f.isBad(txid));
+        env.close();
+        BEAST_EXPECT(env.balance(bob) == XRP(10100));
+
+        // Delivering it again is harmless
+        f.deliver(datagrams, remote);
+        env.app().getJobQueue().rendezvous();
+        env.close();
+        BEAST_EXPECT(env.balance(bob) == XRP(10100));
+    }
+
+    void
+    testBadTransactions()
+    {
+        testcase("XUSHTXNF bad transactions");
+        using namespace jtx;
+        Fixture f(*this, true);
+        auto& env = f.env;
+        auto const remote = ep("203.0.113.4:51235");
+        auto probe = f.probe(remote);
+
+        Account const alice("alice");
+        Account const bob("bob");
+        env.fund(XRP(10000), alice, bob);
+        env.close();
+
+        auto const good = env.jt(pay(alice, bob, XRP(100)));
+        auto const blob = serialize(*good.stx);
+        auto const txid = good.stx->getTransactionID();
+
+        // The ID in the header must match the transaction
+        auto balance = probe.balance();
+        f.deliver(xush::encodeTxn(slice(blob), uint256(1)), remote);
+        BEAST_EXPECT(Fixture::charged(probe, balance, 400));
+        BEAST_EXPECT(!f.isBad(txid));
+
+        // Garbage that doesn't parse as a transaction
+        balance = probe.balance();
+        f.deliver(
+            xush::encodeTxn(makeSlice(std::string("garbage")), uint256(2)),
+            remote);
+        BEAST_EXPECT(Fixture::charged(probe, balance, 400));
+
+        // Fragments of one transaction that disagree with each other
+        balance = probe.balance();
+        f.deliver(xush::encodeTxn(slice(blob), uint256(3), 40).front(), remote);
+        f.deliver(xush::encodeTxn(slice(blob), uint256(3), 60).front(), remote);
+        BEAST_EXPECT(Fixture::charged(probe, balance, 400));
+
+        // A malformed fragment
+        balance = probe.balance();
+        f.deliver(Buffer("XUSHTXNF", 8), remote);
+        BEAST_EXPECT(Fixture::charged(probe, balance, 200));
+
+        // Pseudo-transactions are never accepted
+        balance = probe.balance();
+        auto const pseudo = pseudoTx();
+        f.deliver(
+            xush::encodeTxn(
+                slice(serialize(pseudo)), pseudo.getTransactionID()),
+            remote);
+        BEAST_EXPECT(Fixture::charged(probe, balance, 2000));
+
+        // An expired transaction is remembered as bad
+        auto const expired =
+            env.jt(pay(alice, bob, XRP(100)), last_ledger_seq(1));
+        auto const expiredId = expired.stx->getTransactionID();
+        balance = probe.balance();
+        f.deliver(
+            xush::encodeTxn(slice(serialize(*expired.stx)), expiredId), remote);
+        BEAST_EXPECT(f.isBad(expiredId));
+        BEAST_EXPECT(Fixture::charged(probe, balance, 150));
+
+        // A badly signed transaction is remembered as bad
+        auto corrupted = good.stx->getJson(JsonOptions::none);
+        corrupted[jss::TxnSignature] = std::string(128, '0');
+        STTx const badSig(parse(corrupted));
+        auto const badSigId = badSig.getTransactionID();
+        balance = probe.balance();
+        f.deliver(xush::encodeTxn(slice(serialize(badSig)), badSigId), remote);
+        BEAST_EXPECT(f.isBad(badSigId));
+        BEAST_EXPECT(Fixture::charged(probe, balance, 2000));
+
+        // A known bad transaction is charged for again
+        balance = probe.balance();
+        f.deliver(xush::encodeTxn(slice(serialize(badSig)), badSigId), remote);
+        BEAST_EXPECT(Fixture::charged(probe, balance, 150));
+
+        env.close();
+        BEAST_EXPECT(env.balance(bob) == XRP(10000));
+    }
+
+    void
+    testDisconnect()
+    {
+        testcase("resource limit");
+        Fixture f(*this, true);
+        auto const remote = ep("203.0.113.5:51235");
+        auto probe = f.probe(remote);
+
+        // A sender that keeps misbehaving is ignored
+        Buffer const blob(64);
+        for (int i = 0; i < 10000 && probe.balance() < Resource::dropThreshold;
+             ++i)
+            f.deliver(xush::encodeTxn(slice(blob), uint256(1)), remote);
+        BEAST_EXPECT(probe.balance() >= Resource::dropThreshold);
+
+        f.deliver(xush::encodePeers({ep("8.8.8.8:51235")}), remote);
+        BEAST_EXPECT(f.targets().empty());
+
+        // Other senders are not affected
+        f.deliver(
+            xush::encodePeers({ep("8.8.8.8:51235")}), ep("203.0.113.6:51235"));
+        BEAST_EXPECT(f.targets() == asSet({ep("8.8.8.8:51235")}));
+    }
+
+    void
+    testPublish()
+    {
+        testcase("publish");
+        using namespace jtx;
+        Fixture f(*this, true);
+        auto& env = f.env;
+
+        Account const alice("alice");
+        Account const bob("bob");
+        env.fund(XRP(10000), alice, bob);
+        env.close();
+
+        Sink sink4(boost::asio::ip::make_address("127.0.0.1"));
+        f.peerFinder.addFixedPeer("sink4", {sink4.endpoint()});
+
+        std::optional<Sink> sink6;
+        try
+        {
+            sink6.emplace(boost::asio::ip::make_address("::1"));
+            f.peerFinder.addFixedPeer("sink6", {sink6->endpoint()});
+        }
+        catch (std::exception const&)
+        {
+            log << "IPv6 loopback unavailable, skipping" << std::endl;
+        }
+
+        // A transaction relayed by NetworkOPs is published on the highway
+        env(pay(alice, bob, XRP(100)));
+        auto const stx = env.tx();
+        auto const blob = serialize(*stx);
+
+        auto check = [&](Sink& sink) {
+            auto const d = sink.receive(xush::MessageType::txn);
+            if (!BEAST_EXPECT(d))
+                return;
+            auto const fragment = xush::decodeTxnFragment(makeSlice(*d));
+            if (!BEAST_EXPECT(fragment))
+                return;
+            BEAST_EXPECT(fragment->txid == stx->getTransactionID());
+            BEAST_EXPECT(fragment->count == 1);
+            BEAST_EXPECT(fragment->payload == slice(blob));
+        };
+        check(sink4);
+        if (sink6)
+            check(*sink6);
+
+        // Large transactions are fragmented
+        Buffer big(5000);
+        for (std::size_t i = 0; i < big.size(); ++i)
+            big.data()[i] = static_cast<std::uint8_t>(i);
+        f.overlay.publishTxXUSH(slice(big), uint256(7));
+
+        xush::Reassembler reassembler;
+        std::optional<Buffer> rebuilt;
+        while (!rebuilt)
+        {
+            auto const d = sink4.receive(xush::MessageType::txn);
+            if (!BEAST_EXPECT(d))
+                break;
+            auto const fragment = xush::decodeTxnFragment(makeSlice(*d));
+            if (!BEAST_EXPECT(fragment && fragment->txid == uint256(7)))
+                break;
+            auto result = reassembler.add(
+                sink4.endpoint(),
+                *fragment,
+                xush::Reassembler::clock_type::now());
+            if (result.status == xush::Reassembler::Status::complete)
+                rebuilt = std::move(result.txn);
+        }
+        BEAST_EXPECT(rebuilt && *rebuilt == big);
+
+        // Transactions too large for the highway are not sent
+        Buffer const huge(xush::maxTxnSize + 1);
+        f.overlay.publishTxXUSH(slice(huge), uint256(8));
+        f.overlay.publishTxXUSH(slice(big), uint256(9));
+        auto const d = sink4.receive(xush::MessageType::txn);
+        if (BEAST_EXPECT(d))
+        {
+            auto const fragment = xush::decodeTxnFragment(makeSlice(*d));
+            BEAST_EXPECT(fragment && fragment->txid == uint256(9));
+        }
+    }
+
+    void
+    testSendPeers()
+    {
+        testcase("advertise");
+        Fixture f(*this, true);
+
+        Sink sink(boost::asio::ip::make_address("127.0.0.1"));
+        f.peerFinder.addFixedPeer("sink", {sink.endpoint()});
+
+        // Learned endpoints are advertised to targets once per second, but
+        // fixed peers are never advertised.
+        f.peerFinder.add_highway_peers({ep("8.8.8.8:51235")});
+
+        auto const d = sink.receive(xush::MessageType::peers);
+        if (!BEAST_EXPECT(d))
+            return;
+        auto const endpoints = xush::decodePeers(makeSlice(*d));
+        BEAST_EXPECT(
+            endpoints && asSet(*endpoints) == asSet({ep("8.8.8.8:51235")}));
+    }
+
+public:
+    void
+    run() override
+    {
+        testDisabled();
+        testPeers();
+        testTransaction();
+        testBadTransactions();
+        testDisconnect();
+        testPublish();
+        testSendPeers();
+    }
+};
+
+BEAST_DEFINE_TESTSUITE(OverlayXUSH, overlay, ripple);
+
+}  // namespace test
+}  // namespace ripple

src/test/overlay/XUSH_test.cpp

@@ -0,0 +1,466 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2026 XRPL Labs
+
+    Permission to use, copy, modify, and/or distribute this software for any
+    purpose  with  or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#include <xrpld/overlay/detail/XUSH.h>
+#include <xrpl/beast/unit_test.h>
+
+#include <algorithm>
+#include <chrono>
+#include <cstring>
+#include <set>
+#include <string>
+#include <vector>
+
+namespace ripple {
+namespace test {
+
+class XUSH_test : public beast::unit_test::suite
+{
+    using Status = xush::Reassembler::Status;
+
+    static beast::IP::Endpoint
+    ep(std::string const& s)
+    {
+        return beast::IP::Endpoint::from_string(s);
+    }
+
+    static Slice
+    slice(Buffer const& b)
+    {
+        return Slice(b.data(), b.size());
+    }
+
+    static Buffer
+    makeTxn(std::size_t size)
+    {
+        Buffer b(size);
+        for (std::size_t i = 0; i < size; ++i)
+            b.data()[i] = static_cast<std::uint8_t>(i * 31 + 7);
+        return b;
+    }
+
+    static void
+    putU32(std::uint8_t* p, std::uint32_t v)
+    {
+        p[0] = static_cast<std::uint8_t>(v >> 24);
+        p[1] = static_cast<std::uint8_t>(v >> 16);
+        p[2] = static_cast<std::uint8_t>(v >> 8);
+        p[3] = static_cast<std::uint8_t>(v);
+    }
+
+    // A XUSHTXNF datagram with an arbitrary (possibly invalid) header
+    static Buffer
+    rawFragment(
+        uint256 const& txid,
+        std::uint32_t totalSize,
+        std::uint32_t count,
+        std::uint32_t index,
+        std::size_t payloadSize)
+    {
+        Buffer b(xush::txnHeaderSize + payloadSize);
+        std::memset(b.data(), 0xAB, b.size());
+        std::memcpy(b.data(), "XUSHTXNF", xush::tagSize);
+        std::memcpy(b.data() + xush::tagSize, txid.data(), txid.size());
+        putU32(b.data() + 40, totalSize);
+        putU32(b.data() + 44, count);
+        putU32(b.data() + 48, index);
+        return b;
+    }
+
+    // A XUSHPEER datagram with arbitrary (possibly invalid) counts
+    static Buffer
+    rawPeers(std::uint8_t n4, std::uint8_t n6, std::size_t bodySize)
+    {
+        Buffer b(xush::tagSize + 2 + bodySize);
+        std::memset(b.data(), 0, b.size());
+        std::memcpy(b.data(), "XUSHPEER", xush::tagSize);
+        b.data()[8] = n4;
+        b.data()[9] = n6;
+        return b;
+    }
+
+    void
+    testClassify()
+    {
+        testcase("classify");
+        using namespace xush;
+
+        auto type = [](std::string const& s) { return classify(makeSlice(s)); };
+
+        BEAST_EXPECT(type("XUSHPEER") == MessageType::peers);
+        BEAST_EXPECT(type("XUSHPEER and more") == MessageType::peers);
+        BEAST_EXPECT(type("XUSHTXNF") == MessageType::txn);
+        BEAST_EXPECT(type("XUSHPING") == MessageType::unknown);
+        BEAST_EXPECT(type("XUSHPEE") == MessageType::unknown);
+        BEAST_EXPECT(type("") == MessageType::unknown);
+        BEAST_EXPECT(type("{\"command\":\"ping\"}") == MessageType::unknown);
+    }
+
+    void
+    testPeers()
+    {
+        testcase("XUSHPEER");
+        using namespace xush;
+
+        std::vector<beast::IP::Endpoint> const endpoints{
+            ep("1.2.3.4:51235"), ep("[2001:db8::1]:21337"), ep("5.6.7.8:1")};
+
+        auto const datagram = encodePeers(endpoints);
+        BEAST_EXPECT(datagram.size() == tagSize + 2 + 2 * 8 + 20);
+        BEAST_EXPECT(classify(slice(datagram)) == MessageType::peers);
+
+        if (auto const decoded = decodePeers(slice(datagram));
+            BEAST_EXPECT(decoded))
+        {
+            BEAST_EXPECT(
+                std::set<beast::IP::Endpoint>(
+                    decoded->begin(), decoded->end()) ==
+                std::set<beast::IP::Endpoint>(
+                    endpoints.begin(), endpoints.end()));
+        }
+
+        // An empty advertisement is valid
+        if (auto const decoded = decodePeers(slice(encodePeers({})));
+            BEAST_EXPECT(decoded))
+            BEAST_EXPECT(decoded->empty());
+
+        // At most maxAdvertisedPeers are advertised, and the result fits in
+        // a single datagram even if they are all IPv6
+        {
+            std::vector<beast::IP::Endpoint> many;
+            for (std::uint16_t i = 1; i <= maxAdvertisedPeers + 10; ++i)
+                many.emplace_back(
+                    boost::asio::ip::make_address("2001:db8::1"), i);
+
+            auto const big = encodePeers(many);
+            BEAST_EXPECT(big.size() <= maxDatagramSize);
+
+            auto const decoded = decodePeers(slice(big));
+            BEAST_EXPECT(decoded && decoded->size() == maxAdvertisedPeers);
+        }
+
+        // Truncated or padded
+        BEAST_EXPECT(!decodePeers(Slice(datagram.data(), datagram.size() - 1)));
+        {
+            Buffer padded(datagram.size() + 1);
+            std::memcpy(padded.data(), datagram.data(), datagram.size());
+            BEAST_EXPECT(!decodePeers(slice(padded)));
+        }
+        BEAST_EXPECT(!decodePeers(Slice(datagram.data(), tagSize + 1)));
+
+        // Too many endpoints, even with a consistent size
+        BEAST_EXPECT(!decodePeers(slice(rawPeers(
+            maxAdvertisedPeers + 1, 0, (maxAdvertisedPeers + 1) * 8))));
+        BEAST_EXPECT(decodePeers(
+            slice(rawPeers(maxAdvertisedPeers, 0, maxAdvertisedPeers * 8))));
+
+        // A port that doesn't fit in 16 bits
+        {
+            auto bad = rawPeers(1, 0, 8);
+            putU32(bad.data() + tagSize + 2 + 4, 0x10000);
+            BEAST_EXPECT(!decodePeers(slice(bad)));
+            putU32(bad.data() + tagSize + 2 + 4, 0xFFFF);
+            BEAST_EXPECT(decodePeers(slice(bad)));
+        }
+
+        // Wrong tag
+        BEAST_EXPECT(!decodePeers(slice(rawFragment(uint256{1}, 1, 1, 0, 1))));
+    }
+
+    void
+    testTxnEncoding()
+    {
+        testcase("XUSHTXNF encoding");
+        using namespace xush;
+
+        uint256 const txid{42};
+
+        // A typical transaction fits in a single datagram
+        {
+            auto const txn = makeTxn(200);
+            auto const datagrams = encodeTxn(slice(txn), txid);
+            BEAST_EXPECT(datagrams.size() == 1);
+            BEAST_EXPECT(datagrams[0].size() == txnHeaderSize + 200);
+
+            auto const fragment = decodeTxnFragment(slice(datagrams[0]));
+            if (BEAST_EXPECT(fragment))
+            {
+                BEAST_EXPECT(fragment->txid == txid);
+                BEAST_EXPECT(fragment->totalSize == 200);
+                BEAST_EXPECT(fragment->count == 1);
+                BEAST_EXPECT(fragment->index == 0);
+                BEAST_EXPECT(fragment->payload == slice(txn));
+            }
+        }
+
+        // Larger transactions are split, and no datagram is too large
+        for (auto const size :
+             {maxFragmentPayload,
+              maxFragmentPayload + 1,
+              3 * maxFragmentPayload,
+              3 * maxFragmentPayload + 17})
+        {
+            auto const txn = makeTxn(size);
+            auto const datagrams = encodeTxn(slice(txn), txid);
+            auto const count =
+                (size + maxFragmentPayload - 1) / maxFragmentPayload;
+            BEAST_EXPECT(datagrams.size() == count);
+
+            std::size_t total = 0;
+            for (std::size_t i = 0; i < datagrams.size(); ++i)
+            {
+                BEAST_EXPECT(datagrams[i].size() <= maxDatagramSize);
+                auto const fragment = decodeTxnFragment(slice(datagrams[i]));
+                if (!BEAST_EXPECT(fragment))
+                    continue;
+                BEAST_EXPECT(fragment->index == i);
+                BEAST_EXPECT(fragment->count == count);
+                BEAST_EXPECT(fragment->totalSize == size);
+                BEAST_EXPECT(
+                    fragment->payload ==
+                    Slice(
+                        txn.data() + i * maxFragmentPayload,
+                        fragment->payload.size()));
+                total += fragment->payload.size();
+            }
+            BEAST_EXPECT(total == size);
+        }
+
+        // Limits
+        BEAST_EXPECT(encodeTxn(Slice{}, txid).empty());
+        BEAST_EXPECT(encodeTxn(slice(Buffer(maxTxnSize + 1)), txid).empty());
+        {
+            auto const datagrams = encodeTxn(slice(Buffer(maxTxnSize)), txid);
+            BEAST_EXPECT(!datagrams.empty());
+            BEAST_EXPECT(datagrams.size() <= maxFragments);
+        }
+    }
+
+    void
+    testTxnDecoding()
+    {
+        testcase("XUSHTXNF decoding");
+        using namespace xush;
+
+        uint256 const id{7};
+        auto decode = [](Buffer const& b) {
+            return decodeTxnFragment(slice(b)).has_value();
+        };
+
+        BEAST_EXPECT(decode(rawFragment(id, 10, 2, 1, 5)));
+        BEAST_EXPECT(decode(rawFragment(id, 10, 1, 0, 10)));
+        BEAST_EXPECT(decode(rawFragment(id, maxTxnSize, maxFragments, 0, 1)));
+
+        // No payload
+        BEAST_EXPECT(!decode(rawFragment(id, 10, 1, 0, 0)));
+        // Bad total size
+        BEAST_EXPECT(!decode(rawFragment(id, 0, 1, 0, 1)));
+        BEAST_EXPECT(!decode(rawFragment(id, maxTxnSize + 1, 2, 0, 1)));
+        // Bad fragment count
+        BEAST_EXPECT(!decode(rawFragment(id, 10, 0, 0, 1)));
+        BEAST_EXPECT(!decode(rawFragment(id, 10, 11, 0, 1)));
+        BEAST_EXPECT(
+            !decode(rawFragment(id, maxTxnSize, maxFragments + 1, 0, 1)));
+        // Bad index
+        BEAST_EXPECT(!decode(rawFragment(id, 10, 2, 2, 1)));
+        // Payload larger than the transaction
+        BEAST_EXPECT(!decode(rawFragment(id, 10, 2, 0, 11)));
+        // Wrong tag
+        BEAST_EXPECT(!decodeTxnFragment(slice(encodePeers({}))));
+    }
+
+    void
+    testReassembly()
+    {
+        testcase("reassembly");
+        using namespace xush;
+        using namespace std::chrono_literals;
+
+        auto const now = Reassembler::clock_type::now();
+        auto const alice = ep("1.2.3.4:5000");
+        auto const bob = ep("5.6.7.8:5000");
+        auto const carol = ep("9.9.9.9:5000");
+
+        uint256 const id1{1};
+        uint256 const id2{2};
+
+        auto const txn1 = makeTxn(2 * maxFragmentPayload + 100);
+        auto const txn2 = makeTxn(2 * maxFragmentPayload + 200);
+        auto const d1 = encodeTxn(slice(txn1), id1);
+        auto const d2 = encodeTxn(slice(txn2), id2);
+        BEAST_EXPECT(d1.size() == 3 && d2.size() == 3);
+
+        auto f1 = [&](std::size_t i) {
+            return *decodeTxnFragment(slice(d1[i]));
+        };
+        auto f2 = [&](std::size_t i) {
+            return *decodeTxnFragment(slice(d2[i]));
+        };
+
+        // Out of order, with a repeated fragment
+        {
+            Reassembler r;
+            BEAST_EXPECT(r.add(alice, f1(2), now).status == Status::pending);
+            BEAST_EXPECT(r.add(alice, f1(0), now).status == Status::pending);
+            BEAST_EXPECT(r.add(alice, f1(0), now).status == Status::pending);
+            BEAST_EXPECT(r.pending() == 1);
+
+            auto const result = r.add(alice, f1(1), now);
+            BEAST_EXPECT(result.status == Status::complete);
+            BEAST_EXPECT(result.txn == txn1);
+            BEAST_EXPECT(r.pending() == 0);
+            BEAST_EXPECT(r.bytes() == 0);
+        }
+
+        // A transaction in a single datagram needs no state
+        {
+            Reassembler r;
+            auto const small = makeTxn(100);
+            auto const d = encodeTxn(slice(small), id2);
+            auto const result =
+                r.add(alice, *decodeTxnFragment(slice(d[0])), now);
+            BEAST_EXPECT(result.status == Status::complete);
+            BEAST_EXPECT(result.txn == small);
+            BEAST_EXPECT(r.pending() == 0);
+
+            // ... but must be consistent
+            auto bad = *decodeTxnFragment(slice(d[0]));
+            bad.totalSize += 1;
+            BEAST_EXPECT(r.add(alice, bad, now).status == Status::invalid);
+        }
+
+        // Senders are kept apart
+        {
+            Reassembler r;
+            BEAST_EXPECT(r.add(alice, f1(0), now).status == Status::pending);
+            BEAST_EXPECT(r.add(alice, f1(1), now).status == Status::pending);
+            BEAST_EXPECT(r.add(bob, f1(2), now).status == Status::pending);
+            BEAST_EXPECT(r.pending() == 2);
+            BEAST_EXPECT(r.add(alice, f1(2), now).status == Status::complete);
+            BEAST_EXPECT(r.pending() == 1);
+        }
+
+        // A header that disagrees with earlier fragments
+        {
+            Reassembler r;
+            BEAST_EXPECT(r.add(alice, f1(0), now).status == Status::pending);
+            auto bad = f1(1);
+            bad.count += 1;
+            BEAST_EXPECT(r.add(alice, bad, now).status == Status::invalid);
+            BEAST_EXPECT(r.pending() == 0);
+            BEAST_EXPECT(r.bytes() == 0);
+        }
+
+        // More bytes than the advertised total size
+        {
+            Reassembler r;
+            auto first = f1(0);
+            first.count = 2;
+            first.totalSize = maxFragmentPayload + 10;
+            auto second = f1(1);
+            second.count = 2;
+            second.totalSize = first.totalSize;
+            BEAST_EXPECT(r.add(alice, first, now).status == Status::pending);
+            BEAST_EXPECT(r.add(alice, second, now).status == Status::invalid);
+            BEAST_EXPECT(r.pending() == 0);
+        }
+
+        // Every fragment present but fewer bytes than the total size
+        {
+            Reassembler r;
+            auto first = f1(0);
+            first.count = 2;
+            first.totalSize = 2 * maxFragmentPayload + 5;
+            auto second = f1(1);
+            second.count = 2;
+            second.totalSize = first.totalSize;
+            BEAST_EXPECT(r.add(alice, first, now).status == Status::pending);
+            BEAST_EXPECT(r.add(alice, second, now).status == Status::invalid);
+            BEAST_EXPECT(r.pending() == 0);
+        }
+
+        // Partial transactions time out
+        {
+            Reassembler r(
+                Reassembler::defaultMaxPending,
+                Reassembler::defaultMaxBytes,
+                30s);
+            BEAST_EXPECT(r.add(alice, f1(0), now).status == Status::pending);
+            BEAST_EXPECT(r.add(alice, f1(1), now).status == Status::pending);
+
+            // Not yet
+            BEAST_EXPECT(
+                r.add(bob, f2(0), now + 30s).status == Status::pending);
+            BEAST_EXPECT(r.pending() == 2);
+
+            // Now: the first transaction has to start over
+            BEAST_EXPECT(
+                r.add(bob, f2(1), now + 31s).status == Status::pending);
+            BEAST_EXPECT(r.pending() == 1);
+            BEAST_EXPECT(
+                r.add(alice, f1(2), now + 31s).status == Status::pending);
+            BEAST_EXPECT(r.pending() == 2);
+        }
+
+        // The number of partial transactions is limited
+        {
+            Reassembler r(2, Reassembler::defaultMaxBytes, 30s);
+            BEAST_EXPECT(r.add(alice, f1(0), now).status == Status::pending);
+            BEAST_EXPECT(r.add(bob, f1(0), now).status == Status::pending);
+            BEAST_EXPECT(r.add(carol, f1(0), now).status == Status::overloaded);
+            BEAST_EXPECT(r.pending() == 2);
+
+            // Fragments of transactions already being rebuilt are accepted
+            BEAST_EXPECT(r.add(alice, f1(1), now).status == Status::pending);
+            BEAST_EXPECT(r.add(alice, f1(2), now).status == Status::complete);
+            BEAST_EXPECT(r.add(carol, f1(0), now).status == Status::pending);
+        }
+
+        // The number of bytes buffered is limited
+        {
+            Reassembler r(
+                Reassembler::defaultMaxPending,
+                maxFragmentPayload + Reassembler::fragmentOverhead,
+                30s);
+            BEAST_EXPECT(r.add(alice, f1(0), now).status == Status::pending);
+            BEAST_EXPECT(r.add(alice, f1(1), now).status == Status::overloaded);
+            BEAST_EXPECT(r.add(bob, f1(0), now).status == Status::overloaded);
+            BEAST_EXPECT(r.pending() == 1);
+            BEAST_EXPECT(
+                r.bytes() ==
+                maxFragmentPayload + Reassembler::fragmentOverhead);
+        }
+    }
+
+public:
+    void
+    run() override
+    {
+        testClassify();
+        testPeers();
+        testTxnEncoding();
+        testTxnDecoding();
+        testReassembly();
+    }
+};
+
+BEAST_DEFINE_TESTSUITE(XUSH, overlay, ripple);
+
+}  // namespace test
+}  // namespace ripple

src/test/peerfinder/HighwayPeers_test.cpp

@@ -0,0 +1,276 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2026 XRPL Labs
+
+    Permission to use, copy, modify, and/or distribute this software for any
+    purpose  with  or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#include <test/unit_test/SuiteJournal.h>
+#include <xrpld/peerfinder/PeerfinderManager.h>
+#include <xrpld/peerfinder/detail/Tuning.h>
+#include <xrpld/peerfinder/make_Manager.h>
+#include <xrpl/basics/BasicConfig.h>
+#include <xrpl/basics/chrono.h>
+#include <xrpl/beast/insight/NullCollector.h>
+#include <xrpl/beast/net/IPAddressConversion.h>
+#include <xrpl/beast/unit_test.h>
+
+#include <boost/asio/io_context.hpp>
+
+#include <memory>
+#include <set>
+#include <string>
+#include <vector>
+
+namespace ripple {
+namespace PeerFinder {
+
+// Tests the UDP Superhighway (XUSH) endpoint set kept by PeerFinder::Manager
+class HighwayPeers_test : public beast::unit_test::suite
+{
+    test::SuiteJournal journal_{"HighwayPeers_test", *this};
+
+    struct Fixture
+    {
+        boost::asio::io_context io;
+        TestStopwatch clock;
+        BasicConfig basicConfig;
+        std::unique_ptr<Manager> manager;
+
+        Fixture(beast::Journal journal, bool enabled, bool peerPrivate)
+            : manager(make_Manager(
+                  io,
+                  clock,
+                  journal,
+                  basicConfig,
+                  beast::insight::NullCollector::New(),
+                  false))
+        {
+            Config config;
+            config.udpHighway = enabled;
+            config.peerPrivate = peerPrivate;
+            manager->setConfig(config);
+        }
+
+        ~Fixture()
+        {
+            manager->stop();
+        }
+    };
+
+    static beast::IP::Endpoint
+    ep(std::string const& s)
+    {
+        return beast::IP::Endpoint::from_string(s);
+    }
+
+    // A distinct public IPv4 endpoint
+    static beast::IP::Endpoint
+    publicEp(std::uint32_t i)
+    {
+        return beast::IP::Endpoint(
+            boost::asio::ip::address_v4(0x2C000000 + i), 51235);
+    }
+
+    static std::set<beast::IP::Endpoint>
+    asSet(std::vector<beast::IP::Endpoint> const& v)
+    {
+        return {v.begin(), v.end()};
+    }
+
+    void
+    testDisabled()
+    {
+        testcase("disabled");
+        Fixture f(journal_, false, false);
+
+        f.manager->addFixedPeer("fixed", {ep("10.0.0.1:51235")});
+        f.manager->add_highway_peers({ep("8.8.8.8:51235")});
+        f.manager->onRedirects(
+            beast::IP::to_asio_endpoint(ep("8.8.4.4:51235")),
+            {beast::IP::to_asio_endpoint(ep("1.1.1.1:51235"))});
+
+        BEAST_EXPECT(f.manager->highway_targets(100).empty());
+        BEAST_EXPECT(f.manager->highway_adverts(100).empty());
+    }
+
+    void
+    testLearning()
+    {
+        testcase("learning");
+        Fixture f(journal_, true, false);
+
+        // Only valid public endpoints are learned from other servers
+        f.manager->add_highway_peers(
+            {ep("8.8.8.8:51235"),
+             ep("10.0.0.1:51235"),
+             ep("127.0.0.1:51235"),
+             ep("9.9.9.9:0"),
+             ep("0.0.0.0:51235")});
+        BEAST_EXPECT(
+            asSet(f.manager->highway_targets(100)) ==
+            std::set<beast::IP::Endpoint>{ep("8.8.8.8:51235")});
+
+        // Fixed peers are trusted, even with a private address, but they
+        // are never advertised
+        f.manager->addFixedPeer("fixed", {ep("10.0.0.2:51235")});
+        BEAST_EXPECT(
+            asSet(f.manager->highway_targets(100)) ==
+            (std::set<beast::IP::Endpoint>{
+                ep("8.8.8.8:51235"), ep("10.0.0.2:51235")}));
+        BEAST_EXPECT(
+            asSet(f.manager->highway_adverts(100)) ==
+            std::set<beast::IP::Endpoint>{ep("8.8.8.8:51235")});
+
+        // A learned endpoint that is later configured as fixed stops being
+        // advertised
+        f.manager->addFixedPeer("fixed2", {ep("8.8.8.8:51235")});
+        BEAST_EXPECT(f.manager->highway_adverts(100).empty());
+        BEAST_EXPECT(f.manager->highway_targets(100).size() == 2);
+
+        // Redirects are learned too
+        f.manager->onRedirects(
+            beast::IP::to_asio_endpoint(ep("8.8.4.4:51235")),
+            {beast::IP::to_asio_endpoint(ep("1.1.1.1:51235"))});
+        BEAST_EXPECT(f.manager->highway_targets(100).size() == 3);
+    }
+
+    void
+    testSampling()
+    {
+        testcase("sampling");
+        Fixture f(journal_, true, false);
+
+        std::vector<beast::IP::Endpoint> endpoints;
+        for (std::uint32_t i = 0; i < 100; ++i)
+            endpoints.push_back(publicEp(i));
+        f.manager->add_highway_peers(endpoints);
+
+        for (int i = 0; i < 10; ++i)
+        {
+            auto const targets = f.manager->highway_targets(16);
+            BEAST_EXPECT(targets.size() == 16);
+            BEAST_EXPECT(asSet(targets).size() == 16);
+        }
+
+        BEAST_EXPECT(
+            asSet(f.manager->highway_targets(1000)) == asSet(endpoints));
+        BEAST_EXPECT(f.manager->highway_targets(0).empty());
+    }
+
+    void
+    testPrivate()
+    {
+        testcase("private");
+        Fixture f(journal_, true, true);
+
+        f.manager->addFixedPeer("fixed", {ep("10.0.0.2:51235")});
+        f.manager->add_highway_peers({ep("8.8.8.8:51235")});
+
+        // A private server only talks to its fixed peers, and advertises
+        // nothing
+        BEAST_EXPECT(
+            asSet(f.manager->highway_targets(100)) ==
+            std::set<beast::IP::Endpoint>{ep("10.0.0.2:51235")});
+        BEAST_EXPECT(f.manager->highway_adverts(100).empty());
+    }
+
+    void
+    testExpiry()
+    {
+        testcase("expiry");
+        using namespace std::chrono_literals;
+        Fixture f(journal_, true, false);
+
+        auto const ttl = Tuning::highwayPeerSecondsToLive;
+        auto const advertised = ep("8.8.8.8:51235");
+        auto const redirected = ep("1.1.1.1:51235");
+        auto const fixed = ep("10.0.0.2:51235");
+
+        f.manager->addFixedPeer("fixed", {fixed});
+        f.manager->add_highway_peers({advertised});
+        f.manager->add_highway_peers({redirected});
+
+        // Advertisements don't extend the life of an entry, but
+        // authenticated sources (here, a redirect) do
+        f.clock.advance(ttl / 2);
+        f.manager->add_highway_peers({advertised, redirected});
+        f.manager->onRedirects(
+            beast::IP::to_asio_endpoint(ep("8.8.4.4:51235")),
+            {beast::IP::to_asio_endpoint(redirected)});
+
+        f.clock.advance(ttl / 2);
+        f.manager->once_per_second();
+        BEAST_EXPECT(f.manager->highway_targets(100).size() == 3);
+
+        f.clock.advance(1s);
+        f.manager->once_per_second();
+        BEAST_EXPECT(
+            asSet(f.manager->highway_targets(100)) ==
+            (std::set<beast::IP::Endpoint>{redirected, fixed}));
+
+        f.clock.advance(ttl / 2);
+        f.manager->once_per_second();
+        BEAST_EXPECT(
+            asSet(f.manager->highway_targets(100)) ==
+            std::set<beast::IP::Endpoint>{fixed});
+
+        // An expired entry can be learned again
+        f.manager->add_highway_peers({advertised});
+        BEAST_EXPECT(f.manager->highway_targets(100).size() == 2);
+    }
+
+    void
+    testLimit()
+    {
+        testcase("limit");
+        Fixture f(journal_, true, false);
+
+        auto const max = Tuning::highwayPeersMax;
+
+        std::vector<beast::IP::Endpoint> endpoints;
+        for (std::uint32_t i = 0; i < max + 100; ++i)
+            endpoints.push_back(publicEp(i));
+        f.manager->add_highway_peers(endpoints);
+        BEAST_EXPECT(f.manager->highway_targets(max + 100).size() == max);
+
+        // Fixed peers don't count towards the limit
+        f.manager->addFixedPeer("fixed", {ep("10.0.0.2:51235")});
+        BEAST_EXPECT(f.manager->highway_targets(max + 100).size() == max + 1);
+
+        // Disabling the highway forgets everything
+        Config config;
+        config.udpHighway = false;
+        f.manager->setConfig(config);
+        BEAST_EXPECT(f.manager->highway_targets(max + 100).empty());
+    }
+
+public:
+    void
+    run() override
+    {
+        testDisabled();
+        testLearning();
+        testSampling();
+        testPrivate();
+        testExpiry();
+        testLimit();
+    }
+};
+
+BEAST_DEFINE_TESTSUITE(HighwayPeers, peerfinder, ripple);
+
+}  // namespace PeerFinder
+}  // namespace ripple

src/test/server/Server_test.cpp

@@ -36,6 +36,8 @@
 #include <boost/utility/in_place_factory.hpp>
 
 #include <chrono>
+#include <condition_variable>
+#include <mutex>
 #include <optional>
 #include <stdexcept>
 #include <thread>
@@ -158,6 +160,7 @@ public:
         onUDPMessage(
             std::string const& message,
             boost::asio::ip::tcp::endpoint const& remoteEndpoint,
+            Port const& port,
             std::function<void(std::string const&)> sendResponse)
         {
         }
@@ -371,6 +374,7 @@ public:
             onUDPMessage(
                 std::string const& message,
                 boost::asio::ip::tcp::endpoint const& remoteEndpoint,
+                Port const& port,
                 std::function<void(std::string const&)> sendResponse)
             {
             }
@@ -404,6 +408,96 @@ public:
         pass();
     }
 
+    // Records the datagrams delivered to onUDPMessage
+    struct UDPHandler : TestHandler
+    {
+        std::mutex mutex;
+        std::condition_variable cv;
+        std::vector<std::pair<std::string, bool>> received;
+
+        void
+        onUDPMessage(
+            std::string const& message,
+            boost::asio::ip::tcp::endpoint const& remoteEndpoint,
+            Port const& port,
+            std::function<void(std::string const&)> sendResponse)
+        {
+            std::lock_guard lock(mutex);
+            received.emplace_back(message, port.has_peer());
+            cv.notify_all();
+        }
+
+        bool
+        wait(std::size_t count, std::chrono::milliseconds timeout)
+        {
+            std::unique_lock lock(mutex);
+            return cv.wait_for(
+                lock, timeout, [&] { return received.size() >= count; });
+        }
+    };
+
+    // A peer port with udp_highway set also accepts UDP datagrams on the
+    // same port number; one without it does not.
+    void
+    udpHighwayTests()
+    {
+        testcase("UDP Superhighway door");
+        using namespace std::chrono_literals;
+        using udp = boost::asio::ip::udp;
+
+        TestSink sink{*this};
+        TestThread thread;
+        sink.threshold(beast::severities::Severity::kAll);
+        beast::Journal journal{sink};
+
+        auto const addr = boost::asio::ip::make_address(getEnvLocalhostAddr());
+
+        auto run = [&](bool highway) {
+            UDPHandler handler;
+            auto s = make_Server(handler, thread.get_io_context(), journal);
+
+            std::vector<Port> ports(1);
+            ports.back().name = "peer";
+            ports.back().ip = addr;
+            ports.back().port = 0;
+            ports.back().protocol.insert("peer");
+            ports.back().udp_highway = highway;
+
+            auto const eps = s->ports(ports);
+            if (!BEAST_EXPECT(eps.size() == 1))
+                return;
+
+            // The TCP door picked the port number; the UDP door shares it.
+            auto const port = eps.begin()->second.port();
+            BEAST_EXPECT(port != 0);
+
+            boost::asio::io_context io;
+            udp::socket client(io, udp::endpoint(addr, 0));
+            std::string const message = "XUSHTEST";
+            client.send_to(
+                boost::asio::buffer(message), udp::endpoint(addr, port));
+
+            if (highway)
+            {
+                if (BEAST_EXPECT(handler.wait(1, 5s)))
+                {
+                    std::lock_guard lock(handler.mutex);
+                    BEAST_EXPECT(handler.received.front().first == message);
+                    BEAST_EXPECT(handler.received.front().second);
+                }
+            }
+            else
+            {
+                BEAST_EXPECT(!handler.wait(1, 200ms));
+            }
+
+            s = nullptr;
+        };
+
+        run(true);
+        run(false);
+    }
+
     void
     testBadConfig()
     {
@@ -547,6 +641,7 @@ public:
     {
         basicTests();
         stressTest();
+        udpHighwayTests();
         testBadConfig();
     }
 };

src/xrpld/app/misc/NetworkOPs.cpp

@@ -1725,6 +1725,12 @@ NetworkOPsImp::apply(std::unique_lock<std::mutex>& batchLock)
                     tx.set_deferred(e.result == terQUEUED);
                     // FIXME: This should be when we received it
                     app_.overlay().relay(e.transaction->getID(), tx, *toSkip);
+
+                    // Also fan it out over the UDP Superhighway, if enabled.
+                    // shouldRelay() above ensures we do this at most once.
+                    app_.overlay().publishTxXUSH(
+                        s.slice(), e.transaction->getID());
+
                     e.transaction->setBroadcast();
                 }
             }

src/xrpld/core/Config.h

@@ -197,6 +197,11 @@ public:
 
     // True to ask peers not to relay current IP.
     bool PEER_PRIVATE = false;
+
+    // True to enable the Xahau UDP Superhighway (XUSH): a best-effort UDP
+    // transport on the peer port used to fan transactions out to other
+    // highway-enabled servers in addition to the normal peer relay.
+    bool UDP_HIGHWAY = false;
     // peers_max is a legacy configuration, which is going to be replaced
     // with individual inbound peers peers_in_max and outbound peers
     // peers_out_max configuration. for now we support both the legacy and

src/xrpld/core/ConfigSections.h

@@ -72,6 +72,7 @@ struct ConfigSection
 #define SECTION_PATH_SEARCH_FAST "path_search_fast"
 #define SECTION_PATH_SEARCH_MAX "path_search_max"
 #define SECTION_PEER_PRIVATE "peer_private"
+#define SECTION_UDP_HIGHWAY "udp_highway"
 #define SECTION_PEERS_MAX "peers_max"
 #define SECTION_PEERS_IN_MAX "peers_in_max"
 #define SECTION_PEERS_OUT_MAX "peers_out_max"

src/xrpld/core/detail/Config.cpp

@@ -533,6 +533,9 @@ Config::loadFromString(std::string const& fileContents)
     if (getSingleSection(secConfig, SECTION_PEER_PRIVATE, strTemp, j_))
         PEER_PRIVATE = beast::lexicalCastThrow<bool>(strTemp);
 
+    if (getSingleSection(secConfig, SECTION_UDP_HIGHWAY, strTemp, j_))
+        UDP_HIGHWAY = beast::lexicalCastThrow<bool>(strTemp);
+
     if (getSingleSection(secConfig, SECTION_PEERS_MAX, strTemp, j_))
     {
         PEERS_MAX = beast::lexicalCastThrow<std::size_t>(strTemp);

src/xrpld/overlay/Overlay.h

@@ -235,6 +235,24 @@ public:
      */
     virtual Json::Value
     txMetrics() const = 0;
+
+    /** Process a datagram received on the UDP Superhighway (XUSH).
+
+        Datagrams arrive on the peer port. They are unauthenticated and are
+        accounted for against the sender's IP address.
+    */
+    virtual void
+    processXUSH(
+        std::string const& message,
+        boost::asio::ip::tcp::endpoint const& remoteEndpoint) = 0;
+
+    /** Send a serialized transaction to UDP Superhighway (XUSH) peers.
+
+        Does nothing unless the highway is enabled. Callers must relay each
+        transaction at most once (see HashRouter::shouldRelay).
+    */
+    virtual void
+    publishTxXUSH(Slice const& tx, uint256 const& txid) = 0;
 };
 
 }  // namespace ripple

src/xrpld/overlay/detail/OverlayImpl.cpp

@@ -21,10 +21,12 @@
 #include <xrpld/app/ledger/OpenLedger.h>
 #include <xrpld/app/misc/HashRouter.h>
 #include <xrpld/app/misc/NetworkOPs.h>
+#include <xrpld/app/misc/Transaction.h>
 #include <xrpld/app/misc/ValidatorList.h>
 #include <xrpld/app/misc/ValidatorSite.h>
 #include <xrpld/app/rdb/RelationalDatabase.h>
 #include <xrpld/app/rdb/Wallet.h>
+#include <xrpld/app/tx/apply.h>
 #include <xrpld/overlay/Cluster.h>
 #include <xrpld/overlay/detail/ConnectAttempt.h>
 #include <xrpld/overlay/detail/PeerImp.h>
@@ -36,8 +38,10 @@
 #include <xrpl/basics/make_SSLContext.h>
 #include <xrpl/basics/random.h>
 #include <xrpl/beast/core/LexicalCast.h>
+#include <xrpl/beast/net/IPAddressConversion.h>
 #include <xrpl/protocol/Feature.h>
 #include <xrpl/protocol/STTx.h>
+#include <xrpl/resource/Fees.h>
 #include <xrpl/server/SimpleWriter.h>
 
 #include <xrpld/core/ConfigSections.h>
@@ -107,6 +111,7 @@ OverlayImpl::Timer::on_timer(error_code ec)
 
     overlay_.m_peerFinder->once_per_second();
     overlay_.sendEndpoints();
+    overlay_.sendXUSHPeers();
     overlay_.autoConnect();
     if (overlay_.app_.config().TX_REDUCE_RELAY_ENABLE)
         overlay_.sendTxQueue();
@@ -150,6 +155,8 @@ OverlayImpl::OverlayImpl(
     , next_id_(1)
     , timer_count_(0)
     , slots_(app.logs(), *this)
+    , xushSocket4_(io_context_)
+    , xushSocket6_(io_context_)
     , m_stats(
           std::bind(&OverlayImpl::collect_metrics, this),
           collector,
@@ -495,6 +502,21 @@ OverlayImpl::start()
     m_peerFinder->setConfig(config);
     m_peerFinder->start();
 
+    if (app_.config().UDP_HIGHWAY)
+    {
+        openXUSH();
+
+        // A private server only exchanges highway traffic with its fixed
+        // peers, so without any the highway is enabled but idle.
+        if (config.peerPrivate && app_.config().IPS_FIXED.empty())
+        {
+            JLOG(journal_.warn())
+                << "XUSH: [udp_highway] is enabled, but this server is "
+                   "private (peer_private or validator) and has no "
+                   "[ips_fixed] peers, so no highway traffic will be sent";
+        }
+    }
+
     auto addIps = [this](std::vector<std::string> ips, bool fixed) {
         beast::Journal const& j = app_.journal("Overlay");
         for (auto& ip : ips)
@@ -567,6 +589,7 @@ OverlayImpl::stop()
         cond_.wait(lock, [this] { return list_.empty(); });
     }
     m_peerFinder->stop();
+    closeXUSH();
 }
 
 //------------------------------------------------------------------------------
@@ -1492,6 +1515,372 @@ OverlayImpl::deleteIdlePeers()
     slots_.deleteIdlePeers();
 }
 
+//------------------------------------------------------------------------------
+//
+// XUSH (Xahau UDP Superhighway)
+//
+//------------------------------------------------------------------------------
+
+namespace {
+
+// Called from the JobQueue. Performs the expensive checks and then hands the
+// transaction to NetworkOPs which, if it applies, relays it over both the peer
+// protocol and the highway.
+void
+checkXUSHTransaction(
+    Application& app,
+    beast::Journal j,
+    std::shared_ptr<STTx const> const& stx,
+    Resource::Consumer usage)
+{
+    auto const txID = stx->getTransactionID();
+
+    try
+    {
+        if (stx->isFieldPresent(sfLastLedgerSequence) &&
+            (stx->getFieldU32(sfLastLedgerSequence) <
+             app.getLedgerMaster().getValidLedgerIndex()))
+        {
+            app.getHashRouter().setFlags(txID, SF_BAD);
+            usage.charge(Resource::feeUselessData, "XUSH expired tx");
+            return;
+        }
+
+        // A manifest-authorized SetManifest's Fee follows the voted base fee,
+        // so a miss may be a fee setting this server has not reached yet.
+        // Every ingress path runs this before checkValidity(). Drop it, but
+        // do not mark it bad: the same bytes can become canonical here.
+        if (auto const [fee, feeReason] = checkManifestIngressFee(app, *stx);
+            fee == ManifestIngressFee::Refused)
+        {
+            JLOG(j.trace()) << "XUSH: dropping manifest-authorized tx " << txID
+                            << ": " << feeReason;
+            usage.charge(Resource::feeUselessData, "XUSH manifest fee miss");
+            return;
+        }
+
+        if (auto const [validity, reason] = checkValidity(
+                app.getHashRouter(),
+                *stx,
+                app.getLedgerMaster().getValidatedRules(),
+                app.config());
+            validity != Validity::Valid)
+        {
+            JLOG(j.trace())
+                << "XUSH: transaction " << txID << " failed checks: " << reason;
+            app.getHashRouter().setFlags(txID, SF_BAD);
+            usage.charge(Resource::feeInvalidSignature, "XUSH tx signature");
+            return;
+        }
+
+        std::string why;
+        auto tx = std::make_shared<Transaction>(stx, why, app);
+        if (tx->getStatus() == INVALID)
+        {
+            JLOG(j.trace())
+                << "XUSH: transaction " << txID << " is invalid: " << why;
+            app.getHashRouter().setFlags(txID, SF_BAD);
+            usage.charge(Resource::feeInvalidSignature, "XUSH tx (impossible)");
+            return;
+        }
+
+        app.getOPs().processTransaction(
+            tx, false, false, NetworkOPs::FailHard::no);
+    }
+    catch (std::exception const& ex)
+    {
+        JLOG(j.warn()) << "XUSH: exception checking transaction " << txID
+                       << ": " << ex.what();
+        app.getHashRouter().setFlags(txID, SF_BAD);
+        usage.charge(Resource::feeInvalidData, "XUSH tx exception");
+    }
+}
+
+}  // namespace
+
+void
+OverlayImpl::processXUSH(
+    std::string const& message,
+    boost::asio::ip::tcp::endpoint const& remoteEndpoint)
+{
+    if (!app_.config().UDP_HIGHWAY)
+        return;
+
+    auto const remote = beast::IPAddressConversion::from_asio(remoteEndpoint);
+
+    // Datagrams are unauthenticated, so they are accounted for against the
+    // sender's IP address exactly like an inbound peer connection.
+    auto usage = m_resourceManager.newInboundEndpoint(remote);
+    if (usage.disconnect(journal_))
+        return;
+    usage.charge(Resource::feeTrivialPeer, "XUSH datagram");
+
+    Slice const datagram(message.data(), message.size());
+
+    switch (xush::classify(datagram))
+    {
+        case xush::MessageType::peers:
+            onXUSHPeers(datagram, remote, usage);
+            break;
+
+        case xush::MessageType::txn:
+            onXUSHTxnFragment(datagram, remote, usage);
+            break;
+
+        case xush::MessageType::unknown:
+            // Ignored, so that new datagram types can be added later.
+            JLOG(journal_.trace()) << "XUSH: unknown datagram from " << remote;
+            break;
+    }
+}
+
+void
+OverlayImpl::onXUSHPeers(
+    Slice datagram,
+    beast::IP::Endpoint const& remote,
+    Resource::Consumer& usage)
+{
+    auto const endpoints = xush::decodePeers(datagram);
+    if (!endpoints)
+    {
+        usage.charge(Resource::feeMalformedRequest, "XUSHPEER malformed");
+        return;
+    }
+
+    JLOG(journal_.trace()) << "XUSHPEER from " << remote << " with "
+                           << endpoints->size() << " endpoints";
+
+    m_peerFinder->add_highway_peers(*endpoints);
+}
+
+void
+OverlayImpl::onXUSHTxnFragment(
+    Slice datagram,
+    beast::IP::Endpoint const& remote,
+    Resource::Consumer& usage)
+{
+    auto const fragment = xush::decodeTxnFragment(datagram);
+    if (!fragment)
+    {
+        usage.charge(Resource::feeMalformedRequest, "XUSHTXNF malformed");
+        return;
+    }
+
+    auto result = [&] {
+        std::lock_guard lock(xushMutex_);
+        return xushReassembler_.add(
+            remote, *fragment, xush::Reassembler::clock_type::now());
+    }();
+
+    switch (result.status)
+    {
+        case xush::Reassembler::Status::pending:
+            break;
+
+        case xush::Reassembler::Status::overloaded:
+            JLOG(journal_.debug())
+                << "XUSHTXNF: reassembly limit reached, dropped fragment from "
+                << remote;
+            break;
+
+        case xush::Reassembler::Status::invalid:
+            usage.charge(Resource::feeInvalidData, "XUSHTXNF inconsistent");
+            break;
+
+        case xush::Reassembler::Status::complete:
+            onXUSHTransaction(result.txn, fragment->txid, usage);
+            break;
+    }
+}
+
+void
+OverlayImpl::onXUSHTransaction(
+    Buffer const& blob,
+    uint256 const& txid,
+    Resource::Consumer& usage)
+{
+    // If we've never been in sync, there's nothing we can do with it
+    if (app_.getOPs().isNeedNetworkLedger())
+        return;
+
+    std::shared_ptr<STTx const> stx;
+    try
+    {
+        SerialIter sit(Slice(blob.data(), blob.size()));
+        stx = std::make_shared<STTx const>(sit);
+    }
+    catch (std::exception const& ex)
+    {
+        JLOG(journal_.debug()) << "XUSH: malformed transaction: " << ex.what();
+        usage.charge(Resource::feeInvalidData, "XUSH malformed tx");
+        return;
+    }
+
+    auto const txID = stx->getTransactionID();
+
+    // The ID in the header can't be used for anything (in particular, not to
+    // suppress a transaction) until it's known to match the content.
+    if (txID != txid)
+    {
+        usage.charge(Resource::feeInvalidData, "XUSH tx ID mismatch");
+        return;
+    }
+
+    // Emitted transactions are never relayed, and pseudo-transactions are
+    // never submitted by servers.
+    if (stx->isFieldPresent(sfEmitDetails) || isPseudoTx(*stx))
+    {
+        usage.charge(Resource::feeHeavyBurdenPeer, "XUSH emitted/pseudo tx");
+        return;
+    }
+
+    int flags;
+    if (!app_.getHashRouter().shouldProcess(
+            txID, 0, flags, std::chrono::seconds(10)))
+    {
+        // We have seen this transaction recently
+        if (flags & SF_BAD)
+            usage.charge(Resource::feeUselessData, "XUSH known bad tx");
+        return;
+    }
+
+    if (app_.getLedgerMaster().getValidatedLedgerAge() >
+        std::chrono::minutes(4))
+    {
+        JLOG(journal_.trace()) << "XUSH: no new transactions until synced";
+        return;
+    }
+
+    if (app_.getJobQueue().getJobCount(jtTRANSACTION) >
+        app_.config().MAX_TRANSACTIONS)
+    {
+        incJqTransOverflow();
+        JLOG(journal_.info()) << "XUSH: transaction queue is full";
+        return;
+    }
+
+    app_.getJobQueue().addJob(
+        jtTRANSACTION,
+        "XUSH->checkTransaction",
+        [&app = app_, j = journal_, stx, usage]() {
+            checkXUSHTransaction(app, j, stx, usage);
+        });
+}
+
+void
+OverlayImpl::publishTxXUSH(Slice const& tx, uint256 const& txid)
+{
+    if (!app_.config().UDP_HIGHWAY)
+        return;
+
+    auto const targets = m_peerFinder->highway_targets(xush::txnFanout);
+    if (targets.empty())
+        return;
+
+    auto const datagrams = xush::encodeTxn(tx, txid);
+    if (datagrams.empty())
+    {
+        JLOG(journal_.debug())
+            << "XUSH: transaction " << txid << " is too large for the highway";
+        return;
+    }
+
+    sendXUSH(datagrams, targets);
+}
+
+void
+OverlayImpl::sendXUSHPeers()
+{
+    if (!app_.config().UDP_HIGHWAY)
+        return;
+
+    auto const adverts =
+        m_peerFinder->highway_adverts(xush::maxAdvertisedPeers);
+    if (adverts.empty())
+        return;
+
+    auto const targets = m_peerFinder->highway_targets(xush::advertFanout);
+    if (targets.empty())
+        return;
+
+    sendXUSH({xush::encodePeers(adverts)}, targets);
+}
+
+void
+OverlayImpl::sendXUSH(
+    std::vector<Buffer> const& datagrams,
+    std::vector<beast::IP::Endpoint> const& targets)
+{
+    std::lock_guard lock(xushSendMutex_);
+
+    for (auto const& target : targets)
+    {
+        auto& socket = target.address().is_v4() ? xushSocket4_ : xushSocket6_;
+        if (!socket.is_open())
+            continue;
+
+        boost::asio::ip::udp::endpoint const to(
+            target.address(), target.port());
+
+        for (auto const& datagram : datagrams)
+        {
+            boost::system::error_code ec;
+            socket.send_to(
+                boost::asio::buffer(datagram.data(), datagram.size()),
+                to,
+                0,
+                ec);
+
+            // The socket never blocks: if it can't take a datagram now, the
+            // rest are dropped too. The peer protocol still relays them.
+            if (ec)
+            {
+                JLOG(journal_.trace()) << "XUSH: send to " << target
+                                       << " failed: " << ec.message();
+                break;
+            }
+        }
+    }
+}
+
+void
+OverlayImpl::openXUSH()
+{
+    using boost::asio::ip::udp;
+
+    std::lock_guard lock(xushSendMutex_);
+
+    auto open = [this](udp::socket& socket, udp const& protocol) {
+        boost::system::error_code ec;
+        socket.open(protocol, ec);
+        if (!ec && protocol == udp::v6())
+            socket.set_option(boost::asio::ip::v6_only(true), ec);
+        if (!ec)
+            socket.non_blocking(true, ec);
+
+        if (ec)
+        {
+            JLOG(journal_.warn()) << "XUSH: unable to open "
+                                  << (protocol == udp::v4() ? "IPv4" : "IPv6")
+                                  << " socket: " << ec.message();
+            boost::system::error_code ignored;
+            socket.close(ignored);
+        }
+    };
+
+    open(xushSocket4_, udp::v4());
+    open(xushSocket6_, udp::v6());
+}
+
+void
+OverlayImpl::closeXUSH()
+{
+    std::lock_guard lock(xushSendMutex_);
+    boost::system::error_code ignored;
+    xushSocket4_.close(ignored);
+    xushSocket6_.close(ignored);
+}
+
 //------------------------------------------------------------------------------
 
 Overlay::Setup

src/xrpld/overlay/detail/OverlayImpl.h

@@ -28,6 +28,7 @@
 #include <xrpld/overlay/detail/Handshake.h>
 #include <xrpld/overlay/detail/TrafficCount.h>
 #include <xrpld/overlay/detail/TxMetrics.h>
+#include <xrpld/overlay/detail/XUSH.h>
 #include <xrpld/peerfinder/PeerfinderManager.h>
 #include <xrpld/rpc/ServerHandler.h>
 #include <xrpl/basics/Resolver.h>
@@ -40,6 +41,7 @@
 #include <boost/algorithm/string/predicate.hpp>
 #include <boost/asio/basic_waitable_timer.hpp>
 #include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/ip/udp.hpp>
 #include <boost/asio/ssl/context.hpp>
 #include <boost/asio/strand.hpp>
 #include <boost/container/flat_map.hpp>
@@ -135,6 +137,15 @@ private:
     // Protects the message and the sequence list of manifests
     std::mutex manifestLock_;
 
+    // XUSH (Xahau UDP Superhighway) transaction reassembly
+    std::mutex xushMutex_;
+    xush::Reassembler xushReassembler_;
+
+    // XUSH (Xahau UDP Superhighway) sockets used for sending
+    std::mutex xushSendMutex_;
+    boost::asio::ip::udp::socket xushSocket4_;
+    boost::asio::ip::udp::socket xushSocket6_;
+
     //--------------------------------------------------------------------------
 
 public:
@@ -442,6 +453,14 @@ public:
         txMetrics_.addMetrics(args...);
     }
 
+    void
+    processXUSH(
+        std::string const& message,
+        boost::asio::ip::tcp::endpoint const& remoteEndpoint) override;
+
+    void
+    publishTxXUSH(Slice const& tx, uint256 const& txid) override;
+
 private:
     void
     squelch(
@@ -617,6 +636,49 @@ private:
         }
         m_stats.peerDisconnects = getPeerDisconnect();
     }
+
+    //--------------------------------------------------------------------------
+    //
+    // XUSH (Xahau UDP Superhighway)
+    //
+    //--------------------------------------------------------------------------
+
+    /** Open the sockets used to send highway datagrams. */
+    void
+    openXUSH();
+
+    /** Close the sockets used to send highway datagrams. */
+    void
+    closeXUSH();
+
+    /** Advertise some highway endpoints to some highway peers. */
+    void
+    sendXUSHPeers();
+
+    /** Send every datagram to every target. Sends never block: datagrams
+        that can't be sent immediately are dropped. */
+    void
+    sendXUSH(
+        std::vector<Buffer> const& datagrams,
+        std::vector<beast::IP::Endpoint> const& targets);
+
+    void
+    onXUSHPeers(
+        Slice datagram,
+        beast::IP::Endpoint const& remote,
+        Resource::Consumer& usage);
+
+    void
+    onXUSHTxnFragment(
+        Slice datagram,
+        beast::IP::Endpoint const& remote,
+        Resource::Consumer& usage);
+
+    void
+    onXUSHTransaction(
+        Buffer const& blob,
+        uint256 const& txid,
+        Resource::Consumer& usage);
 };
 
 }  // namespace ripple

src/xrpld/overlay/detail/XUSH.cpp

@@ -0,0 +1,337 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2026 XRPL Labs
+
+    Permission to use, copy, modify, and/or distribute this software for any
+    purpose  with  or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#include <xrpld/overlay/detail/XUSH.h>
+
+#include <boost/asio/ip/address_v4.hpp>
+#include <boost/asio/ip/address_v6.hpp>
+
+#include <algorithm>
+#include <cstring>
+#include <iterator>
+
+namespace ripple {
+namespace xush {
+
+namespace {
+
+char const peersTag[tagSize] = {'X', 'U', 'S', 'H', 'P', 'E', 'E', 'R'};
+char const txnTag[tagSize] = {'X', 'U', 'S', 'H', 'T', 'X', 'N', 'F'};
+
+void
+putU32(std::uint8_t* p, std::uint32_t v)
+{
+    p[0] = static_cast<std::uint8_t>(v >> 24);
+    p[1] = static_cast<std::uint8_t>(v >> 16);
+    p[2] = static_cast<std::uint8_t>(v >> 8);
+    p[3] = static_cast<std::uint8_t>(v);
+}
+
+std::uint32_t
+getU32(std::uint8_t const* p)
+{
+    return (std::uint32_t{p[0]} << 24) | (std::uint32_t{p[1]} << 16) |
+        (std::uint32_t{p[2]} << 8) | std::uint32_t{p[3]};
+}
+
+}  // namespace
+
+MessageType
+classify(Slice datagram)
+{
+    if (datagram.size() < tagSize)
+        return MessageType::unknown;
+
+    if (std::memcmp(datagram.data(), peersTag, tagSize) == 0)
+        return MessageType::peers;
+
+    if (std::memcmp(datagram.data(), txnTag, tagSize) == 0)
+        return MessageType::txn;
+
+    return MessageType::unknown;
+}
+
+Buffer
+encodePeers(std::vector<beast::IP::Endpoint> const& endpoints)
+{
+    std::vector<beast::IP::Endpoint const*> v4;
+    std::vector<beast::IP::Endpoint const*> v6;
+
+    for (auto const& ep : endpoints)
+    {
+        if (v4.size() + v6.size() == maxAdvertisedPeers)
+            break;
+
+        if (ep.address().is_v4())
+            v4.push_back(&ep);
+        else
+            v6.push_back(&ep);
+    }
+
+    Buffer datagram(tagSize + 2 + v4.size() * 8 + v6.size() * 20);
+    auto p = datagram.data();
+
+    std::memcpy(p, peersTag, tagSize);
+    p += tagSize;
+    *p++ = static_cast<std::uint8_t>(v4.size());
+    *p++ = static_cast<std::uint8_t>(v6.size());
+
+    for (auto const ep : v4)
+    {
+        auto const bytes = ep->address().to_v4().to_bytes();
+        std::memcpy(p, bytes.data(), bytes.size());
+        putU32(p + 4, ep->port());
+        p += 8;
+    }
+
+    for (auto const ep : v6)
+    {
+        auto const bytes = ep->address().to_v6().to_bytes();
+        std::memcpy(p, bytes.data(), bytes.size());
+        putU32(p + 16, ep->port());
+        p += 20;
+    }
+
+    return datagram;
+}
+
+std::optional<std::vector<beast::IP::Endpoint>>
+decodePeers(Slice datagram)
+{
+    if (classify(datagram) != MessageType::peers ||
+        datagram.size() < tagSize + 2)
+        return std::nullopt;
+
+    auto p = datagram.data() + tagSize;
+    std::size_t const n4 = p[0];
+    std::size_t const n6 = p[1];
+    p += 2;
+
+    if (n4 + n6 > maxAdvertisedPeers ||
+        datagram.size() != tagSize + 2 + n4 * 8 + n6 * 20)
+        return std::nullopt;
+
+    std::vector<beast::IP::Endpoint> endpoints;
+    endpoints.reserve(n4 + n6);
+
+    for (std::size_t i = 0; i < n4; ++i, p += 8)
+    {
+        auto const port = getU32(p + 4);
+        if (port > 0xFFFF)
+            return std::nullopt;
+
+        boost::asio::ip::address_v4::bytes_type bytes;
+        std::memcpy(bytes.data(), p, bytes.size());
+        endpoints.emplace_back(
+            boost::asio::ip::address_v4(bytes),
+            static_cast<std::uint16_t>(port));
+    }
+
+    for (std::size_t i = 0; i < n6; ++i, p += 20)
+    {
+        auto const port = getU32(p + 16);
+        if (port > 0xFFFF)
+            return std::nullopt;
+
+        boost::asio::ip::address_v6::bytes_type bytes;
+        std::memcpy(bytes.data(), p, bytes.size());
+        endpoints.emplace_back(
+            boost::asio::ip::address_v6(bytes),
+            static_cast<std::uint16_t>(port));
+    }
+
+    return endpoints;
+}
+
+std::vector<Buffer>
+encodeTxn(Slice txn, uint256 const& txid, std::size_t maxPayload)
+{
+    if (txn.empty() || txn.size() > maxTxnSize || maxPayload == 0)
+        return {};
+
+    std::size_t const count = (txn.size() + maxPayload - 1) / maxPayload;
+    if (count > maxFragments)
+        return {};
+
+    std::vector<Buffer> datagrams;
+    datagrams.reserve(count);
+
+    for (std::size_t i = 0; i < count; ++i)
+    {
+        auto const offset = i * maxPayload;
+        auto const size = std::min(maxPayload, txn.size() - offset);
+
+        auto& datagram = datagrams.emplace_back(txnHeaderSize + size);
+        auto const p = datagram.data();
+
+        std::memcpy(p, txnTag, tagSize);
+        std::memcpy(p + tagSize, txid.data(), txid.size());
+        putU32(p + 40, static_cast<std::uint32_t>(txn.size()));
+        putU32(p + 44, static_cast<std::uint32_t>(count));
+        putU32(p + 48, static_cast<std::uint32_t>(i));
+        std::memcpy(p + txnHeaderSize, txn.data() + offset, size);
+    }
+
+    return datagrams;
+}
+
+std::optional<TxnFragment>
+decodeTxnFragment(Slice datagram)
+{
+    if (classify(datagram) != MessageType::txn ||
+        datagram.size() <= txnHeaderSize)
+        return std::nullopt;
+
+    auto const p = datagram.data();
+
+    TxnFragment fragment;
+    fragment.txid = uint256::fromVoid(p + tagSize);
+    fragment.totalSize = getU32(p + 40);
+    fragment.count = getU32(p + 44);
+    fragment.index = getU32(p + 48);
+    fragment.payload =
+        Slice(p + txnHeaderSize, datagram.size() - txnHeaderSize);
+
+    if (fragment.totalSize == 0 || fragment.totalSize > maxTxnSize ||
+        fragment.count == 0 || fragment.count > maxFragments ||
+        fragment.count > fragment.totalSize ||
+        fragment.index >= fragment.count ||
+        fragment.payload.size() > fragment.totalSize)
+        return std::nullopt;
+
+    return fragment;
+}
+
+//------------------------------------------------------------------------------
+
+Reassembler::Reassembler(
+    std::size_t maxPending,
+    std::size_t maxBytes,
+    std::chrono::seconds timeout)
+    : maxPending_(maxPending), maxBytes_(maxBytes), timeout_(timeout)
+{
+}
+
+Reassembler::Result
+Reassembler::add(
+    beast::IP::Endpoint const& source,
+    TxnFragment const& fragment,
+    clock_type::time_point now)
+{
+    // Most transactions fit in a single datagram and need no state.
+    if (fragment.count == 1)
+    {
+        if (fragment.payload.size() != fragment.totalSize)
+            return {Status::invalid, {}};
+
+        return {Status::complete, Buffer(fragment.payload)};
+    }
+
+    expire(now);
+
+    Key const key{fragment.txid, source};
+    auto it = partials_.find(key);
+
+    if (it == partials_.end())
+    {
+        if (partials_.size() >= maxPending_)
+            return {Status::overloaded, {}};
+
+        Partial partial;
+        partial.started = now;
+        partial.totalSize = fragment.totalSize;
+        partial.count = fragment.count;
+        it = partials_.emplace(key, std::move(partial)).first;
+    }
+    else if (
+        it->second.totalSize != fragment.totalSize ||
+        it->second.count != fragment.count)
+    {
+        erase(it);
+        return {Status::invalid, {}};
+    }
+
+    auto& partial = it->second;
+
+    // Repeated fragments are ignored
+    if (partial.fragments.count(fragment.index) != 0)
+        return {Status::pending, {}};
+
+    if (partial.received + fragment.payload.size() > partial.totalSize)
+    {
+        erase(it);
+        return {Status::invalid, {}};
+    }
+
+    auto const cost = fragment.payload.size() + fragmentOverhead;
+    if (bytes_ + cost > maxBytes_)
+    {
+        if (partial.fragments.empty())
+            erase(it);
+        return {Status::overloaded, {}};
+    }
+
+    partial.fragments.emplace(fragment.index, Buffer(fragment.payload));
+    partial.received += fragment.payload.size();
+    partial.cost += cost;
+    bytes_ += cost;
+
+    if (partial.fragments.size() < partial.count)
+        return {Status::pending, {}};
+
+    if (partial.received != partial.totalSize)
+    {
+        erase(it);
+        return {Status::invalid, {}};
+    }
+
+    // All fragments are present and std::map keeps them in index order
+    Buffer txn(partial.totalSize);
+    auto p = txn.data();
+    for (auto const& [index, payload] : partial.fragments)
+    {
+        std::memcpy(p, payload.data(), payload.size());
+        p += payload.size();
+    }
+
+    erase(it);
+    return {Status::complete, std::move(txn)};
+}
+
+void
+Reassembler::expire(clock_type::time_point now)
+{
+    for (auto it = partials_.begin(); it != partials_.end();)
+    {
+        if (now - it->second.started > timeout_)
+            erase(it++);
+        else
+            ++it;
+    }
+}
+
+void
+Reassembler::erase(Map::iterator it)
+{
+    bytes_ -= it->second.cost;
+    partials_.erase(it);
+}
+
+}  // namespace xush
+}  // namespace ripple

src/xrpld/overlay/detail/XUSH.h

@@ -0,0 +1,264 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2026 XRPL Labs
+
+    Permission to use, copy, modify, and/or distribute this software for any
+    purpose  with  or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#ifndef RIPPLE_OVERLAY_XUSH_H_INCLUDED
+#define RIPPLE_OVERLAY_XUSH_H_INCLUDED
+
+#include <xrpl/basics/Buffer.h>
+#include <xrpl/basics/Slice.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/beast/net/IPEndpoint.h>
+
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <map>
+#include <optional>
+#include <utility>
+#include <vector>
+
+namespace ripple {
+
+/** XUSH: the Xahau UDP Superhighway.
+
+    A best-effort UDP transport that runs alongside the peer protocol, on the
+    same port number as the peer port, and is used to fan transactions out to
+    other highway-enabled servers faster than the peer relay alone. Nothing
+    depends on it: every transaction sent over the highway is also relayed
+    over the peer protocol.
+
+    Every datagram starts with an eight byte ASCII tag that identifies its
+    type. All integers are unsigned and big-endian.
+
+    XUSHPEER advertises other highway endpoints:
+
+        [8]         "XUSHPEER"
+        [1]         number of IPv4 endpoints (n4)
+        [1]         number of IPv6 endpoints (n6)
+        [n4 * 8]    4 byte IPv4 address, 4 byte port
+        [n6 * 20]   16 byte IPv6 address, 4 byte port
+
+    XUSHTXNF carries one fragment of a serialized transaction:
+
+        [8]         "XUSHTXNF"
+        [32]        transaction ID
+        [4]         total size of the serialized transaction
+        [4]         number of fragments
+        [4]         index of this fragment
+        [...]       fragment payload
+
+    The payloads of all fragments, concatenated in index order, form the
+    serialized transaction.
+*/
+namespace xush {
+
+enum class MessageType { unknown, peers, txn };
+
+/** Size of the tag at the start of every datagram. */
+std::size_t constexpr tagSize = 8;
+
+/** The largest datagram we send.
+
+    This is the IPv6 minimum MTU (1280) less the IPv6 (40) and UDP (8)
+    headers, so that our datagrams are never fragmented at the IP layer,
+    which many networks drop.
+*/
+std::size_t constexpr maxDatagramSize = 1232;
+
+/** The most endpoints in a XUSHPEER datagram. */
+std::size_t constexpr maxAdvertisedPeers = 50;
+
+/** Size of the XUSHTXNF header. */
+std::size_t constexpr txnHeaderSize = 52;
+
+/** The largest fragment payload we send. */
+std::size_t constexpr maxFragmentPayload = maxDatagramSize - txnHeaderSize;
+
+/** The largest serialized transaction carried over the highway. */
+std::uint32_t constexpr maxTxnSize = 2 * 1024 * 1024;
+
+/** The most fragments a transaction may be split into. */
+std::uint32_t constexpr maxFragments = 2048;
+
+static_assert(
+    tagSize + 2 + maxAdvertisedPeers * 20 <= maxDatagramSize,
+    "XUSHPEER datagrams must fit in maxDatagramSize");
+static_assert(
+    (maxTxnSize + maxFragmentPayload - 1) / maxFragmentPayload <= maxFragments,
+    "The largest transaction must fit in maxFragments fragments");
+
+/** Number of highway endpoints each relayed transaction is sent to.
+
+    Every server relays a transaction at most once (see HashRouter), so this
+    bounds the highway traffic per transaction to roughly this many datagrams
+    per server while still reaching every server with high probability.
+*/
+std::size_t constexpr txnFanout = 16;
+
+/** Number of highway endpoints we send a XUSHPEER datagram to each second. */
+std::size_t constexpr advertFanout = 4;
+
+/** Identify the type of a datagram from its tag. */
+MessageType
+classify(Slice datagram);
+
+/** Build a XUSHPEER datagram.
+
+    At most maxAdvertisedPeers endpoints are included.
+*/
+Buffer
+encodePeers(std::vector<beast::IP::Endpoint> const& endpoints);
+
+/** Parse a XUSHPEER datagram.
+
+    @return The advertised endpoints, or nothing if the datagram is malformed.
+*/
+std::optional<std::vector<beast::IP::Endpoint>>
+decodePeers(Slice datagram);
+
+/** A parsed XUSHTXNF datagram. */
+struct TxnFragment
+{
+    uint256 txid;
+    std::uint32_t totalSize = 0;
+    std::uint32_t count = 0;
+    std::uint32_t index = 0;
+
+    /** Refers to the memory of the parsed datagram. */
+    Slice payload;
+};
+
+/** Split a serialized transaction into XUSHTXNF datagrams.
+
+    @return The datagrams, or nothing if the transaction is empty or larger
+            than maxTxnSize.
+*/
+std::vector<Buffer>
+encodeTxn(
+    Slice txn,
+    uint256 const& txid,
+    std::size_t maxPayload = maxFragmentPayload);
+
+/** Parse a XUSHTXNF datagram.
+
+    @return The fragment, or nothing if the datagram is malformed.
+*/
+std::optional<TxnFragment>
+decodeTxnFragment(Slice datagram);
+
+/** Rebuilds serialized transactions from XUSHTXNF fragments.
+
+    Partial transactions are tracked per (transaction ID, sender), so that a
+    sender can't corrupt a transaction being rebuilt from another sender's
+    fragments. Memory use is bounded by the number of partial transactions,
+    the number of bytes buffered and a timeout.
+
+    This class is not thread-safe.
+*/
+class Reassembler
+{
+public:
+    using clock_type = std::chrono::steady_clock;
+
+    enum class Status {
+        /** The fragment was accepted; the transaction is incomplete. */
+        pending,
+
+        /** The transaction is complete. */
+        complete,
+
+        /** The fragment is inconsistent; the partial transaction was
+            discarded. */
+        invalid,
+
+        /** The fragment was dropped because a limit was reached. */
+        overloaded
+    };
+
+    struct Result
+    {
+        Status status;
+
+        /** The serialized transaction, if status is complete. */
+        Buffer txn;
+    };
+
+    static std::size_t constexpr defaultMaxPending = 256;
+    static std::size_t constexpr defaultMaxBytes = 32 * 1024 * 1024;
+    static constexpr std::chrono::seconds defaultTimeout{30};
+
+    /** Bookkeeping cost charged against the byte limit for each buffered
+        fragment, so that many tiny fragments can't exhaust memory. */
+    static std::size_t constexpr fragmentOverhead = 64;
+
+    explicit Reassembler(
+        std::size_t maxPending = defaultMaxPending,
+        std::size_t maxBytes = defaultMaxBytes,
+        std::chrono::seconds timeout = defaultTimeout);
+
+    Result
+    add(beast::IP::Endpoint const& source,
+        TxnFragment const& fragment,
+        clock_type::time_point now);
+
+    /** Number of partial transactions. */
+    std::size_t
+    pending() const
+    {
+        return partials_.size();
+    }
+
+    /** Bytes charged against the byte limit. */
+    std::size_t
+    bytes() const
+    {
+        return bytes_;
+    }
+
+private:
+    struct Partial
+    {
+        clock_type::time_point started;
+        std::uint32_t totalSize = 0;
+        std::uint32_t count = 0;
+        std::size_t received = 0;
+        std::size_t cost = 0;
+        std::map<std::uint32_t, Buffer> fragments;
+    };
+
+    using Key = std::pair<uint256, beast::IP::Endpoint>;
+    using Map = std::map<Key, Partial>;
+
+    void
+    expire(clock_type::time_point now);
+
+    void
+    erase(Map::iterator it);
+
+    std::size_t const maxPending_;
+    std::size_t const maxBytes_;
+    std::chrono::seconds const timeout_;
+    Map partials_;
+    std::size_t bytes_ = 0;
+};
+
+}  // namespace xush
+}  // namespace ripple
+
+#endif

src/xrpld/peerfinder/PeerfinderManager.h

@@ -75,6 +75,9 @@ struct Config
     /** Limit how many incoming connections we allow per IP */
     int ipLimit;
 
+    /** `true` if the Xahau UDP Superhighway (XUSH) is enabled. */
+    bool udpHighway = false;
+
     //--------------------------------------------------------------------------
 
     /** Create a configuration with default values. */
@@ -272,6 +275,44 @@ public:
     */
     virtual void
     once_per_second() = 0;
+
+    //--------------------------------------------------------------------------
+    //
+    // XUSH (Xahau UDP Superhighway)
+    //
+    // Highway endpoints are learned from fixed peers, from endpoint gossip
+    // and redirects received over authenticated peer connections, and from
+    // XUSHPEER advertisements received over UDP. Nothing is tracked unless
+    // the highway is enabled in the configuration.
+    //
+    //--------------------------------------------------------------------------
+
+    /** Add highway endpoints advertised by another server over UDP.
+
+        Advertisements are unauthenticated, so only valid public addresses
+        are accepted, they never extend the lifetime of an existing entry,
+        and the number of learned (non-fixed) entries is bounded.
+    */
+    virtual void
+    add_highway_peers(std::vector<beast::IP::Endpoint> const& endpoints) = 0;
+
+    /** Returns up to `n` distinct highway endpoints, chosen at random, to
+        send highway traffic to.
+
+        If our own address is private only fixed peers are returned, so that
+        we never reveal our address to servers we did not choose.
+    */
+    virtual std::vector<beast::IP::Endpoint>
+    highway_targets(std::size_t n) = 0;
+
+    /** Returns up to `n` distinct highway endpoints, chosen at random, that
+        may be advertised to other servers.
+
+        Fixed peers are never advertised, and a server whose own address is
+        private advertises nothing.
+    */
+    virtual std::vector<beast::IP::Endpoint>
+    highway_adverts(std::size_t n) = 0;
 };
 
 }  // namespace PeerFinder

src/xrpld/peerfinder/detail/PeerfinderConfig.cpp

@@ -72,6 +72,7 @@ Config::onWrite(beast::PropertyStream::Map& map)
     map["port"] = listeningPort;
     map["features"] = features;
     map["ip_limit"] = ipLimit;
+    map["udp_highway"] = udpHighway;
 }
 
 Config
@@ -84,6 +85,7 @@ Config::makeConfig(
     PeerFinder::Config config;
 
     config.peerPrivate = cfg.PEER_PRIVATE;
+    config.udpHighway = cfg.UDP_HIGHWAY;
 
     // Servers with peer privacy don't want to allow incoming connections
     config.wantIncoming = (!config.peerPrivate) && (port != 0);

src/xrpld/peerfinder/detail/PeerfinderManager.cpp

@@ -24,13 +24,16 @@
 #include <xrpld/peerfinder/detail/Logic.h>
 #include <xrpld/peerfinder/detail/SourceStrings.h>
 #include <xrpld/peerfinder/detail/StoreSqdb.h>
-
+#include <xrpld/peerfinder/detail/Tuning.h>
+#include <xrpl/basics/random.h>
 #include <boost/asio/executor_work_guard.hpp>
 #include <boost/asio/io_context.hpp>
 #include <boost/utility/in_place_factory.hpp>
 
 #include <memory>
+#include <mutex>
 #include <optional>
+#include <random>
 #include <thread>
 
 namespace ripple {
@@ -99,6 +102,17 @@ public:
     void
     setConfig(Config const& config) override
     {
+        {
+            std::lock_guard lock(m_highwayMutex);
+            m_highwayEnabled = config.udpHighway;
+            m_highwayPrivate = config.peerPrivate;
+            if (!m_highwayEnabled)
+            {
+                m_highwayPeers.clear();
+                m_highwayLearned = 0;
+            }
+        }
+
         m_logic.config(config);
     }
 
@@ -113,6 +127,16 @@ public:
         std::string const& name,
         std::vector<beast::IP::Endpoint> const& addresses) override
     {
+        {
+            std::lock_guard lock(m_highwayMutex);
+            if (m_highwayEnabled)
+            {
+                auto const now = m_clock.now();
+                for (auto const& address : addresses)
+                    learnHighwayPeer(address, HighwaySource::fixed, now);
+            }
+        }
+
         m_logic.addFixedPeer(name, addresses);
     }
 
@@ -150,6 +174,17 @@ public:
     on_endpoints(std::shared_ptr<Slot> const& slot, Endpoints const& endpoints)
         override
     {
+        {
+            std::lock_guard lock(m_highwayMutex);
+            if (m_highwayEnabled)
+            {
+                auto const now = m_clock.now();
+                for (auto const& ep : endpoints)
+                    learnHighwayPeer(
+                        ep.address, HighwaySource::authenticated, now);
+            }
+        }
+
         SlotImp::ptr impl(std::dynamic_pointer_cast<SlotImp>(slot));
         m_logic.on_endpoints(impl, endpoints);
     }
@@ -173,6 +208,19 @@ public:
         boost::asio::ip::tcp::endpoint const& remote_address,
         std::vector<boost::asio::ip::tcp::endpoint> const& eps) override
     {
+        {
+            std::lock_guard lock(m_highwayMutex);
+            if (m_highwayEnabled)
+            {
+                auto const now = m_clock.now();
+                for (auto const& ep : eps)
+                    learnHighwayPeer(
+                        beast::IPAddressConversion::from_asio(ep),
+                        HighwaySource::authenticated,
+                        now);
+            }
+        }
+
         m_logic.onRedirects(eps.begin(), eps.end(), remote_address);
     }
 
@@ -214,6 +262,7 @@ public:
     once_per_second() override
     {
         m_logic.once_per_second();
+        expireHighwayPeers();
     }
 
     std::vector<std::pair<std::shared_ptr<Slot>, std::vector<Endpoint>>>
@@ -222,6 +271,47 @@ public:
         return m_logic.buildEndpointsForPeers();
     }
 
+    //--------------------------------------------------------------------------
+    //
+    // XUSH (Xahau UDP Superhighway)
+    //
+    //--------------------------------------------------------------------------
+
+    void
+    add_highway_peers(
+        std::vector<beast::IP::Endpoint> const& endpoints) override
+    {
+        std::lock_guard lock(m_highwayMutex);
+        if (!m_highwayEnabled)
+            return;
+
+        auto const now = m_clock.now();
+        for (auto const& ep : endpoints)
+            learnHighwayPeer(ep, HighwaySource::advertised, now);
+    }
+
+    std::vector<beast::IP::Endpoint>
+    highway_targets(std::size_t n) override
+    {
+        std::lock_guard lock(m_highwayMutex);
+        return sampleHighwayPeers(n, [this](HighwayPeer const& peer) {
+            return peer.fixed || !m_highwayPrivate;
+        });
+    }
+
+    std::vector<beast::IP::Endpoint>
+    highway_adverts(std::size_t n) override
+    {
+        std::lock_guard lock(m_highwayMutex);
+        if (m_highwayPrivate)
+            return {};
+
+        return sampleHighwayPeers(
+            n, [](HighwayPeer const& peer) { return !peer.fixed; });
+    }
+
+    //--------------------------------------------------------------------------
+
     void
     start() override
     {
@@ -243,6 +333,138 @@ public:
     }
 
 private:
+    //--------------------------------------------------------------------------
+    //
+    // XUSH (Xahau UDP Superhighway)
+    //
+    //--------------------------------------------------------------------------
+
+    enum class HighwaySource {
+        // Configured by the operator in [ips_fixed]
+        fixed,
+
+        // Endpoint gossip or a redirect received over a peer connection
+        authenticated,
+
+        // An unauthenticated XUSHPEER datagram
+        advertised
+    };
+
+    struct HighwayPeer
+    {
+        // Fixed peers never expire and are never advertised.
+        bool fixed = false;
+
+        // When the endpoint was last learned from a fixed or authenticated
+        // source. Learned entries expire once this is too old.
+        clock_type::time_point learned;
+    };
+
+    std::mutex m_highwayMutex;
+    bool m_highwayEnabled = false;
+    bool m_highwayPrivate = true;
+    std::size_t m_highwayLearned = 0;  // Number of entries that are not fixed
+    std::map<beast::IP::Endpoint, HighwayPeer> m_highwayPeers;
+
+    // Requires m_highwayMutex
+    void
+    learnHighwayPeer(
+        beast::IP::Endpoint const& ep,
+        HighwaySource source,
+        clock_type::time_point now)
+    {
+        if (is_unspecified(ep) || ep.port() == 0)
+            return;
+
+        auto const it = m_highwayPeers.find(ep);
+
+        if (source == HighwaySource::fixed)
+        {
+            if (it == m_highwayPeers.end())
+            {
+                m_highwayPeers.emplace(ep, HighwayPeer{true, now});
+            }
+            else if (!it->second.fixed)
+            {
+                it->second.fixed = true;
+                --m_highwayLearned;
+            }
+            return;
+        }
+
+        // Endpoints learned from other servers must be publicly routable.
+        if (!is_public(ep))
+            return;
+
+        if (it != m_highwayPeers.end())
+        {
+            // Only authenticated sources extend the life of an entry, so that
+            // unauthenticated advertisements can't keep stale entries alive.
+            if (source == HighwaySource::authenticated)
+                it->second.learned = now;
+            return;
+        }
+
+        if (m_highwayLearned >= Tuning::highwayPeersMax)
+            return;
+
+        m_highwayPeers.emplace(ep, HighwayPeer{false, now});
+        ++m_highwayLearned;
+    }
+
+    void
+    expireHighwayPeers()
+    {
+        std::lock_guard lock(m_highwayMutex);
+        auto const now = m_clock.now();
+
+        for (auto it = m_highwayPeers.begin(); it != m_highwayPeers.end();)
+        {
+            if (!it->second.fixed &&
+                now - it->second.learned > Tuning::highwayPeerSecondsToLive)
+            {
+                it = m_highwayPeers.erase(it);
+                --m_highwayLearned;
+            }
+            else
+            {
+                ++it;
+            }
+        }
+    }
+
+    // Requires m_highwayMutex
+    template <class Predicate>
+    std::vector<beast::IP::Endpoint>
+    sampleHighwayPeers(std::size_t n, Predicate&& eligible)
+    {
+        std::vector<beast::IP::Endpoint const*> candidates;
+        candidates.reserve(m_highwayPeers.size());
+        for (auto const& [ep, peer] : m_highwayPeers)
+        {
+            if (eligible(peer))
+                candidates.push_back(&ep);
+        }
+
+        n = std::min(n, candidates.size());
+
+        // A partial Fisher-Yates shuffle: the first n candidates become a
+        // uniformly random sample.
+        std::vector<beast::IP::Endpoint> result;
+        result.reserve(n);
+        for (std::size_t i = 0; i < n; ++i)
+        {
+            std::uniform_int_distribution<std::size_t> pick(
+                i, candidates.size() - 1);
+            std::swap(candidates[i], candidates[pick(default_prng())]);
+            result.push_back(*candidates[i]);
+        }
+
+        return result;
+    }
+
+    //--------------------------------------------------------------------------
+
     struct Stats
     {
         template <class Handler>

src/xrpld/peerfinder/detail/Tuning.h

@@ -131,6 +131,13 @@ std::chrono::seconds constexpr liveCacheSecondsToLive(30);
 // Note that we ignore the port for purposes of comparison.
 std::chrono::seconds constexpr recentAttemptDuration(60);
 
+// The most learned (non-fixed) UDP Superhighway endpoints we will track.
+std::size_t constexpr highwayPeersMax = 2000;
+
+// How long a learned UDP Superhighway endpoint is kept after it was last
+// learned from an authenticated source (endpoint gossip or a redirect).
+std::chrono::seconds constexpr highwayPeerSecondsToLive(500);
+
 }  // namespace Tuning
 /** @} */
 

src/xrpld/peerfinder/make_Manager.h

@@ -21,9 +21,8 @@
 #define RIPPLE_PEERFINDER_MAKE_MANAGER_H_INCLUDED
 
 #include <xrpld/peerfinder/PeerfinderManager.h>
-
+#include <xrpl/beast/insight/Collector.h>
 #include <boost/asio/io_context.hpp>
-
 #include <memory>
 
 namespace ripple {

src/xrpld/rpc/ServerHandler.h

@@ -286,6 +286,7 @@ public:
     onUDPMessage(
         std::string const& message,
         boost::asio::ip::tcp::endpoint const& remoteEndpoint,
+        Port const& port,
         std::function<void(std::string const&)> sendResponse);
 
     void

src/xrpld/rpc/detail/ServerHandler.cpp

@@ -1119,8 +1119,13 @@ void
 ServerHandler::onUDPMessage(
     std::string const& message,
     boost::asio::ip::tcp::endpoint const& remoteEndpoint,
+    Port const& port,
     std::function<void(std::string const&)> sendResponse)
 {
+    // Datagrams arriving on the peer port belong to the UDP Superhighway
+    if (port.has_peer())
+        return app_.overlay().processXUSH(message, remoteEndpoint);
+
     Json::Value jv;
     if (message.size() > RPC::Tuning::maxRequestSize ||
         !Json::Reader{}.parse(message, jv) || !jv.isObject())
@@ -2256,6 +2261,12 @@ setup_ServerHandler(Config const& config, std::ostream&& log)
     ServerHandler::Setup setup;
     setup.ports = parse_Ports(config, log);
 
+    if (config.UDP_HIGHWAY)
+    {
+        for (auto& port : setup.ports)
+            port.udp_highway = port.has_peer();
+    }
+
     setup_Client(setup);
     setup_Overlay(setup);