feat: Drain proposals and validations ahead of bulk traffic in the peer send queue

include/xrpl/protocol/jss.h

@@ -481,6 +481,7 @@ JSS(ports);                       // out: NetworkOPs
 JSS(previous);                    // out: Reservations
 JSS(previous_ledger);             // out: LedgerPropose
 JSS(price);                       // out: amm_info, AuctionSlot
+JSS(priority_send_queue);         // out: PeerImp
 JSS(proof);                       // in: BookOffers
 JSS(propose_seq);                 // out: LedgerPropose
 JSS(proposers);                   // out: NetworkOPs, LedgerConsensus
@@ -536,6 +537,7 @@ JSS(seed);                        //
 JSS(seed_hex);                    // in: WalletPropose, TransactionSign
 JSS(send_currencies);             // out: AccountCurrencies
 JSS(send_max);                    // in: PathRequest, RipplePathFind
+JSS(send_queue);                  // out: PeerImp
 JSS(seq);                         // in: LedgerEntry
                                   // out: NetworkOPs, RPCSub, AccountOffers, ValidatorList,
                                   //      ValidatorInfo, Manifest

src/test/overlay/PeerSendQueue_test.cpp

@@ -0,0 +1,134 @@
+#include <xrpld/overlay/Message.h>
+#include <xrpld/overlay/detail/PeerSendQueue.h>
+
+#include <xrpl/beast/unit_test/suite.h>
+
+#include <xrpl.pb.h>
+
+#include <memory>
+
+namespace xrpl::test {
+
+class PeerSendQueue_test : public beast::unit_test::Suite
+{
+    static std::shared_ptr<Message>
+    transaction()
+    {
+        protocol::TMTransaction tx;
+        tx.set_rawtransaction("tx");
+        tx.set_status(protocol::tsNEW);
+        return std::make_shared<Message>(tx, protocol::mtTRANSACTION);
+    }
+
+    static std::shared_ptr<Message>
+    proposal()
+    {
+        protocol::TMProposeSet p;
+        p.set_proposeseq(1);
+        p.set_currenttxhash("h");
+        p.set_previousledger("p");
+        p.set_closetime(1);
+        p.set_nodepubkey("k");
+        p.set_signature("s");
+        return std::make_shared<Message>(p, protocol::mtPROPOSE_LEDGER);
+    }
+
+    static std::shared_ptr<Message>
+    validation()
+    {
+        protocol::TMValidation v;
+        v.set_validation("v");
+        return std::make_shared<Message>(v, protocol::mtVALIDATION);
+    }
+
+    void
+    testClassification()
+    {
+        testcase("proposals and validations are priority, the rest is bulk");
+        BEAST_EXPECT(proposal()->isPriority());
+        BEAST_EXPECT(validation()->isPriority());
+        BEAST_EXPECT(!transaction()->isPriority());
+
+        protocol::TMPing ping;
+        ping.set_type(protocol::TMPing::ptPING);
+        BEAST_EXPECT(!Message(ping, protocol::mtPING).isPriority());
+    }
+
+    void
+    testPriorityFirst()
+    {
+        testcase("priority lane drains before bulk regardless of arrival order");
+        PeerSendQueue q;
+        BEAST_EXPECT(q.empty());
+
+        auto const tx1 = transaction();
+        auto const tx2 = transaction();
+        auto const prop = proposal();
+        q.push(tx1);
+        q.push(tx2);
+        q.push(prop);
+        BEAST_EXPECT(q.bulkSize() == 2);
+        BEAST_EXPECT(q.prioritySize() == 1);
+
+        BEAST_EXPECT(q.pop() == prop);
+        BEAST_EXPECT(q.pop() == tx1);
+        BEAST_EXPECT(q.pop() == tx2);
+        BEAST_EXPECT(q.empty());
+    }
+
+    void
+    testInterleaving()
+    {
+        testcase("a priority message pushed mid-stream jumps the remaining bulk");
+        PeerSendQueue q;
+        auto const tx1 = transaction();
+        auto const tx2 = transaction();
+        auto const tx3 = transaction();
+        q.push(tx1);
+        q.push(tx2);
+        q.push(tx3);
+        BEAST_EXPECT(q.pop() == tx1);
+
+        auto const val = validation();
+        q.push(val);
+        BEAST_EXPECT(q.pop() == val);
+        BEAST_EXPECT(q.pop() == tx2);
+        BEAST_EXPECT(q.pop() == tx3);
+        BEAST_EXPECT(q.empty());
+    }
+
+    void
+    testLaneOrder()
+    {
+        testcase("order is preserved within each lane");
+        PeerSendQueue q;
+        auto const p1 = proposal();
+        auto const v1 = validation();
+        auto const p2 = proposal();
+        q.push(p1);
+        q.push(transaction());
+        q.push(v1);
+        q.push(p2);
+        BEAST_EXPECT(q.prioritySize() == 3);
+        BEAST_EXPECT(q.bulkSize() == 1);
+        BEAST_EXPECT(q.pop() == p1);
+        BEAST_EXPECT(q.pop() == v1);
+        BEAST_EXPECT(q.pop() == p2);
+        BEAST_EXPECT(q.prioritySize() == 0);
+        BEAST_EXPECT(q.bulkSize() == 1);
+    }
+
+public:
+    void
+    run() override
+    {
+        testClassification();
+        testPriorityFirst();
+        testInterleaving();
+        testLaneOrder();
+    }
+};
+
+BEAST_DEFINE_TESTSUITE(PeerSendQueue, overlay, xrpl);
+
+}  // namespace xrpl::test

