feat: CLOB Caching

include/xrpl/ledger/DeferredRebuild.h

@@ -1,229 +0,0 @@
-#pragma once
-
-#include <xrpl/basics/base_uint.h>
-#include <xrpl/basics/hardened_hash.h>
-
-#include <array>
-#include <cstdint>
-#include <functional>
-#include <unordered_map>
-#include <vector>
-
-namespace xrpl {
-
-/** An inner-node position in a SHAMap that needs hash recomputation.
-
-    Plan 7's deferred-rebuild algorithm walks bottom-up, recomputing
-    each affected inner node's hash from its children. `AffectedNode`
-    identifies one such node by:
-      * depth:  0 = root, 1..63 = inner nodes, 64 = leaf (not included
-                in the plan output — only inner nodes are rebuilt)
-      * prefix: the leaf-key's first `depth*4` bits, with the remainder
-                zeroed. Two nodes at the same depth with the same
-                prefix are the same node.
-*/
-struct AffectedNode
-{
-    int depth;
-    uint256 prefix;
-
-    [[nodiscard]] bool
-    operator==(AffectedNode const& other) const noexcept
-    {
-        return depth == other.depth && prefix == other.prefix;
-    }
-};
-
-/** Plan which inner nodes need rebuild for a given set of leaf changes.
-
-    Returns the union of ancestor paths of all `modifiedKeys`, sorted
-    by depth descending so a bottom-up rebuild can iterate the result
-    and find each level's nodes before the level above.
-
-    The returned plan contains only INNER nodes (depths 0..63). The
-    leaves themselves are at depth 64 and are not in the plan — they
-    are the modifications, not nodes to be rebuilt.
-
-    Complexity: O(K * 64) where K is `modifiedKeys.size()`. For real
-    workloads (~thousands of modifications), this is microseconds.
-*/
-[[nodiscard]] std::vector<AffectedNode>
-planDeferredRebuild(std::vector<uint256> const& modifiedKeys);
-
-/** Compute the hash of a SHAMap inner node from its 16 children.
-
-    Byte-identical to `SHAMapInnerNode::updateHash()`:
-
-        sha512_half(
-            HashPrefix::InnerNode (4 bytes, big-endian) ||
-            child[0]  (32 bytes) ||
-            child[1]  (32 bytes) ||
-            ...
-            child[15] (32 bytes))
-
-    This is the elementary operation of plan-7's bottom-up rebuild:
-    given the (already-computed) 16 child hashes of an inner node,
-    produce that node's hash. Empty branches are passed as zero
-    uint256 — the same convention SHAMap uses.
-
-    Pure function; no SHAMap state, no allocation beyond a stack buffer.
-*/
-[[nodiscard]] uint256
-computeInnerNodeHash(std::array<uint256, 16> const& childHashes);
-
-/** Hash combiner for AffectedNode keys in unordered_map. */
-struct AffectedNodeHash
-{
-    [[nodiscard]] std::size_t
-    operator()(AffectedNode const& n) const noexcept
-    {
-        // Mix depth into the high bits of a hash of prefix.
-        std::size_t h = 0;
-        for (auto b : n.prefix)
-            h = h * 31 + b;
-        return h ^ (static_cast<std::size_t>(n.depth) << 56);
-    }
-};
-
-/** Map of recomputed inner-node hashes keyed by (depth, prefix). */
-using RebuildResult =
-    std::unordered_map<AffectedNode, uint256, AffectedNodeHash>;
-
-/** Walk a depth-descending plan and compute each affected node's new hash.
-
-    For each AffectedNode in the plan (deepest first), collect its 16
-    child hashes:
-      * If a child position is itself in the plan (and already
-        computed, since we walk deepest-first), use the computed hash.
-      * Otherwise, fall back to `getOriginalChildHash` — the callback
-        is expected to walk the parent SHAMap to find the hash at that
-        position.
-
-    Then `computeInnerNodeHash` over the 16 children yields the
-    affected node's new hash. The result map contains every entry in
-    the plan keyed by its (depth, prefix).
-
-    @param plan
-        Depth-descending plan from `planDeferredRebuild`.
-    @param getOriginalChildHash
-        Callable `uint256(int depth, uint256 const& prefix)` returning
-        the hash at the given position in the parent SHAMap. May
-        return uint256{} for absent positions.
-
-    @note Pure function; safe to call concurrently with disjoint plans.
-*/
-template <typename GetChildHashFn>
-[[nodiscard]] RebuildResult
-executeRebuildPlan(
-    std::vector<AffectedNode> const& plan,
-    GetChildHashFn getOriginalChildHash);
-
-namespace detail {
-
-// Forwarder for template instantiation; declared here, defined in .cpp.
-[[nodiscard]] RebuildResult
-executeRebuildPlanImpl(
-    std::vector<AffectedNode> const& plan,
-    std::function<uint256(int, uint256 const&)> getOriginalChildHash);
-
-}  // namespace detail
-
-template <typename GetChildHashFn>
-[[nodiscard]] RebuildResult
-executeRebuildPlan(
-    std::vector<AffectedNode> const& plan,
-    GetChildHashFn getOriginalChildHash)
-{
-    return detail::executeRebuildPlanImpl(
-        plan,
-        std::function<uint256(int, uint256 const&)>(
-            std::move(getOriginalChildHash)));
-}
-
-/** End-to-end deferred rebuild: produce the new SHAMap root hash.
-
-    Combines `planDeferredRebuild` + `executeRebuildPlan` into one call.
-    This is the consumer-facing API — the integration site only needs
-    to supply the set of modified keys and a callback that reads the
-    parent SHAMap. No AffectedNode plumbing is exposed.
-
-    @param modifiedKeys
-        Keys whose leaves have been added/replaced/deleted in this
-        ledger close. Empty → no rebuild; returns the existing root
-        via the callback at (depth=0, prefix=zero).
-    @param getOriginalChildHash
-        Callable `uint256(int depth, uint256 const& prefix)` returning
-        the hash at the given position in the parent SHAMap. May return
-        uint256{} for absent positions.
-    @return The new SHAMap root hash.
-*/
-template <typename GetChildHashFn>
-[[nodiscard]] uint256
-deferredRebuildRoot(
-    std::vector<uint256> const& modifiedKeys,
-    GetChildHashFn getOriginalChildHash)
-{
-    if (modifiedKeys.empty())
-        return getOriginalChildHash(0, uint256{});
-
-    auto const plan = planDeferredRebuild(modifiedKeys);
-    auto const result = executeRebuildPlan(plan, std::move(getOriginalChildHash));
-    AffectedNode const rootKey{0, uint256{}};
-    auto const it = result.find(rootKey);
-    if (it == result.end())
-        return uint256{};
-    return it->second;
-}
-
-/** Partition modified keys by their first nibble (0..15).
-
-    At depth 1 the root has 16 child subtrees, one per first-nibble
-    value. Keys in different subtrees rebuild independently, so this
-    partition is the basis for plan-7 P7.3's parallel-by-subtree
-    rebuild.
-
-    Returns 16 buckets, one per first-nibble value, each containing
-    only the keys whose first nibble matches the bucket index.
-*/
-[[nodiscard]] std::array<std::vector<uint256>, 16>
-partitionByFirstNibble(std::vector<uint256> const& modifiedKeys);
-
-namespace detail {
-
-[[nodiscard]] uint256
-deferredRebuildRootParallelImpl(
-    std::vector<uint256> const& modifiedKeys,
-    std::function<uint256(int, uint256 const&)> getOriginalChildHash);
-
-}  // namespace detail
-
-/** Parallel deferred rebuild: partition by first nibble, rebuild each
-    of the 16 subtrees in parallel, combine into the root.
-
-    Equivalent to `deferredRebuildRoot` in output; differs only in
-    execution strategy. Useful when the parent SHAMap is large enough
-    that the rebuild cost matters per-close. For small workloads
-    (handful of modifications), the serial path is faster — threading
-    overhead exceeds the work saved.
-
-    @param modifiedKeys
-        Keys whose leaves have been added/replaced/deleted.
-    @param getOriginalChildHash
-        Thread-safe callable; will be invoked concurrently from
-        multiple subtree workers.
-    @return The new SHAMap root hash, byte-identical to
-            `deferredRebuildRoot` over the same inputs.
-*/
-template <typename GetChildHashFn>
-[[nodiscard]] uint256
-deferredRebuildRootParallel(
-    std::vector<uint256> const& modifiedKeys,
-    GetChildHashFn getOriginalChildHash)
-{
-    return detail::deferredRebuildRootParallelImpl(
-        modifiedKeys,
-        std::function<uint256(int, uint256 const&)>(
-            std::move(getOriginalChildHash)));
-}
-
-}  // namespace xrpl

include/xrpl/ledger/FlatStateMap.h

@@ -1,347 +0,0 @@
-#pragma once
-
-#include <xrpl/basics/base_uint.h>
-#include <xrpl/basics/hardened_hash.h>
-#include <xrpl/protocol/Keylet.h>
-#include <xrpl/protocol/STLedgerEntry.h>
-
-#include <cstddef>
-#include <iterator>
-#include <memory>
-#include <shared_mutex>
-#include <unordered_map>
-#include <unordered_set>
-#include <vector>
-
-namespace xrpl {
-
-/** Flat keylet-indexed materialization of XRPL state.
-
-    The SHAMap is XRPL's authoritative state structure — it produces
-    the state root that consensus agrees on. But on its own, it forces
-    every state read to walk the trie: ~6–10 spinlocked inner-node
-    fetches per `read(Keylet)`.
-
-    `FlatStateMap` materializes the same keylet → SLE mapping into a
-    flat hash table. Once populated, lookups are a single
-    `unordered_map::find()` — nanoseconds, not microseconds.
-
-    This is the 2-writes-for-1-read pattern (Plan 6):
-      * On apply, the apply path dual-writes to SHAMap and FlatStateMap.
-      * On read, only FlatStateMap is consulted.
-
-    The flat map is purely auxiliary. The SHAMap remains authoritative,
-    can always be rebuilt from the underlying NodeStore, and is what the
-    network's state-root commitment is computed from. If FlatStateMap is
-    wrong, the differential invariant check at ledger close (Plan 6 P6.5)
-    catches it; there is **no runtime fallback to SHAMap descent on
-    miss** — a miss is a bug.
-
-    Thread-safe via a shared_mutex. Concurrent reads do not block each
-    other; writes are exclusive. Future phases may replace this with
-    a lock-free atomic-pointer-swap design.
-*/
-class FlatStateMap
-{
-public:
-    using key_type = uint256;
-    using value_type = std::shared_ptr<STLedgerEntry const>;
-
-    FlatStateMap() = default;
-    ~FlatStateMap() = default;
-
-    // Non-copyable, non-movable. The map owns a shared_mutex (not movable)
-    // and a potentially large hash table; callers that need ownership
-    // transfer should wrap in std::unique_ptr<FlatStateMap>.
-    FlatStateMap(FlatStateMap const&) = delete;
-    FlatStateMap&
-    operator=(FlatStateMap const&) = delete;
-    FlatStateMap(FlatStateMap&&) = delete;
-    FlatStateMap&
-    operator=(FlatStateMap&&) = delete;
-
-    /** Look up an SLE by its SHAMap key.
-
-        Returns nullptr if the key is not in the map. In Plan 6's pure
-        2w/1r model, callers reading state that *should* exist treat
-        nullptr as a precondition violation — there is no fallback path
-        that would recover from a missed entry.
-    */
-    [[nodiscard]] value_type
-    read(key_type const& key) const;
-
-    /** Test whether an SLE is present. O(1). */
-    [[nodiscard]] bool
-    exists(key_type const& key) const;
-
-    /** Insert a new SLE. Replaces any prior entry under the same key.
-
-        Used by both:
-          * the apply path's `view.insert()` (new ledger object), and
-          * the apply path's `view.update()` (mutating an existing SLE
-            produces a new shared_ptr value).
-    */
-    void
-    insert(key_type const& key, value_type sle);
-
-    /** Remove an SLE. No-op if the key is absent. */
-    void
-    erase(key_type const& key);
-
-    /** Number of SLEs currently materialized. */
-    [[nodiscard]] std::size_t
-    size() const;
-
-    /** Test whether the map is empty. O(1). */
-    [[nodiscard]] bool
-    empty() const;
-
-    /** Remove every entry. Used by tests and by snapshot reset. */
-    void
-    clear();
-
-    /** Deep-copy snapshot of the current state.
-
-        The snapshot is a frozen FlatStateMap (returned by value-like
-        unique_ptr) that shares the underlying SLE objects via
-        shared_ptr but has its own hash table. Subsequent writes to the
-        source FlatStateMap do not affect the snapshot.
-
-        Snapshot cost is O(N) in entry count — ~one pointer copy per
-        entry plus bucket allocation. For current mainnet (~10M SLEs)
-        this is ~100–200 ms; expensive enough that snapshots should be
-        per-ledger-close, not per-transaction.
-
-        A future phase will replace this with a persistent / HAMT
-        structure that gives O(log N) snapshot and structural sharing
-        across versions.
-    */
-    [[nodiscard]] std::unique_ptr<FlatStateMap>
-    snapshot() const;
-
-    /** Visit every (key, SLE) pair under a single shared lock.
-
-        @param visitor  Called as `void(key_type const&, value_type const&)`.
-
-        The lock is held for the duration of the iteration; visitors
-        must not call back into the same FlatStateMap (deadlock /
-        recursive shared_lock UB). Visitors that want to mutate state
-        should collect keys first and apply mutations after iteration
-        returns.
-    */
-    template <typename F>
-    void
-    forEach(F&& visitor) const
-    {
-        std::shared_lock<std::shared_mutex> lock(mutex_);
-        for (auto const& [key, sle] : map_)
-            visitor(key, sle);
-    }
-
-private:
-    using HashFn = HardenedHash<>;
-    using MapType = std::unordered_map<key_type, value_type, HashFn>;
-
-    mutable std::shared_mutex mutex_;
-    MapType map_;
-};
-
-// Forward declarations to avoid pulling heavy headers into this file.
-class ReadView;
-class Ledger;
-
-/** Populate a FlatStateMap from every SLE in a ReadView.
-
-    Used at node startup to build the flat materialization from a
-    SHAMap-backed authoritative ledger. Cost is O(N) iterations of
-    `view.sles`, each of which descends the SHAMap, so this is
-    expected to take seconds-to-minutes for a current mainnet-sized
-    ledger (~10M SLEs). Run once on startup; subsequent ledgers are
-    maintained incrementally via dual-write at apply time (P6.3).
-
-    @pre `target` is empty. (Not enforced — replacing existing entries
-         is well-defined, but mixing populated state with an externally-
-         provided ReadView is a bug-shaped pattern; assert in DEBUG.)
-*/
-void
-populateFromReadView(FlatStateMap& target, ReadView const& source);
-
-/** Populate a FlatStateMap from any forward range of `shared_ptr<SLE const>`.
-
-    Lower-level building block underlying `populateFromReadView`. Useful
-    in tests (the range can be a `std::vector<shared_ptr<SLE const>>`)
-    and in non-ReadView contexts (e.g., reloading a persisted flat-map
-    sidecar at startup).
-
-    The range element type must be `shared_ptr<SLE const>` (or
-    implicitly convertible). Each element's `.key()` becomes the
-    FlatStateMap key.
-*/
-template <typename Range>
-void
-populateFromRange(FlatStateMap& target, Range const& sles)
-{
-    for (auto const& sle : sles)
-        target.insert(sle->key(), sle);
-}
-
-// ---------------------------------------------------------------------------
-// Mirror helpers (P6.3).
-//
-// The xrpld `RawView` interface defines three pure-virtual methods that
-// every state mutation flows through: `rawInsert`, `rawReplace`, and
-// `rawErase`. In plan-6's 2-writes-for-1-read pattern, every such call
-// must also update the flat map. These helpers perform the flat-map side
-// of that dual-write — they exist as standalone functions (rather than
-// methods on FlatStateMap) so the Ledger integration is a one-line
-// addition at each `raw*` override site, with no FlatStateMap class
-// surface added for purely Ledger-specific semantics.
-//
-// Permissive semantics: `mirrorRawReplace` on an absent key inserts;
-// `mirrorRawErase` on an absent key is a no-op. The SHAMap side enforces
-// the precondition (replace requires existence); the flat mirror ensures
-// the post-state matches whatever the SHAMap committed. If the caller
-// gets it wrong, the differential invariant check at close (P6.5) is the
-// stop-the-line gate.
-// ---------------------------------------------------------------------------
-
-void
-mirrorRawInsert(FlatStateMap& map, std::shared_ptr<STLedgerEntry const> sle);
-
-void
-mirrorRawReplace(FlatStateMap& map, std::shared_ptr<STLedgerEntry const> sle);
-
-void
-mirrorRawErase(FlatStateMap& map, std::shared_ptr<STLedgerEntry const> const& sle);
-
-void
-mirrorRawErase(FlatStateMap& map, uint256 const& key);
-
-// ---------------------------------------------------------------------------
-// Keylet-aware read (P6.4).
-//
-// This is the read-side counterpart to `mirrorRaw*` — the testable unit
-// underlying `Ledger::read(Keylet)`'s flat-map path. It looks up the
-// SLE by `k.key` and verifies the SLE matches the keylet's expected
-// type via `Keylet::check`. On either a miss or a type mismatch, it
-// returns nullptr — matching the contract of `Ledger::read`.
-//
-// Plan 6 v2 semantics: this function does not consult a SHAMap or any
-// other source on miss. When a FlatStateMap is the read source of
-// truth, a miss IS the answer. The differential invariant check at
-// close (P6.5) is what makes that safe.
-// ---------------------------------------------------------------------------
-
-std::shared_ptr<STLedgerEntry const>
-readFromFlatStateMap(FlatStateMap const& map, Keylet const& k);
-
-// ---------------------------------------------------------------------------
-// Differential invariant (P6.5).
-//
-// At every ledger close, the flat map's key-set must match the
-// SHAMap's key-set. `diffFlatStateKeys` produces both sides of the
-// disagreement; `flatStateMapMatches` is the boolean predicate the
-// hot-path integration calls (and fails the close if it returns false).
-//
-// Content drift (right keys, wrong SLE bodies) is a separate, stronger
-// invariant. It's prevented by construction: the mirror helpers write
-// exactly the SLE the caller passed to raw*. If mirror helpers and
-// wiring are both correct, the membership check above is sufficient.
-// ---------------------------------------------------------------------------
-
-struct FlatStateKeyDiff
-{
-    std::vector<uint256> missingFromFlat;
-    std::vector<uint256> extraInFlat;
-};
-
-template <typename SourceRange>
-[[nodiscard]] FlatStateKeyDiff
-diffFlatStateKeys(FlatStateMap const& flat, SourceRange const& sourceKeys)
-{
-    FlatStateKeyDiff diff;
-
-    // Pass 1: walk source keys; collect any absent from flat. Record
-    // which keys we've seen so pass 2 can spot phantoms.
-    std::unordered_set<uint256, HardenedHash<>> seen;
-    seen.reserve(static_cast<std::size_t>(std::distance(
-        std::begin(sourceKeys), std::end(sourceKeys))));
-
-    for (auto const& key : sourceKeys)
-    {
-        seen.insert(key);
-        if (!flat.exists(key))
-            diff.missingFromFlat.push_back(key);
-    }
-
-    // Pass 2: walk flat; anything not in `seen` is a phantom.
-    flat.forEach(
-        [&seen, &diff](uint256 const& key, auto const& /*sle*/) {
-            if (!seen.contains(key))
-                diff.extraInFlat.push_back(key);
-        });
-
-    return diff;
-}
-
-template <typename SourceRange>
-[[nodiscard]] bool
-flatStateMapMatches(FlatStateMap const& flat, SourceRange const& sourceKeys)
-{
-    auto const diff = diffFlatStateKeys(flat, sourceKeys);
-    return diff.missingFromFlat.empty() && diff.extraInFlat.empty();
-}
-
-/** Compare a FlatStateMap against a SHAMap-like source.
-
-    `ShaMapLike` is any range whose elements expose a `key()` accessor
-    returning a `uint256`. The real `SHAMap` satisfies this contract
-    (its iterators yield `SHAMapItem`s with `.key()`), as does the
-    `MockShaMapItem` used in tests.
-
-    This is the helper the Ledger integration calls to run the P6.5
-    differential invariant. It extracts keys into a transient buffer
-    (O(N) allocation, ~N pointers' worth of memory) and forwards to
-    `flatStateMapMatches`. The transient buffer is acceptable at close
-    cadence; the integration can later optimize by walking the SHAMap
-    in-place once profiling shows the allocation matters.
-*/
-template <typename ShaMapLike>
-[[nodiscard]] bool
-flatStateMapMatchesShaMap(FlatStateMap const& flat, ShaMapLike const& shaMap)
-{
-    std::vector<uint256> keys;
-    for (auto const& item : shaMap)
-        keys.push_back(item.key());
-    return flatStateMapMatches(flat, keys);
-}
-
-// ---------------------------------------------------------------------------
-// A-phase integration: attach a populated FlatStateMap to a Ledger.
-//
-// This is the public entry point a node or test uses to "turn on" the
-// flat-map read path for a Ledger. The function:
-//   1. Allocates a new FlatStateMap.
-//   2. Eagerly populates it by walking every SLE in the Ledger.
-//   3. Attaches it via `Ledger::setFlatStateMap`.
-//
-// After this call:
-//   * `Ledger::flatStateMap()` returns the populated map
-//   * `Ledger::read(keylet)` routes through the flat map (no SHAMap
-//     descent on the hot path; see P6.4 wiring)
-//   * `Ledger::raw{Insert,Replace,Erase}` mirror writes to the flat
-//     map alongside the SHAMap (see P6.3 wiring)
-//   * `Ledger::validateFlatStateMapMatchesShaMap()` returns true
-//
-// Repeat calls discard the prior map and produce a fresh one.
-//
-// Cost: O(N) walk over the Ledger's state SHAMap to populate the flat
-// map. For mainnet-scale state, this is bounded by SHAMap traversal
-// speed — typically minutes once. Run at node startup or whenever a
-// Ledger first becomes "live" (the one apply writes to).
-// ---------------------------------------------------------------------------
-
-void
-attachFlatStateMapTo(Ledger& ledger);
-
-}  // namespace xrpl

include/xrpl/ledger/Ledger.h

@@ -14,7 +14,6 @@
 
 namespace xrpl {
 
-class FlatStateMap;
 class ServiceRegistry;
 class Job;
 class TransactionMaster;
@@ -365,38 +364,6 @@ public:
     std::shared_ptr<SLE>
     peek(Keylet const& k) const;
 
-    //
-    // Flat-state mirror (Plan 6 P6.3).
-    //
-    // When a FlatStateMap is attached, every successful `raw*` call below
-    // mirrors the operation into the map via the matching `mirrorRaw*`
-    // helper. With no map attached (the default), the ledger behaves
-    // exactly as before — no allocation, no mutex, no observable change.
-    // The map's lifetime is managed by the caller (typically the
-    // Application owns the live ledger's map).
-    //
-
-    void
-    setFlatStateMap(std::shared_ptr<FlatStateMap> map);
-
-    [[nodiscard]] std::shared_ptr<FlatStateMap>
-    flatStateMap() const;
-
-    /** Run the Plan 6 P6.5 differential invariant.
-
-        Returns true iff (a) no FlatStateMap is attached (vacuously
-        true — there is no second source of truth to disagree), or
-        (b) the attached FlatStateMap's key-set matches the SHAMap's
-        key-set exactly.
-
-        Intended to be called at ledger close, before the new state
-        root is published. A false return is a stop-the-line bug —
-        the integrator should crash rather than publish a state root
-        that disagrees with the read path.
-    */
-    [[nodiscard]] bool
-    validateFlatStateMapMatchesShaMap() const;
-
 private:
     class SlesIterImpl;
     class TxsIterImpl;
@@ -433,11 +400,6 @@ private:
     // A SHAMap containing the state objects for this ledger.
     SHAMap mutable stateMap_;
 
-    // Optional flat keylet→SLE mirror. When non-null, every successful
-    // raw* state mutation is mirrored into this map. See FlatStateMap.h
-    // and the Plan 6 docs in tasks/.
-    std::shared_ptr<FlatStateMap> mutable flatStateMap_;
-
     // Protects fee variables
     std::mutex mutable mutex_;
 

include/xrpl/ledger/OpenView.h

@@ -1,7 +1,9 @@
 #pragma once
 
+#include <xrpl/ledger/OrderBookIndex.h>
 #include <xrpl/ledger/RawView.h>
 #include <xrpl/ledger/ReadView.h>
+#include <xrpl/ledger/TopOfBookCache.h>
 #include <xrpl/ledger/detail/RawStateTable.h>
 #include <xrpl/protocol/STArray.h>
 #include <xrpl/protocol/XRPAmount.h>
@@ -89,6 +91,17 @@ private:
 
     bool open_ = true;
 
+    // Per-view top-of-book cache. Lifetime is the view's lifetime; on
+    // OpenView copy (used to snapshot for parallel apply / batch views),
+    // the underlying data is copied but counters reset.
+    mutable TopOfBookCache topOfBookCache_;
+
+    // Per-view ordered order-book index (Plan 9). Generalizes the cache from
+    // "best page" to the full quality-ordered offer sequence, letting the
+    // crossing path iterate via an in-memory cursor instead of re-walking the
+    // SHAMap with succ() per offer. Maintained off the same notifications.
+    mutable OrderBookIndex orderBookIndex_;
+
 public:
     OpenView() = delete;
     OpenView&
@@ -200,6 +213,46 @@ public:
     std::shared_ptr<SLE const>
     read(Keylet const& k) const override;
 
+    // Top-of-book cache hooks
+
+    [[nodiscard]] std::optional<uint256>
+    topOfBookFirstPage(Book const& book) const override;
+
+    void
+    recordTopOfBook(Book const& book, uint256 const& firstPageKey) const override;
+
+    void
+    notifyOfferInserted(Book const& book, uint256 const& dirKey, uint256 const& offerKey)
+        const override;
+
+    void
+    notifyOfferDeleted(Book const& book, uint256 const& dirKey, uint256 const& offerKey)
+        const override;
+
+    [[nodiscard]] std::optional<std::vector<uint256>>
+    orderedBook(Book const& book) const override;
+
+    [[nodiscard]] TopOfBookCache const&
+    topOfBookCache() const noexcept
+    {
+        return topOfBookCache_;
+    }
+
+    [[nodiscard]] OrderBookIndex const&
+    orderBookIndex() const noexcept
+    {
+        return orderBookIndex_;
+    }
+
+    // Non-const access for seeding (rebuild-from-state at attach time) and for
+    // the cursor's lazy populate. The index is auxiliary, so this never affects
+    // the authoritative state.
+    [[nodiscard]] OrderBookIndex&
+    orderBookIndex() noexcept
+    {
+        return orderBookIndex_;
+    }
+
     std::unique_ptr<SlesType::iter_base>
     slesBegin() const override;
 

include/xrpl/ledger/OrderBookIndex.h

@@ -0,0 +1,181 @@
+#pragma once
+
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/ledger/detail/PersistentOrderTree.h>
+#include <xrpl/protocol/Book.h>
+
+#include <atomic>
+#include <cstddef>
+#include <cstdint>
+#include <optional>
+#include <shared_mutex>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+namespace xrpl {
+
+class ReadView;
+
+/** Deterministic, ordered, **persistent** in-memory index of every active order
+    book.
+
+    `BookTip::step()` finds the next offer to cross by calling `ReadView::succ()`
+    — an O(log N) SHAMap successor walk from the book root, re-done once per
+    consumed offer. Profiling shows that walk is ~32% of crossing-apply cost.
+    This index materializes the same quality-ordered offer sequence so iteration
+    becomes an in-memory cursor advance instead of a trie re-walk.
+
+    It generalizes `TopOfBookCache` from "the best directory page" to "the full
+    ordered book". Like `FlatStateMap`, it is **auxiliary**: the SHAMap remains
+    the authoritative state and the source of the consensus root. The index is
+    rebuildable from the SHAMap at any time (`rebuildBook`) and differentially
+    validated against it (`validateMatchesShaMap`); a divergence is a bug in the
+    maintenance hooks, never a fallback.
+
+    **Persistence.** Each book's offers live in an immutable, structurally-shared
+    weight-balanced tree ([[detail/PersistentOrderTree.h]]). `clone()` copies only
+    the per-book `shared_ptr` roots (O(#books)), not the offers — so the
+    open-ledger copy-on-write (`OpenView` copy per `modify()`) preserves the index
+    cheaply and it stays warm across transactions, instead of cold-starting and
+    rebuilding per tx. Immutable nodes also make the COW rollback of a discarded
+    sandbox free: it simply drops its own root pointers.
+
+    Ordering invariant (the load-bearing property for bit-exact crossing):
+
+      - Books are keyed by `Book` (which already carries the permissioned-DEX
+        `domain`), so each book — open or domain — is indexed independently.
+      - Within a book, the tree is keyed by `(dirRoot, insertSeq)`. `dirRoot` is
+        the quality-directory root key; ascending == best-quality-first ==
+        `succ()` order. `insertSeq` is a per-book monotonic counter capturing
+        directory append order; since `dirRemove` preserves relative order and
+        offer keys are never reused, in-order traversal reproduces the SHAMap
+        directory walk byte-for-byte.
+
+    Maintenance drives `insertOffer`/`deleteOffer` from the offer-mutation
+    notifications (`notifyOfferInserted`/`notifyOfferDeleted`), which fire with
+    the quality-directory root key and the offer key.
+*/
+class OrderBookIndex
+{
+public:
+    OrderBookIndex() = default;
+
+    /** Move-construct by locking the source and stealing its book map.
+        Counters are not transferred (a fresh view starts its own accounting). */
+    OrderBookIndex(OrderBookIndex&& other);
+
+    OrderBookIndex(OrderBookIndex const&) = delete;
+    OrderBookIndex&
+    operator=(OrderBookIndex const&) = delete;
+    OrderBookIndex&
+    operator=(OrderBookIndex&&) = delete;
+
+    /** Cheap structural copy: clones the per-book tree roots (O(#books)
+        shared_ptr copies), sharing all offer nodes. Used by the `OpenView` copy
+        ctor so the index stays warm across the open-ledger COW. Counters reset. */
+    [[nodiscard]] OrderBookIndex
+    clone() const;
+
+    // --- maintenance (apply-path hooks) ---
+
+    /** Record that `offerKey` was inserted into `book` at quality-directory root
+        `dirRoot`. Appended (next insertSeq) so it sorts after same-level offers,
+        preserving directory order. */
+    void
+    insertOffer(Book const& book, uint256 const& dirRoot, uint256 const& offerKey);
+
+    /** Record that `offerKey` was removed from `book` at quality-directory root
+        `dirRoot`. The book is dropped when it empties. Removing an absent key is
+        a no-op. */
+    void
+    deleteOffer(Book const& book, uint256 const& dirRoot, uint256 const& offerKey);
+
+    // --- ordered read access (BookTip seam) ---
+
+    /** All offer keys of `book`, best-quality-first, directory order within a
+        level. Empty if the book is absent. */
+    [[nodiscard]] std::vector<uint256>
+    flatten(Book const& book) const;
+
+    /** The best (first) offer key of `book`, or nullopt if absent. */
+    [[nodiscard]] std::optional<uint256>
+    firstOffer(Book const& book) const;
+
+    // --- rebuild / validation (composition with the authoritative SHAMap) ---
+
+    /** Repopulate `book` from `view` by the canonical quality-ordered walk
+        (`succ()` over directory roots + directory iteration within each). */
+    void
+    rebuildBook(ReadView const& view, Book const& book);
+
+    /** True iff the maintained sequence for `book` equals a fresh walk of
+        `view`. The differential invariant. */
+    [[nodiscard]] bool
+    validateMatchesShaMap(ReadView const& view, Book const& book) const;
+
+    // --- bookkeeping ---
+
+    /** True if `book` has an entry (at least one offer). O(1). Present implies
+        non-empty (empty books are dropped). */
+    [[nodiscard]] bool
+    contains(Book const& book) const;
+
+    void
+    eraseBook(Book const& book);
+
+    void
+    clear();
+
+    [[nodiscard]] std::size_t
+    bookCount() const;
+
+    [[nodiscard]] std::size_t
+    offerCount(Book const& book) const;
+
+    [[nodiscard]] std::uint64_t
+    inserts() const noexcept
+    {
+        return inserts_.load(std::memory_order_relaxed);
+    }
+
+    [[nodiscard]] std::uint64_t
+    deletes() const noexcept
+    {
+        return deletes_.load(std::memory_order_relaxed);
+    }
+
+    [[nodiscard]] std::uint64_t
+    rebuilds() const noexcept
+    {
+        return rebuilds_.load(std::memory_order_relaxed);
+    }
+
+    // --- operator-facing kill switch (mirrors TopOfBookCache) ---
+
+    [[nodiscard]] static bool
+    enabled() noexcept;
+
+    static void
+    setEnabled(bool on) noexcept;
+
+private:
+    struct BookState
+    {
+        detail::OrderTreePtr root;  // persistent (dirRoot, insertSeq) -> offerKey
+        std::uint64_t nextSeq{0};   // per-book monotonic append counter
+    };
+
+    // Canonical quality-ordered walk of `book` in `view`: (dirRoot, offerKey)
+    // for each offer, best-quality-first, directory order within a level.
+    [[nodiscard]] static std::vector<std::pair<uint256, uint256>>
+    walkBook(ReadView const& view, Book const& book);
+
+    mutable std::shared_mutex mutex_;
+    std::unordered_map<Book, BookState> books_;
+    std::atomic<std::uint64_t> inserts_{0};
+    std::atomic<std::uint64_t> deletes_{0};
+    std::atomic<std::uint64_t> rebuilds_{0};
+};
+
+}  // namespace xrpl

include/xrpl/ledger/ReadView.h

@@ -3,6 +3,7 @@
 #include <xrpl/basics/chrono.h>
 #include <xrpl/beast/hash/uhash.h>
 #include <xrpl/ledger/detail/ReadViewFwdRange.h>
+#include <xrpl/protocol/Book.h>
 #include <xrpl/protocol/Fees.h>
 #include <xrpl/protocol/IOUAmount.h>
 #include <xrpl/protocol/Indexes.h>
@@ -16,6 +17,7 @@
 #include <cstdint>
 #include <optional>
 #include <unordered_set>
+#include <vector>
 
 namespace xrpl {
 
@@ -188,6 +190,68 @@ public:
         return count;
     }
 
+    //
+    // Top-of-book cache hooks
+    //
+    // The default implementations make every non-overriding view a no-op
+    // pass-through, so non-orderbook code is unaffected. OpenView overrides
+    // these to maintain a real `TopOfBookCache`; views that wrap a base
+    // (ApplyViewBase, PaymentSandbox, ...) delegate to that base.
+
+    /** Return the cached keylet of the best (lowest-keyed) directory page
+        for `book`, if known. std::nullopt forces a `succ()` fallback.
+    */
+    [[nodiscard]] virtual std::optional<uint256>
+    topOfBookFirstPage(Book const& book) const
+    {
+        return std::nullopt;
+    }
+
+    /** Populate the cache after a `succ()`-driven discovery. Called from
+        the cold path of `BookTip::step()`.
+    */
+    virtual void
+    recordTopOfBook(Book const& book, uint256 const& firstPageKey) const
+    {
+    }
+
+    /** Apply-path notification: an offer was inserted into `book` at
+        directory keylet `dirKey`. The cache may use this to update or
+        invalidate its entry; the call must be safe under any base view.
+    */
+    virtual void
+    notifyOfferInserted(Book const& book, uint256 const& dirKey, uint256 const& offerKey) const
+    {
+    }
+
+    /** Apply-path notification: an offer was deleted from `book` at
+        directory keylet `dirKey`. If the deleted offer was on the
+        cached top page, the cache invalidates that entry.
+
+        `offerKey` is the deleted offer's ledger key — unused by the cache,
+        consumed by the order-book index.
+    */
+    virtual void
+    notifyOfferDeleted(Book const& book, uint256 const& dirKey, uint256 const& offerKey) const
+    {
+    }
+
+    /** Return `book`'s offer keys best-quality-first (the order the crossing
+        path consumes them), or std::nullopt to force the `succ()`-based walk.
+
+        Lets `BookTip` iterate the book from an in-memory cursor instead of
+        re-walking the SHAMap with `succ()` per offer. A returned vector is
+        guaranteed complete for `book` — implementations rebuild from the
+        authoritative state on a miss, so the cursor can never under-include.
+        Empty/absent books return nullopt (the cheap `succ()` path finds
+        nothing). Default: no index, always nullopt.
+    */
+    [[nodiscard]] virtual std::optional<std::vector<uint256>>
+    orderedBook(Book const& book) const
+    {
+        return std::nullopt;
+    }
+
     // used by the implementation
     [[nodiscard]] virtual std::unique_ptr<SlesType::iter_base>
     slesBegin() const = 0;

