ARCHIVE/xahaud
3
0
Fork 0
mirror of https://github.com/Xahau/xahaud.git synced 2026-07-25 08:00:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(overlay): relay admitted export shares

Browse Source
This commit is contained in:
Nicholas Dudfield
2026-07-14 14:27:47 +07:00
parent 4c2860a9e1
commit 43a8be7429
11 changed files with 448 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
include/xrpl/proto/ripple.proto
View File

@@ -27,6 +27,7 @@ enum MessageType
mtREPLAY_DELTA_RESPONSE = 60;
mtHAVE_TRANSACTIONS = 63;
mtTRANSACTIONS = 64;
mtEXPORT_SHARES = 65;
}
// token, iterations, target, challenge = issue demand for proof of work
@@ -120,6 +121,15 @@ message TMTransactions
repeated TMTransaction transactions = 1;
}
// Canonical post-validation Export signature contributions. Receivers enforce
// ExportLimits caps for entry count, each frame, aggregate frame payload, and
// encoded protobuf size. The encoded-size cap includes protobuf tags and
// length prefixes but excludes the overlay message header.
message TMExportShares
{
repeated bytes shares = 1;
}
enum NodeStatus
{

192
src/test/overlay/ExportShareTransport_test.cpp Normal file
View File

@@ -0,0 +1,192 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
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 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/Message.h>
#include <xrpld/overlay/detail/ProtocolMessage.h>
#include <xrpld/overlay/detail/TrafficCount.h>
#include <xrpl/beast/unit_test.h>
#include <xrpl/protocol/ExportLimits.h>
#include <xrpl/protocol/ExportShare.h>
#include <xrpl/protocol/SecretKey.h>
#include <xrpl/protocol/Sign.h>
#include <boost/asio/buffer.hpp>
#include <boost/beast/core/multi_buffer.hpp>
#include <memory>
#include <string>
#include <type_traits>
namespace ripple {
namespace test {
class ExportShareTransport_test : public beast::unit_test::suite
{
static ExportShare
makeShare()
{
auto const [key, secret] = randomKeyPair(KeyType::secp256k1);
auto const signature = sign(key, secret, Slice{"export-share", 12});
return ExportShare{
ExportShare::currentVersion,
calcAccountID(key),
uint256{1},
4'200'000,
uint256{2},
uint256{3},
17,
key,
signature};
}
struct Handler
{
bool seen = false;
int shares = 0;
bool
compressionEnabled() const
{
return false;
}
void
onMessageUnknown(std::uint16_t)
{
}
void
onMessageBegin(
std::uint16_t,
std::shared_ptr<::google::protobuf::Message> const&,
std::size_t,
std::size_t,
bool)
{
}
void
onMessageEnd(
std::uint16_t,
std::shared_ptr<::google::protobuf::Message> const&)
{
}
template <class Message>
void
onMessage(std::shared_ptr<Message> const& message)
{
if constexpr (std::is_same_v<Message, protocol::TMExportShares>)
{
seen = true;
shares = message->shares_size();
}
}
};
void
testBatchBounds()
{
testcase("ExportShare batch structural bounds");
auto const encoded = makeShare().serialize();
protocol::TMExportShares batch;
batch.add_shares(encoded.data(), encoded.size());
auto const parsed = detail::parseExportShareBatch(batch);
BEAST_EXPECT(parsed && parsed->size() == 1);
BEAST_EXPECT(
batch.ByteSizeLong() <=
ExportLimits::maxExportShareRelayMessageBytes);
protocol::TMExportShares maxBatch;
for (std::size_t i = 0; i < ExportLimits::maxExportSharesPerRelay; ++i)
maxBatch.add_shares(encoded.data(), encoded.size());
auto const payloadBytes =
encoded.size() * ExportLimits::maxExportSharesPerRelay;
BEAST_EXPECT(
payloadBytes <= ExportLimits::maxExportShareRelayPayloadBytes);
BEAST_EXPECT(maxBatch.ByteSizeLong() > payloadBytes);
BEAST_EXPECT(
maxBatch.ByteSizeLong() <=
ExportLimits::maxExportShareRelayMessageBytes);
BEAST_EXPECT(detail::parseExportShareBatch(maxBatch));
protocol::TMExportShares empty;
BEAST_EXPECT(!detail::parseExportShareBatch(empty));
protocol::TMExportShares tooMany;
for (std::size_t i = 0; i <= ExportLimits::maxExportSharesPerRelay; ++i)
tooMany.add_shares(encoded.data(), encoded.size());
BEAST_EXPECT(!detail::parseExportShareBatch(tooMany));
protocol::TMExportShares oversized;
oversized.add_shares(
std::string(ExportLimits::maxSerializedExportShareBytes + 1, 'x'));
BEAST_EXPECT(!detail::parseExportShareBatch(oversized));
protocol::TMExportShares malformed;
malformed.add_shares("not-an-export-share");
BEAST_EXPECT(!detail::parseExportShareBatch(malformed));
}
void
testDispatchAndTraffic()
{
testcase("ExportShare protocol dispatch and traffic category");
auto const encoded = makeShare().serialize();
protocol::TMExportShares batch;
batch.add_shares(encoded.data(), encoded.size());
BEAST_EXPECT(protocol::mtEXPORT_SHARES == 65);
BEAST_EXPECT(
protocolMessageName(protocol::mtEXPORT_SHARES) == "export_shares");
BEAST_EXPECT(
TrafficCount::categorize(batch, protocol::mtEXPORT_SHARES, true) ==
TrafficCount::category::export_shares);
Message message{batch, protocol::mtEXPORT_SHARES};
auto const& wire = message.getBuffer(compression::Compressed::Off);
boost::beast::multi_buffer buffers;
buffers.commit(boost::asio::buffer_copy(
buffers.prepare(wire.size()), boost::asio::buffer(wire)));
Handler handler;
std::size_t hint = 0;
auto const [consumed, ec] =
invokeProtocolMessage(buffers.data(), handler, hint);
BEAST_EXPECT(!ec);
BEAST_EXPECT(consumed == wire.size());
BEAST_EXPECT(handler.seen);
BEAST_EXPECT(handler.shares == 1);
}
public:
void
run() override
{
testBatchBounds();
testDispatchAndTraffic();
}
};
BEAST_DEFINE_TESTSUITE(ExportShareTransport, overlay, ripple);
} // namespace test
} // namespace ripple

30
src/xrpld/overlay/Overlay.h
View File

@@ -24,6 +24,7 @@
#include <xrpld/overlay/PeerSet.h>
#include <xrpl/beast/utility/PropertyStream.h>
#include <xrpl/json/json_value.h>
#include <xrpl/protocol/ExportShare.h>
#include <xrpl/server/Handoff.h>
#include <boost/asio/buffer.hpp>
#include <boost/asio/ip/tcp.hpp>
@@ -64,6 +65,13 @@ protected:
public:
enum class Promote { automatic, never, always };
/** Application admission for a structurally valid ExportShare.
Returning true means the application has verified the share against
its origin context, committee position/key, and target multisignature.
*/
using ExportShareHandler = std::function<bool(ExportShare const&)>;
struct Setup
{
explicit Setup() = default;
@@ -149,6 +157,28 @@ public:
virtual void
broadcast(protocol::TMValidation& m) = 0;
/** Broadcast application-admitted canonical Export shares. */
virtual void
broadcast(protocol::TMExportShares& m) = 0;
/** Relay application-admitted Export shares using raw-frame suppression. */
virtual void
relay(protocol::TMExportShares& m) = 0;
/** Install the application admission callback for inbound Export shares.
The callback runs on the peer job queue, not an overlay I/O strand.
*/
virtual void
setExportShareHandler(ExportShareHandler handler) = 0;
/** Run application admission for an inbound Export share.
With no callback installed, admission fails closed.
*/
virtual bool
acceptExportShare(ExportShare const& share) = 0;
/** Relay a proposal.
* @param m the serialized proposal
* @param uid the id used to identify this proposal

1
src/xrpld/overlay/detail/Message.cpp
View File

@@ -101,6 +101,7 @@ Message::compress()
case protocol::mtPROOF_PATH_RESPONSE:
case protocol::mtREPLAY_DELTA_REQ:
case protocol::mtHAVE_TRANSACTIONS:
case protocol::mtEXPORT_SHARES:
break;
}
return false;

93
src/xrpld/overlay/detail/OverlayImpl.cpp
View File

@@ -1161,6 +1161,99 @@ OverlayImpl::broadcast(protocol::TMValidation& m)
for_each([sm](std::shared_ptr<PeerImp>&& p) { p->send(sm); });
}
void
OverlayImpl::broadcast(protocol::TMExportShares& m)
{
auto const shares = detail::parseExportShareBatch(m);
if (!shares)
{
JLOG(journal_.error())
<< "Refusing to broadcast malformed ExportShare batch";
return;
}
// Local callers are responsible for application admission before using
// this API. Register raw-wire suppression before routing the batch.
for (auto const& share : *shares)
app_.getHashRouter().addSuppression(share.wireHash());
relay(m);
}
void
OverlayImpl::relay(protocol::TMExportShares& m)
{
auto const shares = detail::parseExportShareBatch(m);
if (!shares)
{
JLOG(journal_.error())
<< "Refusing to relay malformed ExportShare batch";
return;
}
using Route = std::pair<std::size_t, std::set<Peer::id_t>>;
std::vector<Route> routes;
routes.reserve(shares->size());
for (std::size_t i = 0; i < shares->size(); ++i)
{
if (auto const toSkip =
app_.getHashRouter().shouldRelay((*shares)[i].wireHash()))
routes.emplace_back(i, std::move(*toSkip));
}
if (routes.empty())
return;
for_each([&](std::shared_ptr<PeerImp>&& peer) {
protocol::TMExportShares outbound;
outbound.mutable_shares()->Reserve(routes.size());
for (auto const& [index, toSkip] : routes)
{
if (!toSkip.contains(peer->id()))
outbound.add_shares(m.shares(index));
}
if (outbound.shares_size() != 0)
peer->send(
std::make_shared<Message>(outbound, protocol::mtEXPORT_SHARES));
});
}
void
OverlayImpl::setExportShareHandler(ExportShareHandler handler)
{
std::lock_guard lock(exportShareHandlerLock_);
exportShareHandler_ = std::move(handler);
}
bool
OverlayImpl::acceptExportShare(ExportShare const& share)
{
ExportShareHandler handler;
{
std::lock_guard lock(exportShareHandlerLock_);
handler = exportShareHandler_;
}
if (!handler)
return false;
try
{
return handler(share);
}
catch (std::exception const& e)
{
JLOG(journal_.error())
<< "ExportShare admission callback failed: " << e.what();
}
catch (...)
{
JLOG(journal_.error()) << "ExportShare admission callback failed";
}
return false;
}
std::set<Peer::id_t>
OverlayImpl::relay(
protocol::TMValidation& m,

15
src/xrpld/overlay/detail/OverlayImpl.h
View File

@@ -131,6 +131,9 @@ private:
// Protects the message and the sequence list of manifests
std::mutex manifestLock_;
ExportShareHandler exportShareHandler_;
std::mutex exportShareHandlerLock_;
//--------------------------------------------------------------------------
public:
@@ -223,6 +226,18 @@ public:
void
broadcast(protocol::TMValidation& m) override;
void
broadcast(protocol::TMExportShares& m) override;
void
relay(protocol::TMExportShares& m) override;
void
setExportShareHandler(ExportShareHandler handler) override;
bool
acceptExportShare(ExportShare const& share) override;
std::set<Peer::id_t>
relay(
protocol::TMProposeSet& m,

51
src/xrpld/overlay/detail/PeerImp.cpp
View File

@@ -1126,6 +1126,57 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMManifests> const& m)
});
}
void
PeerImp::onMessage(std::shared_ptr<protocol::TMExportShares> const& m)
{
auto shares = detail::parseExportShareBatch(*m);
if (!shares)
{
fee_.update(Resource::feeMalformedRequest, "malformed export shares");
return;
}
std::vector<std::size_t> fresh;
fresh.reserve(shares->size());
for (std::size_t i = 0; i < shares->size(); ++i)
{
// This hash identifies only identical canonical wire frames. Semantic
// contribution deduplication belongs to application admission.
if (app_.getHashRouter().addSuppressionPeer(
(*shares)[i].wireHash(), id_))
fresh.push_back(i);
}
if (fresh.empty())
{
fee_.update(Resource::feeUselessData, "duplicate export shares");
return;
}
std::weak_ptr<PeerImp> weak = shared_from_this();
app_.getJobQueue().addJob(
jtPEER,
"recvExportShares",
[weak, m, shares = std::move(*shares), fresh = std::move(fresh)]() {
auto const peer = weak.lock();
if (!peer)
return;
protocol::TMExportShares accepted;
accepted.mutable_shares()->Reserve(fresh.size());
for (auto const index : fresh)
{
if (peer->overlay_.acceptExportShare(shares[index]))
accepted.add_shares(m->shares(index));
}
// Structural validity is insufficient: only application-admitted
// frames are eligible for relay.
if (accepted.shares_size() != 0)
peer->overlay_.relay(accepted);
});
}
void
PeerImp::onMessage(std::shared_ptr<protocol::TMPing> const& m)
{

2
src/xrpld/overlay/detail/PeerImp.h
View File

@@ -594,6 +594,8 @@ public:
onMessage(std::shared_ptr<protocol::TMReplayDeltaRequest> const& m);
void
onMessage(std::shared_ptr<protocol::TMReplayDeltaResponse> const& m);
void
onMessage(std::shared_ptr<protocol::TMExportShares> const& m);
private:
//--------------------------------------------------------------------------

47
src/xrpld/overlay/detail/ProtocolMessage.h
View File

@@ -25,6 +25,8 @@
#include <xrpld/overlay/detail/ZeroCopyStream.h>
#include <xrpl/basics/ByteUtilities.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/protocol/ExportLimits.h>
#include <xrpl/protocol/ExportShare.h>
#include <xrpl/protocol/messages.h>
#include <boost/asio/buffer.hpp>
#include <boost/asio/buffers_iterator.hpp>
@@ -94,6 +96,8 @@ protocolMessageName(int type)
return "have_transactions";
case protocol::mtTRANSACTIONS:
return "transactions";
case protocol::mtEXPORT_SHARES:
return "export_shares";
case protocol::mtSQUELCH:
return "squelch";
case protocol::mtPROOF_PATH_REQ:
@@ -112,6 +116,45 @@ protocolMessageName(int type)
namespace detail {
/** Parse and structurally validate a canonical ExportShare relay batch.
maxExportShareRelayPayloadBytes applies to the sum of repeated-field value
bytes. maxExportShareRelayMessageBytes additionally includes protobuf tags
and length prefixes. Neither limit includes the overlay message header.
*/
inline std::optional<std::vector<ExportShare>>
parseExportShareBatch(protocol::TMExportShares const& message)
{
auto const count = message.shares_size();
if (count <= 0 ||
static_cast<std::size_t>(count) >
ExportLimits::maxExportSharesPerRelay ||
message.ByteSizeLong() > ExportLimits::maxExportShareRelayMessageBytes)
return std::nullopt;
std::size_t payloadBytes = 0;
std::vector<ExportShare> result;
result.reserve(count);
for (auto const& bytes : message.shares())
{
if (bytes.empty() ||
bytes.size() > ExportLimits::maxSerializedExportShareBytes ||
payloadBytes >
ExportLimits::maxExportShareRelayPayloadBytes - bytes.size())
return std::nullopt;
payloadBytes += bytes.size();
auto share = ExportShare::parse(makeSlice(bytes));
if (!share || share->serialize().slice() != makeSlice(bytes))
return std::nullopt;
result.push_back(std::move(*share));
}
return result;
}
struct MessageHeader
{
/** The size of the message on the wire.
@@ -450,6 +493,10 @@ invokeProtocolMessage(
success = detail::invoke<protocol::TMTransactions>(
*header, buffers, handler);
break;
case protocol::mtEXPORT_SHARES:
success = detail::invoke<protocol::TMExportShares>(
*header, buffers, handler);
break;
case protocol::mtSQUELCH:
success =
detail::invoke<protocol::TMSquelch>(*header, buffers, handler);

3
src/xrpld/overlay/detail/TrafficCount.cpp
View File

@@ -157,6 +157,9 @@ TrafficCount::categorize(
if (type == protocol::mtTRANSACTIONS)
return TrafficCount::category::requested_transactions;
if (type == protocol::mtEXPORT_SHARES)
return TrafficCount::category::export_shares;
return TrafficCount::category::unknown;
}

4
src/xrpld/overlay/detail/TrafficCount.h
View File

@@ -157,6 +157,9 @@ public:
// TMTransactions
requested_transactions,
// TMExportShares
export_shares,
unknown // must be last
};
@@ -249,6 +252,7 @@ protected:
{"replay_delta_response"}, // category::replay_delta_response
{"have_transactions"}, // category::have_transactions
{"requested_transactions"}, // category::transactions
{"export_shares"}, // category::export_shares
{"unknown"} // category::unknown
}};
};