Compare commits

..

38 Commits

Author SHA1 Message Date
Ed Hennis
be0baf1b53 Fix new header to use "#pragma once" 2026-07-10 11:41:11 -04:00
Ed Hennis
b394730bd0 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-07-09 19:50:02 -04:00
Ed Hennis
8290eb3024 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-19 16:53:50 -04:00
Ed Hennis
d537684df5 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-19 10:15:30 -04:00
Ed Hennis
4085361251 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-19 05:16:07 -04:00
Ed Hennis
1de7c130b0 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-15 21:43:33 -04:00
Ed Hennis
2a678a1157 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-14 20:39:23 -04:00
Ed Hennis
5c254aadf2 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-13 22:32:06 -04:00
Ed Hennis
ad2db1962e Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-13 12:03:54 -04:00
Ed Hennis
4dae405762 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-12 19:19:01 -04:00
Ed Hennis
b21b3596f8 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-12 16:26:24 -04:00
Ed Hennis
d648539b62 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-11 13:34:26 -04:00
Ed Hennis
50644c6d30 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-07 18:10:17 -04:00
Ed Hennis
88c248077e Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-07 14:18:43 -04:00
Ed Hennis
bad40d7e10 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-07 13:28:46 -04:00
Ed Hennis
74d0c375ec Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-06 22:34:37 -04:00
Ed Hennis
200ab3f694 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-06 14:18:16 -04:00
Ed Hennis
a7a4357022 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-05 16:46:58 -04:00
Ed Hennis
e0230a1e22 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-05 10:50:49 -04:00
Ed Hennis
5f1a98397c Merge remote-tracking branch 'XRPLF/develop' into ximinez/acquireAsyncDispatch
* XRPLF/develop:
  ci: Rewrite clang-tidy workflow(s) in a reusable manner (7062)
  chore: Ignore identifier-naming update in git blame (7066)
  refactor: Enable clang-tidy `readability-identifier-naming` check (6571)
2026-05-05 00:26:38 -04:00
Ed Hennis
c8ca624974 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-05-01 13:59:18 -04:00
Ed Hennis
4489cda841 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-04-28 16:23:47 -04:00
Ed Hennis
784774005e Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-04-25 14:45:58 -04:00
Ed Hennis
904b39a3d3 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-04-23 15:56:15 -04:00
Ed Hennis
eadfabfe4f Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-04-22 23:40:40 -04:00
Ed Hennis
d991f81e87 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-04-22 14:49:17 -04:00
Ed Hennis
a34a96e233 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-04-22 13:10:48 -04:00
Ed Hennis
a3a277518b Merge remote-tracking branch 'XRPLF/develop' into ximinez/acquireAsyncDispatch
* XRPLF/develop:
  chore: Add -fix to clang-tidy invocation (6990)
  chore: Remove empty Taker.h (6984)
  chore: Enable clang-tidy modernize checks (6975)
  ci: Upload clang-tidy git diff (6983)
  fix: Add rounding to Vault invariants (6217) (6955)
2026-04-21 18:54:46 -04:00
Ed Hennis
8438698c54 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-04-20 17:49:51 -04:00
Ed Hennis
0597047d79 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-04-20 15:45:08 -04:00
Ed Hennis
992c6f525d Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-04-20 11:39:10 -04:00
Ed Hennis
9a870c36e2 Merge remote-tracking branch 'XRPLF/develop' into ximinez/acquireAsyncDispatch
* XRPLF/develop:
  chore: Enable clang-tidy include cleaner (6947)
  fix: Change AMMClawback return code to tecNO_PERMISSION (6946)
  ci: [DEPENDABOT] bump actions/upload-pages-artifact from 4.0.0 to 5.0.0 (6927)
  ci: [DEPENDABOT] bump actions/upload-artifact from 7.0.0 to 7.0.1 (6928)
  chore: Enable clang-tidy readability checks (6930)
2026-04-17 18:13:20 -04:00
Ed Hennis
a4462e65bc Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-04-16 13:44:42 -04:00
Ed Hennis
f0596c0a5d Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-04-15 19:06:33 -04:00
Ed Hennis
dd99bc3ce8 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-04-15 14:29:00 -04:00
Ed Hennis
25cd7730e5 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-04-13 20:28:47 -04:00
Ed Hennis
8dcbcb22c2 Merge branch 'develop' into ximinez/acquireAsyncDispatch 2026-04-10 12:12:54 -04:00
Ed Hennis
f44dfc89cc refactor: acquireAsync will dispatch a job, not the other way around
- Improve job queue collision checks and logging
    - Improve logging related to ledger acquisition and operating mode
      changes
    - Class "CanProcess" to keep track of processing of distinct items
2026-04-10 10:31:21 -04:00
21 changed files with 272 additions and 731 deletions

View File

@@ -58,6 +58,11 @@ public:
cleanup_();
}
CanProcess(CanProcess const&) = delete;
CanProcess&
operator=(CanProcess const&) = delete;
explicit
operator bool() const
{

View File

@@ -44,7 +44,7 @@ template <class T>
inline void
maybeReverseBytes(T& t, std::true_type)
{
reverseBytes(t);
reverse_bytes(t);
}
template <class T, class Hasher>

View File

@@ -145,13 +145,6 @@ private:
return std::move(peers_);
}
/** Return set of peers waiting for reply. Leaves list unchanged. */
std::set<PeerShortID> const&
peekPeerSet()
{
return peers_;
}
/** Return seated relay time point if the message has been relayed */
[[nodiscard]] std::optional<Stopwatch::time_point>
relayed() const
@@ -183,20 +176,6 @@ private:
return true;
}
bool
shouldProcessForPeer(
PeerShortID peer,
Stopwatch::time_point now,
std::chrono::seconds interval)
{
if (peerProcessed_.contains(peer) && ((peerProcessed_[peer] + interval) > now))
return false;
// Peer may already be in the list, but adding it again doesn't hurt
addPeer(peer);
peerProcessed_[peer] = now;
return true;
}
private:
HashRouterFlags flags_ = HashRouterFlags::UNDEFINED;
std::set<PeerShortID> peers_;
@@ -204,7 +183,6 @@ private:
// than one flag needs to expire independently.
std::optional<Stopwatch::time_point> relayed_;
std::optional<Stopwatch::time_point> processed_;
std::map<PeerShortID, Stopwatch::time_point> peerProcessed_;
};
public:
@@ -244,15 +222,6 @@ public:
HashRouterFlags& flags,
std::chrono::seconds txInterval);
/** Determines whether the hashed item should be processed for the given
peer. Could be an incoming or outgoing message.
Items filtered with this function should only be processed for the given
peer once. Unlike shouldProcess, it can be processed for other peers.
*/
bool
shouldProcessForPeer(uint256 const& key, PeerShortID peer, std::chrono::seconds interval);
/** Set the flags on a hash.
@return `true` if the flags were changed. `false` if unchanged.
@@ -278,11 +247,6 @@ public:
std::optional<std::set<PeerShortID>>
shouldRelay(uint256 const& key);
/** Returns a copy of the set of peers in the Entry for the key
*/
std::set<PeerShortID>
getPeers(uint256 const& key);
private:
// pair.second indicates whether the entry was created
std::pair<Entry&, bool>

