feat: CLOB Caching

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/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/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/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

@@ -39,6 +39,10 @@ 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/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/ledger/main.cpp

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