src/xrpld/overlay/Message.h

@@ -127,10 +127,21 @@ public:
         return validatorKey_;
     }
 
+    /**
+     * Whether this message carries consensus state (a proposal or a
+     * validation) and is sent ahead of bulk traffic on a peer connection.
+     */
+    bool
+    isPriority() const
+    {
+        return priority_;
+    }
+
 private:
     std::vector<uint8_t> buffer_;
     std::vector<uint8_t> bufferCompressed_;
     std::size_t category_;
+    bool priority_;
     std::once_flag onceFlag_;
     std::optional<PublicKey> validatorKey_;
 

src/xrpld/overlay/detail/Message.cpp

@@ -23,6 +23,7 @@ Message::Message(
     protocol::MessageType type,
     std::optional<PublicKey> const& validator)
     : category_(static_cast<std::size_t>(TrafficCount::categorize(message, type, false)))
+    , priority_(type == protocol::mtPROPOSE_LEDGER || type == protocol::mtVALIDATION)
     , validatorKey_(validator)
 {
     using namespace xrpl::compression;

src/xrpld/overlay/detail/PeerImp.cpp

@@ -302,7 +302,10 @@ PeerImp::send(std::shared_ptr<Message> const& m)
             TrafficCount::Category::Total,
             static_cast<int>(m->getBuffer(self->compressionEnabled_).size()));
 
-        auto sendqSize = self->sendQueue_.size();
+        // Only bulk traffic counts toward the health checks: priority traffic
+        // is bounded by the validator set and must never look like a peer that
+        // has stopped reading.
+        auto const sendqSize = self->sendQueue_.bulkSize();
 
         if (sendqSize < tuning::kTargetSendQueue)
         {
@@ -321,19 +324,32 @@ PeerImp::send(std::shared_ptr<Message> const& m)
 
         self->sendQueue_.push(m);
 
-        if (sendqSize != 0)
+        if (self->writing_)
             return;
 
-        boost::asio::async_write(
-            self->stream_,
-            boost::asio::buffer(self->sendQueue_.front()->getBuffer(self->compressionEnabled_)),
-            bind_executor(
-                self->strand_, [self](error_code const& ec, std::size_t bytesTransferred) {
-                    self->onWriteMessage(ec, bytesTransferred);
-                }));
+        self->writeNext();
     });
 }
 
+void
+PeerImp::writeNext()
+{
+    XRPL_ASSERT(
+        strand_.running_in_this_thread(), "xrpl::PeerImp::writeNext : strand in this thread");
+    XRPL_ASSERT(!writing_, "xrpl::PeerImp::writeNext : no write in flight");
+    XRPL_ASSERT(!sendQueue_.empty(), "xrpl::PeerImp::writeNext : non-empty send queue");
+
+    writing_ = sendQueue_.pop();
+    boost::asio::async_write(
+        stream_,
+        boost::asio::buffer(writing_->getBuffer(compressionEnabled_)),
+        bind_executor(
+            strand_,
+            [self = shared_from_this()](error_code const& ec, std::size_t bytesTransferred) {
+                self->onWriteMessage(ec, bytesTransferred);
+            }));
+}
+
 void
 PeerImp::sendTxQueue()
 {
@@ -528,6 +544,9 @@ PeerImp::json()
         }
     }
 
+    ret[jss::send_queue] = static_cast<json::UInt>(sendQueue_.bulkSize());
+    ret[jss::priority_send_queue] = static_cast<json::UInt>(sendQueue_.prioritySize());
+
     ret[jss::metrics] = json::Value(json::ValueType::Object);
     ret[jss::metrics][jss::total_bytes_recv] = std::to_string(metrics_.recv.totalBytes());
     ret[jss::metrics][jss::total_bytes_sent] = std::to_string(metrics_.sent.totalBytes());
@@ -651,7 +670,7 @@ PeerImp::gracefulClose()
     XRPL_ASSERT(socket_.is_open(), "xrpl::PeerImp::gracefulClose : socket is open");
     XRPL_ASSERT(!gracefulClose_, "xrpl::PeerImp::gracefulClose : socket is not closing");
     gracefulClose_ = true;