View File

@@ -288,18 +288,8 @@ message TMLedgerData {
required uint32 ledgerSeq = 2;
required TMLedgerInfoType type = 3;
repeated TMLedgerNode nodes = 4;
// If the peer supports "responseCookies", this field will
// never be populated.
optional uint32 requestCookie = 5;
optional TMReplyError error = 6;
// The old field is called "requestCookie", but this is
// a response, so this name makes more sense
repeated uint32 responseCookies = 7;
// If a TMGetLedger request was received without a "requestCookie",
// and the peer supports it, this flag will be set to true to
// indicate that the receiver should process the result in addition
// to forwarding it to its "responseCookies" peers.
optional bool directResponse = 8;
}
message TMPing {

View File

@@ -82,19 +82,6 @@ HashRouter::shouldProcess(
return s.shouldProcess(suppressionMap_.clock().now(), txInterval);
}
bool
HashRouter::shouldProcessForPeer(
uint256 const& key,
PeerShortID peer,
std::chrono::seconds interval)
{
std::lock_guard lock(mutex_);
auto& entry = emplace(key).first;
return entry.shouldProcessForPeer(peer, suppressionMap_.clock().now(), interval);
}
HashRouterFlags
HashRouter::getFlags(uint256 const& key)
{
@@ -132,13 +119,4 @@ HashRouter::shouldRelay(uint256 const& key) -> std::optional<std::set<PeerShortI
return s.releasePeerSet();
}
auto
HashRouter::getPeers(uint256 const& key) -> std::set<PeerShortID>
{
std::lock_guard lock(mutex_);
auto& s = emplace(key).first;
return s.peekPeerSet();
}
} // namespace xrpl

View File

@@ -395,33 +395,6 @@ class HashRouter_test : public beast::unit_test::Suite
BEAST_EXPECT(!any(HF::UNDEFINED));
}
void
testProcessPeer()
{
using namespace std::chrono_literals;
TestStopwatch stopwatch;
HashRouter router(getSetup(5s, 5s), stopwatch);
uint256 const key(1);
HashRouter::PeerShortID peer1 = 1;
HashRouter::PeerShortID peer2 = 2;
auto const timeout = 2s;
BEAST_EXPECT(router.shouldProcessForPeer(key, peer1, timeout));
BEAST_EXPECT(!router.shouldProcessForPeer(key, peer1, timeout));
++stopwatch;
BEAST_EXPECT(!router.shouldProcessForPeer(key, peer1, timeout));
BEAST_EXPECT(router.shouldProcessForPeer(key, peer2, timeout));
BEAST_EXPECT(!router.shouldProcessForPeer(key, peer2, timeout));
++stopwatch;
BEAST_EXPECT(router.shouldProcessForPeer(key, peer1, timeout));
BEAST_EXPECT(!router.shouldProcessForPeer(key, peer2, timeout));
++stopwatch;
BEAST_EXPECT(router.shouldProcessForPeer(key, peer2, timeout));
++stopwatch;
BEAST_EXPECT(router.shouldProcessForPeer(key, peer1, timeout));
BEAST_EXPECT(!router.shouldProcessForPeer(key, peer2, timeout));
}
public:
void
run() override
@@ -434,7 +407,6 @@ public:
testProcess();
testSetup();
testFlagsOps();
testProcessPeer();
}
};

View File

@@ -175,7 +175,12 @@ public:
}
void
acquireAsync(uint256 const& hash, std::uint32_t seq, InboundLedger::Reason reason) override
acquireAsync(
JobType type,
std::string const& name,
uint256 const& hash,
std::uint32_t seq,
InboundLedger::Reason reason) override
{
}
@@ -384,11 +389,6 @@ public:
{
return false;
}
std::set<std::optional<uint64_t>>
releaseRequestCookies(uint256 const& requestHash) override
{
return {};
}
[[nodiscard]] std::string const&
fingerprint() const override

View File

@@ -0,0 +1,165 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2012-2016 Ripple Labs Inc.
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 <xrpl/basics/CanProcess.h>
#include <xrpl/beast/unit_test.h>
#include <memory>
namespace ripple {
namespace test {
struct CanProcess_test : beast::unit_test::Suite
{
template <class Mutex, class Collection, class Item>
void
test(
std::string const& name,
Mutex& mtx,
Collection& collection,
std::vector<Item> const& items)
{
testcase(name);
if (!BEAST_EXPECT(!items.empty()))
return;
if (!BEAST_EXPECT(collection.empty()))
return;
// CanProcess objects can't be copied or moved. To make that easier,
// store shared_ptrs
std::vector<std::shared_ptr<CanProcess>> trackers;
// Fill up the vector with two CanProcess for each Item. The first
// inserts the item into the collection and is "good". The second does
// not and is "bad".
for (int i = 0; i < items.size(); ++i)
{
{
auto const& good =
trackers.emplace_back(std::make_shared<CanProcess>(mtx, collection, items[i]));
BEAST_EXPECT(*good);
}
BEAST_EXPECT(trackers.size() == (2 * i) + 1);
BEAST_EXPECT(collection.size() == i + 1);
{
auto const& bad =
trackers.emplace_back(std::make_shared<CanProcess>(mtx, collection, items[i]));
BEAST_EXPECT(!*bad);
}
BEAST_EXPECT(trackers.size() == 2 * (i + 1));
BEAST_EXPECT(collection.size() == i + 1);
}
BEAST_EXPECT(collection.size() == items.size());
// Now remove the items from the vector<CanProcess> two at a time, and
// try to get another CanProcess for that item.
for (int i = 0; i < items.size(); ++i)
{
// Remove the "bad" one in the second position
// This will have no effect on the collection
{
auto const iter = trackers.begin() + 1;
BEAST_EXPECT(!**iter);
trackers.erase(iter);
}
BEAST_EXPECT(trackers.size() == (2 * items.size()) - 1);
BEAST_EXPECT(collection.size() == items.size());
{
// Append a new "bad" one
auto const& bad =
trackers.emplace_back(std::make_shared<CanProcess>(mtx, collection, items[i]));
BEAST_EXPECT(!*bad);
}
BEAST_EXPECT(trackers.size() == 2 * items.size());
BEAST_EXPECT(collection.size() == items.size());
// Remove the "good" one from the front
{
auto const iter = trackers.begin();
BEAST_EXPECT(**iter);
trackers.erase(iter);
}
BEAST_EXPECT(trackers.size() == (2 * items.size()) - 1);
BEAST_EXPECT(collection.size() == items.size() - 1);
{
// Append a new "good" one
auto const& good =
trackers.emplace_back(std::make_shared<CanProcess>(mtx, collection, items[i]));
BEAST_EXPECT(*good);
}
BEAST_EXPECT(trackers.size() == 2 * items.size());
BEAST_EXPECT(collection.size() == items.size());
}
// Now remove them all two at a time
for (int i = items.size() - 1; i >= 0; --i)
{
// Remove the "bad" one from the front
{
auto const iter = trackers.begin();
BEAST_EXPECT(!**iter);
trackers.erase(iter);
}
BEAST_EXPECT(trackers.size() == (2 * i) + 1);
BEAST_EXPECT(collection.size() == i + 1);
// Remove the "good" one now in front
{
auto const iter = trackers.begin();
BEAST_EXPECT(**iter);
trackers.erase(iter);
}
BEAST_EXPECT(trackers.size() == 2 * i);
BEAST_EXPECT(collection.size() == i);
}
BEAST_EXPECT(trackers.empty());
BEAST_EXPECT(collection.empty());
}
void
run() override
{
{
std::mutex m;
std::set<int> collection;
std::vector<int> const items{1, 2, 3, 4, 5};
test("set of int", m, collection, items);
}
{
std::mutex m;
std::set<std::string> collection;
std::vector<std::string> const items{"one", "two", "three", "four", "five"};
test("set of string", m, collection, items);
}
{
std::mutex m;
std::unordered_set<char> collection;
std::vector<char> const items{'1', '2', '3', '4', '5'};
test("unorderd_set of char", m, collection, items);
}
{
std::mutex m;
std::unordered_set<std::uint64_t> collection;
std::vector<std::uint64_t> const items{100u, 1000u, 150u, 4u, 0u};
test("unordered_set of uint64_t", m, collection, items);
}
}
};
BEAST_DEFINE_TESTSUITE(CanProcess, ripple_basics, ripple);
} // namespace test
} // namespace ripple

