Compare commits

..

47 Commits

Author SHA1 Message Date
Bart
0ad72fae8f Merge branch 'develop' into bthomee/node_depth 2026-03-22 06:33:02 -04:00
Bart
cc383c5fb8 Clang-tidy fixes 2026-03-18 15:09:46 -04:00
Bart
88715f1e5a Merge branch 'develop' into bthomee/node_depth 2026-03-18 13:38:55 -04:00
Bart
e1477cef0c Fix merge conflict 2026-03-18 09:56:08 -04:00
Bart
942874d7b0 Merge branch 'develop' into bthomee/node_depth 2026-03-18 09:47:16 -04:00
Bart
79326fc6b5 Merge branch 'develop' into bthomee/node_depth 2026-03-16 19:48:17 -04:00
Bart
48535d5226 Merge branch 'develop' into bthomee/node_depth 2026-03-10 17:11:14 +01:00
Bart
d1a6558080 Merge branch 'develop' into bthomee/node_depth 2026-03-10 15:39:32 +01:00
Bart
84f86b354f Add defensive check 2026-03-06 13:28:39 +01:00
Bart
40a3985b02 Merge branch 'develop' into bthomee/node_depth 2026-03-06 09:27:59 +01:00
Bart
208bd35d45 Merge branch 'develop' into bthomee/node_depth 2026-03-05 08:36:13 +01:00
Bart
e90fbbf7b2 Copilot review feedback 2026-03-02 13:39:29 -05:00
Bart
277450e648 Merge branch 'develop' into bthomee/node_depth 2026-03-02 12:16:33 -05:00
Bart
e6993524ea Copilot review feedback 2026-03-02 12:10:12 -05:00
Bart
b117ecc6a2 Use std::string_view 2026-03-02 11:58:57 -05:00
Bart
6c3b00c342 Change takeNodes vector argument to r-value 2026-03-02 09:41:23 -05:00
Bart
8c296a935a Improve docstring 2026-03-01 17:44:48 -05:00
Bart
573ba82181 Copilot review feedback 2026-03-01 17:33:03 -05:00
Bart
1542ab7e27 Copilot review feedback 2026-03-01 17:04:04 -05:00
Bart
6374f4886d Support leaf nodes at depth 0, use std::move, simplify tests 2026-03-01 16:44:58 -05:00
Bart
ebf336f472 Copilot review feedback 2026-03-01 15:10:57 -05:00
Bart
ddc15ad612 Copilot review feedback 2026-03-01 14:48:27 -05:00
Bart
82db6ac498 Restore try-catch to protect against other exceptions 2026-03-01 13:33:39 -05:00
Bart
f749c41306 Add log message for consistency 2026-02-28 17:31:22 -05:00
Bart
f25e47a58d Improve comment 2026-02-28 17:27:15 -05:00
Bart
2396799bd8 Update levelization 2026-02-28 16:03:02 -05:00
Bart
4855b9f96a Improve function docstrings, fix tests 2026-02-28 15:46:49 -05:00
Bart
b2f65cb7eb Fix protocol version test 2026-02-28 14:40:55 -05:00
Bart
c523673885 Update levelization 2026-02-28 14:00:13 -05:00
Bart
caac4d63d3 Merge branch 'develop' into bthomee/node_depth 2026-02-28 13:56:37 -05:00
Bart
29b0076fa8 Use new protocol version instead of amendment, add tests 2026-02-28 13:54:00 -05:00
Bart
c9aa1094a7 Update docstrings 2026-02-27 12:55:14 -05:00
Bart
b86f69cb82 Merge branch 'develop' into bthomee/node_depth 2026-02-26 17:33:25 -05:00
Bart
5d0bf78512 Clang-format to 100 line length 2026-02-26 17:09:20 -05:00
Bart
554df631c6 Remove pragma once 2026-02-18 08:28:56 -05:00
Bart
5e704bfdfb Merge branch 'develop' into bthomee/node_depth 2026-02-18 08:02:52 -05:00
Bart
fe8cc02bfa Refine 2026-02-18 07:54:33 -05:00
Bart
061c033f52 Use oneof in proto message 2026-02-16 16:50:40 -05:00
Bart
832a7e7e4a Remove depth, do not include node ID for leaf nodes 2026-02-13 17:05:05 -05:00
Bart
b2371c4c02 Fixes 2026-02-13 15:47:08 -05:00
Bart
b94a7c4b44 Merge branch 'develop' into bthomee/node_depth 2026-02-13 11:46:56 -05:00
Bart
9b9027112d Use helper functions 2026-02-13 11:44:58 -05:00
Bart
8e7889c66e Refactor 2026-02-12 16:55:38 -05:00
Bart
d836c3788d Merge branch 'develop' into bthomee/node_depth 2026-02-12 15:33:13 -05:00
Bart
1cb7c0293f Check if amendment is enabled 2026-02-12 06:31:32 -05:00
Bart
52dabc1f79 Remove deprecated stanza on nodeid field 2026-02-11 16:28:35 -05:00
Bart
2d78d41f7b perf: Replace node ID by depth in TMLedgerNode 2026-02-11 15:55:16 -05:00
29 changed files with 841 additions and 967 deletions

View File

@@ -43,6 +43,7 @@ libxrpl.tx > xrpl.server
libxrpl.tx > xrpl.tx
test.app > test.jtx
test.app > test.rpc
test.app > test.shamap
test.app > test.toplevel
test.app > test.unit_test
test.app > xrpl.basics
@@ -58,6 +59,7 @@ test.app > xrpl.protocol
test.app > xrpl.rdb
test.app > xrpl.resource
test.app > xrpl.server
test.app > xrpl.shamap
test.app > xrpl.tx
test.basics > test.jtx
test.basics > test.unit_test
@@ -177,7 +179,6 @@ test.unit_test > xrpl.protocol
tests.libxrpl > xrpl.basics
tests.libxrpl > xrpl.json
tests.libxrpl > xrpl.net
tests.libxrpl > xrpl.nodestore
tests.libxrpl > xrpl.protocol
tests.libxrpl > xrpl.protocol_autogen
xrpl.conditions > xrpl.basics

View File

@@ -11,6 +11,7 @@
#include <limits>
#include <stdexcept>
#include <string>
#include <string_view>
#include <type_traits>
#include <vector>
@@ -231,4 +232,11 @@ makeSlice(std::basic_string<char, Traits, Alloc> const& s)
return Slice(s.data(), s.size());
}
template <class Traits>
Slice
makeSlice(std::basic_string_view<char, Traits> const& s)
{
return Slice(s.data(), s.size());
}
} // namespace xrpl

View File

@@ -138,22 +138,6 @@ public:
/** Returns the number of file descriptors the backend expects to need. */
virtual int
fdRequired() const = 0;
/** The number of hardware threads to use for compression of a batch. */
static unsigned int const numHardwareThreads;
/** Calculate parallelization parameters for a batch of items.
Determines the number of threads and items per thread needed for parallel batch processing.
@param batchSize Number of items to process
@param maxThreadCount Maximum number of threads to use.
@return A pair of (numThreads, numItems) where numThreads is the exact number of threads to
use, and numItems is the number of items per thread. The last thread may process
fewer items.
*/
static std::pair<unsigned int, unsigned int>
calculateBatchParallelism(unsigned int batchSize, unsigned int maxThreadCount);
};
} // namespace NodeStore

View File

@@ -244,7 +244,15 @@ message TMGetObjectByHash {
message TMLedgerNode {
required bytes nodedata = 1;
optional bytes nodeid = 2; // missing for ledger base data
// Used when protocol version <2.3. Not set for ledger base data.
optional bytes nodeid = 2;
// Used when protocol version >=2.3. Neither value is set for ledger base data.
oneof reference {
bytes id = 3; // Set for inner nodes.
uint32 depth = 4; // Set for leaf nodes.
}
}
enum TMLedgerInfoType {

View File

@@ -16,6 +16,7 @@
#include <set>
#include <stack>
#include <tuple>
#include <vector>
namespace xrpl {
@@ -253,7 +254,7 @@ public:
bool
getNodeFat(
SHAMapNodeID const& wanted,
std::vector<std::pair<SHAMapNodeID, Blob>>& data,
std::vector<std::tuple<SHAMapNodeID, Blob, bool>>& data,
bool fatLeaves,
std::uint32_t depth) const;
@@ -280,10 +281,45 @@ public:
void
serializeRoot(Serializer& s) const;
/** Add a root node to the SHAMap during synchronization.
*
* This function is used when receiving the root node of a SHAMap from a peer during ledger
* synchronization. The node must already have been deserialized.
*
* @param hash The expected hash of the root node.
* @param rootNode A deserialized root node to add.
* @param filter Optional sync filter to track received nodes.
* @return Status indicating whether the node was useful, duplicate, or invalid.
*
* @note This function expects the rootNode to be a valid, deserialized SHAMapTreeNode. The
* caller is responsible for deserialization and basic validation before calling this
* function.
*/
SHAMapAddNode
addRootNode(SHAMapHash const& hash, Slice const& rootNode, SHAMapSyncFilter* filter);
addRootNode(
SHAMapHash const& hash,
intr_ptr::SharedPtr<SHAMapTreeNode> rootNode,
SHAMapSyncFilter const* filter);
/** Add a known node at a specific position in the SHAMap during synchronization.
*
* This function is used when receiving nodes from peers during ledger synchronization. The node
* is inserted at the position specified by nodeID. The node must already have been
* deserialized.
*
* @param nodeID The position in the tree where this node belongs.
* @param treeNode A deserialized tree node to add.
* @param filter Optional sync filter to track received nodes.
* @return Status indicating whether the node was useful, duplicate, or invalid.
*
* @note This function expects that the caller has already validated that the nodeID is
* consistent with the node's content.
*/
SHAMapAddNode
addKnownNode(SHAMapNodeID const& nodeID, Slice const& rawNode, SHAMapSyncFilter* filter);
addKnownNode(
SHAMapNodeID const& nodeID,
intr_ptr::SharedPtr<SHAMapTreeNode> treeNode,
SHAMapSyncFilter const* filter);
// status functions
void
@@ -344,11 +380,11 @@ private:
intr_ptr::SharedPtr<SHAMapTreeNode>
fetchNodeNT(SHAMapHash const& hash) const;
intr_ptr::SharedPtr<SHAMapTreeNode>
fetchNodeNT(SHAMapHash const& hash, SHAMapSyncFilter* filter) const;
fetchNodeNT(SHAMapHash const& hash, SHAMapSyncFilter const* filter) const;
intr_ptr::SharedPtr<SHAMapTreeNode>
fetchNode(SHAMapHash const& hash) const;
intr_ptr::SharedPtr<SHAMapTreeNode>
checkFilter(SHAMapHash const& hash, SHAMapSyncFilter* filter) const;
checkFilter(SHAMapHash const& hash, SHAMapSyncFilter const* filter) const;
/** Update hashes up to the root */
void
@@ -420,7 +456,7 @@ private:
descendAsync(
SHAMapInnerNode* parent,
int branch,
SHAMapSyncFilter* filter,
SHAMapSyncFilter const* filter,
bool& pending,
descendCallback&&) const;
@@ -429,7 +465,7 @@ private:
SHAMapInnerNode* parent,
SHAMapNodeID const& parentID,
int branch,
SHAMapSyncFilter* filter) const;
SHAMapSyncFilter const* filter) const;
// Non-storing
// Does not hook the returned node to its parent

View File

@@ -1,68 +0,0 @@
#include <xrpl/nodestore/Backend.h>
#include <algorithm>
#include <thread>
namespace xrpl {
namespace NodeStore {
// Initialize the static constant for hardware thread count. The `hardware_concurrency` function can
// return 0 on some platforms, in which case we default to 1. We limit the total number of threads
// to 8 to avoid contention.
unsigned int const Backend::numHardwareThreads = []() {
auto const hw = std::thread::hardware_concurrency();
return std::min(std::max(hw, 1u), 8u);
}();
std::pair<unsigned int, unsigned int>
Backend::calculateBatchParallelism(unsigned int batchSize, unsigned int maxThreadCount)
{
XRPL_ASSERT(
maxThreadCount > 0,
"xrpl::NodeStore::Backend::calculateBatchParallelism : maxThreadCount > 0");
if (maxThreadCount == 0)
{
// LCOV_EXCL_START
UNREACHABLE("xrpl::NodeStore::Backend::calculateBatchParallelism : maxThreadCount == 0");
return {1, batchSize};
// LCOV_EXCL_STOP
}
if (batchSize == 0)
{
return {0, 0};
}
// Estimate the number of threads using ceiling division: aim for at least 4 items per thread,
// but don't exceed the number of available threads.
auto const initialThreads = std::min((batchSize + 3u) / 4u, maxThreadCount);
// Calculate number of items per thread.
auto const numItems = (batchSize + initialThreads - 1u) / initialThreads;
// Calculate the actual number of threads needed. After rounding up numItems, we may need fewer
// threads than initially estimated.
auto const actualThreads = (batchSize + numItems - 1u) / numItems;
XRPL_ASSERT(
numItems <= batchSize,
"xrpl::NodeStore::Backend::calculateBatchParallelism : numItems <= batchSize");
XRPL_ASSERT(
actualThreads <= batchSize,
"xrpl::NodeStore::Backend::calculateBatchParallelism : actualThreads <= batchSize");
XRPL_ASSERT(
actualThreads <= maxThreadCount,
"xrpl::NodeStore::Backend::calculateBatchParallelism : actualThreads <= hwThreadCount");
if (numItems > batchSize || actualThreads > batchSize || actualThreads > maxThreadCount)
{
// LCOV_EXCL_START
UNREACHABLE("xrpl::NodeStore::Backend::calculateBatchParallelism : sanity check failed");
return {1, batchSize};
// LCOV_EXCL_STOP
}
return {actualThreads, numItems};
}
} // namespace NodeStore
} // namespace xrpl

View File

@@ -7,22 +7,15 @@
#include <xrpl/nodestore/detail/EncodedBlob.h>
#include <xrpl/nodestore/detail/codec.h>
#include <boost/asio/post.hpp>
#include <boost/asio/thread_pool.hpp>
#include <boost/filesystem.hpp>
#include <nudb/nudb.hpp>
#include <atomic>
#include <chrono>
#include <cstdint>
#include <cstdio>
#include <exception>
#include <latch>
#include <memory>
#include <mutex>
#include <thread>
#include <vector>
namespace xrpl {
namespace NodeStore {
@@ -30,6 +23,21 @@ namespace NodeStore {
class NuDBBackend : public Backend
{
public:
// "appnum" is an application-defined constant stored in the header of a
// NuDB database. We used it to identify shard databases before that code
// was removed. For now, its only use is a sanity check that the database
// was created by xrpld.
static constexpr std::uint64_t appnum = 1;
beast::Journal const j_;
size_t const keyBytes_;
std::size_t const burstSize_;
std::string const name_;
std::size_t const blockSize_;
nudb::store db_;
std::atomic<bool> deletePath_;
Scheduler& scheduler_;
NuDBBackend(
size_t keyBytes,
Section const& keyValues,
@@ -43,7 +51,6 @@ public:
, blockSize_(parseBlockSize(name_, keyValues, journal))
, deletePath_(false)
, scheduler_(scheduler)
, threadPool_(numHardwareThreads)
{
if (name_.empty())
Throw<std::runtime_error>("nodestore: Missing path in NuDB backend");
@@ -64,7 +71,6 @@ public:
, db_(context)
, deletePath_(false)
, scheduler_(scheduler)
, threadPool_(numHardwareThreads)
{
if (name_.empty())
Throw<std::runtime_error>("nodestore: Missing path in NuDB backend");
@@ -74,32 +80,7 @@ public:
{
try
{
// Set shutdown flag to prevent new batch operations from starting. This must happen
// before stop() is called to ensure fetchBatch/storeBatch check the flag before posting
// any new tasks.
shutdown_.store(true, std::memory_order_release);
// Wait for all active operations to complete.
while (pendingReads_.load(std::memory_order_acquire) > 0 ||
pendingWrites_.load(std::memory_order_acquire) > 0)
{
std::this_thread::yield();
}
// Signal the thread pool to stop accepting new work. This ensures no new tasks will be
// posted after this point.
threadPool_.stop();
// Wait for all currently executing thread pool tasks to complete. This prevents worker
// threads from accessing the database after close().
threadPool_.join();
// Verify all writes have completed.
XRPL_ASSERT(
pendingWrites_.load() == 0, "xrpl::NuDBBackend::~NuDBBackend : pendingWrites == 0");
// Close the database. At this point, all threads have stopped and no pending reads and
// writes remain, so it's safe to close the database.
// close can throw and we don't want the destructor to throw.
close();
}
catch (nudb::system_error const&) // NOLINT(bugprone-empty-catch)
@@ -128,7 +109,9 @@ public:
if (db_.is_open())
{
// LCOV_EXCL_START
UNREACHABLE("xrpl::NodeStore::NuDBBackend::open : database is already open");
UNREACHABLE(
"xrpl::NodeStore::NuDBBackend::open : database is already "
"open");
JLOG(j_.error()) << "database is already open";
return;
// LCOV_EXCL_STOP
@@ -144,24 +127,16 @@ public:
nudb::create<nudb::xxhasher>(
dp, kp, lp, appType, uid, salt, keyBytes_, blockSize_, 0.50, ec);
if (ec == nudb::errc::file_exists)
{
ec = {};
}
if (ec)
{
Throw<nudb::system_error>(ec);
}
}
db_.open(dp, kp, lp, ec);
if (ec)
{
Throw<nudb::system_error>(ec);
}
if (db_.appnum() != appnum)
{
Throw<std::runtime_error>("nodestore: unknown appnum");
}
db_.set_burst(burstSize_);
}
@@ -206,22 +181,9 @@ public:
Status
fetch(uint256 const& hash, std::shared_ptr<NodeObject>* pno) override
{
// Increment pending reads counter on entry, decrement on exit. This ensures the destructor
// waits for this operation to complete.
++pendingReads_;
auto guard = [this](void*) { --pendingReads_; };
std::unique_ptr<void, decltype(guard)> opGuard(reinterpret_cast<void*>(1), guard);
// Check if we're shutting down. If so, return immediately instead of doing any work.
if (shutdown_.load(std::memory_order_acquire))
{
return backendError;
}
Status status = ok;
pno->reset();
nudb::error_code ec;
db_.fetch(
hash.data(),
[&hash, pno, &status](void const* data, std::size_t size) {
@@ -237,119 +199,30 @@ public:
status = ok;
},
ec);
if (ec == nudb::error::key_not_found)
{
return notFound;
}
if (ec)
{
Throw<nudb::system_error>(ec);
}
return status;
}
std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
fetchBatch(std::vector<uint256> const& hashes) override
{
if (hashes.empty())
std::vector<std::shared_ptr<NodeObject>> results;
results.reserve(hashes.size());
for (auto const& h : hashes)
{
return {{}, ok};
}
// Increment pending reads counter on entry, decrement on exit. This ensures the destructor
// waits for this operation to complete.
pendingReads_ += hashes.size();
auto guard = [this, &hashes](void*) { pendingReads_ -= hashes.size(); };
std::unique_ptr<void, decltype(guard)> opGuard(reinterpret_cast<void*>(1), guard);
// Check if we're shutting down. If so, return immediately instead of doing any work.
if (shutdown_.load(std::memory_order_acquire))
{
return {{}, backendError};
}
std::vector<std::shared_ptr<NodeObject>> results(hashes.size());
// Calculate parallelization parameters for the batch.
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(hashes.size(), numHardwareThreads);
// If we need only one thread, just do it sequentially. Although it should be impossible to
// get 0 threads here, handle it gracefully just in case.
if (numThreads <= 1u)
{
for (size_t i = 0; i < hashes.size(); ++i)
std::shared_ptr<NodeObject> nObj;
Status status = fetch(h, &nObj);
if (status != ok)
{
std::shared_ptr<NodeObject> nObj;
if (fetch(hashes[i], &nObj) == ok)
{
results[i] = nObj;
}
results.push_back({});
}
return {results, ok};
}
// Use a latch to synchronize task completion.
std::latch taskCompletion(numThreads);
// Track exceptions from worker threads.
std::exception_ptr eptr;
std::mutex emutex;
// Submit fetch tasks to the thread pool.
for (auto t = 0u; t < numThreads; ++t)
{
auto const startIdx = t * numItems;
XRPL_ASSERT(
startIdx < hashes.size(),
"xrpl::NuDBFactory::fetchBatch : startIdx < hashes.size()");
if (startIdx >= hashes.size())
else
{
// This should never happen, but is kept as a safety check.
taskCompletion.count_down();
continue;
results.push_back(nObj);
}
auto const endIdx = std::min<std::size_t>(startIdx + numItems, hashes.size());
auto task =
[this, &hashes, &results, &taskCompletion, &eptr, &emutex, startIdx, endIdx]() {
try
{
// Fetch the items assigned to this task.
for (size_t i = startIdx; i < endIdx; ++i)
{
std::shared_ptr<NodeObject> nObj;
if (fetch(hashes[i], &nObj) == ok)
{
results[i] = nObj;
}
}
}
catch (...)
{
// Store the first exception that occurs. Ensures count_down() is always
// called to prevent deadlock.
std::lock_guard<std::mutex> lock(emutex);
if (!eptr)
{
eptr = std::current_exception();
}
}
// Signal task completion.
taskCompletion.count_down();
};
boost::asio::post(threadPool_, std::move(task));
}
// Wait for all fetch tasks to complete.
taskCompletion.wait();
// Rethrow the first exception if one occurred.
if (eptr)
{
std::rethrow_exception(eptr);
}
return {results, ok};
@@ -359,39 +232,21 @@ public:
do_insert(std::shared_ptr<NodeObject> const& no)
{
EncodedBlob e(no);
nudb::error_code ec;
nudb::detail::buffer bf;
auto const result = nodeobject_compress(e.getData(), e.getSize(), bf);
nudb::error_code ec;
db_.insert(e.getKey(), result.first, result.second, ec);
if (ec && ec != nudb::error::key_exists)
{
Throw<nudb::system_error>(ec);
}
}
void
store(std::shared_ptr<NodeObject> const& no) override
{
// Increment pending writes counter on entry, decrement on exit. This ensures the destructor
// waits for this operation to complete.
++pendingWrites_;
auto guard = [this](void*) { --pendingWrites_; };
std::unique_ptr<void, decltype(guard)> opGuard(reinterpret_cast<void*>(1), guard);
// Check if we're shutting down. If so, return immediately instead of doing any work.
if (shutdown_.load(std::memory_order_acquire))
{
return;
}
BatchWriteReport report{};
report.writeCount = 1;
auto const start = std::chrono::steady_clock::now();
do_insert(no);
report.elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::steady_clock::now() - start);
scheduler_.onBatchWrite(report);
@@ -400,127 +255,11 @@ public:
void
storeBatch(Batch const& batch) override
{
if (batch.empty())
{
return;
}
// Increment pending writes counter on entry, decrement on exit. This ensures the destructor
// waits for this operation to complete.
pendingWrites_ += batch.size();
auto guard = [this, &batch](void*) { pendingWrites_ -= batch.size(); };
std::unique_ptr<void, decltype(guard)> opGuard(reinterpret_cast<void*>(1), guard);
// Check if we're shutting down. If so, return immediately instead of doing any work.
if (shutdown_.load(std::memory_order_acquire))
{
return;
}
BatchWriteReport report{};
report.writeCount = batch.size();
auto const start = std::chrono::steady_clock::now();
// Calculate parallelization parameters for the batch.
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batch.size(), numHardwareThreads);
// If we need only one thread, just do it sequentially. Although it should be impossible to
// get 0 threads here, handle it gracefully just in case.
if (numThreads <= 1u)
{
for (auto const& e : batch)
{
do_insert(e);
}
report.elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::steady_clock::now() - start);
scheduler_.onBatchWrite(report);
return;
}
// Helper struct that stores actual item data, not pointers, to avoid dangling references
// after EncodedBlob and buffer go out of scope in the thread.
struct CompressedData
{
std::vector<std::uint8_t> key;
std::vector<std::uint8_t> data;
std::exception_ptr eptr;
};
std::vector<CompressedData> compressed(batch.size());
// Use a latch to synchronize task completion.
std::latch taskCompletion(numThreads);
// Submit compression tasks to the thread pool.
for (auto t = 0u; t < numThreads; ++t)
{
auto const startIdx = t * numItems;
XRPL_ASSERT(
startIdx < batch.size(), "xrpl::NuDBFactory::storeBatch : startIdx < batch.size()");
if (startIdx >= batch.size())
{
// This should never happen, but is kept as a safety check.
taskCompletion.count_down();
continue;
}
auto const endIdx = std::min<std::size_t>(startIdx + numItems, batch.size());
auto task =
[&batch, &compressed, &taskCompletion, startIdx, endIdx, keyBytes = keyBytes_]() {
// Compress the items assigned to this task.
for (size_t i = startIdx; i < endIdx; ++i)
{
auto& item = compressed[i];
try
{
EncodedBlob e(batch[i]);
// Copy the key data to avoid dangling pointer.
auto const* keyPtr = static_cast<std::uint8_t const*>(e.getKey());
item.key.assign(keyPtr, keyPtr + keyBytes);
// Compress and copy the data to avoid dangling pointer.
nudb::detail::buffer bf;
auto const comp = nodeobject_compress(e.getData(), e.getSize(), bf);
auto const* dataPtr = static_cast<std::uint8_t const*>(comp.first);
item.data.assign(dataPtr, dataPtr + comp.second);
}
catch (...)
{
// Store the exception so it can be rethrown in the sequential phase
// below.
item.eptr = std::current_exception();
}
}
// Signal task completion.
taskCompletion.count_down();
};
boost::asio::post(threadPool_, std::move(task));
}
// Wait for all compression tasks to complete.
taskCompletion.wait();
// Insert the compressed data sequentially, since NuDB is designed as an append-only data
// store that limits concurrent writes.
for (auto const& item : compressed)
{
if (item.eptr)
{
std::rethrow_exception(item.eptr);
}
nudb::error_code ec;
db_.insert(item.key.data(), item.data.data(), item.data.size(), ec);
if (ec && ec != nudb::error::key_exists)
{
Throw<nudb::system_error>(ec);
}
}
for (auto const& e : batch)
do_insert(e);
report.elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::steady_clock::now() - start);
scheduler_.onBatchWrite(report);
@@ -537,7 +276,7 @@ public:
auto const dp = db_.dat_path();
auto const kp = db_.key_path();
auto const lp = db_.log_path();
// auto const appnum = db_.appnum();
nudb::error_code ec;
db_.close(ec);
if (ec)
@@ -571,7 +310,7 @@ public:
int
getWriteLoad() override
{
return pendingWrites_.load();
return 0;
}
void
@@ -646,28 +385,6 @@ private:
Throw<std::runtime_error>(s.str());
}
}
// "appnum" is an application-defined constant stored in the header of a
// NuDB database. We used it to identify shard databases before that code
// was removed. For now, its only use is a sanity check that the database
// was created by xrpld.
static constexpr std::uint64_t appnum = 1;
beast::Journal const j_;
size_t const keyBytes_;
std::size_t const burstSize_;
std::string const name_;
std::size_t const blockSize_;
nudb::store db_;
std::atomic<bool> deletePath_;
Scheduler& scheduler_;
std::atomic<size_t> pendingReads_{
0}; // Declare before threadPool_ to ensure it's destroyed after.
std::atomic<size_t> pendingWrites_{
0}; // Declare before threadPool_ to ensure it's destroyed after.
std::atomic<bool> shutdown_{
false}; // Declare before threadPool_ to ensure it's destroyed after.
boost::asio::thread_pool threadPool_; // Declare after db_ to ensure it's destroyed before.
};
//------------------------------------------------------------------------------

View File

@@ -13,7 +13,6 @@
#include <atomic>
#include <memory>
#include <thread>
namespace xrpl {
namespace NodeStore {
@@ -186,41 +185,6 @@ public:
}
}
// Enable pipelined writes for better write concurrency.
m_options.enable_pipelined_write = true;
// Set background job parallelism for better compaction/flush performance to the number of
// hardware threads, unless the value is explicitly provided in the config. The default is
// 2 (see include/rocksdb/options.h in the Conan dependency directory), so don't use fewer
// than that if no value is explicitly provided.
if (keyValues.exists("max_background_jobs"))
{
m_options.max_background_jobs = get<unsigned int>(keyValues, "max_background_jobs");
}
else if (auto v = numHardwareThreads; v > 2)
{
m_options.max_background_jobs = v;
}
// Set subcompactions for parallel compaction within a job to the number of hardware
// threads, unless the value is explicitly provided in the config. The default is 1 (see
// include/rocksdb/options.h in the Conan dependency directory), so don't use fewer
// than that if no value is explicitly provided.
if (keyValues.exists("max_subcompactions"))
{
m_options.max_subcompactions = get<unsigned int>(keyValues, "max_subcompactions");
}
else if (auto v = numHardwareThreads / 2; v > 1)
{
m_options.max_subcompactions = v;
}
// Enable direct I/O by default unless explicitly disabled in the config. This bypasses the
// OS page cache for better predictable performance on SSDs.
m_options.use_direct_reads = get<bool>(keyValues, "use_direct_io", true);
m_options.use_direct_io_for_flush_and_compaction =
get<bool>(keyValues, "use_direct_io", true);
std::string s1, s2;
rocksdb::GetStringFromDBOptions(&s1, m_options, "; ");
rocksdb::GetStringFromColumnFamilyOptions(&s2, m_options, "; ");
@@ -295,19 +259,23 @@ public:
rocksdb::ReadOptions const options;
rocksdb::Slice const slice(std::bit_cast<char const*>(hash.data()), m_keyBytes);
std::string string;
rocksdb::Status getStatus = m_db->Get(options, slice, &string);
if (getStatus.ok())
{
DecodedBlob decoded(hash.data(), string.data(), string.size());
if (decoded.wasOk())
{
*pObject = decoded.createObject();
}
else
{
// Decoding failed, probably corrupted.
// Decoding failed, probably corrupted!
//
status = dataCorrupt;
}
}
@@ -324,6 +292,7 @@ public:
else
{
status = Status(customCode + unsafe_cast<int>(getStatus.code()));
JLOG(m_journal.error()) << getStatus.ToString();
}
}
@@ -334,44 +303,19 @@ public:
std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
fetchBatch(std::vector<uint256> const& hashes) override
{
XRPL_ASSERT(m_db, "xrpl::NodeStore::RocksDBBackend::fetchBatch : non-null database");
if (hashes.empty())
{
return {{}, ok};
}
// Use MultiGet for parallel reads to allow RocksDB to fetch multiple keys concurrently,
// significantly improving throughput compared to sequential fetch() calls.
std::vector<rocksdb::Slice> keys;
keys.reserve(hashes.size());
std::vector<std::shared_ptr<NodeObject>> results;
results.reserve(hashes.size());
for (auto const& h : hashes)
{
keys.emplace_back(std::bit_cast<char const*>(h.data()), m_keyBytes);
}
rocksdb::ReadOptions options;
options.async_io = true; // Enable for better concurrency on supported platforms.
std::vector<std::string> values(hashes.size());
auto statuses = m_db->MultiGet(options, keys, &values);
std::vector<std::shared_ptr<NodeObject>> results(hashes.size());
for (size_t i = 0; i < hashes.size(); ++i)
{
if (statuses[i].ok())
std::shared_ptr<NodeObject> nObj;
Status status = fetch(h, &nObj);
if (status != ok)
{
DecodedBlob decoded(hashes[i].data(), values[i].data(), values[i].size());
if (decoded.wasOk())
{
results[i] = decoded.createObject();
}
results.push_back({});
}
else if (!statuses[i].IsNotFound())
else
{
// Log other errors but continue processing.
JLOG(m_journal.warn()) << "fetchBatch: MultiGet error for key "
<< keys[i].ToString() << ": " << statuses[i].ToString();
results.push_back(nObj);
}
}
@@ -387,45 +331,25 @@ public:
void
storeBatch(Batch const& batch) override
{
XRPL_ASSERT(m_db, "xrpl::NodeStore::RocksDBBackend::storeBatch : non-null database");
if (batch.empty())
{
return;
}
XRPL_ASSERT(
m_db,
"xrpl::NodeStore::RocksDBBackend::storeBatch : non-null "
"database");
rocksdb::WriteBatch wb;
for (auto const& e : batch)
{
EncodedBlob encoded(e);
wb.Put(
rocksdb::Slice(std::bit_cast<char const*>(encoded.getKey()), m_keyBytes),
rocksdb::Slice(std::bit_cast<char const*>(encoded.getData()), encoded.getSize()));
}
// Configure WriteOptions for high throughput.
// Note: no_slowdown is intentionally NOT set here. When set to true, RocksDB returns an
// error instead of stalling when write buffers are full, which could cause write
// failures during high load. We prefer to accept brief stalls over dropped writes.
rocksdb::WriteOptions options;
// Setting `sync = false` improves write throughput significantly by allowing the OS to
// batch fsync operations, rather than forcing immediate disk synchronization on every
// write. The Write-Ahead Log (WAL) is still written and flushed, so database consistency is
// maintained across clean restarts and crashes.
//
// Note: On hard shutdown up to a few seconds of recent writes (since the last OS-initiated
// flush) may be lost from this node. However, since ledger data is replicated across
// the network, lost writes can be re-synced from peers during startup.
options.sync = false;
// Keep WAL enabled for crash recovery consistency.
options.disableWAL = false;
// Ensure RocksDB will not aggressive throttle the writes.
options.low_pri = false;
rocksdb::WriteOptions const options;
auto ret = m_db->Write(options, &wb);
if (!ret.ok())
Throw<std::runtime_error>("storeBatch failed: " + ret.ToString());
}

View File

@@ -179,7 +179,7 @@ SHAMap::finishFetch(SHAMapHash const& hash, std::shared_ptr<NodeObject> const& o
// See if a sync filter has a node
intr_ptr::SharedPtr<SHAMapTreeNode>
SHAMap::checkFilter(SHAMapHash const& hash, SHAMapSyncFilter* filter) const
SHAMap::checkFilter(SHAMapHash const& hash, SHAMapSyncFilter const* filter) const
{
if (auto nodeData = filter->getNode(hash))
{
@@ -205,7 +205,7 @@ SHAMap::checkFilter(SHAMapHash const& hash, SHAMapSyncFilter* filter) const
// Get a node without throwing
// Used on maps where missing nodes are expected
intr_ptr::SharedPtr<SHAMapTreeNode>
SHAMap::fetchNodeNT(SHAMapHash const& hash, SHAMapSyncFilter* filter) const
SHAMap::fetchNodeNT(SHAMapHash const& hash, SHAMapSyncFilter const* filter) const
{
auto node = cacheLookup(hash);
if (node)
@@ -318,7 +318,7 @@ SHAMap::descend(
SHAMapInnerNode* parent,
SHAMapNodeID const& parentID,
int branch,
SHAMapSyncFilter* filter) const
SHAMapSyncFilter const* filter) const
{
XRPL_ASSERT(parent->isInner(), "xrpl::SHAMap::descend : valid parent input");
XRPL_ASSERT(
@@ -347,7 +347,7 @@ SHAMapTreeNode*
SHAMap::descendAsync(
SHAMapInnerNode* parent,
int branch,
SHAMapSyncFilter* filter,
SHAMapSyncFilter const* filter,
bool& pending,
descendCallback&& callback) const
{

View File

@@ -122,7 +122,9 @@ selectBranch(SHAMapNodeID const& id, uint256 const& hash)
SHAMapNodeID
SHAMapNodeID::createID(int depth, uint256 const& key)
{
XRPL_ASSERT((depth >= 0) && (depth < 65), "xrpl::SHAMapNodeID::createID : valid branch input");
XRPL_ASSERT(
depth >= 0 && depth <= SHAMap::leafDepth,
"xrpl::SHAMapNodeID::createID : valid branch input");
return SHAMapNodeID(depth, key & depthMask(depth));
}

View File

@@ -392,7 +392,7 @@ SHAMap::getMissingNodes(int max, SHAMapSyncFilter* filter)
bool
SHAMap::getNodeFat(
SHAMapNodeID const& wanted,
std::vector<std::pair<SHAMapNodeID, Blob>>& data,
std::vector<std::tuple<SHAMapNodeID, Blob, bool>>& data,
bool fatLeaves,
std::uint32_t depth) const
{
@@ -438,7 +438,7 @@ SHAMap::getNodeFat(
// Add this node to the reply
s.erase();
node->serializeForWire(s);
data.emplace_back(std::make_pair(nodeID, s.getData()));
data.emplace_back(std::make_tuple(nodeID, s.getData(), node->isLeaf()));
if (node->isInner())
{
@@ -468,7 +468,8 @@ SHAMap::getNodeFat(
// Just include this node
s.erase();
childNode->serializeForWire(s);
data.emplace_back(std::make_pair(childID, s.getData()));
data.emplace_back(
std::make_tuple(childID, s.getData(), childNode->isLeaf()));
}
}
}
@@ -486,8 +487,18 @@ SHAMap::serializeRoot(Serializer& s) const
}
SHAMapAddNode
SHAMap::addRootNode(SHAMapHash const& hash, Slice const& rootNode, SHAMapSyncFilter* filter)
SHAMap::addRootNode(
SHAMapHash const& hash,
intr_ptr::SharedPtr<SHAMapTreeNode> rootNode,
SHAMapSyncFilter const* filter)
{
XRPL_ASSERT(rootNode, "xrpl::SHAMap::addRootNode : non-null root node");
if (!rootNode)
{
JLOG(journal_.error()) << "Null node received";
return SHAMapAddNode::invalid();
}
// we already have a root_ node
if (root_->getHash().isNonZero())
{
@@ -497,14 +508,16 @@ SHAMap::addRootNode(SHAMapHash const& hash, Slice const& rootNode, SHAMapSyncFil
}
XRPL_ASSERT(cowid_ >= 1, "xrpl::SHAMap::addRootNode : valid cowid");
auto node = SHAMapTreeNode::makeFromWire(rootNode);
if (!node || node->getHash() != hash)
if (rootNode->getHash() != hash)
{
JLOG(journal_.warn()) << "Corrupt node received";
return SHAMapAddNode::invalid();
}
if (backed_)
canonicalize(hash, node);
canonicalize(hash, rootNode);
root_ = node;
root_ = std::move(rootNode);
if (root_->isLeaf())
clearSynching();
@@ -521,9 +534,23 @@ SHAMap::addRootNode(SHAMapHash const& hash, Slice const& rootNode, SHAMapSyncFil
}
SHAMapAddNode
SHAMap::addKnownNode(SHAMapNodeID const& node, Slice const& rawNode, SHAMapSyncFilter* filter)
SHAMap::addKnownNode(
SHAMapNodeID const& nodeID,
intr_ptr::SharedPtr<SHAMapTreeNode> treeNode,
SHAMapSyncFilter const* filter)
{
XRPL_ASSERT(!node.isRoot(), "xrpl::SHAMap::addKnownNode : valid node input");
XRPL_ASSERT(!nodeID.isRoot(), "xrpl::SHAMap::addKnownNode : valid node input");
if (nodeID.isRoot())
{
JLOG(journal_.error()) << "Root node received";
return SHAMapAddNode::invalid();
}
XRPL_ASSERT(treeNode, "xrpl::SHAMap::addKnownNode : non-null tree node");
if (!treeNode)
{
JLOG(journal_.error()) << "Null node received";
return SHAMapAddNode::invalid();
}
if (!isSynching())
{
@@ -537,14 +564,14 @@ SHAMap::addKnownNode(SHAMapNodeID const& node, Slice const& rawNode, SHAMapSyncF
while (currNode->isInner() &&
!safe_downcast<SHAMapInnerNode*>(currNode)->isFullBelow(generation) &&
(currNodeID.getDepth() < node.getDepth()))
(currNodeID.getDepth() < nodeID.getDepth()))
{
int const branch = selectBranch(currNodeID, node.getNodeID());
int const branch = selectBranch(currNodeID, nodeID.getNodeID());
XRPL_ASSERT(branch >= 0, "xrpl::SHAMap::addKnownNode : valid branch");
auto inner = safe_downcast<SHAMapInnerNode*>(currNode);
if (inner->isEmptyBranch(branch))
{
JLOG(journal_.warn()) << "Add known node for empty branch" << node;
JLOG(journal_.warn()) << "Add known node for empty branch" << nodeID;
return SHAMapAddNode::invalid();
}
@@ -560,67 +587,44 @@ SHAMap::addKnownNode(SHAMapNodeID const& node, Slice const& rawNode, SHAMapSyncF
if (currNode != nullptr)
continue;
auto newNode = SHAMapTreeNode::makeFromWire(rawNode);
if (!newNode || childHash != newNode->getHash())
if (childHash != treeNode->getHash())
{
JLOG(journal_.warn()) << "Corrupt node received";
return SHAMapAddNode::invalid();
}
// In rare cases, a node can still be corrupt even after hash
// validation. For leaf nodes, we perform an additional check to
// ensure the node's position in the tree is consistent with its
// content to prevent inconsistencies that could
// propagate further down the line.
if (newNode->isLeaf())
{
auto const& actualKey =
safe_downcast<SHAMapLeafNode const*>(newNode.get())->peekItem()->key();
// Validate that this leaf belongs at the target position
auto const expectedNodeID = SHAMapNodeID::createID(node.getDepth(), actualKey);
if (expectedNodeID.getNodeID() != node.getNodeID())
{
JLOG(journal_.debug())
<< "Leaf node position mismatch: "
<< "expected=" << expectedNodeID.getNodeID() << ", actual=" << node.getNodeID();
return SHAMapAddNode::invalid();
}
}
// Inner nodes must be at a level strictly less than 64
// but leaf nodes (while notionally at level 64) can be
// at any depth up to and including 64:
if ((currNodeID.getDepth() > leafDepth) ||
(newNode->isInner() && currNodeID.getDepth() == leafDepth))
(treeNode->isInner() && currNodeID.getDepth() == leafDepth))
{
// Map is provably invalid
state_ = SHAMapState::Invalid;
return SHAMapAddNode::useful();
}
if (currNodeID != node)
if (currNodeID != nodeID)
{
// Either this node is broken or we didn't request it (yet)
JLOG(journal_.warn()) << "unable to hook node " << node;
JLOG(journal_.warn()) << "unable to hook node " << nodeID;
JLOG(journal_.info()) << " stuck at " << currNodeID;
JLOG(journal_.info()) << "got depth=" << node.getDepth()
JLOG(journal_.info()) << "got depth=" << nodeID.getDepth()
<< ", walked to= " << currNodeID.getDepth();
return SHAMapAddNode::useful();
}
if (backed_)
canonicalize(childHash, newNode);
canonicalize(childHash, treeNode);
newNode = prevNode->canonicalizeChild(branch, std::move(newNode));
treeNode = prevNode->canonicalizeChild(branch, std::move(treeNode));
if (filter)
{
Serializer s;
newNode->serializeWithPrefix(s);
treeNode->serializeWithPrefix(s);
filter->gotNode(
false, childHash, ledgerSeq_, std::move(s.modData()), newNode->getType());
false, childHash, ledgerSeq_, std::move(s.modData()), treeNode->getType());
}
return SHAMapAddNode::useful();

View File

@@ -0,0 +1,358 @@
#include <test/shamap/common.h>
#include <xrpld/app/ledger/detail/LedgerNodeHelpers.h>
#include <xrpl/beast/unit_test.h>
#include <xrpl/protocol/messages.h>
#include <xrpl/shamap/SHAMap.h>
#include <xrpl/shamap/SHAMapAccountStateLeafNode.h>
#include <xrpl/shamap/SHAMapInnerNode.h>
#include <xrpl/shamap/SHAMapItem.h>
#include <xrpl/shamap/SHAMapTreeNode.h>
#include <bit>
namespace xrpl {
namespace tests {
class LedgerNodeHelpers_test : public beast::unit_test::suite
{
// Helper function to create a simple SHAMapItem for testing.
static boost::intrusive_ptr<SHAMapItem>
makeTestItem(std::uint32_t seed)
{
Serializer s;
s.add32(seed);
s.add32(seed + 1);
s.add32(seed + 2);
return make_shamapitem(s.getSHA512Half(), s.slice());
}
// Helper function to serialize a tree node to wire format.
static std::string
serializeNode(intr_ptr::SharedPtr<SHAMapTreeNode> const& node)
{
Serializer s;
node->serializeForWire(s);
auto const slice = s.slice();
return std::string(std::bit_cast<char const*>(slice.data()), slice.size());
}
void
testValidateLedgerNode()
{
// In the tests below the validity of the content of the node data and ID fields is not
// checked - only that the fields have values when expected. The content of the fields is
// verified in the other tests in this file.
testcase("validateLedgerNode");
// Invalid: missing all fields.
{
protocol::TMLedgerNode node;
BEAST_EXPECT(!validateLedgerNode(node));
}
// Invalid: missing `nodedata` field.
{
protocol::TMLedgerNode node;
node.set_nodeid("test_nodeid");
BEAST_EXPECT(!validateLedgerNode(node));
}
// Invalid: missing `nodedata` field.
{
protocol::TMLedgerNode node;
node.set_id("test_nodeid");
BEAST_EXPECT(!validateLedgerNode(node));
}
// Invalid: missing `nodedata` field.
{
protocol::TMLedgerNode node;
node.set_depth(1);
BEAST_EXPECT(!validateLedgerNode(node));
}
// Valid: legacy `nodeid` field.
{
protocol::TMLedgerNode node;
node.set_nodedata("test_data");
node.set_nodeid("test_nodeid");
BEAST_EXPECT(validateLedgerNode(node));
}
// Invalid: has both legacy `nodeid` and new `id` fields.
{
protocol::TMLedgerNode node;
node.set_nodedata("test_data");
node.set_nodeid("test_nodeid");
node.set_id("test_nodeid");
BEAST_EXPECT(!validateLedgerNode(node));
}
// Invalid: has both legacy `nodeid` and new `depth` fields.
{
protocol::TMLedgerNode node;
node.set_nodedata("test_data");
node.set_nodeid("test_nodeid");
node.set_depth(5);
BEAST_EXPECT(!validateLedgerNode(node));
}
// Valid: new `id` field.
{
protocol::TMLedgerNode node;
node.set_nodedata("test_data");
node.set_id("test_id");
BEAST_EXPECT(validateLedgerNode(node));
}
// Valid: new `depth` field.
{
protocol::TMLedgerNode node;
node.set_nodedata("test_data");
node.set_depth(5);
BEAST_EXPECT(validateLedgerNode(node));
}
// Valid: `depth` at minimum depth.
{
protocol::TMLedgerNode node;
node.set_nodedata("test_data");
node.set_depth(0);
BEAST_EXPECT(validateLedgerNode(node));
}
// Valid: `depth` at arbitrary depth between minimum and maximum.
{
protocol::TMLedgerNode node;
node.set_nodedata("test_data");
node.set_depth(10);
BEAST_EXPECT(validateLedgerNode(node));
}
// Valid: `depth` at maximum depth.
{
protocol::TMLedgerNode node;
node.set_nodedata("test_data");
node.set_depth(SHAMap::leafDepth);
BEAST_EXPECT(validateLedgerNode(node));
}
// Invalid: `depth` is greater than maximum depth.
{
protocol::TMLedgerNode node;
node.set_nodedata("test_data");
node.set_depth(SHAMap::leafDepth + 1);
BEAST_EXPECT(!validateLedgerNode(node));
}
}
void
testGetTreeNode()
{
testcase("getTreeNode");
// Valid: inner node. It must have at least one child for `serializeNode` to work.
{
auto const innerNode = intr_ptr::make_shared<SHAMapInnerNode>(1);
auto const childNode = intr_ptr::make_shared<SHAMapInnerNode>(1);
innerNode->setChild(0, childNode);
auto const innerData = serializeNode(innerNode);
auto const result = getTreeNode(innerData);
BEAST_EXPECT(result.has_value());
BEAST_EXPECT((*result)->isInner());
}
// Valid: leaf node.
{
auto const leafItem = makeTestItem(12345);
auto const leafNode =
intr_ptr::make_shared<SHAMapAccountStateLeafNode>(std::move(leafItem), 1);
auto const leafData = serializeNode(leafNode);
auto result = getTreeNode(leafData);
BEAST_EXPECT(result.has_value());
BEAST_EXPECT((*result)->isLeaf());
}
// Invalid: empty data.
{
auto const result = getTreeNode("");
BEAST_EXPECT(!result.has_value());
}
// Invalid: garbage data.
{
auto const result = getTreeNode("invalid");
BEAST_EXPECT(!result.has_value());
}
// Invalid: truncated data.
{
auto const leafItem = makeTestItem(54321);
auto const leafNode =
intr_ptr::make_shared<SHAMapAccountStateLeafNode>(std::move(leafItem), 1);
// Truncate the data to trigger an exception in SHAMapTreeNode::makeAccountState when
// the data is used to deserialize the node.
uint256 tag;
auto const leafData = serializeNode(leafNode).substr(0, tag.bytes - 1);
auto const result = getTreeNode(leafData);
BEAST_EXPECT(!result.has_value());
}
}
void
testGetSHAMapNodeID()
{
testcase("getSHAMapNodeID");
{
// Tests using inner nodes at various depths.
auto const innerNode = intr_ptr::make_shared<SHAMapInnerNode>(1);
auto const childNode = intr_ptr::make_shared<SHAMapInnerNode>(1);
innerNode->setChild(0, childNode);
auto const innerData = serializeNode(innerNode);
// Valid: legacy `nodeid` field at arbitrary depth.
{
auto const innerDepth = 3;
auto const innerID = SHAMapNodeID::createID(innerDepth, uint256{});
protocol::TMLedgerNode node;
node.set_nodedata(innerData);
node.set_nodeid(innerID.getRawString());
auto const result = getSHAMapNodeID(node, innerNode);
BEAST_EXPECT(result.has_value());
BEAST_EXPECT(*result == innerID);
}
// Valid: new `id` field at minimum depth.
{
auto const innerDepth = 0;
auto const innerID = SHAMapNodeID::createID(innerDepth, uint256{});
protocol::TMLedgerNode node;
node.set_nodedata(innerData);
node.set_id(innerID.getRawString());
auto const result = getSHAMapNodeID(node, innerNode);
BEAST_EXPECT(result.has_value());
BEAST_EXPECT(*result == innerID);
}
// Invalid: new `depth` field should not be used for inner nodes.
{
protocol::TMLedgerNode node;
node.set_nodedata(innerData);
node.set_depth(10);
auto const result = getSHAMapNodeID(node, innerNode);
BEAST_EXPECT(!result.has_value());
}
}
{
// Tests using leaf nodes at various depths.
auto const leafItem = makeTestItem(12345);
auto const leafNode = intr_ptr::make_shared<SHAMapAccountStateLeafNode>(leafItem, 1);
auto const leafData = serializeNode(leafNode);
auto const leafKey = leafItem->key();
// Valid: legacy `nodeid` field at arbitrary depth.
{
auto const leafDepth = 5;
auto const leafID = SHAMapNodeID::createID(leafDepth, leafKey);
protocol::TMLedgerNode ledgerNode;
ledgerNode.set_nodedata(leafData);
ledgerNode.set_nodeid(leafID.getRawString());
auto result = getSHAMapNodeID(ledgerNode, leafNode);
BEAST_EXPECT(result.has_value());
BEAST_EXPECT(*result == leafID);
}
// Invalid: new `id` field should not be used for leaf nodes.
{
auto const leafDepth = 5;
auto const leafID = SHAMapNodeID::createID(leafDepth, leafKey);
protocol::TMLedgerNode ledgerNode;
ledgerNode.set_nodedata(leafData);
ledgerNode.set_id(leafID.getRawString());
auto result = getSHAMapNodeID(ledgerNode, leafNode);
BEAST_EXPECT(!result.has_value());
}
// Valid: new `depth` field at minimum depth.
{
auto const leafDepth = 0;
auto const leafID = SHAMapNodeID::createID(leafDepth, leafKey);
protocol::TMLedgerNode node;
node.set_nodedata(leafData);
node.set_depth(leafDepth);
auto result = getSHAMapNodeID(node, leafNode);
BEAST_EXPECT(result.has_value());
BEAST_EXPECT(*result == leafID);
}
// Valid: new `depth` field at arbitrary depth between minimum and maximum.
{
auto const leafDepth = 10;
auto const leafID = SHAMapNodeID::createID(leafDepth, leafKey);
protocol::TMLedgerNode ledgerNode;
ledgerNode.set_nodedata(leafData);
ledgerNode.set_depth(leafDepth);
auto result = getSHAMapNodeID(ledgerNode, leafNode);
BEAST_EXPECT(result.has_value());
BEAST_EXPECT(*result == leafID);
}
// Valid: new `depth` field at maximum depth.
// Note that we do not test a depth greater than the maximum depth, because the proto
// message is assumed to have been validated by the time the getSHAMapNodeID function is
// called.
{
auto const leafDepth = SHAMap::leafDepth;
auto const leafID = SHAMapNodeID::createID(leafDepth, leafKey);
protocol::TMLedgerNode node;
node.set_nodedata(leafData);
node.set_depth(leafDepth);
auto result = getSHAMapNodeID(node, leafNode);
BEAST_EXPECT(result.has_value());
BEAST_EXPECT(*result == leafID);
}
// Invalid: legacy `nodeid` field where the node ID is inconsistent with the key.
{
auto const otherItem = makeTestItem(54321);
auto const otherNode =
intr_ptr::make_shared<SHAMapAccountStateLeafNode>(otherItem, 1);
auto const otherData = serializeNode(otherNode);
auto const otherKey = otherItem->key();
auto const otherDepth = 1;
auto const otherID = SHAMapNodeID::createID(otherDepth, otherKey);
protocol::TMLedgerNode ledgerNode;
ledgerNode.set_nodedata(otherData);
ledgerNode.set_nodeid(otherID.getRawString());
auto result = getSHAMapNodeID(ledgerNode, leafNode);
BEAST_EXPECT(!result.has_value());
}
}
}
public:
void
run() override
{
testValidateLedgerNode();
testGetTreeNode();
testGetSHAMapNodeID();
}
};
BEAST_DEFINE_TESTSUITE(LedgerNodeHelpers, app, xrpl);
} // namespace tests
} // namespace xrpl

View File

@@ -520,13 +520,6 @@ public:
srcParams.set("type", srcBackendType);
srcParams.set("path", node_db.path());
beast::temp_dir dest_db;
Section destParams;
destParams.set("type", destBackendType);
destParams.set("path", dest_db.path());
testcase("import into '" + destBackendType + "' from '" + srcBackendType + "'");
// Create a batch
auto batch = createPredictableBatch(numObjectsToTest, seedValue);
@@ -545,9 +538,16 @@ public:
Manager::instance().make_Database(megabytes(4), scheduler, 2, srcParams, journal_);
// Set up the destination database
beast::temp_dir dest_db;
Section destParams;
destParams.set("type", destBackendType);
destParams.set("path", dest_db.path());
std::unique_ptr<Database> dest =
Manager::instance().make_Database(megabytes(4), scheduler, 2, destParams, journal_);
testcase("import into '" + destBackendType + "' from '" + srcBackendType + "'");
// Do the import
dest->importDatabase(*src);

View File

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

View File

@@ -103,14 +103,17 @@ public:
destination.setSynching();
{
std::vector<std::pair<SHAMapNodeID, Blob>> a;
std::vector<std::tuple<SHAMapNodeID, Blob, bool>> a;
BEAST_EXPECT(source.getNodeFat(SHAMapNodeID(), a, rand_bool(eng_), rand_int(eng_, 2)));
unexpected(a.empty(), "NodeSize");
BEAST_EXPECT(destination.addRootNode(source.getHash(), makeSlice(a[0].second), nullptr)
.isGood());
auto node = SHAMapTreeNode::makeFromWire(makeSlice(std::get<1>(a[0])));
if (!node)
fail("", __FILE__, __LINE__);
BEAST_EXPECT(
destination.addRootNode(source.getHash(), std::move(node), nullptr).isGood());
}
do
@@ -124,7 +127,7 @@ public:
break;
// get as many nodes as possible based on this information
std::vector<std::pair<SHAMapNodeID, Blob>> b;
std::vector<std::tuple<SHAMapNodeID, Blob, bool>> b;
for (auto& it : nodesMissing)
{
@@ -146,7 +149,10 @@ public:
// Don't use BEAST_EXPECT here b/c it will be called a
// non-deterministic number of times and the number of tests run
// should be deterministic
if (!destination.addKnownNode(b[i].first, makeSlice(b[i].second), nullptr)
auto node = SHAMapTreeNode::makeFromWire(makeSlice(std::get<1>(b[i])));
if (!node)
fail("", __FILE__, __LINE__);
if (!destination.addKnownNode(std::get<0>(b[i]), std::move(node), nullptr)
.isUseful())
fail("", __FILE__, __LINE__);
}

View File

@@ -53,7 +53,3 @@ if(NOT WIN32)
target_link_libraries(xrpl.test.net PRIVATE xrpl.imports.test)
add_dependencies(xrpl.tests xrpl.test.net)
endif()
xrpl_add_test(nodestore)
target_link_libraries(xrpl.test.nodestore PRIVATE xrpl.imports.test)
add_dependencies(xrpl.tests xrpl.test.nodestore)

View File

@@ -1,334 +0,0 @@
#include <xrpl/nodestore/Backend.h>
#include <gtest/gtest.h>
#include <algorithm>
#include <vector>
using namespace xrpl;
using namespace xrpl::NodeStore;
// Helper function to convert the pair result into ranges for testing.
std::vector<std::pair<unsigned int, unsigned int>>
calculateRanges(unsigned int batchSize, unsigned int maxThreadCount)
{
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
std::vector<std::pair<unsigned int, unsigned int>> ranges;
ranges.reserve(numThreads);
for (unsigned int t = 0; t < numThreads; ++t)
{
auto const startIdx = t * numItems;
auto const endIdx = std::min(startIdx + numItems, batchSize);
ranges.emplace_back(startIdx, endIdx);
}
return ranges;
}
TEST(BatchParallelism, EmptyBatch)
{
// Empty batch should return 0 threads.
{
auto const batchSize = 0u;
auto const maxThreadCount = 8u;
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
EXPECT_EQ(numThreads, 0u);
EXPECT_EQ(numItems, 0u);
// Verify ranges calculation.
auto const ranges = calculateRanges(batchSize, maxThreadCount);
EXPECT_EQ(ranges.size(), numThreads);
}
}
TEST(BatchParallelism, SmallBatches)
{
// Batch size 1 should use 1 thread.
{
auto const batchSize = 1u;
auto const maxThreadCount = 8u;
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
EXPECT_EQ(numThreads, 1u);
EXPECT_EQ(numItems, 1u);
auto const ranges = calculateRanges(batchSize, maxThreadCount);
ASSERT_EQ(ranges.size(), numThreads);
EXPECT_EQ(ranges[0].first, 0u);
EXPECT_EQ(ranges[0].second, 1u);
}
// Batch size 2 should use 1 thread.
{
auto const batchSize = 2u;
auto const maxThreadCount = 8u;
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
EXPECT_EQ(numThreads, 1u);
EXPECT_EQ(numItems, 2u);
auto const ranges = calculateRanges(batchSize, maxThreadCount);
ASSERT_EQ(ranges.size(), numThreads);
EXPECT_EQ(ranges[0].first, 0u);
EXPECT_EQ(ranges[0].second, 2u);
}
// Batch size 3 should use 1 thread.
{
auto const batchSize = 3u;
auto const maxThreadCount = 8u;
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
EXPECT_EQ(numThreads, 1u);
EXPECT_EQ(numItems, 3u);
auto const ranges = calculateRanges(batchSize, maxThreadCount);
ASSERT_EQ(ranges.size(), numThreads);
EXPECT_EQ(ranges[0].first, 0u);
EXPECT_EQ(ranges[0].second, 3u);
}
// Batch size 4 should use 1 thread (exactly 4 items).
{
auto const batchSize = 4u;
auto const maxThreadCount = 8u;
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
EXPECT_EQ(numThreads, 1u);
EXPECT_EQ(numItems, 4u);
auto const ranges = calculateRanges(batchSize, maxThreadCount);
ASSERT_EQ(ranges.size(), numThreads);
EXPECT_EQ(ranges[0].first, 0u);
EXPECT_EQ(ranges[0].second, 4u);
}
}
TEST(BatchParallelism, MediumBatches)
{
// Batch size 5 should use 2 threads.
{
auto const batchSize = 5u;
auto const maxThreadCount = 8u;
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
EXPECT_EQ(numThreads, 2u); // ceil(5/4) = 2
EXPECT_EQ(numItems, 3u); // ceil(5/2) = 3
auto const ranges = calculateRanges(batchSize, maxThreadCount);
ASSERT_EQ(ranges.size(), numThreads);
EXPECT_EQ(ranges[0].first, 0u);
EXPECT_EQ(ranges[0].second, 3u);
EXPECT_EQ(ranges[1].first, 3u);
EXPECT_EQ(ranges[1].second, 5u);
}
// Batch size 8 should use 2 threads.
{
auto const batchSize = 8u;
auto const maxThreadCount = 8u;
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
EXPECT_EQ(numThreads, 2u);
EXPECT_EQ(numItems, 4u);
auto const ranges = calculateRanges(batchSize, maxThreadCount);
ASSERT_EQ(ranges.size(), numThreads);
for (size_t i = 0; i < numThreads; ++i)
{
EXPECT_EQ(ranges[i].first, i * numItems);
EXPECT_EQ(ranges[i].second, (i + 1) * numItems);
}
}
// Batch size 15 should use 4 threads (ceil(15/4) = 4).
{
auto const batchSize = 15u;
auto const maxThreadCount = 8u;
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
EXPECT_EQ(numThreads, 4u);
EXPECT_EQ(numItems, 4u);
auto const ranges = calculateRanges(batchSize, maxThreadCount);
ASSERT_EQ(ranges.size(), numThreads);
for (size_t i = 0; i < numThreads - 1; ++i)
{
EXPECT_EQ(ranges[i].first, i * numItems);
EXPECT_EQ(ranges[i].second, (i + 1) * numItems);
}
EXPECT_EQ(ranges[numThreads - 1].first, (numThreads - 1) * numItems);
EXPECT_EQ(ranges[numThreads - 1].second, batchSize); // Last range gets remaining items.
}
// Batch size 22 should use 6 threads.
{
auto const batchSize = 22u;
auto const maxThreadCount = 8u;
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
EXPECT_EQ(numThreads, 6u); // ceil(22/4) = 6
EXPECT_EQ(numItems, 4u);
auto const ranges = calculateRanges(batchSize, maxThreadCount);
ASSERT_EQ(ranges.size(), numThreads);
for (size_t i = 0; i < numThreads - 1; ++i)
{
EXPECT_EQ(ranges[i].first, i * numItems);
EXPECT_EQ(ranges[i].second, (i + 1) * numItems);
}
EXPECT_EQ(ranges[numThreads - 1].first, (numThreads - 1) * numItems);
EXPECT_EQ(ranges[numThreads - 1].second, batchSize);
}
// Batch size 32 should use 8 threads.
{
auto const batchSize = 32u;
auto const maxThreadCount = 8u;
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
EXPECT_EQ(numThreads, 8u);
EXPECT_EQ(numItems, 4u);
auto const ranges = calculateRanges(batchSize, maxThreadCount);
ASSERT_EQ(ranges.size(), numThreads);
for (size_t i = 0; i < numThreads; ++i)
{
EXPECT_EQ(ranges[i].first, i * numItems);
EXPECT_EQ(ranges[i].second, (i + 1) * numItems);
}
}
}
TEST(BatchParallelism, LargeBatches)
{
// Batch size 100 should use 8 threads (max limit).
{
auto const batchSize = 100u;
auto const maxThreadCount = 8u;
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
EXPECT_EQ(numThreads, 8u);
EXPECT_EQ(numItems, 13u); // ceil(100/8) = 13
auto const ranges = calculateRanges(batchSize, maxThreadCount);
ASSERT_EQ(ranges.size(), numThreads);
for (size_t i = 0; i < numThreads - 1; ++i)
{
EXPECT_EQ(ranges[i].first, i * numItems);
EXPECT_EQ(ranges[i].second, (i + 1) * numItems);
}
EXPECT_EQ(ranges[numThreads - 1].first, (numThreads - 1) * numItems);
EXPECT_EQ(ranges[numThreads - 1].second, batchSize);
}
// Batch size 1000 with 8 hw threads.
{
auto const batchSize = 1000u;
auto const maxThreadCount = 8u;
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
EXPECT_EQ(numThreads, 8u);
EXPECT_EQ(numItems, 125u);
auto const ranges = calculateRanges(batchSize, maxThreadCount);
ASSERT_EQ(ranges.size(), numThreads);
for (size_t i = 0; i < numThreads; ++i)
{
EXPECT_EQ(ranges[i].first, i * numItems);
EXPECT_EQ(ranges[i].second, (i + 1) * numItems);
}
}
}
TEST(BatchParallelism, HardwareThreadLimits)
{
// With only 1 thread available.
{
auto const batchSize = 100u;
auto const maxThreadCount = 1u;
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
EXPECT_EQ(numThreads, 1u);
EXPECT_EQ(numItems, 100u);
auto const ranges = calculateRanges(batchSize, maxThreadCount);
ASSERT_EQ(ranges.size(), numThreads);
EXPECT_EQ(ranges[0].first, 0u);
EXPECT_EQ(ranges[0].second, 100u);
}
// With 2 threads.
{
auto const batchSize = 50u;
auto const maxThreadCount = 2u;
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
EXPECT_EQ(numThreads, 2u);
EXPECT_EQ(numItems, 25u);
auto const ranges = calculateRanges(batchSize, maxThreadCount);
ASSERT_EQ(ranges.size(), numThreads);
for (size_t i = 0; i < numThreads; ++i)
{
EXPECT_EQ(ranges[i].first, i * numItems);
EXPECT_EQ(ranges[i].second, (i + 1) * numItems);
}
}
// With 10 threads.
{
auto const batchSize = 50u;
auto const maxThreadCount = 12u;
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
EXPECT_EQ(numThreads, 10u); // ceil(50/4) = 13, but numThreads = 10.
EXPECT_EQ(numItems, 5u);
auto const ranges = calculateRanges(batchSize, maxThreadCount);
ASSERT_EQ(ranges.size(), numThreads);
for (size_t i = 0; i < numThreads; ++i)
{
EXPECT_EQ(ranges[i].first, i * numItems);
EXPECT_EQ(ranges[i].second, (i + 1) * numItems);
}
}
// With many threads.
{
auto const batchSize = 20u;
auto const maxThreadCount = 100u;
auto const [numThreads, numItems] =
Backend::calculateBatchParallelism(batchSize, maxThreadCount);
EXPECT_EQ(numThreads, 5u); // ceil(20/4) = 5, limited by batch size.
EXPECT_EQ(numItems, 4u);
auto const ranges = calculateRanges(batchSize, maxThreadCount);
ASSERT_EQ(ranges.size(), numThreads);
for (size_t i = 0; i < numThreads; ++i)
{
EXPECT_EQ(ranges[i].first, i * numItems);
EXPECT_EQ(ranges[i].second, (i + 1) * numItems);
}
}
}

View File

@@ -1,8 +0,0 @@
#include <gtest/gtest.h>
int
main(int argc, char** argv)
{
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}

View File

@@ -9,6 +9,7 @@
#include <mutex>
#include <set>
#include <string_view>
#include <utility>
namespace xrpl {
@@ -131,16 +132,16 @@ private:
processData(std::shared_ptr<Peer> peer, protocol::TMLedgerData& data);
bool
takeHeader(std::string const& data);
takeHeader(std::string_view data);
void
receiveNode(protocol::TMLedgerData& packet, SHAMapAddNode&);
receiveNode(protocol::TMLedgerData& packet, SHAMapAddNode& san);
bool
takeTxRootNode(Slice const& data, SHAMapAddNode&);
takeTxRootNode(std::string_view data, SHAMapAddNode& san);
bool
takeAsRootNode(Slice const& data, SHAMapAddNode&);
takeAsRootNode(std::string_view data, SHAMapAddNode& san);
std::vector<uint256>
neededTxHashes(int max, SHAMapSyncFilter* filter) const;

View File

@@ -3,6 +3,7 @@
#include <xrpld/app/ledger/InboundLedgers.h>
#include <xrpld/app/ledger/LedgerMaster.h>
#include <xrpld/app/ledger/TransactionStateSF.h>
#include <xrpld/app/ledger/detail/LedgerNodeHelpers.h>
#include <xrpld/app/main/Application.h>
#include <xrpld/overlay/Overlay.h>
@@ -768,7 +769,7 @@ InboundLedger::filterNodes(
*/
// data must not have hash prefix
bool
InboundLedger::takeHeader(std::string const& data)
InboundLedger::takeHeader(std::string_view data)
{
// Return value: true=normal, false=bad data
JLOG(journal_.trace()) << "got header acquiring ledger " << hash_;
@@ -855,20 +856,31 @@ InboundLedger::receiveNode(protocol::TMLedgerData& packet, SHAMapAddNode& san)
{
auto const f = filter.get();
for (auto const& node : packet.nodes())
for (auto const& ledger_node : packet.nodes())
{
auto const nodeID = deserializeSHAMapNodeID(node.nodeid());
auto treeNode = getTreeNode(ledger_node.nodedata());
if (!treeNode)
{
JLOG(journal_.warn()) << "Got invalid node data";
san.incInvalid();
return;
}
auto const nodeID = getSHAMapNodeID(ledger_node, *treeNode);
if (!nodeID)
throw std::runtime_error("data does not properly deserialize");
{
JLOG(journal_.warn()) << "Got invalid node id";
san.incInvalid();
return;
}
if (nodeID->isRoot())
{
san += map.addRootNode(rootHash, makeSlice(node.nodedata()), f);
san += map.addRootNode(rootHash, std::move(*treeNode), f);
}
else
{
san += map.addKnownNode(*nodeID, makeSlice(node.nodedata()), f);
san += map.addKnownNode(*nodeID, std::move(*treeNode), f);
}
if (!san.isGood())
@@ -908,7 +920,7 @@ InboundLedger::receiveNode(protocol::TMLedgerData& packet, SHAMapAddNode& san)
Call with a lock
*/
bool
InboundLedger::takeAsRootNode(Slice const& data, SHAMapAddNode& san)
InboundLedger::takeAsRootNode(std::string_view data, SHAMapAddNode& san)
{
if (failed_ || mHaveState)
{
@@ -924,9 +936,17 @@ InboundLedger::takeAsRootNode(Slice const& data, SHAMapAddNode& san)
// LCOV_EXCL_STOP
}
auto treeNode = getTreeNode(data);
if (!treeNode)
{
JLOG(journal_.warn()) << "Got invalid node data";
san.incInvalid();
return false;
}
AccountStateSF filter(mLedger->stateMap().family().db(), app_.getLedgerMaster());
san +=
mLedger->stateMap().addRootNode(SHAMapHash{mLedger->header().accountHash}, data, &filter);
san += mLedger->stateMap().addRootNode(
SHAMapHash{mLedger->header().accountHash}, std::move(*treeNode), &filter);
return san.isGood();
}
@@ -934,7 +954,7 @@ InboundLedger::takeAsRootNode(Slice const& data, SHAMapAddNode& san)
Call with a lock
*/
bool
InboundLedger::takeTxRootNode(Slice const& data, SHAMapAddNode& san)
InboundLedger::takeTxRootNode(std::string_view data, SHAMapAddNode& san)
{
if (failed_ || mHaveTransactions)
{
@@ -950,8 +970,17 @@ InboundLedger::takeTxRootNode(Slice const& data, SHAMapAddNode& san)
// LCOV_EXCL_STOP
}
auto treeNode = getTreeNode(data);
if (!treeNode)
{
JLOG(journal_.warn()) << "Got invalid node data";
san.incInvalid();
return false;
}
TransactionStateSF filter(mLedger->txMap().family().db(), app_.getLedgerMaster());
san += mLedger->txMap().addRootNode(SHAMapHash{mLedger->header().txHash}, data, &filter);
san += mLedger->txMap().addRootNode(
SHAMapHash{mLedger->header().txHash}, std::move(*treeNode), &filter);
return san.isGood();
}
@@ -1048,13 +1077,13 @@ InboundLedger::processData(std::shared_ptr<Peer> peer, protocol::TMLedgerData& p
}
if (!mHaveState && (packet.nodes().size() > 1) &&
!takeAsRootNode(makeSlice(packet.nodes(1).nodedata()), san))
!takeAsRootNode(packet.nodes(1).nodedata(), san))
{
JLOG(journal_.warn()) << "Included AS root invalid";
}
if (!mHaveTransactions && (packet.nodes().size() > 2) &&
!takeTxRootNode(makeSlice(packet.nodes(2).nodedata()), san))
!takeTxRootNode(packet.nodes(2).nodedata(), san))
{
JLOG(journal_.warn()) << "Included TX root invalid";
}
@@ -1085,13 +1114,13 @@ InboundLedger::processData(std::shared_ptr<Peer> peer, protocol::TMLedgerData& p
ScopedLockType sl(mtx_);
// Verify node IDs and data are complete
for (auto const& node : packet.nodes())
// Verify nodes are complete
for (auto const& ledger_node : packet.nodes())
{
if (!node.has_nodeid() || !node.has_nodedata())
if (!validateLedgerNode(ledger_node))
{
JLOG(journal_.warn()) << "Got bad node";
peer->charge(Resource::feeMalformedRequest, "ledger_data bad node");
JLOG(journal_.warn()) << "Got malformed ledger node";
peer->charge(Resource::feeMalformedRequest, "ledger_node");
return -1;
}
}

View File

@@ -1,5 +1,6 @@
#include <xrpld/app/ledger/InboundLedgers.h>
#include <xrpld/app/ledger/LedgerMaster.h>
#include <xrpld/app/ledger/detail/LedgerNodeHelpers.h>
#include <xrpld/app/main/Application.h>
#include <xrpl/basics/DecayingSample.h>
@@ -224,23 +225,21 @@ public:
Serializer s;
try
{
for (int i = 0; i < packet_ptr->nodes().size(); ++i)
for (auto const& ledger_node : packet_ptr->nodes())
{
auto const& node = packet_ptr->nodes(i);
if (!node.has_nodeid() || !node.has_nodedata())
if (!validateLedgerNode(ledger_node))
return;
auto newNode = SHAMapTreeNode::makeFromWire(makeSlice(node.nodedata()));
if (!newNode)
auto const treeNode = getTreeNode(ledger_node.nodedata());
if (!treeNode)
return;
auto const tn = *treeNode;
s.erase();
newNode->serializeWithPrefix(s);
tn->serializeWithPrefix(s);
app_.getLedgerMaster().addFetchPack(
newNode->getHash().as_uint256(), std::make_shared<Blob>(s.begin(), s.end()));
tn->getHash().as_uint256(), std::make_shared<Blob>(s.begin(), s.end()));
}
}
catch (std::exception const&) // NOLINT(bugprone-empty-catch)

View File

@@ -1,5 +1,6 @@
#include <xrpld/app/ledger/InboundLedgers.h>
#include <xrpld/app/ledger/InboundTransactions.h>
#include <xrpld/app/ledger/detail/LedgerNodeHelpers.h>
#include <xrpld/app/ledger/detail/TransactionAcquire.h>
#include <xrpld/app/main/Application.h>
@@ -132,26 +133,35 @@ public:
return;
}
std::vector<std::pair<SHAMapNodeID, Slice>> data;
std::vector<std::pair<SHAMapNodeID, intr_ptr::SharedPtr<SHAMapTreeNode>>> data;
data.reserve(packet.nodes().size());
for (auto const& node : packet.nodes())
for (auto const& ledger_node : packet.nodes())
{
if (!node.has_nodeid() || !node.has_nodedata())
if (!validateLedgerNode(ledger_node))
{
peer->charge(Resource::feeMalformedRequest, "ledger_data");
JLOG(j_.warn()) << "Got malformed ledger node";
peer->charge(Resource::feeMalformedRequest, "ledger_node");
return;
}
auto const id = deserializeSHAMapNodeID(node.nodeid());
if (!id)
auto treeNode = getTreeNode(ledger_node.nodedata());
if (!treeNode)
{
peer->charge(Resource::feeInvalidData, "ledger_data");
JLOG(j_.warn()) << "Got invalid node data";
peer->charge(Resource::feeInvalidData, "node_data");
return;
}
data.emplace_back(std::make_pair(*id, makeSlice(node.nodedata())));
auto const nodeID = getSHAMapNodeID(ledger_node, *treeNode);
if (!nodeID)
{
JLOG(j_.warn()) << "Got invalid node id";
peer->charge(Resource::feeInvalidData, "node_id");
return;
}
data.emplace_back(std::make_pair(*nodeID, std::move(*treeNode)));
}
if (!ta->takeNodes(data, peer).isUseful())

View File

@@ -0,0 +1,95 @@
#include <xrpld/app/ledger/detail/LedgerNodeHelpers.h>
#include <xrpl/basics/IntrusivePointer.h>
#include <xrpl/basics/Slice.h>
#include <xrpl/basics/safe_cast.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/protocol/messages.h>
#include <xrpl/shamap/SHAMap.h>
#include <xrpl/shamap/SHAMapLeafNode.h>
#include <xrpl/shamap/SHAMapNodeID.h>
#include <xrpl/shamap/SHAMapTreeNode.h>
#include <optional>
#include <string>
namespace xrpl {
bool
validateLedgerNode(protocol::TMLedgerNode const& ledger_node)
{
if (!ledger_node.has_nodedata())
return false;
if (ledger_node.has_nodeid())
return !ledger_node.has_id() && !ledger_node.has_depth();
return ledger_node.has_id() ||
(ledger_node.has_depth() && ledger_node.depth() <= SHAMap::leafDepth);
}
std::optional<intr_ptr::SharedPtr<SHAMapTreeNode>>
getTreeNode(std::string_view data)
{
auto const slice = makeSlice(data);
try
{
auto treeNode = SHAMapTreeNode::makeFromWire(slice);
if (!treeNode)
return std::nullopt;
return treeNode;
}
catch (std::exception const&)
{
return std::nullopt;
}
}
std::optional<SHAMapNodeID>
getSHAMapNodeID(
protocol::TMLedgerNode const& ledger_node,
intr_ptr::SharedPtr<SHAMapTreeNode> const& treeNode)
{
if (ledger_node.has_id() || ledger_node.has_depth())
{
if (treeNode->isInner())
{
if (!ledger_node.has_id())
return std::nullopt;
return deserializeSHAMapNodeID(ledger_node.id());
}
if (treeNode->isLeaf())
{
if (!ledger_node.has_depth())
return std::nullopt;
auto const key =
safe_downcast<SHAMapLeafNode const*>(treeNode.get())->peekItem()->key();
return SHAMapNodeID::createID(ledger_node.depth(), key);
}
UNREACHABLE("xrpl::getSHAMapNodeID : tree node is neither inner nor leaf");
return std::nullopt;
}
if (!ledger_node.has_nodeid())
return std::nullopt;
auto const nodeID = deserializeSHAMapNodeID(ledger_node.nodeid());
if (!nodeID.has_value())
return std::nullopt;
if (treeNode->isLeaf())
{
auto const key = safe_downcast<SHAMapLeafNode const*>(treeNode.get())->peekItem()->key();
auto const expected_id = SHAMapNodeID::createID(static_cast<int>(nodeID->getDepth()), key);
if (nodeID->getNodeID() != expected_id.getNodeID())
return std::nullopt;
}
return nodeID;
}
} // namespace xrpl

View File

@@ -0,0 +1,75 @@
#pragma once
#include <xrpl/basics/IntrusivePointer.h>
#include <xrpl/shamap/SHAMapNodeID.h>
#include <xrpl/shamap/SHAMapTreeNode.h>
#include <optional>
#include <string_view>
namespace protocol {
class TMLedgerNode;
} // namespace protocol
namespace xrpl {
/**
* @brief Validates a ledger node proto message.
*
* This function checks whether a ledger node has the expected fields (for non-ledger base data):
* - The node must have `nodedata`.
* - If the legacy `nodeid` field is present then the new `id` and `depth` fields must not be
* present.
* - If the new `id` or `depth` fields are present (it is a oneof field, so only one of the two can
* be set) then the legacy `nodeid` must not be present.
* - If the `depth` field is present then it must be between 0 and SHAMap::leafDepth (inclusive).
*
* @param ledger_node The ledger node to validate.
* @return true if the ledger node has the expected fields, false otherwise.
*/
[[nodiscard]] bool
validateLedgerNode(protocol::TMLedgerNode const& ledger_node);
/**
* @brief Deserializes a SHAMapTreeNode from wire format data.
*
* This function attempts to create a SHAMapTreeNode from the provided data string. If the data is
* malformed or deserialization fails, the function returns std::nullopt instead of throwing an
* exception.
*
* @param data The serialized node data in wire format.
* @return An optional containing the deserialized tree node if successful, or std::nullopt if
* deserialization fails.
*/
[[nodiscard]] std::optional<intr_ptr::SharedPtr<SHAMapTreeNode>>
getTreeNode(std::string_view data);
/**
* @brief Extracts or reconstructs the SHAMapNodeID from a ledger node proto message.
*
* This function retrieves the SHAMapNodeID for a tree node, with behavior that depends on which
* field is set and the node type (inner vs. leaf).
*
* When the legacy `nodeid` field is set in the message:
* - For all nodes: Deserializes the node ID from the field.
* - For leaf nodes: Validates that the node ID is consistent with the leaf's key.
*
* When the new `id` or `depth` field is set in the message:
* - For inner nodes: Deserializes the node ID from the `id` field. Note that root nodes are also
* inner nodes.
* - For leaf nodes: Reconstructs the node ID using both the depth from the `depth` field and the
* key from the leaf node's item.
*
* @param ledger_node The validated protocol message containing the ledger node data.
* @param treeNode The deserialized tree node (inner or leaf node).
* @return An optional containing the node ID if extraction/reconstruction succeeds, or std::nullopt
* if the required fields are missing or validation fails.
* @note This function expects that the caller has already validated the ledger node by calling the
* `validateLedgerNode` function and obtained a valid tree node by calling `getTreeNode`.
*/
[[nodiscard]] std::optional<SHAMapNodeID>
getSHAMapNodeID(
protocol::TMLedgerNode const& ledger_node,
intr_ptr::SharedPtr<SHAMapTreeNode> const& treeNode);
} // namespace xrpl

View File

@@ -159,7 +159,7 @@ TransactionAcquire::trigger(std::shared_ptr<Peer> const& peer)
SHAMapAddNode
TransactionAcquire::takeNodes(
std::vector<std::pair<SHAMapNodeID, Slice>> const& data,
std::vector<std::pair<SHAMapNodeID, intr_ptr::SharedPtr<SHAMapTreeNode>>> const& data,
std::shared_ptr<Peer> const& peer)
{
ScopedLockType sl(mtx_);
@@ -183,7 +183,7 @@ TransactionAcquire::takeNodes(
ConsensusTransSetSF sf(app_, app_.getTempNodeCache());
for (auto const& d : data)
for (auto& d : data)
{
if (d.first.isRoot())
{
@@ -191,7 +191,8 @@ TransactionAcquire::takeNodes(
{
JLOG(journal_.debug()) << "Got root TXS node, already have it";
}
else if (!mMap->addRootNode(SHAMapHash{hash_}, d.second, nullptr).isGood())
else if (!mMap->addRootNode(SHAMapHash{hash_}, std::move(d.second), nullptr)
.isGood())
{
JLOG(journal_.warn()) << "TX acquire got bad root node";
}
@@ -200,7 +201,7 @@ TransactionAcquire::takeNodes(
mHaveRoot = true;
}
}
else if (!mMap->addKnownNode(d.first, d.second, &sf).isGood())
else if (!mMap->addKnownNode(d.first, std::move(d.second), &sf).isGood())
{
JLOG(journal_.warn()) << "TX acquire got bad non-root node";
return SHAMapAddNode::invalid();

View File

@@ -20,8 +20,8 @@ public:
SHAMapAddNode
takeNodes(
std::vector<std::pair<SHAMapNodeID, Slice>> const& data,
std::shared_ptr<Peer> const&);
std::vector<std::pair<SHAMapNodeID, intr_ptr::SharedPtr<SHAMapTreeNode>>> const& data,
std::shared_ptr<Peer> const& peer);
void
init(int startPeers);

View File

@@ -17,6 +17,7 @@ enum class ProtocolFeature {
ValidatorListPropagation,
ValidatorList2Propagation,
LedgerReplay,
LedgerNodeDepth,
};
/** Represents a peer connection in the overlay. */

View File

@@ -30,6 +30,7 @@
#include <mutex>
#include <numeric>
#include <sstream>
#include <tuple>
using namespace std::chrono_literals;
@@ -498,6 +499,8 @@ PeerImp::supportsFeature(ProtocolFeature f) const
return protocol_ >= make_protocol(2, 1);
case ProtocolFeature::ValidatorList2Propagation:
return protocol_ >= make_protocol(2, 2);
case ProtocolFeature::LedgerNodeDepth:
return protocol_ >= make_protocol(2, 3);
case ProtocolFeature::LedgerReplay:
return ledgerReplayEnabled_;
}
@@ -3388,13 +3391,19 @@ PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m)
std::uint32_t const defaultDepth = isHighLatency() ? 2 : 1;
auto const queryDepth{m->has_querydepth() ? m->querydepth() : defaultDepth};
std::vector<std::pair<SHAMapNodeID, Blob>> data;
std::vector<std::tuple<SHAMapNodeID, Blob, bool>> data;
auto const useLedgerNodeDepth = supportsFeature(ProtocolFeature::LedgerNodeDepth);
for (int i = 0;
i < m->nodeids_size() && ledgerData.nodes_size() < Tuning::softMaxReplyNodes;
++i)
{
auto const shaMapNodeId{deserializeSHAMapNodeID(m->nodeids(i))};
if (!shaMapNodeId.has_value())
{
JLOG(p_journal_.error()) << "processLedgerRequest: Invalid SHAMap node ID";
return;
}
data.clear();
data.reserve(Tuning::softMaxReplyNodes);
@@ -3410,9 +3419,28 @@ PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m)
{
if (ledgerData.nodes_size() >= Tuning::hardMaxReplyNodes)
break;
protocol::TMLedgerNode* node{ledgerData.add_nodes()};
node->set_nodeid(d.first.getRawString());
node->set_nodedata(d.second.data(), d.second.size());
auto const& nodeData = std::get<1>(d);
node->set_nodedata(nodeData.data(), nodeData.size());
// When the LedgerNodeDepth protocol feature is not supported by the peer,
// we always set the `nodeid` field. However, when it is supported then we
// set the `id` field for inner nodes and the `depth` field for leaf nodes.
auto const& nodeID = std::get<0>(d);
if (!useLedgerNodeDepth)
{
node->set_nodeid(nodeID.getRawString());
}
else if (std::get<2>(d))
{
node->set_depth(nodeID.getDepth());
}
else
{
node->set_id(nodeID.getRawString());
}
}
}
else

View File

@@ -21,7 +21,8 @@ namespace xrpl {
constexpr ProtocolVersion const supportedProtocolList[]
{
{2, 1},
{2, 2}
{2, 2},
{2, 3},
};
// clang-format on