-    if (!sendQueue_.empty())
+    if (writing_ || !sendQueue_.empty())
         return;
     setTimer();
     stream_.async_shutdown(bind_executor(
@@ -1007,19 +1026,11 @@ PeerImp::onWriteMessage(error_code ec, std::size_t bytesTransferred)
 
     metrics_.sent.addMessage(bytesTransferred);
 
-    XRPL_ASSERT(!sendQueue_.empty(), "xrpl::PeerImp::onWriteMessage : non-empty send buffer");
-    sendQueue_.pop();
+    XRPL_ASSERT(writing_, "xrpl::PeerImp::onWriteMessage : write was in flight");
+    writing_.reset();
     if (!sendQueue_.empty())
     {
-        // Timeout on writes only
-        boost::asio::async_write(
-            stream_,
-            boost::asio::buffer(sendQueue_.front()->getBuffer(compressionEnabled_)),
-            bind_executor(
-                strand_,
-                [self = shared_from_this()](error_code const& ec, std::size_t bytesTransferred) {
-                    self->onWriteMessage(ec, bytesTransferred);
-                }));
+        writeNext();
         return;
     }
 
@@ -2570,7 +2581,7 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMGetObjectByHash> const& m)
     if (packet.query())
     {
         // this is a query
-        if (sendQueue_.size() >= tuning::kDropSendQueue)
+        if (sendQueue_.bulkSize() >= tuning::kDropSendQueue)
         {
             JLOG(pJournal_.debug()) << "GetObject: Large send queue";
             return;
@@ -3457,7 +3468,7 @@ PeerImp::processLedgerRequest(
     }
     else
     {
-        if (sendQueue_.size() >= tuning::kDropSendQueue)
+        if (sendQueue_.bulkSize() >= tuning::kDropSendQueue)
         {
             JLOG(pJournal_.debug()) << "processLedgerRequest: Large send queue";
             return;

src/xrpld/overlay/detail/PeerImp.h

@@ -8,6 +8,7 @@
 #include <xrpld/overlay/Peer.h>
 #include <xrpld/overlay/Squelch.h>
 #include <xrpld/overlay/detail/OverlayImpl.h>
+#include <xrpld/overlay/detail/PeerSendQueue.h>
 #include <xrpld/overlay/detail/ProtocolVersion.h>
 
 #include <xrpl/basics/Log.h>
@@ -51,7 +52,6 @@
 #include <memory>
 #include <mutex>
 #include <optional>
-#include <queue>
 #include <shared_mutex>
 #include <string>
 #include <type_traits>
@@ -193,7 +193,10 @@ private:
     http_request_type request_;
     http_response_type response_;
     boost::beast::http::fields const& headers_;
-    std::queue<std::shared_ptr<Message>> sendQueue_;
+    PeerSendQueue sendQueue_;
+    // The message whose async_write is in flight. Popped from sendQueue_ when
+    // the write starts so a priority message pushed meanwhile is written next.
+    std::shared_ptr<Message> writing_;
     bool gracefulClose_ = false;
     int largeSendq_ = 0;
     std::unique_ptr<LoadEvent> loadEvent_;
@@ -538,6 +541,10 @@ private:
     onReadMessage(error_code ec, std::size_t bytesTransferred);
 
     // Called when protocol messages bytes are sent
+    // Pop the next message and start its async_write. Strand only, no write in flight.
+    void
+    writeNext();
+
     void
     onWriteMessage(error_code ec, std::size_t bytesTransferred);
 

src/xrpld/overlay/detail/PeerSendQueue.h

@@ -0,0 +1,74 @@
+#pragma once
+
+#include <xrpld/overlay/Message.h>
+
+#include <cstddef>
+#include <memory>
+#include <queue>
+
+namespace xrpl {
+
+/**
+ * Outbound message queue for one peer connection, with a consensus lane.
+ *
+ * Messages that carry consensus state (see Message::isPriority) wait in
+ * their own lane and are popped before any bulk message, so a burst of
+ * relayed transactions cannot delay the proposals and validations queued
+ * behind it. Within each lane order is preserved.
+ *
+ * Not thread safe: the owning peer accesses it from its strand only.
+ */
+class PeerSendQueue
+{
+public:
+    void
+    push(std::shared_ptr<Message> const& m)
+    {
+        if (m->isPriority())
+            priority_.push(m);
+        else
+            bulk_.push(m);
+    }
+
+    /**
+     * Remove and return the next message to write: the priority lane first,
+     * then bulk. Requires !empty().
+     */
+    std::shared_ptr<Message>
+    pop()
+    {
+        auto& lane = priority_.empty() ? bulk_ : priority_;
+        auto m = std::move(lane.front());
+        lane.pop();
+        return m;
+    }
+
+    bool
+    empty() const
+    {
+        return priority_.empty() && bulk_.empty();
+    }
+
+    /**
+     * Bulk lane depth. This is the number the peer health checks use:
+     * priority traffic is bounded by the validator set and must not count
+     * toward a disconnect or a query refusal.
+     */
+    std::size_t
+    bulkSize() const
+    {
+        return bulk_.size();
+    }
+
+    std::size_t
+    prioritySize() const
+    {
+        return priority_.size();
+    }
+
+private:
+    std::queue<std::shared_ptr<Message>> priority_;
+    std::queue<std::shared_ptr<Message>> bulk_;
+};
+
+}  // namespace xrpl