View File

@@ -63,8 +63,8 @@ public:
negotiateProtocolVersion("RTXP/1.2, XRPL/2.0, XRPL/2.1") == makeProtocol(2, 1));
BEAST_EXPECT(negotiateProtocolVersion("XRPL/2.2") == makeProtocol(2, 2));
BEAST_EXPECT(
negotiateProtocolVersion("RTXP/1.2, XRPL/2.2, XRPL/2.3, XRPL/2.4, XRPL/999.999") ==
makeProtocol(2, 3));
negotiateProtocolVersion("RTXP/1.2, XRPL/2.2, XRPL/2.3, XRPL/999.999") ==
makeProtocol(2, 2));
BEAST_EXPECT(negotiateProtocolVersion("XRPL/999.999, WebSocket/1.0") == std::nullopt);
BEAST_EXPECT(negotiateProtocolVersion("") == std::nullopt);
}

View File

@@ -190,11 +190,6 @@ public:
removeTxQueue(uint256 const&) override
{
}
std::set<std::optional<uint64_t>>
releaseRequestCookies(uint256 const& requestHash) override
{
return {};
}
};
/** Manually advanced clock. */

View File

@@ -159,10 +159,8 @@ RCLConsensus::Adaptor::acquireLedger(LedgerHash const& hash)
// Tell the ledger acquire system that we need the consensus ledger
acquiringLedger_ = hash;
app_.getJobQueue().addJob(JtAdvance, "GetConsL1", [id = hash, &app = app_, this]() {
JLOG(j_.debug()) << "JOB advanceLedger getConsensusLedger1 started";
app.getInboundLedgers().acquireAsync(id, 0, InboundLedger::Reason::CONSENSUS);
});
app_.getInboundLedgers().acquireAsync(
JtAdvance, "GetConsL1", hash, 0, InboundLedger::Reason::CONSENSUS);
}
return std::nullopt;
}

View File

@@ -130,12 +130,8 @@ RCLValidationsAdaptor::acquire(LedgerHash const& hash)
{
JLOG(j_.warn()) << "Need validated ledger for preferred ledger analysis " << hash;
Application* pApp = &app_;
app_.getJobQueue().addJob(JtAdvance, "GetConsL2", [pApp, hash, this]() {
JLOG(j_.debug()) << "JOB advanceLedger getConsensusLedger2 started";
pApp->getInboundLedgers().acquireAsync(hash, 0, InboundLedger::Reason::CONSENSUS);
});
app_.getInboundLedgers().acquireAsync(
JtAdvance, "GetConsL2", hash, 0, InboundLedger::Reason::CONSENSUS);
return std::nullopt;
}

View File

@@ -189,22 +189,4 @@ private:
std::unique_ptr<PeerSet> peerSet_;
};
inline std::string
to_string(InboundLedger::Reason reason)
{
using enum InboundLedger::Reason;
switch (reason)
{
case HISTORY:
return "HISTORY";
case GENERIC:
return "GENERIC";
case CONSENSUS:
return "CONSENSUS";
default:
UNREACHABLE("ripple::to_string(InboundLedger::Reason) : unknown value");
return "unknown";
}
}
} // namespace xrpl

View File

@@ -40,7 +40,12 @@ public:
// Queue. TODO review whether all callers of acquire() can use this
// instead. Inbound ledger acquisition is asynchronous anyway.
virtual void
acquireAsync(uint256 const& hash, std::uint32_t seq, InboundLedger::Reason reason) = 0;
acquireAsync(
JobType type,
std::string const& name,
uint256 const& hash,
std::uint32_t seq,
InboundLedger::Reason reason) = 0;
virtual std::shared_ptr<InboundLedger>
find(LedgerHash const& hash) = 0;

View File

