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);