include/xrpl/ledger/Sandbox.h

@@ -35,6 +35,7 @@ public:
     apply(RawView& to)
     {
         items_.apply(to);
+        flushTopOfBookNotifications();
     }
 };
 

include/xrpl/ledger/TopOfBookCache.h

@@ -0,0 +1,163 @@
+#pragma once
+
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/protocol/Book.h>
+#include <xrpl/protocol/Protocol.h>
+
+#include <atomic>
+#include <cstdint>
+#include <mutex>
+#include <optional>
+#include <unordered_map>
+
+namespace xrpl {
+
+/** One entry in the top-of-book cache.
+
+    Records the keylet of the best-quality (lowest-keyed) directory page
+    for a single order book at the time the entry was recorded.
+*/
+struct TopOfBookEntry
+{
+    /// Keylet of the best directory page for the book.
+    uint256 firstPageKey;
+    /// Quality bits encoded in firstPageKey (decoded for fast comparison).
+    std::uint64_t bestQuality{0};
+    /// Ledger sequence at which this entry was populated.
+    LedgerIndex asOfLedger{0};
+};
+
+/** Cache of "best directory page" keylet per active order book.
+
+    Reads of the top of an order book usually return the same directory page
+    over and over, but `BookTip::step()` re-walks the SHAMap on every call.
+    This cache memoizes that result. Lookups become a single hash-map probe;
+    the SHAMap successor walk happens only on cold or invalidated entries.
+
+    The cache is auxiliary — invalidating an entry is always safe, since the
+    next read repopulates lazily via `ReadView::succ()`. That property is what
+    lets the cache ship without an amendment.
+
+    Maintenance rules, applied at the apply path:
+
+    - **Offer inserted**: if the new offer's directory keylet is at-or-better
+      than the cached top, update the entry. Otherwise no-op.
+    - **Offer deleted**: if the deleted offer was on the cached top page,
+      invalidate. Otherwise no-op.
+
+    A best-page key is `keylet::quality(keylet::kBook(book), rate).key`. All
+    pages of a single book share the same prefix, so lower uint256 key =
+    better quality. Comparisons in this file rely on that ordering.
+*/
+class TopOfBookCache
+{
+public:
+    TopOfBookCache() = default;
+
+    /** Copy-construct (used when snapshotting open->closed ledger).
+
+        Hit/miss/invalidation counters are not copied; only the data is.
+    */
+    TopOfBookCache(TopOfBookCache const& other);
+
+    /** Move-construct by locking the source and stealing its map.
+
+        Needed because views that own a cache (OpenView) are moveable;
+        std::mutex is not, so the move is implemented via lock-and-move.
+        Counters are not transferred.
+    */
+    TopOfBookCache(TopOfBookCache&& other);
+
+    TopOfBookCache&
+    operator=(TopOfBookCache const&) = delete;
+    TopOfBookCache&
+    operator=(TopOfBookCache&&) = delete;
+
+    /** Look up the cached top of `book`.
+
+        Returns std::nullopt on miss. Hit/miss counters are updated.
+    */
+    [[nodiscard]] std::optional<TopOfBookEntry>
+    get(Book const& book) const;
+
+    /** Record (or overwrite) a top-of-book entry for `book`.
+
+        Called from the cold path after `succ()` discovers the first page.
+    */
+    void
+    record(Book const& book, uint256 const& firstPageKey, LedgerIndex seq);
+
+    /** Notify the cache that an offer was inserted into `book` at directory
+        keylet `dirKey`.
+
+        If the new keylet is better than (less than) the cached top, the entry
+        is updated. If it is equal, no change. If worse, no change.
+
+        If no entry exists for `book`, this is a no-op: the next read will
+        populate from `succ()`.
+    */
+    void
+    onOfferInsert(Book const& book, uint256 const& dirKey, LedgerIndex seq);
+
+    /** Notify the cache that an offer was deleted from `book` at directory
+        keylet `dirKey`.
+
+        If the delete was on the cached top page, invalidate (the page may
+        now be empty, or the offer count is irrelevant — next read repopulates).
+        Otherwise no-op.
+    */
+    void
+    onOfferDelete(Book const& book, uint256 const& dirKey);
+
+    /** Drop the entry for `book` unconditionally.
+
+        Used as a safety hatch and by tests.
+    */
+    void
+    invalidate(Book const& book);
+
+    /** Drop every entry. */
+    void
+    clear();
+
+    [[nodiscard]] std::size_t
+    size() const;
+
+    [[nodiscard]] std::uint64_t
+    hits() const noexcept
+    {
+        return hits_.load(std::memory_order_relaxed);
+    }
+
+    [[nodiscard]] std::uint64_t
+    misses() const noexcept
+    {
+        return misses_.load(std::memory_order_relaxed);
+    }
+
+    [[nodiscard]] std::uint64_t
+    invalidations() const noexcept
+    {
+        return invalidations_.load(std::memory_order_relaxed);
+    }
+
+    /** Operator-facing kill switch.
+
+        When false, `BookTip` skips cache consults and writes entirely,
+        falling back to plain `succ()`. Default is true.
+    */
+    [[nodiscard]] static bool
+    enabled() noexcept;
+
+    static void
+    setEnabled(bool on) noexcept;
+
+private:
+    mutable std::mutex mutex_;
+    std::unordered_map<Book, TopOfBookEntry> map_;
+    mutable std::atomic<std::uint64_t> hits_{0};
+    mutable std::atomic<std::uint64_t> misses_{0};
+    std::atomic<std::uint64_t> invalidations_{0};
+};
+
+}  // namespace xrpl

include/xrpl/ledger/detail/ApplyViewBase.h

@@ -3,8 +3,13 @@
 #include <xrpl/ledger/ApplyView.h>
 #include <xrpl/ledger/ReadView.h>
 #include <xrpl/ledger/detail/ApplyStateTable.h>
+#include <xrpl/protocol/Book.h>
 #include <xrpl/protocol/XRPAmount.h>
 
+#include <unordered_set>
+#include <utility>
+#include <vector>
+
 namespace xrpl::detail {
 
 class ApplyViewBase : public ApplyView, public RawView
@@ -43,6 +48,26 @@ public:
     [[nodiscard]] std::shared_ptr<SLE const>
     read(Keylet const& k) const override;
 
+    // Top-of-book cache hooks — delegated to the wrapped base view so
+    // sandboxed views share the underlying open-ledger cache.
+
+    [[nodiscard]] std::optional<uint256>
+    topOfBookFirstPage(Book const& book) const override;
+
+    void
+    recordTopOfBook(Book const& book, uint256 const& firstPageKey) const override;
+
+    void
+    notifyOfferInserted(Book const& book, uint256 const& dirKey, uint256 const& offerKey)
+        const override;
+
+    void
+    notifyOfferDeleted(Book const& book, uint256 const& dirKey, uint256 const& offerKey)
+        const override;
+
+    [[nodiscard]] std::optional<std::vector<uint256>>
+    orderedBook(Book const& book) const override;
+
     [[nodiscard]] std::unique_ptr<SlesType::iter_base>
     slesBegin() const override;
 
@@ -95,10 +120,45 @@ public:
     void
     rawDestroyXRP(XRPAmount const& feeDrops) override;
 
+    /** Flush buffered top-of-book notifications to the wrapped base view.
+
+        Called by `Sandbox::apply` (and similar commit points) after the
+        state table itself has been applied. Notifications buffered during
+        the sandbox's lifetime are replayed against `base_` in insertion
+        order so the parent cache only sees changes that actually commit.
+    */
+    void
+    flushTopOfBookNotifications() const;
+
+    /** Discard buffered notifications (e.g. when a sandbox is dropped
+        without applying). Safe to call multiple times.
+    */
+    void
+    discardTopOfBookNotifications() const noexcept;
+
 protected:
     ApplyFlags flags_;
     ReadView const* base_;
     detail::ApplyStateTable items_;
+
+    // Top-of-book cache notifications are buffered here for the lifetime
+    // of the sandbox and only flushed to `base_` on `apply()`. This keeps
+    // rolled-back transactions (e.g. FillOrKill via the sbCancel branch
+    // of OfferCreate) from polluting the parent's cache.
+    //
+    // `dirtyBooks_` records every book mutated by buffered notifications;
+    // reads against `topOfBookFirstPage` skip the cache for these books so
+    // we never observe our own un-committed state. Outside of the dirty
+    // set, the parent's cache is trusted as usual.
+    struct OfferNote
+    {
+        Book book;
+        uint256 dirKey;
+        uint256 offerKey;
+        bool isDelete;
+    };
+    mutable std::vector<OfferNote> pendingTopOfBookNotifications_;
+    mutable std::unordered_set<Book> dirtyBooks_;
 };
 
 }  // namespace xrpl::detail

include/xrpl/ledger/detail/PersistentOrderTree.h

@@ -0,0 +1,257 @@
+#pragma once
+
+#include <xrpl/basics/base_uint.h>
+
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <vector>
+
+namespace xrpl::detail {
+
+/** Persistent (immutable, structurally-shared) ordered tree for the order-book
+    index.
+
+    A weight-balanced BST (Adams BB[α], the family used by Haskell `Data.Map`
+    and std::map-replacement libraries) of immutable `shared_ptr<const Node>`.
+    Keyed by `(dirRoot, insertSeq)`:
+
+      - `dirRoot` ascending == best-quality-first (book directory pages share a
+        prefix, quality is in the low bytes — lower key = better quality).
+      - `insertSeq` ascending within a `dirRoot` == directory append order
+        (the per-book monotonic counter mirrors `dirAppend`; `dirRemove`
+        preserves relative order, so this reproduces the directory walk
+        byte-for-byte).
+
+    Operations are persistent via path-copying: insert/delete reallocate only
+    the O(log n) nodes on the root→leaf path and SHARE every untouched subtree.
+    A "copy" of a tree is just copying the root `shared_ptr` — O(1) — which is
+    what lets the order-book index survive the open-ledger copy-on-write cheaply
+    and stay warm across transactions.
+
+    Immutable nodes are safe to share across threads/snapshots without locking.
+*/
+struct OrderTreeNode
+{
+    uint256 dirRoot;
+    std::uint64_t insertSeq;
+    uint256 offerKey;
+    std::uint32_t size;  // subtree node count (balance + rank)
+    std::shared_ptr<OrderTreeNode const> left;
+    std::shared_ptr<OrderTreeNode const> right;
+};
+
+using OrderTreePtr = std::shared_ptr<OrderTreeNode const>;
+
+// Weight-balance parameters (Adams). delta bounds the size ratio between
+// siblings; gamma chooses single vs double rotation.
+inline constexpr std::uint32_t kOtDelta = 3;
+inline constexpr std::uint32_t kOtGamma = 2;
+
+[[nodiscard]] inline std::uint32_t
+otSize(OrderTreePtr const& t) noexcept
+{
+    return t ? t->size : 0;
+}
+
+// -1 / 0 / +1 ordering on (dirRoot, insertSeq).
+[[nodiscard]] inline int
+otCmp(
+    uint256 const& aDir,
+    std::uint64_t aSeq,
+    uint256 const& bDir,
+    std::uint64_t bSeq) noexcept
+{
+    if (aDir < bDir)
+        return -1;
+    if (bDir < aDir)
+        return 1;
+    if (aSeq < bSeq)
+        return -1;
+    if (bSeq < aSeq)
+        return 1;
+    return 0;
+}
+
+[[nodiscard]] inline OrderTreePtr
+otNode(
+    uint256 const& dir,
+    std::uint64_t seq,
+    uint256 const& off,
+    OrderTreePtr l,
+    OrderTreePtr r)
+{
+    auto n = std::make_shared<OrderTreeNode>();
+    n->dirRoot = dir;
+    n->insertSeq = seq;
+    n->offerKey = off;
+    n->left = std::move(l);
+    n->right = std::move(r);
+    n->size = otSize(n->left) + otSize(n->right) + 1;
+    return n;
+}
+
+// Rebalance a node whose subtrees may violate the weight balance by one step.
+[[nodiscard]] inline OrderTreePtr
+otBalance(
+    uint256 const& dir,
+    std::uint64_t seq,
+    uint256 const& off,
+    OrderTreePtr const& l,
+    OrderTreePtr const& r)
+{
+    auto const ln = otSize(l);
+    auto const rn = otSize(r);
+
+    if (ln + rn <= 1)
+        return otNode(dir, seq, off, l, r);
+
+    if (rn > kOtDelta * ln)
+    {
+        // Right-heavy.
+        auto const& rl = r->left;
+        auto const& rr = r->right;
+        if (otSize(rl) < kOtGamma * otSize(rr))
+            // single left rotation
+            return otNode(
+                r->dirRoot,
+                r->insertSeq,
+                r->offerKey,
+                otNode(dir, seq, off, l, rl),
+                rr);
+        // double left rotation
+        return otNode(
+            rl->dirRoot,
+            rl->insertSeq,
+            rl->offerKey,
+            otNode(dir, seq, off, l, rl->left),
+            otNode(r->dirRoot, r->insertSeq, r->offerKey, rl->right, rr));
+    }
+
+    if (ln > kOtDelta * rn)
+    {
+        // Left-heavy.
+        auto const& ll = l->left;
+        auto const& lr = l->right;
+        if (otSize(lr) < kOtGamma * otSize(ll))
+            // single right rotation
+            return otNode(
+                l->dirRoot,
+                l->insertSeq,
+                l->offerKey,
+                ll,
+                otNode(dir, seq, off, lr, r));
+        // double right rotation
+        return otNode(
+            lr->dirRoot,
+            lr->insertSeq,
+            lr->offerKey,
+            otNode(l->dirRoot, l->insertSeq, l->offerKey, ll, lr->left),
+            otNode(dir, seq, off, lr->right, r));
+    }
+
+    return otNode(dir, seq, off, l, r);
+}
+
+[[nodiscard]] inline OrderTreePtr
+otInsert(OrderTreePtr const& t, uint256 const& dir, std::uint64_t seq, uint256 const& off)
+{
+    if (!t)
+        return otNode(dir, seq, off, nullptr, nullptr);
+    int const c = otCmp(dir, seq, t->dirRoot, t->insertSeq);
+    if (c < 0)
+        return otBalance(
+            t->dirRoot, t->insertSeq, t->offerKey, otInsert(t->left, dir, seq, off), t->right);
+    if (c > 0)
+        return otBalance(
+            t->dirRoot, t->insertSeq, t->offerKey, t->left, otInsert(t->right, dir, seq, off));
+    // Equal key: replace payload (keys are unique in practice; never hit).
+    return otNode(t->dirRoot, t->insertSeq, off, t->left, t->right);
+}
+
+// Remove the minimum node of a non-null tree; write its fields into `outMin`.
+[[nodiscard]] inline OrderTreePtr
+otDeleteMin(OrderTreePtr const& t, OrderTreeNode& outMin)
+{
+    if (!t->left)
+    {
+        outMin = *t;
+        return t->right;
+    }
+    return otBalance(
+        t->dirRoot, t->insertSeq, t->offerKey, otDeleteMin(t->left, outMin), t->right);
+}
+
+// Join two subtrees (all keys in l < all keys in r) by promoting r's minimum.
+[[nodiscard]] inline OrderTreePtr
+otGlue(OrderTreePtr const& l, OrderTreePtr const& r)
+{
+    if (!l)
+        return r;
+    if (!r)
+        return l;
+    OrderTreeNode minN;
+    auto const r2 = otDeleteMin(r, minN);
+    return otBalance(minN.dirRoot, minN.insertSeq, minN.offerKey, l, r2);
+}
+
+[[nodiscard]] inline OrderTreePtr
+otDelete(OrderTreePtr const& t, uint256 const& dir, std::uint64_t seq)
+{
+    if (!t)
+        return nullptr;
+    int const c = otCmp(dir, seq, t->dirRoot, t->insertSeq);
+    if (c < 0)
+        return otBalance(
+            t->dirRoot, t->insertSeq, t->offerKey, otDelete(t->left, dir, seq), t->right);
+    if (c > 0)
+        return otBalance(
+            t->dirRoot, t->insertSeq, t->offerKey, t->left, otDelete(t->right, dir, seq));
+    return otGlue(t->left, t->right);
+}
+
+// In-order traversal: appends offer keys best-quality-first, append order
+// within a level.
+inline void
+otInorder(OrderTreePtr const& t, std::vector<uint256>& out)
+{
+    if (!t)
+        return;
+    otInorder(t->left, out);
+    out.push_back(t->offerKey);
+    otInorder(t->right, out);
+}
+
+// Leftmost (best) offer key.
+[[nodiscard]] inline std::optional<uint256>
+otFirst(OrderTreePtr t)
+{
+    if (!t)
+        return std::nullopt;
+    while (t->left)
+        t = t->left;
+    return t->offerKey;
+}
+
+// Find the insertSeq for (dirRoot, offerKey). All nodes sharing a dirRoot form
+// a contiguous in-order range that may straddle a node's two children, so when
+// dirRoot matches we must check the node and both subtrees. O(level-size) worst
+// case; effectively O(log n) for front deletions (crossing consumes front-first
+// and the target is then the level's leftmost remaining node).
+[[nodiscard]] inline std::optional<std::uint64_t>
+otFindSeq(OrderTreePtr const& t, uint256 const& dir, uint256 const& off)
+{
+    if (!t)
+        return std::nullopt;
+    if (dir < t->dirRoot)
+        return otFindSeq(t->left, dir, off);
+    if (t->dirRoot < dir)
+        return otFindSeq(t->right, dir, off);
+    if (t->offerKey == off)
+        return t->insertSeq;
+    if (auto const l = otFindSeq(t->left, dir, off))
+        return l;
+    return otFindSeq(t->right, dir, off);
+}
+
+}  // namespace xrpl::detail

include/xrpl/shamap/SHAMap.h

@@ -178,31 +178,6 @@ public:
     SHAMapHash
     getHash() const;
 
-    /** Recompute dirty node hashes in parallel and return the root hash.
-
-        Plan 7 Phase 2. Equivalent in output to `getHash()` on a freshly
-        mutated map: it recomputes the hash of every node dirtied since the
-        last hash settle, bottom-up, then returns the root hash. The work is
-        fanned out by top-level subtree — the root's up-to-16 children are
-        independent (a dirty node under one branch is never shared with
-        another), so their subtree recomputations run concurrently with no
-        synchronization, and only the root hash is computed serially after.
-
-        Byte-identical to the serial path by construction: a node's hash is a
-        pure function of its children's hashes, so computation order is
-        irrelevant as long as children precede parents — which the bottom-up
-        walk guarantees.
-
-        Unlike `getHash()`/`unshare()`, this does NOT convert nodes to shared
-        (cowid stays as-is) or flush to the nodestore; it only refreshes hash
-        caches. A subsequent `getHash()` therefore returns immediately.
-
-        @param workers Maximum concurrent subtree recomputations. <= 1 runs
-                       serially. Defaults to the hardware concurrency.
-    */
-    SHAMapHash
-    updateHashesParallel(int workers = 0);
-
     // save a copy if you have a temporary anyway
     bool
     updateGiveItem(SHAMapNodeType type, boost::intrusive_ptr<SHAMapItem const> item);

include/xrpl/tx/paths/BookTip.h

@@ -4,6 +4,10 @@
 #include <xrpl/protocol/Indexes.h>
 #include <xrpl/protocol/Quality.h>
 
+#include <cstddef>
+#include <cstdint>
+#include <vector>
+
 namespace xrpl {
 
 class Logs;
@@ -17,6 +21,7 @@ class BookTip
 private:
     ApplyView& view_;
     bool valid_{false};
+    Book originalBook_;
     uint256 book_;
     uint256 end_;
     uint256 dir_;
@@ -24,6 +29,15 @@ private:
     std::shared_ptr<SLE> entry_;
     Quality quality_{};
 
+    // Plan 9: when the order-book index supplies an ordered offer-key snapshot
+    // for this book, iterate it instead of re-walking the SHAMap with succ()
+    // per offer. `useCursor_` is decided on the first step; thereafter the two
+    // paths are mutually exclusive for the iterator's lifetime.
+    std::vector<uint256> cursor_;
+    std::size_t cursorPos_{0};
+    bool useCursor_{false};
+    std::uint64_t lastCursorQuality_{0};
+
 public:
     /** Create the iterator. */
     BookTip(ApplyView& view, Book const& book);

src/libxrpl/ledger/ApplyViewBase.cpp

@@ -64,6 +64,75 @@ ApplyViewBase::read(Keylet const& k) const
     return items_.read(*base_, k);
 }
 
+std::optional<uint256>
+ApplyViewBase::topOfBookFirstPage(Book const& book) const
+{
+    // Reads inside a sandbox that has already mutated `book` cannot use
+    // the parent's cache: the parent's view of the top doesn't reflect
+    // our buffered changes yet. Fall back to succ() in that case.
+    if (dirtyBooks_.find(book) != dirtyBooks_.end())
+        return std::nullopt;
+    return base_->topOfBookFirstPage(book);
+}
+
+void
+ApplyViewBase::recordTopOfBook(Book const& book, uint256 const& firstPageKey) const
+{
+    // Don't populate the parent cache from inside a dirty sandbox view —
+    // our succ() result may reflect uncommitted mutations from the parent's
+    // perspective.
+    if (dirtyBooks_.find(book) != dirtyBooks_.end())
+        return;
+    base_->recordTopOfBook(book, firstPageKey);
+}
+
+void
+ApplyViewBase::notifyOfferInserted(Book const& book, uint256 const& dirKey, uint256 const& offerKey)
+    const
+{
+    dirtyBooks_.insert(book);
+    pendingTopOfBookNotifications_.emplace_back(book, dirKey, offerKey, /*isDelete=*/false);
+}
+
+void
+ApplyViewBase::notifyOfferDeleted(Book const& book, uint256 const& dirKey, uint256 const& offerKey)
+    const
+{
+    dirtyBooks_.insert(book);
+    pendingTopOfBookNotifications_.emplace_back(book, dirKey, offerKey, /*isDelete=*/true);
+}
+
+std::optional<std::vector<uint256>>
+ApplyViewBase::orderedBook(Book const& book) const
+{
+    // Unlike topOfBookFirstPage, do NOT skip dirty books: the cursor is taken
+    // once and iterated locally, and it self-heals any offer this sandbox has
+    // buffered-deleted via peek()-null-skip in BookTip. So always delegate to
+    // the (immutable-for-this-crossing) base index.
+    return base_->orderedBook(book);
+}
+
+void
+ApplyViewBase::flushTopOfBookNotifications() const
+{
+    for (auto const& note : pendingTopOfBookNotifications_)
+    {
+        if (note.isDelete)
+            base_->notifyOfferDeleted(note.book, note.dirKey, note.offerKey);
+        else
+            base_->notifyOfferInserted(note.book, note.dirKey, note.offerKey);
+    }
+    pendingTopOfBookNotifications_.clear();
+    dirtyBooks_.clear();
+}
+
+void
+ApplyViewBase::discardTopOfBookNotifications() const noexcept
+{
+    pendingTopOfBookNotifications_.clear();
+    dirtyBooks_.clear();
+}
+
 auto
 ApplyViewBase::slesBegin() const -> std::unique_ptr<SlesType::iter_base>
 {

src/libxrpl/ledger/ApplyViewImpl.cpp

@@ -31,7 +31,9 @@ ApplyViewImpl::apply(
     bool isDryRun,
     beast::Journal j)
 {
-    return items_.apply(to, tx, ter, deliver_, parentBatchId, isDryRun, j);
+    auto meta = items_.apply(to, tx, ter, deliver_, parentBatchId, isDryRun, j);
+    flushTopOfBookNotifications();
+    return meta;
 }
 
 std::size_t

src/libxrpl/ledger/DeferredRebuild.cpp

@@ -1,329 +0,0 @@
-#include <xrpl/ledger/DeferredRebuild.h>
-
-#include <xrpl/basics/Slice.h>
-#include <xrpl/basics/hardened_hash.h>
-#include <xrpl/protocol/HashPrefix.h>
-#include <xrpl/protocol/digest.h>
-
-#include <algorithm>
-#include <array>
-#include <cstddef>
-#include <cstdint>
-#include <cstring>
-#include <future>
-#include <mutex>
-#include <unordered_set>
-#include <utility>
-#include <vector>
-
-namespace xrpl {
-
-namespace {
-
-// Zero the trailing (64 - depth) nibbles of `key`, keeping the first
-// `depth` nibbles. Returns the resulting prefix uint256.
-//
-// SHAMap stores keys big-endian. The most significant nibble of the
-// key is the depth-1 branch from root. So at depth N we want to keep
-// the top N nibbles = N*4 bits and zero the rest.
-[[nodiscard]] uint256
-prefixAtDepth(uint256 const& key, int depth) noexcept
-{
-    if (depth >= 64)
-        return key;
-    if (depth <= 0)
-        return uint256{};
-
-    uint256 result = key;
-    auto const totalNibbles = 64;
-    auto const nibblesToZero = totalNibbles - depth;
-
-    // Zero `nibblesToZero` nibbles starting from the least-significant
-    // end. uint256 has 32 bytes (each holds two nibbles, high then low).
-    // Byte index 31 holds the two lowest nibbles; byte 0 holds the two
-    // highest.
-    int nibblesZeroed = 0;
-    for (int byteIdx = uint256::kBytes - 1;
-         byteIdx >= 0 && nibblesZeroed < nibblesToZero;
-         --byteIdx)
-    {
-        if (nibblesZeroed + 2 <= nibblesToZero)
-        {
-            // Zero both nibbles in this byte
-            result.data()[byteIdx] = 0;
-            nibblesZeroed += 2;
-        }
-        else
-        {
-            // Zero only the low nibble in this byte
-            result.data()[byteIdx] &= 0xF0;
-            nibblesZeroed += 1;
-        }
-    }
-    return result;
-}
-
-}  // namespace
-
-namespace {
-
-// Longest common prefix in NIBBLES between two uint256 keys.
-// Returns 0..64. 64 means the keys are identical.
-[[nodiscard]] int
-lcpNibbles(uint256 const& a, uint256 const& b) noexcept
-{
-    for (int byteIdx = 0; byteIdx < uint256::kBytes; ++byteIdx)
-    {
-        if (a.data()[byteIdx] != b.data()[byteIdx])
-        {
-            // They agree in 2*byteIdx whole nibbles. Check whether
-            // the high nibble of this byte also agrees.
-            std::uint8_t aHigh = a.data()[byteIdx] >> 4;
-            std::uint8_t bHigh = b.data()[byteIdx] >> 4;
-            if (aHigh == bHigh)
-                return 2 * byteIdx + 1;
-            return 2 * byteIdx;
-        }
-    }
-    return 64;
-}
-
-}  // namespace
-
-std::vector<AffectedNode>
-planDeferredRebuild(std::vector<uint256> const& modifiedKeys)
-{
-    if (modifiedKeys.empty())
-        return {};
-
-    // LCP-based dedup. For sorted keys, the inner-node ancestors are
-    // grouped by shared prefix; once a key has contributed its
-    // ancestors at depths 0..63, the NEXT sorted key only adds NEW
-    // ancestors at depths > LCP(prev, current). Everything at depth
-    // ≤ LCP is already in the plan from `prev`.
-    //
-    // This replaces O(K × 64) prefix computations with O(K log K)
-    // sort + ~O(K) for sequential-key workloads (where LCP ≈ 63).
-    std::vector<uint256> sorted = modifiedKeys;
-    std::sort(sorted.begin(), sorted.end());
-
-    std::vector<AffectedNode> plan;
-    // Upper bound for arbitrary inputs is K * 64, but realistic
-    // workloads (sequential / clustered) produce ~K entries.
-    plan.reserve(sorted.size() * 2);
-
-    // First key contributes ancestors at every depth.
-    for (int d = 0; d <= 63; ++d)
-        plan.push_back({d, prefixAtDepth(sorted[0], d)});
-
-    // Subsequent keys contribute only ancestors at depths > LCP with
-    // their predecessor.
-    for (std::size_t i = 1; i < sorted.size(); ++i)
-    {
-        // Identical adjacent keys: nothing new to contribute.
-        if (sorted[i] == sorted[i - 1])
-            continue;
-        int const lcp = lcpNibbles(sorted[i - 1], sorted[i]);
-        for (int d = lcp + 1; d <= 63; ++d)
-            plan.push_back({d, prefixAtDepth(sorted[i], d)});
-    }
-
-    // Sort depth-descending so a bottom-up rebuild can iterate in
-    // order. Within a depth, ascending prefix for determinism.
-    std::sort(
-        plan.begin(),
-        plan.end(),
-        [](AffectedNode const& a, AffectedNode const& b) {
-            if (a.depth != b.depth)
-                return a.depth > b.depth;
-            return a.prefix < b.prefix;
-        });
-
-    return plan;
-}
-
-namespace {
-
-// Given a parent inner node at (parentDepth, parentPrefix), compute the
-// child prefix at the given branch (0..15). The child is at depth
-// (parentDepth + 1); its prefix sets the nibble at position parentDepth
-// to the branch value.
-[[nodiscard]] uint256
-childPrefixOf(uint256 const& parentPrefix, int parentDepth, std::uint8_t branch)
-{
-    uint256 result = parentPrefix;
-    int const byteIdx = parentDepth / 2;
-    bool const isHighNibble = (parentDepth % 2) == 0;
-    if (isHighNibble)
-        result.data()[byteIdx] =
-            (result.data()[byteIdx] & 0x0F) |
-            static_cast<std::uint8_t>((branch & 0x0F) << 4);
-    else
-        result.data()[byteIdx] =
-            (result.data()[byteIdx] & 0xF0) |
-            static_cast<std::uint8_t>(branch & 0x0F);
-    return result;
-}
-
-}  // namespace
-
-namespace detail {
-
-RebuildResult
-executeRebuildPlanImpl(
-    std::vector<AffectedNode> const& plan,
-    std::function<uint256(int, uint256 const&)> getOriginalChildHash)
-{
-    RebuildResult result;
-    result.reserve(plan.size());
-
-    // Plan is depth-descending; deepest nodes processed first. Their
-    // hashes are visible to shallower nodes in this same walk.
-    for (auto const& node : plan)
-    {
-        std::array<uint256, 16> children;
-        for (std::uint8_t b = 0; b < 16; ++b)
-        {
-            AffectedNode const childPos{
-                node.depth + 1, childPrefixOf(node.prefix, node.depth, b)};
-            auto it = result.find(childPos);
-            if (it != result.end())
-                children[b] = it->second;
-            else
-                children[b] = getOriginalChildHash(childPos.depth, childPos.prefix);
-        }
-        result.emplace(node, computeInnerNodeHash(children));
-    }
-
-    return result;
-}
-
-}  // namespace detail
-
-std::array<std::vector<uint256>, 16>
-partitionByFirstNibble(std::vector<uint256> const& modifiedKeys)
-{
-    std::array<std::vector<uint256>, 16> buckets;
-    for (auto const& key : modifiedKeys)
-    {
-        // First nibble = high nibble of byte 0
-        std::uint8_t const firstNibble = (key.data()[0] >> 4) & 0x0F;
-        buckets[firstNibble].push_back(key);
-    }
-    return buckets;
-}
-
-namespace detail {
-
-uint256
-deferredRebuildRootParallelImpl(
-    std::vector<uint256> const& modifiedKeys,
-    std::function<uint256(int, uint256 const&)> getOriginalChildHash)
-{
-    if (modifiedKeys.empty())
-        return getOriginalChildHash(0, uint256{});
-
-    auto const buckets = partitionByFirstNibble(modifiedKeys);
-
-    // For each subtree, compute the new depth-1 hash (if any keys
-    // changed in that subtree) in parallel. Empty subtrees fall back
-    // to the parent's depth-1 hash, which is read from the callback.
-    //
-    // Each future captures the relevant bucket and runs an independent
-    // plan-and-execute over just that subtree's keys. The result is
-    // the new hash of the depth-1 inner node rooting that subtree
-    // (which corresponds to the root's branch-b child).
-    std::array<std::future<uint256>, 16> futures;
-    for (std::uint8_t b = 0; b < 16; ++b)
-    {
-        if (buckets[b].empty())
-            continue;
-
-        futures[b] = std::async(
-            std::launch::async,
-            [bucket = buckets[b], &getOriginalChildHash, b]() -> uint256 {
-                // Plan + execute for this subtree's keys. The depth-1
-                // node is the rooting node; we want its new hash.
-                auto const plan = planDeferredRebuild(bucket);
-                auto const result =
-                    executeRebuildPlan(plan, getOriginalChildHash);
-
-                // The depth-1 prefix for this subtree has its first
-                // nibble set to b, rest zero.
-                uint256 prefix{};
-                prefix.data()[0] =
-                    static_cast<std::uint8_t>(b << 4);
-
-                AffectedNode const subtreeRoot{1, prefix};
-                auto const it = result.find(subtreeRoot);
-                if (it == result.end())
-                    return uint256{};
-                return it->second;
-            });
-    }
-
-    // Gather: 16 child hashes for the root.
-    std::array<uint256, 16> rootChildren;
-    for (std::uint8_t b = 0; b < 16; ++b)
-    {
-        if (buckets[b].empty())
-        {
-            // Untouched subtree — read original depth-1 hash from
-            // parent SHAMap via callback.
-            uint256 prefix{};
-            prefix.data()[0] = static_cast<std::uint8_t>(b << 4);
-            rootChildren[b] = getOriginalChildHash(1, prefix);
-        }
-        else
-        {
-            rootChildren[b] = futures[b].get();
-        }
-    }
-
-    return computeInnerNodeHash(rootChildren);
-}
-
-}  // namespace detail
-
-uint256
-computeInnerNodeHash(std::array<uint256, 16> const& childHashes)
-{
-    // SHAMapInnerNode::updateHash short-circuits to a zero hash when
-    // every branch is empty (isBranch_ == 0). Match that convention —
-    // a "node with no children" hashes to zero, not to SHA-512 of a
-    // zero-filled buffer.
-    bool anyNonZero = false;
-    for (auto const& h : childHashes)
-    {
-        if (h.isNonZero())
-        {
-            anyNonZero = true;
-            break;
-        }
-    }
-    if (!anyNonZero)
-        return uint256{};
-
-    // Layout: 4-byte big-endian HashPrefix::InnerNode || 16 × 32-byte
-    // child hashes. Total: 516 bytes. Matches SHAMapInnerNode::updateHash
-    // byte-for-byte; differential-tested against it.
-    constexpr std::size_t kBufSize = 4 + 16 * uint256::kBytes;
-    alignas(64) std::array<std::uint8_t, kBufSize> buf{};
-
-    auto const prefix = static_cast<std::uint32_t>(HashPrefix::InnerNode);
-    buf[0] = static_cast<std::uint8_t>(prefix >> 24);
-    buf[1] = static_cast<std::uint8_t>(prefix >> 16);
-    buf[2] = static_cast<std::uint8_t>(prefix >> 8);
-    buf[3] = static_cast<std::uint8_t>(prefix);
-
-    std::uint8_t* out = buf.data() + 4;
-    for (auto const& h : childHashes)
-    {
-        std::memcpy(out, h.data(), uint256::kBytes);
-        out += uint256::kBytes;
-    }
-
-    return sha512Half(Slice{buf.data(), buf.size()});
-}
-
-}  // namespace xrpl

src/libxrpl/ledger/FlatStateMap.cpp

@@ -1,130 +0,0 @@
-#include <xrpl/ledger/FlatStateMap.h>
-
-#include <xrpl/ledger/Ledger.h>
-#include <xrpl/ledger/ReadView.h>
-
-#include <memory>
-#include <mutex>
-#include <shared_mutex>
-#include <utility>
-
-namespace xrpl {
-
-FlatStateMap::value_type
-FlatStateMap::read(key_type const& key) const
-{
-    std::shared_lock<std::shared_mutex> lock(mutex_);
-    auto const it = map_.find(key);
-    if (it == map_.end())
-        return nullptr;
-    return it->second;
-}
-
-bool
-FlatStateMap::exists(key_type const& key) const
-{
-    std::shared_lock<std::shared_mutex> lock(mutex_);
-    return map_.find(key) != map_.end();
-}
-
-void
-FlatStateMap::insert(key_type const& key, value_type sle)
-{
-    std::unique_lock<std::shared_mutex> lock(mutex_);
-    map_.insert_or_assign(key, std::move(sle));
-}
-
-void
-FlatStateMap::erase(key_type const& key)
-{
-    std::unique_lock<std::shared_mutex> lock(mutex_);
-    map_.erase(key);
-}
-
-std::size_t
-FlatStateMap::size() const
-{
-    std::shared_lock<std::shared_mutex> lock(mutex_);
-    return map_.size();
-}
-
-bool
-FlatStateMap::empty() const
-{
-    std::shared_lock<std::shared_mutex> lock(mutex_);
-    return map_.empty();
-}
-
-void
-FlatStateMap::clear()
-{
-    std::unique_lock<std::shared_mutex> lock(mutex_);
-    map_.clear();
-}
-
-std::unique_ptr<FlatStateMap>
-FlatStateMap::snapshot() const
-{
-    auto out = std::make_unique<FlatStateMap>();
-    std::shared_lock<std::shared_mutex> lock(mutex_);
-    // Reserve to avoid rehash during the bulk copy.
-    out->map_.reserve(map_.size());
-    for (auto const& entry : map_)
-        out->map_.insert(entry);
-    return out;
-}
-
-void
-populateFromReadView(FlatStateMap& target, ReadView const& source)
-{
-    populateFromRange(target, source.sles);
-}
-
-void
-attachFlatStateMapTo(Ledger& ledger)
-{
-    auto map = std::make_shared<FlatStateMap>();
-    populateFromReadView(*map, ledger);
-    ledger.setFlatStateMap(std::move(map));
-}
-
-void
-mirrorRawInsert(FlatStateMap& map, std::shared_ptr<STLedgerEntry const> sle)
-{
-    auto const key = sle->key();
-    map.insert(key, std::move(sle));
-}
-
-void
-mirrorRawReplace(FlatStateMap& map, std::shared_ptr<STLedgerEntry const> sle)
-{
-    auto const key = sle->key();
-    map.insert(key, std::move(sle));  // insert == insert_or_assign here
-}
-
-void
-mirrorRawErase(
-    FlatStateMap& map,
-    std::shared_ptr<STLedgerEntry const> const& sle)
-{
-    map.erase(sle->key());
-}
-
-void
-mirrorRawErase(FlatStateMap& map, uint256 const& key)
-{
-    map.erase(key);
-}
-
-std::shared_ptr<STLedgerEntry const>
-readFromFlatStateMap(FlatStateMap const& map, Keylet const& k)
-{
-    auto sle = map.read(k.key);
-    if (!sle)
-        return nullptr;
-    if (!k.check(*sle))
-        return nullptr;
-    return sle;
-}
-
-}  // namespace xrpl

src/libxrpl/ledger/Ledger.cpp

@@ -9,7 +9,6 @@
 #include <xrpl/beast/utility/Journal.h>
 #include <xrpl/beast/utility/Zero.h>
 #include <xrpl/beast/utility/instrumentation.h>
-#include <xrpl/ledger/FlatStateMap.h>
 #include <xrpl/ledger/LedgerTiming.h>
 #include <xrpl/ledger/ReadView.h>
 #include <xrpl/nodestore/NodeObject.h>
@@ -277,19 +276,6 @@ Ledger::Ledger(Ledger const& prevLedger, NetClock::time_point closeTime)
     {
         header_.closeTime = prevLedger.header_.closeTime + header_.closeTimeResolution;
     }
-
-    // Plan 6 lifecycle: if the parent carries a flat-state mirror, the child
-    // inherits an independent deep-copy snapshot of it. This mirrors the COW
-    // snapshot of `stateMap_` above: the child starts from the parent's final
-    // state and then diverges as the round's transactions mirror into it,
-    // while the parent's map is left untouched. When no map is attached (the
-    // default), this is a no-op and the ledger behaves exactly as before.
-    //
-    // The snapshot is O(N) in entry count today; a persistent/HAMT structure
-    // (see FlatStateMap.h) is the follow-on that makes per-ledger propagation
-    // cheap enough to enable in production.
-    if (prevLedger.flatStateMap_)
-        flatStateMap_ = prevLedger.flatStateMap_->snapshot();
 }
 
 Ledger::Ledger(LedgerHeader const& info, Rules rules, Family& family)
@@ -425,14 +411,6 @@ Ledger::read(Keylet const& k) const
         return nullptr;
         // LCOV_EXCL_STOP
     }
-
-    // Plan 6 P6.4: when a FlatStateMap is attached, it is the read
-    // source of truth. No SHAMap fallback — drift between the two
-    // is caught at close by the differential invariant check (P6.5),
-    // not by silently re-reading from SHAMap.
-    if (flatStateMap_)
-        return readFromFlatStateMap(*flatStateMap_, k);
-
     auto const& item = stateMap_.peekItem(k.key);
     if (!item)
         return nullptr;
@@ -512,8 +490,6 @@ Ledger::rawErase(std::shared_ptr<SLE> const& sle)
 {
     if (!stateMap_.delItem(sle->key()))
         logicError("Ledger::rawErase: key not found");
-    if (flatStateMap_)
-        mirrorRawErase(*flatStateMap_, sle);
 }
 
 void
@@ -521,8 +497,6 @@ Ledger::rawErase(uint256 const& key)
 {
     if (!stateMap_.delItem(key))
         logicError("Ledger::rawErase: key not found");
-    if (flatStateMap_)
-        mirrorRawErase(*flatStateMap_, key);
 }
 
 void
@@ -535,8 +509,6 @@ Ledger::rawInsert(std::shared_ptr<SLE> const& sle)
     {
         logicError("Ledger::rawInsert: key already exists");
     }
-    if (flatStateMap_)
-        mirrorRawInsert(*flatStateMap_, sle);
 }
 
 void
@@ -549,28 +521,6 @@ Ledger::rawReplace(std::shared_ptr<SLE> const& sle)
     {
         logicError("Ledger::rawReplace: key not found");
     }
-    if (flatStateMap_)
-        mirrorRawReplace(*flatStateMap_, sle);
-}
-
-void
-Ledger::setFlatStateMap(std::shared_ptr<FlatStateMap> map)
-{
-    flatStateMap_ = std::move(map);
-}
-
-std::shared_ptr<FlatStateMap>
-Ledger::flatStateMap() const
-{
-    return flatStateMap_;
-}
-
-bool
-Ledger::validateFlatStateMapMatchesShaMap() const
-{
-    if (!flatStateMap_)
-        return true;  // nothing to validate against
-    return flatStateMapMatchesShaMap(*flatStateMap_, stateMap_);
 }
 
 void

src/libxrpl/ledger/OpenView.cpp

@@ -86,7 +86,12 @@ OpenView::OpenView(OpenView const& rhs)
     , base_{rhs.base_}
     , items_{rhs.items_}
     , hold_{rhs.hold_}
-    , open_{rhs.open_} {};
+    , open_{rhs.open_}
+    // Plan 9 P9.6: carry the persistent order-book index forward on the
+    // open-ledger COW (modify() copies the OpenView per tx). clone() is O(#books)
+    // shared_ptr copies sharing all offer nodes, so the index stays warm across
+    // transactions instead of cold-starting and rebuilding per tx.
+    , orderBookIndex_{rhs.orderBookIndex_.clone()} {};
 
 OpenView::OpenView(OpenLedgerT, ReadView const* base, Rules rules, std::shared_ptr<void const> hold)
     : monotonicResource_{
@@ -169,6 +174,71 @@ OpenView::read(Keylet const& k) const
     return items_.read(*base_, k);
 }
 
+std::optional<uint256>
+OpenView::topOfBookFirstPage(Book const& book) const
+{
+    if (!TopOfBookCache::enabled())
+        return std::nullopt;
+    if (auto const entry = topOfBookCache_.get(book))
+        return entry->firstPageKey;
+    return std::nullopt;
+}
+
+void
+OpenView::recordTopOfBook(Book const& book, uint256 const& firstPageKey) const
+{
+    if (!TopOfBookCache::enabled())
+        return;
+    topOfBookCache_.record(book, firstPageKey, header_.seq);
+}
+
+void
+OpenView::notifyOfferInserted(Book const& book, uint256 const& dirKey, uint256 const& offerKey)
+    const
+{
+    // Maintain only books already in the index: a book enters the index only
+    // via rebuildBook (which captures the full authoritative state), so it is
+    // always complete. Inserting into an absent book would create a PARTIAL
+    // entry (missing pre-existing offers) that a later crossing would trust —
+    // wrong. Absent books are populated completely on first read (orderedBook's
+    // rebuild-on-absent). This mirrors TopOfBookCache::onOfferInsert's no-op.
+    if (OrderBookIndex::enabled() && orderBookIndex_.contains(book))
+        orderBookIndex_.insertOffer(book, dirKey, offerKey);
+    if (!TopOfBookCache::enabled())
+        return;
+    topOfBookCache_.onOfferInsert(book, dirKey, header_.seq);
+}
+
+void
+OpenView::notifyOfferDeleted(Book const& book, uint256 const& dirKey, uint256 const& offerKey)
+    const
+{
+    if (OrderBookIndex::enabled())
+        orderBookIndex_.deleteOffer(book, dirKey, offerKey);
+    if (!TopOfBookCache::enabled())
+        return;
+    topOfBookCache_.onOfferDelete(book, dirKey);
+}
+
+std::optional<std::vector<uint256>>
+OpenView::orderedBook(Book const& book) const
+{
+    if (!OrderBookIndex::enabled())
+        return std::nullopt;
+
+    // Guarantee completeness: if the index has no entry for `book`, populate it
+    // from the authoritative state before serving the cursor. A maintained,
+    // already-present book skips this (the steady-state fast path). The index
+    // never holds a partial book, so the cursor can't under-include.
+    if (!orderBookIndex_.contains(book))
+        orderBookIndex_.rebuildBook(*this, book);
+
+    auto offers = orderBookIndex_.flatten(book);
+    if (offers.empty())
+        return std::nullopt;  // genuinely empty book — let succ() find nothing
+    return offers;
+}
+
 auto
 OpenView::slesBegin() const -> std::unique_ptr<SlesType::iter_base>
 {

src/libxrpl/ledger/OrderBookIndex.cpp

@@ -0,0 +1,199 @@
+#include <xrpl/ledger/OrderBookIndex.h>
+
+#include <xrpl/ledger/ReadView.h>
+#include <xrpl/ledger/helpers/DirectoryHelpers.h>
+#include <xrpl/protocol/Indexes.h>
+#include <xrpl/protocol/STLedgerEntry.h>
+
+#include <atomic>
+
+namespace xrpl {
+
+namespace {
+
+// Operator-facing kill switch. Defaults to true; set false via setEnabled()
+// to bypass the index entirely and fall back to baseline succ() iteration
+// without a restart (mirrors TopOfBookCache).
+std::atomic<bool> gEnabled{true};
+
+}  // namespace
+
+bool
+OrderBookIndex::enabled() noexcept
+{
+    return gEnabled.load(std::memory_order_relaxed);
+}
+
+void
+OrderBookIndex::setEnabled(bool on) noexcept
+{
+    gEnabled.store(on, std::memory_order_relaxed);
+}
+
+OrderBookIndex::OrderBookIndex(OrderBookIndex&& other)
+{
+    std::unique_lock lock(other.mutex_);
+    books_ = std::move(other.books_);
+}
+
+OrderBookIndex
+OrderBookIndex::clone() const
+{
+    OrderBookIndex out;
+    std::shared_lock lock(mutex_);
+    // Copying the map copies each BookState — a shared_ptr root (O(1), shares
+    // all offer nodes) + the counter. Total O(#books).
+    out.books_ = books_;
+    return out;
+}
+
+void
+OrderBookIndex::insertOffer(Book const& book, uint256 const& dirRoot, uint256 const& offerKey)
+{
+    std::unique_lock lock(mutex_);
+    auto& st = books_[book];
+    st.root = detail::otInsert(st.root, dirRoot, st.nextSeq++, offerKey);
+    inserts_.fetch_add(1, std::memory_order_relaxed);
+}
+
+void
+OrderBookIndex::deleteOffer(Book const& book, uint256 const& dirRoot, uint256 const& offerKey)
+{
+    std::unique_lock lock(mutex_);
+    auto const it = books_.find(book);
+    if (it == books_.end())
+        return;
+    auto const seq = detail::otFindSeq(it->second.root, dirRoot, offerKey);
+    if (!seq)
+        return;
+    it->second.root = detail::otDelete(it->second.root, dirRoot, *seq);
+    deletes_.fetch_add(1, std::memory_order_relaxed);
+    if (!it->second.root)
+        books_.erase(it);
+}
+
+std::vector<uint256>
+OrderBookIndex::flatten(Book const& book) const
+{
+    std::vector<uint256> out;
+    std::shared_lock lock(mutex_);
+    auto const it = books_.find(book);
+    if (it != books_.end())
+        detail::otInorder(it->second.root, out);
+    return out;
+}
+
+std::optional<uint256>
+OrderBookIndex::firstOffer(Book const& book) const
+{
+    std::shared_lock lock(mutex_);
+    auto const it = books_.find(book);
+    if (it == books_.end())
+        return std::nullopt;
+    return detail::otFirst(it->second.root);
+}
+
+std::vector<std::pair<uint256, uint256>>
+OrderBookIndex::walkBook(ReadView const& view, Book const& book)
+{
+    // Canonical quality-ordered enumeration, mirroring NetworkOPs::getBookPage
+    // and BookTip: succ() over directory roots in [bookBase, bookEnd), then
+    // cdirFirst/cdirNext across each root's pages. uTip advances to the found
+    // root, so the next succ() yields the next-worse quality; a root's overflow
+    // pages live outside [bookBase, bookEnd) and are reached only via sfIndexNext
+    // inside cdirNext, never by succ().
+    std::vector<std::pair<uint256, uint256>> out;
+    uint256 const bookBase = getBookBase(book);
+    uint256 const bookEnd = getQualityNext(bookBase);
+    uint256 uTip = bookBase;
+
+    for (;;)
+    {
+        auto const next = view.succ(uTip, bookEnd);
+        if (!next)
+            break;
+        uint256 const dirRoot = *next;
+
+        std::shared_ptr<SLE const> page;
+        unsigned int index = 0;
+        uint256 offerKey;
+        if (cdirFirst(view, dirRoot, page, index, offerKey))
+        {
+            do
+            {
+                out.emplace_back(dirRoot, offerKey);
+            } while (cdirNext(view, dirRoot, page, index, offerKey));
+        }
+        uTip = dirRoot;
+    }
+    return out;
+}
+
+void
+OrderBookIndex::rebuildBook(ReadView const& view, Book const& book)
+{
+    auto const walked = walkBook(view, book);
+
+    BookState st;
+    // Inserting in walk order assigns ascending insertSeq, so in-order traversal
+    // reproduces the walk exactly.
+    for (auto const& [dirRoot, offerKey] : walked)
+        st.root = detail::otInsert(st.root, dirRoot, st.nextSeq++, offerKey);
+
+    std::unique_lock lock(mutex_);
+    if (!st.root)
+        books_.erase(book);
+    else
+        books_[book] = std::move(st);
+    rebuilds_.fetch_add(1, std::memory_order_relaxed);
+}
+
+bool
+OrderBookIndex::validateMatchesShaMap(ReadView const& view, Book const& book) const
+{
+    std::vector<uint256> fresh;
+    for (auto const& [dirRoot, offerKey] : walkBook(view, book))
+        fresh.push_back(offerKey);
+
+    return fresh == flatten(book);
+}
+
+bool
+OrderBookIndex::contains(Book const& book) const
+{
+    std::shared_lock lock(mutex_);
+    return books_.find(book) != books_.end();
+}
+
+void
+OrderBookIndex::eraseBook(Book const& book)
+{
+    std::unique_lock lock(mutex_);
+    books_.erase(book);
+}
+
+void
+OrderBookIndex::clear()
+{
+    std::unique_lock lock(mutex_);
+    books_.clear();
+}
+
+std::size_t
+OrderBookIndex::bookCount() const
+{
+    std::shared_lock lock(mutex_);
+    return books_.size();
+}
+
+std::size_t
+OrderBookIndex::offerCount(Book const& book) const
+{
+    std::shared_lock lock(mutex_);
+    auto const it = books_.find(book);
+    if (it == books_.end())
+        return 0;
+    return detail::otSize(it->second.root);
+}
+
+}  // namespace xrpl

src/libxrpl/ledger/PaymentSandbox.cpp

@@ -453,6 +453,7 @@ PaymentSandbox::apply(RawView& to)
 {
     XRPL_ASSERT(!ps_, "xrpl::PaymentSandbox::apply : non-null sandbox");
     items_.apply(to);
+    flushTopOfBookNotifications();
 }
 
 void
@@ -461,6 +462,7 @@ PaymentSandbox::apply(PaymentSandbox& to)
     XRPL_ASSERT(ps_ == &to, "xrpl::PaymentSandbox::apply : matching sandbox");
     items_.apply(to);
     tab_.apply(to.tab_);
+    flushTopOfBookNotifications();
 }
 
 XRPAmount

src/libxrpl/ledger/TopOfBookCache.cpp

@@ -0,0 +1,123 @@
+#include <xrpl/ledger/TopOfBookCache.h>
+
+#include <xrpl/protocol/Indexes.h>
+
+#include <atomic>
+#include <mutex>
+
+namespace xrpl {
+
+namespace {
+
+// Operator-facing kill switch. Defaults to true; set false via setEnabled()
+// to bypass the cache entirely (e.g. in case a workload exposes a bug, the
+// node can be brought back to baseline succ() behavior without restart).
+std::atomic<bool> gEnabled{true};
+
+}  // namespace
+
+bool
+TopOfBookCache::enabled() noexcept
+{
+    return gEnabled.load(std::memory_order_relaxed);
+}
+
+void
+TopOfBookCache::setEnabled(bool on) noexcept
+{
+    gEnabled.store(on, std::memory_order_relaxed);
+}
+
+TopOfBookCache::TopOfBookCache(TopOfBookCache const& other)
+{
+    std::lock_guard lock(other.mutex_);
+    map_ = other.map_;
+}
+
+TopOfBookCache::TopOfBookCache(TopOfBookCache&& other)
+{
+    std::lock_guard lock(other.mutex_);
+    map_ = std::move(other.map_);
+}
+
+std::optional<TopOfBookEntry>
+TopOfBookCache::get(Book const& book) const
+{
+    std::lock_guard lock(mutex_);
+    auto it = map_.find(book);
+    if (it == map_.end())
+    {
+        misses_.fetch_add(1, std::memory_order_relaxed);
+        return std::nullopt;
+    }
+    hits_.fetch_add(1, std::memory_order_relaxed);
+    return it->second;
+}
+
+void
+TopOfBookCache::record(Book const& book, uint256 const& firstPageKey, LedgerIndex seq)
+{
+    std::lock_guard lock(mutex_);
+    auto& entry = map_[book];
+    entry.firstPageKey = firstPageKey;
+    entry.bestQuality = getQuality(firstPageKey);
+    entry.asOfLedger = seq;
+}
+
+void
+TopOfBookCache::onOfferInsert(Book const& book, uint256 const& dirKey, LedgerIndex seq)
+{
+    std::lock_guard lock(mutex_);
+    auto it = map_.find(book);
+    if (it == map_.end())
+    {
+        // No cached top — defer to the next reader, which populates lazily.
+        return;
+    }
+    // Lower keylet == better quality (pages share the book prefix, quality
+    // bits are encoded in the low bytes).
+    if (dirKey < it->second.firstPageKey)
+    {
+        it->second.firstPageKey = dirKey;
+        it->second.bestQuality = getQuality(dirKey);
+        it->second.asOfLedger = seq;
+    }
+}
+
+void
+TopOfBookCache::onOfferDelete(Book const& book, uint256 const& dirKey)
+{
+    std::lock_guard lock(mutex_);
+    auto it = map_.find(book);
+    if (it == map_.end())
+        return;
+    if (it->second.firstPageKey == dirKey)
+    {
+        map_.erase(it);
+        invalidations_.fetch_add(1, std::memory_order_relaxed);
+    }
+}
+
+void
+TopOfBookCache::invalidate(Book const& book)
+{
+    std::lock_guard lock(mutex_);
+    if (map_.erase(book) != 0)
+        invalidations_.fetch_add(1, std::memory_order_relaxed);
+}
+
+void
+TopOfBookCache::clear()
+{
+    std::lock_guard lock(mutex_);
+    map_.clear();
+}
+
+std::size_t
+TopOfBookCache::size() const
+{
+    std::lock_guard lock(mutex_);
+    return map_.size();
+}
+
+}  // namespace xrpl

src/libxrpl/ledger/helpers/OfferHelpers.cpp

@@ -5,17 +5,46 @@
 #include <xrpl/beast/utility/instrumentation.h>
 #include <xrpl/ledger/ApplyView.h>
 #include <xrpl/ledger/helpers/AccountRootHelpers.h>
+#include <xrpl/protocol/Book.h>
 #include <xrpl/protocol/Indexes.h>
 #include <xrpl/protocol/LedgerFormats.h>
 #include <xrpl/protocol/SField.h>
+#include <xrpl/protocol/STAmount.h>
 #include <xrpl/protocol/STArray.h>  // IWYU pragma: keep
 #include <xrpl/protocol/STLedgerEntry.h>
 #include <xrpl/protocol/TER.h>
 
 #include <memory>
+#include <optional>
 
 namespace xrpl {
 
+namespace {
+
+// Reconstruct the Book this offer was placed on. The primary directory uses
+// the offer's sfDomainID (if any); the open-book directory of a hybrid offer
+// is the same in/out assets with no domain.
+Book
+primaryBookFromOffer(SLE const& sle)
+{
+    auto const takerPays = sle.getFieldAmount(sfTakerPays);
+    auto const takerGets = sle.getFieldAmount(sfTakerGets);
+    std::optional<uint256> domain;
+    if (sle.isFieldPresent(sfDomainID))
+        domain = sle.getFieldH256(sfDomainID);
+    return Book{takerPays.asset(), takerGets.asset(), domain};
+}
+
+Book
+openBookFromOffer(SLE const& sle)
+{
+    auto const takerPays = sle.getFieldAmount(sfTakerPays);
+    auto const takerGets = sle.getFieldAmount(sfTakerGets);
+    return Book{takerPays.asset(), takerGets.asset(), std::nullopt};
+}
+
+}  // namespace
+
 TER
 offerDelete(ApplyView& view, std::shared_ptr<SLE> const& sle, beast::Journal j)
 {
@@ -37,6 +66,12 @@ offerDelete(ApplyView& view, std::shared_ptr<SLE> const& sle, beast::Journal j)
         return tefBAD_LEDGER;  // LCOV_EXCL_LINE
     }
 
+    // Plan 8: notify the top-of-book cache that the primary book lost an
+    // offer at `uDirectory`. If this was the cached top page the cache
+    // invalidates; otherwise no-op. uDirectory is the first-page keylet of
+    // the offer's quality bucket — i.e. exactly what the cache stores.
+    view.notifyOfferDeleted(primaryBookFromOffer(*sle), uDirectory, offerIndex);
+
     if (sle->isFieldPresent(sfAdditionalBooks))
     {
         XRPL_ASSERT(
@@ -54,6 +89,10 @@ offerDelete(ApplyView& view, std::shared_ptr<SLE> const& sle, beast::Journal j)
             {
                 return tefBAD_LEDGER;  // LCOV_EXCL_LINE
             }
+
+            // Hybrid offers also live on the open (no-domain) book — notify
+            // that cache too.
+            view.notifyOfferDeleted(openBookFromOffer(*sle), dirIndex, offerIndex);
         }
     }
 

src/libxrpl/shamap/SHAMap.cpp

@@ -26,17 +26,13 @@
 
 #include <boost/smart_ptr/intrusive_ptr.hpp>
 
-#include <atomic>
-#include <cstddef>
 #include <cstdint>
 #include <exception>
 #include <functional>
-#include <future>
 #include <memory>
 #include <stack>
 #include <stdexcept>
 #include <string>
-#include <thread>
 #include <tuple>
 #include <type_traits>
 #include <utility>
@@ -855,98 +851,6 @@ SHAMap::getHash() const
     return hash;
 }
 
-namespace {
-
-// Recompute hashes bottom-up for the subtree rooted at `node`, descending
-// only into dirty (cowid != 0) resident children — the hash side of
-// walkSubTree, without flushing/sharing. `node` must itself be dirty.
-//
-// Thread-safety: dirty nodes are uniquely owned by the current cowid, and the
-// subtrees hanging off distinct branches are disjoint, so two callers handed
-// children of different branches never touch the same node. Reads of clean
-// (cowid 0) children's cached hashes via updateHashDeep are read-only and safe
-// to race.
-void
-recomputeSubtreeHashes(SHAMapTreeNode* node)
-{
-    if (node->isLeaf())
-    {
-        node->updateHash();
-        return;
-    }
-
-    auto* inner = safeDowncast<SHAMapInnerNode*>(node);
-    for (int branch = 0; branch < SHAMapInnerNode::kBranchFactor; ++branch)
-    {
-        if (inner->isEmptyBranch(branch))
-            continue;
-        auto* child = inner->getChildPointer(branch);
-        if (child && (child->cowid() != 0))
-            recomputeSubtreeHashes(child);
-    }
-    inner->updateHashDeep();
-}
-
-}  // namespace
-
-SHAMapHash
-SHAMap::updateHashesParallel(int workers)
-{
-    // Nothing dirtied since the last settle: the cached root hash is current.
-    // (root_ is always present, matching getHash()'s invariant.)
-    if (root_->cowid() == 0)
-        return root_->getHash();
-
-    if (root_->isLeaf())
-    {
-        root_->updateHash();
-        return root_->getHash();
-    }
-
-    auto* rootInner = safeDowncast<SHAMapInnerNode*>(root_.get());
-
-    // Gather the root's dirty, resident top-level subtrees. These are
-    // independent and can be recomputed concurrently.
-    std::vector<SHAMapTreeNode*> subtrees;
-    for (int branch = 0; branch < kBranchFactor; ++branch)
-    {
-        if (rootInner->isEmptyBranch(branch))
-            continue;
-        auto* child = rootInner->getChildPointer(branch);
-        if (child && (child->cowid() != 0))
-            subtrees.push_back(child);
-    }
-
-    if (workers <= 0)
-        workers = static_cast<int>(std::thread::hardware_concurrency());
-
-    if (workers <= 1 || subtrees.size() <= 1)
-    {
-        for (auto* s : subtrees)
-            recomputeSubtreeHashes(s);
-    }
-    else
-    {
-        int const nthreads = std::min<int>(workers, static_cast<int>(subtrees.size()));
-        std::atomic<std::size_t> next{0};
-        std::vector<std::future<void>> tasks;
-        tasks.reserve(nthreads);
-        for (int t = 0; t < nthreads; ++t)
-        {
-            tasks.push_back(std::async(std::launch::async, [&subtrees, &next] {
-                for (std::size_t i = next++; i < subtrees.size(); i = next++)
-                    recomputeSubtreeHashes(subtrees[i]);
-            }));
-        }
-        for (auto& task : tasks)
-            task.get();
-    }
-
-    // All top-level subtree hashes are now current; finish at the root.
-    rootInner->updateHashDeep();
-    return rootInner->getHash();
-}
-
 bool
 SHAMap::updateGiveItem(SHAMapNodeType type, boost::intrusive_ptr<SHAMapItem const> item)
 {

src/libxrpl/tx/paths/BookTip.cpp

@@ -1,25 +1,32 @@
 #include <xrpl/tx/paths/BookTip.h>
 
 #include <xrpl/beast/utility/Journal.h>
+#include <xrpl/beast/utility/instrumentation.h>
 #include <xrpl/ledger/ApplyView.h>
+#include <xrpl/ledger/OrderBookIndex.h>
+#include <xrpl/ledger/TopOfBookCache.h>
 #include <xrpl/ledger/helpers/DirectoryHelpers.h>
 #include <xrpl/ledger/helpers/OfferHelpers.h>
 #include <xrpl/protocol/Book.h>
 #include <xrpl/protocol/Indexes.h>
+#include <xrpl/protocol/SField.h>
 #include <xrpl/protocol/STLedgerEntry.h>
 
 #include <memory>
+#include <optional>
 
 namespace xrpl {
 
 BookTip::BookTip(ApplyView& view, Book const& book)
-    : view_(view), book_(getBookBase(book)), end_(getQualityNext(book_))
+    : view_(view), originalBook_(book), book_(getBookBase(book)), end_(getQualityNext(book_))
 {
 }
 
 bool
 BookTip::step(beast::Journal j)
 {
+    bool const firstStep = !valid_;
+
     if (valid_)
     {
         if (entry_)
@@ -29,12 +36,91 @@ BookTip::step(beast::Journal j)
         }
     }
 
+    // Plan 9: on the first step, ask the order-book index for an ordered
+    // snapshot of this book's offers. A returned vector is guaranteed complete
+    // (the index rebuilds from authoritative state on a miss), so iterating it
+    // is equivalent to the succ() walk — but O(1) per offer instead of an
+    // O(log N) trie re-walk. nullopt ⇒ no index ⇒ the succ() path below.
+    if (firstStep && OrderBookIndex::enabled())
+    {
+        if (auto snap = view_.orderedBook(originalBook_))
+        {
+            cursor_ = std::move(*snap);
+            cursorPos_ = 0;
+            useCursor_ = true;
+        }
+    }
+
+    if (useCursor_)
+    {
+        for (;;)
+        {
+            if (cursorPos_ >= cursor_.size())
+                return false;
+
+            uint256 const offerKey = cursor_[cursorPos_++];
+            auto sle = view_.peek(keylet::offer(offerKey));
+            if (!sle)
+                // The snapshot is from the (immutable) base index; this offer
+                // was buffered-deleted by the sandbox (e.g. a pre-crossing
+                // cancel). Skip it — the succ() walk wouldn't see it either.
+                continue;
+
+            index_ = offerKey;
+            dir_ = sle->getFieldH256(sfBookDirectory);
+            quality_ = Quality(getQuality(dir_));
+            entry_ = std::move(sle);
+            valid_ = true;
+
+            // Cursor order must be best-quality-first, exactly like succ().
+            XRPL_ASSERT(
+                getQuality(dir_) >= lastCursorQuality_,
+                "xrpl::BookTip::step : order-book cursor yields non-decreasing quality");
+            lastCursorQuality_ = getQuality(dir_);
+
+            return true;
+        }
+    }
+
+    bool firstIter = firstStep;
     for (;;)
     {
         // See if there's an entry at or worse than current quality. Notice
         // that the quality is encoded only in the index of the first page
         // of a directory.
-        auto const firstPage = view_.succ(book_, end_);
+        std::optional<uint256> firstPage;
+        bool fromCache = false;
+
+        if (firstIter && TopOfBookCache::enabled())
+        {
+            if (auto const cached = view_.topOfBookFirstPage(originalBook_))
+            {
+                firstPage = *cached;
+                fromCache = true;
+            }
+        }
+
+        if (!firstPage)
+        {
+            firstPage = view_.succ(book_, end_);
+            if (firstIter && firstPage && TopOfBookCache::enabled())
+                view_.recordTopOfBook(originalBook_, *firstPage);
+        }
+
+#ifndef NDEBUG
+        // Differential gate (Plan 8 P8.7): in debug builds every cache hit
+        // is shadow-verified against a fresh successor walk. Divergence
+        // here is a bug in the invalidation logic, not a fallback.
+        if (fromCache && firstPage)
+        {
+            auto const verified = view_.succ(book_, end_);
+            XRPL_ASSERT(
+                verified == firstPage,
+                "BookTip::step : top-of-book cache hit diverges from succ()");
+        }
+#endif
+
+        firstIter = false;
 
         if (!firstPage)
             return false;
@@ -60,6 +146,8 @@ BookTip::step(beast::Journal j)
 
         // There should never be an empty directory but just in case,
         // we handle that case by advancing to the next directory.
+        // Also covers the case where a stale cache hit returned a
+        // page that no longer has any indexes.
         book_ = *firstPage;
     }
 

src/libxrpl/tx/paths/OfferStream.cpp

@@ -95,6 +95,12 @@ TOfferStreamBase<TIn, TOut>::erase(ApplyView& view)
     p->setFieldV256(sfIndexes, v);
     view.update(p);
 
+    // Plan 9: this stale-entry cleanup strips sfIndexes directly (not via
+    // dirRemove, which would be a protocol-breaking change), so it bypasses
+    // the usual offerDelete notification. Notify the order-book index here so
+    // it doesn't retain a phantom key. Auxiliary only — no ledger-state change.
+    view.notifyOfferDeleted(book_, tip_.dir(), tip_.index());
+
     JLOG(j_.trace()) << "Missing offer " << tip_.index() << " removed from directory "
                      << tip_.dir();
 }

src/libxrpl/tx/transactors/dex/OfferCreate.cpp

@@ -593,6 +593,11 @@ OfferCreate::applyHybrid(
     if (!bookExists)
         ctx_.registry.get().getOrderBookDB().addOrderBook(book);
 
+    // Plan 8: notify the top-of-book cache that the open book just got a
+    // new offer at `dir.key`. The cache updates its top only if this is
+    // at-or-better than the current cached top; otherwise no-op.
+    sb.notifyOfferInserted(book, dir.key, offerKey.key);
+
     sleOffer->setFieldArray(sfAdditionalBooks, bookArr);
     return tesSUCCESS;
 }
@@ -915,6 +920,11 @@ OfferCreate::applyGuts(Sandbox& sb, Sandbox& sbCancel)
         // LCOV_EXCL_STOP
     }
 
+    // Plan 8: notify the top-of-book cache that `book` got a new offer at
+    // `dir.key`. The cache updates its top only if this is at-or-better
+    // than the current cached top; otherwise no-op.
+    sb.notifyOfferInserted(book, dir.key, offerIndex.key);
+
     auto sleOffer = std::make_shared<SLE>(offerIndex);
     sleOffer->setAccountID(sfAccount, accountID_);
     sleOffer->setFieldU32(sfSequence, offerSequence);

src/test/app/OrderBookCrossing_test.cpp

@@ -0,0 +1,135 @@
+#include <test/jtx/Account.h>
+#include <test/jtx/Env.h>
+#include <test/jtx/amount.h>
+#include <test/jtx/offer.h>
+#include <test/jtx/pay.h>
+#include <test/jtx/trust.h>
+
+#include <xrpl/beast/unit_test/suite.h>
+#include <xrpl/json/json_value.h>
+#include <xrpl/ledger/OrderBookIndex.h>
+#include <xrpl/protocol/UintTypes.h>
+#include <xrpl/protocol/jss.h>
+
+#include <vector>
+
+namespace xrpl::test {
+
+/** Bit-exactness gate for the Plan 9 order-book index seam: a scripted
+    crossing scenario must produce an identical sequence of ledger hashes with
+    the index enabled (BookTip iterates the in-memory cursor) and disabled
+    (BookTip walks the SHAMap with succ()). Any divergence in the cursor's order
+    or contents changes consumed offers/amounts and therefore the ledger hash. */
+class OrderBookCrossing_test : public beast::unit_test::Suite
+{
+    // Run a deterministic crossing scenario and return the ledger hash after
+    // every close. The scenario exercises the cursor-specific paths:
+    // multi-quality books, a multi-offer (shared-quality) level, an unfunded
+    // offer, partial fills, and a pre-crossing cancel (peek-null-skip).
+    std::vector<uint256>
+    runScenario()
+    {
+        using namespace jtx;
+        Env env{*this};
+        std::vector<uint256> hashes;
+        // accountHash is the consensus state root — it reflects every crossing
+        // effect (consumed offers, balances, directories). If the cursor and
+        // succ() paths diverge at all, this differs.
+        auto snap = [&] { hashes.push_back(env.closed()->header().accountHash); };
+
+        auto const gw = Account{"gw"};
+        auto const USD = gw["USD"];
+        Account const alice{"alice"};  // maker, spread of qualities
+        Account const bob{"bob"};      // maker, shared-quality level
+        Account const carol{"carol"};  // maker, becomes unfunded
+        Account const dave{"dave"};    // taker
+
+        env.fund(XRP(10'000'000), gw, alice, bob, carol, dave);
+        env.close();
+        snap();
+        env.trust(USD(100'000'000), alice, bob, carol, dave);
+        env.close();
+        env(pay(gw, alice, USD(1'000'000)));
+        env(pay(gw, bob, USD(1'000'000)));
+        env(pay(gw, carol, USD(1'000'000)));
+        env.close();
+        snap();
+
+        // alice: 8 distinct qualities. bob: 4 offers at one shared quality
+        // (a multi-entry level). carol: one offer she will defund.
+        for (int i = 0; i < 8; ++i)
+            env(offer(alice, XRP(500 + i), USD(100)));
+        for (int i = 0; i < 4; ++i)
+            env(offer(bob, XRP(503), USD(100)));
+        env(offer(carol, XRP(501), USD(100)));
+        env.close();
+        snap();
+
+        // Defund carol: move her USD away so her resting offer is unfunded at
+        // cross time (exercises the unfunded-skip path through the cursor).
+        env(pay(carol, gw, USD(1'000'000)));
+        env.close();
+        snap();
+
+        // dave places an offer, then cancels it via an OfferCreate carrying
+        // OfferSequence (pre-crossing delete → cursor peek-null-skip path).
+        auto const daveOfferSeq = env.seq(dave);
+        env(offer(dave, USD(100), XRP(2'000)));  // far from market: rests
+        env.close();
+        snap();
+
+        // dave crosses: partial and full fills across alice/bob/carol levels.
+        env(offer(dave, USD(250), XRP(1'255)));
+        env.close();
+        snap();
+        env(offer(dave, USD(500), XRP(2'520)));
+        env.close();
+        snap();
+
+        // A crossing OfferCreate that also cancels dave's resting offer.
+        auto cross = offer(dave, USD(100), XRP(505));
+        cross[jss::OfferSequence] = daveOfferSeq;
+        env(cross);
+        env.close();
+        snap();
+
+        return hashes;
+    }
+
+    void
+    testIndexMatchesBaseline()
+    {
+        testcase("ledger hashes identical with order-book index on vs off");
+
+        OrderBookIndex::setEnabled(false);
+        auto const baseline = runScenario();
+
+        OrderBookIndex::setEnabled(true);
+        auto const withIndex = runScenario();
+
+        OrderBookIndex::setEnabled(true);  // restore default
+
+        BEAST_EXPECT(baseline.size() == withIndex.size());
+        bool identical = baseline.size() == withIndex.size();
+        for (std::size_t i = 0; i < baseline.size() && i < withIndex.size(); ++i)
+        {
+            if (baseline[i] != withIndex[i])
+            {
+                identical = false;
+                log << "  ledger-hash divergence at close " << i << "\n";
+            }
+        }
+        BEAST_EXPECT(identical);
+    }
+
+public:
+    void
+    run() override
+    {
+        testIndexMatchesBaseline();
+    }
+};
+
+BEAST_DEFINE_TESTSUITE(OrderBookCrossing, app, xrpl);
+
+}  // namespace xrpl::test

src/test/app/TopOfBookCache_bench.cpp

@@ -0,0 +1,512 @@
+#include <test/jtx/Account.h>
+#include <test/jtx/Env.h>
+#include <test/jtx/amount.h>
+#include <test/jtx/fee.h>
+#include <test/jtx/offer.h>
+#include <test/jtx/pay.h>
+#include <test/jtx/seq.h>
+#include <test/jtx/trust.h>
+
+#include <xrpl/beast/unit_test/suite.h>
+#include <xrpl/ledger/ApplyView.h>
+#include <xrpl/ledger/ApplyViewImpl.h>
+#include <xrpl/ledger/OpenView.h>
+#include <xrpl/ledger/TopOfBookCache.h>
+#include <xrpl/protocol/Book.h>
+#include <xrpl/protocol/Issue.h>
+#include <xrpl/tx/apply.h>
+#include <xrpl/tx/paths/BookTip.h>
+
+#include <algorithm>
+#include <chrono>
+#include <cstdint>
+#include <cstdlib>
+#include <vector>
+
+namespace xrpl::test {
+
+/** Level 1 micro-benchmark for the top-of-book cache.
+
+    A/Bs the cache via its runtime kill switch (TopOfBookCache::setEnabled) over
+    a deep order book, entirely in-process (no network, no synced ledger).
+
+    Two arms:
+
+    - readArm  (isolated): drives BookTip's first-step top-of-book read against
+      an OpenView this benchmark constructs and OWNS, so cache counters are
+      reliable (the open-ledger apply path copies the OpenView per tx and the
+      copy ctor resets counters, so counters read off env.current() are not).
+      This isolates the optimized primitive (succ() walk -> hash probe). It is a
+      BEST-CASE, hot-entry read number — not an end-to-end throughput figure.
+
+    - e2eArm   (end-to-end): times a batch of real crossing OfferCreates through
+      the full Env apply path (real offer consumption => invalidation/repopulate
+      churn). Timing only; also captures the cache's own overhead (the OpenView
+      copy on every modify()). Answers "does the isolated saving show up at all,
+      net of overhead". Realistic hit-rate under MainNet-like mixed load is a
+      later, heavier exercise (Level 1.5 / Level 3), not measured here.
+
+    Registered MANUAL — never runs in normal CI. Invoke explicitly:
+        rippled --unittest=TopOfBookCacheBench
+*/
+class TopOfBookCacheBench_test : public beast::unit_test::Suite
+{
+    using clock = std::chrono::steady_clock;
+
+    // Median of repeated samples — robust to scheduler noise.
+    static double
+    median(std::vector<double> v)
+    {
+        std::sort(v.begin(), v.end());
+        return v.empty() ? 0.0 : v[v.size() / 2];
+    }
+
+    struct ReadArm
+    {
+        double nsPerRead{0};
+        std::uint64_t hits{0};
+        std::uint64_t misses{0};
+        std::uint64_t invalidations{0};
+        bool foundTop{false};
+    };
+
+    // Build a deep order book (XRP <-> USD), close it into the LCL, and return
+    // the Book those offers populate. `pages` distinct qualities => `pages`
+    // directory pages.
+    static Book
+    buildDeepBook(jtx::Env& env, jtx::Account const& gw, int pages)
+    {
+        using namespace jtx;
+        auto const USD = gw["USD"];
+        Account const maker{"maker"};
+
+        env.fund(XRP(1'000'000), gw, maker);
+        env.close();
+        env.trust(USD(10'000'000), maker);
+        env.close();
+        env(pay(gw, maker, USD(1'000'000)));
+        env.close();
+
+        // Each offer: maker receives takerPays (XRP), gives takerGets (USD).
+        // Distinct takerPays => distinct quality => distinct directory page.
+        for (int i = 0; i < pages; ++i)
+            env(offer(maker, XRP(500 + i), USD(100)));
+        env.close();
+
+        // Book{in = takerPays.asset(), out = takerGets.asset()} — see
+        // OfferCreate.cpp:570.
+        return Book{xrpIssue(), USD.issue(), std::nullopt};
+    }
+
+    // Isolated read-path arm. Owns the OpenView so counters are trustworthy.
+    ReadArm
+    runReadArm(jtx::Env& env, Book const& book, bool cacheEnabled, std::size_t reads)
+    {
+        TopOfBookCache::setEnabled(cacheEnabled);
+
+        // Fresh owned view per arm => clean counters (no reset API otherwise).
+        OpenView ov(kOpenLedger, env.closed()->rules(), env.closed());
+
+        // One read = fresh BookTip + a single step() = one top-of-book probe.
+        // BookTip's first step is read-only (it deletes only from the 2nd step
+        // on), so a single ApplyView can be reused across reads.
+        ApplyViewImpl av(&ov, TapNone);
+
+        ReadArm r;
+        {
+            BookTip bt(av, book);
+            r.foundTop = bt.step(env.journal) && bt.entry() != nullptr;
+        }
+
+        auto const once = [&] {
+            for (std::size_t i = 0; i < reads; ++i)
+            {
+                BookTip bt(av, book);
+                bt.step(env.journal);
+            }
+        };
+
+        once();  // warmup (also populates the cache in the enabled arm)
+
+        std::vector<double> samples;
+        for (int rep = 0; rep < 5; ++rep)
+        {
+            auto const t0 = clock::now();
+            once();
+            auto const t1 = clock::now();
+            samples.push_back(
+                static_cast<double>(
+                    std::chrono::duration_cast<std::chrono::nanoseconds>(t1 - t0).count()) /
+                static_cast<double>(reads));
+        }
+
+        r.nsPerRead = median(std::move(samples));
+        r.hits = ov.topOfBookCache().hits();
+        r.misses = ov.topOfBookCache().misses();
+        r.invalidations = ov.topOfBookCache().invalidations();
+        return r;
+    }
+
+    void
+    testReadPath()
+    {
+        testcase("Arm 1: isolated top-of-book read (owned OpenView)");
+        using namespace jtx;
+
+        // Isolate the TopOfBookCache read path: the order-book index, when on,
+        // supersedes the cache in BookTip (cursor instead of cache+succ), so it
+        // must be off for this arm to measure the cache.
+        OrderBookIndex::setEnabled(false);
+
+        int const pages = 64;
+        std::size_t const reads = 200'000;
+
+        Env env{*this};
+        auto const book = buildDeepBook(env, Account{"gw"}, pages);
+
+        auto const off = runReadArm(env, book, /*cacheEnabled=*/false, reads);
+        auto const on = runReadArm(env, book, /*cacheEnabled=*/true, reads);
+        TopOfBookCache::setEnabled(true);  // restore default
+
+        BEAST_EXPECT(off.foundTop);
+        BEAST_EXPECT(on.foundTop);
+        // Disabled arm never consults the cache.
+        BEAST_EXPECT(off.hits == 0 && off.misses == 0);
+        // Enabled arm: 1 cold miss, the rest hits.
+        BEAST_EXPECT(on.hits > 0);
+        BEAST_EXPECT(on.misses >= 1);
+        BEAST_EXPECT(on.invalidations == 0);
+
+        double const speedup = on.nsPerRead > 0 ? off.nsPerRead / on.nsPerRead : 0.0;
+
+#ifndef NDEBUG
+        log << "\n*** DEBUG build: BookTip's differential gate shadow-verifies "
+               "every cache hit with an extra succ() walk, so the cache-ON path "
+               "does MORE work here. Arm 1 timing is only meaningful in a "
+               "Release (NDEBUG) build; counters below are valid regardless. ***\n";
+#endif
+
+        log << "\n=== Arm 1: isolated read (best-case, hot entry) ===\n"
+            << "  book pages           : " << pages << "\n"
+            << "  reads / sample       : " << reads << "\n"
+            << "  cache OFF  ns/read   : " << off.nsPerRead << "\n"
+            << "  cache ON   ns/read   : " << on.nsPerRead << "\n"
+            << "  speedup              : " << speedup << "x\n"
+            << "  cache ON  hits/miss  : " << on.hits << " / " << on.misses << "\n"
+            << std::endl;
+
+        OrderBookIndex::setEnabled(true);  // restore default
+    }
+
+    // End-to-end arm: time real crossing offers through the full apply path.
+    double
+    runE2EArm(bool cacheEnabled, int pages, int crossings)
+    {
+        using namespace jtx;
+        TopOfBookCache::setEnabled(cacheEnabled);
+
+        Env env{*this};
+        auto const gw = Account{"gw"};
+        auto const USD = gw["USD"];
+        buildDeepBook(env, gw, pages);
+
+        // Taker buys USD with XRP, crossing the maker's resting offers.
+        Account const taker{"taker"};
+        env.fund(XRP(1'000'000), taker);
+        env.close();
+        env.trust(USD(10'000'000), taker);
+        env.close();
+
+        auto const t0 = clock::now();
+        for (int i = 0; i < crossings; ++i)
+        {
+            env(offer(taker, USD(100), XRP(500 + (i % pages))));
+            if ((i % 10) == 9)
+                env.close();
+        }
+        env.close();
+        auto const t1 = clock::now();
+
+        return static_cast<double>(
+                   std::chrono::duration_cast<std::chrono::microseconds>(t1 - t0).count()) /
+            crossings;
+    }
+
+    void
+    testEndToEnd()
+    {
+        testcase("Arm 2: end-to-end crossing throughput (timing only)");
+
+        int const pages = 64;
+        int const crossings = 300;
+
+        double const off = runE2EArm(/*cacheEnabled=*/false, pages, crossings);
+        double const on = runE2EArm(/*cacheEnabled=*/true, pages, crossings);
+        TopOfBookCache::setEnabled(true);  // restore default
+
+        BEAST_EXPECT(off > 0 && on > 0);
+
+        log << "\n=== Arm 2: end-to-end crossing (full apply path, real churn) ===\n"
+            << "  book pages           : " << pages << "\n"
+            << "  crossing offers      : " << crossings << "\n"
+            << "  cache OFF  us/cross  : " << off << "\n"
+            << "  cache ON   us/cross  : " << on << "\n"
+            << "  delta                : " << (off - on) << " us/cross"
+            << " (note: dominated by tx machinery + cache copy overhead)\n"
+            << std::endl;
+    }
+
+    // Profiling arm: measure PURE crossing-apply cost with NO ledger close.
+    // env.close() runs full consensus close (flushDirty hashing + SQLite ledger
+    // writes); Arm 2 closed every 10 offers, contaminating its per-crossing
+    // number. Here we pre-sign crossing OfferCreates and replay them against a
+    // fresh owned OpenView per rep (each rep starts with the full book), timing
+    // only xrpl::apply (preflight + preclaim + doApply). Long enough total work
+    // to attach `sample`/Instruments to the running process.
+    void
+    testCrossingApplyProfile()
+    {
+        testcase("Arm 3: pure crossing-apply cost (no ledger close)");
+        using namespace jtx;
+
+        int const pages = 64;
+        int const crossPerRep = 50;  // crossings applied per fresh book
+        // BENCH_PROFILE=1 cranks reps so the apply loop runs ~30s for `sample`.
+        bool const profiling = std::getenv("BENCH_PROFILE") != nullptr;
+        int const reps = profiling ? 5000 : 400;
+
+        Env env{*this};
+        auto const gw = Account{"gw"};
+        auto const USD = gw["USD"];
+        buildDeepBook(env, gw, pages);
+
+        Account const taker{"taker"};
+        env.fund(XRP(10'000'000), taker);
+        env.close();
+        env.trust(USD(100'000'000), taker);
+        env.close();
+
+        // Pre-sign the crossing OfferCreates once, with explicit increasing
+        // sequences starting at taker's current seq. Each fresh accum resets
+        // taker to that same seq, so the identical signed set replays cleanly.
+        std::uint32_t const startSeq = env.seq(taker);
+        std::vector<std::shared_ptr<STTx const>> txns;
+        txns.reserve(crossPerRep);
+        for (int i = 0; i < crossPerRep; ++i)
+        {
+            auto jtx = env.jt(
+                offer(taker, USD(100), XRP(500 + (i % pages))),
+                Seq(startSeq + i),
+                Fee(100));
+            txns.push_back(jtx.stx);
+        }
+
+        auto const base = env.current();  // open view over the closed book
+
+        std::size_t applied = 0, crossed = 0;
+        std::vector<double> samples;
+        for (int rep = 0; rep < reps; ++rep)
+        {
+            OpenView accum(kOpenLedger, base->rules(), base);
+            auto const t0 = clock::now();
+            for (auto const& tx : txns)
+            {
+                auto const r = apply(env.app(), accum, *tx, TapNone, env.journal);
+                if (rep == 0)
+                {
+                    ++applied;
+                    if (r.applied && isTesSuccess(r.ter))
+                        ++crossed;
+                }
+            }
+            auto const t1 = clock::now();
+            samples.push_back(
+                static_cast<double>(
+                    std::chrono::duration_cast<std::chrono::nanoseconds>(t1 - t0).count()) /
+                crossPerRep / 1000.0);  // us/crossing
+        }
+
+        BEAST_EXPECT(applied == static_cast<std::size_t>(crossPerRep));
+        BEAST_EXPECT(crossed > 0);
+
+        log << "\n=== Arm 3: pure crossing-apply (no ledger close) ===\n"
+            << "  book pages              : " << pages << "\n"
+            << "  crossings / rep         : " << crossPerRep << "\n"
+            << "  reps                    : " << reps << "\n"
+            << "  tesSUCCESS (rep 0)      : " << crossed << " / " << applied << "\n"
+            << "  median us / crossing    : " << median(samples) << "\n"
+            << "  (compare to Arm 2's ~780us which INCLUDED ledger close)\n"
+            << std::endl;
+    }
+
+    // Arm 4 (Plan 9 headline): pure crossing-apply with the order-book index
+    // ON vs OFF. OFF = baseline succ()-per-offer walk; ON = BookTip iterates the
+    // in-memory cursor (index pre-seeded per rep, untimed, modelling the
+    // maintained steady state). Same owned-OpenView / no-ledger-close method as
+    // Arm 3, so the delta isolates the succ() cost the cursor removes.
+    void
+    testCrossingIndexArm()
+    {
+        testcase("Arm 4: crossing-apply, order-book index ON vs OFF");
+        using namespace jtx;
+
+        int const pages = 64;
+        int const crossPerRep = 50;
+        int const reps = 400;
+
+        Env env{*this};
+        auto const gw = Account{"gw"};
+        auto const USD = gw["USD"];
+        auto const book = buildDeepBook(env, gw, pages);
+
+        Account const taker{"taker"};
+        env.fund(XRP(10'000'000), taker);
+        env.close();
+        env.trust(USD(100'000'000), taker);
+        env.close();
+
+        std::uint32_t const startSeq = env.seq(taker);
+        std::vector<std::shared_ptr<STTx const>> txns;
+        txns.reserve(crossPerRep);
+        for (int i = 0; i < crossPerRep; ++i)
+            txns.push_back(
+                env.jt(offer(taker, USD(100), XRP(500 + (i % pages))), Seq(startSeq + i), Fee(100))
+                    .stx);
+
+        auto const base = env.current();
+
+        auto runArm = [&](bool indexEnabled) {
+            OrderBookIndex::setEnabled(indexEnabled);
+            std::vector<double> samples;
+            for (int rep = 0; rep < reps; ++rep)
+            {
+                OpenView accum(kOpenLedger, base->rules(), base);
+                // Warm the maintained index outside the timed region (models the
+                // steady state where it is kept in sync, not rebuilt per cross).
+                if (indexEnabled)
+                    accum.orderBookIndex().rebuildBook(accum, book);
+                auto const t0 = clock::now();
+                for (auto const& tx : txns)
+                    apply(env.app(), accum, *tx, TapNone, env.journal);
+                auto const t1 = clock::now();
+                samples.push_back(
+                    static_cast<double>(
+                        std::chrono::duration_cast<std::chrono::nanoseconds>(t1 - t0).count()) /
+                    crossPerRep / 1000.0);
+            }
+            return median(samples);
+        };
+
+        double const off = runArm(false);
+        double const on = runArm(true);
+        OrderBookIndex::setEnabled(true);  // restore default
+
+        double const speedup = on > 0 ? off / on : 0.0;
+
+        log << "\n=== Arm 4: crossing-apply, index ON vs OFF (no ledger close) ===\n"
+            << "  book pages              : " << pages << "\n"
+            << "  crossings / rep         : " << crossPerRep << "\n"
+            << "  index OFF us/crossing   : " << off << "  (baseline succ() walk)\n"
+            << "  index ON  us/crossing   : " << on << "  (in-memory cursor)\n"
+            << "  speedup                 : " << speedup << "x\n"
+            << std::endl;
+    }
+
+    // Arm 5 (P9.6 headline): the REALISTIC per-tx path. Each crossing is applied
+    // to a fresh COW copy of the prior OpenView — exactly what OpenLedger::modify
+    // does per transaction — so the persistent index warms via the clone (no
+    // pre-seed) and the clone cost is INCLUDED in the timing. Index ON should now
+    // beat OFF on this path (the warm cursor amortizes the one cold rebuild),
+    // unlike the non-persistent index which cold-started and rebuilt every tx.
+    void
+    testCrossingWarmArm()
+    {
+        testcase("Arm 5: realistic per-tx-copy crossing, index ON vs OFF");
+        using namespace jtx;
+
+        int const pages = 400;       // deep enough to stay populated across the batch
+        int const crossPerRep = 100;
+        int const reps = 200;
+
+        Env env{*this};
+        auto const gw = Account{"gw"};
+        auto const USD = gw["USD"];
+        auto const book = buildDeepBook(env, gw, pages);
+
+        Account const taker{"taker"};
+        env.fund(XRP(100'000'000), taker);
+        env.close();
+        env.trust(USD(1'000'000'000), taker);
+        env.close();
+
+        std::uint32_t const startSeq = env.seq(taker);
+        std::vector<std::shared_ptr<STTx const>> txns;
+        txns.reserve(crossPerRep);
+        for (int i = 0; i < crossPerRep; ++i)
+            txns.push_back(
+                env.jt(offer(taker, USD(100), XRP(500 + (i % pages))), Seq(startSeq + i), Fee(100))
+                    .stx);
+
+        auto const base = env.current();
+
+        auto runArm = [&](bool indexEnabled) {
+            OrderBookIndex::setEnabled(indexEnabled);
+            std::vector<double> samples;
+            for (int rep = 0; rep < reps; ++rep)
+            {
+                // Fresh cold OpenView over the closed book (index empty).
+                auto current = std::make_shared<OpenView>(kOpenLedger, base->rules(), base);
+                auto const t0 = clock::now();
+                for (auto const& tx : txns)
+                {
+                    // The per-tx COW copy (clones the persistent index) — exactly
+                    // what OpenLedger::modify does per transaction.
+                    auto next = std::make_shared<OpenView>(*current);
+                    apply(env.app(), *next, *tx, TapNone, env.journal);
+                    current = next;
+                }
+                auto const t1 = clock::now();
+                samples.push_back(
+                    static_cast<double>(
+                        std::chrono::duration_cast<std::chrono::nanoseconds>(t1 - t0).count()) /
+                    crossPerRep / 1000.0);
+            }
+            return median(samples);
+        };
+
+        double const off = runArm(false);
+        double const on = runArm(true);
+        OrderBookIndex::setEnabled(true);
+
+        double const speedup = on > 0 ? off / on : 0.0;
+
+        log << "\n=== Arm 5: realistic per-tx-copy crossing (clones index per tx) ===\n"
+            << "  book pages              : " << pages << "\n"
+            << "  crossings / rep         : " << crossPerRep << "\n"
+            << "  index OFF us/crossing   : " << off << "  (succ() per offer, per tx)\n"
+            << "  index ON  us/crossing   : " << on << "  (warm cursor; clone+rebuild amortized)\n"
+            << "  speedup                 : " << speedup << "x\n"
+            << std::endl;
+    }
+
+public:
+    void
+    run() override
+    {
+        // BENCH_PROFILE=1 runs only the crossing-apply loop (long) for `sample`.
+        if (std::getenv("BENCH_PROFILE") != nullptr)
+        {
+            testCrossingApplyProfile();
+            return;
+        }
+        testReadPath();
+        testEndToEnd();
+        testCrossingApplyProfile();
+        testCrossingIndexArm();
+        testCrossingWarmArm();
+    }
+};
+
+BEAST_DEFINE_TESTSUITE_MANUAL_PRIO(TopOfBookCacheBench, app, xrpl, 20);
+
+}  // namespace xrpl::test

src/test/ledger/OrderBookIndex_test.cpp

@@ -0,0 +1,257 @@
+#include <test/jtx/Account.h>
+#include <test/jtx/Env.h>
+#include <test/jtx/amount.h>
+#include <test/jtx/fee.h>
+#include <test/jtx/offer.h>
+#include <test/jtx/pay.h>
+#include <test/jtx/seq.h>
+#include <test/jtx/trust.h>
+
+#include <xrpl/beast/unit_test/suite.h>
+#include <xrpl/ledger/ApplyView.h>
+#include <xrpl/ledger/OpenView.h>
+#include <xrpl/ledger/OrderBookIndex.h>
+#include <xrpl/protocol/Indexes.h>
+#include <xrpl/protocol/Issue.h>
+#include <xrpl/protocol/SField.h>
+#include <xrpl/tx/apply.h>
+
+#include <memory>
+#include <vector>
+
+namespace xrpl::test {
+
+/** Proves OrderBookIndex's rebuild/walk against a real SHAMap-backed book, and
+    that an index maintained by inserting offers in creation order matches the
+    canonical directory walk (the determinism assumption behind P9.3). */
+class OrderBookIndex_test : public beast::unit_test::Suite
+{
+    // Read an offer's quality-directory root (the key the index levels on).
+    static uint256
+    bookDirOf(ReadView const& view, uint256 const& offerKey)
+    {
+        auto const sle = view.read(keylet::offer(offerKey));
+        return sle ? sle->getFieldH256(sfBookDirectory) : uint256{};
+    }
+
+    void
+    testRebuildMatchesWalk()
+    {
+        testcase("rebuild matches SHAMap walk, ordered best-quality-first");
+        using namespace jtx;
+
+        Env env{*this};
+        auto const gw = Account{"gw"};
+        auto const USD = gw["USD"];
+        Account const maker{"maker"};
+
+        env.fund(XRP(10'000'000), gw, maker);
+        env.close();
+        env.trust(USD(100'000'000), maker);
+        env.close();
+        env(pay(gw, maker, USD(10'000'000)));
+        env.close();
+
+        // Book the maker's offers populate: in = TakerPays asset (XRP),
+        // out = TakerGets asset (USD). (OfferCreate.cpp builds it this way.)
+        Book const book{xrpIssue(), USD.issue(), std::nullopt};
+
+        // Place offers, recording each offer's key in creation order.
+        // - 5 distinct qualities (distinct TakerPays => distinct levels)
+        // - one quality with 40 offers to force a multi-page directory level
+        //   (exercises cdirNext across pages in the walk).
+        std::vector<uint256> created;
+        auto place = [&](int xrpPays, int usdGets) {
+            auto const seq = env.seq(maker);
+            env(offer(maker, XRP(xrpPays), USD(usdGets)));
+            created.push_back(keylet::offer(maker, seq).key);
+        };
+
+        for (int q = 0; q < 5; ++q)
+            place(500 + q, 100);          // 5 distinct qualities
+        for (int i = 0; i < 40; ++i)
+            place(800, 100);              // 40 offers at one shared quality
+        env.close();
+
+        auto const view = env.closed();
+
+        // Rebuild from the authoritative state.
+        OrderBookIndex rebuilt;
+        rebuilt.rebuildBook(*view, book);
+
+        BEAST_EXPECT(rebuilt.offerCount(book) == created.size());
+        BEAST_EXPECT(rebuilt.validateMatchesShaMap(*view, book));
+        BEAST_EXPECT(rebuilt.rebuilds() == 1u);
+
+        // Flattened order must be non-decreasing in quality (best first).
+        auto const flat = rebuilt.flatten(book);
+        BEAST_EXPECT(flat.size() == created.size());
+        bool ordered = true;
+        for (std::size_t i = 1; i < flat.size(); ++i)
+        {
+            auto const prev = getQuality(bookDirOf(*view, flat[i - 1]));
+            auto const cur = getQuality(bookDirOf(*view, flat[i]));
+            if (cur < prev)
+                ordered = false;
+        }
+        BEAST_EXPECT(ordered);
+
+        // An index maintained by inserting in creation order (simulating the
+        // P9.3 apply-path hooks, no deletions) must equal the rebuilt index.
+        OrderBookIndex maintained;
+        for (auto const& offerKey : created)
+            maintained.insertOffer(book, bookDirOf(*view, offerKey), offerKey);
+        BEAST_EXPECT(maintained.flatten(book) == flat);
+        BEAST_EXPECT(maintained.validateMatchesShaMap(*view, book));
+    }
+
+    void
+    testEmptyAndAbsentBook()
+    {
+        testcase("rebuild of an empty book yields nothing");
+        using namespace jtx;
+        Env env{*this};
+        env.fund(XRP(10'000), Account{"gw"});
+        env.close();
+
+        Book const book{xrpIssue(), Account{"gw"}["USD"].issue(), std::nullopt};
+        OrderBookIndex idx;
+        idx.rebuildBook(*env.closed(), book);
+        BEAST_EXPECT(idx.offerCount(book) == 0u);
+        BEAST_EXPECT(idx.bookCount() == 0u);
+        BEAST_EXPECT(idx.validateMatchesShaMap(*env.closed(), book));
+    }
+
+    // P9.3: an index seeded from state and then maintained through real
+    // OfferCreate apply (crossings delete offers, placements insert them) must
+    // stay byte-exactly equal to a fresh SHAMap walk. This proves the notify
+    // hooks keep the index in sync without any read-path/seam involvement.
+    void
+    testMaintenanceInSync()
+    {
+        testcase("index stays in sync through real crossing/placement apply");
+        using namespace jtx;
+
+        Env env{*this};
+        auto const gw = Account{"gw"};
+        auto const USD = gw["USD"];
+        Account const maker{"maker"};
+        Account const taker{"taker"};
+
+        env.fund(XRP(10'000'000), gw, maker, taker);
+        env.close();
+        env.trust(USD(100'000'000), maker, taker);
+        env.close();
+        env(pay(gw, maker, USD(10'000'000)));
+        env.close();
+
+        Book const book{xrpIssue(), USD.issue(), std::nullopt};
+
+        // Resting book: 30 offers across distinct qualities.
+        for (int i = 0; i < 30; ++i)
+            env(offer(maker, XRP(500 + i), USD(100)));
+        env.close();
+
+        // Owned OpenView over the closed state; seed the index by rebuild
+        // (the attach-time / startup model).
+        auto const base = env.current();
+        OpenView accum(kOpenLedger, base->rules(), base);
+        accum.orderBookIndex().rebuildBook(accum, book);
+        BEAST_EXPECT(accum.orderBookIndex().validateMatchesShaMap(accum, book));
+        BEAST_EXPECT(accum.orderBookIndex().offerCount(book) == 30u);
+
+        // Pre-sign a mixed batch: taker crossings (consume → delete) and maker
+        // placements at new qualities (insert), with explicit sequences.
+        std::vector<std::shared_ptr<STTx const>> txns;
+        std::uint32_t takerSeq = env.seq(taker);
+        std::uint32_t makerSeq = env.seq(maker);
+        for (int i = 0; i < 15; ++i)
+        {
+            txns.push_back(
+                env.jt(offer(taker, USD(100), XRP(500 + i)), Seq(takerSeq++), Fee(100)).stx);
+            txns.push_back(
+                env.jt(offer(maker, XRP(700 + i), USD(100)), Seq(makerSeq++), Fee(100)).stx);
+        }
+
+        // Apply to the owned view; the index is maintained via the notify
+        // hooks (flushed on each apply). Validate after every tx so a desync
+        // is pinned to the exact transaction that caused it.
+        for (auto const& tx : txns)
+        {
+            auto const r = apply(env.app(), accum, *tx, TapNone, env.journal);
+            BEAST_EXPECT(r.applied);
+            BEAST_EXPECT(accum.orderBookIndex().validateMatchesShaMap(accum, book));
+        }
+
+        // The index actually did work (both directions exercised).
+        BEAST_EXPECT(accum.orderBookIndex().inserts() > 0u);
+        BEAST_EXPECT(accum.orderBookIndex().deletes() > 0u);
+    }
+
+    // P9.6 Stage E: across a ledger close the open-round index is not carried
+    // (the next round starts cold and warms via rebuild-on-touch). Confirm that
+    // after real crossings + a close, the post-close state rebuilds clean — i.e.
+    // the close handoff leaves no index/SHAMap drift.
+    void
+    testCloseHandoff()
+    {
+        testcase("index rebuilds clean across a ledger close");
+        using namespace jtx;
+
+        Env env{*this};
+        auto const gw = Account{"gw"};
+        auto const USD = gw["USD"];
+        Account const maker{"maker"};
+        Account const taker{"taker"};
+
+        env.fund(XRP(10'000'000), gw, maker, taker);
+        env.close();
+        env.trust(USD(100'000'000), maker, taker);
+        env.close();
+        env(pay(gw, maker, USD(10'000'000)));
+        env.close();
+
+        Book const book{xrpIssue(), USD.issue(), std::nullopt};
+
+        for (int i = 0; i < 20; ++i)
+            env(offer(maker, XRP(500 + i), USD(100)));
+        env.close();
+
+        // Round 1: real crossings through the open ledger, then close.
+        for (int i = 0; i < 8; ++i)
+            env(offer(taker, USD(100), XRP(500 + i)));
+        env.close();
+
+        // After the close, a fresh index rebuilt from the post-close ledger must
+        // match the SHAMap walk (no drift left by the round's crossings).
+        {
+            OrderBookIndex idx;
+            idx.rebuildBook(*env.closed(), book);
+            BEAST_EXPECT(idx.validateMatchesShaMap(*env.closed(), book));
+        }
+
+        // Round 2: more crossings on top of the post-close state, then re-check.
+        for (int i = 8; i < 16; ++i)
+            env(offer(taker, USD(100), XRP(500 + i)));
+        env.close();
+        {
+            OrderBookIndex idx;
+            idx.rebuildBook(*env.closed(), book);
+            BEAST_EXPECT(idx.validateMatchesShaMap(*env.closed(), book));
+        }
+    }
+
+public:
+    void
+    run() override
+    {
+        testRebuildMatchesWalk();
+        testEmptyAndAbsentBook();
+        testMaintenanceInSync();
+        testCloseHandoff();
+    }
+};
+
+BEAST_DEFINE_TESTSUITE(OrderBookIndex, ledger, xrpl);
+
+}  // namespace xrpl::test

src/tests/libxrpl/CMakeLists.txt

@@ -35,18 +35,14 @@ xrpl_add_test(json)
 target_link_libraries(xrpl.test.json PRIVATE xrpl.imports.test)
 add_dependencies(xrpl.tests xrpl.test.json)
 
-xrpl_add_test(ledger)
-target_link_libraries(xrpl.test.ledger PRIVATE xrpl.imports.test)
-add_dependencies(xrpl.tests xrpl.test.ledger)
-
-xrpl_add_test(shamap)
-target_link_libraries(xrpl.test.shamap PRIVATE xrpl.imports.test)
-add_dependencies(xrpl.tests xrpl.test.shamap)
-
 xrpl_add_test(tx)
 target_link_libraries(xrpl.test.tx PRIVATE xrpl.imports.test)
 add_dependencies(xrpl.tests xrpl.test.tx)
 
+xrpl_add_test(ledger)
+target_link_libraries(xrpl.test.ledger PRIVATE xrpl.imports.test)
+add_dependencies(xrpl.tests xrpl.test.ledger)
+
 xrpl_add_test(protocol_autogen)
 target_link_libraries(xrpl.test.protocol_autogen PRIVATE xrpl.imports.test)
 add_dependencies(xrpl.tests xrpl.test.protocol_autogen)

src/tests/libxrpl/ledger/DeferredRebuild.cpp

@@ -1,927 +0,0 @@
-// Tests for the Plan 7 deferred-SHAMap rebuild planning kernel.
-//
-// The full plan-7 algorithm (bottom-up parallel rebuild of a SHAMap
-// from a parent SHAMap + delta) decomposes into two layers:
-//
-//   1. PLAN — given a set of modified leaf keys, compute which inner
-//      nodes need their hash recomputed. Pure algorithm; no SHAMap.
-//   2. EXECUTE — given the plan + a parent SHAMap + the delta, produce
-//      the new SHAMap with byte-identical root hash.
-//
-// This file tests (1). Layer (2) requires SHAMap fixtures (Family,
-// NodeStore) and lands in a follow-up.
-
-#include <xrpl/basics/base_uint.h>
-#include <xrpl/basics/SHAMapHash.h>
-#include <xrpl/basics/Slice.h>
-#include <xrpl/ledger/DeferredRebuild.h>
-#include <xrpl/shamap/SHAMapInnerNode.h>
-
-#include <gtest/gtest.h>
-
-#include <algorithm>
-#include <array>
-#include <chrono>
-#include <cstdint>
-#include <cstdio>
-#include <vector>
-
-using namespace xrpl;
-
-namespace {
-
-// Inhibit dead-code elimination in benchmark loops.
-template <typename T>
-inline void
-benchmark_use(T const& v)
-{
-#if defined(__clang__) || defined(__GNUC__)
-    asm volatile("" : : "r,m"(v) : "memory");
-#else
-    (void)v;
-#endif
-}
-
-[[nodiscard]] uint256
-keyOf(std::uint64_t v)
-{
-    return uint256{v};
-}
-
-// Build a key whose first `nibblesIntoKey` nibbles match a given pattern,
-// remaining nibbles set to 0. SHAMap stores keys big-endian; the most
-// significant nibble of the key is the depth-1 branch from root.
-[[nodiscard]] uint256
-keyWithPrefix(std::vector<std::uint8_t> const& prefixNibbles)
-{
-    uint256 k;  // zero-initialised
-    // Each byte of uint256 holds two nibbles, high nibble first.
-    for (std::size_t i = 0; i < prefixNibbles.size(); ++i)
-    {
-        auto const byteIdx = i / 2;
-        auto const isHighNibble = (i % 2) == 0;
-        if (byteIdx >= uint256::kBytes)
-            break;
-        auto const shift = isHighNibble ? 4 : 0;
-        k.data()[byteIdx] |= static_cast<std::uint8_t>(
-            (prefixNibbles[i] & 0x0F) << shift);
-    }
-    return k;
-}
-
-}  // namespace
-
-// ---------------------------------------------------------------------------
-// Empty + single-key shape
-// ---------------------------------------------------------------------------
-
-TEST(DeferredRebuild_Plan, EmptyKeySetYieldsEmptyPlan)
-{
-    std::vector<uint256> keys;
-    auto const plan = planDeferredRebuild(keys);
-    EXPECT_TRUE(plan.empty());
-}
-
-TEST(DeferredRebuild_Plan, SingleKeyTouchesEveryDepth)
-{
-    // A single 256-bit key has 64 nibbles, so the path from root to
-    // the leaf passes through 64 inner nodes (depth 1 through 64).
-    // Plus the root itself at depth 0 — that gives 65 affected
-    // ancestor positions.
-    //
-    // Actually the leaf at depth 64 is the SLE itself, not an inner
-    // node. The inner nodes along the path are at depths 0 (root)
-    // through 63. So 64 inner-node positions total.
-    std::vector<uint256> keys{keyOf(1)};
-    auto const plan = planDeferredRebuild(keys);
-    EXPECT_EQ(plan.size(), 64u);
-}
-
-TEST(DeferredRebuild_Plan, PlanIsDepthDescending)
-{
-    // Bottom-up rebuild walks deepest nodes first. The plan must be
-    // ordered so the consumer can iterate and find each level before
-    // its parent.
-    std::vector<uint256> keys{keyOf(1)};
-    auto const plan = planDeferredRebuild(keys);
-
-    for (std::size_t i = 1; i < plan.size(); ++i)
-    {
-        EXPECT_LE(plan[i].depth, plan[i - 1].depth)
-            << "Plan not depth-descending at index " << i;
-    }
-    EXPECT_EQ(plan.front().depth, 63);  // deepest inner node
-    EXPECT_EQ(plan.back().depth, 0);    // root
-}
-
-// ---------------------------------------------------------------------------
-// Multiple keys — ancestor sharing
-// ---------------------------------------------------------------------------
-
-TEST(DeferredRebuild_Plan, DisjointKeysShareOnlyRoot)
-{
-    // Two keys that differ in their very first nibble share only one
-    // ancestor: the root (depth 0). Each contributes 63 unique
-    // ancestors at depths 1..63, plus the shared root.
-    //
-    // Total affected nodes: 63 + 63 + 1 = 127.
-    auto const keyA = keyWithPrefix({0x0});  // first nibble = 0
-    auto const keyB = keyWithPrefix({0xF});  // first nibble = 15
-
-    auto const plan = planDeferredRebuild({keyA, keyB});
-    EXPECT_EQ(plan.size(), 127u);
-}
-
-TEST(DeferredRebuild_Plan, KeysSharingPrefixShareAncestors)
-{
-    // Two keys that share their first 3 nibbles share their prefixes
-    // at depths 0, 1, 2, AND 3 — at depth N the prefix is the first N
-    // nibbles, so shared-first-3-nibbles means shared at depths 0..3.
-    //
-    // They diverge at depth 4 (the prefix at depth 4 includes the 4th
-    // nibble, which differs). So each contributes unique ancestors at
-    // depths 4..63 = 60 levels.
-    //
-    // Total: 4 shared (depths 0..3) + 60*2 unique = 124.
-    auto const keyA = keyWithPrefix({0x1, 0x2, 0x3, 0x4});
-    auto const keyB = keyWithPrefix({0x1, 0x2, 0x3, 0x5});
-
-    auto const plan = planDeferredRebuild({keyA, keyB});
-    EXPECT_EQ(plan.size(), 124u);
-}
-
-TEST(DeferredRebuild_Plan, IdenticalKeysCountedOnce)
-{
-    // Two identical keys produce the same plan as a single key — the
-    // delta is "this key changed", duplicated or not.
-    auto const k = keyOf(7);
-    auto const plan = planDeferredRebuild({k, k});
-    EXPECT_EQ(plan.size(), 64u);
-}
-
-// ---------------------------------------------------------------------------
-// Correctness of node-identity
-// ---------------------------------------------------------------------------
-
-TEST(DeferredRebuild_Plan, NodesAtSameDepthHaveDistinctPrefixesWhenKeysDiffer)
-{
-    auto const keyA = keyWithPrefix({0x0});
-    auto const keyB = keyWithPrefix({0xF});
-
-    auto const plan = planDeferredRebuild({keyA, keyB});
-
-    // At depth 1, the two keys yield distinct inner-node positions —
-    // they branch at nibble 0 vs nibble F.
-    int depth1NodeCount = 0;
-    for (auto const& node : plan)
-        if (node.depth == 1)
-            ++depth1NodeCount;
-    EXPECT_EQ(depth1NodeCount, 2);
-}
-
-TEST(DeferredRebuild_Plan, RootAlwaysPresent)
-{
-    // Every non-empty plan includes the root (depth 0).
-    std::vector<uint256> keys{keyOf(1), keyOf(2), keyOf(3)};
-    auto const plan = planDeferredRebuild(keys);
-
-    auto const rootCount = std::count_if(
-        plan.begin(),
-        plan.end(),
-        [](AffectedNode const& n) { return n.depth == 0; });
-    EXPECT_EQ(rootCount, 1);
-}
-
-// ---------------------------------------------------------------------------
-// Benchmarks. TDD with benchmarks: gate the plan-generation cost so a
-// regression shows up as a failed test, not a mainnet incident.
-//
-// `planDeferredRebuild` runs at every ledger close to determine which
-// inner nodes to recompute. At realistic mainnet workload (~3000 SLEs
-// modified per ledger), this must be cheap — well under a millisecond.
-// ---------------------------------------------------------------------------
-
-TEST(DeferredRebuild_Bench, PlanGenerationAtLedgerScale)
-{
-    // 3000 modified keys ≈ realistic 1500-TPS-target ledger.
-    constexpr std::size_t N = 3'000;
-    std::vector<uint256> keys;
-    keys.reserve(N);
-    for (std::uint64_t i = 0; i < N; ++i)
-        keys.push_back(keyOf(i));
-
-    auto const t0 = std::chrono::high_resolution_clock::now();
-    auto const plan = planDeferredRebuild(keys);
-    auto const elapsed = std::chrono::high_resolution_clock::now() - t0;
-    auto const us =
-        std::chrono::duration_cast<std::chrono::microseconds>(elapsed).count();
-
-    std::printf(
-        "  planDeferredRebuild N=%zu modified keys : %lld µs (plan size %zu)\n",
-        N,
-        static_cast<long long>(us),
-        plan.size());
-
-    // Post-LCP-optimization: measured ~1.5 ms locally. 15 ms is ~10×
-    // measured — generous for slow CI but tight enough to catch real
-    // regressions (e.g., accidental return to O(K*64) prefix ops).
-    EXPECT_LT(us, 15'000)
-        << "Plan generation at 3000 modified keys should be under 15 ms";
-}
-
-TEST(DeferredRebuild_Bench, PlanGenerationAtTenKKeys)
-{
-    // Stress-test at 10x typical to catch O(N²) regressions early.
-    constexpr std::size_t N = 30'000;
-    std::vector<uint256> keys;
-    keys.reserve(N);
-    for (std::uint64_t i = 0; i < N; ++i)
-        keys.push_back(keyOf(i));
-
-    auto const t0 = std::chrono::high_resolution_clock::now();
-    auto const plan = planDeferredRebuild(keys);
-    auto const elapsed = std::chrono::high_resolution_clock::now() - t0;
-    auto const us =
-        std::chrono::duration_cast<std::chrono::microseconds>(elapsed).count();
-
-    std::printf(
-        "  planDeferredRebuild N=%zu modified keys : %lld µs (plan size %zu)\n",
-        N,
-        static_cast<long long>(us),
-        plan.size());
-
-    // Post-LCP: measured ~16 ms locally. 150 ms threshold = ~10×.
-    EXPECT_LT(us, 150'000)
-        << "Plan generation at 30k keys should be under 150 ms";
-}
-
-TEST(DeferredRebuild_Plan, NoDuplicateNodes)
-{
-    // A given inner node must appear at most once in the plan, even if
-    // many leaves share it as an ancestor.
-    std::vector<uint256> keys;
-    for (std::uint64_t i = 0; i < 100; ++i)
-        keys.push_back(keyOf(i));
-
-    auto const plan = planDeferredRebuild(keys);
-
-    for (std::size_t i = 0; i < plan.size(); ++i)
-        for (std::size_t j = i + 1; j < plan.size(); ++j)
-            EXPECT_FALSE(
-                plan[i].depth == plan[j].depth &&
-                plan[i].prefix == plan[j].prefix)
-                << "Duplicate at indices " << i << " and " << j;
-}
-
-// ---------------------------------------------------------------------------
-// Inner-node hash computation (P7.2.1).
-//
-// The bottom-up rebuild's elementary operation is: given the 16 child
-// hashes of an inner node, compute the inner node's own hash. This is
-// the pure function `computeInnerNodeHash` — the kernel of plan-7's
-// "EXECUTE" layer.
-//
-// We test it against the production code path as oracle:
-// `SHAMapInnerNode::makeFullInner` deserializes 16 hashes from a
-// 512-byte slice and calls `updateHash()` to compute the node's hash.
-// Our function must produce byte-identical output to that path — the
-// safety property that makes plan-7 a pure internal optimization
-// (no protocol change).
-// ---------------------------------------------------------------------------
-
-namespace {
-
-// Pack 16 child hashes into the 512-byte buffer SHAMapInnerNode expects.
-[[nodiscard]] std::vector<std::uint8_t>
-packChildHashes(std::array<uint256, 16> const& children)
-{
-    std::vector<std::uint8_t> buf;
-    buf.reserve(16 * 32);
-    for (auto const& h : children)
-        for (auto b : h)
-            buf.push_back(b);
-    return buf;
-}
-
-// Use SHAMapInnerNode's makeFullInner factory to produce the canonical
-// hash. The returned tree node has updateHash() already invoked.
-[[nodiscard]] uint256
-oracleHash(std::array<uint256, 16> const& children)
-{
-    auto const buf = packChildHashes(children);
-    auto node = SHAMapInnerNode::makeFullInner(
-        Slice{buf.data(), buf.size()}, SHAMapHash{}, /*hashValid=*/false);
-    return node->getHash().asUInt256();
-}
-
-}  // namespace
-
-TEST(DeferredRebuild_InnerHash, AllZeroChildrenMatchesOracle)
-{
-    // An inner node with all-zero children is the same byte pattern
-    // an empty branch produces; computing its hash via either path
-    // must agree.
-    std::array<uint256, 16> children{};  // all zero
-    auto const oracle = oracleHash(children);
-    auto const ours = computeInnerNodeHash(children);
-    EXPECT_EQ(ours, oracle);
-}
-
-TEST(DeferredRebuild_InnerHash, SingleChildMatchesOracle)
-{
-    std::array<uint256, 16> children{};
-    children[7] = uint256{0xDEADBEEF};
-    auto const oracle = oracleHash(children);
-    auto const ours = computeInnerNodeHash(children);
-    EXPECT_EQ(ours, oracle);
-}
-
-TEST(DeferredRebuild_InnerHash, AllChildrenSetMatchesOracle)
-{
-    std::array<uint256, 16> children;
-    for (std::size_t i = 0; i < 16; ++i)
-        children[i] = uint256{0x100ULL + i};
-    auto const oracle = oracleHash(children);
-    auto const ours = computeInnerNodeHash(children);
-    EXPECT_EQ(ours, oracle);
-}
-
-TEST(DeferredRebuild_InnerHash, SwappingChildrenChangesHash)
-{
-    // Order matters — branch position is part of the hash input.
-    std::array<uint256, 16> a;
-    for (std::size_t i = 0; i < 16; ++i)
-        a[i] = uint256{0x200ULL + i};
-
-    std::array<uint256, 16> b = a;
-    std::swap(b[3], b[11]);
-
-    EXPECT_NE(computeInnerNodeHash(a), computeInnerNodeHash(b));
-    EXPECT_EQ(computeInnerNodeHash(a), oracleHash(a));
-    EXPECT_EQ(computeInnerNodeHash(b), oracleHash(b));
-}
-
-TEST(DeferredRebuild_InnerHash, Deterministic)
-{
-    std::array<uint256, 16> children;
-    for (std::size_t i = 0; i < 16; ++i)
-        children[i] = uint256{0x300ULL + i * 0x1234};
-
-    auto const h1 = computeInnerNodeHash(children);
-    auto const h2 = computeInnerNodeHash(children);
-    auto const h3 = computeInnerNodeHash(children);
-    EXPECT_EQ(h1, h2);
-    EXPECT_EQ(h2, h3);
-}
-
-TEST(DeferredRebuild_InnerHash, ChangingOneChildChangesHash)
-{
-    std::array<uint256, 16> base;
-    for (std::size_t i = 0; i < 16; ++i)
-        base[i] = uint256{0x400ULL + i};
-
-    for (std::size_t branchToBump = 0; branchToBump < 16; ++branchToBump)
-    {
-        auto modified = base;
-        modified[branchToBump] = uint256{0xCAFEBABEULL + branchToBump};
-        EXPECT_NE(computeInnerNodeHash(base), computeInnerNodeHash(modified))
-            << "Hash unchanged when modifying branch " << branchToBump;
-    }
-}
-
-// ---------------------------------------------------------------------------
-// Bottom-up plan execution (P7.2.2).
-//
-// Given a depth-descending plan and a callback that supplies "original"
-// child hashes from the parent SHAMap, walk the plan, compute each
-// affected node's new hash, and return them in a map. Hashes computed
-// earlier in the walk (deeper nodes) are visible to later (shallower)
-// nodes that have them as children.
-//
-// The pure-function design takes a callback rather than a SHAMap
-// reference so the algorithm can be tested without SHAMap fixtures.
-// Real integration will pass a callback that walks the actual SHAMap.
-// ---------------------------------------------------------------------------
-
-namespace {
-
-// Given a parent inner node at (parentDepth, parentPrefix), compute the
-// child prefix at the given branch (0..15). The child is at depth
-// (parentDepth + 1) and its prefix has the nibble at position
-// parentDepth set to the branch value.
-[[nodiscard]] uint256
-childPrefixOf(uint256 const& parentPrefix, int parentDepth, std::uint8_t branch)
-{
-    uint256 result = parentPrefix;
-    int const byteIdx = parentDepth / 2;
-    bool const isHighNibble = (parentDepth % 2) == 0;
-    if (isHighNibble)
-        result.data()[byteIdx] =
-            (result.data()[byteIdx] & 0x0F) |
-            static_cast<std::uint8_t>((branch & 0x0F) << 4);
-    else
-        result.data()[byteIdx] =
-            (result.data()[byteIdx] & 0xF0) |
-            static_cast<std::uint8_t>(branch & 0x0F);
-    return result;
-}
-
-}  // namespace
-
-TEST(DeferredRebuild_ChildPrefix, BranchZeroPreservesPrefix)
-{
-    auto const parent = keyWithPrefix({0xA, 0xB});
-    // Setting nibble 2 to 0 — that's already the case
-    auto const child = childPrefixOf(parent, 2, 0);
-    EXPECT_EQ(child, parent);
-}
-
-TEST(DeferredRebuild_ChildPrefix, SetsCorrectNibblePosition)
-{
-    uint256 const empty{};
-    // Parent at depth 0, branch 5 → nibble 0 of result = 5
-    auto const child = childPrefixOf(empty, 0, 5);
-    auto const expected = keyWithPrefix({5});
-    EXPECT_EQ(child, expected);
-}
-
-TEST(DeferredRebuild_ChildPrefix, DepthOneSetsNibbleOne)
-{
-    auto const parent = keyWithPrefix({0xA});
-    // Parent at depth 1 (first nibble set to A), branch 7
-    // → child has nibbles (A, 7, 0, 0, ...)
-    auto const child = childPrefixOf(parent, 1, 7);
-    auto const expected = keyWithPrefix({0xA, 0x7});
-    EXPECT_EQ(child, expected);
-}
-
-TEST(DeferredRebuild_ChildPrefix, DepthSixtyThreeIsLastNibble)
-{
-    uint256 parent{};
-    for (int i = 0; i < uint256::kBytes; ++i)
-        parent.data()[i] = 0xAB;  // arbitrary fill
-    // Parent at depth 63 → set the last nibble (low nibble of last byte)
-    auto const child = childPrefixOf(parent, 63, 0xC);
-    uint256 expected = parent;
-    expected.data()[31] = (expected.data()[31] & 0xF0) | 0x0C;
-    EXPECT_EQ(child, expected);
-}
-
-// ---------------------------------------------------------------------------
-// Plan execution — the actual rebuild walk
-// ---------------------------------------------------------------------------
-
-TEST(DeferredRebuild_Execute, EmptyPlanReturnsEmptyResult)
-{
-    std::vector<AffectedNode> plan;
-    auto const result = executeRebuildPlan(
-        plan,
-        [](int /*depth*/, uint256 const& /*prefix*/) { return uint256{}; });
-    EXPECT_TRUE(result.empty());
-}
-
-TEST(DeferredRebuild_Execute, SingleRootNodePlanComputesRootHash)
-{
-    // Plan: just the root at depth 0. All 16 children come from the
-    // parent SHAMap (none are themselves affected). The rebuild
-    // should produce a root hash equal to computeInnerNodeHash over
-    // those children.
-    std::vector<AffectedNode> plan{{0, uint256{}}};
-
-    // Mock parent: child branch b has hash 0x100 + b
-    auto const parentLookup = [](int depth, uint256 const& prefix) {
-        if (depth != 1)
-            return uint256{};
-        // Branch is the high nibble of the first byte of prefix
-        std::uint8_t branch = (prefix.data()[0] >> 4) & 0x0F;
-        return uint256{0x100ULL + branch};
-    };
-
-    auto const result = executeRebuildPlan(plan, parentLookup);
-    ASSERT_EQ(result.size(), 1u);
-
-    // Expected: compute the root hash from the 16 child hashes
-    std::array<uint256, 16> children;
-    for (std::uint8_t b = 0; b < 16; ++b)
-        children[b] = uint256{0x100ULL + b};
-    auto const expectedRoot = computeInnerNodeHash(children);
-
-    AffectedNode const rootKey{0, uint256{}};
-    auto it = result.find(rootKey);
-    ASSERT_NE(it, result.end());
-    EXPECT_EQ(it->second, expectedRoot);
-}
-
-TEST(DeferredRebuild_Execute, ChildInPlanShadowsParentLookup)
-{
-    // Plan contains:
-    //   - depth 1, prefix (5,0,0,...)  — this child of root is affected
-    //   - depth 0, root — the root, which uses the depth-1 result as
-    //                     its branch-5 child
-    //
-    // The plan is depth-descending so the depth-1 node is processed
-    // first, and its computed hash is used as branch-5 of root.
-    //
-    // For the depth-1 node, all 16 of ITS children come from parent
-    // lookup (depth=2).
-    auto const depth1Prefix = keyWithPrefix({0x5});
-    std::vector<AffectedNode> plan{{1, depth1Prefix}, {0, uint256{}}};
-
-    // Mock parent:
-    //   - depth 1 children (depth 2 lookup): return distinct hashes per branch
-    //   - depth 0 children OTHER THAN branch 5 (depth 1 lookup): return
-    //     distinct hashes per branch
-    int parentLookupCalls = 0;
-    auto const parentLookup =
-        [&parentLookupCalls](int depth, uint256 const& prefix) -> uint256 {
-        ++parentLookupCalls;
-        if (depth == 2)
-        {
-            // High nibble of byte 0 is the parent's prefix nibble (5),
-            // low nibble of byte 0 is the branch within that parent.
-            std::uint8_t branch = prefix.data()[0] & 0x0F;
-            return uint256{0xA00ULL + branch};
-        }
-        if (depth == 1)
-        {
-            std::uint8_t branch = (prefix.data()[0] >> 4) & 0x0F;
-            return uint256{0xB00ULL + branch};
-        }
-        return uint256{};
-    };
-
-    auto const result = executeRebuildPlan(plan, parentLookup);
-    ASSERT_EQ(result.size(), 2u);
-
-    // Compute expected depth-1 hash: 16 children from depth-2 lookup
-    std::array<uint256, 16> depth1Children;
-    for (std::uint8_t b = 0; b < 16; ++b)
-        depth1Children[b] = uint256{0xA00ULL + b};
-    auto const expectedDepth1Hash = computeInnerNodeHash(depth1Children);
-
-    AffectedNode const depth1Key{1, depth1Prefix};
-    auto it1 = result.find(depth1Key);
-    ASSERT_NE(it1, result.end());
-    EXPECT_EQ(it1->second, expectedDepth1Hash);
-
-    // Compute expected root hash: branch 5 is the depth-1 result, all
-    // other branches come from parentLookup at depth 1
-    std::array<uint256, 16> rootChildren;
-    for (std::uint8_t b = 0; b < 16; ++b)
-        rootChildren[b] = (b == 5) ? expectedDepth1Hash : uint256{0xB00ULL + b};
-    auto const expectedRoot = computeInnerNodeHash(rootChildren);
-
-    AffectedNode const rootKey{0, uint256{}};
-    auto it0 = result.find(rootKey);
-    ASSERT_NE(it0, result.end());
-    EXPECT_EQ(it0->second, expectedRoot);
-
-    // Sanity: the parentLookup should NOT have been called for the
-    // branch-5 child of root (depth=1, prefix=depth1Prefix), because
-    // that child IS the affected depth-1 node. Verify by counting:
-    //  - depth 2 lookups: 16 (one per branch of the depth-1 node)
-    //  - depth 1 lookups: 15 (all branches of root EXCEPT branch 5)
-    //  - total: 31
-    EXPECT_EQ(parentLookupCalls, 31);
-}
-
-TEST(DeferredRebuild_Execute, DeterministicAcrossRuns)
-{
-    auto const depth1Prefix = keyWithPrefix({0x3});
-    std::vector<AffectedNode> plan{{1, depth1Prefix}, {0, uint256{}}};
-
-    auto const parentLookup = [](int depth, uint256 const& prefix) {
-        std::uint64_t v = 0;
-        for (int i = 0; i < 8; ++i)
-            v = (v << 8) | prefix.data()[i];
-        return uint256{v + static_cast<std::uint64_t>(depth) * 0xABCDEF};
-    };
-
-    auto const r1 = executeRebuildPlan(plan, parentLookup);
-    auto const r2 = executeRebuildPlan(plan, parentLookup);
-    auto const r3 = executeRebuildPlan(plan, parentLookup);
-    EXPECT_EQ(r1, r2);
-    EXPECT_EQ(r2, r3);
-}
-
-// ---------------------------------------------------------------------------
-// Unified entry point: deferredRebuildRoot
-//
-// The consumer-facing API: given (modified keys, parent-state callback),
-// return the new SHAMap root hash. Equivalent to plan + execute + pluck
-// the depth-0 entry, but exposed as a single call so the integration
-// site doesn't have to know about the AffectedNode plumbing.
-// ---------------------------------------------------------------------------
-
-TEST(DeferredRebuild_Root, EmptyKeysReturnsExistingRootFromCallback)
-{
-    // No modifications → no rebuild needed → the new root equals the
-    // existing root, which the parent-state callback supplies at
-    // (depth=0, prefix=zero).
-    uint256 const expectedExistingRoot{0xDEADBEEF};
-    auto const parentLookup = [&expectedExistingRoot](
-                                  int depth, uint256 const& prefix) {
-        if (depth == 0 && prefix == uint256{})
-            return expectedExistingRoot;
-        return uint256{};
-    };
-
-    auto const newRoot = deferredRebuildRoot({}, parentLookup);
-    EXPECT_EQ(newRoot, expectedExistingRoot);
-}
-
-TEST(DeferredRebuild_Root, SingleKeyMatchesPlanAndExecuteComposition)
-{
-    // The unified entry point must be byte-identical to the explicit
-    // plan + execute composition. This is the safety guarantee that
-    // consumers can switch to the unified entry point with no
-    // observable change.
-    std::vector<uint256> keys{keyOf(42)};
-
-    auto const parentLookup = [](int depth, uint256 const& prefix) {
-        std::uint64_t v = static_cast<std::uint64_t>(depth) * 0x1000;
-        for (int i = 0; i < 4; ++i)
-            v += prefix.data()[i];
-        return uint256{v};
-    };
-
-    // Via composition
-    auto const plan = planDeferredRebuild(keys);
-    auto const result = executeRebuildPlan(plan, parentLookup);
-    AffectedNode const rootKey{0, uint256{}};
-    auto const composedRoot = result.at(rootKey);
-
-    // Via unified API
-    auto const unifiedRoot = deferredRebuildRoot(keys, parentLookup);
-
-    EXPECT_EQ(unifiedRoot, composedRoot);
-}
-
-TEST(DeferredRebuild_Root, ManyKeysMatchesPlanAndExecuteComposition)
-{
-    std::vector<uint256> keys;
-    for (std::uint64_t i = 0; i < 100; ++i)
-        keys.push_back(keyOf(i * 13));
-
-    auto const parentLookup = [](int depth, uint256 const& prefix) {
-        std::uint64_t v = static_cast<std::uint64_t>(depth);
-        for (int i = 0; i < 8; ++i)
-            v = (v * 257) + prefix.data()[i];
-        return uint256{v};
-    };
-
-    auto const plan = planDeferredRebuild(keys);
-    auto const result = executeRebuildPlan(plan, parentLookup);
-    AffectedNode const rootKey{0, uint256{}};
-    auto const composedRoot = result.at(rootKey);
-
-    auto const unifiedRoot = deferredRebuildRoot(keys, parentLookup);
-
-    EXPECT_EQ(unifiedRoot, composedRoot);
-}
-
-TEST(DeferredRebuild_Root, IdenticalInputsProduceIdenticalRoots)
-{
-    std::vector<uint256> keys{keyOf(1), keyOf(2), keyOf(3)};
-    auto const parentLookup = [](int depth, uint256 const& prefix) {
-        return uint256{
-            static_cast<std::uint64_t>(depth) * 1000 + prefix.data()[0]};
-    };
-
-    auto const a = deferredRebuildRoot(keys, parentLookup);
-    auto const b = deferredRebuildRoot(keys, parentLookup);
-    EXPECT_EQ(a, b);
-}
-
-// ---------------------------------------------------------------------------
-// End-to-end benchmark
-// ---------------------------------------------------------------------------
-
-// ---------------------------------------------------------------------------
-// Parallel rebuild by subtree (P7.3).
-//
-// The root has 16 children at depth 1 — one per first-nibble value of
-// the keys. Modified keys partition disjointly into these 16 subtrees;
-// each subtree's rebuild touches only its own ancestor paths and never
-// reads or writes another subtree's nodes. So we can rebuild all 16
-// subtrees in parallel, then combine the 16 child hashes into the
-// root in one final step.
-//
-// The combine step needs:
-//   * For each non-empty subtree: the new depth-1 hash from the rebuild
-//   * For each empty subtree:    the existing depth-1 hash from the
-//                                parent SHAMap (unchanged)
-// — then compute the root via computeInnerNodeHash.
-//
-// The pure-function design keeps parallelism orthogonal to correctness:
-// the caller chooses serial or parallel execution, but the result must
-// be byte-identical to the single-threaded deferredRebuildRoot.
-// ---------------------------------------------------------------------------
-
-TEST(DeferredRebuild_Partition, EmptyKeysProduceEmptyBuckets)
-{
-    auto const buckets = partitionByFirstNibble({});
-    for (auto const& b : buckets)
-        EXPECT_TRUE(b.empty());
-}
-
-TEST(DeferredRebuild_Partition, KeysGoToCorrectBucketByFirstNibble)
-{
-    auto const keyA = keyWithPrefix({0x0});
-    auto const keyB = keyWithPrefix({0x5});
-    auto const keyC = keyWithPrefix({0xF});
-
-    auto const buckets = partitionByFirstNibble({keyA, keyB, keyC});
-
-    EXPECT_EQ(buckets[0x0].size(), 1u);
-    EXPECT_EQ(buckets[0x5].size(), 1u);
-    EXPECT_EQ(buckets[0xF].size(), 1u);
-    for (std::size_t i = 0; i < 16; ++i)
-        if (i != 0x0 && i != 0x5 && i != 0xF)
-            EXPECT_TRUE(buckets[i].empty()) << "Bucket " << i << " unexpectedly populated";
-}
-
-TEST(DeferredRebuild_Partition, MultipleKeysShareBuckets)
-{
-    auto const keyA = keyWithPrefix({0x3, 0x1});
-    auto const keyB = keyWithPrefix({0x3, 0x2});
-    auto const keyC = keyWithPrefix({0x7, 0x0});
-
-    auto const buckets = partitionByFirstNibble({keyA, keyB, keyC});
-
-    EXPECT_EQ(buckets[0x3].size(), 2u);
-    EXPECT_EQ(buckets[0x7].size(), 1u);
-}
-
-TEST(DeferredRebuild_Parallel, EmptyKeysMatchSerial)
-{
-    auto const parentLookup = [](int depth, uint256 const& prefix) {
-        if (depth == 0 && prefix == uint256{})
-            return uint256{0xDEAD};
-        return uint256{};
-    };
-
-    EXPECT_EQ(
-        deferredRebuildRootParallel({}, parentLookup),
-        deferredRebuildRoot({}, parentLookup));
-}
-
-TEST(DeferredRebuild_Parallel, SingleKeyMatchesSerial)
-{
-    std::vector<uint256> keys{keyOf(42)};
-
-    auto const parentLookup = [](int depth, uint256 const& prefix) {
-        std::uint64_t v = static_cast<std::uint64_t>(depth) * 0x1000;
-        for (int i = 0; i < 4; ++i)
-            v += prefix.data()[i];
-        return uint256{v};
-    };
-
-    EXPECT_EQ(
-        deferredRebuildRootParallel(keys, parentLookup),
-        deferredRebuildRoot(keys, parentLookup));
-}
-
-TEST(DeferredRebuild_Parallel, ManyKeysAcrossManySubtreesMatchSerial)
-{
-    // Keys spread across all 16 first-nibble buckets to exercise the
-    // multi-subtree case.
-    std::vector<uint256> keys;
-    for (std::uint64_t n = 0; n < 16; ++n)
-    {
-        for (std::uint64_t j = 0; j < 30; ++j)
-        {
-            // First nibble = n, rest scattered
-            auto k = keyWithPrefix({static_cast<std::uint8_t>(n)});
-            // Mix in some entropy for nibbles 1+
-            for (int i = 1; i < 8; ++i)
-                k.data()[i / 2] ^= static_cast<std::uint8_t>(j * 0x37 + i);
-            keys.push_back(k);
-        }
-    }
-
-    auto const parentLookup = [](int depth, uint256 const& prefix) {
-        std::uint64_t v = static_cast<std::uint64_t>(depth);
-        for (int i = 0; i < 8; ++i)
-            v = v * 257 + prefix.data()[i];
-        return uint256{v};
-    };
-
-    auto const serialRoot = deferredRebuildRoot(keys, parentLookup);
-    auto const parallelRoot = deferredRebuildRootParallel(keys, parentLookup);
-    EXPECT_EQ(parallelRoot, serialRoot);
-}
-
-TEST(DeferredRebuild_Parallel, AllKeysInSingleSubtreeMatchSerial)
-{
-    // Adversarial case: every key has the same first nibble, so 15
-    // subtrees are empty and the workload doesn't parallelize. The
-    // parallel implementation must still produce the same result.
-    std::vector<uint256> keys;
-    for (std::uint64_t j = 0; j < 50; ++j)
-    {
-        auto k = keyWithPrefix({0x7});  // all keys start with 7
-        for (int i = 1; i < 8; ++i)
-            k.data()[i / 2] ^= static_cast<std::uint8_t>(j * 0x11 + i);
-        keys.push_back(k);
-    }
-
-    auto const parentLookup = [](int depth, uint256 const& prefix) {
-        std::uint64_t v = static_cast<std::uint64_t>(depth) * 0xABCDEF;
-        for (int i = 0; i < 8; ++i)
-            v += prefix.data()[i];
-        return uint256{v};
-    };
-
-    EXPECT_EQ(
-        deferredRebuildRootParallel(keys, parentLookup),
-        deferredRebuildRoot(keys, parentLookup));
-}
-
-TEST(DeferredRebuild_Bench, EndToEndAtLedgerScale)
-{
-    // Realistic mainnet workload: ~3000 modifications per ledger.
-    // Measure the full plan + execute pipeline as a single number,
-    // since that's what production close-time will pay.
-    //
-    // The parent lookup is a constant-time computation — in production
-    // it's a SHAMap walk, which dominates the cost. This benchmark
-    // measures only the algorithmic overhead of the deferred rebuild
-    // itself, isolated from SHAMap traversal cost.
-    constexpr std::size_t N = 3'000;
-    std::vector<uint256> keys;
-    keys.reserve(N);
-    for (std::uint64_t i = 0; i < N; ++i)
-        keys.push_back(keyOf(i));
-
-    auto const parentLookup = [](int depth, uint256 const& prefix) {
-        return uint256{
-            static_cast<std::uint64_t>(depth) * 0xABCDEF +
-            prefix.data()[0] * 0x100 + prefix.data()[1]};
-    };
-
-    auto const t0 = std::chrono::high_resolution_clock::now();
-    auto const newRoot = deferredRebuildRoot(keys, parentLookup);
-    auto const elapsed = std::chrono::high_resolution_clock::now() - t0;
-    auto const us =
-        std::chrono::duration_cast<std::chrono::microseconds>(elapsed).count();
-
-    std::printf(
-        "  deferredRebuildRoot N=%zu keys : %lld µs total\n",
-        N,
-        static_cast<long long>(us));
-
-    benchmark_use(newRoot);
-
-    // Post-LCP: measured ~2.3 ms locally — comfortably within
-    // plan-7's ~7 ms close-time budget. 25 ms threshold = ~10×.
-    EXPECT_LT(us, 25'000)
-        << "End-to-end rebuild at 3k keys exceeded 25 ms (plan-7 budget ~7 ms)";
-}
-
-TEST(DeferredRebuild_Bench, ParallelVsSerialAtLedgerScale)
-{
-    constexpr std::size_t N = 3'000;
-    std::vector<uint256> keys;
-    keys.reserve(N);
-    // Spread keys across first-nibble buckets so all 16 subtrees see work.
-    for (std::uint64_t i = 0; i < N; ++i)
-    {
-        auto k = keyOf(i);
-        // Force first nibble to vary by i % 16
-        k.data()[0] = (k.data()[0] & 0x0F) |
-            static_cast<std::uint8_t>((i % 16) << 4);
-        keys.push_back(k);
-    }
-
-    auto const parentLookup = [](int depth, uint256 const& prefix) {
-        return uint256{
-            static_cast<std::uint64_t>(depth) * 0xABCDEF +
-            prefix.data()[0] * 0x100 + prefix.data()[1]};
-    };
-
-    auto const t0 = std::chrono::high_resolution_clock::now();
-    auto const serialRoot = deferredRebuildRoot(keys, parentLookup);
-    auto const t1 = std::chrono::high_resolution_clock::now();
-    auto const parallelRoot = deferredRebuildRootParallel(keys, parentLookup);
-    auto const t2 = std::chrono::high_resolution_clock::now();
-
-    auto const serialUs =
-        std::chrono::duration_cast<std::chrono::microseconds>(t1 - t0).count();
-    auto const parallelUs =
-        std::chrono::duration_cast<std::chrono::microseconds>(t2 - t1).count();
-
-    std::printf(
-        "  serial   : %lld µs\n"
-        "  parallel : %lld µs  (speedup %.2fx)\n",
-        static_cast<long long>(serialUs),
-        static_cast<long long>(parallelUs),
-        static_cast<double>(serialUs) / std::max<long long>(1, parallelUs));
-
-    EXPECT_EQ(serialRoot, parallelRoot);
-    // Parallel must not be slower than serial by more than 2× (which
-    // would indicate the threading overhead dominates the work and
-    // we're shipping the wrong implementation).
-    EXPECT_LT(parallelUs, serialUs * 2)
-        << "Parallel slower than 2× serial — threading overhead unexpected";
-}

src/tests/libxrpl/ledger/FlatStateMap.cpp

@@ -1,995 +0,0 @@
-#include <xrpl/basics/base_uint.h>
-#include <xrpl/ledger/FlatStateMap.h>
-#include <xrpl/protocol/Indexes.h>
-#include <xrpl/protocol/Keylet.h>
-#include <xrpl/protocol/LedgerFormats.h>
-#include <xrpl/protocol/STLedgerEntry.h>
-
-#include <gtest/gtest.h>
-
-#include <algorithm>
-#include <atomic>
-#include <chrono>
-#include <cstdio>
-#include <memory>
-#include <random>
-#include <thread>
-#include <vector>
-
-using namespace xrpl;
-
-namespace {
-
-// Construct a synthetic SLE for testing. The contents are not meaningful —
-// we only need a distinct shared_ptr<STLedgerEntry const> per key to exercise
-// the map's storage and retrieval semantics.
-[[nodiscard]] std::shared_ptr<STLedgerEntry const>
-makeSle(std::uint64_t keyValue)
-{
-    uint256 key{keyValue};
-    return std::make_shared<STLedgerEntry const>(ltACCOUNT_ROOT, key);
-}
-
-[[nodiscard]] uint256
-keyOf(std::uint64_t v)
-{
-    return uint256{v};
-}
-
-}  // namespace
-
-// ---------------------------------------------------------------------------
-// Basic read/write semantics
-// ---------------------------------------------------------------------------
-
-TEST(FlatStateMap, EmptyOnConstruction)
-{
-    FlatStateMap m;
-    EXPECT_TRUE(m.empty());
-    EXPECT_EQ(m.size(), 0u);
-    EXPECT_FALSE(m.exists(keyOf(0)));
-    EXPECT_EQ(m.read(keyOf(0)), nullptr);
-}
-
-TEST(FlatStateMap, InsertThenRead)
-{
-    FlatStateMap m;
-    auto const sle = makeSle(42);
-
-    m.insert(keyOf(42), sle);
-
-    EXPECT_FALSE(m.empty());
-    EXPECT_EQ(m.size(), 1u);
-    EXPECT_TRUE(m.exists(keyOf(42)));
-    EXPECT_EQ(m.read(keyOf(42)), sle);
-}
-
-TEST(FlatStateMap, ReadMissReturnsNullptr)
-{
-    FlatStateMap m;
-    m.insert(keyOf(1), makeSle(1));
-    EXPECT_EQ(m.read(keyOf(2)), nullptr);
-    EXPECT_FALSE(m.exists(keyOf(2)));
-}
-
-TEST(FlatStateMap, InsertReplacesPriorEntry)
-{
-    // update() semantics in apply: mutating an SLE produces a new shared_ptr
-    // value; insert() must replace the prior pointer cleanly.
-    FlatStateMap m;
-    auto const first = makeSle(1);
-    auto const second = makeSle(1);  // same key, different SLE object
-
-    m.insert(keyOf(1), first);
-    EXPECT_EQ(m.read(keyOf(1)), first);
-
-    m.insert(keyOf(1), second);
-    EXPECT_EQ(m.size(), 1u);
-    EXPECT_EQ(m.read(keyOf(1)), second);
-    EXPECT_NE(m.read(keyOf(1)), first);
-}
-
-TEST(FlatStateMap, EraseRemovesEntry)
-{
-    FlatStateMap m;
-    m.insert(keyOf(7), makeSle(7));
-    EXPECT_TRUE(m.exists(keyOf(7)));
-
-    m.erase(keyOf(7));
-    EXPECT_FALSE(m.exists(keyOf(7)));
-    EXPECT_EQ(m.read(keyOf(7)), nullptr);
-    EXPECT_TRUE(m.empty());
-}
-
-TEST(FlatStateMap, EraseAbsentKeyIsNoop)
-{
-    FlatStateMap m;
-    m.insert(keyOf(1), makeSle(1));
-    m.erase(keyOf(99));  // no-op; must not throw, must not affect other keys
-    EXPECT_EQ(m.size(), 1u);
-    EXPECT_TRUE(m.exists(keyOf(1)));
-}
-
-TEST(FlatStateMap, Clear)
-{
-    FlatStateMap m;
-    for (std::uint64_t i = 0; i < 100; ++i)
-        m.insert(keyOf(i), makeSle(i));
-    EXPECT_EQ(m.size(), 100u);
-
-    m.clear();
-    EXPECT_TRUE(m.empty());
-    EXPECT_EQ(m.size(), 0u);
-    EXPECT_FALSE(m.exists(keyOf(50)));
-}
-
-// ---------------------------------------------------------------------------
-// Iteration
-// ---------------------------------------------------------------------------
-
-TEST(FlatStateMap, ForEachVisitsAllEntries)
-{
-    FlatStateMap m;
-    std::vector<std::shared_ptr<STLedgerEntry const>> inserted;
-    constexpr std::size_t N = 50;
-    for (std::uint64_t i = 0; i < N; ++i)
-    {
-        auto sle = makeSle(i);
-        inserted.push_back(sle);
-        m.insert(keyOf(i), sle);
-    }
-
-    std::size_t visited = 0;
-    m.forEach([&](uint256 const& /*key*/, auto const& /*sle*/) { ++visited; });
-    EXPECT_EQ(visited, N);
-}
-
-// ---------------------------------------------------------------------------
-// Snapshot semantics
-// ---------------------------------------------------------------------------
-
-TEST(FlatStateMap, SnapshotPreservesEntries)
-{
-    FlatStateMap source;
-    for (std::uint64_t i = 0; i < 10; ++i)
-        source.insert(keyOf(i), makeSle(i));
-
-    auto snap = source.snapshot();
-    ASSERT_NE(snap, nullptr);
-    EXPECT_EQ(snap->size(), 10u);
-    for (std::uint64_t i = 0; i < 10; ++i)
-    {
-        ASSERT_TRUE(snap->exists(keyOf(i)));
-        EXPECT_EQ(snap->read(keyOf(i)), source.read(keyOf(i)));  // shared SLE
-    }
-}
-
-TEST(FlatStateMap, SnapshotIsIndependentOfSubsequentWrites)
-{
-    FlatStateMap source;
-    source.insert(keyOf(1), makeSle(1));
-    source.insert(keyOf(2), makeSle(2));
-
-    auto snap = source.snapshot();
-
-    // Mutate source after snapshot.
-    source.insert(keyOf(3), makeSle(3));
-    source.erase(keyOf(1));
-    source.insert(keyOf(2), makeSle(99));  // replace key 2 with a different SLE
-
-    // Snapshot must reflect state at snapshot time, not source's current state.
-    EXPECT_EQ(snap->size(), 2u);
-    EXPECT_TRUE(snap->exists(keyOf(1)));
-    EXPECT_TRUE(snap->exists(keyOf(2)));
-    EXPECT_FALSE(snap->exists(keyOf(3)));
-    EXPECT_NE(snap->read(keyOf(2)), source.read(keyOf(2)));
-}
-
-TEST(FlatStateMap, SnapshotSharesUnderlyingSleObjects)
-{
-    // Snapshot performs a shallow copy of the map (shared_ptr values).
-    // It does NOT deep-copy SLE bodies; both source and snapshot point at
-    // the same immutable SLE instance.
-    FlatStateMap source;
-    auto const sle = makeSle(1);
-    source.insert(keyOf(1), sle);
-
-    auto snap = source.snapshot();
-    EXPECT_EQ(snap->read(keyOf(1)).get(), sle.get());
-    EXPECT_EQ(source.read(keyOf(1)).get(), sle.get());
-}
-
-// ---------------------------------------------------------------------------
-// Ownership: FlatStateMap is non-copyable and non-movable (owns a mutex).
-// Callers that need ownership transfer wrap in std::unique_ptr<FlatStateMap>.
-// ---------------------------------------------------------------------------
-
-TEST(FlatStateMap, UniquePtrOwnershipTransfer)
-{
-    auto a = std::make_unique<FlatStateMap>();
-    a->insert(keyOf(1), makeSle(1));
-    a->insert(keyOf(2), makeSle(2));
-
-    auto b = std::move(a);  // pointer move, not map move
-    ASSERT_NE(b, nullptr);
-    EXPECT_EQ(a, nullptr);  // a is now null
-    EXPECT_EQ(b->size(), 2u);
-    EXPECT_TRUE(b->exists(keyOf(1)));
-    EXPECT_TRUE(b->exists(keyOf(2)));
-}
-
-// ---------------------------------------------------------------------------
-// Population from a range of SLEs (P6.2). The ReadView-taking overload
-// `populateFromReadView` is the same one-line forwarder; we cover the
-// templated range form directly so the test doesn't need a live ReadView.
-// ---------------------------------------------------------------------------
-
-TEST(FlatStateMap, PopulateFromRange)
-{
-    std::vector<std::shared_ptr<STLedgerEntry const>> sles;
-    constexpr std::size_t N = 25;
-    for (std::uint64_t i = 0; i < N; ++i)
-        sles.push_back(makeSle(i));
-
-    FlatStateMap m;
-    populateFromRange(m, sles);
-
-    EXPECT_EQ(m.size(), N);
-    for (std::size_t i = 0; i < N; ++i)
-    {
-        ASSERT_TRUE(m.exists(sles[i]->key()));
-        EXPECT_EQ(m.read(sles[i]->key()).get(), sles[i].get());
-    }
-}
-
-TEST(FlatStateMap, PopulateFromRangeOnEmptyRangeLeavesMapEmpty)
-{
-    std::vector<std::shared_ptr<STLedgerEntry const>> empty;
-    FlatStateMap m;
-    populateFromRange(m, empty);
-    EXPECT_TRUE(m.empty());
-}
-
-TEST(FlatStateMap, PopulateFromRangePreservesSleIdentity)
-{
-    // The map must store the exact shared_ptr the caller provided —
-    // not a deep copy of the SLE. This matters because SLE objects are
-    // logically immutable; any "copy" would risk subtle observer drift.
-    std::vector<std::shared_ptr<STLedgerEntry const>> sles{makeSle(1)};
-    auto const expected = sles[0].get();
-
-    FlatStateMap m;
-    populateFromRange(m, sles);
-
-    EXPECT_EQ(m.read(sles[0]->key()).get(), expected);
-}
-
-// ---------------------------------------------------------------------------
-// Dual-write mirroring (P6.3 from plan-6).
-//
-// In plan-6's 2-writes-for-1-read pattern, every state mutation must go to
-// both the SHAMap (authoritative for the state root) and the FlatStateMap
-// (read-side materialization). The integration point in xrpld is the
-// RawView interface — every state mutation goes through one of three
-// methods: rawInsert(sle), rawReplace(sle), or rawErase(sle).
-//
-// `mirrorRawInsert/mirrorRawReplace/mirrorRawErase` are the testable
-// units that perform the flat-map side of the dual write. They take a
-// FlatStateMap and an SLE (or key, for erase) and update the map to
-// reflect the operation. The Ledger integration (a separate change)
-// wires each `raw*` override to call the matching `mirror*` helper.
-//
-// These tests describe the contract:
-//   * mirrorRawInsert(map, sle)  — adds the sle keyed by sle->key()
-//   * mirrorRawReplace(map, sle) — replaces the sle for sle->key()
-//   * mirrorRawErase(map, sle)   — removes sle->key() from the map
-//   * mirrorRawErase(map, key)   — removes the key from the map
-//
-// All four are write-side ops; they take a unique_lock under the hood.
-// ---------------------------------------------------------------------------
-
-TEST(FlatStateMap_Mirror, MirrorRawInsertAddsEntry)
-{
-    FlatStateMap map;
-    auto const sle = makeSle(1);
-
-    mirrorRawInsert(map, sle);
-
-    EXPECT_EQ(map.size(), 1u);
-    EXPECT_EQ(map.read(sle->key()), sle);
-}
-
-TEST(FlatStateMap_Mirror, MirrorRawReplaceReplacesEntry)
-{
-    FlatStateMap map;
-    auto const original = makeSle(1);
-    auto const replacement = makeSle(1);  // same key, distinct object
-    map.insert(original->key(), original);
-
-    mirrorRawReplace(map, replacement);
-
-    EXPECT_EQ(map.size(), 1u);
-    EXPECT_EQ(map.read(original->key()), replacement);
-    EXPECT_NE(map.read(original->key()), original);
-}
-
-TEST(FlatStateMap_Mirror, MirrorRawEraseBySleRemovesEntry)
-{
-    FlatStateMap map;
-    auto const sle = makeSle(1);
-    map.insert(sle->key(), sle);
-
-    mirrorRawErase(map, sle);
-
-    EXPECT_TRUE(map.empty());
-    EXPECT_FALSE(map.exists(sle->key()));
-}
-
-TEST(FlatStateMap_Mirror, MirrorRawEraseByKeyRemovesEntry)
-{
-    FlatStateMap map;
-    auto const sle = makeSle(1);
-    map.insert(sle->key(), sle);
-
-    mirrorRawErase(map, sle->key());
-
-    EXPECT_TRUE(map.empty());
-}
-
-TEST(FlatStateMap_Mirror, MirrorOpsAreNoopOnAbsentKeys)
-{
-    FlatStateMap map;
-    // Erasing keys not in the map must not throw and must not alter the map.
-    mirrorRawErase(map, keyOf(99));
-    EXPECT_TRUE(map.empty());
-
-    // Replacing a key not in the map is semantically equivalent to an
-    // insert: in xrpld, rawReplace asserts the prior SLE exists in the
-    // SHAMap, so the SHAMap side handles the precondition. The flat
-    // mirror is permissive — it ensures the post-state matches what the
-    // SHAMap will have. If the caller upstream got it wrong, the
-    // differential invariant check at close (P6.5) is what catches it.
-    auto const sle = makeSle(5);
-    mirrorRawReplace(map, sle);
-    EXPECT_EQ(map.size(), 1u);
-    EXPECT_EQ(map.read(sle->key()), sle);
-}
-
-TEST(FlatStateMap_Mirror, MirroredSequenceMatchesIntendedState)
-{
-    // Simulate a sequence of raw operations as they would happen during
-    // a transaction's apply path, and assert the flat map ends in the
-    // state matching the SHAMap-equivalent view.
-    FlatStateMap map;
-
-    auto const a = makeSle(1);
-    auto const b = makeSle(2);
-    auto const c = makeSle(3);
-    auto const aPrime = makeSle(1);  // updated version of a
-
-    mirrorRawInsert(map, a);
-    mirrorRawInsert(map, b);
-    mirrorRawInsert(map, c);
-    mirrorRawReplace(map, aPrime);
-    mirrorRawErase(map, b);
-
-    // Expected end state: { 1 -> aPrime, 3 -> c }
-    EXPECT_EQ(map.size(), 2u);
-    EXPECT_EQ(map.read(a->key()), aPrime);
-    EXPECT_FALSE(map.exists(b->key()));
-    EXPECT_EQ(map.read(c->key()), c);
-}
-
-// ---------------------------------------------------------------------------
-// Keylet-aware read (P6.4).
-//
-// `readFromFlatStateMap(map, keylet)` is the testable unit underlying
-// the `Ledger::read(Keylet)` integration. It performs three steps:
-//   1. lookup by keylet.key in the FlatStateMap
-//   2. if missing, return nullptr (no SLE under that key)
-//   3. if present, verify the SLE matches the keylet's expected type via
-//      Keylet::check; on mismatch, return nullptr
-//
-// The type check mirrors the existing Ledger::read behavior — a keylet
-// query for the wrong type returns nullptr, not the wrong-typed SLE.
-// This preserves the contract: callers ask "is there an X at this key?"
-// and the read either yields an X or yields nothing.
-// ---------------------------------------------------------------------------
-
-TEST(FlatStateMap_KeyletRead, HitReturnsSle)
-{
-    FlatStateMap map;
-    auto const sle = makeSle(1);
-    map.insert(sle->key(), sle);
-
-    Keylet const k{ltACCOUNT_ROOT, sle->key()};
-    auto const result = readFromFlatStateMap(map, k);
-    EXPECT_EQ(result, sle);
-}
-
-TEST(FlatStateMap_KeyletRead, MissReturnsNullptr)
-{
-    FlatStateMap map;
-    Keylet const k{ltACCOUNT_ROOT, keyOf(42)};
-
-    auto const result = readFromFlatStateMap(map, k);
-    EXPECT_EQ(result, nullptr);
-}
-
-TEST(FlatStateMap_KeyletRead, TypeMismatchReturnsNullptr)
-{
-    // SLE stored with ltACCOUNT_ROOT, queried as ltRIPPLE_STATE: must
-    // return nullptr, not the wrong-typed SLE.
-    FlatStateMap map;
-    auto const sle = makeSle(1);  // ltACCOUNT_ROOT (see makeSle helper)
-    map.insert(sle->key(), sle);
-
-    Keylet const wrongType{ltRIPPLE_STATE, sle->key()};
-    auto const result = readFromFlatStateMap(map, wrongType);
-    EXPECT_EQ(result, nullptr);
-}
-
-TEST(FlatStateMap_KeyletRead, AbsenceIsAuthoritativeUnderPlan6V2)
-{
-    // Plan 6 v2 semantics: when a FlatStateMap is the read source of
-    // truth, a miss IS the answer. No fallback. The differential
-    // invariant check at close (P6.5) is what makes this safe.
-    //
-    // This test exists to document the contract: a populated map that
-    // doesn't contain key K reports nullptr for K, period. No probing
-    // into a SHAMap or other source.
-    FlatStateMap map;
-    auto const sleA = makeSle(1);
-    auto const sleB = makeSle(2);
-    map.insert(sleA->key(), sleA);
-    map.insert(sleB->key(), sleB);
-
-    Keylet const absent{ltACCOUNT_ROOT, keyOf(99)};
-    auto const result = readFromFlatStateMap(map, absent);
-    EXPECT_EQ(result, nullptr);
-    // Map state unchanged (no implicit population on read miss).
-    EXPECT_EQ(map.size(), 2u);
-    EXPECT_FALSE(map.exists(absent.key));
-}
-
-// ---------------------------------------------------------------------------
-// Concurrency: many concurrent readers do not block each other; writes
-// interleave with reads safely. We're not benchmarking, just checking that
-// no race trips a sanitizer and that final state is consistent.
-// ---------------------------------------------------------------------------
-
-// ---------------------------------------------------------------------------
-// Differential invariant (P6.5).
-//
-// Once the flat map is the read source of truth (P6.4), the safety
-// property that lets the no-fallback design ship is: at every ledger
-// close, the flat map and the SHAMap must agree on which keys are
-// present. `diffFlatStateKeys(flat, sourceKeys)` performs that check —
-// returning the sets of (a) keys in the source that are missing from
-// the flat map and (b) keys in the flat map that aren't in the source.
-//
-// Both lists empty == invariant holds. Anything else is a stop-the-line
-// bug — the Ledger integration crashes rather than publishing a state
-// root that disagrees with reality.
-//
-// Content drift (right keys, wrong SLE bodies) is a separate, stronger
-// invariant. It's prevented by construction: the mirror helpers (tested
-// in isolation) write exactly the SLE the caller passed to raw*. If the
-// mirror helpers are correct and the wiring is correct, content can't
-// drift. P6.5 catches the membership-drift case where wiring is broken.
-// ---------------------------------------------------------------------------
-
-TEST(FlatStateMap_Diff, EmptyVsEmptyHasNoDiff)
-{
-    FlatStateMap flat;
-    std::vector<uint256> sourceKeys;
-
-    auto const diff = diffFlatStateKeys(flat, sourceKeys);
-    EXPECT_TRUE(diff.missingFromFlat.empty());
-    EXPECT_TRUE(diff.extraInFlat.empty());
-}
-
-TEST(FlatStateMap_Diff, IdenticalKeySetsHaveNoDiff)
-{
-    FlatStateMap flat;
-    std::vector<uint256> sourceKeys;
-    for (std::uint64_t i = 0; i < 50; ++i)
-    {
-        auto const sle = makeSle(i);
-        flat.insert(sle->key(), sle);
-        sourceKeys.push_back(sle->key());
-    }
-
-    auto const diff = diffFlatStateKeys(flat, sourceKeys);
-    EXPECT_TRUE(diff.missingFromFlat.empty());
-    EXPECT_TRUE(diff.extraInFlat.empty());
-}
-
-TEST(FlatStateMap_Diff, KeysInSourceButNotFlatAreFlagged)
-{
-    FlatStateMap flat;
-    auto const sleA = makeSle(1);
-    auto const sleB = makeSle(2);
-    auto const sleC = makeSle(3);
-    flat.insert(sleA->key(), sleA);
-    // sleB intentionally not in flat
-    flat.insert(sleC->key(), sleC);
-
-    std::vector<uint256> sourceKeys{sleA->key(), sleB->key(), sleC->key()};
-
-    auto const diff = diffFlatStateKeys(flat, sourceKeys);
-    ASSERT_EQ(diff.missingFromFlat.size(), 1u);
-    EXPECT_EQ(diff.missingFromFlat[0], sleB->key());
-    EXPECT_TRUE(diff.extraInFlat.empty());
-}
-
-TEST(FlatStateMap_Diff, KeysInFlatButNotSourceAreFlagged)
-{
-    FlatStateMap flat;
-    auto const sleA = makeSle(1);
-    auto const sleB = makeSle(2);
-    auto const sleC = makeSle(3);
-    flat.insert(sleA->key(), sleA);
-    flat.insert(sleB->key(), sleB);  // phantom — not in source
-    flat.insert(sleC->key(), sleC);
-
-    std::vector<uint256> sourceKeys{sleA->key(), sleC->key()};
-
-    auto const diff = diffFlatStateKeys(flat, sourceKeys);
-    EXPECT_TRUE(diff.missingFromFlat.empty());
-    ASSERT_EQ(diff.extraInFlat.size(), 1u);
-    EXPECT_EQ(diff.extraInFlat[0], sleB->key());
-}
-
-TEST(FlatStateMap_Diff, BothSidesFlaggedSimultaneously)
-{
-    FlatStateMap flat;
-    auto const inBoth = makeSle(1);
-    auto const onlyFlat = makeSle(2);
-    auto const onlySource = makeSle(3);
-    flat.insert(inBoth->key(), inBoth);
-    flat.insert(onlyFlat->key(), onlyFlat);
-
-    std::vector<uint256> sourceKeys{inBoth->key(), onlySource->key()};
-
-    auto const diff = diffFlatStateKeys(flat, sourceKeys);
-    ASSERT_EQ(diff.missingFromFlat.size(), 1u);
-    EXPECT_EQ(diff.missingFromFlat[0], onlySource->key());
-    ASSERT_EQ(diff.extraInFlat.size(), 1u);
-    EXPECT_EQ(diff.extraInFlat[0], onlyFlat->key());
-}
-
-TEST(FlatStateMap_Diff, FlatMapsAgreeReturnsTrueWhenNoDiff)
-{
-    // Convenience predicate built on diffFlatStateKeys for the hot
-    // path: at every close, the integration calls this. Returns true
-    // iff both diff lists are empty.
-    FlatStateMap flat;
-    std::vector<uint256> sourceKeys;
-    for (std::uint64_t i = 0; i < 10; ++i)
-    {
-        auto const sle = makeSle(i);
-        flat.insert(sle->key(), sle);
-        sourceKeys.push_back(sle->key());
-    }
-
-    EXPECT_TRUE(flatStateMapMatches(flat, sourceKeys));
-
-    // After divergence, must return false.
-    flat.erase(sourceKeys[0]);
-    EXPECT_FALSE(flatStateMapMatches(flat, sourceKeys));
-}
-
-// ---------------------------------------------------------------------------
-// SHAMap-like adapter (P6.5 integration).
-//
-// The Ledger integration of the differential invariant needs to compare
-// the FlatStateMap against the live SHAMap's key-set. SHAMap iterators
-// yield `SHAMapItem` objects, not raw `uint256`s — so we need a thin
-// adapter that walks any "SHAMap-like" range (anything with begin()/
-// end() yielding items with a `.key()` method) and feeds the keys
-// through `flatStateMapMatches`.
-//
-// This is the helper Ledger::validateFlatStateMapMatchesShaMap() will
-// call. It's testable here with a mock SHAMap-like, so the Ledger
-// integration becomes a one-line forwarder.
-// ---------------------------------------------------------------------------
-
-namespace {
-
-// Minimal mock that satisfies the contract:
-//   * iterable via begin/end
-//   * each element exposes a `key()` returning uint256
-struct MockShaMapItem
-{
-    uint256 k;
-    [[nodiscard]] uint256 const&
-    key() const noexcept
-    {
-        return k;
-    }
-};
-
-}  // namespace
-
-TEST(FlatStateMap_ShaMapAdapter, EmptyShaMapMatchesEmptyFlat)
-{
-    FlatStateMap flat;
-    std::vector<MockShaMapItem> shaMap;
-    EXPECT_TRUE(flatStateMapMatchesShaMap(flat, shaMap));
-}
-
-TEST(FlatStateMap_ShaMapAdapter, IdenticalContentsMatch)
-{
-    FlatStateMap flat;
-    std::vector<MockShaMapItem> shaMap;
-    for (std::uint64_t i = 0; i < 25; ++i)
-    {
-        auto const sle = makeSle(i);
-        flat.insert(sle->key(), sle);
-        shaMap.push_back({sle->key()});
-    }
-    EXPECT_TRUE(flatStateMapMatchesShaMap(flat, shaMap));
-}
-
-TEST(FlatStateMap_ShaMapAdapter, MissingFromFlatFails)
-{
-    FlatStateMap flat;
-    auto const sleA = makeSle(1);
-    flat.insert(sleA->key(), sleA);
-
-    std::vector<MockShaMapItem> shaMap{
-        {sleA->key()}, {keyOf(99)}};  // 99 is in shaMap, missing from flat
-    EXPECT_FALSE(flatStateMapMatchesShaMap(flat, shaMap));
-}
-
-TEST(FlatStateMap_ShaMapAdapter, PhantomInFlatFails)
-{
-    FlatStateMap flat;
-    auto const sleA = makeSle(1);
-    auto const phantom = makeSle(99);
-    flat.insert(sleA->key(), sleA);
-    flat.insert(phantom->key(), phantom);
-
-    std::vector<MockShaMapItem> shaMap{{sleA->key()}};  // phantom isn't there
-    EXPECT_FALSE(flatStateMapMatchesShaMap(flat, shaMap));
-}
-
-// ---------------------------------------------------------------------------
-// Benchmarks. TDD with benchmarks: assert performance regressions fail
-// the test, not just correctness regressions. Thresholds are set
-// generously (10x slack vs. measured locally) so CI on under-spec
-// machines doesn't flake. Reported numbers are printed so a real
-// regression shows up as a measured slowdown even before the threshold
-// trips.
-//
-// What we're proving:
-//   * read() is O(1) — average latency does not grow with map size
-//   * write throughput is bounded but not pathological
-//   * readFromFlatStateMap (the Keylet-typed read) adds negligible
-//     overhead over the bare map.read() call
-//
-// All benchmarks measure on a single thread; concurrent scaling is
-// covered by ConcurrentReadersAndWritersAreConsistent.
-// ---------------------------------------------------------------------------
-
-namespace {
-
-// Inhibit dead-code elimination of the value `v` in benchmark loops.
-// The empty inline-asm "uses" v as an input, forcing the compiler to
-// materialize it. Cheap; no observable side effect.
-template <typename T>
-inline void
-benchmark_use(T const& v)
-{
-#if defined(__clang__) || defined(__GNUC__)
-    asm volatile("" : : "r,m"(v) : "memory");
-#else
-    (void)v;
-#endif
-}
-
-struct BenchResult
-{
-    double nsPerOp;
-    std::size_t ops;
-};
-
-template <typename Fn>
-BenchResult
-timeOps(std::size_t opCount, Fn&& fn)
-{
-    auto const t0 = std::chrono::high_resolution_clock::now();
-    for (std::size_t i = 0; i < opCount; ++i)
-        fn(i);
-    auto const t1 = std::chrono::high_resolution_clock::now();
-    auto const elapsedNs =
-        std::chrono::duration_cast<std::chrono::nanoseconds>(t1 - t0).count();
-    return {static_cast<double>(elapsedNs) / static_cast<double>(opCount),
-            opCount};
-}
-
-void
-populate(FlatStateMap& m, std::size_t n)
-{
-    for (std::uint64_t i = 0; i < n; ++i)
-        m.insert(keyOf(i), makeSle(i));
-}
-
-}  // namespace
-
-TEST(FlatStateMap_Bench, ReadIsO1AtVariousSizes)
-{
-    // Measure read() average latency at three map sizes. With O(1)
-    // semantics (hash table), the per-op time should be roughly flat.
-    constexpr std::size_t kOpsPerRun = 100'000;
-    std::vector<std::size_t> sizes{1'000, 10'000, 100'000};
-    std::vector<double> nsPerOpAtSize;
-
-    for (auto const n : sizes)
-    {
-        FlatStateMap m;
-        populate(m, n);
-
-        // Shuffle the access pattern so we don't accidentally
-        // measure a sequential cache-friendly access pattern.
-        std::vector<uint256> keys;
-        keys.reserve(n);
-        for (std::uint64_t i = 0; i < n; ++i)
-            keys.push_back(keyOf(i));
-        std::mt19937_64 rng(12345);
-        std::shuffle(keys.begin(), keys.end(), rng);
-
-        auto const result = timeOps(kOpsPerRun, [&](std::size_t i) {
-            auto sle = m.read(keys[i % n]);
-            // Prevent the compiler from optimizing the read away.
-            benchmark_use(sle);
-        });
-
-        std::printf(
-            "  FlatStateMap::read at N=%zu : %.1f ns/op (%zu ops)\n",
-            n,
-            result.nsPerOp,
-            result.ops);
-        nsPerOpAtSize.push_back(result.nsPerOp);
-
-        // Regression gate: even on a slow CI box, hash-map reads of a
-        // 100k-entry map should be well under 1 µs. We pick 2 µs as a
-        // generous threshold (~10x measured local) to avoid flakes.
-        EXPECT_LT(result.nsPerOp, 2000.0)
-            << "Read latency at N=" << n << " exceeded 2 µs/op";
-    }
-
-    // O(1) sanity: read at 100k should not be more than 4x slower than
-    // read at 1k (cache effects + memory bandwidth give some headroom,
-    // but not a real log-factor). 4x is generous; tighten if it's
-    // stable in CI.
-    EXPECT_LT(nsPerOpAtSize.back(), nsPerOpAtSize.front() * 4.0)
-        << "Read latency grew super-constantly with map size — "
-        << "expected O(1), got " << nsPerOpAtSize.front() << " ns at N=1k vs "
-        << nsPerOpAtSize.back() << " ns at N=100k";
-}
-
-TEST(FlatStateMap_Bench, KeyletReadOverheadIsSmall)
-{
-    // readFromFlatStateMap adds a Keylet::check call on top of map.read.
-    // The overhead should be a small constant — well under 100 ns —
-    // because Keylet::check is just a type tag comparison.
-    constexpr std::size_t N = 10'000;
-    constexpr std::size_t kOpsPerRun = 100'000;
-    FlatStateMap m;
-    populate(m, N);
-
-    std::vector<Keylet> keylets;
-    keylets.reserve(N);
-    for (std::uint64_t i = 0; i < N; ++i)
-        keylets.emplace_back(ltACCOUNT_ROOT, keyOf(i));
-
-    auto const bare = timeOps(kOpsPerRun, [&](std::size_t i) {
-        auto sle = m.read(keylets[i % N].key);
-        benchmark_use(sle);
-    });
-
-    auto const wrapped = timeOps(kOpsPerRun, [&](std::size_t i) {
-        auto sle = readFromFlatStateMap(m, keylets[i % N]);
-        benchmark_use(sle);
-    });
-
-    std::printf(
-        "  bare map.read         : %.1f ns/op\n"
-        "  readFromFlatStateMap  : %.1f ns/op  (+%.1f ns)\n",
-        bare.nsPerOp,
-        wrapped.nsPerOp,
-        wrapped.nsPerOp - bare.nsPerOp);
-
-    EXPECT_LT(wrapped.nsPerOp - bare.nsPerOp, 500.0)
-        << "Keylet check added more overhead than expected";
-}
-
-TEST(FlatStateMap_Bench, WriteThroughput)
-{
-    // Insert throughput is bounded by the cost of an unordered_map
-    // insert under a unique_lock. We're not optimizing this; we're
-    // gating it so a regression in the lock or allocator shows up.
-    constexpr std::size_t N = 100'000;
-    FlatStateMap m;
-
-    std::vector<std::shared_ptr<STLedgerEntry const>> sles;
-    sles.reserve(N);
-    for (std::uint64_t i = 0; i < N; ++i)
-        sles.push_back(makeSle(i));
-
-    auto const result =
-        timeOps(N, [&](std::size_t i) { m.insert(sles[i]->key(), sles[i]); });
-
-    std::printf(
-        "  FlatStateMap::insert  : %.1f ns/op (%zu ops)\n",
-        result.nsPerOp,
-        result.ops);
-
-    EXPECT_LT(result.nsPerOp, 5000.0) << "Insert latency exceeded 5 µs/op";
-    EXPECT_EQ(m.size(), N);
-}
-
-TEST(FlatStateMap_Bench, SnapshotCostAtLedgerScale)
-{
-    // P6.6 runs snapshot() at every close — capturing the live open
-    // ledger's flat map as the immutable base for the new closed
-    // ledger. The cost is O(N) in entry count (shallow copy of N
-    // shared_ptrs) and must be a small fraction of the close budget.
-    //
-    // Mainnet target: ~10M SLEs. We measure at 100k here and report
-    // ns/entry so extrapolation is clear. With ~50 ns/entry, 10M
-    // ledger snapshots in ~500 ms — borderline; a persistent HAMT
-    // (Plan 6 follow-on) is the long-term answer if this proves too
-    // expensive.
-    constexpr std::size_t N = 100'000;
-    FlatStateMap source;
-    for (std::uint64_t i = 0; i < N; ++i)
-        source.insert(keyOf(i), makeSle(i));
-
-    auto const start = std::chrono::high_resolution_clock::now();
-    auto snap = source.snapshot();
-    auto const elapsed = std::chrono::high_resolution_clock::now() - start;
-    auto const ms =
-        std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count();
-    auto const nsPerEntry =
-        std::chrono::duration_cast<std::chrono::nanoseconds>(elapsed).count() /
-        static_cast<double>(N);
-
-    std::printf(
-        "  FlatStateMap::snapshot N=%zu : %lld ms total (%.1f ns/entry)\n",
-        N,
-        static_cast<long long>(ms),
-        nsPerEntry);
-
-    ASSERT_NE(snap, nullptr);
-    EXPECT_EQ(snap->size(), N);
-    EXPECT_LT(ms, 500)
-        << "Snapshot at 100k entries should complete under 500 ms";
-}
-
-TEST(FlatStateMap_Bench, DifferentialInvariantCheckIsCheap)
-{
-    // P6.5 runs the diff at every ledger close. If it's slow it adds
-    // latency to the close path, defeating the point of Plan 6.
-    // Threshold: a 100k-entry map must validate in well under 100 ms
-    // on a typical validator. Real mainnet has ~10M SLEs, so this
-    // extrapolates to ~10 s at 100M-mapping. That would be too slow
-    // for real deployment; we'll need a partial / incremental check
-    // for production scale, but at this layer we just want a bounded
-    // O(N) walk.
-    constexpr std::size_t N = 100'000;
-    FlatStateMap m;
-    std::vector<uint256> sourceKeys;
-    sourceKeys.reserve(N);
-    for (std::uint64_t i = 0; i < N; ++i)
-    {
-        auto const sle = makeSle(i);
-        m.insert(sle->key(), sle);
-        sourceKeys.push_back(sle->key());
-    }
-
-    auto const start = std::chrono::high_resolution_clock::now();
-    bool const ok = flatStateMapMatches(m, sourceKeys);
-    auto const elapsed = std::chrono::high_resolution_clock::now() - start;
-    auto const ms =
-        std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count();
-    auto const nsPerKey =
-        std::chrono::duration_cast<std::chrono::nanoseconds>(elapsed).count() /
-        static_cast<double>(N);
-
-    std::printf(
-        "  flatStateMapMatches N=%zu : %lld ms total (%.1f ns/key)\n",
-        N,
-        static_cast<long long>(ms),
-        nsPerKey);
-
-    EXPECT_TRUE(ok);
-    EXPECT_LT(ms, 200) << "Diff at 100k entries should complete under 200 ms";
-}
-
-TEST(FlatStateMap_Bench, MirrorOverheadOverDirectInsert)
-{
-    // mirrorRawInsert forwards to map.insert with an extra shared_ptr
-    // load to extract the key. The wrapper overhead should be near
-    // zero — within noise of the bare insert.
-    constexpr std::size_t N = 50'000;
-
-    std::vector<std::shared_ptr<STLedgerEntry const>> sles;
-    sles.reserve(N);
-    for (std::uint64_t i = 0; i < N; ++i)
-        sles.push_back(makeSle(i));
-
-    FlatStateMap direct;
-    auto const bareResult = timeOps(
-        N, [&](std::size_t i) { direct.insert(sles[i]->key(), sles[i]); });
-
-    FlatStateMap mirrored;
-    auto const mirrorResult =
-        timeOps(N, [&](std::size_t i) { mirrorRawInsert(mirrored, sles[i]); });
-
-    std::printf(
-        "  direct insert         : %.1f ns/op\n"
-        "  mirrorRawInsert       : %.1f ns/op  (%+.1f ns)\n",
-        bareResult.nsPerOp,
-        mirrorResult.nsPerOp,
-        mirrorResult.nsPerOp - bareResult.nsPerOp);
-
-    // Mirror wrapper should not double the insert cost; 50% slack is
-    // very generous given they do the same thing.
-    EXPECT_LT(mirrorResult.nsPerOp, bareResult.nsPerOp * 1.5)
-        << "mirrorRawInsert wrapper added more overhead than expected";
-    EXPECT_EQ(mirrored.size(), N);
-}
-
-// ---------------------------------------------------------------------------
-
-TEST(FlatStateMap, ConcurrentReadersAndWritersAreConsistent)
-{
-    FlatStateMap m;
-    constexpr std::size_t N = 1000;
-
-    // Pre-populate with even keys.
-    for (std::uint64_t i = 0; i < N; i += 2)
-        m.insert(keyOf(i), makeSle(i));
-
-    std::atomic<bool> stop{false};
-    std::atomic<std::uint64_t> readsObserved{0};
-
-    auto reader = [&] {
-        while (!stop.load(std::memory_order_relaxed))
-        {
-            for (std::uint64_t i = 0; i < N; i += 2)
-            {
-                if (m.exists(keyOf(i)))
-                    readsObserved.fetch_add(1, std::memory_order_relaxed);
-            }
-        }
-    };
-
-    auto writer = [&] {
-        // Insert odd keys; do not modify the even keys readers observe.
-        for (std::uint64_t i = 1; i < N; i += 2)
-            m.insert(keyOf(i), makeSle(i));
-    };
-
-    std::vector<std::thread> readers;
-    for (int i = 0; i < 4; ++i)
-        readers.emplace_back(reader);
-
-    std::thread w(writer);
-    w.join();
-    stop.store(true, std::memory_order_relaxed);
-    for (auto& r : readers)
-        r.join();
-
-    // Every pre-populated even key must still be present.
-    for (std::uint64_t i = 0; i < N; i += 2)
-        EXPECT_TRUE(m.exists(keyOf(i)));
-    // Every written odd key must be present.
-    for (std::uint64_t i = 1; i < N; i += 2)
-        EXPECT_TRUE(m.exists(keyOf(i)));
-    EXPECT_EQ(m.size(), N);
-    EXPECT_GT(readsObserved.load(), 0u);  // readers made progress
-}

src/tests/libxrpl/ledger/FlatStateMapIntegration.cpp

@@ -1,596 +0,0 @@
-// Plan 6 A-phase integration tests.
-//
-// Exercises the public `attachFlatStateMapTo(Ledger&)` helper —
-// the one explicit entry point a node or test integration uses to
-// turn on the flat-map read path for a given Ledger. After attach:
-//   * Ledger::flatStateMap() returns non-null
-//   * Ledger::validateFlatStateMapMatchesShaMap() returns true
-//   * Ledger::read(keylet) consults the flat map (verified by the
-//     existing P6.4 wiring + null-safety regression tests)
-//
-// These are real-Ledger tests, not data-structure unit tests. They
-// catch wiring issues the libxrpl/ledger/FlatStateMap.cpp tests
-// cannot — bugs that only surface when an actual Ledger walks its
-// SHAMap to build the flat map.
-
-#include <helpers/TestFamily.h>
-
-#include <xrpl/basics/UnorderedContainers.h>
-#include <xrpl/beast/utility/Journal.h>
-#include <xrpl/ledger/DeferredRebuild.h>
-#include <xrpl/ledger/FlatStateMap.h>
-#include <xrpl/ledger/Ledger.h>
-#include <xrpl/protocol/Fees.h>
-#include <xrpl/protocol/Indexes.h>
-#include <xrpl/protocol/LedgerFormats.h>
-#include <xrpl/protocol/Rules.h>
-#include <xrpl/protocol/STLedgerEntry.h>
-#include <xrpl/shamap/SHAMap.h>
-
-#include <gtest/gtest.h>
-
-#include <memory>
-#include <unordered_set>
-#include <vector>
-
-namespace xrpl::test {
-
-class FlatStateMapIntegration : public ::testing::Test
-{
-protected:
-    TestFamily family_{beast::Journal{beast::Journal::getNullSink()}};
-
-    [[nodiscard]] std::shared_ptr<Ledger>
-    makeGenesisLedger()
-    {
-        // Genesis ledger with default rules (no amendments enabled).
-        // This ledger is IMMUTABLE — suitable for read-side tests but
-        // not for raw{Insert,Replace,Erase} which require a mutable
-        // SHAMap. For write-side tests, use `makeMutableChildLedger`.
-        Rules const rules{std::unordered_set<uint256, beast::Uhash<>>{}};
-        Fees const fees{XRPAmount{10}, XRPAmount{10'000'000}, XRPAmount{2'000'000}};
-        std::vector<uint256> const amendments;
-
-        return std::make_shared<Ledger>(
-            kCreateGenesis, rules, fees, amendments, family_);
-    }
-
-    [[nodiscard]] std::shared_ptr<Ledger>
-    makeMutableChildLedger()
-    {
-        // Build a child Ledger atop genesis. Child ledgers are
-        // constructed mutable so the apply path can write to them.
-        // This matches the production lifecycle: closed ledger N is
-        // immutable; closed ledger N+1 is built from N (mutable until
-        // it's itself closed).
-        auto genesis = makeGenesisLedger();
-        return std::make_shared<Ledger>(
-            *genesis,
-            NetClock::time_point{NetClock::duration{0}});
-    }
-};
-
-TEST_F(FlatStateMapIntegration, NoFlatMapByDefault)
-{
-    auto const ledger = makeGenesisLedger();
-    EXPECT_EQ(ledger->flatStateMap(), nullptr);
-    EXPECT_TRUE(ledger->validateFlatStateMapMatchesShaMap());
-}
-
-TEST_F(FlatStateMapIntegration, AttachPopulatesAndMatches)
-{
-    auto const ledger = makeGenesisLedger();
-
-    attachFlatStateMapTo(*ledger);
-
-    auto const map = ledger->flatStateMap();
-    ASSERT_NE(map, nullptr);
-    EXPECT_TRUE(ledger->validateFlatStateMapMatchesShaMap());
-
-    // The flat map should have one entry per SLE in the ledger.
-    std::size_t countViaWalk = 0;
-    for (auto const& sle : ledger->sles)
-    {
-        (void)sle;
-        ++countViaWalk;
-    }
-    EXPECT_EQ(map->size(), countViaWalk);
-    EXPECT_GT(map->size(), 0u)
-        << "Genesis ledger should have at least amendment+fee SLEs";
-}
-
-TEST_F(FlatStateMapIntegration, ReadsAfterAttachReturnSameSles)
-{
-    auto const ledger = makeGenesisLedger();
-
-    // Collect SLE pointers via SHAMap-descent path (pre-attach).
-    std::vector<std::shared_ptr<SLE const>> preAttachReads;
-    for (auto const& sle : ledger->sles)
-        preAttachReads.push_back(sle);
-
-    attachFlatStateMapTo(*ledger);
-
-    // After attach, reading via the flat map must yield SLEs that
-    // serialize byte-equal to the originals.
-    for (auto const& origSle : preAttachReads)
-    {
-        auto const fromMap = ledger->flatStateMap()->read(origSle->key());
-        ASSERT_NE(fromMap, nullptr) << "Missing key: " << origSle->key();
-        EXPECT_EQ(fromMap->getFullText(), origSle->getFullText());
-    }
-}
-
-TEST_F(FlatStateMapIntegration, RepeatedAttachReplaces)
-{
-    auto const ledger = makeGenesisLedger();
-
-    attachFlatStateMapTo(*ledger);
-    auto const firstMap = ledger->flatStateMap();
-    ASSERT_NE(firstMap, nullptr);
-
-    attachFlatStateMapTo(*ledger);
-    auto const secondMap = ledger->flatStateMap();
-    ASSERT_NE(secondMap, nullptr);
-
-    EXPECT_NE(firstMap.get(), secondMap.get());
-    EXPECT_EQ(firstMap->size(), secondMap->size());
-    EXPECT_TRUE(ledger->validateFlatStateMapMatchesShaMap());
-}
-
-// ---------------------------------------------------------------------------
-// Write-path verification — the P6.3 mirror wiring through Ledger::raw*
-// ---------------------------------------------------------------------------
-
-namespace {
-
-// Construct a minimal SLE we can rawInsert into a Ledger for testing.
-// We use ltACCOUNT_ROOT with a unique key — content doesn't need to be
-// realistic; we're testing the mirror plumbing, not transactor logic.
-[[nodiscard]] std::shared_ptr<SLE>
-makeTestSle(std::uint64_t keyValue)
-{
-    uint256 const key{keyValue};
-    return std::make_shared<SLE>(ltACCOUNT_ROOT, key);
-}
-
-}  // namespace
-
-TEST_F(FlatStateMapIntegration, RawInsertMirrorsToFlatMap)
-{
-    auto const ledger = makeMutableChildLedger();
-    attachFlatStateMapTo(*ledger);
-    auto const map = ledger->flatStateMap();
-    ASSERT_NE(map, nullptr);
-
-    auto const sle = makeTestSle(0xDEAD'BEEFu);
-    auto const sizeBefore = map->size();
-
-    ledger->rawInsert(sle);
-
-    EXPECT_EQ(map->size(), sizeBefore + 1u);
-    auto const fromMap = map->read(sle->key());
-    ASSERT_NE(fromMap, nullptr);
-    EXPECT_EQ(fromMap.get(), sle.get())
-        << "Flat map should hold the exact same shared_ptr we inserted";
-
-    // Differential invariant must still hold after the write.
-    EXPECT_TRUE(ledger->validateFlatStateMapMatchesShaMap());
-}
-
-TEST_F(FlatStateMapIntegration, RawEraseMirrorsToFlatMap)
-{
-    auto const ledger = makeMutableChildLedger();
-    attachFlatStateMapTo(*ledger);
-    auto const map = ledger->flatStateMap();
-    ASSERT_NE(map, nullptr);
-
-    // Insert, then erase.
-    auto const sle = makeTestSle(0xCAFEu);
-    ledger->rawInsert(sle);
-    ASSERT_TRUE(map->exists(sle->key()));
-
-    ledger->rawErase(sle);
-
-    EXPECT_FALSE(map->exists(sle->key()))
-        << "Erase should clear the flat-map entry";
-    EXPECT_TRUE(ledger->validateFlatStateMapMatchesShaMap());
-}
-
-TEST_F(FlatStateMapIntegration, RawReplaceUpdatesFlatMap)
-{
-    auto const ledger = makeMutableChildLedger();
-    attachFlatStateMapTo(*ledger);
-    auto const map = ledger->flatStateMap();
-    ASSERT_NE(map, nullptr);
-
-    // Insert first version, then replace with a different SLE object
-    // at the same key.
-    auto const firstSle = makeTestSle(0xC0DEu);
-    ledger->rawInsert(firstSle);
-    auto const fromMapFirst = map->read(firstSle->key());
-    ASSERT_NE(fromMapFirst, nullptr);
-    EXPECT_EQ(fromMapFirst.get(), firstSle.get());
-
-    auto const secondSle = makeTestSle(0xC0DEu);  // same key
-    ASSERT_NE(firstSle.get(), secondSle.get())
-        << "Test setup: replacement SLE must be a distinct object";
-
-    ledger->rawReplace(secondSle);
-
-    auto const fromMapSecond = map->read(secondSle->key());
-    ASSERT_NE(fromMapSecond, nullptr);
-    EXPECT_EQ(fromMapSecond.get(), secondSle.get())
-        << "Replace should swap the flat-map pointer to the new SLE";
-    EXPECT_NE(fromMapSecond.get(), firstSle.get());
-
-    EXPECT_TRUE(ledger->validateFlatStateMapMatchesShaMap());
-}
-
-TEST_F(FlatStateMapIntegration, ManyWritesPreserveInvariant)
-{
-    auto const ledger = makeMutableChildLedger();
-    attachFlatStateMapTo(*ledger);
-    auto const map = ledger->flatStateMap();
-    ASSERT_NE(map, nullptr);
-
-    std::vector<std::shared_ptr<SLE>> sles;
-    for (std::uint64_t i = 0; i < 100; ++i)
-    {
-        auto sle = makeTestSle(0x1000u + i);
-        ledger->rawInsert(sle);
-        sles.push_back(sle);
-    }
-
-    // Erase every other one.
-    for (std::size_t i = 0; i < sles.size(); i += 2)
-        ledger->rawErase(sles[i]);
-
-    // Replace the rest.
-    for (std::size_t i = 1; i < sles.size(); i += 2)
-    {
-        auto replacement = makeTestSle(0x1000u + i);
-        ledger->rawReplace(replacement);
-    }
-
-    // After 100 inserts + 50 erases + 50 replaces, the invariant
-    // must still hold.
-    EXPECT_TRUE(ledger->validateFlatStateMapMatchesShaMap());
-}
-
-TEST_F(FlatStateMapIntegration, WritesBeforeAttachDontGetMirrored)
-{
-    // Sanity: if writes happen BEFORE the flat map is attached, they
-    // hit the SHAMap only. Then attach + validate populates from the
-    // SHAMap and the invariant holds. Catches the bootstrapping case.
-    auto const ledger = makeMutableChildLedger();
-    auto const sle = makeTestSle(0xBEEFu);
-    ledger->rawInsert(sle);
-
-    EXPECT_EQ(ledger->flatStateMap(), nullptr);
-
-    attachFlatStateMapTo(*ledger);
-
-    // After attach, the flat map sees the previously-inserted SLE
-    // because populate walked the SHAMap.
-    auto const map = ledger->flatStateMap();
-    ASSERT_NE(map, nullptr);
-    EXPECT_TRUE(map->exists(sle->key()));
-    EXPECT_TRUE(ledger->validateFlatStateMapMatchesShaMap());
-}
-
-// ---------------------------------------------------------------------------
-// Lifecycle propagation (P6.3): a child ledger built from a parent that
-// carries a flat-state mirror must inherit an INDEPENDENT snapshot — the
-// flat-map analogue of the SHAMap COW snapshot. This is what lets the flat
-// map survive across ledgers instead of dying after one round.
-// ---------------------------------------------------------------------------
-
-TEST_F(FlatStateMapIntegration, ChildWithoutParentMapInheritsNone)
-{
-    // Default lifecycle: parent has no flat map → child has none. Zero
-    // behavior change when the feature is not turned on.
-    auto const parent = makeGenesisLedger();
-    ASSERT_EQ(parent->flatStateMap(), nullptr);
-
-    auto const child = std::make_shared<Ledger>(
-        *parent, NetClock::time_point{NetClock::duration{0}});
-    EXPECT_EQ(child->flatStateMap(), nullptr);
-    EXPECT_TRUE(child->validateFlatStateMapMatchesShaMap());
-}
-
-TEST_F(FlatStateMapIntegration, ChildInheritsIndependentSnapshot)
-{
-    auto const parent = makeGenesisLedger();
-    attachFlatStateMapTo(*parent);
-    auto const parentMap = parent->flatStateMap();
-    ASSERT_NE(parentMap, nullptr);
-
-    auto const child = std::make_shared<Ledger>(
-        *parent, NetClock::time_point{NetClock::duration{0}});
-
-    auto const childMap = child->flatStateMap();
-    ASSERT_NE(childMap, nullptr);
-    // Distinct object, equal contents, and consistent with the child's own
-    // (COW-snapshotted) SHAMap.
-    EXPECT_NE(childMap.get(), parentMap.get());
-    EXPECT_EQ(childMap->size(), parentMap->size());
-    EXPECT_TRUE(child->validateFlatStateMapMatchesShaMap());
-}
-
-TEST_F(FlatStateMapIntegration, ChildWritesDoNotCorruptParentSnapshot)
-{
-    auto const parent = makeGenesisLedger();
-    attachFlatStateMapTo(*parent);
-    auto const parentMap = parent->flatStateMap();
-    auto const parentSizeBefore = parentMap->size();
-
-    auto const child = std::make_shared<Ledger>(
-        *parent, NetClock::time_point{NetClock::duration{0}});
-
-    // Mutate the child; the parent's snapshot must be untouched.
-    auto const sle = makeTestSle(0x5117'5157u);
-    child->rawInsert(sle);
-
-    EXPECT_TRUE(child->flatStateMap()->exists(sle->key()));
-    EXPECT_FALSE(parentMap->exists(sle->key()))
-        << "Child write leaked into the parent's flat map";
-    EXPECT_EQ(parentMap->size(), parentSizeBefore);
-    EXPECT_TRUE(child->validateFlatStateMapMatchesShaMap());
-    EXPECT_TRUE(parent->validateFlatStateMapMatchesShaMap());
-}
-
-TEST_F(FlatStateMapIntegration, PropagationChainsAcrossGenerations)
-{
-    // parent -> child -> grandchild, each inheriting + mutating. Every
-    // generation's invariant must hold and each map stays independent.
-    auto const parent = makeGenesisLedger();
-    attachFlatStateMapTo(*parent);
-
-    auto const child = std::make_shared<Ledger>(
-        *parent, NetClock::time_point{NetClock::duration{0}});
-    auto const cSle = makeTestSle(0xC111'D000u);
-    child->rawInsert(cSle);
-    ASSERT_TRUE(child->validateFlatStateMapMatchesShaMap());
-
-    auto const grandchild = std::make_shared<Ledger>(
-        *child, NetClock::time_point{NetClock::duration{0}});
-    // Grandchild inherits the child's mutation...
-    EXPECT_TRUE(grandchild->flatStateMap()->exists(cSle->key()));
-    auto const gSle = makeTestSle(0x6111'D000u);
-    grandchild->rawInsert(gSle);
-
-    EXPECT_TRUE(grandchild->validateFlatStateMapMatchesShaMap());
-    // ...but the grandchild's own write doesn't reach back up.
-    EXPECT_FALSE(child->flatStateMap()->exists(gSle->key()));
-    EXPECT_TRUE(child->validateFlatStateMapMatchesShaMap());
-}
-
-// ---------------------------------------------------------------------------
-// Read routing (P6.4): with a map attached, Ledger::read(Keylet) must take
-// the flat path and return results identical to the SHAMap-descent path.
-// ---------------------------------------------------------------------------
-
-TEST_F(FlatStateMapIntegration, LedgerReadFlatPathMatchesShaMapPath)
-{
-    auto const ledger = makeMutableChildLedger();
-    // Add a few typed SLEs so there's something to read.
-    std::vector<std::shared_ptr<SLE>> inserted;
-    for (std::uint64_t i = 0; i < 25; ++i)
-    {
-        auto sle = makeTestSle(0x7000u + i);
-        ledger->rawInsert(sle);
-        inserted.push_back(sle);
-    }
-
-    // Capture SHAMap-path reads BEFORE attaching the flat map.
-    std::vector<std::string> shaMapReads;
-    for (auto const& s : inserted)
-    {
-        auto const r = ledger->read(Keylet{s->getType(), s->key()});
-        ASSERT_NE(r, nullptr);
-        shaMapReads.push_back(r->getFullText());
-    }
-
-    attachFlatStateMapTo(*ledger);
-    ASSERT_NE(ledger->flatStateMap(), nullptr);
-
-    // Now reads route through the flat map and must match byte-for-byte.
-    for (std::size_t i = 0; i < inserted.size(); ++i)
-    {
-        auto const r = ledger->read(Keylet{inserted[i]->getType(), inserted[i]->key()});
-        ASSERT_NE(r, nullptr) << "flat read miss for key " << inserted[i]->key();
-        EXPECT_EQ(r->getFullText(), shaMapReads[i]);
-    }
-
-    // An absent key returns nullptr on the flat path too.
-    EXPECT_EQ(
-        ledger->read(Keylet{ltACCOUNT_ROOT, uint256{0xAB5E'0000ull}}), nullptr);
-}
-
-// ---------------------------------------------------------------------------
-// The close-time differential invariant (P6.5) must actually FAIL on drift,
-// not just pass on correct state — otherwise it's not a safety gate.
-// ---------------------------------------------------------------------------
-
-TEST_F(FlatStateMapIntegration, InvariantFailsOnPhantomEntry)
-{
-    auto const ledger = makeMutableChildLedger();
-    attachFlatStateMapTo(*ledger);
-    ASSERT_TRUE(ledger->validateFlatStateMapMatchesShaMap());
-
-    // Inject a flat-map entry the SHAMap does not have.
-    ledger->flatStateMap()->insert(
-        uint256{0xDEAD'0001ull}, makeTestSle(0xDEAD'0001ull));
-
-    EXPECT_FALSE(ledger->validateFlatStateMapMatchesShaMap())
-        << "Invariant must catch an entry present in flat but not SHAMap";
-}
-
-TEST_F(FlatStateMapIntegration, InvariantFailsOnMissingEntry)
-{
-    auto const ledger = makeMutableChildLedger();
-    auto const sle = makeTestSle(0xFEED'0001u);
-    ledger->rawInsert(sle);
-    attachFlatStateMapTo(*ledger);
-    ASSERT_TRUE(ledger->validateFlatStateMapMatchesShaMap());
-
-    // Drop an entry the SHAMap still has.
-    ledger->flatStateMap()->erase(sle->key());
-
-    EXPECT_FALSE(ledger->validateFlatStateMapMatchesShaMap())
-        << "Invariant must catch an entry present in SHAMap but not flat";
-}
-
-// ---------------------------------------------------------------------------
-// Plan 7 against a real Ledger's SHAMap (A-phase milestone 2)
-//
-// First end-to-end composition test: deferredRebuildRoot driven by a
-// callback that reads from an actual Ledger's stateMap. The empty-
-// modifications case is the smallest meaningful integration — the
-// rebuild trivially returns the current root, which must match the
-// SHAMap's root via getHash().
-//
-// Future slices extend this with a real after-modifications byte-
-// identical assertion (the merge gate the plan-7 doc requires for any
-// production ship).
-// ---------------------------------------------------------------------------
-
-TEST_F(FlatStateMapIntegration, Plan7EmptyDeltaProducesShaMapRoot)
-{
-    auto const ledger = makeMutableChildLedger();
-    // Add some state so the SHAMap root is non-trivial.
-    for (std::uint64_t i = 0; i < 20; ++i)
-        ledger->rawInsert(makeTestSle(0x2000u + i));
-
-    auto const shaMapRoot = ledger->stateMap().getHash().asUInt256();
-
-    // The callback only needs to handle (depth=0, prefix=zero) for
-    // the empty-delta case — it returns the SHAMap root.
-    auto const callback =
-        [&shaMapRoot](int depth, uint256 const& prefix) -> uint256 {
-        if (depth == 0 && prefix == uint256{})
-            return shaMapRoot;
-        // Any other position is a bug for the empty-delta path.
-        return uint256{};
-    };
-
-    auto const rebuiltRoot = deferredRebuildRoot({}, callback);
-    EXPECT_EQ(rebuiltRoot, shaMapRoot);
-}
-
-TEST_F(FlatStateMapIntegration, Plan7EmptyDeltaOnEmptyLedger)
-{
-    // Same as above but on a freshly-constructed mutable child with
-    // no extra state. The SHAMap root may already be non-trivial due
-    // to inherited genesis SLEs.
-    auto const ledger = makeMutableChildLedger();
-    auto const shaMapRoot = ledger->stateMap().getHash().asUInt256();
-
-    auto const callback =
-        [&shaMapRoot](int depth, uint256 const& prefix) -> uint256 {
-        if (depth == 0 && prefix == uint256{})
-            return shaMapRoot;
-        return uint256{};
-    };
-
-    EXPECT_EQ(deferredRebuildRoot({}, callback), shaMapRoot);
-}
-
-// ---------------------------------------------------------------------------
-// CRITICAL LIMITATION TEST — documents an architectural gap surfaced
-// by A-phase integration.
-//
-// Plan 7's `executeRebuildPlan` algorithm assumes leaves live at depth
-// 64 (a fully-expanded radix tree). REAL SHAMap uses PATH COMPRESSION:
-// leaves are stored at the shallowest depth where they're unambiguous
-// (see SHAMap.cpp addGiveItem — when the path hits an empty branch or
-// a leaf, the new leaf is placed at that depth, not deeper).
-//
-// Consequence: when actual SHAMap modifications cause new leaves to be
-// placed at depths < 64 (the common case), the SHAMap's root hash
-// computation differs from Plan 7's. They will NOT agree.
-//
-// Implications:
-//   * Plan 7 as implemented is NOT byte-identical to real SHAMap.
-//     The merge gate plan-7-deferred-shamap.md describes cannot
-//     close on this implementation.
-//   * Production deployment of Plan 7 requires either:
-//       (a) extending the algorithm to handle path compression (know
-//           where each leaf lives in the parent tree; account for
-//           new inner-node creation at split points), OR
-//       (b) replacing SHAMap with a non-path-compressed structure
-//           (a much bigger amendment-class change).
-//   * Option (a) is substantially more complex than the current
-//     callback-based design — the algorithm needs to track tree
-//     topology, not just position-keyed hashes.
-//
-// This test asserts the discrepancy explicitly so future readers see
-// the issue. The library code (planDeferredRebuild, computeInnerNodeHash,
-// executeRebuildPlan, deferredRebuildRoot, parallel variants) remains
-// in place as a working kernel for the depth-64-leaves model, which
-// is still useful as a reference and starting point for the refactor.
-// ---------------------------------------------------------------------------
-
-TEST_F(
-    FlatStateMapIntegration,
-    Plan7DoesNotMatchPathCompressedShaMap_DocumentedLimitation)
-{
-    // Build a Ledger with a few SLEs whose keys differ in their high
-    // nibbles (forces SHAMap path compression — they're stored as
-    // direct children of root or shallow inner nodes).
-    auto const ledger = makeMutableChildLedger();
-    std::vector<std::shared_ptr<SLE>> sles;
-    for (std::uint64_t i = 0; i < 5; ++i)
-    {
-        // Spread keys across the top of the tree.
-        uint256 k;
-        k.data()[0] = static_cast<std::uint8_t>(i << 4);
-        sles.push_back(std::make_shared<SLE>(ltACCOUNT_ROOT, k));
-        ledger->rawInsert(sles.back());
-    }
-
-    auto const realShaMapRoot = ledger->stateMap().getHash().asUInt256();
-
-    // Construct a Plan-7 rebuild treating these leaves as if they
-    // lived at depth 64. The callback returns:
-    //   - depth 64: the leaf hash for the modified key (else zero)
-    //   - other depths: zero (assume empty parent)
-    std::vector<uint256> modKeys;
-    for (auto const& s : sles)
-        modKeys.push_back(s->key());
-
-    // Pre-compute each leaf's SHAMap hash for the callback.
-    std::unordered_map<uint256, uint256, beast::Uhash<>> leafHashes;
-    for (auto const& s : sles)
-    {
-        uint256 const leafKey = s->key();
-        // We can ask the SHAMap for the leaf's actual hash
-        SHAMapHash itemHash;
-        auto const item = ledger->stateMap().peekItem(leafKey, itemHash);
-        if (item)
-            leafHashes[leafKey] = itemHash.asUInt256();
-    }
-
-    auto const plan7Callback =
-        [&leafHashes](int depth, uint256 const& prefix) -> uint256 {
-        if (depth == 64)
-        {
-            auto it = leafHashes.find(prefix);
-            if (it != leafHashes.end())
-                return it->second;
-        }
-        return uint256{};  // empty parent at non-leaf depths
-    };
-
-    auto const plan7Root = deferredRebuildRoot(modKeys, plan7Callback);
-
-    // Document the discrepancy. Plan 7's depth-64 model produces a
-    // different hash than path-compressed SHAMap — this MUST be the
-    // case until Plan 7 is refactored to handle path compression.
-    EXPECT_NE(plan7Root, realShaMapRoot)
-        << "If this assertion ever starts failing, Plan 7's path-"
-        << "compression limitation may have been fixed — update this "
-        << "test to assert equality and remove the documented limitation.";
-}
-
-}  // namespace xrpl::test

src/tests/libxrpl/ledger/OrderBookIndex.cpp

@@ -0,0 +1,175 @@
+#include <xrpl/ledger/OrderBookIndex.h>
+
+#include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/Asset.h>
+#include <xrpl/protocol/Book.h>
+#include <xrpl/protocol/Indexes.h>
+#include <xrpl/protocol/Issue.h>
+#include <xrpl/protocol/UintTypes.h>
+
+#include <gtest/gtest.h>
+
+namespace xrpl::test {
+
+namespace {
+
+// Synthetic-but-consistent IOU book (XRP <-> tagged currency), matching the
+// TopOfBookCache test helper so the two suites stay comparable.
+Book
+makeIOUBook(std::uint8_t tag)
+{
+    Currency c{};
+    c.data()[19] = tag;
+    AccountID issuer{};
+    issuer.data()[19] = tag;
+    Issue const inIssue{c, issuer};
+    return Book{Asset{inIssue}, Asset{Issue{xrpCurrency(), xrpAccount()}}, std::nullopt};
+}
+
+// Quality-directory root key for a book at a given rate. Lower rate => lower
+// key => better quality (the ordering the index relies on).
+uint256
+dirKey(Book const& book, std::uint64_t rate)
+{
+    return keylet::quality(keylet::kBook(book), rate).key;
+}
+
+// Arbitrary distinct offer key.
+uint256
+offerKey(std::uint8_t tag)
+{
+    uint256 k{};
+    k.data()[0] = tag;
+    return k;
+}
+
+}  // namespace
+
+TEST(OrderBookIndex, EmptyBook)
+{
+    OrderBookIndex idx;
+    Book const book = makeIOUBook(1);
+    EXPECT_TRUE(idx.flatten(book).empty());
+    EXPECT_FALSE(idx.firstOffer(book).has_value());
+    EXPECT_EQ(idx.bookCount(), 0u);
+    EXPECT_EQ(idx.offerCount(book), 0u);
+}
+
+TEST(OrderBookIndex, InsertWithinLevelPreservesAppendOrder)
+{
+    OrderBookIndex idx;
+    Book const book = makeIOUBook(2);
+    uint256 const lvl = dirKey(book, 1'000'000u);
+
+    idx.insertOffer(book, lvl, offerKey(1));
+    idx.insertOffer(book, lvl, offerKey(2));
+    idx.insertOffer(book, lvl, offerKey(3));
+
+    std::vector<uint256> const expect{offerKey(1), offerKey(2), offerKey(3)};
+    EXPECT_EQ(idx.flatten(book), expect);
+    EXPECT_EQ(idx.firstOffer(book), offerKey(1));
+    EXPECT_EQ(idx.offerCount(book), 3u);
+    EXPECT_EQ(idx.inserts(), 3u);
+}
+
+TEST(OrderBookIndex, LevelsOrderedBestQualityFirstRegardlessOfInsertOrder)
+{
+    OrderBookIndex idx;
+    Book const book = makeIOUBook(3);
+    uint256 const best = dirKey(book, 1'000'000u);
+    uint256 const mid = dirKey(book, 2'000'000u);
+    uint256 const worst = dirKey(book, 3'000'000u);
+    ASSERT_LT(best, mid);
+    ASSERT_LT(mid, worst);
+
+    // Insert worst-first to prove ordering is by quality, not insertion.
+    idx.insertOffer(book, worst, offerKey(30));
+    idx.insertOffer(book, best, offerKey(10));
+    idx.insertOffer(book, mid, offerKey(20));
+
+    std::vector<uint256> const expect{offerKey(10), offerKey(20), offerKey(30)};
+    EXPECT_EQ(idx.flatten(book), expect);
+    EXPECT_EQ(idx.firstOffer(book), offerKey(10));
+}
+
+TEST(OrderBookIndex, DeletePreservesOrderAndDropsEmptyLevel)
+{
+    OrderBookIndex idx;
+    Book const book = makeIOUBook(4);
+    uint256 const a = dirKey(book, 1'000u);
+    uint256 const b = dirKey(book, 2'000u);
+
+    idx.insertOffer(book, a, offerKey(1));
+    idx.insertOffer(book, a, offerKey(2));
+    idx.insertOffer(book, a, offerKey(3));
+    idx.insertOffer(book, b, offerKey(4));
+
+    // Remove a middle offer: relative order of the rest is preserved.
+    idx.deleteOffer(book, a, offerKey(2));
+    std::vector<uint256> const expect1{offerKey(1), offerKey(3), offerKey(4)};
+    EXPECT_EQ(idx.flatten(book), expect1);
+    EXPECT_EQ(idx.deletes(), 1u);
+
+    // Empty the first level: it is dropped, second becomes the front.
+    idx.deleteOffer(book, a, offerKey(1));
+    idx.deleteOffer(book, a, offerKey(3));
+    EXPECT_EQ(idx.firstOffer(book), offerKey(4));
+    EXPECT_EQ(idx.flatten(book), std::vector<uint256>{offerKey(4)});
+
+    // Empty the book entirely: it is removed from the index.
+    idx.deleteOffer(book, b, offerKey(4));
+    EXPECT_TRUE(idx.flatten(book).empty());
+    EXPECT_EQ(idx.bookCount(), 0u);
+}
+
+TEST(OrderBookIndex, DeleteAbsentIsNoOp)
+{
+    OrderBookIndex idx;
+    Book const book = makeIOUBook(5);
+    uint256 const lvl = dirKey(book, 1'000u);
+    idx.insertOffer(book, lvl, offerKey(1));
+
+    idx.deleteOffer(book, lvl, offerKey(99));        // absent key
+    idx.deleteOffer(book, dirKey(book, 9u), offerKey(1));  // absent level
+    idx.deleteOffer(makeIOUBook(6), lvl, offerKey(1));     // absent book
+
+    EXPECT_EQ(idx.flatten(book), std::vector<uint256>{offerKey(1)});
+    EXPECT_EQ(idx.deletes(), 0u);
+}
+
+TEST(OrderBookIndex, DistinctBooksIndependent)
+{
+    OrderBookIndex idx;
+    Book const a = makeIOUBook(7);
+    Book const b = makeIOUBook(8);
+
+    idx.insertOffer(a, dirKey(a, 100u), offerKey(1));
+    idx.insertOffer(b, dirKey(b, 100u), offerKey(2));
+    EXPECT_EQ(idx.bookCount(), 2u);
+
+    idx.eraseBook(a);
+    EXPECT_TRUE(idx.flatten(a).empty());
+    EXPECT_EQ(idx.flatten(b), std::vector<uint256>{offerKey(2)});
+    EXPECT_EQ(idx.bookCount(), 1u);
+}
+
+TEST(OrderBookIndex, ClearEmptiesEverything)
+{
+    OrderBookIndex idx;
+    Book const book = makeIOUBook(9);
+    idx.insertOffer(book, dirKey(book, 1u), offerKey(1));
+    idx.clear();
+    EXPECT_EQ(idx.bookCount(), 0u);
+    EXPECT_TRUE(idx.flatten(book).empty());
+}
+
+TEST(OrderBookIndex, KillSwitchToggleable)
+{
+    EXPECT_TRUE(OrderBookIndex::enabled());
+    OrderBookIndex::setEnabled(false);
+    EXPECT_FALSE(OrderBookIndex::enabled());
+    OrderBookIndex::setEnabled(true);
+    EXPECT_TRUE(OrderBookIndex::enabled());
+}
+
+}  // namespace xrpl::test

src/tests/libxrpl/ledger/PersistentOrderTree.cpp

@@ -0,0 +1,151 @@
+#include <xrpl/ledger/detail/PersistentOrderTree.h>
+
+#include <gtest/gtest.h>
+
+#include <cmath>
+#include <cstring>
+#include <map>
+#include <random>
+#include <utility>
+#include <vector>
+
+namespace xrpl::detail {
+
+namespace {
+
+// 256-bit value whose numeric order matches integer order (n written
+// big-endian into the low 8 bytes; base_uint compares MSB-first).
+uint256
+u256(std::uint64_t n)
+{
+    uint256 k;
+    std::memset(k.data(), 0, k.size());
+    auto* end = k.data() + k.size();
+    for (int i = 0; i < 8; ++i)
+        end[-1 - i] = static_cast<unsigned char>((n >> (8 * i)) & 0xff);
+    return k;
+}
+
+using RefKey = std::pair<uint256, std::uint64_t>;  // (dirRoot, insertSeq)
+
+// Reference inorder: std::map orders by (dirRoot, insertSeq); collect offers.
+std::vector<uint256>
+refInorder(std::map<RefKey, uint256> const& ref)
+{
+    std::vector<uint256> out;
+    out.reserve(ref.size());
+    for (auto const& [k, off] : ref)
+        out.push_back(off);
+    return out;
+}
+
+std::vector<uint256>
+treeInorder(OrderTreePtr const& t)
+{
+    std::vector<uint256> out;
+    otInorder(t, out);
+    return out;
+}
+
+int
+height(OrderTreePtr const& t)
+{
+    if (!t)
+        return 0;
+    return 1 + std::max(height(t->left), height(t->right));
+}
+
+// Verify subtree size fields are consistent.
+std::uint32_t
+checkSize(OrderTreePtr const& t)
+{
+    if (!t)
+        return 0;
+    auto const s = checkSize(t->left) + checkSize(t->right) + 1;
+    EXPECT_EQ(s, t->size);
+    return s;
+}
+
+}  // namespace
+
+TEST(PersistentOrderTree, MatchesStdMapUnderRandomOps)
+{
+    std::mt19937_64 rng(0xC0FFEEu);  // fixed seed → deterministic
+    std::map<RefKey, uint256> ref;
+    OrderTreePtr tree;
+
+    // A small set of dirRoots (quality levels) so levels hold multiple offers,
+    // exercising within-level ordering and the dirRoot-range delete search.
+    constexpr std::uint64_t kDirs = 8;
+    std::uint64_t seqCounter = 0;
+    std::uint64_t offerCounter = 0;
+    std::vector<RefKey> live;
+
+    for (int op = 0; op < 4000; ++op)
+    {
+        bool const doInsert = live.empty() || (rng() % 100) < 60;
+        if (doInsert)
+        {
+            uint256 const dir = u256(rng() % kDirs);
+            std::uint64_t const seq = ++seqCounter;  // unique → unique key
+            uint256 const off = u256(1'000'000 + (++offerCounter));
+            RefKey const key{dir, seq};
+            ref.emplace(key, off);
+            tree = otInsert(tree, dir, seq, off);
+            live.push_back(key);
+        }
+        else
+        {
+            // Delete a random live key by (dirRoot, offerKey) lookup, exactly
+            // like OrderBookIndex::deleteOffer does.
+            auto const idx = rng() % live.size();
+            RefKey const key = live[idx];
+            uint256 const off = ref.at(key);
+
+            auto const foundSeq = otFindSeq(tree, key.first, off);
+            ASSERT_TRUE(foundSeq.has_value());
+            EXPECT_EQ(*foundSeq, key.second);
+
+            tree = otDelete(tree, key.first, *foundSeq);
+            ref.erase(key);
+            live[idx] = live.back();
+            live.pop_back();
+        }
+
+        // Inorder equivalence after every op.
+        ASSERT_EQ(treeInorder(tree), refInorder(ref));
+        // Size field integrity + element count.
+        EXPECT_EQ(otSize(tree), ref.size());
+        checkSize(tree);
+        // first == reference begin's offer.
+        if (ref.empty())
+            EXPECT_FALSE(otFirst(tree).has_value());
+        else
+            EXPECT_EQ(otFirst(tree), ref.begin()->second);
+    }
+
+    // Weight-balanced height stays logarithmic (loose bound).
+    auto const n = otSize(tree);
+    if (n > 0)
+        EXPECT_LE(height(tree), 3 * (static_cast<int>(std::log2(n)) + 1) + 3);
+}
+
+TEST(PersistentOrderTree, StructuralSharingImmutability)
+{
+    OrderTreePtr base;
+    for (std::uint64_t i = 0; i < 200; ++i)
+        base = otInsert(base, u256(i % 4), i + 1, u256(10'000 + i));
+
+    auto const before = treeInorder(base);
+
+    // Mutate copies; the captured `base` must be unaffected (immutable nodes).
+    auto inserted = otInsert(base, u256(2), 99'999, u256(42));
+    auto deleted = otDelete(base, u256(0), 1);
+
+    EXPECT_EQ(treeInorder(base), before);             // base unchanged by insert
+    EXPECT_EQ(otSize(inserted), otSize(base) + 1);    // derived tree grew
+    EXPECT_EQ(otSize(deleted), otSize(base) - 1);     // derived tree shrank
+    EXPECT_EQ(treeInorder(base), before);             // base unchanged by delete
+}
+
+}  // namespace xrpl::detail

src/tests/libxrpl/ledger/TopOfBookCache.cpp

@@ -0,0 +1,200 @@
+#include <xrpl/ledger/TopOfBookCache.h>
+
+#include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/Asset.h>
+#include <xrpl/protocol/Book.h>
+#include <xrpl/protocol/Indexes.h>
+#include <xrpl/protocol/Issue.h>
+#include <xrpl/protocol/UintTypes.h>
+
+#include <gtest/gtest.h>
+
+#include <optional>
+
+namespace xrpl::test {
+
+namespace {
+
+// Construct a synthetic-but-consistent IOU book. The currency byte
+// distinguishes books for cache lookups; pairs are XRP <-> <currency>.
+Book
+makeIOUBook(std::uint8_t tag)
+{
+    Currency c{};
+    c.data()[19] = tag;
+    AccountID issuer{};
+    issuer.data()[19] = tag;
+    Issue const inIssue{c, issuer};
+    return Book{Asset{inIssue}, Asset{Issue{xrpCurrency(), xrpAccount()}}, std::nullopt};
+}
+
+// Derive the directory keylet (first-page key) for a given book at a given
+// quality rate. Two distinct rates produce two distinct, prefix-comparable
+// keys for the same book.
+uint256
+dirKey(Book const& book, std::uint64_t rate)
+{
+    return keylet::quality(keylet::kBook(book), rate).key;
+}
+
+}  // namespace
+
+TEST(TopOfBookCache, EmptyCacheMisses)
+{
+    TopOfBookCache cache;
+    EXPECT_FALSE(cache.get(makeIOUBook(1)).has_value());
+    EXPECT_EQ(cache.size(), 0u);
+    EXPECT_EQ(cache.hits(), 0u);
+    EXPECT_EQ(cache.misses(), 1u);
+}
+
+TEST(TopOfBookCache, RecordThenHit)
+{
+    TopOfBookCache cache;
+    Book const book = makeIOUBook(2);
+    uint256 const key = dirKey(book, 1'000'000u);
+
+    cache.record(book, key, /*seq=*/42);
+
+    auto const got = cache.get(book);
+    ASSERT_TRUE(got.has_value());
+    EXPECT_EQ(got->firstPageKey, key);
+    EXPECT_EQ(got->bestQuality, getQuality(key));
+    EXPECT_EQ(got->asOfLedger, 42u);
+    EXPECT_EQ(cache.hits(), 1u);
+    EXPECT_EQ(cache.misses(), 0u);
+}
+
+TEST(TopOfBookCache, OnOfferInsertBetterReplacesTop)
+{
+    TopOfBookCache cache;
+    Book const book = makeIOUBook(3);
+    uint256 const worse = dirKey(book, 2'000'000u);
+    uint256 const better = dirKey(book, 1'000'000u);
+    // Higher rate keys sort higher (worse quality). Sanity check.
+    ASSERT_LT(better, worse);
+
+    cache.record(book, worse, 1);
+    cache.onOfferInsert(book, better, 2);
+
+    auto const got = cache.get(book);
+    ASSERT_TRUE(got.has_value());
+    EXPECT_EQ(got->firstPageKey, better);
+    EXPECT_EQ(got->asOfLedger, 2u);
+}
+
+TEST(TopOfBookCache, OnOfferInsertSameLeavesTop)
+{
+    TopOfBookCache cache;
+    Book const book = makeIOUBook(4);
+    uint256 const key = dirKey(book, 1'000'000u);
+
+    cache.record(book, key, 5);
+    cache.onOfferInsert(book, key, 6);
+
+    auto const got = cache.get(book);
+    ASSERT_TRUE(got.has_value());
+    EXPECT_EQ(got->firstPageKey, key);
+    // asOfLedger preserved — same-quality insert is a no-op.
+    EXPECT_EQ(got->asOfLedger, 5u);
+}
+
+TEST(TopOfBookCache, OnOfferInsertWorseLeavesTop)
+{
+    TopOfBookCache cache;
+    Book const book = makeIOUBook(5);
+    uint256 const best = dirKey(book, 1'000'000u);
+    uint256 const worse = dirKey(book, 3'000'000u);
+    ASSERT_LT(best, worse);
+
+    cache.record(book, best, 5);
+    cache.onOfferInsert(book, worse, 9);
+
+    auto const got = cache.get(book);
+    ASSERT_TRUE(got.has_value());
+    EXPECT_EQ(got->firstPageKey, best);
+    EXPECT_EQ(got->asOfLedger, 5u);
+}
+
+TEST(TopOfBookCache, OnOfferInsertWithoutEntryIsNoOp)
+{
+    TopOfBookCache cache;
+    Book const book = makeIOUBook(6);
+    uint256 const key = dirKey(book, 1'000u);
+
+    // No prior entry — we don't speculatively populate.
+    cache.onOfferInsert(book, key, 1);
+    EXPECT_FALSE(cache.get(book).has_value());
+}
+
+TEST(TopOfBookCache, OnOfferDeleteOfTopInvalidates)
+{
+    TopOfBookCache cache;
+    Book const book = makeIOUBook(7);
+    uint256 const top = dirKey(book, 1'000u);
+
+    cache.record(book, top, 1);
+    cache.onOfferDelete(book, top);
+
+    EXPECT_FALSE(cache.get(book).has_value());
+    EXPECT_EQ(cache.invalidations(), 1u);
+}
+
+TEST(TopOfBookCache, OnOfferDeleteOfOtherPageLeavesTop)
+{
+    TopOfBookCache cache;
+    Book const book = makeIOUBook(8);
+    uint256 const top = dirKey(book, 1'000u);
+    uint256 const worsePage = dirKey(book, 4'000u);
+
+    cache.record(book, top, 1);
+    cache.onOfferDelete(book, worsePage);
+
+    auto const got = cache.get(book);
+    ASSERT_TRUE(got.has_value());
+    EXPECT_EQ(got->firstPageKey, top);
+    EXPECT_EQ(cache.invalidations(), 0u);
+}
+
+TEST(TopOfBookCache, DistinctBooksIndependent)
+{
+    TopOfBookCache cache;
+    Book const a = makeIOUBook(10);
+    Book const b = makeIOUBook(11);
+
+    cache.record(a, dirKey(a, 100u), 1);
+    cache.record(b, dirKey(b, 200u), 2);
+
+    EXPECT_TRUE(cache.get(a).has_value());
+    EXPECT_TRUE(cache.get(b).has_value());
+    EXPECT_EQ(cache.size(), 2u);
+
+    cache.onOfferDelete(a, dirKey(a, 100u));
+    EXPECT_FALSE(cache.get(a).has_value());
+    EXPECT_TRUE(cache.get(b).has_value());
+    EXPECT_EQ(cache.size(), 1u);
+}
+
+TEST(TopOfBookCache, InvalidateUnconditional)
+{
+    TopOfBookCache cache;
+    Book const book = makeIOUBook(12);
+    cache.record(book, dirKey(book, 100u), 1);
+    cache.invalidate(book);
+    EXPECT_FALSE(cache.get(book).has_value());
+    EXPECT_EQ(cache.invalidations(), 1u);
+    // Re-invalidating doesn't double-count.
+    cache.invalidate(book);
+    EXPECT_EQ(cache.invalidations(), 1u);
+}
+
+TEST(TopOfBookCache, KillSwitchToggleable)
+{
+    EXPECT_TRUE(TopOfBookCache::enabled());
+    TopOfBookCache::setEnabled(false);
+    EXPECT_FALSE(TopOfBookCache::enabled());
+    TopOfBookCache::setEnabled(true);
+    EXPECT_TRUE(TopOfBookCache::enabled());
+}
+
+}  // namespace xrpl::test

src/tests/libxrpl/shamap/ShaMapCostBreakdown.cpp

@@ -1,417 +0,0 @@
-// Plan 7 — Phase 0 cost-breakdown benchmark.
-//
-// Splits per-close SHAMap cost into three buckets so the Phase-1 vs Phase-2
-// build decision rests on measured numbers, not the (corrected) cost model in
-// tasks/plan-7-deferred-shamap.md. See tasks/plan-7-quantify.md for the why.
-//
-//   COW              = clone allocations on first touch  (already deduped today)
-//   traversal+dirty  = descent + setItem/setChild        (Phase-1 bulkApply target)
-//   serial hashing   = bottom-up unshare() at close       (Phase-2 parallel target)
-//
-// Gated behind the SHAMAP_BENCH env var so it never runs in normal CI.
-// Run with:  SHAMAP_BENCH=1 ./xrpl.test.shamap
-
-#include <helpers/TestFamily.h>
-
-#include <xrpl/basics/Slice.h>
-#include <xrpl/basics/base_uint.h>
-#include <xrpl/beast/utility/Journal.h>
-#include <xrpl/nodestore/NodeObject.h>
-#include <xrpl/shamap/SHAMap.h>
-#include <xrpl/shamap/SHAMapItem.h>
-#include <xrpl/shamap/SHAMapMissingNode.h>
-#include <xrpl/shamap/SHAMapTreeNode.h>
-
-#include <gtest/gtest.h>
-
-#include <algorithm>
-#include <array>
-#include <chrono>
-#include <cstdint>
-#include <cstdlib>
-#include <cstring>
-#include <iomanip>
-#include <iostream>
-#include <random>
-#include <vector>
-
-namespace xrpl::test {
-
-namespace {
-
-using Clock = std::chrono::steady_clock;
-using Ns = std::chrono::nanoseconds;
-
-[[nodiscard]] bool
-benchEnabled()
-{
-    char const* v = std::getenv("SHAMAP_BENCH");
-    return v != nullptr && v[0] != '\0' && v[0] != '0';
-}
-
-// Fill a uint256 with 32 pseudo-random bytes. State keys on mainnet are
-// SHA-512Half digests, i.e. uniformly distributed — random bytes give a
-// representative (uniform, depth ~log16 N) tree shape.
-[[nodiscard]] uint256
-randomKey(std::mt19937_64& rng)
-{
-    uint256 k;
-    auto* p = k.data();
-    for (std::size_t i = 0; i < k.size(); i += 8)
-    {
-        std::uint64_t const r = rng();
-        std::memcpy(p + i, &r, 8);
-    }
-    return k;
-}
-
-// A ~128-byte value whose leading bytes encode `salt`, so successive
-// replacements of the same key always differ (forcing setItem to re-dirty).
-constexpr std::size_t kValueBytes = 128;
-
-[[nodiscard]] boost::intrusive_ptr<SHAMapItem const>
-makeItem(uint256 const& key, std::uint64_t salt)
-{
-    std::array<std::uint8_t, kValueBytes> buf{};
-    std::memcpy(buf.data(), &salt, sizeof(salt));
-    std::memcpy(buf.data() + sizeof(salt), key.data(), 16);
-    return makeShamapitem(key, Slice(buf.data(), buf.size()));
-}
-
-struct Trial
-{
-    double tColdNs = 0;     // traversal + COW + dirty
-    double tWarmNs = 0;     // traversal + dirty (no COW)
-    double tHashNs = 0;     // serial bottom-up hash recompute (unshare)
-    double tHashParNs = 0;  // updateHashesParallel(kParWorkers)
-    int dirtyNodes = 0;     // nodes processed by the close-time recompute
-};
-
-constexpr int kParWorkers = 8;
-
-[[nodiscard]] double
-median(std::vector<double> v)
-{
-    std::sort(v.begin(), v.end());
-    return v.empty() ? 0.0 : v[v.size() / 2];
-}
-
-// One (N, M) measurement: build a base map of N entries, then on fresh
-// snapshots time M random replacements (cold), the same again warm, and the
-// close-time recompute.
-[[nodiscard]] Trial
-measure(std::size_t N, std::size_t M, int iters, beast::Journal j)
-{
-    std::mt19937_64 rng(0xC0FFEEull ^ (N * 1000003ull + M));
-
-    TestFamily family(j);
-    auto base = std::make_shared<SHAMap>(SHAMapType::STATE, family);
-    base->setUnbacked();
-
-    std::vector<uint256> keys;
-    keys.reserve(N);
-    for (std::size_t i = 0; i < N; ++i)
-    {
-        uint256 const k = randomKey(rng);
-        keys.push_back(k);
-        base->addItem(SHAMapNodeType::TnAccountState, makeItem(k, 0));
-    }
-    base->getHash();   // settle: all base nodes become shared (cowid 0)
-
-    std::uniform_int_distribution<std::size_t> pick(0, N - 1);
-
-    std::vector<double> cold, warm, hash, hashPar;
-    std::vector<int> dirty;
-    cold.reserve(iters);
-    warm.reserve(iters);
-    hash.reserve(iters);
-    hashPar.reserve(iters);
-    dirty.reserve(iters);
-
-    for (int it = 0; it < iters; ++it)
-    {
-        // Choose M distinct existing keys for this iteration.
-        std::vector<uint256> sel;
-        sel.reserve(M);
-        {
-            std::vector<bool> seen(N, false);
-            while (sel.size() < M)
-            {
-                std::size_t const idx = pick(rng);
-                if (!seen[idx])
-                {
-                    seen[idx] = true;
-                    sel.push_back(keys[idx]);
-                }
-            }
-        }
-
-        // --- COLD: traversal + COW + dirty, on a fresh snapshot ---
-        {
-            auto m = base->snapShot(/*isMutable=*/true);
-            auto const t0 = Clock::now();
-            std::uint64_t salt = 1;
-            for (auto const& k : sel)
-                m->updateGiveItem(
-                    SHAMapNodeType::TnAccountState, makeItem(k, salt++));
-            auto const t1 = Clock::now();
-            cold.push_back(
-                std::chrono::duration_cast<Ns>(t1 - t0).count());
-
-            // --- HASH: the serial recompute getHash() would drive ---
-            auto const h0 = Clock::now();
-            int const flushed = m->unshare();
-            auto const h1 = Clock::now();
-            hash.push_back(
-                std::chrono::duration_cast<Ns>(h1 - h0).count());
-            dirty.push_back(flushed);
-        }
-
-        // --- WARM: traversal + dirty, no COW (nodes already at cowid) ---
-        {
-            auto m = base->snapShot(/*isMutable=*/true);
-            std::uint64_t salt = 1;
-            for (auto const& k : sel)  // warm-up pass clones every path
-                m->updateGiveItem(
-                    SHAMapNodeType::TnAccountState, makeItem(k, salt++));
-
-            auto const w0 = Clock::now();
-            for (auto const& k : sel)  // timed pass: no clones
-                m->updateGiveItem(
-                    SHAMapNodeType::TnAccountState, makeItem(k, salt++));
-            auto const w1 = Clock::now();
-            warm.push_back(
-                std::chrono::duration_cast<Ns>(w1 - w0).count());
-        }
-
-        // --- PARALLEL HASH: the Phase-2 path on a fresh dirty snapshot ---
-        {
-            auto m = base->snapShot(/*isMutable=*/true);
-            std::uint64_t salt = 1;
-            for (auto const& k : sel)
-                m->updateGiveItem(
-                    SHAMapNodeType::TnAccountState, makeItem(k, salt++));
-
-            auto const p0 = Clock::now();
-            m->updateHashesParallel(kParWorkers);
-            auto const p1 = Clock::now();
-            hashPar.push_back(
-                std::chrono::duration_cast<Ns>(p1 - p0).count());
-        }
-    }
-
-    Trial r;
-    r.tColdNs = median(cold);
-    r.tWarmNs = median(warm);
-    r.tHashNs = median(hash);
-    r.tHashParNs = median(hashPar);
-    r.dirtyNodes = dirty.empty() ? 0 : dirty[dirty.size() / 2];
-    return r;
-}
-
-void
-printRow(std::size_t N, std::size_t M, Trial const& t)
-{
-    double const cowNs = std::max(0.0, t.tColdNs - t.tWarmNs);
-    auto us = [](double ns) { return ns / 1000.0; };
-    std::cout << std::fixed << std::setprecision(1) << "  " << std::setw(9) << N
-              << std::setw(7) << M << " |" << std::setw(9) << us(t.tWarmNs)
-              << std::setw(9) << us(cowNs) << std::setw(10) << us(t.tHashNs)
-              << std::setw(10) << us(t.tHashParNs) << " |" << std::setw(7)
-              << t.dirtyNodes << std::setw(9)
-              << (t.tHashParNs > 0 ? t.tHashNs / t.tHashParNs : 0.0) << "x\n";
-}
-
-// --- Backed-map flush split (Phase-3 sizing) ---------------------------------
-//
-// The real close path computes the state root via flushDirty() == hash + write
-// to the nodestore, in one serial walk — NOT via getHash(). updateHashesParallel
-// only parallelizes the hash half; the write half (writeNode → canonicalize is
-// serialized by the TreeNodeCache's single mutex) stays serial. So the
-// realizable close win is bounded by the hash fraction of flush.
-//
-// We measure on a BACKED map (memory nodestore, like production) per (N,M):
-//   flush      = flushDirty()      — today's serial close cost (hash + write)
-//   s.hash     = unshare()         — serial hash only
-//   p.hash     = updateHashesParallel
-// Projected Phase-3 close = flush - (s.hash - p.hash)   [replace serial hash
-// with parallel hash; write half unchanged].
-
-struct BackedTrial
-{
-    double flushNs = 0;
-    double sHashNs = 0;
-    double pHashNs = 0;
-    int dirtyNodes = 0;
-};
-
-[[nodiscard]] BackedTrial
-measureBacked(std::size_t N, std::size_t M, int iters, beast::Journal j)
-{
-    std::mt19937_64 rng(0xF1A7ull ^ (N * 1000003ull + M));
-
-    TestFamily family(j);  // backed: do NOT call setUnbacked()
-    auto base = std::make_shared<SHAMap>(SHAMapType::STATE, family);
-
-    std::vector<uint256> keys;
-    keys.reserve(N);
-    for (std::size_t i = 0; i < N; ++i)
-    {
-        uint256 const k = randomKey(rng);
-        keys.push_back(k);
-        base->addItem(SHAMapNodeType::TnAccountState, makeItem(k, 0));
-    }
-    base->flushDirty(NodeObjectType::AccountNode);  // settle + persist base
-
-    std::uniform_int_distribution<std::size_t> pick(0, N - 1);
-    std::vector<double> flush, sHash, pHash;
-    std::vector<int> dirty;
-
-    auto dirtySnapshot = [&](std::vector<uint256> const& sel) {
-        auto m = base->snapShot(/*isMutable=*/true);
-        std::uint64_t salt = 1;
-        for (auto const& k : sel)
-            m->updateGiveItem(SHAMapNodeType::TnAccountState, makeItem(k, salt++));
-        return m;
-    };
-
-    for (int it = 0; it < iters; ++it)
-    {
-        std::vector<uint256> sel;
-        sel.reserve(M);
-        std::vector<bool> seen(N, false);
-        while (sel.size() < M)
-        {
-            std::size_t const idx = pick(rng);
-            if (!seen[idx])
-            {
-                seen[idx] = true;
-                sel.push_back(keys[idx]);
-            }
-        }
-
-        {
-            auto m = dirtySnapshot(sel);
-            auto const t0 = Clock::now();
-            int const f = m->flushDirty(NodeObjectType::AccountNode);
-            flush.push_back(
-                std::chrono::duration_cast<Ns>(Clock::now() - t0).count());
-            dirty.push_back(f);
-        }
-        {
-            auto m = dirtySnapshot(sel);
-            auto const t0 = Clock::now();
-            m->unshare();
-            sHash.push_back(
-                std::chrono::duration_cast<Ns>(Clock::now() - t0).count());
-        }
-        {
-            auto m = dirtySnapshot(sel);
-            auto const t0 = Clock::now();
-            m->updateHashesParallel(kParWorkers);
-            pHash.push_back(
-                std::chrono::duration_cast<Ns>(Clock::now() - t0).count());
-        }
-    }
-
-    BackedTrial r;
-    r.flushNs = median(flush);
-    r.sHashNs = median(sHash);
-    r.pHashNs = median(pHash);
-    r.dirtyNodes = dirty.empty() ? 0 : dirty[dirty.size() / 2];
-    return r;
-}
-
-void
-printBackedRow(std::size_t N, std::size_t M, BackedTrial const& t)
-{
-    auto us = [](double ns) { return ns / 1000.0; };
-    double const projected = std::max(0.0, t.flushNs - (t.sHashNs - t.pHashNs));
-    double const hashFrac = t.flushNs > 0 ? t.sHashNs / t.flushNs : 0.0;
-    std::cout << std::fixed << std::setprecision(1) << "  " << std::setw(9) << N
-              << std::setw(7) << M << " |" << std::setw(10) << us(t.flushNs)
-              << std::setw(10) << us(t.sHashNs) << std::setw(10) << us(t.pHashNs)
-              << std::setw(11) << us(projected) << " |" << std::setw(7)
-              << std::setprecision(0) << (hashFrac * 100) << "%"
-              << std::setw(8) << std::setprecision(2)
-              << (projected > 0 ? t.flushNs / projected : 0.0) << "x\n";
-}
-
-}  // namespace
-
-TEST(ShaMapCostBreakdown, Report)
-{
-    if (!benchEnabled())
-        GTEST_SKIP() << "set SHAMAP_BENCH=1 to run the cost-breakdown benchmark";
-
-    beast::Journal const j{beast::Journal::getNullSink()};
-
-    std::cout << "\nPlan 7 Phase-0 — SHAMap per-close cost breakdown\n"
-              << "(median over iterations; times in microseconds for the whole "
-                 "batch of M replaces)\n\n"
-              << "        N      M | travrsl      COW    s.hash   p.hash |"
-                 "  dirty  speedup\n"
-              << "  -----------------------------------------------------------"
-                 "---------------\n";
-
-    struct Case
-    {
-        std::size_t N;
-        std::size_t M;
-        int iters;
-    };
-    std::array<Case, 4> const cases{
-        {{50'000, 1'000, 7},
-         {50'000, 3'000, 7},
-         {200'000, 1'000, 5},
-         {200'000, 3'000, 5}}};
-
-    for (auto const& c : cases)
-        printRow(c.N, c.M, measure(c.N, c.M, c.iters, j));
-
-    std::cout << "\n  travrsl = traversal+dirty (Phase-1 bulkApply ceiling)\n"
-                 "  COW     = clone allocs (already deduped by cowid today)\n"
-                 "  s.hash  = serial bottom-up recompute (status quo at close)\n"
-                 "  p.hash  = updateHashesParallel(" << kParWorkers
-              << ") (Phase-2)\n"
-                 "  speedup = s.hash / p.hash\n\n";
-}
-
-TEST(ShaMapCostBreakdown, BackedFlush)
-{
-    if (!benchEnabled())
-        GTEST_SKIP() << "set SHAMAP_BENCH=1 to run the cost-breakdown benchmark";
-
-    beast::Journal const j{beast::Journal::getNullSink()};
-
-    std::cout << "\nPlan 7 Phase-3 sizing — backed-map flush split\n"
-                 "(real close computes the root via flushDirty = hash + write; "
-                 "memory nodestore)\n\n"
-                 "        N      M |    flush    s.hash    p.hash  projected |"
-                 " hashfr speedup\n"
-                 "  --------------------------------------------------------"
-                 "------------------\n";
-
-    struct Case
-    {
-        std::size_t N;
-        std::size_t M;
-        int iters;
-    };
-    std::array<Case, 4> const cases{
-        {{50'000, 1'000, 5},
-         {50'000, 3'000, 5},
-         {200'000, 1'000, 4},
-         {200'000, 3'000, 4}}};
-
-    for (auto const& c : cases)
-        printBackedRow(c.N, c.M, measureBacked(c.N, c.M, c.iters, j));
-
-    std::cout << "\n  flush     = flushDirty() — today's serial close (hash+write)\n"
-                 "  projected = flush - (s.hash - p.hash)  [parallel hash, "
-                 "write half unchanged]\n"
-                 "  hashfr    = s.hash / flush  (hash share of close)\n"
-                 "  speedup   = flush / projected  (realizable Phase-3 close lift)\n\n";
-}
-
-}  // namespace xrpl::test

src/tests/libxrpl/shamap/UpdateHashesParallel.cpp

@@ -1,222 +0,0 @@
-// Plan 7 Phase-2 — differential test for SHAMap::updateHashesParallel.
-//
-// The contract: updateHashesParallel(W) must return the *byte-identical* root
-// hash that the serial getHash() produces, for any workload and any worker
-// count. We assert this against the production serial path on independently
-// mutated twin maps, over replace / insert / erase / mixed workloads, many
-// randomized seeds, and W in {1,2,4,8,16}.
-
-#include <helpers/TestFamily.h>
-
-#include <xrpl/basics/Slice.h>
-#include <xrpl/basics/base_uint.h>
-#include <xrpl/beast/utility/Journal.h>
-#include <xrpl/shamap/SHAMap.h>
-#include <xrpl/shamap/SHAMapItem.h>
-#include <xrpl/shamap/SHAMapMissingNode.h>
-#include <xrpl/shamap/SHAMapTreeNode.h>
-
-#include <gtest/gtest.h>
-
-#include <array>
-#include <cstdint>
-#include <cstring>
-#include <memory>
-#include <random>
-#include <vector>
-
-namespace xrpl::test {
-
-namespace {
-
-constexpr std::array<int, 5> kWorkerCounts{1, 2, 4, 8, 16};
-
-[[nodiscard]] uint256
-randomKey(std::mt19937_64& rng)
-{
-    uint256 k;
-    auto* p = k.data();
-    for (std::size_t i = 0; i < k.size(); i += 8)
-    {
-        std::uint64_t const r = rng();
-        std::memcpy(p + i, &r, 8);
-    }
-    return k;
-}
-
-[[nodiscard]] boost::intrusive_ptr<SHAMapItem const>
-makeItem(uint256 const& key, std::uint64_t salt)
-{
-    std::array<std::uint8_t, 96> buf{};
-    std::memcpy(buf.data(), &salt, sizeof(salt));
-    std::memcpy(buf.data() + sizeof(salt), key.data(), 16);
-    return makeShamapitem(key, Slice(buf.data(), buf.size()));
-}
-
-}  // namespace
-
-class UpdateHashesParallel : public ::testing::Test
-{
-protected:
-    TestFamily family_{beast::Journal{beast::Journal::getNullSink()}};
-
-    // A settled (hashes computed, nodes shared) base map of N random entries.
-    // Snapshots of it are the mutation targets — each snapshot clones on first
-    // touch, exactly like a live ledger inheriting its parent's state.
-    std::shared_ptr<SHAMap>
-    makeBase(std::size_t N, std::uint64_t seed, std::vector<uint256>& keysOut)
-    {
-        std::mt19937_64 rng(seed);
-        auto base = std::make_shared<SHAMap>(SHAMapType::STATE, family_);
-        base->setUnbacked();
-        keysOut.clear();
-        keysOut.reserve(N);
-        for (std::size_t i = 0; i < N; ++i)
-        {
-            uint256 const k = randomKey(rng);
-            keysOut.push_back(k);
-            base->addItem(SHAMapNodeType::TnAccountState, makeItem(k, 0));
-        }
-        base->getHash();  // settle
-        return base;
-    }
-
-    // Assert: for every worker count, a freshly mutated snapshot hashed in
-    // parallel equals an identically mutated snapshot hashed serially.
-    template <class Mutate>
-    void
-    expectParallelMatchesSerial(
-        std::shared_ptr<SHAMap> const& base,
-        Mutate&& mutate,
-        char const* label)
-    {
-        auto serialMap = base->snapShot(/*isMutable=*/true);
-        mutate(*serialMap);
-        SHAMapHash const serial = serialMap->getHash();
-
-        for (int w : kWorkerCounts)
-        {
-            auto parMap = base->snapShot(/*isMutable=*/true);
-            mutate(*parMap);
-            SHAMapHash const par = parMap->updateHashesParallel(w);
-            EXPECT_EQ(serial, par)
-                << label << " mismatch at workers=" << w;
-            // A second hash must agree with the cached result it left behind.
-            EXPECT_EQ(serial, parMap->getHash())
-                << label << " post-parallel getHash mismatch at workers=" << w;
-        }
-    }
-};
-
-TEST_F(UpdateHashesParallel, ReplaceWorkload)
-{
-    std::vector<uint256> keys;
-    auto base = makeBase(/*N=*/5000, /*seed=*/0x11, keys);
-    std::mt19937_64 rng(0xA1);
-    std::uniform_int_distribution<std::size_t> pick(0, keys.size() - 1);
-
-    for (std::size_t M : {1u, 50u, 500u, 2000u})
-    {
-        std::vector<uint256> sel;
-        for (std::size_t i = 0; i < M; ++i)
-            sel.push_back(keys[pick(rng)]);
-
-        expectParallelMatchesSerial(
-            base,
-            [&](SHAMap& m) {
-                std::uint64_t salt = 1;
-                for (auto const& k : sel)
-                    m.updateGiveItem(
-                        SHAMapNodeType::TnAccountState, makeItem(k, salt++));
-            },
-            "replace");
-    }
-}
-
-TEST_F(UpdateHashesParallel, InsertWorkload)
-{
-    std::vector<uint256> keys;
-    auto base = makeBase(/*N=*/3000, /*seed=*/0x22, keys);
-
-    for (std::size_t M : {1u, 100u, 1500u})
-    {
-        // Distinct fresh keys generated from a fixed seed so both twin
-        // snapshots receive the identical insert set.
-        std::mt19937_64 keyRng(0xBEEF + M);
-        std::vector<uint256> fresh;
-        for (std::size_t i = 0; i < M; ++i)
-            fresh.push_back(randomKey(keyRng));
-
-        expectParallelMatchesSerial(
-            base,
-            [&](SHAMap& m) {
-                std::uint64_t salt = 1;
-                for (auto const& k : fresh)
-                    m.addItem(
-                        SHAMapNodeType::TnAccountState, makeItem(k, salt++));
-            },
-            "insert");
-    }
-}
-
-TEST_F(UpdateHashesParallel, EraseWorkload)
-{
-    std::vector<uint256> keys;
-    auto base = makeBase(/*N=*/4000, /*seed=*/0x33, keys);
-
-    for (std::size_t M : {1u, 100u, 1000u})
-    {
-        // Erase the first M keys (a deterministic, distinct subset).
-        std::vector<uint256> sel(keys.begin(), keys.begin() + M);
-        expectParallelMatchesSerial(
-            base,
-            [&](SHAMap& m) {
-                for (auto const& k : sel)
-                    m.delItem(k);
-            },
-            "erase");
-    }
-}
-
-TEST_F(UpdateHashesParallel, MixedWorkload)
-{
-    std::vector<uint256> keys;
-    auto base = makeBase(/*N=*/6000, /*seed=*/0x44, keys);
-    std::mt19937_64 keyRng(0xC0DE);
-    std::vector<uint256> fresh;
-    for (int i = 0; i < 800; ++i)
-        fresh.push_back(randomKey(keyRng));
-
-    expectParallelMatchesSerial(
-        base,
-        [&](SHAMap& m) {
-            std::uint64_t salt = 1;
-            for (std::size_t i = 0; i < 800; ++i)
-            {
-                m.updateGiveItem(
-                    SHAMapNodeType::TnAccountState, makeItem(keys[i], salt++));
-                m.delItem(keys[3000 + i]);
-                m.addItem(
-                    SHAMapNodeType::TnAccountState, makeItem(fresh[i], salt++));
-            }
-        },
-        "mixed");
-}
-
-TEST_F(UpdateHashesParallel, NoMutationsAndEmpty)
-{
-    // Clean snapshot: no dirty nodes, must return the inherited root hash.
-    std::vector<uint256> keys;
-    auto base = makeBase(/*N=*/2000, /*seed=*/0x55, keys);
-    auto clean = base->snapShot(/*isMutable=*/true);
-    for (int w : kWorkerCounts)
-        EXPECT_EQ(base->getHash(), clean->updateHashesParallel(w));
-
-    // Empty map.
-    auto empty = std::make_shared<SHAMap>(SHAMapType::STATE, family_);
-    empty->setUnbacked();
-    for (int w : kWorkerCounts)
-        EXPECT_EQ(empty->getHash(), empty->updateHashesParallel(w));
-}
-
-}  // namespace xrpl::test

src/tests/libxrpl/shamap/main.cpp

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