@@ -79,78 +79,12 @@ public:
XRPL_ASSERT(
hash.isNonZero(), "xrpl::InboundLedgersImp::acquire::doAcquire : nonzero hash");
bool const needNetworkLedger = app_.getOPs().isNeedNetworkLedger();
bool const shouldAcquire = [&]() {
if (!needNetworkLedger)
return true;
if (reason == InboundLedger::Reason::GENERIC)
return true;
if (reason == InboundLedger::Reason::CONSENSUS)
return true;
return false;
}();
// probably not the right rule
if (app_.getOPs().isNeedNetworkLedger() && (reason != InboundLedger::Reason::GENERIC) &&
(reason != InboundLedger::Reason::CONSENSUS))
return {};
std::stringstream ss;
ss << "InboundLedger::acquire: "
<< "Request: " << to_string(hash) << ", " << seq
<< " NeedNetworkLedger: " << (needNetworkLedger ? "yes" : "no")
<< " Reason: " << to_string(reason)
<< " Should acquire: " << (shouldAcquire ? "true." : "false.");
/* Acquiring ledgers is somewhat expensive. It requires lots of
* computation and network communication. Avoid it when it's not
* appropriate. Every validation from a peer for a ledger that
* we do not have locally results in a call to this function: even
* if we are moments away from validating the same ledger.
*/
bool const shouldBroadcast = [&]() {
// If the node is not in "full" state, it needs to sync to
// the network, and doesn't have the necessary tx's and
// ledger entries to build the ledger.
bool const isFull = app_.getOPs().isFull();
// If everything else is ok, don't try to acquire the ledger
// if the requested seq is in the near future relative to
// the validated ledger. If the requested ledger is between
// 1 and 19 inclusive ledgers ahead of the valid ledger this
// node has not built it yet, but it's possible/likely it
// has the tx's necessary to build it and get caught up.
// Plus it might not become validated. On the other hand, if
// it's more than 20 in the future, this node should request
// it so that it can jump ahead and get caught up.
LedgerIndex const validSeq = app_.getLedgerMaster().getValidLedgerIndex();
constexpr std::size_t lagLeeway = 20;
bool const nearFuture = (seq > validSeq) && (seq < validSeq + lagLeeway);
// If everything else is ok, don't try to acquire the ledger
// if the request is related to consensus. (Note that
// consensus calls usually pass a seq of 0, so nearFuture
// will be false other than on a brand new network.)
bool const consensus = reason == InboundLedger::Reason::CONSENSUS;
ss << " Evaluating whether to broadcast requests to peers"
<< ". full: " << (isFull ? "true" : "false") << ". ledger sequence " << seq
<< ". Valid sequence: " << validSeq << ". Lag leeway: " << lagLeeway
<< ". request for near future ledger: " << (nearFuture ? "true" : "false")
<< ". Consensus: " << (consensus ? "true" : "false");
// If the node is not synced, send requests.
if (!isFull)
return true;
// If the ledger is in the near future, do NOT send requests.
// This node is probably about to build it.
if (nearFuture)
return false;
// If the request is because of consensus, do NOT send requests.
// This node is probably about to build it.
if (consensus)
return false;
return true;
}();
ss << ". Would broadcast to peers? " << (shouldBroadcast ? "true." : "false.");
if (!shouldAcquire)
{
JLOG(j_.debug()) << "Abort(rule): " << ss.str();
return {};
}
bool isNew = true;
std::shared_ptr<InboundLedger> inbound;
@@ -202,25 +136,35 @@ public:
}
void
acquireAsync(uint256 const& hash, std::uint32_t seq, InboundLedger::Reason reason) override
acquireAsync(
JobType type,
std::string const& name,
uint256 const& hash,
std::uint32_t seq,
InboundLedger::Reason reason) override
{
if (CanProcess const check{acquiresMutex_, pendingAcquires_, hash})
if (auto check = std::make_shared<CanProcess const>(acquiresMutex_, pendingAcquires_, hash);
*check)
{
try
{
acquire(hash, seq, reason);
}
catch (std::exception const& e)
{
JLOG(j_.warn()) << "Exception thrown for acquiring new inbound ledger " << hash
<< ": " << e.what();
}
catch (...)
{
JLOG(j_.warn()) << "Unknown exception thrown for acquiring new "
"inbound ledger "
<< hash;
}
app_.getJobQueue().addJob(type, name, [check, name, hash, seq, reason, this]() {
JLOG(j_.debug()) << "JOB acquireAsync " << name << " started ";
try
{
acquire(hash, seq, reason);
}
catch (std::exception const& e)
{
JLOG(j_.warn()) << "Exception thrown for acquiring new "
"inbound ledger "
<< hash << ": " << e.what();
}
catch (...)
{
JLOG(j_.warn()) << "Unknown exception thrown for acquiring new "
"inbound ledger "
<< hash;
}
});
}
}

View File

@@ -23,7 +23,6 @@ enum class ProtocolFeature {
ValidatorListPropagation,
ValidatorList2Propagation,
LedgerReplay,
LedgerDataCookies
};
/** Represents a peer connection in the overlay. */
@@ -123,13 +122,6 @@ public:
[[nodiscard]] virtual bool
txReduceRelayEnabled() const = 0;
//
// Messages
//
virtual std::set<std::optional<uint64_t>>
releaseRequestCookies(uint256 const& requestHash) = 0;
};
} // namespace xrpl

View File

@@ -111,9 +111,6 @@ constexpr std::chrono::milliseconds kPeerHighLatency{300};
/** How often we PING the peer to check for latency and sendq probe */
constexpr std::chrono::seconds kPeerTimerInterval{60};
/** How often we process duplicate incoming TMGetLedger messages */
std::chrono::seconds constexpr kGetLedgerInterval{15};
} // namespace
// TODO: Remove this exclusion once unit tests are added after the hotfix
@@ -544,8 +541,6 @@ PeerImp::supportsFeature(ProtocolFeature f) const
return protocol_ >= makeProtocol(2, 2);
case ProtocolFeature::LedgerReplay:
return ledgerReplayEnabled_;
case ProtocolFeature::LedgerDataCookies:
return protocol_ >= makeProtocol(2, 3);
}
return false;
}
@@ -1415,9 +1410,8 @@ PeerImp::handleTransaction(
void
PeerImp::onMessage(std::shared_ptr<protocol::TMGetLedger> const& m)
{
auto badData = [&](std::string const& msg, bool chargefee = true) {
if (chargefee)
fee_.update(Resource::kFeeInvalidData, "get_ledger " + msg);
auto badData = [&](std::string const& msg) {
fee_.update(Resource::kFeeInvalidData, "get_ledger " + msg);
JLOG(pJournal_.warn()) << "TMGetLedger: " << msg;
};
auto const itype{m->itype()};
@@ -1515,68 +1509,11 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMGetLedger> const& m)
}
}
// Drop duplicate requests from the same peer for at least
// `kGetLedgerInterval` seconds.
// Append a little junk to prevent the hash of an incoming messsage
// from matching the hash of the same outgoing message.
// `shouldProcessForPeer` does not distingish between incoming and
// outgoing, and some of the message relay logic checks the hash to see
// if the message has been relayed already. If the hashes are the same,
// a duplicate will be detected when sending the message is attempted,
// so it will fail.
auto const messageHash = sha512Half(*m, nullptr);
// Request cookies are not included in the hash. Track them here.
auto const requestCookie = [&m]() -> std::optional<uint64_t> {
if (m->has_requestcookie())
return m->requestcookie();
return std::nullopt;
}();
auto const [inserted, pending] = [&] {
std::lock_guard lock{cookieLock_};
auto& cookies = messageRequestCookies_[messageHash];
bool const pending = !cookies.empty();
return std::pair{cookies.emplace(requestCookie).second, pending};
}();
// Check if the request has been seen from this peer.
if (!app_.getHashRouter().shouldProcessForPeer(messageHash, id_, kGetLedgerInterval))
{
// This request has already been seen from this peer.
// Has it been seen with this request cookie (or lack thereof)?
if (inserted)
{
// This is a duplicate request, but with a new cookie. When a
// response is ready, one will be sent for each request cookie.
JLOG(pJournal_.debug())
<< "TMGetLedger: duplicate request with new request cookie: "
<< requestCookie.value_or(0) << ". Job pending: " << (pending ? "yes" : "no")
<< ": " << messageHash;
if (pending)
{
// Don't bother queueing up a new job if other requests are
// already pending. This should limit entries in the job queue
// to one per peer per unique request.
JLOG(pJournal_.debug()) << "TMGetLedger: Suppressing recvGetLedger job, since one "
"is pending: "
<< messageHash;
return;
}
}
else
{
// Don't punish nodes that don't know any better
return badData(
"duplicate request: " + to_string(messageHash),
supportsFeature(ProtocolFeature::LedgerDataCookies));
}
}
// Queue a job to process the request
JLOG(pJournal_.debug()) << "TMGetLedger: Adding recvGetLedger job: " << messageHash;
std::weak_ptr<PeerImp> const weak = shared_from_this();
app_.getJobQueue().addJob(JtLedgerReq, "RcvGetLedger", [weak, m, messageHash]() {
app_.getJobQueue().addJob(JtLedgerReq, "RcvGetLedger", [weak, m]() {
if (auto peer = weak.lock())
peer->processLedgerRequest(m, messageHash);
peer->processLedgerRequest(m);
});
}
@@ -1683,9 +1620,8 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMReplayDeltaResponse> const& m)
void
PeerImp::onMessage(std::shared_ptr<protocol::TMLedgerData> const& m)
{
auto badData = [&](std::string const& msg, bool charge = true) {
if (charge)
fee_.update(Resource::kFeeInvalidData, msg);
auto badData = [&](std::string const& msg) {
fee_.update(Resource::kFeeInvalidData, msg);
JLOG(pJournal_.warn()) << "TMLedgerData: " << msg;
};
@@ -1742,93 +1678,19 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMLedgerData> const& m)
return;
}
auto const messageHash = sha512Half(*m);
if (!app_.getHashRouter().addSuppressionPeer(messageHash, id_))
// If there is a request cookie, attempt to relay the message
if (m->has_requestcookie())
{
// Don't punish nodes that don't know any better
return badData(
"Duplicate message: " + to_string(messageHash),
supportsFeature(ProtocolFeature::LedgerDataCookies));
}
bool const routed =
m->has_directresponse() || m->responsecookies_size() || m->has_requestcookie();
{
// Check if this message needs to be forwarded to one or more peers.
// Maximum of one of the relevant fields should be populated.
XRPL_ASSERT(
!m->has_requestcookie() || !m->responsecookies_size(),
"ripple::PeerImp::onMessage(TMLedgerData) : valid cookie fields");
// Make a copy of the response cookies, then wipe the list so it can be
// forwarded cleanly
auto const responseCookies = m->responsecookies();
m->clear_responsecookies();
// Flag indicating if this response should be processed locally,
// possibly in addition to being forwarded.
bool const directResponse = m->has_directresponse() && m->directresponse();
m->clear_directresponse();
auto const relay = [this, m, &messageHash](auto const cookie) {
if (auto peer = overlay_.findPeerByShortID(cookie))
{
XRPL_ASSERT(
!m->has_requestcookie() && !m->responsecookies_size(),
"ripple::PeerImp::onMessage(TMLedgerData) relay : no "
"cookies");
if (peer->supportsFeature(ProtocolFeature::LedgerDataCookies))
// Setting this flag is not _strictly_ necessary for peers
// that support it if there are no cookies included in the
// message, but it is more accurate.
m->set_directresponse(true);
else
m->clear_directresponse();
peer->send(std::make_shared<Message>(*m, protocol::mtLEDGER_DATA));
}
else
JLOG(pJournal_.info()) << "Unable to route TX/ledger data reply to peer [" << cookie
<< "]: " << messageHash;
};
// If there is a request cookie, attempt to relay the message
if (m->has_requestcookie())
if (auto peer = overlay_.findPeerByShortID(m->requestcookie()))
{
XRPL_ASSERT(
responseCookies.empty(),
"ripple::PeerImp::onMessage(TMLedgerData) : no response "
"cookies");
m->clear_requestcookie();
relay(m->requestcookie());
if (!directResponse && responseCookies.empty())
return;
peer->send(std::make_shared<Message>(*m, protocol::mtLEDGER_DATA));
}
// If there's a list of request cookies, attempt to relay the message to
// all of them.
if (responseCookies.size())
else
{
for (auto const cookie : responseCookies)
relay(cookie);
if (!directResponse)
return;
}
}
// Now that any forwarding is done check the base message (data only, no
// routing info for duplicates)
if (routed)
{
m->clear_directresponse();
XRPL_ASSERT(
!m->has_requestcookie() && !m->responsecookies_size(),
"ripple::PeerImp::onMessage(TMLedgerData) : no cookies");
auto const baseMessageHash = sha512Half(*m);
if (!app_.getHashRouter().addSuppressionPeer(baseMessageHash, id_))
{
// Don't punish nodes that don't know any better
return badData(
"Duplicate message: " + to_string(baseMessageHash),
supportsFeature(ProtocolFeature::LedgerDataCookies));
JLOG(pJournal_.info()) << "Unable to route TX/ledger data reply";
}
return;
}
uint256 const ledgerHash = uint256::fromRaw(m->ledgerhash());
@@ -3219,21 +3081,16 @@ PeerImp::checkValidation(
// the TX tree with the specified root hash.
//
static std::shared_ptr<PeerImp>
getPeerWithTree(
OverlayImpl& ov,
uint256 const& rootHash,
PeerImp const* skip,
std::function<bool(Peer::id_t)> shouldProcessCallback)
getPeerWithTree(OverlayImpl& ov, uint256 const& rootHash, PeerImp const* skip)
{
std::shared_ptr<PeerImp> ret;
int retScore = 0;
XRPL_ASSERT(shouldProcessCallback, "ripple::getPeerWithTree : callback provided");
ov.forEach([&](std::shared_ptr<PeerImp>&& p) {
if (p->hasTxSet(rootHash) && p.get() != skip)
{
auto score = p->getScore(true);
if (!ret || (score > retScore && shouldProcessCallback(p->id())))
if (!ret || (score > retScore))
{
ret = std::move(p);
retScore = score;
@@ -3252,18 +3109,16 @@ getPeerWithLedger(
OverlayImpl& ov,
uint256 const& ledgerHash,
LedgerIndex ledger,
PeerImp const* skip,
std::function<bool(Peer::id_t)> shouldProcessCallback)
PeerImp const* skip)
{
std::shared_ptr<PeerImp> ret;
int retScore = 0;
XRPL_ASSERT(shouldProcessCallback, "ripple::getPeerWithLedger : callback provided");
ov.forEach([&](std::shared_ptr<PeerImp>&& p) {
if (p->hasLedger(ledgerHash, ledger) && p.get() != skip)
{
auto score = p->getScore(true);
if (!ret || (score > retScore && shouldProcessCallback(p->id())))
if (!ret || (score > retScore))
{
ret = std::move(p);
retScore = score;
@@ -3277,8 +3132,7 @@ getPeerWithLedger(
void
PeerImp::sendLedgerBase(
std::shared_ptr<Ledger const> const& ledger,
protocol::TMLedgerData& ledgerData,
PeerCookieMap const& destinations)
protocol::TMLedgerData& ledgerData)
{
JLOG(pJournal_.trace()) << "sendLedgerBase: Base data";
@@ -3308,92 +3162,14 @@ PeerImp::sendLedgerBase(
}
}
sendToMultiple(ledgerData, destinations);
}
void
PeerImp::sendToMultiple(protocol::TMLedgerData& ledgerData, PeerCookieMap const& destinations)
{
bool foundSelf = false;
for (auto const& [peer, cookies] : destinations)
{
if (peer.get() == this)
foundSelf = true;
bool const multipleCookies = peer->supportsFeature(ProtocolFeature::LedgerDataCookies);
std::vector<std::uint64_t> sendCookies;
bool directResponse = false;
if (!multipleCookies)
{
JLOG(pJournal_.debug()) << "sendToMultiple: Sending " << cookies.size()
<< " TMLedgerData messages to peer [" << peer->id()
<< "]: " << sha512Half(ledgerData);
}
for (auto const& cookie : cookies)
{
// Unfortunately, need a separate Message object for every
// combination
if (cookie)
{
if (multipleCookies)
{
// Save this one for later to send a single message
sendCookies.emplace_back(*cookie);
continue;
}
// Feature not supported, so send a single message with a
// single cookie
ledgerData.set_requestcookie(*cookie);
}
else
{
if (multipleCookies)
{
// Set this flag later on the single message
directResponse = true;
continue;
}
ledgerData.clear_requestcookie();
}
XRPL_ASSERT(
!multipleCookies,
"ripple::PeerImp::sendToMultiple : ledger data cookies "
"unsupported");
auto message{std::make_shared<Message>(ledgerData, protocol::mtLEDGER_DATA)};
peer->send(message);
}
if (multipleCookies)
{
// Send a single message with all the cookies and/or the direct
// response flag, so the receiver can farm out the single message to
// multiple peers and/or itself
XRPL_ASSERT(
sendCookies.size() || directResponse,
"ripple::PeerImp::sendToMultiple : valid response options");
ledgerData.clear_requestcookie();
ledgerData.clear_responsecookies();
ledgerData.set_directresponse(directResponse);
for (auto const& cookie : sendCookies)
ledgerData.add_responsecookies(cookie);
auto message{std::make_shared<Message>(ledgerData, protocol::mtLEDGER_DATA)};
peer->send(message);
JLOG(pJournal_.debug())
<< "sendToMultiple: Sent 1 TMLedgerData message to peer [" << peer->id()
<< "]: including " << (directResponse ? "the direct response flag and " : "")
<< sendCookies.size() << " response cookies. "
<< ": " << sha512Half(ledgerData);
}
}
XRPL_ASSERT(foundSelf, "ripple::PeerImp::sendToMultiple : current peer included");
auto message{std::make_shared<Message>(ledgerData, protocol::mtLEDGER_DATA)};
send(message);
}
std::shared_ptr<Ledger const>
PeerImp::getLedger(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 const& mHash)
PeerImp::getLedger(std::shared_ptr<protocol::TMGetLedger> const& m)
{
JLOG(pJournal_.trace()) << "getLedger: Ledger " << mHash;
JLOG(pJournal_.trace()) << "getLedger: Ledger";
std::shared_ptr<Ledger const> ledger;
@@ -3409,30 +3185,16 @@ PeerImp::getLedger(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 cons
if (m->has_querytype() && !m->has_requestcookie())
{
// Attempt to relay the request to a peer
// Note repeated messages will not relay to the same peer
// before `kGetLedgerInterval` seconds. This prevents one
// peer from getting flooded, and distributes the request
// load. If a request has been relayed to all eligible
// peers, then this message will not be relayed.
if (auto const peer = getPeerWithLedger(
overlay_,
ledgerHash,
m->has_ledgerseq() ? m->ledgerseq() : 0,
this,
[&](Peer::id_t id) {
return app_.getHashRouter().shouldProcessForPeer(
mHash, id, kGetLedgerInterval);
}))
overlay_, ledgerHash, m->has_ledgerseq() ? m->ledgerseq() : 0, this))
{
m->set_requestcookie(id());
peer->send(std::make_shared<Message>(*m, protocol::mtGET_LEDGER));
JLOG(pJournal_.debug())
<< "getLedger: Request relayed to peer [" << peer->id() << "]: " << mHash;
JLOG(pJournal_.debug()) << "getLedger: Request relayed to peer";
return ledger;
}
JLOG(pJournal_.trace())
<< "getLedger: Don't have ledger with hash " << ledgerHash << ": " << mHash;
JLOG(pJournal_.trace()) << "getLedger: Failed to find peer to relay request";
}
}
}
@@ -3441,15 +3203,15 @@ PeerImp::getLedger(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 cons
// Attempt to find ledger by sequence
if (m->ledgerseq() < app_.getLedgerMaster().getEarliestFetch())
{
JLOG(pJournal_.debug()) << "getLedger: Early ledger sequence request " << mHash;
JLOG(pJournal_.debug()) << "getLedger: Early ledger sequence request";
}
else
{
ledger = app_.getLedgerMaster().getLedgerBySeq(m->ledgerseq());
if (!ledger)
{
JLOG(pJournal_.debug()) << "getLedger: Don't have ledger with sequence "
<< m->ledgerseq() << ": " << mHash;
JLOG(pJournal_.debug())
<< "getLedger: Don't have ledger with sequence " << m->ledgerseq();
}
}
}
@@ -3471,29 +3233,27 @@ PeerImp::getLedger(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 cons
charge(Resource::kFeeMalformedRequest, "get_ledger ledgerSeq");
ledger.reset();
JLOG(pJournal_.warn())
<< "getLedger: Invalid ledger sequence " << ledgerSeq << ": " << mHash;
JLOG(pJournal_.warn()) << "getLedger: Invalid ledger sequence " << ledgerSeq;
}
}
else if (ledgerSeq < app_.getLedgerMaster().getEarliestFetch())
{
ledger.reset();
JLOG(pJournal_.debug())
<< "getLedger: Early ledger sequence request " << ledgerSeq << ": " << mHash;
JLOG(pJournal_.debug()) << "getLedger: Early ledger sequence request " << ledgerSeq;
}
}
else
{
JLOG(pJournal_.debug()) << "getLedger: Unable to find ledger " << mHash;
JLOG(pJournal_.debug()) << "getLedger: Unable to find ledger";
}
return ledger;
}
std::shared_ptr<SHAMap const>
PeerImp::getTxSet(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 const& mHash) const
PeerImp::getTxSet(std::shared_ptr<protocol::TMGetLedger> const& m) const
{
JLOG(pJournal_.trace()) << "getTxSet: TX set " << mHash;
JLOG(pJournal_.trace()) << "getTxSet: TX set";
uint256 const txSetHash = uint256::fromRaw(m->ledgerhash());
std::shared_ptr<SHAMap> shaMap{app_.getInboundTransactions().getSet(txSetHash, false)};
@@ -3502,28 +3262,20 @@ PeerImp::getTxSet(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 const
if (m->has_querytype() && !m->has_requestcookie())
{
// Attempt to relay the request to a peer
// Note repeated messages will not relay to the same peer
// before `kGetLedgerInterval` seconds. This prevents one
// peer from getting flooded, and distributes the request
// load. If a request has been relayed to all eligible
// peers, then this message will not be relayed.
if (auto const peer = getPeerWithTree(overlay_, txSetHash, this, [&](Peer::id_t id) {
return app_.getHashRouter().shouldProcessForPeer(mHash, id, kGetLedgerInterval);
}))
if (auto const peer = getPeerWithTree(overlay_, txSetHash, this))
{
m->set_requestcookie(id());
peer->send(std::make_shared<Message>(*m, protocol::mtGET_LEDGER));
JLOG(pJournal_.debug())
<< "getTxSet: Request relayed to peer [" << peer->id() << "]: " << mHash;
JLOG(pJournal_.debug()) << "getTxSet: Request relayed";
}
else
{
JLOG(pJournal_.debug()) << "getTxSet: Failed to find relay peer: " << mHash;
JLOG(pJournal_.debug()) << "getTxSet: Failed to find relay peer";
}
}
else
{
JLOG(pJournal_.debug()) << "getTxSet: Failed to find TX set " << mHash;
JLOG(pJournal_.debug()) << "getTxSet: Failed to find TX set";
}
}
@@ -3531,7 +3283,7 @@ PeerImp::getTxSet(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 const
}
void
PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 const& mHash)
PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m)
{
// Do not resource charge a peer responding to a relay
if (!m->has_requestcookie())
@@ -3544,72 +3296,9 @@ PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m, u
bool fatLeaves{true};
auto const itype{m->itype()};
auto getDestinations = [&] {
// If a ledger data message is generated, it's going to be sent to every
// peer that is waiting for it.
PeerCookieMap result;
std::size_t numCookies = 0;
{
// Don't do the work under this peer if this peer is not waiting for
// any replies
auto myCookies = releaseRequestCookies(mHash);
if (myCookies.empty())
{
JLOG(pJournal_.debug()) << "TMGetLedger: peer is no longer "
"waiting for response to request: "
<< mHash;
return result;
}
numCookies += myCookies.size();
result[shared_from_this()] = myCookies;
}
std::set<HashRouter::PeerShortID> const peers = app_.getHashRouter().getPeers(mHash);
for (auto const peerID : peers)
{
// This loop does not need to be done under the HashRouter
// lock because findPeerByShortID and releaseRequestCookies
// are thread safe, and everything else is local
if (auto p = overlay_.findPeerByShortID(peerID))
{
auto cookies = p->releaseRequestCookies(mHash);
numCookies += cookies.size();
if (result.contains(p))
{
// Unlikely, but if a request came in to this peer while
// iterating, add the items instead of copying /
// overwriting.
XRPL_ASSERT(
p.get() == this,
"ripple::PeerImp::processLedgerRequest : found self in "
"map");
for (auto const& cookie : cookies)
result[p].emplace(cookie);
}
else if (cookies.size())
result[p] = cookies;
}
}
JLOG(pJournal_.debug()) << "TMGetLedger: Processing request for " << result.size()
<< " peers. Will send " << numCookies
<< " messages if successful: " << mHash;
return result;
};
// Will only populate this if we're going to do work.
PeerCookieMap destinations;
if (itype == protocol::liTS_CANDIDATE)
{
destinations = getDestinations();
if (destinations.empty())
// Nowhere to send the response!
return;
if (sharedMap = getTxSet(m, mHash); !sharedMap)
if (sharedMap = getTxSet(m); !sharedMap)
return;
map = sharedMap.get();
@@ -3617,6 +3306,8 @@ PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m, u
ledgerData.set_ledgerseq(0);
ledgerData.set_ledgerhash(m->ledgerhash());
ledgerData.set_type(protocol::liTS_CANDIDATE);
if (m->has_requestcookie())
ledgerData.set_requestcookie(m->requestcookie());
// We'll already have most transactions
fatLeaves = false;
@@ -3634,12 +3325,7 @@ PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m, u
return;
}
destinations = getDestinations();
if (destinations.empty())
// Nowhere to send the response!
return;
if (ledger = getLedger(m, mHash); !ledger)
if (ledger = getLedger(m); !ledger)
return;
// Fill out the reply
@@ -3647,11 +3333,13 @@ PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m, u
ledgerData.set_ledgerhash(ledgerHash.begin(), ledgerHash.size());
ledgerData.set_ledgerseq(ledger->header().seq);
ledgerData.set_type(itype);
if (m->has_requestcookie())
ledgerData.set_requestcookie(m->requestcookie());
switch (itype)
{
case protocol::liBASE:
sendLedgerBase(ledger, ledgerData, destinations);
sendLedgerBase(ledger, ledgerData);
return;
case protocol::liTX_NODE:
@@ -3762,7 +3450,7 @@ PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m, u
if (ledgerData.nodes_size() == 0)
return;
sendToMultiple(ledgerData, destinations);
send(std::make_shared<Message>(ledgerData, protocol::mtLEDGER_DATA));
}
// Differential pricing helper. Returns only the *dynamic* component
@@ -3861,19 +3549,6 @@ PeerImp::isHighLatency() const
return latency_ >= kPeerHighLatency;
}
std::set<std::optional<uint64_t>>
PeerImp::releaseRequestCookies(uint256 const& requestHash)
{
std::set<std::optional<uint64_t>> result;
std::lock_guard lock(cookieLock_);
if (messageRequestCookies_.contains(requestHash))
{
std::swap(result, messageRequestCookies_[requestHash]);
messageRequestCookies_.erase(requestHash);
}
return result;
};
void
PeerImp::Metrics::addMessage(std::uint64_t bytes)
{

View File

@@ -209,13 +209,6 @@ private:
bool ledgerReplayEnabled_ = false;
LedgerReplayMsgHandler ledgerReplayMsgHandler_;
// Track message requests and responses
// TODO: Use an expiring cache or something
using MessageCookieMap = std::map<uint256, std::set<std::optional<uint64_t>>>;
using PeerCookieMap = std::map<std::shared_ptr<Peer>, std::set<std::optional<uint64_t>>>;
std::mutex mutable cookieLock_;
MessageCookieMap messageRequestCookies_;
friend class OverlayImpl;
class Metrics
@@ -459,13 +452,6 @@ public:
return txReduceRelayEnabled_;
}
//
// Messages
//
std::set<std::optional<uint64_t>>
releaseRequestCookies(uint256 const& requestHash) override;
private:
void
close();
@@ -664,22 +650,16 @@ private:
std::shared_ptr<protocol::TMValidation> const& packet);
void
sendLedgerBase(
std::shared_ptr<Ledger const> const& ledger,
protocol::TMLedgerData& ledgerData,
PeerCookieMap const& destinations);
void
sendToMultiple(protocol::TMLedgerData& ledgerData, PeerCookieMap const& destinations);
sendLedgerBase(std::shared_ptr<Ledger const> const& ledger, protocol::TMLedgerData& ledgerData);
std::shared_ptr<Ledger const>
getLedger(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 const& mHash);
getLedger(std::shared_ptr<protocol::TMGetLedger> const& m);
std::shared_ptr<SHAMap const>
getTxSet(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 const& mHash) const;
getTxSet(std::shared_ptr<protocol::TMGetLedger> const& m) const;
void
processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m, uint256 const& mHash);
processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m);
protected:
// Kept `protected` so test subclasses (see

