[WIP] Toying with the idea of VL exchange

* Log Validator Lists received from peers

Author: ximinez

src/xrpld/app/misc/ValidatorList.h

@@ -363,7 +363,7 @@ class ValidatorList
     static std::vector<ValidatorBlobInfo>
     parseBlobs(protocol::TMValidatorListCollection const& body);
 
-    static void
+    static std::optional<std::string>
     sendValidatorList(
         Peer& peer,
         std::uint64_t peerSequence,
@@ -375,6 +375,18 @@ class ValidatorList
         HashRouter& hashRouter,
         beast::Journal j);
 
+    std::tuple<
+        std::string,
+        std::uint32_t,
+        std::map<std::size_t, ValidatorBlobInfo>,
+        uint256>
+    sendLatestValidatorLists(
+        Peer& peer,
+        std::uint64_t peerSequence,
+        PublicKey const& publisherKey,
+        HashRouter& hashRouter,
+        beast::Journal j) const;
+
     [[nodiscard]] static std::pair<std::size_t, std::size_t>
     buildValidatorListMessages(
         std::size_t messageVersion,
@@ -815,7 +827,7 @@ class ValidatorList
         HashRouter& hashRouter,
         beast::Journal j);
 
-    static void
+    static std::optional<std::string>
     sendValidatorList(
         Peer& peer,
         std::uint64_t peerSequence,

src/xrpld/app/misc/detail/ValidatorList.cpp

@@ -735,8 +735,62 @@ ValidatorList::buildValidatorListMessages(
     return {0, 0};
 }
 
+std::tuple<
+    std::string,
+    std::uint32_t,
+    std::map<std::size_t, ValidatorBlobInfo>,
+    uint256>
+ValidatorList::sendLatestValidatorLists(
+    Peer& peer,
+    std::uint64_t peerSequence,
+    PublicKey const& publisherKey,
+    HashRouter& hashRouter,
+    beast::Journal j) const
+{
+    std::vector<ValidatorList::MessageWithHash> messages;
+    std::map<std::size_t, ValidatorBlobInfo> blobInfos;
+
+    if (publisherLists_.count(publisherKey) == 0)
+        return {};
+    ValidatorList::PublisherListCollection const& lists =
+        publisherLists_.at(publisherKey);
+
+    auto const maxSequence = lists.current.sequence;
+    ASSERT(
+        lists.current.sequence == maxSequence ||
+            lists.remaining.count(maxSequence) == 1,
+        "ripple::ValidatorList::sendLatestValidatorLists : valid sequence");
+
+    if (peerSequence < maxSequence)
+    {
+        buildBlobInfos(blobInfos, lists);
+        sendValidatorList(
+            peer,
+            peerSequence,
+            publisherKey,
+            maxSequence,
+            lists.rawVersion,
+            lists.rawManifest,
+            blobInfos,
+            messages,
+            hashRouter,
+            j);
+
+        // Suppress the messages so they'll be ignored next time.
+        uint256 lasthash;
+        for (auto const& m : messages)
+        {
+            lasthash = m.hash;
+            hashRouter.addSuppressionPeer(lasthash, peer.id());
+        }
+        return std::make_tuple(
+            lists.rawManifest, lists.rawVersion, blobInfos, lasthash);
+    }
+    return {};
+}
+
 // static
-void
+std::optional<std::string>
 ValidatorList::sendValidatorList(
     Peer& peer,
     std::uint64_t peerSequence,
@@ -754,7 +808,7 @@ ValidatorList::sendValidatorList(
         : peer.supportsFeature(ProtocolFeature::ValidatorListPropagation) ? 1
                                                                           : 0;
     if (!messageVersion)
-        return;
+        return {};
     auto const [newPeerSequence, numVLs] = buildValidatorListMessages(
         messageVersion,
         peerSequence,
@@ -790,6 +844,7 @@ ValidatorList::sendValidatorList(
         if (sent)
         {
             if (messageVersion > 1)
+            {
                 JLOG(j.debug())
                     << "Sent " << messages.size()
                     << " validator list collection(s) containing " << numVLs
@@ -798,6 +853,8 @@ ValidatorList::sendValidatorList(
                     << newPeerSequence << " to "
                     << peer.getRemoteAddress().to_string() << " [" << peer.id()
                     << "]";
+                return "ValidatorListCollection";
+            }
             else
             {
                 XRPL_ASSERT(
@@ -809,13 +866,15 @@ ValidatorList::sendValidatorList(
                     << " with sequence " << newPeerSequence << " to "
                     << peer.getRemoteAddress().to_string() << " [" << peer.id()
                     << "]";
+                return "ValidatorList";
             }
         }
     }
+    return {};
 }
 
 // static
-void
+std::optional<std::string>
 ValidatorList::sendValidatorList(
     Peer& peer,
     std::uint64_t peerSequence,
@@ -828,7 +887,7 @@ ValidatorList::sendValidatorList(
     beast::Journal j)
 {
     std::vector<ValidatorList::MessageWithHash> messages;
-    sendValidatorList(
+    return sendValidatorList(
         peer,
         peerSequence,
         publisherKey,
@@ -899,7 +958,7 @@ ValidatorList::broadcastBlobs(
         std::map<std::size_t, ValidatorBlobInfo> blobInfos;
 
         XRPL_ASSERT(
-            lists.current.sequence == maxSequence ||
+            lists.current.sequence <= maxSequence ||
                 lists.remaining.count(maxSequence) == 1,
             "ripple::ValidatorList::broadcastBlobs : valid sequence");
         // Can't use overlay.foreach here because we need to modify

src/xrpld/overlay/detail/PeerImp.cpp

@@ -848,6 +848,46 @@ PeerImp::domain() const
 
 // Protocol logic
 
+void
+logVLBlob(beast::Journal j, ValidatorBlobInfo const& blob, std::size_t count)
+{
+    auto const stream = j.trace();
+    JLOG(stream) << "Blob " << count << " Signature: " << blob.signature;
+    JLOG(stream) << "Blob " << count << " blob: " << base64_decode(blob.blob);
+    JLOG(stream) << "Blob " << count << " manifest: "
+                 << (blob.manifest ? base64_decode(*blob.manifest) : "NONE");
+}
+
+void
+logVLBlob(
+    beast::Journal j,
+    std::pair<std::size_t, ValidatorBlobInfo> const& blob,
+    std::size_t count)
+{
+    logVLBlob(j, blob.second, count);
+}
+
+template <class TBlobs>
+void
+logVL(
+    beast::Journal j,
+    std::string const& manifest,
+    std::uint32_t version,
+    TBlobs const& blobs,
+    uint256 const& hash)
+{
+    auto const stream = j.trace();
+    JLOG(stream) << "Manifest: " << manifest;
+    JLOG(stream) << "Version: " << version;
+    JLOG(stream) << "Hash: " << hash;
+    std::size_t count = 1;
+    for (auto const& blob : blobs)
+    {
+        logVLBlob(j, blob, count);
+        ++count;
+    }
+}
+
 void
 PeerImp::doProtocolStart()
 {
@@ -2066,6 +2106,8 @@ PeerImp::onValidatorListMessage(
         return;
     }
 
+    logVL(p_journal_, manifest, version, blobs, hash);
+
     auto const applyResult = app_.validators().applyListsAndBroadcast(
         manifest,
         version,
@@ -2108,7 +2150,8 @@ PeerImp::onValidatorListMessage(
                     "ripple::PeerImp::onValidatorListMessage : lower sequence");
             }
 #endif
-            publisherListSequences_[pubKey] = applyResult.sequence;
+            if (publisherListSequences_[pubKey] < applyResult.sequence)
+                publisherListSequences_[pubKey] = applyResult.sequence;
         }
         break;
         case ListDisposition::same_sequence:
@@ -2127,8 +2170,37 @@ PeerImp::onValidatorListMessage(
         }
 #endif  // !NDEBUG
 
+            [[fallthrough]];
+        case ListDisposition::stale: {
+            auto const [pubKey, currentPeerSeq] = [&]() {
+                std::lock_guard<std::mutex> sl(recentLock_);
+                ASSERT(
+                    applyResult.sequence && applyResult.publisherKey,
+                    "ripple::PeerImp::onValidatorListMessage : (stale) nonzero "
+                    "sequence");
+                auto const& pubKey = *applyResult.publisherKey;
+                auto const& current = publisherListSequences_[pubKey];
+                ASSERT(
+                    current <= applyResult.sequence,
+                    "ripple::PeerImp::onValidatorListMessage : (stale) valid "
+                    "sequence");
+                return std::make_pair(
+                    pubKey, current ? current : applyResult.sequence);
+            }();
+            if (currentPeerSeq <= applyResult.sequence)
+            {
+                auto const [sentmanifest, sentversion, sentblobs, senthash] =
+                    app_.validators().sendLatestValidatorLists(
+                        *this,
+                        currentPeerSeq,
+                        pubKey,
+                        app_.getHashRouter(),
+                        p_journal_);
+                logVL(
+                    p_journal_, sentmanifest, sentversion, sentblobs, senthash);
+            }
+        }
         break;
-        case ListDisposition::stale:
         case ListDisposition::untrusted:
         case ListDisposition::invalid:
         case ListDisposition::unsupported_version:
@@ -2219,7 +2291,7 @@ PeerImp::onValidatorListMessage(
                 break;
             case ListDisposition::stale:
                 JLOG(p_journal_.warn())
-                    << "Ignored " << count << "stale " << messageType
+                    << "Ignored " << count << " stale " << messageType
                     << "(s) from peer " << remote_address_;
                 break;
             case ListDisposition::untrusted:
@@ -2229,12 +2301,12 @@ PeerImp::onValidatorListMessage(
                 break;
             case ListDisposition::unsupported_version:
                 JLOG(p_journal_.warn())
-                    << "Ignored " << count << "unsupported version "
+                    << "Ignored " << count << " unsupported version "
                     << messageType << "(s) from peer " << remote_address_;
                 break;
             case ListDisposition::invalid:
                 JLOG(p_journal_.warn())
-                    << "Ignored " << count << "invalid " << messageType
+                    << "Ignored " << count << " invalid " << messageType
                     << "(s) from peer " << remote_address_;
                 break;
             default: