diff --git a/.gitignore b/.gitignore index 6bd34ece04..5a700456ec 100644 --- a/.gitignore +++ b/.gitignore @@ -86,3 +86,5 @@ __pycache__ # clangd cache /.cache +labrun/ +build-release/ diff --git a/include/xrpl/ledger/detail/ApplyStateTable.h b/include/xrpl/ledger/detail/ApplyStateTable.h index 7b18f742b4..021ee98d8c 100644 --- a/include/xrpl/ledger/detail/ApplyStateTable.h +++ b/include/xrpl/ledger/detail/ApplyStateTable.h @@ -4,11 +4,15 @@ #include #include #include +#include #include #include #include #include +#ifndef NDEBUG +#include +#endif namespace xrpl::detail { @@ -103,7 +107,39 @@ public: return dropsDestroyed_; } +#ifndef NDEBUG + /** Every ledger entry this table has read or written, mapped to its type. + + Populated in DEBUG builds by the access methods below (reads via + read/exists/peek and writes via insert/update/replace/erase). Directory + iteration via succ() is deliberately NOT recorded — see AccessSet for + why directory entries are out of scope for conflict tracking. Used by + the parallel-apply access-set assertion to verify a transactor's + declared footprint is a superset of what it actually touched. + */ + [[nodiscard]] std::map const& + touchedEntries() const + { + return touched_; + } +#endif + private: +#ifndef NDEBUG + void + recordTouch(key_type const& key, LedgerEntryType type) const + { + // Prefer ltDIR_NODE on collision so directory pages stay identifiable + // regardless of access order; non-dir objects are never accessed via a + // directory keylet, so this never mislabels a real object. + auto const [it, inserted] = touched_.try_emplace(key, type); + if (!inserted && type == ltDIR_NODE) + it->second = ltDIR_NODE; + } + + mutable std::map touched_; +#endif + using Mods = hash_map>; static void diff --git a/include/xrpl/ledger/detail/ApplyViewBase.h b/include/xrpl/ledger/detail/ApplyViewBase.h index 558c9e5d4d..559c86072f 100644 --- a/include/xrpl/ledger/detail/ApplyViewBase.h +++ b/include/xrpl/ledger/detail/ApplyViewBase.h @@ -95,6 +95,16 @@ public: void rawDestroyXRP(XRPAmount const& feeDrops) override; +#ifndef NDEBUG + /** Every ledger entry touched (read or written) by this view, with type. + DEBUG-only; backs the parallel-apply access-set assertion. */ + [[nodiscard]] std::map const& + touchedEntries() const + { + return items_.touchedEntries(); + } +#endif + protected: ApplyFlags flags_; ReadView const* base_; diff --git a/include/xrpl/tx/AccessSet.h b/include/xrpl/tx/AccessSet.h new file mode 100644 index 0000000000..c34809e3a7 --- /dev/null +++ b/include/xrpl/tx/AccessSet.h @@ -0,0 +1,92 @@ +#pragma once + +#include + +#include + +#include + +namespace xrpl { + +/** The statically-declared ledger footprint of a single transaction. + + An AccessSet enumerates the ledger entries a transaction may read or write, + grouped by category for readability. It is produced by `accessSetOf()` from + the signed transaction body (and, where required, a read-only snapshot of + the closed ledger) — never from another transaction's in-flight writes. + + Two transactions are independent (safe to apply concurrently) iff their + AccessSets do not conflict; see `conflictsWith()`. A transaction whose + footprint cannot be enumerated statically sets `touchesGlobal` and is + serialized against everything. + + Directory-node (ltDIR_NODE) entries are intentionally NOT represented here. + Owner-directory bookkeeping is a consequence of mutating the owning account, + whose AccountRoot is already declared; shared (book) directories belong only + to transactors that are `touchesGlobal` in this version. Conflict detection + on directories is therefore subsumed by the account/object entries. + + The category split is for human readability and future scheduler heuristics; + conflict detection itself operates on the flat union, `keys()`. +*/ +struct AccessSet +{ + boost::container::flat_set accounts; // AccountRoot entries + boost::container::flat_set trustlines; // RippleState entries + boost::container::flat_set offerBooks; // book directory roots + boost::container::flat_set ammPools; // AMM root entries + boost::container::flat_set nftPages; // NFToken page roots + boost::container::flat_set miscObjects; // escrow/check/ticket/etc. + + /** When true, the footprint is not statically known; serialize this tx. */ + bool touchesGlobal = false; + + /** The conservative "I touch everything" footprint. */ + [[nodiscard]] static AccessSet + global() + { + AccessSet a; + a.touchesGlobal = true; + return a; + } + + /** The flat union of every declared entry key, across all categories. */ + [[nodiscard]] std::set + keys() const + { + std::set out; + for (auto const* s : + {&accounts, &trustlines, &offerBooks, &ammPools, &nftPages, &miscObjects}) + out.insert(s->begin(), s->end()); + return out; + } + + /** True if these two transactions cannot safely apply concurrently. + + Either side touching global state forces a conflict; otherwise the two + conflict iff their declared key sets intersect. + */ + [[nodiscard]] bool + conflictsWith(AccessSet const& other) const + { + if (touchesGlobal || other.touchesGlobal) + return true; + + auto const a = keys(); + auto const b = other.keys(); + auto i = a.begin(); + auto j = b.begin(); + while (i != a.end() && j != b.end()) + { + if (*i < *j) + ++i; + else if (*j < *i) + ++j; + else + return true; + } + return false; + } +}; + +} // namespace xrpl diff --git a/include/xrpl/tx/ApplyContext.h b/include/xrpl/tx/ApplyContext.h index 910ec6be42..5230fdfbf8 100644 --- a/include/xrpl/tx/ApplyContext.h +++ b/include/xrpl/tx/ApplyContext.h @@ -112,6 +112,24 @@ public: TER checkInvariants(TER const result, XRPAmount const fee); +#ifndef NDEBUG + /** The read-only base (closed-ledger) snapshot this tx applies against. */ + [[nodiscard]] ReadView const& + baseView() const + { + return base_; + } + + /** Every ledger entry touched during apply, with type. Backs the + access-set assertion in Transactor::operator(). */ + [[nodiscard]] std::map const& + touchedEntries() const + { + // NOLINTNEXTLINE(bugprone-unchecked-optional-access) view_ emplaced in ctor + return view_->touchedEntries(); + } +#endif + private: static TER failInvariantCheck(TER const result); diff --git a/include/xrpl/tx/Schedule.h b/include/xrpl/tx/Schedule.h new file mode 100644 index 0000000000..911d28eee3 --- /dev/null +++ b/include/xrpl/tx/Schedule.h @@ -0,0 +1,128 @@ +#pragma once + +#include +#include +#include +#include +#include + +#include +#include +#include + +namespace xrpl { + +class ServiceRegistry; + +/** A maximal set of transactions that conflict (transitively) with each other. + + The transactions are held in canonical (input) order, which is significant: + within a group they must be applied serially in this order. Two distinct + groups are independent by construction — their declared access sets are + disjoint — so groups may be applied concurrently, and the order in which + groups are applied does not affect the resulting state. +*/ +struct ConflictGroup +{ + std::vector> txns; +}; + +/** The partitioning of a canonically-ordered transaction set for parallel + application. + + Either the set was partitioned into independent `groups` (the parallel + case), or scheduling fell back to a single serial ordering in `serial` + (the conservative case — see `scheduleApply`). Exactly one of the two is + populated; `fullySerial` says which. +*/ +struct Schedule +{ + std::vector groups; + std::vector> serial; + bool fullySerial = false; + + /** Total transactions scheduled, across groups or the serial list. */ + [[nodiscard]] std::size_t + size() const + { + if (fullySerial) + return serial.size(); + std::size_t n = 0; + for (auto const& g : groups) + n += g.txns.size(); + return n; + } +}; + +/** Partition a canonically-ordered transaction set into independent groups by + static access-set conflict, for parallel application. + + For each transaction, `accessSetOf` is consulted against `base` (the closed + -ledger snapshot the round applies to). Transactions are unioned into the + same group iff their access sets conflict — i.e. they share any declared + ledger entry, or (implicitly, via the access set) act on the same account. + + If ANY transaction declares `touchesGlobal` (a pseudo-transaction such as + SetFee/EnableAmendment/UNLModify, or any not-yet-migrated transactor), the + schedule falls back to fully serial in canonical order. This is the + conservative flag-ledger handling: correctness over throughput, at a cost of + ~1/256 of ledgers. A later version may apply the global transactions first + and parallelize the remainder. + + Deterministic: identical input yields an identical schedule (groups are + ordered by their lowest canonical index, transactions within a group keep + canonical order). Applies nothing and reads only `base`. +*/ +Schedule +scheduleApply(std::vector> const& txns, ReadView const& base); + +/** Outcome of applyScheduled. */ +struct ScheduledApplyResult +{ + std::size_t groupCount = 0; // independent groups (1 if fully serial) + std::size_t applied = 0; // transactions that applied successfully + bool fullySerial = false; +}; + +/** Apply a canonically-ordered transaction set via its schedule. + + Schedules `txns` (see `scheduleApply`), then applies each independent group + in its own `OpenView` layered over the immutable `closed` snapshot, and + merges the per-group write-sets into `to`. Because distinct groups have + disjoint access sets, their write-sets are disjoint and the merge is + order-independent — so the resulting state in `to` is byte-identical to a + serial canonical apply. (That equivalence is what the differential test + asserts, and it is the correctness contract a parallel executor relies on.) + + `workers` controls concurrency of the per-group apply phase: + - `workers <= 1`: groups applied sequentially (deterministic baseline). + - `workers > 1`: up to `workers` groups apply concurrently, each in its own + view over the immutable `closed` snapshot; the write-sets are then merged + into `to` sequentially in deterministic group order. + The result is identical for any `workers` value (the merge is order- + independent because groups are disjoint, and is performed in a fixed order). + + NOTE on the threaded path: it is correct by construction here, but it is NOT + certified for production consensus use. A non-deterministic apply forks the + network, so before the threaded path may be trusted it needs ThreadSanitizer + clean runs, adversarial scheduling (Antithesis), and a mainnet-history replay + differential — none of which a unit test establishes. The concurrent reads + of `closed` and the shared `registry` are the surfaces that must be cleared. + + @param registry service registry used by the apply pipeline. + @param closed the immutable closed-ledger snapshot to read and schedule against. + @param to the target ledger receiving the merged writes. + @param txns transactions in canonical order. + @param j journal. + @param workers max concurrent group-apply threads (default 1 = sequential). +*/ +ScheduledApplyResult +applyScheduled( + ServiceRegistry& registry, + ReadView const& closed, + TxsRawView& to, + std::vector> const& txns, + beast::Journal j, + unsigned workers = 1); + +} // namespace xrpl diff --git a/include/xrpl/tx/Transactor.h b/include/xrpl/tx/Transactor.h index 1440a5097f..cb38fb8aa9 100644 --- a/include/xrpl/tx/Transactor.h +++ b/include/xrpl/tx/Transactor.h @@ -4,6 +4,7 @@ #include #include #include +#include #include #include @@ -222,10 +223,39 @@ public: return tesSUCCESS; } + /** The static ledger footprint this transaction may read or write. + + Consumed by the parallel-apply scheduler to decide which transactions + can apply concurrently. The base implementation is fail-safe: it + declares `touchesGlobal`, so an un-migrated transactor is serialized + against everything. A transactor opts into concurrency by hiding this + with its own static `accessSetOf` that enumerates exactly what it + touches. Like preclaim/calculateBaseFee, this is dispatched by + compile-time name hiding, not virtual dispatch. + + @param tx the signed transaction. + @param base a read-only snapshot of the ledger the tx applies to, for + the few transactors whose footprint needs a state lookup + (e.g. resolving an AMM pseudo-account). Most ignore it. + */ + static AccessSet + accessSetOf(STTx const& tx, ReadView const& base) + { + return AccessSet::global(); + } + static NotTEC checkPermission(ReadView const& view, STTx const& tx); ///////////////////////////////////////////////////// + /** The ledger footprint every transaction incurs, regardless of type: the + actor's AccountRoot, the fee-payer's AccountRoot (when delegated), and + the consumed Ticket object (when ticket-sequenced). Directory pages are + excluded by design — see AccessSet. A migrated `accessSetOf` seeds its + result with this and adds its type-specific entries. */ + static AccessSet + commonAccountFootprint(STTx const& tx); + // Interface used by AccountDelete static TER ticketDelete( @@ -380,6 +410,15 @@ private: void trapTransaction(uint256) const; +#ifndef NDEBUG + /** DEBUG: assert the transaction's actual ledger footprint is a subset of + the access set declared by `accessSetOf`, or — for `touchesGlobal` + transactors — optionally record the measured footprint for the + hard-transactor audit. Called only on a clean tesSUCCESS apply. */ + void + verifyAccessSet() const; +#endif + /** Performs early sanity checks on the account and fee fields. (And passes flagMask to preflight0) diff --git a/include/xrpl/tx/applySteps.h b/include/xrpl/tx/applySteps.h index 897cf500d6..d4a0082ceb 100644 --- a/include/xrpl/tx/applySteps.h +++ b/include/xrpl/tx/applySteps.h @@ -2,6 +2,7 @@ #include #include +#include namespace xrpl { @@ -308,6 +309,19 @@ preclaim(PreflightResult const& preflightResult, ServiceRegistry& registry, Open XRPAmount calculateBaseFee(ReadView const& view, STTx const& tx); +/** Compute the static ledger footprint of a transaction. + + Dispatches to the concrete transactor's `accessSetOf`. Un-migrated + transactors return `touchesGlobal`. Used by the parallel-apply scheduler + (and the DEBUG access-set assertion) to reason about which transactions can + apply concurrently. No validation is implied. + + @param tx the transaction. + @param base a read-only snapshot of the ledger the tx applies to. +*/ +AccessSet +accessSetOf(STTx const& tx, ReadView const& base); + /** Return the minimum fee that an "ordinary" transaction would pay. When computing the FeeLevel for a transaction the TxQ sometimes needs diff --git a/include/xrpl/tx/transactors/account/AccountSet.h b/include/xrpl/tx/transactors/account/AccountSet.h index 002779db64..32bb3c4c49 100644 --- a/include/xrpl/tx/transactors/account/AccountSet.h +++ b/include/xrpl/tx/transactors/account/AccountSet.h @@ -29,6 +29,9 @@ public: static TER preclaim(PreclaimContext const& ctx); + static AccessSet + accessSetOf(STTx const& tx, ReadView const& base); + TER doApply() override; diff --git a/include/xrpl/tx/transactors/account/SetRegularKey.h b/include/xrpl/tx/transactors/account/SetRegularKey.h index a9f1ce715d..c38b4ad298 100644 --- a/include/xrpl/tx/transactors/account/SetRegularKey.h +++ b/include/xrpl/tx/transactors/account/SetRegularKey.h @@ -19,6 +19,9 @@ public: static XRPAmount calculateBaseFee(ReadView const& view, STTx const& tx); + static AccessSet + accessSetOf(STTx const& tx, ReadView const& base); + TER doApply() override; diff --git a/include/xrpl/tx/transactors/account/SignerListSet.h b/include/xrpl/tx/transactors/account/SignerListSet.h index 760f6e9358..0e1452ba47 100644 --- a/include/xrpl/tx/transactors/account/SignerListSet.h +++ b/include/xrpl/tx/transactors/account/SignerListSet.h @@ -36,6 +36,9 @@ public: static NotTEC preflight(PreflightContext const& ctx); + static AccessSet + accessSetOf(STTx const& tx, ReadView const& base); + TER doApply() override; void diff --git a/include/xrpl/tx/transactors/delegate/DelegateSet.h b/include/xrpl/tx/transactors/delegate/DelegateSet.h index c93c48e970..8e5ad89584 100644 --- a/include/xrpl/tx/transactors/delegate/DelegateSet.h +++ b/include/xrpl/tx/transactors/delegate/DelegateSet.h @@ -19,6 +19,9 @@ public: static TER preclaim(PreclaimContext const& ctx); + static AccessSet + accessSetOf(STTx const& tx, ReadView const& base); + TER doApply() override; diff --git a/include/xrpl/tx/transactors/did/DIDDelete.h b/include/xrpl/tx/transactors/did/DIDDelete.h index c750d4c95e..f6564bcc59 100644 --- a/include/xrpl/tx/transactors/did/DIDDelete.h +++ b/include/xrpl/tx/transactors/did/DIDDelete.h @@ -22,6 +22,9 @@ public: static TER deleteSLE(ApplyView& view, std::shared_ptr sle, AccountID const owner, beast::Journal j); + static AccessSet + accessSetOf(STTx const& tx, ReadView const& base); + TER doApply() override; diff --git a/include/xrpl/tx/transactors/did/DIDSet.h b/include/xrpl/tx/transactors/did/DIDSet.h index b2c3d97c81..afc5a42667 100644 --- a/include/xrpl/tx/transactors/did/DIDSet.h +++ b/include/xrpl/tx/transactors/did/DIDSet.h @@ -16,6 +16,9 @@ public: static NotTEC preflight(PreflightContext const& ctx); + static AccessSet + accessSetOf(STTx const& tx, ReadView const& base); + TER doApply() override; diff --git a/include/xrpl/tx/transactors/oracle/OracleDelete.h b/include/xrpl/tx/transactors/oracle/OracleDelete.h index e83a334b87..6cad2d6f23 100644 --- a/include/xrpl/tx/transactors/oracle/OracleDelete.h +++ b/include/xrpl/tx/transactors/oracle/OracleDelete.h @@ -28,6 +28,9 @@ public: static TER preclaim(PreclaimContext const& ctx); + static AccessSet + accessSetOf(STTx const& tx, ReadView const& base); + TER doApply() override; diff --git a/include/xrpl/tx/transactors/oracle/OracleSet.h b/include/xrpl/tx/transactors/oracle/OracleSet.h index 12c022470b..3a3cf105f7 100644 --- a/include/xrpl/tx/transactors/oracle/OracleSet.h +++ b/include/xrpl/tx/transactors/oracle/OracleSet.h @@ -28,6 +28,9 @@ public: static TER preclaim(PreclaimContext const& ctx); + static AccessSet + accessSetOf(STTx const& tx, ReadView const& base); + TER doApply() override; diff --git a/include/xrpl/tx/transactors/payment/DepositPreauth.h b/include/xrpl/tx/transactors/payment/DepositPreauth.h index fbf22f8ca8..1957145919 100644 --- a/include/xrpl/tx/transactors/payment/DepositPreauth.h +++ b/include/xrpl/tx/transactors/payment/DepositPreauth.h @@ -22,6 +22,9 @@ public: static TER preclaim(PreclaimContext const& ctx); + static AccessSet + accessSetOf(STTx const& tx, ReadView const& base); + TER doApply() override; diff --git a/include/xrpl/tx/transactors/payment/Payment.h b/include/xrpl/tx/transactors/payment/Payment.h index ef42b67fa4..23d451cb0e 100644 --- a/include/xrpl/tx/transactors/payment/Payment.h +++ b/include/xrpl/tx/transactors/payment/Payment.h @@ -37,6 +37,9 @@ public: static TER preclaim(PreclaimContext const& ctx); + static AccessSet + accessSetOf(STTx const& tx, ReadView const& base); + TER doApply() override; diff --git a/include/xrpl/tx/transactors/permissioned_domain/PermissionedDomainDelete.h b/include/xrpl/tx/transactors/permissioned_domain/PermissionedDomainDelete.h index 77834f7683..e7388255ef 100644 --- a/include/xrpl/tx/transactors/permissioned_domain/PermissionedDomainDelete.h +++ b/include/xrpl/tx/transactors/permissioned_domain/PermissionedDomainDelete.h @@ -19,6 +19,9 @@ public: static TER preclaim(PreclaimContext const& ctx); + static AccessSet + accessSetOf(STTx const& tx, ReadView const& base); + /** Attempt to delete the Permissioned Domain. */ TER doApply() override; diff --git a/include/xrpl/tx/transactors/system/TicketCreate.h b/include/xrpl/tx/transactors/system/TicketCreate.h index 4991d1d08b..1dd31991d4 100644 --- a/include/xrpl/tx/transactors/system/TicketCreate.h +++ b/include/xrpl/tx/transactors/system/TicketCreate.h @@ -56,6 +56,9 @@ public: static TER preclaim(PreclaimContext const& ctx); + static AccessSet + accessSetOf(STTx const& tx, ReadView const& base); + /** Precondition: fee collection is likely. Attempt to create ticket(s). */ TER doApply() override; diff --git a/include/xrpl/tx/transactors/token/TrustSet.h b/include/xrpl/tx/transactors/token/TrustSet.h index d439b676b5..ad5bc1105c 100644 --- a/include/xrpl/tx/transactors/token/TrustSet.h +++ b/include/xrpl/tx/transactors/token/TrustSet.h @@ -26,6 +26,9 @@ public: static TER preclaim(PreclaimContext const& ctx); + static AccessSet + accessSetOf(STTx const& tx, ReadView const& base); + TER doApply() override; diff --git a/src/libxrpl/ledger/ApplyStateTable.cpp b/src/libxrpl/ledger/ApplyStateTable.cpp index 70fa0aef5d..2ad0b7ba01 100644 --- a/src/libxrpl/ledger/ApplyStateTable.cpp +++ b/src/libxrpl/ledger/ApplyStateTable.cpp @@ -287,6 +287,9 @@ ApplyStateTable::apply( bool ApplyStateTable::exists(ReadView const& base, Keylet const& k) const { +#ifndef NDEBUG + recordTouch(k.key, k.type); +#endif auto const iter = items_.find(k.key); if (iter == items_.end()) return base.exists(k); @@ -342,6 +345,11 @@ ApplyStateTable::succ( std::shared_ptr ApplyStateTable::read(ReadView const& base, Keylet const& k) const { +#ifndef NDEBUG + // Record even on a nullptr result: an absence probe still conflicts with a + // concurrent transaction that would create the key. + recordTouch(k.key, k.type); +#endif auto const iter = items_.find(k.key); if (iter == items_.end()) return base.read(k); @@ -364,6 +372,9 @@ ApplyStateTable::read(ReadView const& base, Keylet const& k) const std::shared_ptr ApplyStateTable::peek(ReadView const& base, Keylet const& k) { +#ifndef NDEBUG + recordTouch(k.key, k.type); +#endif auto iter = items_.lower_bound(k.key); if (iter == items_.end() || iter->first != k.key) { @@ -398,6 +409,9 @@ ApplyStateTable::peek(ReadView const& base, Keylet const& k) void ApplyStateTable::erase(ReadView const& base, std::shared_ptr const& sle) { +#ifndef NDEBUG + recordTouch(sle->key(), sle->getType()); +#endif auto const iter = items_.find(sle->key()); if (iter == items_.end()) Throw("ApplyStateTable::erase: missing key"); @@ -422,6 +436,9 @@ ApplyStateTable::erase(ReadView const& base, std::shared_ptr const& sle) void ApplyStateTable::rawErase(ReadView const& base, std::shared_ptr const& sle) { +#ifndef NDEBUG + recordTouch(sle->key(), sle->getType()); +#endif using namespace std; auto const result = items_.emplace( piecewise_construct, forward_as_tuple(sle->key()), forward_as_tuple(Action::Erase, sle)); @@ -447,6 +464,9 @@ ApplyStateTable::rawErase(ReadView const& base, std::shared_ptr const& sle) void ApplyStateTable::insert(ReadView const& base, std::shared_ptr const& sle) { +#ifndef NDEBUG + recordTouch(sle->key(), sle->getType()); +#endif auto const iter = items_.lower_bound(sle->key()); if (iter == items_.end() || iter->first != sle->key()) { @@ -477,6 +497,9 @@ ApplyStateTable::insert(ReadView const& base, std::shared_ptr const& sle) void ApplyStateTable::replace(ReadView const& base, std::shared_ptr const& sle) { +#ifndef NDEBUG + recordTouch(sle->key(), sle->getType()); +#endif auto const iter = items_.lower_bound(sle->key()); if (iter == items_.end() || iter->first != sle->key()) { @@ -506,6 +529,9 @@ ApplyStateTable::replace(ReadView const& base, std::shared_ptr const& sle) void ApplyStateTable::update(ReadView const& base, std::shared_ptr const& sle) { +#ifndef NDEBUG + recordTouch(sle->key(), sle->getType()); +#endif auto const iter = items_.find(sle->key()); if (iter == items_.end()) Throw("ApplyStateTable::update: missing key"); diff --git a/src/libxrpl/tx/Schedule.cpp b/src/libxrpl/tx/Schedule.cpp new file mode 100644 index 0000000000..b497854234 --- /dev/null +++ b/src/libxrpl/tx/Schedule.cpp @@ -0,0 +1,217 @@ +#include + +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace xrpl { + +namespace { + +// Disjoint-set (union-find) with path compression + union by size. +class UnionFind +{ +public: + explicit UnionFind(std::size_t n) : parent_(n), size_(n, 1) + { + std::iota(parent_.begin(), parent_.end(), std::size_t{0}); + } + + std::size_t + find(std::size_t x) + { + while (parent_[x] != x) + x = parent_[x] = parent_[parent_[x]]; + return x; + } + + void + unite(std::size_t a, std::size_t b) + { + a = find(a); + b = find(b); + if (a == b) + return; + if (size_[a] < size_[b]) + std::swap(a, b); + parent_[b] = a; + size_[a] += size_[b]; + } + +private: + std::vector parent_; + std::vector size_; +}; + +} // namespace + +Schedule +scheduleApply(std::vector> const& txns, ReadView const& base) +{ + Schedule sched; + + std::vector access; + access.reserve(txns.size()); + bool anyGlobal = false; + for (auto const& tx : txns) + { + access.push_back(accessSetOf(*tx, base)); + anyGlobal = anyGlobal || access.back().touchesGlobal; + } + + // Conservative fallback: any global-touching (or pseudo) transaction forces + // a single serial ordering. Correctness over throughput on flag ledgers. + if (anyGlobal) + { + sched.fullySerial = true; + sched.serial = txns; + return sched; + } + + // Union transactions that share any declared ledger entry. An inverted + // index (key -> first transaction seen touching it) makes this near-linear + // in the total number of declared keys, rather than O(n^2) pairwise. + UnionFind uf(txns.size()); + std::unordered_map> ownerOfKey; + for (std::size_t i = 0; i < access.size(); ++i) + { + for (auto const& key : access[i].keys()) + { + auto const [it, inserted] = ownerOfKey.try_emplace(key, i); + if (!inserted) + uf.unite(i, it->second); + } + } + + // Collect components. Iterating i ascending preserves canonical order within + // each group; groups are then ordered deterministically by the transaction + // id of their first (lowest canonical index) member. + std::map> components; + for (std::size_t i = 0; i < txns.size(); ++i) + components[uf.find(i)].push_back(i); + + sched.groups.reserve(components.size()); + for (auto const& [root, members] : components) + { + ConflictGroup group; + group.txns.reserve(members.size()); + for (auto const idx : members) + group.txns.push_back(txns[idx]); + sched.groups.push_back(std::move(group)); + } + + // Deterministic group order: by each group's first (lowest-index) member. + std::sort( + sched.groups.begin(), + sched.groups.end(), + [](ConflictGroup const& a, ConflictGroup const& b) { + return a.txns.front()->getTransactionID() < b.txns.front()->getTransactionID(); + }); + + return sched; +} + +namespace { + +// Apply a group's transactions into a fresh isolated view over `closed`, +// returning the view (with its accumulated, not-yet-merged write-set) and the +// count that applied. Reads only `closed`, so this is independent of every +// other group. +std::pair +applyGroupIsolated( + ServiceRegistry& registry, + ReadView const& closed, + std::vector> const& group, + beast::Journal j) +{ + OpenView gv(&closed); + std::size_t applied = 0; + for (auto const& tx : group) + { + if (apply(registry, gv, *tx, TapNone, j).applied) + ++applied; + } + return {std::move(gv), applied}; +} + +} // namespace + +ScheduledApplyResult +applyScheduled( + ServiceRegistry& registry, + ReadView const& closed, + TxsRawView& to, + std::vector> const& txns, + beast::Journal j, + unsigned workers) +{ + auto const sched = scheduleApply(txns, closed); + + ScheduledApplyResult res; + res.fullySerial = sched.fullySerial; + + if (sched.fullySerial) + { + res.groupCount = 1; + auto [gv, applied] = applyGroupIsolated(registry, closed, sched.serial, j); + gv.apply(to); + res.applied = applied; + return res; + } + + res.groupCount = sched.groups.size(); + + // Phase 1 — apply each group in isolation. Results are written into a + // pre-sized, index-keyed slot vector so the subsequent merge order is the + // (deterministic) schedule group order regardless of completion order. + std::vector>> slots(sched.groups.size()); + + unsigned const nThreads = + std::min(std::max(workers, 1u), static_cast(slots.size())); + + if (nThreads <= 1) + { + for (std::size_t i = 0; i < sched.groups.size(); ++i) + slots[i].emplace(applyGroupIsolated(registry, closed, sched.groups[i].txns, j)); + } + else + { + std::atomic next{0}; + auto worker = [&]() { + for (std::size_t i = next.fetch_add(1); i < sched.groups.size(); + i = next.fetch_add(1)) + slots[i].emplace(applyGroupIsolated(registry, closed, sched.groups[i].txns, j)); + }; + std::vector pool; + pool.reserve(nThreads); + for (unsigned t = 0; t < nThreads; ++t) + pool.emplace_back(worker); + for (auto& th : pool) + th.join(); + } + + // Phase 2 — merge the disjoint write-sets into the target, in fixed group + // order. Sequential by design: `to` is a single mutable ledger. + for (auto& slot : slots) + { + slot->first.apply(to); + res.applied += slot->second; + } + return res; +} + +} // namespace xrpl diff --git a/src/libxrpl/tx/Transactor.cpp b/src/libxrpl/tx/Transactor.cpp index 28fa059902..dcac38a25c 100644 --- a/src/libxrpl/tx/Transactor.cpp +++ b/src/libxrpl/tx/Transactor.cpp @@ -44,10 +44,12 @@ #include #include +#include #include #include #include #include +#include #include #include @@ -276,6 +278,21 @@ Transactor::Transactor(ApplyContext& ctx) { } +AccessSet +Transactor::commonAccountFootprint(STTx const& tx) +{ + AccessSet acc; + // The actor and the fee-payer (which differ only under delegation; + // getFeePayer() returns sfDelegate when present, else sfAccount). + acc.accounts.insert(keylet::account(tx.getAccountID(sfAccount)).key); + acc.accounts.insert(keylet::account(tx.getFeePayer()).key); + // A ticket-sequenced transaction consumes (erases) its Ticket object. + if (tx.isFieldPresent(sfTicketSequence)) + acc.miscObjects.insert( + keylet::kTicket(tx.getAccountID(sfAccount), tx.getFieldU32(sfTicketSequence)).key); + return acc; +} + bool Transactor::validDataLength(std::optional const& slice, std::size_t maxLength) { @@ -284,6 +301,56 @@ Transactor::validDataLength(std::optional const& slice, std::size_t maxLe return !slice->empty() && slice->length() <= maxLength; } +#ifndef NDEBUG +namespace { +// Opt-in via the XRPL_ACCESS_AUDIT env var: log the measured footprint of +// touchesGlobal transactors to feed the hard-transactor audit (Phase 1.5). +bool +accessSetAuditEnabled() +{ + static bool const enabled = [] { + auto const* v = std::getenv("XRPL_ACCESS_AUDIT"); + return v != nullptr && std::string_view{v} != "" && std::string_view{v} != "0"; + }(); + return enabled; +} +} // namespace + +void +Transactor::verifyAccessSet() const +{ + // Qualify: the unqualified name would bind to the static member + // Transactor::accessSetOf (the global default), not the free dispatcher. + auto const declared = xrpl::accessSetOf(ctx_.tx, ctx_.baseView()); + + if (declared.touchesGlobal) + { + if (accessSetAuditEnabled()) + { + std::size_t nonDir = 0; + for (auto const& [key, type] : ctx_.touchedEntries()) + if (type != ltDIR_NODE) + ++nonDir; + JLOG(j_.warn()) << "ACCESS_AUDIT txType=" << static_cast(ctx_.tx.getTxnType()) + << " touched=" << ctx_.touchedEntries().size() << " nonDir=" << nonDir; + } + return; + } + + auto const allowed = declared.keys(); + for (auto const& [key, type] : ctx_.touchedEntries()) + { + // Directory bookkeeping is out of scope for the access set (see + // AccessSet); a directory conflict is subsumed by its owner's account. + if (type == ltDIR_NODE) + continue; + XRPL_ASSERT( + allowed.contains(key), + "xrpl::Transactor::verifyAccessSet : touched entry within declared access set"); + } +} +#endif + std::uint32_t Transactor::getFlagsMask(PreflightContext const& ctx) { @@ -1233,6 +1300,15 @@ Transactor::operator()() if (isTesSuccess(result)) result = apply(); +#ifndef NDEBUG + // Validate (or audit) the declared access set against the actual footprint, + // but only on a clean success — tec/reset paths below intentionally touch + // extra state (removeUnfundedOffers, etc.) that the access set does not + // model. + if (isTesSuccess(result)) + verifyAccessSet(); +#endif + // No transaction can return temUNKNOWN from apply, // and it can't be passed in from a preclaim. XRPL_ASSERT(result != temUNKNOWN, "xrpl::Transactor::operator() : result is not temUNKNOWN"); diff --git a/src/libxrpl/tx/applySteps.cpp b/src/libxrpl/tx/applySteps.cpp index 217fdd717f..1a0882b354 100644 --- a/src/libxrpl/tx/applySteps.cpp +++ b/src/libxrpl/tx/applySteps.cpp @@ -420,6 +420,22 @@ calculateBaseFee(ReadView const& view, STTx const& tx) return invokeCalculateBaseFee(view, tx); } +AccessSet +accessSetOf(STTx const& tx, ReadView const& base) +{ + try + { + return withTxnType(base.rules(), tx.getTxnType(), [&]() { + return T::accessSetOf(tx, base); + }); + } + catch (UnknownTxnType const&) + { + // Unknown type — fail safe to global so a scheduler serializes it. + return AccessSet::global(); + } +} + XRPAmount calculateDefaultBaseFee(ReadView const& view, STTx const& tx) { diff --git a/src/libxrpl/tx/transactors/account/AccountSet.cpp b/src/libxrpl/tx/transactors/account/AccountSet.cpp index b52db14720..1fd5bb0473 100644 --- a/src/libxrpl/tx/transactors/account/AccountSet.cpp +++ b/src/libxrpl/tx/transactors/account/AccountSet.cpp @@ -276,6 +276,13 @@ AccountSet::preclaim(PreclaimContext const& ctx) return tesSUCCESS; } +AccessSet +AccountSet::accessSetOf(STTx const& tx, ReadView const& base) +{ + // AccountSet only mutates the actor's own AccountRoot. + return commonAccountFootprint(tx); +} + TER AccountSet::doApply() { diff --git a/src/libxrpl/tx/transactors/account/SetRegularKey.cpp b/src/libxrpl/tx/transactors/account/SetRegularKey.cpp index 66a0bef336..2051aa0736 100644 --- a/src/libxrpl/tx/transactors/account/SetRegularKey.cpp +++ b/src/libxrpl/tx/transactors/account/SetRegularKey.cpp @@ -52,6 +52,13 @@ SetRegularKey::preflight(PreflightContext const& ctx) return tesSUCCESS; } +AccessSet +SetRegularKey::accessSetOf(STTx const& tx, ReadView const& base) +{ + // SetRegularKey only mutates the actor's own AccountRoot. + return commonAccountFootprint(tx); +} + TER SetRegularKey::doApply() { diff --git a/src/libxrpl/tx/transactors/account/SignerListSet.cpp b/src/libxrpl/tx/transactors/account/SignerListSet.cpp index 7a5d2c60b0..58f782034a 100644 --- a/src/libxrpl/tx/transactors/account/SignerListSet.cpp +++ b/src/libxrpl/tx/transactors/account/SignerListSet.cpp @@ -107,6 +107,15 @@ SignerListSet::preflight(PreflightContext const& ctx) return tesSUCCESS; } +AccessSet +SignerListSet::accessSetOf(STTx const& tx, ReadView const& base) +{ + // Touches the actor's AccountRoot and its SignerList object. + AccessSet acc = commonAccountFootprint(tx); + acc.miscObjects.insert(keylet::signers(tx.getAccountID(sfAccount)).key); + return acc; +} + TER SignerListSet::doApply() { diff --git a/src/libxrpl/tx/transactors/delegate/DelegateSet.cpp b/src/libxrpl/tx/transactors/delegate/DelegateSet.cpp index 30d703686c..060b666848 100644 --- a/src/libxrpl/tx/transactors/delegate/DelegateSet.cpp +++ b/src/libxrpl/tx/transactors/delegate/DelegateSet.cpp @@ -65,6 +65,17 @@ DelegateSet::preclaim(PreclaimContext const& ctx) return tesSUCCESS; } +AccessSet +DelegateSet::accessSetOf(STTx const& tx, ReadView const& base) +{ + // Touches the actor's AccountRoot and the single Delegate object it owns + // for the authorized account. + AccessSet acc = commonAccountFootprint(tx); + acc.miscObjects.insert( + keylet::delegate(tx.getAccountID(sfAccount), tx[sfAuthorize]).key); + return acc; +} + TER DelegateSet::doApply() { diff --git a/src/libxrpl/tx/transactors/did/DIDDelete.cpp b/src/libxrpl/tx/transactors/did/DIDDelete.cpp index 4b4fb47169..cd01e9ba23 100644 --- a/src/libxrpl/tx/transactors/did/DIDDelete.cpp +++ b/src/libxrpl/tx/transactors/did/DIDDelete.cpp @@ -64,6 +64,15 @@ DIDDelete::deleteSLE( return tesSUCCESS; } +AccessSet +DIDDelete::accessSetOf(STTx const& tx, ReadView const& base) +{ + // Touches the actor's AccountRoot and its single DID object. + AccessSet acc = commonAccountFootprint(tx); + acc.miscObjects.insert(keylet::did(tx.getAccountID(sfAccount)).key); + return acc; +} + TER DIDDelete::doApply() { diff --git a/src/libxrpl/tx/transactors/did/DIDSet.cpp b/src/libxrpl/tx/transactors/did/DIDSet.cpp index a930c3c754..67415bacbd 100644 --- a/src/libxrpl/tx/transactors/did/DIDSet.cpp +++ b/src/libxrpl/tx/transactors/did/DIDSet.cpp @@ -95,6 +95,15 @@ addSLE(ApplyContext& ctx, std::shared_ptr const& sle, AccountID const& owne return tesSUCCESS; } +AccessSet +DIDSet::accessSetOf(STTx const& tx, ReadView const& base) +{ + // Touches the actor's AccountRoot and its single DID object. + AccessSet acc = commonAccountFootprint(tx); + acc.miscObjects.insert(keylet::did(tx.getAccountID(sfAccount)).key); + return acc; +} + TER DIDSet::doApply() { diff --git a/src/libxrpl/tx/transactors/oracle/OracleDelete.cpp b/src/libxrpl/tx/transactors/oracle/OracleDelete.cpp index 839b2b4e24..1206eaa48b 100644 --- a/src/libxrpl/tx/transactors/oracle/OracleDelete.cpp +++ b/src/libxrpl/tx/transactors/oracle/OracleDelete.cpp @@ -79,6 +79,16 @@ OracleDelete::deleteOracle( return tesSUCCESS; } +AccessSet +OracleDelete::accessSetOf(STTx const& tx, ReadView const& base) +{ + // Touches the actor's AccountRoot and its own Oracle object. + AccessSet acc = commonAccountFootprint(tx); + acc.miscObjects.insert( + keylet::oracle(tx.getAccountID(sfAccount), tx[sfOracleDocumentID]).key); + return acc; +} + TER OracleDelete::doApply() { diff --git a/src/libxrpl/tx/transactors/oracle/OracleSet.cpp b/src/libxrpl/tx/transactors/oracle/OracleSet.cpp index 66ce95f46c..02afd964f5 100644 --- a/src/libxrpl/tx/transactors/oracle/OracleSet.cpp +++ b/src/libxrpl/tx/transactors/oracle/OracleSet.cpp @@ -198,6 +198,16 @@ setPriceDataInnerObjTemplate(STObject& obj) obj.set(*elements); } +AccessSet +OracleSet::accessSetOf(STTx const& tx, ReadView const& base) +{ + // Touches the actor's AccountRoot and its own Oracle object. + AccessSet acc = commonAccountFootprint(tx); + acc.miscObjects.insert( + keylet::oracle(tx.getAccountID(sfAccount), tx[sfOracleDocumentID]).key); + return acc; +} + TER OracleSet::doApply() { diff --git a/src/libxrpl/tx/transactors/payment/DepositPreauth.cpp b/src/libxrpl/tx/transactors/payment/DepositPreauth.cpp index 8b4ad2f025..ddd965279a 100644 --- a/src/libxrpl/tx/transactors/payment/DepositPreauth.cpp +++ b/src/libxrpl/tx/transactors/payment/DepositPreauth.cpp @@ -148,6 +148,31 @@ DepositPreauth::preclaim(PreclaimContext const& ctx) return tesSUCCESS; } +AccessSet +DepositPreauth::accessSetOf(STTx const& tx, ReadView const& base) +{ + // Touches the actor's AccountRoot and the single DepositPreauth object it + // creates or removes. The object key is fully derivable from the tx body + // (by authorized account, or by the sorted credential set). + AccessSet acc = commonAccountFootprint(tx); + auto const owner = tx.getAccountID(sfAccount); + if (tx.isFieldPresent(sfAuthorize)) + acc.miscObjects.insert(keylet::depositPreauth(owner, tx[sfAuthorize]).key); + else if (tx.isFieldPresent(sfUnauthorize)) + acc.miscObjects.insert(keylet::depositPreauth(owner, tx[sfUnauthorize]).key); + else if (tx.isFieldPresent(sfAuthorizeCredentials)) + acc.miscObjects.insert( + keylet::depositPreauth( + owner, credentials::makeSorted(tx.getFieldArray(sfAuthorizeCredentials))) + .key); + else if (tx.isFieldPresent(sfUnauthorizeCredentials)) + acc.miscObjects.insert( + keylet::depositPreauth( + owner, credentials::makeSorted(tx.getFieldArray(sfUnauthorizeCredentials))) + .key); + return acc; +} + TER DepositPreauth::doApply() { diff --git a/src/libxrpl/tx/transactors/payment/Payment.cpp b/src/libxrpl/tx/transactors/payment/Payment.cpp index 1848d34786..882ab1d598 100644 --- a/src/libxrpl/tx/transactors/payment/Payment.cpp +++ b/src/libxrpl/tx/transactors/payment/Payment.cpp @@ -395,6 +395,32 @@ Payment::preclaim(PreclaimContext const& ctx) return tesSUCCESS; } +AccessSet +Payment::accessSetOf(STTx const& tx, ReadView const& base) +{ + // Only the direct XRP->XRP case has a statically-enumerable footprint. + // Path / cross-currency / IOU / MPT payments route through the flow engine, + // which touches state-dependent trust lines and offers, so they stay global. + // A SendMax (even in XRP) also forces the rippling path. + if (tx.isFieldPresent(sfPaths) || tx.isFieldPresent(sfSendMax) || + !tx.getFieldAmount(sfAmount).native()) + return AccessSet::global(); + + AccessSet acc = commonAccountFootprint(tx); + auto const src = tx.getAccountID(sfAccount); + auto const dst = tx.getAccountID(sfDestination); + acc.accounts.insert(keylet::account(dst).key); + // The destination's deposit-authorization check may read this preauth + // object (declared unconditionally — pessimistic but a narrow conflict + // surface) and, when authorized by credentials, the credentials named in + // the tx. + acc.miscObjects.insert(keylet::depositPreauth(dst, src).key); + if (tx.isFieldPresent(sfCredentialIDs)) + for (auto const& h : tx.getFieldV256(sfCredentialIDs)) + acc.miscObjects.insert(h); + return acc; +} + TER Payment::doApply() { diff --git a/src/libxrpl/tx/transactors/permissioned_domain/PermissionedDomainDelete.cpp b/src/libxrpl/tx/transactors/permissioned_domain/PermissionedDomainDelete.cpp index e36a0eb584..da1bc938ca 100644 --- a/src/libxrpl/tx/transactors/permissioned_domain/PermissionedDomainDelete.cpp +++ b/src/libxrpl/tx/transactors/permissioned_domain/PermissionedDomainDelete.cpp @@ -44,6 +44,16 @@ PermissionedDomainDelete::preclaim(PreclaimContext const& ctx) return tesSUCCESS; } +AccessSet +PermissionedDomainDelete::accessSetOf(STTx const& tx, ReadView const& base) +{ + // Touches the actor's AccountRoot and the single PermissionedDomain object + // named directly by the transaction. + AccessSet acc = commonAccountFootprint(tx); + acc.miscObjects.insert(keylet::permissionedDomain(tx[sfDomainID]).key); + return acc; +} + /** Attempt to delete the Permissioned Domain. */ TER PermissionedDomainDelete::doApply() diff --git a/src/libxrpl/tx/transactors/system/TicketCreate.cpp b/src/libxrpl/tx/transactors/system/TicketCreate.cpp index f442ac223b..2afd76ed8f 100644 --- a/src/libxrpl/tx/transactors/system/TicketCreate.cpp +++ b/src/libxrpl/tx/transactors/system/TicketCreate.cpp @@ -64,6 +64,26 @@ TicketCreate::preclaim(PreclaimContext const& ctx) return tesSUCCESS; } +AccessSet +TicketCreate::accessSetOf(STTx const& tx, ReadView const& base) +{ + AccessSet acc = commonAccountFootprint(tx); + auto const account = tx.getAccountID(sfAccount); + // Tickets are numbered from the account's current Sequence, which is ledger + // state — read it from the closed-ledger snapshot. The apply machinery + // advances Sequence by one before creating tickets for a sequence-based tx, + // so declare the inclusive range [seq, seq + count] to cover both the + // sequence-based (start = seq + 1) and ticket-based (start = seq) cases. + auto const sleAccount = base.read(keylet::account(account)); + if (!sleAccount) + return AccessSet::global(); + std::uint32_t const firstSeq = (*sleAccount)[sfSequence]; + std::uint32_t const count = tx[sfTicketCount]; + for (std::uint32_t i = 0; i <= count; ++i) + acc.miscObjects.insert(keylet::kTicket(account, firstSeq + i).key); + return acc; +} + TER TicketCreate::doApply() { diff --git a/src/libxrpl/tx/transactors/token/TrustSet.cpp b/src/libxrpl/tx/transactors/token/TrustSet.cpp index a8ea786347..c3cd015a86 100644 --- a/src/libxrpl/tx/transactors/token/TrustSet.cpp +++ b/src/libxrpl/tx/transactors/token/TrustSet.cpp @@ -323,6 +323,20 @@ TrustSet::preclaim(PreclaimContext const& ctx) return tesSUCCESS; } +AccessSet +TrustSet::accessSetOf(STTx const& tx, ReadView const& base) +{ + // Touches the actor and issuer AccountRoots and the single shared trust + // line between them; both endpoints come from sfLimitAmount. + AccessSet acc = commonAccountFootprint(tx); + auto const account = tx.getAccountID(sfAccount); + auto const limit = tx.getFieldAmount(sfLimitAmount); + auto const issuer = limit.getIssuer(); + acc.accounts.insert(keylet::account(issuer).key); + acc.trustlines.insert(keylet::line(account, issuer, limit.get().currency).key); + return acc; +} + TER TrustSet::doApply() { diff --git a/src/tests/libxrpl/helpers/TxTest.h b/src/tests/libxrpl/helpers/TxTest.h index 1b3ce460a2..c586e23a87 100644 --- a/src/tests/libxrpl/helpers/TxTest.h +++ b/src/tests/libxrpl/helpers/TxTest.h @@ -347,6 +347,19 @@ public: return registry_; } + /** + * @brief Get the closed (base) ledger as a shared Ledger pointer. + * + * Exposes the underlying Ledger (not just the ReadView) so tests can build + * sibling ledgers from it — e.g. to apply a transaction set in a controlled + * order outside the canonicalizing close() path. + */ + [[nodiscard]] std::shared_ptr + getClosedLedgerPtr() const + { + return closedLedger_; + } + private: TestServiceRegistry registry_; std::unordered_set> featureSet_; diff --git a/src/tests/libxrpl/tx/AccessSet.cpp b/src/tests/libxrpl/tx/AccessSet.cpp new file mode 100644 index 0000000000..f2807040ab --- /dev/null +++ b/src/tests/libxrpl/tx/AccessSet.cpp @@ -0,0 +1,417 @@ +// Tests for per-transactor static access-set extraction (Plan 1, Phase 1). +// +// Two layers: +// 1. AccessSet semantics (conflictsWith / keys) — pure, no ledger. +// 2. accessSetOf(tx, view) content — the declared footprint of each migrated +// transactor matches expectation, and un-migrated / dynamic ones report +// touchesGlobal. +// +// The *subset* safety net (declared footprint ⊇ what apply actually touched) is +// enforced continuously: in DEBUG builds Transactor::operator() asserts it on +// every successful apply, so every env.submit() below — and every other test in +// the suite that exercises a migrated transactor — validates it for free. + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +#include +#include + +namespace xrpl::test { + +namespace { + +// Build a signed STTx from a builder with an explicit sequence (the signature +// is irrelevant to accessSetOf; it never checks it). +template +std::shared_ptr +sttxOf(Builder builder, Account const& signer, std::uint32_t seq) +{ + return builder.setSequence(seq).setFee(XRPAmount{10}).build(signer.pk(), signer.sk()).getSTTx(); +} + +std::set +keysOf(std::initializer_list ks) +{ + std::set out; + for (auto const& k : ks) + out.insert(k.key); + return out; +} + +} // namespace + +//------------------------------------------------------------------------------ +// 1. AccessSet semantics +//------------------------------------------------------------------------------ + +TEST(AccessSet, ConflictDisjoint) +{ + Account const a("a"); + Account const b("b"); + + AccessSet s1; + s1.accounts.insert(keylet::account(a.id()).key); + AccessSet s2; + s2.accounts.insert(keylet::account(b.id()).key); + + EXPECT_FALSE(s1.conflictsWith(s2)); + EXPECT_FALSE(s2.conflictsWith(s1)); +} + +TEST(AccessSet, ConflictSharedAccount) +{ + Account const a("a"); + Account const b("b"); + + AccessSet s1; + s1.accounts.insert(keylet::account(a.id()).key); + s1.accounts.insert(keylet::account(b.id()).key); + + AccessSet s2; // shares account a + s2.accounts.insert(keylet::account(a.id()).key); + + EXPECT_TRUE(s1.conflictsWith(s2)); + EXPECT_TRUE(s2.conflictsWith(s1)); +} + +TEST(AccessSet, ConflictAcrossCategories) +{ + // The same key appearing under different categories still conflicts: + // conflict is decided on the flat union, not per-category. + Account const a("a"); + + AccessSet s1; + s1.accounts.insert(keylet::account(a.id()).key); + AccessSet s2; + s2.miscObjects.insert(keylet::account(a.id()).key); + + EXPECT_TRUE(s1.conflictsWith(s2)); +} + +TEST(AccessSet, GlobalConflictsWithEverything) +{ + Account const a("a"); + AccessSet local; + local.accounts.insert(keylet::account(a.id()).key); + + AccessSet const g = AccessSet::global(); + EXPECT_TRUE(g.touchesGlobal); + EXPECT_TRUE(g.conflictsWith(local)); + EXPECT_TRUE(local.conflictsWith(g)); + EXPECT_TRUE(g.conflictsWith(AccessSet{})); // even with an empty set +} + +TEST(AccessSet, KeysIsUnionAcrossCategories) +{ + Account const a("a"); + Account const b("b"); + + AccessSet s; + s.accounts.insert(keylet::account(a.id()).key); + s.trustlines.insert(keylet::account(b.id()).key); // any key; category is cosmetic + s.accounts.insert(keylet::account(a.id()).key); // duplicate + + EXPECT_EQ(s.keys().size(), 2u); +} + +//------------------------------------------------------------------------------ +// 2. accessSetOf content +//------------------------------------------------------------------------------ + +TEST(AccessSet, AccountSetTouchesOnlyActor) +{ + TxTest env; + Account const alice("alice"); + env.createAccount(alice, XRP(10000)); + env.close(); + + auto const stx = sttxOf( + transactions::AccountSetBuilder{alice}, alice, env.getAccountRoot(alice.id()).getSequence()); + auto const acc = xrpl::accessSetOf(*stx, env.getClosedLedger()); + + EXPECT_FALSE(acc.touchesGlobal); + EXPECT_EQ(acc.keys(), keysOf({keylet::account(alice.id())})); +} + +TEST(AccessSet, SetRegularKeyTouchesOnlyActor) +{ + TxTest env; + Account const alice("alice"); + Account const reg("reg"); + env.createAccount(alice, XRP(10000)); + env.close(); + + auto const stx = sttxOf( + transactions::SetRegularKeyBuilder{alice}.setRegularKey(reg.id()), + alice, + env.getAccountRoot(alice.id()).getSequence()); + auto const acc = xrpl::accessSetOf(*stx, env.getClosedLedger()); + + EXPECT_FALSE(acc.touchesGlobal); + EXPECT_EQ(acc.keys(), keysOf({keylet::account(alice.id())})); +} + +TEST(AccessSet, SignerListSetTouchesActorAndSignerList) +{ + TxTest env; + Account const alice("alice"); + Account const bob("bob"); + env.createAccount(alice, XRP(10000)); + env.close(); + + STArray signerEntries(1); + signerEntries.push_back(STObject::makeInnerObject(sfSignerEntry)); + signerEntries.back()[sfAccount] = bob.id(); + signerEntries.back()[sfSignerWeight] = std::uint16_t{1}; + + auto const stx = sttxOf( + transactions::SignerListSetBuilder{alice, 1}.setSignerEntries(signerEntries), + alice, + env.getAccountRoot(alice.id()).getSequence()); + auto const acc = xrpl::accessSetOf(*stx, env.getClosedLedger()); + + EXPECT_FALSE(acc.touchesGlobal); + EXPECT_EQ(acc.keys(), keysOf({keylet::account(alice.id()), keylet::signers(alice.id())})); +} + +TEST(AccessSet, DepositPreauthTouchesActorAndPreauthObject) +{ + TxTest env; + Account const alice("alice"); + Account const bob("bob"); + env.createAccount(alice, XRP(10000)); + env.close(); + + auto const stx = sttxOf( + transactions::DepositPreauthBuilder{alice}.setAuthorize(bob.id()), + alice, + env.getAccountRoot(alice.id()).getSequence()); + auto const acc = xrpl::accessSetOf(*stx, env.getClosedLedger()); + + EXPECT_FALSE(acc.touchesGlobal); + EXPECT_EQ( + acc.keys(), + keysOf({keylet::account(alice.id()), keylet::depositPreauth(alice.id(), bob.id())})); +} + +TEST(AccessSet, TrustSetTouchesBothEndpointsAndLine) +{ + TxTest env; + Account const alice("alice"); + Account const gw("gateway"); + IOU const usd("USD", gw); + env.createAccount(alice, XRP(10000)); + env.createAccount(gw, XRP(10000)); + env.close(); + + auto const stx = sttxOf( + transactions::TrustSetBuilder{alice}.setLimitAmount(usd.amount(10)), + alice, + env.getAccountRoot(alice.id()).getSequence()); + auto const acc = xrpl::accessSetOf(*stx, env.getClosedLedger()); + + EXPECT_FALSE(acc.touchesGlobal); + EXPECT_EQ( + acc.keys(), + keysOf( + {keylet::account(alice.id()), + keylet::account(gw.id()), + keylet::line(alice.id(), gw.id(), usd.issue().currency)})); +} + +TEST(AccessSet, PaymentXrpDirectTouchesSrcDstAndPreauth) +{ + TxTest env; + Account const alice("alice"); + Account const bob("bob"); + env.createAccount(alice, XRP(10000)); + env.createAccount(bob, XRP(10000)); + env.close(); + + auto const stx = sttxOf( + transactions::PaymentBuilder{alice, bob, XRP(1)}, + alice, + env.getAccountRoot(alice.id()).getSequence()); + auto const acc = xrpl::accessSetOf(*stx, env.getClosedLedger()); + + EXPECT_FALSE(acc.touchesGlobal); + EXPECT_EQ( + acc.keys(), + keysOf( + {keylet::account(alice.id()), + keylet::account(bob.id()), + keylet::depositPreauth(bob.id(), alice.id())})); +} + +TEST(AccessSet, TicketCreateDeclaresSequenceRange) +{ + TxTest env; + Account const alice("alice"); + env.createAccount(alice, XRP(10000)); + env.close(); + + std::uint32_t const seq = env.getAccountRoot(alice.id()).getSequence(); + std::uint32_t const count = 3; + + auto const stx = + sttxOf(transactions::TicketCreateBuilder{alice, count}, alice, seq); + auto const acc = xrpl::accessSetOf(*stx, env.getClosedLedger()); + + EXPECT_FALSE(acc.touchesGlobal); + // Inclusive range [seq, seq + count] covers both sequence- and ticket-based + // apply (the machinery may advance the sequence by one before creating). + std::set expected{keylet::account(alice.id()).key}; + for (std::uint32_t i = 0; i <= count; ++i) + expected.insert(keylet::kTicket(alice.id(), seq + i).key); + EXPECT_EQ(acc.keys(), expected); +} + +//------------------------------------------------------------------------------ +// 3. Fail-safe: dynamic-footprint transactions report touchesGlobal +//------------------------------------------------------------------------------ + +TEST(AccessSet, PaymentWithSendMaxIsGlobal) +{ + // Even an all-XRP payment with SendMax routes through the flow engine. + TxTest env; + Account const alice("alice"); + Account const bob("bob"); + env.createAccount(alice, XRP(10000)); + env.createAccount(bob, XRP(10000)); + env.close(); + + auto const stx = sttxOf( + transactions::PaymentBuilder{alice, bob, XRP(1)}.setSendMax(XRP(2)), + alice, + env.getAccountRoot(alice.id()).getSequence()); + EXPECT_TRUE(xrpl::accessSetOf(*stx, env.getClosedLedger()).touchesGlobal); +} + +TEST(AccessSet, IouPaymentIsGlobal) +{ + TxTest env; + Account const gw("gateway"); + Account const alice("alice"); + IOU const usd("USD", gw); + env.createAccount(gw, XRP(10000), asfDefaultRipple); + env.createAccount(alice, XRP(10000), asfDefaultRipple); + env.close(); + + auto const stx = sttxOf( + transactions::PaymentBuilder{gw, alice, usd.amount(5)}, + gw, + env.getAccountRoot(gw.id()).getSequence()); + EXPECT_TRUE(xrpl::accessSetOf(*stx, env.getClosedLedger()).touchesGlobal); +} + +TEST(AccessSet, UnmigratedTransactorIsGlobal) +{ + // OfferCreate is intentionally not migrated (offer crossing has a + // state-dependent footprint); it must fall back to the global default. + TxTest env; + Account const alice("alice"); + Account const gw("gateway"); + IOU const usd("USD", gw); + env.createAccount(alice, XRP(10000)); + env.close(); + + auto const stx = sttxOf( + transactions::OfferCreateBuilder{alice, usd.amount(10), XRP(10)}, + alice, + env.getAccountRoot(alice.id()).getSequence()); + EXPECT_TRUE(xrpl::accessSetOf(*stx, env.getClosedLedger()).touchesGlobal); +} + +//------------------------------------------------------------------------------ +// 4. Subset safety net (explicit). In DEBUG these apply paths assert the +// declared footprint ⊇ the touched footprint; tesSUCCESS means it held. +//------------------------------------------------------------------------------ + +TEST(AccessSet, SubsetPaymentToNewAccount) +{ + // Exercises the absence-probe path: the brand-new destination is read + // (returns nullptr) and then inserted; both must fall within the declared + // {src, dst, depositPreauth(dst, src)}. + TxTest env; + Account const alice("alice"); + Account const carol("carol"); // never created + env.createAccount(alice, XRP(10000)); + env.close(); + + EXPECT_EQ( + env.submit(transactions::PaymentBuilder{alice, carol, XRP(100)}, alice).ter, tesSUCCESS); +} + +TEST(AccessSet, SubsetTicketCreateThenUse) +{ + TxTest env; + Account const alice("alice"); + env.createAccount(alice, XRP(10000)); + env.close(); + + std::uint32_t const seq = env.getAccountRoot(alice.id()).getSequence(); + EXPECT_EQ(env.submit(transactions::TicketCreateBuilder{alice, 1}, alice).ter, tesSUCCESS); + env.close(); + + // Use the ticket (a ticket-based AccountSet) — exercises common-footprint + // ticket handling on the consume side. + std::uint32_t const ticketSeq = seq + 1; + EXPECT_EQ( + env.submit(transactions::AccountSetBuilder{alice}.setTicketSequence(ticketSeq), alice).ter, + tesSUCCESS); +} + +TEST(AccessSet, SubsetTrustSetAndSignerList) +{ + TxTest env; + Account const alice("alice"); + Account const bob("bob"); + Account const gw("gateway"); + IOU const usd("USD", gw); + env.createAccount(alice, XRP(10000), asfDefaultRipple); + env.createAccount(gw, XRP(10000), asfDefaultRipple); + env.close(); + + EXPECT_EQ( + env.submit(transactions::TrustSetBuilder{alice}.setLimitAmount(usd.amount(10)), alice).ter, + tesSUCCESS); + env.close(); + + STArray signerEntries(1); + signerEntries.push_back(STObject::makeInnerObject(sfSignerEntry)); + signerEntries.back()[sfAccount] = bob.id(); + signerEntries.back()[sfSignerWeight] = std::uint16_t{1}; + + EXPECT_EQ( + env.submit( + transactions::SignerListSetBuilder{alice, 1}.setSignerEntries(signerEntries), alice) + .ter, + tesSUCCESS); + env.close(); + + // Remove the signer list (destroy path). + EXPECT_EQ(env.submit(transactions::SignerListSetBuilder{alice, 0}, alice).ter, tesSUCCESS); +} + +} // namespace xrpl::test diff --git a/src/tests/libxrpl/tx/Schedule.cpp b/src/tests/libxrpl/tx/Schedule.cpp new file mode 100644 index 0000000000..baa3494b46 --- /dev/null +++ b/src/tests/libxrpl/tx/Schedule.cpp @@ -0,0 +1,190 @@ +// Tests for scheduleApply (Plan 1, Phase 2): partitioning a canonical +// transaction set into independent conflict groups for parallel application. +// +// The headline guarantee is the differential test: applying a workload in the +// scheduler's (reordered) group order produces a byte-identical account-state +// root to applying it in canonical order. That is the correctness contract a +// parallel executor depends on. + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +#include +#include +#include + +namespace xrpl::test { + +namespace { + +// Build a signed Payment STTx with an explicit sequence. +std::shared_ptr +payment(Account const& from, Account const& to, STAmount const& amount, std::uint32_t seq) +{ + return transactions::PaymentBuilder{from, to, amount} + .setSequence(seq) + .setFee(XRPAmount{10}) + .build(from.pk(), from.sk()) + .getSTTx(); +} + +} // namespace + +TEST(Schedule, DisjointPaymentsFormSeparateGroups) +{ + TxTest env; + Account const a("a"), b("b"), c("c"), d("d"), e("e"), f("f"); + for (auto const* acct : {&a, &b, &c, &d, &e, &f}) + env.createAccount(*acct, XRP(10000)); + env.close(); + + std::vector> txns{ + payment(a, b, XRP(1), env.getAccountRoot(a.id()).getSequence()), + payment(c, d, XRP(1), env.getAccountRoot(c.id()).getSequence()), + payment(e, f, XRP(1), env.getAccountRoot(e.id()).getSequence()), + }; + + auto const sched = scheduleApply(txns, env.getClosedLedger()); + + EXPECT_FALSE(sched.fullySerial); + EXPECT_EQ(sched.groups.size(), 3u); + EXPECT_EQ(sched.size(), 3u); + for (auto const& g : sched.groups) + EXPECT_EQ(g.txns.size(), 1u); +} + +TEST(Schedule, SameSourceFormsOneOrderedGroup) +{ + TxTest env; + Account const a("a"), b("b"), c("c"); + for (auto const* acct : {&a, &b, &c}) + env.createAccount(*acct, XRP(10000)); + env.close(); + + std::uint32_t const seq = env.getAccountRoot(a.id()).getSequence(); + std::vector> txns{ + payment(a, b, XRP(1), seq), + payment(a, c, XRP(1), seq + 1), + }; + + auto const sched = scheduleApply(txns, env.getClosedLedger()); + + ASSERT_EQ(sched.groups.size(), 1u); + ASSERT_EQ(sched.groups[0].txns.size(), 2u); + // Canonical (sequence) order preserved within the group. + EXPECT_EQ(sched.groups[0].txns[0]->getSeqValue(), seq); + EXPECT_EQ(sched.groups[0].txns[1]->getSeqValue(), seq + 1); +} + +TEST(Schedule, SharedDestinationConflicts) +{ + TxTest env; + Account const a("a"), b("b"), z("z"); + for (auto const* acct : {&a, &b, &z}) + env.createAccount(*acct, XRP(10000)); + env.close(); + + // Both pay the same destination z -> they share z's AccountRoot -> 1 group. + std::vector> txns{ + payment(a, z, XRP(1), env.getAccountRoot(a.id()).getSequence()), + payment(b, z, XRP(1), env.getAccountRoot(b.id()).getSequence()), + }; + + auto const sched = scheduleApply(txns, env.getClosedLedger()); + EXPECT_FALSE(sched.fullySerial); + ASSERT_EQ(sched.groups.size(), 1u); + EXPECT_EQ(sched.groups[0].txns.size(), 2u); +} + +TEST(Schedule, GlobalTransactionForcesFullySerial) +{ + TxTest env; + Account const a("a"), b("b"), gw("gw"); + IOU const usd("USD", gw); + for (auto const* acct : {&a, &b, &gw}) + env.createAccount(*acct, XRP(10000)); + env.close(); + + // An OfferCreate is touchesGlobal (dynamic footprint) -> whole set serial. + auto const offer = transactions::OfferCreateBuilder{a, usd.amount(10), XRP(10)} + .setSequence(env.getAccountRoot(a.id()).getSequence()) + .setFee(XRPAmount{10}) + .build(a.pk(), a.sk()) + .getSTTx(); + + std::vector> txns{ + payment(b, gw, XRP(1), env.getAccountRoot(b.id()).getSequence()), + offer, + }; + + auto const sched = scheduleApply(txns, env.getClosedLedger()); + EXPECT_TRUE(sched.fullySerial); + EXPECT_TRUE(sched.groups.empty()); + EXPECT_EQ(sched.serial.size(), 2u); +} + +// The headline correctness property the scheduler must guarantee: any two +// transactions placed in DIFFERENT groups have non-conflicting access sets. +// Together with the (separately, continuously verified) fact that each access +// set is a superset of what the transaction actually touches, this is exactly +// what makes applying distinct groups concurrently state-equivalent to a serial +// apply — independent writes commute. We assert the partition property directly, +// which is stronger and more honest than an apply-order replay through this test +// harness (whose close() re-canonicalizes by tx hash regardless of submission +// order, so it cannot observe a reordering). The full apply-in-schedule-order +// state-root differential belongs to Phase 3, where a parallel executor applies +// outside the canonicalizing close path. +TEST(Schedule, GroupsArePairwiseIndependent) +{ + TxTest env; + Account const a("a"), b("b"), c("c"), d("d"), e("e"), f("f"), z("z"); + for (auto const* acct : {&a, &b, &c, &d, &e, &f, &z}) + env.createAccount(*acct, XRP(10000)); + env.close(); + + std::uint32_t const seqA = env.getAccountRoot(a.id()).getSequence(); + std::vector> txns{ + payment(a, b, XRP(7), seqA), + payment(c, d, XRP(3), env.getAccountRoot(c.id()).getSequence()), + payment(e, f, XRP(5), env.getAccountRoot(e.id()).getSequence()), + payment(a, b, XRP(2), seqA + 1), // same source a -> same group as #1 + payment(c, z, XRP(1), env.getAccountRoot(c.id()).getSequence() + 1), // shares c -> #2 + }; + + auto const& base = env.getClosedLedger(); + auto const sched = scheduleApply(txns, base); + ASSERT_FALSE(sched.fullySerial); + + // Every transaction is scheduled exactly once. + EXPECT_EQ(sched.size(), txns.size()); + + // Within a group, canonical (input) order is preserved — verify per-source + // sequences are monotonic. + for (auto const& g : sched.groups) + for (std::size_t i = 1; i < g.txns.size(); ++i) + if (g.txns[i]->getAccountID(sfAccount) == g.txns[i - 1]->getAccountID(sfAccount)) + EXPECT_LT(g.txns[i - 1]->getSeqValue(), g.txns[i]->getSeqValue()); + + // The core invariant: any two transactions in DIFFERENT groups do not + // conflict. (Equivalently: every real conflict is contained within a group.) + for (std::size_t i = 0; i < sched.groups.size(); ++i) + for (std::size_t j = i + 1; j < sched.groups.size(); ++j) + for (auto const& ta : sched.groups[i].txns) + for (auto const& tb : sched.groups[j].txns) + EXPECT_FALSE( + accessSetOf(*ta, base).conflictsWith(accessSetOf(*tb, base))) + << "transactions in different groups must not conflict"; +} + +} // namespace xrpl::test diff --git a/src/tests/libxrpl/tx/ScheduledApply.cpp b/src/tests/libxrpl/tx/ScheduledApply.cpp new file mode 100644 index 0000000000..7ac6f02e4e --- /dev/null +++ b/src/tests/libxrpl/tx/ScheduledApply.cpp @@ -0,0 +1,254 @@ +// Differential test for applyScheduled (Plan 1, Phase 3 core): applying a +// transaction set via its conflict-group schedule (each group isolated over the +// closed snapshot, write-sets merged) must yield a byte-identical account-state +// root to a serial canonical apply. +// +// Unlike a test routed through TxTest::close() (which re-canonicalizes by tx +// hash and so cannot observe a reordering), this test builds BOTH ledgers +// itself, so the comparison is real: if the scheduler ever placed two +// conflicting transactions in different groups, the roots would diverge here. + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace xrpl::test { + +namespace { + +std::shared_ptr +payment(Account const& from, Account const& to, STAmount const& amount, std::uint32_t seq) +{ + return transactions::PaymentBuilder{from, to, amount} + .setSequence(seq) + .setFee(XRPAmount{10}) + .build(from.pk(), from.sk()) + .getSTTx(); +} + +// Build a fresh ledger from `closed`, apply `txns` in the given exact order in a +// single accumulating view (the serial ground truth), and return its +// account-state root. +uint256 +serialStateRoot( + TxTest& env, + Ledger const& closed, + std::vector> const& txns) +{ + auto const closeTime = env.getCloseTime() + closed.header().closeTimeResolution; + auto next = std::make_shared(closed, closeTime); + { + OpenView accum(&closed); + for (auto const& tx : txns) + apply(env.getServiceRegistry(), accum, *tx, TapNone, env.getServiceRegistry().getJournal("apply")); + accum.apply(*next); + } + next->setAccepted(closeTime, closed.header().closeTimeResolution, true); + return next->header().accountHash; +} + +// Build a fresh ledger from `closed`, apply `txns` via applyScheduled (grouped + +// merged), and return its account-state root, along with the schedule result. +std::pair +scheduledStateRoot( + TxTest& env, + Ledger const& closed, + std::vector> const& txns, + unsigned workers = 1) +{ + auto const closeTime = env.getCloseTime() + closed.header().closeTimeResolution; + auto next = std::make_shared(closed, closeTime); + auto const res = applyScheduled( + env.getServiceRegistry(), + closed, + *next, + txns, + env.getServiceRegistry().getJournal("apply"), + workers); + next->setAccepted(closeTime, closed.header().closeTimeResolution, true); + return {next->header().accountHash, res}; +} + +} // namespace + +TEST(ScheduledApply, ParallelGroupedMatchesSerial) +{ + TxTest env; + Account const a("a"), b("b"), c("c"), d("d"), e("e"), f("f"), g("g"); + for (auto const* acct : {&a, &b, &c, &d, &e, &f, &g}) + env.createAccount(*acct, XRP(10000)); + env.close(); + + auto const& closed = *env.getClosedLedgerPtr(); + std::uint32_t const seqA = env.getAccountRoot(a.id()).getSequence(); + + // Groups: {p1,p4} (source a, ordered), {p2}, {p3} — three independent groups. + std::vector> txns{ + payment(a, b, XRP(7), seqA), + payment(c, d, XRP(3), env.getAccountRoot(c.id()).getSequence()), + payment(e, f, XRP(5), env.getAccountRoot(e.id()).getSequence()), + payment(a, g, XRP(2), seqA + 1), + }; + + auto const serialRoot = serialStateRoot(env, closed, txns); + auto const [scheduledRoot, res] = scheduledStateRoot(env, closed, txns); + + EXPECT_FALSE(res.fullySerial); + EXPECT_EQ(res.groupCount, 3u); + EXPECT_EQ(res.applied, 4u); + + // The state root must be non-trivial (real accounts changed) AND identical + // regardless of the parallel grouping — the determinism guarantee. + EXPECT_NE(serialRoot, uint256{}); + EXPECT_EQ(serialRoot, scheduledRoot); +} + +TEST(ScheduledApply, ThreadedMatchesSerialAcrossManyGroups) +{ + TxTest env; + // 24 accounts -> 12 disjoint payment pairs -> 12 independent groups, applied + // across a thread pool. Repeated to give nondeterminism/races a chance to + // surface. (A clean unit pass is necessary, not sufficient, for production — + // see applyScheduled's note on certification.) + constexpr int kPairs = 12; + std::vector accts; + accts.reserve(kPairs * 2); + for (int i = 0; i < kPairs * 2; ++i) + accts.emplace_back("acct" + std::to_string(i)); + for (auto const& a : accts) + env.createAccount(a, XRP(10000)); + env.close(); + + auto const& closed = *env.getClosedLedgerPtr(); + std::vector> txns; + for (int p = 0; p < kPairs; ++p) + { + auto const& from = accts[2 * p]; + auto const& to = accts[2 * p + 1]; + txns.push_back( + payment(from, to, XRP(p + 1), env.getAccountRoot(from.id()).getSequence())); + } + + auto const serialRoot = serialStateRoot(env, closed, txns); + EXPECT_NE(serialRoot, uint256{}); + + for (int iter = 0; iter < 8; ++iter) + { + auto const [threadedRoot, res] = scheduledStateRoot(env, closed, txns, /*workers=*/8); + EXPECT_FALSE(res.fullySerial); + EXPECT_EQ(res.groupCount, static_cast(kPairs)); + EXPECT_EQ(res.applied, static_cast(kPairs)); + EXPECT_EQ(threadedRoot, serialRoot) << "threaded apply diverged on iteration " << iter; + } +} + +TEST(ScheduledApply, FullySerialPathAlsoMatches) +{ + TxTest env; + Account const a("a"), b("b"), gw("gw"); + IOU const usd("USD", gw); + for (auto const* acct : {&a, &b, &gw}) + env.createAccount(*acct, XRP(10000)); + env.close(); + + auto const& closed = *env.getClosedLedgerPtr(); + + // An OfferCreate is touchesGlobal -> scheduleApply falls back to fully + // serial; applyScheduled then applies the whole set in canonical order. + auto const offer = transactions::OfferCreateBuilder{a, usd.amount(10), XRP(10)} + .setSequence(env.getAccountRoot(a.id()).getSequence()) + .setFee(XRPAmount{10}) + .build(a.pk(), a.sk()) + .getSTTx(); + std::vector> txns{ + payment(b, gw, XRP(1), env.getAccountRoot(b.id()).getSequence()), + offer, + }; + + auto const serialRoot = serialStateRoot(env, closed, txns); + auto const [scheduledRoot, res] = scheduledStateRoot(env, closed, txns); + + EXPECT_TRUE(res.fullySerial); + EXPECT_NE(serialRoot, uint256{}); + EXPECT_EQ(serialRoot, scheduledRoot); +} + +// Throughput benchmark: time applyScheduled over many disjoint payments at +// increasing worker counts. Run explicitly: +// xrpl.test.tx --gtest_filter='ScheduledApply.ThroughputBenchmark' +// Build type matters enormously — Debug numbers (assertions on, unoptimized) are +// directional only; use a Release build for representative figures. +TEST(ScheduledApply, ThroughputBenchmark) +{ + constexpr int kPairs = 400; // -> kPairs independent groups, 2*kPairs accounts + constexpr int kReps = 5; + + TxTest env; + std::vector accts; + accts.reserve(kPairs * 2); + for (int i = 0; i < kPairs * 2; ++i) + accts.emplace_back("ba" + std::to_string(i)); + // Fund in batches with a single close per batch (createAccount closes each). + for (auto const& a : accts) + env.createAccount(a, XRP(10000)); + env.close(); + + auto const& closed = *env.getClosedLedgerPtr(); + std::vector> txns; + txns.reserve(kPairs); + for (int p = 0; p < kPairs; ++p) + txns.push_back(payment( + accts[2 * p], accts[2 * p + 1], XRP(1), env.getAccountRoot(accts[2 * p].id()).getSequence())); + + auto const closeTime = env.getCloseTime() + closed.header().closeTimeResolution; + auto bestNanos = [&](unsigned workers) { + long long best = -1; + for (int r = 0; r < kReps; ++r) + { + auto next = std::make_shared(closed, closeTime); + auto const t0 = std::chrono::steady_clock::now(); + auto const res = applyScheduled( + env.getServiceRegistry(), closed, *next, txns, + env.getServiceRegistry().getJournal("bench"), workers); + auto const t1 = std::chrono::steady_clock::now(); + EXPECT_EQ(res.applied, static_cast(kPairs)); + auto const ns = std::chrono::duration_cast(t1 - t0).count(); + if (best < 0 || ns < best) + best = ns; + } + return best; + }; + + std::printf("\n=== applyScheduled throughput: %d disjoint payments (best of %d) ===\n", + kPairs, kReps); + long long base = 0; + for (unsigned w : {1u, 2u, 4u, 8u}) + { + auto const ns = bestNanos(w); + if (w == 1) + base = ns; + std::printf(" workers=%u: %8.3f ms total, %7.1f us/tx, speedup %.2fx\n", + w, ns / 1e6, ns / 1e3 / kPairs, base / double(ns)); + } + std::printf("(build type dominates these numbers; Debug is directional only)\n"); +} + +} // namespace xrpl::test diff --git a/src/xrpld/app/ledger/detail/BuildLedger.cpp b/src/xrpld/app/ledger/detail/BuildLedger.cpp index a77c1c9c50..53a4932c12 100644 --- a/src/xrpld/app/ledger/detail/BuildLedger.cpp +++ b/src/xrpld/app/ledger/detail/BuildLedger.cpp @@ -17,12 +17,18 @@ #include #include #include +#include #include +#include #include +#include #include #include #include +#include +#include +#include namespace xrpl { @@ -90,6 +96,39 @@ buildLedgerImpl( @return number of transactions applied; transactions to retry left in txns */ +namespace { + +// Experimental, operator-opt-in parallel apply (Plan 1). Gated by the +// XRPL_PARALLEL_APPLY env var; default OFF, so default behaviour is unchanged. +// A proper amendment/Config flag is the productionization step. +bool +parallelApplyEnabled() +{ + static bool const enabled = [] { + auto const* v = std::getenv("XRPL_PARALLEL_APPLY"); + return v != nullptr && std::string_view{v} != "" && std::string_view{v} != "0"; + }(); + return enabled; +} + +unsigned +parallelApplyWorkers() +{ + static unsigned const workers = [] { + if (auto const* v = std::getenv("XRPL_PARALLEL_APPLY_WORKERS")) + { + auto const n = std::atoi(v); + if (n > 0) + return static_cast(n); + } + unsigned const hw = std::thread::hardware_concurrency(); + return std::clamp(hw > 2 ? hw - 2 : 2, 2u, 8u); + }(); + return workers; +} + +} // namespace + std::size_t applyTransactions( Application& app, @@ -99,6 +138,36 @@ applyTransactions( OpenView& view, beast::Journal j) { + // Plan 1 parallel apply: schedule the canonical set into independent groups + // and apply them (optionally across a thread pool), merging the disjoint + // write-sets into `view`. Behind XRPL_PARALLEL_APPLY; the resulting state is + // byte-identical to the serial path for conflict-free reorderings (the + // schedule guarantees this; see ScheduledApply differential tests). On flag + // ledgers / any global tx, the scheduler falls back to fully serial. + if (parallelApplyEnabled()) + { + std::vector> ordered; + ordered.reserve(txns.size()); + for (auto const& item : txns) + { + if (built->txExists(item.first.getTXID())) + continue; + ordered.push_back(item.second); + } + + auto const res = applyScheduled(app, *built, view, ordered, j, parallelApplyWorkers()); + + JLOG(j.debug()) << "Parallel apply: " << res.applied << " applied across " + << res.groupCount << " group(s)" + << (res.fullySerial ? " (fell back to serial)" : ""); + + // Every transaction has been consumed; the parallel path needs no retry + // passes — by access-set disjointness, no cross-group dependency exists. + for (auto it = txns.begin(); it != txns.end();) + it = txns.erase(it); + return res.applied; + } + bool certainRetry = true; std::size_t count = 0; diff --git a/tasks/plan-1-status.md b/tasks/plan-1-status.md new file mode 100644 index 0000000000..616c4b40a1 --- /dev/null +++ b/tasks/plan-1-status.md @@ -0,0 +1,200 @@ +# Plan 1 (parallel transaction apply) — implementation status + +Branch: `feat/parallel-apply-access-set`. This file tracks what is implemented, +verified, and what remains, so any engineer can continue without re-deriving. + +## Implemented & verified + +### Phase 1 foundation (commit "static access-set extraction + DEBUG assertion gate") +- `include/xrpl/tx/AccessSet.h` — the declared per-tx footprint (categorised key + sets + `touchesGlobal`), `keys()`, `conflictsWith()`. +- `Transactor::accessSetOf(STTx, ReadView)` — name-hidden dispatch hook, base + default `touchesGlobal=true` (fail-safe). Free `accessSetOf` dispatcher in + `applySteps.cpp`. `Transactor::commonAccountFootprint` helper. +- DEBUG touched-key instrumentation in `detail::ApplyStateTable` (the single + chokepoint for read/exists/peek/insert/update/erase), exposed up through + `ApplyViewBase`/`ApplyContext`. +- DEBUG subset assertion + `XRPL_ACCESS_AUDIT` footprint dump in + `Transactor::operator()` (success path only). +- Tests: `src/tests/libxrpl/tx/AccessSet.cpp` (18). Verified: full + `xrpl.test.protocol_autogen` (495 tests) passes with the assertion live. + +### Phase 2 scheduler (commit "access-set scheduler ... independent groups") +- `include/xrpl/tx/Schedule.h` + `src/libxrpl/tx/Schedule.cpp` — `scheduleApply` + partitions a canonical tx set into independent `ConflictGroup`s via union-find + over an inverted key→tx index. Any `touchesGlobal` tx ⇒ conservative + fully-serial fallback (flag-ledger handling). +- Tests: `src/tests/libxrpl/tx/Schedule.cpp` (5), incl. the core invariant + **GroupsArePairwiseIndependent** (txns in different groups never conflict). + +### Migrated transactors (13) — `accessSetOf` declared, assertion-verified +AccountSet, SetRegularKey, DepositPreauth, SignerListSet, TicketCreate, TrustSet, +DIDSet, DIDDelete, Payment (XRP→XRP only), OracleSet, OracleDelete, DelegateSet, +PermissionedDomainDelete. + +## Load-bearing design rules (apply to every future migration) +1. **Directory exclusion + owner declaration.** The assertion excludes + `ltDIR_NODE` entries (owner-dir pages are derived bookkeeping). This is sound + ONLY if, for every owner directory a transactor modifies, that owner's + `keylet::account` is declared. So: whenever `doApply` does `dirInsert`/ + `dirRemove` on `ownerDir(X)`, declare `account(X)` — even if X's root isn't + otherwise written (e.g. CredentialCreate touches the subject's owner dir). +2. **Shared (non-owner) directories ⇒ global.** Book directories and NFT + buy/sell directories (`nftSells`/`nftBuys`, keyed by NFTokenID) are shared + across accounts; they are genuine cross-account conflict surfaces with no + single owning account, so any transactor touching them stays `touchesGlobal`. +3. **`succ()`/range scans ⇒ not statically declarable.** A footprint discovered + by walking a chain (NFT page chains) is not a static superset ⇒ global. +4. **Snapshot reads are allowed.** `accessSetOf(tx, base)` may read `base` to + resolve a field stored inside an object SLE (e.g. an escrow's Destination). +5. The DEBUG subset assertion is the gate: migrate, run `xrpl.test.protocol_autogen` + in Debug; any under-declaration aborts the relevant `*Tests` suite. + +## Remaining Phase-1 migrations — turnkey (footprints analysed, verdicts fixed) + +STATIC (derivable from tx body): +- CredentialCreate → `credential(sfSubject, src, sfCredentialType)` + `account(sfSubject)` +- MPTokenIssuanceCreate → `mptIssuance(tx.getSeqValue(), src)` +- CheckCreate → `check(src, tx.getSeqValue())` + `account(sfDestination)` +- PaymentChannelCreate → `payChan(src, sfDestination, tx.getSeqValue())` + `account(sfDestination)` +- Clawback → `account(src)` + `account(sfHolder)` + (IOU: `line(holder, issuer, ccy)`; + MPT: `mptIssuance(id)` + `mptoken(id, holder)`) +- EscrowCreate (XRP) → `escrow(src, tx.getSeqValue())` + `account(sfDestination)` + (IOU/MPT variant: + issuer account + `line(src,issuer)` / mpt objects, or keep global) + +STATIC_WITH_SNAPSHOT (read the object SLE in `base` to resolve owners): +- CredentialAccept → `credential(src, sfIssuer, type)` + `account(sfIssuer)` +- CredentialDelete → `credential(subject|src, issuer|src, type)` + `account(issuer)` + `account(subject)` +- MPTokenIssuanceDestroy → `mptIssuance(id)` + `account(issuer-from-SLE)` +- MPTokenIssuanceSet → `mptIssuance(id)` or `mptoken(id, sfHolder)` (+ `permissionedDomain(sfDomainID)`) +- MPTokenAuthorize → `mptIssuance(id)` + `account(holder)` + `mptoken(id, holder)` (holder = src or sfHolder) +- CheckCancel → `check(sfCheckID)` + `account(check.Account)` + `account(check.Destination)` +- PaymentChannelFund → `payChan(sfChannel)` + `account(chan.Account)` + `account(chan.Destination)` +- PaymentChannelClaim → `payChan(sfChannel)` + `account(chan.Account)` + `account(chan.Destination)` + + `depositPreauth(chan.Destination, src)` + each `sfCredentialIDs` key +- EscrowFinish (XRP; else global) → `escrow(sfOwner, sfOfferSequence)` + escrow.Account + escrow.Destination + + `depositPreauth(dst, src)` + each `sfCredentialIDs` key +- EscrowCancel (XRP; else global) → `escrow(sfOwner, sfOfferSequence)` + escrow.Account + escrow.Destination + +## DYNAMIC — must stay `touchesGlobal` in v1 (reason) +- OfferCreate, OfferCancel — shared book directory + offer crossing modifies + counterparty accounts not in the tx. +- All AMM* (Create/Deposit/Withdraw/Vote/Bid/Delete/Clawback) — pool + pseudo-account + crossing. +- Payment with paths / cross-currency / IOU / MPT, CheckCash — flow engine + (unbounded trustline/offer traversal). +- All NFToken* — NFT page-chain `succ()` search (Mint/Burn/Modify/AcceptOffer) + and shared NFT offer directories (CreateOffer/CancelOffer). +- AccountDelete — deletes every src-owned object (unbounded footprint). +- Batch — meta-transaction; footprint is the union of its inner txs. +- Vault*, LoanBroker*/Loan* (lending), XChain*/bridge — pseudo-accounts and + cross-chain/compound state; unaudited dynamic footprints. +- PermissionedDomainSet — references arbitrary credential-issuer accounts in + `sfAcceptedCredentials`; keep global pending a deeper audit. +- Change (SetFee/EnableAmendment/UNLModify), LedgerStateFix — pseudo/global. + +## Phase 3 core — BUILT & verified +- `applyScheduled` (`Schedule.h`/`.cpp`): schedules a tx set, applies each + independent group in an isolated `OpenView` over the immutable closed + snapshot, and merges the disjoint write-sets into the target ledger. +- `src/tests/libxrpl/tx/ScheduledApply.cpp`: the **serial-vs-scheduled + differential** — both ledgers built by the test itself (bypassing the + canonicalising `TxTest::close`), asserting a byte-identical, non-trivial + account-state root. This is the determinism-critical contract; it passes. + +## Phase 3 — threaded executor + server integration — BUILT +- **Threaded execution.** `applyScheduled(..., unsigned workers)` applies the + independent groups across a thread pool (each group in its own view over the + immutable closed snapshot; disjoint write-sets merged sequentially in fixed + group order). `ScheduledApply.ThreadedMatchesSerialAcrossManyGroups` runs 12 + groups across 8 threads, 8 iterations, each byte-identical to serial. +- **Server integration.** `BuildLedger.cpp::applyTransactions` has a flag-gated + branch (`XRPL_PARALLEL_APPLY`, default OFF → unchanged default behaviour) that + schedules the canonical set and applies it via `applyScheduled`, merging into + the close `OpenView`. Works because `Application` *is-a* `ServiceRegistry` and + `OpenView` *is-a* `TxsRawView`. Compiles under `xrpld=ON`. + `XRPL_PARALLEL_APPLY_WORKERS` overrides the worker count + (default `clamp(cores-2, 2, 8)`). + +## Network determinism test via xrpld-lab (the live oracle) +A local multi-validator network is itself a determinism check: if parallel apply +were non-deterministic, validators would compute different ledger hashes and +fail to validate. Procedure: +1. Build: `cmake --build build --target xrpld` (xrpld=ON). +2. `cp build/xrpld /Users/infinityworks/projects/xrplf/xrpld-lab/xrpld` +3. `cd xrpld-lab && xrpld-lab create:network --protocol xrpl --local --num_validators 3 --num_peers 1 --genesis True` +4. `export XRPL_PARALLEL_APPLY=1` then run the cluster's `start.sh` (the env var + propagates to all `nohup ./xrpld` children). +5. Push disjoint-payment load at `ws://127.0.0.1:6016`. +6. Assert all validators agree on `ledger_hash`/`account_hash` each round (e.g. + poll `server_info`/`ledger` across nodes). Divergence ⇒ a determinism bug. + +## Network determinism test — RUN & PASSED (xrpld-lab, 3 validators) +Built the full `xrpld` binary from this branch and ran a 3-validator local network +via xrpld-lab with `XRPL_PARALLEL_APPLY=1` (4 workers) on every node: +- Consensus advanced normally (proposers=2, ~2s converge), parallel path active + on all nodes (logs: `Parallel apply: N applied across K group(s)`). +- Under disjoint-payment load, every validator independently scheduled identical + groups — `5 applied across 5 group(s)`, `10 across 1` (same-source funding) — + and produced **byte-identical `ledger_hash` AND `account_hash` on every ledger**. +- This is the within-run determinism oracle: 3 independent validators each + scheduling + thread-pool-applying the same tx sets agreed on every state root. + A nondeterministic apply would have diverged and stalled validation. It didn't. + +(The earlier standalone cross-restart hash comparison is an INVALID method — +two pure-serial runs also differ because standalone ledger composition varies +with submit/accept timing. The multi-validator within-run agreement above is the +valid test; the unit `ScheduledApply` differential is the controlled complement.) + +## Measured speedup (Release, apply engine in isolation) — MODEST, overhead-bound +`ScheduledApply.ThroughputBenchmark`, 400 disjoint payments, best of 5, Release +(NDEBUG → access-set assertion + instrumentation compiled out): + +| workers | us/tx | speedup | +|---|---|---| +| 1 | 35.4 | 1.00x | +| 2 | 30.4 | 1.17x | +| 4 | 24.9 | 1.42x (peak) | +| 8 | 35.2 | 1.01x (regressed) | + +Honest reading — this is NOT the headline 5–10x: +1. **Overhead-bound.** The current engine spawns `std::thread`s *per ledger* and + builds an `OpenView` per group with a sequential merge. For cheap payments + (~35 us/tx serial) that fixed overhead rivals the work, so it peaks ~1.4x at + 4 workers and goes net-negative by 8. A persistent thread pool + lower + per-group overhead is required to scale; the per-ledger spawn is the first + thing to fix. +2. **Apply isn't the payment bottleneck.** ~35 us/tx serial ≈ 28k apply/s on one + core — far above the ~159 TPS network ceiling. For payment-dominated load the + ceiling is elsewhere in the pipeline (consensus, relay, admission), so + parallelizing *apply* alone yields limited end-to-end gain. The plan's big + numbers require parallelizing the EXPENSIVE transactors (AMM/DEX/paths) — and + those are exactly the ones still `touchesGlobal` (serial) in v1. +3. **Worker default needs rethinking.** `clamp(cores-2, 2, 8)` over-threads this + workload; ~4 was best here. Tune per measurement, not a fixed default. + +Net: the parallelism is real and deterministic (verified), but v1's economic case +is weak — modest payment speedup, with the large wins gated behind parallelizing +the dynamic-footprint transactors and a lower-overhead executor. + +## What remains genuinely uncertifiable in a coding run +- **Network/production certification.** A green lab run is strong evidence but + not proof. Shipping parallel apply to mainnet still needs: ThreadSanitizer-clean + runs, adversarial scheduling (Antithesis), and a 12-month mainnet-replay + differential CI gate (hundreds of GB, out-of-repo). A determinism bug forks the + network, so these are non-negotiable before the flag becomes an amendment. +- **Phase 5 (amendment).** `ParallelApply` amendment + validator governance vote. +- **Productionization of the flag** (Config stanza / amendment gate instead of an + env var) and faithful failed/retry-set parity for adversarial (invalid-tx) + workloads — the current branch clears the set after a successful grouped apply, + which is correct for valid load but not yet a full match of the serial path's + failure bookkeeping. +- **Phase 4 (testnet load).** Operational: testnet + load harness + weeks of + runtime to hit the ≥1000 TPS target. Cannot be done from the repo. +- **Phase 5 (amendment).** `ParallelApply` amendment + governance vote. + +## How to verify locally +Debug build, `-Dtests=ON -Dxrpld=OFF`; build `xrpl.test.tx` and +`xrpl.test.protocol_autogen`; run both. The DEBUG assertion validates every +migrated transactor against real apply tests. `XRPL_ACCESS_AUDIT=1` logs the +measured footprint of `touchesGlobal` transactors (Phase-1.5 audit data).