View File

@@ -7,9 +7,6 @@
#include <xrpl/basics/Log.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/core/HashRouter.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/protocol/digest.h>
#include <google/protobuf/message.h>
@@ -100,45 +97,16 @@ PeerSetImpl::sendRequest(
std::shared_ptr<Peer> const& peer)
{
auto packet = std::make_shared<Message>(message, type);
auto const messageHash = [&]() {
auto const packetBuffer = packet->getBuffer(compression::Compressed::Off);
return sha512Half(Slice(packetBuffer.data(), packetBuffer.size()));
}();
// Allow messages to be re-sent to the same peer after a delay
using namespace std::chrono_literals;
constexpr std::chrono::seconds interval = 30s;
if (peer)
{
if (app_.getHashRouter().shouldProcessForPeer(messageHash, peer->id(), interval))
{
JLOG(journal_.trace()) << "Sending " << protocolMessageName(type) << " message to ["
<< peer->id() << "]: " << messageHash;
peer->send(packet);
}
else
JLOG(journal_.debug()) << "Suppressing sending duplicate " << protocolMessageName(type)
<< " message to [" << peer->id() << "]: " << messageHash;
peer->send(packet);
return;
}
for (auto id : peers_)
{
if (auto p = app_.getOverlay().findPeerByShortID(id))
{
if (app_.getHashRouter().shouldProcessForPeer(messageHash, p->id(), interval))
{
JLOG(journal_.trace()) << "Sending " << protocolMessageName(type) << " message to ["
<< p->id() << "]: " << messageHash;
p->send(packet);
}
else
JLOG(journal_.debug())
<< "Suppressing sending duplicate " << protocolMessageName(type)
<< " message to [" << p->id() << "]: " << messageHash;
}
p->send(packet);
}
}

View File

@@ -30,12 +30,6 @@ protocolMessageType(protocol::TMGetLedger const&)
return protocol::mtGET_LEDGER;
}
inline protocol::MessageType
protocolMessageType(protocol::TMLedgerData const&)
{
return protocol::mtLEDGER_DATA;
}
inline protocol::MessageType
protocolMessageType(protocol::TMReplayDeltaRequest const&)
{
@@ -442,63 +436,3 @@ invokeProtocolMessage(Buffers const& buffers, Handler& handler, std::size_t& hin
}
} // namespace xrpl
namespace protocol {
template <class Hasher>
void
hash_append(Hasher& h, TMGetLedger const& msg)
{
using beast::hash_append;
using namespace xrpl;
hash_append(h, safeCast<int>(protocolMessageType(msg)));
hash_append(h, safeCast<int>(msg.itype()));
if (msg.has_ltype())
hash_append(h, safeCast<int>(msg.ltype()));
if (msg.has_ledgerhash())
hash_append(h, msg.ledgerhash());
if (msg.has_ledgerseq())
hash_append(h, msg.ledgerseq());
for (auto const& nodeId : msg.nodeids())
hash_append(h, nodeId);
hash_append(h, msg.nodeids_size());
// Do NOT include the request cookie. It does not affect the content of the
// request, but only where to route the results.
// if (msg.has_requestcookie())
// hash_append(h, msg.requestcookie());
if (msg.has_querytype())
hash_append(h, safeCast<int>(msg.querytype()));
if (msg.has_querydepth())
hash_append(h, msg.querydepth());
}
template <class Hasher>
void
hash_append(Hasher& h, TMLedgerData const& msg)
{
using beast::hash_append;
using namespace xrpl;
hash_append(h, safeCast<int>(protocolMessageType(msg)));
hash_append(h, msg.ledgerhash());
hash_append(h, msg.ledgerseq());
hash_append(h, safeCast<int>(msg.type()));
for (auto const& node : msg.nodes())
{
hash_append(h, node.nodedata());
if (node.has_nodeid())
hash_append(h, node.nodeid());
}
hash_append(h, msg.nodes_size());
if (msg.has_requestcookie())
hash_append(h, msg.requestcookie());
if (msg.has_error())
hash_append(h, safeCast<int>(msg.error()));
}
} // namespace protocol

View File

@@ -28,8 +28,6 @@ namespace xrpl {
constexpr ProtocolVersion const kSupportedProtocolList[]{
{2, 1},
{2, 2},
// Adds TMLedgerData::responseCookies and directResponse
{2, 3},
};
// This ugly construct ensures that supportedProtocolList is sorted in strictly